Throw startup error if TeleportReadyEvent is not emitted (#11725)

* Throw startup error if `TeleportReadyEvent` is not emitted

Before this commit, the `TeleportReadyEvent` was only waited for when a
process reload occurred. Thus, if a bug exists in the code that emits
this event (as it's currently the case since the `MetricsReady` and
`WindowsDesktopReady` events are never emitted), such a bug may go
unnoticed for a while.

This commit ensures that the `TeleportReadyEvent` is always waited for
on startup, and throws an error if the event is not emitted (after some
timeout).

This commit also:
- removes the `MetricsReady` event (as this is not produced by a
  component that sends heartbeats, which is the case of every other
  event required by the `TeleportReadyEvent` event mapping)
- ensures that `WindowsDesktopReady` event is emitted
- refactors some of the code in `lib/service/supervisor.go`
- moves the event mapping registration to a new `registerTeleportReadyEvent` function

integration/helpers.go

@@ -1103,26 +1103,15 @@ func (i *TeleInstance) Start() error {
 	// Build a list of expected events to wait for before unblocking based off
 	// the configuration passed in.
 	expectedEvents := []string{}
-	if i.Config.Auth.Enabled {
-		expectedEvents = append(expectedEvents, service.AuthTLSReady)
-	}
+	// Always wait for TeleportReadyEvent.
+	expectedEvents = append(expectedEvents, service.TeleportReadyEvent)
 	if i.Config.Proxy.Enabled {
 		expectedEvents = append(expectedEvents, service.ProxyReverseTunnelReady)
-		expectedEvents = append(expectedEvents, service.ProxySSHReady)
 		expectedEvents = append(expectedEvents, service.ProxyAgentPoolReady)
 		if !i.Config.Proxy.DisableWebService {
 			expectedEvents = append(expectedEvents, service.ProxyWebServerReady)
 		}
 	}
-	if i.Config.SSH.Enabled {
-		expectedEvents = append(expectedEvents, service.NodeSSHReady)
-	}
-	if i.Config.Apps.Enabled {
-		expectedEvents = append(expectedEvents, service.AppsReady)
-	}
-	if i.Config.Databases.Enabled {
-		expectedEvents = append(expectedEvents, service.DatabasesReady)
-	}
 
 	// Start the process and block until the expected events have arrived.
 	receivedEvents, err := startAndWait(i.Process, expectedEvents)

integration/integration_test.go

@@ -4359,6 +4359,17 @@ func waitForProcessStart(serviceC chan *service.TeleportProcess) (*service.Telep
 		dumpGoroutineProfile()
 		return nil, trace.BadParameter("timeout waiting for service to start")
 	}
+
+	eventC := make(chan service.Event, 1)
+	svc.WaitForEvent(context.TODO(), service.TeleportReadyEvent, eventC)
+	select {
+	case <-eventC:
+
+	case <-time.After(20 * time.Second):
+		dumpGoroutineProfile()
+		return nil, trace.BadParameter("timeout waiting for service to broadcast ready status")
+	}
+
 	return svc, nil
 }
 

lib/service/desktop.go

@@ -248,6 +248,10 @@ func (process *TeleportProcess) initWindowsDesktopServiceRegistered(log *logrus.
 				"Windows desktop service %s:%s is starting on %v.",
 				teleport.Version, teleport.Gitref, listener.Addr())
 		}
+
+		// since srv.Serve is a blocking call, we emit this event right before
+		// the service has started
+		process.BroadcastEvent(Event{Name: WindowsDesktopReady, Payload: nil})
 		err := srv.Serve(listener)
 		if err != nil {
 			if err == http.ErrServerClosed {

lib/service/kubernetes.go

@@ -271,6 +271,8 @@ func (process *TeleportProcess) initKubernetesService(log *logrus.Entry, conn *C
 				"Kubernetes service %s:%s is starting on %v.",
 				teleport.Version, teleport.Gitref, listener.Addr())
 		}
+		// since kubeServer.Serve is a blocking call, we emit this event right before
+		// the service has started
 		process.BroadcastEvent(Event{Name: KubernetesReady, Payload: nil})
 		err := kubeServer.Serve(listener)
 		if err != nil {

lib/service/service.go

@@ -173,10 +173,6 @@ const (
 	// is ready to start accepting connections.
 	DatabasesReady = "DatabasesReady"
 
-	// MetricsReady is generated when the Teleport metrics service is ready to
-	// start accepting connections.
-	MetricsReady = "MetricsReady"
-
 	// WindowsDesktopReady is generated when the Teleport windows desktop
 	// service is ready to start accepting connections.
 	WindowsDesktopReady = "WindowsDesktopReady"
@@ -513,6 +509,20 @@ func Run(ctx context.Context, cfg Config, newTeleport NewProcess) error {
 	if err := srv.Start(); err != nil {
 		return trace.Wrap(err, "startup failed")
 	}
+
+	// Wait for the service to report that it has started.
+	startTimeoutCtx, startCancel := context.WithTimeout(ctx, signalPipeTimeout)
+	defer startCancel()
+	eventC := make(chan Event, 1)
+	srv.WaitForEvent(startTimeoutCtx, TeleportReadyEvent, eventC)
+	select {
+	case <-eventC:
+		cfg.Log.Infof("Service has started successfully.")
+	case <-startTimeoutCtx.Done():
+		warnOnErr(srv.Close(), cfg.Log)
+		return trace.BadParameter("service has failed to start")
+	}
+
 	// Wait and reload until called exit.
 	for {
 		srv, err = waitAndReload(ctx, cfg, srv, newTeleport)
@@ -761,36 +771,8 @@ func NewTeleport(cfg *Config) (*TeleportProcess, error) {
 		cfg.Keygen = native.New(process.ExitContext(), native.PrecomputeKeys(precomputeCount))
 	}
 
-	// Produce global TeleportReadyEvent
-	// when all components have started
-	eventMapping := EventMapping{
-		Out: TeleportReadyEvent,
-	}
-	if cfg.Auth.Enabled {
-		eventMapping.In = append(eventMapping.In, AuthTLSReady)
-	}
-	if cfg.SSH.Enabled {
-		eventMapping.In = append(eventMapping.In, NodeSSHReady)
-	}
-	if cfg.Proxy.Enabled {
-		eventMapping.In = append(eventMapping.In, ProxySSHReady)
-	}
-	if cfg.Kube.Enabled {
-		eventMapping.In = append(eventMapping.In, KubernetesReady)
-	}
-	if cfg.Apps.Enabled {
-		eventMapping.In = append(eventMapping.In, AppsReady)
-	}
-	if cfg.Databases.Enabled {
-		eventMapping.In = append(eventMapping.In, DatabasesReady)
-	}
-	if cfg.Metrics.Enabled {
-		eventMapping.In = append(eventMapping.In, MetricsReady)
-	}
-	if cfg.WindowsDesktop.Enabled {
-		eventMapping.In = append(eventMapping.In, WindowsDesktopReady)
-	}
-	process.RegisterEventMapping(eventMapping)
+	// Produce global TeleportReadyEvent when all components have started
+	process.registerTeleportReadyEvent(cfg)
 
 	if cfg.Auth.Enabled {
 		if err := process.initAuthService(); err != nil {
@@ -1381,7 +1363,7 @@ func (process *TeleportProcess) initAuthService() error {
 		utils.Consolef(cfg.Console, log, teleport.ComponentAuth, "Auth service %s:%s is starting on %v.",
 			teleport.Version, teleport.Gitref, authAddr)
 
-		// since tlsServer.Serve is a blocking call, we emit this even right before
+		// since tlsServer.Serve is a blocking call, we emit this event right before
 		// the service has started
 		process.BroadcastEvent(Event{Name: AuthTLSReady, Payload: nil})
 		err := tlsServer.Serve()
@@ -3464,6 +3446,44 @@ func (process *TeleportProcess) waitForAppDepend() {
 	}
 }
 
+// registerTeleportReadyEvent ensures that a TeleportReadyEvent is produced
+// when all components have started.
+func (process *TeleportProcess) registerTeleportReadyEvent(cfg *Config) {
+	eventMapping := EventMapping{
+		Out: TeleportReadyEvent,
+	}
+
+	if cfg.Auth.Enabled {
+		eventMapping.In = append(eventMapping.In, AuthTLSReady)
+	}
+
+	if cfg.SSH.Enabled {
+		eventMapping.In = append(eventMapping.In, NodeSSHReady)
+	}
+
+	if cfg.Proxy.Enabled {
+		eventMapping.In = append(eventMapping.In, ProxySSHReady)
+	}
+
+	if cfg.Kube.Enabled {
+		eventMapping.In = append(eventMapping.In, KubernetesReady)
+	}
+
+	if cfg.Apps.Enabled {
+		eventMapping.In = append(eventMapping.In, AppsReady)
+	}
+
+	if cfg.Databases.Enabled {
+		eventMapping.In = append(eventMapping.In, DatabasesReady)
+	}
+
+	if cfg.WindowsDesktop.Enabled {
+		eventMapping.In = append(eventMapping.In, WindowsDesktopReady)
+	}
+
+	process.RegisterEventMapping(eventMapping)
+}
+
 // appDependEvents is a list of events that the application service depends on.
 var appDependEvents = []string{
 	AuthTLSReady,

lib/service/signals.go

@@ -153,7 +153,7 @@ func (process *TeleportProcess) WaitForSignals(ctx context.Context) error {
 }
 
 // ErrTeleportReloading is returned when signal waiter exits
-// because the teleport process has initiaded shutdown
+// because the teleport process has initiated shutdown
 var ErrTeleportReloading = &trace.CompareFailedError{Message: "teleport process is reloading"}
 
 // ErrTeleportExited means that teleport has exited

lib/service/supervisor.go

@@ -55,7 +55,7 @@ type Supervisor interface {
 	Wait() error
 
 	// Run starts and waits for the service to complete
-	// it's a combinatioin Start() and Wait()
+	// it's a combination Start() and Wait()
 	Run() error
 
 	// Services returns list of running services
@@ -362,33 +362,26 @@ func (s *LocalSupervisor) BroadcastEvent(event Event) {
 		s.signalReload()
 	}
 
-	s.events[event.Name] = event
+	sendEvent := func(e Event) {
+		select {
+		case s.eventsC <- e:
+		case <-s.closeContext.Done():
+		}
+	}
 
+	s.events[event.Name] = event
+	go sendEvent(event)
 	// Log all events other than recovered events to prevent the logs from
 	// being flooded.
 	if event.String() != TeleportOKEvent {
 		s.log.WithField("event", event.String()).Debug("Broadcasting event.")
 	}
 
-	go func() {
-		select {
-		case s.eventsC <- event:
-		case <-s.closeContext.Done():
-			return
-		}
-	}()
-
 	for _, m := range s.eventMappings {
 		if m.matches(event.Name, s.events) {
 			mappedEvent := Event{Name: m.Out}
 			s.events[mappedEvent.Name] = mappedEvent
-			go func(e Event) {
-				select {
-				case s.eventsC <- e:
-				case <-s.closeContext.Done():
-					return
-				}
-			}(mappedEvent)
+			go sendEvent(mappedEvent)
 			s.log.WithFields(logrus.Fields{
 				"in":  event.String(),
 				"out": m.String(),
@@ -416,7 +409,7 @@ func (s *LocalSupervisor) WaitForEvent(ctx context.Context, name string, eventC
 	waiter := &waiter{eventC: eventC, context: ctx}
 	event, ok := s.events[name]
 	if ok {
-		go s.notifyWaiter(waiter, event)
+		go waiter.notify(event)
 		return
 	}
 	s.eventWaiters[name] = append(s.eventWaiters[name], waiter)
@@ -432,20 +425,13 @@ func (s *LocalSupervisor) getWaiters(name string) []*waiter {
 	return out
 }
 
-func (s *LocalSupervisor) notifyWaiter(w *waiter, event Event) {
-	select {
-	case w.eventC <- event:
-	case <-w.context.Done():
-	}
-}
-
 func (s *LocalSupervisor) fanOut() {
 	for {
 		select {
 		case event := <-s.eventsC:
 			waiters := s.getWaiters(event.Name)
 			for _, waiter := range waiters {
-				go s.notifyWaiter(waiter, event)
+				go waiter.notify(event)
 			}
 		case <-s.closeContext.Done():
 			return
@@ -458,6 +444,13 @@ type waiter struct {
 	context context.Context
 }
 
+func (w *waiter) notify(event Event) {
+	select {
+	case w.eventC <- event:
+	case <-w.context.Done():
+	}
+}
+
 // Service is a running teleport service function
 type Service interface {
 	// Serve starts the function