diff --git a/.gitignore b/.gitignore index 009e10da1..480a9b5b4 100644 --- a/.gitignore +++ b/.gitignore @@ -8,7 +8,7 @@ bin charts/**/charts charts/koperator/requirements.lock - +charts/kafka-operator/ingress # Test binary, build with `go test -c` *.test diff --git a/api/v1beta1/kafkacluster_types.go b/api/v1beta1/kafkacluster_types.go index cdb7671e1..e337c9772 100644 --- a/api/v1beta1/kafkacluster_types.go +++ b/api/v1beta1/kafkacluster_types.go @@ -157,9 +157,22 @@ 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"` + // Allows ScaleOps to manage Memory and CPU Resource Requests for Kafka Broker Pods. + // This Disables CPU and Memory request reconciliation from the desired state defined in + // the KafkaCluster to the current state in the Kubernetes Cluster + // +kubebuilder:default=false + // +optional + ScaleOpsEnabled bool `json:"scaleOpsEnabled"` + 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 diff --git a/charts/kafka-operator/crds/kafkaclusters.yaml b/charts/kafka-operator/crds/kafkaclusters.yaml index e3fd3e25d..0a2a2d103 100644 --- a/charts/kafka-operator/crds/kafkaclusters.yaml +++ b/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 @@ -23735,6 +23743,13 @@ spec: required: - failureThreshold type: object + scaleOpsEnabled: + default: false + description: |- + Allows ScaleOps to manage Memory and CPU Resource Requests for Kafka Broker Pods. + This Disables CPU and Memory request reconciliation from the desired state defined in + the KafkaCluster to the current state in the Kubernetes Cluster + type: boolean taintedBrokersSelector: description: Selector for broker pods that need to be recycled/reconciled properties: diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml index e3fd3e25d..0a2a2d103 100644 --- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml +++ b/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 @@ -23735,6 +23743,13 @@ spec: required: - failureThreshold type: object + scaleOpsEnabled: + default: false + description: |- + Allows ScaleOps to manage Memory and CPU Resource Requests for Kafka Broker Pods. + This Disables CPU and Memory request reconciliation from the desired state defined in + the KafkaCluster to the current state in the Kubernetes Cluster + type: boolean taintedBrokersSelector: description: Selector for broker pods that need to be recycled/reconciled properties: diff --git a/config/samples/simpleZookeeper.yaml b/config/samples/simpleZookeeper.yaml new file mode 100644 index 000000000..6bf70aa9c --- /dev/null +++ b/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 + diff --git a/config/samples/simplekafkacluster.yaml b/config/samples/simplekafkacluster.yaml index d890f8551..cf08d8980 100644 --- a/config/samples/simplekafkacluster.yaml +++ b/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 diff --git a/config/scaleops/CustomOwnerGrouping.yaml b/config/scaleops/CustomOwnerGrouping.yaml new file mode 100644 index 000000000..7e9760d82 --- /dev/null +++ b/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 diff --git a/pkg/resources/cruisecontrol/service.go b/pkg/resources/cruisecontrol/service.go index d868eacf4..2c1c64439 100644 --- a/pkg/resources/cruisecontrol/service.go +++ b/pkg/resources/cruisecontrol/service.go @@ -26,7 +26,7 @@ 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), @@ -34,6 +34,7 @@ func (r *Reconciler) service() runtime.Object { ), 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 } diff --git a/pkg/resources/kafka/allBrokerService.go b/pkg/resources/kafka/allBrokerService.go index ecfdd5b7b..ed0eed60c 100644 --- a/pkg/resources/kafka/allBrokerService.go +++ b/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 } diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go index eec273a7a..8fc5e042c 100644 --- a/pkg/resources/kafka/kafka.go +++ b/pkg/resources/kafka/kafka.go @@ -834,6 +834,7 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod, return errorfactory.New(errorfactory.APIFailure{}, err, "getting resource failed", "kind", desiredType) } switch { + //initial run - Create Pod case len(podList.Items) == 0: if err := patch.DefaultAnnotator.SetLastAppliedAnnotation(desiredPod); err != nil { return errors.WrapIf(err, "could not apply last state to annotation") @@ -941,6 +942,42 @@ func (r *Reconciler) updateStatusWithDockerImageAndVersion(brokerId int32, broke return nil } +// syncResourceRequests overwrites CPU and memory requests in desiredPod's containers +// with the values from currentPod so that request-only changes do not trigger a pod restart. +func syncResourceRequests(desiredPod, currentPod *corev1.Pod) { + syncContainerResourceRequests(desiredPod.Spec.Containers, currentPod.Spec.Containers) + syncContainerResourceRequests(desiredPod.Spec.InitContainers, currentPod.Spec.InitContainers) + syncPodAffinities(desiredPod, currentPod) +} + +func syncPodAffinities(desiredPod, currentPod *corev1.Pod) { + panic("unimplemented") +} + +func syncContainerResourceRequests(desired, current []corev1.Container) { + index := make(map[string]corev1.ResourceList, len(current)) + for _, c := range current { + index[c.Name] = c.Resources.Requests + } + for i := range desired { + c := &desired[i] + reqs, ok := index[c.Name] + if !ok { + continue + } + if c.Resources.Requests == nil { + c.Resources.Requests = make(corev1.ResourceList) + } + for _, res := range []corev1.ResourceName{corev1.ResourceCPU, corev1.ResourceMemory} { + if val, exists := reqs[res]; exists { + c.Resources.Requests[res] = val + } else { + delete(c.Resources.Requests, res) + } + } + } +} + //gocyclo:ignore func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPod *corev1.Pod, desiredType reflect.Type) error { // Since toleration does not support patchStrategy:"merge,retainKeys", @@ -957,6 +994,10 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo } desiredPod.Spec.Tolerations = uniqueTolerations } + // Ignore CPU/memory request diffs — changing requests does not require a pod restart. + if r.KafkaCluster.Spec.ScaleOpsEnabled { + syncResourceRequests(desiredPod, currentPod) + } // Check if the resource actually updated or if labels match TaintedBrokersSelector patchResult, err := patch.DefaultPatchMaker.Calculate(currentPod, desiredPod) switch { diff --git a/pkg/resources/kafka/kafka_test.go b/pkg/resources/kafka/kafka_test.go index ad9e6db4b..f636715e5 100644 --- a/pkg/resources/kafka/kafka_test.go +++ b/pkg/resources/kafka/kafka_test.go @@ -1986,3 +1986,6 @@ func TestGetBrokerAzMap(t *testing.T) { }) } } + +func TestScaleOps(t. *testing.T) { + \ No newline at end of file diff --git a/pkg/resources/kafka/service.go b/pkg/resources/kafka/service.go index fed334635..fa9dca5cf 100644 --- a/pkg/resources/kafka/service.go +++ b/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 } diff --git a/pkg/resources/kafka/service_test.go b/pkg/resources/kafka/service_test.go new file mode 100644 index 000000000..cdb4e6650 --- /dev/null +++ b/pkg/resources/kafka/service_test.go @@ -0,0 +1,231 @@ +// Copyright © 2023 Cisco Systems, Inc. and/or its affiliates +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package kafka + +import ( + "fmt" + "testing" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" + + "go.uber.org/mock/gomock" + + apiutil "github.com/banzaicloud/koperator/api/util" + "github.com/banzaicloud/koperator/api/v1beta1" + "github.com/banzaicloud/koperator/pkg/resources" + mocks "github.com/banzaicloud/koperator/pkg/resources/kafka/mocks" + "github.com/banzaicloud/koperator/pkg/util" +) + +func TestService(t *testing.T) { + testCases := []struct { + testName string + r *Reconciler + expectedService *corev1.Service + }{ + { + testName: "Basic Internal And External Service", + r: &Reconciler{ + Reconciler: resources.Reconciler{ + KafkaCluster: &v1beta1.KafkaCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: "kafka", + Namespace: "kafka", + }, + Spec: v1beta1.KafkaClusterSpec{ + DebugEnabled: false, + KRaftMode: false, + ListenersConfig: v1beta1.ListenersConfig{ + InternalListeners: []v1beta1.InternalListenerConfig{ + { + CommonListenerSpec: v1beta1.CommonListenerSpec{ + Name: "internal", + ContainerPort: 29092, + Type: "plaintext", + UsedForInnerBrokerCommunication: true, + }, + }, + }, + ExternalListeners: []v1beta1.ExternalListenerConfig{ + { + CommonListenerSpec: v1beta1.CommonListenerSpec{ + Name: "plaintext", + ContainerPort: 29094, + Type: "plaintext", + UsedForInnerBrokerCommunication: false, + }, + AccessMethod: corev1.ServiceTypeLoadBalancer, + }, + }, + }, + }, + }, + }, + }, + expectedService: &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "kafka-1", + Namespace: "kafka", + Labels: map[string]string{"app": "kafka", "brokerId": "1", "kafka_cr": "kafka"}, + Annotations: map[string]string{}, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "", + Kind: "", + Name: "kafka", + UID: "", + Controller: util.BoolPointer(true), + BlockOwnerDeletion: util.BoolPointer(true), + }, + }, + }, + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeClusterIP, + SessionAffinity: corev1.ServiceAffinityNone, + Selector: apiutil.MergeLabels(apiutil.LabelsForKafka("kafka"), map[string]string{v1beta1.BrokerIdLabelKey: fmt.Sprintf("1")}), + Ports: []corev1.ServicePort{ + { + Name: "tcp-internal", + Protocol: "TCP", + Port: 29092, + TargetPort: intstr.FromInt(29092), + NodePort: 0, + }, + { + Name: "tcp-plaintext", + Protocol: "TCP", + Port: 29094, + TargetPort: intstr.FromInt(29094), + NodePort: 0, + }, + { + Name: "metrics", + Protocol: "TCP", + Port: 9020, + TargetPort: intstr.FromInt(9020), + NodePort: 0, + }, + }, + ClusterIP: "", + PublishNotReadyAddresses: false, + }, + }, + }, + { + testName: "Basic Internal And External Service", + r: &Reconciler{ + Reconciler: resources.Reconciler{ + KafkaCluster: &v1beta1.KafkaCluster{ + ObjectMeta: metav1.ObjectMeta{ + Name: "kafka", + Namespace: "kafka", + }, + Spec: v1beta1.KafkaClusterSpec{ + DebugEnabled: true, + KRaftMode: false, + ListenersConfig: v1beta1.ListenersConfig{ + InternalListeners: []v1beta1.InternalListenerConfig{ + { + CommonListenerSpec: v1beta1.CommonListenerSpec{ + Name: "internal", + ContainerPort: 29092, + Type: "plaintext", + UsedForInnerBrokerCommunication: true, + }, + }, + }, + ExternalListeners: []v1beta1.ExternalListenerConfig{ + { + CommonListenerSpec: v1beta1.CommonListenerSpec{ + Name: "plaintext", + ContainerPort: 29094, + Type: "plaintext", + UsedForInnerBrokerCommunication: false, + }, + AccessMethod: corev1.ServiceTypeLoadBalancer, + }, + }, + }, + }, + }, + }, + }, + expectedService: &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "kafka-1", + Namespace: "kafka", + Labels: map[string]string{"app": "kafka", "brokerId": "1", "kafka_cr": "kafka"}, + Annotations: map[string]string{}, + OwnerReferences: []metav1.OwnerReference{ + { + APIVersion: "", + Kind: "", + Name: "kafka", + UID: "", + Controller: util.BoolPointer(true), + BlockOwnerDeletion: util.BoolPointer(true), + }, + }, + }, + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeLoadBalancer, + SessionAffinity: corev1.ServiceAffinityNone, + Selector: apiutil.MergeLabels(apiutil.LabelsForKafka("kafka"), map[string]string{v1beta1.BrokerIdLabelKey: fmt.Sprintf("1")}), + Ports: []corev1.ServicePort{ + { + Name: "tcp-internal", + Protocol: "TCP", + Port: 29092, + TargetPort: intstr.FromInt(29092), + NodePort: 0, + }, + { + Name: "tcp-plaintext", + Protocol: "TCP", + Port: 29094, + TargetPort: intstr.FromInt(29094), + NodePort: 0, + }, + { + Name: "metrics", + Protocol: "TCP", + Port: 9020, + TargetPort: intstr.FromInt(9020), + NodePort: 0, + }, + }, + ClusterIP: "", + PublishNotReadyAddresses: false, + }, + }, + }, + } + mockCtrl := gomock.NewController(t) + + for _, test := range testCases { + t.Run(test.testName, func(t *testing.T) { + mockClient := mocks.NewMockClient(mockCtrl) + mockClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).AnyTimes() + r := test.r + + actualService := r.service(1, nil) + + require.Equal(t, test.expectedService, actualService) + }) + } +} diff --git a/run-local.sh b/run-local.sh new file mode 100755 index 000000000..a9e5d7cb8 --- /dev/null +++ b/run-local.sh @@ -0,0 +1,149 @@ +#!/bin/bash + +## PREREQUISITES +# 1. Install Kind: https://kind.sigs.k8s.io/docs/user/quick-start/ +# 2. Start Docker Daemon and ensure it's running +# 3. If using SCALEOPS, set SCALEOPS_TOKEN env variable with your ScaleOps API token +# 4. Install and Start cloud-provider-kind to enable LoadBalancer services on Kind (Required for Local Debugging). https://github.com/kubernetes-sigs/cloud-provider-kind + +## USAGE +# ./run-local.sh [--local] [--scaleops] +# +# --local Run koperator as a local process instead of as a container on Kind. +# Starts cloud-provider-kind and runs `make install && make run`. +# --scaleops Install the ScaleOps helm chart. Requires SCALEOPS_TOKEN to be set. + + +## IMPORTANT NOTES (for running koperator locally with --local flag) +# +# Make sure to set `debugEnabled: true` in your KafkaCluster spec. This will +# create LoadBalancer services for the Kafka and Cruise Control pods, allowing +# your local koperator to access services running on the Kind cluster. +# +# Cloud Provider KIND is required to enable LoadBalancer services on Kind. +# If you don't want to run it, you can port-forward the services instead. If you are running in local +# mode and notice that your kafka services don't have an external IP, it's because cloud-provider-kind +# either isn't running or has some issue. Local koperator won't be able to communicate +# with kafka pods without these. +# +# Finally, you'll need to update your /etc/hosts file to direct requests from +# Koperator to the LoadBalancer IPs. You can find the LoadBalancer IPs by running: +# kubectl get svc -n kafka +# +# Your /etc/hosts entries should look something like this: +# 172.18.0.7 kafka-0.kafka.svc.cluster.local +# 172.18.0.9 kafka-1.kafka.svc.cluster.local +# 172.18.0.10 kafka-2.kafka.svc.cluster.local +# 172.18.0.11 kafka-all-broker.kafka.svc.cluster.local +# 172.18.0.8 kafka-cruisecontrol-svc.kafka.svc.cluster.local + + +## ATTACHING A DEBUGGER TO LOCAL KOPERATOR +# If you need to debug your local koperator, you can find the logs in /tmp/koperator.log. +# Additionally, you can attach a debugger to the koperator process using VSCODE. Instead of running `make run`, +# start koperator as a Go application with debug enabled from VSCode, and set breakpoints as needed. +# This can be done by simply opening main.go in VSCode, going to the DEBUG Tab, and clicking Run and Debug. + +LOCAL=false +SCALEOPS=false + +while [[ $# -gt 0 ]]; do + case $1 in + --local) LOCAL=true; shift ;; + --scaleops) SCALEOPS=true; shift ;; + *) echo "Unknown flag: $1"; exit 1 ;; + esac +done + +if $SCALEOPS && [[ -n "${SCALEOPS_TOKEN}" ]]; then + echo "Error: --scaleops requires SCALEOPS_TOKEN to be set" + exit 1 +fi + +## Check if Docker daemon is running +if ! docker ps &>/dev/null; then + echo "Error: Docker daemon is not running. Please start Docker and try again." + exit 1 +fi + +## Create kind cluster +kind delete clusters kind-kafka +kind create cluster --config=./tests/e2e/platforms/kind/kind_config.yaml --name=kind-kafka + +## Validate kubectl context is set to kind +CURRENT_CONTEXT=$(kubectl config current-context) +if [[ ! "$CURRENT_CONTEXT" =~ kind ]]; then + echo "Error: kubectl context is not set to a kind cluster. Current context: $CURRENT_CONTEXT" + exit 1 +fi + +## Build/Load images (Kafka 3.7.0) +kind load docker-image docker-pipeline-upstream-mirror.dr-uw2.adobeitc.com/adobe/kafka:2.13-3.7.0 --name kind-kafka + +if ! $LOCAL; then + docker build . -t koperator_e2e_test + kind load docker-image koperator_e2e_test:latest --name kind-kafka +fi + +## Install Helm Charts and CRDs +### project contour +helm repo add contour https://projectcontour.github.io/helm-charts/ --force-update +helm upgrade --install contour contour/contour --namespace projectcontour --create-namespace + +### cert-manager +helm repo add jetstack https://charts.jetstack.io --force-update +helm upgrade --install cert-manager jetstack/cert-manager --namespace cert-manager --create-namespace --version v1.16.2 --set crds.enabled=true + +### zookeeper-operator +helm repo add pravega https://charts.pravega.io --force-update +helm upgrade --install zookeeper-operator pravega/zookeeper-operator --version 0.2.15 --namespace zookeeper --create-namespace --set crd.create=true + +### prometheus +helm repo add prometheus https://prometheus-community.github.io/helm-charts --force-update +helm upgrade --install prometheus prometheus/kube-prometheus-stack --version 54.1.0 --namespace prometheus --create-namespace + +### scaleops +if $SCALEOPS; then + helm upgrade --install --create-namespace -n scaleops-system \ + --repo https://registry.scaleops.com/charts/ \ + --username scaleops --password "${SCALEOPS_TOKEN}" \ + --set scaleopsToken="${SCALEOPS_TOKEN}" \ + --set clusterName="$(kubectl config current-context)" \ + scaleops scaleops + kubectl apply -f config/scaleops/CustomOwnerGrouping.yaml +fi + +## Run Koperator +if $LOCAL; then + ## Check if cloud-provider-kind started successfully + if ! pgrep -f cloud-provider-kind &>/dev/null; then + echo "Warning: cloud-provider-kind failed to start. LoadBalancer services may not work properly." + echo "Check /tmp/cloudproviderkind.log for details." + fi + + kubectl get namespace kafka &>/dev/null || kubectl create namespace kafka + kubectl config set-context --current --namespace=kafka + make install + +else + helm upgrade --install kafka-operator charts/kafka-operator \ + --set operator.image.repository=koperator_e2e_test \ + --set operator.image.tag=latest \ + --set prometheusMetrics.enabled=false \ + --namespace kafka --create-namespace +fi + +## Initialize Zookeeper and Kafka Cluster +kubectl apply -f config/samples/simplezookeeper.yaml -n zookeeper + +if ! $LOCAL; then + kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=kafka-operator -n kafka --timeout=120s + sleep 5 +fi + +kubectl apply -f config/samples/simplekafkacluster.yaml -n kafka + +## Start Local Koperator +if $LOCAL; then + make run +fi diff --git a/tests/e2e/platforms/kind/kind_config.yaml b/tests/e2e/platforms/kind/kind_config.yaml index 65d601b47..15a139f3f 100644 --- a/tests/e2e/platforms/kind/kind_config.yaml +++ b/tests/e2e/platforms/kind/kind_config.yaml @@ -3,6 +3,7 @@ # topology.kubernetes.io/zone (e.g. config/samples/simplekafkacluster_affinity.yaml). kind: Cluster apiVersion: kind.x-k8s.io/v1alpha4 +name: kind-kafka nodes: - role: control-plane kubeadmConfigPatches: @@ -32,9 +33,11 @@ nodes: nodeRegistration: kubeletExtraArgs: node-labels: "topology.kubernetes.io/zone=zone-c" -containerdConfigPatches: -- |- - [plugins."io.containerd.grpc.v1.cri".containerd] - snapshotter = "overlayfs" - [plugins."io.containerd.grpc.v1.cri".registry.mirrors."localhost:5000"] - endpoint = ["http://localhost:5000"] + extraPortMappings: + - containerPort: 80 + hostPort: 80 + listenAddress: "0.0.0.0" + - containerPort: 443 + hostPort: 443 + listenAddress: "0.0.0.0" + \ No newline at end of file