feat: add opentelemetry trace to record the sync trace

Author: yafeiaa

go.mod

@@ -18,6 +18,8 @@ require (
 	k8s.io/apimachinery v0.32.2
 	k8s.io/cli-runtime v0.32.2
 	k8s.io/client-go v0.32.2
+	go.opentelemetry.io/otel v1.28.0
+	go.opentelemetry.io/otel/trace v1.28.0
 	k8s.io/klog/v2 v2.130.1
 	k8s.io/kube-aggregator v0.32.2
 	k8s.io/kube-openapi v0.0.0-20241212222426-2c72e554b1e7

pkg/engine/engine.go

@@ -23,6 +23,7 @@ import (
 	"github.com/argoproj/gitops-engine/pkg/sync"
 	"github.com/argoproj/gitops-engine/pkg/sync/common"
 	"github.com/argoproj/gitops-engine/pkg/utils/kube"
+	"github.com/argoproj/gitops-engine/pkg/utils/tracing"
 )
 
 const (
@@ -84,7 +85,7 @@ func (e *gitOpsEngine) Sync(ctx context.Context,
 		return nil, err
 	}
 	opts = append(opts, sync.WithSkipHooks(!diffRes.Modified))
-	syncCtx, cleanup, err := sync.NewSyncContext(revision, result, e.config, e.config, e.kubectl, namespace, e.cache.GetOpenAPISchema(), opts...)
+	syncCtx, cleanup, err := sync.NewSyncContext(revision, result, e.config, e.config, e.kubectl, namespace, e.cache.GetOpenAPISchema(), tracing.NopTracer{}, "", "", opts...)
 	if err != nil {
 		return nil, err
 	}

pkg/sync/sync_context.go

@@ -32,6 +32,7 @@ import (
 	"github.com/argoproj/gitops-engine/pkg/sync/hook"
 	resourceutil "github.com/argoproj/gitops-engine/pkg/sync/resource"
 	kubeutil "github.com/argoproj/gitops-engine/pkg/utils/kube"
+	"github.com/argoproj/gitops-engine/pkg/utils/tracing"
 )
 
 type reconciledResource struct {
@@ -209,6 +210,8 @@ func NewSyncContext(
 	kubectl kubeutil.Kubectl,
 	namespace string,
 	openAPISchema openapi.Resources,
+	syncTracer tracing.Tracer,
+	syncTraceID, syncTraceRootSpanID string,
 	opts ...SyncOpt,
 ) (SyncContext, func(), error) {
 	dynamicIf, err := dynamic.NewForConfig(restConfig)
@@ -246,6 +249,9 @@ func NewSyncContext(
 		permissionValidator: func(_ *unstructured.Unstructured, _ *metav1.APIResource) error {
 			return nil
 		},
+		syncTracer:          syncTracer,
+		syncTraceID:         syncTraceID,
+		syncTraceRootSpanID: syncTraceRootSpanID,
 	}
 	for _, opt := range opts {
 		opt(ctx)
@@ -357,6 +363,11 @@ type syncContext struct {
 	// lock to protect concurrent updates of the result list
 	lock sync.Mutex
 
+	// tracer for tracing the sync operation
+	syncTraceID         string
+	syncTraceRootSpanID string
+	syncTracer          tracing.Tracer
+
 	// syncNamespace is a function that will determine if the managed
 	// namespace should be synced
 	syncNamespace func(*unstructured.Unstructured, *unstructured.Unstructured) (bool, error)
@@ -1262,6 +1273,8 @@ func (sc *syncContext) runTasks(tasks syncTasks, dryRun bool) runState {
 			ss.Go(func(state runState) runState {
 				logCtx := sc.log.WithValues("dryRun", dryRun, "task", t)
 				logCtx.V(1).Info("Pruning")
+				span := sc.syncTracer.StartSpanFromTraceParent("pruneObject", sc.syncTraceID, sc.syncTraceRootSpanID)
+				defer span.Finish()
 				result, message := sc.pruneObject(t.liveObj, sc.prune, dryRun)
 				if result == common.ResultCodeSyncFailed {
 					state = failed
@@ -1270,6 +1283,7 @@ func (sc *syncContext) runTasks(tasks syncTasks, dryRun bool) runState {
 				if !dryRun || sc.dryRun || result == common.ResultCodeSyncFailed {
 					sc.setResourceResult(t, result, operationPhases[result], message)
 				}
+				sc.setBaggageItemForTasks(&span, t, message, result, operationPhases[result])
 				return state
 			})
 		}
@@ -1289,19 +1303,27 @@ func (sc *syncContext) runTasks(tasks syncTasks, dryRun bool) runState {
 			ss.Go(func(state runState) runState {
 				sc.log.WithValues("dryRun", dryRun, "task", t).V(1).Info("Deleting")
 				if !dryRun {
+					span := sc.syncTracer.StartSpanFromTraceParent("hooksDeletion", sc.syncTraceID, sc.syncTraceRootSpanID)
+					defer span.Finish()
 					err := sc.deleteResource(t)
+					message := "deleted"
+					operationPhase := common.OperationRunning
 					if err != nil {
 						// it is possible to get a race condition here, such that the resource does not exist when
 						// delete is requested, we treat this as a nop
 						if !apierrors.IsNotFound(err) {
 							state = failed
-							sc.setResourceResult(t, "", common.OperationError, fmt.Sprintf("failed to delete resource: %v", err))
+							message = fmt.Sprintf("failed to delete resource: %v", err)
+							operationPhase = common.OperationError
+							sc.setResourceResult(t, "", operationPhase, message)
 						}
 					} else {
 						// if there is anything that needs deleting, we are at best now in pending and
 						// want to return and wait for sync to be invoked again
 						state = pending
+						operationPhase = common.OperationSucceeded
 					}
+					sc.setBaggageItemForTasks(&span, t, message, "", operationPhase)
 				}
 				return state
 			})
@@ -1330,6 +1352,24 @@ func (sc *syncContext) runTasks(tasks syncTasks, dryRun bool) runState {
 	return state
 }
 
+func (sc *syncContext) createSpan(operation string, dryrun bool) tracing.Span {
+	// skip tracing if dryrun
+	if dryrun {
+		return tracing.NopTracer{}.StartSpan(operation)
+	}
+	return sc.syncTracer.StartSpanFromTraceParent(operation, sc.syncTraceID, sc.syncTraceRootSpanID)
+}
+
+func (sc *syncContext) setBaggageItemForTasks(span *tracing.Span, t *syncTask, message string, result common.ResultCode, operationPhase common.OperationPhase) {
+	resourceKey := t.resourceKey()
+	(*span).SetBaggageItem("resource", resourceKey.String())
+	(*span).SetBaggageItem("result", string(result))
+	(*span).SetBaggageItem("operationPhase", string(operationPhase))
+	(*span).SetBaggageItem("message", message)
+	(*span).SetBaggageItem("phase", string(t.phase))
+	(*span).SetBaggageItem("wave", fmt.Sprint(t.wave()))
+}
+
 func (sc *syncContext) processCreateTasks(state runState, tasks syncTasks, dryRun bool) runState {
 	ss := newStateSync(state)
 	for _, task := range tasks {
@@ -1341,11 +1381,14 @@ func (sc *syncContext) processCreateTasks(state runState, tasks syncTasks, dryRu
 			logCtx := sc.log.WithValues("dryRun", dryRun, "task", t)
 			logCtx.V(1).Info("Applying")
 			validate := sc.validate && !resourceutil.HasAnnotationOption(t.targetObj, common.AnnotationSyncOptions, common.SyncOptionsDisableValidation)
+			span := sc.syncTracer.StartSpanFromTraceParent("applyObject", sc.syncTraceID, sc.syncTraceRootSpanID)
+			defer span.Finish()
 			result, message := sc.applyObject(t, dryRun, validate)
 			if result == common.ResultCodeSyncFailed {
 				logCtx.WithValues("message", message).Info("Apply failed")
 				state = failed
 			}
+			var phase common.OperationPhase
 			if !dryRun || sc.dryRun || result == common.ResultCodeSyncFailed {
 				phase := operationPhases[result]
 				// no resources are created in dry-run, so running phase means validation was
@@ -1355,6 +1398,7 @@ func (sc *syncContext) processCreateTasks(state runState, tasks syncTasks, dryRu
 				}
 				sc.setResourceResult(t, result, phase, message)
 			}
+			sc.setBaggageItemForTasks(&span, t, message, result, phase)
 			return state
 		})
 	}

pkg/utils/tracing/api.go

@@ -8,9 +8,12 @@ package tracing
 
 type Tracer interface {
 	StartSpan(operationName string) Span
+	StartSpanFromTraceParent(operationName string, parentTraceId, parentSpanId string) Span
 }
 
 type Span interface {
 	SetBaggageItem(key string, value any)
 	Finish()
+	SpanID() string
+	TraceID() string
 }

pkg/utils/tracing/logging.go

@@ -30,6 +30,12 @@ func (l LoggingTracer) StartSpan(operationName string) Span {
 	}
 }
 
+// loggingSpan is not a real distributed tracing system.
+// so no need to implement real StartSpanFromTraceParent method.
+func (l LoggingTracer) StartSpanFromTraceParent(operationName string, parentTraceId, parentSpanId string) Span {
+	return l.StartSpan(operationName)
+}
+
 type loggingSpan struct {
 	logger        logr.Logger
 	operationName string
@@ -54,3 +60,11 @@ func baggageToVals(baggage map[string]any) []any {
 	}
 	return result
 }
+
+func (s loggingSpan) TraceID() string {
+	return ""
+}
+
+func (s loggingSpan) SpanID() string {
+	return ""
+}

pkg/utils/tracing/nop.go

@@ -11,10 +11,22 @@ func (n NopTracer) StartSpan(_ string) Span {
 	return nopSpan{}
 }
 
+func (n NopTracer) StartSpanFromTraceParent(_, _, _ string) Span {
+	return nopSpan{}
+}
+
 type nopSpan struct{}
 
 func (n nopSpan) SetBaggageItem(_ string, _ any) {
 }
 
 func (n nopSpan) Finish() {
 }
+
+func (s nopSpan) TraceID() string {
+	return ""
+}
+
+func (s nopSpan) SpanID() string {
+	return ""
+}

pkg/utils/tracing/opentelemetry.go

@@ -0,0 +1,56 @@
+package tracing
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type OpenTelemetryTracer struct {
+	realTracer trace.Tracer
+}
+
+func NewOpenTelemetryTracer(t trace.Tracer) Tracer {
+	return &OpenTelemetryTracer{
+		realTracer: t,
+	}
+}
+
+func (t OpenTelemetryTracer) StartSpan(operationName string) Span {
+	_, realspan := t.realTracer.Start(context.Background(), operationName)
+	return openTelemetrySpan{realSpan: realspan}
+}
+
+func (t OpenTelemetryTracer) StartSpanFromTraceParent(operationName string, parentTraceId, parentSpanId string) Span {
+	traceID, _ := trace.TraceIDFromHex(parentTraceId)
+	parentSpanID, _ := trace.SpanIDFromHex(parentSpanId)
+	spanCtx := trace.NewSpanContext(trace.SpanContextConfig{
+		TraceID:    traceID,
+		SpanID:     parentSpanID,
+		TraceFlags: trace.FlagsSampled,
+	})
+	ctx := trace.ContextWithSpanContext(context.Background(), spanCtx)
+	_, realSpan := t.realTracer.Start(ctx, operationName)
+	return openTelemetrySpan{realSpan: realSpan}
+}
+
+type openTelemetrySpan struct {
+	realSpan trace.Span
+}
+
+func (s openTelemetrySpan) SetBaggageItem(key string, value interface{}) {
+	s.realSpan.SetAttributes(attribute.Key(key).String(value.(string)))
+}
+
+func (s openTelemetrySpan) Finish() {
+	s.realSpan.End()
+}
+
+func (s openTelemetrySpan) TraceID() string {
+	return s.realSpan.SpanContext().TraceID().String()
+}
+
+func (s openTelemetrySpan) SpanID() string {
+	return s.realSpan.SpanContext().SpanID().String()
+}