build/devenv: observability component

build/devenv/components/committeeccv/component.go

@@ -12,6 +12,7 @@ import (
 	"github.com/smartcontractkit/chainlink-ccv/build/devenv/chainreg"
 	devenvcommon "github.com/smartcontractkit/chainlink-ccv/build/devenv/common"
 	blockchainscomp "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/blockchains"
+	"github.com/smartcontractkit/chainlink-ccv/build/devenv/components/observability"
 	ccdeploy "github.com/smartcontractkit/chainlink-ccv/build/devenv/deploy"
 	"github.com/smartcontractkit/chainlink-ccv/build/devenv/jobs"
 	devenvruntime "github.com/smartcontractkit/chainlink-ccv/build/devenv/runtime"
@@ -96,6 +97,10 @@ func (c *component) RunPhase3(
 	if !ok || topology == nil {
 		return nil, nil, fmt.Errorf("committeeccv: environment_topology not found in phase outputs")
 	}
+	obs, ok := priorOutputs["observability"].(*observability.Observability)
+	if !ok || obs == nil {
+		return nil, nil, fmt.Errorf("committeeccv: observability not found in phase outputs")
+	}
 	ds, ok := priorOutputs["_ds"].(datastore.MutableDataStore)
 	if !ok {
 		return nil, nil, fmt.Errorf("committeeccv: _ds not found in phase outputs")
@@ -220,7 +225,7 @@ func (c *component) RunPhase3(
 	}
 
 	// Step 8: Generate verifier job specs and emit job proposal effects.
-	effects, err := buildVerifierJobSpecEffects(e, verifiers, topology, sharedTLSCerts, blockchainOutputs, ds)
+	effects, err := buildVerifierJobSpecEffects(e, verifiers, topology, obs, sharedTLSCerts, blockchainOutputs, ds)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -307,6 +312,7 @@ func buildVerifierJobSpecEffects(
 	e *deployment.Environment,
 	verifiers []*committeeverifier.Input,
 	topology *ccvdeployment.EnvironmentTopology,
+	obs *observability.Observability,
 	sharedTLSCerts *services.TLSCertPaths,
 	blockchainOutputs []*ctfblockchain.Output,
 	ds datastore.MutableDataStore,
@@ -350,8 +356,8 @@ func buildVerifierJobSpecEffects(
 				DefaultExecutorQualifier: devenvcommon.DefaultExecutorQualifier,
 				NOPs:                     ccvchangesets.NOPInputsFromTopology(topology),
 				Committee:                ccvchangesets.CommitteeInputFromTopologyPerFamily(committee, family),
-				PyroscopeURL:             topology.PyroscopeURL,
-				Monitoring:               topology.Monitoring,
+				PyroscopeURL:             obs.PyroscopeURL,
+				Monitoring:               obs.Monitoring,
 				TargetNOPs:               verNOPAliases,
 				DisableFinalityCheckers:  disableFinalityCheckersPerFamily[family],
 			})

build/devenv/components/executor/component.go

@@ -15,6 +15,7 @@ import (
 	"github.com/smartcontractkit/chainlink-ccv/build/devenv/chainreg"
 	devenvcommon "github.com/smartcontractkit/chainlink-ccv/build/devenv/common"
 	blockchainscomp "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/blockchains"
+	"github.com/smartcontractkit/chainlink-ccv/build/devenv/components/observability"
 	"github.com/smartcontractkit/chainlink-ccv/build/devenv/jobs"
 	devenvruntime "github.com/smartcontractkit/chainlink-ccv/build/devenv/runtime"
 	"github.com/smartcontractkit/chainlink-ccv/build/devenv/services"
@@ -144,12 +145,16 @@ func (c *component) RunPhase3(
 	if !ok || topology == nil {
 		return nil, nil, fmt.Errorf("executor: environment_topology not found in phase outputs")
 	}
+	obs, ok := priorOutputs["observability"].(*observability.Observability)
+	if !ok || obs == nil {
+		return nil, nil, fmt.Errorf("executor: observability not found in phase outputs")
+	}
 	ds, ok := priorOutputs["_ds"].(datastore.MutableDataStore)
 	if !ok {
 		return nil, nil, fmt.Errorf("executor: _ds not found in phase outputs")
 	}
 
-	jobSpecs, err := buildExecutorJobSpecs(e, executors, topology, ds)
+	jobSpecs, err := buildExecutorJobSpecs(e, executors, topology, obs, ds)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -252,6 +257,7 @@ func buildExecutorJobSpecs(
 	e *deployment.Environment,
 	executors []*executorsvc.Input,
 	topology *ccvdeployment.EnvironmentTopology,
+	obs *observability.Observability,
 	ds datastore.MutableDataStore,
 ) (map[string]bootstrap.JobSpec, error) {
 	result := make(map[string]bootstrap.JobSpec)
@@ -283,8 +289,8 @@ func buildExecutorJobSpecs(
 			NOPs:              ccvchangesets.NOPInputsFromTopology(topology),
 			Pool:              ccvchangesets.ExecutorPoolInputFromTopology(pool),
 			IndexerAddress:    topology.IndexerAddress,
-			PyroscopeURL:      topology.PyroscopeURL,
-			Monitoring:        topology.Monitoring,
+			PyroscopeURL:      obs.PyroscopeURL,
+			Monitoring:        obs.Monitoring,
 			TargetNOPs:        ccvshared.ConvertStringToNopAliases(execNOPAliases),
 		})
 		if err != nil {

build/devenv/components/observability/component.go

@@ -0,0 +1,97 @@
+package observability
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/pelletier/go-toml/v2"
+
+	devenvruntime "github.com/smartcontractkit/chainlink-ccv/build/devenv/runtime"
+	ccvdeployment "github.com/smartcontractkit/chainlink-ccv/deployment"
+)
+
+const configKey = "observability"
+
+// Version is the observability component config schema version. Exactly this
+// version is supported; configs declaring any other version are rejected.
+const Version = 1
+
+// Observability holds cross-cutting observability settings — the beholder
+// monitoring config and the pyroscope profiling URL. These are consumed by both
+// verifier and executor configs but are not part of the topology graph. The
+// observability component publishes this as its phase output for the committeeccv
+// and executor components to consume. The toml tags keep the serialized phased
+// output stable.
+type Observability struct {
+	PyroscopeURL string                         `toml:"pyroscope_url"`
+	Monitoring   ccvdeployment.MonitoringConfig `toml:"monitoring"`
+}
+
+func init() {
+	if err := devenvruntime.Register(configKey, factory); err != nil {
+		panic(fmt.Sprintf("observability component: %v", err))
+	}
+}
+
+func factory(_ map[string]any) (devenvruntime.Component, error) {
+	return &component{}, nil
+}
+
+type component struct{}
+
+func (c *component) ValidateConfig(componentConfig any) error {
+	_, err := decode(componentConfig)
+	return err
+}
+
+// RunPhase1 publishes the cross-cutting observability settings (beholder
+// monitoring + pyroscope URL) as a phase output for later phases to consume. It
+// has no dependencies on other components and runs in Phase 1 alongside
+// blockchains and jd.
+//
+// Output:
+//   - "observability" — *Observability, read by the committeeccv and executor
+//     components when generating verifier/executor configs.
+func (c *component) RunPhase1(
+	_ context.Context,
+	_ map[string]any,
+	componentConfig any,
+) (map[string]any, []devenvruntime.Effect, error) {
+	obs, err := decode(componentConfig)
+	if err != nil {
+		return nil, nil, err
+	}
+	return map[string]any{configKey: obs}, nil, nil
+}
+
+// config is the [observability] component config. Version is the component
+// schema version; the remaining fields populate the published Observability.
+type config struct {
+	Version      int                            `toml:"version"`
+	PyroscopeURL string                         `toml:"pyroscope_url"`
+	Monitoring   ccvdeployment.MonitoringConfig `toml:"monitoring"`
+}
+
+// decode round-trips the raw TOML map[string]any into *Observability, verifying
+// the declared component version.
+func decode(raw any) (*Observability, error) {
+	b, err := toml.Marshal(struct {
+		V any `toml:"observability"`
+	}{V: raw})
+	if err != nil {
+		return nil, fmt.Errorf("re-encoding observability config: %w", err)
+	}
+	var wrapper struct {
+		V config `toml:"observability"`
+	}
+	if err := toml.Unmarshal(b, &wrapper); err != nil {
+		return nil, fmt.Errorf("decoding observability config: %w", err)
+	}
+	if err := devenvruntime.CheckConfigVersion(wrapper.V.Version, Version); err != nil {
+		return nil, err
+	}
+	return &Observability{
+		PyroscopeURL: wrapper.V.PyroscopeURL,
+		Monitoring:   wrapper.V.Monitoring,
+	}, nil
+}

build/devenv/components/observability/component_test.go

@@ -0,0 +1,49 @@
+package observability
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func validConfig() map[string]any {
+	return map[string]any{
+		"version":       int64(1),
+		"pyroscope_url": "http://host.docker.internal:4040",
+		"monitoring": map[string]any{
+			"Enabled": true,
+			"Type":    "beholder",
+			"Beholder": map[string]any{
+				"InsecureConnection":       true,
+				"OtelExporterHTTPEndpoint": "host.docker.internal:4318",
+			},
+		},
+	}
+}
+
+func TestValidateConfig_Valid(t *testing.T) {
+	c := &component{}
+	require.NoError(t, c.ValidateConfig(validConfig()))
+}
+
+func TestValidateConfig_RejectsWrongVersion(t *testing.T) {
+	cfg := validConfig()
+	cfg["version"] = int64(2)
+	c := &component{}
+	err := c.ValidateConfig(cfg)
+	require.Error(t, err)
+}
+
+func TestRunPhase1_PublishesObservability(t *testing.T) {
+	c := &component{}
+	out, effects, err := c.RunPhase1(context.Background(), nil, validConfig())
+	require.NoError(t, err)
+	require.Nil(t, effects)
+
+	obs, ok := out[configKey].(*Observability)
+	require.True(t, ok, "output %q should be *Observability", configKey)
+	require.Equal(t, "http://host.docker.internal:4040", obs.PyroscopeURL)
+	require.True(t, obs.Monitoring.Enabled)
+	require.Equal(t, "host.docker.internal:4318", obs.Monitoring.Beholder.OtelExporterHTTPEndpoint)
+}

build/devenv/env-phased.toml

@@ -7,27 +7,35 @@
 # below in sync with env.toml until the two configs are formally split.
 version = 1
 
-[protocol_contracts]
+## Cross-cutting observability settings (beholder monitoring + pyroscope),
+## owned by the phased observability component (Phase 1) and consumed by the
+## committeeccv and executor components. These are intentionally NOT part of the
+## topology; the legacy monolith still carries the equivalent keys under
+## [environment_topology] in env.toml.
+[observability]
 version = 1
-use_legacy_configure_lane = false
-
-## Environment configuration define the topology. Eventually the topology should be the only config that is needed.
-[protocol_contracts.environment_topology]
-indexer_address = ["http://indexer-1:8100"]
 pyroscope_url = "http://host.docker.internal:4040"
 
-[protocol_contracts.environment_topology.monitoring]
+[observability.monitoring]
 Enabled = true
 Type = "beholder"
 
-[protocol_contracts.environment_topology.monitoring.Beholder]
+[observability.monitoring.Beholder]
 InsecureConnection = true
 OtelExporterHTTPEndpoint = "host.docker.internal:4318"
 LogStreamingEnabled = false
 MetricReaderInterval = 5
 TraceSampleRatio = 1.0
 TraceBatchTimeout = 10
 
+[protocol_contracts]
+version = 1
+use_legacy_configure_lane = false
+
+## Environment configuration define the topology. Eventually the topology should be the only config that is needed.
+[protocol_contracts.environment_topology]
+indexer_address = ["http://indexer-1:8100"]
+
 [[protocol_contracts.environment_topology.nop_topology.nops]]
 alias = "default-verifier-1"
 name = "default-verifier-1"

build/devenv/environment.go

@@ -32,6 +32,7 @@ import (
 	_ "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/fake"
 	_ "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/indexer"
 	_ "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/jd"
+	_ "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/observability"
 	_ "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/pricer"
 	_ "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/protocol_contracts"
 	_ "github.com/smartcontractkit/chainlink-ccv/build/devenv/components/tokenverifier"

build/devenv/runtime/environment.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"maps"
+	"slices"
 	"sort"
 
 	"github.com/rs/zerolog"
@@ -69,12 +70,8 @@ func NewEnvironmentWithRegistry(ctx context.Context, rawConfig map[string]any, r
 
 	unclaimed := unclaimedKeys(rawConfig, r.factories)
 	if len(unclaimed) > 0 {
-		keys := make([]string, 0, len(unclaimed))
-		for k := range unclaimed {
-			keys = append(keys, k)
-		}
-		// TODO: Make this an error.
-		logger.Warn().Strs("keys", keys).Msg("unclaimed config keys")
+		keys := slices.Sorted(maps.Keys(unclaimed))
+		return nil, fmt.Errorf("unclaimed config keys: %v", keys)
 	}
 	accumulated := map[string]any{}
 

build/devenv/runtime/environment_test.go

@@ -270,3 +270,15 @@ func TestVersionKeyRejectsNonInteger(t *testing.T) {
 	require.Contains(t, err.Error(), "version")
 	require.Contains(t, err.Error(), "integer")
 }
+
+// TestUnclaimedKeyFailsFast verifies a top-level config key that no registered
+// component claims (e.g. a typo or stale key) is a hard error naming the key.
+// The check runs before any phase executes, so no component need be registered.
+func TestUnclaimedKeyFailsFast(t *testing.T) {
+	r := devenvruntime.NewRegistry()
+
+	_, err := runEnv(t, r, map[string]any{"typoo": nil})
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "unclaimed config keys")
+	require.Contains(t, err.Error(), "typoo")
+}