Repository: banzaicloud/kafka-operator
Pull request #361 (2020-04-24 16:22:40)

Allow override cluster domain
| Q               | A
| --------------- | ---
| Bug fix?        | no
| New feature?    | yes
| API breaks?     | no (I added a new field that defaults to the existing behavior, not sure if that counts)
| Deprecations?   | no
| Related tickets | 
| License         | Apache 2.0


### What's in this PR?
I've added a field to the KafkaCluster for `clusterDomain`, which defaults to `cluster.local` but can now be overridden.


### Why?
My clusters do not use `cluster.local`, we have a different cluster domain for each cluster. This PR allows me to set the cluster domain, but keeps the existing functionality for everyone else.


### Checklist
<!-- Place an '[x]' (no spaces) in all applicable fields. Please remove unrelated fields. -->

- [x] Implementation tested
- [x] User guide and development docs updated (if needed)
 (I couldn't find anywhere where the settings in the KafkaCluster were documented, so I just added a description field in the CRD)

### To Do
<!-- (Please remove this section if you don't need it.) -->
- [x] If the PR is not complete but you want to discuss the approach, list what remains to be done here

1) Is KafkaCluster.Spec the right place for this? It's likely this will be the same for all clusters in a given deployment, I couldn't find an easy place to put that.
2) I recognize there's potential for confusion regarding the k8s term "cluster" and the kafka term "cluster". I could make the setting something more like `kubernetesClusterDomain` if that helps?
3) I noticed that most of the functions in `pkg/scale/scale.go` took the same arguments, so I had to copy them around many times. Would it be worth it to encapsulate the parameters into a `struct` and change these functions into methods?

CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/banzaicloud/kafka-operator?pullRequest=361) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/banzaicloud/kafka-operator?pullRequest=361) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/banzaicloud/kafka-operator?pullRequest=361) it.</sub>
lukemassa: @baluchicken Great sounds good!

Do you want me to do the refactor of `scale.go` as part of this PR or in a separate one?
baluchicken: Please do in this PR. Thanks!
baluchicken: Hi, @lukemassa any update on this? Do you need any further help?
lukemassa: Hi sorry I got distracted with other things, I should be able to make those edits by the end of the week
lukemassa: OK I made some progress. I created a `CruiseControlScaler` struct and converted all the functions there to methods. 

I'm not totally happy about the state now, though, since a lot of places especially in `cruisecontroltask_controller.go` I create an object and immediately use it. I was thinking of making CruiseControlScaler a property on KafkaCluster, but that felt a bit strange for such a central object. Any suggestions where to go from here?
baluchicken: Thanks for the "refactor". I see what you mean, but I think let's address that in a different PR. Also we need some time to think this through. I don't like making a CruiseControlScaler a property on KafkaCluster. Let's leave this one as is, I don't want to delay merging this PR any more:) Can you please create an issue about this, so we can follow up? Thanks.
lukemassa: I added the issue (linked above).

Thanks!

Patch:
diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
index f37789a2..4aacd3f3 100644
--- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
+++ b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml
@@ -1489,6 +1489,8 @@ spec:
                     type: string
                   type: object
               type: object
+            kubernetesClusterDomain:
+              type: string
             listenersConfig:
               description: ListenersConfig defines the Kafka listener types
               properties:
diff --git a/controllers/cruisecontroltask_controller.go b/controllers/cruisecontroltask_controller.go
index 16d10a56..9dacfde8 100644
--- a/controllers/cruisecontroltask_controller.go
+++ b/controllers/cruisecontroltask_controller.go
@@ -141,7 +141,8 @@ func (r *CruiseControlTaskReconciler) Reconcile(request ctrl.Request) (ctrl.Resu
 		err = r.handlePodDeleteCCTask(instance, brokersWithDownscaleRequired, log)
 	} else if len(brokersWithDiskRebalanceRequired) > 0 {
 		// create new cc task, set status to running
-		taskId, startTime, err = scale.RebalanceDisks(brokersWithDiskRebalanceRequired, instance.Namespace, instance.Spec.CruiseControlConfig.CruiseControlEndpoint, instance.Name)
+		cc := scale.NewCruiseControlScaler(instance.Namespace, instance.Spec.GetKubernetesClusterDomain(), instance.Spec.CruiseControlConfig.CruiseControlEndpoint, instance.Name)
+		taskId, startTime, err = cc.RebalanceDisks(brokersWithDiskRebalanceRequired)
 		if err != nil {
 			log.Error(err, "executing disk rebalance cc task failed")
 		} else {
@@ -192,7 +193,8 @@ func (r *CruiseControlTaskReconciler) Reconcile(request ctrl.Request) (ctrl.Resu
 	return reconciled()
 }
 func (r *CruiseControlTaskReconciler) handlePodAddCCTask(kafkaCluster *v1beta1.KafkaCluster, brokerIds []string, log logr.Logger) error {
-	uTaskId, taskStartTime, scaleErr := scale.UpScaleCluster(brokerIds, kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	cc := scale.NewCruiseControlScaler(kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(), kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	uTaskId, taskStartTime, scaleErr := cc.UpScaleCluster(brokerIds)
 	if scaleErr != nil {
 		log.Info("cruise control communication error during upscaling broker(s)", "brokerId(s)", brokerIds)
 		return errorfactory.New(errorfactory.CruiseControlNotReady{}, scaleErr, fmt.Sprintf("broker id(s): %s", brokerIds))
@@ -206,8 +208,9 @@ func (r *CruiseControlTaskReconciler) handlePodAddCCTask(kafkaCluster *v1beta1.K
 	return nil
 }
 func (r *CruiseControlTaskReconciler) handlePodDeleteCCTask(kafkaCluster *v1beta1.KafkaCluster, brokerIds []string, log logr.Logger) error {
-	uTaskId, taskStartTime, err := scale.DownsizeCluster(brokerIds,
-		kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+
+	cc := scale.NewCruiseControlScaler(kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(), kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	uTaskId, taskStartTime, err := cc.DownsizeCluster(brokerIds)
 	if err != nil {
 		log.Info("cruise control communication error during downscaling broker(s)", "id(s)", brokerIds)
 		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, fmt.Sprintf("broker(s) id(s): %s", brokerIds))
@@ -243,7 +246,8 @@ func (r *CruiseControlTaskReconciler) checkCCTaskState(kafkaCluster *v1beta1.Kaf
 	}
 
 	// check cc task status
-	status, err := scale.GetCCTaskState(ccTaskId, kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	cc := scale.NewCruiseControlScaler(kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(), kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	status, err := cc.GetCCTaskState(ccTaskId)
 	if err != nil {
 		log.Info("Cruise control communication error checking running task", "taskId", ccTaskId)
 		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
@@ -324,7 +328,9 @@ func (r *CruiseControlTaskReconciler) checkCCTaskState(kafkaCluster *v1beta1.Kaf
 	// task timed out
 	if len(brokersWithTimedOutCCTask) > 0 {
 		log.Info("Killing Cruise control task", "taskId", ccTaskId)
-		err = scale.KillCCTask(kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+		cc := scale.NewCruiseControlScaler(kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(), kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+		err = cc.KillCCTask()
+
 		if err != nil {
 			return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
 		}
@@ -374,7 +380,8 @@ func (r *CruiseControlTaskReconciler) checkVolumeCCTaskState(kafkaCluster *v1bet
 	}
 
 	// check cc task status
-	status, err := scale.GetCCTaskState(ccTaskId, kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	cc := scale.NewCruiseControlScaler(kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(), kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+	status, err := cc.GetCCTaskState(ccTaskId)
 	if err != nil {
 		log.Info("Cruise control communication error checking running task", "taskId", ccTaskId)
 		return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
@@ -465,7 +472,9 @@ func (r *CruiseControlTaskReconciler) checkVolumeCCTaskState(kafkaCluster *v1bet
 	// task timed out
 	if len(brokersWithTimedOutCCTask) > 0 {
 		log.Info("Killing Cruise control task", "taskId", ccTaskId)
-		err = scale.KillCCTask(kafkaCluster.Namespace, kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+		cc := scale.NewCruiseControlScaler(kafkaCluster.Namespace, kafkaCluster.Spec.GetKubernetesClusterDomain(), kafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, kafkaCluster.Name)
+
+		err = cc.KillCCTask()
 		if err != nil {
 			return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, "cc communication error")
 		}
diff --git a/internal/alertmanager/currentalert/process.go b/internal/alertmanager/currentalert/process.go
index 1dcce684..068297f3 100644
--- a/internal/alertmanager/currentalert/process.go
+++ b/internal/alertmanager/currentalert/process.go
@@ -263,7 +263,8 @@ func downScale(log logr.Logger, labels model.LabelSet, client client.Client) err
 		return nil
 	}
 
-	brokerId, err := scale.GetBrokerIDWithLeastPartition(string(labels["namespace"]), cr.Spec.CruiseControlConfig.CruiseControlEndpoint, cr.Name)
+	cc := scale.NewCruiseControlScaler(string(labels["namespace"]), cr.Spec.GetKubernetesClusterDomain(), cr.Spec.CruiseControlConfig.CruiseControlEndpoint, cr.Name)
+	brokerId, err := cc.GetBrokerIDWithLeastPartition()
 	if err != nil {
 		return err
 	}
diff --git a/pkg/kafkaclient/config.go b/pkg/kafkaclient/config.go
index 4ee260c8..880d1d6c 100644
--- a/pkg/kafkaclient/config.go
+++ b/pkg/kafkaclient/config.go
@@ -66,15 +66,17 @@ func determineInternalListenerForInnerCom(internalListeners []v1beta1.InternalLi
 
 func generateKafkaAddress(cluster *v1beta1.KafkaCluster) string {
 	if cluster.Spec.HeadlessServiceEnabled {
-		return fmt.Sprintf("%s.%s.svc.cluster.local:%d",
+		return fmt.Sprintf("%s.%s.svc.%s:%d",
 			fmt.Sprintf(kafka.HeadlessServiceTemplate, cluster.Name),
 			cluster.Namespace,
+			cluster.Spec.GetKubernetesClusterDomain(),
 			cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].ContainerPort,
 		)
 	}
-	return fmt.Sprintf("%s.%s.svc.cluster.local:%d",
+	return fmt.Sprintf("%s.%s.svc.%s:%d",
 		fmt.Sprintf(kafka.AllBrokerServiceTemplate, cluster.Name),
 		cluster.Namespace,
+		cluster.Spec.GetKubernetesClusterDomain(),
 		cluster.Spec.ListenersConfig.InternalListeners[determineInternalListenerForInnerCom(cluster.Spec.ListenersConfig.InternalListeners)].ContainerPort,
 	)
 }
diff --git a/pkg/resources/envoy/configmap.go b/pkg/resources/envoy/configmap.go
index 2270c17b..ef135316 100644
--- a/pkg/resources/envoy/configmap.go
+++ b/pkg/resources/envoy/configmap.go
@@ -103,7 +103,7 @@ func generateEnvoyConfig(kc *v1beta1.KafkaCluster, log logr.Logger) string {
 				{
 					Address: &envoycore.Address_SocketAddress{
 						SocketAddress: &envoycore.SocketAddress{
-							Address: fmt.Sprintf("%s-%d.%s-headless.%s.svc.cluster.local", kc.Name, broker.Id, kc.Name, kc.Namespace),
+							Address: fmt.Sprintf("%s-%d.%s-headless.%s.svc.%s", kc.Name, broker.Id, kc.Name, kc.Namespace, kc.Spec.GetKubernetesClusterDomain()),
 							PortSpecifier: &envoycore.SocketAddress_PortValue{
 								PortValue: uint32(kc.Spec.ListenersConfig.ExternalListeners[0].ContainerPort),
 							},
diff --git a/pkg/resources/kafka/configmap.go b/pkg/resources/kafka/configmap.go
index cdcf3150..a8e1fdee 100644
--- a/pkg/resources/kafka/configmap.go
+++ b/pkg/resources/kafka/configmap.go
@@ -83,9 +83,9 @@ func (r *Reconciler) getConfigString(bConfig *v1beta1.BrokerConfig, id int32, lo
 		"ListenerConfig":                     generateListenerSpecificConfig(&r.KafkaCluster.Spec.ListenersConfig, log),
 		"SSLEnabledForInternalCommunication": r.KafkaCluster.Spec.ListenersConfig.SSLSecrets != nil && util.IsSSLEnabledForInternalCommunication(r.KafkaCluster.Spec.ListenersConfig.InternalListeners),
 		"ZookeeperConnectString":             zookeeperutils.PrepareConnectionAddress(r.KafkaCluster.Spec.ZKAddresses, r.KafkaCluster.Spec.GetZkPath()),
-		"CruiseControlBootstrapServers":      getInternalListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners, id, r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
+		"CruiseControlBootstrapServers":      getInternalListener(r.KafkaCluster.Spec.ListenersConfig.InternalListeners, id, r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
 		"StorageConfig":                      generateStorageConfig(bConfig.StorageConfigs),
-		"AdvertisedListenersConfig":          generateAdvertisedListenerConfig(id, r.KafkaCluster.Spec.ListenersConfig, loadBalancerIPs, r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
+		"AdvertisedListenersConfig":          generateAdvertisedListenerConfig(id, r.KafkaCluster.Spec.ListenersConfig, loadBalancerIPs, r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.Namespace, r.KafkaCluster.Name, r.KafkaCluster.Spec.HeadlessServiceEnabled),
 		"SuperUsers":                         strings.Join(generateSuperUsers(superUsers), ";"),
 		"ServerKeystorePath":                 serverKeystorePath,
 		"ClientKeystorePath":                 clientKeystorePath,
@@ -121,7 +121,7 @@ func (r *Reconciler) configMap(id int32, brokerConfig *v1beta1.BrokerConfig, loa
 	}
 }
 
-func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadBalancerIPs []string, namespace, crName string, headlessServiceEnabled bool) string {
+func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadBalancerIPs []string, domain, namespace, crName string, headlessServiceEnabled bool) string {
 	advertisedListenerConfig := []string{}
 	for _, eListener := range l.ExternalListeners {
 		// use first element of loadBalancerIPs slice for external listener name
@@ -131,10 +131,10 @@ func generateAdvertisedListenerConfig(id int32, l v1beta1.ListenersConfig, loadB
 	for _, iListener := range l.InternalListeners {
 		if headlessServiceEnabled {
 			advertisedListenerConfig = append(advertisedListenerConfig,
-				fmt.Sprintf("%s://%s-%d.%s-headless.%s.svc.cluster.local:%d", strings.ToUpper(iListener.Name), crName, id, crName, namespace, iListener.ContainerPort))
+				fmt.Sprintf("%s://%s-%d.%s-headless.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, crName, namespace, domain, iListener.ContainerPort))
 		} else {
 			advertisedListenerConfig = append(advertisedListenerConfig,
-				fmt.Sprintf("%s://%s-%d.%s.svc.cluster.local:%d", strings.ToUpper(iListener.Name), crName, id, namespace, iListener.ContainerPort))
+				fmt.Sprintf("%s://%s-%d.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, namespace, domain, iListener.ContainerPort))
 		}
 	}
 	return fmt.Sprintf("advertised.listeners=%s\n", strings.Join(advertisedListenerConfig, ","))
@@ -190,16 +190,16 @@ func generateListenerSpecificConfig(l *v1beta1.ListenersConfig, log logr.Logger)
 		"listeners=" + strings.Join(listenerConfig, ",") + "\n"
 }
 
-func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32, namespace, crName string, headlessServiceEnabled bool) string {
+func getInternalListener(iListeners []v1beta1.InternalListenerConfig, id int32, domain, namespace, crName string, headlessServiceEnabled bool) string {
 
 	internalListener := ""
 
 	for _, iListener := range iListeners {
 		if iListener.UsedForInnerBrokerCommunication {
 			if headlessServiceEnabled {
-				internalListener = fmt.Sprintf("%s://%s-%d.%s-headless.%s.svc.cluster.local:%d", strings.ToUpper(iListener.Name), crName, id, crName, namespace, iListener.ContainerPort)
+				internalListener = fmt.Sprintf("%s://%s-%d.%s-headless.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, crName, namespace, domain, iListener.ContainerPort)
 			} else {
-				internalListener = fmt.Sprintf("%s://%s-%d.%s.svc.cluster.local:%d", strings.ToUpper(iListener.Name), crName, id, namespace, iListener.ContainerPort)
+				internalListener = fmt.Sprintf("%s://%s-%d.%s.svc.%s:%d", strings.ToUpper(iListener.Name), crName, id, namespace, domain, iListener.ContainerPort)
 			}
 		}
 	}
diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go
index 78940c0d..1e230429 100644
--- a/pkg/resources/kafka/kafka.go
+++ b/pkg/resources/kafka/kafka.go
@@ -322,9 +322,9 @@ OUTERLOOP:
 
 	if len(deletedBrokers) > 0 {
 		if !arePodsAlreadyDeleted(deletedBrokers, log) {
-			liveBrokers, err := scale.GetLiveKafkaBrokersFromCruiseControl(
-				generateBrokerIdsFromPodSlice(deletedBrokers),
-				r.KafkaCluster.Namespace, r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
+			cc := scale.NewCruiseControlScaler(r.KafkaCluster.Namespace, r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint, r.KafkaCluster.Name)
+			liveBrokers, err := cc.GetLiveKafkaBrokersFromCruiseControl(generateBrokerIdsFromPodSlice(deletedBrokers))
+
 			if err != nil {
 				log.Error(err, "could not query CC for ALIVE brokers")
 				return errorfactory.New(errorfactory.CruiseControlNotReady{}, err, fmt.Sprintf("broker(s) id(s): %s", strings.Join(liveBrokers, ",")))
diff --git a/pkg/scale/scale.go b/pkg/scale/scale.go
index aba0016a..f3136613 100644
--- a/pkg/scale/scale.go
+++ b/pkg/scale/scale.go
@@ -46,22 +46,38 @@ var errCruiseControlNotReturned200 = errors.New("non 200 response from cruise-co
 
 var log = logf.Log.WithName("cruise-control-methods")
 
-func generateUrlForCC(action, namespace string, options map[string]string, ccEndpoint, clusterName string) string {
+type CruiseControlScaler struct {
+	namespace               string
+	kubernetesClusterDomain string
+	endpoint                string
+	clusterName             string
+}
+
+func NewCruiseControlScaler(namespace, kubernetesClusterDomain, endpoint, clusterName string) CruiseControlScaler {
+	return CruiseControlScaler{
+		namespace:               namespace,
+		kubernetesClusterDomain: kubernetesClusterDomain,
+		endpoint:                endpoint,
+		clusterName:             clusterName,
+	}
+}
+
+func (cc *CruiseControlScaler) generateUrlForCC(action string, options map[string]string) string {
 	optionURL := ""
 	for option, value := range options {
 		optionURL = optionURL + option + "=" + value + "&"
 	}
-	if ccEndpoint != "" {
-		return "http://" + ccEndpoint + "/" + basePath + "/" + action + "?" + strings.TrimSuffix(optionURL, "&")
+	if cc.endpoint != "" {
+		return "http://" + cc.endpoint + "/" + basePath + "/" + action + "?" + strings.TrimSuffix(optionURL, "&")
 	}
-	return "http://" + fmt.Sprintf(serviceNameTemplate, clusterName) + "." + namespace + ".svc.cluster.local:8090/" + basePath + "/" + action + "?" + strings.TrimSuffix(optionURL, "&")
+	return "http://" + fmt.Sprintf(serviceNameTemplate, cc.clusterName) + "." + cc.namespace + ".svc." + cc.kubernetesClusterDomain + ":8090/" + basePath + "/" + action + "?" + strings.TrimSuffix(optionURL, "&")
 	//TODO only for testing
 	//return "http://localhost:8090/" + basePath + "/" + action + "?" + strings.TrimSuffix(optionURL, "&")
 }
 
-func postCruiseControl(action, namespace string, options map[string]string, ccEndpoint, clusterName string) (*http.Response, error) {
+func (cc *CruiseControlScaler) postCruiseControl(action string, options map[string]string) (*http.Response, error) {
 
-	requestURl := generateUrlForCC(action, namespace, options, ccEndpoint, clusterName)
+	requestURl := cc.generateUrlForCC(action, options)
 	rsp, err := http.Post(requestURl, "text/plain", nil)
 	if err != nil {
 		log.Error(err, "error during talking to cruise-control")
@@ -75,9 +91,9 @@ func postCruiseControl(action, namespace string, options map[string]string, ccEn
 	return rsp, nil
 }
 
-func getCruiseControl(action, namespace string, options map[string]string, ccEndpoint, clusterName string) (*http.Response, error) {
+func (cc *CruiseControlScaler) getCruiseControl(action string, options map[string]string) (*http.Response, error) {
 
-	requestURl := generateUrlForCC(action, namespace, options, ccEndpoint, clusterName)
+	requestURl := cc.generateUrlForCC(action, options)
 	rsp, err := http.Get(requestURl)
 	if err != nil {
 		log.Error(err, "error during talking to cruise-control")
@@ -108,16 +124,16 @@ func parseCCErrorFromResp(input io.Reader) (string, error) {
 	return errorFromResponse.ErrorMessage, err
 }
 
-func isKafkaBrokerDiskReady(brokerIdsWithMountPath map[string][]string, namespace, ccEndpoint, clusterName string) (bool, error) {
+func (cc *CruiseControlScaler) isKafkaBrokerDiskReady(brokerIdsWithMountPath map[string][]string) (bool, error) {
 	options := map[string]string{
 		"json": "true",
 	}
 
-	rsp, err := getCruiseControl(kafkaClusterStateAction, namespace, options, ccEndpoint, clusterName)
+	rsp, err := cc.getCruiseControl(kafkaClusterStateAction, options)
 	if err != nil {
 		keyVals := []interface{}{
-			"namespace", namespace,
-			"clusterName", clusterName,
+			"namespace", cc.namespace,
+			"clusterName", cc.clusterName,
 			//"brokerId", brokerIds,
 			//"path", mountPath,
 		}
@@ -170,13 +186,13 @@ func isKafkaBrokerDiskReady(brokerIdsWithMountPath map[string][]string, namespac
 }
 
 // Get brokers status from CC from a provided list of broker ids
-func GetLiveKafkaBrokersFromCruiseControl(brokerIds []string, namespace, ccEndpoint, clusterName string) ([]string, error) {
+func (cc *CruiseControlScaler) GetLiveKafkaBrokersFromCruiseControl(brokerIds []string) ([]string, error) {
 
 	options := map[string]string{
 		"json": "true",
 	}
 
-	rsp, err := getCruiseControl(clusterLoadAction, namespace, options, ccEndpoint, clusterName)
+	rsp, err := cc.getCruiseControl(clusterLoadAction, options)
 	if err != nil {
 		log.Error(err, "can't work with cruise-control because it is not ready")
 		return nil, err
@@ -217,7 +233,7 @@ func GetLiveKafkaBrokersFromCruiseControl(brokerIds []string, namespace, ccEndpo
 }
 
 // GetBrokerIDWithLeastPartition returns
-func GetBrokerIDWithLeastPartition(namespace, ccEndpoint, clusterName string) (string, error) {
+func (cc *CruiseControlScaler) GetBrokerIDWithLeastPartition() (string, error) {
 
 	brokerWithLeastPartition := ""
 
@@ -225,7 +241,7 @@ func GetBrokerIDWithLeastPartition(namespace, ccEndpoint, clusterName string) (s
 		"json": "true",
 	}
 
-	rsp, err := getCruiseControl(kafkaClusterStateAction, namespace, options, ccEndpoint, clusterName)
+	rsp, err := cc.getCruiseControl(kafkaClusterStateAction, options)
 	if err != nil {
 		log.Error(err, "can't work with cruise-control because it is not ready")
 		return brokerWithLeastPartition, err
@@ -266,9 +282,9 @@ func GetBrokerIDWithLeastPartition(namespace, ccEndpoint, clusterName string) (s
 }
 
 // UpScaleCluster upscales Kafka cluster
-func UpScaleCluster(brokerIds []string, namespace, ccEndpoint, clusterName string) (string, string, error) {
+func (cc *CruiseControlScaler) UpScaleCluster(brokerIds []string) (string, string, error) {
 
-	liveBrokers, err := GetLiveKafkaBrokersFromCruiseControl(brokerIds, namespace, ccEndpoint, clusterName)
+	liveBrokers, err := cc.GetLiveKafkaBrokersFromCruiseControl(brokerIds)
 	if err != nil {
 		return "", "", err
 	}
@@ -284,7 +300,7 @@ func UpScaleCluster(brokerIds []string, namespace, ccEndpoint, clusterName strin
 		"brokerid": strings.Join(brokerIds, ","),
 	}
 
-	uResp, err := postCruiseControl(addBrokerAction, namespace, options, ccEndpoint, clusterName)
+	uResp, err := cc.postCruiseControl(addBrokerAction, options)
 	if err != nil && err != errCruiseControlNotReturned200 {
 		log.Error(err, "can't upscale cluster gracefully since post to cruise-control failed")
 		return "", "", err
@@ -310,7 +326,7 @@ func UpScaleCluster(brokerIds []string, namespace, ccEndpoint, clusterName strin
 }
 
 // DownsizeCluster downscales Kafka cluster
-func DownsizeCluster(brokerIds []string, namespace, ccEndpoint, clusterName string) (string, string, error) {
+func (cc *CruiseControlScaler) DownsizeCluster(brokerIds []string) (string, string, error) {
 
 	options := map[string]string{
 		"brokerid": strings.Join(brokerIds, ","),
@@ -320,7 +336,7 @@ func DownsizeCluster(brokerIds []string, namespace, ccEndpoint, clusterName stri
 
 	var dResp *http.Response
 
-	dResp, err := postCruiseControl(removeBrokerAction, namespace, options, ccEndpoint, clusterName)
+	dResp, err := cc.postCruiseControl(removeBrokerAction, options)
 	if err != nil && err != errCruiseControlNotReturned200 {
 		log.Error(err, "downsize cluster gracefully failed since CC returned non 200")
 		return "", "", err
@@ -345,9 +361,9 @@ func DownsizeCluster(brokerIds []string, namespace, ccEndpoint, clusterName stri
 }
 
 // RebalanceDisks rebalances Kafka broker replicas between disks using CC
-func RebalanceDisks(brokerIdsWithMountPath map[string][]string, namespace, ccEndpoint, clusterName string) (string, string, error) {
+func (cc *CruiseControlScaler) RebalanceDisks(brokerIdsWithMountPath map[string][]string) (string, string, error) {
 
-	ready, err := isKafkaBrokerDiskReady(brokerIdsWithMountPath, namespace, ccEndpoint, clusterName)
+	ready, err := cc.isKafkaBrokerDiskReady(brokerIdsWithMountPath)
 	if err != nil {
 		return "", "", err
 	}
@@ -361,7 +377,7 @@ func RebalanceDisks(brokerIdsWithMountPath map[string][]string, namespace, ccEnd
 		"rebalance_disk": "true",
 	}
 
-	rResp, err := postCruiseControl(rebalanceAction, namespace, options, ccEndpoint, clusterName)
+	rResp, err := cc.postCruiseControl(rebalanceAction, options)
 	if err != nil && err != errCruiseControlNotReturned200 {
 		log.Error(err, "can't rebalance brokers disk gracefully since post to cruise-control failed")
 		return "", "", err
@@ -386,14 +402,14 @@ func RebalanceDisks(brokerIdsWithMountPath map[string][]string, namespace, ccEnd
 }
 
 // RebalanceCluster rebalances Kafka cluster using CC
-func RebalanceCluster(namespace, ccEndpoint, clusterName string) (string, error) {
+func (cc *CruiseControlScaler) RebalanceCluster() (string, error) {
 
 	options := map[string]string{
 		"dryrun": "false",
 		"json":   "true",
 	}
 
-	dResp, err := postCruiseControl(rebalanceAction, namespace, options, ccEndpoint, clusterName)
+	dResp, err := cc.postCruiseControl(rebalanceAction, options)
 	if err != nil {
 		log.Error(err, "can't rebalance cluster gracefully since post to cruise-control failed")
 		return "", err
@@ -406,7 +422,7 @@ func RebalanceCluster(namespace, ccEndpoint, clusterName string) (string, error)
 }
 
 // RunPreferedLeaderElectionInCluster runs leader election in  Kafka cluster using CC
-func RunPreferedLeaderElectionInCluster(namespace, ccEndpoint, clusterName string) (string, error) {
+func (cc *CruiseControlScaler) RunPreferedLeaderElectionInCluster() (string, error) {
 
 	options := map[string]string{
 		"dryrun": "false",
@@ -414,7 +430,7 @@ func RunPreferedLeaderElectionInCluster(namespace, ccEndpoint, clusterName strin
 		"goals":  "PreferredLeaderElectionGoal",
 	}
 
-	dResp, err := postCruiseControl(rebalanceAction, namespace, options, ccEndpoint, clusterName)
+	dResp, err := cc.postCruiseControl(rebalanceAction, options)
 	if err != nil {
 		log.Error(err, "can't rebalance cluster gracefully since post to cruise-control failed")
 		return "", err
@@ -427,12 +443,12 @@ func RunPreferedLeaderElectionInCluster(namespace, ccEndpoint, clusterName strin
 }
 
 // KillCCTask kills the specified CC task
-func KillCCTask(namespace, ccEndpoint, clusterName string) error {
+func (cc *CruiseControlScaler) KillCCTask() error {
 	options := map[string]string{
 		"json": "true",
 	}
 
-	_, err := postCruiseControl(killProposalAction, namespace, options, ccEndpoint, clusterName)
+	_, err := cc.postCruiseControl(killProposalAction, options)
 	if err != nil {
 		log.Error(err, "can't kill running tasks since post to cruise-control failed")
 		return err
@@ -443,12 +459,12 @@ func KillCCTask(namespace, ccEndpoint, clusterName string) error {
 }
 
 // GetCCTaskState checks whether the given CC Task ID finished or not
-func GetCCTaskState(uTaskId, namespace, ccEndpoint, clusterName string) (banzaicloudv1beta1.CruiseControlUserTaskState, error) {
+func (cc *CruiseControlScaler) GetCCTaskState(uTaskId string) (banzaicloudv1beta1.CruiseControlUserTaskState, error) {
 
-	gResp, err := getCruiseControl(getTaskListAction, namespace, map[string]string{
+	gResp, err := cc.getCruiseControl(getTaskListAction, map[string]string{
 		"json":          "true",
 		"user_task_ids": uTaskId,
-	}, ccEndpoint, clusterName)
+	})
 	if err != nil {
 		log.Error(err, "can't get task list from cruise-control")
 		return "", err
diff --git a/pkg/sdk/v1beta1/kafkacluster_types.go b/pkg/sdk/v1beta1/kafkacluster_types.go
index d5a72cdc..0ee06471 100644
--- a/pkg/sdk/v1beta1/kafkacluster_types.go
+++ b/pkg/sdk/v1beta1/kafkacluster_types.go
@@ -50,16 +50,17 @@ type KafkaClusterSpec struct {
 	Brokers              []Broker                `json:"brokers"`
 	RollingUpgradeConfig RollingUpgradeConfig    `json:"rollingUpgradeConfig"`
 	// +kubebuilder:validation:Enum=envoy;istioingress
-	IngressController   string              `json:"ingressController,omitempty"`
-	OneBrokerPerNode    bool                `json:"oneBrokerPerNode"`
-	PropagateLabels     bool                `json:"propagateLabels,omitempty"`
-	CruiseControlConfig CruiseControlConfig `json:"cruiseControlConfig"`
-	EnvoyConfig         EnvoyConfig         `json:"envoyConfig,omitempty"`
-	MonitoringConfig    MonitoringConfig    `json:"monitoringConfig,omitempty"`
-	VaultConfig         VaultConfig         `json:"vaultConfig,omitempty"`
-	AlertManagerConfig  *AlertManagerConfig `json:"alertManagerConfig,omitempty"`
-	IstioIngressConfig  IstioIngressConfig  `json:"istioIngressConfig,omitempty"`
-	Envs                []corev1.EnvVar     `json:"envs,omitempty"`
+	IngressController       string              `json:"ingressController,omitempty"`
+	OneBrokerPerNode        bool                `json:"oneBrokerPerNode"`
+	PropagateLabels         bool                `json:"propagateLabels,omitempty"`
+	CruiseControlConfig     CruiseControlConfig `json:"cruiseControlConfig"`
+	EnvoyConfig             EnvoyConfig         `json:"envoyConfig,omitempty"`
+	MonitoringConfig        MonitoringConfig    `json:"monitoringConfig,omitempty"`
+	VaultConfig             VaultConfig         `json:"vaultConfig,omitempty"`
+	AlertManagerConfig      *AlertManagerConfig `json:"alertManagerConfig,omitempty"`
+	IstioIngressConfig      IstioIngressConfig  `json:"istioIngressConfig,omitempty"`
+	Envs                    []corev1.EnvVar     `json:"envs,omitempty"`
+	KubernetesClusterDomain string              `json:"kubernetesClusterDomain,omitempty"`
 }
 
 // KafkaClusterStatus defines the observed state of KafkaCluster
@@ -310,6 +311,14 @@ func (kSpec *KafkaClusterSpec) GetIngressController() string {
 	return kSpec.IngressController
 }
 
+// GetDomain returns the default domain if not specified otherwise
+func (kSpec *KafkaClusterSpec) GetKubernetesClusterDomain() string {
+	if kSpec.KubernetesClusterDomain == "" {
+		return "cluster.local"
+	}
+	return kSpec.KubernetesClusterDomain
+}
+
 // GetZkPath returns the default "/" ZkPath if not specified otherwise
 func (kSpec *KafkaClusterSpec) GetZkPath() string {
 	const prefix = "/"
diff --git a/pkg/util/pki/common.go b/pkg/util/pki/common.go
index 7403be13..e2ea9019 100644
--- a/pkg/util/pki/common.go
+++ b/pkg/util/pki/common.go
@@ -41,9 +41,9 @@ const (
 	BrokerControllerTemplate = "%s-controller"
 	// BrokerControllerFQDNTemplate is combined with the above and cluster namespace
 	// to create a 'fake' full-name for the controller user
-	BrokerControllerFQDNTemplate = "%s.%s.mgt.cluster.local"
+	BrokerControllerFQDNTemplate = "%s.%s.mgt.%s"
 	// CAIntermediateTemplate is the template used for intermediate CA resources
-	CAIntermediateTemplate = "%s-intermediate.%s.cluster.local"
+	CAIntermediateTemplate = "%s-intermediate.%s.%s"
 	// CAFQDNTemplate is the template used for the FQDN of a CA
 	CAFQDNTemplate = "%s-ca.%s.cluster.local"
 )
@@ -100,9 +100,9 @@ func GetInternalDNSNames(cluster *v1beta1.KafkaCluster) (dnsNames []string) {
 // GetCommonName returns the full FQDN for the internal Kafka listener
 func GetCommonName(cluster *v1beta1.KafkaCluster) string {
 	if cluster.Spec.HeadlessServiceEnabled {
-		return fmt.Sprintf("%s.%s.svc.cluster.local", fmt.Sprintf(kafka.HeadlessServiceTemplate, cluster.Name), cluster.Namespace)
+		return fmt.Sprintf("%s.%s.svc.%s", fmt.Sprintf(kafka.HeadlessServiceTemplate, cluster.Name), cluster.Namespace, cluster.Spec.GetKubernetesClusterDomain())
 	}
-	return fmt.Sprintf("%s.%s.svc.cluster.local", fmt.Sprintf(kafka.AllBrokerServiceTemplate, cluster.Name), cluster.Namespace)
+	return fmt.Sprintf("%s.%s.svc.%s", fmt.Sprintf(kafka.AllBrokerServiceTemplate, cluster.Name), cluster.Namespace, cluster.Spec.GetKubernetesClusterDomain())
 }
 
 // clusterDNSNames returns all the possible DNS Names for a Kafka Cluster
@@ -177,7 +177,7 @@ func BrokerUserForCluster(cluster *v1beta1.KafkaCluster, additionalHostnames []s
 func ControllerUserForCluster(cluster *v1beta1.KafkaCluster) *v1alpha1.KafkaUser {
 	return &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(
-			fmt.Sprintf(BrokerControllerFQDNTemplate, fmt.Sprintf(BrokerControllerTemplate, cluster.Name), cluster.Namespace),
+			fmt.Sprintf(BrokerControllerFQDNTemplate, fmt.Sprintf(BrokerControllerTemplate, cluster.Name), cluster.Namespace, cluster.Spec.GetKubernetesClusterDomain()),
 			LabelsForKafkaPKI(cluster.Name), cluster,
 		),
 		Spec: v1alpha1.KafkaUserSpec{


Test patch:
diff --git a/pkg/resources/kafka/configmap_test.go b/pkg/resources/kafka/configmap_test.go
index 767bfdb3..7783773e 100644
--- a/pkg/resources/kafka/configmap_test.go
+++ b/pkg/resources/kafka/configmap_test.go
@@ -30,6 +30,7 @@ func TestGenerateBrokerConfig(t *testing.T) {
 		readOnlyConfig          string
 		zkAddresses             []string
 		zkPath                  string
+		kubernetesClusterDomain string
 		clusterWideConfig       string
 		perBrokerReadOnlyConfig string
 		perBrokerConfig         string
@@ -41,6 +42,7 @@ func TestGenerateBrokerConfig(t *testing.T) {
 			readOnlyConfig:          ``,
 			zkAddresses:             []string{"example.zk:2181"},
 			zkPath:                  ``,
+			kubernetesClusterDomain: ``,
 			clusterWideConfig:       ``,
 			perBrokerConfig:         ``,
 			perBrokerReadOnlyConfig: ``,
@@ -57,6 +59,7 @@ zookeeper.connect=example.zk:2181/`,
 			testName:                "basicConfigWithZKPath",
 			readOnlyConfig:          ``,
 			zkPath:                  `/kafka`,
+			kubernetesClusterDomain: ``,
 			zkAddresses:             []string{"example.zk:2181"},
 			clusterWideConfig:       ``,
 			perBrokerConfig:         ``,
@@ -74,6 +77,7 @@ zookeeper.connect=example.zk:2181/kafka`,
 			testName:                "basicConfigWithSimpleZkPath",
 			readOnlyConfig:          ``,
 			zkPath:                  `/`,
+			kubernetesClusterDomain: ``,
 			zkAddresses:             []string{"example.zk:2181"},
 			clusterWideConfig:       ``,
 			perBrokerConfig:         ``,
@@ -92,6 +96,7 @@ zookeeper.connect=example.zk:2181/`,
 			readOnlyConfig:          ``,
 			zkPath:                  `/kafka`,
 			zkAddresses:             []string{"example.zk:2181", "example.zk-1:2181"},
+			kubernetesClusterDomain: ``,
 			clusterWideConfig:       ``,
 			perBrokerConfig:         ``,
 			perBrokerReadOnlyConfig: ``,
@@ -109,6 +114,7 @@ zookeeper.connect=example.zk:2181,example.zk-1:2181/kafka`,
 			readOnlyConfig:          ``,
 			zkAddresses:             []string{"example.zk:2181"},
 			zkPath:                  ``,
+			kubernetesClusterDomain: ``,
 			clusterWideConfig:       ``,
 			perBrokerConfig:         ``,
 			perBrokerReadOnlyConfig: ``,
@@ -128,9 +134,28 @@ metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlM
 zookeeper.connect=example.zk:2181/`,
 		},
 		{
-			testName:    "readOnlyRedefinedInOneBroker",
-			zkAddresses: []string{"example.zk:2181"},
-			zkPath:      ``,
+			testName:                "basicConfigWithClusterDomain",
+			readOnlyConfig:          ``,
+			zkAddresses:             []string{"example.zk:2181"},
+			zkPath:                  ``,
+			kubernetesClusterDomain: `foo.bar`,
+			clusterWideConfig:       ``,
+			perBrokerConfig:         ``,
+			perBrokerReadOnlyConfig: ``,
+			expectedConfig: `advertised.listeners=INTERNAL://kafka-0.kafka.svc.foo.bar:9092
+broker.id=0
+cruise.control.metrics.reporter.bootstrap.servers=INTERNAL://kafka-0.kafka.svc.foo.bar:9092
+inter.broker.listener.name=INTERNAL
+listener.security.protocol.map=INTERNAL:PLAINTEXT
+listeners=INTERNAL://:9092
+metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter
+zookeeper.connect=example.zk:2181/`,
+		},
+		{
+			testName:                "readOnlyRedefinedInOneBroker",
+			zkAddresses:             []string{"example.zk:2181"},
+			zkPath:                  ``,
+			kubernetesClusterDomain: ``,
 			readOnlyConfig: `
 auto.create.topics.enable=false
 control.plane.listener.name=thisisatest
@@ -184,8 +209,9 @@ zookeeper.connect=example.zk:2181/`,
 									},
 								},
 							},
-							ReadOnlyConfig:    test.readOnlyConfig,
-							ClusterWideConfig: test.clusterWideConfig,
+							ReadOnlyConfig:          test.readOnlyConfig,
+							KubernetesClusterDomain: test.kubernetesClusterDomain,
+							ClusterWideConfig:       test.clusterWideConfig,
 							Brokers: []v1beta1.Broker{{
 								Id:             0,
 								ReadOnlyConfig: test.perBrokerReadOnlyConfig,
diff --git a/pkg/util/pki/pki_common_test.go b/pkg/util/pki/pki_common_test.go
index f0962ffb..1451242e 100644
--- a/pkg/util/pki/pki_common_test.go
+++ b/pkg/util/pki/pki_common_test.go
@@ -66,6 +66,13 @@ func TestGetCommonName(t *testing.T) {
 	if allBrokerCN != expected {
 		t.Error("Expected:", expected, "Got:", allBrokerCN)
 	}
+
+	cluster.Spec = v1beta1.KafkaClusterSpec{HeadlessServiceEnabled: true, KubernetesClusterDomain: "foo.bar"}
+	kubernetesClusterDomainCN := GetCommonName(cluster)
+	expected = "test-cluster-headless.test-namespace.svc.foo.bar"
+	if kubernetesClusterDomainCN != expected {
+		t.Error("Expected:", expected, "Got:", kubernetesClusterDomainCN)
+	}
 }
 
 func TestLabelsForKafkaPKI(t *testing.T) {
@@ -141,7 +148,7 @@ func TestControllerUserForCluster(t *testing.T) {
 
 	expected := &v1alpha1.KafkaUser{
 		ObjectMeta: templates.ObjectMeta(
-			fmt.Sprintf(BrokerControllerFQDNTemplate, fmt.Sprintf(BrokerControllerTemplate, cluster.Name), cluster.Namespace),
+			fmt.Sprintf(BrokerControllerFQDNTemplate, fmt.Sprintf(BrokerControllerTemplate, cluster.Name), cluster.Namespace, cluster.Spec.GetKubernetesClusterDomain()),
 			LabelsForKafkaPKI(cluster.Name), cluster,
 		),
 		Spec: v1alpha1.KafkaUserSpec{