Refactor healthcheck signalling between server and service

Simplifies the signalling of healthcheck status from the "servers" to the "service": instead of using 2 channels to feed healthcheck status back to the Service.HealthCheck, we just give the server components direct access to the Healthcheck which they can update directly. This is possible because the Healthcheck package is threadsafe (uses `atomic.Value` for state). This pattern is consisten with how the service's Healtcheck is passed directly to cmd/collector/app package. closes #5307 Signed-off-by: Will Sewell <[email protected]>
jaegertracing · Mar 28, 2024 · df0ad22 · df0ad22
1 parent 0be0cb9
commit df0ad22
Show file tree

Hide file tree

Showing 11 changed files with 62 additions and 186 deletions.
diff --git a/cmd/all-in-one/main.go b/cmd/all-in-one/main.go
@@ -283,15 +283,10 @@ func startQuery(
 ) *queryApp.Server {
 	spanReader = storageMetrics.NewReadMetricsDecorator(spanReader, metricsFactory)
 	qs := querysvc.NewQueryService(spanReader, depReader, *queryOpts)
-	server, err := queryApp.NewServer(svc.Logger, qs, metricsQueryService, qOpts, tm, jt)
+	server, err := queryApp.NewServer(svc.Logger, svc.HC(), qs, metricsQueryService, qOpts, tm, jt)
 	if err != nil {
 		svc.Logger.Fatal("Could not create jaeger-query", zap.Error(err))
 	}
-	go func() {
-		for s := range server.HealthCheckStatus() {
-			svc.SetHealthCheckStatus(s)
-		}
-	}()
 	if err := server.Start(); err != nil {
 		svc.Logger.Fatal("Could not start jaeger-query", zap.Error(err))
 	}

diff --git a/cmd/internal/flags/service.go b/cmd/internal/flags/service.go
@@ -50,20 +50,16 @@ type Service struct {
 	MetricsFactory metrics.Factory
 
 	signalsChannel chan os.Signal
-
-	hcStatusChannel chan healthcheck.Status
 }
 
 // NewService creates a new Service.
 func NewService(adminPort int) *Service {
 	signalsChannel := make(chan os.Signal, 1)
-	hcStatusChannel := make(chan healthcheck.Status)
 	signal.Notify(signalsChannel, os.Interrupt, syscall.SIGTERM)
 
 	return &Service{
-		Admin:           NewAdminServer(ports.PortToHostPort(adminPort)),
-		signalsChannel:  signalsChannel,
-		hcStatusChannel: hcStatusChannel,
+		Admin:          NewAdminServer(ports.PortToHostPort(adminPort)),
+		signalsChannel: signalsChannel,
 	}
 }
 
@@ -79,11 +75,6 @@ func (s *Service) AddFlags(flagSet *flag.FlagSet) {
 	s.Admin.AddFlags(flagSet)
 }
 
-// SetHealthCheckStatus sets status of healthcheck
-func (s *Service) SetHealthCheckStatus(status healthcheck.Status) {
-	s.hcStatusChannel <- status
-}
-
 // Start bootstraps the service and starts the admin server.
 func (s *Service) Start(v *viper.Viper) error {
 	if err := TryLoadConfigFile(v); err != nil {
@@ -143,15 +134,7 @@ func (s *Service) HC() *healthcheck.HealthCheck {
 func (s *Service) RunAndThen(shutdown func()) {
 	s.HC().Ready()
 
-statusLoop:
-	for {
-		select {
-		case status := <-s.hcStatusChannel:
-			s.HC().Set(status)
-		case <-s.signalsChannel:
-			break statusLoop
-		}
-	}
+	<-s.signalsChannel
 
 	s.Logger.Info("Shutting down")
 	s.HC().Set(healthcheck.Unavailable)

diff --git a/cmd/internal/flags/service_test.go b/cmd/internal/flags/service_test.go
@@ -93,7 +93,7 @@ func TestStartErrors(t *testing.T) {
 			go s.RunAndThen(shutdown)
 
 			waitForEqual(t, healthcheck.Ready, func() interface{} { return s.HC().Get() })
-			s.SetHealthCheckStatus(healthcheck.Unavailable)
+			s.HC().Set(healthcheck.Unavailable)
 			waitForEqual(t, healthcheck.Unavailable, func() interface{} { return s.HC().Get() })
 
 			s.signalsChannel <- os.Interrupt

diff --git a/cmd/jaeger/internal/extension/jaegerquery/server.go b/cmd/jaeger/internal/extension/jaegerquery/server.go
@@ -14,6 +14,7 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/jaeger/internal/extension/jaegerstorage"
 	queryApp "github.com/jaegertracing/jaeger/cmd/query/app"
 	"github.com/jaegertracing/jaeger/cmd/query/app/querysvc"
+	"github.com/jaegertracing/jaeger/pkg/healthcheck"
 	"github.com/jaegertracing/jaeger/pkg/jtracer"
 	"github.com/jaegertracing/jaeger/pkg/tenancy"
 	"github.com/jaegertracing/jaeger/plugin/metrics/disabled"
@@ -81,6 +82,7 @@ func (s *server) Start(ctx context.Context, host component.Host) error {
 	//nolint
 	s.server, err = queryApp.NewServer(
 		s.logger,
+		healthcheck.New(),
 		qs,
 		metricsQueryService,
 		s.makeQueryOptions(),

diff --git a/cmd/query/app/server.go b/cmd/query/app/server.go
@@ -48,23 +48,23 @@ import (
 // Server runs HTTP, Mux and a grpc server
 type Server struct {
 	logger       *zap.Logger
+	healthCheck  *healthcheck.HealthCheck
 	querySvc     *querysvc.QueryService
 	queryOptions *QueryOptions
 
 	tracer *jtracer.JTracer // TODO make part of flags.Service
 
-	conn               net.Listener
-	grpcConn           net.Listener
-	httpConn           net.Listener
-	cmuxServer         cmux.CMux
-	grpcServer         *grpc.Server
-	httpServer         *http.Server
-	separatePorts      bool
-	unavailableChannel chan healthcheck.Status
+	conn          net.Listener
+	grpcConn      net.Listener
+	httpConn      net.Listener
+	cmuxServer    cmux.CMux
+	grpcServer    *grpc.Server
+	httpServer    *http.Server
+	separatePorts bool
 }
 
 // NewServer creates and initializes Server
-func NewServer(logger *zap.Logger, querySvc *querysvc.QueryService, metricsQuerySvc querysvc.MetricsQueryService, options *QueryOptions, tm *tenancy.Manager, tracer *jtracer.JTracer) (*Server, error) {
+func NewServer(logger *zap.Logger, healthCheck *healthcheck.HealthCheck, querySvc *querysvc.QueryService, metricsQuerySvc querysvc.MetricsQueryService, options *QueryOptions, tm *tenancy.Manager, tracer *jtracer.JTracer) (*Server, error) {
 	_, httpPort, err := net.SplitHostPort(options.HTTPHostPort)
 	if err != nil {
 		return nil, fmt.Errorf("invalid HTTP server host:port: %w", err)
@@ -89,22 +89,17 @@ func NewServer(logger *zap.Logger, querySvc *querysvc.QueryService, metricsQuery
 	}
 
 	return &Server{
-		logger:             logger,
-		querySvc:           querySvc,
-		queryOptions:       options,
-		tracer:             tracer,
-		grpcServer:         grpcServer,
-		httpServer:         httpServer,
-		separatePorts:      grpcPort != httpPort,
-		unavailableChannel: make(chan healthcheck.Status),
+		logger:        logger,
+		healthCheck:   healthCheck,
+		querySvc:      querySvc,
+		queryOptions:  options,
+		tracer:        tracer,
+		grpcServer:    grpcServer,
+		httpServer:    httpServer,
+		separatePorts: grpcPort != httpPort,
 	}, nil
 }
 
-// HealthCheckStatus returns health check status channel a client can subscribe to
-func (s Server) HealthCheckStatus() chan healthcheck.Status {
-	return s.unavailableChannel
-}
-
 func createGRPCServer(querySvc *querysvc.QueryService, metricsQuerySvc querysvc.MetricsQueryService, options *QueryOptions, tm *tenancy.Manager, logger *zap.Logger, tracer *jtracer.JTracer) (*grpc.Server, error) {
 	var grpcOpts []grpc.ServerOption
 
@@ -292,7 +287,7 @@ func (s *Server) Start() error {
 			s.logger.Error("Could not start HTTP server", zap.Error(err))
 		}
 
-		s.unavailableChannel <- healthcheck.Unavailable
+		s.healthCheck.Set(healthcheck.Unavailable)
 	}()
 
 	// Start GRPC server concurrently
@@ -302,7 +297,7 @@ func (s *Server) Start() error {
 		if err := s.grpcServer.Serve(s.grpcConn); err != nil {
 			s.logger.Error("Could not start GRPC server", zap.Error(err))
 		}
-		s.unavailableChannel <- healthcheck.Unavailable
+		s.healthCheck.Set(healthcheck.Unavailable)
 	}()
 
 	// Start cmux server concurrently.
@@ -315,7 +310,7 @@ func (s *Server) Start() error {
 			if err != nil && !strings.Contains(err.Error(), "use of closed network connection") {
 				s.logger.Error("Could not start multiplexed server", zap.Error(err))
 			}
-			s.unavailableChannel <- healthcheck.Unavailable
+			s.healthCheck.Set(healthcheck.Unavailable)
 		}()
 	}