Merge branch 'main' into infoplat-14703-chip_client-label

Author: thomaska

observability-lib/go.mod

@@ -3,8 +3,8 @@ module github.com/smartcontractkit/chainlink-common/observability-lib
 go 1.26.2
 
 require (
-	github.com/go-resty/resty/v2 v2.17.1
-	github.com/grafana/grafana-foundation-sdk/go v0.0.12
+	github.com/go-resty/resty/v2 v2.17.2
+	github.com/grafana/grafana-foundation-sdk/go v0.0.18
 	github.com/spf13/cobra v1.10.2
 	github.com/stretchr/testify v1.11.1
 	gopkg.in/yaml.v3 v3.0.1

observability-lib/go.sum

@@ -76,6 +76,7 @@ type MathExpression struct {
 
 type ResampleExpression struct {
 	Expression  string
+	Window      string
 	DownSampler expr.TypeResampleDownsampler
 	UpSampler   expr.TypeResampleUpsampler
 }
@@ -137,47 +138,57 @@ func newConditionQuery(options ConditionQuery) *alerting.QueryBuilder {
 		DatasourceUid("__expr__")
 
 	if options.ReduceExpression != nil {
-		reduceBuider := expr.NewTypeReduceBuilder().
+		reduceBuilder := expr.NewTypeReduceBuilder().
 			RefId(options.RefID).
 			Expression(options.ReduceExpression.Expression).
 			IntervalMs(*options.IntervalMs).
 			MaxDataPoints(*options.MaxDataPoints).
 			Reducer(options.ReduceExpression.Reducer)
 
 		if options.ReduceExpression.ReduceSettings != nil && options.ReduceExpression.ReduceSettings.Mode != "" {
-			reduceBuider.Settings(newReduceSettingsOptions(*options.ReduceExpression.ReduceSettings))
+			reduceBuilder.Settings(newReduceSettingsOptions(*options.ReduceExpression.ReduceSettings))
 		}
-		res.Model(reduceBuider)
+		res.Model(expr.NewTypeMathOrTypeReduceOrTypeResampleOrTypeClassicConditionsOrTypeThresholdOrTypeSqlBuilder().
+			TypeReduce(reduceBuilder))
 	}
 
 	if options.MathExpression != nil {
-		res.Model(expr.NewTypeMathBuilder().
-			RefId(options.RefID).
-			Expression(options.MathExpression.Expression).
-			IntervalMs(*options.IntervalMs).
-			MaxDataPoints(*options.MaxDataPoints),
-		)
+		res.Model(expr.NewTypeMathOrTypeReduceOrTypeResampleOrTypeClassicConditionsOrTypeThresholdOrTypeSqlBuilder().
+			TypeMath(expr.NewTypeMathBuilder().
+				RefId(options.RefID).
+				Expression(options.MathExpression.Expression).
+				IntervalMs(*options.IntervalMs).
+				MaxDataPoints(*options.MaxDataPoints),
+			))
 	}
 
 	if options.ResampleExpression != nil {
-		res.Model(expr.NewTypeResampleBuilder().
-			RefId(options.RefID).
-			Expression(options.ResampleExpression.Expression).
-			IntervalMs(*options.IntervalMs).
-			MaxDataPoints(*options.MaxDataPoints).
-			Downsampler(options.ResampleExpression.DownSampler).
-			Upsampler(options.ResampleExpression.UpSampler),
-		)
+		window := options.ResampleExpression.Window
+		if window == "" {
+			// Window is required (>= 1 char) by the SDK; default to the relative time range.
+			window = "10m"
+		}
+		res.Model(expr.NewTypeMathOrTypeReduceOrTypeResampleOrTypeClassicConditionsOrTypeThresholdOrTypeSqlBuilder().
+			TypeResample(expr.NewTypeResampleBuilder().
+				RefId(options.RefID).
+				Expression(options.ResampleExpression.Expression).
+				IntervalMs(*options.IntervalMs).
+				MaxDataPoints(*options.MaxDataPoints).
+				Window(window).
+				Downsampler(options.ResampleExpression.DownSampler).
+				Upsampler(options.ResampleExpression.UpSampler),
+			))
 	}
 
 	if options.ThresholdExpression != nil {
-		res.Model(expr.NewTypeThresholdBuilder().
-			RefId(options.RefID).
-			Expression(options.ThresholdExpression.Expression).
-			IntervalMs(*options.IntervalMs).
-			MaxDataPoints(*options.MaxDataPoints).
-			Conditions(newThresholdConditionsOptions(options.ThresholdExpression.ThresholdConditionsOptions)),
-		)
+		res.Model(expr.NewTypeMathOrTypeReduceOrTypeResampleOrTypeClassicConditionsOrTypeThresholdOrTypeSqlBuilder().
+			TypeThreshold(expr.NewTypeThresholdBuilder().
+				RefId(options.RefID).
+				Expression(options.ThresholdExpression.Expression).
+				IntervalMs(*options.IntervalMs).
+				MaxDataPoints(*options.MaxDataPoints).
+				Conditions(newThresholdConditionsOptions(options.ThresholdExpression.ThresholdConditionsOptions)),
+			))
 	}
 
 	return res

observability-lib/grafana/alerts.go

@@ -1,6 +1,7 @@
 package grafana_test
 
 import (
+	"encoding/json"
 	"testing"
 
 	"github.com/grafana/grafana-foundation-sdk/go/alerting"
@@ -481,6 +482,81 @@ func TestBuilder_AddPanel(t *testing.T) {
 	})
 }
 
+func TestNewAlertRule_ConditionExpressionModels(t *testing.T) {
+	t.Run("builds a model for every expression type", func(t *testing.T) {
+		ruleBuilder := grafana.NewAlertRule(&grafana.AlertOptions{
+			Title: "Expression Alert",
+			Query: []grafana.RuleQuery{
+				{
+					Expr:       `my_metric`,
+					Instant:    true,
+					RefID:      "A",
+					Datasource: "datasource-uid",
+				},
+			},
+			QueryRefCondition: "E",
+			Condition: []grafana.ConditionQuery{
+				{
+					RefID: "B",
+					ReduceExpression: &grafana.ReduceExpression{
+						Expression: "A",
+						Reducer:    expr.TypeReduceReducerSum,
+					},
+				},
+				{
+					RefID: "C",
+					MathExpression: &grafana.MathExpression{
+						Expression: "$B * 2",
+					},
+				},
+				{
+					RefID: "D",
+					ResampleExpression: &grafana.ResampleExpression{
+						Expression:  "A",
+						Window:      "10s",
+						DownSampler: expr.TypeResampleDownsamplerMean,
+						UpSampler:   expr.TypeResampleUpsamplerPad,
+					},
+				},
+				{
+					RefID: "E",
+					ThresholdExpression: &grafana.ThresholdExpression{
+						Expression: "D",
+						ThresholdConditionsOptions: grafana.ThresholdConditionsOption{
+							Params: []float64{2},
+							Type:   expr.ExprTypeThresholdConditionsEvaluatorTypeLt,
+						},
+					},
+				},
+			},
+		})
+
+		rule, err := ruleBuilder.Build()
+		require.NoError(t, err)
+
+		models := map[string]string{}
+		for _, query := range rule.Data {
+			require.NotNil(t, query.RefId)
+			raw, errMarshal := json.Marshal(query.Model)
+			require.NoError(t, errMarshal)
+			models[*query.RefId] = string(raw)
+		}
+
+		require.Contains(t, models["B"], `"type":"reduce"`)
+
+		require.Contains(t, models["C"], `"type":"math"`)
+		require.Contains(t, models["C"], `"expression":"$B * 2"`)
+
+		require.Contains(t, models["D"], `"type":"resample"`)
+		require.Contains(t, models["D"], `"downsampler":"mean"`)
+		require.Contains(t, models["D"], `"upsampler":"pad"`)
+		require.Contains(t, models["D"], `"expression":"A"`)
+		require.Contains(t, models["D"], `"window":"10s"`)
+
+		require.Contains(t, models["E"], `"type":"threshold"`)
+	})
+}
+
 func TestBuilder_AddTimeSeriesPanelWithAlert(t *testing.T) {
 	t.Run("AddPanel adds a panel to the dashboard", func(t *testing.T) {
 		builder := grafana.NewBuilder(&grafana.BuilderOptions{

observability-lib/grafana/builder_test.go

@@ -123,8 +123,8 @@ func (l *Link) UnmarshalJSON(data []byte) error {
 // UnmarshalText implements the encoding.TextUnmarshaler interface.
 func (l *Link) UnmarshalText(text []byte) error {
 	s := string(text)
-	if strings.HasSuffix(s, "link") {
-		s = strings.TrimSuffix(s, "link")
+	if before, ok := strings.CutSuffix(s, "link"); ok {
+		s = before
 		s = strings.TrimSuffix(s, " ")
 		d, err := decimal.NewFromString(s)
 		if err != nil {
@@ -138,8 +138,8 @@ func (l *Link) UnmarshalText(text []byte) error {
 		l.Set((*Link)(d.Rat().Num()))
 		return nil
 	}
-	if strings.HasSuffix(s, "juels") {
-		s = strings.TrimSuffix(s, "juels")
+	if before, ok := strings.CutSuffix(s, "juels"); ok {
+		s = before
 		s = strings.TrimSuffix(s, " ")
 	}
 	if _, ok := l.SetString(s, 10); !ok {

pkg/assets/link.go

@@ -95,7 +95,7 @@ func TestChipIngressBatchEmitterService_Emit(t *testing.T) {
 		require.NoError(t, err)
 		require.NoError(t, emitter.Start(t.Context()))
 
-		for i := 0; i < 3; i++ {
+		for range 3 {
 			err = emitter.Emit(t.Context(), []byte("body"),
 				beholder.AttrKeyDomain, "platform",
 				beholder.AttrKeyEntity, "TestEvent",
@@ -190,7 +190,7 @@ func TestChipIngressBatchEmitterService_PublishBatchError(t *testing.T) {
 	require.NoError(t, err)
 	require.NoError(t, emitter.Start(t.Context()))
 
-	for i := 0; i < 3; i++ {
+	for range 3 {
 		err = emitter.Emit(t.Context(), []byte("body"),
 			beholder.AttrKeyDomain, "platform",
 			beholder.AttrKeyEntity, "TestEvent",
@@ -401,7 +401,7 @@ func TestChipIngressBatchEmitterService_EmitWithCallback(t *testing.T) {
 		<-firstCallSignal
 		time.Sleep(100 * time.Millisecond)
 
-		for i := 0; i < 10; i++ {
+		for range 10 {
 			_ = emitter.Emit(t.Context(), []byte("filler"),
 				beholder.AttrKeyDomain, "platform",
 				beholder.AttrKeyEntity, "TestEvent",

pkg/beholder/batch_emitter_service_test.go

@@ -549,11 +549,11 @@ func newMeterProvider(cfg Config, resource *sdkresource.Resource, auth Auth, cre
 	for _, p := range cfg.MetricProducers {
 		readerOpts = append(readerOpts, sdkmetric.WithProducer(p))
 	}
-	return sdkmetric.NewMeterProvider(
+	mpOpts := append(cfg.metricOptions(),
 		sdkmetric.WithReader(sdkmetric.NewPeriodicReader(exporter, readerOpts...)),
 		sdkmetric.WithResource(resource),
-		sdkmetric.WithView(cfg.MetricViews...),
-	), nil
+	)
+	return sdkmetric.NewMeterProvider(mpOpts...), nil
 }
 
 // newLoggerOpts creates options for a logger exporter

pkg/beholder/client.go

@@ -785,7 +785,7 @@ func TestChipIngressClient(t *testing.T) {
 	t.Run("noop client PublishBatch returns empty response", func(t *testing.T) {
 		noopClient := &chipingress.NoopClient{}
 		events := []chipingress.CloudEvent{}
-		for i := 0; i < 3; i++ {
+		for i := range 3 {
 			event, err := chipingress.NewEvent(fmt.Sprintf("domain-%d", i), "test.type", []byte("test"), nil)
 			require.NoError(t, err)
 			events = append(events, event)
@@ -868,7 +868,7 @@ func TestChipIngressClient(t *testing.T) {
 		mockClient := chipmocks.NewClient(t)
 
 		events := []chipingress.CloudEvent{}
-		for i := 0; i < 2; i++ {
+		for i := range 2 {
 			event, err := chipingress.NewEvent(fmt.Sprintf("domain-%d", i), "test.type", []byte("test"), nil)
 			require.NoError(t, err)
 			events = append(events, event)
@@ -936,7 +936,7 @@ func TestClient_batchEmitterService(t *testing.T) {
 	newBatchClient := func(t *testing.T) *beholder.Client {
 		t.Helper()
 		client, err := beholder.NewClient(beholder.Config{
-			OtelExporterGRPCEndpoint:       "localhost:4317",
+			OtelExporterGRPCEndpoint: "localhost:4317",
 			// Use simple exporter in this lifecycle test to avoid batch flush/shutdown delays.
 			EmitterBatchProcessor:          false,
 			LogBatchProcessor:              false,

pkg/beholder/client_test.go

@@ -26,7 +26,7 @@ type Config struct {
 	EmitterMaxQueueSize       int
 	// EmitterBatchProcessor controls custom-message export mode:
 	// true = batched async export; false = immediate per-record export.
-	EmitterBatchProcessor     bool
+	EmitterBatchProcessor bool
 
 	// OTel Trace
 	TraceSampleRatio  float64
@@ -39,7 +39,10 @@ type Config struct {
 	// OTel Metric
 	MetricReaderInterval time.Duration
 	MetricRetryConfig    *RetryConfig
-	MetricViews          []metric.View
+	MetricViews []metric.View
+	// MetricCardinalityLimit sets the SDK per-instrument attribute-set limit (0 = disabled).
+	// DefaultConfig uses DefaultMetricCardinalityLimit as a production safety valve for high-cardinality workloads.
+	MetricCardinalityLimit int
 	// MetricCompressor sets the gRPC compressor for metrics. Valid values: "gzip" (default), "none".
 	MetricCompressor string
 	MetricProducers  []metric.Producer // For example, a prometheus bridge
@@ -108,6 +111,9 @@ var defaultRetryConfig = RetryConfig{
 const (
 	defaultPackageName        = "beholder"
 	defaultMaxConcurrentSends = 10
+
+	// DefaultMetricCardinalityLimit is the production safety valve for high-cardinality workloads.
+	DefaultMetricCardinalityLimit = 100000
 )
 
 var defaultOtelAttributes = []attribute.KeyValue{
@@ -127,7 +133,7 @@ func DefaultConfig() Config {
 		EmitterExportInterval:     1 * time.Second,
 		EmitterMaxQueueSize:       2048,
 		// Keep batched export enabled by default for throughput.
-		EmitterBatchProcessor:     true,
+		EmitterBatchProcessor: true,
 		// OTel message log exporter retry config
 		LogRetryConfig: defaultRetryConfig.Copy(),
 		// Trace
@@ -137,8 +143,9 @@ func DefaultConfig() Config {
 		// OTel trace exporter retry config
 		TraceRetryConfig: defaultRetryConfig.Copy(),
 		// Metric
-		MetricReaderInterval: 1 * time.Second,
-		MetricCompressor:     "gzip",
+		MetricReaderInterval:   1 * time.Second,
+		MetricCompressor:       "gzip",
+		MetricCardinalityLimit: DefaultMetricCardinalityLimit,
 		// OTel metric exporter retry config
 		MetricRetryConfig: defaultRetryConfig.Copy(),
 		// Log
@@ -148,8 +155,8 @@ func DefaultConfig() Config {
 		LogMaxQueueSize:       2048,
 		LogBatchProcessor:     true,
 		LogStreamingEnabled:   true, // Enable logs streaming by default
-		LogLevel:      zapcore.InfoLevel,
-		LogCompressor: "gzip",
+		LogLevel:              zapcore.InfoLevel,
+		LogCompressor:         "gzip",
 		// Chip Ingress Batch Emitter
 		ChipIngressBatchEmitterEnabled: false,
 		ChipIngressBufferSize:          1000,
@@ -173,6 +180,7 @@ func TestDefaultConfig() Config {
 	config.LogRetryConfig.MaxElapsedTime = 0    // Retry is disabled
 	config.TraceRetryConfig.MaxElapsedTime = 0  // Retry is disabled
 	config.MetricRetryConfig.MaxElapsedTime = 0 // Retry is disabled
+	config.MetricCardinalityLimit = 0           // Disable overflow aggregation in unit tests
 	// Auth disabled for testing (TTL=0 means static auth mode)
 	config.AuthHeadersTTL = 0
 	return config
@@ -186,6 +194,7 @@ func TestDefaultConfigHTTPClient() Config {
 	config.LogBatchProcessor = false
 	config.OtelExporterGRPCEndpoint = ""
 	config.OtelExporterHTTPEndpoint = "localhost:4318"
+	config.MetricCardinalityLimit = 0
 	// Auth disabled for testing (TTL=0 means static auth mode)
 	config.AuthHeadersTTL = 0
 	return config

pkg/beholder/config.go

@@ -53,8 +53,9 @@ func TestConfig(t *testing.T) {
 		// OTel trace exporter retry config
 		TraceRetryConfig: nil,
 		// Metric
-		MetricReaderInterval: 1 * time.Second,
-		MetricCompressor:     "gzip",
+		MetricReaderInterval:   1 * time.Second,
+		MetricCompressor:       "gzip",
+		MetricCardinalityLimit: 0,
 		// OTel metric exporter retry config
 		MetricRetryConfig: nil,
 		// Log