Merge branch 'main' into SRE-1613-add-time-range-to-alert-rule-specs

observability-lib/grafana/panels.go

@@ -88,21 +88,53 @@ func newTransform(options *TransformOptions) dashboard.DataTransformerConfig {
 	}
 }
 
+type ToolTipOptions struct {
+	Mode      common.TooltipDisplayMode
+	Sort      common.SortOrder
+	MaxWidth  *float64
+	MaxHeight *float64
+}
+
+func newToolTip(options *ToolTipOptions) *common.VizTooltipOptionsBuilder {
+	if options.Mode == "" {
+		options.Mode = common.TooltipDisplayModeSingle
+	}
+
+	if options.Sort == "" {
+		options.Sort = common.SortOrderNone
+	}
+
+	builder := common.NewVizTooltipOptionsBuilder().
+		Mode(options.Mode).
+		Sort(options.Sort)
+
+	if options.MaxWidth != nil {
+		builder.MaxWidth(*options.MaxWidth)
+	}
+
+	if options.MaxHeight != nil {
+		builder.MaxHeight(*options.MaxHeight)
+	}
+
+	return builder
+}
+
 type PanelOptions struct {
-	Datasource  string
-	Title       string
-	Description string
-	Span        uint32
-	Height      uint32
-	Decimals    float64
-	Unit        string
-	NoValue     string
-	Min         *float64
-	Max         *float64
-	Query       []Query
-	Threshold   *ThresholdOptions
-	Transform   *TransformOptions
-	ColorScheme dashboard.FieldColorModeId
+	Datasource    string
+	Title         string
+	Description   string
+	Span          uint32
+	Height        uint32
+	Decimals      float64
+	Unit          string
+	NoValue       string
+	Min           *float64
+	Max           *float64
+	MaxDataPoints *float64
+	Query         []Query
+	Threshold     *ThresholdOptions
+	Transform     *TransformOptions
+	ColorScheme   dashboard.FieldColorModeId
 }
 
 type Panel struct {
@@ -183,6 +215,10 @@ func NewStatPanel(options *StatPanelOptions) *Panel {
 		Mappings(options.Mappings).
 		ReduceOptions(common.NewReduceDataOptionsBuilder().Calcs([]string{"last"}))
 
+	if options.MaxDataPoints != nil {
+		newPanel.MaxDataPoints(*options.MaxDataPoints)
+	}
+
 	if options.Min != nil {
 		newPanel.Min(*options.Min)
 	}
@@ -230,6 +266,7 @@ type TimeSeriesPanelOptions struct {
 	FillOpacity       float64
 	ScaleDistribution common.ScaleDistribution
 	LegendOptions     *LegendOptions
+	ToolTipOptions    *ToolTipOptions
 	ThresholdStyle    common.GraphThresholdsStyleMode
 }
 
@@ -244,6 +281,10 @@ func NewTimeSeriesPanel(options *TimeSeriesPanelOptions) *Panel {
 		options.LegendOptions = &LegendOptions{}
 	}
 
+	if options.ToolTipOptions == nil {
+		options.ToolTipOptions = &ToolTipOptions{}
+	}
+
 	newPanel := timeseries.NewPanelBuilder().
 		Datasource(datasourceRef(options.Datasource)).
 		Title(options.Title).
@@ -257,7 +298,12 @@ func NewTimeSeriesPanel(options *TimeSeriesPanelOptions) *Panel {
 		Legend(newLegend(options.LegendOptions)).
 		ScaleDistribution(common.NewScaleDistributionConfigBuilder().
 			Type(options.ScaleDistribution),
-		)
+		).
+		Tooltip(newToolTip(options.ToolTipOptions))
+
+	if options.MaxDataPoints != nil {
+		newPanel.MaxDataPoints(*options.MaxDataPoints)
+	}
 
 	if options.Min != nil {
 		newPanel.Min(*options.Min)
@@ -326,6 +372,10 @@ func NewGaugePanel(options *GaugePanelOptions) *Panel {
 				Calcs([]string{"lastNotNull"}).Values(false),
 		)
 
+	if options.MaxDataPoints != nil {
+		newPanel.MaxDataPoints(*options.MaxDataPoints)
+	}
+
 	if options.Min != nil {
 		newPanel.Min(*options.Min)
 	}
@@ -368,6 +418,10 @@ func NewTablePanel(options *TablePanelOptions) *Panel {
 		Unit(options.Unit).
 		NoValue(options.NoValue)
 
+	if options.MaxDataPoints != nil {
+		newPanel.MaxDataPoints(*options.MaxDataPoints)
+	}
+
 	if options.Min != nil {
 		newPanel.Min(*options.Min)
 	}
@@ -414,6 +468,10 @@ func NewLogPanel(options *LogPanelOptions) *Panel {
 		NoValue(options.NoValue).
 		PrettifyLogMessage(options.PrettifyJSON)
 
+	if options.MaxDataPoints != nil {
+		newPanel.MaxDataPoints(*options.MaxDataPoints)
+	}
+
 	if options.Min != nil {
 		newPanel.Min(*options.Min)
 	}

pkg/beholder/client.go

@@ -115,6 +115,15 @@ func newGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 		if cfg.LogExportTimeout > 0 {
 			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportTimeout(cfg.LogExportTimeout)) // Default is 30s
 		}
+		if cfg.LogExportMaxBatchSize > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportMaxBatchSize(cfg.LogExportMaxBatchSize)) // Default is 512, must be <= maxQueueSize
+		}
+		if cfg.LogExportInterval > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportInterval(cfg.LogExportInterval)) // Default is 1s
+		}
+		if cfg.LogMaxQueueSize > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithMaxQueueSize(cfg.LogMaxQueueSize)) // Default is 2048
+		}
 		loggerProcessor = sdklog.NewBatchProcessor(
 			sharedLogExporter,
 			batchProcessorOpts...,
@@ -159,6 +168,15 @@ func newGRPCClient(cfg Config, otlploggrpcNew otlploggrpcFactory) (*Client, erro
 		if cfg.EmitterExportTimeout > 0 {
 			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportTimeout(cfg.EmitterExportTimeout)) // Default is 30s
 		}
+		if cfg.EmitterExportMaxBatchSize > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportMaxBatchSize(cfg.EmitterExportMaxBatchSize)) // Default is 512, must be <= maxQueueSize
+		}
+		if cfg.EmitterExportInterval > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportInterval(cfg.EmitterExportInterval)) // Default is 1s
+		}
+		if cfg.EmitterMaxQueueSize > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithMaxQueueSize(cfg.EmitterMaxQueueSize)) // Default is 2048
+		}
 		messageLogProcessor = sdklog.NewBatchProcessor(
 			sharedLogExporter,
 			batchProcessorOpts...,
@@ -361,7 +379,6 @@ func newMeterProvider(config Config, resource *sdkresource.Resource, creds crede
 	if err != nil {
 		return nil, err
 	}
-
 	mp := sdkmetric.NewMeterProvider(
 		sdkmetric.WithReader(
 			sdkmetric.NewPeriodicReader(

pkg/beholder/config.go

@@ -17,24 +17,29 @@ type Config struct {
 	// OTel Resource
 	ResourceAttributes []otelattr.KeyValue
 	// Message Emitter
-	EmitterExportTimeout time.Duration
-	// Batch processing is enabled by default
-	// Disable it only for testing
-	EmitterBatchProcessor bool
+	EmitterExportTimeout      time.Duration
+	EmitterExportInterval     time.Duration
+	EmitterExportMaxBatchSize int
+	EmitterMaxQueueSize       int
+	EmitterBatchProcessor     bool // Enabled by default. Disable only for testing.
+
 	// OTel Trace
 	TraceSampleRatio  float64
 	TraceBatchTimeout time.Duration
 	TraceSpanExporter sdktrace.SpanExporter // optional additional exporter
 	TraceRetryConfig  *RetryConfig
+
 	// OTel Metric
 	MetricReaderInterval time.Duration
 	MetricRetryConfig    *RetryConfig
 	MetricViews          []sdkmetric.View
+
 	// OTel Log
-	LogExportTimeout time.Duration
-	// Batch processing is enabled by default
-	// Disable it only for testing
-	LogBatchProcessor bool
+	LogExportTimeout      time.Duration
+	LogExportInterval     time.Duration
+	LogExportMaxBatchSize int
+	LogMaxQueueSize       int
+	LogBatchProcessor     bool // Enabled by default. Disable only for testing.
 	// Retry config for shared log exporter, used by Emitter and Logger
 	LogRetryConfig *RetryConfig
 
@@ -81,8 +86,11 @@ func DefaultConfig() Config {
 		// Resource
 		ResourceAttributes: defaultOtelAttributes,
 		// Message Emitter
-		EmitterExportTimeout:  1 * time.Second,
-		EmitterBatchProcessor: true,
+		EmitterExportTimeout:      30 * time.Second,
+		EmitterExportMaxBatchSize: 512,
+		EmitterExportInterval:     1 * time.Second,
+		EmitterMaxQueueSize:       2048,
+		EmitterBatchProcessor:     true,
 		// OTel message log exporter retry config
 		LogRetryConfig: defaultRetryConfig.Copy(),
 		// Trace
@@ -95,8 +103,11 @@ func DefaultConfig() Config {
 		// OTel metric exporter retry config
 		MetricRetryConfig: defaultRetryConfig.Copy(),
 		// Log
-		LogExportTimeout:  1 * time.Second,
-		LogBatchProcessor: true,
+		LogExportTimeout:      30 * time.Second,
+		LogExportMaxBatchSize: 512,
+		LogExportInterval:     1 * time.Second,
+		LogMaxQueueSize:       2048,
+		LogBatchProcessor:     true,
 	}
 }
 

pkg/beholder/config_test.go

@@ -25,8 +25,11 @@ func ExampleConfig() {
 			otelattr.String("sender", "beholderclient"),
 		},
 		// Message Emitter
-		EmitterExportTimeout:  1 * time.Second,
-		EmitterBatchProcessor: true,
+		EmitterExportTimeout:      1 * time.Second,
+		EmitterExportMaxBatchSize: 512,
+		EmitterExportInterval:     1 * time.Second,
+		EmitterMaxQueueSize:       2048,
+		EmitterBatchProcessor:     true,
 		// OTel message log exporter retry config
 		LogRetryConfig: nil,
 		// Trace
@@ -39,8 +42,11 @@ func ExampleConfig() {
 		// OTel metric exporter retry config
 		MetricRetryConfig: nil,
 		// Log
-		LogExportTimeout:  1 * time.Second,
-		LogBatchProcessor: true,
+		LogExportTimeout:      1 * time.Second,
+		LogExportMaxBatchSize: 512,
+		LogExportInterval:     1 * time.Second,
+		LogMaxQueueSize:       2048,
+		LogBatchProcessor:     true,
 	}
 	fmt.Printf("%+v\n", config)
 	config.LogRetryConfig = &beholder.RetryConfig{
@@ -50,6 +56,6 @@ func ExampleConfig() {
 	}
 	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 EmitterBatchProcessor:true TraceSampleRatio:1 TraceBatchTimeout:1s TraceSpanExporter:<nil> TraceRetryConfig:<nil> MetricReaderInterval:1s MetricRetryConfig:<nil> MetricViews:[] LogExportTimeout:1s LogBatchProcessor:true LogRetryConfig:<nil> AuthPublicKeyHex: AuthHeaders:map[]}
+	// {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> MetricReaderInterval:1s MetricRetryConfig:<nil> MetricViews:[] LogExportTimeout:1s LogExportInterval:1s LogExportMaxBatchSize:512 LogMaxQueueSize:2048 LogBatchProcessor:true LogRetryConfig:<nil> AuthPublicKeyHex: AuthHeaders:map[]}
 	// {InitialInterval:5s MaxInterval:30s MaxElapsedTime:1m0s}
 }

pkg/beholder/httpclient.go

@@ -80,6 +80,16 @@ func newHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro
 		if cfg.LogExportTimeout > 0 {
 			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportTimeout(cfg.LogExportTimeout)) // Default is 30s
 		}
+		if cfg.LogExportMaxBatchSize > 0 {
+
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportMaxBatchSize(cfg.LogExportMaxBatchSize)) // Default is 512, must be <= maxQueueSize
+		}
+		if cfg.LogExportInterval > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportInterval(cfg.LogExportInterval)) // Default is 1s
+		}
+		if cfg.LogMaxQueueSize > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithMaxQueueSize(cfg.LogMaxQueueSize)) // Default is 2048
+		}
 		loggerProcessor = sdklog.NewBatchProcessor(
 			sharedLogExporter,
 			batchProcessorOpts...,
@@ -124,9 +134,18 @@ func newHTTPClient(cfg Config, otlploghttpNew otlploghttpFactory) (*Client, erro
 		if cfg.EmitterExportTimeout > 0 {
 			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportTimeout(cfg.EmitterExportTimeout)) // Default is 30s
 		}
+		if cfg.EmitterExportMaxBatchSize > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportMaxBatchSize(cfg.EmitterExportMaxBatchSize)) // Default is 512, must be <= maxQueueSize
+		}
+		if cfg.EmitterExportInterval > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithExportInterval(cfg.EmitterExportInterval)) // Default is 1s
+		}
+		if cfg.EmitterMaxQueueSize > 0 {
+			batchProcessorOpts = append(batchProcessorOpts, sdklog.WithMaxQueueSize(cfg.EmitterMaxQueueSize)) // Default is 2048
+		}
 		messageLogProcessor = sdklog.NewBatchProcessor(
 			sharedLogExporter,
-			batchProcessorOpts..., // Default is 30s
+			batchProcessorOpts...,
 		)
 	} else {
 		messageLogProcessor = sdklog.NewSimpleProcessor(sharedLogExporter)

pkg/capabilities/capabilities.go

@@ -67,6 +67,8 @@ type RequestMetadata struct {
 	WorkflowDonConfigVersion uint32
 	// The step reference ID of the workflow
 	ReferenceID string
+	// Use DecodedWorkflowName if the human readable name needs to be exposed, such as for logging purposes.
+	DecodedWorkflowName string
 }
 
 type RegistrationMetadata struct {

pkg/capabilities/consensus/ocr3/types/aggregator.go

@@ -13,22 +13,23 @@ const MetadataFieldName = "INTERNAL_METADATA"
 
 type Metadata struct {
 	Version          uint32 //  1 byte
-	ExecutionID      string // 32 hex bytes
+	ExecutionID      string // 32 hex bytes (string len  = 64)
 	Timestamp        uint32 //  4 bytes
 	DONID            uint32 //  4 bytes
 	DONConfigVersion uint32 //  4 bytes
-	WorkflowID       string // 32 hex bytes
-	WorkflowName     string // 10 hex bytes
-	WorkflowOwner    string // 20 hex bytes
-	ReportID         string //  2 hex bytes
+	WorkflowID       string // 32 hex bytes (string len = 64)
+	WorkflowName     string // 10 hex bytes (string len = 20)
+	WorkflowOwner    string // 20 hex bytes (string len = 40)
+	ReportID         string //  2 hex bytes (string len = 4)
 }
 
 // the contract requires exactly 10 bytes for the workflow name
-// the json schema allows for a variable length string <= len(10)
-// pad with trailing spaces to meet the contract requirements
+// the resulting workflow name should be up to 10 bytes long
+// so pad accordingly to meet the contract requirements
 func (m *Metadata) padWorkflowName() {
-	if len(m.WorkflowName) < 10 {
-		suffix := strings.Repeat(" ", 10-len(m.WorkflowName))
+	// it should have 10 hex bytes, so 20 characters total
+	if len(m.WorkflowName) < 20 {
+		suffix := strings.Repeat("0", 20-len(m.WorkflowName))
 		m.WorkflowName += suffix
 	}
 }