add e2e-test

Author: kobzonega

Makefile

@@ -70,11 +70,11 @@ kind-init:
 	--image=kindest/node:v1.31.2@sha256:18fbefc20a7113353c7b75b5c869d7145a6abd6269154825872dc59c1329912e \
 	--name kind-ydb-operator; \
     docker pull k8s.gcr.io/ingress-nginx/kube-webhook-certgen:v1.0; \
-    kind load docker-image k8s.gcr.io/ingress-nginx/kube-webhook-certgen:v1.0 --nodes kind-worker,kind-worker2,kind-worker3 --name kind-ydb-operator; \
-    YDB_IMAGE=$(grep "anchor_for_fetching_image_from_workflow" ./tests/**/*.go | grep -o -E '"cr\.yandex.*"'); \
-    YDB_IMAGE=${YDB_IMAGE:1:-1}; \
-    docker pull ${YDB_IMAGE}; \
-    kind load docker-image ${YDB_IMAGE} --nodes kind-worker,kind-worker2,kind-worker3 --name kind-ydb-operator;
+	kind load docker-image k8s.gcr.io/ingress-nginx/kube-webhook-certgen:v1.0 --nodes kind-worker,kind-worker2,kind-worker3 --name kind-ydb-operator; \
+	YDB_IMAGE=$(grep "anchor_for_fetching_image_from_workflow" ./tests/**/*.go | grep -o -E '"cr\.yandex.*"'); \
+	YDB_IMAGE=${YDB_IMAGE:1:-1}; \
+	docker pull ${YDB_IMAGE}; \
+	kind load docker-image ${YDB_IMAGE} --nodes kind-worker,kind-worker2,kind-worker3 --name kind-ydb-operator;
 
 kind-load:
 	docker tag cr.yandex/yc/ydb-operator:latest kind/ydb-operator:current

internal/controllers/database/controller_test.go

@@ -310,7 +310,6 @@ var _ = Describe("Database controller medium tests", func() {
 
 	It("Check externalHost and externalPort GRPC Service field propagation", func() {
 		By("Create test database")
-		publicHost := fmt.Sprintf(v1alpha1.InterconnectServiceFQDNFormat, testobjects.DatabaseName, testobjects.YdbNamespace, v1alpha1.DefaultDomainName)
 		databaseSample = *testobjects.DefaultDatabase()
 		Expect(k8sClient.Create(ctx, &databaseSample)).Should(Succeed())
 
@@ -350,14 +349,14 @@ var _ = Describe("Database controller medium tests", func() {
 		}
 
 		By("Check that args `--grpc-public-host` and `--grpc-public-port` propagated to StatefulSet pods...")
+		publicHost := fmt.Sprintf(v1alpha1.InterconnectServiceFQDNFormat, testobjects.DatabaseName, testobjects.YdbNamespace, v1alpha1.DefaultDomainName)
 		Eventually(
 			checkPublicArgs(fmt.Sprintf("%s.%s", "$(NODE_NAME)", publicHost), fmt.Sprintf("%d", v1alpha1.GRPCPort)),
 			test.Timeout,
 			test.Interval).ShouldNot(HaveOccurred())
 
 		externalHost := fmt.Sprintf("%s.%s.%s", testobjects.DatabaseName, testobjects.YdbNamespace, "example.com")
 		externalPort := int32(31001)
-
 		By("Update externalHost and externalPort for Database GRPC Service...", func() {
 			database := v1alpha1.Database{}
 			Expect(k8sClient.Get(ctx, types.NamespacedName{

internal/resources/database.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/rest"
 
 	api "github.com/ydb-platform/ydb-kubernetes-operator/api/v1alpha1"
@@ -152,9 +153,10 @@ func (b *DatabaseBuilder) GetResourceBuilders(restConfig *rest.Config) []Resourc
 			SelectorLabels: databaseLabels,
 			Annotations:    b.Spec.Service.GRPC.AdditionalAnnotations,
 			Ports: []corev1.ServicePort{{
-				Name:     api.GRPCServicePortName,
-				Port:     api.GRPCPort,
-				NodePort: b.Spec.Service.GRPC.ExternalPort,
+				Name:       api.GRPCServicePortName,
+				Port:       api.GRPCPort,
+				TargetPort: intstr.IntOrString{Type: intstr.String, StrVal: "grpc"},
+				NodePort:   b.Spec.Service.GRPC.ExternalPort,
 			}},
 			IPFamilies:     b.Spec.Service.GRPC.IPFamilies,
 			IPFamilyPolicy: b.Spec.Service.GRPC.IPFamilyPolicy,

internal/resources/database_statefulset.go

@@ -427,8 +427,15 @@ func (b *DatabaseStatefulSetBuilder) buildContainer() corev1.Container {
 		}
 	}
 
+	var grpcPublicPort int32
+	if b.Spec.Service.GRPC.ExternalPort != 0 {
+		grpcPublicPort = b.Spec.Service.GRPC.ExternalPort
+	} else {
+		grpcPublicPort = api.GRPCPort
+	}
+
 	ports := []corev1.ContainerPort{{
-		Name: "grpc", ContainerPort: api.GRPCPort,
+		Name: "grpc", ContainerPort: grpcPublicPort,
 	}, {
 		Name: "interconnect", ContainerPort: api.InterconnectPort,
 	}, {

tests/cfg/kind-cluster-config.yaml

@@ -4,16 +4,31 @@ nodes:
 - role: control-plane
 
 - role: worker
+  extraPortMappings:
+  - containerPort: 30001
+    hostPort: 2135
+    listenAddress: "127.0.0.1"
+    protocol: tcp
   labels:
     topology.kubernetes.io/zone: az-1
     worker: true
 
 - role: worker
+  extraPortMappings:
+  - containerPort: 30001
+    hostPort: 2135
+    listenAddress: "127.0.0.1"
+    protocol: tcp
   labels:
     topology.kubernetes.io/zone: az-2
     worker: true
 
 - role: worker
+  extraPortMappings:
+  - containerPort: 30001
+    hostPort: 2135
+    listenAddress: "127.0.0.1"
+    protocol: tcp
   labels:
     topology.kubernetes.io/zone: az-3
     worker: true

tests/e2e/smoke_test.go

@@ -733,6 +733,34 @@ var _ = Describe("Operator smoke test", func() {
 		ExecuteSimpleTableE2ETest(podName, testobjects.YdbNamespace, storageEndpoint, databasePath)
 	})
 
+	It("Check externalPort for Database", func() {
+		By("create storage...")
+		Expect(k8sClient.Create(ctx, storageSample)).Should(Succeed())
+		defer DeleteStorageSafely(ctx, k8sClient, storageSample)
+		By("create database...")
+		databaseSample.Spec.Service.GRPC.ExternalPort = 30001
+		Expect(k8sClient.Create(ctx, databaseSample)).Should(Succeed())
+		defer func() {
+			Expect(k8sClient.Delete(ctx, databaseSample)).Should(Succeed())
+		}()
+
+		By("waiting until Storage is ready...")
+		WaitUntilStorageReady(ctx, k8sClient, storageSample.Name, testobjects.YdbNamespace)
+
+		By("checking that all the storage pods are running and ready...")
+		CheckPodsRunningAndReady(ctx, k8sClient, "ydb-cluster", "kind-storage", storageSample.Spec.Nodes)
+
+		By("waiting until database is ready...")
+		WaitUntilDatabaseReady(ctx, k8sClient, databaseSample.Name, testobjects.YdbNamespace)
+
+		By("checking that all the database pods are running and ready...")
+		CheckPodsRunningAndReady(ctx, k8sClient, "ydb-cluster", "kind-database", databaseSample.Spec.Nodes)
+
+		By("execute simple query with ydb-go-sdk...")
+		databasePath := DatabasePathWithDefaultDomain(databaseSample)
+		ExecuteSimpleTableE2ETestWithSDK(databasePath)
+	})
+
 	AfterEach(func() {
 		UninstallOperatorWithHelm(testobjects.YdbNamespace)
 		Expect(k8sClient.Delete(ctx, &namespace)).Should(Succeed())

tests/test-utils/test-utils.go

@@ -3,6 +3,7 @@ package testutils
 import (
 	"bufio"
 	"context"
+	"database/sql"
 	"fmt"
 	"io"
 	"os"
@@ -23,6 +24,8 @@ import (
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 
+	ydb "github.com/ydb-platform/ydb-go-sdk/v3"
+	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
 	v1alpha1 "github.com/ydb-platform/ydb-kubernetes-operator/api/v1alpha1"
 	. "github.com/ydb-platform/ydb-kubernetes-operator/internal/controllers/constants"
 	testobjects "github.com/ydb-platform/ydb-kubernetes-operator/tests/test-k8s-objects"
@@ -34,11 +37,17 @@ const (
 	Interval                   = time.Second * 2
 	YdbOperatorRemoteChart     = "ydb/ydb-operator"
 	YdbOperatorReleaseName     = "ydb-operator"
+	TestTablePath              = "testfolder/testtable"
 )
 
 var (
 	pathToHelmValuesInLocalInstall  = filepath.Join("..", "cfg", "operator-local-values.yaml")
 	pathToHelmValuesInRemoteInstall = filepath.Join("..", "cfg", "operator-values.yaml")
+
+	createTableQuery = fmt.Sprintf("CREATE TABLE `%s` (testColumnA Utf8, testColumnB Utf8, PRIMARY KEY (testColumnA));", TestTablePath)
+	insertQuery      = fmt.Sprintf("INSERT INTO `%s` (testColumnA, testColumnB) VALUES ('valueA', 'valueB');", TestTablePath)
+	selectQuery      = fmt.Sprintf("SELECT testColumnA, testColumnB FROM `%s`;", TestTablePath)
+	dropTableQuery   = fmt.Sprintf("DROP TABLE `%s`;", TestTablePath)
 )
 
 func InstallLocalOperatorWithHelm(namespace string) {
@@ -203,8 +212,6 @@ func BringYdbCliToPod(podName, podNamespace string) {
 }
 
 func ExecuteSimpleTableE2ETest(podName, podNamespace, storageEndpoint string, databasePath string) {
-	tablePath := "testfolder/testtable"
-
 	tableCreatingInterval := time.Second * 10
 
 	Eventually(func(g Gomega) {
@@ -217,7 +224,7 @@ func ExecuteSimpleTableE2ETest(podName, podNamespace, storageEndpoint string, da
 			"-e", storageEndpoint,
 			"yql",
 			"-s",
-			fmt.Sprintf("CREATE TABLE `%s` (testColumnA Utf8, testColumnB Utf8, PRIMARY KEY (testColumnA));", tablePath),
+			createTableQuery,
 		}
 		output, _ := exec.Command("kubectl", args...).CombinedOutput()
 		fmt.Println(string(output))
@@ -232,7 +239,7 @@ func ExecuteSimpleTableE2ETest(podName, podNamespace, storageEndpoint string, da
 		"-e", storageEndpoint,
 		"yql",
 		"-s",
-		fmt.Sprintf("INSERT INTO `%s` (testColumnA, testColumnB) VALUES ('valueA', 'valueB');", tablePath),
+		insertQuery,
 	}
 	output, err := exec.Command("kubectl", argsInsert...).CombinedOutput()
 	Expect(err).ShouldNot(HaveOccurred(), string(output))
@@ -247,7 +254,7 @@ func ExecuteSimpleTableE2ETest(podName, podNamespace, storageEndpoint string, da
 		"yql",
 		"--format", "csv",
 		"-s",
-		fmt.Sprintf("SELECT * FROM `%s`;", tablePath),
+		selectQuery,
 	}
 	output, err = exec.Command("kubectl", argsSelect...).CombinedOutput()
 	Expect(err).ShouldNot(HaveOccurred(), string(output))
@@ -262,12 +269,74 @@ func ExecuteSimpleTableE2ETest(podName, podNamespace, storageEndpoint string, da
 		"-e", storageEndpoint,
 		"yql",
 		"-s",
-		fmt.Sprintf("DROP TABLE `%s`;", tablePath),
+		dropTableQuery,
 	}
 	output, err = exec.Command("kubectl", argsDrop...).CombinedOutput()
 	Expect(err).ShouldNot(HaveOccurred(), string(output))
 }
 
+func ExecuteSimpleTableE2ETestWithSDK(databasePath string) {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	cc, err := ydb.Open(ctx, fmt.Sprintf("grpc://localhost:2135/%s", databasePath))
+	Expect(err).ShouldNot(HaveOccurred(), string(err.Error()))
+	defer func() { _ = cc.Close(ctx) }()
+
+	c, err := ydb.Connector(cc,
+		ydb.WithAutoDeclare(),
+		ydb.WithTablePathPrefix(TestTablePath),
+	)
+	Expect(err).ShouldNot(HaveOccurred())
+	defer func() { _ = c.Close() }()
+
+	db := sql.OpenDB(c)
+	defer func() { _ = db.Close() }()
+
+	err = retry.Do(ctx, db, func(ctx context.Context, cc *sql.Conn) error {
+		_, err = cc.ExecContext(ydb.WithQueryMode(ctx, ydb.SchemeQueryMode), createTableQuery)
+		if err != nil {
+			return err
+		}
+		return nil
+	}, retry.WithIdempotent(true))
+	Expect(err).ShouldNot(HaveOccurred())
+
+	err = retry.DoTx(ctx, db, func(ctx context.Context, tx *sql.Tx) error {
+		if _, err = tx.ExecContext(ctx, insertQuery); err != nil {
+			return err
+		}
+		return nil
+	}, retry.WithIdempotent(true))
+	Expect(err).ShouldNot(HaveOccurred())
+
+	var (
+		testColumnA string
+		testColumnB string
+	)
+	err = retry.Do(ctx, db, func(ctx context.Context, cc *sql.Conn) (err error) {
+		row := cc.QueryRowContext(ctx, selectQuery)
+		if err = row.Scan(&testColumnA, &testColumnB); err != nil {
+			return err
+		}
+
+		return nil
+	}, retry.WithIdempotent(true))
+	Expect(err).ShouldNot(HaveOccurred())
+	Expect(testColumnA).To(BeEquivalentTo("valueA"))
+	Expect(testColumnB).To(BeEquivalentTo("valueB"))
+
+	err = retry.Do(ctx, db, func(ctx context.Context, cc *sql.Conn) error {
+		_, err = cc.ExecContext(ydb.WithQueryMode(ctx, ydb.SchemeQueryMode), dropTableQuery)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	}, retry.WithIdempotent(true))
+	Expect(err).ShouldNot(HaveOccurred())
+}
+
 func PortForward(
 	ctx context.Context,
 	svcName, svcNamespace, serverName string,