diff --git a/core/scripts/cre/environment/configs/extensions/README.md b/core/scripts/cre/environment/configs/extensions/README.md new file mode 100644 index 00000000000..a305676a4aa --- /dev/null +++ b/core/scripts/cre/environment/configs/extensions/README.md @@ -0,0 +1,22 @@ +# Local CRE extensions + +Centralized CRE services run from their own repos' compose files, through their chainlink-testing-framework components +(`framework/components/dockercompose/cre_storage_service`). With an `[extensions]` section (`extensions.toml`), +`env start` starts them on `ctf` after the DONs, passing the registry chain, its CapabilitiesRegistry and the workflow +DON family, and points the workflow nodes at the roles they answer to. `env stop` removes them. + +| Role (alias on `ctf`) | Port on `ctf` | Workflow node config | +| --- | --- | --- | +| `cre-workflow-source` | 50052 | `Capabilities.WorkflowRegistry.AdditionalSources` | +| `cre-artifact-api` | 50052 | `Capabilities.WorkflowRegistry.WorkflowStorage.URL` | +| `cre-artifact-store` | 4566 | `WorkflowStorage.ArtifactStorageHost`, allowed for the gateway | +| `cre-deploy-api` | 50051 | deploy tooling (`registryclient`) | + +`storage_compose` defaults to a `cre-storage-service` checkout next to this repo; its images are built locally (see its +`local-stack/README.md`). Workflows then come from the extensions, not `env workflow deploy`; deploy tooling runs in a +container on `ctf`, since artifact URLs use `cre-artifact-store`. From the host, the roles are on free ports that +`env start` logs. + +```bash +CTF_CONFIGS=configs/workflow-gateway-don.toml,configs/extensions/extensions.toml go run . env start +``` diff --git a/core/scripts/cre/environment/configs/extensions/extensions.toml b/core/scripts/cre/environment/configs/extensions/extensions.toml new file mode 100644 index 00000000000..e3dd9099c6d --- /dev/null +++ b/core/scripts/cre/environment/configs/extensions/extensions.toml @@ -0,0 +1,3 @@ +# Starts the extension stacks (README.md) with any topology. Add it after the topology in CTF_CONFIGS. +[extensions] + storage_compose = "../../../../../cre-storage-service/local-stack/compose.yaml" diff --git a/core/scripts/cre/environment/environment/environment.go b/core/scripts/cre/environment/environment/environment.go index bc843fac1b1..adc2e23197e 100644 --- a/core/scripts/cre/environment/environment/environment.go +++ b/core/scripts/cre/environment/environment/environment.go @@ -348,6 +348,14 @@ func startCmd() *cobra.Command { return errors.Wrap(err, "failed to validate test configuration") } + var extensionsDonFamily string + if in.Extensions != nil { + var extErr error + if extensionsDonFamily, extErr = validateExtensions(cmdContext, in); extErr != nil { + return errors.Wrap(extErr, "invalid extensions config") + } + } + topologySummary, _, topErr := generateTopologyArtifactsForLoadedConfig(in) if topErr != nil { framework.L.Warn().Err(topErr).Msg("failed to generate topology visualization artifacts") @@ -401,6 +409,12 @@ func startCmd() *cobra.Command { fmt.Fprintf(os.Stderr, "failed to track startup: %s\n", dxErr) } + if in.Extensions != nil { + if err := startExtensions(cmdContext, in.Extensions, extensionsDonFamily, output); err != nil { + return errors.Wrap(err, "failed to start extensions") + } + } + if withChipIngressStack { startChipIngressStackErr := startChipIngressStack( cmdContext, @@ -643,6 +657,10 @@ func stopCmd() *cobra.Command { return errors.Wrap(removeErr, "failed to remove environment containers. Please remove them manually") } + if err := stopExtensions(); err != nil { + framework.L.Warn().Msgf("failed to stop extensions: %s", err) + } + if allFlag { stopChipIngressStackErr := stopChipIngressStack() if stopChipIngressStackErr != nil { @@ -814,6 +832,9 @@ func defaultGatewayWhitelistConfig(in *envconfig.Config, extraAllowedPorts []int if in.FakeHTTP != nil { ports = append(ports, in.FakeHTTP.Port) } + if in.Extensions != nil { + ports = append(ports, artifactStorePort) + } return gateway.WhitelistConfig{ ExtraAllowedPorts: ports, @@ -861,6 +882,9 @@ func StartCLIEnvironment( BlockchainDeployers: blockchains_sets.NewDeployerSet(testLogger, in.Infra), ContractVersions: env.ContractVersions(), } + if in.Extensions != nil { + universalSetupInput.ConfigFactoryFunctions = append(universalSetupInput.ConfigFactoryFunctions, extensionsNodeConfig) + } ctx, cancel := context.WithTimeout(cmdContext, 20*time.Minute) defer cancel() diff --git a/core/scripts/cre/environment/environment/extensions.go b/core/scripts/cre/environment/environment/extensions.go new file mode 100644 index 00000000000..0a64d61682f --- /dev/null +++ b/core/scripts/cre/environment/environment/extensions.go @@ -0,0 +1,193 @@ +package environment + +import ( + "context" + "os" + "os/exec" + "slices" + "strconv" + "strings" + + "github.com/pelletier/go-toml/v2" + "github.com/pkg/errors" + + "github.com/smartcontractkit/chainlink-testing-framework/framework" + storage "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/cre_storage_service" + keystone_changeset "github.com/smartcontractkit/chainlink/deployment/keystone/changeset" + "github.com/smartcontractkit/chainlink/system-tests/lib/cre" + "github.com/smartcontractkit/chainlink/system-tests/lib/cre/contracts" + creenv "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment" + envconfig "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/config" + coretoml "github.com/smartcontractkit/chainlink/v2/core/config/toml" + corechainlink "github.com/smartcontractkit/chainlink/v2/core/services/chainlink" +) + +// Role addresses the extension stacks answer to on ctf, as defined by their CTF components; configs/extensions/README.md +// lists them. +const ( + extensionsSourceName = "cre-extensions" + + workflowSourceRole = storage.WorkflowSourceRole + ":" + storage.NodeServicePort + artifactAPIRole = storage.ArtifactAPIRole + ":" + storage.NodeServicePort + artifactStoreRole = storage.ArtifactStoreRole + ":" + storage.ArtifactStorePort +) + +var artifactStorePort = mustPort(storage.ArtifactStorePort) + +// extensionsNodeConfig points workflow DON nodes at the extension roles: workflows come from cre-workflow-source and +// artifacts from cre-artifact-api / cre-artifact-store. +func extensionsNodeConfig(input cre.GenerateConfigsInput, configs cre.NodeIndexToConfigOverride) (cre.NodeIndexToConfigOverride, error) { + if !input.DonMetadata.HasFlag(cre.WorkflowDON) { + return configs, nil + } + + out := make(cre.NodeIndexToConfigOverride, len(configs)) + for idx, raw := range configs { + var c corechainlink.Config + if err := toml.Unmarshal([]byte(raw), &c); err != nil { + return nil, errors.Wrapf(err, "failed to unmarshal config of node %d in DON %s", idx, input.DonMetadata.Name) + } + + c.CRE.WorkflowFetcher = &coretoml.WorkflowFetcherConfig{URL: new("")} + c.Capabilities.WorkflowRegistry.WorkflowStorage = coretoml.WorkflowStorage{ + URL: new(artifactAPIRole), + TLSEnabled: new(false), + ArtifactStorageHost: new(artifactStoreRole), + } + c.Capabilities.WorkflowRegistry.AdditionalSourcesConfig = append(c.Capabilities.WorkflowRegistry.AdditionalSourcesConfig, coretoml.AdditionalWorkflowSource{ + URL: new(workflowSourceRole), + TLSEnabled: new(false), + Name: new(extensionsSourceName), + }) + + marshalled, err := toml.Marshal(c) + if err != nil { + return nil, errors.Wrapf(err, "failed to marshal config of node %d in DON %s", idx, input.DonMetadata.Name) + } + out[idx] = string(marshalled) + } + return out, nil +} + +// validateExtensions rejects, before the environment starts, extensions config that can't work: a non-Docker provider, +// a missing compose file, other than one workflow DON family, or stack images that are neither local nor pullable. It +// returns the workflow DON family the stacks serve. +func validateExtensions(ctx context.Context, in *envconfig.Config) (string, error) { + if in.Infra == nil || !in.Infra.IsDocker() { + return "", errors.New("extensions require the docker infra provider") + } + if _, err := os.Stat(in.Extensions.StorageCompose); err != nil { + return "", errors.Wrap(err, "extensions.storage_compose") + } + + var families []string + for _, ns := range in.NodeSets { + if slices.Contains(ns.DONTypes, cre.WorkflowDON) && len(ns.DonFamilies) > 0 && !slices.Contains(families, ns.DonFamilies[0]) { + families = append(families, ns.DonFamilies[0]) + } + } + if len(families) != 1 { + return "", errors.Errorf("extensions support exactly one workflow DON family, found %v", families) + } + + return families[0], ensureExtensionsImages(ctx, in.Extensions) +} + +// ensureExtensionsImages pulls the stacks' images that aren't local, and names the ones that can't be pulled (images +// built from the stack's repo) instead of failing later in docker compose up. +func ensureExtensionsImages(ctx context.Context, ext *envconfig.Extensions) error { + cmd := docker(ctx, "compose", "-f", ext.StorageCompose, "config", "--images") + // the stack's required inputs; their values don't affect the images + cmd.Env = append(os.Environ(), + "CRE_LOCAL_REGISTRY_RPC_URL=placeholder", + "CRE_LOCAL_REGISTRY_CHAIN_SELECTOR=0", + "CRE_LOCAL_REGISTRY_ADDRESS=placeholder", + "CRE_LOCAL_DON_FAMILY=placeholder", + ) + cmd.Stderr = os.Stderr + out, err := cmd.Output() + if err != nil { + return errors.Wrapf(err, "failed to list images in %s", ext.StorageCompose) + } + + images := strings.Fields(string(out)) + slices.Sort(images) + var missing []string + for _, image := range slices.Compact(images) { + if docker(ctx, "image", "inspect", image).Run() != nil && docker(ctx, "pull", "-q", image).Run() != nil { + missing = append(missing, image) + } + } + if len(missing) > 0 { + return errors.Errorf("images %v are neither local nor pullable; build them as described in the README next to %s", missing, ext.StorageCompose) + } + return nil +} + +// startExtensions starts the extension stacks on the environment's network, with the registry chain, its +// CapabilitiesRegistry and the workflow DON family as inputs. +func startExtensions(ctx context.Context, ext *envconfig.Extensions, donFamily string, output *creenv.SetupOutput) error { + in, err := storageInput(ext, output, donFamily) + if err != nil { + return err + } + + _ = storage.Stop() + + out, err := storage.NewWithContext(ctx, in) + if err != nil { + return errors.Wrapf(err, "failed to start %s", ext.StorageCompose) + } + framework.L.Info(). + Str("cre-deploy-api", out.UserService.ExternalURL). + Str("cre-workflow-source", out.NodeService.ExternalURL). + Str("cre-artifact-store", out.ArtifactStore.ExternalURL). + Msg("Extensions started; reachable from the host at") + return nil +} + +func stopExtensions() error { + return storage.Stop() +} + +func storageInput(ext *envconfig.Extensions, output *creenv.SetupOutput, donFamily string) (*storage.Input, error) { + creEnv := output.CreEnvironment + selector := creEnv.RegistryChainSelector + + var rpcURL string + for _, bc := range creEnv.Blockchains { + if bc.ChainSelector() == selector && len(bc.CtfOutput().Nodes) > 0 { + rpcURL = bc.CtfOutput().Nodes[0].InternalHTTPUrl + } + } + if rpcURL == "" { + return nil, errors.Errorf("no RPC URL for registry chain %d", selector) + } + + capReg := contracts.MightGetAddressFromDataStore(creEnv.CldfEnvironment.DataStore, selector, + keystone_changeset.CapabilitiesRegistry.String(), creEnv.ContractVersions[keystone_changeset.CapabilitiesRegistry.String()], "") + if capReg == nil { + return nil, errors.Errorf("no CapabilitiesRegistry on registry chain %d", selector) + } + + return &storage.Input{ + ComposeFile: "file://" + ext.StorageCompose, + Network: framework.DefaultNetworkName, + RegistryRPCURL: rpcURL, + RegistryChainSelector: selector, + RegistryAddress: capReg.Hex(), + DonFamily: donFamily, + }, nil +} + +func mustPort(port string) int { + p, err := strconv.Atoi(port) + if err != nil { + panic(errors.Wrapf(err, "invalid port %q", port)) + } + return p +} + +func docker(ctx context.Context, args ...string) *exec.Cmd { + return exec.CommandContext(ctx, "docker", args...) +} diff --git a/core/scripts/cre/environment/environment/extensions_test.go b/core/scripts/cre/environment/environment/extensions_test.go new file mode 100644 index 00000000000..b6b5d25fc8c --- /dev/null +++ b/core/scripts/cre/environment/environment/extensions_test.go @@ -0,0 +1,97 @@ +package environment + +import ( + "os" + "path/filepath" + "testing" + + "github.com/pelletier/go-toml/v2" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink/system-tests/lib/cre" + envconfig "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/config" + "github.com/smartcontractkit/chainlink/system-tests/lib/infra" + corechainlink "github.com/smartcontractkit/chainlink/v2/core/services/chainlink" +) + +const extensionsTestNodeConfig = ` +[CRE.WorkflowFetcher] +URL = 'file:///home/chainlink/workflows' + +[[Capabilities.WorkflowRegistry.AdditionalSources]] +URL = 'grpc-source:50051' +Name = 'existing' +` + +func TestExtensionsNodeConfig_workflowDON(t *testing.T) { + t.Parallel() + + out, err := extensionsNodeConfig(extensionsTestInput(cre.WorkflowDON), cre.NodeIndexToConfigOverride{0: extensionsTestNodeConfig, 1: extensionsTestNodeConfig}) + require.NoError(t, err) + require.Len(t, out, 2) + + var c corechainlink.Config + require.NoError(t, toml.Unmarshal([]byte(out[1]), &c)) + require.Empty(t, *c.CRE.WorkflowFetcher.URL) + + storage := c.Capabilities.WorkflowRegistry.WorkflowStorage + require.Equal(t, "cre-artifact-api:50052", *storage.URL) + require.Equal(t, "cre-artifact-store:4566", *storage.ArtifactStorageHost) + require.False(t, *storage.TLSEnabled) + + sources := c.Capabilities.WorkflowRegistry.AdditionalSourcesConfig + require.Len(t, sources, 2) + require.Equal(t, "existing", *sources[0].Name) + require.Equal(t, "cre-workflow-source:50052", *sources[1].URL) + require.False(t, *sources[1].TLSEnabled) +} + +func TestExtensionsNodeConfig_otherDONsUnchanged(t *testing.T) { + t.Parallel() + + configs := cre.NodeIndexToConfigOverride{0: extensionsTestNodeConfig} + out, err := extensionsNodeConfig(extensionsTestInput(cre.GatewayDON), configs) + require.NoError(t, err) + require.Equal(t, configs, out) +} + +func extensionsTestInput(flag cre.CapabilityFlag) cre.GenerateConfigsInput { + return cre.GenerateConfigsInput{DonMetadata: &cre.DonMetadata{Name: "don", Flags: []string{flag}}} +} + +func TestValidateExtensions_rejectsBeforeStart(t *testing.T) { + t.Parallel() + + compose := filepath.Join(t.TempDir(), "compose.yaml") + require.NoError(t, os.WriteFile(compose, []byte("services: {}\n"), 0o600)) + workflowNodeSet := func(family string) *cre.NodeSet { + return &cre.NodeSet{DONTypes: []string{cre.WorkflowDON}, DonFamilies: []string{family}} + } + + tests := []struct { + name string + in *envconfig.Config + want string + }{ + {"kubernetes", &envconfig.Config{ + Infra: &infra.Provider{Type: infra.Kubernetes}, + Extensions: &envconfig.Extensions{StorageCompose: compose}, + }, "docker infra provider"}, + {"missing compose file", &envconfig.Config{ + Infra: &infra.Provider{Type: infra.Docker}, + Extensions: &envconfig.Extensions{StorageCompose: filepath.Join(t.TempDir(), "missing.yaml")}, + }, "extensions.storage_compose"}, + {"two workflow DON families", &envconfig.Config{ + Infra: &infra.Provider{Type: infra.Docker}, + NodeSets: []*cre.NodeSet{workflowNodeSet("zone-a"), workflowNodeSet("zone-b")}, + Extensions: &envconfig.Extensions{StorageCompose: compose}, + }, "exactly one workflow DON family"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + _, err := validateExtensions(t.Context(), tt.in) + require.ErrorContains(t, err, tt.want) + }) + } +} diff --git a/core/scripts/cre/environment/environment/workflow.go b/core/scripts/cre/environment/environment/workflow.go index db29b0a2303..b30cc96965c 100644 --- a/core/scripts/cre/environment/environment/workflow.go +++ b/core/scripts/cre/environment/environment/workflow.go @@ -126,6 +126,9 @@ func deployWorkflowCmd() *cobra.Command { if resolverErr != nil { return errors.Wrap(resolverErr, "failed to load local CRE state") } + if resolver != nil && resolver.cfg.Extensions != nil { + return errors.New("❌ this environment reads workflows from its extensions, not the file fetcher; deploy through cre-deploy-api instead (see configs/extensions/README.md)") + } defer func() { metaData := map[string]any{} diff --git a/core/scripts/go.mod b/core/scripts/go.mod index d7d204ecf9d..6ec2261c435 100644 --- a/core/scripts/go.mod +++ b/core/scripts/go.mod @@ -40,7 +40,7 @@ require ( github.com/smartcontractkit/chainlink-evm/gethwrappers v0.0.0-20261005112317-b723176adfe8 github.com/smartcontractkit/chainlink-protos/job-distributor v0.20.1-0.20260701185448-696c075849ea github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.8 - github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.23 + github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.24-0.20261022155844-32538607cda4 github.com/smartcontractkit/chainlink-testing-framework/seth v1.51.5 github.com/smartcontractkit/chainlink/deployment v0.0.0-20260126202327-6be9a05f0caf github.com/smartcontractkit/chainlink/system-tests/lib v0.0.0-00010101000000-000000000000 diff --git a/core/scripts/go.sum b/core/scripts/go.sum index 9ba6a48e00b..1d60c27c888 100644 --- a/core/scripts/go.sum +++ b/core/scripts/go.sum @@ -1770,8 +1770,8 @@ github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.7 h1:2j1 github.com/smartcontractkit/chainlink-testing-framework/framework v0.16.7/go.mod h1:nyOjn4ADJGqRMe3+4ZXSV+J/7nWb1H2Vx8Qk57eLRYA= github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.4 h1:bGicxBPndwy9NeB79n+CgyNxA8aeWoMudC84krz6QGM= github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.4/go.mod h1:TsZMdVIPeIBzFwVIUmU7jkXOTHSpyvCJGeLtjuBxa8E= -github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.23 h1:FYZZ2U6h2y4sITrEyTKPHTzjJrrsqCqN3zGqkpk7p3s= -github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.23/go.mod h1:S6VqtFkFyrNaPOodfiJTEHSu7wgj8bFfWtvfdFM+ztc= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.24-0.20261022155844-32538607cda4 h1:zkbVIJML0Dkn4K45Butdp+5rTa8PYA56tSVTae7X7DM= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.24-0.20261022155844-32538607cda4/go.mod h1:Iyo5qLQujzsQ32DXD3Kg82ftgGPMr1g4uUpz2UGt2hc= github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.15.0 h1:qELAsHYwbNWfJDJsp+3yqAxvmBjB2/zLPZhR91t2cg8= github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.15.0/go.mod h1:5xdlcXFhjagMeRA6q01FKEMv5OqvfVknI5qwXGriS5M= github.com/smartcontractkit/chainlink-testing-framework/lib v1.54.9 h1:/5fgG2lZqqTJ7mr3mUtxbWiUxd/gmDHSEGUERIUPvyU= diff --git a/system-tests/lib/cre/environment/config/config.go b/system-tests/lib/cre/environment/config/config.go index 3af59afaedb..76442bc050d 100644 --- a/system-tests/lib/cre/environment/config/config.go +++ b/system-tests/lib/cre/environment/config/config.go @@ -68,12 +68,19 @@ type Config struct { ChipRouter *ctfchiprouter.Input `toml:"chip_router" validate:"required"` S3ProviderInput *s3provider.Input `toml:"s3provider"` CapabilityConfigs map[string]cre.CapabilityConfig `toml:"capability_configs"` // capability flag -> capability config + Extensions *Extensions `toml:"extensions"` Addresses []string `toml:"addresses"` mu sync.Mutex loaded bool } +// Extensions are centralized CRE services started alongside the DONs, answering to fixed role aliases on the +// Docker network (see core/scripts/cre/environment/configs/extensions/README.md). +type Extensions struct { + StorageCompose string `toml:"storage_compose" validate:"required"` // path to cre-storage-service/local-stack/compose.yaml +} + // Validate performs validation checks on the configuration, ensuring all required fields // are present and all referenced capabilities are known to the system. func (c *Config) Validate(envDependencies cre.CLIEnvironmentDependencies) error { diff --git a/system-tests/lib/cre/features/vault/vault.go b/system-tests/lib/cre/features/vault/vault.go index 53c485a9310..a92915c6785 100644 --- a/system-tests/lib/cre/features/vault/vault.go +++ b/system-tests/lib/cre/features/vault/vault.go @@ -145,14 +145,13 @@ func updateNodeConfig(workerNode *cre.NodeMetadata, currentConfig string, regist return nil, errors.Wrapf(unmarshallErr, "failed to unmarshal config for node index %d", workerNode.Index) } - // enable workflow registry syncer - typedConfig.Capabilities.WorkflowRegistry = coretoml.WorkflowRegistry{ - Address: new(workflowRegistryAddress.Hex()), - NetworkID: new("evm"), - ChainID: new(strconv.FormatUint(registryChainID, 10)), - SyncStrategy: new("reconciliation"), - ContractVersion: new(wfRegVersion.String()), - } + // enable workflow registry syncer, keeping other WorkflowRegistry settings (e.g. WorkflowStorage, AdditionalSources) + wfReg := &typedConfig.Capabilities.WorkflowRegistry + wfReg.Address = new(workflowRegistryAddress.Hex()) + wfReg.NetworkID = new("evm") + wfReg.ChainID = new(strconv.FormatUint(registryChainID, 10)) + wfReg.SyncStrategy = new("reconciliation") + wfReg.ContractVersion = new(wfRegVersion.String()) typedConfig.CRE.Linking = &coretoml.LinkingConfig{ URL: new(strings.TrimPrefix(framework.HostDockerInternal(), "http://") + ":18124"), TLSEnabled: new(false),