diff --git a/README.md b/README.md index 9c30662dde..acc4c9ca34 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,12 @@ The v2 `/status` endpoint would be `/api/v2/status`. If `--web.route-prefix` is prefixed with that as well, so `--web.route-prefix=/alertmanager/` would relate to `/alertmanager/api/v2/status`. +The experimental ConnectRPC API serves Connect and gRPC-Web under the route-prefix-aware `/api/` +path and native gRPC, health, and reflection at the server root. The listener accepts native gRPC +over exporter-toolkit TLS with HTTP/2 ALPN and over plaintext h2c. Because h2c provides neither +encryption nor peer authentication, expose a plaintext listener only on a trusted network or behind +a trusted TLS-terminating proxy; use `--web.config.file` to configure TLS for direct exposure. + ## amtool `amtool` is a cli tool for interacting with the Alertmanager API. It is bundled with all releases of Alertmanager. 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..70494ba617 100644 --- a/app/app.go +++ b/app/app.go @@ -107,6 +107,7 @@ type App struct { coordinator *config.Coordinator tracingMgr *tracing.Manager server *http.Server + servers []*http.Server listeners []net.Listener // webReload is the channel exposed by httpserver.Register for the @@ -383,16 +384,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) @@ -515,17 +524,26 @@ func (a *App) setup() error { mux := apih.Register(router, routePrefix) - protocols := new(http.Protocols) - protocols.SetHTTP1(true) - protocols.SetHTTP2(true) - protocols.SetUnencryptedHTTP2(true) - a.server = &http.Server{ - // Instrument all handlers with tracing. - Handler: tracing.Middleware(mux), - Protocols: protocols, - ReadHeaderTimeout: 10 * time.Second, - IdleTimeout: 90 * time.Second, + newServer := func() *http.Server { + protocols := new(http.Protocols) + protocols.SetHTTP1(true) + protocols.SetHTTP2(true) + protocols.SetUnencryptedHTTP2(true) + server := &http.Server{ + // Instrument all handlers with tracing. + Handler: tracing.Middleware(mux), + Protocols: protocols, + ReadHeaderTimeout: 10 * time.Second, + IdleTimeout: 90 * time.Second, + } + server.RegisterOnShutdown(apih.Shutdown) + return server + } + a.servers = make([]*http.Server, 0, len(a.listeners)) + for range a.listeners { + a.servers = append(a.servers, newServer()) } + a.server = a.servers[0] return nil } diff --git a/app/lifecycle.go b/app/lifecycle.go index 1be4aaa6cd..344a242a6f 100644 --- a/app/lifecycle.go +++ b/app/lifecycle.go @@ -17,6 +17,7 @@ import ( "context" "errors" "fmt" + "net" "net/http" "slices" "time" @@ -61,11 +62,24 @@ func (a *App) Start() error { // on an unbuffered channel has no receiver. go a.reloadRouter() + http2Enabled := configuredHTTP2(*a.opts.WebConfig.WebConfigFile) go func() { - err := web.ServeMultiple(a.listeners, a.server, a.opts.WebConfig, a.logger) - if err != nil && !errors.Is(err, http.ErrServerClosed) { - a.logger.Error("Listen error", "err", err) - a.serveErrc <- err + errCh := make(chan error, len(a.listeners)) + for idx, listener := range a.listeners { + server := a.servers[idx] + go func() { + errCh <- web.Serve(withReloadableTLSALPN([]net.Listener{listener}, server, http2Enabled)[0], server, a.opts.WebConfig, a.logger) + }() + } + var serveErr error + for range a.listeners { + if err := <-errCh; err != nil && !errors.Is(err, http.ErrServerClosed) { + serveErr = errors.Join(serveErr, err) + } + } + if serveErr != nil { + a.logger.Error("Listen error", "err", serveErr) + a.serveErrc <- serveErr } close(a.serveErrc) }() @@ -206,10 +220,24 @@ func (a *App) Stop(ctx context.Context) error { // reload router is still running at this point so any in-flight // /-/reload handler can complete its send/receive cycle and unblock // Shutdown. - if a.server != nil { - if err := a.server.Shutdown(shutdownCtx); err != nil { + servers := a.servers + if len(servers) == 0 && a.server != nil { + servers = []*http.Server{a.server} + } + shutdownErrCh := make(chan error, len(servers)) + for _, server := range servers { + go func() { + err := server.Shutdown(shutdownCtx) + if err != nil { + err = errors.Join(err, server.Close()) + } + shutdownErrCh <- err + }() + } + for range servers { + if err := <-shutdownErrCh; err != nil { a.logger.Warn("graceful HTTP shutdown failed", "err", err) - stopErr = err + stopErr = errors.Join(stopErr, err) } } // HTTP is fully drained; no new /-/reload requests can arrive. diff --git a/app/listen.go b/app/listen.go index c73b2a5362..aa8d64d3ba 100644 --- a/app/listen.go +++ b/app/listen.go @@ -17,13 +17,17 @@ import ( "errors" "fmt" "net" + "net/http" "net/url" + "os" "strconv" "strings" + "sync" "github.com/coreos/go-systemd/v22/activation" "github.com/mdlayher/vsock" "github.com/prometheus/exporter-toolkit/web" + "gopkg.in/yaml.v2" ) // listenAll eagerly binds every listener described by flags so that the @@ -94,3 +98,44 @@ func parseVsockPort(address string) (uint32, error) { } return uint32(port), nil } + +type reloadableTLSALPNListener struct { + net.Listener + server *http.Server + once *sync.Once + http2Enabled bool +} + +func (l reloadableTLSALPNListener) Accept() (net.Conn, error) { + l.once.Do(func() { + if !l.http2Enabled || l.server.TLSConfig == nil { + return + } + if _, ok := l.server.TLSNextProto["h2"]; ok { + l.server.TLSConfig.NextProtos = []string{"h2", "http/1.1"} + } + }) + return l.Listener.Accept() +} + +func configuredHTTP2(path string) bool { + config := struct { + HTTP struct { + HTTP2 *bool `yaml:"http2"` + } `yaml:"http_server_config"` + }{} + content, err := os.ReadFile(path) + if err != nil || yaml.Unmarshal(content, &config) != nil || config.HTTP.HTTP2 == nil { + return true + } + return *config.HTTP.HTTP2 +} + +func withReloadableTLSALPN(listeners []net.Listener, server *http.Server, http2Enabled bool) []net.Listener { + once := &sync.Once{} + wrapped := make([]net.Listener, 0, len(listeners)) + for _, listener := range listeners { + wrapped = append(wrapped, reloadableTLSALPNListener{Listener: listener, server: server, once: once, http2Enabled: http2Enabled}) + } + return wrapped +} 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/docs/https.md b/docs/https.md index 6fa3ee8115..f78b1230ec 100644 --- a/docs/https.md +++ b/docs/https.md @@ -80,7 +80,8 @@ tls_server_config: [ - ] ] http_server_config: - # Enable HTTP/2 support. Note that HTTP/2 is only supported with TLS. + # Enable HTTP/2 support over TLS. Alertmanager also accepts plaintext h2c, + # independently of this setting; expose plaintext only on a trusted network. # This can not be changed on the fly. [ http2: | default = true ] # List of headers that can be added to HTTP responses. diff --git a/test/e2e/harness_test.go b/test/e2e/harness_test.go index a0fe5d0ed2..cd3de91246 100644 --- a/test/e2e/harness_test.go +++ b/test/e2e/harness_test.go @@ -15,6 +15,9 @@ package e2e import ( "context" + "crypto/tls" + "crypto/x509" + "fmt" "net/http" "os" "path/filepath" @@ -46,13 +49,18 @@ type instance struct { baseURL string routePrefix string httpClient *http.Client - h2cClient *http.Client + rpcClient *http.Client + tlsConfig *tls.Config } // startInstance boots an Alertmanager and registers its teardown (and // temp-dir removal) via Ginkgo's DeferCleanup. -func startInstance(routePrefix string) *instance { +func startInstance(routePrefix string, tlsEnabled bool, enableHTTP2 ...bool) *instance { GinkgoHelper() + http2Enabled := true + if len(enableHTTP2) > 0 { + http2Enabled = enableHTTP2[0] + } dir, err := os.MkdirTemp("", "am-e2e-") Expect(err).NotTo(HaveOccurred()) @@ -71,6 +79,22 @@ func startInstance(routePrefix string) *instance { addrs := []string{"127.0.0.1:0"} systemd := false webCfg := "" + var clientTLS *tls.Config + if tlsEnabled { + cert, err := os.ReadFile(filepath.Join("..", "..", "cluster", "testdata", "certs", "node1.pem")) + Expect(err).NotTo(HaveOccurred()) + key, err := os.ReadFile(filepath.Join("..", "..", "cluster", "testdata", "certs", "node1-key.pem")) + Expect(err).NotTo(HaveOccurred()) + ca, err := os.ReadFile(filepath.Join("..", "..", "cluster", "testdata", "certs", "ca.pem")) + Expect(err).NotTo(HaveOccurred()) + Expect(os.WriteFile(filepath.Join(dir, "server.pem"), cert, 0o600)).To(Succeed()) + Expect(os.WriteFile(filepath.Join(dir, "server-key.pem"), key, 0o600)).To(Succeed()) + webCfg = filepath.Join(dir, "web.yml") + Expect(os.WriteFile(webCfg, []byte(fmt.Sprintf("tls_server_config:\n cert_file: %s\n key_file: %s\nhttp_server_config:\n http2: %t\n", filepath.Join(dir, "server.pem"), filepath.Join(dir, "server-key.pem"), http2Enabled)), 0o600)).To(Succeed()) + roots := x509.NewCertPool() + Expect(roots.AppendCertsFromPEM(ca)).To(BeTrue()) + clientTLS = &tls.Config{RootCAs: roots, MinVersion: tls.VersionTLS12} + } opts := app.DefaultOptions() opts.ConfigFile = configPath @@ -95,19 +119,33 @@ func startInstance(routePrefix string) *instance { Expect(a.Start()).To(Succeed()) client := &http.Client{Timeout: 5 * time.Second} - DeferCleanup(client.CloseIdleConnections) - protocols := new(http.Protocols) - protocols.SetUnencryptedHTTP2(true) - h2cTransport := &http.Transport{Protocols: protocols} - h2cClient := &http.Client{Transport: h2cTransport, Timeout: 5 * time.Second} - DeferCleanup(h2cTransport.CloseIdleConnections) + var rpcClient *http.Client + scheme := "http://" + if tlsEnabled { + protocols := new(http.Protocols) + protocols.SetHTTP1(true) + protocols.SetHTTP2(true) + transport := &http.Transport{TLSClientConfig: clientTLS, Protocols: protocols} + client = &http.Client{Transport: transport, Timeout: 5 * time.Second} + rpcClient = client + scheme = "https://" + DeferCleanup(transport.CloseIdleConnections) + } else { + DeferCleanup(client.CloseIdleConnections) + protocols := new(http.Protocols) + protocols.SetUnencryptedHTTP2(true) + transport := &http.Transport{Protocols: protocols} + rpcClient = &http.Client{Transport: transport, Timeout: 5 * time.Second} + DeferCleanup(transport.CloseIdleConnections) + } inst := &instance{ app: a, - baseURL: "http://" + a.Addr(), + baseURL: scheme + a.Addr(), routePrefix: routePrefix, httpClient: client, - h2cClient: h2cClient, + rpcClient: rpcClient, + tlsConfig: clientTLS, } inst.waitHealthy() return inst diff --git a/test/e2e/routing_test.go b/test/e2e/routing_test.go index ffa476cf53..ab06716164 100644 --- a/test/e2e/routing_test.go +++ b/test/e2e/routing_test.go @@ -22,16 +22,20 @@ import ( var _ = Describe("API routing", func() { DescribeTable("serves v1 and v2 alongside the Connect API", - func(routePrefix, path string, expectedStatus int) { - inst := startInstance(routePrefix) + func(routePrefix, path string, tlsEnabled bool, expectedStatus int) { + inst := startInstance(routePrefix, tlsEnabled) resp, err := inst.httpClient.Get(inst.webURL(path)) Expect(err).NotTo(HaveOccurred()) DeferCleanup(resp.Body.Close) Expect(resp.StatusCode).To(Equal(expectedStatus)) }, - Entry("v2 at the root", "", "/api/v2/status", http.StatusOK), - Entry("v1 at the root", "", "/api/v1/status", http.StatusGone), - Entry("v2 under a route prefix", "/alertmanager", "/api/v2/status", http.StatusOK), - Entry("v1 under a route prefix", "/alertmanager", "/api/v1/status", http.StatusGone), + Entry("v2 over h2c at the root", "", "/api/v2/status", false, http.StatusOK), + Entry("v1 over h2c at the root", "", "/api/v1/status", false, http.StatusGone), + Entry("v2 over h2c under a route prefix", "/alertmanager", "/api/v2/status", false, http.StatusOK), + Entry("v1 over h2c under a route prefix", "/alertmanager", "/api/v1/status", false, http.StatusGone), + Entry("v2 over TLS at the root", "", "/api/v2/status", true, http.StatusOK), + Entry("v1 over TLS at the root", "", "/api/v1/status", true, http.StatusGone), + Entry("v2 over TLS under a route prefix", "/alertmanager", "/api/v2/status", true, http.StatusOK), + Entry("v1 over TLS under a route prefix", "/alertmanager", "/api/v1/status", true, http.StatusGone), ) }) diff --git a/test/e2e/status_test.go b/test/e2e/status_test.go index a0d148631f..9079090459 100644 --- a/test/e2e/status_test.go +++ b/test/e2e/status_test.go @@ -15,7 +15,7 @@ package e2e import ( "context" - "strings" + "crypto/tls" "time" "connectrpc.com/connect" @@ -23,8 +23,10 @@ import ( . "github.com/onsi/gomega" "github.com/prometheus/common/version" "google.golang.org/grpc" + "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" healthv1 "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/peer" reflectionv1 "google.golang.org/grpc/reflection/grpc_reflection_v1" statusv3alpha "github.com/prometheus/alertmanager/api/status/v3alpha" @@ -33,12 +35,12 @@ import ( var _ = Describe("StatusService", func() { DescribeTable("GetStatus succeeds over supported transports", - func(routePrefix string, nativeGRPC bool, opts []connect.ClientOption) { - inst := startInstance(routePrefix) + func(routePrefix string, tlsEnabled, nativeGRPC bool, opts []connect.ClientOption) { + inst := startInstance(routePrefix, tlsEnabled) httpClient := connect.HTTPClient(inst.httpClient) basePath := inst.apiPath() if nativeGRPC { - httpClient = inst.h2cClient + httpClient = inst.rpcClient basePath = "" } client := inst.statusClient(httpClient, basePath, opts...) @@ -54,25 +56,33 @@ var _ = Describe("StatusService", func() { Expect(status.GetStartTime().AsTime()).NotTo(BeZero()) Expect(status.GetCluster().GetState()).To(Equal(statusv3alpha.ClusterStatus_STATE_DISABLED)) }, - Entry("Connect POST at the root prefix", "", false, []connect.ClientOption{}), - Entry("Connect HTTP GET at the root prefix", "", false, []connect.ClientOption{connect.WithHTTPGet()}), - Entry("gRPC-Web at the root prefix", "", false, []connect.ClientOption{connect.WithGRPCWeb()}), - Entry("native gRPC at the server root", "", true, []connect.ClientOption{connect.WithGRPC()}), - Entry("Connect POST under a route prefix", "/alertmanager", false, []connect.ClientOption{}), - Entry("Connect HTTP GET under a route prefix", "/alertmanager", false, []connect.ClientOption{connect.WithHTTPGet()}), - Entry("gRPC-Web under a route prefix", "/alertmanager", false, []connect.ClientOption{connect.WithGRPCWeb()}), - Entry("native gRPC with a route prefix configured", "/alertmanager", true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over h2c at the root prefix", "", false, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over h2c at the root prefix", "", false, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over h2c at the root prefix", "", false, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over h2c at the server root", "", false, true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over h2c under a route prefix", "/alertmanager", false, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over h2c under a route prefix", "/alertmanager", false, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over h2c under a route prefix", "/alertmanager", false, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over h2c with a route prefix configured", "/alertmanager", false, true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over TLS at the root prefix", "", true, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over TLS at the root prefix", "", true, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over TLS at the root prefix", "", true, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over TLS at the server root", "", true, true, []connect.ClientOption{connect.WithGRPC()}), + Entry("Connect POST over TLS under a route prefix", "/alertmanager", true, false, []connect.ClientOption{}), + Entry("Connect HTTP GET over TLS under a route prefix", "/alertmanager", true, false, []connect.ClientOption{connect.WithHTTPGet()}), + Entry("gRPC-Web over TLS under a route prefix", "/alertmanager", true, false, []connect.ClientOption{connect.WithGRPCWeb()}), + Entry("native gRPC over TLS with a route prefix configured", "/alertmanager", true, true, []connect.ClientOption{connect.WithGRPC()}), ) DescribeTable("rejects transports outside their configured prefix", func(routePrefix, basePath string, nativeGRPC bool, opts []connect.ClientOption) { - inst := startInstance(routePrefix) + inst := startInstance(routePrefix, false) httpClient := connect.HTTPClient(inst.httpClient) if basePath == "api" { basePath = inst.apiPath() } if nativeGRPC { - httpClient = inst.h2cClient + httpClient = inst.rpcClient } client := inst.statusClient(httpClient, basePath, opts...) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) @@ -90,18 +100,31 @@ var _ = Describe("StatusService", func() { ) DescribeTable("exposes native health and reflection at the server root", - func(routePrefix string) { - inst := startInstance(routePrefix) + func(routePrefix string, tlsEnabled bool) { + inst := startInstance(routePrefix, tlsEnabled) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - conn, err := grpc.NewClient(strings.TrimPrefix(inst.baseURL, "http://"), grpc.WithTransportCredentials(insecure.NewCredentials())) + transportCredentials := credentials.TransportCredentials(insecure.NewCredentials()) + if tlsEnabled { + transportCredentials = credentials.NewTLS(inst.tlsConfig.Clone()) + } + conn, err := grpc.NewClient(inst.app.Addr(), grpc.WithTransportCredentials(transportCredentials)) Expect(err).NotTo(HaveOccurred()) DeferCleanup(conn.Close) - health, err := healthv1.NewHealthClient(conn).Check(ctx, &healthv1.HealthCheckRequest{}) - Expect(err).NotTo(HaveOccurred()) - Expect(health.GetStatus()).To(Equal(healthv1.HealthCheckResponse_SERVING)) + healthClient := healthv1.NewHealthClient(conn) + for _, service := range []string{"", statusv3alphaconnect.StatusServiceName} { + var remote peer.Peer + health, err := healthClient.Check(ctx, &healthv1.HealthCheckRequest{Service: service}, grpc.Peer(&remote)) + Expect(err).NotTo(HaveOccurred()) + Expect(health.GetStatus()).To(Equal(healthv1.HealthCheckResponse_SERVING)) + if tlsEnabled { + info, ok := remote.AuthInfo.(credentials.TLSInfo) + Expect(ok).To(BeTrue()) + Expect(info.State.NegotiatedProtocol).To(Equal("h2")) + } + } stream, err := reflectionv1.NewServerReflectionClient(conn).ServerReflectionInfo(ctx) Expect(err).NotTo(HaveOccurred()) @@ -118,7 +141,45 @@ var _ = Describe("StatusService", func() { } Expect(names).To(ContainElement(statusv3alphaconnect.StatusServiceName)) }, - Entry("without a route prefix", ""), - Entry("with a route prefix", "/alertmanager"), + Entry("over h2c without a route prefix", "", false), + Entry("over h2c with a route prefix", "/alertmanager", false), + Entry("over TLS without a route prefix", "", true), + Entry("over TLS with a route prefix", "/alertmanager", true), ) + + It("honors exporter-toolkit HTTP/2 disablement", func() { + inst := startInstance("", true, false) + config := inst.tlsConfig.Clone() + config.NextProtos = []string{"h2", "http/1.1"} + conn, err := tls.Dial("tcp", inst.app.Addr(), config) + Expect(err).NotTo(HaveOccurred()) + DeferCleanup(conn.Close) + Expect(conn.ConnectionState().NegotiatedProtocol).NotTo(Equal("h2")) + }) + + It("cancels active streams during shutdown", func() { + inst := startInstance("", false) + 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()) + }) })