From 6c3bf7819d71241753b5ad7622cbc7a21ab32113 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 7a65ee38..ef9fbb2c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,6 +14,7 @@ Changelog for Cass Operator, new PRs should update the `main / unreleased` secti
 * [CHANGE] [#720](https://github.com/k8ssandra/cass-operator/issues/720) Always use ObjectMeta.Name for the PodDisruptionBudget resource name, not the DatacenterName
 * [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] [#722](https://github.com/k8ssandra/cass-operator/issues/722) Add back the ability to track cleanup task before marking scale up as done. This is controlled by an annotation cassandra.datastax.com/track-cleanup-tasks
+* [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. Add ability to override namespace of all components and to override single HCD version 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 07eb9a69..eaf4c4eb 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 38ea804a..537299d4 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 025b3edd790c1ea6a97c6acaa9ae7a15f36aadc9 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 eaf4c4eb..eae898fe 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 f53983c1..fd1ccf8a 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 5f6a382f..c0c62a4f 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 a9d8e2eb5b24f1b9f859b89ed67c94eea60a0052 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 2e85d348..60542c60 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 eae898fe..e1df3edd 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 c0c62a4f..15d6d5cd 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: "-ubi"
   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 0728aef316718eeed945bc788c39cd734147eac0 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                                 |  3 +-
 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, 55 insertions(+), 29 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index ef9fbb2c..346c360c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -14,7 +14,8 @@ Changelog for Cass Operator, new PRs should update the `main / unreleased` secti
 * [CHANGE] [#720](https://github.com/k8ssandra/cass-operator/issues/720) Always use ObjectMeta.Name for the PodDisruptionBudget resource name, not the DatacenterName
 * [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] [#722](https://github.com/k8ssandra/cass-operator/issues/722) Add back the ability to track cleanup task before marking scale up as done. This is controlled by an annotation cassandra.datastax.com/track-cleanup-tasks
-* [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. Add ability to override namespace of all components and to override single HCD version image. 
+* [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.  Also add ability to override a single HCD version 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 e1df3edd..e064f941 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 3cb7b511a3543281c2154fe7dd56007c910ea2ff 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 e064f941..acec66fc 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 4f26efa00c81ddf165607999c95f01fd2ebb0ea4 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"