Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[v17] Move most of lib/srv/discovery from logrus to slog #48426

Merged
merged 2 commits into from
Nov 5, 2024
+148 −125
Merged

[v17] Move most of lib/srv/discovery from logrus to slog #48426

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
567ab90
Move most of `lib/srv/discovery` from logrus to slog
marcoandredinis Nov 1, 2024
22a8087
use local context
marcoandredinis Nov 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion lib/service/discovery.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
24 changes: 12 additions & 12 deletions lib/srv/discovery/access_graph.go
View file
Open in desktop
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 @@ -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

Expand All @@ -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
Expand All @@ -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 {
Expand Down
14 changes: 7 additions & 7 deletions lib/srv/discovery/database_watcher.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down Expand Up @@ -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()
}
Expand Down Expand Up @@ -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
}

Expand All @@ -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(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 @@ -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(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
Loading