Attach required type label to all custmsg emitters

core/capabilities/compute/compute.go

@@ -30,6 +30,7 @@ import (
 	wasmpb "github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/pb"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/validation"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/webapi"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink/v2/core/platform"
 	ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
 )
@@ -453,7 +454,7 @@ func NewAction(
 
 	var (
 		lggr    = logger.Named(log, "CustomCompute")
-		labeler = custmsg.NewLabeler()
+		labeler = custmsg.NewLabeler().WithType(custmsgtypes.TypeCompute)
 		compute = &Compute{
 			stopCh:               make(services.StopChan),
 			log:                  lggr,

core/capabilities/compute/transformer_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/pkg/capabilities"
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink-common/pkg/logger"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows/wasm/host"
 	"github.com/smartcontractkit/chainlink-protos/cre/go/values"
@@ -89,7 +90,7 @@ func Test_transformer(t *testing.T) {
 	t.Parallel()
 	var (
 		lgger   = logger.Test(t)
-		emitter = custmsg.NewLabeler()
+		emitter = custmsg.NewLabeler().WithType(custmsgtypes.TypeCompute)
 	)
 	t.Run("success", func(t *testing.T) {
 		t.Parallel()

core/cmd/shell_local.go

@@ -47,6 +47,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/pg"
 	"github.com/smartcontractkit/chainlink/v2/core/sessions"
 	"github.com/smartcontractkit/chainlink/v2/core/shutdown"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink/v2/core/static"
 	"github.com/smartcontractkit/chainlink/v2/core/store"
 	"github.com/smartcontractkit/chainlink/v2/core/store/migrate"
@@ -335,13 +336,12 @@ const ownerPermsMask = os.FileMode(0o700)
 
 // EmitNodeConfig emits the node configuration through beholder as a pb.BaseMessage
 func (s *Shell) EmitNodeConfig(ctx context.Context) {
-	cme := custmsg.NewLabeler()
 	labels := map[string]string{
 		"system":  "Application",
 		"version": static.Version,
 		"commit":  static.Sha,
 	}
-	emitter := cme.WithMapLabels(labels)
+	emitter := custmsg.NewLabeler().WithLabelsAndType(labels, custmsgtypes.TypeNodeConfig)
 
 	// Get the effective TOML configuration (with defaults applied)
 	_, effectiveTOML := s.Config.ConfigTOML()

core/cmd/shell_test.go

@@ -622,4 +622,5 @@ func TestShell_emitNodeConfig(t *testing.T) {
 	require.Equal(t, "Application", baseMsg.Labels["system"])
 	require.Equal(t, static.Version, baseMsg.Labels["version"])
 	require.Equal(t, static.Sha, baseMsg.Labels["commit"])
+	require.Equal(t, "NodeConfig", baseMsg.Labels["type"])
 }

core/custmsgtypes/types.go

@@ -0,0 +1,11 @@
+// Package custmsgtypes defines stable type label values for custmsg BaseMessage emissions.
+package custmsgtypes
+
+const (
+	TypeNodeConfig         = "NodeConfig"
+	TypeHeartbeat          = "Heartbeat"
+	TypeWorkflowEngine     = "WorkflowEngine"
+	TypeWorkflowSyncer     = "WorkflowSyncer"
+	TypeWorkflowArtifact   = "WorkflowArtifact"
+	TypeCompute            = "Compute"
+)

core/services/chainlink/heartbeat.go

@@ -10,6 +10,7 @@ import (
 	commonservices "github.com/smartcontractkit/chainlink-common/pkg/services"
 	"github.com/smartcontractkit/chainlink-common/pkg/timeutil"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink/v2/core/static"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/beholder"
@@ -59,7 +60,6 @@ func NewHeartbeatConfig(cfg ApplicationOpts) HeartbeatConfig {
 // Update the constructor to accept optional emitter and meter
 func NewHeartbeat(cfg HeartbeatConfig, opts ...HeartbeatOpt) Heartbeat {
 	// setup default emitter and meter
-	cme := custmsg.NewLabeler()
 	labels := map[string]string{"system": "Application", "version": static.Version, "commit": static.Sha}
 	if cfg.P2P != "" {
 		labels["peer_id"] = cfg.P2P
@@ -71,11 +71,10 @@ func NewHeartbeat(cfg HeartbeatConfig, opts ...HeartbeatOpt) Heartbeat {
 		labels["csa_key"] = cfg.CSAPublicKey
 	}
 
-	cme.WithMapLabels(labels)
 	h := Heartbeat{
 		beat:    cfg.Beat,
 		opts:    cfg,
-		emitter: cme.WithMapLabels(labels),
+		emitter: custmsg.NewLabeler().WithLabelsAndType(labels, custmsgtypes.TypeHeartbeat),
 		meter:   beholder.GetMeter(),
 	}
 

core/services/cre/cre.go

@@ -35,6 +35,7 @@ import (
 	linkingclient "github.com/smartcontractkit/chainlink-protos/linking-service/go/v1"
 
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/compute"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/confidentialrelay"
 	gatewayconnector "github.com/smartcontractkit/chainlink/v2/core/capabilities/gateway_connector"
@@ -804,7 +805,7 @@ func newWorkflowRegistrySyncerV1(
 		fetcherFunc,
 		clockwork.NewRealClock(),
 		key,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 		artifactsV1.WithMaxArtifactSize(
 			artifactsV1.ArtifactConfig{
 				MaxBinarySize:  uint64(capCfg.WorkflowRegistry().MaxBinarySize()),
@@ -838,7 +839,7 @@ func newWorkflowRegistrySyncerV1(
 		dontimeStore,
 		opts.UseLocalTimeProvider,
 		engineRegistry,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowSyncer),
 		engineLimiters,
 		featureFlags,
 		workflowRateLimiter,
@@ -950,7 +951,7 @@ func newWorkflowRegistrySyncerV2(
 		retrieverFunc,
 		clockwork.NewRealClock(),
 		key,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 		lf,
 		artifactsV2.WithConfig(artifactsV2.StoreConfig{
 			ArtifactStorageHost: wfReg.WorkflowStorage().ArtifactStorageHost(),
@@ -1083,7 +1084,7 @@ func newWorkflowRegistrySyncerV2(
 		opts.CapabilitiesRegistry,
 		opts.ExecutionHandlers,
 		engineRegistry,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowSyncer),
 		engineLimiters,
 		featureFlags,
 		workflowRateLimiter,

core/services/workflows/artifacts/store_test.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink-common/keystore/corekeys/workflowkey"
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink-common/pkg/workflows/secrets"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 	ghcapabilities "github.com/smartcontractkit/chainlink/v2/core/services/gateway/handlers/capabilities"
@@ -72,7 +73,7 @@ func Test_Handler_SecretsFor(t *testing.T) {
 		fetcher.Fetch,
 		clockwork.NewFakeClock(),
 		encryptionKey,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 	)
 	expectedSecrets := map[string]string{
 		"Foo": "Bar",
@@ -136,7 +137,7 @@ func Test_Handler_SecretsFor_RefreshesSecrets(t *testing.T) {
 		fetcher.Fetch,
 		clockwork.NewFakeClock(),
 		encryptionKey,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 	)
 
 	expectedSecrets := map[string]string{
@@ -203,7 +204,7 @@ func Test_Handler_SecretsFor_RefreshLogic(t *testing.T) {
 		fetcher.Fetch,
 		clock,
 		encryptionKey,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 	)
 
 	expectedSecrets := map[string]string{

core/services/workflows/artifacts/v2/store_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/smartcontractkit/chainlink-common/keystore/corekeys/workflowkey"
 	"github.com/smartcontractkit/chainlink-common/pkg/contexts"
 	"github.com/smartcontractkit/chainlink-common/pkg/custmsg"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink-common/pkg/settings/limits"
 	storage_service "github.com/smartcontractkit/chainlink-protos/storage-service/go"
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
@@ -78,7 +79,7 @@ func Test_Store_DeleteWorkflowArtifacts(t *testing.T) {
 		fetcher.RetrieveURL,
 		clockwork.NewFakeClock(),
 		encryptionKey,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 		limits.Factory{Logger: lggr},
 		WithConfig(StoreConfig{
 			ArtifactStorageHost: "example.com",
@@ -127,7 +128,7 @@ func Test_Store_DeleteWorkflowArtifactsBatch(t *testing.T) {
 	fetcher := &mockFetcher{}
 	h, err := NewStore(
 		lggr, orm, fetcher.Fetch, fetcher.RetrieveURL,
-		clockwork.NewFakeClock(), encryptionKey, custmsg.NewLabeler(),
+		clockwork.NewFakeClock(), encryptionKey, custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 		limits.Factory{Logger: lggr},
 		WithConfig(StoreConfig{ArtifactStorageHost: "example.com"}),
 	)
@@ -178,7 +179,7 @@ func Test_Store_FetchWorkflowArtifacts_WithStorage(t *testing.T) {
 		fetcher.RetrieveURL,
 		clockwork.NewFakeClock(),
 		encryptionKey,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 		limits.Factory{Logger: lggr},
 		WithConfig(StoreConfig{
 			ArtifactStorageHost: "storage.chain.link",
@@ -222,7 +223,7 @@ func Test_Store_FetchWorkflowArtifacts_WithoutStorage(t *testing.T) {
 		fetcher.RetrieveURL,
 		clockwork.NewFakeClock(),
 		encryptionKey,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 		limits.Factory{Logger: lggr},
 		WithConfig(StoreConfig{
 			ArtifactStorageHost: "storage.chain.link",
@@ -266,7 +267,7 @@ func Test_Store_FetchWorkflowArtifacts_SkipsRetrieving(t *testing.T) {
 		nil, // No retrieval function provided, so it should skip retrieving
 		clockwork.NewFakeClock(),
 		encryptionKey,
-		custmsg.NewLabeler(),
+		custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowArtifact),
 		limits.Factory{Logger: lggr},
 		WithConfig(StoreConfig{
 			ArtifactStorageHost: "example.com",

core/services/workflows/cmd/cre/utils/standalone_engine.go

@@ -66,7 +66,7 @@
 	workflowSettingsCfgFn func(*cresettings.Workflows),
 ) (services.Service, []*sdkpb.TriggerSubscription, error) {
 	ctx = contexts.WithCRE(ctx, contexts.CRE{Owner: defaultOwner, Workflow: defaultWorkflowID})
-	labeler := custmsg.NewLabeler()
+	labeler := custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowEngine)
 	moduleConfig := &host.ModuleConfig{
 		Logger:                  lggr,
 		Labeler:                 labeler,
@@ -198,7 +198,7 @@
 		FeatureFlags:        featureFlags,
 		GlobalWorkflowLimit: workflowLimits,
 
-		BeholderEmitter: custmsg.NewLabeler(),
+		BeholderEmitter: custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowEngine),
 
 		BillingClient: billingClient,
 		Hooks:         lifecycleHooks,

core/services/workflows/delegate.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/smartcontractkit/chainlink/v2/core/logger"
 	"github.com/smartcontractkit/chainlink/v2/core/platform"
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink/v2/core/services/job"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/metering"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/store"
@@ -65,7 +66,7 @@ func (d *Delegate) OnDeleteJob(context.Context, job.Job) error { return nil }
 
 // ServicesForSpec satisfies the job.Delegate interface.
 func (d *Delegate) ServicesForSpec(ctx context.Context, spec job.Job) ([]job.ServiceCtx, error) {
-	cma := custmsg.NewLabeler().With(platform.KeyWorkflowID, spec.WorkflowSpec.WorkflowID, platform.KeyWorkflowOwner, spec.WorkflowSpec.WorkflowOwner, platform.KeyWorkflowName, spec.WorkflowSpec.WorkflowName)
+	cma := custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowEngine).With(platform.KeyWorkflowID, spec.WorkflowSpec.WorkflowID, platform.KeyWorkflowOwner, spec.WorkflowSpec.WorkflowOwner, platform.KeyWorkflowName, spec.WorkflowSpec.WorkflowName)
 	sdkSpec, err := spec.WorkflowSpec.SDKSpec(ctx)
 	if err != nil {
 		logCustMsg(ctx, cma, fmt.Sprintf("failed to start workflow engine: failed to get workflow sdk spec: %v", err), d.logger)

core/services/workflows/engine.go

@@ -30,6 +30,7 @@ import (
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/ratelimiter"
 	"github.com/smartcontractkit/chainlink/v2/core/services/workflows/shardownership"
 
+	"github.com/smartcontractkit/chainlink/v2/core/custmsgtypes"
 	"github.com/smartcontractkit/chainlink/v2/core/capabilities/transmission"
 	"github.com/smartcontractkit/chainlink/v2/core/platform"
 	"github.com/smartcontractkit/chainlink/v2/core/services/shardorchestrator"
@@ -1489,7 +1490,7 @@ func NewEngine(ctx context.Context, cfg Config) (engine *Engine, err error) {
 		cfg.WorkflowRegistryAddress = "0xv1EngineDefault"
 	}
 
-	cma := custmsg.NewLabeler().With(platform.KeyWorkflowID, cfg.WorkflowID,
+	cma := custmsg.NewLabeler().WithType(custmsgtypes.TypeWorkflowEngine).With(platform.KeyWorkflowID, cfg.WorkflowID,
 		platform.KeyWorkflowOwner, cfg.WorkflowOwner,
 		platform.KeyWorkflowName, cfg.WorkflowName.String(),
 		platform.KeyWorkflowVersion, platform.ValueWorkflowVersion,