Merge dev.servicegraphs into main

CHANGELOG.md

@@ -50,6 +50,8 @@ for specific instructions.
 - [FEATURE] Add `operator-detach` command to agentctl to allow zero-downtime
   upgrades when removing an Operator CRD. (@rfratto)
 
+- [FEATURE] Service graphs processor (@mapno)
+
 - [ENHANCEMENT] The Grafana Agent Operator will now default to deploying
   the matching release version of the Grafana Agent instead of v0.14.0.
   (@rfratto)

docs/configuration/traces-config.md

@@ -195,7 +195,7 @@ spanmetrics:
 # a complete trace. This is achieved by waiting a given time for all the spans
 # before evaluating the trace.
 #
-# Tail sampling also supports multiple agent deployments, allowing to group all
+# Tail sampling also supports multi agent deployments, allowing to group all
 # spans of a trace in the same agent by load balancing the spans by trace ID
 # between the instances.
 # * To make use of this feature, check load_balancing below *
@@ -255,6 +255,33 @@ load_balancing:
       [ username: <string> ]
       [ password: <secret> ]
       [ password_file: <string> ]
+
+# service_graphs configures processing of traces for building service graphs in
+# the form of prometheus metrics. The generated metrics represent edges between
+# nodes in the graph. Nodes are represented by `client` and `server` labels.
+#
+#  e.g. tempo_service_graph_request_total{client="app", server="db"} 20
+#
+# Service graphs works by inspecting spans and looking for the tag `span.kind`.
+# If it finds the span kind to be client or server, it stores the request in a
+# local in-memory store.
+#
+# That request waits until its corresponding client or server pair span is
+# processed or until the maximum waiting time has passed.
+# When either of those conditions is reached, the request is processed and
+# removed from the local store. If the request is complete by that time, it'll
+# be recorded as an edge in the graph.
+#
+# Service graphs supports multi agent deployments, allowing to group all spans
+# of a trace in the same agent by load balancing the spans by trace ID between
+# the instances.
+# * To make use of this feature, check load_balancing above *
+service_graphs:
+  [ enabled: <bool> | default = false ]
+
+  [ wait: <duration> | default = "10s"]
+
+  [ max_items: <integer> | default = 10_000 ]
 ```
 
 > **Note:** More information on the following types can be found on the

example/docker-compose/agent/config/agent.yaml

@@ -23,7 +23,7 @@ metrics:
 
         - job_name: local_scrape
           static_configs:
-            - targets: ['127.0.0.1:12345']
+            - targets: ['127.0.0.1:12345', '0.0.0.0:8889']
               labels:
                 cluster: 'docker_compose'
                 container: 'agent'
@@ -97,4 +97,6 @@ traces:
       processes: true
       roots: true
     spanmetrics:
-      metrics_instance: test
+      handler_endpoint: 0.0.0.0:8889
+    service_graphs:
+      enabled: true

go.mod

@@ -24,10 +24,11 @@ require (
 	github.com/google/go-jsonnet v0.17.0
 	github.com/gorilla/mux v1.8.0
 	github.com/grafana/loki v1.6.2-0.20210429132126-d88f3996eaa2
+	github.com/grafana/tempo v1.0.1
 	github.com/hashicorp/consul/api v1.10.1
 	github.com/hashicorp/go-cleanhttp v0.5.2
 	github.com/hashicorp/go-getter v1.5.3
-	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-multierror v1.1.1
 	github.com/infinityworks/github-exporter v0.0.0-20201016091012-831b72461034
 	github.com/jsternberg/zap-logfmt v1.2.0
 	github.com/miekg/dns v1.1.42
@@ -40,6 +41,7 @@ require (
 	github.com/onsi/gomega v1.11.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/loadbalancingexporter v0.36.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter v0.36.0
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal v0.36.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor v0.36.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/spanmetricsprocessor v0.36.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/processor/tailsamplingprocessor v0.36.0
@@ -50,6 +52,7 @@ require (
 	github.com/opentracing-contrib/go-grpc v0.0.0-20210225150812-73cb765af46e
 	github.com/opentracing-contrib/go-stdlib v1.0.0
 	github.com/opentracing/opentracing-go v1.2.0
+	github.com/patrickmn/go-cache v0.0.0-20180527043350-9f6ff22cfff8
 	github.com/percona/mongodb_exporter v0.0.0-00010101000000-000000000000
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus-community/elasticsearch_exporter v1.2.1

pkg/traces/config.go

@@ -15,6 +15,7 @@ import (
 	"github.com/grafana/agent/pkg/traces/noopreceiver"
 	"github.com/grafana/agent/pkg/traces/promsdprocessor"
 	"github.com/grafana/agent/pkg/traces/remotewriteexporter"
+	"github.com/grafana/agent/pkg/traces/servicegraphprocessor"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/loadbalancingexporter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusexporter"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/attributesprocessor"
@@ -122,10 +123,13 @@ type InstanceConfig struct {
 	AutomaticLogging *automaticloggingprocessor.AutomaticLoggingConfig `yaml:"automatic_logging,omitempty"`
 
 	// TailSampling defines a sampling strategy for the pipeline
-	TailSampling *tailSamplingConfig `yaml:"tail_sampling"`
+	TailSampling *tailSamplingConfig `yaml:"tail_sampling,omitempty"`
 
 	// LoadBalancing is used to distribute spans of the same trace to the same agent instance
 	LoadBalancing *loadBalancingConfig `yaml:"load_balancing"`
+
+	// ServiceGraphs
+	ServiceGraphs *serviceGraphsConfig `yaml:"service_graphs,omitempty"`
 }
 
 const (
@@ -243,6 +247,12 @@ type exporterConfig struct {
 	BasicAuth          *prom_config.BasicAuth `yaml:"basic_auth,omitempty"`
 }
 
+type serviceGraphsConfig struct {
+	Enabled  bool          `yaml:"enabled,omitempty"`
+	Wait     time.Duration `yaml:"wait,omitempty"`
+	MaxItems int           `yaml:"max_items,omitempty"`
+}
+
 // exporter builds an OTel exporter from RemoteWriteConfig
 func exporter(rwCfg RemoteWriteConfig) (map[string]interface{}, error) {
 	if len(rwCfg.Endpoint) == 0 {
@@ -564,6 +574,14 @@ func (c *InstanceConfig) otelConfig() (*config.Config, error) {
 		}
 	}
 
+	if c.ServiceGraphs != nil && c.ServiceGraphs.Enabled {
+		processors[servicegraphprocessor.TypeStr] = map[string]interface{}{
+			"wait":      c.ServiceGraphs.Wait,
+			"max_items": c.ServiceGraphs.MaxItems,
+		}
+		processorNames = append(processorNames, servicegraphprocessor.TypeStr)
+	}
+
 	// Build Pipelines
 	splitPipeline := c.LoadBalancing != nil
 	orderedSplitProcessors := orderProcessors(processorNames, splitPipeline)
@@ -660,6 +678,7 @@ func tracingFactories() (component.Factories, error) {
 		spanmetricsprocessor.NewFactory(),
 		automaticloggingprocessor.NewFactory(),
 		tailsamplingprocessor.NewFactory(),
+		servicegraphprocessor.NewFactory(),
 	)
 	if err != nil {
 		return component.Factories{}, err
@@ -680,9 +699,10 @@ func orderProcessors(processors []string, splitPipelines bool) [][]string {
 	order := map[string]int{
 		"attributes":        0,
 		"spanmetrics":       1,
-		"tail_sampling":     2,
-		"automatic_logging": 3,
-		"batch":             4,
+		"service_graphs":    2,
+		"tail_sampling":     3,
+		"automatic_logging": 4,
+		"batch":             5,
 	}
 
 	sort.Slice(processors, func(i, j int) bool {

pkg/traces/config_test.go

@@ -869,6 +869,39 @@ service:
       exporters: ["otlphttp/0", "otlp/1"]
       processors: []
       receivers: ["jaeger"]
+`,
+		},
+		{
+			name: "service graphs",
+			cfg: `
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+remote_write:
+  - endpoint: example.com:12345
+service_graphs:
+  enabled: true
+`,
+			expectedConfig: `
+receivers:
+  jaeger:
+    protocols:
+      grpc:
+exporters:
+  otlp/0:
+    endpoint: example.com:12345
+    compression: gzip
+    retry_on_failure:
+      max_elapsed_time: 60s
+processors:
+  service_graphs:
+service:
+  pipelines:
+    traces:
+      exporters: ["otlp/0"]
+      processors: ["service_graphs"]
+      receivers: ["jaeger"]
 `,
 		},
 		{
@@ -1006,11 +1039,14 @@ tail_sampling:
         values:
           - value1
           - value2
+service_graphs:
+  enabled: true
 `,
 			expectedProcessors: map[string][]config.ComponentID{
 				"traces": {
 					config.NewID("attributes"),
 					config.NewID("spanmetrics"),
+					config.NewID("service_graphs"),
 					config.NewID("tail_sampling"),
 					config.NewID("automatic_logging"),
 					config.NewID("batch"),
@@ -1062,11 +1098,14 @@ load_balancing:
     dns:
       hostname: agent
       port: 4318
+service_graphs:
+  enabled: true
 `,
 			expectedProcessors: map[string][]config.ComponentID{
 				"traces/0": {
 					config.NewID("attributes"),
 					config.NewID("spanmetrics"),
+					config.NewID("service_graphs"),
 				},
 				"traces/1": {
 					config.NewID("tail_sampling"),

pkg/traces/contextkeys/keys.go

@@ -8,4 +8,7 @@ const (
 
 	// Metrics is used to pass instance.Manager through the context
 	Metrics
+
+	// PrometheusRegisterer is used to pass prometheus.Registerer through the context
+	PrometheusRegisterer
 )

pkg/traces/instance.go

@@ -9,6 +9,7 @@ import (
 	"github.com/grafana/agent/pkg/build"
 	"github.com/grafana/agent/pkg/logs"
 	"github.com/grafana/agent/pkg/metrics/instance"
+	"github.com/grafana/agent/pkg/traces/automaticloggingprocessor"
 	"github.com/grafana/agent/pkg/traces/contextkeys"
 	"github.com/grafana/agent/pkg/util"
 	"github.com/prometheus/client_golang/prometheus"
@@ -44,14 +45,14 @@ func NewInstance(logsSubsystem *logs.Logs, reg prometheus.Registerer, cfg Instan
 		return nil, fmt.Errorf("failed to create metric views: %w", err)
 	}
 
-	if err := instance.ApplyConfig(logsSubsystem, promInstanceManager, cfg); err != nil {
+	if err := instance.ApplyConfig(logsSubsystem, promInstanceManager, reg, cfg); err != nil {
 		return nil, err
 	}
 	return instance, nil
 }
 
 // ApplyConfig updates the configuration of the Instance.
-func (i *Instance) ApplyConfig(logsSubsystem *logs.Logs, promInstanceManager instance.Manager, cfg InstanceConfig) error {
+func (i *Instance) ApplyConfig(logsSubsystem *logs.Logs, promInstanceManager instance.Manager, reg prometheus.Registerer, cfg InstanceConfig) error {
 	i.mut.Lock()
 	defer i.mut.Unlock()
 
@@ -64,8 +65,7 @@ func (i *Instance) ApplyConfig(logsSubsystem *logs.Logs, promInstanceManager ins
 	// Shut down any existing pipeline
 	i.stop()
 
-	createCtx := context.WithValue(context.Background(), contextkeys.Logs, logsSubsystem)
-	err := i.buildAndStartPipeline(createCtx, cfg, promInstanceManager)
+	err := i.buildAndStartPipeline(context.Background(), cfg, logsSubsystem, promInstanceManager, reg)
 	if err != nil {
 		return fmt.Errorf("failed to create pipeline: %w", err)
 	}
@@ -131,7 +131,7 @@ func (i *Instance) stop() {
 	i.exporter = nil
 }
 
-func (i *Instance) buildAndStartPipeline(ctx context.Context, cfg InstanceConfig, promManager instance.Manager) error {
+func (i *Instance) buildAndStartPipeline(ctx context.Context, cfg InstanceConfig, logs *logs.Logs, instManager instance.Manager, reg prometheus.Registerer) error {
 	// create component factories
 	otelConfig, err := cfg.otelConfig()
 	if err != nil {
@@ -150,12 +150,20 @@ func (i *Instance) buildAndStartPipeline(ctx context.Context, cfg InstanceConfig
 	}
 
 	if cfg.SpanMetrics != nil && len(cfg.SpanMetrics.MetricsInstance) != 0 {
-		ctx = context.WithValue(ctx, contextkeys.Metrics, promManager)
+		ctx = context.WithValue(ctx, contextkeys.Metrics, instManager)
 	}
 
-	if cfg.TailSampling != nil && cfg.LoadBalancing == nil {
-		i.logger.Warn("Configuring tail_sampling without load_balance." +
-			"Load balancing is required for tail sampling to properly work in multi instance deployments")
+	if cfg.LoadBalancing == nil && (cfg.TailSampling != nil || cfg.ServiceGraphs != nil) {
+		i.logger.Warn("Configuring tail_sampling and/or service_graphs without load_balance." +
+			"Load balancing is required for those features to properly work in multi agent deployments")
+	}
+
+	if cfg.AutomaticLogging != nil && cfg.AutomaticLogging.Backend != automaticloggingprocessor.BackendStdout {
+		ctx = context.WithValue(ctx, contextkeys.Logs, logs)
+	}
+
+	if cfg.ServiceGraphs != nil {
+		ctx = context.WithValue(ctx, contextkeys.PrometheusRegisterer, reg)
 	}
 
 	factories, err := tracingFactories()

pkg/traces/remotewriteexporter/exporter.go

@@ -70,7 +70,7 @@ func newRemoteWriteExporter(cfg *Config) (component.MetricsExporter, error) {
 func (e *remoteWriteExporter) Start(ctx context.Context, _ component.Host) error {
 	manager, ok := ctx.Value(contextkeys.Metrics).(instance.Manager)
 	if !ok || manager == nil {
-		return fmt.Errorf("key does not contain a Prometheus instance")
+		return fmt.Errorf("key does not contain a InstanceManager instance")
 	}
 	e.manager = manager
 	return nil

pkg/traces/remotewriteexporter/factory.go

@@ -20,7 +20,7 @@ type label struct {
 
 var _ config.Exporter = (*Config)(nil)
 
-// Config holds the configuration for the Prometheus SD processor.
+// Config holds the configuration for the Prometheus remote write processor.
 type Config struct {
 	config.ExporterSettings `mapstructure:",squash"`
 
@@ -29,7 +29,7 @@ type Config struct {
 	PromInstance string  `mapstructure:"metrics_instance"`
 }
 
-// NewFactory returns a new factory for the Attributes processor.
+// NewFactory returns a new factory for the Prometheus remote write processor.
 func NewFactory() component.ExporterFactory {
 	return exporterhelper.NewFactory(
 		TypeStr,