github/kubevela
1
0
Fork 0
mirror of https://github.com/kubevela/kubevela.git synced 2026-07-29 02:11:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix: update apps with topology policy during cluster join (#6768)

Browse Source 
* functionality to get all application with a topology in cluster

Signed-off-by: vishal210893 <vishal210893@gmail.com>
Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>

* refactor code and unit tests

Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>

* refactor code and unit tests

Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>

* rearrange imports

Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>

* remove calling of goroutine

Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>

* update logic to set publich version annotation

Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>

* removed unused constants

Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>

* make reviewable

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* add license info for cluster_test.go

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* ignore errors in updateAppsWithTopologyPolicy

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* modify error message

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* gofmt

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* add retry logic to handle conflict errors

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* using cmd to print and add log for retried applications

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* using context as first argument

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* log namespace in error

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* optimize retry logic

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

* add pagination for listing applications

Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>

---------

Signed-off-by: vishal210893 <vishal210893@gmail.com>
Signed-off-by: Shivin Gopalani <sgopalani@guidewire.com>
Signed-off-by: Shivin Gopalani <gopalanishivin@gmail.com>
Co-authored-by: vishal210893 <vishal210893@gmail.com>
This commit is contained in:
shivin
2025-05-01 14:31:11 +05:30
committed by GitHub
parent edf3be272e
commit a1145f21fe
2 changed files with 298 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
references/cli/cluster.go
View File

@@ -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{

189
references/cli/cluster_test.go Normal file
View File

@@ -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
}