From 65852d9a2e45fa4c40b372ef96f0ee12c7f8ccd7 Mon Sep 17 00:00:00 2001
From: Michael Burman <michael.burman@datastax.com>
Date: Mon, 13 Mar 2023 16:41:04 +0200
Subject: [PATCH 1/6] Extend current ImageConfig to allow more values to images
 as well as additional configuration options for each image type

Add modified PullPolicy to containers

Add ability to restrict pullSecrets added per pod

Make internal imageConfig to be non-pointer to allow easier usage in tests

Make some pointer accesses safer
---
 CHANGELOG.md                                  |   1 +
 apis/config/v1beta1/imageconfig_types.go      |  85 ++++++++++--
 apis/config/v1beta1/zz_generated.deepcopy.go  |  60 ++++++++-
 pkg/images/images.go                          | 122 ++++++++++++++----
 pkg/images/images_test.go                     |  56 +++++---
 .../construct_podtemplatespec.go              |  25 ++--
 .../image_config_parsing_more_options.yaml    |  45 +++++++
 7 files changed, 328 insertions(+), 66 deletions(-)
 create mode 100644 tests/testdata/image_config_parsing_more_options.yaml

diff --git a/CHANGELOG.md b/CHANGELOG.md
index a17178e3..caef50ec 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -12,6 +12,7 @@ Changelog for Cass Operator, new PRs should update the `main / unreleased` secti
 ## unreleased
 
 * [FEATURE] [#651](https://github.com/k8ssandra/cass-operator/issues/651) Add tsreload task for DSE deployments and ability to check if sync operation is available on the mgmt-api side
+* [ENHANCEMENT] [#532](https://github.com/k8ssandra/k8ssandra-operator/issues/532) Extend ImageConfig type to allow additional parameters for k8ssandra-operator requirements. These include per-image PullPolicy / PullSecrets as well as additional image
 * [BUGFIX] [#705](https://github.com/k8ssandra/cass-operator/issues/705) Ensure ConfigSecret has annotations map before trying to set a value
 
 ## v1.22.4
diff --git a/apis/config/v1beta1/imageconfig_types.go b/apis/config/v1beta1/imageconfig_types.go
index 3cdf6e20..3ff451b3 100644
--- a/apis/config/v1beta1/imageconfig_types.go
+++ b/apis/config/v1beta1/imageconfig_types.go
@@ -17,13 +17,12 @@ limitations under the License.
 package v1beta1
 
 import (
+	"encoding/json"
+
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
-// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
-
 //+kubebuilder:object:root=true
 //+kubebuilder:subresource:images
 
@@ -35,6 +34,10 @@ type ImageConfig struct {
 
 	DefaultImages *DefaultImages `json:"defaults,omitempty"`
 
+	ImagePolicy
+}
+
+type ImagePolicy struct {
 	ImageRegistry string `json:"imageRegistry,omitempty"`
 
 	ImagePullSecret corev1.LocalObjectReference `json:"imagePullSecret,omitempty"`
@@ -52,26 +55,86 @@ type Images struct {
 
 	DSEVersions map[string]string `json:"dse,omitempty"`
 
-	SystemLogger string `json:"system-logger"`
-
-	ConfigBuilder string `json:"config-builder"`
+	SystemLogger string `json:"system-logger,omitempty"`
 
 	Client string `json:"k8ssandra-client,omitempty"`
+
+	ConfigBuilder string `json:"config-builder,omitempty"`
+
+	Others map[string]string `json:",inline,omitempty"`
 }
 
-type DefaultImages struct {
-	metav1.TypeMeta `json:",inline"`
+type _Images Images
+
+func (i *Images) UnmarshalJSON(b []byte) error {
+	var imagesTemp _Images
+	if err := json.Unmarshal(b, &imagesTemp); err != nil {
+		return err
+	}
+	*i = Images(imagesTemp)
+
+	var otherFields map[string]interface{}
+	if err := json.Unmarshal(b, &otherFields); err != nil {
+		return err
+	}
+
+	delete(otherFields, CassandraImageComponent)
+	delete(otherFields, DSEImageComponent)
+	delete(otherFields, SystemLoggerImageComponent)
+	delete(otherFields, ConfigBuilderImageComponent)
+	delete(otherFields, ClientImageComponent)
+
+	others := make(map[string]string, len(otherFields))
+	for k, v := range otherFields {
+		others[k] = v.(string)
+	}
+
+	i.Others = others
+	return nil
+}
+
+const (
+	CassandraImageComponent     string = "cassandra"
+	DSEImageComponent           string = "dse"
+	HCDImageComponent           string = "hcd"
+	SystemLoggerImageComponent  string = "system-logger"
+	ConfigBuilderImageComponent string = "config-builder"
+	ClientImageComponent        string = "k8ssandra-client"
+)
 
-	CassandraImageComponent ImageComponent `json:"cassandra,omitempty"`
+type ImageComponents map[string]ImageComponent
 
-	DSEImageComponent ImageComponent `json:"dse,omitempty"`
+type DefaultImages struct {
+	ImageComponents
+}
+
+func (d *DefaultImages) MarshalJSON() ([]byte, error) {
+	// This shouldn't be required, just like it's not with ImagePolicy, but this is Go..
+	return json.Marshal(d.ImageComponents)
+}
 
-	HCDImageComponent ImageComponent `json:"hcd,omitempty"`
+func (d *DefaultImages) UnmarshalJSON(b []byte) error {
+	d.ImageComponents = make(map[string]ImageComponent)
+	var input map[string]json.RawMessage
+	if err := json.Unmarshal(b, &input); err != nil {
+		return err
+	}
+
+	for k, v := range input {
+		var component ImageComponent
+		if err := json.Unmarshal(v, &component); err != nil {
+			return err
+		}
+		d.ImageComponents[k] = component
+	}
+
+	return nil
 }
 
 type ImageComponent struct {
 	Repository string `json:"repository,omitempty"`
 	Suffix     string `json:"suffix,omitempty"`
+	ImagePolicy
 }
 
 func init() {
diff --git a/apis/config/v1beta1/zz_generated.deepcopy.go b/apis/config/v1beta1/zz_generated.deepcopy.go
index c160419b..cb23b64a 100644
--- a/apis/config/v1beta1/zz_generated.deepcopy.go
+++ b/apis/config/v1beta1/zz_generated.deepcopy.go
@@ -27,10 +27,13 @@ import (
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *DefaultImages) DeepCopyInto(out *DefaultImages) {
 	*out = *in
-	out.TypeMeta = in.TypeMeta
-	out.CassandraImageComponent = in.CassandraImageComponent
-	out.DSEImageComponent = in.DSEImageComponent
-	out.HCDImageComponent = in.HCDImageComponent
+	if in.ImageComponents != nil {
+		in, out := &in.ImageComponents, &out.ImageComponents
+		*out = make(ImageComponents, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new DefaultImages.
@@ -46,6 +49,7 @@ func (in *DefaultImages) DeepCopy() *DefaultImages {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ImageComponent) DeepCopyInto(out *ImageComponent) {
 	*out = *in
+	out.ImagePolicy = in.ImagePolicy
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ImageComponent.
@@ -58,6 +62,27 @@ func (in *ImageComponent) DeepCopy() *ImageComponent {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in ImageComponents) DeepCopyInto(out *ImageComponents) {
+	{
+		in := &in
+		*out = make(ImageComponents, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ImageComponents.
+func (in ImageComponents) DeepCopy() ImageComponents {
+	if in == nil {
+		return nil
+	}
+	out := new(ImageComponents)
+	in.DeepCopyInto(out)
+	return *out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ImageConfig) DeepCopyInto(out *ImageConfig) {
 	*out = *in
@@ -70,9 +95,9 @@ func (in *ImageConfig) DeepCopyInto(out *ImageConfig) {
 	if in.DefaultImages != nil {
 		in, out := &in.DefaultImages, &out.DefaultImages
 		*out = new(DefaultImages)
-		**out = **in
+		(*in).DeepCopyInto(*out)
 	}
-	out.ImagePullSecret = in.ImagePullSecret
+	out.ImagePolicy = in.ImagePolicy
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ImageConfig.
@@ -93,6 +118,22 @@ func (in *ImageConfig) DeepCopyObject() runtime.Object {
 	return nil
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *ImagePolicy) DeepCopyInto(out *ImagePolicy) {
+	*out = *in
+	out.ImagePullSecret = in.ImagePullSecret
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ImagePolicy.
+func (in *ImagePolicy) DeepCopy() *ImagePolicy {
+	if in == nil {
+		return nil
+	}
+	out := new(ImagePolicy)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *Images) DeepCopyInto(out *Images) {
 	*out = *in
@@ -111,6 +152,13 @@ func (in *Images) DeepCopyInto(out *Images) {
 			(*out)[key] = val
 		}
 	}
+	if in.Others != nil {
+		in, out := &in.Others, &out.Others
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new Images.
diff --git a/pkg/images/images.go b/pkg/images/images.go
index c6e7f1c7..e64f5240 100644
--- a/pkg/images/images.go
+++ b/pkg/images/images.go
@@ -19,7 +19,7 @@ import (
 )
 
 var (
-	imageConfig *configv1beta1.ImageConfig
+	imageConfig configv1beta1.ImageConfig
 	scheme      = runtime.NewScheme()
 )
 
@@ -58,7 +58,7 @@ func LoadImageConfig(content []byte) (*configv1beta1.ImageConfig, error) {
 		return nil, fmt.Errorf("could not decode file into runtime.Object: %v", err)
 	}
 
-	imageConfig = parsedImageConfig
+	imageConfig = *parsedImageConfig
 
 	return parsedImageConfig, nil
 }
@@ -88,8 +88,7 @@ func stripRegistry(image string) string {
 	}
 }
 
-func applyDefaultRegistryOverride(image string) string {
-	customRegistry := GetImageConfig().ImageRegistry
+func applyDefaultRegistryOverride(customRegistry, image string) string {
 	customRegistry = strings.TrimSuffix(customRegistry, "/")
 
 	if customRegistry == "" {
@@ -100,13 +99,33 @@ func applyDefaultRegistryOverride(image string) string {
 	}
 }
 
-func ApplyRegistry(image string) string {
-	return applyDefaultRegistryOverride(image)
+func getRegistryOverride(imageType string) string {
+	customRegistry := ""
+	defaults := GetImageConfig().DefaultImages
+	if defaults != nil {
+		if component, found := defaults.ImageComponents[imageType]; found {
+			customRegistry = component.ImageRegistry
+		}
+	}
+
+	defaultRegistry := GetImageConfig().ImageRegistry
+
+	if customRegistry != "" {
+		return customRegistry
+	}
+
+	return defaultRegistry
+}
+
+func applyRegistry(imageType, image string) string {
+	registry := getRegistryOverride(imageType)
+
+	return applyDefaultRegistryOverride(registry, image)
 }
 
 func GetImageConfig() *configv1beta1.ImageConfig {
 	// For now, this is static configuration (updated only on start of the pod), even if the actual ConfigMap underneath is updated.
-	return imageConfig
+	return &imageConfig
 }
 
 func getCassandraContainerImageOverride(serverType, version string) (bool, string) {
@@ -140,15 +159,14 @@ func getImageComponents(serverType string) (string, string) {
 	defaults := GetImageConfig().DefaultImages
 	if defaults != nil {
 		var component configv1beta1.ImageComponent
-		switch serverType {
-		case "dse":
-			component = defaults.DSEImageComponent
-		case "cassandra":
-			component = defaults.CassandraImageComponent
-		case "hcd":
-			component = defaults.HCDImageComponent
-		default:
-			component = defaults.CassandraImageComponent
+		if serverType == "dse" {
+			component = defaults.ImageComponents[configv1beta1.DSEImageComponent]
+		}
+		if serverType == "cassandra" {
+			component = defaults.ImageComponents[configv1beta1.CassandraImageComponent]
+		}
+		if serverType == "hcd" {
+			component = defaults.ImageComponents[configv1beta1.HCDImageComponent]
 		}
 
 		if component.Repository != "" {
@@ -161,7 +179,7 @@ func getImageComponents(serverType string) (string, string) {
 
 func GetCassandraImage(serverType, version string) (string, error) {
 	if found, image := getCassandraContainerImageOverride(serverType, version); found {
-		return ApplyRegistry(image), nil
+		return applyRegistry(serverType, image), nil
 	}
 
 	switch serverType {
@@ -183,28 +201,82 @@ func GetCassandraImage(serverType, version string) (string, error) {
 
 	prefix, suffix := getImageComponents(serverType)
 
-	return ApplyRegistry(fmt.Sprintf("%s:%s%s", prefix, version, suffix)), nil
+	return applyRegistry(serverType, fmt.Sprintf("%s:%s%s", prefix, version, suffix)), nil
+}
+
+func GetConfiguredImage(imageType, image string) string {
+	return applyRegistry(imageType, image)
+}
+
+func GetImage(imageType string) string {
+	return applyRegistry(imageType, GetImageConfig().Images.Others[imageType])
+}
+
+func GetImagePullPolicy(imageType string) corev1.PullPolicy {
+	var customPolicy corev1.PullPolicy
+	defaults := GetImageConfig().DefaultImages
+	if defaults != nil {
+		if component, found := defaults.ImageComponents[imageType]; found {
+			customPolicy = component.ImagePullPolicy
+		}
+	}
+
+	defaultOverridePolicy := GetImageConfig().ImagePullPolicy
+
+	if customPolicy != "" {
+		return customPolicy
+	} else if defaultOverridePolicy != "" {
+		return defaultOverridePolicy
+	}
+
+	return ""
 }
 
 func GetConfigBuilderImage() string {
-	return ApplyRegistry(GetImageConfig().Images.ConfigBuilder)
+	return applyRegistry(configv1beta1.ConfigBuilderImageComponent, GetImageConfig().Images.ConfigBuilder)
 }
 
 func GetClientImage() string {
-	return ApplyRegistry(GetImageConfig().Images.Client)
+	return applyRegistry(configv1beta1.ClientImageComponent, GetImageConfig().Images.Client)
 }
 
 func GetSystemLoggerImage() string {
-	return ApplyRegistry(GetImageConfig().Images.SystemLogger)
+	return applyRegistry(configv1beta1.SystemLoggerImageComponent, GetImageConfig().Images.SystemLogger)
 }
 
-func AddDefaultRegistryImagePullSecrets(podSpec *corev1.PodSpec) bool {
+func AddDefaultRegistryImagePullSecrets(podSpec *corev1.PodSpec, imageTypes ...string) {
+	secretNames := make([]string, 0)
 	secretName := GetImageConfig().ImagePullSecret.Name
 	if secretName != "" {
+		secretNames = append(secretNames, secretName)
+	}
+
+	imageTypesToAdd := make(map[string]bool, len(imageTypes))
+	if len(imageTypes) < 1 {
+		if GetImageConfig().DefaultImages != nil {
+			for name := range GetImageConfig().DefaultImages.ImageComponents {
+				imageTypesToAdd[name] = true
+			}
+		}
+	} else {
+		for _, image := range imageTypes {
+			imageTypesToAdd[image] = true
+		}
+	}
+
+	if GetImageConfig().DefaultImages != nil {
+		for name, component := range GetImageConfig().DefaultImages.ImageComponents {
+			if _, found := imageTypesToAdd[name]; found {
+				if component.ImagePullSecret.Name != "" {
+					secretNames = append(secretNames, component.ImagePullSecret.Name)
+				}
+			}
+		}
+	}
+
+	for _, s := range secretNames {
 		podSpec.ImagePullSecrets = append(
 			podSpec.ImagePullSecrets,
-			corev1.LocalObjectReference{Name: secretName})
-		return true
+			corev1.LocalObjectReference{Name: s})
 	}
-	return false
 }
diff --git a/pkg/images/images_test.go b/pkg/images/images_test.go
index 13582e19..a2182446 100644
--- a/pkg/images/images_test.go
+++ b/pkg/images/images_test.go
@@ -20,9 +20,10 @@ import (
 
 func TestDefaultRegistryOverride(t *testing.T) {
 	assert := assert.New(t)
-	imageConfig = &configv1beta1.ImageConfig{}
+	imageConfig = configv1beta1.ImageConfig{}
 	imageConfig.ImageRegistry = "localhost:5000"
 	imageConfig.Images = &configv1beta1.Images{}
+	imageConfig.DefaultImages = &configv1beta1.DefaultImages{}
 	imageConfig.Images.ConfigBuilder = "k8ssandra/config-builder-temp:latest"
 
 	image := GetConfigBuilderImage()
@@ -38,8 +39,9 @@ func TestCassandraOverride(t *testing.T) {
 
 	customImageName := "my-custom-image:4.0.0"
 
-	imageConfig = &configv1beta1.ImageConfig{}
+	imageConfig = configv1beta1.ImageConfig{}
 	imageConfig.Images = &configv1beta1.Images{}
+	imageConfig.DefaultImages = &configv1beta1.DefaultImages{}
 
 	cassImage, err := GetCassandraImage("cassandra", "4.0.0")
 	assert.NoError(err, "getting Cassandra image should succeed")
@@ -81,8 +83,8 @@ func TestDefaultImageConfigParsing(t *testing.T) {
 	assert.True(strings.Contains(GetImageConfig().Images.ConfigBuilder, "datastax/cass-config-builder:"))
 	assert.True(strings.Contains(GetImageConfig().Images.Client, "k8ssandra/k8ssandra-client:"))
 
-	assert.Equal("k8ssandra/cass-management-api", GetImageConfig().DefaultImages.CassandraImageComponent.Repository)
-	assert.Equal("datastax/dse-mgmtapi-6_8", GetImageConfig().DefaultImages.DSEImageComponent.Repository)
+	assert.Equal("k8ssandra/cass-management-api", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.CassandraImageComponent].Repository)
+	assert.Equal("datastax/dse-mgmtapi-6_8", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.DSEImageComponent].Repository)
 
 	path, err := GetCassandraImage("dse", "6.8.47")
 	assert.NoError(err)
@@ -109,16 +111,16 @@ func TestImageConfigParsing(t *testing.T) {
 	assert.True(strings.HasPrefix(GetImageConfig().Images.SystemLogger, "k8ssandra/system-logger:"))
 	assert.True(strings.HasPrefix(GetImageConfig().Images.ConfigBuilder, "datastax/cass-config-builder:"))
 
-	assert.Equal("k8ssandra/cass-management-api", GetImageConfig().DefaultImages.CassandraImageComponent.Repository)
-	assert.Equal("datastax/dse-mgmtapi-6_8", GetImageConfig().DefaultImages.DSEImageComponent.Repository)
+	assert.Equal("k8ssandra/cass-management-api", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.CassandraImageComponent].Repository)
+	assert.Equal("datastax/dse-mgmtapi-6_8", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.DSEImageComponent].Repository)
 
 	assert.Equal("localhost:5000", GetImageConfig().ImageRegistry)
 	assert.Equal(corev1.PullAlways, GetImageConfig().ImagePullPolicy)
 	assert.Equal("my-secret-pull-registry", GetImageConfig().ImagePullSecret.Name)
 
-	path, err := GetCassandraImage("dse", "6.8.17")
+	path, err := GetCassandraImage("dse", "6.8.43")
 	assert.NoError(err)
-	assert.Equal("localhost:5000/datastax/dse-mgmtapi-6_8:6.8.17-ubi8", path)
+	assert.Equal("localhost:5000/datastax/dse-mgmtapi-6_8:6.8.43-ubi8", path)
 
 	path, err = GetCassandraImage("dse", "6.8.999")
 	assert.NoError(err)
@@ -129,10 +131,31 @@ func TestImageConfigParsing(t *testing.T) {
 	assert.Equal("localhost:5000/k8ssandra/cassandra-ubi:latest", path)
 }
 
+func TestExtendedImageConfigParsing(t *testing.T) {
+	assert := require.New(t)
+	imageConfigFile := filepath.Join("..", "..", "tests", "testdata", "image_config_parsing_more_options.yaml")
+	err := ParseImageConfig(imageConfigFile)
+	assert.NoError(err, "imageConfig parsing should succeed")
+
+	// Verify some default values are set
+	assert.NotNil(GetImageConfig())
+	assert.NotNil(GetImageConfig().Images)
+	assert.NotNil(GetImageConfig().DefaultImages)
+
+	medusaImage := GetImage("medusa")
+	assert.Equal("localhost:5005/k8ssandra/medusa:latest", medusaImage)
+	reaperImage := GetImage("reaper")
+	assert.Equal("localhost:5000/k8ssandra/reaper:latest", reaperImage)
+
+	assert.Equal(corev1.PullAlways, GetImagePullPolicy(configv1beta1.SystemLoggerImageComponent))
+	assert.Equal(corev1.PullIfNotPresent, GetImagePullPolicy(configv1beta1.CassandraImageComponent))
+}
+
 func TestDefaultRepositories(t *testing.T) {
 	assert := assert.New(t)
-	imageConfig = &configv1beta1.ImageConfig{}
+	imageConfig = configv1beta1.ImageConfig{}
 	imageConfig.Images = &configv1beta1.Images{}
+	imageConfig.DefaultImages = &configv1beta1.DefaultImages{}
 
 	path, err := GetCassandraImage("cassandra", "4.0.1")
 	assert.NoError(err)
@@ -159,8 +182,7 @@ func TestPullPolicyOverride(t *testing.T) {
 	assert.NoError(err, "imageConfig parsing should succeed")
 
 	podSpec := &corev1.PodSpec{}
-	added := AddDefaultRegistryImagePullSecrets(podSpec)
-	assert.True(added)
+	AddDefaultRegistryImagePullSecrets(podSpec)
 	assert.Equal(1, len(podSpec.ImagePullSecrets))
 	assert.Equal("my-secret-pull-registry", podSpec.ImagePullSecrets[0].Name)
 }
@@ -173,11 +195,15 @@ func TestImageConfigByteParsing(t *testing.T) {
 			ConfigBuilder: "k8ssandra/config-builder:next",
 		},
 		DefaultImages: &configv1beta1.DefaultImages{
-			CassandraImageComponent: configv1beta1.ImageComponent{
-				Repository: "k8ssandra/management-api:next",
+			ImageComponents: configv1beta1.ImageComponents{
+				configv1beta1.CassandraImageComponent: configv1beta1.ImageComponent{
+					Repository: "k8ssandra/management-api:next",
+				},
 			},
 		},
-		ImageRegistry: "localhost:5000",
+		ImagePolicy: configv1beta1.ImagePolicy{
+			ImageRegistry: "localhost:5000",
+		},
 	}
 
 	b, err := json.Marshal(imageConfig)
@@ -191,7 +217,7 @@ func TestImageConfigByteParsing(t *testing.T) {
 	require.Equal("localhost:5000", parsedImageConfig.ImageRegistry)
 	require.Equal(imageConfig.Images.SystemLogger, parsedImageConfig.Images.SystemLogger)
 	require.Equal(imageConfig.Images.ConfigBuilder, parsedImageConfig.Images.ConfigBuilder)
-	require.Equal(imageConfig.DefaultImages.CassandraImageComponent.Repository, parsedImageConfig.DefaultImages.CassandraImageComponent.Repository)
+	require.Equal(imageConfig.DefaultImages.ImageComponents[configv1beta1.CassandraImageComponent].Repository, parsedImageConfig.DefaultImages.ImageComponents[configv1beta1.CassandraImageComponent].Repository)
 	require.Equal(imageConfig.ImageRegistry, parsedImageConfig.ImageRegistry)
 
 	// And now check that images.GetImageConfig() works also..
diff --git a/pkg/reconciliation/construct_podtemplatespec.go b/pkg/reconciliation/construct_podtemplatespec.go
index 4caa80ed..4f41ae91 100644
--- a/pkg/reconciliation/construct_podtemplatespec.go
+++ b/pkg/reconciliation/construct_podtemplatespec.go
@@ -14,6 +14,7 @@ import (
 	"github.com/pkg/errors"
 
 	api "github.com/k8ssandra/cass-operator/apis/cassandra/v1beta1"
+	configapi "github.com/k8ssandra/cass-operator/apis/config/v1beta1"
 	"github.com/k8ssandra/cass-operator/pkg/cdc"
 	"github.com/k8ssandra/cass-operator/pkg/httphelper"
 	"github.com/k8ssandra/cass-operator/pkg/images"
@@ -452,6 +453,10 @@ func buildInitContainers(dc *api.CassandraDatacenter, rackName string, baseTempl
 					"config",
 					"build",
 				}
+				pullPolicy := images.GetImagePullPolicy(configapi.ClientImageComponent)
+				if pullPolicy != "" {
+					serverCfg.ImagePullPolicy = pullPolicy
+				}
 			}
 		} else {
 			// Use older config-builder
@@ -460,10 +465,10 @@ func buildInitContainers(dc *api.CassandraDatacenter, rackName string, baseTempl
 			} else {
 				serverCfg.Image = images.GetConfigBuilderImage()
 			}
-		}
-
-		if images.GetImageConfig() != nil && images.GetImageConfig().ImagePullPolicy != "" {
-			serverCfg.ImagePullPolicy = images.GetImageConfig().ImagePullPolicy
+			pullPolicy := images.GetImagePullPolicy(configapi.ConfigBuilderImageComponent)
+			if pullPolicy != "" {
+				serverCfg.ImagePullPolicy = pullPolicy
+			}
 		}
 	}
 
@@ -674,8 +679,9 @@ func buildContainers(dc *api.CassandraDatacenter, baseTemplate *corev1.PodTempla
 		}
 
 		cassContainer.Image = serverImage
-		if images.GetImageConfig() != nil && images.GetImageConfig().ImagePullPolicy != "" {
-			cassContainer.ImagePullPolicy = images.GetImageConfig().ImagePullPolicy
+		pullPolicy := images.GetImagePullPolicy(dc.Spec.ServerType)
+		if pullPolicy != "" {
+			cassContainer.ImagePullPolicy = pullPolicy
 		}
 	}
 
@@ -858,8 +864,9 @@ func buildContainers(dc *api.CassandraDatacenter, baseTemplate *corev1.PodTempla
 		} else {
 			loggerContainer.Image = images.GetSystemLoggerImage()
 		}
-		if images.GetImageConfig() != nil && images.GetImageConfig().ImagePullPolicy != "" {
-			loggerContainer.ImagePullPolicy = images.GetImageConfig().ImagePullPolicy
+		pullPolicy := images.GetImagePullPolicy(configapi.SystemLoggerImageComponent)
+		if pullPolicy != "" {
+			loggerContainer.ImagePullPolicy = pullPolicy
 		}
 	}
 
@@ -931,7 +938,7 @@ func buildPodTemplateSpec(dc *api.CassandraDatacenter, rack api.Rack, addLegacyI
 
 	// Adds custom registry pull secret if needed
 
-	_ = images.AddDefaultRegistryImagePullSecrets(&baseTemplate.Spec)
+	images.AddDefaultRegistryImagePullSecrets(&baseTemplate.Spec)
 
 	// Labels
 
diff --git a/tests/testdata/image_config_parsing_more_options.yaml b/tests/testdata/image_config_parsing_more_options.yaml
new file mode 100644
index 00000000..302e78d4
--- /dev/null
+++ b/tests/testdata/image_config_parsing_more_options.yaml
@@ -0,0 +1,45 @@
+apiVersion: config.k8ssandra.io/v1beta1
+kind: ImageConfig
+metadata:
+  name: image-config
+images:
+  system-logger: "k8ssandra/system-logger:latest"
+  config-builder: "datastax/cass-config-builder:1.0-ubi7"
+  cassandra:
+      "4.0.0": "k8ssandra/cassandra-ubi:latest"
+  dse:
+    # How to detect between two different formats?
+      "6.8.999": "datastax/dse-server-prototype:latest"
+  medusa: "k8ssandra/medusa:latest"
+  reaper: "k8ssandra/reaper:latest"
+imageRegistry: "localhost:5000"
+imagePullPolicy: Always
+imagePullSecret:
+  name: my-secret-pull-registry
+defaults:
+  # Note, suffix is ignored if repository is not set
+  cassandra:
+    repository: "k8ssandra/cass-management-api"
+    imageRegistry: "localhost:5001"
+    imagePullPolicy: IfNotPresent
+    imagePullSecret:
+      name: my-secret-pull-registry-cassandra
+  dse:
+    repository: "datastax/dse-server"
+    imageRegistry: "localhost:5002"
+    imagePullPolicy: IfNotPresent
+    imagePullSecret:
+      name: my-secret-pull-registry-dse
+    suffix: "-ubi7"
+  config-builder:
+    imageRegistry: "localhost:5003"
+    imagePullPolicy: IfNotPresent
+    imagePullSecret:
+      name: my-secret-pull-registry-builder
+  system-logger: 
+    imageRegistry: "localhost:5004"
+    imagePullPolicy: Always
+    imagePullSecret:
+      name: my-secret-pull-registry-logger
+  medusa:
+    imageRegistry: "localhost:5005"

From a4b86f0d52995a2751352859914de5485a873c3e Mon Sep 17 00:00:00 2001
From: Michael Burman <yak@iki.fi>
Date: Fri, 12 Apr 2024 17:43:44 +0300
Subject: [PATCH 2/6] Some rebase fixes

---
 pkg/images/images_test.go                             | 1 +
 pkg/reconciliation/construct_statefulset_test.go      | 2 +-
 tests/testdata/image_config_parsing.yaml              | 4 ++--
 tests/testdata/image_config_parsing_more_options.yaml | 3 ++-
 4 files changed, 6 insertions(+), 4 deletions(-)

diff --git a/pkg/images/images_test.go b/pkg/images/images_test.go
index a2182446..18a852a9 100644
--- a/pkg/images/images_test.go
+++ b/pkg/images/images_test.go
@@ -110,6 +110,7 @@ func TestImageConfigParsing(t *testing.T) {
 	assert.NotNil(GetImageConfig().Images)
 	assert.True(strings.HasPrefix(GetImageConfig().Images.SystemLogger, "k8ssandra/system-logger:"))
 	assert.True(strings.HasPrefix(GetImageConfig().Images.ConfigBuilder, "datastax/cass-config-builder:"))
+	assert.True(strings.Contains(GetImageConfig().Images.Client, "k8ssandra/k8ssandra-client:"))
 
 	assert.Equal("k8ssandra/cass-management-api", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.CassandraImageComponent].Repository)
 	assert.Equal("datastax/dse-mgmtapi-6_8", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.DSEImageComponent].Repository)
diff --git a/pkg/reconciliation/construct_statefulset_test.go b/pkg/reconciliation/construct_statefulset_test.go
index a0530b9e..63fed2d6 100644
--- a/pkg/reconciliation/construct_statefulset_test.go
+++ b/pkg/reconciliation/construct_statefulset_test.go
@@ -416,7 +416,7 @@ func Test_newStatefulSetForCassandraDatacenterWithAdditionalVolumes(t *testing.T
 		assert.Equal(t, "/var/log/cassandra", got.Spec.Template.Spec.InitContainers[0].VolumeMounts[0].MountPath)
 
 		assert.Equal(t, "server-config-init", got.Spec.Template.Spec.InitContainers[1].Name)
-		assert.Equal(t, "localhost:5000/datastax/cass-config-builder:1.0-ubi7", got.Spec.Template.Spec.InitContainers[1].Image)
+		assert.Equal(t, "localhost:5000/datastax/cass-config-builder:1.0-ubi8", got.Spec.Template.Spec.InitContainers[1].Image)
 		assert.Equal(t, 1, len(got.Spec.Template.Spec.InitContainers[1].VolumeMounts))
 		assert.Equal(t, "server-config", got.Spec.Template.Spec.InitContainers[1].VolumeMounts[0].Name)
 		assert.Equal(t, "/config", got.Spec.Template.Spec.InitContainers[1].VolumeMounts[0].MountPath)
diff --git a/tests/testdata/image_config_parsing.yaml b/tests/testdata/image_config_parsing.yaml
index 1f8caf26..15a97fd5 100644
--- a/tests/testdata/image_config_parsing.yaml
+++ b/tests/testdata/image_config_parsing.yaml
@@ -4,8 +4,8 @@ metadata:
   name: image-config
 images:
   system-logger: "k8ssandra/system-logger:latest"
-  config-builder: "datastax/cass-config-builder:1.0-ubi7"
-  k8ssandra-client: "k8ssandra/k8ssandra-client:v0.2.1"
+  config-builder: "datastax/cass-config-builder:1.0-ubi8"
+  k8ssandra-client: "k8ssandra/k8ssandra-client:v0.2.2"
   cassandra:
     "4.0.0": "k8ssandra/cassandra-ubi:latest"
   dse:
diff --git a/tests/testdata/image_config_parsing_more_options.yaml b/tests/testdata/image_config_parsing_more_options.yaml
index 302e78d4..7f44ad28 100644
--- a/tests/testdata/image_config_parsing_more_options.yaml
+++ b/tests/testdata/image_config_parsing_more_options.yaml
@@ -4,7 +4,8 @@ metadata:
   name: image-config
 images:
   system-logger: "k8ssandra/system-logger:latest"
-  config-builder: "datastax/cass-config-builder:1.0-ubi7"
+  config-builder: "datastax/cass-config-builder:1.0-ubi8"
+  k8ssandra-client: "k8ssandra/k8ssandra-client:v0.2.2"
   cassandra:
       "4.0.0": "k8ssandra/cassandra-ubi:latest"
   dse:

From da98e9698750bcaebff6bad63dd0fefe46aa99bf Mon Sep 17 00:00:00 2001
From: Michael Burman <yak@iki.fi>
Date: Wed, 17 Apr 2024 18:24:55 +0300
Subject: [PATCH 3/6] Add new field (imageNamespace) and tests for it

---
 apis/config/v1beta1/imageconfig_types.go      |  2 +
 config/manager/image_config.yaml              |  1 +
 pkg/images/images.go                          | 34 +++++++++---
 pkg/images/images_test.go                     | 54 +++++++++++++++++--
 tests/testdata/image_config_parsing.yaml      |  4 +-
 .../image_config_parsing_more_options.yaml    |  1 +
 6 files changed, 82 insertions(+), 14 deletions(-)

diff --git a/apis/config/v1beta1/imageconfig_types.go b/apis/config/v1beta1/imageconfig_types.go
index 3ff451b3..fddf1ef0 100644
--- a/apis/config/v1beta1/imageconfig_types.go
+++ b/apis/config/v1beta1/imageconfig_types.go
@@ -43,6 +43,8 @@ type ImagePolicy struct {
 	ImagePullSecret corev1.LocalObjectReference `json:"imagePullSecret,omitempty"`
 
 	ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy,omitempty"`
+
+	ImageNamespace string `json:"imageNamespace,omitempty"`
 }
 
 //+kubebuilder:object:root=true
diff --git a/config/manager/image_config.yaml b/config/manager/image_config.yaml
index 04aed7f0..83482126 100644
--- a/config/manager/image_config.yaml
+++ b/config/manager/image_config.yaml
@@ -11,6 +11,7 @@ images:
   # dse:
   #   "6.8.999": "datastax/dse-server-prototype:latest"
 # imageRegistry: "localhost:5000"
+# imageNamespace: "internal"
 # imagePullPolicy: Always
 # imagePullSecret:
 #   name: my-secret-pull-registry
diff --git a/pkg/images/images.go b/pkg/images/images.go
index e64f5240..91a132c5 100644
--- a/pkg/images/images.go
+++ b/pkg/images/images.go
@@ -88,6 +88,24 @@ func stripRegistry(image string) string {
 	}
 }
 
+func applyNamespaceOverride(image string) string {
+	namespace := GetImageConfig().ImageNamespace
+
+	if namespace == "" {
+		return image
+	}
+
+	// It can be first or second..
+	imageNoRegistry := stripRegistry(image)
+	comps := strings.Split(imageNoRegistry, "/")
+	if len(comps) > 1 {
+		noNamespace := strings.Join(comps[1:], "/")
+		return fmt.Sprintf("%s/%s", namespace, noNamespace)
+	} else {
+		return image // We can't process this correctly, we only have 1 component
+	}
+}
+
 func applyDefaultRegistryOverride(customRegistry, image string) string {
 	customRegistry = strings.TrimSuffix(customRegistry, "/")
 
@@ -117,7 +135,7 @@ func getRegistryOverride(imageType string) string {
 	return defaultRegistry
 }
 
-func applyRegistry(imageType, image string) string {
+func applyOverrides(imageType, image string) string {
 	registry := getRegistryOverride(imageType)
 
 	return applyDefaultRegistryOverride(registry, image)
@@ -179,7 +197,7 @@ func getImageComponents(serverType string) (string, string) {
 
 func GetCassandraImage(serverType, version string) (string, error) {
 	if found, image := getCassandraContainerImageOverride(serverType, version); found {
-		return applyRegistry(serverType, image), nil
+		return applyOverrides(serverType, image), nil
 	}
 
 	switch serverType {
@@ -201,15 +219,15 @@ func GetCassandraImage(serverType, version string) (string, error) {
 
 	prefix, suffix := getImageComponents(serverType)
 
-	return applyRegistry(serverType, fmt.Sprintf("%s:%s%s", prefix, version, suffix)), nil
+	return applyOverrides(serverType, fmt.Sprintf("%s:%s%s", prefix, version, suffix)), nil
 }
 
 func GetConfiguredImage(imageType, image string) string {
-	return applyRegistry(imageType, image)
+	return applyOverrides(imageType, image)
 }
 
 func GetImage(imageType string) string {
-	return applyRegistry(imageType, GetImageConfig().Images.Others[imageType])
+	return applyOverrides(imageType, GetImageConfig().Images.Others[imageType])
 }
 
 func GetImagePullPolicy(imageType string) corev1.PullPolicy {
@@ -233,15 +251,15 @@ func GetImagePullPolicy(imageType string) corev1.PullPolicy {
 }
 
 func GetConfigBuilderImage() string {
-	return applyRegistry(configv1beta1.ConfigBuilderImageComponent, GetImageConfig().Images.ConfigBuilder)
+	return applyOverrides(configv1beta1.ConfigBuilderImageComponent, GetImageConfig().Images.ConfigBuilder)
 }
 
 func GetClientImage() string {
-	return applyRegistry(configv1beta1.ClientImageComponent, GetImageConfig().Images.Client)
+	return applyOverrides(configv1beta1.ClientImageComponent, GetImageConfig().Images.Client)
 }
 
 func GetSystemLoggerImage() string {
-	return applyRegistry(configv1beta1.SystemLoggerImageComponent, GetImageConfig().Images.SystemLogger)
+	return applyOverrides(configv1beta1.SystemLoggerImageComponent, GetImageConfig().Images.SystemLogger)
 }
 
 func AddDefaultRegistryImagePullSecrets(podSpec *corev1.PodSpec, imageTypes ...string) {
diff --git a/pkg/images/images_test.go b/pkg/images/images_test.go
index 18a852a9..582f4ee8 100644
--- a/pkg/images/images_test.go
+++ b/pkg/images/images_test.go
@@ -112,8 +112,8 @@ func TestImageConfigParsing(t *testing.T) {
 	assert.True(strings.HasPrefix(GetImageConfig().Images.ConfigBuilder, "datastax/cass-config-builder:"))
 	assert.True(strings.Contains(GetImageConfig().Images.Client, "k8ssandra/k8ssandra-client:"))
 
-	assert.Equal("k8ssandra/cass-management-api", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.CassandraImageComponent].Repository)
-	assert.Equal("datastax/dse-mgmtapi-6_8", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.DSEImageComponent].Repository)
+	assert.Equal("cr.k8ssandra.io/k8ssandra/cass-management-api", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.CassandraImageComponent].Repository)
+	assert.Equal("cr.dtsx.io/datastax/dse-mgmtapi-6_8", GetImageConfig().DefaultImages.ImageComponents[configv1beta1.DSEImageComponent].Repository)
 
 	assert.Equal("localhost:5000", GetImageConfig().ImageRegistry)
 	assert.Equal(corev1.PullAlways, GetImageConfig().ImagePullPolicy)
@@ -144,9 +144,9 @@ func TestExtendedImageConfigParsing(t *testing.T) {
 	assert.NotNil(GetImageConfig().DefaultImages)
 
 	medusaImage := GetImage("medusa")
-	assert.Equal("localhost:5005/k8ssandra/medusa:latest", medusaImage)
+	assert.Equal("localhost:5005/enterprise/medusa:latest", medusaImage)
 	reaperImage := GetImage("reaper")
-	assert.Equal("localhost:5000/k8ssandra/reaper:latest", reaperImage)
+	assert.Equal("localhost:5000/enterprise/reaper:latest", reaperImage)
 
 	assert.Equal(corev1.PullAlways, GetImagePullPolicy(configv1beta1.SystemLoggerImageComponent))
 	assert.Equal(corev1.PullIfNotPresent, GetImagePullPolicy(configv1beta1.CassandraImageComponent))
@@ -188,6 +188,52 @@ func TestPullPolicyOverride(t *testing.T) {
 	assert.Equal("my-secret-pull-registry", podSpec.ImagePullSecrets[0].Name)
 }
 
+func TestRepositoryAndNamespaceOverride(t *testing.T) {
+	assert := assert.New(t)
+	imageConfig = configv1beta1.ImageConfig{}
+	imageConfig.Images = &configv1beta1.Images{}
+	imageConfig.DefaultImages = &configv1beta1.DefaultImages{}
+
+	path, err := GetCassandraImage("dse", "6.8.44")
+	assert.NoError(err)
+	assert.Equal("datastax/dse-mgmtapi-6_8:6.8.44", path)
+
+	imageConfig.ImageRegistry = "ghcr.io"
+	path, err = GetCassandraImage("dse", "6.8.44")
+	assert.NoError(err)
+	assert.Equal("ghcr.io/datastax/dse-mgmtapi-6_8:6.8.44", path)
+
+	imageConfig.ImageNamespace = "enterprise"
+	path, err = GetCassandraImage("dse", "6.8.44")
+	assert.NoError(err)
+	assert.Equal("ghcr.io/enterprise/dse-mgmtapi-6_8:6.8.44", path)
+
+	imageConfig = configv1beta1.ImageConfig{}
+	imageConfig.Images = &configv1beta1.Images{}
+	imageConfig.DefaultImages = &configv1beta1.DefaultImages{}
+	imageConfig.ImageNamespace = "enterprise"
+	path, err = GetCassandraImage("dse", "6.8.44")
+	assert.NoError(err)
+	assert.Equal("enterprise/dse-mgmtapi-6_8:6.8.44", path)
+
+	imageConfig = configv1beta1.ImageConfig{}
+	imageConfig.Images = &configv1beta1.Images{}
+	imageConfig.DefaultImages = &configv1beta1.DefaultImages{
+		ImageComponents: map[string]configv1beta1.ImageComponent{
+			configv1beta1.DSEImageComponent: {
+				Repository: "cr.dtsx.io/datastax/dse-mgmtapi-6_8",
+			},
+		},
+	}
+	path, err = GetCassandraImage("dse", "6.8.44")
+	assert.NoError(err)
+	assert.Equal("cr.dtsx.io/datastax/dse-mgmtapi-6_8:6.8.44", path)
+	imageConfig.ImageNamespace = "internal"
+	path, err = GetCassandraImage("dse", "6.8.44")
+	assert.NoError(err)
+	assert.Equal("cr.dtsx.io/internal/dse-mgmtapi-6_8:6.8.44", path)
+}
+
 func TestImageConfigByteParsing(t *testing.T) {
 	require := require.New(t)
 	imageConfig := configv1beta1.ImageConfig{
diff --git a/tests/testdata/image_config_parsing.yaml b/tests/testdata/image_config_parsing.yaml
index 15a97fd5..dfd1788d 100644
--- a/tests/testdata/image_config_parsing.yaml
+++ b/tests/testdata/image_config_parsing.yaml
@@ -17,8 +17,8 @@ imagePullSecret:
 defaults:
   # Note, postfix is ignored if repository is not set
   cassandra:
-    repository: "k8ssandra/cass-management-api"
+    repository: "cr.k8ssandra.io/k8ssandra/cass-management-api"
     suffix: "-ubi8"
   dse:
-    repository: "datastax/dse-mgmtapi-6_8"
+    repository: "cr.dtsx.io/datastax/dse-mgmtapi-6_8"
     suffix: "-ubi8"
diff --git a/tests/testdata/image_config_parsing_more_options.yaml b/tests/testdata/image_config_parsing_more_options.yaml
index 7f44ad28..f36522a6 100644
--- a/tests/testdata/image_config_parsing_more_options.yaml
+++ b/tests/testdata/image_config_parsing_more_options.yaml
@@ -17,6 +17,7 @@ imageRegistry: "localhost:5000"
 imagePullPolicy: Always
 imagePullSecret:
   name: my-secret-pull-registry
+imageNamespace: "enterprise"
 defaults:
   # Note, suffix is ignored if repository is not set
   cassandra:

From 85e07a0b3e56e5755c7448d38298109a1da82fd4 Mon Sep 17 00:00:00 2001
From: Michael Burman <yak@iki.fi>
Date: Mon, 23 Sep 2024 16:54:01 +0300
Subject: [PATCH 4/6] Add support for namespace overrides in image pulling

---
 CHANGELOG.md                                 |  1 +
 apis/config/v1beta1/imageconfig_types.go     |  2 +-
 apis/config/v1beta1/zz_generated.deepcopy.go | 13 +++--
 pkg/images/images.go                         | 53 ++++++++++++--------
 pkg/images/images_test.go                    | 13 +++--
 5 files changed, 54 insertions(+), 28 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index caef50ec..b8f5bf9f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,7 @@ Changelog for Cass Operator, new PRs should update the `main / unreleased` secti
 
 * [FEATURE] [#651](https://github.com/k8ssandra/cass-operator/issues/651) Add tsreload task for DSE deployments and ability to check if sync operation is available on the mgmt-api side
 * [ENHANCEMENT] [#532](https://github.com/k8ssandra/k8ssandra-operator/issues/532) Extend ImageConfig type to allow additional parameters for k8ssandra-operator requirements. These include per-image PullPolicy / PullSecrets as well as additional image
+* [ENHANCEMENT] [#636](https://github.com/k8ssandra/cass-operator/issues/636) Add support for new field in ImageConfig, imageNamespace. This will allow to override namespace of all images when using private registries. Setting it to empty will remove the namespace entirely.
 * [BUGFIX] [#705](https://github.com/k8ssandra/cass-operator/issues/705) Ensure ConfigSecret has annotations map before trying to set a value
 
 ## v1.22.4
diff --git a/apis/config/v1beta1/imageconfig_types.go b/apis/config/v1beta1/imageconfig_types.go
index fddf1ef0..f06afda9 100644
--- a/apis/config/v1beta1/imageconfig_types.go
+++ b/apis/config/v1beta1/imageconfig_types.go
@@ -44,7 +44,7 @@ type ImagePolicy struct {
 
 	ImagePullPolicy corev1.PullPolicy `json:"imagePullPolicy,omitempty"`
 
-	ImageNamespace string `json:"imageNamespace,omitempty"`
+	ImageNamespace *string `json:"imageNamespace,omitempty"`
 }
 
 //+kubebuilder:object:root=true
diff --git a/apis/config/v1beta1/zz_generated.deepcopy.go b/apis/config/v1beta1/zz_generated.deepcopy.go
index cb23b64a..e7903da9 100644
--- a/apis/config/v1beta1/zz_generated.deepcopy.go
+++ b/apis/config/v1beta1/zz_generated.deepcopy.go
@@ -31,7 +31,7 @@ func (in *DefaultImages) DeepCopyInto(out *DefaultImages) {
 		in, out := &in.ImageComponents, &out.ImageComponents
 		*out = make(ImageComponents, len(*in))
 		for key, val := range *in {
-			(*out)[key] = val
+			(*out)[key] = *val.DeepCopy()
 		}
 	}
 }
@@ -49,7 +49,7 @@ func (in *DefaultImages) DeepCopy() *DefaultImages {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *ImageComponent) DeepCopyInto(out *ImageComponent) {
 	*out = *in
-	out.ImagePolicy = in.ImagePolicy
+	in.ImagePolicy.DeepCopyInto(&out.ImagePolicy)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ImageComponent.
@@ -68,7 +68,7 @@ func (in ImageComponents) DeepCopyInto(out *ImageComponents) {
 		in := &in
 		*out = make(ImageComponents, len(*in))
 		for key, val := range *in {
-			(*out)[key] = val
+			(*out)[key] = *val.DeepCopy()
 		}
 	}
 }
@@ -97,7 +97,7 @@ func (in *ImageConfig) DeepCopyInto(out *ImageConfig) {
 		*out = new(DefaultImages)
 		(*in).DeepCopyInto(*out)
 	}
-	out.ImagePolicy = in.ImagePolicy
+	in.ImagePolicy.DeepCopyInto(&out.ImagePolicy)
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ImageConfig.
@@ -122,6 +122,11 @@ func (in *ImageConfig) DeepCopyObject() runtime.Object {
 func (in *ImagePolicy) DeepCopyInto(out *ImagePolicy) {
 	*out = *in
 	out.ImagePullSecret = in.ImagePullSecret
+	if in.ImageNamespace != nil {
+		in, out := &in.ImageNamespace, &out.ImageNamespace
+		*out = new(string)
+		**out = **in
+	}
 }
 
 // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ImagePolicy.
diff --git a/pkg/images/images.go b/pkg/images/images.go
index 91a132c5..bd575589 100644
--- a/pkg/images/images.go
+++ b/pkg/images/images.go
@@ -78,43 +78,42 @@ func IsHCDVersionSupported(version string) bool {
 	return validVersions.MatchString(version)
 }
 
-func stripRegistry(image string) string {
+func splitRegistry(image string) (registry string, imageNoRegistry string) {
 	comps := strings.Split(image, "/")
 
 	if len(comps) > 1 && (strings.Contains(comps[0], ".") || strings.Contains(comps[0], ":")) {
-		return strings.Join(comps[1:], "/")
+		return comps[0], strings.Join(comps[1:], "/")
 	} else {
-		return image
+		return "", image
 	}
 }
 
-func applyNamespaceOverride(image string) string {
-	namespace := GetImageConfig().ImageNamespace
-
-	if namespace == "" {
-		return image
+// applyNamespaceOverride takes only input without registry
+func applyNamespaceOverride(imageNoRegistry string) string {
+	if GetImageConfig().ImageNamespace == nil {
+		return imageNoRegistry
 	}
 
-	// It can be first or second..
-	imageNoRegistry := stripRegistry(image)
+	namespace := *GetImageConfig().ImageNamespace
+
 	comps := strings.Split(imageNoRegistry, "/")
 	if len(comps) > 1 {
 		noNamespace := strings.Join(comps[1:], "/")
+		if namespace == "" {
+			return noNamespace
+		}
 		return fmt.Sprintf("%s/%s", namespace, noNamespace)
 	} else {
-		return image // We can't process this correctly, we only have 1 component
+		// We can't process this correctly, we only have 1 component. We do not support a case where the original image has no registry and no namespace.
+		return imageNoRegistry
 	}
 }
 
-func applyDefaultRegistryOverride(customRegistry, image string) string {
-	customRegistry = strings.TrimSuffix(customRegistry, "/")
-
+func applyDefaultRegistryOverride(customRegistry, imageNoRegistry string) string {
 	if customRegistry == "" {
-		return image
-	} else {
-		imageNoRegistry := stripRegistry(image)
-		return fmt.Sprintf("%s/%s", customRegistry, imageNoRegistry)
+		return imageNoRegistry
 	}
+	return fmt.Sprintf("%s/%s", customRegistry, imageNoRegistry)
 }
 
 func getRegistryOverride(imageType string) string {
@@ -136,9 +135,23 @@ func getRegistryOverride(imageType string) string {
 }
 
 func applyOverrides(imageType, image string) string {
-	registry := getRegistryOverride(imageType)
+	registryOverride := getRegistryOverride(imageType)
+	registryOverride = strings.TrimSuffix(registryOverride, "/")
+	registry, imageNoRegistry := splitRegistry(image)
+
+	if registryOverride == "" && GetImageConfig().ImageNamespace == nil {
+		return image
+	}
+
+	if GetImageConfig().ImageNamespace != nil {
+		imageNoRegistry = applyNamespaceOverride(imageNoRegistry)
+	}
+
+	if registryOverride != "" {
+		return applyDefaultRegistryOverride(registryOverride, imageNoRegistry)
+	}
 
-	return applyDefaultRegistryOverride(registry, image)
+	return applyDefaultRegistryOverride(registry, imageNoRegistry)
 }
 
 func GetImageConfig() *configv1beta1.ImageConfig {
diff --git a/pkg/images/images_test.go b/pkg/images/images_test.go
index 582f4ee8..cbc473a8 100644
--- a/pkg/images/images_test.go
+++ b/pkg/images/images_test.go
@@ -14,6 +14,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/utils/ptr"
 
 	configv1beta1 "github.com/k8ssandra/cass-operator/apis/config/v1beta1"
 )
@@ -203,7 +204,7 @@ func TestRepositoryAndNamespaceOverride(t *testing.T) {
 	assert.NoError(err)
 	assert.Equal("ghcr.io/datastax/dse-mgmtapi-6_8:6.8.44", path)
 
-	imageConfig.ImageNamespace = "enterprise"
+	imageConfig.ImageNamespace = ptr.To[string]("enterprise")
 	path, err = GetCassandraImage("dse", "6.8.44")
 	assert.NoError(err)
 	assert.Equal("ghcr.io/enterprise/dse-mgmtapi-6_8:6.8.44", path)
@@ -211,7 +212,7 @@ func TestRepositoryAndNamespaceOverride(t *testing.T) {
 	imageConfig = configv1beta1.ImageConfig{}
 	imageConfig.Images = &configv1beta1.Images{}
 	imageConfig.DefaultImages = &configv1beta1.DefaultImages{}
-	imageConfig.ImageNamespace = "enterprise"
+	imageConfig.ImageNamespace = ptr.To[string]("enterprise")
 	path, err = GetCassandraImage("dse", "6.8.44")
 	assert.NoError(err)
 	assert.Equal("enterprise/dse-mgmtapi-6_8:6.8.44", path)
@@ -228,10 +229,16 @@ func TestRepositoryAndNamespaceOverride(t *testing.T) {
 	path, err = GetCassandraImage("dse", "6.8.44")
 	assert.NoError(err)
 	assert.Equal("cr.dtsx.io/datastax/dse-mgmtapi-6_8:6.8.44", path)
-	imageConfig.ImageNamespace = "internal"
+
+	imageConfig.ImageNamespace = ptr.To[string]("internal")
 	path, err = GetCassandraImage("dse", "6.8.44")
 	assert.NoError(err)
 	assert.Equal("cr.dtsx.io/internal/dse-mgmtapi-6_8:6.8.44", path)
+
+	imageConfig.ImageNamespace = ptr.To[string]("")
+	path, err = GetCassandraImage("dse", "6.8.44")
+	assert.NoError(err)
+	assert.Equal("cr.dtsx.io/dse-mgmtapi-6_8:6.8.44", path)
 }
 
 func TestImageConfigByteParsing(t *testing.T) {

From fb9eb8cef3f0852df4b56419b5f9e4e39a34cc94 Mon Sep 17 00:00:00 2001
From: Michael Burman <yak@iki.fi>
Date: Fri, 11 Oct 2024 10:39:53 +0300
Subject: [PATCH 5/6] Add HCD version override test

---
 apis/config/v1beta1/imageconfig_types.go     |  2 ++
 apis/config/v1beta1/zz_generated.deepcopy.go |  7 +++++++
 pkg/images/images.go                         |  5 +++++
 pkg/images/images_test.go                    | 20 +++++++++++++++++---
 4 files changed, 31 insertions(+), 3 deletions(-)

diff --git a/apis/config/v1beta1/imageconfig_types.go b/apis/config/v1beta1/imageconfig_types.go
index f06afda9..14b3a368 100644
--- a/apis/config/v1beta1/imageconfig_types.go
+++ b/apis/config/v1beta1/imageconfig_types.go
@@ -57,6 +57,8 @@ type Images struct {
 
 	DSEVersions map[string]string `json:"dse,omitempty"`
 
+	HCDVersions map[string]string `json:"hcd,omitempty"`
+
 	SystemLogger string `json:"system-logger,omitempty"`
 
 	Client string `json:"k8ssandra-client,omitempty"`
diff --git a/apis/config/v1beta1/zz_generated.deepcopy.go b/apis/config/v1beta1/zz_generated.deepcopy.go
index e7903da9..8559fadc 100644
--- a/apis/config/v1beta1/zz_generated.deepcopy.go
+++ b/apis/config/v1beta1/zz_generated.deepcopy.go
@@ -157,6 +157,13 @@ func (in *Images) DeepCopyInto(out *Images) {
 			(*out)[key] = val
 		}
 	}
+	if in.HCDVersions != nil {
+		in, out := &in.HCDVersions, &out.HCDVersions
+		*out = make(map[string]string, len(*in))
+		for key, val := range *in {
+			(*out)[key] = val
+		}
+	}
 	if in.Others != nil {
 		in, out := &in.Others, &out.Others
 		*out = make(map[string]string, len(*in))
diff --git a/pkg/images/images.go b/pkg/images/images.go
index bd575589..07bca5fa 100644
--- a/pkg/images/images.go
+++ b/pkg/images/images.go
@@ -173,6 +173,11 @@ func getCassandraContainerImageOverride(serverType, version string) (bool, strin
 				return true, value
 			}
 		}
+		if serverType == "hcd" {
+			if value, found := images.HCDVersions[version]; found {
+				return true, value
+			}
+		}
 	}
 	return false, ""
 }
diff --git a/pkg/images/images_test.go b/pkg/images/images_test.go
index cbc473a8..04fbb452 100644
--- a/pkg/images/images_test.go
+++ b/pkg/images/images_test.go
@@ -61,14 +61,28 @@ func TestCassandraOverride(t *testing.T) {
 	assert.NoError(err, "getting Cassandra image with overrides should succeed")
 	assert.Equal(fmt.Sprintf("ghcr.io/%s", customImageName), cassImage)
 
-	customImageWithOrg := "k8ssandra/cass-management-api:4.0.0"
+	customImageNamespace := "modified"
 	imageConfig.Images.CassandraVersions = map[string]string{
-		"4.0.0": fmt.Sprintf("us-docker.pkg.dev/%s", customImageWithOrg),
+		"4.0.0": fmt.Sprintf("us-docker.pkg.dev/%s/cass-management-api:4.0.0", customImageNamespace),
+	}
+	imageConfig.Images.DSEVersions = map[string]string{
+		"6.8.0": fmt.Sprintf("us-docker.pkg.dev/%s/dse-mgmtapi-6_8:6.8.0", customImageNamespace),
+	}
+	imageConfig.Images.HCDVersions = map[string]string{
+		"1.0.0": fmt.Sprintf("us-docker.pkg.dev/%s/hcd:1.0.0", customImageNamespace),
 	}
 
 	cassImage, err = GetCassandraImage("cassandra", "4.0.0")
 	assert.NoError(err, "getting Cassandra image with overrides should succeed")
-	assert.Equal(fmt.Sprintf("ghcr.io/%s", customImageWithOrg), cassImage)
+	assert.Equal("ghcr.io/modified/cass-management-api:4.0.0", cassImage)
+
+	cassImage, err = GetCassandraImage("dse", "6.8.0")
+	assert.NoError(err, "getting Cassandra image with overrides should succeed")
+	assert.Equal("ghcr.io/modified/dse-mgmtapi-6_8:6.8.0", cassImage)
+
+	cassImage, err = GetCassandraImage("hcd", "1.0.0")
+	assert.NoError(err, "getting Cassandra image with overrides should succeed")
+	assert.Equal("ghcr.io/modified/hcd:1.0.0", cassImage)
 }
 
 func TestDefaultImageConfigParsing(t *testing.T) {

From 74ec96d2be14914fc8beb541ded735fd771b94aa Mon Sep 17 00:00:00 2001
From: Michael Burman <yak@iki.fi>
Date: Tue, 22 Oct 2024 09:56:34 +0300
Subject: [PATCH 6/6] Fix UnmarshalJSON to remove hcd field before doing double
 unmarshalling

---
 apis/config/v1beta1/imageconfig_types.go              | 1 +
 tests/testdata/image_config_parsing_more_options.yaml | 3 ++-
 2 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/apis/config/v1beta1/imageconfig_types.go b/apis/config/v1beta1/imageconfig_types.go
index 14b3a368..221ee0c3 100644
--- a/apis/config/v1beta1/imageconfig_types.go
+++ b/apis/config/v1beta1/imageconfig_types.go
@@ -84,6 +84,7 @@ func (i *Images) UnmarshalJSON(b []byte) error {
 
 	delete(otherFields, CassandraImageComponent)
 	delete(otherFields, DSEImageComponent)
+	delete(otherFields, HCDImageComponent)
 	delete(otherFields, SystemLoggerImageComponent)
 	delete(otherFields, ConfigBuilderImageComponent)
 	delete(otherFields, ClientImageComponent)
diff --git a/tests/testdata/image_config_parsing_more_options.yaml b/tests/testdata/image_config_parsing_more_options.yaml
index f36522a6..5e21bd52 100644
--- a/tests/testdata/image_config_parsing_more_options.yaml
+++ b/tests/testdata/image_config_parsing_more_options.yaml
@@ -9,8 +9,9 @@ images:
   cassandra:
       "4.0.0": "k8ssandra/cassandra-ubi:latest"
   dse:
-    # How to detect between two different formats?
       "6.8.999": "datastax/dse-server-prototype:latest"
+  hcd:
+      "1.0.0": "datastax/hcd:latest"
   medusa: "k8ssandra/medusa:latest"
   reaper: "k8ssandra/reaper:latest"
 imageRegistry: "localhost:5000"