Merge branch 'main' into balance-metrics-by-resources

.chloggen/grafanacloudconnector.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: grafanacloudconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Adds a connector to generate metrics for Grafana Cloud.
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [31647]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

.github/CODEOWNERS

@@ -29,6 +29,7 @@ connector/countconnector/                                @open-telemetry/collect
 connector/datadogconnector/                              @open-telemetry/collector-contrib-approvers @mx-psi @dineshg13
 connector/exceptionsconnector/                           @open-telemetry/collector-contrib-approvers @jpkrohling @marctc
 connector/failoverconnector/                             @open-telemetry/collector-contrib-approvers @akats7 @djaglowski @fatsheep9146
+connector/grafanacloudconnector/                         @open-telemetry/collector-contrib-approvers @jpkrohling @rlankfo @jcreixell
 connector/routingconnector/                              @open-telemetry/collector-contrib-approvers @jpkrohling @mwear
 connector/servicegraphconnector/                         @open-telemetry/collector-contrib-approvers @jpkrohling @mapno
 connector/spanmetricsconnector/                          @open-telemetry/collector-contrib-approvers @portertech

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -30,6 +30,7 @@ body:
       - connector/datadog
       - connector/exceptions
       - connector/failover
+      - connector/grafanacloud
       - connector/routing
       - connector/servicegraph
       - connector/spanmetrics

.github/ISSUE_TEMPLATE/feature_request.yaml

@@ -24,6 +24,7 @@ body:
       - connector/datadog
       - connector/exceptions
       - connector/failover
+      - connector/grafanacloud
       - connector/routing
       - connector/servicegraph
       - connector/spanmetrics

.github/ISSUE_TEMPLATE/other.yaml

@@ -24,6 +24,7 @@ body:
       - connector/datadog
       - connector/exceptions
       - connector/failover
+      - connector/grafanacloud
       - connector/routing
       - connector/servicegraph
       - connector/spanmetrics

connector/grafanacloudconnector/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

connector/grafanacloudconnector/README.md

@@ -0,0 +1,39 @@
+# Grafana Cloud Connector
+
+<!-- status autogenerated section -->
+| Status        |           |
+| ------------- |-----------|
+| Distributions | [contrib], [grafana] |
+| Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aconnector%2Fgrafanacloud%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aconnector%2Fgrafanacloud) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aconnector%2Fgrafanacloud%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aconnector%2Fgrafanacloud) |
+| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@jpkrohling](https://www.github.com/jpkrohling), [@rlankfo](https://www.github.com/rlankfo), [@jcreixell](https://www.github.com/jcreixell) \| Seeking more code owners! |
+
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
+[contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
+[grafana]: https://github.com/grafana/agent
+
+## Supported Pipeline Types
+
+| [Exporter Pipeline Type] | [Receiver Pipeline Type] | [Stability Level] |
+| ------------------------ | ------------------------ | ----------------- |
+| traces | metrics | [development] |
+
+[Exporter Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#exporter-pipeline-type
+[Receiver Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#receiver-pipeline-type
+[Stability Level]: https://github.com/open-telemetry/opentelemetry-collector#stability-levels
+<!-- end autogenerated section -->
+
+## Overview
+
+The Grafana Cloud Connector (grafanacloudconnector) is a connector component that can analyze telemetry in pipelines to generate usage metrics for the following Grafana Cloud products:
+* Application Observability
+
+#### Example configuration for the component
+
+```yaml
+connectors:
+  grafanacloud:
+    host_identifiers: ["host.id"]
+    metrics_flush_interval: 60s
+```
+
+This connector will generate a host info metric based on the first "host_identifiers" resource attribute found on spans. The rest are skipped. Valid flush intervals are between 15s and 5m.

connector/grafanacloudconnector/config.go

@@ -0,0 +1,34 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package grafanacloudconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/grafanacloudconnector"
+
+import (
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+)
+
+// Config defines the configuration options for the Grafana Cloud connector.
+type Config struct {
+	// HostIdentifiers defines the list of resource attributes used to derive
+	// a unique `grafana.host.id` value. In most cases, this should be [ "host.id" ]
+	HostIdentifiers      []string      `mapstructure:"host_identifiers"`
+	MetricsFlushInterval time.Duration `mapstructure:"metrics_flush_interval"`
+}
+
+var _ component.ConfigValidator = (*Config)(nil)
+
+// Validate checks if the configuration is valid
+func (c Config) Validate() error {
+	if len(c.HostIdentifiers) == 0 {
+		return fmt.Errorf("at least one host identifier is required")
+	}
+
+	if c.MetricsFlushInterval > 5*time.Minute || c.MetricsFlushInterval < 15*time.Second {
+		return fmt.Errorf("%q is not a valid flush interval between 15s and 5m", c.MetricsFlushInterval)
+	}
+
+	return nil
+}

connector/grafanacloudconnector/config_test.go

@@ -0,0 +1,88 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package grafanacloudconnector
+
+import (
+	"path/filepath"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap/confmaptest"
+	"gotest.tools/assert"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/grafanacloudconnector/internal/metadata"
+)
+
+func TestLoadConfig(t *testing.T) {
+	testCases := []struct {
+		name   string
+		expect *Config
+	}{
+		{
+			name:   "",
+			expect: createDefaultConfig().(*Config),
+		},
+		{
+			name: "custom",
+			expect: &Config{
+				HostIdentifiers:      []string{"k8s.node.name", "host.name", "host.id"},
+				MetricsFlushInterval: 30 * time.Second,
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			cm, err := confmaptest.LoadConf(filepath.Join("testdata", "config.yaml"))
+			assert.NilError(t, err)
+
+			factory := NewFactory()
+			cfg := factory.CreateDefaultConfig()
+
+			sub, err := cm.Sub(component.NewIDWithName(metadata.Type, tc.name).String())
+			assert.NilError(t, err)
+			assert.NilError(t, component.UnmarshalConfig(sub, cfg))
+			assert.DeepEqual(t, tc.expect, cfg)
+		})
+	}
+}
+
+func TestValidate(t *testing.T) {
+	testCases := []struct {
+		name   string
+		cfg    *Config
+		errMsg string
+	}{
+		{
+			name:   "valid",
+			cfg:    createDefaultConfig().(*Config),
+			errMsg: "",
+		},
+		{
+			name:   "missing identifiers",
+			cfg:    &Config{},
+			errMsg: "at least one host identifier is required",
+		},
+		{
+			name: "invalid flush interval",
+			cfg: &Config{
+				HostIdentifiers:      []string{"host.id"},
+				MetricsFlushInterval: time.Hour,
+			},
+			errMsg: "\"1h0m0s\" is not a valid flush interval between 15s and 5m",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.cfg.Validate()
+			if tc.errMsg != "" {
+				assert.Equal(t, tc.errMsg, err.Error())
+			} else {
+				assert.NilError(t, err)
+			}
+		})
+	}
+}

connector/grafanacloudconnector/connector.go

@@ -0,0 +1,163 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package grafanacloudconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/grafanacloudconnector"
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/connector"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/otel/metric"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/grafanacloudconnector/internal/metadata"
+)
+
+const (
+	hostInfoMetric     = "traces_host_info"
+	hostIdentifierAttr = "grafana.host.id"
+)
+
+var _ connector.Traces = (*connectorImp)(nil)
+
+type connectorImp struct {
+	config Config
+	logger *zap.Logger
+
+	started      bool
+	done         chan struct{}
+	shutdownOnce sync.Once
+
+	metricsConsumer consumer.Metrics
+	hostMetrics     *hostMetrics
+
+	metricHostCount      metric.Int64ObservableGauge
+	metricFlushCount     metric.Int64Counter
+	metricDatapointCount metric.Int64Counter
+}
+
+func newConnector(logger *zap.Logger, set component.TelemetrySettings, config component.Config) (*connectorImp, error) {
+	hm := newHostMetrics()
+	mHostCount, err := metadata.Meter(set).Int64ObservableGauge(
+		"grafanacloud_host_count",
+		metric.WithDescription("Number of unique hosts"),
+		metric.WithUnit("1"),
+		metric.WithInt64Callback(func(ctx context.Context, result metric.Int64Observer) error {
+			result.Observe(int64(hm.count()))
+			return nil
+		}),
+	)
+
+	if err != nil {
+		return nil, err
+	}
+
+	mFlushCount, err := metadata.Meter(set).Int64Counter(
+		"grafanacloud_flush_count",
+		metric.WithDescription("Number of metrics flushes"),
+		metric.WithUnit("1"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	mDatapointCount, err := metadata.Meter(set).Int64Counter(
+		"grafanacloud_datapoint_count",
+		metric.WithDescription("Number of datapoints sent to Grafana Cloud"),
+		metric.WithUnit("1"),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	cfg := config.(*Config)
+	return &connectorImp{
+		config:               *cfg,
+		logger:               logger,
+		done:                 make(chan struct{}),
+		hostMetrics:          hm,
+		metricHostCount:      mHostCount,
+		metricFlushCount:     mFlushCount,
+		metricDatapointCount: mDatapointCount,
+	}, nil
+}
+
+// Capabilities implements connector.Traces.
+func (c *connectorImp) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+// ConsumeTraces implements connector.Traces.
+func (c *connectorImp) ConsumeTraces(_ context.Context, td ptrace.Traces) error {
+	for i := 0; i < td.ResourceSpans().Len(); i++ {
+		resourceSpan := td.ResourceSpans().At(i)
+		attrs := resourceSpan.Resource().Attributes()
+		mapping := attrs.AsRaw()
+
+		for _, attrName := range c.config.HostIdentifiers {
+			if val, ok := mapping[attrName]; ok {
+				if v, ok := val.(string); ok {
+					c.hostMetrics.add(v)
+				}
+				break
+			}
+		}
+	}
+	return nil
+}
+
+// Start implements connector.Traces.
+func (c *connectorImp) Start(ctx context.Context, _ component.Host) error {
+	c.logger.Info("Starting Grafana Cloud connector")
+	c.started = true
+	ticker := time.NewTicker(c.config.MetricsFlushInterval)
+	go func() {
+		for {
+			select {
+			case <-c.done:
+				ticker.Stop()
+				return
+			case <-ticker.C:
+				if err := c.flush(ctx); err != nil {
+					c.logger.Error("Error consuming metrics", zap.Error(err))
+				}
+			}
+		}
+	}()
+	return nil
+}
+
+// Shutdown implements connector.Traces.
+func (c *connectorImp) Shutdown(ctx context.Context) error {
+	c.shutdownOnce.Do(func() {
+		c.logger.Info("Stopping Grafana Cloud connector")
+		if c.started {
+			// flush metrics on shutdown
+			if err := c.flush(ctx); err != nil {
+				c.logger.Error("Error consuming metrics", zap.Error(err))
+			}
+			c.done <- struct{}{}
+			c.started = false
+		}
+	})
+	return nil
+}
+
+func (c *connectorImp) flush(ctx context.Context) error {
+	var err error
+
+	metrics, count := c.hostMetrics.metrics()
+	if count > 0 {
+		c.hostMetrics.reset()
+		c.logger.Debug("Flushing metrics", zap.Int("count", count))
+		c.metricDatapointCount.Add(ctx, int64(metrics.DataPointCount()))
+		err = c.metricsConsumer.ConsumeMetrics(ctx, *metrics)
+	}
+	c.metricFlushCount.Add(ctx, int64(1))
+	return err
+}