fix(ingress-controller): Add headers for tracing in requests to Konnect to upload configuration (#4062)

Co-authored-by: Jintao Zhang <[email protected]>

Author: randmonkey

CHANGELOG.md

@@ -46,6 +46,14 @@
 
 ### Added
 
+- Add the following headers in requests of `ingress-controller` sent to Konnect
+  for uploading configuration for tracing:
+  - `X-Kic-Konnect-Sync-Instance-Id` for instance ID of Konnect config synchronizer.
+    It is set to use the `ControlPlane`'s instance ID.
+  - `X-Kic-Konnect-Sync-Serial-Number` for serial number of config sync round.
+  - `X-Kic-Konnect-Sync-Start-Timestamp` for the timestamp of starting the config sync round.
+  - `X-Kic-Konnect-Sync-Round-Id` for the ID to mark the config sync round.
+  [#4062](https://github.com/Kong/kong-operator/pull/4062)
 - Add `spec.konnect.configUploadConcurrency` to set the concurrency of uploading
   configuration to Konnect in the on-prem gateway integration wit Konnect and
   decrease the default concurrency to `4` to reduce the possibility of triggering

ingress-controller/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/kong-operator/v2/ingress-controller/internal/dataplane/deckgen"
 	"github.com/kong/kong-operator/v2/ingress-controller/internal/dataplane/kongstate"
 	"github.com/kong/kong-operator/v2/ingress-controller/internal/dataplane/sendconfig"
+	"github.com/kong/kong-operator/v2/ingress-controller/internal/konnect/tracing"
 	"github.com/kong/kong-operator/v2/ingress-controller/internal/logging"
 	"github.com/kong/kong-operator/v2/ingress-controller/internal/metrics"
 	"github.com/kong/kong-operator/v2/ingress-controller/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(tracing.SyncRoundIDNamespace, 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.String())
+	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,

ingress-controller/internal/konnect/tracing/datadog.go

@@ -3,15 +3,43 @@ package tracing
 import (
 	"context"
 	"net/http"
+	"strconv"
 	"strings"
 
 	"github.com/go-logr/logr"
+	"github.com/google/uuid"
 	ctrllog "sigs.k8s.io/controller-runtime/pkg/log"
 
 	"github.com/kong/kong-operator/v2/ingress-controller/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"
+)
+
+// SyncRoundIDNamespace is the UUID namespace used to generate deterministic IDs for Konnect sync rounds.
+var SyncRoundIDNamespace = uuid.NameSpaceDNS
+
+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 +53,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 +82,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 sets 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 {

ingress-controller/internal/konnect/tracing/datadog_test.go

@@ -2,6 +2,7 @@ package tracing_test
 
 import (
 	"bytes"
+	"context"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -97,3 +98,42 @@ func TestDoRequest(t *testing.T) {
 		})
 	}
 }
+
+func TestDoRequest_SyncHeaders(t *testing.T) {
+	var (
+		testInstanceID     = "test-instance-id"
+		testSerialNumber   = uint32(42)
+		testStartTimestamp = int64(1700000000)
+		testSyncRoundID    = "test-sync-round-id"
+	)
+
+	var receivedHeaders http.Header
+	testServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		receivedHeaders = r.Header.Clone()
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(testServer.Close)
+
+	client := testServer.Client()
+	request, err := http.NewRequest(http.MethodGet, testServer.URL, nil)
+	require.NoError(t, err)
+
+	loggerBuf := &bytes.Buffer{}
+	logger := buflogr.NewWithBuffer(loggerBuf)
+	ctx := ctrllog.IntoContext(t.Context(), logger)
+	ctx = context.WithValue(ctx, tracing.SynchronizerIDKey, testInstanceID)
+	ctx = context.WithValue(ctx, tracing.SyncSerialNumberKey, testSerialNumber)
+	ctx = context.WithValue(ctx, tracing.SyncStartTimestampKey, testStartTimestamp)
+	ctx = context.WithValue(ctx, tracing.SyncRoundIDKey, testSyncRoundID)
+
+	resp, err := tracing.DoRequest(ctx, client, request)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		_ = resp.Body.Close()
+	})
+
+	require.Equal(t, testInstanceID, receivedHeaders.Get(tracing.InstanceIDHeader))
+	require.Equal(t, "42", receivedHeaders.Get(tracing.SyncSerialNumberHeader))
+	require.Equal(t, "1700000000", receivedHeaders.Get(tracing.SyncStartTimestampHeader))
+	require.Equal(t, testSyncRoundID, receivedHeaders.Get(tracing.SyncRoundIDHeader))
+}

ingress-controller/internal/manager/run.go

@@ -354,6 +354,7 @@ func New(
 			updateStrategyResolver,
 			configStatusNotifier,
 			metricsRecorder,
+			instanceID,
 		)
 		if err != nil {
 			setupLog.Error(err, "Failed to setup Konnect configuration synchronizer with manager, skipping")

ingress-controller/internal/manager/setup.go

@@ -499,6 +499,7 @@ func setupKonnectConfigSynchronizerWithMgr(
 	updateStrategyResolver sendconfig.UpdateStrategyResolver,
 	configStatusNotifier clients.ConfigStatusNotifier,
 	metricsRecorder metrics.Recorder,
+	instanceID InstanceID,
 ) (*konnect.ConfigSynchronizer, error) {
 	s := konnect.NewConfigSynchronizer(
 		konnect.ConfigSynchronizerParams{
@@ -510,6 +511,7 @@ func setupKonnectConfigSynchronizerWithMgr(
 			ConfigChangeDetector:   sendconfig.NewKonnectConfigurationChangeDetector(),
 			ConfigStatusNotifier:   configStatusNotifier,
 			MetricsRecorder:        metricsRecorder,
+			SynchronizerID:         instanceID.String(),
 		},
 	)
 	err := mgr.Add(s)