DPA-1556 Implement distribute LLO jobs. (#17082)

* Implement distribute LLO jobs.

* Fix lint.

* Fix lint.

* Configurator address is now defined by the migration, not the changeset.

* Proposing jobs emits the proposed jobs in ChangesetOutput.

Author: ro-tex

deployment/data-streams/changeset/common.go

@@ -0,0 +1,68 @@
+package changeset
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	chainsel "github.com/smartcontractkit/chain-selectors"
+
+	"github.com/smartcontractkit/chainlink-protos/job-distributor/v1/job"
+
+	"github.com/smartcontractkit/chainlink/deployment"
+)
+
+const (
+	defaultJobSpecsTimeout = 120 * time.Second
+)
+
+func chainAndAddresses(e deployment.Environment, chainSel uint64) (chainID string, addresses map[string]deployment.TypeAndVersion, err error) {
+	chainID, err = chainsel.GetChainIDFromSelector(chainSel)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to get chain ID from selector: %w", err)
+	}
+
+	addresses, err = e.ExistingAddresses.AddressesForChain(chainSel)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to get existing addresses: %w", err)
+	}
+	return chainID, addresses, nil
+}
+
+// proposeAllOrNothing proposes all jobs in the list and if any of them fail, it will revoke all already made proposals.
+// TODO We need a test here.
+func proposeAllOrNothing(ctx context.Context, oc deployment.OffchainClient, prs []*job.ProposeJobRequest) (proposedJobs []deployment.ProposedJob, err error) {
+	var proposals []*job.ProposeJobResponse
+	var p *job.ProposeJobResponse
+	for _, pr := range prs {
+		p, err = oc.ProposeJob(ctx, pr)
+		if err != nil {
+			break
+		}
+		proposedJobs = append(proposedJobs, deployment.ProposedJob{
+			JobID: p.Proposal.JobId,
+			Node:  pr.NodeId,
+			Spec:  pr.Spec,
+		})
+		proposals = append(proposals, p)
+	}
+
+	if err != nil {
+		// There's an error, so we need to revoke all proposals we just made.
+		var errs []error
+		for _, pr := range proposals {
+			if _, errRevoke := oc.RevokeJob(ctx, &job.RevokeJobRequest{
+				IdOneof: &job.RevokeJobRequest_Id{Id: pr.Proposal.JobId},
+			}); errRevoke != nil {
+				errs = append(errs, fmt.Errorf("failed to revoke job %s: %w", pr.Proposal.JobId, errRevoke))
+			}
+		}
+		// If we got any errors while trying to cancel, we need to return them, so we know we sent some jobs twice.
+		if len(errs) > 0 {
+			err = errors.Join(err, errors.Join(errs...))
+		}
+	}
+
+	return proposedJobs, err
+}

deployment/data-streams/changeset/jd_distribute_bootstrap_jobs.go

@@ -3,67 +3,43 @@ package changeset
 import (
 	"context"
 	"fmt"
-	"time"
-
-	chainsel "github.com/smartcontractkit/chain-selectors"
 
 	jobv1 "github.com/smartcontractkit/chainlink-protos/job-distributor/v1/job"
 	"github.com/smartcontractkit/chainlink-protos/job-distributor/v1/shared/ptypes"
 
 	"github.com/smartcontractkit/chainlink/deployment"
 	"github.com/smartcontractkit/chainlink/deployment/data-streams/jd"
 	"github.com/smartcontractkit/chainlink/deployment/data-streams/jobs"
-)
-
-const (
-	defaultBootstrapJobSpecsTimeout = 120 * time.Second
+	"github.com/smartcontractkit/chainlink/deployment/data-streams/utils"
 )
 
 var _ deployment.ChangeSetV2[CsDistributeBootstrapJobSpecsConfig] = CsDistributeBootstrapJobSpecs{}
 
 type CsDistributeBootstrapJobSpecsConfig struct {
-	ChainSelectorEVM uint64
-	Filter           *jd.ListFilter
-}
-
-func findConfiguratorAddressByDON(addresses map[string]deployment.TypeAndVersion, donID uint64) (string, error) {
-	for address, contract := range addresses {
-		if contract.Type == "Configurator" && contract.Labels.Contains(fmt.Sprintf("don-%d", donID)) {
-			return address, nil
-		}
-	}
-	return "", fmt.Errorf("Configurator contract not found for DON %d", donID)
+	ChainSelectorEVM    uint64
+	Filter              *jd.ListFilter
+	ConfiguratorAddress string
 }
 
 type CsDistributeBootstrapJobSpecs struct{}
 
 func (CsDistributeBootstrapJobSpecs) Apply(e deployment.Environment, cfg CsDistributeBootstrapJobSpecsConfig) (deployment.ChangesetOutput, error) {
-	ctx, cancel := context.WithTimeout(e.GetContext(), defaultBootstrapJobSpecsTimeout)
+	ctx, cancel := context.WithTimeout(e.GetContext(), defaultJobSpecsTimeout)
 	defer cancel()
 
-	bootstrapNodes, err := jd.FetchDONBootstrappersFromJD(ctx, e.Offchain, cfg.Filter)
-	if err != nil {
-		return deployment.ChangesetOutput{}, fmt.Errorf("failed to get workflow don nodes: %w", err)
-	}
-
-	chainID, err := chainsel.GetChainIDFromSelector(cfg.ChainSelectorEVM)
-	if err != nil {
-		return deployment.ChangesetOutput{}, fmt.Errorf("failed to get chain ID from selector: %w", err)
-	}
-
-	addresses, err := e.ExistingAddresses.AddressesForChain(cfg.ChainSelectorEVM)
-	if err != nil {
-		return deployment.ChangesetOutput{}, fmt.Errorf("failed to get existing addresses: %w", err)
-	}
-
-	// Search for the Configurator address in the address book by DON ID
-	configuratorAddress, err := findConfiguratorAddressByDON(addresses, cfg.Filter.DONID)
+	chainID, _, err := chainAndAddresses(e, cfg.ChainSelectorEVM)
 	if err != nil {
 		return deployment.ChangesetOutput{}, err
 	}
 
+	// Add a label to the job spec to identify the related DON
+	labels := append([]*ptypes.Label(nil),
+		&ptypes.Label{
+			Key: utils.DonIdentifier(cfg.Filter.DONID, cfg.Filter.DONName),
+		})
+
 	bootstrapSpec := jobs.NewBootstrapSpec(
-		configuratorAddress,
+		cfg.ConfiguratorAddress,
 		cfg.Filter.DONID,
 		jobs.RelayTypeEVM,
 		jobs.RelayConfig{
@@ -76,28 +52,27 @@ func (CsDistributeBootstrapJobSpecs) Apply(e deployment.Environment, cfg CsDistr
 		return deployment.ChangesetOutput{}, fmt.Errorf("failed to marshal bootstrap spec: %w", err)
 	}
 
-	// Add a label to the job spec to identify the related DON
-	labels := append([]*ptypes.Label(nil),
-		&ptypes.Label{
-			Key: fmt.Sprintf("don-%d-%s", cfg.Filter.DONID, cfg.Filter.DONName),
-		})
+	bootstrapNodes, err := jd.FetchDONBootstrappersFromJD(ctx, e.Offchain, cfg.Filter)
+	if err != nil {
+		return deployment.ChangesetOutput{}, fmt.Errorf("failed to get workflow don nodes: %w", err)
+	}
 
-	// TODO: For now the implementation uses a very simple approach, in case of partial failures there is a risk
-	// of sending the same job spec multiple times to the same node. We need to understand the implications of this
-	// and decide if we need to implement a more complex approach.
+	var proposals []*jobv1.ProposeJobRequest
 	for _, node := range bootstrapNodes {
-		_, err = e.Offchain.ProposeJob(ctx, &jobv1.ProposeJobRequest{
+		proposals = append(proposals, &jobv1.ProposeJobRequest{
 			NodeId: node.Id,
 			Spec:   string(renderedSpec),
 			Labels: labels,
 		})
-
-		if err != nil {
-			return deployment.ChangesetOutput{}, fmt.Errorf("failed to propose job: %w", err)
-		}
+	}
+	proposedJobs, err := proposeAllOrNothing(ctx, e.Offchain, proposals)
+	if err != nil {
+		return deployment.ChangesetOutput{}, fmt.Errorf("failed to propose all bootstrap jobs: %w", err)
 	}
 
-	return deployment.ChangesetOutput{}, nil
+	return deployment.ChangesetOutput{
+		Jobs: proposedJobs,
+	}, nil
 }
 
 func (f CsDistributeBootstrapJobSpecs) VerifyPreconditions(e deployment.Environment, config CsDistributeBootstrapJobSpecsConfig) error {

deployment/data-streams/changeset/jd_distribute_bootstrap_jobs_test.go

@@ -12,7 +12,6 @@ import (
 )
 
 func TestDistributeBootstrapJobSpecs(t *testing.T) {
-	t.Skip("Flaky Test: https://smartcontract-it.atlassian.net/browse/DX-196")
 	t.Parallel()
 
 	e := testutil.NewMemoryEnv(t, false, 1)
@@ -21,7 +20,8 @@ func TestDistributeBootstrapJobSpecs(t *testing.T) {
 	chainSelector := e.AllChainSelectors()[0]
 
 	// insert a Configurator address for the given DON
-	err := e.ExistingAddresses.Save(chainSelector, "0x4170ed0880ac9a755fd29b2688956bd959f923f4",
+	configuratorAddr := "0x4170ed0880ac9a755fd29b2688956bd959f923f4"
+	err := e.ExistingAddresses.Save(chainSelector, configuratorAddr,
 		deployment.TypeAndVersion{
 			Type:    "Configurator",
 			Version: deployment.Version1_0_0,
@@ -37,6 +37,7 @@ func TestDistributeBootstrapJobSpecs(t *testing.T) {
 			EnvLabel: "env",
 			Size:     0,
 		},
+		ConfiguratorAddress: configuratorAddr,
 	}
 
 	tests := []struct {
@@ -66,7 +67,7 @@ func TestDistributeBootstrapJobSpecs(t *testing.T) {
 				require.NoError(t, err)
 			} else {
 				require.Error(t, err)
-				require.Contains(t, err.Error(), tt.wantErr)
+				require.Contains(t, err.Error(), *tt.wantErr)
 			}
 		})
 	}