Run Koperator on Kind Cluster Locally

.gitignore

@@ -8,7 +8,7 @@
 bin
 charts/**/charts
 charts/koperator/requirements.lock
-
+charts/kafka-operator/ingress
 # Test binary, build with `go test -c`
 *.test
 

api/v1beta1/kafkacluster_types.go

@@ -157,9 +157,16 @@ type KafkaClusterSpec struct {
 	// This is default to be true; if set to false, the Kafka cluster is in ZooKeeper mode.
 	// +kubebuilder:default=false
 	// +optional
-	KRaftMode              bool            `json:"kRaft"`
-	HeadlessServiceEnabled bool            `json:"headlessServiceEnabled"`
-	ListenersConfig        ListenersConfig `json:"listenersConfig"`
+	KRaftMode              bool `json:"kRaft"`
+	HeadlessServiceEnabled bool `json:"headlessServiceEnabled"`
+	// DebugEnabled is used to decide whether to create a separate loadbalancer services for the
+	// Kafka and Cruise Control Pods. These services will expose the internal listener ports of the Kafka
+	// cluster with LoadBalancer type, which can be used for running Koperator on a local machine against
+	// a kafkaCluster instance on a Kind Cluster.
+	// +kubebuilder:default=false
+	// +optional
+	DebugEnabled    bool            `json:"debugEnabled"`
+	ListenersConfig ListenersConfig `json:"listenersConfig"`
 	// Custom ports to expose in the container. Example use case: a custom kafka distribution, that includes an integrated metrics api endpoint
 	AdditionalPorts []corev1.ContainerPort `json:"additionalPorts,omitempty"`
 	// ZKAddresses specifies the ZooKeeper connection string

charts/kafka-operator/crds/kafkaclusters.yaml

@@ -19231,6 +19231,14 @@ spec:
                       type: object
                     type: array
                 type: object
+              debugEnabled:
+                default: false
+                description: |-
+                  DebugEnabled is used to decide whether to create a separate loadbalancer services for the
+                  Kafka and Cruise Control Pods. These services will expose the internal listener ports of the Kafka
+                  cluster with LoadBalancer type, which can be used for running Koperator on a local machine against
+                  a kafkaCluster instance on a Kind Cluster.
+                type: boolean
               disruptionBudget:
                 description: DisruptionBudget defines the configuration for PodDisruptionBudget
                   where the workload is managed by the kafka-operator

config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml

@@ -19231,6 +19231,14 @@ spec:
                       type: object
                     type: array
                 type: object
+              debugEnabled:
+                default: false
+                description: |-
+                  DebugEnabled is used to decide whether to create a separate loadbalancer services for the
+                  Kafka and Cruise Control Pods. These services will expose the internal listener ports of the Kafka
+                  cluster with LoadBalancer type, which can be used for running Koperator on a local machine against
+                  a kafkaCluster instance on a Kind Cluster.
+                type: boolean
               disruptionBudget:
                 description: DisruptionBudget defines the configuration for PodDisruptionBudget
                   where the workload is managed by the kafka-operator

config/samples/simpleZookeeper.yaml

@@ -0,0 +1,10 @@
+apiVersion: zookeeper.pravega.io/v1beta1
+kind: ZookeeperCluster
+metadata:
+    name: zookeeper-server
+    namespace: zookeeper
+spec:
+    replicas: 3
+    persistence:
+        reclaimPolicy: Delete
+

config/samples/simplekafkacluster.yaml

@@ -5,10 +5,11 @@ metadata:
     controller-tools.k8s.io: "1.0"
   name: kafka
 spec:
+  debugEnabled: true
   kRaft: false
   monitoringConfig:
     jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.4.0"
-  headlessServiceEnabled: true
+  headlessServiceEnabled: false
   zkAddresses:
     - "zookeeper-server-client.zookeeper:2181"
   propagateLabels: false

config/scaleops/CustomOwnerGrouping.yaml

@@ -0,0 +1,22 @@
+
+kind: CustomOwnerGrouping
+apiVersion: analysis.scaleops.sh/v1alpha1
+metadata:
+  name: kafkabroker
+  namespace: scaleops-system
+spec:
+  groupBy:
+    positiveRegexMatch: false
+  groupBys:
+    - labels:
+        - 'isBrokerNode: true'
+      positiveRegexMatch: false
+      topOwnerController:
+        apiVersion: kafka.banzaicloud.io/v1beta1
+        kind: KafkaCluster
+  displayOptions:
+    hideGeneratedSuffix: true
+    fields:
+      - ownerName
+  defaultPolicy: kafka-brokers
+  enabled: true

pkg/resources/cruisecontrol/service.go

@@ -26,14 +26,15 @@ import (
 )
 
 func (r *Reconciler) service() runtime.Object {
-	return &corev1.Service{
+	svc := &corev1.Service{
 		ObjectMeta: templates.ObjectMeta(
 			fmt.Sprintf(serviceNameTemplate, r.KafkaCluster.Name),
 			apiutil.MergeLabels(ccLabelSelector(r.KafkaCluster.Name), r.KafkaCluster.Labels),
 			r.KafkaCluster,
 		),
 		Spec: corev1.ServiceSpec{
 			Selector: ccLabelSelector(r.KafkaCluster.Name),
+			Type:     corev1.ServiceTypeClusterIP,
 			Ports: []corev1.ServicePort{
 				{
 					Name:       "cc",
@@ -50,4 +51,10 @@ func (r *Reconciler) service() runtime.Object {
 			},
 		},
 	}
+
+	if r.KafkaCluster.Spec.DebugEnabled {
+		svc.Spec.Type = corev1.ServiceTypeLoadBalancer
+	}
+
+	return svc
 }

pkg/resources/kafka/allBrokerService.go

@@ -39,7 +39,7 @@ func (r *Reconciler) allBrokerService() runtime.Object {
 	usedPorts = append(usedPorts,
 		generateServicePortForAdditionalPorts(r.KafkaCluster.Spec.AdditionalPorts)...)
 
-	return &corev1.Service{
+	svc := &corev1.Service{
 		ObjectMeta: templates.ObjectMetaWithAnnotations(
 			fmt.Sprintf(kafkautils.AllBrokerServiceTemplate, r.KafkaCluster.GetName()),
 			apiutil.LabelsForKafka(r.KafkaCluster.GetName()),
@@ -52,4 +52,10 @@ func (r *Reconciler) allBrokerService() runtime.Object {
 			Ports:           usedPorts,
 		},
 	}
+
+	if r.KafkaCluster.Spec.DebugEnabled {
+		svc.Spec.Type = corev1.ServiceTypeLoadBalancer
+	}
+
+	return svc
 }

pkg/resources/kafka/service.go

@@ -46,7 +46,7 @@ func (r *Reconciler) service(id int32, _ *v1beta1.BrokerConfig) runtime.Object {
 		Protocol:   corev1.ProtocolTCP,
 	})
 
-	return &corev1.Service{
+	svc := &corev1.Service{
 		ObjectMeta: templates.ObjectMetaWithAnnotations(fmt.Sprintf("%s-%d", r.KafkaCluster.Name, id),
 			apiutil.MergeLabels(
 				apiutil.LabelsForKafka(r.KafkaCluster.Name),
@@ -61,4 +61,8 @@ func (r *Reconciler) service(id int32, _ *v1beta1.BrokerConfig) runtime.Object {
 			Ports:           usedPorts,
 		},
 	}
+	if r.KafkaCluster.Spec.DebugEnabled {
+		svc.Spec.Type = corev1.ServiceTypeLoadBalancer
+	}
+	return svc
 }