fix(konnect/tracing): Add values about Konnect sync round for tracing and logging in Konnect config syncer (#7838)

* add values about Konnect sync round for tracing and logging

* update changelog

Author: randmonkey

CHANGELOG.md

@@ -123,16 +123,9 @@ Adding a new version? You'll need three changes:
 - Separate Konnect config sync concurrency from gateway Admin API concurrency via
   `--konnect-config-sync-concurrency` to reduce Konnect 429 rate-limit errors.
   [#7834](https://github.com/Kong/kubernetes-ingress-controller/pull/7834)
-- Do not cleanup `null`s in the configuration of plugins with Kong running in
-  DBLess mode in the translator of ingress-controller. This enables user to use
-  explicit `null`s in plugins.
-  [#7751](https://github.com/Kong/kubernetes-ingress-controller/pull/7751)
-- Translate `healtchchecks.thershold` in `KongUpstreamPolicy` to the
-  `healthchecks.thershold` field in Kong upstreams.
-  (Thanks to [@elbrogan-vizio](https://github.com/elbrogan-vizio) who contributed to the fix.)
-  [#7784](https://github.com/Kong/kubernetes-ingress-controller/pull/7784)
-- Reject CA Secrets with multiple PEM certs.
-  [#7763](https://github.com/Kong/kubernetes-ingress-controller/pull/7763)
+- Add headers to mark the synchronization loop run in requests sent to Konnect in
+  configuration synchronization for better tracing.
+  [#7838](https://github.com/Kong/kubernetes-ingress-controller/pull/7838)
 
 ## [3.5.3]
 

internal/konnect/config_synchronizer.go

@@ -5,8 +5,11 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/go-logr/logr"
+	"github.com/google/uuid"
 	"github.com/kong/go-database-reconciler/pkg/file"
 	"github.com/kong/go-kong/kong"
 	"github.com/samber/mo"
@@ -18,6 +21,7 @@ import (
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/deckgen"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/kongstate"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/dataplane/sendconfig"
+	"github.com/kong/kubernetes-ingress-controller/v3/internal/konnect/tracing"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/logging"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/metrics"
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/util"
@@ -45,6 +49,11 @@ type ConfigSynchronizer struct {
 
 	targetKongState mo.Option[TargetKongState]
 	configLock      sync.RWMutex
+
+	// synchronizerID is the identifier to mark the ConfigSynchronizer instance.
+	synchronizerID string
+	// serialNumber is the serial number to mark the loop round of config synchronization.
+	serialNumber atomic.Uint32
 }
 
 // TargetKongState wraps the Kong state to be uploaded to Konnect and indicates whether the configuration is a fallback
@@ -74,18 +83,23 @@ type ConfigSynchronizerParams struct {
 	ConfigChangeDetector   sendconfig.ConfigurationChangeDetector
 	ConfigStatusNotifier   clients.ConfigStatusNotifier
 	MetricsRecorder        metrics.Recorder
+	SynchronizerID         string
 }
 
 func NewConfigSynchronizer(p ConfigSynchronizerParams) *ConfigSynchronizer {
+	if p.SynchronizerID == "" {
+		p.SynchronizerID = uuid.NewString()
+	}
 	return &ConfigSynchronizer{
-		logger:                 p.Logger,
+		logger:                 p.Logger.WithValues("synchronizerID", p.SynchronizerID),
 		kongConfig:             p.KongConfig,
 		syncTicker:             p.ConfigUploadTicker,
 		konnectClientFactory:   p.KonnectClientFactory,
 		updateStrategyResolver: p.UpdateStrategyResolver,
 		configChangeDetector:   p.ConfigChangeDetector,
 		configStatusNotifier:   p.ConfigStatusNotifier,
 		metricsRecorder:        p.MetricsRecorder,
+		synchronizerID:         p.SynchronizerID,
 	}
 }
 
@@ -94,6 +108,7 @@ var _ manager.LeaderElectionRunnable = &ConfigSynchronizer{}
 // Start starts the loop to receive configuration and upload configuration to Konnect.
 func (s *ConfigSynchronizer) Start(ctx context.Context) error {
 	s.logger.Info("Starting Konnect configuration synchronizer")
+	ctx = context.WithValue(ctx, tracing.SynchronizerIDKey, s.synchronizerID)
 
 	konnectAdminClient, err := s.konnectClientFactory.NewKonnectClient(ctx)
 	if err != nil {
@@ -199,26 +214,36 @@ func (s *ConfigSynchronizer) run(ctx context.Context) {
 }
 
 func (s *ConfigSynchronizer) handleConfigSynchronizationTick(ctx context.Context) {
-	s.logger.V(logging.DebugLevel).Info("Start uploading configuration to Konnect")
+	// Add values about the sync round in the context.
+	serialNumber := s.serialNumber.Add(1)
+	startTimestamp := time.Now().Unix()
+	syncRoundID := uuid.NewSHA1(uuid.Nil, fmt.Appendf([]byte{}, "%s:%d:%d", s.synchronizerID, serialNumber, startTimestamp))
+	ctx = context.WithValue(ctx, tracing.SyncSerialNumberKey, serialNumber)
+	ctx = context.WithValue(ctx, tracing.SyncStartTimestampKey, startTimestamp)
+	ctx = context.WithValue(ctx, tracing.SyncRoundIDKey, syncRoundID)
+	logger := s.logger.WithValues("syncRoundID", syncRoundID, "serialNumber", serialNumber)
+
+	logger.V(logging.DebugLevel).Info("Start uploading configuration to Konnect")
 
 	// Get the latest configuration copy to upload to Konnect. We don't want to hold the lock for a long time to prevent
 	// blocking the update of the configuration.
 	targetCfg, ok := s.currentContent(ctx)
 	if !ok {
-		s.logger.Info("No configuration received yet, skipping Konnect configuration synchronization")
+		logger.Info("No configuration received yet, skipping Konnect configuration synchronization")
 		return
 	}
 
 	// Upload the configuration to Konnect.
-	if err := s.uploadConfig(ctx, s.konnectAdminClient, targetCfg); err != nil {
-		s.logger.Error(err, "Failed to upload configuration to Konnect")
-		logKonnectErrors(s.logger, err)
+	if err := s.uploadConfig(ctx, logger, s.konnectAdminClient, targetCfg); err != nil {
+		logger.Error(err, "Failed to upload configuration to Konnect")
+		logKonnectErrors(logger, err)
 	}
 }
 
 // uploadConfig sends the given configuration to Konnect.
 func (s *ConfigSynchronizer) uploadConfig(
 	ctx context.Context,
+	logger logr.Logger,
 	client *adminapi.KonnectClient,
 	targetContent TargetContent,
 ) error {
@@ -229,7 +254,7 @@ func (s *ConfigSynchronizer) uploadConfig(
 
 	newSHA, err := sendconfig.PerformUpdate(
 		ctx,
-		s.logger,
+		logger,
 		client,
 		s.kongConfig,
 		targetContent.Content,

internal/konnect/tracing/datadog.go

@@ -3,6 +3,7 @@ package tracing
 import (
 	"context"
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/go-logr/logr"
@@ -11,7 +12,30 @@ import (
 	"github.com/kong/kubernetes-ingress-controller/v3/internal/logging"
 )
 
+// ContextKey is the key to carry values to trace Konnect configuration sync in the context.
+type ContextKey string
+
+const (
+	// SynchronizerIDKey is the context key for the ID of KIC's Konnect configuration synchronizer instance.
+	SynchronizerIDKey ContextKey = "KonnectSynchronizerID"
+	// SyncSerialNumberKey is the context key for serial number to mark a round of configuration sync to Konnect.
+	SyncSerialNumberKey ContextKey = "KonnectSyncSerialNumber"
+	// SyncRoundIDKey is the context key to mark the ID of a round of configuration sync.
+	SyncRoundIDKey ContextKey = "KonnectSyncRoundID"
+	// SyncStartTimestampKey is the context key for timestamp (in seconds) of starting a round of configuration sync.
+	SyncStartTimestampKey ContextKey = "KonnectSyncStartTimestamp"
+)
+
 const (
+	// InstanceIDHeader is the header to mark the ID of KIC's Konnect configuration synchronizer instance.
+	InstanceIDHeader = "X-Kic-Konnect-Sync-Instance-Id"
+	// SyncSerialNumberHeader is the header for serial number to mark a round of configuration sync to Konnect.
+	SyncSerialNumberHeader = "X-Kic-Konnect-Sync-Serial-Number"
+	// SyncStartTimestampHeader is the header for timestamp (in seconds) of starting a round of configuration sync.
+	SyncStartTimestampHeader = "X-Kic-Konnect-Sync-Start-Timestamp"
+	// SyncRoundIDHeader is the header to mark the ID of a round of configuration sync.
+	SyncRoundIDHeader = "X-Kic-Konnect-Sync-Round-Id"
+
 	// B3TraceIDHeader is the header used by the B3 propagation format to pass the trace ID.
 	B3TraceIDHeader = "X-B3-TraceId"
 	// B3SpanIDHeader is the header used by the B3 propagation format to pass the span ID.
@@ -25,6 +49,8 @@ const (
 
 // DoRequest is a helper function that sends an HTTP request and logs the result with DataDog trace ID.
 func DoRequest(ctx context.Context, httpClient *http.Client, req *http.Request) (*http.Response, error) {
+	req = addHeaderFromContext(ctx, req)
+
 	httpResp, err := httpClient.Do(req)
 	if err != nil {
 		return nil, err
@@ -52,6 +78,25 @@ func DoRequest(ctx context.Context, httpClient *http.Client, req *http.Request)
 	return httpResp, nil
 }
 
+// addHeaderFromContext extracts the values to mark a configuration sync round in context
+// and set the headers in the request for tracing.
+func addHeaderFromContext(ctx context.Context, req *http.Request) *http.Request {
+	if instanceID, ok := ctx.Value(SynchronizerIDKey).(string); ok {
+		req.Header.Add(InstanceIDHeader, instanceID)
+	}
+	if syncRoundID, ok := ctx.Value(SyncRoundIDKey).(string); ok {
+		req.Header.Add(SyncRoundIDHeader, syncRoundID)
+	}
+	if serialNumber, ok := ctx.Value(SyncSerialNumberKey).(uint32); ok {
+		req.Header.Add(SyncSerialNumberHeader, strconv.FormatUint(uint64(serialNumber), 10))
+	}
+	if startTimestamp, ok := ctx.Value(SyncStartTimestampKey).(int64); ok {
+		req.Header.Add(SyncStartTimestampHeader, strconv.FormatInt(startTimestamp, 10))
+	}
+
+	return req
+}
+
 // loggerWithDataDogTraceID creates a new logger with the DataDog tracing information extracted from the HTTP response's
 // headers. This data is useful for correlating logs with traces and logs in DataDog.
 func loggerWithDataDogTraceID(logger logr.Logger, resp *http.Response) logr.Logger {