Add e2e tests for external datastores (#1141)

* Add e2e tests for external datastores Signed-off-by: galal-hussein <hussein.galal.ahmed.11@gmail.com>
2026-08-21 13:26:29 +00:00 · 2026-08-20 02:32:18 +03:00
parent 45d6bc2ec3
commit 9ae5eabd04
3 changed files with 443 additions and 1 deletions
@@ -37,7 +37,7 @@ jobs:
        - name: sync
          label-filter: "pod || app || sync || policy || status"
        - name: features
-          label-filter: "networking || certificates || registry || addons"
+          label-filter: "networking || certificates || registry || addons || datastore"

    name: tests-e2e (${{ matrix.name }})

@@ -0,0 +1,328 @@
+package k3k_test
+
+import (
+	"fmt"
+	"net/url"
+	"strconv"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
+	fwk3k "github.com/rancher/k3k/tests/framework/k3k"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = When("creating a shared mode cluster with postgres datastore via server args", Ordered, Label(datastoreTestsLabel), Label(slowTestsLabel), func() {
+	var virtualCluster *VirtualCluster
+
+	BeforeAll(func() {
+		namespace := fwk3k.CreateNamespace(k8s)
+
+		DeferCleanup(func() {
+			fwk3k.DeleteNamespaces(k8s, namespace.Name)
+		})
+
+		postgresEndpoint := fmt.Sprintf("postgres-k3k.%s.svc.cluster.local:5432", namespace.Name)
+
+		dsn := url.URL{
+			Scheme:   "postgres",
+			User:     url.UserPassword(postgresUser, postgresPassword),
+			Host:     postgresEndpoint,
+			Path:     postgresDatabase,
+			RawQuery: "sslmode=disable",
+		}
+
+		cluster := NewCluster(namespace.Name, func(c *v1beta1.Cluster) {
+			c.Spec.ServerArgs = []string{
+				fmt.Sprintf("--datastore-endpoint=%q", dsn.String()),
+				"--cluster-init=false",
+			}
+		})
+
+		deployPostgresInCluster(cluster)
+
+		CreateCluster(cluster)
+		client, restConfig := NewVirtualK8sClientAndConfig(cluster)
+
+		virtualCluster = &VirtualCluster{
+			Cluster:    cluster,
+			RestConfig: restConfig,
+			Client:     client,
+		}
+	})
+	It("creates and writes to a database for the cluster", func() {
+		Eventually(func(g Gomega) {
+			ctx := GinkgoT().Context()
+
+			count, err := queryPostgres(ctx, virtualCluster.Cluster.Namespace, "SELECT count(*) FROM kine WHERE name LIKE '/registry/namespaces/%'")
+			g.Expect(err).To(Not(HaveOccurred()))
+
+			g.Expect(strconv.Atoi(count)).To(BeNumerically(">", 0))
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+	})
+	It("creates server pods with no etcd finalizers", func() {
+		Consistently(func(g Gomega) {
+			serverPods := listServerPods(GinkgoT().Context(), virtualCluster)
+			for _, s := range serverPods {
+				g.Expect(s.Finalizers).To(BeEmpty())
+			}
+		}).
+			WithTimeout(time.Second * 30).
+			WithPolling(time.Second * 2).
+			Should(Succeed())
+	})
+	It("can create a nginx pod", func() {
+		_, _ = virtualCluster.NewNginxPod("")
+	})
+})
+
+var _ = When("creating a shared mode cluster with postgres datastore via drop-in config file", Ordered, Label(datastoreTestsLabel), Label(slowTestsLabel), func() {
+	var virtualCluster *VirtualCluster
+
+	BeforeAll(func() {
+		ctx := GinkgoT().Context()
+
+		namespace := fwk3k.CreateNamespace(k8s)
+
+		DeferCleanup(func() {
+			fwk3k.DeleteNamespaces(k8s, namespace.Name)
+		})
+
+		postgresEndpoint := fmt.Sprintf("postgres-k3k.%s.svc.cluster.local:5432", namespace.Name)
+
+		dsn := url.URL{
+			Scheme:   "postgres",
+			User:     url.UserPassword(postgresUser, postgresPassword),
+			Host:     postgresEndpoint,
+			Path:     postgresDatabase,
+			RawQuery: "sslmode=disable",
+		}
+
+		configSecretName := "datastore-config-secret"
+		configSecret := corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      configSecretName,
+				Namespace: namespace.Name,
+			},
+			Data: map[string][]byte{
+				"config.yaml": []byte(fmt.Sprintf("datastore-endpoint: %q\ncluster-init: false\n", dsn.String())),
+			},
+		}
+
+		err := k8sClient.Create(ctx, &configSecret)
+		Expect(err).NotTo(HaveOccurred())
+
+		cluster := NewCluster(namespace.Name, func(c *v1beta1.Cluster) {
+			c.Spec.SecretMounts = []v1beta1.SecretMount{
+				{
+					Name: "external-datastore-init-config",
+					SecretVolumeSource: corev1.SecretVolumeSource{
+						SecretName: configSecretName,
+					},
+					MountPath: "/opt/rancher/k3s/init/config.yaml.d/",
+				},
+			}
+		})
+
+		deployPostgresInCluster(cluster)
+
+		CreateCluster(cluster)
+		client, restConfig := NewVirtualK8sClientAndConfig(cluster)
+
+		virtualCluster = &VirtualCluster{
+			Cluster:    cluster,
+			RestConfig: restConfig,
+			Client:     client,
+		}
+	})
+	It("creates and writes to a database for the cluster", func() {
+		Eventually(func(g Gomega) {
+			ctx := GinkgoT().Context()
+
+			count, err := queryPostgres(ctx, virtualCluster.Cluster.Namespace, "SELECT count(*) FROM kine WHERE name LIKE '/registry/namespaces/%'")
+			g.Expect(err).To(Not(HaveOccurred()))
+
+			g.Expect(strconv.Atoi(count)).To(BeNumerically(">", 0))
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+	})
+	It("creates server pods with no etcd finalizers", func() {
+		Consistently(func(g Gomega) {
+			serverPods := listServerPods(GinkgoT().Context(), virtualCluster)
+			for _, s := range serverPods {
+				g.Expect(s.Finalizers).To(BeEmpty())
+			}
+		}).
+			WithTimeout(time.Second * 30).
+			WithPolling(time.Second * 2).
+			Should(Succeed())
+	})
+	It("can create a nginx pod", func() {
+		_, _ = virtualCluster.NewNginxPod("")
+	})
+})
+
+var _ = When("creating a virtual mode cluster with postgres datastore via server args", Ordered, Label(datastoreTestsLabel), Label(slowTestsLabel), func() {
+	var virtualCluster *VirtualCluster
+
+	BeforeAll(func() {
+		namespace := fwk3k.CreateNamespace(k8s)
+
+		DeferCleanup(func() {
+			fwk3k.DeleteNamespaces(k8s, namespace.Name)
+		})
+
+		postgresEndpoint := fmt.Sprintf("postgres-k3k.%s.svc.cluster.local:5432", namespace.Name)
+
+		dsn := url.URL{
+			Scheme:   "postgres",
+			User:     url.UserPassword(postgresUser, postgresPassword),
+			Host:     postgresEndpoint,
+			Path:     postgresDatabase,
+			RawQuery: "sslmode=disable",
+		}
+
+		cluster := NewCluster(namespace.Name, func(c *v1beta1.Cluster) {
+			c.Spec.ServerArgs = []string{
+				fmt.Sprintf("--datastore-endpoint=%q", dsn.String()),
+				"--cluster-init=false",
+			}
+		})
+
+		deployPostgresInCluster(cluster)
+
+		CreateCluster(cluster)
+		client, restConfig := NewVirtualK8sClientAndConfig(cluster)
+
+		virtualCluster = &VirtualCluster{
+			Cluster:    cluster,
+			RestConfig: restConfig,
+			Client:     client,
+		}
+	})
+	It("creates and writes to a database for the cluster", func() {
+		Eventually(func(g Gomega) {
+			ctx := GinkgoT().Context()
+
+			count, err := queryPostgres(ctx, virtualCluster.Cluster.Namespace, "SELECT count(*) FROM kine WHERE name LIKE '/registry/namespaces/%'")
+			g.Expect(err).To(Not(HaveOccurred()))
+
+			g.Expect(strconv.Atoi(count)).To(BeNumerically(">", 0))
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+	})
+	It("creates server pods with no etcd finalizers", func() {
+		Consistently(func(g Gomega) {
+			serverPods := listServerPods(GinkgoT().Context(), virtualCluster)
+			for _, s := range serverPods {
+				g.Expect(s.Finalizers).To(BeEmpty())
+			}
+		}).
+			WithTimeout(time.Second * 30).
+			WithPolling(time.Second * 2).
+			Should(Succeed())
+	})
+	It("can create a nginx pod", func() {
+		_, _ = virtualCluster.NewNginxPod("")
+	})
+})
+
+var _ = When("creating a virtual mode cluster with postgres datastore via drop-in config file", Ordered, Label(datastoreTestsLabel), Label(slowTestsLabel), func() {
+	var virtualCluster *VirtualCluster
+
+	BeforeAll(func() {
+		ctx := GinkgoT().Context()
+
+		namespace := fwk3k.CreateNamespace(k8s)
+
+		DeferCleanup(func() {
+			fwk3k.DeleteNamespaces(k8s, namespace.Name)
+		})
+
+		postgresEndpoint := fmt.Sprintf("postgres-k3k.%s.svc.cluster.local:5432", namespace.Name)
+
+		dsn := url.URL{
+			Scheme:   "postgres",
+			User:     url.UserPassword(postgresUser, postgresPassword),
+			Host:     postgresEndpoint,
+			Path:     postgresDatabase,
+			RawQuery: "sslmode=disable",
+		}
+
+		configSecretName := "datastore-config-secret"
+		configSecret := corev1.Secret{
+			ObjectMeta: metav1.ObjectMeta{
+				Name:      configSecretName,
+				Namespace: namespace.Name,
+			},
+			Data: map[string][]byte{
+				"config.yaml": []byte(fmt.Sprintf("datastore-endpoint: %q\ncluster-init: false\n", dsn.String())),
+			},
+		}
+
+		err := k8sClient.Create(ctx, &configSecret)
+		Expect(err).NotTo(HaveOccurred())
+
+		cluster := NewCluster(namespace.Name, func(c *v1beta1.Cluster) {
+			c.Spec.Mode = v1beta1.VirtualClusterMode
+			c.Spec.SecretMounts = []v1beta1.SecretMount{
+				{
+					Name: "external-datastore-init-config",
+					SecretVolumeSource: corev1.SecretVolumeSource{
+						SecretName: configSecretName,
+					},
+					MountPath: "/opt/rancher/k3s/init/config.yaml.d/",
+				},
+			}
+		})
+
+		deployPostgresInCluster(cluster)
+
+		CreateCluster(cluster)
+		client, restConfig := NewVirtualK8sClientAndConfig(cluster)
+
+		virtualCluster = &VirtualCluster{
+			Cluster:    cluster,
+			RestConfig: restConfig,
+			Client:     client,
+		}
+	})
+	It("creates and writes to a database for the cluster", func() {
+		Eventually(func(g Gomega) {
+			ctx := GinkgoT().Context()
+
+			count, err := queryPostgres(ctx, virtualCluster.Cluster.Namespace, "SELECT count(*) FROM kine WHERE name LIKE '/registry/namespaces/%'")
+			g.Expect(err).To(Not(HaveOccurred()))
+
+			g.Expect(strconv.Atoi(count)).To(BeNumerically(">", 0))
+		}).
+			WithTimeout(time.Minute).
+			WithPolling(time.Second).
+			Should(Succeed())
+	})
+	It("creates server pods with no etcd finalizers", func() {
+		Consistently(func(g Gomega) {
+			serverPods := listServerPods(GinkgoT().Context(), virtualCluster)
+			for _, s := range serverPods {
+				g.Expect(s.Finalizers).To(BeEmpty())
+			}
+		}).
+			WithTimeout(time.Second * 30).
+			WithPolling(time.Second * 2).
+			Should(Succeed())
+	})
+	It("can create a nginx pod", func() {
+		_, _ = virtualCluster.NewNginxPod("")
+	})
+})
@@ -9,6 +9,7 @@ import (
 	"os/exec"
 	"path"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"

@@ -28,6 +29,7 @@ import (
 	networkingv1 "k8s.io/api/networking/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

+	"github.com/rancher/k3k/pkg/apis/k3k.io/v1beta1"
 	fwclient "github.com/rancher/k3k/tests/framework/client"

 	. "github.com/onsi/ginkgo/v2"
@@ -64,12 +66,19 @@ const (
 	certificatesTestsLabel = "certificates"
 	registryTestsLabel     = "registry"
 	addonsTestsLabel       = "addons"
+	datastoreTestsLabel    = "datastore"

 	registryImage               = "registry:2"
 	registryCACertSecretName    = "private-registry-ca-cert"
 	registryCertSecretName      = "private-registry-cert"
 	registryConfigSecretName    = "private-registry-config"
 	k3sRegistryConfigSecretName = "k3s-registry-config"
+
+	postgresImage    = "postgres:17-alpine"
+	postgresUser     = "postgres"
+	postgresPassword = "passw0rd"
+	postgresDatabase = "k3k"
+	postgresPort     = 5432
 )

 func TestTests(t *testing.T) {
@@ -358,6 +367,30 @@ func podExec(ctx context.Context, clientset *kubernetes.Clientset, config *rest.
 	return stderr.Bytes(), nil
 }

+func queryPostgres(ctx context.Context, namespace, query string) (string, error) {
+	podList, err := k8s.CoreV1().Pods(namespace).List(ctx, metav1.ListOptions{
+		LabelSelector: "app=postgres-k3k",
+	})
+	if err != nil {
+		return "", err
+	}
+
+	if len(podList.Items) != 1 {
+		return "", fmt.Errorf("expected 1 postgres pod in namespace %s, found %d", namespace, len(podList.Items))
+	}
+
+	command := []string{"psql", "-U", postgresUser, "-d", postgresDatabase, "-tAc", query}
+
+	var stdout bytes.Buffer
+
+	stderr, err := podExec(ctx, k8s, restcfg, namespace, podList.Items[0].Name, command, nil, &stdout)
+	if err != nil {
+		return "", fmt.Errorf("error running query %q: %v: %s", query, err, stderr)
+	}
+
+	return strings.TrimSpace(stdout.String()), nil
+}
+
 func caCertSecret(name, namespace string, crt, key []byte) *corev1.Secret {
 	return &corev1.Secret{
 		ObjectMeta: metav1.ObjectMeta{
@@ -598,3 +631,84 @@ func buildRegistryConfigSecret(tlsMap map[string]string, namespace, name string,

 	return secret, nil
 }
+
+func deployPostgresInCluster(cluster *v1beta1.Cluster) {
+	ctx := context.Background()
+
+	labels := map[string]string{"app": "postgres-k3k"}
+
+	dep := &appsv1.Deployment{
+		ObjectMeta: metav1.ObjectMeta{Name: "postgres-k3k", Namespace: cluster.Namespace},
+		Spec: appsv1.DeploymentSpec{
+			Replicas: new(int32(1)),
+			Selector: &metav1.LabelSelector{MatchLabels: labels},
+			Template: corev1.PodTemplateSpec{
+				ObjectMeta: metav1.ObjectMeta{Labels: labels},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{{
+						Name:  "postgres",
+						Image: postgresImage,
+						Env: []corev1.EnvVar{
+							{Name: "POSTGRES_USER", Value: postgresUser},
+							{Name: "POSTGRES_PASSWORD", Value: postgresPassword},
+							{Name: "POSTGRES_DB", Value: postgresDatabase},
+						},
+						ReadinessProbe: &corev1.Probe{
+							ProbeHandler: corev1.ProbeHandler{
+								Exec: &corev1.ExecAction{
+									Command: []string{"pg_isready", "-U", postgresUser, "-d", postgresDatabase},
+								},
+							},
+							PeriodSeconds: 2,
+						},
+					}},
+				},
+			},
+		},
+	}
+	err := k8sClient.Create(ctx, dep)
+	Expect(err).To(Not(HaveOccurred()))
+
+	svc := &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{Name: "postgres-k3k", Namespace: cluster.Namespace},
+		Spec: corev1.ServiceSpec{
+			Selector: labels,
+			Ports:    []corev1.ServicePort{{Port: postgresPort, TargetPort: intstr.FromInt(postgresPort)}},
+		},
+	}
+
+	err = k8sClient.Create(ctx, svc)
+	Expect(err).To(Not(HaveOccurred()))
+
+	Eventually(func() bool {
+		podList, err := k8s.CoreV1().Pods(cluster.Namespace).List(ctx, metav1.ListOptions{
+			LabelSelector: "app=postgres-k3k",
+		})
+		Expect(err).To(Not(HaveOccurred()))
+
+		if len(podList.Items) != 1 {
+			return false
+		}
+
+		postgresPod := podList.Items[0]
+
+		_, cond := pod.GetPodCondition(&postgresPod.Status, corev1.PodReady)
+
+		// pod not ready
+		if cond == nil || cond.Status != corev1.ConditionTrue {
+			GinkgoLogr.Info("Waiting for postgres pod to be Ready",
+				"name", postgresPod.Name, "namespace", postgresPod.Namespace,
+				"time", time.Now().Format(time.DateTime),
+			)
+
+			return false
+		}
+
+		return true
+	}).
+		WithTimeout(time.Minute * 5).
+		WithPolling(time.Second * 10).
+		Should(BeTrue())
+
+	By("Postgres is ready")
+}