middleware

Documentation

Overview

Package middleware provides connection between listeners and handlers.

Index

• Variables
• type CommandMetrics
• type Handler
• type Metrics
  • func NewMetrics() *Metrics
  • func (m *Metrics) Collect(ch chan<- prometheus.Metric)
  • func (m *Metrics) Describe(ch chan<- *prometheus.Desc)
  • func (m *Metrics) GetResponses() map[string]map[string]map[string]CommandMetrics
• type Middleware
  • func New(opts *NewOpts) *Middleware
  • func (m *Middleware) Collect(ch chan<- prometheus.Metric)
  • func (m *Middleware) Describe(ch chan<- *prometheus.Desc)
  • func (m *Middleware) Handle(ctx context.Context, req *Request) (resp *Response)
  • func (m *Middleware) Run(ctx context.Context)
• type Mode
• type NewOpts
• type Request
  • func RequestDoc(doc wirebson.AnyDocument) (*Request, error)
  • func RequestWire(header *wire.MsgHeader, body wire.MsgBody) (*Request, error)
  • func (req *Request) Document() *wirebson.Document
  • func (req *Request) DocumentDeep() (*wirebson.Document, error)
  • func (req *Request) DocumentRaw() wirebson.RawDocument
  • func (req *Request) WireBody() wire.MsgBody
  • func (req *Request) WireHeader() *wire.MsgHeader
• type Response
  • func ResponseDoc(req *Request, doc wirebson.AnyDocument) (*Response, error)
  • func ResponseErr(req *Request, err *mongoerrors.Error) *Response
  • func ResponseWire(header *wire.MsgHeader, body wire.MsgBody) (*Response, error)
  • func (resp *Response) Document() *wirebson.Document
  • func (resp *Response) DocumentDeep() (*wirebson.Document, error)
  • func (resp *Response) DocumentRaw() wirebson.RawDocument
  • func (resp *Response) ErrorCode() mongoerrors.Code
  • func (resp *Response) ErrorName() string
  • func (resp *Response) MongoError() *mongoerrors.Error
  • func (resp *Response) OK() bool
  • func (resp *Response) WireBody() wire.MsgBody
  • func (resp *Response) WireHeader() *wire.MsgHeader

Variables

var AllModes = []string{
	string(NormalMode),
	string(ProxyMode),
	string(DiffNormalMode),
	string(DiffProxyMode),
}

AllModes includes all operation modes, with the first one being the default.

Types

type CommandMetrics

type CommandMetrics struct {
	Failures map[string]int // count by result, except "ok"
	Total    int            // both "ok" and failures
}

CommandMetrics represents command results metrics.

type Handler

type Handler interface {
	// Run runs the handler until ctx is canceled and all requests are processed.
	// When this method returns, the handler is fully stopped.
	//
	// It is handler's responsibility to wait for all in-progress requests to be processed,
	// and to return from the future calls to Handle with an error early.
	Run(ctx context.Context)

	// Handle processes a single request.
	//
	// The passed context is canceled when the client disconnects.
	// Canceling ctx should stop the processing of the request.
	// Handle should also exit early when ctx passed to Run is canceled.
	//
	// Response is a normal or error response produced by the handler.
	//
	// Error is returned when the handler cannot process the request;
	// for example, when connection with PostgreSQL or proxy is lost.
	// Returning an error generally means that the listener should
	// close the client connection without responding.
	// Error should not be [*mongoerrors.Error] or have that type in its chain.
	//
	// Exactly one of Response or error should be non-nil.
	Handle(ctx context.Context, req *Request) (resp *Response, err error)

	// Handler should expose its metrics, but not metrics of passed dependencies.
	prometheus.Collector
}

Handler is a common interface for [handler.Handler] and [proxy.Handler] (but not for Middleware).

type Metrics

type Metrics struct {
	// contains filtered or unexported fields
}

Metrics represents middleware Metrics.

func NewMetrics

func NewMetrics() *Metrics

NewMetrics creates new metrics.

func (*Metrics) Collect

func (m *Metrics) Collect(ch chan<- prometheus.Metric)

Collect implements prometheus.Collector.

func (*Metrics) Describe

func (m *Metrics) Describe(ch chan<- *prometheus.Desc)

Describe implements prometheus.Collector.

func (*Metrics) GetResponses

func (m *Metrics) GetResponses() map[string]map[string]map[string]CommandMetrics

GetResponses returns a map with all response metrics:

opcode (e.g. "OP_MSG", "OP_QUERY") -> command (e.g. "find", "aggregate") -> argument that caused an error (e.g. "sort", "$count (stage)"; or "unknown") -> result (e.g. "ok", "NotImplemented", "error", or "panic") -> count.

type Middleware

type Middleware struct {
	// contains filtered or unexported fields
}

Middleware connects listeners and handlers.

func New

func New(opts *NewOpts) *Middleware

New returns a new middleware. Middleware.Run must be called on the returned value.

func (*Middleware) Collect

func (m *Middleware) Collect(ch chan<- prometheus.Metric)

Collect implements prometheus.Collector.

func (*Middleware) Describe

func (m *Middleware) Describe(ch chan<- *prometheus.Desc)

Describe implements prometheus.Collector.

func (*Middleware) Handle

func (m *Middleware) Handle(ctx context.Context, req *Request) (resp *Response)

Handle implements Handler, except that it returns nil for unrecoverable errors.

func (*Middleware) Run

func (m *Middleware) Run(ctx context.Context)

Run implements Handler.

type Mode

type Mode string

Mode represents DocDB mode of operation.

const (
	// NormalMode only handles requests.
	NormalMode Mode = "normal"

	// ProxyMode only proxies requests to another wire protocol compatible service.
	ProxyMode Mode = "proxy"

	// DiffNormalMode both handles requests and proxies them, then logs the diff.
	// Only the DocDB response is sent to the client.
	DiffNormalMode Mode = "diff-normal"

	// DiffProxyMode both handles requests and proxies them, then logs the diff.
	// Only the proxy response is sent to the client.
	DiffProxyMode Mode = "diff-proxy"
)

type NewOpts

type NewOpts struct {
	Mode    Mode
	DocDB   Handler
	Proxy   Handler
	Metrics *Metrics
	L       *slog.Logger
}

NewOpts represents middleware configuration.

type Request

type Request struct {
	// contains filtered or unexported fields
}

Request represents an incoming command from the client.

It may be constructed from [*wire.MsgHeader]/[wire.MsgBody] (for the wire protocol listener) or from *wirebson.Document (for data API and MCP listeners). The other value is extracted/generated and cached during Request creation, because we need both:

• raw documents carried by [wire.MsgBody] are used by both DocumentDB and proxy handlers;
• (non-deeply) decoded documents are used by routing, metrics, etc.

func RequestDoc

func RequestDoc(doc wirebson.AnyDocument) (*Request, error)

RequestDoc creates a new request from the given document. Error is returned if it cannot be decoded.

If it is *wirebson.Document, it freezes it.

func RequestWire

func RequestWire(header *wire.MsgHeader, body wire.MsgBody) (*Request, error)

RequestWire creates a new request from the given wire protocol header and body. Error is returned if the body cannot be decoded.

func (*Request) Document

func (req *Request) Document() *wirebson.Document

Document returns the request document (only section 0 for OpMsg).

func (*Request) DocumentDeep

func (req *Request) DocumentDeep() (*wirebson.Document, error)

DocumentDeep returns the deeply decoded request document. Callers should use it instead of `resp.DocumentRaw().DecodeDeep()`.

func (*Request) DocumentRaw

func (req *Request) DocumentRaw() wirebson.RawDocument

DocumentRaw returns the raw request document (only section 0 for OpMsg).

func (*Request) WireBody

func (req *Request) WireBody() wire.MsgBody

WireBody returns the request body for the wire protocol.

func (*Request) WireHeader

func (req *Request) WireHeader() *wire.MsgHeader

WireHeader returns the request header for the wire protocol.

type Response

type Response struct {
	// contains filtered or unexported fields
}

Response is a normal or error response produced by the handler.

It may be constructed from [*wire.MsgHeader]/[wire.MsgBody] (for the proxy handler), from wirebson.AnyDocument (for the DocDB handler), or from *mongoerrors.Error (for both).

func ResponseDoc

func ResponseDoc(req *Request, doc wirebson.AnyDocument) (*Response, error)

ResponseDoc creates a new response from the given document. Error is returned if it cannot be decoded.

If it is *wirebson.Document, it freezes it.

func ResponseErr

func ResponseErr(req *Request, err *mongoerrors.Error) *Response

ResponseErr creates a new error response from the given *mongoerrors.Error.

func ResponseWire

func ResponseWire(header *wire.MsgHeader, body wire.MsgBody) (*Response, error)

ResponseWire creates a new response from the given wire protocol header and body. Error is returned if the body cannot be decoded.

func (*Response) Document

func (resp *Response) Document() *wirebson.Document

Document returns the response document (only section 0 for OpMsg).

func (*Response) DocumentDeep

func (resp *Response) DocumentDeep() (*wirebson.Document, error)

DocumentDeep returns the deeply decoded response document (only section 0 for OpMsg). Callers should use it instead of `resp.DocumentRaw().DecodeDeep()`.

func (*Response) DocumentRaw

func (resp *Response) DocumentRaw() wirebson.RawDocument

DocumentRaw returns the raw response document (only section 0 for OpMsg).

func (*Response) ErrorCode

func (resp *Response) ErrorCode() mongoerrors.Code

ErrorCode returns the error code of the response, or mongoerrors.ErrUnset (zero).

func (*Response) ErrorName

func (resp *Response) ErrorName() string

ErrorName returns the error name of the response, or an empty string.

func (*Response) MongoError

func (resp *Response) MongoError() *mongoerrors.Error

MongoError returns *mongoerrors.Error, if the response was created with ResponseErr.

func (*Response) OK

func (resp *Response) OK() bool

OK returns true if the response document contains the "ok" field with numeric value 1 (and, temporarily, boolean value true).

func (*Response) WireBody

func (resp *Response) WireBody() wire.MsgBody

WireBody returns the response body for the wire protocol.

func (*Response) WireHeader

func (resp *Response) WireHeader() *wire.MsgHeader

WireHeader returns the response header for the wire protocol.