feat(kuma-cp) add datadog traffic tracing (#2247)

* feat(kuma-cp) Add datadog traffic tracing Signed-off-by: Paul Parkanzky <paul.parkanzky@konghq.com> * feat(kuma-cp) Address datadog integration PR comments Signed-off-by: Paul Parkanzky <paul.parkanzky@konghq.com> * feat(kuma-cp) Fix tracing perspective Signed-off-by: Paul Parkanzky <paul.parkanzky@konghq.com>
kumahq · Jul 13, 2021 · 6a38918 · 6a38918
1 parent 0f1e4ab
commit 6a38918
Show file tree

Hide file tree

Showing 13 changed files with 327 additions and 123 deletions.
diff --git a/DEVELOPER.md b/DEVELOPER.md
@@ -50,7 +50,7 @@ docker run --name kuma-build -ti \
   --volume `pwd`:/go/src/github.com/kumahq/kuma \
   --workdir /go/src/github.com/kumahq/kuma \
   --env GO111MODULE=on \
-  golang:1.12.12 \
+  golang:1.16 \
   bash -c 'apt update && apt install -y unzip && export PATH=$HOME/bin:$PATH && bash'
 ```
 

diff --git a/api/DEVELOPER.md b/api/DEVELOPER.md
@@ -16,7 +16,7 @@ docker run --rm -ti \
   --workdir /go/src/github.com/kumahq/kuma/api \
   --env HOME=/tmp/home \
   --env GO111MODULE=on \
-  golang:1.12.12 bash
+  golang:1.16 bash
 export PATH=$HOME/bin:$PATH
 apt update && apt install unzip
 ```
@@ -49,4 +49,4 @@ Run:
 
 ```bash
 make build
-```
+```
diff --git a/api/mesh/v1alpha1/known_backends.go b/api/mesh/v1alpha1/known_backends.go
@@ -4,7 +4,8 @@ const (
 	LoggingTcpType  = "tcp"
 	LoggingFileType = "file"
 
-	TracingZipkinType = "zipkin"
+	TracingZipkinType  = "zipkin"
+	TracingDatadogType = "datadog"
 
 	MetricsPrometheusType = "prometheus"
 )
diff --git a/api/mesh/v1alpha1/mesh.pb.go b/api/mesh/v1alpha1/mesh.pb.go
diff --git a/api/mesh/v1alpha1/mesh.proto b/api/mesh/v1alpha1/mesh.proto
@@ -123,6 +123,14 @@ message TracingBackend {
   google.protobuf.Struct conf = 4;
 }
 
+message DatadogTracingBackendConfig {
+  // Address of datadog collector.
+  string address = 1;
+
+  // Port of datadog collector
+  uint32 port = 2;
+}
+
 message ZipkinTracingBackendConfig {
   // Address of Zipkin collector.
   string url = 1;

diff --git a/pkg/core/resources/apis/mesh/mesh_validator.go b/pkg/core/resources/apis/mesh/mesh_validator.go
@@ -139,15 +139,36 @@ func validateTracingBackend(backend *mesh_proto.TracingBackend) validators.Valid
 	if backend.Name == "" {
 		verr.AddViolation("name", "cannot be empty")
 	}
-	if backend.GetType() != mesh_proto.TracingZipkinType {
-		verr.AddViolation("type", fmt.Sprintf("unknown backend type. Available backends: %q", mesh_proto.TracingZipkinType))
-	}
 	if backend.Sampling.GetValue() < 0.0 || backend.Sampling.GetValue() > 100.0 {
 		verr.AddViolation("sampling", "has to be in [0.0 - 100.0] range")
 	}
-	if backend.GetType() == mesh_proto.TracingZipkinType {
+	switch backend.GetType() {
+	case mesh_proto.TracingZipkinType:
 		verr.AddError("config", validateZipkin(backend.Conf))
+	case mesh_proto.TracingDatadogType:
+		verr.AddError("config", validateDatadog(backend.Conf))
+	default:
+		verr.AddViolation("type", fmt.Sprintf("unknown backend type. Available backends: %q, %q", mesh_proto.TracingZipkinType, mesh_proto.TracingDatadogType))
+	}
+	return verr
+}
+
+func validateDatadog(cfgStr *structpb.Struct) validators.ValidationError {
+	var verr validators.ValidationError
+	cfg := mesh_proto.DatadogTracingBackendConfig{}
+	if err := proto.ToTyped(cfgStr, &cfg); err != nil {
+		verr.AddViolation("", fmt.Sprintf("could not parse config: %s", err.Error()))
+		return verr
 	}
+
+	if cfg.Address == "" {
+		verr.AddViolation("address", "cannot be empty")
+	}
+
+	if cfg.Port > 0xFFFF || cfg.Port < 1 {
+		verr.AddViolation("port", "must be in the range 1 to 65535")
+	}
+
 	return verr
 }
 

diff --git a/pkg/core/resources/apis/mesh/mesh_validator_test.go b/pkg/core/resources/apis/mesh/mesh_validator_test.go
@@ -404,7 +404,7 @@ var _ = Describe("Mesh", func() {
                 - field: logging.backends[0].type
                   message: 'unknown backend type. Available backends: "tcp", "file"'
                 - field: tracing.backends[0].type
-                  message: 'unknown backend type. Available backends: "zipkin"'
+                  message: 'unknown backend type. Available backends: "zipkin", "datadog"'
                 - field: metrics.backends[0].type
                   message: 'unknown backend type. Available backends: "prometheus"'`,
 			}),

diff --git a/pkg/xds/envoy/listeners/configurers.go b/pkg/xds/envoy/listeners/configurers.go
@@ -25,10 +25,11 @@ func Kafka(statsName string) FilterChainBuilderOpt {
 	})
 }
 
-func Tracing(backend *mesh_proto.TracingBackend) FilterChainBuilderOpt {
+func Tracing(backend *mesh_proto.TracingBackend, service string) FilterChainBuilderOpt {
 	return FilterChainBuilderOptFunc(func(config *FilterChainBuilderConfig) {
 		config.AddV3(&v3.TracingConfigurer{
 			Backend: backend,
+			Service: service,
 		})
 	})
 }

diff --git a/pkg/xds/envoy/listeners/v3/tracing_configurer.go b/pkg/xds/envoy/listeners/v3/tracing_configurer.go
@@ -18,6 +18,10 @@ import (
 
 type TracingConfigurer struct {
 	Backend *mesh_proto.TracingBackend
+
+	// Opaque string which envoy will assign to tracer collector cluster, on those
+	// which support association of named "service" tags on traces. Consumed by datadog.
+	Service string
 }
 
 var _ FilterChainConfigurer = &TracingConfigurer{}
@@ -41,11 +45,40 @@ func (c *TracingConfigurer) Configure(filterChain *envoy_listener.FilterChain) e
 				return err
 			}
 			hcm.Tracing.Provider = tracing
+		case mesh_proto.TracingDatadogType:
+			tracing, err := datadogConfig(c.Backend.Conf, c.Backend.Name, c.Service)
+			if err != nil {
+				return err
+			}
+			hcm.Tracing.Provider = tracing
 		}
 		return nil
 	})
 }
 
+func datadogConfig(cfgStr *structpb.Struct, backendName string, serviceName string) (*envoy_trace.Tracing_Http, error) {
+	cfg := mesh_proto.DatadogTracingBackendConfig{}
+	if err := proto.ToTyped(cfgStr, &cfg); err != nil {
+		return nil, errors.Wrap(err, "could not convert backend")
+	}
+
+	datadogConfig := envoy_trace.DatadogConfig{
+		CollectorCluster: names.GetTracingClusterName(backendName),
+		ServiceName:      serviceName,
+	}
+	datadogConfigAny, err := proto.MarshalAnyDeterministic(&datadogConfig)
+	if err != nil {
+		return nil, err
+	}
+	tracingConfig := &envoy_trace.Tracing_Http{
+		Name: "envoy.datadog",
+		ConfigType: &envoy_trace.Tracing_Http_TypedConfig{
+			TypedConfig: datadogConfigAny,
+		},
+	}
+	return tracingConfig, nil
+}
+
 func zipkinConfig(cfgStr *structpb.Struct, backendName string) (*envoy_trace.Tracing_Http, error) {
 	cfg := mesh_proto.ZipkinTracingBackendConfig{}
 	if err := proto.ToTyped(cfgStr, &cfg); err != nil {

diff --git a/pkg/xds/envoy/listeners/v3/tracing_configurer_test.go b/pkg/xds/envoy/listeners/v3/tracing_configurer_test.go
@@ -29,7 +29,7 @@ var _ = Describe("TracingConfigurer", func() {
 				Configure(InboundListener("inbound:192.168.0.1:8080", "192.168.0.1", 8080, xds.SocketAddressProtocolTCP)).
 				Configure(FilterChain(NewFilterChainBuilder(envoy.APIV3).
 					Configure(HttpConnectionManager("localhost:8080", false)).
-					Configure(Tracing(given.backend)))).
+					Configure(Tracing(given.backend, "service")))).
 				Build()
 			// then
 			Expect(err).ToNot(HaveOccurred())
@@ -40,7 +40,7 @@ var _ = Describe("TracingConfigurer", func() {
 			// and
 			Expect(actual).To(MatchYAML(given.expected))
 		},
-		Entry("backend specified with sampling", testCase{
+		Entry("zipkin backend specified with sampling", testCase{
 			backend: &mesh_proto.TracingBackend{
 				Name:     "zipkin",
 				Sampling: &wrapperspb.DoubleValue{Value: 30.5},
@@ -76,7 +76,7 @@ var _ = Describe("TracingConfigurer", func() {
             name: inbound:192.168.0.1:8080
             trafficDirection: INBOUND`,
 		}),
-		Entry("backend specified without sampling", testCase{
+		Entry("zipkin backend specified without sampling", testCase{
 			backend: &mesh_proto.TracingBackend{
 				Name: "zipkin",
 				Type: mesh_proto.TracingZipkinType,
@@ -109,6 +109,39 @@ var _ = Describe("TracingConfigurer", func() {
             name: inbound:192.168.0.1:8080
             trafficDirection: INBOUND`,
 		}),
+		Entry("datadog backend specified", testCase{
+			backend: &mesh_proto.TracingBackend{
+				Name: "datadog",
+				Type: mesh_proto.TracingDatadogType,
+				Conf: util_proto.MustToStruct(&mesh_proto.DatadogTracingBackendConfig{
+					Address: "1.1.1.1",
+					Port:    1111,
+				}),
+			},
+			expected: `
+        address:
+          socketAddress:
+            address: 192.168.0.1
+            portValue: 8080
+        filterChains:
+        - filters:
+          - name: envoy.filters.network.http_connection_manager
+            typedConfig:
+              '@type': type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
+              httpFilters:
+              - name: envoy.filters.http.router
+              statPrefix: localhost_8080
+              tracing:
+                provider:
+                  name: envoy.datadog
+                  typedConfig:
+                    '@type': type.googleapis.com/envoy.config.trace.v3.DatadogConfig
+                    collectorCluster: tracing:datadog
+                    serviceName: service
+        name: inbound:192.168.0.1:8080
+        trafficDirection: INBOUND`,
+		}),
+
 		Entry("no backend specified", testCase{
 			backend: nil,
 			expected: `

diff --git a/pkg/xds/generator/inbound_proxy_generator.go b/pkg/xds/generator/inbound_proxy_generator.go
@@ -76,15 +76,15 @@ func (g InboundProxyGenerator) Generate(ctx xds_context.Context, proxy *model.Pr
 					Configure(envoy_listeners.HttpConnectionManager(localClusterName, true)).
 					Configure(envoy_listeners.FaultInjection(proxy.Policies.FaultInjections[endpoint])).
 					Configure(envoy_listeners.RateLimit(proxy.Policies.RateLimits[endpoint])).
-					Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend)).
+					Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend, service)).
 					Configure(envoy_listeners.HttpInboundRoutes(service, routes))
 			case mesh_core.ProtocolGRPC:
 				filterChainBuilder.
 					Configure(envoy_listeners.HttpConnectionManager(localClusterName, true)).
 					Configure(envoy_listeners.GrpcStats()).
 					Configure(envoy_listeners.FaultInjection(proxy.Policies.FaultInjections[endpoint])).
 					Configure(envoy_listeners.RateLimit(proxy.Policies.RateLimits[endpoint])).
-					Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend)).
+					Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend, service)).
 					Configure(envoy_listeners.HttpInboundRoutes(service, routes))
 			case mesh_core.ProtocolKafka:
 				filterChainBuilder.

diff --git a/pkg/xds/generator/outbound_proxy_generator.go b/pkg/xds/generator/outbound_proxy_generator.go
@@ -105,15 +105,15 @@ func (_ OutboundProxyGenerator) generateLDS(proxy *model.Proxy, routes envoy_com
 		case mesh_core.ProtocolGRPC:
 			filterChainBuilder.
 				Configure(envoy_listeners.HttpConnectionManager(serviceName, false)).
-				Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend)).
+				Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend, sourceService)).
 				Configure(envoy_listeners.HttpAccessLog(meshName, envoy_common.TrafficDirectionOutbound, sourceService, serviceName, proxy.Policies.Logs[serviceName], proxy)).
 				Configure(envoy_listeners.HttpOutboundRoute(serviceName, routes, proxy.Dataplane.Spec.TagSet())).
 				Configure(envoy_listeners.Retry(retryPolicy, protocol)).
 				Configure(envoy_listeners.GrpcStats())
 		case mesh_core.ProtocolHTTP, mesh_core.ProtocolHTTP2:
 			filterChainBuilder.
 				Configure(envoy_listeners.HttpConnectionManager(serviceName, false)).
-				Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend)).
+				Configure(envoy_listeners.Tracing(proxy.Policies.TracingBackend, sourceService)).
 				Configure(envoy_listeners.HttpAccessLog(
 					meshName,
 					envoy_common.TrafficDirectionOutbound,

diff --git a/pkg/xds/generator/tracing_proxy_generator.go b/pkg/xds/generator/tracing_proxy_generator.go
@@ -35,7 +35,14 @@ func (t TracingProxyGenerator) Generate(_ xds_context.Context, proxy *core_xds.P
 			return nil, errors.Wrap(err, "could not generate zipkin cluster")
 		}
 		resources.Add(res)
+	case mesh_proto.TracingDatadogType:
+		res, err := t.datadogCluster(proxy.Policies.TracingBackend, proxy.APIVersion)
+		if err != nil {
+			return nil, errors.Wrap(err, "could not generate datadog cluster")
+		}
+		resources.Add(res)
 	}
+
 	return resources, nil
 }
 
@@ -67,3 +74,28 @@ func (t TracingProxyGenerator) zipkinCluster(backend *mesh_proto.TracingBackend,
 		Resource: cluster,
 	}, nil
 }
+
+func (t TracingProxyGenerator) datadogCluster(backend *mesh_proto.TracingBackend, apiVersion envoy.APIVersion) (*core_xds.Resource, error) {
+	cfg := mesh_proto.DatadogTracingBackendConfig{}
+	if err := proto.ToTyped(backend.Conf, &cfg); err != nil {
+		return nil, errors.Wrap(err, "could not convert backend")
+	}
+
+	if cfg.Port > 0xFFFF || cfg.Port < 1 {
+		return nil, errors.Errorf("invalid Datadog port number %d. Must be in range 1-65535", cfg.Port)
+	}
+
+	clusterName := names.GetTracingClusterName(backend.Name)
+	cluster, err := clusters.NewClusterBuilder(apiVersion).
+		Configure(clusters.DNSCluster(clusterName, cfg.Address, cfg.Port)).
+		Build()
+	if err != nil {
+		return nil, err
+	}
+
+	return &core_xds.Resource{
+		Name:     clusterName,
+		Origin:   OriginTracing,
+		Resource: cluster,
+	}, nil
+}