feat: replace image registry dynamically (#8018)

cmd/dataprotection/main.go

@@ -182,9 +182,16 @@ func main() {
 	if err := viper.ReadInConfig(); err != nil { // Handle errors reading the config file
 		setupLog.Info("unable to read in config, errors ignored")
 	}
+	if err := intctrlutil.LoadRegistryConfig(); err != nil {
+		setupLog.Error(err, "unable to reload registry config")
+		os.Exit(1)
+	}
 	setupLog.Info(fmt.Sprintf("config file: %s", viper.GetViper().ConfigFileUsed()))
 	viper.OnConfigChange(func(e fsnotify.Event) {
 		setupLog.Info(fmt.Sprintf("config file changed: %s", e.Name))
+		if err := intctrlutil.LoadRegistryConfig(); err != nil {
+			setupLog.Error(err, "unable to reload registry config")
+		}
 	})
 	viper.WatchConfig()
 

cmd/manager/main.go

@@ -302,9 +302,16 @@ func main() {
 	if err := viper.ReadInConfig(); err != nil { // Handle errors reading the config file
 		setupLog.Info("unable to read in config, errors ignored")
 	}
+	if err := intctrlutil.LoadRegistryConfig(); err != nil {
+		setupLog.Error(err, "unable to reload registry config")
+		os.Exit(1)
+	}
 	setupLog.Info(fmt.Sprintf("config file: %s", viper.GetViper().ConfigFileUsed()))
 	viper.OnConfigChange(func(e fsnotify.Event) {
 		setupLog.Info(fmt.Sprintf("config file changed: %s", e.Name))
+		if err := intctrlutil.LoadRegistryConfig(); err != nil {
+			setupLog.Error(err, "unable to reload registry config")
+		}
 	})
 	viper.WatchConfig()
 

controllers/apps/component_controller_test.go

@@ -2105,4 +2105,92 @@ var _ = Describe("Component Controller", func() {
 			testImageUnchangedAfterNewReleasePublished(release)
 		})
 	})
+
+	Context("with registry replace enabled", func() {
+		registry := "foo.bar"
+		setRegistryConfig := func() {
+			viper.Set(constant.CfgRegistries, map[string]any{
+				"defaultRegistry": registry,
+			})
+			Expect(intctrlutil.LoadRegistryConfig()).Should(Succeed())
+		}
+
+		BeforeEach(func() {
+			createAllDefinitionObjects()
+		})
+
+		AfterEach(func() {
+			viper.Set(constant.CfgRegistries, nil)
+			Expect(intctrlutil.LoadRegistryConfig()).Should(Succeed())
+		})
+
+		It("replaces image registry", func() {
+			setRegistryConfig()
+
+			createClusterObj(defaultCompName, compDefName, nil)
+
+			itsKey := compKey
+			Eventually(testapps.CheckObj(&testCtx, itsKey, func(g Gomega, its *workloads.InstanceSet) {
+				// check the image
+				c := its.Spec.Template.Spec.Containers[0]
+				g.Expect(c.Image).To(HavePrefix(registry))
+			})).Should(Succeed())
+		})
+
+		It("handles running its and upgrade", func() {
+			createClusterObj(defaultCompName, compDefName, nil)
+			itsKey := compKey
+			Eventually(testapps.CheckObj(&testCtx, itsKey, func(g Gomega, its *workloads.InstanceSet) {
+				// check the image
+				c := its.Spec.Template.Spec.Containers[0]
+				g.Expect(c.Image).To(Equal(compVerObj.Spec.Releases[0].Images[c.Name]))
+			})).Should(Succeed())
+
+			setRegistryConfig()
+			By("trigger component reconcile")
+			now := time.Now().Format(time.RFC3339)
+			Expect(testapps.GetAndChangeObj(&testCtx, compKey, func(comp *kbappsv1.Component) {
+				comp.Annotations["now"] = now
+			})()).Should(Succeed())
+
+			Consistently(testapps.CheckObj(&testCtx, itsKey, func(g Gomega, its *workloads.InstanceSet) {
+				// check the image
+				c := its.Spec.Template.Spec.Containers[0]
+				g.Expect(c.Image).NotTo(HavePrefix(registry))
+			})).Should(Succeed())
+
+			By("replaces registry when upgrading")
+			release := kbappsv1.ComponentVersionRelease{
+				Name:           "8.0.31",
+				ServiceVersion: "8.0.31",
+				Images: map[string]string{
+					testapps.DefaultMySQLContainerName: "docker.io/apecloud/mysql:8.0.31",
+				},
+			}
+
+			By("publish a new release")
+			compVerKey := client.ObjectKeyFromObject(compVerObj)
+			Expect(testapps.GetAndChangeObj(&testCtx, compVerKey, func(compVer *kbappsv1.ComponentVersion) {
+				compVer.Spec.Releases = append(compVer.Spec.Releases, release)
+				compVer.Spec.CompatibilityRules[0].Releases = append(compVer.Spec.CompatibilityRules[0].Releases, release.Name)
+			})()).Should(Succeed())
+
+			By("update serviceversion in cluster")
+			Expect(testapps.GetAndChangeObj(&testCtx, clusterKey, func(cluster *kbappsv1.Cluster) {
+				cluster.Spec.ComponentSpecs[0].ServiceVersion = "8.0.31"
+			})()).Should(Succeed())
+
+			By("trigger component reconcile")
+			now = time.Now().Format(time.RFC3339)
+			Expect(testapps.GetAndChangeObj(&testCtx, compKey, func(comp *kbappsv1.Component) {
+				comp.Annotations["now"] = now
+			})()).Should(Succeed())
+
+			Eventually(testapps.CheckObj(&testCtx, itsKey, func(g Gomega, its *workloads.InstanceSet) {
+				// check the image
+				c := its.Spec.Template.Spec.Containers[0]
+				g.Expect(c.Image).To(HavePrefix(registry))
+			})).Should(Succeed())
+		})
+	})
 })

controllers/apps/suite_test.go

@@ -169,7 +169,7 @@ var _ = BeforeSuite(func() {
 	Expect(err).ToNot(HaveOccurred())
 
 	viper.SetDefault("CERT_DIR", "/tmp/k8s-webhook-server/serving-certs")
-	viper.SetDefault(constant.KBToolsImage, "apecloud/kubeblocks-tools:latest")
+	viper.SetDefault(constant.KBToolsImage, "docker.io/apecloud/kubeblocks-tools:latest")
 	viper.SetDefault("PROBE_SERVICE_PORT", 3501)
 	viper.SetDefault("PROBE_SERVICE_LOG_LEVEL", "info")
 	viper.SetDefault("CM_NAMESPACE", "default")

controllers/apps/transformer_component_workload.go

@@ -151,6 +151,16 @@ func (t *componentWorkloadTransformer) runningInstanceSetObject(ctx graph.Transf
 
 func (t *componentWorkloadTransformer) reconcileWorkload(ctx context.Context, cli client.Reader,
 	synthesizedComp *component.SynthesizedComponent, comp *appsv1.Component, runningITS, protoITS *workloads.InstanceSet) error {
+	// if runningITS already exists, the image changes in protoITS will be
+	// rollback to the original image in `checkNRollbackProtoImages`.
+	// So changing registry configs won't affect existing clusters.
+	for i, container := range protoITS.Spec.Template.Spec.Containers {
+		protoITS.Spec.Template.Spec.Containers[i].Image = intctrlutil.ReplaceImageRegistry(container.Image)
+	}
+	for i, container := range protoITS.Spec.Template.Spec.InitContainers {
+		protoITS.Spec.Template.Spec.InitContainers[i].Image = intctrlutil.ReplaceImageRegistry(container.Image)
+	}
+
 	buildInstanceSetPlacementAnnotation(comp, protoITS)
 
 	if err := t.reconcileReplicasStatus(ctx, cli, synthesizedComp, runningITS, protoITS); err != nil {

controllers/extensions/addon_controller_stages.go

@@ -514,7 +514,7 @@ func setInitContainer(addon *extensionsv1alpha1.Addon, helmJobPodSpec *corev1.Po
 	}
 	copyChartsContainer := corev1.Container{
 		Name:    "copy-charts",
-		Image:   addon.Spec.Helm.ChartsImage,
+		Image:   intctrlutil.ReplaceImageRegistry(addon.Spec.Helm.ChartsImage),
 		Command: []string{"sh", "-c", fmt.Sprintf("cp %s/* /mnt/charts", fromPath)},
 		VolumeMounts: []corev1.VolumeMount{
 			{

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/aws/aws-sdk-go v1.50.8
 	github.com/bhmj/jsonslice v1.1.2
 	github.com/clbanning/mxj/v2 v2.5.7
+	github.com/distribution/reference v0.6.0
 	github.com/docker/docker v25.0.6+incompatible
 	github.com/evanphx/json-patch v5.6.0+incompatible
 	github.com/evanphx/json-patch/v5 v5.8.0
@@ -104,7 +105,6 @@ require (
 	github.com/containerd/log v0.1.0 // indirect
 	github.com/cyphar/filepath-securejoin v0.2.4 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/distribution/reference v0.5.0 // indirect
 	github.com/docker/cli v25.0.1+incompatible // indirect
 	github.com/docker/distribution v2.8.3+incompatible // indirect
 	github.com/docker/docker-credential-helpers v0.8.0 // indirect

go.sum

@@ -190,8 +190,8 @@ github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZm
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
 github.com/distribution/distribution/v3 v3.0.0-20221208165359-362910506bc2 h1:aBfCb7iqHmDEIp6fBvC/hQUddQfg+3qdYjwzaiP9Hnc=
 github.com/distribution/distribution/v3 v3.0.0-20221208165359-362910506bc2/go.mod h1:WHNsWjnIn2V1LYOrME7e8KxSeKunYHsxEm4am0BUtcI=
-github.com/distribution/reference v0.5.0 h1:/FUIFXtfc/x2gpa5/VGfiGLuOIdYa1t65IKK2OFGvA0=
-github.com/distribution/reference v0.5.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
+github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
+github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
 github.com/docker/cli v25.0.1+incompatible h1:mFpqnrS6Hsm3v1k7Wa/BO23oz0k121MTbTO1lpcGSkU=
 github.com/docker/cli v25.0.1+incompatible/go.mod h1:JLrzqnKDaYBop7H2jaqPtU4hHvMKP+vjCwu2uszcLI8=
 github.com/docker/distribution v2.8.3+incompatible h1:AtKxIZ36LoNK51+Z6RpzLpddBirtxJnzDrHLEKxTAYk=

pkg/constant/viper_config.go

@@ -53,5 +53,6 @@ const (
 	CfgClientQPS          = "CLIENT_QPS"
 	CfgClientBurst        = "CLIENT_BURST"
 
+	CfgRegistries     = "registries"
 	I18nResourcesName = "I18N_RESOURCES_NAME"
 )