Skip to content

Commit

Permalink
Move most of lib/srv/discovery from logrus to slog (#48292) (#48428)
Browse files Browse the repository at this point in the history 
* Move most of `lib/srv/discovery` from logrus to slog

* use local context
  • Loading branch information
@marcoandredinis
marcoandredinis authored Nov 5, 2024
1 parent 3700c1c commit 92861f1
Show file tree
Hide file tree
Showing 11 changed files with 151 additions and 128 deletions.
3 changes: 2 additions & 1 deletion lib/service/discovery.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -98,7 +98,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.ClientIdentity.ClusterName,
ClusterFeatures: process.GetClusterFeatures,
PollInterval: process.Config.Discovery.PollInterval,
Expand Down
24 changes: 12 additions & 12 deletions lib/srv/discovery/access_graph.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
Expand Down Expand Up @@ -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
Expand All @@ -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.
Expand All @@ -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
Expand Down Expand Up @@ -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)
}
}()

Expand All @@ -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 (
Expand All @@ -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")
}
}()

Expand Down Expand Up @@ -411,7 +411,7 @@ func grpcCredentials(config AccessGraphConfig, certs []tls.Certificate) (grpc.Di
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

Expand All @@ -424,7 +424,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
Expand All @@ -435,10 +435,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 {
Expand Down
16 changes: 8 additions & 8 deletions lib/srv/discovery/database_watcher.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ func (s *Server) startDatabaseWatchers() error {
defer mu.Unlock()
return utils.FromSlice(newDatabases, types.Database.GetName)
},
Log: s.Log.WithField("kind", types.KindDatabase),
Log: s.LegacyLogger.WithField("kind", types.KindDatabase),
OnCreate: s.onDatabaseCreate,
OnUpdate: s.onDatabaseUpdate,
OnDelete: s.onDatabaseDelete,
Expand All @@ -64,7 +64,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(),
Expand Down Expand Up @@ -94,7 +94,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()
}
Expand Down Expand Up @@ -126,7 +126,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
}

Expand All @@ -136,7 +136,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(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.
Expand All @@ -161,18 +161,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(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(ctx, "Deleting database", "database", database.GetName())
return trace.Wrap(s.AccessPoint.DeleteDatabase(ctx, database.GetName()))
}

Expand Down
Loading

0 comments on commit 92861f1

Please sign in to comment.