[WIP] OTel log with ctx

internal/evaluator/logger.go

@@ -18,8 +18,11 @@
 package evaluator
 
 import (
+	"context"
+
 	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/open-policy-agent/opa/v1/logging"
+	"go.opentelemetry.io/otel/trace"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 
@@ -54,7 +57,7 @@ func (l *logger) Info(fmt string, a ...any) {
 }
 
 func (l *logger) Error(fmt string, a ...any) {
-	l.log.Errorf(fmt, a...)
+	l.log.Errorf(context.TODO(), fmt, a...)
 }
 
 func (l *logger) Warn(fmt string, a ...any) {
@@ -84,12 +87,12 @@ func mapToArray(m map[string]any) []any {
 	return ret
 }
 
-func newLogger() logging.Logger {
+func newLogger(ctx context.Context) logging.Logger {
 	lvl := zap.NewAtomicLevelAt(logp.GetLevel())
 	log := clog.NewLogger("opa").WithOptions(
 		zap.IncreaseLevel(lvl),
 		zap.AddCallerSkip(1),
-	)
+	).WithSpanContext(trace.SpanContextFromContext(ctx))
 
 	return &logger{
 		log: log,

internal/evaluator/logger_test.go

@@ -46,7 +46,7 @@ func (s *LoggerTestSuite) SetupSuite() {
 }
 
 func (s *LoggerTestSuite) TestLogFormat() {
-	logger := newLogger()
+	logger := newLogger(s.T().Context())
 	logger.SetLevel(logging.Warn)
 	logger.Warn("warn %s", "warn")
 	logs := logp.ObserverLogs().TakeAll()
@@ -56,7 +56,7 @@ func (s *LoggerTestSuite) TestLogFormat() {
 }
 
 func (s *LoggerTestSuite) TestLogFields() {
-	logger := newLogger()
+	logger := newLogger(s.T().Context())
 	logger.SetLevel(logging.Debug)
 	logger = logger.WithFields(map[string]any{
 		"key": "val",
@@ -72,7 +72,7 @@ func (s *LoggerTestSuite) TestLogFields() {
 }
 
 func (s *LoggerTestSuite) TestLogMultipleFields() {
-	logger := newLogger()
+	logger := newLogger(s.T().Context())
 	logger.SetLevel(logging.Debug)
 	logger = logger.WithFields(map[string]any{
 		"key1": "val1",
@@ -93,7 +93,7 @@ func (s *LoggerTestSuite) TestLogMultipleFields() {
 }
 
 func (s *LoggerTestSuite) TestLoggerGetLevel() {
-	logger := newLogger()
+	logger := newLogger(s.T().Context())
 	tests := []logging.Level{
 		logging.Debug,
 		logging.Info,
@@ -108,7 +108,7 @@ func (s *LoggerTestSuite) TestLoggerGetLevel() {
 }
 
 func (s *LoggerTestSuite) TestLoggerSetLevel() {
-	logger := newLogger()
+	logger := newLogger(s.T().Context())
 	logger.SetLevel(logging.Debug)
 	logger.Debug("debug")
 	logs := logp.ObserverLogs().TakeAll()

internal/evaluator/opa.go

@@ -72,14 +72,11 @@ func NewOpaEvaluator(ctx context.Context, log *clog.Logger, cfg *config.Config)
 	plugin := fmt.Sprintf(logPlugin, dlogger.PluginName, dlogger.PluginName)
 	opaCfg := fmt.Sprintf(opaConfig, cfg.BundlePath, plugin)
 
-	decisonLogger := newLogger()
-	stdLogger := newLogger()
-
 	// create an instance of the OPA object
 	opa, err := sdk.New(ctx, sdk.Options{
 		Config:        bytes.NewReader([]byte(opaCfg)),
-		Logger:        stdLogger,
-		ConsoleLogger: decisonLogger,
+		Logger:        newLogger(ctx),
+		ConsoleLogger: newLogger(ctx),
 		Plugins: map[string]plugins.Factory{
 			dlogger.PluginName: &dlogger.Factory{},
 		},

internal/flavors/benchmark/aws_org.go

@@ -161,7 +161,7 @@ func (a *AWSOrg) getAwsAccounts(ctx context.Context, log *clog.Logger, cfgCloudb
 		if identity.Account == rootIdentity.Account {
 			cfg, err := a.pickManagementAccountRole(ctx, log, stsClient, cfgCloudbeatRoot, identity)
 			if err != nil {
-				log.Errorf("error picking roles for account %s: %s", identity.Account, err)
+				log.Errorf(ctx, "error picking roles for account %s: %s", identity.Account, err)
 				continue
 			}
 			awsConfig = cfg
@@ -218,7 +218,7 @@ func (a *AWSOrg) pickManagementAccountRole(ctx context.Context, log *clog.Logger
 	if foundTagValue == scanSettingTagValue {
 		_, err := a.IAMProvider.GetRole(ctx, memberRole)
 		if err != nil {
-			log.Errorf("Management Account should be scanned (%s: %s), but %q role is missing: %s", scanSettingTagKey, foundTagValue, memberRole, err)
+			log.Errorf(ctx, "Management Account should be scanned (%s: %s), but %q role is missing: %s", scanSettingTagKey, foundTagValue, memberRole, err)
 		}
 	}
 

internal/flavors/benchmark/k8s_helper.go

@@ -48,7 +48,7 @@ func NewK8sBenchmarkHelper(log *clog.Logger, cfg *config.Config, client client_g
 func (h *K8SBenchmarkHelper) GetK8sDataProvider(ctx context.Context, clusterNameProvider k8s.ClusterNameProviderAPI) (dataprovider.CommonDataProvider, error) {
 	clusterName, err := clusterNameProvider.GetClusterName(ctx, h.cfg)
 	if err != nil {
-		h.log.Errorf("failed to get cluster name: %v", err)
+		h.log.Errorf(ctx, "failed to get cluster name: %v", err)
 	}
 
 	serverVersion, err := h.client.Discovery().ServerVersion()

internal/flavors/vulnerability.go

@@ -129,7 +129,7 @@ func (bt *vulnerability) Run(*beat.Beat) error {
 func (bt *vulnerability) runIteration() error {
 	worker, err := vuln.NewVulnerabilityWorker(bt.ctx, bt.log, bt.config, bt.bdp, bt.cdp)
 	if err != nil {
-		bt.log.Error("vulnerability.runIteration worker creation failed")
+		bt.log.Error(context.TODO(), "vulnerability.runIteration worker creation failed")
 		bt.cancel()
 		return err
 	}

internal/infra/clog/clog.go

@@ -31,22 +31,24 @@ type Logger struct {
 	*logp.Logger
 }
 
-func (l *Logger) Errorf(template string, args ...any) {
+func (l *Logger) Errorf(ctx context.Context, template string, args ...any) {
+	spanCtx := trace.SpanContextFromContext(ctx)
 	// Downgrade context.Canceled errors to warning level
 	if hasErrorType(context.Canceled, args...) {
-		l.Warnf(template, args...)
+		l.WithSpanContext(spanCtx).Warnf(template, args...)
 		return
 	}
-	l.Logger.Errorf(template, args...)
+	l.WithSpanContext(spanCtx).Logger.Errorf(template, args...)
 }
 
-func (l *Logger) Error(args ...any) {
+func (l *Logger) Error(ctx context.Context, args ...any) {
+	spanCtx := trace.SpanContextFromContext(ctx)
 	// Downgrade context.Canceled errors to warning level
 	if hasErrorType(context.Canceled, args...) {
-		l.Warn(args...)
+		l.WithSpanContext(spanCtx).Warn(args...)
 		return
 	}
-	l.Logger.Error(args...)
+	l.WithSpanContext(spanCtx).Logger.Error(args...)
 }
 
 func (l *Logger) Named(name string) *Logger {

internal/infra/clog/clog_test.go

@@ -46,8 +46,8 @@ func (s *LoggerTestSuite) TestErrorfWithContextCanceled() {
 	logger := NewLogger("test")
 
 	err := context.Canceled
-	logger.Errorf("some error: %s", err)         // error with context.Canceled
-	logger.Errorf("some error: %s", err.Error()) // error string with context Canceled
+	logger.Errorf(s.T().Context(), "some error: %s", err)         // error with context.Canceled
+	logger.Errorf(s.T().Context(), "some error: %s", err.Error()) // error string with context Canceled
 
 	logs := logp.ObserverLogs().TakeAll()
 	if s.Len(logs, 2) {
@@ -62,7 +62,7 @@ func (s *LoggerTestSuite) TestLogErrorfWithoutContextCanceled() {
 	logger := NewLogger("test")
 
 	err := errors.New("oops")
-	logger.Errorf("some error: %s", err)
+	logger.Errorf(s.T().Context(), "some error: %s", err)
 
 	logs := logp.ObserverLogs().TakeAll()
 	if s.Len(logs, 1) {

internal/inventory/awsfetcher/fetcher_ec2_instance.go

@@ -57,7 +57,7 @@ func (e *ec2InstanceFetcher) Fetch(ctx context.Context, assetChannel chan<- inve
 
 	instances, err := e.provider.DescribeInstances(ctx)
 	if err != nil {
-		e.logger.Errorf("Could not list ec2 instances: %v", err)
+		e.logger.Errorf(ctx, "Could not list ec2 instances: %v", err)
 		awslib.ReportMissingPermission(e.statusHandler, err)
 		return
 	}

internal/inventory/awsfetcher/fetcher_elb.go

@@ -76,8 +76,8 @@ func (f *elbFetcher) fetch(ctx context.Context, resourceName string, function el
 
 	awsResources, err := function(ctx)
 	if err != nil {
+		f.logger.Errorf(ctx, "Could not fetch %s: %v", resourceName, err)
 		awslib.ReportMissingPermission(f.statusHandler, err)
-		f.logger.Errorf("Could not fetch %s: %v", resourceName, err)
 		return
 	}
 

internal/inventory/awsfetcher/fetcher_iam_policy.go

@@ -57,7 +57,7 @@ func (i *iamPolicyFetcher) Fetch(ctx context.Context, assetChannel chan<- invent
 
 	policies, err := i.provider.GetPolicies(ctx)
 	if err != nil {
-		i.logger.Errorf("Could not list policies: %v", err)
+		i.logger.Errorf(ctx, "Could not list policies: %v", err)
 		if len(policies) == 0 {
 			return
 		}
@@ -71,7 +71,7 @@ func (i *iamPolicyFetcher) Fetch(ctx context.Context, assetChannel chan<- invent
 
 		policy, ok := resource.(iam.Policy)
 		if !ok {
-			i.logger.Errorf("Could not get info about policy: %s", resource.GetResourceArn())
+			i.logger.Errorf(ctx, "Could not get info about policy: %s", resource.GetResourceArn())
 			continue
 		}
 

internal/inventory/awsfetcher/fetcher_iam_role.go

@@ -57,7 +57,7 @@ func (i *iamRoleFetcher) Fetch(ctx context.Context, assetChannel chan<- inventor
 
 	roles, err := i.provider.ListRoles(ctx)
 	if err != nil {
-		i.logger.Errorf("Could not list roles: %v", err)
+		i.logger.Errorf(ctx, "Could not list roles: %v", err)
 		if len(roles) == 0 {
 			return
 		}

internal/inventory/awsfetcher/fetcher_iam_user.go

@@ -56,8 +56,8 @@ func (i *iamUserFetcher) Fetch(ctx context.Context, assetChannel chan<- inventor
 
 	users, err := i.provider.GetUsers(ctx)
 	if err != nil {
+		i.logger.Errorf(ctx, "Could not list users: %v", err)
 		awslib.ReportMissingPermission(i.statusHandler, err)
-		i.logger.Errorf("Could not list users: %v", err)
 		if len(users) == 0 {
 			return
 		}
@@ -70,7 +70,7 @@ func (i *iamUserFetcher) Fetch(ctx context.Context, assetChannel chan<- inventor
 
 		user, ok := resource.(iam.User)
 		if !ok {
-			i.logger.Errorf("Could not get info about user: %s", resource.GetResourceArn())
+			i.logger.Errorf(ctx, "Could not get info about user: %s", resource.GetResourceArn())
 			continue
 		}
 

internal/inventory/awsfetcher/fetcher_lambda.go

@@ -76,8 +76,8 @@ func (s *lambdaFetcher) fetch(ctx context.Context, resourceName string, function
 
 	awsResources, err := function(ctx)
 	if err != nil {
+		s.logger.Errorf(ctx, "Could not fetch %s: %v", resourceName, err)
 		awslib.ReportMissingPermission(s.statusHandler, err)
-		s.logger.Errorf("Could not fetch %s: %v", resourceName, err)
 		return
 	}
 

internal/inventory/awsfetcher/fetcher_networking.go

@@ -91,8 +91,8 @@ func (s *networkingFetcher) fetch(ctx context.Context, resourceName string, func
 
 	awsResources, err := function(ctx)
 	if err != nil {
+		s.logger.Errorf(ctx, "Could not fetch %s: %v", resourceName, err)
 		awslib.ReportMissingPermission(s.statusHandler, err)
-		s.logger.Errorf("Could not fetch %s: %v", resourceName, err)
 		return
 	}
 

internal/inventory/awsfetcher/fetcher_rds.go

@@ -58,7 +58,7 @@ func (s *rdsFetcher) Fetch(ctx context.Context, assetChannel chan<- inventory.As
 
 	awsResources, err := s.provider.DescribeDBInstances(ctx)
 	if err != nil {
-		s.logger.Errorf("Could not list RDS Instances: %v", err)
+		s.logger.Errorf(ctx, "Could not list RDS Instances: %v", err)
 		if len(awsResources) == 0 {
 			return
 		}

internal/inventory/awsfetcher/fetcher_s3_bucket.go

@@ -58,7 +58,7 @@ func (s *s3BucketFetcher) Fetch(ctx context.Context, assetChannel chan<- invento
 
 	awsBuckets, err := s.provider.DescribeBuckets(ctx)
 	if err != nil {
-		s.logger.Errorf("Could not list s3 buckets: %v", err)
+		s.logger.Errorf(ctx, "Could not list s3 buckets: %v", err)
 		if len(awsBuckets) == 0 {
 			return
 		}

internal/inventory/awsfetcher/fetcher_sns.go

@@ -55,7 +55,7 @@ func (s *snsFetcher) Fetch(ctx context.Context, assetChannel chan<- inventory.As
 
 	awsResources, err := s.provider.ListTopicsWithSubscriptions(ctx)
 	if err != nil {
-		s.logger.Errorf("Could not fetch SNS Topics: %v", err)
+		s.logger.Errorf(ctx, "Could not fetch SNS Topics: %v", err)
 		awslib.ReportMissingPermission(s.statusHandler, err)
 		return
 	}

internal/inventory/azurefetcher/fetcher_account.go

@@ -67,7 +67,7 @@ func (f *accountFetcher) fetch(ctx context.Context, resourceName string, functio
 
 	azureAssets, err := function(ctx)
 	if err != nil {
-		f.logger.Errorf("Could not fetch %s: %v", resourceName, err)
+		f.logger.Errorf(ctx, "Could not fetch %s: %v", resourceName, err)
 		return
 	}
 

internal/inventory/azurefetcher/fetcher_activedirectory.go

@@ -63,7 +63,7 @@ func (f *activedirectoryFetcher) fetchServicePrincipals(ctx context.Context, ass
 
 	items, err := f.provider.ListServicePrincipals(ctx)
 	if err != nil {
-		f.logger.Errorf("Could not fetch Service Principals: %v", err)
+		f.logger.Errorf(ctx, "Could not fetch Service Principals: %v", err)
 	}
 
 	for _, item := range items {
@@ -94,7 +94,7 @@ func (f *activedirectoryFetcher) fetchDirectoryRoles(ctx context.Context, assetC
 
 	items, err := f.provider.ListDirectoryRoles(ctx)
 	if err != nil {
-		f.logger.Errorf("Could not fetch Directory Roles: %v", err)
+		f.logger.Errorf(ctx, "Could not fetch Directory Roles: %v", err)
 	}
 
 	for _, item := range items {
@@ -124,7 +124,7 @@ func (f *activedirectoryFetcher) fetchGroups(ctx context.Context, assetChan chan
 
 	items, err := f.provider.ListGroups(ctx)
 	if err != nil {
-		f.logger.Errorf("Could not fetch Groups: %v", err)
+		f.logger.Errorf(ctx, "Could not fetch Groups: %v", err)
 	}
 
 	for _, item := range items {
@@ -154,7 +154,7 @@ func (f *activedirectoryFetcher) fetchUsers(ctx context.Context, assetChan chan<
 
 	items, err := f.provider.ListUsers(ctx)
 	if err != nil {
-		f.logger.Errorf("Could not fetch Users: %v", err)
+		f.logger.Errorf(ctx, "Could not fetch Users: %v", err)
 	}
 
 	for _, item := range items {

internal/inventory/azurefetcher/fetcher_resource_graph.go

@@ -79,7 +79,7 @@ func (f *resourceGraphFetcher) fetch(ctx context.Context, resourceName, serviceN
 
 	azureAssets, err := f.provider.ListAllAssetTypesByName(ctx, resourceGroup, []string{resourceType})
 	if err != nil {
-		f.logger.Errorf("Could not fetch %s: %v", resourceName, err)
+		f.logger.Errorf(ctx, "Could not fetch %s: %v", resourceName, err)
 		return
 	}
 

internal/inventory/azurefetcher/fetcher_storage.go

@@ -74,7 +74,7 @@ func (f *storageFetcher) Fetch(ctx context.Context, assetChan chan<- inventory.A
 
 	storageAccounts, err := f.listStorageAccounts(ctx)
 	if err != nil {
-		f.logger.Errorf("Could not fetch anything: %v", err)
+		f.logger.Errorf(ctx, "Could not fetch anything: %v", err)
 		return
 	}
 
@@ -108,7 +108,7 @@ func (f *storageFetcher) fetch(ctx context.Context, storageAccounts []azurelib.A
 
 	azureAssets, err := function(ctx, storageAccounts)
 	if err != nil {
-		f.logger.Errorf("Could not fetch %s: %v", resourceName, err)
+		f.logger.Errorf(ctx, "Could not fetch %s: %v", resourceName, err)
 		return
 	}