Fix remaining bugs on service mesh support

Author: baluchicken

pkg/apis/banzaicloud/v1alpha1/kafkacluster_types.go

@@ -42,8 +42,8 @@ type KafkaClusterSpec struct {
 
 // KafkaClusterStatus defines the observed state of KafkaCluster
 type KafkaClusterStatus struct {
-	BrokersState             map[int32]*BrokerState    `json:"brokersState,omitempty"`
-	CruiseControlTopicStatus *CruiseControlTopicStatus `json:"cruiseControlTopicStatus,omitempty"`
+	BrokersState             map[int32]*BrokerState   `json:"brokersState,omitempty"`
+	CruiseControlTopicStatus CruiseControlTopicStatus `json:"cruiseControlTopicStatus,omitempty"`
 }
 
 // BrokerConfig defines the broker configuration

pkg/apis/banzaicloud/v1alpha1/zz_generated.deepcopy.go

@@ -132,7 +132,7 @@ func updateGracefulScaleStatus(c client.Client, brokerId int32, cluster *banzaic
 func UpdateCCTopicStatus(c client.Client, cluster *banzaicloudv1alpha1.KafkaCluster, ccTopicStatus banzaicloudv1alpha1.CruiseControlTopicStatus, logger logr.Logger) error {
 	typeMeta := cluster.TypeMeta
 
-	cluster.Status.CruiseControlTopicStatus = &ccTopicStatus
+	cluster.Status.CruiseControlTopicStatus = ccTopicStatus
 
 	err := c.Status().Update(context.Background(), cluster)
 	if errors.IsNotFound(err) {
@@ -150,7 +150,7 @@ func UpdateCCTopicStatus(c client.Client, cluster *banzaicloudv1alpha1.KafkaClus
 			return emperror.Wrap(err, "could not get config for updating status")
 		}
 
-		cluster.Status.CruiseControlTopicStatus = &ccTopicStatus
+		cluster.Status.CruiseControlTopicStatus = ccTopicStatus
 
 		err = c.Status().Update(context.Background(), cluster)
 		if errors.IsNotFound(err) {

pkg/k8sutil/status.go

@@ -63,7 +63,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 
 	if r.KafkaCluster.Spec.CruiseControlConfig.CruiseControlEndpoint == "" {
 
-		if *r.KafkaCluster.Status.CruiseControlTopicStatus != banzaicloudv1alpha1.CruiseControlTopicReady {
+		if r.KafkaCluster.Status.CruiseControlTopicStatus == "" || r.KafkaCluster.Status.CruiseControlTopicStatus == banzaicloudv1alpha1.CruiseControlTopicNotReady {
 			err := generateCCTopic(r.KafkaCluster)
 			if err != nil {
 				k8sutil.UpdateCCTopicStatus(r.Client, r.KafkaCluster, banzaicloudv1alpha1.CruiseControlTopicNotReady, log)
@@ -75,7 +75,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 			}
 		}
 
-		if *r.KafkaCluster.Status.CruiseControlTopicStatus == banzaicloudv1alpha1.CruiseControlTopicReady {
+		if r.KafkaCluster.Status.CruiseControlTopicStatus == banzaicloudv1alpha1.CruiseControlTopicReady {
 			for _, res := range []resources.ResourceWithLogs{
 				r.service,
 				r.configMap,

pkg/resources/cruisecontrol/cruisecontrol.go

@@ -26,7 +26,7 @@ import (
 func generateCCTopic(cluster *banzaicloudv1alpha1.KafkaCluster) error {
 
 	conn, err := kafkaGo.Dial("tcp",
-		fmt.Sprintf("%s.%s.svc.cluster.local:%d", fmt.Sprintf(kafka.AllBrokerServiceTemplate, cluster.Name), cluster.Namespace, cluster.Spec.ListenersConfig.InternalListeners[0].ContainerPort))
+		generateKafkaAddress(cluster))
 	if err != nil {
 		return emperror.Wrap(err, "could not create topic for CC because kafka is unavailable")
 	}
@@ -42,3 +42,10 @@ func generateCCTopic(cluster *banzaicloudv1alpha1.KafkaCluster) error {
 	}
 	return nil
 }
+
+func generateKafkaAddress(cluster *banzaicloudv1alpha1.KafkaCluster) string {
+	if cluster.Spec.HeadlessServiceEnabled {
+		return fmt.Sprintf("%s.%s.svc.cluster.local:%d", fmt.Sprintf(kafka.HeadlessServiceTemplate, cluster.Name), cluster.Namespace, cluster.Spec.ListenersConfig.InternalListeners[0].ContainerPort)
+	}
+	return fmt.Sprintf("%s.%s.svc.cluster.local:%d", fmt.Sprintf(kafka.AllBrokerServiceTemplate, cluster.Name), cluster.Namespace, cluster.Spec.ListenersConfig.InternalListeners[0].ContainerPort)
+}