Pass DeployTarget in SDK

pkg/app/pipedv1/plugin/kubernetes/deployment/rollback.go

@@ -83,9 +83,9 @@
 	}
 
 	// Get the deploy target config.
-	targets, err := input.GetDeployment().GetDeployTargets(a.pluginConfig.Name)
-	if err != nil {
-		lp.Errorf("Failed while finding deploy target config (%v)", err)
+	targets := input.GetDeployment().GetDeployTargets(a.pluginConfig.Name)
+	if len(targets) == 0 {
+		lp.Errorf("No deploy target was found for the plugin %s", a.pluginConfig.Name)
 		return model.StageStatus_STAGE_FAILURE
 	}
 

pkg/app/pipedv1/plugin/kubernetes/deployment/sync.go

@@ -82,9 +82,9 @@
 	}
 
 	// Get the deploy target config.
-	targets, err := input.GetDeployment().GetDeployTargets(a.pluginConfig.Name)
-	if err != nil {
-		lp.Errorf("Failed while finding deploy target config (%v)", err)
+	targets := input.GetDeployment().GetDeployTargets(a.pluginConfig.Name)
+	if len(targets) == 0 {
+		lp.Errorf("No deploy target was found for the plugin %s", a.pluginConfig.Name)
 		return model.StageStatus_STAGE_FAILURE
 	}
 	deployTargetConfig, err := kubeconfig.FindDeployTarget(a.pluginConfig, targets[0]) // TODO: consider multiple targets

pkg/model/deployment.go

@@ -214,13 +214,13 @@ func (d *Deployment) SetUpdatedAt(t int64) {
 	d.UpdatedAt = t
 }
 
-func (d *Deployment) GetDeployTargets(pluginName string) ([]string, error) {
+func (d *Deployment) GetDeployTargets(pluginName string) []string {
 	dps, ok := d.GetDeployTargetsByPlugin()[pluginName]
 	if !ok || len(dps.GetDeployTargets()) == 0 {
-		return nil, fmt.Errorf("deploy target not found for plugin %v", pluginName)
+		return []string{}
 	}
 
-	return dps.GetDeployTargets(), nil
+	return dps.GetDeployTargets()
 }
 
 // Implement sort.Interface for PipelineStages.

pkg/model/deployment_test.go

@@ -658,3 +658,68 @@ func TestSortPipelineStagesByIndex(t *testing.T) {
 		{Index: 4},
 	}, stages)
 }
+
+func TestDeployment_GetDeployTargets(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name       string
+		deployment *Deployment
+		pluginName string
+		expected   []string
+	}{
+		{
+			name: "plugin with deploy targets",
+			deployment: &Deployment{
+				DeployTargetsByPlugin: map[string]*DeployTargets{
+					"pluginA": {
+						DeployTargets: []string{"target1", "target2"},
+					},
+				},
+			},
+			pluginName: "pluginA",
+			expected:   []string{"target1", "target2"},
+		},
+		{
+			name: "plugin without deploy targets",
+			deployment: &Deployment{
+				DeployTargetsByPlugin: map[string]*DeployTargets{
+					"pluginA": {
+						DeployTargets: []string{},
+					},
+				},
+			},
+			pluginName: "pluginA",
+			expected:   []string{},
+		},
+		{
+			name: "plugin not found",
+			deployment: &Deployment{
+				DeployTargetsByPlugin: map[string]*DeployTargets{
+					"pluginA": {
+						DeployTargets: []string{"target1", "target2"},
+					},
+				},
+			},
+			pluginName: "pluginB",
+			expected:   []string{},
+		},
+		{
+			name: "no deploy targets by plugin",
+			deployment: &Deployment{
+				DeployTargetsByPlugin: map[string]*DeployTargets{},
+			},
+			pluginName: "pluginA",
+			expected:   []string{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+
+			got := tt.deployment.GetDeployTargets(tt.pluginName)
+			assert.Equal(t, tt.expected, got)
+		})
+	}
+}

pkg/plugin/sdk/deployment.go

@@ -206,7 +206,34 @@
 		stageID:       request.GetInput().GetStage().GetId(),
 		logPersister:  lp,
 	}
-	return executeStage(ctx, s.base, &s.config, nil, client, request, s.logger) // TODO: pass the deployTargets
+
+	// Get the deploy targets set on the deployment from the piped plugin config.
+	dtNames := request.GetInput().GetDeployment().GetDeployTargets(s.commonFields.config.Name)
+	deployTargets := make([]*DeployTarget[DeployTargetConfig], 0, len(dtNames))
+	for _, name := range dtNames {
+		dt := s.commonFields.config.FindDeployTarget(name)
+		if dt == nil {
+			continue
+		}
+
+		// TODO: cache the unmarshaled config to avoid unmarshaling it multiple times.
+		var sdkDt DeployTargetConfig
+		if err := json.Unmarshal(dt.Config, &sdkDt); err != nil {
+			return nil, status.Errorf(codes.Internal, "failed to unmarshal deploy target config: %v", err)
+		}
+
+		deployTargets = append(deployTargets, &DeployTarget[DeployTargetConfig]{
+			Name:   name,
+			Labels: dt.Labels,
+			Config: sdkDt,
+		})
+	}
+
+	if len(dtNames) != len(deployTargets) {
+		return nil, status.Errorf(codes.Internal, "the number of deploy targets in the piped plugin config should be the same as the ones set on the deployment: in the piped config = %d, in the deployment= %d", len(dtNames), len(deployTargets))
+	}
+
+	return executeStage(ctx, s.base, &s.config, deployTargets, client, request, s.logger)
 }
 
 // StagePluginServiceServer is the gRPC server that handles requests from the piped.