From bc82b9de473874ee9b267825a0acda2c96bb6335 Mon Sep 17 00:00:00 2001 From: skudasov Date: Fri, 24 Jul 2026 12:52:57 +0200 Subject: [PATCH 1/3] wip --- core/services/chainlink/application.go | 17 ++ core/services/cljobinfo/cljobinfo.go | 207 +++++++++++++++++++++++++ core/services/cljobinfo/emit_test.go | 116 ++++++++++++++ 3 files changed, 340 insertions(+) create mode 100644 core/services/cljobinfo/cljobinfo.go create mode 100644 core/services/cljobinfo/emit_test.go diff --git a/core/services/chainlink/application.go b/core/services/chainlink/application.go index 468b0c6529b..8e19b3cbe53 100644 --- a/core/services/chainlink/application.go +++ b/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") diff --git a/core/services/cljobinfo/cljobinfo.go b/core/services/cljobinfo/cljobinfo.go new file mode 100644 index 00000000000..0e251fb651c --- /dev/null +++ b/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) + } +} diff --git a/core/services/cljobinfo/emit_test.go b/core/services/cljobinfo/emit_test.go new file mode 100644 index 00000000000..c58f327bad2 --- /dev/null +++ b/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) +} From a66cc6d89ea0c84be7f26f607398a84b402e1efd Mon Sep 17 00:00:00 2001 From: gheorghestrimtu Date: Fri, 4 Sep 2026 15:00:10 +0300 Subject: [PATCH 2/3] refactor(nodestatusreporter): report every job type from the existing reporter Fold core/services/cljobinfo into the existing nodestatusreporter/jobspec service instead of adding a parallel reporter. CLJobInfo is now emitted for every job regardless of type, alongside the unchanged OCR2-only JobSpecEvent track. application.go and the [JobSpecReporter] config are unchanged. Requires: https://github.com/smartcontractkit/chainlink-protos/pull/432 RANE-4655 --- core/services/chainlink/application.go | 17 -- core/services/cljobinfo/cljobinfo.go | 207 ------------------ .../nodestatusreporter/jobspec/cl_job_info.go | 134 ++++++++++++ .../jobspec/cl_job_info_test.go} | 76 +++++-- .../jobspec/job_spec_reporter.go | 100 +++++++-- 5 files changed, 277 insertions(+), 257 deletions(-) delete mode 100644 core/services/cljobinfo/cljobinfo.go create mode 100644 core/services/nodestatusreporter/jobspec/cl_job_info.go rename core/services/{cljobinfo/emit_test.go => nodestatusreporter/jobspec/cl_job_info_test.go} (51%) diff --git a/core/services/chainlink/application.go b/core/services/chainlink/application.go index 8e19b3cbe53..468b0c6529b 100644 --- a/core/services/chainlink/application.go +++ b/core/services/chainlink/application.go @@ -62,7 +62,6 @@ 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" @@ -590,9 +589,6 @@ 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) @@ -866,8 +862,6 @@ 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, @@ -880,17 +874,6 @@ 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") diff --git a/core/services/cljobinfo/cljobinfo.go b/core/services/cljobinfo/cljobinfo.go deleted file mode 100644 index 0e251fb651c..00000000000 --- a/core/services/cljobinfo/cljobinfo.go +++ /dev/null @@ -1,207 +0,0 @@ -// 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) - } -} diff --git a/core/services/nodestatusreporter/jobspec/cl_job_info.go b/core/services/nodestatusreporter/jobspec/cl_job_info.go new file mode 100644 index 00000000000..f5852cc36a0 --- /dev/null +++ b/core/services/nodestatusreporter/jobspec/cl_job_info.go @@ -0,0 +1,134 @@ +package jobspec + +import ( + "context" + "fmt" + "time" + + "github.com/pelletier/go-toml" + "google.golang.org/protobuf/proto" + + "github.com/smartcontractkit/chainlink-common/pkg/beholder" + commonv1 "github.com/smartcontractkit/chainlink-protos/node-platform/common/v1" + + "github.com/smartcontractkit/chainlink/v2/core/services/job" +) + +// CLJobInfo is the generic, job-type-agnostic half of this reporter. Where +// JobSpecEvent models one job type (OCR2) field by field, CLJobInfo carries the +// job's common identity plus the complete definition as a raw TOML string, so +// every job the node runs is reported through a single schema with no +// per-type code here and none needed for future job types. +// +// It is emitted on the same triggers and from the same service as JobSpecEvent +// rather than from a parallel one, so there is exactly one place in the node +// that reports what jobs it runs. Once consumers have migrated, the OCR2-only +// half can be deleted from here without touching the wiring. +const ( + // Domain, Entity and DataSchema identify CLJobInfo telemetry on Beholder. + Domain = "node-platform" + Entity = "common.v1.CLJobInfo" + DataSchema = "/node-platform/common/v1" +) + +// NodeIdentity is the node-level context attached to every emitted CLJobInfo. +type NodeIdentity struct { + CSAPublicKey string + NodeVersion string + Hostname string +} + +// JobProposal is the Job Distributor provenance for a job that arrived as an +// approved job proposal. Jobs created directly (CLI, UI, TOML on disk) have no +// proposal, and the zero value leaves the corresponding CLJobInfo fields unset +// — which is how a consumer tells a managed job from an unmanaged one. +type JobProposal struct { + FeedsManagerID int64 + RemoteUUID string + SpecVersion int32 + ProposedAt time.Time + ApprovedAt time.Time +} + +// BuildCLJobInfo converts any job.Job into its generic CLJobInfo representation. +// +// prop is optional: pass nil for a job with no Job Distributor proposal. +// +// If the job cannot be TOML-encoded, BuildCLJobInfo 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 BuildCLJobInfo(jb job.Job, trigger commonv1.CLJobInfoTrigger, id NodeIdentity, prop *JobProposal, 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: formatCLJobInfoTime(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) + } + if prop != nil { + info.FeedsManagerId = &prop.FeedsManagerID + info.RemoteUuid = &prop.RemoteUUID + info.SpecVersion = &prop.SpecVersion + info.ProposedAt = new(formatCLJobInfoTime(prop.ProposedAt)) + info.ApprovedAt = new(formatCLJobInfoTime(prop.ApprovedAt)) + } + + 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 +} + +// EmitCLJobInfo marshals a CLJobInfo and publishes it to Beholder. +func EmitCLJobInfo(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 formatCLJobInfoTime(t time.Time) string { + if t.IsZero() { + return "" + } + return t.UTC().Format(time.RFC3339Nano) +} diff --git a/core/services/cljobinfo/emit_test.go b/core/services/nodestatusreporter/jobspec/cl_job_info_test.go similarity index 51% rename from core/services/cljobinfo/emit_test.go rename to core/services/nodestatusreporter/jobspec/cl_job_info_test.go index c58f327bad2..0b326c7e1c1 100644 --- a/core/services/cljobinfo/emit_test.go +++ b/core/services/nodestatusreporter/jobspec/cl_job_info_test.go @@ -1,4 +1,4 @@ -package cljobinfo_test +package jobspec_test import ( "testing" @@ -16,12 +16,12 @@ import ( 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/nodestatusreporter/jobspec" "github.com/smartcontractkit/chainlink/v2/core/services/pipeline" ) -func sampleJob() job.Job { +func clJobInfoSampleJob() job.Job { streamID := uint32(42) return job.Job{ ID: 7, @@ -42,20 +42,19 @@ func sampleJob() job.Job { "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"} +// TestBuildCLJobInfo_EncodesFullSpecAsTOML is the load-bearing check: an +// arbitrary job must round-trip to TOML with no per-type code. +func TestBuildCLJobInfo_EncodesFullSpecAsTOML(t *testing.T) { + jb := clJobInfoSampleJob() + id := jobspec.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)) + info, err := jobspec.BuildCLJobInfo(jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, id, nil, time.Date(2026, 7, 24, 12, 0, 0, 0, time.UTC)) require.NoError(t, err) require.Equal(t, "csa", info.CsaPublicKey) @@ -79,7 +78,7 @@ func TestBuild_EncodesFullSpecAsTOML(t *testing.T) { require.Contains(t, info.SpecToml, "0xcccccccccccccccccccccccccccccccccccccccc") } -func TestBuild_HandlesMultipleJobTypesGenerically(t *testing.T) { +func TestBuildCLJobInfo_HandlesMultipleJobTypesGenerically(t *testing.T) { jobs := []job.Job{ {Type: job.VRF, VRFSpec: &job.VRFSpec{ EVMChainID: sqlutil.NewI(4), @@ -88,25 +87,66 @@ func TestBuild_HandlesMultipleJobTypesGenerically(t *testing.T) { {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()) + info, err := jobspec.BuildCLJobInfo(jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_HEARTBEAT, jobspec.NodeIdentity{}, nil, 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) { +// TestBuildCLJobInfo_CarriesJobDistributorProvenance covers the JD join key: +// remote_uuid is what links this event back to api.job.v1.Job.uuid. +func TestBuildCLJobInfo_CarriesJobDistributorProvenance(t *testing.T) { + proposedAt := time.Date(2026, 7, 20, 9, 0, 0, 0, time.UTC) + approvedAt := time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC) + prop := &jobspec.JobProposal{ + FeedsManagerID: 3, + RemoteUUID: "6d7d9d1a-0d0f-4b3f-9a2f-2e4a1c0b8d55", + SpecVersion: 2, + ProposedAt: proposedAt, + ApprovedAt: approvedAt, + } + + info, err := jobspec.BuildCLJobInfo(clJobInfoSampleJob(), commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, jobspec.NodeIdentity{}, prop, time.Now()) + require.NoError(t, err) + + require.NotNil(t, info.FeedsManagerId) + require.Equal(t, int64(3), *info.FeedsManagerId) + require.NotNil(t, info.RemoteUuid) + require.Equal(t, "6d7d9d1a-0d0f-4b3f-9a2f-2e4a1c0b8d55", *info.RemoteUuid) + require.NotNil(t, info.SpecVersion) + require.Equal(t, int32(2), *info.SpecVersion) + require.NotNil(t, info.ProposedAt) + require.Equal(t, proposedAt.Format(time.RFC3339Nano), *info.ProposedAt) + require.NotNil(t, info.ApprovedAt) + require.Equal(t, approvedAt.Format(time.RFC3339Nano), *info.ApprovedAt) +} + +// TestBuildCLJobInfo_UnmanagedJobHasNoProvenance: an unset feeds_manager_id is +// how a consumer tells a directly-created job from a JD-managed one. +func TestBuildCLJobInfo_UnmanagedJobHasNoProvenance(t *testing.T) { + info, err := jobspec.BuildCLJobInfo(clJobInfoSampleJob(), commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, jobspec.NodeIdentity{}, nil, time.Now()) + require.NoError(t, err) + + require.Nil(t, info.FeedsManagerId) + require.Nil(t, info.RemoteUuid) + require.Nil(t, info.SpecVersion) + require.Nil(t, info.ProposedAt) + require.Nil(t, info.ApprovedAt) +} + +func TestEmitCLJobInfo_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()) + info, err := jobspec.BuildCLJobInfo(clJobInfoSampleJob(), commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, jobspec.NodeIdentity{CSAPublicKey: "csa"}, nil, time.Now()) require.NoError(t, err) - require.NoError(t, cljobinfo.Emit(t.Context(), beholder.GetEmitter(), info)) + require.NoError(t, jobspec.EmitCLJobInfo(t.Context(), beholder.GetEmitter(), info)) - msgs := obs.Messages(t, beholder.AttrKeyEntity, cljobinfo.Entity) + msgs := obs.Messages(t, beholder.AttrKeyEntity, jobspec.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]) + require.Equal(t, jobspec.Domain, msg.Attrs[beholder.AttrKeyDomain]) + require.Equal(t, jobspec.DataSchema, msg.Attrs[beholder.AttrKeyDataSchema]) var payload commonv1.CLJobInfo require.NoError(t, proto.Unmarshal(msg.Body, &payload)) diff --git a/core/services/nodestatusreporter/jobspec/job_spec_reporter.go b/core/services/nodestatusreporter/jobspec/job_spec_reporter.go index 9ee6b91b8de..ea3ed5969a3 100644 --- a/core/services/nodestatusreporter/jobspec/job_spec_reporter.go +++ b/core/services/nodestatusreporter/jobspec/job_spec_reporter.go @@ -14,6 +14,7 @@ import ( "github.com/smartcontractkit/chainlink-common/pkg/beholder" "github.com/smartcontractkit/chainlink-common/pkg/services" commontypes "github.com/smartcontractkit/chainlink-common/pkg/types" + commonv1 "github.com/smartcontractkit/chainlink-protos/node-platform/common/v1" coreconfig "github.com/smartcontractkit/chainlink/v2/core/config" "github.com/smartcontractkit/chainlink/v2/core/logger" @@ -29,6 +30,17 @@ var _ job.Listener = (*Service)(nil) // Service polls active jobs and pushes their specs to Beholder, and also emits // on job create/delete via the job.Listener interface. +// +// It emits two payloads on every trigger: +// +// - CLJobInfo, for every job the node runs regardless of type, carrying the +// complete definition as TOML (see cl_job_info.go). +// - JobSpecEvent, the original OCR2-only projection, for jobs passing the +// EnabledOCR2PluginTypes gate. +// +// The second is superseded by the first and is retained only until its +// consumers migrate; once they have, ShouldEmit, EmitForJob and the events +// package can be deleted from here without touching the service wiring. type Service struct { services.Service eng *services.Engine @@ -88,34 +100,92 @@ func (s *Service) HealthReport() map[string]error { // AfterJobStarted emits a create event when a job starts. func (s *Service) AfterJobStarted(ctx context.Context, jb job.Job) { - if !s.ShouldEmit(&jb) { - return - } - if err := s.EmitForJob(ctx, jb, events.EmissionTrigger_EMISSION_TRIGGER_CREATE); err != nil { - s.eng.Warnw("Failed to emit job spec telemetry on create", "jobID", jb.ID, "error", err) - } + s.emit(ctx, jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_CREATE, events.EmissionTrigger_EMISSION_TRIGGER_CREATE) } // AfterJobStopped emits a delete event when a job is removed. func (s *Service) AfterJobStopped(ctx context.Context, jb job.Job) { + s.emit(ctx, jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_DELETE, events.EmissionTrigger_EMISSION_TRIGGER_DELETE) +} + +// pollAllJobs emits heartbeat telemetry for every active job. +func (s *Service) pollAllJobs(ctx context.Context) { + for _, jb := range s.spawner.ActiveJobs() { + s.emit(ctx, jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_HEARTBEAT, events.EmissionTrigger_EMISSION_TRIGGER_HEARTBEAT) + } +} + +// emit reports jb on both tracks: CLJobInfo unconditionally, and the legacy +// OCR2 JobSpecEvent only for jobs passing the plugin-type gate. A failure on +// one track never suppresses the other. +func (s *Service) emit(ctx context.Context, jb job.Job, clTrigger commonv1.CLJobInfoTrigger, trigger events.EmissionTrigger) { + if err := s.EmitCLJobInfoForJob(ctx, jb, clTrigger); err != nil { + s.eng.Warnw("Failed to emit CLJobInfo", "jobID", jb.ID, "trigger", clTrigger, "error", err) + } + if !s.ShouldEmit(&jb) { return } - if err := s.EmitForJob(ctx, jb, events.EmissionTrigger_EMISSION_TRIGGER_DELETE); err != nil { - s.eng.Warnw("Failed to emit job spec telemetry on delete", "jobID", jb.ID, "error", err) + if err := s.EmitForJob(ctx, jb, trigger); err != nil { + s.eng.Warnw("Failed to emit job spec telemetry", "jobID", jb.ID, "trigger", trigger, "error", err) } } -// pollAllJobs emits heartbeat telemetry for every active job that passes the emit gate. -func (s *Service) pollAllJobs(ctx context.Context) { - for _, jb := range s.spawner.ActiveJobs() { - if !s.ShouldEmit(&jb) { - continue +// EmitCLJobInfoForJob builds and emits the generic CLJobInfo for any job type. +// +// A job whose spec cannot be TOML-encoded is still reported: the identity +// envelope is emitted without spec_toml so the job is accounted for, and the +// encoding failure is returned for logging rather than dropping the event. +func (s *Service) EmitCLJobInfoForJob(ctx context.Context, jb job.Job, trigger commonv1.CLJobInfoTrigger) error { + prop, err := s.jobProposal(ctx, jb) + if err != nil { + // Provenance is an enrichment, not a precondition: a job with no + // proposal is a valid, unmanaged job. + s.eng.Warnw("Failed to resolve job proposal provenance for CLJobInfo", + "jobID", jb.ID, "externalJobID", jb.ExternalJobID, "error", err) + } + + identity := NodeIdentity{CSAPublicKey: s.csaPublicKey, NodeVersion: s.nodeVersion, Hostname: s.hostname} + info, buildErr := BuildCLJobInfo(jb, trigger, identity, prop, time.Now()) + + if emitErr := EmitCLJobInfo(ctx, s.emitter, info); emitErr != nil { + return emitErr + } + return buildErr +} + +// jobProposal resolves the Job Distributor provenance for jb, or nil if the job +// did not arrive as an approved job proposal. +func (s *Service) jobProposal(ctx context.Context, jb job.Job) (*JobProposal, error) { + if s.feedsORM == nil || jb.ExternalJobID == uuid.Nil { + return nil, nil + } + + prop, err := s.feedsORM.GetJobProposalByExternalJobID(ctx, jb.ExternalJobID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil } - if err := s.EmitForJob(ctx, jb, events.EmissionTrigger_EMISSION_TRIGGER_HEARTBEAT); err != nil { - s.eng.Warnw("Failed to emit job spec telemetry", "jobID", jb.ID, "error", err) + return nil, fmt.Errorf("fetching job proposal: %w", err) + } + + spec, err := s.feedsORM.GetApprovedSpec(ctx, prop.ID) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + // The proposal exists but has no approved spec, e.g. we are + // reporting a job that is mid-cancellation. + return &JobProposal{FeedsManagerID: prop.FeedsManagerID, RemoteUUID: prop.RemoteUUID.String()}, nil } + return nil, fmt.Errorf("fetching approved spec: %w", err) } + + return &JobProposal{ + FeedsManagerID: prop.FeedsManagerID, + RemoteUUID: prop.RemoteUUID.String(), + SpecVersion: spec.Version, + ProposedAt: spec.CreatedAt, + ApprovedAt: spec.StatusUpdatedAt, + }, nil } // ShouldEmit reports whether the job passes the config-driven emit gate. From 4e0d6fd33cc82c009fcc72d7fd93fba9a2043abb Mon Sep 17 00:00:00 2001 From: gheorghestrimtu Date: Fri, 4 Sep 2026 17:49:13 +0300 Subject: [PATCH 3/3] refactor(nodestatusreporter): emit CLJobInfo times as protobuf Timestamps --- .../nodestatusreporter/jobspec/cl_job_info.go | 21 ++++++--- .../jobspec/cl_job_info_test.go | 46 +++++++++++++++++-- 2 files changed, 57 insertions(+), 10 deletions(-) diff --git a/core/services/nodestatusreporter/jobspec/cl_job_info.go b/core/services/nodestatusreporter/jobspec/cl_job_info.go index f5852cc36a0..d6d158aafc8 100644 --- a/core/services/nodestatusreporter/jobspec/cl_job_info.go +++ b/core/services/nodestatusreporter/jobspec/cl_job_info.go @@ -7,6 +7,7 @@ import ( "github.com/pelletier/go-toml" "google.golang.org/protobuf/proto" + "google.golang.org/protobuf/types/known/timestamppb" "github.com/smartcontractkit/chainlink-common/pkg/beholder" commonv1 "github.com/smartcontractkit/chainlink-protos/node-platform/common/v1" @@ -69,9 +70,9 @@ func BuildCLJobInfo(jb job.Job, trigger commonv1.CLJobInfoTrigger, id NodeIdenti JobType: string(jb.Type), SchemaVersion: jb.SchemaVersion, ForwardingAllowed: jb.ForwardingAllowed, - CreatedAt: formatCLJobInfoTime(jb.CreatedAt), + CreatedAt: timestampOrNil(jb.CreatedAt), Trigger: trigger, - Timestamp: now.UTC().Format(time.RFC3339Nano), + Timestamp: timestamppb.New(now), } if jb.GasLimit.Valid { info.GasLimit = new(jb.GasLimit.Uint32) @@ -83,8 +84,8 @@ func BuildCLJobInfo(jb job.Job, trigger commonv1.CLJobInfoTrigger, id NodeIdenti info.FeedsManagerId = &prop.FeedsManagerID info.RemoteUuid = &prop.RemoteUUID info.SpecVersion = &prop.SpecVersion - info.ProposedAt = new(formatCLJobInfoTime(prop.ProposedAt)) - info.ApprovedAt = new(formatCLJobInfoTime(prop.ApprovedAt)) + info.ProposedAt = timestampOrNil(prop.ProposedAt) + info.ApprovedAt = timestampOrNil(prop.ApprovedAt) } specTOML, err := jobTOML(jb) @@ -126,9 +127,15 @@ func jobTOML(jb job.Job) (string, error) { return string(out), nil } -func formatCLJobInfoTime(t time.Time) string { +// timestampOrNil converts t to a protobuf Timestamp, leaving an unset time as +// nil rather than mapping it onto the epoch. google.protobuf.Timestamp is used +// throughout the Job Distributor protos and, unlike an RFC3339Nano string, +// orders correctly for consumers: Go trims trailing zeros from the fractional +// seconds, so those strings are variable-width and do not sort lexicographically +// in chronological order. +func timestampOrNil(t time.Time) *timestamppb.Timestamp { if t.IsZero() { - return "" + return nil } - return t.UTC().Format(time.RFC3339Nano) + return timestamppb.New(t) } diff --git a/core/services/nodestatusreporter/jobspec/cl_job_info_test.go b/core/services/nodestatusreporter/jobspec/cl_job_info_test.go index 0b326c7e1c1..79fe68592e8 100644 --- a/core/services/nodestatusreporter/jobspec/cl_job_info_test.go +++ b/core/services/nodestatusreporter/jobspec/cl_job_info_test.go @@ -68,7 +68,10 @@ func TestBuildCLJobInfo_EncodesFullSpecAsTOML(t *testing.T) { 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) + require.NotNil(t, info.Timestamp) + require.Equal(t, time.Date(2026, 7, 24, 12, 0, 0, 0, time.UTC), info.Timestamp.AsTime()) + require.NotNil(t, info.CreatedAt) + require.Equal(t, time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC), info.CreatedAt.AsTime()) // spec_toml must be valid TOML and contain type-specific spec data. require.NotEmpty(t, info.SpecToml) @@ -116,9 +119,9 @@ func TestBuildCLJobInfo_CarriesJobDistributorProvenance(t *testing.T) { require.NotNil(t, info.SpecVersion) require.Equal(t, int32(2), *info.SpecVersion) require.NotNil(t, info.ProposedAt) - require.Equal(t, proposedAt.Format(time.RFC3339Nano), *info.ProposedAt) + require.Equal(t, proposedAt, info.ProposedAt.AsTime()) require.NotNil(t, info.ApprovedAt) - require.Equal(t, approvedAt.Format(time.RFC3339Nano), *info.ApprovedAt) + require.Equal(t, approvedAt, info.ApprovedAt.AsTime()) } // TestBuildCLJobInfo_UnmanagedJobHasNoProvenance: an unset feeds_manager_id is @@ -154,3 +157,40 @@ func TestEmitCLJobInfo_PublishesToBeholder(t *testing.T) { require.Equal(t, "offchainreporting2", payload.JobType) require.NotEmpty(t, payload.SpecToml) } + +// TestBuildCLJobInfo_TimestampsRoundTripExactly guards the reason these fields +// are google.protobuf.Timestamp rather than RFC3339Nano strings: Go trims +// trailing zeros from the fractional seconds, so string-encoded times are +// variable-width and do not sort lexicographically in chronological order — a +// whole-second time sorts after every sub-second one in the same second. +func TestBuildCLJobInfo_TimestampsRoundTripExactly(t *testing.T) { + for _, tc := range []struct { + name string + at time.Time + }{ + {"whole second", time.Date(2026, 7, 24, 10, 0, 0, 0, time.UTC)}, + {"tenth of a second", time.Date(2026, 7, 24, 10, 0, 0, 100000000, time.UTC)}, + {"sub-millisecond", time.Date(2026, 7, 24, 10, 0, 0, 123400000, time.UTC)}, + {"nanosecond", time.Date(2026, 7, 24, 10, 0, 0, 123456789, time.UTC)}, + } { + t.Run(tc.name, func(t *testing.T) { + jb := clJobInfoSampleJob() + jb.CreatedAt = tc.at + + info, err := jobspec.BuildCLJobInfo(jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_HEARTBEAT, jobspec.NodeIdentity{}, nil, time.Now()) + require.NoError(t, err) + require.NotNil(t, info.CreatedAt) + require.Equal(t, tc.at, info.CreatedAt.AsTime()) + }) + } +} + +// TestBuildCLJobInfo_ZeroTimeIsUnset: an absent time must be nil, not the epoch. +func TestBuildCLJobInfo_ZeroTimeIsUnset(t *testing.T) { + jb := clJobInfoSampleJob() + jb.CreatedAt = time.Time{} + + info, err := jobspec.BuildCLJobInfo(jb, commonv1.CLJobInfoTrigger_CL_JOB_INFO_TRIGGER_HEARTBEAT, jobspec.NodeIdentity{}, nil, time.Now()) + require.NoError(t, err) + require.Nil(t, info.CreatedAt) +}