Implement resource pruning in k8s plugin

pkg/app/pipedv1/plugin/kubernetes/config/application.go

@@ -36,6 +36,9 @@ type KubernetesApplicationSpec struct {
 	// Input for Kubernetes deployment such as kubectl version, helm version, manifests filter...
 	Input KubernetesDeploymentInput `json:"input"`
 
+	// Configuration for quick sync.
+	QuickSync K8sSyncStageOptions `json:"quickSync"`
+
 	// Which resources should be considered as the Workload of application.
 	// Empty means all Deployments.
 	// e.g.
@@ -100,6 +103,14 @@ type KubernetesDeployTargetConfig struct {
 	KubectlVersion string `json:"kubectlVersion"`
 }
 
+// K8sSyncStageOptions contains all configurable values for a K8S_SYNC stage.
+type K8sSyncStageOptions struct {
+	// Whether the PRIMARY variant label should be added to manifests if they were missing.
+	AddVariantLabelToSelector bool `json:"addVariantLabelToSelector"`
+	// Whether the resources that are no longer defined in Git should be removed or not.
+	Prune bool `json:"prune"`
+}
+
 // FindDeployTarget finds the deploy target configuration by the given name.
 func FindDeployTarget(cfg *config.PipedPlugin, name string) (KubernetesDeployTargetConfig, error) {
 	if cfg == nil {

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

@@ -32,11 +32,11 @@ func annotateConfigHash(manifests []provider.Manifest) error {
 	secrets := make(map[string]provider.Manifest)
 	for _, m := range manifests {
 		if m.IsConfigMap() {
-			configMaps[m.Key().Name()] = m
+			configMaps[m.Name()] = m
 			continue
 		}
 		if m.IsSecret() {
-			secrets[m.Key().Name()] = m
+			secrets[m.Name()] = m
 		}
 	}
 

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

@@ -73,10 +73,10 @@ func determineVersions(manifests []provider.Manifest) ([]*model.ArtifactVersion,
 func findManifests(kind, name string, manifests []provider.Manifest) []provider.Manifest {
 	out := make([]provider.Manifest, 0, len(manifests))
 	for _, m := range manifests {
-		if m.Key().Kind() != kind {
+		if m.Kind() != kind {
 			continue
 		}
-		if name != "" && m.Key().Name() != name {
+		if name != "" && m.Name() != name {
 			continue
 		}
 		out = append(out, m)
@@ -186,7 +186,7 @@ func determineStrategy(olds, news []provider.Manifest, workloadRefs []config.K8s
 			if msg, changed := checkImageChange(templateDiffs); changed {
 				return model.SyncStrategy_PIPELINE, msg
 			}
-			return model.SyncStrategy_PIPELINE, fmt.Sprintf("Sync progressively because pod template of workload %s was changed", w.New.Key().Name())
+			return model.SyncStrategy_PIPELINE, fmt.Sprintf("Sync progressively because pod template of workload %s was changed", w.New.Name())
 		}
 	}
 
@@ -203,14 +203,14 @@ func determineStrategy(olds, news []provider.Manifest, workloadRefs []config.K8s
 	for k, oc := range oldConfigs {
 		nc, ok := newConfigs[k]
 		if !ok {
-			return model.SyncStrategy_PIPELINE, fmt.Sprintf("Sync progressively because %s %s was deleted", oc.Key().Kind(), oc.Key().Name())
+			return model.SyncStrategy_PIPELINE, fmt.Sprintf("Sync progressively because %s %s was deleted", oc.Kind(), oc.Name())
 		}
 		result, err := provider.Diff(oc, nc, logger)
 		if err != nil {
 			return model.SyncStrategy_PIPELINE, fmt.Sprintf("Sync progressively due to an error while calculating the diff (%v)", err)
 		}
 		if result.HasDiff() {
-			return model.SyncStrategy_PIPELINE, fmt.Sprintf("Sync progressively because %s %s was updated", oc.Key().Kind(), oc.Key().Name())
+			return model.SyncStrategy_PIPELINE, fmt.Sprintf("Sync progressively because %s %s was updated", oc.Kind(), oc.Name())
 		}
 	}
 

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

@@ -17,6 +17,8 @@ package deployment
 import (
 	"cmp"
 	"context"
+	"errors"
+	"fmt"
 	"time"
 
 	kubeconfig "github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/kubernetes/config"
@@ -266,6 +268,9 @@ func (a *DeploymentService) executeK8sSyncStage(ctx context.Context, lp logpersi
 
 	// Add variant annotations to all manifests.
 	for i := range manifests {
+		manifests[i].AddLabels(map[string]string{
+			variantLabel: primaryVariant,
+		})
 		manifests[i].AddAnnotations(map[string]string{
 			variantLabel: primaryVariant,
 		})
@@ -290,17 +295,85 @@ func (a *DeploymentService) executeK8sSyncStage(ctx context.Context, lp logpersi
 		return model.StageStatus_STAGE_FAILURE
 	}
 
+	// Create the kubectl wrapper for the target cluster.
+	kubectl := provider.NewKubectl(kubectlPath)
+
 	// Create the applier for the target cluster.
-	applier := provider.NewApplier(provider.NewKubectl(kubectlPath), cfg.Spec.Input, deployTargetConfig, a.logger)
+	applier := provider.NewApplier(kubectl, cfg.Spec.Input, deployTargetConfig, a.logger)
 
 	// Start applying all manifests to add or update running resources.
 	if err := applyManifests(ctx, applier, manifests, cfg.Spec.Input.Namespace, lp); err != nil {
 		lp.Errorf("Failed while applying manifests (%v)", err)
 		return model.StageStatus_STAGE_FAILURE
 	}
 
-	// TODO: implement prune resources
+	// TODO: treat the stage options specified under "with"
+	if !cfg.Spec.QuickSync.Prune {
+		lp.Info("Resource GC was skipped because sync.prune was not configured")
+		return model.StageStatus_STAGE_SUCCESS
+	}
+
+	// Wait for all applied manifests to be stable.
+	// In theory, we don't need to wait for them to be stable before going to the next step
+	// but waiting for a while reduces the number of Kubernetes changes in a short time.
+	lp.Info("Waiting for the applied manifests to be stable")
+	select {
+	case <-time.After(15 * time.Second):
+		break
+	case <-ctx.Done():
+		break
+	}
+
+	lp.Info("Start finding all running resources but no longer defined in Git")
+
+	namespacedLiveResources, err := kubectl.GetAll(ctx, deployTargetConfig.KubeConfigPath,
+		"",
+		fmt.Sprintf("%s=%s", provider.LabelManagedBy, provider.ManagedByPiped),
+		fmt.Sprintf("%s=%s", provider.LabelApplication, input.GetDeployment().GetApplicationId()),
+	)
+	if err != nil {
+		lp.Errorf("Failed while listing all resources (%v)", err)
+		return model.StageStatus_STAGE_FAILURE
+	}
+
+	clusterLiveResources, err := kubectl.GetAllClusterScoped(ctx, deployTargetConfig.KubeConfigPath,
+		fmt.Sprintf("%s=%s", provider.LabelManagedBy, provider.ManagedByPiped),
+		fmt.Sprintf("%s=%s", provider.LabelApplication, input.GetDeployment().GetApplicationId()),
+	)
+	if err != nil {
+		lp.Errorf("Failed while listing all cluster-scoped resources (%v)", err)
+		return model.StageStatus_STAGE_FAILURE
+	}
+
+	if len(namespacedLiveResources)+len(clusterLiveResources) == 0 {
+		lp.Info("There is no data about live resource so no resource will be removed")
+		return model.StageStatus_STAGE_SUCCESS
+	}
+
+	lp.Successf("Successfully loaded %d live resources", len(namespacedLiveResources)+len(clusterLiveResources))
+
+	removeKeys := provider.FindRemoveResources(manifests, namespacedLiveResources, clusterLiveResources)
+	if len(removeKeys) == 0 {
+		lp.Info("There are no live resources should be removed")
+		return model.StageStatus_STAGE_SUCCESS
+	}
+
+	lp.Infof("Start pruning %d resources", len(removeKeys))
+	var deletedCount int
+	for _, key := range removeKeys {
+		if err := kubectl.Delete(ctx, deployTargetConfig.KubeConfigPath, key.Namespace(), key); err != nil {
+			if errors.Is(err, provider.ErrNotFound) {
+				lp.Infof("Specified resource does not exist, so skip deleting the resource: %s (%v)", key.ReadableString(), err)
+				continue
+			}
+			lp.Errorf("Failed while deleting resource %s (%v)", key.ReadableString(), err)
+			continue // continue to delete other resources
+		}
+		deletedCount++
+		lp.Successf("- deleted resource: %s", key.ReadableString())
+	}
 
+	lp.Successf("Successfully deleted %d resources", deletedCount)
 	return model.StageStatus_STAGE_SUCCESS
 }