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) +}