Add kafka cluster create test

pkg/resources/kafka/configmap_test.go

@@ -45,7 +45,7 @@ func TestGetMountPathsFromBrokerConfigMap(t *testing.T) {
 listeners=INTERNAL://:29092,CONTROLLER://:29093
 log.dirs=/kafka-logs3/kafka,/kafka-logs/kafka,/kafka-logs2/kafka,/kafka-logs4/kafka
 metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter\noffsets.topic.replication.factor=2
-zookeeper.connect=zookeeper-server-client.zookeeper:2181/
+zookeeper.connect=zookeeper-client.zookeeper:2181/
 `},
 			},
 			expectedLogDirs: []string{"/kafka-logs3/kafka", "/kafka-logs/kafka", "/kafka-logs2/kafka", "/kafka-logs4/kafka"},

tests/e2e/const.go

@@ -41,16 +41,23 @@ const (
 	zookeeperClusterName       = "zookeeper-server"
 	managedByHelmLabelTemplate = "app.kubernetes.io/managed-by=Helm,app.kubernetes.io/instance=%s"
 
+	cruiseControlPodReadinessTimeout       = 50 * time.Second
+	kafkaClusterResourceReadinessTimeout   = 60 * time.Second
 	defaultDeletionTimeout                 = 20 * time.Second
 	defaultPodReadinessWaitTime            = 10 * time.Second
 	defaultTopicCreationWaitTime           = 10 * time.Second
-	kafkaClusterResourceCleanupTimeout     = 30 * time.Second
+	kafkaClusterCreateTimeout              = 500 * time.Second
+	kafkaClusterResourceCleanupTimeout     = 120 * time.Second
+	zookeeperClusterCreateTimeout          = 4 * time.Minute
 	zookeeperClusterResourceCleanupTimeout = 60 * time.Second
 	externalConsumerTimeout                = 5 * time.Second
 	externalProducerTimeout                = 5 * time.Second
 
-	kcatPodTemplate    = "templates/kcat.yaml.tmpl"
-	kafkaTopicTemplate = "templates/topic.yaml.tmpl"
+	zookeeperReplicaCount = 1
+
+	kcatPodTemplate          = "templates/kcat.yaml.tmpl"
+	kafkaTopicTemplate       = "templates/topic.yaml.tmpl"
+	zookeeperClusterTemplate = "templates/zookeeper_cluster.yaml.tmpl"
 
 	kubectlNotFoundErrorMsg = "NotFound"
 )

tests/e2e/helm.go

@@ -311,6 +311,12 @@ func listHelmReleases(kubectlOptions k8s.KubectlOptions) ([]*HelmRelease, error)
 		"list",
 		"--output", "json",
 	)
+
+	fmt.Println("OUTPUT: " + output)
+	if strings.Contains(output, "[]") || strings.Contains(output, "WARNING") {
+		return nil, nil
+	}
+
 	if err != nil {
 		return nil, errors.WrapIf(err, "listing Helm releases failed")
 	}

tests/e2e/install_cluster.go

@@ -0,0 +1,89 @@
+// 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 e2e
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/banzaicloud/koperator/api/v1beta1"
+	"github.com/gruntwork-io/terratest/modules/k8s"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+// requireCreatingKafkaCluster creates a KafkaCluster and
+// checks the success of that operation.
+func requireCreatingKafkaCluster(kubectlOptions k8s.KubectlOptions, koperatorVersion string, sampleFile string) {
+	It("Deploying a KafkaCluster", func() {
+
+		By("Checking existing KafkaClusters")
+		err := checkExistenceOfK8sResource(kubectlOptions, kafkaKind, kafkaClusterName)
+		if err == nil {
+			By(fmt.Sprintf("KafkaCluster %s already exists\n", kafkaClusterName))
+		} else {
+			By("Deploying a KafkaCluster")
+			err = applyKoperatorSampleResource(kubectlOptions, koperatorVersion, sampleFile)
+			Expect(err).NotTo(HaveOccurred())
+		}
+
+		By("Verifying the KafkaCluster state")
+		err = waitK8sResourceCondition(kubectlOptions, kafkaKind, fmt.Sprintf("jsonpath={.status.state}=%s", string(v1beta1.KafkaClusterRunning)), kafkaClusterCreateTimeout, "", kafkaClusterName)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Verifying the CruiseControl pod")
+		Eventually(context.Background(), func() error {
+			return waitK8sResourceCondition(kubectlOptions, "pod", "condition=Ready", cruiseControlPodReadinessTimeout, v1beta1.KafkaCRLabelKey+"="+kafkaClusterName+",app=cruisecontrol", "")
+		}, kafkaClusterResourceReadinessTimeout, 3*time.Second).ShouldNot(HaveOccurred())
+
+		By("Verifying all Kafka pods")
+		err = waitK8sResourceCondition(kubectlOptions, "pod", "condition=Ready", defaultPodReadinessWaitTime, v1beta1.KafkaCRLabelKey+"="+kafkaClusterName, "")
+		Expect(err).NotTo(HaveOccurred())
+	})
+}
+
+// requireCreatingZookeeperCluster creates a ZookeeperCluster and
+// checks the success of that operation.
+func requireCreatingZookeeperCluster(kubectlOptions k8s.KubectlOptions, zkClusterReplicaCount int) {
+	It("Deploying a ZookeeperCluster", func() {
+
+		By("Checking existing ZookeeperClusters")
+		err := checkExistenceOfK8sResource(kubectlOptions, zookeeperKind, zookeeperClusterName)
+		if err == nil {
+			By(fmt.Sprintf("ZookeeperCluster %s already exists\n", zookeeperClusterName))
+		} else {
+			By("Deploying the sample ZookeeperCluster")
+			err = applyK8sResourceFromTemplate(kubectlOptions,
+				zookeeperClusterTemplate,
+				map[string]interface{}{
+					"Name":      zookeeperClusterName,
+					"Namespace": kubectlOptions.Namespace,
+					"Replicas":  zkClusterReplicaCount,
+				},
+			)
+			Expect(err).NotTo(HaveOccurred())
+		}
+
+		By("Verifying the ZookeeperCluster resource")
+		err = waitK8sResourceCondition(kubectlOptions, zookeeperKind, "jsonpath={.status.readyReplicas}=1", zookeeperClusterCreateTimeout, "", zookeeperClusterName)
+		Expect(err).NotTo(HaveOccurred())
+
+		By("Verifying the ZookeeperCluster's pods")
+		err = waitK8sResourceCondition(kubectlOptions, "pod", "condition=Ready", defaultPodReadinessWaitTime, "app="+zookeeperClusterName, "")
+		Expect(err).NotTo(HaveOccurred())
+	})
+}

tests/e2e/k8s.go

@@ -51,6 +51,17 @@ func applyK8sResourceManifest(kubectlOptions k8s.KubectlOptions, manifestPath st
 	k8s.KubectlApply(GinkgoT(), &kubectlOptions, manifestPath)
 }
 
+// checkExistenceOfK8sResource queries a Resource by it's kind, namespace and name and
+// returns the output of stderr
+func checkExistenceOfK8sResource(
+	kubectlOptions k8s.KubectlOptions,
+	resourceKind string,
+	resourceName string,
+) error {
+	By(fmt.Sprintf("Checking the existence of resource %s in namespace %s (kind: %s)", resourceName, kubectlOptions.Namespace, resourceKind))
+	return k8s.RunKubectlE(GinkgoT(), &kubectlOptions, "get", resourceKind, resourceName)
+}
+
 // createOrReplaceK8sResourcesFromManifest creates non-existent Kubernetes
 // resources or replaces existing ones from the specified manifest to the
 // provided kubectl context and namespace.

tests/e2e/koperator.go

@@ -0,0 +1,79 @@
+// 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 e2e
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"time"
+
+	"emperror.dev/errors"
+	"github.com/gruntwork-io/terratest/modules/k8s"
+	. "github.com/onsi/ginkgo/v2"
+)
+
+// applyKoperatorSampleResource deploys the specified sample resource (config/samples).
+// The full path of the manifest also can be specified.
+// It supports different versions that can be specified with the koperatorVersion parameter.
+func applyKoperatorSampleResource(kubectlOptions k8s.KubectlOptions, koperatorVersion Version, sampleFile string) error {
+	By(fmt.Sprintf("Retrieving Koperator sample resource: '%s' with version: '%s' ", sampleFile, koperatorVersion))
+
+	sampleFileSplit := strings.Split(sampleFile, "/")
+	if len(sampleFileSplit) == 0 {
+		return errors.Errorf("sample file path shouldn't be empty")
+	}
+
+	var err error
+	var rawKoperatorSampleResource []byte
+
+	switch koperatorVersion {
+	case LocalVersion:
+		if len(sampleFileSplit) == 1 {
+			sampleFile = fmt.Sprintf("../../config/samples/%s", sampleFile)
+		}
+
+		rawKoperatorSampleResource, err = os.ReadFile(sampleFile)
+		if err != nil {
+			return err
+		}
+	default:
+		httpClient := new(http.Client)
+		httpClient.Timeout = 5 * time.Second
+
+		if len(sampleFileSplit) != 1 {
+			return errors.Errorf("sample file path shouldn't contain a \"/\" character")
+		}
+
+		response, err := httpClient.Get("https://raw.githubusercontent.com/banzaicloud/koperator/" + koperatorVersion + "/config/samples/" + sampleFile)
+		if response != nil {
+			defer func() { _ = response.Body.Close() }()
+		}
+		if err != nil {
+			return err
+		}
+
+		rawKoperatorSampleResource, err = io.ReadAll(response.Body)
+		if err != nil {
+			return err
+		}
+	}
+
+	By(fmt.Sprintf("Applying K8s manifest %s", sampleFile))
+	k8s.KubectlApplyFromString(GinkgoT(), &kubectlOptions, string(rawKoperatorSampleResource))
+	return nil
+}

tests/e2e/koperator_suite_test.go

@@ -54,9 +54,10 @@ var _ = BeforeSuite(func() {
 })
 
 var _ = When("Testing e2e test altogether", Ordered, func() {
-	//testInstall()
-	//testProduceConsumeInternal()
-	testUninstallZookeeperCluster()
+	testInstall()
+	testInstallZookeeperCluster()
+	testInstallKafkaCluster("local", "simplekafkacluster.yaml")
 	testUninstallKafkaCluster()
+	testUninstallZookeeperCluster()
 	testUninstall()
 })

tests/e2e/templates/zookeeper_cluster.yaml.tmpl

@@ -0,0 +1,9 @@
+apiVersion: zookeeper.pravega.io/v1beta1
+kind: ZookeeperCluster
+metadata:
+    name: {{ .Name }}
+    namespace: {{ or .Namespace "zookeeper" }}
+spec:
+    replicas: {{ or .Replicas 1 }}
+    persistence:
+        reclaimPolicy: Delete

tests/e2e/test_install_cluster.go

@@ -0,0 +1,51 @@
+// 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 e2e
+
+import (
+	"github.com/gruntwork-io/terratest/modules/k8s"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func testInstallZookeeperCluster() bool {
+	return When("Installing Zookeeper cluster", func() {
+		var kubectlOptions k8s.KubectlOptions
+		var err error
+
+		It("Acquiring K8s config and context", func() {
+			kubectlOptions, err = kubectlOptionsForCurrentContext()
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		kubectlOptions.Namespace = zookeeperOperatorHelmDescriptor.Namespace
+		requireCreatingZookeeperCluster(kubectlOptions, zookeeperReplicaCount)
+	})
+}
+
+func testInstallKafkaCluster(koperatorVersion string, kafkaClusterSample string) bool {
+	return When("Installing Kafka cluster", func() {
+		var kubectlOptions k8s.KubectlOptions
+		var err error
+
+		It("Acquiring K8s config and context", func() {
+			kubectlOptions, err = kubectlOptionsForCurrentContext()
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		kubectlOptions.Namespace = koperatorLocalHelmDescriptor.Namespace
+		requireCreatingKafkaCluster(kubectlOptions, koperatorVersion, kafkaClusterSample)
+	})
+}