refactor telemetry backendRefs

internal/gatewayapi/contexts.go

@@ -187,6 +187,7 @@ func GetRouteType(route RouteContext) gwapiv1.Kind {
 
 // TODO: [v1alpha2-gwapiv1] This should not be required once all Route
 // objects being implemented are of type gwapiv1.
+
 // GetHostnames returns the hosts targeted by the Route object.
 func GetHostnames(route RouteContext) []string {
 	rv := reflect.ValueOf(route).Elem()

internal/gatewayapi/listener.go

@@ -6,8 +6,10 @@
 package gatewayapi
 
 import (
+	"errors"
 	"fmt"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	gwapiv1 "sigs.k8s.io/gateway-api/apis/v1"
 
@@ -16,7 +18,6 @@ import (
 	"github.com/envoyproxy/gateway/internal/ir"
 	"github.com/envoyproxy/gateway/internal/utils"
 	"github.com/envoyproxy/gateway/internal/utils/naming"
-	"github.com/envoyproxy/gateway/internal/utils/net"
 )
 
 var _ ListenersTranslator = (*Translator)(nil)
@@ -44,7 +45,7 @@ func (t *Translator) ProcessListeners(gateways []*GatewayContext, xdsIR XdsIRMap
 		if resources.EnvoyProxy != nil {
 			infraIR[irKey].Proxy.Config = resources.EnvoyProxy
 		}
-		t.processProxyObservability(gateway.Gateway, xdsIR[irKey], infraIR[irKey].Proxy.Config)
+		t.processProxyObservability(gateway, xdsIR[irKey], infraIR[irKey].Proxy.Config, resources)
 
 		for _, listener := range gateway.listeners {
 			// Process protocol & supported kinds
@@ -136,10 +137,31 @@ func (t *Translator) ProcessListeners(gateways []*GatewayContext, xdsIR XdsIRMap
 	}
 }
 
-func (t *Translator) processProxyObservability(gw *gwapiv1.Gateway, xdsIR *ir.Xds, envoyProxy *egv1a1.EnvoyProxy) {
-	xdsIR.AccessLog = processAccessLog(envoyProxy)
-	xdsIR.Tracing = processTracing(gw, envoyProxy, t.MergeGateways)
-	xdsIR.Metrics = processMetrics(envoyProxy)
+func (t *Translator) processProxyObservability(gwCtx *GatewayContext, xdsIR *ir.Xds, envoyProxy *egv1a1.EnvoyProxy, resources *Resources) {
+	var err error
+
+	xdsIR.AccessLog, err = t.processAccessLog(envoyProxy, resources)
+	if err != nil {
+		status.UpdateGatewayListenersNotValidCondition(gwCtx.Gateway, gwapiv1.GatewayReasonInvalid, metav1.ConditionFalse,
+			fmt.Sprintf("Invalid access log backendRefs: %v", err))
+		return
+	}
+
+	xdsIR.Tracing, err = t.processTracing(gwCtx.Gateway, envoyProxy, t.MergeGateways, resources)
+	if err != nil {
+		status.UpdateGatewayListenersNotValidCondition(gwCtx.Gateway, gwapiv1.GatewayReasonInvalid, metav1.ConditionFalse,
+			fmt.Sprintf("Invalid tracing backendRefs: %v", err))
+		return
+	}
+
+	xdsIR.Metrics, err = t.processMetrics(envoyProxy, resources)
+	if err != nil {
+		status.UpdateGatewayListenersNotValidCondition(gwCtx.Gateway, gwapiv1.GatewayReasonInvalid, metav1.ConditionFalse,
+			fmt.Sprintf("Invalid metrics backendRefs: %v", err))
+		return
+	}
+
+	status.RemoveGatewayListenersNotValidCondition(gwCtx.Gateway)
 }
 
 func (t *Translator) processInfraIRListener(listener *ListenerContext, infraIR InfraIRMap, irKey string, servicePort *protocolPort) {
@@ -172,7 +194,7 @@ func (t *Translator) processInfraIRListener(listener *ListenerContext, infraIR I
 	infraIR[irKey].Proxy.Listeners = append(infraIR[irKey].Proxy.Listeners, proxyListener)
 }
 
-func processAccessLog(envoyproxy *egv1a1.EnvoyProxy) *ir.AccessLog {
+func (t *Translator) processAccessLog(envoyproxy *egv1a1.EnvoyProxy, resources *Resources) (*ir.AccessLog, error) {
 	if envoyproxy == nil ||
 		envoyproxy.Spec.Telemetry == nil ||
 		envoyproxy.Spec.Telemetry.AccessLog == nil ||
@@ -184,11 +206,11 @@ func processAccessLog(envoyproxy *egv1a1.EnvoyProxy) *ir.AccessLog {
 					Path: "/dev/stdout",
 				},
 			},
-		}
+		}, nil
 	}
 
 	if envoyproxy.Spec.Telemetry.AccessLog.Disable {
-		return nil
+		return nil, nil
 	}
 
 	irAccessLog := &ir.AccessLog{}
@@ -236,7 +258,11 @@ func processAccessLog(envoyproxy *egv1a1.EnvoyProxy) *ir.AccessLog {
 				}
 
 				if len(sink.OpenTelemetry.BackendRefs) > 0 {
-					al.Host, al.Port = net.BackendHostAndPort(sink.OpenTelemetry.BackendRefs[0].BackendObjectReference, envoyproxy.Namespace)
+					ds, err := t.processBackendRefs(sink.OpenTelemetry.BackendRefs, envoyproxy.Namespace, resources)
+					if err != nil {
+						return nil, err
+					}
+					al.Destinations = ds
 				}
 
 				switch accessLog.Format.Type {
@@ -251,14 +277,14 @@ func processAccessLog(envoyproxy *egv1a1.EnvoyProxy) *ir.AccessLog {
 		}
 	}
 
-	return irAccessLog
+	return irAccessLog, nil
 }
 
-func processTracing(gw *gwapiv1.Gateway, envoyproxy *egv1a1.EnvoyProxy, mergeGateways bool) *ir.Tracing {
+func (t *Translator) processTracing(gw *gwapiv1.Gateway, envoyproxy *egv1a1.EnvoyProxy, mergeGateways bool, resources *Resources) (*ir.Tracing, error) {
 	if envoyproxy == nil ||
 		envoyproxy.Spec.Telemetry == nil ||
 		envoyproxy.Spec.Telemetry.Tracing == nil {
-		return nil
+		return nil, nil
 	}
 	tracing := envoyproxy.Spec.Telemetry.Tracing
 
@@ -268,8 +294,10 @@ func processTracing(gw *gwapiv1.Gateway, envoyproxy *egv1a1.EnvoyProxy, mergeGat
 	if tracing.Provider.Host != nil {
 		host, port = *tracing.Provider.Host, uint32(tracing.Provider.Port)
 	}
-	if len(tracing.Provider.BackendRefs) > 0 {
-		host, port = net.BackendHostAndPort(tracing.Provider.BackendRefs[0].BackendObjectReference, gw.Namespace)
+
+	ds, err := t.processBackendRefs(tracing.Provider.BackendRefs, envoyproxy.Namespace, resources)
+	if err != nil {
+		return nil, err
 	}
 
 	samplingRate := 100.0
@@ -283,22 +311,56 @@ func processTracing(gw *gwapiv1.Gateway, envoyproxy *egv1a1.EnvoyProxy, mergeGat
 	}
 
 	return &ir.Tracing{
-		ServiceName:  serviceName,
-		Host:         host,
-		Port:         port,
-		SamplingRate: samplingRate,
-		CustomTags:   tracing.CustomTags,
-	}
+		ServiceName:         serviceName,
+		Host:                host,
+		Port:                port,
+		SamplingRate:        samplingRate,
+		CustomTags:          tracing.CustomTags,
+		DestinationSettings: ds,
+	}, nil
 }
 
-func processMetrics(envoyproxy *egv1a1.EnvoyProxy) *ir.Metrics {
+func (t *Translator) processMetrics(envoyproxy *egv1a1.EnvoyProxy, resources *Resources) (*ir.Metrics, error) {
 	if envoyproxy == nil ||
 		envoyproxy.Spec.Telemetry == nil ||
 		envoyproxy.Spec.Telemetry.Metrics == nil {
-		return nil
+		return nil, nil
 	}
+
+	for _, sink := range envoyproxy.Spec.Telemetry.Metrics.Sinks {
+		if sink.OpenTelemetry == nil {
+			continue
+		}
+
+		_, err := t.processBackendRefs(sink.OpenTelemetry.BackendRefs, envoyproxy.Namespace, resources)
+		if err != nil {
+			return nil, err
+		}
+	}
+
 	return &ir.Metrics{
 		EnableVirtualHostStats: envoyproxy.Spec.Telemetry.Metrics.EnableVirtualHostStats,
 		EnablePerEndpointStats: envoyproxy.Spec.Telemetry.Metrics.EnablePerEndpointStats,
+	}, nil
+}
+
+func (t *Translator) processBackendRefs(backendRefs []egv1a1.BackendRef, namespace string, resources *Resources) ([]*ir.DestinationSetting, error) {
+	result := make([]*ir.DestinationSetting, 0, len(backendRefs))
+	for _, ref := range backendRefs {
+		ns := NamespaceDerefOr(ref.Namespace, namespace)
+		kind := KindDerefOr(ref.Kind, KindService)
+		if kind != KindService {
+			return nil, errors.New("only service kind is supported for backendRefs")
+		}
+		if err := validateBackendService(ref.BackendObjectReference, resources, ns, corev1.ProtocolTCP); err != nil {
+			return nil, err
+		}
+
+		ds := t.serviceDestinationSetting(ref.BackendObjectReference, ns, ir.GRPC, resources)
+		result = append(result, ds)
+	}
+	if len(result) == 0 {
+		return nil, nil
 	}
+	return result, nil
 }