Add force-sync-by-replace annotation option

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

@@ -240,27 +240,48 @@ func applyManifests(ctx context.Context, ag applierGetter, manifests []provider.
 			return err
 		}
 
-		annotation := m.GetAnnotations()[provider.LabelSyncReplace]
-		if annotation != provider.UseReplaceEnabled {
-			if err := applier.ApplyManifest(ctx, m); err != nil {
-				lp.Errorf("Failed to apply manifest: %s (%w)", m.Key.ReadableString(), err)
+		// The force annotation has higher priority, so we need to check the annotation in the following order:
+		// 1. force-sync-by-replace
+		// 2. sync-by-replace
+		// 3. others
+		if annotation := m.GetAnnotations()[provider.LabelForceSyncReplace]; annotation == provider.UseReplaceEnabled {
+			// 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.ForceReplaceManifest(ctx, m)
+			if errors.Is(err, provider.ErrNotFound) {
+				lp.Infof("Specified resource does not exist, so create the resource: %s (%w)", m.Key.ReadableString(), err)
+				err = applier.CreateManifest(ctx, m)
+			}
+			if err != nil {
+				lp.Errorf("Failed to forcefully replace or create manifest: %s (%w)", m.Key.ReadableString(), err)
 				return err
 			}
-			lp.Successf("- applied manifest: %s", m.Key.ReadableString())
+			lp.Successf("- forcefully replaced or created manifest: %s", m.Key.ReadableString())
 			continue
 		}
-		// 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)
-		if errors.Is(err, provider.ErrNotFound) {
-			lp.Infof("Specified resource does not exist, so create the resource: %s (%w)", m.Key.ReadableString(), err)
-			err = applier.CreateManifest(ctx, m)
+
+		if annotation := m.GetAnnotations()[provider.LabelSyncReplace]; annotation == provider.UseReplaceEnabled {
+			// 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)
+			if errors.Is(err, provider.ErrNotFound) {
+				lp.Infof("Specified resource does not exist, so create the resource: %s (%w)", m.Key.ReadableString(), err)
+				err = applier.CreateManifest(ctx, m)
+			}
+			if err != nil {
+				lp.Errorf("Failed to replace or create manifest: %s (%w)", m.Key.ReadableString(), err)
+				return err
+			}
+			lp.Successf("- replaced or created manifest: %s", m.Key.ReadableString())
+			continue
 		}
-		if err != nil {
-			lp.Errorf("Failed to replace or create manifest: %s (%w)", m.Key.ReadableString(), err)
+
+		if err := applier.ApplyManifest(ctx, m); err != nil {
+			lp.Errorf("Failed to apply manifest: %s (%w)", m.Key.ReadableString(), err)
 			return err
 		}
-		lp.Successf("- replaced or created manifest: %s", m.Key.ReadableString())
+		lp.Successf("- applied manifest: %s", m.Key.ReadableString())
+		continue
 
 	}
 	lp.Successf("Successfully applied %d manifests", len(manifests))

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

@@ -355,6 +355,32 @@ spec:
     metadata:
       labels:
         app: simple
+`,
+			namespace: "",
+			wantErr:   true,
+		},
+		{
+			name: "unable to force replace manifest",
+			applier: func() provider.Applier {
+				p := kubernetestest.NewMockApplier(ctrl)
+				p.EXPECT().ForceReplaceManifest(gomock.Any(), gomock.Any()).Return(errors.New("unexpected error"))
+				return p
+			}(),
+			manifest: `
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: simple
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+spec:
+  selector:
+    matchLabels:
+      app: simple
+  template:
+    metadata:
+      labels:
+        app: simple
 `,
 			namespace: "",
 			wantErr:   true,
@@ -382,6 +408,33 @@ spec:
     metadata:
       labels:
         app: simple
+`,
+			namespace: "",
+			wantErr:   true,
+		},
+		{
+			name: "unable to create manifest",
+			applier: func() provider.Applier {
+				p := kubernetestest.NewMockApplier(ctrl)
+				p.EXPECT().ForceReplaceManifest(gomock.Any(), gomock.Any()).Return(provider.ErrNotFound)
+				p.EXPECT().CreateManifest(gomock.Any(), gomock.Any()).Return(errors.New("unexpected error"))
+				return p
+			}(),
+			manifest: `
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: simple
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+spec:
+  selector:
+    matchLabels:
+      app: simple
+  template:
+    metadata:
+      labels:
+        app: simple
 `,
 			namespace: "",
 			wantErr:   true,
@@ -432,6 +485,32 @@ spec:
     metadata:
       labels:
         app: simple
+`,
+			namespace: "",
+			wantErr:   false,
+		},
+		{
+			name: "successfully force replace manifest",
+			applier: func() provider.Applier {
+				p := kubernetestest.NewMockApplier(ctrl)
+				p.EXPECT().ForceReplaceManifest(gomock.Any(), gomock.Any()).Return(nil)
+				return p
+			}(),
+			manifest: `
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: simple
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+spec:
+  selector:
+    matchLabels:
+      app: simple
+  template:
+    metadata:
+      labels:
+        app: simple
 `,
 			namespace: "",
 			wantErr:   false,
@@ -459,6 +538,33 @@ spec:
     metadata:
       labels:
         app: simple
+`,
+			namespace: "",
+			wantErr:   false,
+		},
+		{
+			name: "successfully force create manifest",
+			applier: func() provider.Applier {
+				p := kubernetestest.NewMockApplier(ctrl)
+				p.EXPECT().ForceReplaceManifest(gomock.Any(), gomock.Any()).Return(provider.ErrNotFound)
+				p.EXPECT().CreateManifest(gomock.Any(), gomock.Any()).Return(nil)
+				return p
+			}(),
+			manifest: `
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: simple
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+spec:
+  selector:
+    matchLabels:
+      app: simple
+  template:
+    metadata:
+      labels:
+        app: simple
 `,
 			namespace: "",
 			wantErr:   false,

pkg/app/piped/platformprovider/kubernetes/applier.go

@@ -33,6 +33,8 @@
 	CreateManifest(ctx context.Context, manifest Manifest) error
 	// ReplaceManifest does replacing resource from given manifest.
 	ReplaceManifest(ctx context.Context, manifest Manifest) error
+	// ForceReplaceManifest does force replacing resource from given manifest.
+	ForceReplaceManifest(ctx context.Context, manifest Manifest) error
 	// Delete deletes the given resource from Kubernetes cluster.
 	Delete(ctx context.Context, key ResourceKey) error
 }
@@ -137,6 +139,32 @@
 	return err
 }
 
+// ForceReplaceManifest uses kubectl to forcefully replace the given manifests.
+func (a *applier) ForceReplaceManifest(ctx context.Context, manifest Manifest) error {
+	a.initOnce.Do(func() {
+		a.kubectl, a.initErr = a.findKubectl(ctx, a.getToolVersionToRun())
+	})
+	if a.initErr != nil {
+		return a.initErr
+	}
+
+	err := a.kubectl.ForceReplace(
+		ctx,
+		a.platformProvider.KubeConfigPath,
+		a.getNamespaceToRun(manifest.Key),
+		manifest,
+	)
+	if err == nil {
+		return nil
+	}
+
+	if errors.Is(err, errorReplaceNotFound) {
+		return ErrNotFound
+	}
+
+	return err
+}
+
 // Delete deletes the given resource from Kubernetes cluster.
 // If the resource key is different, this returns ErrNotFound.
 func (a *applier) Delete(ctx context.Context, k ResourceKey) (err error) {
@@ -237,6 +265,15 @@
 	return nil
 }
 
+func (a *multiApplier) ForceReplaceManifest(ctx context.Context, manifest Manifest) error {
+	for _, a := range a.appliers {
+		if err := a.ForceReplaceManifest(ctx, manifest); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (a *multiApplier) Delete(ctx context.Context, key ResourceKey) error {
 	for _, a := range a.appliers {
 		if err := a.Delete(ctx, key); err != nil {

pkg/app/piped/platformprovider/kubernetes/kubectl.go

@@ -159,6 +159,45 @@
 	return fmt.Errorf("failed to replace: %s (%w)", string(out), err)
 }
 
+func (c *Kubectl) ForceReplace(ctx context.Context, kubeconfig, namespace string, manifest Manifest) (err error) {
+	defer func() {
+		kubernetesmetrics.IncKubectlCallsCounter(
+			c.version,
+			kubernetesmetrics.LabelReplaceCommand,
+			err == nil,
+		)
+	}()
+
+	data, err := manifest.YamlBytes()
+	if err != nil {
+		return err
+	}
+
+	args := make([]string, 0, 7)
+	if kubeconfig != "" {
+		args = append(args, "--kubeconfig", kubeconfig)
+	}
+	if namespace != "" {
+		args = append(args, "--namespace", namespace)
+	}
+	args = append(args, "replace", "--force", "-f", "-")
+
+	cmd := exec.CommandContext(ctx, c.execPath, args...)
+	r := bytes.NewReader(data)
+	cmd.Stdin = r
+
+	out, err := cmd.CombinedOutput()
+	if err == nil {
+		return nil
+	}
+
+	if strings.Contains(string(out), errorNotFoundLiteral) {
+		return errorReplaceNotFound
+	}
+
+	return fmt.Errorf("failed to replace: %s (%w)", string(out), err)
+}
+
 func (c *Kubectl) Delete(ctx context.Context, kubeconfig, namespace string, r ResourceKey) (err error) {
 	defer func() {
 		kubernetesmetrics.IncKubectlCallsCounter(

pkg/app/piped/platformprovider/kubernetes/kubernetes.go

@@ -31,6 +31,7 @@ const (
 	LabelOriginalAPIVersion   = "pipecd.dev/original-api-version"   // The api version defined in git configuration. e.g. apps/v1
 	LabelIgnoreDriftDirection = "pipecd.dev/ignore-drift-detection" // Whether the drift detection should ignore this resource.
 	LabelSyncReplace          = "pipecd.dev/sync-by-replace"        // Use replace instead of apply.
+	LabelForceSyncReplace     = "pipecd.dev/force-sync-by-replace"  // Use replace --force instead of apply.
 	LabelServerSideApply      = "pipecd.dev/server-side-apply"      // Use server side apply instead of client side apply.
 	AnnotationConfigHash      = "pipecd.dev/config-hash"            // The hash value of all mouting config resources.
 	AnnotationOrder           = "pipecd.dev/order"                  // The order number of resource used to sort them before using.