Add JobSpecReporter service for OCR2 job spec telemetry via Beholder

core/config/app_config.go

@@ -64,6 +64,7 @@ type AppConfig interface {
 	CCV() CCV
 	Billing() Billing
 	BridgeStatusReporter() BridgeStatusReporter
+	JobSpecReporter() JobSpecReporter
 	Sharding() Sharding
 	LOOPP() LOOPP
 }

core/config/docs/core.toml

@@ -949,6 +949,17 @@ IgnoreInvalidBridges = true # Default
 # IgnoreJoblessBridges skips bridges that have no associated jobs.
 IgnoreJoblessBridges = false # Default
 
+# JobSpecReporter holds settings for the Job Spec Reporter service, which periodically emits job spec telemetry.
+[JobSpecReporter]
+# Enabled enables the Job Spec Reporter service.
+Enabled = false # Default
+# PollingInterval is how often to emit a heartbeat event for each tracked job.
+PollingInterval = "1h" # Default
+# EnabledOCR2PluginTypes restricts OCR2 telemetry to jobs with these plugin types.
+EnabledOCR2PluginTypes = ["median"] # Default
+# EmitNonOCR2Jobs emits telemetry for non-OCR2 job types (OCR1, Flux Monitor, Keeper).
+EmitNonOCR2Jobs = false # Default
+
 [CRE]
 # UseLocalTimeProvider should be set true if the DON Time OCR Plugin is not running
 UseLocalTimeProvider = true # Default

core/config/job_spec_reporter_config.go

@@ -0,0 +1,14 @@
+package config
+
+import "time"
+
+type JobSpecReporter interface {
+	Enabled() bool
+	PollingInterval() time.Duration
+	// EnabledOCR2PluginTypes is the allowlist of OCR2 plugin types to emit for
+	// (e.g. "median", "ocr2keeper"). An empty slice means emit for all types.
+	EnabledOCR2PluginTypes() []string
+	// EmitNonOCR2Jobs toggles emission for non-OCR2 job types (VRF, Keeper,
+	// Functions, CCIP, Workflow, …). Defaults to false.
+	EmitNonOCR2Jobs() bool
+}

core/config/toml/types.go

@@ -9,6 +9,7 @@ import (
 	"reflect"
 	"regexp"
 	"strings"
+	"time"
 
 	"github.com/google/uuid"
 	"go.uber.org/zap/zapcore"
@@ -65,6 +66,7 @@ type Core struct {
 	CRE                  CreConfig            `toml:",omitempty"`
 	Billing              Billing              `toml:",omitempty"`
 	BridgeStatusReporter BridgeStatusReporter `toml:",omitempty"`
+	JobSpecReporter      JobSpecReporter      `toml:",omitempty"`
 	Sharding             Sharding             `toml:",omitempty"`
 	LOOPP                LOOPP                `toml:",omitempty"`
 }
@@ -112,6 +114,7 @@ func (c *Core) SetFrom(f *Core) {
 	c.CRE.setFrom(&f.CRE)
 	c.Billing.setFrom(&f.Billing)
 	c.BridgeStatusReporter.setFrom(&f.BridgeStatusReporter)
+	c.JobSpecReporter.setFrom(&f.JobSpecReporter)
 
 	c.Sharding.setFrom(&f.Sharding)
 	c.LOOPP.setFrom(&f.LOOPP)
@@ -3086,6 +3089,55 @@ func (e *BridgeStatusReporter) ValidateConfig() error {
 	return nil
 }
 
+type JobSpecReporter struct {
+	Enabled                *bool
+	PollingInterval        *commonconfig.Duration
+	EnabledOCR2PluginTypes *[]string
+	EmitNonOCR2Jobs        *bool
+}
+
+func (e *JobSpecReporter) setFrom(f *JobSpecReporter) {
+	if f.Enabled != nil {
+		e.Enabled = f.Enabled
+	}
+	if f.PollingInterval != nil {
+		e.PollingInterval = f.PollingInterval
+	}
+	if f.EnabledOCR2PluginTypes != nil {
+		e.EnabledOCR2PluginTypes = f.EnabledOCR2PluginTypes
+	}
+	if f.EmitNonOCR2Jobs != nil {
+		e.EmitNonOCR2Jobs = f.EmitNonOCR2Jobs
+	}
+}
+
+func (e *JobSpecReporter) ValidateConfig() error {
+	if e.Enabled == nil || !*e.Enabled {
+		return nil
+	}
+
+	if e.PollingInterval == nil {
+		defaultInterval := commonconfig.MustNewDuration(time.Hour)
+		e.PollingInterval = defaultInterval
+	}
+
+	if e.PollingInterval.Duration() < config.MinimumPollingInterval {
+		return configutils.ErrInvalid{Name: "PollingInterval", Value: e.PollingInterval.Duration(), Msg: "must be greater than or equal to: " + config.MinimumPollingInterval.String()}
+	}
+
+	if e.EnabledOCR2PluginTypes == nil {
+		defaultTypes := []string{"median"}
+		e.EnabledOCR2PluginTypes = &defaultTypes
+	}
+
+	if e.EmitNonOCR2Jobs == nil {
+		defaultEmitNonOCR2 := false
+		e.EmitNonOCR2Jobs = &defaultEmitNonOCR2
+	}
+
+	return nil
+}
+
 type JobDistributor struct {
 	DisplayName *string
 }

core/scripts/go.mod

@@ -506,6 +506,7 @@ require (
 	github.com/smartcontractkit/chainlink-protos/chainlink-ccv/heartbeat v0.0.0-20260115142640-f6b99095c12e // indirect
 	github.com/smartcontractkit/chainlink-protos/chainlink-ccv/message-discovery v0.0.0-20251211142334-5c3421fe2c8d // indirect
 	github.com/smartcontractkit/chainlink-protos/chainlink-ccv/verifier v0.0.0-20251211142334-5c3421fe2c8d // indirect
+	github.com/smartcontractkit/chainlink-protos/data-feeds v0.1.1-0.20260427191023-4c493ae93432 // indirect
 	github.com/smartcontractkit/chainlink-protos/linking-service/go v0.0.0-20251002192024-d2ad9222409b // indirect
 	github.com/smartcontractkit/chainlink-protos/node-platform v0.0.0-20260319180422-b5808c964785 // indirect
 	github.com/smartcontractkit/chainlink-protos/orchestrator v0.10.0 // indirect

core/services/chainlink/application.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"math/big"
 	"net/http"
+	"os"
 	"strconv"
 	"sync"
 	"time"
@@ -71,6 +72,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/keystore"
 	"github.com/smartcontractkit/chainlink/v2/core/services/llo/retirement"
 	"github.com/smartcontractkit/chainlink/v2/core/services/nodestatusreporter/bridgestatus"
+	"github.com/smartcontractkit/chainlink/v2/core/services/nodestatusreporter/jobspec"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocr2"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ocrbootstrap"
@@ -800,8 +802,9 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 	srvcs = append(srvcs, jobSpawner, pipelineRunner)
 
 	var feedsService feeds.Service
+	var feedsORM feeds.ORM
 	if cfg.Feature().FeedsManager() {
-		feedsORM := feeds.NewORM(opts.DS, globalLogger)
+		feedsORM = feeds.NewORM(opts.DS, globalLogger)
 		feedsService = feeds.NewService(
 			feedsORM,
 			jobORM,
@@ -824,6 +827,19 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 		feedsService = &feeds.NullService{}
 	}
 
+	hostname, _ := os.Hostname()
+	jobSpecReporter := jobspec.NewJobSpecReporter(
+		cfg.JobSpecReporter(),
+		jobSpawner,
+		feedsORM,
+		beholder.GetEmitter(),
+		csaPubKeyHex,
+		static.Version,
+		hostname,
+		globalLogger,
+	)
+	srvcs = append(srvcs, jobSpecReporter)
+
 	for _, s := range srvcs {
 		if s == nil {
 			panic("service unexpectedly nil")

core/services/chainlink/config_general.go

@@ -596,6 +596,10 @@ func (g *generalConfig) BridgeStatusReporter() coreconfig.BridgeStatusReporter {
 	return &bridgeStatusReporterConfig{c: g.c.BridgeStatusReporter}
 }
 
+func (g *generalConfig) JobSpecReporter() coreconfig.JobSpecReporter {
+	return &jobSpecReporterConfig{c: g.c.JobSpecReporter}
+}
+
 func (g *generalConfig) Sharding() coreconfig.Sharding {
 	return &shardingConfig{s: g.c.Sharding}
 }

core/services/chainlink/config_job_spec_reporter.go

@@ -0,0 +1,42 @@
+package chainlink
+
+import (
+	"time"
+
+	"github.com/smartcontractkit/chainlink/v2/core/config"
+	"github.com/smartcontractkit/chainlink/v2/core/config/toml"
+)
+
+var _ config.JobSpecReporter = (*jobSpecReporterConfig)(nil)
+
+type jobSpecReporterConfig struct {
+	c toml.JobSpecReporter
+}
+
+func (e *jobSpecReporterConfig) Enabled() bool {
+	if e.c.Enabled == nil {
+		return false
+	}
+	return *e.c.Enabled
+}
+
+func (e *jobSpecReporterConfig) PollingInterval() time.Duration {
+	if e.c.PollingInterval == nil {
+		return time.Hour
+	}
+	return e.c.PollingInterval.Duration()
+}
+
+func (e *jobSpecReporterConfig) EnabledOCR2PluginTypes() []string {
+	if e.c.EnabledOCR2PluginTypes == nil {
+		return []string{"median"}
+	}
+	return *e.c.EnabledOCR2PluginTypes
+}
+
+func (e *jobSpecReporterConfig) EmitNonOCR2Jobs() bool {
+	if e.c.EmitNonOCR2Jobs == nil {
+		return false
+	}
+	return *e.c.EmitNonOCR2Jobs
+}

core/services/chainlink/config_test.go

@@ -678,6 +678,13 @@ func TestConfig_Marshal(t *testing.T) {
 		IgnoreInvalidBridges: ptr(true),
 		IgnoreJoblessBridges: ptr(false),
 	}
+	enabledOCR2PluginTypes := []string{"median"}
+	full.JobSpecReporter = toml.JobSpecReporter{
+		Enabled:                ptr(true),
+		PollingInterval:        commoncfg.MustNewDuration(time.Hour),
+		EnabledOCR2PluginTypes: &enabledOCR2PluginTypes,
+		EmitNonOCR2Jobs:        ptr(false),
+	}
 	full.Sharding = toml.Sharding{
 		ShardingEnabled:          ptr(false),
 		ArbiterPort:              ptr[uint16](9876),

core/services/chainlink/testdata/config-empty-effective.toml

@@ -399,6 +399,12 @@ PollingInterval = '5m0s'
 IgnoreInvalidBridges = true
 IgnoreJoblessBridges = false
 
+[JobSpecReporter]
+Enabled = false
+PollingInterval = '1h0m0s'
+EnabledOCR2PluginTypes = ['median']
+EmitNonOCR2Jobs = false
+
 [Sharding]
 ShardingEnabled = false
 ArbiterPort = 9876

core/services/chainlink/testdata/config-full.toml

@@ -437,6 +437,12 @@ PollingInterval = '5m0s'
 IgnoreInvalidBridges = true
 IgnoreJoblessBridges = false
 
+[JobSpecReporter]
+Enabled = true
+PollingInterval = '1h0m0s'
+EnabledOCR2PluginTypes = ['median']
+EmitNonOCR2Jobs = false
+
 [Sharding]
 ShardingEnabled = false
 ArbiterPort = 9876

core/services/chainlink/testdata/config-multi-chain-effective.toml

@@ -399,6 +399,12 @@ PollingInterval = '5m0s'
 IgnoreInvalidBridges = true
 IgnoreJoblessBridges = false
 
+[JobSpecReporter]
+Enabled = false
+PollingInterval = '1h0m0s'
+EnabledOCR2PluginTypes = ['median']
+EmitNonOCR2Jobs = false
+
 [Sharding]
 ShardingEnabled = false
 ArbiterPort = 9876