add support for enabling telemetry forwarding to consul-telemetry-collector

internal/commands/exec/command.go

@@ -59,9 +59,10 @@ type Command struct {
 	flagGatewayNamespace string // Gateway namespace.
 
 	// Envoy params
-	flagBootstrapPath    string // Path for config file for bootstrapping envoy
-	flagSDSServerAddress string // Address for the SDS server
-	flagSDSServerPort    int    // Port for the SDS server
+	flagBootstrapPath               string // Path for config file for bootstrapping envoy
+	flagSDSServerAddress            string // Address for the SDS server
+	flagSDSServerPort               int    // Port for the SDS server
+	flagEnvoyTelemetryBindSocketDir string // Enables telemetry forwarding to consul-telemetry-collector
 
 	// ACL Auth
 	flagACLAuthMethod       string // Auth Method to use for ACLs, if enabled.
@@ -101,6 +102,7 @@ func (c *Command) init() {
 		c.flagSet.StringVar(&c.flagBootstrapPath, "envoy-bootstrap-path", "", "Path to the config file for bootstrapping Envoy.")
 		c.flagSet.StringVar(&c.flagSDSServerAddress, "envoy-sds-address", "", "Address of the SDS server.")
 		c.flagSet.IntVar(&c.flagSDSServerPort, "envoy-sds-port", 9090, "Port of the SDS server.")
+		c.flagSet.StringVar(&c.flagEnvoyTelemetryBindSocketDir, "envoy-telemetry-bind-socket-dir", "", "Dir to bind socket for telemetry forwarding to consul-telemetry-collector")
 	}
 	{
 		// Gateway
@@ -201,6 +203,11 @@ func (c *Command) Run(args []string) (ret int) {
 		}
 	}
 
+	proxyConfig := map[string]interface{}{}
+	if c.flagEnvoyTelemetryBindSocketDir != "" {
+		proxyConfig["envoy_telemetry_bind_socket_dir"] = c.flagEnvoyTelemetryBindSocketDir
+	}
+
 	return RunExec(ExecConfig{
 		Context:           c.ctx,
 		Logger:            logger,
@@ -230,6 +237,7 @@ func (c *Command) Run(args []string) (ret int) {
 			Output:            c.output,
 		},
 		ConsulClientConfig: consulClientConfig,
+		ProxyConfig:        proxyConfig,
 		isTest:             c.isTest,
 	})
 }

internal/commands/exec/exec.go

@@ -57,6 +57,7 @@ type ExecConfig struct {
 	EnvoyConfig        EnvoyConfig
 	ConsulClientConfig consul.ClientConfig
 	PrimaryDatacenter  string
+	ProxyConfig        map[string]interface{}
 
 	// for testing only
 	isTest bool
@@ -104,7 +105,7 @@ func RunExec(config ExecConfig) (ret int) {
 		config.GatewayConfig.Namespace,
 		config.GatewayConfig.Partition,
 		config.GatewayConfig.Host,
-	)
+	).WithProxyConfig(config.ProxyConfig)
 	if config.isTest {
 		registry = registry.WithTries(1)
 	}

internal/commands/server/command.go

@@ -56,6 +56,8 @@ type Command struct {
 	flagK8sContext    string // context to use
 	flagK8sNamespace  string // namespace we're run in
 
+	flagEnableTelemetryCollector bool // toggles adding proxy config to enable telemetry collection by consul-telemetry-collector
+
 	// Consul namespaces
 	flagConsulDestinationNamespace string
 	flagMirrorK8SNamespaces        bool
@@ -87,6 +89,7 @@ func (c *Command) init() {
 	c.flagSet.IntVar(&c.flagSDSServerPort, "sds-server-port", defaultSDSServerPort, "SDS Server Port.")
 	c.flagSet.IntVar(&c.flagMetricsPort, "metrics-port", 0, "Metrics port, if not set, metrics are not enabled.")
 	c.flagSet.IntVar(&c.flagPprofPort, "pprof-port", 0, "Go pprof port, if not set, profiling is not enabled.")
+	c.flagSet.BoolVar(&c.flagEnableTelemetryCollector, "enable-telemetry-collector", false, "Enables telemetry forwarding to consul-telemetry-collector")
 
 	{
 		// Consul namespaces
@@ -229,15 +232,16 @@ func (c *Command) Run(args []string) int {
 	}
 
 	return RunServer(ServerConfig{
-		Context:            context.Background(),
-		Logger:             logger,
-		ConsulConfig:       consulCfg,
-		K8sConfig:          cfg,
-		ProfilingPort:      c.flagPprofPort,
-		MetricsPort:        c.flagMetricsPort,
-		PrimaryDatacenter:  c.flagPrimaryDatacenter,
-		isTest:             c.isTest,
-		ConsulClientConfig: consulClientConfig,
+		Context:                  context.Background(),
+		Logger:                   logger,
+		ConsulConfig:             consulCfg,
+		K8sConfig:                cfg,
+		ProfilingPort:            c.flagPprofPort,
+		MetricsPort:              c.flagMetricsPort,
+		PrimaryDatacenter:        c.flagPrimaryDatacenter,
+		isTest:                   c.isTest,
+		ConsulClientConfig:       consulClientConfig,
+		EnableTelemetryCollector: c.flagEnableTelemetryCollector,
 	})
 }
 

internal/commands/server/server.go

@@ -43,6 +43,8 @@ type ServerConfig struct {
 	MetricsPort        int
 	PrimaryDatacenter  string
 
+	EnableTelemetryCollector bool
+
 	// for testing only
 	isTest bool
 }

internal/consul/registration.go

@@ -31,12 +31,13 @@ type ServiceRegistry struct {
 	client Client
 	logger hclog.Logger
 
-	id        string
-	name      string
-	namespace string
-	partition string
-	host      string
-	tags      []string
+	id          string
+	name        string
+	namespace   string
+	partition   string
+	host        string
+	tags        []string
+	proxyConfig map[string]interface{}
 
 	cancel                 context.CancelFunc
 	tries                  uint64
@@ -74,6 +75,12 @@ func (s *ServiceRegistry) WithTries(tries uint64) *ServiceRegistry {
 	return s
 }
 
+// WithProxyConfig adds proxy config map to the gateway service registration
+func (s *ServiceRegistry) WithProxyConfig(cfg map[string]interface{}) *ServiceRegistry {
+	s.proxyConfig = cfg
+	return s
+}
+
 // Register registers a Gateway service with Consul.
 func (s *ServiceRegistry) RegisterGateway(ctx context.Context, ttl bool) error {
 	serviceChecks := api.AgentServiceChecks{{
@@ -103,6 +110,9 @@ func (s *ServiceRegistry) RegisterGateway(ctx context.Context, ttl bool) error {
 			"external-source": "consul-api-gateway",
 		},
 		Checks: serviceChecks,
+		Proxy: &api.AgentServiceConnectProxyConfig{
+			Config: s.proxyConfig,
+		},
 	}, ttl)
 }
 

internal/k8s/builder/builder.go

@@ -41,6 +41,8 @@ const (
 	consulCALocalPath = "/consul/tls"
 	consulCAFilename  = "ca.pem"
 
+	envoyTelemetryBindSocketDir = "/consul/envoy-telemetry"
+
 	k8sHostnameTopologyKey = "kubernetes.io/hostname"
 )
 

internal/k8s/builder/gateway.go

@@ -80,13 +80,14 @@ func filterAnnotations(annotations map[string]string, allowed []string) map[stri
 }
 
 type GatewayDeploymentBuilder struct {
-	gateway                 *gwv1beta1.Gateway
-	gwConfig                *v1alpha1.GatewayClassConfig
-	sdsHost                 string
-	sdsPort                 int
-	consulCAData            string
-	consulGatewayNamespace  string
-	consulPrimaryDatacenter string
+	gateway                  *gwv1beta1.Gateway
+	gwConfig                 *v1alpha1.GatewayClassConfig
+	sdsHost                  string
+	sdsPort                  int
+	consulCAData             string
+	consulGatewayNamespace   string
+	consulPrimaryDatacenter  string
+	enableTelemetryCollector bool
 }
 
 func NewGatewayDeployment(gw *gwv1beta1.Gateway) *GatewayDeploymentBuilder {
@@ -119,6 +120,11 @@ func (b *GatewayDeploymentBuilder) WithPrimaryConsulDatacenter(datacenter string
 	return b
 }
 
+func (b *GatewayDeploymentBuilder) WithEnableTelemetryCollector(enabled bool) *GatewayDeploymentBuilder {
+	b.enableTelemetryCollector = enabled
+	return b
+}
+
 func (b *GatewayDeploymentBuilder) Build(currentReplicas *int32) *v1.Deployment {
 	labels := utils.LabelsForGateway(b.gateway)
 
@@ -263,6 +269,9 @@ func (b *GatewayDeploymentBuilder) execArgs() []string {
 	if method := b.gwConfig.Spec.ConsulSpec.AuthSpec.Method; method != "" {
 		data.ACLAuthMethod = method
 	}
+	if b.enableTelemetryCollector {
+		data.EnvoyTelemetryBindSocketDir = envoyTelemetryBindSocketDir
+	}
 	var buf bytes.Buffer
 	err := template.Must(template.New("root").
 		Parse(strings.TrimSpace(gwContainerArgsTpl))).
@@ -369,6 +378,19 @@ func (b *GatewayDeploymentBuilder) volumes() ([]corev1.Volume, []corev1.VolumeMo
 			MountPath: consulCALocalPath,
 		})
 	}
+
+	if b.enableTelemetryCollector {
+		volumes = append(volumes, corev1.Volume{
+			Name: "envoy_telemetry",
+			VolumeSource: corev1.VolumeSource{
+				EmptyDir: &corev1.EmptyDirVolumeSource{},
+			},
+		})
+		mounts = append(mounts, corev1.VolumeMount{
+			Name:      "envoy_telemetry",
+			MountPath: envoyTelemetryBindSocketDir,
+		})
+	}
 	return volumes, mounts
 }
 
@@ -397,19 +419,20 @@ func (b *GatewayDeploymentBuilder) requiresCA() bool {
 }
 
 type gwContainerCommandData struct {
-	ConsulCAFile      string
-	ConsulCAData      string
-	ConsulHTTPAddr    string
-	ConsulHTTPPort    string
-	ConsulGRPCPort    string
-	ACLAuthMethod     string
-	LogLevel          string
-	GatewayHost       string
-	GatewayName       string
-	GatewayNamespace  string
-	PrimaryDatacenter string
-	SDSHost           string
-	SDSPort           int
+	ConsulCAFile                string
+	ConsulCAData                string
+	ConsulHTTPAddr              string
+	ConsulHTTPPort              string
+	ConsulGRPCPort              string
+	ACLAuthMethod               string
+	LogLevel                    string
+	GatewayHost                 string
+	GatewayName                 string
+	GatewayNamespace            string
+	PrimaryDatacenter           string
+	SDSHost                     string
+	SDSPort                     int
+	EnvoyTelemetryBindSocketDir string
 }
 
 // gwContainerArgsTpl is the template for the command arguments executed in the Envoy container.
@@ -447,4 +470,8 @@ const gwContainerArgsTpl = `
 {{ .SDSHost }}
 -envoy-sds-port
 {{ .SDSPort }}
+{{- if .EnvoyTelemetryBindSocketDir }}
+-envoy-telemetry-bind-socket-dir
+{{ .EnvoyTelemetryBindSocketDir }}
+{{- end }}
 `

internal/k8s/controller.go

@@ -90,6 +90,8 @@ type Config struct {
 	RestConfig          *rest.Config
 	Namespace           string
 
+	EnableTelemetryCollector bool
+
 	// ConsulNamespaceConfig
 	ConsulNamespaceConfig ConsulNamespaceConfig
 }
@@ -177,6 +179,7 @@ func (k *Kubernetes) Start(ctx context.Context) error {
 		Store:                    k.store,
 		ConsulNamespaceMapper:    k.config.ConsulNamespaceConfig.Namespace,
 		ConsulNamespaceMirroring: k.config.ConsulNamespaceConfig.MirrorKubernetesNamespaces,
+		EnableTelemetryCollector: k.config.EnableTelemetryCollector,
 	})
 
 	err := (&controllers.GatewayClassConfigReconciler{

internal/k8s/reconciler/deployer.go

@@ -33,6 +33,7 @@ type GatewayDeployer struct {
 	consul                   consul.Client
 	consulNamespaceMirroring bool
 	partitionInfo            consul.PartitionInfo
+	enableTelemetryCollector bool
 
 	logger hclog.Logger
 }
@@ -47,6 +48,7 @@ type DeployerConfig struct {
 	Consul                   consul.Client
 	ConsulNamespaceMirroring bool
 	ConsulPartitionInfo      consul.PartitionInfo
+	EnableTelemetryCollector bool
 }
 
 func NewDeployer(config DeployerConfig) *GatewayDeployer {
@@ -60,6 +62,7 @@ func NewDeployer(config DeployerConfig) *GatewayDeployer {
 		consul:                   config.Consul,
 		consulNamespaceMirroring: config.ConsulNamespaceMirroring,
 		partitionInfo:            config.ConsulPartitionInfo,
+		enableTelemetryCollector: config.EnableTelemetryCollector,
 	}
 }
 
@@ -215,6 +218,7 @@ func (d *GatewayDeployer) Deployment(namespace string, config apigwv1alpha1.Gate
 		WithConsulCA(d.consulCA).
 		WithConsulGatewayNamespace(namespace).
 		WithPrimaryConsulDatacenter(d.primaryDatacenter).
+		WithEnableTelemetryCollector(d.enableTelemetryCollector).
 		Build(currentReplicas)
 }
 

internal/k8s/reconciler/manager.go

@@ -79,6 +79,7 @@ type ManagerConfig struct {
 	Logger                   hclog.Logger
 	ConsulNamespaceMapper    common.ConsulNamespaceMapper
 	ConsulNamespaceMirroring bool
+	EnableTelemetryCollector bool
 }
 
 func NewReconcileManager(config ManagerConfig) *GatewayReconcileManager {
@@ -92,6 +93,7 @@ func NewReconcileManager(config ManagerConfig) *GatewayReconcileManager {
 		Client:                   config.Client,
 		Consul:                   config.Consul,
 		ConsulNamespaceMirroring: config.ConsulNamespaceMirroring,
+		EnableTelemetryCollector: config.EnableTelemetryCollector,
 	})
 
 	return &GatewayReconcileManager{