pkg/beholder: bridge prometheus metrics through otel

go.mod

@@ -16,7 +16,7 @@ require (
 	github.com/go-json-experiment/json v0.0.0-20250223041408-d3c622f1b874
 	github.com/go-playground/validator/v10 v10.26.0
 	github.com/go-viper/mapstructure/v2 v2.4.0
-	github.com/golang-jwt/jwt/v5 v5.2.3
+	github.com/golang-jwt/jwt/v5 v5.3.0
 	github.com/google/go-cmp v0.7.0
 	github.com/google/uuid v1.6.0
 	github.com/grafana/otel-profiling-go v0.5.1
@@ -37,7 +37,8 @@ require (
 	github.com/mr-tron/base58 v1.2.0
 	github.com/pelletier/go-toml v1.9.5
 	github.com/pelletier/go-toml/v2 v2.2.4
-	github.com/prometheus/client_golang v1.22.0
+	github.com/prometheus/client_golang v1.23.2
+	github.com/prometheus/client_model v0.6.2
 	github.com/santhosh-tekuri/jsonschema/v5 v5.3.1
 	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
@@ -54,6 +55,7 @@ require (
 	github.com/smartcontractkit/libocr v0.0.0-20250912173940-f3ab0246e23d
 	github.com/stellar/go-stellar-sdk v0.5.0
 	github.com/stretchr/testify v1.11.1
+	go.opentelemetry.io/contrib/bridges/prometheus v0.68.0
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.63.0
 	go.opentelemetry.io/otel v1.43.0
 	go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.12.2
@@ -133,9 +135,8 @@ require (
 	github.com/pierrec/lz4/v4 v4.1.22 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
-	github.com/prometheus/client_model v0.6.2 // indirect
-	github.com/prometheus/common v0.65.0 // indirect
-	github.com/prometheus/procfs v0.16.1 // indirect
+	github.com/prometheus/common v0.67.5 // indirect
+	github.com/prometheus/procfs v0.20.1 // indirect
 	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/ryanuber/go-glob v1.0.0 // indirect
 	github.com/sanity-io/litter v1.5.5 // indirect
@@ -148,6 +149,7 @@ require (
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.10.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
+	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	golang.org/x/mod v0.33.0 // indirect
 	golang.org/x/net v0.52.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect

pkg/beholder/client.go

@@ -567,16 +567,18 @@ func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, cre
 	if err != nil {
 		return nil, err
 	}
-	mp := sdkmetric.NewMeterProvider(
-		sdkmetric.WithReader(
-			sdkmetric.NewPeriodicReader(
-				exporter,
-				sdkmetric.WithInterval(cfg.MetricReaderInterval), // Default is 10s
-			)),
+
+	readerOpts := []sdkmetric.PeriodicReaderOption{
+		sdkmetric.WithInterval(cfg.MetricReaderInterval), // Default is 10s
+	}
+	for _, p := range cfg.MetricProducers {
+		readerOpts = append(readerOpts, sdkmetric.WithProducer(p))
+	}
+	return sdkmetric.NewMeterProvider(
+		sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exporter, readerOpts...)),
 		sdkmetric.WithResource(resource),
 		sdkmetric.WithView(cfg.MetricViews...),
-	)
-	return mp, nil
+	), nil
 }
 
 // newLoggerOpts creates options for a logger exporter

pkg/beholder/config.go

@@ -42,6 +42,7 @@ type Config struct {
 	MetricViews          []metric.View
 	// MetricCompressor sets the gRPC compressor for metrics. Valid values: "gzip" (default), "none".
 	MetricCompressor string
+	MetricProducers  []metric.Producer // For example, a prometheus bridge
 
 	// Custom Events via Chip Ingress Emitter
 	ChipIngressEmitterEnabled      bool

pkg/beholder/config_test.go

@@ -1,9 +1,16 @@
 package beholder_test
 
 import (
+	_ "embed"
+	"encoding/json"
+	"flag"
 	"fmt"
+	"os"
+	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	otelattr "go.opentelemetry.io/otel/attribute"
 	"go.uber.org/zap/zapcore"
 
@@ -14,7 +21,12 @@ const (
 	packageName = "beholder"
 )
 
-func ExampleConfig() {
+//go:embed testdata/config-example.json
+var configExample string
+
+var update = flag.Bool("update", false, "update golden test files")
+
+func TestConfig(t *testing.T) {
 	config := beholder.Config{
 		InsecureConnection:       true,
 		CACertFile:               "",
@@ -31,7 +43,7 @@ func ExampleConfig() {
 		EmitterExportInterval:     1 * time.Second,
 		EmitterMaxQueueSize:       2048,
 		// true uses batched async export for custom messages.
-		EmitterBatchProcessor:     true,
+		EmitterBatchProcessor: true,
 		// OTel message log exporter retry config
 		LogRetryConfig: nil,
 		// Trace
@@ -60,14 +72,20 @@ func ExampleConfig() {
 		AuthKeySigner:    nil,
 		AuthHeadersTTL:   0,
 	}
-	fmt.Printf("%+v\n", config)
+
+	b, err := json.MarshalIndent(config, "", "  ")
+	require.NoError(t, err)
+
+	if *update {
+		require.NoError(t, os.WriteFile("testdata/config-example.json", b, 0644))
+	} else {
+		assert.Equal(t, configExample, string(b))
+	}
+
 	config.LogRetryConfig = &beholder.RetryConfig{
 		InitialInterval: 5 * time.Second,
 		MaxInterval:     30 * time.Second,
 		MaxElapsedTime:  1 * time.Minute, // Set to zero to disable retry
 	}
-	fmt.Printf("%+v\n", *config.LogRetryConfig)
-	// Output:
-	// {InsecureConnection:true CACertFile: OtelExporterGRPCEndpoint:localhost:4317 OtelExporterHTTPEndpoint:localhost:4318 ResourceAttributes:[{Key:package_name Value:{vtype:4 numeric:0 stringly:beholder slice:<nil>}} {Key:sender Value:{vtype:4 numeric:0 stringly:beholderclient slice:<nil>}}] EmitterExportTimeout:1s EmitterExportInterval:1s EmitterExportMaxBatchSize:512 EmitterMaxQueueSize:2048 EmitterBatchProcessor:true TraceSampleRatio:1 TraceBatchTimeout:1s TraceSpanExporter:<nil> TraceRetryConfig:<nil> TraceCompressor:gzip MetricReaderInterval:1s MetricRetryConfig:<nil> MetricViews:[] MetricCompressor:gzip ChipIngressEmitterEnabled:false ChipIngressEmitterGRPCEndpoint: ChipIngressInsecureConnection:false ChipIngressBatchEmitterEnabled:false ChipIngressBufferSize:0 ChipIngressMaxBatchSize:0 ChipIngressSendInterval:0s ChipIngressSendTimeout:0s ChipIngressDrainTimeout:0s ChipIngressMaxConcurrentSends:0 ChipIngressLogger:<nil> LogExportTimeout:1s LogExportInterval:1s LogExportMaxBatchSize:512 LogMaxQueueSize:2048 LogBatchProcessor:true LogRetryConfig:<nil> LogStreamingEnabled:false LogLevel:info LogCompressor:gzip AuthHeaders:map[] AuthHeadersTTL:0s AuthKeySigner:<nil> AuthPublicKeyHex:}
-	// {InitialInterval:5s MaxInterval:30s MaxElapsedTime:1m0s}
+	assert.Equal(t, "{InitialInterval:5s MaxInterval:30s MaxElapsedTime:1m0s}", fmt.Sprintf("%+v", *config.LogRetryConfig))
 }

pkg/beholder/testdata/config-example.json

@@ -0,0 +1,61 @@
+{
+  "InsecureConnection": true,
+  "CACertFile": "",
+  "OtelExporterGRPCEndpoint": "localhost:4317",
+  "OtelExporterHTTPEndpoint": "localhost:4318",
+  "ResourceAttributes": [
+    {
+      "Key": "package_name",
+      "Value": {
+        "Type": "STRING",
+        "Value": "beholder"
+      }
+    },
+    {
+      "Key": "sender",
+      "Value": {
+        "Type": "STRING",
+        "Value": "beholderclient"
+      }
+    }
+  ],
+  "EmitterExportTimeout": 1000000000,
+  "EmitterExportInterval": 1000000000,
+  "EmitterExportMaxBatchSize": 512,
+  "EmitterMaxQueueSize": 2048,
+  "EmitterBatchProcessor": true,
+  "TraceSampleRatio": 1,
+  "TraceBatchTimeout": 1000000000,
+  "TraceSpanExporter": null,
+  "TraceRetryConfig": null,
+  "TraceCompressor": "gzip",
+  "MetricReaderInterval": 1000000000,
+  "MetricRetryConfig": null,
+  "MetricViews": null,
+  "MetricCompressor": "gzip",
+  "MetricProducers": null,
+  "ChipIngressEmitterEnabled": false,
+  "ChipIngressEmitterGRPCEndpoint": "",
+  "ChipIngressInsecureConnection": false,
+  "ChipIngressBatchEmitterEnabled": false,
+  "ChipIngressBufferSize": 0,
+  "ChipIngressMaxBatchSize": 0,
+  "ChipIngressSendInterval": 0,
+  "ChipIngressSendTimeout": 0,
+  "ChipIngressDrainTimeout": 0,
+  "ChipIngressMaxConcurrentSends": 0,
+  "ChipIngressLogger": null,
+  "LogExportTimeout": 1000000000,
+  "LogExportInterval": 1000000000,
+  "LogExportMaxBatchSize": 512,
+  "LogMaxQueueSize": 2048,
+  "LogBatchProcessor": true,
+  "LogRetryConfig": null,
+  "LogStreamingEnabled": false,
+  "LogLevel": "info",
+  "LogCompressor": "gzip",
+  "AuthHeaders": {},
+  "AuthHeadersTTL": 0,
+  "AuthKeySigner": null,
+  "AuthPublicKeyHex": ""
+}

pkg/loop/config.go

@@ -82,11 +82,13 @@ const (
 	envTelemetryLogMaxQueueSize           = "CL_TELEMETRY_LOG_MAX_QUEUE_SIZE"
 	envTelemetryTraceCompressor           = "CL_TELEMETRY_TRACE_COMPRESSOR"
 	envTelemetryMetricCompressor          = "CL_TELEMETRY_METRIC_COMPRESSOR"
+	envTelemetryPrometheusBridgeEnabled   = "CL_TELEMETRY_PROMETHEUS_BRIDGE_ENABLED"
+	envTelemetryPrometheusBridgePrefixes  = "CL_TELEMETRY_PROMETHEUS_BRIDGE_PREFIXES"
 	envTelemetryLogCompressor             = "CL_TELEMETRY_LOG_COMPRESSOR"
 
-	envChipIngressEndpoint             = "CL_CHIP_INGRESS_ENDPOINT"
-	envChipIngressInsecureConnection   = "CL_CHIP_INGRESS_INSECURE_CONNECTION"
-	envChipIngressBatchEmitterEnabled  = "CL_CHIP_INGRESS_BATCH_EMITTER_ENABLED"
+	envChipIngressEndpoint            = "CL_CHIP_INGRESS_ENDPOINT"
+	envChipIngressInsecureConnection  = "CL_CHIP_INGRESS_INSECURE_CONNECTION"
+	envChipIngressBatchEmitterEnabled = "CL_CHIP_INGRESS_BATCH_EMITTER_ENABLED"
 
 	envCRESettings        = cresettings.EnvNameSettings
 	envCRESettingsDefault = cresettings.EnvNameSettingsDefault
@@ -97,8 +99,8 @@ const (
 type EnvConfig struct {
 	AppID string
 
-	ChipIngressEndpoint           string
-	ChipIngressInsecureConnection bool
+	ChipIngressEndpoint            string
+	ChipIngressInsecureConnection  bool
 	ChipIngressBatchEmitterEnabled bool
 
 	CRESettings        string
@@ -139,15 +141,15 @@ type EnvConfig struct {
 	PyroscopePPROFBlockProfileRate     int
 	PyroscopePPROFMutexProfileFraction int
 
-	TelemetryEnabled                   bool
-	TelemetryEndpoint                  string
-	TelemetryInsecureConnection        bool
-	TelemetryCACertFile                string
-	TelemetryAttributes                OtelAttributes
-	TelemetryTraceSampleRatio          float64
-	TelemetryAuthHeaders               map[string]string
-	TelemetryAuthPubKeyHex             string
-	TelemetryAuthHeadersTTL            time.Duration
+	TelemetryEnabled            bool
+	TelemetryEndpoint           string
+	TelemetryInsecureConnection bool
+	TelemetryCACertFile         string
+	TelemetryAttributes         OtelAttributes
+	TelemetryTraceSampleRatio   float64
+	TelemetryAuthHeaders        map[string]string
+	TelemetryAuthPubKeyHex      string
+	TelemetryAuthHeadersTTL     time.Duration
 	// TelemetryEmitterBatchProcessor maps to beholder Config.EmitterBatchProcessor
 	// (batched async custom-message export vs immediate per-record export).
 	TelemetryEmitterBatchProcessor     bool
@@ -164,6 +166,8 @@ type EnvConfig struct {
 	TelemetryLogMaxQueueSize           int
 	TelemetryTraceCompressor           string
 	TelemetryMetricCompressor          string
+	TelemetryPrometheusBridgeEnabled   bool
+	TelemetryPrometheusBridgePrefixes  []string
 	TelemetryLogCompressor             string
 
 	TracingEnabled         bool
@@ -255,6 +259,8 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 	add(envTelemetryLogMaxQueueSize, strconv.Itoa(e.TelemetryLogMaxQueueSize))
 	add(envTelemetryTraceCompressor, e.TelemetryTraceCompressor)
 	add(envTelemetryMetricCompressor, e.TelemetryMetricCompressor)
+	add(envTelemetryPrometheusBridgeEnabled, strconv.FormatBool(e.TelemetryPrometheusBridgeEnabled))
+	add(envTelemetryPrometheusBridgePrefixes, strings.Join(e.TelemetryPrometheusBridgePrefixes, ","))
 	add(envTelemetryLogCompressor, e.TelemetryLogCompressor)
 
 	add(envChipIngressEndpoint, e.ChipIngressEndpoint)
@@ -484,6 +490,11 @@ func (e *EnvConfig) parse() error {
 		}
 		e.TelemetryTraceCompressor = os.Getenv(envTelemetryTraceCompressor)
 		e.TelemetryMetricCompressor = os.Getenv(envTelemetryMetricCompressor)
+		e.TelemetryPrometheusBridgeEnabled, err = getBool(envTelemetryPrometheusBridgeEnabled)
+		if err != nil {
+			return fmt.Errorf("failed to parse %s: %w", envTelemetryPrometheusBridgeEnabled, err)
+		}
+		e.TelemetryPrometheusBridgePrefixes = strings.Split(os.Getenv(envTelemetryPrometheusBridgePrefixes), ",")
 		e.TelemetryLogCompressor = os.Getenv(envTelemetryLogCompressor)
 		// Optional
 		e.ChipIngressEndpoint = os.Getenv(envChipIngressEndpoint)

pkg/loop/config_test.go

@@ -83,6 +83,8 @@ func TestEnvConfig_parse(t *testing.T) {
 				envTelemetryEmitterExportMaxBatchSize: "100",
 				envTelemetryEmitterMaxQueueSize:       "1000",
 				envTelemetryLogStreamingEnabled:       "false",
+				envTelemetryPrometheusBridgeEnabled:   "true",
+				envTelemetryPrometheusBridgePrefixes:  "foo,bar",
 
 				envChipIngressEndpoint:            "chip-ingress.example.com:50051",
 				envChipIngressInsecureConnection:  "true",
@@ -195,6 +197,8 @@ var envCfgFull = EnvConfig{
 	TelemetryEmitterExportMaxBatchSize: 100,
 	TelemetryEmitterMaxQueueSize:       1000,
 	TelemetryLogStreamingEnabled:       false,
+	TelemetryPrometheusBridgeEnabled:   true,
+	TelemetryPrometheusBridgePrefixes:  []string{"foo", "bar"},
 
 	ChipIngressEndpoint:            "chip-ingress.example.com:50051",
 	ChipIngressInsecureConnection:  true,
@@ -257,6 +261,8 @@ func TestEnvConfig_AsCmdEnv(t *testing.T) {
 	assert.Equal(t, "100", got[envTelemetryEmitterExportMaxBatchSize])
 	assert.Equal(t, "1000", got[envTelemetryEmitterMaxQueueSize])
 	assert.Equal(t, "false", got[envTelemetryLogStreamingEnabled])
+	assert.Equal(t, "true", got[envTelemetryPrometheusBridgeEnabled])
+	assert.Equal(t, "foo,bar", got[envTelemetryPrometheusBridgePrefixes])
 
 	// Assert ChipIngress environment variables
 	assert.Equal(t, "chip-ingress.example.com:50051", got[envChipIngressEndpoint])