Add functionalities for applying k8s manifests (#5359)

* Add applier interface for manifest operations in Kubernetes deployment Signed-off-by: Shinnosuke Sawada-Dazai <[email protected]> * Add new error and labels for sync operations in Kubernetes provider Signed-off-by: Shinnosuke Sawada-Dazai <[email protected]> * Implement applyManifests function for kubernetes Signed-off-by: Shinnosuke Sawada-Dazai <[email protected]> * Add unit tests for applyManifests function Signed-off-by: Shinnosuke Sawada-Dazai <[email protected]> --------- Signed-off-by: Shinnosuke Sawada-Dazai <[email protected]>
pipe-cd · Nov 26, 2024 · 839c3d8 · 839c3d8
1 parent 44dd631
commit 839c3d8
Show file tree

Hide file tree

Showing 4 changed files with 391 additions and 1 deletion.
diff --git a/pkg/app/pipedv1/plugin/kubernetes/deployment/apply.go b/pkg/app/pipedv1/plugin/kubernetes/deployment/apply.go
@@ -0,0 +1,79 @@
+// Copyright 2024 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 deployment
+
+import (
+	"context"
+	"errors"
+
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/kubernetes/provider"
+	"github.com/pipe-cd/pipecd/pkg/plugin/logpersister"
+)
+
+func applyManifests(ctx context.Context, applier applier, manifests []provider.Manifest, namespace string, lp logpersister.StageLogPersister) 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 {
+		// 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.Body.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("- forcefully replaced or created manifest: %s", m.Key.ReadableString())
+			continue
+		}
+
+		if annotation := m.Body.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 := applier.ApplyManifest(ctx, m); err != nil {
+			lp.Errorf("Failed to apply manifest: %s (%w)", m.Key.ReadableString(), err)
+			return err
+		}
+		lp.Successf("- applied manifest: %s", m.Key.ReadableString())
+		continue
+
+	}
+	lp.Successf("Successfully applied %d manifests", len(manifests))
+	return nil
+}
diff --git a/pkg/app/pipedv1/plugin/kubernetes/deployment/apply_test.go b/pkg/app/pipedv1/plugin/kubernetes/deployment/apply_test.go
@@ -0,0 +1,290 @@
+// Copyright 2024 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 deployment
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin/kubernetes/provider"
+)
+
+type mockStageLogPersister struct {
+	logs      []string
+	completed bool
+}
+
+func (m *mockStageLogPersister) Write(log []byte) (int, error) {
+	m.logs = append(m.logs, string(log))
+	return len(log), nil
+}
+
+func (m *mockStageLogPersister) Info(log string) {
+	m.logs = append(m.logs, log)
+}
+
+func (m *mockStageLogPersister) Infof(format string, a ...interface{}) {
+	m.logs = append(m.logs, fmt.Sprintf(format, a...))
+}
+
+func (m *mockStageLogPersister) Success(log string) {
+	m.logs = append(m.logs, log)
+}
+
+func (m *mockStageLogPersister) Successf(format string, a ...interface{}) {
+	m.logs = append(m.logs, fmt.Sprintf(format, a...))
+}
+
+func (m *mockStageLogPersister) Error(log string) {
+	m.logs = append(m.logs, log)
+}
+
+func (m *mockStageLogPersister) Errorf(format string, a ...interface{}) {
+	m.logs = append(m.logs, fmt.Sprintf(format, a...))
+}
+
+func (m *mockStageLogPersister) Complete(timeout time.Duration) error {
+	m.completed = true
+	return nil
+}
+
+type mockApplier struct {
+	applyErr        error
+	forceReplaceErr error
+	replaceErr      error
+	createErr       error
+}
+
+func (m *mockApplier) ApplyManifest(ctx context.Context, manifest provider.Manifest) error {
+	return m.applyErr
+}
+
+func (m *mockApplier) ForceReplaceManifest(ctx context.Context, manifest provider.Manifest) error {
+	return m.forceReplaceErr
+}
+
+func (m *mockApplier) ReplaceManifest(ctx context.Context, manifest provider.Manifest) error {
+	return m.replaceErr
+}
+
+func (m *mockApplier) CreateManifest(ctx context.Context, manifest provider.Manifest) error {
+	return m.createErr
+}
+
+func TestApplyManifests(t *testing.T) {
+	tests := []struct {
+		name      string
+		manifests []provider.Manifest
+		namespace string
+		applier   *mockApplier
+		wantErr   bool
+	}{
+		{
+			name: "apply manifests successfully",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				applyErr: nil,
+			},
+			wantErr: false,
+		},
+		{
+			name: "force replace manifests successfully",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				forceReplaceErr: nil,
+			},
+			wantErr: false,
+		},
+		{
+			name: "replace manifests successfully",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				replaceErr: nil,
+			},
+			wantErr: false,
+		},
+		{
+			name: "apply manifests with error",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				applyErr: errors.New("apply error"),
+			},
+			wantErr: true,
+		},
+		{
+			name: "force replace manifests with error",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				forceReplaceErr: errors.New("force replace error"),
+			},
+			wantErr: true,
+		},
+		{
+			name: "replace manifests with error",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				replaceErr: errors.New("replace error"),
+			},
+			wantErr: true,
+		},
+		{
+			name: "create manifests successfully after force replace not found",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				forceReplaceErr: provider.ErrNotFound,
+				createErr:       nil,
+			},
+			wantErr: false,
+		},
+		{
+			name: "create manifests successfully after replace not found",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				replaceErr: provider.ErrNotFound,
+				createErr:  nil,
+			},
+			wantErr: false,
+		},
+		{
+			name: "create manifests with error after force replace not found",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/force-sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				forceReplaceErr: provider.ErrNotFound,
+				createErr:       errors.New("create error"),
+			},
+			wantErr: true,
+		},
+		{
+			name: "create manifests with error after replace not found",
+			manifests: mustParseManifests(t, `
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-config
+  annotations:
+    pipecd.dev/sync-by-replace: "enabled"
+data:
+  key: value
+`),
+			namespace: "",
+			applier: &mockApplier{
+				replaceErr: provider.ErrNotFound,
+				createErr:  errors.New("create error"),
+			},
+			wantErr: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			lp := new(mockStageLogPersister)
+			err := applyManifests(context.Background(), tt.applier, tt.manifests, tt.namespace, lp)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("applyManifests() error = %v, wantErr %v", err, tt.wantErr)
+			}
+		})
+	}
+}
diff --git a/pkg/app/pipedv1/plugin/kubernetes/deployment/server.go b/pkg/app/pipedv1/plugin/kubernetes/deployment/server.go
@@ -39,6 +39,17 @@ type loader interface {
 	LoadManifests(ctx context.Context, input provider.LoaderInput) ([]provider.Manifest, error)
 }
 
+type applier interface {
+	// ApplyManifest does applying the given manifest.
+	ApplyManifest(ctx context.Context, manifest provider.Manifest) error
+	// CreateManifest does creating resource from given manifest.
+	CreateManifest(ctx context.Context, manifest provider.Manifest) error
+	// ReplaceManifest does replacing resource from given manifest.
+	ReplaceManifest(ctx context.Context, manifest provider.Manifest) error
+	// ForceReplaceManifest does force replacing resource from given manifest.
+	ForceReplaceManifest(ctx context.Context, manifest provider.Manifest) error
+}
+
 type DeploymentService struct {
 	deployment.UnimplementedDeploymentServiceServer