Move most of lib/srv/discovery from logrus to slog

lib/service/discovery.go

@@ -87,7 +87,8 @@ func (process *TeleportProcess) initDiscoveryService() error {
 		Emitter:           asyncEmitter,
 		AccessPoint:       accessPoint,
 		ServerID:          process.Config.HostUUID,
-		Log:               process.log,
+		Log:               process.logger,
+		LegacyLogger:      process.log,
 		ClusterName:       conn.ClusterName(),
 		ClusterFeatures:   process.GetClusterFeatures,
 		PollInterval:      process.Config.Discovery.PollInterval,

lib/srv/discovery/access_graph.go

@@ -66,7 +66,7 @@ func (s *Server) reconcileAccessGraph(ctx context.Context, currentTAGResources *
 		upsert, toDel := aws_sync.ReconcileResults(currentTAGResources, &aws_sync.Resources{})
 
 		if err := push(stream, upsert, toDel); err != nil {
-			s.Log.WithError(err).Error("Error pushing empty resources to TAGs")
+			s.Log.ErrorContext(ctx, "Error pushing empty resources to TAGs", "error", err)
 		}
 		return trace.Wrap(errNoAccessGraphFetchers)
 	}
@@ -109,7 +109,7 @@ func (s *Server) reconcileAccessGraph(ctx context.Context, currentTAGResources *
 	// Aggregate all errors into a single error.
 	err := trace.NewAggregate(errs...)
 	if err != nil {
-		s.Log.WithError(err).Error("Error polling TAGs")
+		s.Log.ErrorContext(ctx, "Error polling TAGs", "error", err)
 	}
 	result := aws_sync.MergeResources(results...)
 	// Merge all results into a single result
@@ -122,7 +122,7 @@ func (s *Server) reconcileAccessGraph(ctx context.Context, currentTAGResources *
 	}
 
 	if pushErr != nil {
-		s.Log.WithError(pushErr).Error("Error pushing TAGs")
+		s.Log.ErrorContext(ctx, "Error pushing TAGs", "error", pushErr)
 		return nil
 	}
 	// Update the currentTAGResources with the result of the reconciliation.
@@ -135,7 +135,7 @@ func (s *Server) reconcileAccessGraph(ctx context.Context, currentTAGResources *
 			},
 		},
 	}); err != nil {
-		s.Log.WithError(err).Error("Error submitting usage event")
+		s.Log.ErrorContext(ctx, "Error submitting usage event", "error", err)
 	}
 
 	return nil
@@ -315,7 +315,7 @@ func (s *Server) initializeAndWatchAccessGraph(ctx context.Context, reloadCh <-c
 	defer func() {
 		lease.Stop()
 		if err := lease.Wait(); err != nil {
-			s.Log.WithError(err).Warn("error cleaning up semaphore")
+			s.Log.WarnContext(ctx, "Error cleaning up semaphore", "error", err)
 		}
 	}()
 
@@ -336,12 +336,12 @@ func (s *Server) initializeAndWatchAccessGraph(ctx context.Context, reloadCh <-c
 
 	stream, err := client.AWSEventsStream(ctx)
 	if err != nil {
-		s.Log.WithError(err).Error("Failed to get access graph service stream")
+		s.Log.ErrorContext(ctx, "Failed to get access graph service stream", "error", err)
 		return trace.Wrap(err)
 	}
 	header, err := stream.Header()
 	if err != nil {
-		s.Log.WithError(err).Error("Failed to get access graph service stream header")
+		s.Log.ErrorContext(ctx, "Failed to get access graph service stream header", "error", err)
 		return trace.Wrap(err)
 	}
 	const (
@@ -361,7 +361,7 @@ func (s *Server) initializeAndWatchAccessGraph(ctx context.Context, reloadCh <-c
 		defer wg.Done()
 		defer cancel()
 		if !accessGraphConn.WaitForStateChange(ctx, connectivity.Ready) {
-			s.Log.Info("access graph service connection was closed")
+			s.Log.InfoContext(ctx, "Access graph service connection was closed")
 		}
 	}()
 
@@ -418,7 +418,7 @@ func grpcCredentials(config AccessGraphConfig, getCert func() (*tls.Certificate,
 func (s *Server) initAccessGraphWatchers(ctx context.Context, cfg *Config) error {
 	fetchers, err := s.accessGraphFetchersFromMatchers(ctx, cfg.Matchers, "" /* discoveryConfigName */)
 	if err != nil {
-		s.Log.WithError(err).Error("Error initializing access graph fetchers")
+		s.Log.ErrorContext(ctx, "Error initializing access graph fetchers", "error", err)
 	}
 	s.staticTAGSyncFetchers = fetchers
 
@@ -431,7 +431,7 @@ func (s *Server) initAccessGraphWatchers(ctx context.Context, cfg *Config) error
 				// We will wait for the config to change and re-evaluate the fetchers
 				// before starting the sync.
 				if len(allFetchers) == 0 {
-					s.Log.Debug("No AWS sync fetchers configured. Access graph sync will not be enabled.")
+					s.Log.DebugContext(ctx, "No AWS sync fetchers configured. Access graph sync will not be enabled.")
 					select {
 					case <-ctx.Done():
 						return
@@ -442,10 +442,10 @@ func (s *Server) initAccessGraphWatchers(ctx context.Context, cfg *Config) error
 				}
 				// reset the currentTAGResources to force a full sync
 				if err := s.initializeAndWatchAccessGraph(ctx, reloadCh); errors.Is(err, errTAGFeatureNotEnabled) {
-					s.Log.Warn("Access Graph specified in config, but the license does not include Teleport Policy. Access graph sync will not be enabled.")
+					s.Log.WarnContext(ctx, "Access Graph specified in config, but the license does not include Teleport Policy. Access graph sync will not be enabled.")
 					break
 				} else if err != nil {
-					s.Log.Warnf("Error initializing and watching access graph: %v", err)
+					s.Log.WarnContext(ctx, "Error initializing and watching access graph", "error", err)
 				}
 
 				select {

lib/srv/discovery/database_watcher.go

@@ -66,7 +66,7 @@ func (s *Server) startDatabaseWatchers() error {
 
 	watcher, err := common.NewWatcher(s.ctx, common.WatcherConfig{
 		FetchersFn:     s.getAllDatabaseFetchers,
-		Log:            s.Log.WithField("kind", types.KindDatabase),
+		Log:            s.LegacyLogger.WithField("kind", types.KindDatabase),
 		DiscoveryGroup: s.DiscoveryGroup,
 		Interval:       s.PollInterval,
 		TriggerFetchC:  s.newDiscoveryConfigChangedSub(),
@@ -96,7 +96,7 @@ func (s *Server) startDatabaseWatchers() error {
 				mu.Unlock()
 
 				if err := reconciler.Reconcile(s.ctx); err != nil {
-					s.Log.WithError(err).Warn("Unable to reconcile database resources.")
+					s.Log.WarnContext(s.ctx, "Unable to reconcile database resources", "error", err)
 				} else if s.onDatabaseReconcile != nil {
 					s.onDatabaseReconcile()
 				}
@@ -128,7 +128,7 @@ func (s *Server) getAllDatabaseFetchers() []common.Fetcher {
 func (s *Server) getCurrentDatabases() map[string]types.Database {
 	databases, err := s.AccessPoint.GetDatabases(s.ctx)
 	if err != nil {
-		s.Log.WithError(err).Warn("Failed to get databases from cache.")
+		s.Log.WarnContext(s.ctx, "Failed to get databases from cache", "error", err)
 		return nil
 	}
 
@@ -138,7 +138,7 @@ func (s *Server) getCurrentDatabases() map[string]types.Database {
 }
 
 func (s *Server) onDatabaseCreate(ctx context.Context, database types.Database) error {
-	s.Log.Debugf("Creating database %s.", database.GetName())
+	s.Log.DebugContext(s.ctx, "Creating database", "database", database.GetName())
 	err := s.AccessPoint.CreateDatabase(ctx, database)
 	// If the database already exists but has cloud origin and an empty
 	// discovery group, then update it.
@@ -163,18 +163,18 @@ func (s *Server) onDatabaseCreate(ctx context.Context, database types.Database)
 		},
 	})
 	if err != nil {
-		s.Log.WithError(err).Debug("Error emitting usage event.")
+		s.Log.DebugContext(ctx, "Error emitting usage event", "error", err)
 	}
 	return nil
 }
 
 func (s *Server) onDatabaseUpdate(ctx context.Context, database, _ types.Database) error {
-	s.Log.Debugf("Updating database %s.", database.GetName())
+	s.Log.DebugContext(s.ctx, "Updating database", "database", database.GetName())
 	return trace.Wrap(s.AccessPoint.UpdateDatabase(ctx, database))
 }
 
 func (s *Server) onDatabaseDelete(ctx context.Context, database types.Database) error {
-	s.Log.Debugf("Deleting database %s.", database.GetName())
+	s.Log.DebugContext(s.ctx, "Deleting database", "database", database.GetName())
 	return trace.Wrap(s.AccessPoint.DeleteDatabase(ctx, database.GetName()))
 }