diff --git a/references/cli/cluster.go b/references/cli/cluster.go index 17db981c7..b36e5e854 100644 --- a/references/cli/cluster.go +++ b/references/cli/cluster.go @@ -18,9 +18,11 @@ package cli import ( "context" + "encoding/json" "fmt" "sort" "strings" + "time" "github.com/crossplane/crossplane-runtime/pkg/meta" "github.com/fatih/color" @@ -37,9 +39,17 @@ import ( "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1alpha1" + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + + apierrors "k8s.io/apimachinery/pkg/api/errors" + apitypes "k8s.io/apimachinery/pkg/types" + "github.com/oam-dev/kubevela/apis/types" velacmd "github.com/oam-dev/kubevela/pkg/cmd" "github.com/oam-dev/kubevela/pkg/multicluster" + "github.com/oam-dev/kubevela/pkg/oam" + "github.com/oam-dev/kubevela/pkg/oam/util" "github.com/oam-dev/kubevela/pkg/utils/common" cmdutil "github.com/oam-dev/kubevela/pkg/utils/util" ) @@ -213,11 +223,17 @@ func NewClusterJoinCommand(c *common.Args, ioStreams cmdutil.IOStreams) *cobra.C if err != nil { return err } - cmd.Printf("Successfully add cluster %s, endpoint: %s.\n", clusterName, clusterConfig.Cluster.Server) if len(labels) > 0 { - return addClusterLabels(cmd, c, clusterName, labels) + if err := addClusterLabels(cmd, c, clusterName, labels); err != nil { + return fmt.Errorf("error in adding cluster labels: %w", err) + } } + if err := updateAppsWithTopologyPolicy(ctx, cmd, client); err != nil { + return fmt.Errorf("error in updating apps with topology policy: %w", err) + } + + cmd.Printf("Successfully add cluster %s, endpoint: %s.\n", clusterName, clusterConfig.Cluster.Server) return nil }, } @@ -232,6 +248,97 @@ func NewClusterJoinCommand(c *common.Args, ioStreams cmdutil.IOStreams) *cobra.C return cmd } +// updateAppsWithTopologyPolicy iterates through all Application resources in the cluster, +// and updates those that have a cluster-level label selector defined in topology policy. +// For each matching application, it sets or updates publish version annotation. +func updateAppsWithTopologyPolicy(ctx context.Context, cmd *cobra.Command, k8sClient client.Client) error { + var continueToken string + const pageSize = 100 // Adjust based on performance needs + for { + // List every Application once, update only those with a cluster label selector. + applicationList := &v1beta1.ApplicationList{} + listOpts := &client.ListOptions{ + Limit: pageSize, + Continue: continueToken, + } + if err := k8sClient.List(ctx, applicationList, listOpts); err != nil { + return fmt.Errorf("failed to list applications: %w", err) + } + + for i := range applicationList.Items { // index-based to avoid copies + app := &applicationList.Items[i] + + matched, err := hasClusterLabelSelector(app.Spec.Policies) + if err != nil { + return fmt.Errorf("failed to check clusterlabelselector for application %s in namespace %s: %w", app.Name, app.Namespace, err) + } + if !matched { + continue + } + + // Retry loop for conflict handling + const maxRetries = 5 + for attempt := 0; attempt < maxRetries; attempt++ { + // Refresh the object to get the latest resourceVersion (only after 1st attempt) + if attempt > 0 { + key := apitypes.NamespacedName{Namespace: app.Namespace, Name: app.Name} + if err := k8sClient.Get(ctx, key, app); err != nil { + return fmt.Errorf("failed to refetch app %s in namespace %s: %w", app.Name, app.Namespace, err) + } + } + + // Update logic + oam.SetPublishVersion(app, util.GenerateVersion("clusterjoin")) + + if err := k8sClient.Update(ctx, app); err != nil { + if apierrors.IsConflict(err) { + // Retry if there's a conflict + if attempt == maxRetries-1 { + return fmt.Errorf("conflict error updating app %s in namespace %s after %d retries: %w", app.Name, app.Namespace, maxRetries, err) + } + cmd.Printf("Conflict updating app %s in namespace %s, retrying (%d/%d)...\n", app.Name, app.Namespace, attempt+1, maxRetries) + time.Sleep(500 * time.Millisecond) + continue + } + // Non-conflict error, return it + return fmt.Errorf("error updating app %s in namespace %s: %w", app.Name, app.Namespace, err) + } + + if attempt > 0 { + cmd.Printf("Successfully updated app %s in namespace %s after %d retries.\n", app.Name, app.Namespace, attempt) + } + // Successful update + break + } + } + continueToken = applicationList.Continue + if continueToken == "" { + break // No more pages + } + } + return nil +} + +// hasClusterLabelSelector returns true when at least one topology policy +// has an explicit clusterLabelSelector. +func hasClusterLabelSelector(policies []v1beta1.AppPolicy) (bool, error) { + for _, p := range policies { + if p.Type != "topology" || p.Properties == nil || len(p.Properties.Raw) == 0 { + continue + } + + var tp v1alpha1.Placement + if err := json.Unmarshal(p.Properties.Raw, &tp); err != nil { + return false, fmt.Errorf("error in unmarshalling policy %v: %w", p, err) + } + + if tp.ClusterLabelSelector != nil { + return true, nil + } + } + return false, nil +} + // NewClusterRenameCommand create command to help user rename cluster func NewClusterRenameCommand(c *common.Args) *cobra.Command { cmd := &cobra.Command{ diff --git a/references/cli/cluster_test.go b/references/cli/cluster_test.go new file mode 100644 index 000000000..01b7616c1 --- /dev/null +++ b/references/cli/cluster_test.go @@ -0,0 +1,189 @@ +/* +Copyright 2022 The KubeVela 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 cli + +import ( + "context" + "fmt" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/spf13/cobra" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/yaml" + + "github.com/oam-dev/kubevela/apis/core.oam.dev/v1beta1" + "github.com/oam-dev/kubevela/pkg/oam" +) + +var ( + appWithoutTopologyPolicyYaml = ` +apiVersion: core.oam.dev/v1beta1 +kind: Application +metadata: + name: app-without-policies + namespace: vela-system +spec: + components: + - name: nginx-basic + type: webservice + properties: + image: nginx +` + appWithTopologyClustersYaml = ` +apiVersion: core.oam.dev/v1beta1 +kind: Application +metadata: + name: basic-topology + namespace: default +spec: + components: + - name: nginx-basic + type: webservice + properties: + image: nginx + policies: + - name: topology-hangzhou-clusters + type: topology + properties: + clusters: ["hangzhou-1", "hangzhou-2"] +` + appWithTopologyClusterLabelSelectorYaml = ` +apiVersion: core.oam.dev/v1beta1 +kind: Application +metadata: + name: region-selector + namespace: vela-system +spec: + components: + - name: nginx-basic + type: webservice + properties: + image: nginx + policies: + - name: topology-hangzhou-clusters + type: topology + properties: + clusterLabelSelector: + region: hangzhou +` + appWithEmptyTopologyClusterLabelSelectorYaml = ` +apiVersion: core.oam.dev/v1beta1 +kind: Application +metadata: + name: empty-cluster-selector + namespace: default +spec: + components: + - name: nginx-basic + type: webservice + properties: + image: nginx + policies: + - name: topology-hangzhou-clusters + type: topology + properties: + clusterLabelSelector: {} +` +) + +var _ = Describe("Test updateAppsWithTopologyPolicy", func() { + + var _ = When("app does not have topology policy", func() { + It("app should not have publish version annotation set", func() { + err := createApplication(appWithoutTopologyPolicyYaml) + Expect(err).Should(BeNil()) + + cmd := &cobra.Command{} + err = updateAppsWithTopologyPolicy(context.Background(), cmd, k8sClient) + Expect(err).Should(BeNil()) + + matched, err := hasPublishVersionAnnotation("app-without-policies", "vela-system") + Expect(err).Should(BeNil()) + Expect(matched).Should(BeFalse()) + }) + }) + + var _ = When("app has topology policy without clusterLabelSelector", func() { + It("app should not have publish version annotation set", func() { + err := createApplication(appWithTopologyClustersYaml) + Expect(err).Should(BeNil()) + + cmd := &cobra.Command{} + err = updateAppsWithTopologyPolicy(context.Background(), cmd, k8sClient) + Expect(err).Should(BeNil()) + + matched, err := hasPublishVersionAnnotation("basic-topology", "default") + Expect(err).Should(BeNil()) + Expect(matched).Should(BeFalse()) + }) + }) + + var _ = When("app has topology policy with clusterLabelSelector", func() { + It("app should have publish version annotation set", func() { + err := createApplication(appWithTopologyClusterLabelSelectorYaml) + Expect(err).Should(BeNil()) + + cmd := &cobra.Command{} + err = updateAppsWithTopologyPolicy(context.Background(), cmd, k8sClient) + Expect(err).Should(BeNil()) + + matched, err := hasPublishVersionAnnotation("region-selector", "vela-system") + Expect(err).Should(BeNil()) + Expect(matched).Should(BeTrue()) + }) + }) + + var _ = When("app has topology policy with empty clusterLabelSelector", func() { + It("app should have publish version annotation set", func() { + err := createApplication(appWithEmptyTopologyClusterLabelSelectorYaml) + Expect(err).Should(BeNil()) + + cmd := &cobra.Command{} + err = updateAppsWithTopologyPolicy(context.Background(), cmd, k8sClient) + Expect(err).Should(BeNil()) + + matched, err := hasPublishVersionAnnotation("empty-cluster-selector", "default") + Expect(err).Should(BeNil()) + Expect(matched).Should(BeTrue()) + }) + }) + +}) + +func createApplication(appYaml string) error { + app := v1beta1.Application{} + if err := yaml.Unmarshal([]byte(appYaml), &app); err != nil { + return fmt.Errorf("unmarshal error for yaml %s: %w", appYaml, err) + } + if err := k8sClient.Create(context.Background(), &app); err != nil { + return fmt.Errorf("error in creating app %s in namespace %s: %w", app.Name, app.Namespace, err) + } + return nil +} + +func hasPublishVersionAnnotation(name, namespace string) (bool, error) { + app := &v1beta1.Application{} + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: name, Namespace: namespace}, app); err != nil { + return false, fmt.Errorf("error in getting application %s in namespace %s: %w", name, namespace, err) + } + annotations := app.GetAnnotations() + if annotations != nil && annotations[oam.AnnotationPublishVersion] != "" { + return true, nil + } + return false, nil +}