From 0c47cc11cdc0bdd227280a15de51e28683807a9b Mon Sep 17 00:00:00 2001 From: zirain Date: Wed, 5 Aug 2026 11:38:10 +0800 Subject: [PATCH 1/3] fix: make control plane trace more readable Signed-off-by: zirain --- internal/gatewayapi/resource/resource.go | 9 +++++++++ internal/gatewayapi/runner/runner.go | 23 +++++++++++++++++------ internal/globalratelimit/runner/runner.go | 5 +---- internal/message/types.go | 8 ++++++++ internal/xds/runner/runner.go | 5 +---- 5 files changed, 36 insertions(+), 14 deletions(-) diff --git a/internal/gatewayapi/resource/resource.go b/internal/gatewayapi/resource/resource.go index 6d2c05a800..4529f7f47a 100644 --- a/internal/gatewayapi/resource/resource.go +++ b/internal/gatewayapi/resource/resource.go @@ -155,6 +155,15 @@ type ControllerResourcesContext struct { Context context.Context } +// ParentContext returns the trace context stashed on c, or fallback if c is nil or has none +// (e.g. before any Reconcile has stored a context yet). +func (c *ControllerResourcesContext) ParentContext(fallback context.Context) context.Context { + if c != nil && c.Context != nil { + return c.Context + } + return fallback +} + // DeepCopy creates a new ControllerResourcesContext. // The Context field is preserved (not deep copied) since contexts are meant to be passed around. func (c *ControllerResourcesContext) DeepCopy() *ControllerResourcesContext { diff --git a/internal/gatewayapi/runner/runner.go b/internal/gatewayapi/runner/runner.go index 810a744a82..b7dfd46455 100644 --- a/internal/gatewayapi/runner/runner.go +++ b/internal/gatewayapi/runner/runner.go @@ -210,10 +210,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re func(update message.Update[string, *resource.ControllerResourcesContext], errChan chan error) { message.PublishRunnerEventMetric(r.Name(), update.Delete) - parentCtx := context.Background() - if update.Value != nil && update.Value.Context != nil { - parentCtx = update.Value.Context - } + parentCtx := update.Value.ParentContext(context.Background()) traceCtx, span := tracer.Start(parentCtx, "GatewayApiRunner.subscribeAndTranslate") defer span.End() @@ -287,7 +284,19 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re } span.AddEvent("translate", trace.WithAttributes(attribute.Int("resources.count", len(*val)))) + + rtcTraceCtx, rtcSpan := tracer.Start(traceCtx, "GatewayApiRunner.ResoureTranslationCycle") for _, resources := range *val { + // The GatewayClass name is deliberately kept out of the span name and passed as + // an attribute instead: span names are what most trace backends group/aggregate + // by (latency percentiles, error rates, etc.), and this loop runs once per + // GatewayClass in the cluster. Baking the name into the span name would fragment + // those aggregates into one bucket per class - unbounded and growing over the + // cluster's lifetime - for no benefit, since the attribute already makes the span + // filterable/searchable by GatewayClass in any trace UI. + translateGCCtx, translateGCSpan := tracer.Start(rtcTraceCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateGatewayClass", + trace.WithAttributes(attribute.String("gatewayclass.name", string(resources.GatewayClass.Name))), + ) // Translate and publish IRs. t := &gatewayapi.Translator{ GatewayControllerName: r.EnvoyGateway.Gateway.ControllerName, @@ -324,7 +333,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re traceLogger.Info("extension resources", "GVKs count", len(extGKs)) } // Translate to IR - _, translateToIRSpan := tracer.Start(traceCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateToIR") + _, translateToIRSpan := tracer.Start(translateGCCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateToIR") result, err := t.Translate(resources) translateToIRSpan.End() if err != nil { @@ -373,7 +382,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re } // Update Status - _, statusUpdateSpan := tracer.Start(traceCtx, "GatewayApiRunner.ResoureTranslationCycle.UpdateStatus") + _, statusUpdateSpan := tracer.Start(translateGCCtx, "GatewayApiRunner.ResoureTranslationCycle.UpdateStatus") if result.GatewayClass != nil { key := utils.NamespacedName(result.GatewayClass) r.ProviderResources.GatewayClassStatuses.Store(key, &result.GatewayClass.Status) @@ -488,6 +497,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)] = mergeEnvoyProxyStatus(aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)], &ep.Status) } statusUpdateSpan.End() + translateGCSpan.End() } // Store the stauses of all objects atomically with the aggregated status. @@ -604,6 +614,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re // Delete keys using mark and sweep r.deleteKeys(keysToDelete) + rtcSpan.End() }, ) r.Logger.Info("shutting down") diff --git a/internal/globalratelimit/runner/runner.go b/internal/globalratelimit/runner/runner.go index fd1bab3e02..45b7aae1ee 100644 --- a/internal/globalratelimit/runner/runner.go +++ b/internal/globalratelimit/runner/runner.go @@ -147,10 +147,7 @@ func (r *Runner) translateFromSubscription(ctx context.Context, c <-chan watchab func(update message.Update[string, *message.XdsIRWithContext], errChan chan error) { message.PublishRunnerEventMetric(r.Name(), update.Delete) - parentCtx := ctx - if update.Value != nil && update.Value.Context != nil { - parentCtx = update.Value.Context - } + parentCtx := update.Value.ParentContext(ctx) traceCtx, span := tracer.Start(parentCtx, "GlobalRateLimitRunner.translateFromSubscription") defer span.End() diff --git a/internal/message/types.go b/internal/message/types.go index c23d1fc440..0fb09a6248 100644 --- a/internal/message/types.go +++ b/internal/message/types.go @@ -143,6 +143,14 @@ type XdsIRWithContext struct { Context context.Context } +// ParentContext returns the trace context stashed on x, or fallback if x is nil or has none. +func (x *XdsIRWithContext) ParentContext(fallback context.Context) context.Context { + if x != nil && x.Context != nil { + return x.Context + } + return fallback +} + // DeepCopy creates a new ControllerResourcesContext. // The Context field is preserved (not deep copied) since contexts are meant to be passed around. func (x *XdsIRWithContext) DeepCopy() *XdsIRWithContext { diff --git a/internal/xds/runner/runner.go b/internal/xds/runner/runner.go index 7864adfa54..8d6644ac7e 100644 --- a/internal/xds/runner/runner.go +++ b/internal/xds/runner/runner.go @@ -274,10 +274,7 @@ func (r *Runner) translateFromSubscription(sub <-chan watchable.Snapshot[string, func(update message.Update[string, *message.XdsIRWithContext], errChan chan error) { message.PublishRunnerEventMetric(r.Name(), update.Delete) - parentCtx := context.Background() - if update.Value != nil && update.Value.Context != nil { - parentCtx = update.Value.Context - } + parentCtx := update.Value.ParentContext(context.Background()) traceCtx, span := tracer.Start(parentCtx, "XdsRunner.subscribeAndTranslate") defer span.End() From c0312dc984730ab1b2b2f6f773339cf1e5acb53e Mon Sep 17 00:00:00 2001 From: zirain Date: Wed, 5 Aug 2026 17:06:28 +0800 Subject: [PATCH 2/3] update Signed-off-by: zirain --- internal/gatewayapi/runner/runner.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/gatewayapi/runner/runner.go b/internal/gatewayapi/runner/runner.go index b7dfd46455..7e0018ff2f 100644 --- a/internal/gatewayapi/runner/runner.go +++ b/internal/gatewayapi/runner/runner.go @@ -295,7 +295,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re // cluster's lifetime - for no benefit, since the attribute already makes the span // filterable/searchable by GatewayClass in any trace UI. translateGCCtx, translateGCSpan := tracer.Start(rtcTraceCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateGatewayClass", - trace.WithAttributes(attribute.String("gatewayclass.name", string(resources.GatewayClass.Name))), + trace.WithAttributes(attribute.String("gatewayclass.name", resources.GatewayClass.Name)), ) // Translate and publish IRs. t := &gatewayapi.Translator{ @@ -374,7 +374,7 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re } else { m := message.XdsIRWithContext{ XdsIR: val, - Context: traceCtx, + Context: translateGCCtx, } r.XdsIR.Store(key, &m) xdsIRCount++ From 1fd321f7dac670a59694cf31b04410b5b9eb6a3e Mon Sep 17 00:00:00 2001 From: zirain Date: Thu, 6 Aug 2026 07:42:08 +0800 Subject: [PATCH 3/3] make sure that span end on panic Signed-off-by: zirain --- internal/gatewayapi/runner/runner.go | 376 +++++++++--------- .../gatewayapi/runner/runner_race_test.go | 89 +++++ 2 files changed, 278 insertions(+), 187 deletions(-) diff --git a/internal/gatewayapi/runner/runner.go b/internal/gatewayapi/runner/runner.go index 7e0018ff2f..cf04c7b1ee 100644 --- a/internal/gatewayapi/runner/runner.go +++ b/internal/gatewayapi/runner/runner.go @@ -286,218 +286,221 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re span.AddEvent("translate", trace.WithAttributes(attribute.Int("resources.count", len(*val)))) rtcTraceCtx, rtcSpan := tracer.Start(traceCtx, "GatewayApiRunner.ResoureTranslationCycle") + defer rtcSpan.End() for _, resources := range *val { - // The GatewayClass name is deliberately kept out of the span name and passed as - // an attribute instead: span names are what most trace backends group/aggregate - // by (latency percentiles, error rates, etc.), and this loop runs once per - // GatewayClass in the cluster. Baking the name into the span name would fragment - // those aggregates into one bucket per class - unbounded and growing over the - // cluster's lifetime - for no benefit, since the attribute already makes the span - // filterable/searchable by GatewayClass in any trace UI. - translateGCCtx, translateGCSpan := tracer.Start(rtcTraceCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateGatewayClass", - trace.WithAttributes(attribute.String("gatewayclass.name", resources.GatewayClass.Name)), - ) - // Translate and publish IRs. - t := &gatewayapi.Translator{ - GatewayControllerName: r.EnvoyGateway.Gateway.ControllerName, - GatewayClassName: gwapiv1.ObjectName(resources.GatewayClass.Name), - GlobalRateLimitEnabled: r.EnvoyGateway.RateLimit != nil, - EnvoyPatchPolicyEnabled: r.EnvoyGateway.ExtensionAPIs != nil && r.EnvoyGateway.ExtensionAPIs.EnableEnvoyPatchPolicy, - BackendEnabled: r.EnvoyGateway.ExtensionAPIs != nil && r.EnvoyGateway.ExtensionAPIs.EnableBackend, - SDSSecretRefEnabled: r.EnvoyGateway.ExtensionAPIs != nil && r.EnvoyGateway.ExtensionAPIs.EnableSDSSecretRef, - ControllerNamespace: r.ControllerNamespace, - GatewayNamespaceMode: r.EnvoyGateway.GatewayNamespaceMode(), - MergeGateways: gatewayapi.IsMergeGatewaysEnabled(resources), - MergeBackends: gatewayapi.IsMergeBackendsEnabled(resources), - PerResourceSystemCASecret: r.EnvoyGateway.RuntimeFlags.IsEnabled(egv1a1.PerResourceSystemCASecret), - WasmCache: r.wasmCache, - RunningOnHost: r.EnvoyGateway.Provider != nil && r.EnvoyGateway.Provider.IsRunningOnHost(), - InfraRemotelyManaged: r.EnvoyGateway.Provider != nil && r.EnvoyGateway.Provider.IsInfraManagedRemotely(), - Logger: traceLogger, - LuaEnvoyExtensionPolicyDisabled: r.EnvoyGateway.ExtensionAPIs.LuaDisabled(), - } - - // If extensions are loaded, pass their supported groups/kinds to the translator - if extensions := r.EnvoyGateway.GetExtensionManagers(); len(extensions) > 0 { - var extGKs []schema.GroupKind - for _, em := range extensions { - for _, gvk := range em.Resources { - extGKs = append(extGKs, schema.GroupKind{Group: gvk.Group, Kind: gvk.Kind}) - } - // Include backend resources in extension group kinds for custom backend support - for _, gvk := range em.BackendResources { - extGKs = append(extGKs, schema.GroupKind{Group: gvk.Group, Kind: gvk.Kind}) - } + func() { + // The GatewayClass name is deliberately kept out of the span name and passed as + // an attribute instead: span names are what most trace backends group/aggregate + // by (latency percentiles, error rates, etc.), and this loop runs once per + // GatewayClass in the cluster. Baking the name into the span name would fragment + // those aggregates into one bucket per class - unbounded and growing over the + // cluster's lifetime - for no benefit, since the attribute already makes the span + // filterable/searchable by GatewayClass in any trace UI. + translateGCCtx, translateGCSpan := tracer.Start(rtcTraceCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateGatewayClass", + trace.WithAttributes(attribute.String("gatewayclass.name", resources.GatewayClass.Name)), + ) + defer translateGCSpan.End() + // Translate and publish IRs. + t := &gatewayapi.Translator{ + GatewayControllerName: r.EnvoyGateway.Gateway.ControllerName, + GatewayClassName: gwapiv1.ObjectName(resources.GatewayClass.Name), + GlobalRateLimitEnabled: r.EnvoyGateway.RateLimit != nil, + EnvoyPatchPolicyEnabled: r.EnvoyGateway.ExtensionAPIs != nil && r.EnvoyGateway.ExtensionAPIs.EnableEnvoyPatchPolicy, + BackendEnabled: r.EnvoyGateway.ExtensionAPIs != nil && r.EnvoyGateway.ExtensionAPIs.EnableBackend, + SDSSecretRefEnabled: r.EnvoyGateway.ExtensionAPIs != nil && r.EnvoyGateway.ExtensionAPIs.EnableSDSSecretRef, + ControllerNamespace: r.ControllerNamespace, + GatewayNamespaceMode: r.EnvoyGateway.GatewayNamespaceMode(), + MergeGateways: gatewayapi.IsMergeGatewaysEnabled(resources), + MergeBackends: gatewayapi.IsMergeBackendsEnabled(resources), + PerResourceSystemCASecret: r.EnvoyGateway.RuntimeFlags.IsEnabled(egv1a1.PerResourceSystemCASecret), + WasmCache: r.wasmCache, + RunningOnHost: r.EnvoyGateway.Provider != nil && r.EnvoyGateway.Provider.IsRunningOnHost(), + InfraRemotelyManaged: r.EnvoyGateway.Provider != nil && r.EnvoyGateway.Provider.IsInfraManagedRemotely(), + Logger: traceLogger, + LuaEnvoyExtensionPolicyDisabled: r.EnvoyGateway.ExtensionAPIs.LuaDisabled(), } - t.ExtensionGroupKinds = extGKs - traceLogger.Info("extension resources", "GVKs count", len(extGKs)) - } - // Translate to IR - _, translateToIRSpan := tracer.Start(translateGCCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateToIR") - result, err := t.Translate(resources) - translateToIRSpan.End() - 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) - // Notify the main control loop about translation errors. This may be a critical error in standalone mode, so - // notify the control loop in case this needs to be handled. - r.RunnerErrors.Store(r.Name(), message.NewWatchableError(err)) - } - // Publish the IRs. - // Also validate the ir before sending it. - for key, val := range result.InfraIR { - logV := traceLogger.V(1).WithValues(string(message.InfraIRMessageName), key) - if logV.Enabled() { - logV.Info(val.JSONString()) + // If extensions are loaded, pass their supported groups/kinds to the translator + if extensions := r.EnvoyGateway.GetExtensionManagers(); len(extensions) > 0 { + var extGKs []schema.GroupKind + for _, em := range extensions { + for _, gvk := range em.Resources { + extGKs = append(extGKs, schema.GroupKind{Group: gvk.Group, Kind: gvk.Kind}) + } + // Include backend resources in extension group kinds for custom backend support + for _, gvk := range em.BackendResources { + extGKs = append(extGKs, schema.GroupKind{Group: gvk.Group, Kind: gvk.Kind}) + } + } + t.ExtensionGroupKinds = extGKs + traceLogger.Info("extension resources", "GVKs count", len(extGKs)) } - if err := val.Validate(); err != nil { - traceLogger.Error(err, "unable to validate infra ir, skipped sending it") - errChan <- err - } else { - r.InfraIR.Store(key, val) - infraIRCount++ - // Track IR key for mark and sweep - r.keyCache.IR[key] = true - delete(keysToDelete.IR, key) + // Translate to IR + _, translateToIRSpan := tracer.Start(translateGCCtx, "GatewayApiRunner.ResoureTranslationCycle.TranslateToIR") + defer translateToIRSpan.End() + result, err := t.Translate(resources) + 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) + // Notify the main control loop about translation errors. This may be a critical error in standalone mode, so + // notify the control loop in case this needs to be handled. + r.RunnerErrors.Store(r.Name(), message.NewWatchableError(err)) } - } - for key, val := range result.XdsIR { - logV := traceLogger.V(1).WithValues(string(message.XDSIRMessageName), key) - if logV.Enabled() { - logV.Info(val.JSONString()) + // Publish the IRs. + // Also validate the ir before sending it. + for key, val := range result.InfraIR { + logV := traceLogger.V(1).WithValues(string(message.InfraIRMessageName), key) + if logV.Enabled() { + logV.Info(val.JSONString()) + } + if err := val.Validate(); err != nil { + traceLogger.Error(err, "unable to validate infra ir, skipped sending it") + errChan <- err + } else { + r.InfraIR.Store(key, val) + infraIRCount++ + // Track IR key for mark and sweep + r.keyCache.IR[key] = true + delete(keysToDelete.IR, key) + } } - if err := val.Validate(); err != nil { - traceLogger.Error(err, "unable to validate xds ir, skipped sending it") - errChan <- err - } else { - m := message.XdsIRWithContext{ - XdsIR: val, - Context: translateGCCtx, + + for key, val := range result.XdsIR { + logV := traceLogger.V(1).WithValues(string(message.XDSIRMessageName), key) + if logV.Enabled() { + logV.Info(val.JSONString()) + } + if err := val.Validate(); err != nil { + traceLogger.Error(err, "unable to validate xds ir, skipped sending it") + errChan <- err + } else { + m := message.XdsIRWithContext{ + XdsIR: val, + Context: translateGCCtx, + } + r.XdsIR.Store(key, &m) + xdsIRCount++ } - r.XdsIR.Store(key, &m) - xdsIRCount++ } - } - // Update Status - _, statusUpdateSpan := tracer.Start(translateGCCtx, "GatewayApiRunner.ResoureTranslationCycle.UpdateStatus") - if result.GatewayClass != nil { - key := utils.NamespacedName(result.GatewayClass) - r.ProviderResources.GatewayClassStatuses.Store(key, &result.GatewayClass.Status) - } + // Update Status + _, statusUpdateSpan := tracer.Start(translateGCCtx, "GatewayApiRunner.ResoureTranslationCycle.UpdateStatus") + defer statusUpdateSpan.End() + if result.GatewayClass != nil { + key := utils.NamespacedName(result.GatewayClass) + r.ProviderResources.GatewayClassStatuses.Store(key, &result.GatewayClass.Status) + } - // Resources which can only belong to 1 GatewayClass (at most) get their statuses stored right away. - for _, gateway := range result.Gateways { - key := utils.NamespacedName(gateway) - r.ProviderResources.GatewayStatuses.Store(key, &gateway.Status) - gatewayStatusCount++ - delete(keysToDelete.GatewayStatus, key) - r.keyCache.GatewayStatus[key] = true - } - for _, listenerSet := range result.ListenerSets { - key := utils.NamespacedName(listenerSet) - r.ProviderResources.ListenerSetStatuses.Store(key, &listenerSet.Status) - listenerSetStatusCount++ - delete(keysToDelete.ListenerSetStatus, key) - r.keyCache.ListenerSetStatus[key] = true - } + // Resources which can only belong to 1 GatewayClass (at most) get their statuses stored right away. + for _, gateway := range result.Gateways { + key := utils.NamespacedName(gateway) + r.ProviderResources.GatewayStatuses.Store(key, &gateway.Status) + gatewayStatusCount++ + delete(keysToDelete.GatewayStatus, key) + r.keyCache.GatewayStatus[key] = true + } + for _, listenerSet := range result.ListenerSets { + key := utils.NamespacedName(listenerSet) + r.ProviderResources.ListenerSetStatuses.Store(key, &listenerSet.Status) + listenerSetStatusCount++ + delete(keysToDelete.ListenerSetStatus, key) + r.keyCache.ListenerSetStatus[key] = true + } - // Backend statuses have no parents, so they are not aggregated. - for _, backend := range result.Backends { - key := utils.NamespacedName(backend) - if len(backend.Status.Conditions) > 0 { - r.ProviderResources.BackendStatuses.Store(key, &backend.Status) - backendStatusCount++ + // Backend statuses have no parents, so they are not aggregated. + for _, backend := range result.Backends { + key := utils.NamespacedName(backend) + if len(backend.Status.Conditions) > 0 { + r.ProviderResources.BackendStatuses.Store(key, &backend.Status) + backendStatusCount++ + } + delete(keysToDelete.BackendStatus, key) + r.keyCache.BackendStatus[key] = true } - delete(keysToDelete.BackendStatus, key) - r.keyCache.BackendStatus[key] = true - } - // Resources which can belong to multiple GatewayClasses get their statuses aggregated, - // then stored once after iterating over all GatewayClasses. - for _, httpRoute := range result.HTTPRoutes { - if len(httpRoute.Status.Parents) != 0 { - key := utils.NamespacedName(httpRoute) - aggregatedStatuses.HTTPRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.HTTPRoutes[key], &httpRoute.Status.RouteStatus, httpRoute.Generation) + // Resources which can belong to multiple GatewayClasses get their statuses aggregated, + // then stored once after iterating over all GatewayClasses. + for _, httpRoute := range result.HTTPRoutes { + if len(httpRoute.Status.Parents) != 0 { + key := utils.NamespacedName(httpRoute) + aggregatedStatuses.HTTPRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.HTTPRoutes[key], &httpRoute.Status.RouteStatus, httpRoute.Generation) + } } - } - for _, grpcRoute := range result.GRPCRoutes { - if len(grpcRoute.Status.Parents) != 0 { - key := utils.NamespacedName(grpcRoute) - aggregatedStatuses.GRPCRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.GRPCRoutes[key], &grpcRoute.Status.RouteStatus, grpcRoute.Generation) + for _, grpcRoute := range result.GRPCRoutes { + if len(grpcRoute.Status.Parents) != 0 { + key := utils.NamespacedName(grpcRoute) + aggregatedStatuses.GRPCRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.GRPCRoutes[key], &grpcRoute.Status.RouteStatus, grpcRoute.Generation) + } } - } - for _, tlsRoute := range result.TLSRoutes { - if len(tlsRoute.Status.Parents) != 0 { - key := utils.NamespacedName(tlsRoute) - aggregatedStatuses.TLSRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.TLSRoutes[key], &tlsRoute.Status.RouteStatus, tlsRoute.Generation) + for _, tlsRoute := range result.TLSRoutes { + if len(tlsRoute.Status.Parents) != 0 { + key := utils.NamespacedName(tlsRoute) + aggregatedStatuses.TLSRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.TLSRoutes[key], &tlsRoute.Status.RouteStatus, tlsRoute.Generation) + } } - } - for _, tcpRoute := range result.TCPRoutes { - if len(tcpRoute.Status.Parents) != 0 { - key := utils.NamespacedName(tcpRoute) - aggregatedStatuses.TCPRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.TCPRoutes[key], &tcpRoute.Status.RouteStatus, tcpRoute.Generation) + for _, tcpRoute := range result.TCPRoutes { + if len(tcpRoute.Status.Parents) != 0 { + key := utils.NamespacedName(tcpRoute) + aggregatedStatuses.TCPRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.TCPRoutes[key], &tcpRoute.Status.RouteStatus, tcpRoute.Generation) + } } - } - for _, udpRoute := range result.UDPRoutes { - if len(udpRoute.Status.Parents) != 0 { - key := utils.NamespacedName(udpRoute) - aggregatedStatuses.UDPRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.UDPRoutes[key], &udpRoute.Status.RouteStatus, udpRoute.Generation) + for _, udpRoute := range result.UDPRoutes { + if len(udpRoute.Status.Parents) != 0 { + key := utils.NamespacedName(udpRoute) + aggregatedStatuses.UDPRoutes[key] = mergeAggregatedRouteStatus(aggregatedStatuses.UDPRoutes[key], &udpRoute.Status.RouteStatus, udpRoute.Generation) + } } - } - for _, backendTLSPolicy := range result.BackendTLSPolicies { - if len(backendTLSPolicy.Status.Ancestors) != 0 { - key := utils.NamespacedName(backendTLSPolicy) - aggregatedStatuses.BackendTLSPolicies[key] = mergePolicyStatus(aggregatedStatuses.BackendTLSPolicies[key], &backendTLSPolicy.Status, backendTLSPolicy.Generation) + for _, backendTLSPolicy := range result.BackendTLSPolicies { + if len(backendTLSPolicy.Status.Ancestors) != 0 { + key := utils.NamespacedName(backendTLSPolicy) + aggregatedStatuses.BackendTLSPolicies[key] = mergePolicyStatus(aggregatedStatuses.BackendTLSPolicies[key], &backendTLSPolicy.Status, backendTLSPolicy.Generation) + } } - } - for _, clientTrafficPolicy := range result.ClientTrafficPolicies { - if len(clientTrafficPolicy.Status.Ancestors) != 0 { - key := utils.NamespacedName(clientTrafficPolicy) - aggregatedStatuses.ClientTrafficPolicies[key] = mergePolicyStatus(aggregatedStatuses.ClientTrafficPolicies[key], &clientTrafficPolicy.Status, clientTrafficPolicy.Generation) + for _, clientTrafficPolicy := range result.ClientTrafficPolicies { + if len(clientTrafficPolicy.Status.Ancestors) != 0 { + key := utils.NamespacedName(clientTrafficPolicy) + aggregatedStatuses.ClientTrafficPolicies[key] = mergePolicyStatus(aggregatedStatuses.ClientTrafficPolicies[key], &clientTrafficPolicy.Status, clientTrafficPolicy.Generation) + } } - } - for _, backendTrafficPolicy := range result.BackendTrafficPolicies { - if len(backendTrafficPolicy.Status.Ancestors) != 0 { - key := utils.NamespacedName(backendTrafficPolicy) - aggregatedStatuses.BackendTrafficPolicies[key] = mergePolicyStatus(aggregatedStatuses.BackendTrafficPolicies[key], &backendTrafficPolicy.Status, backendTrafficPolicy.Generation) + for _, backendTrafficPolicy := range result.BackendTrafficPolicies { + if len(backendTrafficPolicy.Status.Ancestors) != 0 { + key := utils.NamespacedName(backendTrafficPolicy) + aggregatedStatuses.BackendTrafficPolicies[key] = mergePolicyStatus(aggregatedStatuses.BackendTrafficPolicies[key], &backendTrafficPolicy.Status, backendTrafficPolicy.Generation) + } } - } - for _, securityPolicy := range result.SecurityPolicies { - if len(securityPolicy.Status.Ancestors) != 0 { - key := utils.NamespacedName(securityPolicy) - aggregatedStatuses.SecurityPolicies[key] = mergePolicyStatus(aggregatedStatuses.SecurityPolicies[key], &securityPolicy.Status, securityPolicy.Generation) + for _, securityPolicy := range result.SecurityPolicies { + if len(securityPolicy.Status.Ancestors) != 0 { + key := utils.NamespacedName(securityPolicy) + aggregatedStatuses.SecurityPolicies[key] = mergePolicyStatus(aggregatedStatuses.SecurityPolicies[key], &securityPolicy.Status, securityPolicy.Generation) + } } - } - for _, envoyExtensionPolicy := range result.EnvoyExtensionPolicies { - if len(envoyExtensionPolicy.Status.Ancestors) != 0 { - key := utils.NamespacedName(envoyExtensionPolicy) - aggregatedStatuses.EnvoyExtensionPolicies[key] = mergePolicyStatus(aggregatedStatuses.EnvoyExtensionPolicies[key], &envoyExtensionPolicy.Status, envoyExtensionPolicy.Generation) + for _, envoyExtensionPolicy := range result.EnvoyExtensionPolicies { + if len(envoyExtensionPolicy.Status.Ancestors) != 0 { + key := utils.NamespacedName(envoyExtensionPolicy) + aggregatedStatuses.EnvoyExtensionPolicies[key] = mergePolicyStatus(aggregatedStatuses.EnvoyExtensionPolicies[key], &envoyExtensionPolicy.Status, envoyExtensionPolicy.Generation) + } } - } - for _, extServerPolicy := range result.ExtensionServerPolicies { - policyStatus := gatewayapi.ExtServerPolicyStatusAsPolicyStatus(&extServerPolicy) - if len(policyStatus.Ancestors) != 0 { - key := message.NamespacedNameAndGVK{ - NamespacedName: utils.NamespacedName(&extServerPolicy), - GroupVersionKind: extServerPolicy.GroupVersionKind(), + for _, extServerPolicy := range result.ExtensionServerPolicies { + policyStatus := gatewayapi.ExtServerPolicyStatusAsPolicyStatus(&extServerPolicy) + if len(policyStatus.Ancestors) != 0 { + key := message.NamespacedNameAndGVK{ + NamespacedName: utils.NamespacedName(&extServerPolicy), + GroupVersionKind: extServerPolicy.GroupVersionKind(), + } + aggregatedStatuses.ExtensionServerPolicies[key] = mergePolicyStatus(aggregatedStatuses.ExtensionServerPolicies[key], &policyStatus, extServerPolicy.GetGeneration()) } - aggregatedStatuses.ExtensionServerPolicies[key] = mergePolicyStatus(aggregatedStatuses.ExtensionServerPolicies[key], &policyStatus, extServerPolicy.GetGeneration()) } - } - // EnvoyProxy status - for _, ep := range result.EnvoyProxiesForGateways { - r.Logger.Info("update envoyproxy status", "key", utils.NamespacedName(ep)) - aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)] = mergeEnvoyProxyStatus(aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)], &ep.Status) - } - if ep := result.EnvoyProxyForGatewayClass; ep != nil { - r.Logger.Info("update envoyproxy status", "key", utils.NamespacedName(ep)) - aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)] = mergeEnvoyProxyStatus(aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)], &ep.Status) - } - statusUpdateSpan.End() - translateGCSpan.End() + // EnvoyProxy status + for _, ep := range result.EnvoyProxiesForGateways { + r.Logger.Info("update envoyproxy status", "key", utils.NamespacedName(ep)) + aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)] = mergeEnvoyProxyStatus(aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)], &ep.Status) + } + if ep := result.EnvoyProxyForGatewayClass; ep != nil { + r.Logger.Info("update envoyproxy status", "key", utils.NamespacedName(ep)) + aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)] = mergeEnvoyProxyStatus(aggregatedStatuses.EnvoyProxies[utils.NamespacedName(ep)], &ep.Status) + } + }() } // Store the stauses of all objects atomically with the aggregated status. @@ -614,7 +617,6 @@ func (r *Runner) subscribeAndTranslate(sub <-chan watchable.Snapshot[string, *re // Delete keys using mark and sweep r.deleteKeys(keysToDelete) - rtcSpan.End() }, ) r.Logger.Info("shutting down") diff --git a/internal/gatewayapi/runner/runner_race_test.go b/internal/gatewayapi/runner/runner_race_test.go index 6964a168d4..4fa1a5501c 100644 --- a/internal/gatewayapi/runner/runner_race_test.go +++ b/internal/gatewayapi/runner/runner_race_test.go @@ -7,12 +7,17 @@ package runner import ( "context" + "io" "sync" "testing" "time" "github.com/stretchr/testify/require" "github.com/telepresenceio/watchable" + "go.opentelemetry.io/otel" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + gwapiv1 "sigs.k8s.io/gateway-api/apis/v1" egv1a1 "github.com/envoyproxy/gateway/api/v1alpha1" "github.com/envoyproxy/gateway/internal/envoygateway/config" @@ -233,3 +238,87 @@ func TestRunnerDataRaceImmediate(t *testing.T) { // Race window is maximized here } + +func TestSubscribeAndTranslateEndsGatewayClassSpansOnPanic(t *testing.T) { + sr := tracetest.NewSpanRecorder() + tp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr)) + oldProvider := otel.GetTracerProvider() + otel.SetTracerProvider(tp) + t.Cleanup(func() { + otel.SetTracerProvider(oldProvider) + }) + tracer = otel.Tracer("envoy-gateway/gateway-api") + + serverCfg := &config.Server{ + EnvoyGateway: &egv1a1.EnvoyGateway{ + EnvoyGatewaySpec: egv1a1.EnvoyGatewaySpec{ + Gateway: &egv1a1.Gateway{ControllerName: "test-controller"}, + }, + }, + Logger: logging.DefaultLogger(io.Discard, egv1a1.LogLevelInfo), + } + + providerResources := new(message.ProviderResources) + providerResources.GatewayAPIResources = watchable.Map[string, *resource.ControllerResourcesContext]{} + + cfg := &Config{ + Server: *serverCfg, + ProviderResources: providerResources, + XdsIR: new(message.XdsIR), + InfraIR: new(message.InfraIR), + RunnerErrors: new(message.RunnerErrors), + } + + runner := New(cfg) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + sub := providerResources.GatewayAPIResources.Subscribe(ctx) + done := make(chan struct{}) + go func() { + runner.subscribeAndTranslate(sub) + close(done) + }() + + providerResources.GatewayAPIResources.Store("test-controller", &resource.ControllerResourcesContext{ + Context: ctx, + Resources: &resource.ControllerResources{ + &resource.Resources{ + GatewayClass: &gwapiv1.GatewayClass{}, + Gateways: []*gwapiv1.Gateway{nil}, + }, + }, + }) + + require.Eventually(t, func() bool { + for _, span := range sr.Ended() { + if span.Name() == "GatewayApiRunner.ResoureTranslationCycle.TranslateGatewayClass" || span.Name() == "GatewayApiRunner.ResoureTranslationCycle" { + return true + } + } + return false + }, time.Second, 10*time.Millisecond) + + cancel() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timed out waiting for subscription handler to exit") + } + + ended := sr.Ended() + require.NotEmpty(t, ended) + + var gcSpanEnded, rtcSpanEnded bool + for _, span := range ended { + switch span.Name() { + case "GatewayApiRunner.ResoureTranslationCycle.TranslateGatewayClass": + gcSpanEnded = true + case "GatewayApiRunner.ResoureTranslationCycle": + rtcSpanEnded = true + } + } + + require.True(t, gcSpanEnded, "expected GatewayClass translation span to be ended") + require.True(t, rtcSpanEnded, "expected translation-cycle span to be ended") +}