feat: add option to send forwarded metrics to Observe via OTLP (#212)

### Description

Add option to send forwarded metrics to Observe via OTLP. Going forward,
we want to be able to send all custom metrics to Observe in the format
the user prefers (between OTLP or Prometheus)

Author: obs-gh-mattcotter

go.mod

@@ -17,7 +17,6 @@ require (
 	go.opentelemetry.io/collector/otelcol v0.124.0
 	go.uber.org/zap v1.27.0
 	golang.org/x/sys v0.32.0
-	gopkg.in/yaml.v2 v2.4.0
 	gopkg.in/yaml.v3 v3.0.1
 )
 
@@ -442,6 +441,7 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
 	k8s.io/api v0.32.3 // indirect
 	k8s.io/apimachinery v0.32.3 // indirect
 	k8s.io/client-go v0.32.3 // indirect

integration/scripts/snapshots/docker.yaml

@@ -14,6 +14,17 @@ exporters:
         sending_queue:
             num_consumers: 4
             queue_size: 100
+    otlphttp/observemetrics:
+        compression: zstd
+        endpoint: https://123456789.collect.observe-eng.com/v2/otel
+        headers:
+            authorization: Bearer abcdefghijklmnopqrst:OWt0SXV4YTlqYUhpSHZjSlhXUDVkRXpl
+            x-observe-target-package: Metrics
+        retry_on_failure:
+            enabled: true
+        sending_queue:
+            num_consumers: 4
+            queue_size: 100
     otlphttp/observetracing:
         compression: zstd
         endpoint: https://123456789.collect.observe-eng.com/v2/otel
@@ -279,10 +290,13 @@ service:
                 - otlphttp/observe
                 - count
             processors:
+                - memory_limiter
+                - transform/truncate
                 - resourcedetection
                 - resourcedetection/cloud
                 - attributes/observe_global_attributes
                 - resource/observe_global_resource_attributes
+                - batch
             receivers:
                 - otlp
         logs/host_monitoring-file:
@@ -354,6 +368,7 @@ service:
             exporters:
                 - prometheusremotewrite/observe
             processors:
+                - memory_limiter
                 - resourcedetection
                 - resourcedetection/cloud
                 - attributes/observe_global_attributes
@@ -390,10 +405,13 @@ service:
             exporters:
                 - otlphttp/observetracing
             processors:
+                - memory_limiter
+                - transform/truncate
                 - resourcedetection
                 - resourcedetection/cloud
                 - attributes/observe_global_attributes
                 - resource/observe_global_resource_attributes
+                - batch
             receivers:
                 - otlp
     telemetry:

integration/scripts/snapshots/linux.yaml

@@ -14,6 +14,17 @@ exporters:
         sending_queue:
             num_consumers: 4
             queue_size: 100
+    otlphttp/observemetrics:
+        compression: zstd
+        endpoint: https://123456789.collect.observe-eng.com/v2/otel
+        headers:
+            authorization: Bearer abcdefghijklmnopqrst:OWt0SXV4YTlqYUhpSHZjSlhXUDVkRXpl
+            x-observe-target-package: Metrics
+        retry_on_failure:
+            enabled: true
+        sending_queue:
+            num_consumers: 4
+            queue_size: 100
     otlphttp/observetracing:
         compression: zstd
         endpoint: https://123456789.collect.observe-eng.com/v2/otel
@@ -273,10 +284,13 @@ service:
                 - otlphttp/observe
                 - count
             processors:
+                - memory_limiter
+                - transform/truncate
                 - resourcedetection
                 - resourcedetection/cloud
                 - attributes/observe_global_attributes
                 - resource/observe_global_resource_attributes
+                - batch
             receivers:
                 - otlp
         logs/host_monitoring-file:
@@ -348,6 +362,7 @@ service:
             exporters:
                 - prometheusremotewrite/observe
             processors:
+                - memory_limiter
                 - resourcedetection
                 - resourcedetection/cloud
                 - attributes/observe_global_attributes
@@ -384,10 +399,13 @@ service:
             exporters:
                 - otlphttp/observetracing
             processors:
+                - memory_limiter
+                - transform/truncate
                 - resourcedetection
                 - resourcedetection/cloud
                 - attributes/observe_global_attributes
                 - resource/observe_global_resource_attributes
+                - batch
             receivers:
                 - otlp
     telemetry:

integration/scripts/snapshots/windows.yaml

@@ -14,6 +14,17 @@ exporters:
         sending_queue:
             num_consumers: 4
             queue_size: 100
+    otlphttp/observemetrics:
+        compression: zstd
+        endpoint: https://123456789.collect.observe-eng.com/v2/otel
+        headers:
+            authorization: Bearer abcdefghijklmnopqrst:OWt0SXV4YTlqYUhpSHZjSlhXUDVkRXpl
+            x-observe-target-package: Metrics
+        retry_on_failure:
+            enabled: true
+        sending_queue:
+            num_consumers: 4
+            queue_size: 100
     otlphttp/observetracing:
         compression: zstd
         endpoint: https://123456789.collect.observe-eng.com/v2/otel
@@ -277,7 +288,6 @@ service:
                 - prometheusremotewrite/observe
             processors:
                 - memory_limiter
-                - transform/truncate
                 - resourcedetection
                 - resourcedetection/cloud
                 - attributes/observe_global_attributes

internal/commands/diagnose/configcheck.go

@@ -2,12 +2,13 @@ package diagnose
 
 import (
 	"embed"
+	"errors"
 	"fmt"
 	"os"
 
 	"github.com/observeinc/observe-agent/internal/config"
 	"github.com/spf13/viper"
-	"gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v3"
 )
 
 type ConfigTestResult struct {
@@ -18,31 +19,52 @@ type ConfigTestResult struct {
 }
 
 func checkConfig(v *viper.Viper) (bool, any, error) {
+	// Ensure there is an observe-agent config file.
 	configFile := v.ConfigFileUsed()
 	if configFile == "" {
 		return false, nil, fmt.Errorf("no config file defined")
 	}
+	if _, err := os.Stat(configFile); err != nil && errors.Is(err, os.ErrNotExist) {
+		return false, nil, fmt.Errorf("config file %s does not exist", configFile)
+	}
+
+	// Ensure the file is valid yaml.
 	contents, err := os.ReadFile(configFile)
 	if err != nil {
-		return false, nil, err
+		return false, nil, fmt.Errorf("error reading config file %s: %w", configFile, err)
 	}
-	var conf config.AgentConfig
-	if err = yaml.Unmarshal(contents, &conf); err != nil {
+	var yamlMap map[string]any
+	if err = yaml.Unmarshal(contents, &yamlMap); err != nil {
 		return false, ConfigTestResult{
 			ConfigFile:     configFile,
 			ParseSucceeded: false,
 			IsValid:        false,
 			Error:          err.Error(),
 		}, nil
 	}
-	if err = conf.Validate(); err != nil {
+
+	// Ensure the agent config can be loaded via viper.
+	agentConfig, err := config.AgentConfigFromViper(v)
+	if err != nil {
+		return false, ConfigTestResult{
+			ConfigFile:     configFile,
+			ParseSucceeded: false,
+			IsValid:        false,
+			Error:          err.Error(),
+		}, nil
+	}
+
+	// Ensure the agent config is valid.
+	if err = agentConfig.Validate(); err != nil {
 		return false, ConfigTestResult{
 			ConfigFile:     configFile,
 			ParseSucceeded: true,
 			IsValid:        false,
 			Error:          err.Error(),
 		}, nil
 	}
+
+	// All checks passed.
 	return true, ConfigTestResult{
 		ConfigFile:     configFile,
 		ParseSucceeded: true,

internal/commands/diagnose/configcheck_test.go

@@ -18,6 +18,11 @@ observe_url: "https://collect.observeinc.com"
 # Debug mode - Sets agent log level to debug
 debug: false
 
+forwarding:
+  enabled: true
+  metrics:
+    output_format: otel
+
 # collect metrics and logs pertaining to the agent itself
 self_monitoring:
   enabled: true
@@ -62,10 +67,15 @@ func Test_checkConfig(t *testing.T) {
 		f, err := os.CreateTemp("", "test-config-*.yaml")
 		assert.NoError(t, err)
 		defer os.Remove(f.Name())
-		f.Write([]byte(tc.confStr))
+		_, err = f.Write([]byte(tc.confStr))
+		assert.NoError(t, err)
 
 		v := viper.New()
 		v.SetConfigFile(f.Name())
+		err = v.ReadInConfig()
+		if tc.shouldParse {
+			assert.NoError(t, err)
+		}
 		success, resultAny, err := checkConfig(v)
 		assert.NoError(t, err)
 		result, ok := resultAny.(ConfigTestResult)

internal/commands/initconfig/initconfig.go

@@ -19,6 +19,7 @@ var (
 	observe_url                             string
 	cloud_resource_detectors                []string
 	resource_attributes                     map[string]string
+	forwarding_metrics_format               string
 	self_monitoring_enabled                 bool
 	host_monitoring_enabled                 bool
 	host_monitoring_logs_enabled            bool
@@ -87,6 +88,9 @@ func RegisterConfigFlags(cmd *cobra.Command, v *viper.Viper) {
 	cmd.PersistentFlags().StringToStringVar(&resource_attributes, "resource_attributes", map[string]string{}, "The cloud environments from which to detect resources")
 	v.BindPFlag("resource_attributes", cmd.PersistentFlags().Lookup("resource_attributes"))
 
+	cmd.PersistentFlags().StringVar(&forwarding_metrics_format, "forwarding::metrics::output_format", "", "Format for sending app metrics to Observe, valid options are 'prometheus' and 'otel'")
+	v.BindPFlag("forwarding::metrics::output_format", cmd.PersistentFlags().Lookup("forwarding::metrics::output_format"))
+
 	cmd.PersistentFlags().BoolVar(&self_monitoring_enabled, "self_monitoring::enabled", true, "Enable self monitoring")
 	v.BindPFlag("self_monitoring::enabled", cmd.PersistentFlags().Lookup("self_monitoring::enabled"))
 	v.SetDefault("self_monitoring::enabled", true)

internal/commands/initconfig/initconfig_test.go

@@ -74,6 +74,33 @@ func Test_InitConfigCommand(t *testing.T) {
 			}),
 			expectErr: "",
 		},
+		{
+			args: []string{"--config_path=./test-config.yaml", "--token=test-token", "--observe_url=test-url", "--forwarding::metrics::output_format=otel"},
+			expectedConfig: setConfigDefaults(config.AgentConfig{
+				Token:      "test-token",
+				ObserveURL: "test-url",
+				Forwarding: config.ForwardingConfig{
+					Metrics: config.ForwardingMetricsConfig{
+						OutputFormat: "otel",
+					},
+				},
+				SelfMonitoring: config.SelfMonitoringConfig{
+					Enabled: true,
+				},
+				HostMonitoring: config.HostMonitoringConfig{
+					Enabled: true,
+					Logs: config.HostMonitoringLogsConfig{
+						Enabled: true,
+					},
+					Metrics: config.HostMonitoringMetricsConfig{
+						Host: config.HostMonitoringHostMetricsConfig{
+							Enabled: true,
+						},
+					},
+				},
+			}),
+			expectErr: "",
+		},
 	}
 	for _, tc := range testcases {
 		v := viper.NewWithOptions(viper.KeyDelimiter("::"))

internal/config/configschema.go

@@ -46,8 +46,17 @@ type HealthCheckConfig struct {
 	Path     string `yaml:"path" mapstructure:"path" default:"/status"`
 }
 
+type ForwardingMetricsConfig struct {
+	OutputFormat string `yaml:"output_format,omitempty" mapstructure:"output_format" default:"prometheus"`
+}
+
+func (config *ForwardingMetricsConfig) OtlpMetrics() bool {
+	return config.OutputFormat == "otel"
+}
+
 type ForwardingConfig struct {
-	Enabled bool `yaml:"enabled" mapstructure:"enabled" default:"true"`
+	Enabled bool                    `yaml:"enabled" mapstructure:"enabled" default:"true"`
+	Metrics ForwardingMetricsConfig `yaml:"metrics,omitempty" mapstructure:"metrics"`
 }
 
 type InternalTelemetryMetricsConfig struct {
@@ -145,5 +154,10 @@ func (config *AgentConfig) Validate() error {
 	if !strings.Contains(config.Token, ":") {
 		return errors.New("invalid Token, the provided value may be the token ID instead of the token itself")
 	}
+
+	if config.Forwarding.Metrics.OutputFormat != "prometheus" && config.Forwarding.Metrics.OutputFormat != "otel" {
+		return fmt.Errorf("invalid metrics forwarding output format '%s' - valid options are 'prometheus' and 'otel'", config.Forwarding.Metrics.OutputFormat)
+	}
+
 	return nil
 }