feat: trace watchable messages

internal/cmd/egctl/translate.go

@@ -7,6 +7,7 @@ package egctl
 
 import (
 	"bufio"
+	"context"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -291,7 +292,8 @@ func translateGatewayAPIToIR(resources *resource.Resources) (*gatewayapi.Transla
 		}
 	}
 
-	result, _ := t.Translate(resources)
+	ctx := context.Background()
+	result, _ := t.Translate(resources, ctx)
 
 	return result, nil
 }
@@ -311,7 +313,7 @@ func translateGatewayAPIToGatewayAPI(resources *resource.Resources) (resource.Re
 		BackendEnabled:          true,
 		Logger:                  logging.DefaultLogger(io.Discard, egv1a1.LogLevelInfo),
 	}
-	gRes, _ := gTranslator.Translate(resources)
+	gRes, _ := gTranslator.Translate(resources, context.Background())
 	// Update the status of the GatewayClass based on EnvoyProxy validation
 	epInvalid := false
 	if resources.EnvoyProxyForGatewayClass != nil {
@@ -351,7 +353,7 @@ func TranslateGatewayAPIToXds(namespace, dnsDomain, resourceType string, resourc
 		BackendEnabled:          true,
 		Logger:                  logging.DefaultLogger(io.Discard, egv1a1.LogLevelInfo),
 	}
-	gRes, _ := gTranslator.Translate(resources)
+	gRes, _ := gTranslator.Translate(resources, context.Background())
 
 	keys := []string{}
 	for key := range gRes.XdsIR {
@@ -374,7 +376,7 @@ func TranslateGatewayAPIToXds(namespace, dnsDomain, resourceType string, resourc
 		if resources.EnvoyProxyForGatewayClass != nil {
 			xTranslator.FilterOrder = resources.EnvoyProxyForGatewayClass.Spec.FilterOrder
 		}
-		xRes, err := xTranslator.Translate(val)
+		xRes, err := xTranslator.Translate(val, context.Background())
 		if err != nil {
 			return nil, fmt.Errorf("failed to translate xds ir for key %s value %+v, error:%w", key, val, err)
 		}

internal/gatewayapi/runner/runner.go

@@ -18,6 +18,7 @@ import (
 	"github.com/telepresenceio/watchable"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -144,7 +145,6 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
 			// There is only 1 key which is the controller name
 			// so when a delete is triggered, delete all keys
 			if update.Delete || valWrapper == nil || valWrapper.Resources == nil {
-				span.AddEvent("delete_all_keys")
 				r.deleteAllKeys()
 				return
 			}
@@ -169,6 +169,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
 			var backendTLSPolicyStatusCount, clientTrafficPolicyStatusCount, backendTrafficPolicyStatusCount int
 			var securityPolicyStatusCount, envoyExtensionPolicyStatusCount, backendStatusCount, extensionServerPolicyStatusCount int
 
+			span.AddEvent("gateway_resources_translation_cycle", trace.WithAttributes(attribute.Int("resources.count", len(*val))))
 			for _, resources := range *val {
 				// Translate and publish IRs.
 				t := &gatewayapi.Translator{
@@ -199,11 +200,13 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
 					traceLogger.Info("extension resources", "GVKs count", len(extGKs))
 				}
 				// Translate to IR
-				result, err := t.Translate(resources)
+				_, translateToIRSpan := tracer.Start(parentCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateToIR")
+				result, err := t.Translate(resources, parentCtx)
 				if err != nil {
 					// Currently all errors that Translate returns should just be logged
 					traceLogger.Error(err, "errors detected during translation", "gateway-class", resources.GatewayClass.Name)
 				}
+				translateToIRSpan.End()
 
 				// Publish the IRs.
 				// Also validate the ir before sending it.
@@ -243,6 +246,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
 				}
 
 				// Update Status
+				_, statusUpdateSpan := tracer.Start(parentCtx, "GatewayApiRunner.ResoureTranslationCycle.UpdateStatus")
 				if result.GatewayClass != nil {
 					key := utils.NamespacedName(result.GatewayClass)
 					r.ProviderResources.GatewayClassStatuses.Store(key, &result.GatewayClass.Status)
@@ -365,6 +369,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re
 					delete(keysToDelete.ExtensionServerPolicyStatus, key)
 					r.keyCache.ExtensionServerPolicyStatus[key] = true
 				}
+				statusUpdateSpan.End()
 			}
 
 			// Publish aggregated metrics

internal/gatewayapi/translator.go

@@ -6,9 +6,12 @@
 package gatewayapi
 
 import (
+	"context"
 	"errors"
 	"fmt"
 
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"golang.org/x/exp/maps"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -49,10 +52,13 @@ const (
 	wellKnownPortShift = 10000
 )
 
-var _ TranslatorManager = (*Translator)(nil)
+var (
+	_      TranslatorManager = (*Translator)(nil)
+	tracer                   = otel.Tracer("envoy-gateway/gateway-api/translator")
+)
 
 type TranslatorManager interface {
-	Translate(resources *resource.Resources) (*TranslateResult, error)
+	Translate(resources *resource.Resources, ctx context.Context) (*TranslateResult, error)
 	GetRelevantGateways(resources *resource.Resources) (acceptedGateways, failedGateways []*GatewayContext)
 
 	RoutesTranslator
@@ -217,7 +223,10 @@ func newTranslateResult(
 	return translateResult
 }
 
-func (t *Translator) Translate(resources *resource.Resources) (*TranslateResult, error) {
+func (t *Translator) Translate(resources *resource.Resources, ctx context.Context) (*TranslateResult, error) {
+	_, span := tracer.Start(ctx, "Translator.Translate")
+	defer span.End()
+	span.SetAttributes(getAttributes(resources)...)
 	var errs error
 
 	// Get Gateways belonging to our GatewayClass.
@@ -529,3 +538,16 @@ func (t *Translator) IRKey(gatewayNN types.NamespacedName) string {
 	}
 	return irStringKey(gatewayNN.Namespace, gatewayNN.Name)
 }
+
+func getAttributes(resources *resource.Resources) []attribute.KeyValue {
+	attrs := []attribute.KeyValue{}
+	if resources.GatewayClass == nil {
+		return attrs
+	}
+	attrs = append(attrs, attribute.String("gateway-class", resources.GatewayClass.Name))
+	attrs = append(attrs, attribute.String("gateway-class-namespace", resources.GatewayClass.Namespace))
+	if resources.GatewayClass.Spec.ControllerName != "" {
+		attrs = append(attrs, attribute.String("gateway-class-controller-name", string(resources.GatewayClass.Spec.ControllerName)))
+	}
+	return attrs
+}

internal/gatewayapi/translator_test.go

@@ -421,7 +421,7 @@ func TestTranslate(t *testing.T) {
 				},
 			})
 
-			got, _ := translator.Translate(resources)
+			got, _ := translator.Translate(resources, context.Background())
 			require.NoError(t, field.SetValue(got, "LastTransitionTime", metav1.NewTime(time.Time{})))
 			outputFilePath := strings.ReplaceAll(inputFile, ".in.yaml", ".out.yaml")
 			out, err := yaml.Marshal(got)
@@ -696,7 +696,7 @@ func TestTranslateWithExtensionKinds(t *testing.T) {
 				},
 			})
 
-			got, _ := translator.Translate(resources)
+			got, _ := translator.Translate(resources, context.Background())
 			require.NoError(t, field.SetValue(got, "LastTransitionTime", metav1.NewTime(time.Time{})))
 			// Also fix lastTransitionTime in unstructured members
 			for i := range got.ExtensionServerPolicies {

internal/globalratelimit/runner/runner.go

@@ -20,6 +20,8 @@ import (
 	resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 	serverv3 "github.com/envoyproxy/go-control-plane/pkg/server/v3"
 	"github.com/telepresenceio/watchable"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 
@@ -47,6 +49,8 @@ const (
 	rateLimitTLSCACertFilepath = "/certs/ca.crt"
 )
 
+var tracer = otel.Tracer("envoy-gateway/global-rate-limit/runner")
+
 type Config struct {
 	config.Server
 	XdsIR           *message.XdsIR
@@ -138,7 +142,20 @@ func (r *Runner) translateFromSubscription(ctx context.Context, c <-chan watchab
 
 	message.HandleSubscription(message.Metadata{Runner: r.Name(), Message: message.XDSIRMessageName}, c,
 		func(update message.Update[string, *message.XdsIRWithContext], errChan chan error) {
-			r.Logger.Info("received a notification")
+			parentCtx := context.Background()
+			if update.Value != nil && update.Value.Context != nil {
+				parentCtx = update.Value.Context
+			}
+			traceLogger := r.Logger.WithTrace(parentCtx)
+			traceLogger.Info("received a notification")
+
+			_, span := tracer.Start(parentCtx, "GlobalRateLimitRunner.translateFromSubscription")
+			defer span.End()
+
+			span.SetAttributes(
+				attribute.String("controller.key", update.Key),
+				attribute.Bool("update.delete", update.Delete),
+			)
 
 			if update.Delete {
 				delete(rateLimitConfigsCache, update.Key)
@@ -183,6 +200,9 @@ func (r *Runner) translate(xdsIR *ir.Xds) (*types.ResourceVersionTable, error) {
 }
 
 func (r *Runner) updateSnapshot(ctx context.Context, resource types.XdsResources) {
+	_, span := tracer.Start(ctx, "GlobalRateLimitRunner.updateSnapshot")
+	defer span.End()
+
 	if r.cache == nil {
 		r.Logger.Error(nil, "failed to init the snapshot cache")
 		return

internal/xds/cache/snapshotcache.go

@@ -35,7 +35,7 @@ import (
 
 var (
 	Hash   = cachev3.IDHash{}
-	tracer = otel.Tracer("envoy-gateway/gateway-api")
+	tracer = otel.Tracer("envoy-gateway/xds/snapshotcache")
 )
 
 // SnapshotCacheWithCallbacks uses the go-control-plane SimpleCache to store snapshots of

internal/xds/runner/runner.go

@@ -66,7 +66,7 @@ const (
 	defaultMaxConnectionAgeGrace = 2 * time.Minute
 )
 
-var tracer = otel.Tracer("envoy-gateway/gateway-api")
+var tracer = otel.Tracer("envoy-gateway/xds")
 
 var maxConnectionAgeValues = []time.Duration{
 	10 * time.Hour,
@@ -316,7 +316,7 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string,
 					}
 				}
 
-				result, err := t.Translate(val.XdsIR)
+				result, err := t.Translate(val.XdsIR, parentCtx)
 				if err != nil {
 					traceLogger.Error(err, "failed to translate xds ir")
 					errChan <- err

internal/xds/translator/translator.go

@@ -6,6 +6,7 @@
 package translator
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"runtime"
@@ -22,6 +23,7 @@ import (
 	matcherv3 "github.com/envoyproxy/go-control-plane/envoy/type/matcher/v3"
 	resourcev3 "github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 	"github.com/envoyproxy/go-control-plane/pkg/wellknown"
+	"go.opentelemetry.io/otel"
 	protobuf "google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
 	"google.golang.org/protobuf/types/known/wrapperspb"
@@ -44,6 +46,8 @@ const (
 	emptyClusterName = "EmptyCluster"
 )
 
+var tracer = otel.Tracer("envoy-gateway/xds/translator")
+
 // The dummy cluster for TCP/UDP listeners that have no routes
 var emptyRouteCluster = &clusterv3.Cluster{
 	Name:                 emptyClusterName,
@@ -94,7 +98,10 @@ type GlobalRateLimitSettings struct {
 }
 
 // Translate translates the XDS IR into xDS resources
-func (t *Translator) Translate(xdsIR *ir.Xds) (*types.ResourceVersionTable, error) {
+func (t *Translator) Translate(xdsIR *ir.Xds, ctx context.Context) (*types.ResourceVersionTable, error) {
+	_, span := tracer.Start(ctx, "Translator.Translate")
+	defer span.End()
+
 	if xdsIR == nil {
 		return nil, errors.New("ir is nil")
 	}

internal/xds/translator/translator_test.go

@@ -6,6 +6,7 @@
 package translator
 
 import (
+	"context"
 	"embed"
 	"encoding/json"
 	"os"
@@ -180,7 +181,7 @@ func TestTranslateXds(t *testing.T) {
 				FilterOrder:  x.FilterOrder,
 				RuntimeFlags: cfg.runtimeFlags,
 			}
-			tCtx, err := tr.Translate(x)
+			tCtx, err := tr.Translate(x, context.Background())
 			if !strings.HasSuffix(inputFileName, "partial-invalid") && len(cfg.errMsg) == 0 {
 				t.Log(inputFileName)
 				require.NoError(t, err)
@@ -384,7 +385,7 @@ func TestTranslateXdsWithExtensionErrorsWhenFailOpen(t *testing.T) {
 			defer closeFunc()
 			tr.ExtensionManager = &extMgr
 
-			tCtx, err := tr.Translate(x)
+			tCtx, err := tr.Translate(x, context.Background())
 			if len(cfg.errMsg) > 0 {
 				require.EqualError(t, err, cfg.errMsg)
 			} else {
@@ -525,7 +526,7 @@ func TestTranslateXdsWithExtensionErrorsWhenFailClosed(t *testing.T) {
 			defer closeFunc()
 			tr.ExtensionManager = &extMgr
 
-			_, err = tr.Translate(x)
+			_, err = tr.Translate(x, context.Background())
 			require.EqualError(t, err, cfg.errMsg)
 		})
 	}

test/helm/gateway-addons-helm/e2e.in.yaml

@@ -4,12 +4,5 @@ grafana:
   enabled: false
 opentelemetry-collector:
   enabled: true
-  mode: deployment
-  service:
-    type: LoadBalancer
 fluent-bit:
   enabled: false
-tempo:
-  enabled: true
-  service:
-    type: LoadBalancer

test/helm/gateway-addons-helm/e2e.out.yaml

@@ -10334,7 +10334,7 @@ metadata:
     app.kubernetes.io/component: standalone-collector
     component: standalone-collector
 spec:
-  type: LoadBalancer
+  type: ClusterIP
   ports:
 
     - name: datadog