Common Job spec reporting via Beholder

core/services/chainlink/application.go

@@ -62,6 +62,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/blockheaderfeeder"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ccv/ccvcommitteeverifier"
 	"github.com/smartcontractkit/chainlink/v2/core/services/ccv/ccvexecutor"
+	"github.com/smartcontractkit/chainlink/v2/core/services/cljobinfo"
 	"github.com/smartcontractkit/chainlink/v2/core/services/cre"
 	"github.com/smartcontractkit/chainlink/v2/core/services/cresettings"
 	"github.com/smartcontractkit/chainlink/v2/core/services/cron"
@@ -589,6 +590,9 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 	)
 	srvcs = append(srvcs, workflowORM)
 
+	// Superseded by cljobinfo.Reporter (wired below), which emits a full,
+	// type-agnostic job definition; retained until consumers migrate off the
+	// flat submitter-address projection.
 	nodePlatformJobInfo := NewNodePlatformJobInfoService(NewNodePlatformJobInfoConfig(opts, jobORM, relayChainInterops))
 	srvcs = append(srvcs, &nodePlatformJobInfo)
 
@@ -862,6 +866,8 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 	}
 
 	hostname, _ := os.Hostname()
+	// Superseded by cljobinfo.Reporter (wired below); this OCR2-only reporter is
+	// retained until its consumers migrate to the generic CLJobInfo schema.
 	jobSpecReporter := jobspec.NewJobSpecReporter(
 		cfg.JobSpecReporter(),
 		jobSpawner,
@@ -874,6 +880,17 @@ func NewApplication(ctx context.Context, opts ApplicationOpts) (Application, err
 	)
 	srvcs = append(srvcs, jobSpecReporter)
 
+	// CLJobInfo: single, type-agnostic reporter that emits a full job
+	// definition (as TOML) on create/delete/heartbeat for every job type.
+	clJobInfoReporter := cljobinfo.NewReporter(
+		jobSpawner,
+		beholder.GetEmitter(),
+		cljobinfo.NodeIdentity{CSAPublicKey: csaPubKeyHex, NodeVersion: static.Version, Hostname: hostname},
+		cljobinfo.DefaultPollInterval,
+		globalLogger,
+	)
+	srvcs = append(srvcs, clJobInfoReporter)
+
 	for _, s := range srvcs {
 		if s == nil {
 			panic("service unexpectedly nil")

core/services/cljobinfo/cljobinfo.go

@@ -0,0 +1,207 @@
+// Package cljobinfo provides a single, generic way for any part of the core
+// node to emit a full job definition as telemetry.
+//
+// It supersedes the two prior, divergent approaches:
+//
+//   - NodePlatformJobInfoService (core/services/chainlink/node_platform.go)
+//     emits a flat, denormalized projection (chain_id/job_type/field_path ->
+//     addresses) aggregated across all jobs. Generic schema, but only carries
+//     submitter addresses and has hardcoded per-spec extractors.
+//   - JobSpecReporter (core/services/nodestatusreporter/jobspec) emits a
+//     rich, per-job, create/delete/heartbeat event, but models each spec as a
+//     dedicated proto message and only supports OCR2.
+//
+// cljobinfo keeps the better half of each: the event-driven, per-job lifecycle
+// and node identity of JobSpecReporter, and a schema-agnostic payload like
+// NodePlatformJobInfo. The full, type-specific job definition is carried as a
+// raw TOML string, so any job type is supported with no per-type code here and
+// none for future job types.
+package cljobinfo
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/pelletier/go-toml"
+	"google.golang.org/protobuf/proto"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/services"
+	commonv1 "github.com/smartcontractkit/chainlink-protos/node-platform/common/v1"
+
+	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	"github.com/smartcontractkit/chainlink/v2/core/services/job"
+)
+
+const (
+	// Domain, Entity and DataSchema identify CLJobInfo telemetry on Beholder.
+	Domain     = "node-platform"
+	Entity     = "common.v1.CLJobInfo"
+	DataSchema = "/node-platform/common/v1"
+
+	ServiceName = "CLJobInfoReporter"
+
+	// DefaultPollInterval is the heartbeat cadence when a caller does not
+	// specify one, matching the node-platform build/job info beat.
+	DefaultPollInterval = 3 * time.Minute
+)
+
+// NodeIdentity is the node-level context attached to every emitted CLJobInfo.
+type NodeIdentity struct {
+	CSAPublicKey string
+	NodeVersion  string
+	Hostname     string
+}
+
+// Build converts any job.Job into its generic CLJobInfo representation.
+//
+// The complete, type-specific job definition is captured as TOML, so no
+// per-job-type code lives here and none is needed for future job types. If the
+// job cannot be TOML-encoded, Build still returns a fully populated identity
+// payload (with an empty SpecToml) alongside the encoding error, so callers can
+// choose to emit the envelope and log the failure rather than drop the event.
+func Build(jb job.Job, trigger commonv1.CLJobInfoTrigger, id NodeIdentity, now time.Time) (*commonv1.CLJobInfo, error) {
+	info := &commonv1.CLJobInfo{
+		CsaPublicKey:      id.CSAPublicKey,
+		NodeVersion:       id.NodeVersion,
+		Hostname:          id.Hostname,
+		ExternalJobId:     jb.ExternalJobID.String(),
+		JobId:             jb.ID,
+		Name:              jb.Name.ValueOrZero(),
+		JobType:           string(jb.Type),
+		SchemaVersion:     jb.SchemaVersion,
+		ForwardingAllowed: jb.ForwardingAllowed,
+		CreatedAt:         formatTime(jb.CreatedAt),
+		Trigger:           trigger,
+		Timestamp:         now.UTC().Format(time.RFC3339Nano),
+	}
+	if jb.GasLimit.Valid {
+		info.GasLimit = new(jb.GasLimit.Uint32)
+	}
+	if jb.StreamID != nil {
+		info.StreamId = new(*jb.StreamID)
+	}
+
+	specTOML, err := jobTOML(jb)
+	if err != nil {
+		return info, fmt.Errorf("encoding job %s (%d) spec to TOML: %w", jb.ExternalJobID, jb.ID, err)
+	}
+	info.SpecToml = specTOML
+
+	return info, nil
+}
+
+// Emit marshals a CLJobInfo and publishes it to Beholder.
+func Emit(ctx context.Context, emitter beholder.Emitter, info *commonv1.CLJobInfo) error {
+	payload, err := proto.Marshal(info)
+	if err != nil {
+		return fmt.Errorf("marshaling CLJobInfo: %w", err)
+	}
+
+	err = emitter.Emit(ctx, payload,
+		beholder.AttrKeyDomain, Domain,
+		beholder.AttrKeyEntity, Entity,
+		beholder.AttrKeyDataSchema, DataSchema,
+	)
+	if err != nil {
+		return fmt.Errorf("emitting CLJobInfo: %w", err)
+	}
+	return nil
+}
+
+// jobTOML serializes the entire job definition to TOML. Marshaling the whole
+// job.Job captures both the common top-level fields and the single active
+// type-specific spec, so all fields for any job type are included without
+// enumerating them.
+func jobTOML(jb job.Job) (string, error) {
+	out, err := toml.Marshal(jb)
+	if err != nil {
+		return "", err
+	}
+	return string(out), nil
+}
+
+func formatTime(t time.Time) string {
+	if t.IsZero() {
+		return ""
+	}
+	return t.UTC().Format(time.RFC3339Nano)
+}
+
+var _ job.Listener = (*Reporter)(nil)
+
+// Reporter emits a CLJobInfo for every job on create, delete, and on a
+// recurring heartbeat. It is job-type agnostic: any job the node runs is
+// reported through the single generic schema.
+type Reporter struct {
+	services.Service
+	eng *services.Engine
+
+	spawner      job.Spawner
+	emitter      beholder.Emitter
+	identity     NodeIdentity
+	pollInterval time.Duration
+}
+
+// NewReporter builds a Reporter that reports every job the node runs.
+func NewReporter(
+	spawner job.Spawner,
+	emitter beholder.Emitter,
+	identity NodeIdentity,
+	pollInterval time.Duration,
+	lggr logger.Logger,
+) *Reporter {
+	r := &Reporter{
+		spawner:      spawner,
+		emitter:      emitter,
+		identity:     identity,
+		pollInterval: pollInterval,
+	}
+	r.Service, r.eng = services.Config{
+		Name:  ServiceName,
+		Start: r.start,
+	}.NewServiceEngine(lggr)
+	return r
+}
+
+func (r *Reporter) start(_ context.Context) error {
+	r.spawner.RegisterListener(r)
+	r.eng.GoTick(services.NewTicker(r.pollInterval), r.pollAllJobs)
+	return nil
+}
+
+func (r *Reporter) HealthReport() map[string]error {
+	return map[string]error{ServiceName: r.Ready()}
+}
+
+// AfterJobStarted emits a create event when a job starts.
+func (r *Reporter) AfterJobStarted(ctx context.Context, jb job.Job) {
+	r.emitForJob(ctx, jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE)
+}
+
+// AfterJobStopped emits a delete event when a job is removed.
+func (r *Reporter) AfterJobStopped(ctx context.Context, jb job.Job) {
+	r.emitForJob(ctx, jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_DELETE)
+}
+
+// pollAllJobs emits a heartbeat event for every active job.
+func (r *Reporter) pollAllJobs(ctx context.Context) {
+	for _, jb := range r.spawner.ActiveJobs() {
+		r.emitForJob(ctx, jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_HEARTBEAT)
+	}
+}
+
+func (r *Reporter) emitForJob(ctx context.Context, jb job.Job, trigger commonv1.CLJobInfoTrigger) {
+	info, err := Build(jb, trigger, r.identity, time.Now())
+	if err != nil {
+		// Spec encoding failed; still emit the identity envelope so the job is
+		// accounted for, but flag the gap.
+		r.eng.Warnw("Failed to encode job spec for CLJobInfo; emitting without spec_toml",
+			"jobID", jb.ID, "externalJobID", jb.ExternalJobID, "error", err)
+	}
+
+	if err := Emit(ctx, r.emitter, info); err != nil {
+		r.eng.Warnw("Failed to emit CLJobInfo", "jobID", jb.ID, "trigger", trigger, "error", err)
+	}
+}

core/services/cljobinfo/emit_test.go

@@ -0,0 +1,116 @@
+package cljobinfo_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pelletier/go-toml"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/protobuf/proto"
+	"gopkg.in/guregu/null.v4"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
+	"github.com/smartcontractkit/chainlink-common/pkg/beholder/beholdertest"
+	"github.com/smartcontractkit/chainlink-common/pkg/sqlutil"
+	commontypes "github.com/smartcontractkit/chainlink-common/pkg/types"
+	evmtypes "github.com/smartcontractkit/chainlink-evm/pkg/types"
+	commonv1 "github.com/smartcontractkit/chainlink-protos/node-platform/common/v1"
+
+	"github.com/smartcontractkit/chainlink/v2/core/services/cljobinfo"
+	"github.com/smartcontractkit/chainlink/v2/core/services/job"
+	"github.com/smartcontractkit/chainlink/v2/core/services/pipeline"
+)
+
+func sampleJob() job.Job {
+	streamID := uint32(42)
+	return job.Job{
+		ID:                7,
+		Name:              null.StringFrom("my-ocr2-job"),
+		Type:              job.OffchainReporting2,
+		SchemaVersion:     1,
+		ForwardingAllowed: true,
+		StreamID:          &streamID,
+		CreatedAt:         time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC),
+		OCR2OracleSpec: &job.OCR2OracleSpec{
+			Relay:         "evm",
+			ChainID:       "1",
+			PluginType:    commontypes.Median,
+			ContractID:    "0xcccccccccccccccccccccccccccccccccccccccc",
+			TransmitterID: null.StringFrom("0x1111111111111111111111111111111111111111"),
+			RelayConfig: job.JSONConfig{
+				"chainID":     "1",
+				"sendingKeys": []any{"0x1111111111111111111111111111111111111111"},
+			},
+		},
+		VRFSpec: nil,
+		Pipeline: pipeline.Pipeline{Tasks: []pipeline.Task{
+			&pipeline.ETHTxTask{From: "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
+		}},
+	}
+}
+
+// TestBuild_EncodesFullSpecAsTOML is the load-bearing check: an arbitrary job
+// must round-trip to TOML with no per-type code.
+func TestBuild_EncodesFullSpecAsTOML(t *testing.T) {
+	jb := sampleJob()
+	id := cljobinfo.NodeIdentity{CSAPublicKey: "csa", NodeVersion: "1.2.3", Hostname: "host-1"}
+
+	info, err := cljobinfo.Build(jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, id, time.Date(2026, 7, 24, 12, 0, 0, 0, time.UTC))
+	require.NoError(t, err)
+
+	require.Equal(t, "csa", info.CsaPublicKey)
+	require.Equal(t, "1.2.3", info.NodeVersion)
+	require.Equal(t, "host-1", info.Hostname)
+	require.Equal(t, int32(7), info.JobId)
+	require.Equal(t, "my-ocr2-job", info.Name)
+	require.Equal(t, "offchainreporting2", info.JobType)
+	require.Equal(t, uint32(1), info.SchemaVersion)
+	require.True(t, info.ForwardingAllowed)
+	require.NotNil(t, info.StreamId)
+	require.Equal(t, uint32(42), *info.StreamId)
+	require.Equal(t, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, info.Trigger)
+	require.NotEmpty(t, info.Timestamp)
+
+	// spec_toml must be valid TOML and contain type-specific spec data.
+	require.NotEmpty(t, info.SpecToml)
+	var decoded map[string]any
+	require.NoError(t, toml.Unmarshal([]byte(info.SpecToml), &decoded))
+	require.Contains(t, info.SpecToml, "median")
+	require.Contains(t, info.SpecToml, "0xcccccccccccccccccccccccccccccccccccccccc")
+}
+
+func TestBuild_HandlesMultipleJobTypesGenerically(t *testing.T) {
+	jobs := []job.Job{
+		{Type: job.VRF, VRFSpec: &job.VRFSpec{
+			EVMChainID:    sqlutil.NewI(4),
+			FromAddresses: []evmtypes.EIP55Address{evmtypes.MustEIP55Address("0x6666666666666666666666666666666666666666")},
+		}},
+		{Type: job.BlockhashStore, BlockhashStoreSpec: &job.BlockhashStoreSpec{EVMChainID: sqlutil.NewI(5)}},
+	}
+	for _, jb := range jobs {
+		info, err := cljobinfo.Build(jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_HEARTBEAT, cljobinfo.NodeIdentity{}, time.Now())
+		require.NoErrorf(t, err, "job type %s should encode without per-type code", jb.Type)
+		require.NotEmpty(t, info.SpecToml)
+	}
+}
+
+func TestEmit_PublishesToBeholder(t *testing.T) {
+	obs := beholdertest.NewObserver(t)
+
+	info, err := cljobinfo.Build(sampleJob(), commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, cljobinfo.NodeIdentity{CSAPublicKey: "csa"}, time.Now())
+	require.NoError(t, err)
+	require.NoError(t, cljobinfo.Emit(t.Context(), beholder.GetEmitter(), info))
+
+	msgs := obs.Messages(t, beholder.AttrKeyEntity, cljobinfo.Entity)
+	require.NotEmpty(t, msgs)
+
+	msg := msgs[0]
+	require.Equal(t, cljobinfo.Domain, msg.Attrs[beholder.AttrKeyDomain])
+	require.Equal(t, cljobinfo.DataSchema, msg.Attrs[beholder.AttrKeyDataSchema])
+
+	var payload commonv1.CLJobInfo
+	require.NoError(t, proto.Unmarshal(msg.Body, &payload))
+	require.Equal(t, "csa", payload.CsaPublicKey)
+	require.Equal(t, "offchainreporting2", payload.JobType)
+	require.NotEmpty(t, payload.SpecToml)
+}