test(e2e):Run controller in e2e tests process, dump namespace obejcts and pod logs on failure.

Author: GrigoryPervakov

.github/workflows/ci.yaml

@@ -225,9 +225,6 @@ jobs:
       - name: Go Mod
         run: go mod download
 
-      - name: Build image
-        run: make docker-build IMG="clickhouse.com/clickhouse-operator:v0.0.1" BUILD_TIME=e2e
-
       - name: Run compatibility e2e tests
         run: make test-compat-e2e
         env:
@@ -238,7 +235,7 @@ jobs:
         if: ${{ !cancelled() }}
         with:
           name: compat-e2e-report-${{ strategy.job-index }}
-          path: "**/report/*.xml"
+          path: "**/report/*"
           if-no-files-found: error
           overwrite: true
 
@@ -261,9 +258,6 @@ jobs:
       - name: Go Mod
         run: go mod download
 
-      - name: Build image
-        run: make docker-build IMG="clickhouse.com/clickhouse-operator:v0.0.1" BUILD_TIME=e2e
-
       - name: Create k8s Kind Cluster
         uses: helm/kind-action@v1
         with:
@@ -280,7 +274,7 @@ jobs:
         if: ${{ !cancelled() }}
         with:
           name: e2e-report-${{ matrix.scope }}
-          path: "**/report/*.xml"
+          path: "**/report/*"
           if-no-files-found: error
           overwrite: true
 

.golangci.yml

@@ -103,6 +103,10 @@ linters:
               - "github.com/onsi/gomega"
         - name: package-comments
           disabled: true
+    staticcheck:
+      dot-import-whitelist:
+        - "github.com/onsi/ginkgo/v2"
+        - "github.com/onsi/gomega"
     gosec:
       excludes:
         - G204

cmd/main.go

@@ -195,11 +195,11 @@ func run() error {
 		upgradeChecker = upgrade.NewChecker(updater)
 	}
 
-	if err = keeper.SetupWithManager(mgr, zapLogger, upgradeChecker); err != nil {
+	if err = keeper.SetupWithManager(mgr, zapLogger, upgradeChecker, nil); err != nil {
 		return fmt.Errorf("unable to setup KeeperCluster controller: %w", err)
 	}
 
-	if err = clickhouse.SetupWithManager(mgr, zapLogger, upgradeChecker); err != nil {
+	if err = clickhouse.SetupWithManager(mgr, zapLogger, upgradeChecker, nil); err != nil {
 		return fmt.Errorf("unable to setup ClickHouseCluster controller: %w", err)
 	}
 

internal/controller/clickhouse/commands.go

@@ -5,6 +5,8 @@ import (
 	"database/sql"
 	"errors"
 	"fmt"
+	"net"
+	"strconv"
 	"sync"
 
 	"github.com/ClickHouse/clickhouse-go/v2"
@@ -53,16 +55,18 @@ type commander struct {
 	log     controllerutil.Logger
 	cluster *v1.ClickHouseCluster
 	auth    clickhouse.Auth
+	dialer  controllerutil.DialContextFunc
 
 	lock  sync.RWMutex
 	conns map[v1.ClickHouseReplicaID]clickhouse.Conn
 }
 
-func newCommander(log controllerutil.Logger, cluster *v1.ClickHouseCluster, secret *corev1.Secret) *commander {
+func newCommander(log controllerutil.Logger, cluster *v1.ClickHouseCluster, secret *corev1.Secret, dialer controllerutil.DialContextFunc) *commander {
 	return &commander{
 		log:     log.Named("commander"),
 		conns:   map[v1.ClickHouseReplicaID]clickhouse.Conn{},
 		cluster: cluster,
+		dialer:  dialer,
 		auth: clickhouse.Auth{
 			Username: OperatorManagementUsername,
 			Password: string(secret.Data[SecretKeyManagementPassword]),
@@ -409,8 +413,9 @@ func (cmd *commander) getConn(id v1.ClickHouseReplicaID) (clickhouse.Conn, error
 	cmd.log.Debug("creating new ClickHouse connection", "replica_id", id)
 
 	conn, err := clickhouse.Open(&clickhouse.Options{
-		Addr: []string{fmt.Sprintf("%s:%d", cmd.cluster.HostnameByID(id), PortManagement)},
-		Auth: cmd.auth,
+		Addr:        []string{net.JoinHostPort(cmd.cluster.HostnameByID(id), strconv.FormatInt(int64(PortManagement), 10))},
+		Auth:        cmd.auth,
+		DialContext: cmd.dialer,
 		Debugf: func(format string, args ...any) {
 			cmd.log.Debug(fmt.Sprintf(format, args...))
 		},

internal/controller/clickhouse/controller.go

@@ -9,7 +9,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
@@ -37,6 +36,7 @@ type ClusterController struct {
 	Logger   controllerutil.Logger
 	Webhook  webhookv1.ClickHouseClusterWebhook
 	Checker  *upgrade.Checker
+	Dialer   controllerutil.DialContextFunc
 }
 
 // +kubebuilder:rbac:groups=clickhouse.com,resources=clickhouseclusters,verbs=get;list;watch;create;update;patch;delete
@@ -75,7 +75,7 @@ func (cc *ClusterController) Reconcile(ctx context.Context, req ctrl.Request) (c
 	}
 
 	if _, err := cc.Webhook.ValidateCreate(ctx, cluster); err != nil {
-		meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
+		chctrl.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
 			Type:               string(v1.ConditionTypeSpecValid),
 			Status:             metav1.ConditionFalse,
 			Reason:             string(v1.ConditionReasonSpecInvalid),
@@ -90,7 +90,7 @@ func (cc *ClusterController) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, nil
 	}
 
-	meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
+	chctrl.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
 		Type:               string(v1.ConditionTypeSpecValid),
 		Status:             metav1.ConditionTrue,
 		Reason:             string(v1.ConditionReasonSpecValid),
@@ -129,8 +129,13 @@ func (cc *ClusterController) GetVersionChecker() *upgrade.Checker {
 	return cc.Checker
 }
 
+// GetDialer returns the custom dialer, or nil.
+func (cc *ClusterController) GetDialer() controllerutil.DialContextFunc {
+	return cc.Dialer
+}
+
 // SetupWithManager sets up the controller with the Manager.
-func SetupWithManager(mgr ctrl.Manager, log controllerutil.Logger, checker *upgrade.Checker) error {
+func SetupWithManager(mgr ctrl.Manager, log controllerutil.Logger, checker *upgrade.Checker, dialer controllerutil.DialContextFunc) error {
 	namedLogger := log.Named("clickhouse")
 
 	clickhouseController := &ClusterController{
@@ -140,6 +145,7 @@ func SetupWithManager(mgr ctrl.Manager, log controllerutil.Logger, checker *upgr
 		Logger:   namedLogger,
 		Webhook:  webhookv1.ClickHouseClusterWebhook{Log: namedLogger},
 		Checker:  checker,
+		Dialer:   dialer,
 	}
 
 	err := ctrl.NewControllerManagedBy(mgr).

internal/controller/clickhouse/controller_test.go

@@ -2,6 +2,8 @@ package clickhouse
 
 import (
 	"context"
+	"errors"
+	"net"
 	"testing"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -75,6 +77,9 @@ var _ = When("reconciling ClickHouseCluster", Ordered, func() {
 			Webhook: webhookv1.ClickHouseClusterWebhook{
 				Log: suite.Log.Named("clickhouse-webhook"),
 			},
+			Dialer: func(context.Context, string) (net.Conn, error) {
+				return nil, errors.New("disabled")
+			},
 		}
 
 		keeper := &v1.KeeperCluster{

internal/controller/clickhouse/sync.go

@@ -118,6 +118,7 @@ type clickhouseReconciler struct {
 	commander *commander
 
 	versionProbe           chctrl.VersionProbeResult
+	readyReplicas          int
 	databasesInSync        bool
 	staleReplicasCleanedUp bool
 	pvcRevision            string
@@ -173,7 +174,7 @@ func (r *clickhouseReconciler) sync(ctx context.Context, log ctrlutil.Logger) (c
 				unknownConditions = append(unknownConditions, r.NewCondition(cond, metav1.ConditionUnknown, v1.ConditionReasonStepFailed, errMsg))
 			}
 
-			meta.SetStatusCondition(&unknownConditions, r.NewCondition(v1.ConditionTypeReconcileSucceeded, metav1.ConditionFalse, v1.ConditionReasonStepFailed, errMsg))
+			chctrl.SetStatusCondition(&unknownConditions, r.NewCondition(v1.ConditionTypeReconcileSucceeded, metav1.ConditionFalse, v1.ConditionReasonStepFailed, errMsg))
 			r.SetConditions(log, unknownConditions)
 
 			if updateErr := r.UpsertStatus(ctx, log); updateErr != nil {
@@ -272,7 +273,7 @@ func (r *clickhouseReconciler) reconcileCommonResources(ctx context.Context, log
 		log.Debug("cluster secret is up to date")
 	}
 
-	r.commander = newCommander(log, r.Cluster, &r.secret)
+	r.commander = newCommander(log, r.Cluster, &r.secret, r.GetDialer())
 
 	return nil, nil
 }
@@ -297,14 +298,20 @@ func (r *clickhouseReconciler) reconcileClusterRevisions(ctx context.Context, lo
 		Namespace: r.Cluster.Namespace,
 		Name:      r.Cluster.Spec.KeeperClusterRef.Name,
 	}, &r.keeper); err != nil {
+		if k8serrors.IsNotFound(err) {
+			log.Debug("keeper cluster not found, waiting")
+
+			return &ctrl.Result{RequeueAfter: chctrl.RequeueOnRefreshTimeout}, nil
+		}
+
 		return nil, fmt.Errorf("get keeper cluster: %w", err)
 	}
 
 	if cond := meta.FindStatusCondition(r.keeper.Status.Conditions, string(v1.ConditionTypeReady)); cond == nil || cond.Status != metav1.ConditionTrue {
 		if cond == nil {
-			log.Warn("keeper cluster is not ready")
+			log.Info("keeper cluster is not ready")
 		} else {
-			log.Warn("keeper cluster is not ready", "reason", cond.Reason, "message", cond.Message)
+			log.Info("keeper cluster is not ready", "reason", cond.Reason, "message", cond.Message)
 		}
 	}
 
@@ -377,7 +384,7 @@ func (r *clickhouseReconciler) reconcileActiveReplicaStatus(ctx context.Context,
 		pinged := false
 		version := ""
 
-		if !hasError {
+		if !hasError && sts.Status.ReadyReplicas > 0 {
 			ctx, cancel := context.WithTimeout(ctx, chctrl.LoadReplicaStateTimeout)
 			defer cancel()
 
@@ -419,6 +426,10 @@ func (r *clickhouseReconciler) reconcileActiveReplicaStatus(ctx context.Context,
 	}
 
 	for id, state := range states {
+		if state.Ready() {
+			r.readyReplicas++
+		}
+
 		if exists := r.SetReplica(id, state); exists {
 			log.Debug(fmt.Sprintf("multiple StatefulSets for single replica %v", id),
 				"replica_id", id, "statefulset", state.StatefulSet.Name)
@@ -491,7 +502,12 @@ func (r *clickhouseReconciler) reconcileReplicaResources(ctx context.Context, lo
 
 func (r *clickhouseReconciler) reconcileReplicateSchema(ctx context.Context, log ctrlutil.Logger) (*ctrl.Result, error) {
 	if !r.Cluster.Spec.Settings.EnableDatabaseSync {
-		log.Info("database sync is disabled, skipping")
+		log.Debug("database sync is disabled, skipping")
+		return nil, nil
+	}
+
+	if r.readyReplicas < 2 {
+		log.Info("no ready replicas to replicate schema, skipping")
 		return nil, nil
 	}
 
@@ -502,11 +518,6 @@ func (r *clickhouseReconciler) reconcileReplicateSchema(ctx context.Context, log
 		}
 	}
 
-	if readyReplicas == nil {
-		log.Info("no ready replicas to replicate schema, skipping")
-		return nil, nil
-	}
-
 	hasNotSynced := false
 	replicaDatabases := ctrlutil.ExecuteParallel(readyReplicas, func(id v1.ClickHouseReplicaID) (v1.ClickHouseReplicaID, map[string]databaseDescriptor, error) {
 		if err := r.commander.EnsureDefaultDatabaseEngine(ctx, log, id); err != nil {
@@ -675,7 +686,7 @@ func (r *clickhouseReconciler) reconcileCleanUp(ctx context.Context, log ctrluti
 		}
 	}
 
-	if r.Cluster.Spec.Settings.EnableDatabaseSync {
+	if r.Cluster.Spec.Settings.EnableDatabaseSync && r.readyReplicas > 0 {
 		if err := r.commander.CleanupDatabaseReplicas(ctx, log, runningStaleReplicas); err != nil {
 			log.Warn("failed to cleanup database replicas", "error", err)
 

internal/controller/keeper/commands.go

@@ -10,8 +10,6 @@ import (
 	"strconv"
 	"strings"
 
-	"golang.org/x/net/proxy"
-
 	"github.com/ClickHouse/clickhouse-operator/internal/controllerutil"
 )
 
@@ -34,26 +32,26 @@ type serverStatus struct {
 	Version     string
 }
 
-func getConnection(ctx context.Context, hostname string, tlsRequired bool) (net.Conn, error) {
-	var d proxy.ContextDialer = &net.Dialer{}
-
+func getConnection(ctx context.Context, dialer controllerutil.DialContextFunc, hostname string, tlsRequired bool) (net.Conn, error) {
 	port := PortNative
 	if tlsRequired {
-		d = &tls.Dialer{
-			NetDialer: &net.Dialer{},
-			Config: &tls.Config{
-				//nolint:gosec // User managed certificate may be outdated or issued for other hostnames.
-				InsecureSkipVerify: true,
-			},
-		}
 		port = PortNativeSecure
 	}
 
-	conn, err := d.DialContext(ctx, "tcp", fmt.Sprintf("%s:%d", hostname, port))
+	addr := net.JoinHostPort(hostname, strconv.FormatInt(int64(port), 10))
+
+	conn, err := dialer(ctx, addr)
 	if err != nil {
 		return nil, fmt.Errorf("connect to %s: %w", hostname, err)
 	}
 
+	if tlsRequired {
+		return tls.Client(conn, &tls.Config{
+			//nolint:gosec // User managed certificate may be outdated or issued for other hostnames.
+			InsecureSkipVerify: true,
+		}), nil
+	}
+
 	return conn, nil
 }
 
@@ -126,8 +124,8 @@ func queryKeeper(ctx context.Context, log controllerutil.Logger, conn net.Conn)
 	return result, nil
 }
 
-func getServerStatus(ctx context.Context, log controllerutil.Logger, hostname string, tlsRequired bool) serverStatus {
-	conn, err := getConnection(ctx, hostname, tlsRequired)
+func getServerStatus(ctx context.Context, log controllerutil.Logger, dialer controllerutil.DialContextFunc, hostname string, tlsRequired bool) serverStatus {
+	conn, err := getConnection(ctx, dialer, hostname, tlsRequired)
 	if err != nil {
 		log.Info("failed to get keeper connection", "error", err)
 		return serverStatus{}

internal/controller/keeper/controller.go

@@ -9,7 +9,6 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	policyv1 "k8s.io/api/policy/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/tools/events"
@@ -34,6 +33,7 @@ type ClusterController struct {
 	Logger   controllerutil.Logger
 	Webhook  webhookv1.KeeperClusterWebhook
 	Checker  *upgrade.Checker
+	Dialer   controllerutil.DialContextFunc
 }
 
 // +kubebuilder:rbac:groups=clickhouse.com,resources=keeperclusters,verbs=get;list;watch;create;update;patch;delete
@@ -72,7 +72,7 @@ func (cc *ClusterController) Reconcile(ctx context.Context, req ctrl.Request) (c
 	}
 
 	if _, err := cc.Webhook.ValidateCreate(ctx, cluster); err != nil {
-		meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
+		chctrl.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
 			Type:               string(v1.ConditionTypeSpecValid),
 			Status:             metav1.ConditionFalse,
 			Reason:             string(v1.ConditionReasonSpecInvalid),
@@ -87,7 +87,7 @@ func (cc *ClusterController) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, nil
 	}
 
-	meta.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
+	chctrl.SetStatusCondition(&cluster.Status.Conditions, metav1.Condition{
 		Type:               string(v1.ConditionTypeSpecValid),
 		Status:             metav1.ConditionTrue,
 		Reason:             string(v1.ConditionReasonSpecValid),
@@ -127,8 +127,13 @@ func (cc *ClusterController) GetVersionChecker() *upgrade.Checker {
 	return cc.Checker
 }
 
+// GetDialer returns the custom dialer, or the default TCP dialer if none was set.
+func (cc *ClusterController) GetDialer() controllerutil.DialContextFunc {
+	return cc.Dialer
+}
+
 // SetupWithManager sets up the controller with the Manager.
-func SetupWithManager(mgr ctrl.Manager, log controllerutil.Logger, checker *upgrade.Checker) error {
+func SetupWithManager(mgr ctrl.Manager, log controllerutil.Logger, checker *upgrade.Checker, dialer controllerutil.DialContextFunc) error {
 	namedLogger := log.Named("keeper")
 
 	keeperController := &ClusterController{
@@ -138,6 +143,7 @@ func SetupWithManager(mgr ctrl.Manager, log controllerutil.Logger, checker *upgr
 		Logger:   namedLogger,
 		Webhook:  webhookv1.KeeperClusterWebhook{Log: namedLogger},
 		Checker:  checker,
+		Dialer:   dialer,
 	}
 
 	err := ctrl.NewControllerManagedBy(mgr).