Update k8s executor to follow specified routes when applying manifests

pkg/app/piped/executor/kubernetes/applier_group.go

@@ -0,0 +1,92 @@
+// Copyright 2022 The PipeCD Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kubernetes
+
+import (
+	"fmt"
+
+	"go.uber.org/zap"
+
+	provider "github.com/pipe-cd/pipecd/pkg/app/piped/cloudprovider/kubernetes"
+	"github.com/pipe-cd/pipecd/pkg/config"
+	"github.com/pipe-cd/pipecd/pkg/model"
+)
+
+type applierGetter interface {
+	Get(k provider.ResourceKey) (provider.Applier, error)
+}
+
+type applierGroup struct {
+	resourceRoutes []config.KubernetesResourceRoute
+	appliers       map[string]provider.Applier
+	defaultApplier provider.Applier
+}
+
+func newApplierGroup(defaultProvider string, appCfg config.KubernetesApplicationSpec, pipedCfg *config.PipedSpec, logger *zap.Logger) (*applierGroup, error) {
+	cp, ok := pipedCfg.FindCloudProvider(defaultProvider, model.ApplicationKind_KUBERNETES)
+	if !ok {
+		return nil, fmt.Errorf("provider %s was not found", defaultProvider)
+	}
+
+	defaultApplier := provider.NewApplier(
+		appCfg.Input,
+		*cp.KubernetesConfig,
+		logger,
+	)
+	d := &applierGroup{
+		resourceRoutes: appCfg.ResourceRoutes,
+		appliers:       map[string]provider.Applier{defaultProvider: defaultApplier},
+		defaultApplier: defaultApplier,
+	}
+
+	for _, r := range appCfg.ResourceRoutes {
+		if _, ok := d.appliers[r.Provider]; ok {
+			continue
+		}
+
+		cp, ok := pipedCfg.FindCloudProvider(r.Provider, model.ApplicationKind_KUBERNETES)
+		if !ok {
+			return nil, fmt.Errorf("provider %s specified in resourceRoutes was not found", r.Provider)
+		}
+
+		d.appliers[r.Provider] = provider.NewApplier(appCfg.Input, *cp.KubernetesConfig, logger)
+	}
+
+	return d, nil
+}
+
+// TODO: Add test for this applierGroup function.
+func (d applierGroup) Get(rk provider.ResourceKey) (provider.Applier, error) {
+	for _, r := range d.resourceRoutes {
+		if r.Match == nil {
+			if a, ok := d.appliers[r.Provider]; ok {
+				return a, nil
+			}
+			return nil, fmt.Errorf("provider %s specified in resourceRoutes was not found", r.Provider)
+		}
+		if k := r.Match.Kind; k != "" && k != rk.Kind {
+			continue
+		}
+		if n := r.Match.Name; n != "" && n != rk.Name {
+			continue
+		}
+		if a, ok := d.appliers[r.Provider]; ok {
+			return a, nil
+		}
+		return nil, fmt.Errorf("provider %s specified in resourceRoutes was not found", r.Provider)
+	}
+
+	return d.defaultApplier, nil
+}

pkg/app/piped/executor/kubernetes/baseline.go

@@ -85,7 +85,7 @@ func (e *deployExecutor) ensureBaselineRollout(ctx context.Context) model.StageS
 
 	// Start rolling out the resources for BASELINE variant.
 	e.LogPersister.Info("Start rolling out BASELINE variant...")
-	if err := applyManifests(ctx, e.applier, baselineManifests, e.appCfg.Input.Namespace, e.LogPersister); err != nil {
+	if err := applyManifests(ctx, e.applierGetter, baselineManifests, e.appCfg.Input.Namespace, e.LogPersister); err != nil {
 		return model.StageStatus_STAGE_FAILURE
 	}
 
@@ -101,7 +101,7 @@ func (e *deployExecutor) ensureBaselineClean(ctx context.Context) model.StageSta
 	}
 
 	resources := strings.Split(value, ",")
-	if err := removeBaselineResources(ctx, e.applier, resources, e.LogPersister); err != nil {
+	if err := removeBaselineResources(ctx, e.applierGetter, resources, e.LogPersister); err != nil {
 		e.LogPersister.Errorf("Unable to remove baseline resources: %v", err)
 		return model.StageStatus_STAGE_FAILURE
 	}
@@ -157,7 +157,7 @@ func (e *deployExecutor) generateBaselineManifests(manifests []provider.Manifest
 	return baselineManifests, nil
 }
 
-func removeBaselineResources(ctx context.Context, applier provider.Applier, resources []string, lp executor.LogPersister) error {
+func removeBaselineResources(ctx context.Context, ag applierGetter, resources []string, lp executor.LogPersister) error {
 	if len(resources) == 0 {
 		return nil
 	}
@@ -181,13 +181,13 @@ func removeBaselineResources(ctx context.Context, applier provider.Applier, reso
 
 	// We delete the service first to close all incoming connections.
 	lp.Info("Starting finding and deleting service resources of BASELINE variant")
-	if err := deleteResources(ctx, applier, serviceKeys, lp); err != nil {
+	if err := deleteResources(ctx, ag, serviceKeys, lp); err != nil {
 		return err
 	}
 
 	// Next, delete all workloads.
 	lp.Info("Starting finding and deleting workload resources of BASELINE variant")
-	if err := deleteResources(ctx, applier, workloadKeys, lp); err != nil {
+	if err := deleteResources(ctx, ag, workloadKeys, lp); err != nil {
 		return err
 	}
 

pkg/app/piped/executor/kubernetes/canary.go

@@ -102,7 +102,7 @@ func (e *deployExecutor) ensureCanaryRollout(ctx context.Context) model.StageSta
 
 	// Start rolling out the resources for CANARY variant.
 	e.LogPersister.Info("Start rolling out CANARY variant...")
-	if err := applyManifests(ctx, e.applier, canaryManifests, e.appCfg.Input.Namespace, e.LogPersister); err != nil {
+	if err := applyManifests(ctx, e.applierGetter, canaryManifests, e.appCfg.Input.Namespace, e.LogPersister); err != nil {
 		return model.StageStatus_STAGE_FAILURE
 	}
 
@@ -118,7 +118,7 @@ func (e *deployExecutor) ensureCanaryClean(ctx context.Context) model.StageStatu
 	}
 
 	resources := strings.Split(value, ",")
-	if err := removeCanaryResources(ctx, e.applier, resources, e.LogPersister); err != nil {
+	if err := removeCanaryResources(ctx, e.applierGetter, resources, e.LogPersister); err != nil {
 		e.LogPersister.Errorf("Unable to remove canary resources: %v", err)
 		return model.StageStatus_STAGE_FAILURE
 	}
@@ -187,7 +187,7 @@ func (e *deployExecutor) generateCanaryManifests(manifests []provider.Manifest,
 	return canaryManifests, nil
 }
 
-func removeCanaryResources(ctx context.Context, applier provider.Applier, resources []string, lp executor.LogPersister) error {
+func removeCanaryResources(ctx context.Context, ag applierGetter, resources []string, lp executor.LogPersister) error {
 	if len(resources) == 0 {
 		return nil
 	}
@@ -211,13 +211,13 @@ func removeCanaryResources(ctx context.Context, applier provider.Applier, resour
 
 	// We delete the service first to close all incoming connections.
 	lp.Info("Starting finding and deleting service resources of CANARY variant")
-	if err := deleteResources(ctx, applier, serviceKeys, lp); err != nil {
+	if err := deleteResources(ctx, ag, serviceKeys, lp); err != nil {
 		return err
 	}
 
 	// Next, delete all workloads.
 	lp.Info("Starting finding and deleting workload resources of CANARY variant")
-	if err := deleteResources(ctx, applier, workloadKeys, lp); err != nil {
+	if err := deleteResources(ctx, ag, workloadKeys, lp); err != nil {
 		return err
 	}
 

pkg/app/piped/executor/kubernetes/canary_test.go

@@ -175,11 +175,13 @@ func TestEnsureCanaryRollout(t *testing.T) {
 					}, nil)
 					return p
 				}(),
-				applier: func() provider.Applier {
-					p := kubernetestest.NewMockApplier(ctrl)
-					p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(fmt.Errorf("error"))
-					return p
-				}(),
+				applierGetter: &applierGroup{
+					defaultApplier: func() provider.Applier {
+						p := kubernetestest.NewMockApplier(ctrl)
+						p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(fmt.Errorf("error"))
+						return p
+					}(),
+				},
 				appCfg: &config.KubernetesApplicationSpec{},
 			},
 		},
@@ -228,11 +230,13 @@ func TestEnsureCanaryRollout(t *testing.T) {
 					}, nil)
 					return p
 				}(),
-				applier: func() provider.Applier {
-					p := kubernetestest.NewMockApplier(ctrl)
-					p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
-					return p
-				}(),
+				applierGetter: &applierGroup{
+					defaultApplier: func() provider.Applier {
+						p := kubernetestest.NewMockApplier(ctrl)
+						p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
+						return p
+					}(),
+				},
 				appCfg: &config.KubernetesApplicationSpec{},
 			},
 		},

pkg/app/piped/executor/kubernetes/kubernetes.go

@@ -39,8 +39,8 @@ type deployExecutor struct {
 	commit string
 	appCfg *config.KubernetesApplicationSpec
 
-	loader  provider.Loader
-	applier provider.Applier
+	loader        provider.Loader
+	applierGetter applierGetter
 }
 
 type registerer interface {
@@ -94,9 +94,9 @@ func (e *deployExecutor) Execute(sig executor.StopSignal) model.StageStatus {
 		}
 	}
 
-	cp, ok := e.PipedConfig.FindCloudProvider(e.Deployment.CloudProvider, model.ApplicationKind_KUBERNETES)
-	if !ok {
-		e.LogPersister.Errorf("Not found cloud provider %q", e.Deployment.CloudProvider)
+	e.applierGetter, err = newApplierGroup(e.Deployment.CloudProvider, *e.appCfg, e.PipedConfig, e.Logger)
+	if err != nil {
+		e.LogPersister.Error(err.Error())
 		return model.StageStatus_STAGE_FAILURE
 	}
 
@@ -109,11 +109,7 @@ func (e *deployExecutor) Execute(sig executor.StopSignal) model.StageStatus {
 		e.GitClient,
 		e.Logger,
 	)
-	e.applier = provider.NewApplier(
-		e.appCfg.Input,
-		*cp.KubernetesConfig,
-		e.Logger,
-	)
+
 	e.Logger.Info("start executing kubernetes stage",
 		zap.String("stage-name", e.Stage.Name),
 		zap.String("app-dir", ds.AppDir),
@@ -226,13 +222,20 @@ func addBuiltinAnnotations(manifests []provider.Manifest, variantLabel, variant,
 	}
 }
 
-func applyManifests(ctx context.Context, applier provider.Applier, manifests []provider.Manifest, namespace string, lp executor.LogPersister) error {
+func applyManifests(ctx context.Context, ag applierGetter, manifests []provider.Manifest, namespace string, lp executor.LogPersister) error {
 	if namespace == "" {
 		lp.Infof("Start applying %d manifests", len(manifests))
 	} else {
 		lp.Infof("Start applying %d manifests to %q namespace", len(manifests), namespace)
 	}
+
 	for _, m := range manifests {
+		applier, err := ag.Get(m.Key)
+		if err != nil {
+			lp.Error(err.Error())
+			return err
+		}
+
 		annotation := m.GetAnnotations()[provider.LabelSyncReplace]
 		if annotation != provider.UseReplaceEnabled {
 			if err := applier.ApplyManifest(ctx, m); err != nil {
@@ -244,7 +247,7 @@ func applyManifests(ctx context.Context, applier provider.Applier, manifests []p
 		}
 		// Always try to replace first and create if it fails due to resource not found error.
 		// This is because we cannot know whether resource already exists before executing command.
-		err := applier.ReplaceManifest(ctx, m)
+		err = applier.ReplaceManifest(ctx, m)
 		if errors.Is(err, provider.ErrNotFound) {
 			lp.Infof("Specified resource does not exist, so create the resource: %s (%w)", m.Key.ReadableLogString(), err)
 			err = applier.CreateManifest(ctx, m)
@@ -260,7 +263,7 @@ func applyManifests(ctx context.Context, applier provider.Applier, manifests []p
 	return nil
 }
 
-func deleteResources(ctx context.Context, applier provider.Applier, resources []provider.ResourceKey, lp executor.LogPersister) error {
+func deleteResources(ctx context.Context, ag applierGetter, resources []provider.ResourceKey, lp executor.LogPersister) error {
 	resourcesLen := len(resources)
 	if resourcesLen == 0 {
 		lp.Info("No resources to delete")
@@ -271,7 +274,13 @@ func deleteResources(ctx context.Context, applier provider.Applier, resources []
 	var deletedCount int
 
 	for _, k := range resources {
-		err := applier.Delete(ctx, k)
+		applier, err := ag.Get(k)
+		if err != nil {
+			lp.Error(err.Error())
+			return err
+		}
+
+		err = applier.Delete(ctx, k)
 		if err == nil {
 			lp.Successf("- deleted resource: %s", k.ReadableLogString())
 			deletedCount++

pkg/app/piped/executor/kubernetes/kubernetes_test.go

@@ -470,7 +470,8 @@ spec:
 			ctx := context.Background()
 			manifests, err := provider.ParseManifests(tc.manifest)
 			require.NoError(t, err)
-			err = applyManifests(ctx, tc.applier, manifests, tc.namespace, &fakeLogPersister{})
+			ag := &applierGroup{defaultApplier: tc.applier}
+			err = applyManifests(ctx, ag, manifests, tc.namespace, &fakeLogPersister{})
 			assert.Equal(t, tc.wantErr, err != nil)
 		})
 	}
@@ -538,7 +539,8 @@ func TestDeleteResources(t *testing.T) {
 	for _, tc := range testcases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx := context.Background()
-			err := deleteResources(ctx, tc.applier, tc.resources, &fakeLogPersister{})
+			ag := &applierGroup{defaultApplier: tc.applier}
+			err := deleteResources(ctx, ag, tc.resources, &fakeLogPersister{})
 			assert.Equal(t, tc.wantErr, err != nil)
 		})
 	}

pkg/app/piped/executor/kubernetes/primary.go

@@ -133,7 +133,7 @@ func (e *deployExecutor) ensurePrimaryRollout(ctx context.Context) model.StageSt
 
 	// Start applying all manifests to add or update running resources.
 	e.LogPersister.Info("Start rolling out PRIMARY variant...")
-	if err := applyManifests(ctx, e.applier, primaryManifests, e.appCfg.Input.Namespace, e.LogPersister); err != nil {
+	if err := applyManifests(ctx, e.applierGetter, primaryManifests, e.appCfg.Input.Namespace, e.LogPersister); err != nil {
 		return model.StageStatus_STAGE_FAILURE
 	}
 	e.LogPersister.Success("Successfully rolled out PRIMARY variant")
@@ -171,7 +171,7 @@ func (e *deployExecutor) ensurePrimaryRollout(ctx context.Context) model.StageSt
 
 	// Start deleting all running resources that are not defined in Git.
 	e.LogPersister.Infof("Start deleting %d resources", len(removeKeys))
-	if err := deleteResources(ctx, e.applier, removeKeys, e.LogPersister); err != nil {
+	if err := deleteResources(ctx, e.applierGetter, removeKeys, e.LogPersister); err != nil {
 		return model.StageStatus_STAGE_FAILURE
 	}
 

pkg/app/piped/executor/kubernetes/primary_test.go

@@ -137,11 +137,13 @@ func TestEnsurePrimaryRollout(t *testing.T) {
 					}, nil)
 					return p
 				}(),
-				applier: func() provider.Applier {
-					p := kubernetestest.NewMockApplier(ctrl)
-					p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
-					return p
-				}(),
+				applierGetter: &applierGroup{
+					defaultApplier: func() provider.Applier {
+						p := kubernetestest.NewMockApplier(ctrl)
+						p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
+						return p
+					}(),
+				},
 				appCfg: &config.KubernetesApplicationSpec{},
 			},
 		},
@@ -190,12 +192,14 @@ func TestEnsurePrimaryRollout(t *testing.T) {
 					}, nil)
 					return p
 				}(),
-				applier: func() provider.Applier {
-					p := kubernetestest.NewMockApplier(ctrl)
-					p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
-					p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
-					return p
-				}(),
+				applierGetter: &applierGroup{
+					defaultApplier: func() provider.Applier {
+						p := kubernetestest.NewMockApplier(ctrl)
+						p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
+						p.EXPECT().ApplyManifest(gomock.Any(), gomock.Any()).Return(nil)
+						return p
+					}(),
+				},
 				appCfg: &config.KubernetesApplicationSpec{
 					Service: config.K8sResourceReference{
 						Kind: "Service",