From 7f7811c79cc9f84d3b4139cb48854f30222fdea5 Mon Sep 17 00:00:00 2001 From: Siavash Safi Date: Tue, 1 Sep 2026 13:16:54 +0200 Subject: [PATCH] api: bound Connect admission and message resources Extend the procedure catalog with service, procedure, and stream metadata, move RPC admission ahead of decoding, and expose configurable resource limits with bounded lifecycle metrics. Preserve successful and specifically coded handler results when request contexts expire. Signed-off-by: Siavash Safi --- api/api.go | 37 +++- api/api_test.go | 3 +- api/connect/connect.go | 341 ++++++++++++++++++++++++++---- api/connect/connect_suite_test.go | 7 + api/connect/health_test.go | 2 +- api/connect/status_test.go | 147 ++++++++++++- app/app.go | 26 ++- app/options.go | 16 +- app/options_test.go | 3 + cmd/alertmanager/main.go | 32 ++- 10 files changed, 527 insertions(+), 87 deletions(-) diff --git a/api/api.go b/api/api.go index 9336241594..a58292dff2 100644 --- a/api/api.go +++ b/api/api.go @@ -75,7 +75,13 @@ type Options struct { // limit of GOMAXPROCS or 8, whichever is larger. Status code 503 is served // for GET requests that would exceed the concurrency limit; Connect calls // receive ResourceExhausted. - Concurrency int + Concurrency int + ConnectUnaryConcurrency int + ConnectStreamConcurrency int + ConnectUnaryTimeout time.Duration + ConnectReadMaxBytes int + ConnectSendMaxBytes int + ConnectMaxRequestBodyBytes int64 // Logger is used for logging, if nil, no logging will happen. Logger *slog.Logger // Registry is used to register Prometheus metrics. If nil, no metrics @@ -133,12 +139,31 @@ func New(opts Options) (*API, error) { if err != nil { return nil, err } - connect := apiconnect.NewAPI(apiconnect.Options{ - Peer: opts.Peer, - UnaryConcurrency: concurrency, - StreamConcurrency: concurrency, - UnaryTimeout: opts.Timeout, + unaryConcurrency := opts.ConnectUnaryConcurrency + if unaryConcurrency < 1 { + unaryConcurrency = concurrency + } + streamConcurrency := opts.ConnectStreamConcurrency + if streamConcurrency < 1 { + streamConcurrency = concurrency + } + unaryTimeout := opts.ConnectUnaryTimeout + if unaryTimeout == 0 { + unaryTimeout = opts.Timeout + } + connect, err := apiconnect.NewAPI(apiconnect.Options{ + Peer: opts.Peer, + Registerer: opts.Registry, + UnaryConcurrency: unaryConcurrency, + StreamConcurrency: streamConcurrency, + UnaryTimeout: unaryTimeout, + ReadMaxBytes: opts.ConnectReadMaxBytes, + SendMaxBytes: opts.ConnectSendMaxBytes, + MaxRequestBodyBytes: opts.ConnectMaxRequestBodyBytes, }) + if err != nil { + return nil, err + } requestsInFlight := prometheus.NewGauge(prometheus.GaugeOpts{ Name: "alertmanager_http_requests_in_flight", diff --git a/api/api_test.go b/api/api_test.go index 4a86faf6f6..ce6ffa08fc 100644 --- a/api/api_test.go +++ b/api/api_test.go @@ -85,7 +85,8 @@ func TestConcurrencyLimitHandler(t *testing.T) { func TestConnectProceduresRegistered(t *testing.T) { for _, routePrefix := range []string{"/", "/alertmanager"} { t.Run(routePrefix, func(t *testing.T) { - connectAPI := apiconnect.NewAPI(apiconnect.Options{}) + connectAPI, err := apiconnect.NewAPI(apiconnect.Options{}) + require.NoError(t, err) requestDuration := prometheus.NewHistogramVec( prometheus.HistogramOpts{Name: "test_registered_http_request_duration_seconds"}, []string{"handler", "method", "code"}, diff --git a/api/connect/connect.go b/api/connect/connect.go index 684694c9a0..edf7a92ad4 100644 --- a/api/connect/connect.go +++ b/api/connect/connect.go @@ -21,6 +21,7 @@ package apiconnect import ( "context" "errors" + "fmt" "net/http" "runtime" "sync/atomic" @@ -29,6 +30,7 @@ import ( "connectrpc.com/connect" "connectrpc.com/grpchealth" "connectrpc.com/grpcreflect" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/alertmanager/api/status/v3alpha/statusv3alphaconnect" "github.com/prometheus/alertmanager/cluster" @@ -37,14 +39,21 @@ import ( // Options configures the Connect API. type Options struct { - Peer cluster.ClusterPeer - UnaryConcurrency int - StreamConcurrency int - UnaryTimeout time.Duration + Peer cluster.ClusterPeer + Registerer prometheus.Registerer + UnaryConcurrency int + StreamConcurrency int + UnaryTimeout time.Duration + ReadMaxBytes int + SendMaxBytes int + MaxRequestBodyBytes int64 } type procedureDescriptor struct { - path string + path string + service string + procedure string + streamType connect.StreamType } type serviceDescriptor struct { @@ -53,6 +62,62 @@ type serviceDescriptor struct { handler func(...connect.HandlerOption) (string, http.Handler) } +type rpcMetrics struct { + unaryInFlight *prometheus.GaugeVec + unaryRejected *prometheus.CounterVec + unaryDuration *prometheus.HistogramVec + unaryDeadlines *prometheus.CounterVec + streamsActive *prometheus.GaugeVec + streamsRejected *prometheus.CounterVec +} + +func newRPCMetrics(reg prometheus.Registerer) (*rpcMetrics, error) { + labels := []string{"service", "procedure"} + outcomeLabels := []string{"service", "procedure", "outcome"} + metrics := &rpcMetrics{ + unaryInFlight: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "alertmanager_api_connect_unary_requests_in_flight", + Help: "Current number of admitted Connect unary RPCs.", + }, labels), + unaryRejected: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "alertmanager_api_connect_unary_admission_rejections_total", + Help: "Total number of Connect unary RPCs rejected by admission control.", + }, labels), + unaryDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "alertmanager_api_connect_unary_request_duration_seconds", + Help: "Duration of Connect unary RPCs.", + Buckets: prometheus.DefBuckets, + }, outcomeLabels), + unaryDeadlines: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "alertmanager_api_connect_unary_deadline_exceeded_total", + Help: "Total number of Connect unary RPCs that exceeded their deadline.", + }, labels), + streamsActive: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "alertmanager_api_connect_streams_active", + Help: "Current number of admitted Connect streams.", + }, labels), + streamsRejected: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "alertmanager_api_connect_stream_admission_rejections_total", + Help: "Total number of Connect streams rejected by admission control.", + }, labels), + } + if reg != nil { + for _, collector := range []prometheus.Collector{ + metrics.unaryInFlight, + metrics.unaryRejected, + metrics.unaryDuration, + metrics.unaryDeadlines, + metrics.streamsActive, + metrics.streamsRejected, + } { + if err := reg.Register(collector); err != nil { + return nil, fmt.Errorf("register Connect API metrics: %w", err) + } + } + } + return metrics, nil +} + // API implements the ConnectRPC service handlers for the Connect API. type API struct { peer cluster.ClusterPeer @@ -61,25 +126,28 @@ type API struct { peerSnapshotSem chan struct{} services []serviceDescriptor procedures map[string]procedureDescriptor + readMaxBytes int + sendMaxBytes int + maxRequestBytes int64 configSnapshot atomic.Pointer[string] } // NewAPI returns a new Connect API handler. Peer may be nil when clustering // is disabled. -func NewAPI(opts Options) *API { - unaryConcurrency := defaultConcurrency(opts.UnaryConcurrency) - streamConcurrency := defaultConcurrency(opts.StreamConcurrency) +func NewAPI(opts Options) (*API, error) { + metrics, err := newRPCMetrics(opts.Registerer) + if err != nil { + return nil, err + } api := &API{ peer: opts.Peer, uptime: time.Now(), peerSnapshotSem: make(chan struct{}, 1), procedures: make(map[string]procedureDescriptor), - admission: &admissionInterceptor{ - unary: make(chan struct{}, unaryConcurrency), - streams: make(chan struct{}, streamConcurrency), - unaryTimeout: opts.UnaryTimeout, - }, + readMaxBytes: opts.ReadMaxBytes, + sendMaxBytes: opts.SendMaxBytes, + maxRequestBytes: opts.MaxRequestBodyBytes, } api.services = api.serviceDescriptors() for _, service := range api.services { @@ -87,7 +155,14 @@ func NewAPI(opts Options) *API { api.procedures[procedure.path] = procedure } } - return api + api.admission = &admissionInterceptor{ + unary: make(chan struct{}, defaultConcurrency(opts.UnaryConcurrency)), + streams: make(chan struct{}, defaultConcurrency(opts.StreamConcurrency)), + unaryTimeout: opts.UnaryTimeout, + procedures: api.procedures, + metrics: metrics, + } + return api, nil } func defaultConcurrency(concurrency int) int { @@ -97,15 +172,20 @@ func defaultConcurrency(concurrency int) int { return concurrency } -func procedure(service, name string) procedureDescriptor { - return procedureDescriptor{path: "/" + service + "/" + name} +func procedure(service, name string, streamType connect.StreamType) procedureDescriptor { + return procedureDescriptor{ + path: "/" + service + "/" + name, + service: service, + procedure: name, + streamType: streamType, + } } func (api *API) serviceDescriptors() []serviceDescriptor { status := serviceDescriptor{ name: statusv3alphaconnect.StatusServiceName, procedures: []procedureDescriptor{ - procedure(statusv3alphaconnect.StatusServiceName, "GetStatus"), + procedure(statusv3alphaconnect.StatusServiceName, "GetStatus", connect.StreamTypeUnary), }, handler: func(opts ...connect.HandlerOption) (string, http.Handler) { return statusv3alphaconnect.NewStatusServiceHandler(api, opts...) @@ -119,8 +199,8 @@ func (api *API) serviceDescriptors() []serviceDescriptor { { name: grpchealth.HealthV1ServiceName, procedures: []procedureDescriptor{ - procedure(grpchealth.HealthV1ServiceName, "Check"), - procedure(grpchealth.HealthV1ServiceName, "Watch"), + procedure(grpchealth.HealthV1ServiceName, "Check", connect.StreamTypeUnary), + procedure(grpchealth.HealthV1ServiceName, "Watch", connect.StreamTypeServer), }, handler: func(opts ...connect.HandlerOption) (string, http.Handler) { return grpchealth.NewHandler(checker, opts...) @@ -129,7 +209,7 @@ func (api *API) serviceDescriptors() []serviceDescriptor { { name: grpcreflect.ReflectV1ServiceName, procedures: []procedureDescriptor{ - procedure(grpcreflect.ReflectV1ServiceName, "ServerReflectionInfo"), + procedure(grpcreflect.ReflectV1ServiceName, "ServerReflectionInfo", connect.StreamTypeBidi), }, handler: func(opts ...connect.HandlerOption) (string, http.Handler) { return grpcreflect.NewHandlerV1(reflector, opts...) @@ -138,7 +218,7 @@ func (api *API) serviceDescriptors() []serviceDescriptor { { name: grpcreflect.ReflectV1AlphaServiceName, procedures: []procedureDescriptor{ - procedure(grpcreflect.ReflectV1AlphaServiceName, "ServerReflectionInfo"), + procedure(grpcreflect.ReflectV1AlphaServiceName, "ServerReflectionInfo", connect.StreamTypeBidi), }, handler: func(opts ...connect.HandlerOption) (string, http.Handler) { return grpcreflect.NewHandlerV1Alpha(reflector, opts...) @@ -147,39 +227,141 @@ func (api *API) serviceDescriptors() []serviceDescriptor { } } +type ( + admittedContextKey struct{} + observationContextKey struct{} + requestStartContextKey struct{} + responseControllerContextKey struct{} +) + // admissionInterceptor gives unary RPCs and streams independent capacity so // slow Connect clients cannot consume the API v2 GET request allowance. type admissionInterceptor struct { unary chan struct{} streams chan struct{} unaryTimeout time.Duration + procedures map[string]procedureDescriptor + metrics *rpcMetrics } -func (i *admissionInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc { - return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { +func (i *admissionInterceptor) descriptor(path string, streamType connect.StreamType) procedureDescriptor { + if procedure, ok := i.procedures[path]; ok { + return procedure + } + return procedureDescriptor{service: "unknown", procedure: "unknown", streamType: streamType} +} + +func (i *admissionInterceptor) enter(desc procedureDescriptor) (func(), error) { + labels := prometheus.Labels{"service": desc.service, "procedure": desc.procedure} + if desc.streamType == connect.StreamTypeUnary { select { case i.unary <- struct{}{}: - defer func() { <-i.unary }() + if i.metrics != nil { + i.metrics.unaryInFlight.With(labels).Inc() + } + return func() { + <-i.unary + if i.metrics != nil { + i.metrics.unaryInFlight.With(labels).Dec() + } + }, nil default: + if i.metrics != nil { + i.metrics.unaryRejected.With(labels).Inc() + } return nil, connect.NewError(connect.CodeResourceExhausted, errors.New("maximum concurrent unary RPCs reached")) } - if i.unaryTimeout > 0 { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, i.unaryTimeout) - defer cancel() + } + select { + case i.streams <- struct{}{}: + if i.metrics != nil { + i.metrics.streamsActive.With(labels).Inc() + } + return func() { + <-i.streams + if i.metrics != nil { + i.metrics.streamsActive.With(labels).Dec() + } + }, nil + default: + if i.metrics != nil { + i.metrics.streamsRejected.With(labels).Inc() + } + return nil, connect.NewError(connect.CodeResourceExhausted, errors.New("maximum concurrent streams reached")) + } +} + +func (i *admissionInterceptor) startTime(ctx context.Context) time.Time { + if started, ok := ctx.Value(requestStartContextKey{}).(time.Time); ok { + return started + } + return time.Now() +} + +func (i *admissionInterceptor) observe(desc procedureDescriptor, started time.Time, err error) { + if i.metrics == nil { + return + } + outcome := "ok" + if err != nil { + outcome = connect.CodeOf(err).String() + } + if desc.streamType == connect.StreamTypeUnary { + labels := prometheus.Labels{"service": desc.service, "procedure": desc.procedure, "outcome": outcome} + i.metrics.unaryDuration.With(labels).Observe(time.Since(started).Seconds()) + } +} + +func normalizeContextError(ctx context.Context, err error) error { + if err == nil { + return nil + } + if connect.CodeOf(err) != connect.CodeUnknown { + return err + } + cause := context.Cause(ctx) + if errors.Is(cause, context.DeadlineExceeded) { + return connect.NewError(connect.CodeDeadlineExceeded, context.DeadlineExceeded) + } + switch { + case errors.Is(err, context.DeadlineExceeded): + return connect.NewError(connect.CodeDeadlineExceeded, context.DeadlineExceeded) + case errors.Is(err, context.Canceled), errors.Is(cause, context.Canceled): + return connect.NewError(connect.CodeCanceled, context.Canceled) + default: + return err + } +} + +func (i *admissionInterceptor) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + desc := i.descriptor("", connect.StreamTypeUnary) + if req != nil { + desc = i.descriptor(req.Spec().Procedure, connect.StreamTypeUnary) + } + started := i.startTime(ctx) + if controller, ok := ctx.Value(responseControllerContextKey{}).(*http.ResponseController); ok { + _ = controller.SetReadDeadline(time.Time{}) + } + if _, admitted := ctx.Value(admittedContextKey{}).(struct{}); !admitted { + release, err := i.enter(desc) + if err != nil { + return nil, err + } + defer release() + if i.unaryTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, i.unaryTimeout) + defer cancel() + } } response, err := next(ctx, req) - if err == nil || connect.CodeOf(err) != connect.CodeUnknown { - return response, err - } - switch { - case errors.Is(err, context.DeadlineExceeded): - return nil, connect.NewError(connect.CodeDeadlineExceeded, err) - case errors.Is(err, context.Canceled): - return nil, connect.NewError(connect.CodeCanceled, err) - default: - return response, err + err = normalizeContextError(ctx, err) + i.observe(desc, started, err) + if observed, ok := ctx.Value(observationContextKey{}).(*atomic.Bool); ok { + observed.Store(true) } + return response, err } } @@ -189,13 +371,18 @@ func (i *admissionInterceptor) WrapStreamingClient(next connect.StreamingClientF func (i *admissionInterceptor) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc { return func(ctx context.Context, conn connect.StreamingHandlerConn) error { - select { - case i.streams <- struct{}{}: - defer func() { <-i.streams }() - default: - return connect.NewError(connect.CodeResourceExhausted, errors.New("maximum concurrent streams reached")) + desc := i.descriptor("", connect.StreamTypeBidi) + if conn != nil { + desc = i.descriptor(conn.Spec().Procedure, conn.Spec().StreamType) + } + if _, admitted := ctx.Value(admittedContextKey{}).(struct{}); !admitted { + release, err := i.enter(desc) + if err != nil { + return err + } + defer release() } - return next(ctx, conn) + return normalizeContextError(ctx, next(ctx, conn)) } } @@ -233,9 +420,71 @@ func (api *API) Procedures() []string { return procedures } +func (api *API) controlHandler(next http.Handler, errorWriter *connect.ErrorWriter) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + desc, ok := api.procedures[r.URL.Path] + if !ok { + next.ServeHTTP(w, r) + return + } + release, err := api.admission.enter(desc) + if err != nil { + _ = r.Body.Close() + _ = errorWriter.Write(w, r, err) + return + } + defer release() + + controller := http.NewResponseController(w) + started := time.Now() + observed := &atomic.Bool{} + ctx := context.WithValue(r.Context(), admittedContextKey{}, struct{}{}) + ctx = context.WithValue(ctx, observationContextKey{}, observed) + ctx = context.WithValue(ctx, requestStartContextKey{}, started) + ctx = context.WithValue(ctx, responseControllerContextKey{}, controller) + if desc.streamType == connect.StreamTypeUnary && api.admission.unaryTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, api.admission.unaryTimeout) + defer cancel() + if deadline, ok := ctx.Deadline(); ok { + _ = controller.SetReadDeadline(deadline) + } + } + defer func() { + if desc.streamType == connect.StreamTypeUnary && !observed.Load() { + err := normalizeContextError(ctx, errors.New("rpc ended before handler execution")) + api.admission.observe(desc, started, err) + } + }() + if desc.streamType == connect.StreamTypeUnary { + defer func() { + if errors.Is(context.Cause(ctx), context.DeadlineExceeded) { + api.admission.metrics.unaryDeadlines.With(prometheus.Labels{"service": desc.service, "procedure": desc.procedure}).Inc() + } + }() + } + defer func() { + if ctx.Err() == nil { + _ = controller.SetReadDeadline(time.Time{}) + } + }() + + request := r.WithContext(ctx) + handler := next + if desc.streamType == connect.StreamTypeUnary && api.maxRequestBytes > 0 { + handler = http.MaxBytesHandler(handler, api.maxRequestBytes) + } + handler.ServeHTTP(w, request) + }) +} + // buildHandler registers every ConnectRPC service on a fresh mux. func (api *API) buildHandler(opts ...connect.HandlerOption) http.Handler { - opts = append([]connect.HandlerOption{connect.WithInterceptors(api.admission)}, opts...) + opts = append(opts, + connect.WithReadMaxBytes(api.readMaxBytes), + connect.WithSendMaxBytes(api.sendMaxBytes), + connect.WithInterceptors(api.admission), + ) mux := http.NewServeMux() for _, service := range api.services { @@ -245,5 +494,5 @@ func (api *API) buildHandler(opts ...connect.HandlerOption) http.Handler { } mux.Handle(path, handler) } - return mux + return api.controlHandler(mux, connect.NewErrorWriter(opts...)) } diff --git a/api/connect/connect_suite_test.go b/api/connect/connect_suite_test.go index b12041a4b0..96bc497d0f 100644 --- a/api/connect/connect_suite_test.go +++ b/api/connect/connect_suite_test.go @@ -24,3 +24,10 @@ func TestConnectAPI(t *testing.T) { RegisterFailHandler(Fail) RunSpecs(t, "Connect API Suite") } + +func newTestAPI(opts Options) *API { + GinkgoHelper() + api, err := NewAPI(opts) + Expect(err).NotTo(HaveOccurred()) + return api +} diff --git a/api/connect/health_test.go b/api/connect/health_test.go index bd0090912b..2bba462e1c 100644 --- a/api/connect/health_test.go +++ b/api/connect/health_test.go @@ -33,7 +33,7 @@ import ( // protocol with JSON, which needs only an HTTP/1.1 client. var _ = Describe("gRPC health", func() { It("reports serving for the server and StatusService", func() { - api := NewAPI(Options{}) + api := newTestAPI(Options{}) api.Update(&config.Config{}) srv := httptest.NewServer(api.Handler()) diff --git a/api/connect/status_test.go b/api/connect/status_test.go index d5dbeb33f8..b1cd1edace 100644 --- a/api/connect/status_test.go +++ b/api/connect/status_test.go @@ -15,6 +15,7 @@ package apiconnect import ( "context" + "io" "net/http" "net/http/httptest" "sync" @@ -24,6 +25,7 @@ import ( "connectrpc.com/connect" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/common/version" statusv3alpha "github.com/prometheus/alertmanager/api/status/v3alpha" @@ -72,7 +74,7 @@ func (p *blockingPeer) unblock() { p.releaseOnce.Do(func() { close(p.release) }) var _ = Describe("StatusService", func() { It("returns status when clustering is disabled", func() { - api := NewAPI(Options{}) + api := newTestAPI(Options{}) api.Update(&config.Config{}) resp, err := api.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{})) @@ -102,7 +104,7 @@ var _ = Describe("StatusService", func() { }, } - api := NewAPI(Options{Peer: peer}) + api := newTestAPI(Options{Peer: peer}) api.Update(&config.Config{}) resp, err := api.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{})) @@ -126,7 +128,7 @@ var _ = Describe("StatusService", func() { } DeferCleanup(peer.unblock) - api := NewAPI(Options{Peer: peer}) + api := newTestAPI(Options{Peer: peer}) api.Update(&config.Config{}) statusDone := make(chan error, 1) @@ -154,7 +156,7 @@ var _ = Describe("StatusService", func() { entered: make(chan struct{}), release: make(chan struct{}), } - api := NewAPI(Options{Peer: peer, UnaryTimeout: 20 * time.Millisecond}) + api := newTestAPI(Options{Peer: peer, UnaryConcurrency: 1, UnaryTimeout: 20 * time.Millisecond}) api.Update(&config.Config{}) srv := httptest.NewServer(api.Handler()) @@ -188,7 +190,7 @@ var _ = Describe("StatusService", func() { // standard library's http.Protocols. DescribeTable("serves status over HTTP", func(wantMethod string, opts []connect.ClientOption) { - api := NewAPI(Options{}) + api := newTestAPI(Options{}) api.Update(&config.Config{}) methods := make(chan string, 1) @@ -228,11 +230,87 @@ var _ = Describe("StatusService", func() { Entry("gRPC-Web", http.MethodPost, []connect.ClientOption{connect.WithGRPCWeb()}), Entry("gRPC", http.MethodPost, []connect.ClientOption{connect.WithGRPC()}), ) + + It("rejects oversized incoming messages before handler execution", func() { + peer := &blockingPeer{entered: make(chan struct{}), release: make(chan struct{})} + api := newTestAPI(Options{Peer: peer, ReadMaxBytes: 1, MaxRequestBodyBytes: 1024}) + api.Update(&config.Config{}) + srv := httptest.NewServer(api.Handler()) + DeferCleanup(srv.Close) + DeferCleanup(peer.unblock) + + request := &statusv3alpha.GetStatusRequest{} + request.ProtoReflect().SetUnknown([]byte{0x08, 0x01}) + client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL) + _, err := client.GetStatus(context.Background(), connect.NewRequest(request)) + Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted)) + Consistently(peer.entered, 50*time.Millisecond).ShouldNot(BeClosed()) + }) + + It("rejects oversized unary request bodies", func() { + api := newTestAPI(Options{ReadMaxBytes: 1024, MaxRequestBodyBytes: 1}) + api.Update(&config.Config{}) + srv := httptest.NewServer(api.Handler()) + DeferCleanup(srv.Close) + + request := &statusv3alpha.GetStatusRequest{} + request.ProtoReflect().SetUnknown([]byte{0x08, 0x01}) + client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL) + _, err := client.GetStatus(context.Background(), connect.NewRequest(request)) + Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted)) + }) + + It("rejects oversized outgoing messages", func() { + api := newTestAPI(Options{SendMaxBytes: 1}) + api.Update(&config.Config{}) + srv := httptest.NewServer(api.Handler()) + DeferCleanup(srv.Close) + client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL) + + _, err := client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{})) + Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted)) + }) + + It("bounds slow unary uploads before handler execution", func() { + peer := &blockingPeer{entered: make(chan struct{}), release: make(chan struct{})} + api := newTestAPI(Options{Peer: peer, UnaryConcurrency: 1, UnaryTimeout: 500 * time.Millisecond}) + api.Update(&config.Config{}) + srv := httptest.NewServer(api.Handler()) + DeferCleanup(srv.Close) + DeferCleanup(peer.unblock) + + reader, writer := io.Pipe() + DeferCleanup(writer.Close) + req, err := http.NewRequest(http.MethodPost, srv.URL+statusv3alphaconnect.StatusServiceGetStatusProcedure, reader) + Expect(err).NotTo(HaveOccurred()) + req.Header.Set("Content-Type", "application/proto") + req.ContentLength = 1 + done := make(chan *http.Response, 1) + go func() { + resp, _ := srv.Client().Do(req) + done <- resp + }() + Eventually(func() int { return len(api.admission.unary) }).Should(Equal(1)) + + client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL) + _, err = client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{})) + Expect(connect.CodeOf(err)).To(Equal(connect.CodeResourceExhausted)) + + var resp *http.Response + Eventually(done, time.Second).Should(Receive(&resp)) + Expect(resp).NotTo(BeNil()) + Expect(resp.Body.Close()).To(Succeed()) + Expect(peer.calls.Load()).To(BeZero()) + + peer.unblock() + _, err = client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{})) + Expect(err).NotTo(HaveOccurred()) + }) }) var _ = Describe("Connect API", func() { It("pins registered procedures", func() { - Expect(NewAPI(Options{}).Procedures()).To(Equal([]string{ + Expect(newTestAPI(Options{}).Procedures()).To(Equal([]string{ "/status.v3alpha.StatusService/GetStatus", "/grpc.health.v1.Health/Check", "/grpc.health.v1.Health/Watch", @@ -240,15 +318,59 @@ var _ = Describe("Connect API", func() { "/grpc.reflection.v1alpha.ServerReflection/ServerReflectionInfo", })) }) + + It("allows unlimited message and request body sizes by default", func() { + api := newTestAPI(Options{}) + Expect(api.readMaxBytes).To(BeZero()) + Expect(api.sendMaxBytes).To(BeZero()) + Expect(api.maxRequestBytes).To(BeZero()) + }) + + It("returns metric registration errors", func() { + reg := prometheus.NewRegistry() + newTestAPI(Options{Registerer: reg}) + _, err := NewAPI(Options{Registerer: reg}) + Expect(err).To(HaveOccurred()) + }) + + It("registers bounded unary lifecycle metrics", func() { + reg := prometheus.NewRegistry() + api := newTestAPI(Options{Registerer: reg}) + api.Update(&config.Config{}) + srv := httptest.NewServer(api.Handler()) + DeferCleanup(srv.Close) + client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL) + + _, err := client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{})) + Expect(err).NotTo(HaveOccurred()) + families, err := reg.Gather() + Expect(err).NotTo(HaveOccurred()) + + var labels map[string]string + for _, family := range families { + if family.GetName() != "alertmanager_api_connect_unary_request_duration_seconds" { + continue + } + labels = map[string]string{} + for _, pair := range family.GetMetric()[0].GetLabel() { + labels[pair.GetName()] = pair.GetValue() + } + } + Expect(labels).To(Equal(map[string]string{ + "outcome": "ok", + "procedure": "GetStatus", + "service": statusv3alphaconnect.StatusServiceName, + })) + }) }) var _ = Describe("RPC admission", func() { It("defaults unary and stream concurrency independently", func() { - api := NewAPI(Options{UnaryConcurrency: 1}) + api := newTestAPI(Options{UnaryConcurrency: 1}) Expect(cap(api.admission.unary)).To(Equal(1)) Expect(cap(api.admission.streams)).To(BeNumerically(">=", 8)) - api = NewAPI(Options{StreamConcurrency: 1}) + api = newTestAPI(Options{StreamConcurrency: 1}) Expect(cap(api.admission.unary)).To(BeNumerically(">=", 8)) Expect(cap(api.admission.streams)).To(Equal(1)) }) @@ -323,6 +445,15 @@ var _ = Describe("RPC admission", func() { Expect(wrapped(context.Background(), nil)).To(Succeed()) }) + It("preserves successful and specifically coded results after context expiration", func() { + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(context.DeadlineExceeded) + + Expect(normalizeContextError(ctx, nil)).To(Succeed()) + specific := connect.NewError(connect.CodeInvalidArgument, context.Canceled) + Expect(normalizeContextError(ctx, specific)).To(BeIdenticalTo(specific)) + }) + It("sets configured unary deadlines", func() { admission := &admissionInterceptor{ unary: make(chan struct{}, 1), diff --git a/app/app.go b/app/app.go index 67e3ec80aa..01df329878 100644 --- a/app/app.go +++ b/app/app.go @@ -383,16 +383,22 @@ func (a *App) setup() error { } apih, err := api.New(api.Options{ - Alerts: alerts, - Silences: silences, - GroupMutedFunc: groupMarker.Muted, - Peer: clusterPeer, - Timeout: opts.HTTPTimeout, - Concurrency: opts.GetConcurrency, - Logger: logger.With("component", "api"), - Registry: reg, - RequestDuration: m.requestDuration, - GroupFunc: groupFn, + Alerts: alerts, + Silences: silences, + GroupMutedFunc: groupMarker.Muted, + Peer: clusterPeer, + Timeout: opts.HTTPTimeout, + Concurrency: opts.GetConcurrency, + ConnectUnaryConcurrency: opts.ConnectUnaryConcurrency, + ConnectStreamConcurrency: opts.ConnectStreamConcurrency, + ConnectUnaryTimeout: opts.ConnectUnaryTimeout, + ConnectReadMaxBytes: opts.ConnectReadMaxBytes, + ConnectSendMaxBytes: opts.ConnectSendMaxBytes, + ConnectMaxRequestBodyBytes: opts.ConnectMaxRequestBodyBytes, + Logger: logger.With("component", "api"), + Registry: reg, + RequestDuration: m.requestDuration, + GroupFunc: groupFn, }) if err != nil { return fmt.Errorf("failed to create API: %w", err) diff --git a/app/options.go b/app/options.go index aba831f8a3..d1fe90361e 100644 --- a/app/options.go +++ b/app/options.go @@ -64,11 +64,17 @@ type Options struct { DispatchStartDelay time.Duration // Web server. - WebConfig *web.FlagConfig - ExternalURL string - RoutePrefix string - GetConcurrency int - HTTPTimeout time.Duration + WebConfig *web.FlagConfig + ExternalURL string + RoutePrefix string + GetConcurrency int + HTTPTimeout time.Duration + ConnectUnaryConcurrency int + ConnectStreamConcurrency int + ConnectUnaryTimeout time.Duration + ConnectReadMaxBytes int + ConnectSendMaxBytes int + ConnectMaxRequestBodyBytes int64 // Cluster. ClusterBindAddr string diff --git a/app/options_test.go b/app/options_test.go index eb40a18127..aa255e8340 100644 --- a/app/options_test.go +++ b/app/options_test.go @@ -55,6 +55,9 @@ func TestOptions_Validate(t *testing.T) { base := valid() require.NoError(t, base.validate()) + require.Zero(t, base.ConnectReadMaxBytes) + require.Zero(t, base.ConnectSendMaxBytes) + require.Zero(t, base.ConnectMaxRequestBodyBytes) for _, tc := range []struct { name string diff --git a/cmd/alertmanager/main.go b/cmd/alertmanager/main.go index cd2afd6972..cd722ccec0 100644 --- a/cmd/alertmanager/main.go +++ b/cmd/alertmanager/main.go @@ -60,11 +60,17 @@ func run() int { dispatchMaintenanceInterval = kingpin.Flag("dispatch.maintenance-interval", "Interval between maintenance of aggregation groups in the dispatcher.").Default("30s").Duration() dispatchStartDelay = kingpin.Flag("dispatch.start-delay", "Minimum amount of time to wait before dispatching alerts. This option should be synced with value of --rules.alert.resend-delay on Prometheus.").Default("0s").Duration() - webConfig = webflag.AddFlags(kingpin.CommandLine, ":9093") - externalURL = kingpin.Flag("web.external-url", "The URL under which Alertmanager is externally reachable (for example, if Alertmanager is served via a reverse proxy). Used for generating relative and absolute links back to Alertmanager itself. If the URL has a path portion, it will be used to prefix all HTTP endpoints served by Alertmanager. If omitted, relevant URL components will be derived automatically.").String() - routePrefix = kingpin.Flag("web.route-prefix", "Prefix for the internal routes of web endpoints. Defaults to path of --web.external-url.").String() - getConcurrency = kingpin.Flag("web.get-concurrency", "Maximum number of GET requests processed concurrently. If negative or zero, the limit is GOMAXPROC or 8, whichever is larger.").Default("0").Int() - httpTimeout = kingpin.Flag("web.timeout", "Timeout for HTTP requests. If negative or zero, no timeout is set.").Default("0").Duration() + webConfig = webflag.AddFlags(kingpin.CommandLine, ":9093") + externalURL = kingpin.Flag("web.external-url", "The URL under which Alertmanager is externally reachable (for example, if Alertmanager is served via a reverse proxy). Used for generating relative and absolute links back to Alertmanager itself. If the URL has a path portion, it will be used to prefix all HTTP endpoints served by Alertmanager. If omitted, relevant URL components will be derived automatically.").String() + routePrefix = kingpin.Flag("web.route-prefix", "Prefix for the internal routes of web endpoints. Defaults to path of --web.external-url.").String() + getConcurrency = kingpin.Flag("web.get-concurrency", "Maximum number of GET requests processed concurrently. If negative or zero, the limit is GOMAXPROC or 8, whichever is larger.").Default("0").Int() + httpTimeout = kingpin.Flag("web.timeout", "Timeout for HTTP requests. If negative or zero, no timeout is set.").Default("0").Duration() + connectUnaryConcurrency = kingpin.Flag("api.connect.unary-concurrency", "Maximum number of Connect unary RPCs processed concurrently. Defaults to --web.get-concurrency.").Default("0").Int() + connectStreamConcurrency = kingpin.Flag("api.connect.stream-concurrency", "Maximum number of Connect streams processed concurrently. Defaults to --web.get-concurrency.").Default("0").Int() + connectUnaryTimeout = kingpin.Flag("api.connect.unary-timeout", "Timeout for Connect unary RPCs, including request reads. Defaults to --web.timeout.").Default("0").Duration() + connectReadMaxBytes = kingpin.Flag("api.connect.read-max-bytes", "Maximum size of each incoming Connect protobuf message. If zero or negative, no limit is set.").Default("0").Int() + connectSendMaxBytes = kingpin.Flag("api.connect.send-max-bytes", "Maximum size of each outgoing Connect protobuf message. If zero or negative, no limit is set.").Default("0").Int() + connectMaxRequestBodyBytes = kingpin.Flag("api.connect.max-request-body-bytes", "Maximum wire size of a Connect unary request body. If zero or negative, no limit is set.").Default("0").Int64() memlimitEnable = kingpin.Flag("auto-gomemlimit", "Automatically set GOMEMLIMIT to match Linux container or system memory limit"). Default("false").Bool() @@ -175,11 +181,17 @@ func run() int { DispatchMaintenanceInterval: *dispatchMaintenanceInterval, DispatchStartDelay: *dispatchStartDelay, - WebConfig: webConfig, - ExternalURL: *externalURL, - RoutePrefix: *routePrefix, - GetConcurrency: *getConcurrency, - HTTPTimeout: *httpTimeout, + WebConfig: webConfig, + ExternalURL: *externalURL, + RoutePrefix: *routePrefix, + GetConcurrency: *getConcurrency, + HTTPTimeout: *httpTimeout, + ConnectUnaryConcurrency: *connectUnaryConcurrency, + ConnectStreamConcurrency: *connectStreamConcurrency, + ConnectUnaryTimeout: *connectUnaryTimeout, + ConnectReadMaxBytes: *connectReadMaxBytes, + ConnectSendMaxBytes: *connectSendMaxBytes, + ConnectMaxRequestBodyBytes: *connectMaxRequestBodyBytes, ClusterBindAddr: *clusterBindAddr, ClusterAdvertiseAddr: *clusterAdvertiseAddr,