[connector/signaltometrics]Add config validation and custom ottl funcs (

#36671)  #### Description Adds config validation logic and custom OTTL functions required for signal to metrics connector.  #### Link to tracking issue Related to: #35930  #### Testing Uni tests added.  #### Documentation N/A  --------- Co-authored-by: Christos Markou <[email protected]>
open-telemetry · Dec 16, 2024 · 19f550c · 19f550c
1 parent 9aadead
commit 19f550c
Show file tree

Hide file tree

Showing 21 changed files with 1,046 additions and 39 deletions.
diff --git a/.chloggen/signaltometrics-config-validation.yaml b/.chloggen/signaltometrics-config-validation.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: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: signaltometrics
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add config validation and custom OTTL functions
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35930]
+
+# (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: Adds config validation for the signal to metrics connector. Also introduces `AdjustedCount` OTTL function.
+
+# 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]
diff --git a/connector/signaltometricsconnector/README.md b/connector/signaltometricsconnector/README.md
@@ -103,9 +103,9 @@ histogram:
 
 - [**Optional**] `count` represents an OTTL expression to extract the count to be
   recorded in the histogram from the incoming data. If no expression is provided
-  then it defaults to the count of the signal i.e. [adjusted count](https://opentelemetry.io/docs/specs/otel/trace/tracestate-probability-sampling-experimental/#adjusted-count)
-  for spans and count for others. [OTTL converters](https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs#readme-converters)
-  can be used to transform the data.
+  then it defaults to the count of the signal. [OTTL converters](https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs#readme-converters)
+  can be used to transform the data. For spans, a special converter [adjusted count](#custom-ottl-functions),
+  is provided to help calculte the span's [adjusted count](https://opentelemetry.io/docs/specs/otel/trace/tracestate-probability-sampling-experimental/#adjusted-count).
 - [**Required**] `value` represents an OTTL expression to extract the value to be
   recorded in the histogram from the incoming data. [OTTL converters](https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs#readme-converters)
   can be used to transform the data.
@@ -124,13 +124,13 @@ exponential_histogram:
 - [**Optional**] `max_size` represents the maximum number of buckets per positive
   or negative number range. Defaults to `160`.
 - [**Optional**] `count` represents an OTTL expression to extract the count to be
-  recorded in the exponential histogram from the incoming data. If no expression
-  is provided then it defaults to the count of the signal i.e. [adjusted count](https://opentelemetry.io/docs/specs/otel/trace/tracestate-probability-sampling-experimental/#adjusted-count)
-  for spans and count for others.
-  [OTTL converters](https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs#readme-converters) can be used to transform the data.
-- [**Required**] `value` represents an OTTL expression to extract the value to be recorded
-  in the exponential histogram from the incoming data.
-  [OTTL converters](https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs#readme-converters) can be used to transform the data.
+  recorded in the expoential histogram from the incoming data. If no expression
+  is provided then it defaults to the count of the signal. [OTTL converters](https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs#readme-converters)
+  can be used to transform the data. For spans, a special converter [adjusted count](#custom-ottl-functions),
+  is provided to help calculte the span's [adjusted count](https://opentelemetry.io/docs/specs/otel/trace/tracestate-probability-sampling-experimental/#adjusted-count).
+- [**Required**] `value` represents an OTTL expression to extract the value to be
+  recorded in the exponential histogram from the incoming data. [OTTL converters](https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/ottlfuncs#readme-converters)
+  can be used to transform the data.
 
 ### Attributes
 
@@ -225,3 +225,12 @@ signaltometrics.service.name: <service_name_of_the_otel_collector>
 signaltometrics.service.namespace: <service_namespace_of_the_otel_collector>
 signaltometrics.service.instance.id: <service_instance_id_of_the_otel_collector>
 ```
+
+### Custom OTTL functions
+
+The component implements a couple of custom OTTL functions:
+
+1. `AdjustedCount`: a converter capable of calculating [adjusted count for a span](https://github.com/open-telemetry/oteps/blob/main/text/trace/0235-sampling-threshold-in-trace-state.md).
+2. `get`: a temporary solution to parse OTTL expressions with only values. This is
+only for internal usage and MUST NOT be used explicitly as it is a stopgap measure
+([see this for more details](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/35621)).
diff --git a/connector/signaltometricsconnector/config/config.go b/connector/signaltometricsconnector/config/config.go
@@ -3,7 +3,38 @@
 
 package config // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/signaltometricsconnector/config"
 
-import "fmt"
+import (
+	"errors"
+	"fmt"
+
+	"github.com/lightstep/go-expohisto/structure"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.uber.org/zap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/signaltometricsconnector/internal/customottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
+)
+
+const (
+	// defaultExponentialHistogramMaxSize is the default maximum number
+	// of buckets per positive or negative number range. 160 buckets
+	// default supports a high-resolution histogram able to cover a
+	// long-tail latency distribution from 1ms to 100s with a relative
+	// error of less than 5%.
+	// Ref: https://opentelemetry.io/docs/specs/otel/metrics/sdk/#base2-exponential-bucket-histogram-aggregation
+	defaultExponentialHistogramMaxSize = 160
+)
+
+var defaultHistogramBuckets = []float64{
+	2, 4, 6, 8, 10, 50, 100, 200, 400, 800, 1000, 1400, 2000, 5000, 10_000, 15_000,
+}
+
+var _ confmap.Unmarshaler = (*Config)(nil)
 
 // Config for the connector. Each configuration field describes the metrics
 // to produce from a specific signal.
@@ -17,9 +48,98 @@ func (c *Config) Validate() error {
 	if len(c.Spans) == 0 && len(c.Datapoints) == 0 && len(c.Logs) == 0 {
 		return fmt.Errorf("no configuration provided, at least one should be specified")
 	}
+	var multiError error // collect all errors at once
+	if len(c.Spans) > 0 {
+		parser, err := ottlspan.NewParser(
+			customottl.SpanFuncs(),
+			component.TelemetrySettings{Logger: zap.NewNop()},
+		)
+		if err != nil {
+			return fmt.Errorf("failed to create parser for OTTL spans: %w", err)
+		}
+		for _, span := range c.Spans {
+			if err := validateMetricInfo(span, parser); err != nil {
+				multiError = errors.Join(multiError, fmt.Errorf("failed to validate spans configuration: %w", err))
+			}
+		}
+	}
+	if len(c.Datapoints) > 0 {
+		parser, err := ottldatapoint.NewParser(
+			customottl.DatapointFuncs(),
+			component.TelemetrySettings{Logger: zap.NewNop()},
+		)
+		if err != nil {
+			return fmt.Errorf("failed to create parser for OTTL datapoints: %w", err)
+		}
+		for _, dp := range c.Datapoints {
+			if err := validateMetricInfo(dp, parser); err != nil {
+				multiError = errors.Join(multiError, fmt.Errorf("failed to validate datapoints configuration: %w", err))
+			}
+		}
+	}
+	if len(c.Logs) > 0 {
+		parser, err := ottllog.NewParser(
+			customottl.LogFuncs(),
+			component.TelemetrySettings{Logger: zap.NewNop()},
+		)
+		if err != nil {
+			return fmt.Errorf("failed to create parser for OTTL logs: %w", err)
+		}
+		for _, log := range c.Logs {
+			if err := validateMetricInfo(log, parser); err != nil {
+				multiError = errors.Join(multiError, fmt.Errorf("failed to validate logs configuration: %w", err))
+			}
+		}
+	}
+	return multiError
+}
+
+// Unmarshal implements the confmap.Unmarshaler interface. It allows
+// unmarshaling the config with a custom logic to allow setting
+// default values when/if required.
+func (c *Config) Unmarshal(collectorCfg *confmap.Conf) error {
+	if collectorCfg == nil {
+		return nil
+	}
+	if err := collectorCfg.Unmarshal(c, confmap.WithIgnoreUnused()); err != nil {
+		return err
+	}
+	for i, info := range c.Spans {
+		info.ensureDefaults()
+		c.Spans[i] = info
+	}
+	for i, info := range c.Datapoints {
+		info.ensureDefaults()
+		c.Datapoints[i] = info
+	}
+	for i, info := range c.Logs {
+		info.ensureDefaults()
+		c.Logs[i] = info
+	}
 	return nil
 }
 
+type Attribute struct {
+	Key          string `mapstructure:"key"`
+	DefaultValue any    `mapstructure:"default_value"`
+}
+
+type Histogram struct {
+	Buckets []float64 `mapstructure:"buckets"`
+	Count   string    `mapstructure:"count"`
+	Value   string    `mapstructure:"value"`
+}
+
+type ExponentialHistogram struct {
+	MaxSize int32  `mapstructure:"max_size"`
+	Count   string `mapstructure:"count"`
+	Value   string `mapstructure:"value"`
+}
+
+type Sum struct {
+	Value string `mapstructure:"value"`
+}
+
 // MetricInfo defines the structure of the metric produced by the connector.
 type MetricInfo struct {
 	Name        string `mapstructure:"name"`
@@ -40,23 +160,120 @@ type MetricInfo struct {
 	Sum                  *Sum                  `mapstructure:"sum"`
 }
 
-type Attribute struct {
-	Key          string `mapstructure:"key"`
-	DefaultValue any    `mapstructure:"default_value"`
+func (mi *MetricInfo) ensureDefaults() {
+	if mi.Histogram != nil {
+		// Add default buckets if explicit histogram is defined
+		if len(mi.Histogram.Buckets) == 0 {
+			mi.Histogram.Buckets = defaultHistogramBuckets
+		}
+	}
+	if mi.ExponentialHistogram != nil {
+		if mi.ExponentialHistogram.MaxSize == 0 {
+			mi.ExponentialHistogram.MaxSize = defaultExponentialHistogramMaxSize
+		}
+	}
 }
 
-type Histogram struct {
-	Buckets []float64 `mapstructure:"buckets"`
-	Count   string    `mapstructure:"count"`
-	Value   string    `mapstructure:"value"`
+func (mi *MetricInfo) validateAttributes() error {
+	tmp := pcommon.NewValueEmpty()
+	duplicate := map[string]struct{}{}
+	for _, attr := range mi.Attributes {
+		if attr.Key == "" {
+			return fmt.Errorf("attribute key missing")
+		}
+		if _, ok := duplicate[attr.Key]; ok {
+			return fmt.Errorf("duplicate key found in attributes config: %s", attr.Key)
+		}
+		if err := tmp.FromRaw(attr.DefaultValue); err != nil {
+			return fmt.Errorf("invalid default value specified for attribute %s", attr.Key)
+		}
+		duplicate[attr.Key] = struct{}{}
+	}
+	return nil
 }
 
-type ExponentialHistogram struct {
-	MaxSize int32  `mapstructure:"max_size"`
-	Count   string `mapstructure:"count"`
-	Value   string `mapstructure:"value"`
+func (mi *MetricInfo) validateHistogram() error {
+	if mi.Histogram != nil {
+		if len(mi.Histogram.Buckets) == 0 {
+			return errors.New("histogram buckets missing")
+		}
+		if mi.Histogram.Value == "" {
+			return errors.New("value OTTL statement is required")
+		}
+	}
+	if mi.ExponentialHistogram != nil {
+		if _, err := structure.NewConfig(
+			structure.WithMaxSize(mi.ExponentialHistogram.MaxSize),
+		).Validate(); err != nil {
+			return err
+		}
+		if mi.ExponentialHistogram.Value == "" {
+			return errors.New("value OTTL statement is required")
+		}
+	}
+	return nil
 }
 
-type Sum struct {
-	Value string `mapstructure:"value"`
+func (mi *MetricInfo) validateSum() error {
+	if mi.Sum != nil {
+		if mi.Sum.Value == "" {
+			return errors.New("value must be defined for sum metrics")
+		}
+	}
+	return nil
+}
+
+// validateMetricInfo is an utility method validate all supported metric
+// types defined for the metric info including any ottl expressions.
+func validateMetricInfo[K any](mi MetricInfo, parser ottl.Parser[K]) error {
+	if mi.Name == "" {
+		return errors.New("missing required metric name configuration")
+	}
+	if err := mi.validateAttributes(); err != nil {
+		return fmt.Errorf("attributes validation failed: %w", err)
+	}
+	if err := mi.validateHistogram(); err != nil {
+		return fmt.Errorf("histogram validation failed: %w", err)
+	}
+	if err := mi.validateSum(); err != nil {
+		return fmt.Errorf("sum validation failed: %w", err)
+	}
+
+	// Exactly one metric should be defined
+	var (
+		metricsDefinedCount int
+		statements          []string
+	)
+	if mi.Histogram != nil {
+		metricsDefinedCount++
+		if mi.Histogram.Count != "" {
+			statements = append(statements, customottl.ConvertToStatement(mi.Histogram.Count))
+		}
+		statements = append(statements, customottl.ConvertToStatement(mi.Histogram.Value))
+	}
+	if mi.ExponentialHistogram != nil {
+		metricsDefinedCount++
+		if mi.ExponentialHistogram.Count != "" {
+			statements = append(statements, customottl.ConvertToStatement(mi.ExponentialHistogram.Count))
+		}
+		statements = append(statements, customottl.ConvertToStatement(mi.ExponentialHistogram.Value))
+	}
+	if mi.Sum != nil {
+		metricsDefinedCount++
+		statements = append(statements, customottl.ConvertToStatement(mi.Sum.Value))
+	}
+	if metricsDefinedCount != 1 {
+		return fmt.Errorf("exactly one of the metrics must be defined, %d found", metricsDefinedCount)
+	}
+
+	// validate OTTL statements, note that, here we only evalaute if statements
+	// are valid. Check for required statements is left to the other validations.
+	if _, err := parser.ParseStatements(statements); err != nil {
+		return fmt.Errorf("failed to parse OTTL statements: %w", err)
+	}
+	// validate OTTL conditions
+	if _, err := parser.ParseConditions(mi.Conditions); err != nil {
+		return fmt.Errorf("failed to parse OTTL conditions: %w", err)
+	}
+	return nil
 }