diff --git a/api/api.go b/api/api.go index 9336241594..85fd9f6f79 100644 --- a/api/api.go +++ b/api/api.go @@ -75,7 +75,15 @@ 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 + ConnectStreamIdleTimeout time.Duration + ConnectStreamLifetime 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 +141,33 @@ 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, + StreamIdleTimeout: opts.ConnectStreamIdleTimeout, + StreamLifetime: opts.ConnectStreamLifetime, + 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", @@ -264,6 +293,13 @@ func (api *API) Update(cfg *config.Config, setAlertStatus func(ctx context.Conte } } +// Shutdown rejects new Connect RPCs and cancels active RPCs. +func (api *API) Shutdown() { + if api.connect != nil { + api.connect.Shutdown() + } +} + func (api *API) limitHandler(h http.Handler) http.Handler { limited := api.concurrencyLimitHandler(h) if api.timeout <= 0 { 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..71cccb5ba7 100644 --- a/api/connect/connect.go +++ b/api/connect/connect.go @@ -21,14 +21,17 @@ package apiconnect import ( "context" "errors" + "fmt" "net/http" "runtime" + "sync" "sync/atomic" "time" "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 +40,23 @@ 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 + StreamIdleTimeout time.Duration + StreamLifetime 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 +65,69 @@ 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 + streamLifetime *prometheus.HistogramVec +} + +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), + streamLifetime: prometheus.NewHistogramVec(prometheus.HistogramOpts{ + Name: "alertmanager_api_connect_stream_lifetime_seconds", + Help: "Lifetime of Connect streams.", + Buckets: prometheus.ExponentialBuckets(1, 2, 14), + }, outcomeLabels), + } + if reg != nil { + for _, collector := range []prometheus.Collector{ + metrics.unaryInFlight, + metrics.unaryRejected, + metrics.unaryDuration, + metrics.unaryDeadlines, + metrics.streamsActive, + metrics.streamsRejected, + metrics.streamLifetime, + } { + 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 +136,32 @@ type API struct { peerSnapshotSem chan struct{} services []serviceDescriptor procedures map[string]procedureDescriptor + readMaxBytes int + sendMaxBytes int + maxRequestBytes int64 + activeMutex sync.Mutex + activeRPCs map[*rpcLifecycle]struct{} + draining atomic.Bool 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, + activeRPCs: make(map[*rpcLifecycle]struct{}), } api.services = api.serviceDescriptors() for _, service := range api.services { @@ -87,7 +169,16 @@ 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, + streamIdleTimeout: opts.StreamIdleTimeout, + streamLifetime: opts.StreamLifetime, + procedures: api.procedures, + metrics: metrics, + } + return api, nil } func defaultConcurrency(concurrency int) int { @@ -97,15 +188,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 +215,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 +225,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 +234,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 +243,196 @@ func (api *API) serviceDescriptors() []serviceDescriptor { } } +type ( + admittedContextKey struct{} + requestStartContextKey struct{} + rpcLifecycleContextKey struct{} +) + +type rpcLifecycle struct { + cancel context.CancelCauseFunc + idleTimeout time.Duration + writeTimeout time.Duration + controller *http.ResponseController + mutex sync.Mutex + idleTimer *time.Timer + decoded atomic.Bool + observed atomic.Bool + stream bool +} + +func (l *rpcLifecycle) terminate(cause error) { + l.cancel(cause) + if l.controller != nil { + now := time.Now() + if l.stream || !l.decoded.Load() { + _ = l.controller.SetReadDeadline(now) + } + if l.stream { + _ = l.controller.SetWriteDeadline(now) + } else if l.writeTimeout > 0 { + _ = l.controller.SetWriteDeadline(now.Add(l.writeTimeout)) + } + } +} + +func (l *rpcLifecycle) expire() { + l.terminate(context.DeadlineExceeded) +} + +func (l *rpcLifecycle) touch() { + l.mutex.Lock() + defer l.mutex.Unlock() + if l.idleTimer != nil { + l.idleTimer.Stop() + l.idleTimer.Reset(l.idleTimeout) + } +} + +func (l *rpcLifecycle) stop() { + l.mutex.Lock() + defer l.mutex.Unlock() + if l.idleTimer != nil { + l.idleTimer.Stop() + } +} + // 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 + unary chan struct{} + streams chan struct{} + unaryTimeout time.Duration + streamIdleTimeout time.Duration + streamLifetime 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() + } + labels := prometheus.Labels{"service": desc.service, "procedure": desc.procedure, "outcome": outcome} + if desc.streamType == connect.StreamTypeUnary { + i.metrics.unaryDuration.With(labels).Observe(time.Since(started).Seconds()) + return + } + i.metrics.streamLifetime.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) + lifecycle, _ := ctx.Value(rpcLifecycleContextKey{}).(*rpcLifecycle) + if lifecycle != nil { + lifecycle.decoded.Store(true) + if lifecycle.controller != nil { + _ = lifecycle.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 lifecycle != nil { + lifecycle.observed.Store(true) } + return response, err } } @@ -187,15 +440,118 @@ func (i *admissionInterceptor) WrapStreamingClient(next connect.StreamingClientF return next } +type activityConn struct { + connect.StreamingHandlerConn + lifecycle *rpcLifecycle +} + +func (c *activityConn) Receive(message any) error { + err := c.StreamingHandlerConn.Receive(message) + if err == nil { + c.lifecycle.touch() + } + return err +} + +func (c *activityConn) Send(message any) error { + err := c.StreamingHandlerConn.Send(message) + if err == nil { + c.lifecycle.touch() + } + return err +} + +func (i *admissionInterceptor) unaryContext(ctx context.Context, controller *http.ResponseController) (context.Context, *rpcLifecycle, func()) { + unaryCtx, cancel := context.WithCancelCause(ctx) + lifecycle := &rpcLifecycle{cancel: cancel, writeTimeout: i.unaryTimeout, controller: controller} + var timeoutTimer *time.Timer + if i.unaryTimeout > 0 { + timeoutTimer = time.AfterFunc(i.unaryTimeout, lifecycle.expire) + } + return context.WithValue(unaryCtx, rpcLifecycleContextKey{}, lifecycle), lifecycle, func() { + if timeoutTimer != nil { + timeoutTimer.Stop() + } + cancel(context.Canceled) + } +} + +func (i *admissionInterceptor) streamContext(ctx context.Context, controller *http.ResponseController) (context.Context, *rpcLifecycle, func()) { + streamCtx, cancel := context.WithCancelCause(ctx) + lifecycle := &rpcLifecycle{cancel: cancel, idleTimeout: i.streamIdleTimeout, controller: controller, stream: true} + if lifecycle.idleTimeout > 0 { + lifecycle.idleTimer = time.AfterFunc(lifecycle.idleTimeout, lifecycle.expire) + } + var lifetimeTimer *time.Timer + if i.streamLifetime > 0 { + lifetimeTimer = time.AfterFunc(i.streamLifetime, lifecycle.expire) + } + return context.WithValue(streamCtx, rpcLifecycleContextKey{}, lifecycle), lifecycle, func() { + if lifetimeTimer != nil { + lifetimeTimer.Stop() + } + lifecycle.stop() + cancel(context.Canceled) + } +} + 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) + } + started := i.startTime(ctx) + if _, admitted := ctx.Value(admittedContextKey{}).(struct{}); !admitted { + release, err := i.enter(desc) + if err != nil { + return err + } + defer release() + var cleanup func() + ctx, _, cleanup = i.streamContext(ctx, nil) + defer cleanup() } - return next(ctx, conn) + lifecycle, _ := ctx.Value(rpcLifecycleContextKey{}).(*rpcLifecycle) + if lifecycle != nil && lifecycle.idleTimeout > 0 && conn != nil { + conn = &activityConn{StreamingHandlerConn: conn, lifecycle: lifecycle} + } + err := normalizeContextError(ctx, next(ctx, conn)) + i.observe(desc, started, err) + if lifecycle != nil { + lifecycle.observed.Store(true) + } + return err + } +} + +func (api *API) registerRPC(lifecycle *rpcLifecycle) bool { + api.activeMutex.Lock() + defer api.activeMutex.Unlock() + if api.draining.Load() { + return false + } + api.activeRPCs[lifecycle] = struct{}{} + return true +} + +func (api *API) unregisterRPC(lifecycle *rpcLifecycle) { + api.activeMutex.Lock() + delete(api.activeRPCs, lifecycle) + api.activeMutex.Unlock() +} + +// Shutdown rejects new RPCs and cancels active RPCs. +func (api *API) Shutdown() { + api.draining.Store(true) + api.activeMutex.Lock() + streams := make([]*rpcLifecycle, 0, len(api.activeRPCs)) + for lifecycle := range api.activeRPCs { + streams = append(streams, lifecycle) + } + api.activeMutex.Unlock() + for _, lifecycle := range streams { + lifecycle.terminate(context.Canceled) } } @@ -233,9 +589,80 @@ 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 + } + if api.draining.Load() { + _ = r.Body.Close() + _ = errorWriter.Write(w, r, connect.NewError(connect.CodeUnavailable, errors.New("connect API is shutting down"))) + 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() + ctx := context.WithValue(r.Context(), admittedContextKey{}, struct{}{}) + ctx = context.WithValue(ctx, requestStartContextKey{}, started) + var lifecycle *rpcLifecycle + var cleanup func() + if desc.streamType == connect.StreamTypeUnary { + ctx, lifecycle, cleanup = api.admission.unaryContext(ctx, controller) + } else { + ctx, lifecycle, cleanup = api.admission.streamContext(ctx, controller) + } + if !api.registerRPC(lifecycle) { + cleanup() + _ = r.Body.Close() + _ = errorWriter.Write(w, r, connect.NewError(connect.CodeUnavailable, errors.New("connect API is shutting down"))) + return + } + defer api.unregisterRPC(lifecycle) + defer cleanup() + defer func() { + if !lifecycle.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 +672,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..4ebecb47d6 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" @@ -70,9 +72,40 @@ func (p *blockingPeer) Peers() []cluster.ClusterMember { } func (p *blockingPeer) unblock() { p.releaseOnce.Do(func() { close(p.release) }) } +type fakeStreamingConn struct{} + +func (fakeStreamingConn) Spec() connect.Spec { + return connect.Spec{Procedure: "/grpc.reflection.v1.ServerReflection/ServerReflectionInfo", StreamType: connect.StreamTypeBidi} +} +func (fakeStreamingConn) Peer() connect.Peer { return connect.Peer{} } +func (fakeStreamingConn) Receive(any) error { return nil } +func (fakeStreamingConn) RequestHeader() http.Header { return http.Header{} } +func (fakeStreamingConn) Send(any) error { return nil } +func (fakeStreamingConn) ResponseHeader() http.Header { return http.Header{} } +func (fakeStreamingConn) ResponseTrailer() http.Header { return http.Header{} } + +type deadlineResponseWriter struct { + header http.Header + readDeadline time.Time + writeDeadline time.Time +} + +func (w *deadlineResponseWriter) Header() http.Header { return w.header } +func (*deadlineResponseWriter) Write(p []byte) (int, error) { return len(p), nil } +func (*deadlineResponseWriter) WriteHeader(int) {} +func (w *deadlineResponseWriter) SetReadDeadline(t time.Time) error { + w.readDeadline = t + return nil +} + +func (w *deadlineResponseWriter) SetWriteDeadline(t time.Time) error { + w.writeDeadline = t + return nil +} + 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 +135,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 +159,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) @@ -149,12 +182,34 @@ var _ = Describe("StatusService", func() { Expect(statusErr).NotTo(HaveOccurred()) }) + It("cancels active unary RPCs during shutdown", func() { + peer := &blockingPeer{entered: make(chan struct{}), release: make(chan struct{})} + api := newTestAPI(Options{Peer: peer, UnaryConcurrency: 1, UnaryTimeout: -1}) + api.Update(&config.Config{}) + srv := httptest.NewServer(api.Handler()) + DeferCleanup(srv.Close) + DeferCleanup(peer.unblock) + client := statusv3alphaconnect.NewStatusServiceClient(srv.Client(), srv.URL) + done := make(chan error, 1) + go func() { + _, err := client.GetStatus(context.Background(), connect.NewRequest(&statusv3alpha.GetStatusRequest{})) + done <- err + }() + Eventually(peer.entered).Should(BeClosed()) + + api.Shutdown() + var err error + Eventually(done).Should(Receive(&err)) + Expect(connect.CodeOf(err)).To(Equal(connect.CodeCanceled)) + Expect(api.admission.unary).To(HaveLen(0)) + }) + It("bounds peer snapshots when the unary deadline expires", func() { peer := &blockingPeer{ 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 +243,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 +283,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, 2*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 +371,65 @@ 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("allows streams without idle or lifetime limits by default", func() { + api := newTestAPI(Options{}) + Expect(api.admission.streamIdleTimeout).To(BeZero()) + Expect(api.admission.streamLifetime).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 +504,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), @@ -337,4 +527,89 @@ var _ = Describe("RPC admission", func() { _, err := wrapped(context.Background(), nil) Expect(connect.CodeOf(err)).To(Equal(connect.CodeDeadlineExceeded)) }) + + It("bounds writes when terminating decoded unary RPCs", func() { + ctx, cancel := context.WithCancelCause(context.Background()) + writer := &deadlineResponseWriter{header: http.Header{}} + lifecycle := &rpcLifecycle{ + cancel: cancel, + writeTimeout: time.Second, + controller: http.NewResponseController(writer), + } + lifecycle.decoded.Store(true) + + started := time.Now() + lifecycle.terminate(context.DeadlineExceeded) + Expect(writer.readDeadline).To(BeZero()) + Expect(writer.writeDeadline).To(BeTemporally(">", started)) + Expect(context.Cause(ctx)).To(MatchError(context.DeadlineExceeded)) + }) + + It("bounds idle streams before the first message is decoded", func() { + api := newTestAPI(Options{StreamConcurrency: 1, StreamIdleTimeout: 500 * time.Millisecond, StreamLifetime: 2 * time.Second}) + handler := api.Handler() + srv := httptest.NewUnstartedServer(handler) + serverProtocols := new(http.Protocols) + serverProtocols.SetUnencryptedHTTP2(true) + srv.Config.Protocols = serverProtocols + srv.Start() + DeferCleanup(srv.Close) + clientProtocols := new(http.Protocols) + clientProtocols.SetUnencryptedHTTP2(true) + transport := &http.Transport{Protocols: clientProtocols} + client := &http.Client{Transport: transport, Timeout: time.Second} + DeferCleanup(transport.CloseIdleConnections) + reader, writer := io.Pipe() + DeferCleanup(writer.Close) + req, err := http.NewRequest(http.MethodPost, srv.URL+"/grpc.reflection.v1.ServerReflection/ServerReflectionInfo", reader) + Expect(err).NotTo(HaveOccurred()) + req.Header.Set("Content-Type", "application/grpc") + done := make(chan struct{}) + go func() { + resp, _ := client.Do(req) + if resp != nil { + _ = resp.Body.Close() + } + close(done) + }() + _, err = writer.Write([]byte{0}) + Expect(err).NotTo(HaveOccurred()) + Eventually(func() int { return len(api.admission.streams) }).Should(Equal(1)) + Eventually(done, 2*time.Second).Should(BeClosed()) + Expect(api.admission.streams).To(HaveLen(0)) + }) + + It("releases stream capacity after lifetime expiration", func() { + api := newTestAPI(Options{StreamConcurrency: 1, StreamLifetime: 10 * time.Millisecond}) + wrapped := api.admission.WrapStreamingHandler(func(ctx context.Context, _ connect.StreamingHandlerConn) error { + <-ctx.Done() + return context.Cause(ctx) + }) + + Expect(connect.CodeOf(wrapped(context.Background(), nil))).To(Equal(connect.CodeDeadlineExceeded)) + Expect(connect.CodeOf(wrapped(context.Background(), nil))).To(Equal(connect.CodeDeadlineExceeded)) + }) + + It("releases stream capacity after cancellation", func() { + api := newTestAPI(Options{StreamConcurrency: 1}) + wrapped := api.admission.WrapStreamingHandler(func(ctx context.Context, _ connect.StreamingHandlerConn) error { + <-ctx.Done() + return context.Cause(ctx) + }) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + Expect(connect.CodeOf(wrapped(ctx, nil))).To(Equal(connect.CodeCanceled)) + }) + + It("releases stream capacity after idle expiration", func() { + api := newTestAPI(Options{StreamConcurrency: 1, StreamIdleTimeout: 10 * time.Millisecond, StreamLifetime: time.Second}) + wrapped := api.admission.WrapStreamingHandler(func(ctx context.Context, _ connect.StreamingHandlerConn) error { + <-ctx.Done() + return context.Cause(ctx) + }) + + Expect(connect.CodeOf(wrapped(context.Background(), fakeStreamingConn{}))).To(Equal(connect.CodeDeadlineExceeded)) + Expect(connect.CodeOf(wrapped(context.Background(), fakeStreamingConn{}))).To(Equal(connect.CodeDeadlineExceeded)) + }) }) diff --git a/app/app.go b/app/app.go index 67e3ec80aa..7e39dcd8c8 100644 --- a/app/app.go +++ b/app/app.go @@ -383,16 +383,24 @@ 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, + ConnectStreamIdleTimeout: opts.ConnectStreamIdleTimeout, + ConnectStreamLifetime: opts.ConnectStreamLifetime, + 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) @@ -526,6 +534,7 @@ func (a *App) setup() error { ReadHeaderTimeout: 10 * time.Second, IdleTimeout: 90 * time.Second, } + a.server.RegisterOnShutdown(apih.Shutdown) return nil } diff --git a/app/lifecycle.go b/app/lifecycle.go index 1be4aaa6cd..6d3c74773d 100644 --- a/app/lifecycle.go +++ b/app/lifecycle.go @@ -210,6 +210,9 @@ func (a *App) Stop(ctx context.Context) error { if err := a.server.Shutdown(shutdownCtx); err != nil { a.logger.Warn("graceful HTTP shutdown failed", "err", err) stopErr = err + if closeErr := a.server.Close(); closeErr != nil { + stopErr = errors.Join(stopErr, closeErr) + } } } // HTTP is fully drained; no new /-/reload requests can arrive. diff --git a/app/options.go b/app/options.go index aba831f8a3..f3ee2c5d1a 100644 --- a/app/options.go +++ b/app/options.go @@ -64,11 +64,19 @@ 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 + ConnectStreamIdleTimeout time.Duration + ConnectStreamLifetime 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..c8b5df9490 100644 --- a/app/options_test.go +++ b/app/options_test.go @@ -55,6 +55,11 @@ 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) + require.Zero(t, base.ConnectStreamIdleTimeout) + require.Zero(t, base.ConnectStreamLifetime) for _, tc := range []struct { name string diff --git a/cmd/alertmanager/main.go b/cmd/alertmanager/main.go index cd2afd6972..32926a94a8 100644 --- a/cmd/alertmanager/main.go +++ b/cmd/alertmanager/main.go @@ -60,11 +60,19 @@ 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() + connectStreamIdleTimeout = kingpin.Flag("api.connect.stream-idle-timeout", "Maximum time between messages on a Connect stream. If zero or negative, no idle timeout is set.").Default("0").Duration() + connectStreamLifetime = kingpin.Flag("api.connect.stream-lifetime", "Maximum lifetime of a Connect stream. If zero or negative, no lifetime limit is set.").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 +183,19 @@ 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, + ConnectStreamIdleTimeout: *connectStreamIdleTimeout, + ConnectStreamLifetime: *connectStreamLifetime, + ConnectReadMaxBytes: *connectReadMaxBytes, + ConnectSendMaxBytes: *connectSendMaxBytes, + ConnectMaxRequestBodyBytes: *connectMaxRequestBodyBytes, ClusterBindAddr: *clusterBindAddr, ClusterAdvertiseAddr: *clusterAdvertiseAddr, diff --git a/test/e2e/status_test.go b/test/e2e/status_test.go index a0d148631f..695f3f60ab 100644 --- a/test/e2e/status_test.go +++ b/test/e2e/status_test.go @@ -121,4 +121,30 @@ var _ = Describe("StatusService", func() { Entry("without a route prefix", ""), Entry("with a route prefix", "/alertmanager"), ) + + It("cancels active streams during shutdown", func() { + inst := startInstance("") + conn, err := grpc.NewClient(inst.app.Addr(), grpc.WithTransportCredentials(insecure.NewCredentials())) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(conn.Close) + stream, err := reflectionv1.NewServerReflectionClient(conn).ServerReflectionInfo(context.Background()) + Expect(err).NotTo(HaveOccurred()) + Expect(stream.Send(&reflectionv1.ServerReflectionRequest{ + MessageRequest: &reflectionv1.ServerReflectionRequest_ListServices{}, + })).To(Succeed()) + _, err = stream.Recv() + Expect(err).NotTo(HaveOccurred()) + + stopDone := make(chan error, 1) + go func() { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + stopDone <- inst.app.Stop(ctx) + }() + var stopErr error + Eventually(stopDone, 2*time.Second).Should(Receive(&stopErr)) + Expect(stopErr).NotTo(HaveOccurred()) + _, err = stream.Recv() + Expect(err).To(HaveOccurred()) + }) })