diff --git a/.github/workflows/cre-local-env-tests.yaml b/.github/workflows/cre-local-env-tests.yaml index 19f024d4d1b..620a08c7fe4 100644 --- a/.github/workflows/cre-local-env-tests.yaml +++ b/.github/workflows/cre-local-env-tests.yaml @@ -112,7 +112,7 @@ jobs: uses: aws-actions/configure-aws-credentials@010d0da01d0b5a38af31e9c3470dbfdabdecca3a # v4.0.1 with: aws-region: ${{ secrets.QA_AWS_REGION }} - role-to-assume: ${{ secrets.QA_AWS_ROLE_TO_ASSUME }} + role-to-assume: ${{ secrets.AWS_CTF_READ_ACCESS_ROLE_ARN }} role-duration-seconds: 1800 mask-aws-account-id: true @@ -151,7 +151,8 @@ jobs: env: DISABLE_DX_TRACKING: true GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - AWS_ECR: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.us-west-2.amazonaws.com + MAIN_AWS_ECR: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.us-west-2.amazonaws.com + SDLC_AWS_ECR: ${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.us-west-2.amazonaws.com run: | # Remove chip_ingress/chip_config sections since CI role lacks ECR permissions for the Atlas repo awk '/^\[chip_ingress\.build_config\]/,/^$/{next} /^\[chip_ingress\.pull_config\]/,/^$/{next} /^\[chip_config\.build_config\]/,/^$/{next} /^\[chip_config\.pull_config\]/,/^$/{next} {print}' configs/setup.toml > configs/setup.toml.tmp && mv configs/setup.toml.tmp configs/setup.toml @@ -165,6 +166,7 @@ jobs: CTF_CONFIGS: "./configs/workflow-gateway-don.toml" CTF_JD_IMAGE: "${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/job-distributor:0.22.1" CTF_CHAINLINK_IMAGE: "${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/chainlink:${{ github.event_name == 'pull_request' && format('nightly-{0}-plugins', steps.set-date.outputs.date) || inputs.chainlink_image_tag }}" + CTF_CHIP_ROUTER_IMAGE: "${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/local-cre-chip-router:v1.0.1" DISABLE_DX_TRACKING: "true" CI: "true" run: | diff --git a/.github/workflows/cre-regression-system-tests.yaml b/.github/workflows/cre-regression-system-tests.yaml index 5422fd0b413..95c16df5020 100644 --- a/.github/workflows/cre-regression-system-tests.yaml +++ b/.github/workflows/cre-regression-system-tests.yaml @@ -117,6 +117,7 @@ jobs: # Beholder stack will be started only for the Beholder tests CHIP_INGRESS_IMAGE: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/atlas-chip-ingress:da84cb72d3a160e02896247d46ab4b9806ebee2f CHIP_CONFIG_IMAGE: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/atlas-chip-config:7b4e9ee68fd1c737dd3480b5a3ced0188f29b969 + CTF_CHIP_ROUTER_IMAGE: "${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/local-cre-chip-router:v1.0.1" BILLING_PLATFORM_SERVICE_IMAGE: ${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/billing-platform-service:v1.45.0 steps: @@ -221,12 +222,9 @@ jobs: continue-on-error: ${{ env.ENABLE_AUTO_QUARANTINE == 'true' }} env: TEST_NAME: ${{ matrix.tests.test_name }} - TEST_TIMEOUT: 30m + TEST_TIMEOUT: 7m # let's leave 3 minutes for other steps (the whole job times out after 10 minutes) PARALLEL_COUNT: "10" - # parallelisation flags for tests - # fanout is necessary, because multiple tests would otherwise start chip test sinks at the same port, causing conflicts CRE_TEST_PARALLEL_ENABLED: "true" - CRE_TEST_CHIP_SINK_FANOUT_ENABLED: "true" run: | echo "Starting test: '${TEST_NAME}'" echo "⚠️⚠️⚠️ Add 'skip-e2e-regression' label to skip this step if necessary ⚠️⚠️⚠️" @@ -260,12 +258,12 @@ jobs: artifact-name: ${{ matrix.tests.test_id }}_test_logs - name: Show Docker containers status - if: failure() + if: failure() || cancelled() shell: bash run: docker ps -a - name: Save Regression tests Docker logs - if: failure() + if: failure() || cancelled() shell: bash working-directory: system-tests/tests/regression/cre run: | @@ -275,7 +273,7 @@ jobs: done - name: Upload all artifacts as single package - if: failure() + if: failure() || cancelled() uses: actions/upload-artifact@v7 with: name: test-logs-${{ matrix.tests.test_name }}-${{ matrix.tests.topology }} diff --git a/.github/workflows/cre-soak-memory-leak.yml b/.github/workflows/cre-soak-memory-leak.yml index c5f68be2c6b..4e8912e327b 100644 --- a/.github/workflows/cre-soak-memory-leak.yml +++ b/.github/workflows/cre-soak-memory-leak.yml @@ -40,6 +40,7 @@ jobs: CTF_CHAINLINK_IMAGE: "${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/${{ inputs.ecr_name || 'chainlink' }}:${{ inputs.chainlink_image_tag }}" CHIP_INGRESS_IMAGE: "${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/atlas-chip-ingress:da84cb72d3a160e02896247d46ab4b9806ebee2f" CHIP_CONFIG_IMAGE: "${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/atlas-chip-config:7b4e9ee68fd1c737dd3480b5a3ced0188f29b969" + CTF_CHIP_ROUTER_IMAGE: "${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/local-cre-chip-router:v1.0.1" steps: - name: Enable S3 Cache for Self-Hosted Runners @@ -76,28 +77,12 @@ jobs: registries: ${{ format('{0},{1}', secrets.QA_AWS_ACCOUNT_NUMBER, secrets.AWS_ACCOUNT_ID_PROD) }} env: AWS_REGION: ${{ secrets.QA_AWS_REGION }} - - - name: Install CTF binary - shell: bash - working-directory: core/scripts/cre/environment - env: - GH_TOKEN: ${{ secrets.GITHUB_TOKEN }} - CTF_TAG: "framework/v0.15.2" - run: | - echo "::startgroup::Install CTF binary" - mkdir -p bin - gh release download "${CTF_TAG}" --pattern "framework-v*-linux-amd64.tar.gz" --clobber --repo smartcontractkit/chainlink-testing-framework -O ctf.tar.gz - tar -xzf ctf.tar.gz - mv ctf bin/ctf - chmod +x bin/ctf - echo "::endgroup::" - - name: Start observability stack shell: bash working-directory: core/scripts/cre/environment run: | echo "::startgroup::Starting observability stack (required by leak package)" - ./bin/ctf obs up -f + go run . obs up -f echo "::endgroup::" - name: Start local CRE diff --git a/.github/workflows/cre-system-tests.yaml b/.github/workflows/cre-system-tests.yaml index 34881fe913e..60ec33e105e 100644 --- a/.github/workflows/cre-system-tests.yaml +++ b/.github/workflows/cre-system-tests.yaml @@ -248,6 +248,7 @@ jobs: env: CTF_JD_IMAGE: "${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/job-distributor:0.22.1" CTF_CHAINLINK_IMAGE: "${{ steps.resolve-chainlink-image.outputs.resolved_image }}" + CTF_CHIP_ROUTER_IMAGE: "${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/local-cre-chip-router:v1.0.1" CTF_CONFIGS: ${{ matrix.tests.configs }} CRE_VERSION: ${{ matrix.tests.cre_version }} TEST_NAME: ${{ matrix.tests.test_name }} @@ -327,15 +328,12 @@ jobs: continue-on-error: ${{ env.ENABLE_AUTO_QUARANTINE == 'true' }} env: TEST_NAME: ${{ matrix.tests.test_name }} - TEST_TIMEOUT: 30m + TEST_TIMEOUT: 7m # let's leave 3 minutes for other steps (the whole job times out after 10 minutes) RUN_QUARANTINED_TESTS: "true" # always run quarantined tests in CI TOPOLOGY_NAME: ${{ matrix.tests.topology }} GITHUB_TOKEN: ${{ steps.github-token.outputs.access-token || '' }} # to avoid rate limiting when downloading protobuf files from GitHub PARALLEL_COUNT: "10" - # parallelisation flags for tests - # fanout is necessary, because multiple tests would otherwise start chip test sinks at the same port, causing conflicts CRE_TEST_PARALLEL_ENABLED: "true" - CRE_TEST_CHIP_SINK_FANOUT_ENABLED: "true" run: | echo "Starting test: '${TEST_NAME}'" gotestsum \ @@ -365,12 +363,12 @@ jobs: artifact-name: ${{ matrix.tests.test_id }}_test_logs - name: Show Docker containers status - if: failure() + if: failure() || cancelled() shell: bash run: docker ps -a - name: Save Smoke tests Docker logs - if: failure() + if: failure() || cancelled() shell: bash working-directory: system-tests/tests/smoke/cre run: | @@ -383,7 +381,7 @@ jobs: done - name: Upload all artifacts as single package - if: failure() + if: failure() || cancelled() uses: actions/upload-artifact@v7 with: name: test-logs-${{ matrix.tests.test_name }}-${{ matrix.tests.topology }} diff --git a/.github/workflows/devenv-compat.yml b/.github/workflows/devenv-compat.yml index 149aa337d07..69075e32d32 100644 --- a/.github/workflows/devenv-compat.yml +++ b/.github/workflows/devenv-compat.yml @@ -99,6 +99,8 @@ jobs: CTF_LOG_LEVEL: ${{ inputs.ctf-log-level }} # JD is required by the local CRE-based tests CTF_JD_IMAGE: "${{ secrets.AWS_ACCOUNT_ID_PROD }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/job-distributor:0.22.1" + # ChIP Router is required by the local CRE-based tests + CTF_CHIP_ROUTER_IMAGE: "${{ secrets.QA_AWS_ACCOUNT_NUMBER }}.dkr.ecr.${{ secrets.QA_AWS_REGION }}.amazonaws.com/local-cre-chip-router:v1.0.1" steps: - name: Checkout code uses: actions/checkout@v5 diff --git a/core/scripts/cre/environment/configs/setup.toml b/core/scripts/cre/environment/configs/setup.toml index 313c6e82996..f4a50dbae8a 100644 --- a/core/scripts/cre/environment/configs/setup.toml +++ b/core/scripts/cre/environment/configs/setup.toml @@ -12,7 +12,19 @@ local_image = "job-distributor:0.22.1" [job_distributor.pull_config] local_image = "job-distributor:0.22.1" -ecr_image = "{{.ECR}}/job-distributor:0.22.1" +ecr_image = "{{.MAIN_ECR}}/job-distributor:0.22.1" + +[chip_router.build_config] +repository = "https://github.com/smartcontractkit/chainlink-testing-framework" +branch = "main" +commit = "dd420c0d953334e3fef9a109b6816de706f6be90" +dockerfile = "framework/components/chiprouter/Dockerfile" +docker_ctx = "framework/components/chiprouter" +local_image = "local-cre-chip-router:v1.0.1" + +[chip_router.pull_config] +local_image = "local-cre-chip-router:v1.0.1" +ecr_image = "{{.SDLC_ECR}}/local-cre-chip-router:v1.0.1" [chip_ingress.build_config] repository = "https://github.com/smartcontractkit/atlas" @@ -25,7 +37,7 @@ pre_run = "pushd chip-ingress && go mod vendor && popd" [chip_ingress.pull_config] local_image = "chip-ingress:da84cb72d3a160e02896247d46ab4b9806ebee2f" -ecr_image = "{{.ECR}}/atlas-chip-ingress:da84cb72d3a160e02896247d46ab4b9806ebee2f" +ecr_image = "{{.MAIN_ECR}}/atlas-chip-ingress:da84cb72d3a160e02896247d46ab4b9806ebee2f" [chip_config.build_config] repository = "https://github.com/smartcontractkit/atlas" @@ -38,7 +50,7 @@ pre_run = "pushd chip-config && go mod vendor && popd" [chip_config.pull_config] local_image = "chip-config:7b4e9ee68fd1c737dd3480b5a3ced0188f29b969" -ecr_image = "{{.ECR}}/atlas-chip-config:7b4e9ee68fd1c737dd3480b5a3ced0188f29b969" +ecr_image = "{{.MAIN_ECR}}/atlas-chip-config:7b4e9ee68fd1c737dd3480b5a3ced0188f29b969" [billing_platform_service.build_config] repository = "https://github.com/smartcontractkit/billing-platform-service" @@ -50,7 +62,7 @@ local_image = "billing-platform-service:local-cre" [billing_platform_service.pull_config] local_image = "billing-platform-service:local-cre" -ecr_image = "{{.ECR}}/billing-platform-service:1.36.1" +ecr_image = "{{.MAIN_ECR}}/billing-platform-service:1.36.1" [observability] repository = "https://github.com/smartcontractkit/chainlink-observability" diff --git a/core/scripts/cre/environment/configs/workflow-don-solana.toml b/core/scripts/cre/environment/configs/workflow-don-solana.toml index 947ef9e5eef..14a6452c83a 100644 --- a/core/scripts/cre/environment/configs/workflow-don-solana.toml +++ b/core/scripts/cre/environment/configs/workflow-don-solana.toml @@ -1,4 +1,7 @@ +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-don-tron.toml b/core/scripts/cre/environment/configs/workflow-don-tron.toml index 711e6efe0b7..bceba577cdb 100755 --- a/core/scripts/cre/environment/configs/workflow-don-tron.toml +++ b/core/scripts/cre/environment/configs/workflow-don-tron.toml @@ -1,4 +1,7 @@ +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-capabilities-don.toml b/core/scripts/cre/environment/configs/workflow-gateway-capabilities-don.toml index b94d80787b6..8b70049da33 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-capabilities-don.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-capabilities-don.toml @@ -1,4 +1,7 @@ +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-don-aptos.toml b/core/scripts/cre/environment/configs/workflow-gateway-don-aptos.toml index 2748039bff0..6746b4e6997 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-don-aptos.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-don-aptos.toml @@ -1,6 +1,9 @@ # Same as workflow-gateway-don.toml but with Aptos chain and a single Aptos capability. # Anvil 1337: registry and gateway. Aptos: local devnet (chain_id 4). Run: env config path , then env start. +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-don-grpc-source.toml b/core/scripts/cre/environment/configs/workflow-gateway-don-grpc-source.toml index 01ae469e74d..09b6571935b 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-don-grpc-source.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-don-grpc-source.toml @@ -4,6 +4,9 @@ # # Used by: system-tests/tests/smoke/cre/v2_grpc_source_test.go +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-don.toml b/core/scripts/cre/environment/configs/workflow-gateway-don.toml index 197853faa28..5bbb9f55a1f 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-don.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-don.toml @@ -1,4 +1,7 @@ +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-legacy-vault-don.toml b/core/scripts/cre/environment/configs/workflow-gateway-legacy-vault-don.toml index 40c5cb85d16..b352240acd3 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-legacy-vault-don.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-legacy-vault-don.toml @@ -1,5 +1,8 @@ # NOTE: Identical to workflow-gatewway-capabilities.toml but with a vault capability config override # to disable the new pending queue feature. +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-mock-don.toml b/core/scripts/cre/environment/configs/workflow-gateway-mock-don.toml index 0e66be77870..bf3d098e09e 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-mock-don.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-mock-don.toml @@ -1,3 +1,6 @@ +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] chain_id = "1337" container_name = "anvil-1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-sharded-5-dons.toml b/core/scripts/cre/environment/configs/workflow-gateway-sharded-5-dons.toml index e9453a70c50..5f4feb374fc 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-sharded-5-dons.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-sharded-5-dons.toml @@ -1,4 +1,7 @@ +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/configs/workflow-gateway-sharded-don.toml b/core/scripts/cre/environment/configs/workflow-gateway-sharded-don.toml index c1943d030ba..992212c2b7d 100644 --- a/core/scripts/cre/environment/configs/workflow-gateway-sharded-don.toml +++ b/core/scripts/cre/environment/configs/workflow-gateway-sharded-don.toml @@ -1,4 +1,7 @@ +[chip_router] + image = "local-cre-chip-router:v1.0.1" + [[blockchains]] type = "anvil" chain_id = "1337" diff --git a/core/scripts/cre/environment/environment/beholder.go b/core/scripts/cre/environment/environment/beholder.go index 826498bcf80..36904fcf356 100644 --- a/core/scripts/cre/environment/environment/beholder.go +++ b/core/scripts/cre/environment/environment/beholder.go @@ -23,7 +23,9 @@ import ( "github.com/spf13/cobra" "github.com/smartcontractkit/chainlink-testing-framework/framework" + ctfchiprouter "github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter" chipingressset "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/chip_ingress_set" + "github.com/smartcontractkit/chainlink/system-tests/lib/cre/chiprouter" envconfig "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/config" "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/stagegen" libformat "github.com/smartcontractkit/chainlink/system-tests/lib/format" @@ -287,6 +289,10 @@ func startBeholderCmd() *cobra.Command { return fmt.Errorf("failed to set TESTCONTAINERS_RYUK_DISABLED environment variable: %w", setErr) } + if routerErr := chiprouter.EnsureStarted(cmd.Context()); routerErr != nil { + return errors.Wrap(routerErr, "failed to ensure chip ingress router is running. Please make sure that local CRE environment is started and that the chip ingress router is running") + } + startBeholderErr = startBeholder(cmd.Context(), timeout, port) if startBeholderErr != nil { // remove the stack if the error is not related to proto registration @@ -305,7 +311,7 @@ func startBeholderCmd() *cobra.Command { } cmd.Flags().DurationVarP(&timeout, "wait-on-error-timeout", "w", 15*time.Second, "Time to wait before removing Docker containers if environment fails to start (e.g. 10s, 1m, 1h)") - cmd.Flags().IntVarP(&port, "grpc-port", "g", mustStringToInt(chipingressset.DEFAULT_CHIP_INGRESS_GRPC_PORT), "GRPC port for Chip Ingress") + cmd.Flags().IntVarP(&port, "grpc-port", "g", ctfchiprouter.DefaultBeholderGRPCPort, "GRPC port for downstream Chip Ingress") return cmd } @@ -319,6 +325,47 @@ func mustStringToInt(in string) int { return out } +func loadPersistedBeholderState(relativePathToRepoRoot string) (*envconfig.ChipIngressConfig, error) { + absPath := envconfig.MustChipIngressStateFileAbsPath(relativePathToRepoRoot) + if _, err := os.Stat(absPath); err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, errors.Wrap(err, "failed to stat persisted Beholder state") + } + + cfg := &envconfig.ChipIngressConfig{} + if err := cfg.Load(absPath); err != nil { + return nil, errors.Wrap(err, "failed to load persisted Beholder state") + } + + return cfg, nil +} + +func persistedBeholderGRPCEndpoint(cfg *envconfig.ChipIngressConfig) string { + if cfg == nil || cfg.ChipIngress == nil || cfg.ChipIngress.Output == nil || cfg.ChipIngress.Output.ChipIngress == nil { + return "" + } + + return strings.TrimSpace(cfg.ChipIngress.Output.ChipIngress.GRPCExternalURL) +} + +func restorePersistedBeholderState(relativePathToRepoRoot string, cfg *envconfig.ChipIngressConfig) error { + if cfg == nil { + return nil + } + return cfg.Store(envconfig.MustChipIngressStateFileAbsPath(relativePathToRepoRoot)) +} + +func reconcilePersistedBeholderWithRouter(ctx context.Context, cfg *envconfig.ChipIngressConfig) error { + endpoint := persistedBeholderGRPCEndpoint(cfg) + if endpoint == "" { + return errors.New("persisted Beholder state is missing chip ingress grpc endpoint") + } + + return registerBeholderEndpointWithRouter(ctx, endpoint) +} + var stopBeholderCmd = &cobra.Command{ Use: "stop", Short: "Stop the Beholder", @@ -330,6 +377,17 @@ var stopBeholderCmd = &cobra.Command{ } func stopBeholder() error { + subscriberID, loadSubscriberErr := loadBeholderSubscriberID(relativePathToRepoRoot) + if loadSubscriberErr != nil && !os.IsNotExist(loadSubscriberErr) { + framework.L.Warn().Err(loadSubscriberErr).Msg("failed to load Beholder router subscriber id") + } + if subscriberID != "" { + unregisterErr := chiprouter.UnregisterSubscriber(context.Background(), subscriberID) + if unregisterErr != nil && !os.IsNotExist(unregisterErr) && !strings.Contains(unregisterErr.Error(), "local CRE state file not found") && !strings.Contains(unregisterErr.Error(), "no such file or directory") { + framework.L.Warn().Err(unregisterErr).Msg("failed to unregister Beholder from chip ingress router") + } + } + setErr := os.Setenv("CTF_CONFIGS", DefaultBeholderConfigFile) if setErr != nil { return fmt.Errorf("failed to set CTF_CONFIGS environment variable: %w", setErr) @@ -350,7 +408,13 @@ func removeBeholderStateFiles(relativePathToRepoRoot string) error { return errors.Wrap(absErr, "error getting absolute path for chip ingress state file") } - return os.Remove(absPath) + if err := os.Remove(absPath); err != nil && !os.IsNotExist(err) { + return err + } + if err := os.Remove(beholderSubscriberIDPath(relativePathToRepoRoot)); err != nil && !os.IsNotExist(err) { + return err + } + return nil } func isPortAvailable(addr string) bool { @@ -368,12 +432,20 @@ var protoRegistrationErrMsg = "proto registration failed" // MissingImage represents an image that needs to be built or pulled type MissingImage struct { Name string - Tag string FullImage string BuildConfig BuildConfig PullConfig PullConfig } +func newMissingImage(name string, cfg ImageConfig) MissingImage { + return MissingImage{ + Name: name, + FullImage: cfg.BuildConfig.LocalImage, + BuildConfig: cfg.BuildConfig, + PullConfig: cfg.PullConfig, + } +} + // ensureChipImagesExist checks if required chip images exist and auto-builds them if missing. // In CI environments (CI=true), this check is skipped as images will be pulled at runtime. func ensureChipImagesExist(ctx context.Context, cfg *SetupConfigFile) error { @@ -383,42 +455,35 @@ func ensureChipImagesExist(ctx context.Context, cfg *SetupConfigFile) error { return nil } + var requiredImages []MissingImage + if cfg.ChipIngress != nil { + requiredImages = append(requiredImages, newMissingImage("chip-ingress", ImageConfig{ + BuildConfig: cfg.ChipIngress.BuildConfig, + PullConfig: cfg.ChipIngress.PullConfig, + })) + } + if cfg.ChipConfig != nil { + requiredImages = append(requiredImages, newMissingImage("chip-config", ImageConfig{ + BuildConfig: cfg.ChipConfig.BuildConfig, + PullConfig: cfg.ChipConfig.PullConfig, + })) + } + + return ensureManagedImagesExist(ctx, cfg.General.AWSProfile, requiredImages) +} + +func ensureManagedImagesExist(ctx context.Context, awsProfile string, requiredImages []MissingImage) error { dockerClient, err := client.NewClientWithOpts(client.WithAPIVersionNegotiation()) if err != nil { return errors.Wrap(err, "failed to create Docker client") } defer dockerClient.Close() - // Check if Docker is running _, err = dockerClient.Ping(ctx) if err != nil { return errors.Wrap(err, "Docker is not running") } - // Collect required images - var requiredImages []MissingImage - - if cfg.ChipIngress != nil { - requiredImages = append(requiredImages, MissingImage{ - Name: "chip-ingress", - Tag: cfg.ChipIngress.BuildConfig.Commit, - FullImage: cfg.ChipIngress.BuildConfig.LocalImage, - BuildConfig: cfg.ChipIngress.BuildConfig, - PullConfig: cfg.ChipIngress.PullConfig, - }) - } - - if cfg.ChipConfig != nil { - requiredImages = append(requiredImages, MissingImage{ - Name: "chip-config", - Tag: cfg.ChipConfig.BuildConfig.Commit, - FullImage: cfg.ChipConfig.BuildConfig.LocalImage, - BuildConfig: cfg.ChipConfig.BuildConfig, - PullConfig: cfg.ChipConfig.PullConfig, - }) - } - - // Find missing images var missing []MissingImage for _, img := range requiredImages { _, err := dockerClient.ImageInspect(ctx, img.FullImage) @@ -434,23 +499,21 @@ func ensureChipImagesExist(ctx context.Context, cfg *SetupConfigFile) error { return nil } - ecrURL := os.Getenv("AWS_ECR") + missingRegistryVars := missingRegistryEnvVars(missing) interactive := isInteractiveTerminal() // Non-interactive mode handling if !interactive { - if ecrURL != "" { - // Non-interactive with AWS_ECR - pull images - framework.L.Info().Msgf("Non-interactive mode with AWS_ECR set. Pulling %d missing image(s) from ECR...", len(missing)) - return pullAllImages(ctx, cfg, missing) + if len(missingRegistryVars) == 0 { + framework.L.Info().Msgf("Non-interactive mode with required ECR env vars set. Pulling %d missing image(s) from ECR...", len(missing)) + return pullAllImages(ctx, awsProfile, missing) } - // Non-interactive without AWS_ECR - fail with instructions - framework.L.Error().Msgf("Missing %d required image(s) and AWS_ECR is not set:", len(missing)) + framework.L.Error().Msgf("Missing %d required image(s) and required ECR env vars are not set:", len(missing)) for _, img := range missing { framework.L.Error().Msgf(" - %s", img.FullImage) } - printChipImagePullInstructions() - return errors.Errorf("missing %d required image(s). Set AWS_ECR to enable auto-pull or run 'go run . env setup' manually", len(missing)) + printChipImagePullInstructions(missingRegistryVars) + return errors.Errorf("missing %d required image(s). Set %s to enable auto-pull or run 'go run . env setup' manually", len(missing), strings.Join(missingRegistryVars, ", ")) } // Interactive mode - try building first @@ -478,14 +541,14 @@ func ensureChipImagesExist(ctx context.Context, cfg *SetupConfigFile) error { } // Some builds failed - offer to pull all failed images - return handleChipImageBuildFailures(ctx, cfg, failedBuilds, buildErrors) + return handleChipImageBuildFailures(ctx, awsProfile, failedBuilds, buildErrors) } // pullAllImages pulls all specified images from ECR -func pullAllImages(ctx context.Context, cfg *SetupConfigFile, images []MissingImage) error { +func pullAllImages(ctx context.Context, awsProfile string, images []MissingImage) error { for _, img := range images { framework.L.Info().Msgf("Pulling %s from ECR...", img.Name) - _, pullErr := img.PullConfig.Pull(ctx, cfg.General.AWSProfile) + _, pullErr := img.PullConfig.Pull(ctx, awsProfile) if pullErr != nil { return errors.Wrapf(pullErr, "failed to pull %s", img.Name) } @@ -505,7 +568,7 @@ func isInteractiveTerminal() bool { } // handleChipImageBuildFailures handles build failures by offering to pull all failed images -func handleChipImageBuildFailures(ctx context.Context, cfg *SetupConfigFile, failedImages []MissingImage, buildErrors []error) error { +func handleChipImageBuildFailures(ctx context.Context, awsProfile string, failedImages []MissingImage, buildErrors []error) error { // List all failed images fmt.Println() framework.L.Error().Msgf("Failed to build %d image(s):", len(failedImages)) @@ -513,14 +576,14 @@ func handleChipImageBuildFailures(ctx context.Context, cfg *SetupConfigFile, fai framework.L.Error().Msgf(" - %s: %v", img.FullImage, buildErrors[i]) } - ecrURL := os.Getenv("AWS_ECR") - if ecrURL != "" { + missingRegistryVars := missingRegistryEnvVars(failedImages) + if len(missingRegistryVars) == 0 { shouldPull := false if isInteractiveTerminal() { // Interactive mode - ask user fmt.Println() - fmt.Printf("AWS_ECR is set. Would you like to pull all %d failed image(s) from ECR instead? [Y/n] ", len(failedImages)) + fmt.Printf("Required ECR env vars are set. Would you like to pull all %d failed image(s) from ECR instead? [Y/n] ", len(failedImages)) reader := bufio.NewReader(os.Stdin) input, _ := reader.ReadString('\n') @@ -537,7 +600,7 @@ func handleChipImageBuildFailures(ctx context.Context, cfg *SetupConfigFile, fai // Pull all failed images for _, img := range failedImages { framework.L.Info().Msgf("Pulling %s from ECR...", img.Name) - _, pullErr := img.PullConfig.Pull(ctx, cfg.General.AWSProfile) + _, pullErr := img.PullConfig.Pull(ctx, awsProfile) if pullErr != nil { return errors.Wrapf(pullErr, "failed to pull %s", img.Name) } @@ -548,19 +611,41 @@ func handleChipImageBuildFailures(ctx context.Context, cfg *SetupConfigFile, fai } // Show manual instructions - printChipImagePullInstructions() + printChipImagePullInstructions(missingRegistryVars) return errors.Errorf("failed to build %d image(s)", len(failedImages)) } -// printChipImagePullInstructions prints helpful instructions for pulling images manually -func printChipImagePullInstructions() { +func missingRegistryEnvVars(images []MissingImage) []string { + seen := make(map[string]struct{}) + var missing []string + for _, img := range images { + for _, envVar := range img.PullConfig.MissingRegistryEnvVars() { + if _, ok := seen[envVar]; ok { + continue + } + seen[envVar] = struct{}{} + missing = append(missing, envVar) + } + } + return missing +} + +// printChipImagePullInstructions prints helpful instructions for pulling images manually. +func printChipImagePullInstructions(requiredEnvVars []string) { fmt.Println() fmt.Println("────────────────────────────────────────────────────────────────") fmt.Println("To pull pre-built images instead, run:") fmt.Println() - fmt.Println(" AWS_ECR=.dkr.ecr.us-west-2.amazonaws.com go run . env setup") + if len(requiredEnvVars) == 0 { + requiredEnvVars = []string{mainECREnvVarName, sdlcECREnvVarName} + } + assignments := make([]string, 0, len(requiredEnvVars)) + for _, envVar := range requiredEnvVars { + assignments = append(assignments, envVar+"=") + } + fmt.Printf(" %s go run . env setup\n", strings.Join(assignments, " ")) fmt.Println() - fmt.Println("Replace with prod AWS account number.") + fmt.Printf("Set the required registry env vars: %s.\n", strings.Join(requiredEnvVars, ", ")) fmt.Println("See: https://smartcontract-it.atlassian.net/wiki/spaces/INFRA/pages/1045495923") fmt.Println("────────────────────────────────────────────────────────────────") fmt.Println() @@ -603,9 +688,8 @@ func startBeholder(cmdContext context.Context, cleanupWait time.Duration, port i fmt.Print(libformat.PurpleText("%s", stageGen.Wrap("Starting Chip Ingress stack"))) if !isPortAvailable(":" + strconv.Itoa(port)) { - return fmt.Errorf(`port %d is already in use. Most probably an instance of ChIP Test Sink is already running. -If you want to use both together start ChIP Ingress on a different port with '--grpc-port' flag -and make sure that the sink is pointing to correct upstream endpoint ('localhost:' in most cases)`, port) + return fmt.Errorf(`port %d is already in use. Either an instance of CHiP Router or ChIP Test Sink is already running. +If you want to use both together start ChIP Ingress on a different port with '--grpc-port' flag`, port) } // Load setup config to check for required images @@ -696,6 +780,10 @@ and make sure that the sink is pointing to correct upstream endpoint ('localhost fmt.Println() framework.L.Info().Msgf("Red Panda Console URL: %s", out.RedPanda.ConsoleExternalURL) + if err := registerBeholderWithRouter(cmdContext, port); err != nil { + return errors.Wrap(err, "failed to register Beholder with chip ingress router") + } + topicsErr := chipingressset.CreateTopics(cmdContext, out.RedPanda.KafkaExternalURL, in.Kafka.Topics) if topicsErr != nil { return errors.Wrap(topicsErr, "failed to create topics") @@ -714,6 +802,45 @@ and make sure that the sink is pointing to correct upstream endpoint ('localhost return in.Store(envconfig.MustChipIngressStateFileAbsPath(relativePathToRepoRoot)) } +func registerBeholderWithRouter(ctx context.Context, port int) error { + return registerBeholderEndpointWithRouter(ctx, fmt.Sprintf("127.0.0.1:%d", port)) +} + +func registerBeholderEndpointWithRouter(ctx context.Context, endpoint string) error { + previousID, err := loadBeholderSubscriberID(relativePathToRepoRoot) + if err == nil && previousID != "" { + _ = chiprouter.UnregisterSubscriber(ctx, previousID) + } + + id, err := chiprouter.RegisterSubscriber(ctx, "beholder", endpoint) + if err != nil { + return err + } + + // Persist the fixed alias so stopBeholder can remove it without reading transient test output. + if id == "" { + return errors.New("empty subscriber id returned when registering Beholder") + } + + statePath := beholderSubscriberIDPath(relativePathToRepoRoot) + if writeErr := os.WriteFile(statePath, []byte(id), 0o600); writeErr != nil { + return errors.Wrap(writeErr, "failed to persist Beholder router subscriber id") + } + return nil +} + +func loadBeholderSubscriberID(relativePathToRepoRoot string) (string, error) { + raw, err := os.ReadFile(beholderSubscriberIDPath(relativePathToRepoRoot)) + if err != nil { + return "", err + } + return strings.TrimSpace(string(raw)), nil +} + +func beholderSubscriberIDPath(relativePathToRepoRoot string) string { + return filepath.Join(relativePathToRepoRoot, envconfig.StateDirname, "chip_ingress_router_beholder_subscriber") +} + func parseConfigsAndRegisterProtos(ctx context.Context, schemaSets []chipingressset.SchemaSet, chipIngressOutput *chipingressset.ChipIngressOutput) error { if len(schemaSets) == 0 { framework.L.Warn().Msg("no proto configs provided, skipping proto registration") diff --git a/core/scripts/cre/environment/environment/environment.go b/core/scripts/cre/environment/environment/environment.go index 9da6c66a263..a187b8070b9 100644 --- a/core/scripts/cre/environment/environment/environment.go +++ b/core/scripts/cre/environment/environment/environment.go @@ -257,6 +257,11 @@ func startCmd() *cobra.Command { return fmt.Errorf("with-plugins-docker-image flag is no longer supported. Set Docker image in TOML config instead (%s) for each nodeset under the [nodesets.nodesets.node_specs.node.image] field", effectiveConfig) } + persistedBeholderState, persistedBeholderStateErr := loadPersistedBeholderState(relativePathToRepoRoot) + if persistedBeholderStateErr != nil { + framework.L.Warn().Err(persistedBeholderStateErr).Msg("failed to load persisted Beholder state before startup cleanup") + } + cleanUpErr := envconfig.RemoveAllEnvironmentStateDir(relativePathToRepoRoot) if cleanUpErr != nil { return errors.Wrap(cleanUpErr, "failed to clean up environment state files") @@ -274,6 +279,7 @@ func startCmd() *cobra.Command { if err := in.Load(os.Getenv("CTF_CONFIGS")); err != nil { return errors.Wrap(err, "failed to load environment configuration") } + applyChipRouterImageOverride(in) // Skip Docker operations for Kubernetes provider (Docker not needed) isDocker := in.Infra != nil && !in.Infra.IsKubernetes() @@ -285,6 +291,10 @@ func startCmd() *cobra.Command { return err } + if err := ensureChipRouterImageExists(cmdContext, in, setupConfig.ConfigPath); err != nil { + return err + } + // This will not work with remote images that require authentication, but it will catch early most of the issues with missing env setup if err := ensureDockerImagesExist(cmdContext, framework.L, in); err != nil { return err @@ -369,6 +379,19 @@ func startCmd() *cobra.Command { return errors.Wrap(startErr, "failed to start environment") } + storeErr := in.Store(envconfig.MustLocalCREStateFileAbsPath(relativePathToRepoRoot)) + if storeErr != nil { + return errors.Wrap(storeErr, "failed to store local CRE state") + } + + if !withBeholder && persistedBeholderState != nil { + if err := reconcilePersistedBeholderWithRouter(cmdContext, persistedBeholderState); err != nil { + framework.L.Warn().Err(err).Msg("failed to re-register persisted Beholder with chip ingress router") + } else if err := restorePersistedBeholderState(relativePathToRepoRoot, persistedBeholderState); err != nil { + framework.L.Warn().Err(err).Msg("failed to restore persisted Beholder state after router re-registration") + } + } + registryChainOut := output.CreEnvironment.Blockchains[0] sErr := StartCmdGenerateSettingsFile(registryChainOut, output) @@ -495,7 +518,7 @@ func startCmd() *cobra.Command { if stErr != nil { return errors.Wrap(stErr, "failed to set addresses on Config") } - storeErr := in.Store(envconfig.MustLocalCREStateFileAbsPath(relativePathToRepoRoot)) + storeErr = in.Store(envconfig.MustLocalCREStateFileAbsPath(relativePathToRepoRoot)) if storeErr != nil { return errors.Wrap(storeErr, "failed to store local CRE state") } @@ -820,6 +843,7 @@ func StartCLIEnvironment( universalSetupInput := &creenv.SetupInput{ NodeSets: in.NodeSets, BlockchainsInput: in.Blockchains, + ChipRouterInput: in.ChipRouter, ContractVersions: env.ContractVersions(), WithV2Registries: env.WithV2Registries(), JdInput: in.JD, @@ -881,7 +905,7 @@ func PrintCRELogo() { func setDefaultCtfConfigs() error { if os.Getenv("CTF_CONFIGS") == "" { - if err := os.Setenv("CTF_CONFIGS", "configs/workflow-gateway-don.toml"); err != nil { + if err := os.Setenv("CTF_CONFIGS", "configs/workflow-gateway-capabilities-don.toml"); err != nil { return fmt.Errorf("failed to set CTF_CONFIGS environment variable: %w", err) } @@ -897,6 +921,50 @@ func setDefaultCtfConfigs() error { return nil } +func applyChipRouterImageOverride(in *envconfig.Config) { + if in == nil || in.ChipRouter == nil { + return + } + + override := strings.TrimSpace(os.Getenv(envconfig.CTFChipRouterImageEnvVar)) + if override == "" { + return + } + + in.ChipRouter.Image = override + framework.L.Info().Msgf("Using Chip Router image override from %s: %s", envconfig.CTFChipRouterImageEnvVar, override) +} + +func ensureChipRouterImageExists(ctx context.Context, in *envconfig.Config, setupConfigPath string) error { + if os.Getenv("CI") == "true" { + framework.L.Info().Msg("CI environment detected, skipping chip image pre-check") + return nil + } + + if in == nil || in.ChipRouter == nil || (in.Infra != nil && in.Infra.IsKubernetes()) { + return nil + } + + setupCfg, err := ReadSetupConfig(setupConfigPath) + if err != nil { + return errors.Wrap(err, "failed to read setup config for chip router image validation") + } + if setupCfg.ChipRouter == nil { + return errors.New("chip_router configuration is missing from setup config") + } + + routerImage := newMissingImage("chip-router", ImageConfig{ + BuildConfig: setupCfg.ChipRouter.BuildConfig, + PullConfig: setupCfg.ChipRouter.PullConfig, + }.WithLocalImage(in.ChipRouter.Image)) + + if err := ensureManagedImagesExist(ctx, setupCfg.General.AWSProfile, []MissingImage{routerImage}); err != nil { + return errors.Wrapf(err, "Chip Router image '%s' is not available", in.ChipRouter.Image) + } + + return nil +} + func hasBuiltDockerImage(in *envconfig.Config) bool { for _, nodeset := range in.NodeSets { for _, nodeSpec := range nodeset.NodeSpecs { @@ -1162,6 +1230,9 @@ func allEnvironmentStateFiles() ([]string, error) { func initLocalCREStageGen(in *envconfig.Config) *stagegen.StageGen { stages := 9 + if in.ChipRouter != nil { + stages++ + } if in.S3ProviderInput != nil { stages++ } diff --git a/core/scripts/cre/environment/environment/setup.go b/core/scripts/cre/environment/environment/setup.go index 4e15f3d5d04..cb1cfb74e62 100644 --- a/core/scripts/cre/environment/environment/setup.go +++ b/core/scripts/cre/environment/environment/setup.go @@ -60,6 +60,7 @@ func init() { type SetupConfigFile struct { General GeneralConfig `toml:"general"` JobDistributor JobDistributorConfig `toml:"job_distributor"` + ChipRouter *ChipRouterConfig `toml:"chip_router"` ChipIngress *ChipIngressConfig `toml:"chip_ingress"` ChipConfig *ChipConfigConfig `toml:"chip_config"` BillingService *BillingServiceConfig `toml:"billing_platform_service"` @@ -78,6 +79,12 @@ type JobDistributorConfig struct { PullConfig PullConfig `toml:"pull_config"` } +// ChipRouterConfig contains chip router image configuration +type ChipRouterConfig struct { + BuildConfig BuildConfig `toml:"build_config"` + PullConfig PullConfig `toml:"pull_config"` +} + // ChipIngressConfig contains chip ingress image configuration type ChipIngressConfig struct { BuildConfig BuildConfig `toml:"build_config"` @@ -103,11 +110,20 @@ type ObservabilityConfig struct { TargetPath string `toml:"target_path"` } -var ( - ECR = os.Getenv("AWS_ECR") // TODO this can be moved to an env file +const DefaultSetupConfigPath = "configs/setup.toml" + +const ( + mainECREnvVarName = "MAIN_AWS_ECR" + sdlcECREnvVarName = "SDLC_AWS_ECR" ) -const DefaultSetupConfigPath = "configs/setup.toml" +func mainECR() string { + return os.Getenv(mainECREnvVarName) +} + +func sdlcECR() string { + return os.Getenv(sdlcECREnvVarName) +} type EnsureOption = string @@ -225,6 +241,10 @@ func (c BuildConfig) Build(ctx context.Context) (localImage string, err error) { tag = c.Branch commit = c.Commit ) + if strings.TrimSpace(c.LocalRepo) != "" { + repo = c.LocalRepo + } + logger := framework.L name := strings.ReplaceAll(strings.Split(c.LocalImage, ":")[0], "-", " ") name = cases.Title(language.English).String(name) @@ -291,9 +311,20 @@ type PullConfig struct { EcrImage string `toml:"ecr_image"` } +func (c PullConfig) MissingRegistryEnvVars() []string { + var missing []string + if strings.Contains(c.EcrImage, "{{.MAIN_ECR}}") && mainECR() == "" { + missing = append(missing, mainECREnvVarName) + } + if strings.Contains(c.EcrImage, "{{.SDLC_ECR}}") && sdlcECR() == "" { + missing = append(missing, sdlcECREnvVarName) + } + return missing +} + func (c PullConfig) Pull(ctx context.Context, awsProfile string) (localImage string, err error) { - if ECR == "" { - return "", errors.New("AWS_ECR environment variable is not set. See README for more details and references to find the correct ECR URL or visit https://smartcontract-it.atlassian.net/wiki/spaces/INFRA/pages/1045495923/Configure+the+AWS+CLI") + if missing := c.MissingRegistryEnvVars(); len(missing) > 0 { + return "", fmt.Errorf("%s environment variable(s) must be set. See README for setup details and https://smartcontract-it.atlassian.net/wiki/spaces/INFRA/pages/1045495923/Configure+the+AWS+CLI", strings.Join(missing, ", ")) } tmpl, tmplErr := template.New("ecr-image").Parse(c.EcrImage) @@ -302,7 +333,8 @@ func (c PullConfig) Pull(ctx context.Context, awsProfile string) (localImage str } templateData := map[string]string{ - "ECR": ECR, + "MAIN_ECR": mainECR(), + "SDLC_ECR": sdlcECR(), } var configBuffer bytes.Buffer @@ -319,6 +351,13 @@ type ImageConfig struct { PullConfig PullConfig } +func (c ImageConfig) WithLocalImage(localImage string) ImageConfig { + out := c + out.BuildConfig.LocalImage = localImage + out.PullConfig.LocalImage = localImage + return out +} + func (c ImageConfig) Ensure(ctx context.Context, dockerClient *client.Client, awsProfile string, noPrompt bool, defaultOption EnsureOption, purge bool) (localImage string, err error) { // If purge flag is set, remove existing images first if purge { @@ -333,7 +372,7 @@ func (c ImageConfig) Ensure(ctx context.Context, dockerClient *client.Client, aw logger.Warn().Msgf("Failed to remove local image %s: %v", c.BuildConfig.LocalImage, err) } - // Remove ECR image if it exists + // Remove remote-tagged image if it exists _, err = dockerClient.ImageRemove(ctx, c.PullConfig.EcrImage, image.RemoveOptions{Force: true}) if err != nil { logger.Warn().Msgf("Failed to remove ECR image %s: %v", c.PullConfig.EcrImage, err) @@ -487,6 +526,23 @@ func RunSetup(ctx context.Context, config SetupConfig, noPrompt, purge, withBill return } + var chipRouterLocalImage string + if cfg.ChipRouter != nil { + chipRouterConfig := ImageConfig{ + BuildConfig: cfg.ChipRouter.BuildConfig, + PullConfig: cfg.ChipRouter.PullConfig, + } + + var err error + chipRouterLocalImage, err = chipRouterConfig.Ensure(ctx, dockerClient, cfg.General.AWSProfile, noPrompt, PullOption, purge) + if err != nil { + setupErr = errors.Wrap(err, "failed to ensure Chip Router image") + return + } + } else { + logger.Warn().Str("config file", config.ConfigPath).Msg("Skipping Chip Router setup, because configuration is not provided in the config file") + } + var chipIngressLocalImage string if cfg.ChipIngress != nil { chipConfig := ImageConfig{ @@ -560,6 +616,9 @@ func RunSetup(ctx context.Context, config SetupConfig, noPrompt, purge, withBill logger.Info().Msg("✅ Setup Summary:") logger.Info().Msg(" ✓ Docker is installed and configured correctly") logger.Info().Msgf(" ✓ Job Distributor image %s is available", jdLocalImage) + if chipRouterLocalImage != "" { + logger.Info().Msgf(" ✓ Chip Router image %s is available", chipRouterLocalImage) + } if chipIngressLocalImage != "" { logger.Info().Msgf(" ✓ Atlas Chip Ingress image %s is available", chipIngressLocalImage) } @@ -744,8 +803,8 @@ func checkDockerConfiguration() error { return nil } -// localImageExists checks if the local image or ECR image exists -// if ECR image exists, it tags it as the local image +// localImageExists checks if the local image or rendered remote image exists +// if the rendered remote image exists, it tags it as the local image func localImageExists(ctx context.Context, dockerClient *client.Client, localImage, ecrImage string) (bool, error) { logger := framework.L name := strings.ReplaceAll(strings.Split(localImage, ":")[0], "-", " ") @@ -757,7 +816,7 @@ func localImageExists(ctx context.Context, dockerClient *client.Client, localIma return true, nil } - // Check if ECR image exists + // Check if rendered remote image exists _, err = dockerClient.ImageInspect(ctx, ecrImage) if err == nil { logger.Info().Msgf("✓ %s image (%s) is available", name, ecrImage) @@ -771,7 +830,7 @@ func localImageExists(ctx context.Context, dockerClient *client.Client, localIma return false, nil } -// pullImage pulls the Job Distributor image from ECR +// pullImage pulls the configured image from its remote registry and retags it locally. func pullImage(ctx context.Context, awsProfile string, localImage, ecrImage string) (string, error) { logger := framework.L name := strings.ReplaceAll(strings.Split(localImage, ":")[0], "-", " ") diff --git a/core/scripts/go.mod b/core/scripts/go.mod index 84a4e5d1472..30155a6cad0 100644 --- a/core/scripts/go.mod +++ b/core/scripts/go.mod @@ -55,6 +55,7 @@ require ( github.com/smartcontractkit/chainlink-protos/cre/go v0.0.0-20260409211238-5b99921cbc7c github.com/smartcontractkit/chainlink-protos/job-distributor v0.18.0 github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16 + github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2 github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.20 github.com/smartcontractkit/chainlink-testing-framework/lib v1.54.5 github.com/smartcontractkit/chainlink-testing-framework/seth v1.51.5 diff --git a/core/scripts/go.sum b/core/scripts/go.sum index b09e2f5c97d..69bed2624d4 100644 --- a/core/scripts/go.sum +++ b/core/scripts/go.sum @@ -1715,6 +1715,8 @@ github.com/smartcontractkit/chainlink-sui/deployment v0.0.0-20260304150206-c64e4 github.com/smartcontractkit/chainlink-sui/deployment v0.0.0-20260304150206-c64e48eb0cb0/go.mod h1:IfeW6t5Yc5293H5ixuooAft+wYBMSFQWKjbBTwYiKr4= github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16 h1:pzrAgF6QFMQLS/kukXenLN87PCa48SEMlE7QvJxTOHs= github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16/go.mod h1:BALK9cj8sk12e15UF6uDhifHgIApa+6N11TcQfInEro= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2 h1:c8T3cnZXrwOo6YU4A/VNj941mZo6n9VbfY3POP74OoA= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2/go.mod h1:MjUJAyU+kLvLLPRPBs3X7zYadds5umZcjTLHjfNhpUc= github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.20 h1:8D2DUnn7mLUZOLhPDGGFKKvBrgU6LQd00tq2VOprvfI= github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.20/go.mod h1:98jNYBOPuKWJw9a8x0LgQuudp5enrHhQQP5Hq0YwRB8= github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.10.0 h1:PWAMYu0WaAMBfbpxCpFJGRIDHmcgmYin6a+UQC0OdtY= diff --git a/docs/local-cre/environment/index.md b/docs/local-cre/environment/index.md index d6c4967f7bc..60ebb7b6c11 100644 --- a/docs/local-cre/environment/index.md +++ b/docs/local-cre/environment/index.md @@ -74,6 +74,20 @@ By default Local CRE builds the Chainlink image from the local branch. To use a The deprecated `-p/--with-plugins-docker-image` flag still exists, but contributors should use TOML-based image selection instead. +`env setup` ensures required managed images are present, including Job Distributor, Chip Router, Chip Ingress, and Chip Config. + +When pulling managed images from ECR, configure both registries: + +- `MAIN_AWS_ECR` for core managed CRE images +- `SDLC_AWS_ECR` for Chip Router images + +Chip Router image resolution during startup is: + +1. `CTF_CHIP_ROUTER_IMAGE` (if set) +2. `chip_router.image` from the active topology TOML + +If the resolved router image is missing locally, startup follows the same build-or-pull fallback path as the Beholder images. + ## Beholder and Observability Use `--with-beholder` when you need the ChIP ingress stack and Red Panda. Use `--with-observability` or `--with-dashboards` when you need the Grafana-based observability stack. @@ -85,6 +99,27 @@ Important related flags: When dashboards are enabled, the CLI waits for Grafana at `http://localhost:3000`. +### Chip Router Topology + +Chip Router is the ingress owner on `50051`. Nodes emit workflow telemetry to the router, and the router fans out to downstream subscribers. + +Current local ports: + +- `50050`: Chip Router admin API +- `50051`: Chip Router ingress gRPC +- `50052`: chip-config +- `50053`: real ChIP / Beholder ingress gRPC + +In tests, sink-backed scenarios register a test sink with Chip Router. Beholder-backed scenarios register real ChIP / Beholder with Chip Router. + +To override the router image without changing committed TOMLs: + +```bash +export CTF_CHIP_ROUTER_IMAGE=chip-router: +``` + +This override takes precedence over `chip_router.image`. + ## Storage and State Local CRE persists state to the repo-local state file that the system tests later reuse. This is why the smoke-test helpers can detect an existing environment and avoid recreating it. @@ -109,7 +144,7 @@ Hot-swapping guidance and workflow-specific commands are covered in: The Local CRE stack supports: - OTel-based observability -- Chip ingress / Beholder integration +- Chip Router fanout with Beholder integration - DX tracing If you need the full tracing stack for debugging or demos, enable observability during startup and follow the environment-specific tracing configuration described in the advanced page. diff --git a/docs/local-cre/getting-started/index.md b/docs/local-cre/getting-started/index.md index 4893cdac758..d74a6338ff5 100644 --- a/docs/local-cre/getting-started/index.md +++ b/docs/local-cre/getting-started/index.md @@ -42,6 +42,12 @@ cd core/scripts/cre/environment go run . env start --auto-setup ``` +If you need setup/startup to pull managed images from ECR, provide both registries: + +```bash +MAIN_AWS_ECR= SDLC_AWS_ECR= go run . env start --auto-setup +``` + Deploy a first workflow: ```bash @@ -84,13 +90,13 @@ Once the environment is up, run the CRE smoke package: go test ./system-tests/tests/smoke/cre -timeout 20m -run '^Test_CRE_' ``` -For the default smoke-test flow, start Local CRE without `--with-beholder`. Most tests start the ChIP test sink on the default gRPC port (`50051`), and Beholder uses that same port through Chip Ingress. +For the default smoke-test flow, start Local CRE without `--with-beholder`. Chip Router owns ingress on `50051`, and tests register downstream subscribers behind the router (test sink by default, Beholder for Beholder-backed scenarios). -Enable Beholder only when: +Enable Beholder when: - you are running Beholder-specific tests - you intentionally need the Beholder stack for debugging -- you move Beholder to a different port with `--grpc-port` so it does not conflict with the test sink +- you want to inspect workflow events in the Beholder stack during debugging The smoke tests default to the capability-enabled topology when you do not override `CTF_CONFIGS`, and the test helpers can start Local CRE automatically if the state file does not exist yet. diff --git a/go.md b/go.md index 3bedafbab43..5d3c09e7b55 100644 --- a/go.md +++ b/go.md @@ -402,6 +402,9 @@ flowchart LR click chainlink-sui/deployment href "https://github.com/smartcontractkit/chainlink-sui" chainlink-testing-framework/framework --> chainlink-testing-framework/wasp click chainlink-testing-framework/framework href "https://github.com/smartcontractkit/chainlink-testing-framework" + chainlink-testing-framework/framework/components/chiprouter --> chainlink-common/pkg/chipingress + chainlink-testing-framework/framework/components/chiprouter --> chainlink-testing-framework/framework + click chainlink-testing-framework/framework/components/chiprouter href "https://github.com/smartcontractkit/chainlink-testing-framework" chainlink-testing-framework/framework/components/dockercompose --> chainlink-common/pkg/chipingress chainlink-testing-framework/framework/components/dockercompose --> chainlink-testing-framework/framework chainlink-testing-framework/framework/components/dockercompose --> freeport @@ -463,6 +466,7 @@ flowchart LR chainlink/load-tests --> chainlink-testing-framework/havoc chainlink/load-tests --> chainlink/integration-tests click chainlink/load-tests href "https://github.com/smartcontractkit/chainlink" + chainlink/system-tests/lib --> chainlink-testing-framework/framework/components/chiprouter chainlink/system-tests/lib --> chainlink-testing-framework/framework/components/dockercompose chainlink/system-tests/lib --> chainlink-testing-framework/framework/components/fake chainlink/system-tests/lib --> chainlink/deployment @@ -680,6 +684,7 @@ flowchart LR subgraph chainlink-testing-framework-repo[chainlink-testing-framework] chainlink-testing-framework/framework + chainlink-testing-framework/framework/components/chiprouter chainlink-testing-framework/framework/components/dockercompose chainlink-testing-framework/framework/components/fake chainlink-testing-framework/havoc diff --git a/system-tests/lib/cre/chiprouter/router.go b/system-tests/lib/cre/chiprouter/router.go new file mode 100644 index 00000000000..38eb74c4077 --- /dev/null +++ b/system-tests/lib/cre/chiprouter/router.go @@ -0,0 +1,113 @@ +package chiprouter + +import ( + "context" + "errors" + "net" + "net/netip" + "os" + "strings" + "sync" + + pkgerrors "github.com/pkg/errors" + + "github.com/smartcontractkit/chainlink-testing-framework/framework" + ctfchiprouterclient "github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter/client" + envconfig "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/config" +) + +var ( + clientOnce sync.Once + clientInst *ctfchiprouterclient.Client + errClient error +) + +func getClient(ctx context.Context) (*ctfchiprouterclient.Client, error) { + clientOnce.Do(func() { + in := &envconfig.Config{} + err := in.Load(os.Getenv("CTF_CONFIGS")) + if err != nil { + errClient = err + return + } + + if in.ChipRouter == nil || in.ChipRouter.Out == nil { + errClient = errors.New("chip router output not found in local CRE state") + return + } + + // st, err := envconfig.LoadChipIngressRouterStateFromLocalCRE(relativePathToRepoRoot) + // if err != nil { + // errClient = err + // return + // } + clientInst, errClient = ctfchiprouterclient.New(ctx, in.ChipRouter.Out.ExternalAdminURL, in.ChipRouter.Out.ExternalGRPCURL) + }) + + return clientInst, errClient +} + +func EnsureStarted(ctx context.Context) error { + _, err := getClient(ctx) + if err != nil { + if os.IsNotExist(err) { + return pkgerrors.New("local CRE state file not found; start the environment first") + } + return err + } + + return nil +} + +func RegisterSubscriber(ctx context.Context, name, endpoint string) (string, error) { + c, err := getClient(ctx) + if err != nil { + return "", err + } + + return c.RegisterSubscriber(ctx, name, normalizeEndpointForRouter(endpoint)) +} + +func UnregisterSubscriber(ctx context.Context, id string) error { + if strings.TrimSpace(id) == "" { + return nil + } + + c, err := getClient(ctx) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return err + } + + return c.UnregisterSubscriber(ctx, id) +} + +func normalizeEndpointForRouter(endpoint string) string { + host, port, err := net.SplitHostPort(strings.TrimSpace(endpoint)) + if err != nil { + return endpoint + } + + if !requiresHostGateway(host) { + return endpoint + } + + dockerHost := strings.TrimPrefix(framework.HostDockerInternal(), "http://") + return net.JoinHostPort(dockerHost, port) +} + +func requiresHostGateway(host string) bool { + switch strings.TrimSpace(host) { + case "", "localhost": + return true + } + + addr, err := netip.ParseAddr(host) + if err != nil { + return false + } + + return addr.IsLoopback() || addr.IsUnspecified() +} diff --git a/system-tests/lib/cre/don.go b/system-tests/lib/cre/don.go index ae310de3350..ce0b1a107d4 100644 --- a/system-tests/lib/cre/don.go +++ b/system-tests/lib/cre/don.go @@ -3,7 +3,6 @@ package cre import ( "context" "fmt" - "math" "net/url" "slices" "strconv" @@ -288,15 +287,9 @@ func registerWithJD(ctx context.Context, d *Don, supportedChains []blockchains.B for _, role := range node.Roles { switch role { case RoleWorker, RoleBootstrap: - chainConfigStart := time.Now() if err := createJDChainConfigs(ctx, node, supportedChains, jd); err != nil { return fmt.Errorf("failed to create supported chains in node %s: %w", node.Name, err) } - framework.L.Info(). - Str("don", d.Name). - Str("node", node.Name). - Float64("duration_s", roundSeconds(time.Since(chainConfigStart))). - Msg("JD chain-config setup completed") case RoleGateway: // no chains configuration needed for gateway nodes default: @@ -846,7 +839,6 @@ func LinkToJobDistributor(ctx context.Context, input *LinkDonsToJDInput) error { return errors.New("input is nil") } - start := time.Now() dons := input.Dons.List() donMetadata := input.Topology.DonsMetadata.List() nodeIDsByDON := make([][]string, len(dons)) @@ -854,7 +846,6 @@ func LinkToJobDistributor(ctx context.Context, input *LinkDonsToJDInput) error { errGroup, groupCtx := errgroup.WithContext(ctx) for idx, don := range dons { errGroup.Go(func() error { - donStart := time.Now() supportedChains, schErr := findDonSupportedChains(donMetadata[idx], input.Blockchains) if schErr != nil { return errors.Wrap(schErr, "failed to find supported chains for DON") @@ -865,10 +856,6 @@ func LinkToJobDistributor(ctx context.Context, input *LinkDonsToJDInput) error { } nodeIDsByDON[idx] = don.JDNodeIDs() - framework.L.Info(). - Str("don", don.Name). - Float64("duration_s", roundSeconds(time.Since(donStart))). - Msg("JD registration completed for DON") return nil }) } @@ -884,7 +871,6 @@ func LinkToJobDistributor(ctx context.Context, input *LinkDonsToJDInput) error { input.CldfEnvironment.NodeIDs = nodeIDs framework.L.Info(). - Float64("duration_s", roundSeconds(time.Since(start))). Msg("Post-start JD linking completed") return nil @@ -933,10 +919,6 @@ func findDonSupportedChains(donMetadata *DonMetadata, bcs []blockchains.Blockcha return chains, nil } -func roundSeconds(d time.Duration) float64 { - return math.Round(d.Seconds()*10) / 10 -} - // Make DonMetadata also implement it, just in case? type KeystoneDON interface { KeystoneDONConfig() ks_contracts_op.ConfigureKeystoneDON diff --git a/system-tests/lib/cre/environment/config/config.go b/system-tests/lib/cre/environment/config/config.go index b31df9e1b6b..ef1ceb9044f 100644 --- a/system-tests/lib/cre/environment/config/config.go +++ b/system-tests/lib/cre/environment/config/config.go @@ -18,6 +18,7 @@ import ( "github.com/smartcontractkit/chainlink-deployments-framework/datastore" "github.com/smartcontractkit/chainlink-testing-framework/framework" "github.com/smartcontractkit/chainlink-testing-framework/framework/components/blockchain" + ctfchiprouter "github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter" billingplatformservice "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/billing_platform_service" chipingressset "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/chip_ingress_set" "github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake" @@ -56,13 +57,16 @@ func (c *Config) SetAddresses(refs []datastore.AddressRef) error { return nil } +const CTFChipRouterImageEnvVar = "CTF_CHIP_ROUTER_IMAGE" + type Config struct { - Blockchains []*blockchain.Input `toml:"blockchains" validate:"required"` - NodeSets []*cre.NodeSet `toml:"nodesets" validate:"required"` + Blockchains []*blockchain.Input `toml:"blockchains" validate:"required,min=1"` + NodeSets []*cre.NodeSet `toml:"nodesets" validate:"required,min=1"` JD *jd.Input `toml:"jd" validate:"required"` Infra *infra.Provider `toml:"infra" validate:"required"` Fake *fake.Input `toml:"fake"` FakeHTTP *fake.Input `toml:"fake_http"` + 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 Addresses []string `toml:"addresses"` @@ -78,18 +82,6 @@ func (c *Config) Validate(envDependencies cre.CLIEnvironmentDependencies) error return errors.New("jd.csa_encryption_key must be provided") } - if len(c.Blockchains) == 0 { - return errors.New("at least one blockchain must be configured") - } - - if len(c.NodeSets) == 0 { - return errors.New("at least one nodeset must be configured") - } - - if c.Infra == nil { - return errors.New("infra configuration must be provided") - } - for _, nodeSet := range c.NodeSets { for _, capability := range nodeSet.Capabilities { capability = removeChainIDFromFlag(capability) diff --git a/system-tests/lib/cre/environment/environment.go b/system-tests/lib/cre/environment/environment.go index da702cbb2ab..1390d5dd6ae 100644 --- a/system-tests/lib/cre/environment/environment.go +++ b/system-tests/lib/cre/environment/environment.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "maps" - "os" "github.com/Masterminds/semver/v3" "github.com/ethereum/go-ethereum/common" @@ -22,6 +21,7 @@ import ( "github.com/smartcontractkit/chainlink-deployments-framework/operations" "github.com/smartcontractkit/chainlink-testing-framework/framework/components/blockchain" + ctfchiprouter "github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter" "github.com/smartcontractkit/chainlink-testing-framework/framework/components/jd" "github.com/smartcontractkit/chainlink-testing-framework/framework/components/s3provider" @@ -55,6 +55,7 @@ type SetupOutput struct { type SetupInput struct { NodeSets []*cre.NodeSet BlockchainsInput []*blockchain.Input + ChipRouterInput *ctfchiprouter.Input JdInput *jd.Input Provider infra.Provider ContractVersions map[cre.ContractType]*semver.Version @@ -108,15 +109,6 @@ func SetupTestEnvironment( return nil, pkgerrors.New("input is nil") } - //TODO: remove these checks in December 2025, when everyone has migrated - if val := os.Getenv("E2E_JD_IMAGE"); val != "" { - return nil, errors.New("E2E_JD_IMAGE and E2E_JD_VERSION are deprecated, please use CTF_JD_IMAGE instead to specify the Job Distributor image with tag") - } - - if val := os.Getenv("E2E_TEST_CHAINLINK_IMAGE"); val != "" { - return nil, errors.New("E2E_TEST_CHAINLINK_IMAGE and E2E_TEST_CHAINLINK_VERSION are deprecated, please use CTF_CHAINLINK_IMAGE instead to specify the Chainlink Node image with tag") - } - if err := input.Validate(); err != nil { return nil, pkgerrors.Wrap(err, "input validation failed") } @@ -126,6 +118,13 @@ func SetupTestEnvironment( return nil, pkgerrors.Wrap(s3Err, "failed to start S3 provider") } + fmt.Print(libformat.PurpleText("%s", input.StageGen.Wrap("Starting Chip Router"))) + _, err := ctfchiprouter.NewWithContext(ctx, input.ChipRouterInput) + if err != nil { + return nil, pkgerrors.Wrap(err, "failed to start chip router") + } + + fmt.Print(libformat.PurpleText("%s", input.StageGen.WrapAndNext("Chip Router started in %.2f seconds", input.StageGen.Elapsed().Seconds()))) fmt.Print(libformat.PurpleText("%s", input.StageGen.Wrap("Starting %d blockchain(s)", len(input.BlockchainsInput)))) deployedBlockchains, startErr := blockchains.Start( diff --git a/system-tests/lib/cre/types.go b/system-tests/lib/cre/types.go index 6402d41e2a8..fca444f2514 100644 --- a/system-tests/lib/cre/types.go +++ b/system-tests/lib/cre/types.go @@ -10,7 +10,6 @@ import ( "slices" "strconv" "strings" - "time" "github.com/Masterminds/semver/v3" "github.com/ethereum/go-ethereum/common" @@ -586,7 +585,6 @@ func NewDonMetadata(c *NodeSet, id uint64, provider infra.Provider, capabilityCo cfgs[i] = cfg } - newNodesStart := time.Now() nodes, err := newNodes(cfgs) if err != nil { return nil, fmt.Errorf("failed to create nodes metadata: %w", err) @@ -594,7 +592,6 @@ func NewDonMetadata(c *NodeSet, id uint64, provider infra.Provider, capabilityCo framework.L.Info(). Str("don", c.Name). Int("nodes", len(cfgs)). - Float64("duration_s", roundSeconds(time.Since(newNodesStart))). Msg("Node metadata generation completed") capConfigs, capErr := processCapabilityConfigs(c, capabilityConfigs) @@ -1465,7 +1462,6 @@ type NodeKeyInput struct { } func NewNodeKeys(input NodeKeyInput) (*secrets.NodeKeys, error) { - start := time.Now() out := &secrets.NodeKeys{ EVM: make(map[uint64]*crypto.EVMKey), Solana: make(map[string]*crypto.SolKey), @@ -1525,7 +1521,6 @@ func NewNodeKeys(input NodeKeyInput) (*secrets.NodeKeys, error) { Int("evm_chains", len(input.EVMChainIDs)). Int("solana_chains", len(input.SolanaChainIDs)). Bool("imported", input.ImportedSecrets != ""). - Float64("duration_s", roundSeconds(time.Since(start))). Msg("Node key generation completed") return out, nil } diff --git a/system-tests/lib/go.mod b/system-tests/lib/go.mod index eefc3ef9af6..9007fe8e948 100644 --- a/system-tests/lib/go.mod +++ b/system-tests/lib/go.mod @@ -41,6 +41,7 @@ require ( github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260323124644-faea187e6997 github.com/smartcontractkit/chainlink-solana v1.1.2-0.20260407162454-407b2a207dcc github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16 + github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2 github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.15 github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.10.0 github.com/smartcontractkit/chainlink-testing-framework/lib v1.54.5 @@ -455,7 +456,7 @@ require ( github.com/smartcontractkit/chainlink-ccip/chains/solana/gobindings v0.0.0-20260310183131-8d0f0e383288 // indirect github.com/smartcontractkit/chainlink-ccip/deployment v0.0.0-20260331203127-b8dc0a6b85c5 // indirect github.com/smartcontractkit/chainlink-ccv v0.0.0-20260408181529-b5080e662563 // indirect - github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10 // indirect + github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20251211140724-319861e514c4 // indirect github.com/smartcontractkit/chainlink-data-streams v0.1.13 // indirect github.com/smartcontractkit/chainlink-evm/contracts/cre/gobindings v0.0.0-20260403151002-2c91155b5501 // indirect github.com/smartcontractkit/chainlink-feeds v0.1.2-0.20250227211209-7cd000095135 // indirect diff --git a/system-tests/lib/go.sum b/system-tests/lib/go.sum index dab6e81de9d..9a68821a836 100644 --- a/system-tests/lib/go.sum +++ b/system-tests/lib/go.sum @@ -1614,8 +1614,8 @@ github.com/smartcontractkit/chainlink-common v0.11.2-0.20260413191736-29c0bf1edb github.com/smartcontractkit/chainlink-common v0.11.2-0.20260413191736-29c0bf1edbbc/go.mod h1:RnNTmxoheJYec/Gl/9t3wPLtFIHrlYjmWDdwZZJjchw= github.com/smartcontractkit/chainlink-common/keystore v1.0.2 h1:AWisx4JT3QV8tcgh6J5NCrex+wAgTYpWyHsyNPSXzsQ= github.com/smartcontractkit/chainlink-common/keystore v1.0.2/go.mod h1:rSkIHdomyak3YnUtXLenl6poIq8q0V3UZPiiyYqPdGA= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10 h1:FJAFgXS9oqASnkS03RE1HQwYQQxrO4l46O5JSzxqLgg= -github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.10/go.mod h1:oiDa54M0FwxevWwyAX773lwdWvFYYlYHHQV1LQ5HpWY= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20251211140724-319861e514c4 h1:NOUsjsMzNecbjiPWUQGlRSRAutEvCFrqqyETDJeh5q4= +github.com/smartcontractkit/chainlink-common/pkg/chipingress v0.0.11-0.20251211140724-319861e514c4/go.mod h1:Zpvul9sTcZNAZOVzt5vBl1XZGNvQebFpnpn3/KOQvOQ= github.com/smartcontractkit/chainlink-common/pkg/monitoring v0.0.0-20251215152504-b1e41f508340 h1:PsjEI+5jZIz9AS4eOsLS5VpSWJINf38clXV3wryPyMk= github.com/smartcontractkit/chainlink-common/pkg/monitoring v0.0.0-20251215152504-b1e41f508340/go.mod h1:P/0OSXUlFaxxD4B/P6HWbxYtIRmmWGDJAvanq19879c= github.com/smartcontractkit/chainlink-data-streams v0.1.13 h1:YOmt545DW6U0SyaqBf+NTGDLm1yMurVI7yOvxP5hlJk= @@ -1680,6 +1680,8 @@ github.com/smartcontractkit/chainlink-sui/deployment v0.0.0-20260304150206-c64e4 github.com/smartcontractkit/chainlink-sui/deployment v0.0.0-20260304150206-c64e48eb0cb0/go.mod h1:IfeW6t5Yc5293H5ixuooAft+wYBMSFQWKjbBTwYiKr4= github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16 h1:pzrAgF6QFMQLS/kukXenLN87PCa48SEMlE7QvJxTOHs= github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16/go.mod h1:BALK9cj8sk12e15UF6uDhifHgIApa+6N11TcQfInEro= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2 h1:c8T3cnZXrwOo6YU4A/VNj941mZo6n9VbfY3POP74OoA= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2/go.mod h1:MjUJAyU+kLvLLPRPBs3X7zYadds5umZcjTLHjfNhpUc= github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.15 h1:usf6YCNmSO8R1/rU28wUfIdp7zXlqGGOAttXW5mgkXU= github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.15/go.mod h1:YqrpawYGRkT/jcvXcmaZeZPOtu0erIenrHl5Mb8+U/c= github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.10.0 h1:PWAMYu0WaAMBfbpxCpFJGRIDHmcgmYin6a+UQC0OdtY= diff --git a/system-tests/tests/go.mod b/system-tests/tests/go.mod index edc6e820d42..9a7df391277 100644 --- a/system-tests/tests/go.mod +++ b/system-tests/tests/go.mod @@ -69,6 +69,7 @@ require ( github.com/smartcontractkit/chainlink-protos/ring/go v0.0.0-20260331131315-f08a616d8dcd github.com/smartcontractkit/chainlink-protos/workflows/go v0.0.0-20260323124644-faea187e6997 github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16 + github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2 github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.10.0 github.com/smartcontractkit/chainlink-testing-framework/havoc v1.50.7 github.com/smartcontractkit/chainlink-testing-framework/lib v1.54.5 @@ -613,7 +614,7 @@ require ( github.com/smartcontractkit/chainlink-protos/storage-service v0.3.0 // indirect github.com/smartcontractkit/chainlink-protos/svr v1.1.1-0.20260203131522-bb8bc5c423b3 // indirect github.com/smartcontractkit/chainlink-sui v0.0.0-20260409162403-8bca8f7c9de5 // indirect - github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.18 + github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.18 // indirect github.com/smartcontractkit/chainlink-testing-framework/lib/grafana v1.50.0 // indirect github.com/smartcontractkit/chainlink-testing-framework/parrot v0.6.2 // indirect github.com/smartcontractkit/chainlink-ton v0.0.0-20260410201811-4e2e3b68a59e // indirect diff --git a/system-tests/tests/go.sum b/system-tests/tests/go.sum index 09568405c67..00b529f3a14 100644 --- a/system-tests/tests/go.sum +++ b/system-tests/tests/go.sum @@ -1895,6 +1895,8 @@ github.com/smartcontractkit/chainlink-sui/deployment v0.0.0-20260304150206-c64e4 github.com/smartcontractkit/chainlink-sui/deployment v0.0.0-20260304150206-c64e48eb0cb0/go.mod h1:IfeW6t5Yc5293H5ixuooAft+wYBMSFQWKjbBTwYiKr4= github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16 h1:pzrAgF6QFMQLS/kukXenLN87PCa48SEMlE7QvJxTOHs= github.com/smartcontractkit/chainlink-testing-framework/framework v0.15.16/go.mod h1:BALK9cj8sk12e15UF6uDhifHgIApa+6N11TcQfInEro= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2 h1:c8T3cnZXrwOo6YU4A/VNj941mZo6n9VbfY3POP74OoA= +github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter v1.0.2/go.mod h1:MjUJAyU+kLvLLPRPBs3X7zYadds5umZcjTLHjfNhpUc= github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.18 h1:1ng+p/+85zcVLHB050PiWUAjOcxyd4KjwkUlJy34rgE= github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose v0.1.18/go.mod h1:2+OrSz56pdgtY0Oc20nCS9LH/bEksFDBQjoR82De5PI= github.com/smartcontractkit/chainlink-testing-framework/framework/components/fake v0.10.0 h1:PWAMYu0WaAMBfbpxCpFJGRIDHmcgmYin6a+UQC0OdtY= diff --git a/system-tests/tests/regression/cre/cre_regression_suite_test.go b/system-tests/tests/regression/cre/cre_regression_suite_test.go index 5b5f22f76a4..e4d8aa1145a 100644 --- a/system-tests/tests/regression/cre/cre_regression_suite_test.go +++ b/system-tests/tests/regression/cre/cre_regression_suite_test.go @@ -12,7 +12,6 @@ import ( var ( parallelEnabled = t_helpers.ParallelEnabled() - fanoutEnabled = t_helpers.ChipSinkFanoutEnabled() ) // REGRESSION TESTS target edge cases, negative conditions, etc., all happy path and sanity checks should go to a `smoke` package. @@ -33,7 +32,7 @@ func Test_CRE_V2_Consensus_Regression(t *testing.T) { for _, tCase := range consensusNegativeTestsGenerateReport { testName := fmt.Sprintf(consensusTestNameTemplate, tCase.caseToTrigger, tCase.name) t.Run(testName, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) @@ -47,7 +46,10 @@ func Test_CRE_V2_Cron_Regression(t *testing.T) { for _, tCase := range cronInvalidSchedulesTests { testName := "[v2] Cron (Beholder) fails when schedule is " + tCase.name t.Run(testName, func(t *testing.T) { - testEnv := t_helpers.SetupTestEnvironmentWithConfig(t, t_helpers.GetDefaultTestConfig(t)) + if parallelEnabled { + t.Parallel() + } + testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) CronBeholderFailsWithInvalidScheduleTest(t, testEnv, tCase.invalidSchedule) }) @@ -58,7 +60,7 @@ func Test_CRE_V2_HTTP_Regression(t *testing.T) { for _, tCase := range httpNegativeTests { testName := "[v2] HTTP Trigger fails with " + tCase.name t.Run(testName, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) @@ -75,7 +77,7 @@ func runEVMNegativeTestSuite(t *testing.T, testCases []evmNegativeTest) { for _, tCase := range testCases { testName := fmt.Sprintf(evmTestNameTemplate, tCase.functionToTest, tCase.name) t.Run(testName, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) @@ -160,7 +162,7 @@ func Test_CRE_V2_HTTP_Action_CRUD_Regression(t *testing.T) { for _, tCase := range httpActionFailureTests { testName := "[v2] HTTP Action fails with " + tCase.name t.Run(testName, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) diff --git a/system-tests/tests/regression/cre/v2_consensus_regression_test.go b/system-tests/tests/regression/cre/v2_consensus_regression_test.go index d5f34977573..8dfea4a5694 100644 --- a/system-tests/tests/regression/cre/v2_consensus_regression_test.go +++ b/system-tests/tests/regression/cre/v2_consensus_regression_test.go @@ -1,6 +1,7 @@ package cre import ( + "context" "fmt" "testing" "time" @@ -44,9 +45,10 @@ func ConsensusFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, consensus server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) for _, bcOutput := range testEnv.CreEnvironment.Blockchains { diff --git a/system-tests/tests/regression/cre/v2_cron_beholder_regression_test.go b/system-tests/tests/regression/cre/v2_cron_beholder_regression_test.go index f08912558a0..b1e59988da9 100644 --- a/system-tests/tests/regression/cre/v2_cron_beholder_regression_test.go +++ b/system-tests/tests/regression/cre/v2_cron_beholder_regression_test.go @@ -44,7 +44,6 @@ func CronBeholderFailsWithInvalidScheduleTest(t *testing.T, testEnv *ttypes.Test testLogger.Warn().Msgf("Expecting Cron workflow to fail with invalid schedule: %s", invalidSchedule) expectedBeholderLog := "beholder found engine initialization failure message!" - timeout := 75 * time.Second expectedError := t_helpers.AssertBeholderMessage(listenerCtx, t, expectedBeholderLog, testLogger, messageChan, kafkaErrChan, timeout) require.Error(t, expectedError, "Cron (Beholder) test failed. This test expects to fail with an error, but did not.") diff --git a/system-tests/tests/regression/cre/v2_evm_regression_test.go b/system-tests/tests/regression/cre/v2_evm_regression_test.go index 3f2d11ed798..7a4b61097d2 100644 --- a/system-tests/tests/regression/cre/v2_evm_regression_test.go +++ b/system-tests/tests/regression/cre/v2_evm_regression_test.go @@ -1,6 +1,7 @@ package cre import ( + "context" "fmt" "math/rand" "strconv" @@ -198,9 +199,15 @@ func EVMReadFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, evmNegative server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain( + ctx, + server, + userLogsCh, + baseMessageCh, + ) }) for _, bcOutput := range testEnv.CreEnvironment.Blockchains { @@ -265,9 +272,15 @@ func EVMLogTriggerFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, evmNe server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain( + ctx, + server, + userLogsCh, + baseMessageCh, + ) }) // drain user logs channel in the background, we are not asserting anything on it t_helpers.IgnoreUserLogs(t.Context(), userLogsCh) @@ -364,9 +377,15 @@ func EVMWriteFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, evmNegativ server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain( + ctx, + server, + userLogsCh, + baseMessageCh, + ) }) for _, bcOutput := range testEnv.CreEnvironment.Blockchains { diff --git a/system-tests/tests/regression/cre/v2_http_action_regression_test.go b/system-tests/tests/regression/cre/v2_http_action_regression_test.go index 1de153765f7..67a8a47542f 100644 --- a/system-tests/tests/regression/cre/v2_http_action_regression_test.go +++ b/system-tests/tests/regression/cre/v2_http_action_regression_test.go @@ -1,6 +1,7 @@ package cre import ( + "context" "strings" "testing" "time" @@ -125,9 +126,10 @@ func HTTPActionFailureTest(t *testing.T, testEnv *ttypes.TestEnvironment, httpAc server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) // Now register and deploy the workflow diff --git a/system-tests/tests/regression/cre/v2_http_trigger_regression_test.go b/system-tests/tests/regression/cre/v2_http_trigger_regression_test.go index e3984025901..9a52b8f6d5a 100644 --- a/system-tests/tests/regression/cre/v2_http_trigger_regression_test.go +++ b/system-tests/tests/regression/cre/v2_http_trigger_regression_test.go @@ -2,6 +2,7 @@ package cre import ( "bytes" + "context" "crypto/ecdsa" "encoding/json" "io" @@ -109,9 +110,10 @@ func HTTPTriggerFailsTest(t *testing.T, testEnv *ttypes.TestEnvironment, httpNeg server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) // drain user logs channel in the background, we are not asserting anything on it t_helpers.IgnoreUserLogs(t.Context(), userLogsCh) diff --git a/system-tests/tests/smoke/cre/cre_suite_test.go b/system-tests/tests/smoke/cre/cre_suite_test.go index 6a92d66427e..e039bfbe083 100644 --- a/system-tests/tests/smoke/cre/cre_suite_test.go +++ b/system-tests/tests/smoke/cre/cre_suite_test.go @@ -22,7 +22,6 @@ var v1RegistriesFlags = []string{"--with-contracts-version", "v1"} var ( parallelEnabled = t_helpers.ParallelEnabled() - fanoutEnabled = t_helpers.ChipSinkFanoutEnabled() // topology is used in test names topology = os.Getenv("TOPOLOGY_NAME") ) @@ -152,6 +151,9 @@ func runV2SuiteScenario(t *testing.T, topology string, scenario v2suite_config.S // NOTE: this test is not easily parallelisable, because it uses "real" ChIP Ingress stack // we don't want to plug it into ChIP fanout, at least not yet t.Run("[v2] Cron Beholder - "+topology, func(t *testing.T) { + if parallelEnabled { + t.Parallel() + } testEnv := t_helpers.SetupTestEnvironmentWithConfig(t, t_helpers.GetDefaultTestConfig(t)) ExecuteCronBeholderTest(t, testEnv) }) @@ -165,7 +167,7 @@ func runV2SuiteScenario(t *testing.T, topology string, scenario v2suite_config.S }) case v2suite_config.SuiteScenarioHTTPActionCRUD: t.Run("[v2] HTTP Action CRUD - "+topology, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) @@ -173,7 +175,7 @@ func runV2SuiteScenario(t *testing.T, topology string, scenario v2suite_config.S }) case v2suite_config.SuiteScenarioDONTime: t.Run("[v2] DON Time - "+topology, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) @@ -181,7 +183,7 @@ func runV2SuiteScenario(t *testing.T, topology string, scenario v2suite_config.S }) case v2suite_config.SuiteScenarioConsensus: t.Run("[v2] Consensus - "+topology, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) @@ -194,7 +196,7 @@ func runV2SuiteScenario(t *testing.T, topology string, scenario v2suite_config.S func Test_CRE_V2_EVM_Write_LogTrigger(t *testing.T) { t.Run("[v2] EVM Write - "+topology, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) @@ -203,7 +205,7 @@ func Test_CRE_V2_EVM_Write_LogTrigger(t *testing.T) { }) t.Run("[v2] EVM LogTrigger - "+topology, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } testEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, t_helpers.GetDefaultTestConfig(t)) diff --git a/system-tests/tests/smoke/cre/v2_aptos_capability_test.go b/system-tests/tests/smoke/cre/v2_aptos_capability_test.go index 5a182b0e1d2..b9a9dac67ec 100644 --- a/system-tests/tests/smoke/cre/v2_aptos_capability_test.go +++ b/system-tests/tests/smoke/cre/v2_aptos_capability_test.go @@ -157,7 +157,7 @@ func executeAptosScenarios(t *testing.T, tenv *configuration.TestEnvironment, sc for _, scenario := range scenarios { t.Run(scenario.name, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } @@ -168,9 +168,10 @@ func executeAptosScenarios(t *testing.T, tenv *configuration.TestEnvironment, sc baseMessageCh := make(chan *commonevents.BaseMessage, 1000) server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(lggr, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) scenario.run(t, scenarioEnv, scenarioAptosChain, userLogsCh, baseMessageCh) diff --git a/system-tests/tests/smoke/cre/v2_consensus_capability_test.go b/system-tests/tests/smoke/cre/v2_consensus_capability_test.go index f5a8b5c1d46..e41b3f7389d 100644 --- a/system-tests/tests/smoke/cre/v2_consensus_capability_test.go +++ b/system-tests/tests/smoke/cre/v2_consensus_capability_test.go @@ -1,6 +1,7 @@ package cre import ( + "context" "testing" "time" @@ -22,9 +23,10 @@ func ExecuteConsensusTest(t *testing.T, testEnv *ttypes.TestEnvironment) { server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) workflowName := t_helpers.UniqueWorkflowName(testEnv, "consensustest") diff --git a/system-tests/tests/smoke/cre/v2_dontime_test.go b/system-tests/tests/smoke/cre/v2_dontime_test.go index be323f07437..b4ec0d59a99 100644 --- a/system-tests/tests/smoke/cre/v2_dontime_test.go +++ b/system-tests/tests/smoke/cre/v2_dontime_test.go @@ -1,6 +1,7 @@ package cre import ( + "context" "testing" "time" @@ -26,9 +27,10 @@ func ExecuteDonTimeTest(t *testing.T, testEnv *ttypes.TestEnvironment) { server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) testLogger.Info().Msg("Creating Cron workflow configuration file...") diff --git a/system-tests/tests/smoke/cre/v2_evm_capability_test.go b/system-tests/tests/smoke/cre/v2_evm_capability_test.go index be182de6377..d17c248013e 100644 --- a/system-tests/tests/smoke/cre/v2_evm_capability_test.go +++ b/system-tests/tests/smoke/cre/v2_evm_capability_test.go @@ -67,7 +67,7 @@ func ExecuteEVMReadTestForCases(t *testing.T, testEnv *ttypes.TestEnvironment, t for _, tc := range testCases { t.Run("Read "+tc.String(), func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } @@ -82,9 +82,10 @@ func ExecuteEVMReadTestForCases(t *testing.T, testEnv *ttypes.TestEnvironment, t // `./logs` folder inside `smoke/cre` is uploaded as artifact in GH server := t_helpers.StartChipTestSink(t, t_helpers.GetLoggingPublishFn(lggr, userLogsCh, baseMessageCh, evmReadLogFilePath(t, perCaseEnv))) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) for _, bcOutput := range perCaseEnv.CreEnvironment.Blockchains { @@ -391,9 +392,10 @@ func ExecuteEVMLogTriggerTest(t *testing.T, testEnv *ttypes.TestEnvironment) { server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(lggr, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) for _, bcOutput := range testEnv.CreEnvironment.Blockchains { diff --git a/system-tests/tests/smoke/cre/v2_grpc_source_test.go b/system-tests/tests/smoke/cre/v2_grpc_source_test.go index cd9bd2a95f9..6b1c18ee1ac 100644 --- a/system-tests/tests/smoke/cre/v2_grpc_source_test.go +++ b/system-tests/tests/smoke/cre/v2_grpc_source_test.go @@ -267,8 +267,10 @@ func startWorkflowEventSink(t *testing.T) (context.Context, <-chan proto.Message messageChan := make(chan proto.Message, 1000) server := t_helpers.StartChipTestSink(t, t_helpers.GetWorkflowV2LifecyclePublishFn(framework.L, messageChan)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(messageChan) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, messageChan) }) timeout := 5 * time.Minute diff --git a/system-tests/tests/smoke/cre/v2_http_action_test.go b/system-tests/tests/smoke/cre/v2_http_action_test.go index 280effae9b1..bccbf731330 100644 --- a/system-tests/tests/smoke/cre/v2_http_action_test.go +++ b/system-tests/tests/smoke/cre/v2_http_action_test.go @@ -1,6 +1,7 @@ package cre import ( + "context" "net/http" "testing" "time" @@ -127,9 +128,10 @@ func HTTPActionRegressionTest(t *testing.T, testEnv *ttypes.TestEnvironment, url baseMessageCh := make(chan *commonevents.BaseMessage, 1000) server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) testLogger.Info().Msg("Waiting for HTTP Action regression workflow to complete...") @@ -187,7 +189,7 @@ func ExecuteHTTPActionCRUDSuccessTest(t *testing.T, testEnv *ttypes.TestEnvironm testName := "[v2] HTTP Action " + testCase.name t.Run(testName, func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } // Each case gets its own per-test execution context to avoid shared-signer nonce collisions @@ -222,9 +224,10 @@ func HTTPActionSuccessTest(t *testing.T, testEnv *ttypes.TestEnvironment, httpAc server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) // Wait for workflow execution to complete and verify success diff --git a/system-tests/tests/smoke/cre/v2_sharding_test.go b/system-tests/tests/smoke/cre/v2_sharding_test.go index 0ab68542bfe..b6cd90c078b 100644 --- a/system-tests/tests/smoke/cre/v2_sharding_test.go +++ b/system-tests/tests/smoke/cre/v2_sharding_test.go @@ -328,9 +328,10 @@ func validateShardingScaleScenario(t *testing.T, testEnv *ttypes.TestEnvironment baseMessageCh := make(chan *commonevents.BaseMessage, 1000) server := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(logger, userLogsCh, baseMessageCh)) t.Cleanup(func() { - server.Shutdown(t.Context()) - close(userLogsCh) - close(baseMessageCh) + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, server, userLogsCh, baseMessageCh) }) execTimeout := 3 * time.Minute diff --git a/system-tests/tests/smoke/cre/v2_vault_don_test.go b/system-tests/tests/smoke/cre/v2_vault_don_test.go index 2570ff7d26f..d4277dce631 100644 --- a/system-tests/tests/smoke/cre/v2_vault_don_test.go +++ b/system-tests/tests/smoke/cre/v2_vault_don_test.go @@ -1,6 +1,7 @@ package cre import ( + "context" "encoding/hex" "encoding/json" "math/big" @@ -79,7 +80,7 @@ func ExecuteVaultTest(t *testing.T, testEnv *ttypes.TestEnvironment) { gwURL := gatewayURL.String() t.Run("basic_crud", func(t *testing.T) { - if parallelEnabled && fanoutEnabled { + if parallelEnabled { t.Parallel() } subEnv := t_helpers.SetupTestEnvironmentWithPerTestKeys(t, testEnv.TestConfig) @@ -95,7 +96,12 @@ func ExecuteVaultTest(t *testing.T, testEnv *ttypes.TestEnvironment) { ulCh := make(chan *workflowevents.UserLogs, 1000) bmCh := make(chan *commonevents.BaseMessage, 1000) sink := t_helpers.StartChipTestSink(t, t_helpers.GetPublishFn(testLogger, ulCh, bmCh)) - t.Cleanup(func() { sink.Shutdown(t.Context()); close(ulCh); close(bmCh) }) + t.Cleanup(func() { + // can't use t.Context() here because it will have been cancelled before the cleanup function is called + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + t_helpers.ShutdownChipSinkWithDrain(ctx, sink, ulCh, bmCh) + }) namespaces := []string{"main", "alt"} executeVaultSecretsCreateTest(t, enc, secretID, owner, gwURL, namespaces, sc, wfReg) diff --git a/system-tests/tests/test-helpers/before_suite.go b/system-tests/tests/test-helpers/before_suite.go index 21258778566..ac6a73e896c 100644 --- a/system-tests/tests/test-helpers/before_suite.go +++ b/system-tests/tests/test-helpers/before_suite.go @@ -9,6 +9,7 @@ import ( "os/exec" "path/filepath" "slices" + "strconv" "strings" "sync" "testing" @@ -26,13 +27,14 @@ import ( cldf "github.com/smartcontractkit/chainlink-deployments-framework/deployment" "github.com/smartcontractkit/chainlink-testing-framework/framework" "github.com/smartcontractkit/chainlink-testing-framework/framework/components/blockchain" - chipingressset "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/chip_ingress_set" + ctfchiprouter "github.com/smartcontractkit/chainlink-testing-framework/framework/components/chiprouter" "github.com/smartcontractkit/chainlink-testing-framework/seth" keystone_changeset "github.com/smartcontractkit/chainlink/deployment/keystone/changeset" cldlogger "github.com/smartcontractkit/chainlink/deployment/logger" workflow_registry_v2_wrapper "github.com/smartcontractkit/chainlink-evm/gethwrappers/workflow/generated/workflow_registry_wrapper_v2" + "github.com/smartcontractkit/chainlink/system-tests/lib/cre/chiprouter" crecontracts "github.com/smartcontractkit/chainlink/system-tests/lib/cre/contracts" "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment" "github.com/smartcontractkit/chainlink/system-tests/lib/cre/environment/blockchains" @@ -124,6 +126,7 @@ func getOrCreateSharedEnvironment(t *testing.T, tconf *ttypes.TestConfig, flags entry.once.Do(func() { createEnvironment(t, tconf, flags...) + require.NoError(t, chiprouter.EnsureStarted(t.Context()), "failed to ensure chip ingress router is running") in := getEnvironmentConfig(t) creEnvironment, dons, err := environment.BuildFromSavedState(t.Context(), cldlogger.NewSingleFileLogger(t), in) if err != nil { @@ -295,7 +298,7 @@ func GetTestConfig(t *testing.T, configPath string) *ttypes.TestConfig { EnvironmentDirPath: environmentDirPath, EnvironmentConfigPath: filepath.Join(environmentDirPath, configPath), // change to your desired config, if you want to use another topology EnvironmentStateFile: filepath.Join(environmentDirPath, envconfig.StateDirname, envconfig.LocalCREStateFilename), - ChipIngressGRPCPort: chipingressset.DEFAULT_CHIP_INGRESS_GRPC_PORT, + ChipIngressGRPCPort: strconv.Itoa(ctfchiprouter.DefaultBeholderGRPCPort), } } diff --git a/system-tests/tests/test-helpers/beholder_provider.go b/system-tests/tests/test-helpers/beholder_provider.go index f98e98f550a..978183ee734 100644 --- a/system-tests/tests/test-helpers/beholder_provider.go +++ b/system-tests/tests/test-helpers/beholder_provider.go @@ -7,6 +7,7 @@ import ( "os" "os/exec" "strings" + "sync" "time" "github.com/avast/retry-go/v4" @@ -46,6 +47,8 @@ const ( errBeholderOrConfigNil = "beholder or config is nil" ) +var beholderStartupMu sync.Mutex + type Beholder struct { cfg *config.ChipIngressConfig lggr zerolog.Logger @@ -63,6 +66,9 @@ type ConsumerOptions struct { // NewBeholder creates a Beholder instance, even if it's not already running. func NewBeholder(lggr zerolog.Logger, testConfig *configuration.TestConfig) (*Beholder, error) { + beholderStartupMu.Lock() + defer beholderStartupMu.Unlock() + if err := startBeholderIfNotRunning(testConfig.RelativePathToRepoRoot, testConfig.EnvironmentDirPath, testConfig.ChipIngressGRPCPort); err != nil { return nil, errors.Wrap(err, "Beholder failed to start") } diff --git a/system-tests/tests/test-helpers/chip-testsink/server.go b/system-tests/tests/test-helpers/chip-testsink/server.go index 724dbe9a48f..8360bc92074 100644 --- a/system-tests/tests/test-helpers/chip-testsink/server.go +++ b/system-tests/tests/test-helpers/chip-testsink/server.go @@ -32,6 +32,9 @@ type Config struct { // Started optionally receives a signal once the gRPC listener is bound. Started chan<- struct{} + + // ActualAddr optionally receives the resolved listen address after binding. + ActualAddr chan<- string } // Server implements the ChipIngress gRPC service + a tiny HTTP API. @@ -88,6 +91,7 @@ func (s *Server) Run() error { s.grpcServer.Stop() return err } + notifyAddr(s.cfg.ActualAddr, addr) notifyStarted(s.cfg.Started) if s.cfg.UpstreamEndpoint != "" { @@ -108,7 +112,7 @@ func (s *Server) Run() error { func (s *Server) Publish(ctx context.Context, event *pb.CloudEvent) (*chippb.PublishResponse, error) { go func() { if s.cfg.UpstreamEndpoint != "" { - forwardCtx, cancelFn := context.WithTimeout(context.Background(), 10*time.Second) + forwardCtx, cancelFn := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) defer cancelFn() _, err := s.upstream.Publish(forwardCtx, event) if err != nil { @@ -120,6 +124,33 @@ func (s *Server) Publish(ctx context.Context, event *pb.CloudEvent) (*chippb.Pub return s.cfg.PublishFunc(ctx, event) } +func (s *Server) PublishBatch(ctx context.Context, batch *chippb.CloudEventBatch) (*chippb.PublishResponse, error) { + if batch == nil || len(batch.Events) == 0 { + return &chippb.PublishResponse{}, nil + } + + go func() { + if s.cfg.UpstreamEndpoint == "" { + return + } + + forwardCtx, cancelFn := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancelFn() + _, err := s.upstream.PublishBatch(forwardCtx, batch) + if err != nil { + log.Printf("failed to forward batch to upstream: %v", err) + } + }() + + for _, event := range batch.Events { + if _, err := s.cfg.PublishFunc(ctx, event); err != nil { + return nil, err + } + } + + return &chippb.PublishResponse{}, nil +} + func (s *Server) Shutdown(ctx context.Context) { s.grpcServer.GracefulStop() log.Println("[chip-testsink] Server shutdown") @@ -155,3 +186,14 @@ func notifyStarted(ch chan<- struct{}) { default: } } + +func notifyAddr(ch chan<- string, addr string) { + if ch == nil || addr == "" { + return + } + + select { + case ch <- addr: + default: + } +} diff --git a/system-tests/tests/test-helpers/chip_testsink_helpers.go b/system-tests/tests/test-helpers/chip_testsink_helpers.go index 9e4ed2e8ec2..03d3282c41b 100644 --- a/system-tests/tests/test-helpers/chip_testsink_helpers.go +++ b/system-tests/tests/test-helpers/chip_testsink_helpers.go @@ -3,9 +3,9 @@ package helpers import ( "context" "encoding/json" - "net" "os" "path/filepath" + "reflect" "strings" "sync" "testing" @@ -19,21 +19,181 @@ import ( "google.golang.org/protobuf/proto" chippb "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" - chipingressset "github.com/smartcontractkit/chainlink-testing-framework/framework/components/dockercompose/chip_ingress_set" + "github.com/smartcontractkit/chainlink-testing-framework/framework" commonevents "github.com/smartcontractkit/chainlink-protos/workflows/go/common" workflowevents "github.com/smartcontractkit/chainlink-protos/workflows/go/events" workfloweventsv2 "github.com/smartcontractkit/chainlink-protos/workflows/go/v2" + "github.com/smartcontractkit/chainlink/system-tests/lib/cre/chiprouter" chiptestsink "github.com/smartcontractkit/chainlink/system-tests/tests/test-helpers/chip-testsink" ) const testSinkStartupTimeout = 10 * time.Second +const relativePathToRepoRoot = "../../../../" type ChipSink interface { Shutdown(ctx context.Context) } +type registeredChipSink struct { + server *chiptestsink.Server + subscriberID string + relativePath string +} + +func (s *registeredChipSink) Shutdown(ctx context.Context) { + if s == nil { + return + } + if err := chiprouter.UnregisterSubscriber(ctx, s.subscriberID); err != nil && !os.IsNotExist(err) { + framework.L.Warn().Msgf("failed to unregister chip sink subscriber: %s", err) + } + if s.server != nil { + s.server.Shutdown(ctx) + } +} + +// StartChannelDrainers starts one goroutine per channel and drains messages until stop is called +// or the channel is closed. This is useful during teardown to avoid producer goroutines blocking +// on full channels while infrastructure is shutting down. +func StartChannelDrainers[T any](channels ...<-chan T) func() { + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + + for _, ch := range channels { + if ch == nil { + continue + } + + wg.Add(1) + go func(ch <-chan T) { + defer wg.Done() + for { + select { + case <-ctx.Done(): + return + case _, ok := <-ch: + if !ok { + return + } + } + } + }(ch) + } + + return func() { + cancel() + wg.Wait() + } +} + +// CloseChannels closes all non-nil channels in order. +func CloseChannels[T any](channels ...chan T) { + for _, ch := range channels { + if ch != nil { + close(ch) + } + } +} + +// collectChannels flattens channel arguments and slices/arrays of channels into a single list. +func collectChannels(args ...any) []reflect.Value { + channels := make([]reflect.Value, 0, len(args)) + for _, arg := range args { + v := reflect.ValueOf(arg) + if !v.IsValid() { + continue + } + + switch v.Kind() { + case reflect.Chan: + channels = append(channels, v) + case reflect.Slice, reflect.Array: + for i := 0; i < v.Len(); i++ { + elem := v.Index(i) + if elem.IsValid() && elem.Kind() == reflect.Chan { + channels = append(channels, elem) + } + } + default: + framework.L.Warn().Msgf("unsupported arg to ShutdownChipSinkWithDrain: %T", arg) + } + } + return channels +} + +// startReflectChannelDrainers drains channels until they are closed or stop() is called. +func startReflectChannelDrainers(channels []reflect.Value) func() { + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + doneCh := reflect.ValueOf(ctx.Done()) + + for _, ch := range channels { + if !ch.IsValid() || ch.Kind() != reflect.Chan || ch.IsNil() { + continue + } + + wg.Add(1) + go func(ch reflect.Value) { + defer wg.Done() + cases := []reflect.SelectCase{ + {Dir: reflect.SelectRecv, Chan: ch}, + {Dir: reflect.SelectRecv, Chan: doneCh}, + } + for { + chosen, _, ok := reflect.Select(cases) + if chosen == 1 { + return + } + if !ok { + return + } + } + }(ch) + } + + return func() { + cancel() + wg.Wait() + } +} + +func closeReflectChannels(channels []reflect.Value) { + for _, ch := range channels { + if !ch.IsValid() || ch.Kind() != reflect.Chan || ch.IsNil() { + continue + } + // Only close channels that have send capability. + if ch.Type().ChanDir()&reflect.SendDir == 0 { + continue + } + func(ch reflect.Value) { + defer func() { _ = recover() }() + ch.Close() + }(ch) + } +} + +// ShutdownChipSinkWithDrain performs a safer sink teardown for tests. +// It starts background drainers for provided channels, shuts down the sink, +// stops drainers, and then closes channels. +func ShutdownChipSinkWithDrain( + shutdownCtx context.Context, + sink ChipSink, + channels ...any, +) { + flattenedChannels := collectChannels(channels...) + stopDrainers := startReflectChannelDrainers(flattenedChannels) + + if sink != nil { + sink.Shutdown(shutdownCtx) + } + + stopDrainers() + closeReflectChannels(flattenedChannels) +} + type baseMessageWatchCfg struct { workflowID string labelEq map[string]string @@ -111,25 +271,6 @@ func WithUserLogWorkflowID(workflowID string) UserLogWatchOpt { } } -type fanoutSubscription struct { - id string -} - -func (s *fanoutSubscription) Shutdown(_ context.Context) { - fanoutSubMu.Lock() - defer fanoutSubMu.Unlock() - delete(fanoutSubs, s.id) -} - -var ( - fanoutOnce sync.Once - fanoutServer *chiptestsink.Server - errFanout error - - fanoutSubMu sync.Mutex - fanoutSubs = make(map[string]chiptestsink.PublishFn) -) - func safeSendUserLogs(ch chan *workflowevents.UserLogs, msg *workflowevents.UserLogs) { // In fanout mode, tests may close their log channels immediately after // unsubscribing during cleanup. An in-flight publish can race with that close, @@ -159,58 +300,6 @@ func safeSendProtoMessage(ch chan proto.Message, msg proto.Message) { ch <- msg } -func ChipSinkFanoutEnabled() bool { - v := strings.TrimSpace(strings.ToLower(os.Getenv("CRE_TEST_CHIP_SINK_FANOUT_ENABLED"))) - return v == "1" || v == "true" || v == "yes" -} - -func ensureFanoutServer(t *testing.T) { - t.Helper() - - fanoutOnce.Do(func() { - grpcListenAddr := ":" + chipingressset.DEFAULT_CHIP_INGRESS_GRPC_PORT - startCh := make(chan struct{}, 1) - fanoutServer, errFanout = chiptestsink.NewServer(chiptestsink.Config{ - GRPCListen: grpcListenAddr, - Started: startCh, - PublishFunc: func(ctx context.Context, event *pb.CloudEvent) (*chippb.PublishResponse, error) { - fanoutSubMu.Lock() - snapshot := make([]chiptestsink.PublishFn, 0, len(fanoutSubs)) - for _, fn := range fanoutSubs { - snapshot = append(snapshot, fn) - } - fanoutSubMu.Unlock() - - for _, fn := range snapshot { - if _, err := fn(ctx, event); err != nil { - // Best-effort delivery: one subscriber must not fail all. - continue - } - } - return &chippb.PublishResponse{}, nil - }, - }) - if errFanout != nil { - return - } - - errCh := make(chan error, 1) - go func() { - errCh <- fanoutServer.Run() - }() - - select { - case <-startCh: - case err := <-errCh: - errFanout = err - case <-time.After(testSinkStartupTimeout): - errFanout = errors.New("timeout waiting for fanout sink server to start") - } - }) - - require.NoError(t, errFanout, "failed to start fanout sink server") -} - // WaitForUserLog monitors workflow user logs until one contains needle or the context ends. func WaitForUserLog( ctx context.Context, @@ -509,34 +598,18 @@ func GetLoggingPublishFn( } } -// StartChipTestSink boots the CHiP test sink and waits until it is accepting traffic. -// In fanout mode (CRE_TEST_CHIP_SINK_FANOUT_ENABLED=1), a singleton sink is started and each test -// registers its own publish function as a fanout subscriber. +// StartChipTestSink boots a per-test CHiP sink on an ephemeral port and registers it with the +// shared chip ingress router, which owns the default ingress port. func StartChipTestSink(t *testing.T, publishFn chiptestsink.PublishFn) ChipSink { - if ChipSinkFanoutEnabled() { - ensureFanoutServer(t) - subID := t.Name() + "-" + time.Now().Format("150405.000000000") - fanoutSubMu.Lock() - fanoutSubs[subID] = publishFn - fanoutSubMu.Unlock() - return &fanoutSubscription{id: subID} - } - - grpcListenAddr := ":" + chipingressset.DEFAULT_CHIP_INGRESS_GRPC_PORT - if !isPortAvailable(grpcListenAddr) { - t.Fatalf(`failed to start ChIP Ingress Test Sink. Port %s is already taken. Most probably an instance of ChIP Ingress is already running. -If you want to use both together start ChIP Ingress on a different port with '--grpc-port' flag -and make sure that the sink is pointing to correct upstream endpoint ('localhost:' in most cases)`, chipingressset.DEFAULT_CHIP_INGRESS_GRPC_PORT) - } - startCh := make(chan struct{}, 1) - server, err := chiptestsink.NewServer(chiptestsink.Config{ + addrCh := make(chan string, 1) + server, sErr := chiptestsink.NewServer(chiptestsink.Config{ PublishFunc: publishFn, - GRPCListen: grpcListenAddr, - Started: startCh, // signals that server is indeed listening on the GRPC port - // UpstreamEndpoint: "localhost:50052", // uncomment to forward events to ChIP, remember to start ChIP on a different port config.DefaultChipIngressPort (=50051) + GRPCListen: "0.0.0.0:0", + Started: startCh, + ActualAddr: addrCh, }) - require.NoError(t, err, "failed to create new test sink server") + require.NoError(t, sErr, "failed to create new test sink server") errCh := make(chan error, 1) go func() { @@ -551,17 +624,24 @@ and make sure that the sink is pointing to correct upstream endpoint ('localhost require.FailNow(t, "timeout waiting for test sink server to start") } - return server -} + var actualAddr string + select { + case actualAddr = <-addrCh: + case <-time.After(testSinkStartupTimeout): + server.Shutdown(t.Context()) + require.FailNow(t, "timeout waiting for test sink listen address") + } + + require.NoError(t, chiprouter.EnsureStarted(t.Context()), "failed to ensure chip ingress router is running") + + subscriberID, err := chiprouter.RegisterSubscriber(t.Context(), t.Name(), actualAddr) + require.NoError(t, err, "failed to register test sink with chip ingress router") -func isPortAvailable(addr string) bool { - lc := net.ListenConfig{} - l, err := lc.Listen(context.Background(), "tcp", addr) - if err != nil { - return false // already in use or permission denied + return ®isteredChipSink{ + server: server, + subscriberID: subscriberID, + relativePath: relativePathToRepoRoot, } - _ = l.Close() - return true } // WatchWorkflowLogs enforces that the expected log appears before timeout and that poison logs abort the test.