feat - object etcd migration

cmd/migrate.go

@@ -0,0 +1,55 @@
+// Copyright 2024 The prometheus-operator 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 cmd
+
+import (
+	"fmt"
+
+	"github.com/prometheus-operator/poctl/internal/etcdmigrate"
+	"github.com/prometheus-operator/poctl/internal/k8sutil"
+	"github.com/prometheus-operator/poctl/internal/log"
+	"github.com/spf13/cobra"
+)
+
+// migrateCmd represents the etcd store objects migration command.
+var migrateCmd = &cobra.Command{
+	Use:   "migrate",
+	Short: "Automatically update Custom Resources to the latest storage version.",
+	Long:  `The migrate command in poctl automates the process of updating Kubernetes Custom Resources to the latest storage version. This is essential when upgrading a CRD that supports multiple API versions.`,
+	RunE:  runMigration,
+}
+
+func init() {
+	rootCmd.AddCommand(migrateCmd)
+}
+
+func runMigration(cmd *cobra.Command, _ []string) error {
+	logger, err := log.NewLogger()
+	if err != nil {
+		return fmt.Errorf("error while creating logger: %v", err)
+	}
+	clientSets, err := k8sutil.GetClientSets(kubeconfig)
+	if err != nil {
+		logger.Error("error while getting client sets", "err", err)
+		return err
+	}
+
+	if err := etcdmigrate.MigrateCRDs(cmd.Context(), clientSets); err != nil {
+		logger.Error("error while updating etcd store", "err", err)
+	}
+
+	logger.Info("Prometheus Operator CRD were update in etcd store ")
+	return nil
+}

internal/etcdmigrate/etcdmigrate.go

@@ -0,0 +1,106 @@
+// Copyright 2024 The prometheus-operator 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 etcdmigrate
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/prometheus-operator/poctl/internal/k8sutil"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+)
+
+func MigrateCRDs(ctx context.Context, clientSets *k8sutil.ClientSets) error {
+	crds, err := clientSets.APIExtensionsClient.ApiextensionsV1().CustomResourceDefinitions().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to list CRDs: %w", err)
+	}
+
+	for _, crd := range crds.Items {
+		if crd.Spec.Group != "monitoring.coreos.com" {
+			continue
+		}
+
+		var storageVersion string
+		for _, version := range crd.Spec.Versions {
+			if version.Storage {
+				storageVersion = version.Name
+				break
+			}
+		}
+		if storageVersion == "" {
+			continue
+		}
+
+		crdResourceVersion := schema.GroupVersionResource{
+			Group:    crd.Spec.Group,
+			Version:  storageVersion,
+			Resource: crd.Spec.Names.Plural,
+		}
+
+		namespaces, err := clientSets.KClient.CoreV1().Namespaces().List(ctx, metav1.ListOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to list Namespaces %v", err)
+		}
+		for _, namespace := range namespaces.Items {
+			ns := namespace.Name
+
+			customResourcesInstances, err := clientSets.DClient.Resource(crdResourceVersion).Namespace(ns).List(ctx, metav1.ListOptions{})
+			if err != nil {
+				continue
+			}
+
+			for _, cri := range customResourcesInstances.Items {
+				name := cri.GetName()
+				apiVersion := cri.GetAPIVersion()
+
+				expectedAPIVersion := fmt.Sprintf("%s/%s", crd.Spec.Group, storageVersion)
+				if apiVersion == expectedAPIVersion {
+					continue
+				}
+
+				crdJSON, err := json.Marshal(cri.Object)
+				if err != nil {
+					continue
+				}
+
+				var unstructuredeObject map[string]interface{}
+				if err := json.Unmarshal(crdJSON, &unstructuredeObject); err != nil {
+					continue
+				}
+
+				unstructuredeObject["apiVersion"] = expectedAPIVersion
+
+				updatedStorageObject := &unstructured.Unstructured{Object: unstructuredeObject}
+
+				_, err = clientSets.DClient.Resource(crdResourceVersion).Namespace(ns).Create(ctx, updatedStorageObject, metav1.CreateOptions{})
+				if err != nil {
+					return fmt.Errorf("failed to create new version %s %s: %v", ns, name, err)
+				}
+
+				err = clientSets.DClient.Resource(crdResourceVersion).Namespace(ns).Delete(ctx, name, metav1.DeleteOptions{})
+				if err != nil {
+					return fmt.Errorf("failed to delete old version of %s/%s: %v", ns, name, err)
+				}
+			}
+		}
+	}
+
+	fmt.Println("CRD migration completed.")
+	return nil
+}

internal/etcdmigrate/etcdmigrate_test.go

@@ -0,0 +1,158 @@
+// Copyright 2024 The prometheus-operator 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 etcdmigrate
+
+import (
+	"context"
+	"testing"
+
+	"github.com/prometheus-operator/poctl/internal/k8sutil"
+	"github.com/stretchr/testify/assert"
+	apiextensions "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	fakeApiExtensions "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/fake"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
+	fakeDynamicClient "k8s.io/client-go/dynamic/fake"
+	clienttesting "k8s.io/client-go/testing"
+)
+
+func TestMigrateCRDs(t *testing.T) {
+	type testCase struct {
+		name                string
+		namespace           string
+		getMockedClientSets func(tc testCase) k8sutil.ClientSets
+		shouldFail          bool
+	}
+
+	tests := []testCase{
+		{
+			name:       "FailCRDList",
+			shouldFail: true,
+			getMockedClientSets: func(_ testCase) k8sutil.ClientSets {
+				apiExtensionsClient := fakeApiExtensions.NewSimpleClientset()
+				apiExtensionsClient.PrependReactor("list", "customresourcedefinitions", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+					return true, nil, errors.NewInternalError(nil)
+				})
+
+				return k8sutil.ClientSets{
+					APIExtensionsClient: apiExtensionsClient,
+				}
+			},
+		},
+		{
+			name:       "FailObjectUpdate",
+			namespace:  "test",
+			shouldFail: true,
+			getMockedClientSets: func(tc testCase) k8sutil.ClientSets {
+				crd := &apiextensions.CustomResourceDefinition{
+					ObjectMeta: metav1.ObjectMeta{Name: "testcrd"},
+					Spec: apiextensions.CustomResourceDefinitionSpec{
+						Group: "monitoring.coreos.com",
+						Names: apiextensions.CustomResourceDefinitionNames{
+							Plural: "probes",
+						},
+						Versions: []apiextensions.CustomResourceDefinitionVersion{},
+					},
+				}
+
+				crInstance := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"apiVersion": "monitoring.coreos.com/v1beta1",
+						"kind":       "Probe",
+						"metadata": map[string]interface{}{
+							"name":      "probe",
+							"namespace": tc.namespace,
+						},
+					},
+				}
+
+				apiExtensionsClient := fakeApiExtensions.NewSimpleClientset(crd)
+				dClient := fakeDynamicClient.NewSimpleDynamicClient(runtime.NewScheme(), crInstance)
+				dClient.PrependReactor("update", "probes", func(_ clienttesting.Action) (bool, runtime.Object, error) {
+					return true, nil, errors.NewInternalError(nil)
+				})
+
+				return k8sutil.ClientSets{
+					APIExtensionsClient: apiExtensionsClient,
+					DClient:             dClient,
+				}
+			},
+		},
+		{
+			name:       "SuccessUpdateStorageVersion",
+			namespace:  "test",
+			shouldFail: false,
+			getMockedClientSets: func(tc testCase) k8sutil.ClientSets {
+				crd := &apiextensions.CustomResourceDefinition{
+					ObjectMeta: metav1.ObjectMeta{Name: "testcrd"},
+					Spec: apiextensions.CustomResourceDefinitionSpec{
+						Group: "monitoring.coreos.com",
+						Names: apiextensions.CustomResourceDefinitionNames{
+							Plural: "probes",
+						},
+						Versions: []apiextensions.CustomResourceDefinitionVersion{
+							{Name: "v1", Storage: true},
+						},
+					},
+				}
+
+				crInstance := &unstructured.Unstructured{
+					Object: map[string]interface{}{
+						"apiVersion": "monitoring.coreos.com/v1beta1",
+						"kind":       "Probe",
+						"metadata": map[string]interface{}{
+							"name":      "probe",
+							"namespace": tc.namespace,
+						},
+					},
+				}
+
+				apiExtensionsClient := fakeApiExtensions.NewSimpleClientset(crd)
+				dClient := fakeDynamicClient.NewSimpleDynamicClient(runtime.NewScheme(), crInstance)
+
+				dClient.PrependReactor("update", "probes", func(action clienttesting.Action) (bool, runtime.Object, error) {
+					updateAction, _ := action.(clienttesting.UpdateAction)
+					obj := updateAction.GetObject().(*unstructured.Unstructured)
+					apiVersion := obj.GetAPIVersion()
+
+					if apiVersion != "monitoring.coreos.com/v1" {
+						return true, nil, errors.NewInternalError(nil)
+					}
+
+					return true, obj, nil
+				})
+
+				return k8sutil.ClientSets{
+					APIExtensionsClient: apiExtensionsClient,
+					DClient:             dClient,
+				}
+			},
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			clientSets := tc.getMockedClientSets(tc)
+
+			err := MigrateCRDs(context.Background(), &clientSets)
+			if tc.shouldFail {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}