Add support clickhouse

Author: ilyario

.github/workflows/build-and-push.yml

@@ -57,6 +57,9 @@ jobs:
           push: true
           tags: ${{ steps.meta.outputs.tags }}
           labels: ${{ steps.meta.outputs.labels }}
+          platforms: linux/amd64,linux/arm64
+          build-args: |
+            VERSION=${{ github.ref_name }}
           cache-from: type=gha
           cache-to: type=gha,mode=max
 

Taskfile.yml

@@ -57,17 +57,17 @@ tasks:
     cmds:
       - |
         KUBEBUILDER_ASSETS="$$({{.LOCALBIN}}/setup-envtest use {{.ENVTEST_K8S_VERSION}} -p path)" \
-        go test ./internal/... ./test/... -coverprofile cover.out || true
+        GOTOOLCHAIN=go1.25.0 go test ./internal/... ./test/... -coverprofile cover.out || true
 
   test-unit:
     desc: Run unit tests without envtest with statistics
     cmds:
-      - ./scripts/run-tests-with-stats.sh ./...
+      - GOTOOLCHAIN=go1.25.0 ./scripts/run-tests-with-stats.sh ./...
 
   test-unit-simple:
     desc: Run unit tests without envtest (simple output)
     cmds:
-      - go test ./... -v
+      - GOTOOLCHAIN=go1.25.0 go test ./... -v
 
   test-all:
     desc: Run all unit tests (alias for test-unit)
@@ -77,12 +77,12 @@ tasks:
   test-integration:
     desc: Run integration tests with statistics
     cmds:
-      - ./scripts/run-tests-with-stats.sh ./test/integration/...
+      - GOTOOLCHAIN=go1.25.0 ./scripts/run-tests-with-stats.sh ./test/integration/...
 
   test-integration-simple:
     desc: Run integration tests (simple output)
     cmds:
-      - go test ./test/integration/... -v
+      - GOTOOLCHAIN=go1.25.0 go test ./test/integration/... -v
 
   # Build tasks
   build:

api/v1/dataflow_types.go

@@ -70,6 +70,10 @@ type SourceSpec struct {
 	// Trino source configuration
 	// +optional
 	Trino *TrinoSourceSpec `json:"trino,omitempty"`
+
+	// ClickHouse source configuration
+	// +optional
+	ClickHouse *ClickHouseSourceSpec `json:"clickhouse,omitempty"`
 }
 
 // KafkaSourceSpec defines Kafka source configuration
@@ -230,6 +234,31 @@ type TrinoSourceSpec struct {
 	TableSecretRef *SecretRef `json:"tableSecretRef,omitempty"`
 }
 
+// ClickHouseSourceSpec defines ClickHouse source configuration
+type ClickHouseSourceSpec struct {
+	// ConnectionString for ClickHouse database (e.g., clickhouse://host:9000?username=default&password=xxx&database=default)
+	ConnectionString string `json:"connectionString"`
+
+	// Table to read from
+	Table string `json:"table"`
+
+	// Query for custom SQL query (optional, if not provided, reads from table)
+	// +optional
+	Query string `json:"query,omitempty"`
+
+	// PollInterval in seconds for polling mode
+	// +optional
+	PollInterval *int32 `json:"pollInterval,omitempty"`
+
+	// ConnectionStringSecretRef references a Kubernetes secret for connection string
+	// +optional
+	ConnectionStringSecretRef *SecretRef `json:"connectionStringSecretRef,omitempty"`
+
+	// TableSecretRef references a Kubernetes secret for table name
+	// +optional
+	TableSecretRef *SecretRef `json:"tableSecretRef,omitempty"`
+}
+
 // KeycloakConfig defines Keycloak OAuth2/OIDC authentication configuration
 type KeycloakConfig struct {
 	// ServerURL is the Keycloak server URL (e.g., https://keycloak.example.com/auth)
@@ -304,6 +333,10 @@ type SinkSpec struct {
 	// Trino sink configuration
 	// +optional
 	Trino *TrinoSinkSpec `json:"trino,omitempty"`
+
+	// ClickHouse sink configuration
+	// +optional
+	ClickHouse *ClickHouseSinkSpec `json:"clickhouse,omitempty"`
 }
 
 // KafkaSinkSpec defines Kafka sink configuration
@@ -409,6 +442,35 @@ type TrinoSinkSpec struct {
 	TableSecretRef *SecretRef `json:"tableSecretRef,omitempty"`
 }
 
+// ClickHouseSinkSpec defines ClickHouse sink configuration
+type ClickHouseSinkSpec struct {
+	// ConnectionString for ClickHouse database (e.g., clickhouse://host:9000?username=default&password=xxx&database=default)
+	ConnectionString string `json:"connectionString"`
+
+	// Table to write to
+	Table string `json:"table"`
+
+	// BatchSize for batch inserts
+	// +optional
+	BatchSize *int32 `json:"batchSize,omitempty"`
+
+	// BatchFlushIntervalSeconds flushes the batch after this many seconds even if BatchSize is not reached (default: 10)
+	// +optional
+	BatchFlushIntervalSeconds *int32 `json:"batchFlushIntervalSeconds,omitempty"`
+
+	// AutoCreateTable automatically creates the table if it doesn't exist
+	// +optional
+	AutoCreateTable *bool `json:"autoCreateTable,omitempty"`
+
+	// ConnectionStringSecretRef references a Kubernetes secret for connection string
+	// +optional
+	ConnectionStringSecretRef *SecretRef `json:"connectionStringSecretRef,omitempty"`
+
+	// TableSecretRef references a Kubernetes secret for table name
+	// +optional
+	TableSecretRef *SecretRef `json:"tableSecretRef,omitempty"`
+}
+
 // SecretRef references a Kubernetes secret
 type SecretRef struct {
 	// Name of the secret

api/v1/dataflow_validation.go

@@ -22,10 +22,10 @@ import (
 )
 
 // Valid source types (must match connectors/factory.go).
-var validSourceTypes = map[string]bool{"kafka": true, "postgresql": true, "trino": true}
+var validSourceTypes = map[string]bool{"kafka": true, "postgresql": true, "trino": true, "clickhouse": true}
 
 // Valid sink types (must match connectors/factory.go).
-var validSinkTypes = map[string]bool{"kafka": true, "postgresql": true, "trino": true}
+var validSinkTypes = map[string]bool{"kafka": true, "postgresql": true, "trino": true, "clickhouse": true}
 
 // Valid transformation types (must match transformers/factory.go).
 var validTransformationTypes = map[string]bool{
@@ -61,7 +61,7 @@ func validateSource(s *SourceSpec, f *field.Path) field.ErrorList {
 		return all
 	}
 	if !validSourceTypes[s.Type] {
-		all = append(all, field.NotSupported(f.Child("type"), s.Type, []string{"kafka", "postgresql", "trino"}))
+		all = append(all, field.NotSupported(f.Child("type"), s.Type, []string{"kafka", "postgresql", "trino", "clickhouse"}))
 		return all
 	}
 	switch s.Type {
@@ -83,6 +83,12 @@ func validateSource(s *SourceSpec, f *field.Path) field.ErrorList {
 		} else {
 			all = append(all, validateTrinoSource(s.Trino, f.Child("trino"))...)
 		}
+	case "clickhouse":
+		if s.ClickHouse == nil {
+			all = append(all, field.Required(f.Child("clickhouse"), "clickhouse source configuration is required"))
+		} else {
+			all = append(all, validateClickHouseSource(s.ClickHouse, f.Child("clickhouse"))...)
+		}
 	}
 	return all
 }
@@ -169,7 +175,7 @@ func validateSink(s *SinkSpec, f *field.Path) field.ErrorList {
 		return all
 	}
 	if !validSinkTypes[s.Type] {
-		all = append(all, field.NotSupported(f.Child("type"), s.Type, []string{"kafka", "postgresql", "trino"}))
+		all = append(all, field.NotSupported(f.Child("type"), s.Type, []string{"kafka", "postgresql", "trino", "clickhouse"}))
 		return all
 	}
 	switch s.Type {
@@ -191,6 +197,12 @@ func validateSink(s *SinkSpec, f *field.Path) field.ErrorList {
 		} else {
 			all = append(all, validateTrinoSink(s.Trino, f.Child("trino"))...)
 		}
+	case "clickhouse":
+		if s.ClickHouse == nil {
+			all = append(all, field.Required(f.Child("clickhouse"), "clickhouse sink configuration is required"))
+		} else {
+			all = append(all, validateClickHouseSink(s.ClickHouse, f.Child("clickhouse"))...)
+		}
 	}
 	return all
 }
@@ -266,6 +278,44 @@ func validateTrinoSink(t *TrinoSinkSpec, f *field.Path) field.ErrorList {
 	return all
 }
 
+func validateClickHouseSource(c *ClickHouseSourceSpec, f *field.Path) field.ErrorList {
+	var all field.ErrorList
+	hasConn := c.ConnectionString != "" || c.ConnectionStringSecretRef != nil
+	if !hasConn {
+		all = append(all, field.Required(f.Child("connectionString"), "connectionString or connectionStringSecretRef is required"))
+	}
+	hasTable := c.Table != "" || c.TableSecretRef != nil
+	if !hasTable {
+		all = append(all, field.Required(f.Child("table"), "table or tableSecretRef is required"))
+	}
+	if c.ConnectionStringSecretRef != nil {
+		all = append(all, validateSecretRef(c.ConnectionStringSecretRef, f.Child("connectionStringSecretRef"))...)
+	}
+	if c.TableSecretRef != nil {
+		all = append(all, validateSecretRef(c.TableSecretRef, f.Child("tableSecretRef"))...)
+	}
+	return all
+}
+
+func validateClickHouseSink(c *ClickHouseSinkSpec, f *field.Path) field.ErrorList {
+	var all field.ErrorList
+	hasConn := c.ConnectionString != "" || c.ConnectionStringSecretRef != nil
+	if !hasConn {
+		all = append(all, field.Required(f.Child("connectionString"), "connectionString or connectionStringSecretRef is required"))
+	}
+	hasTable := c.Table != "" || c.TableSecretRef != nil
+	if !hasTable {
+		all = append(all, field.Required(f.Child("table"), "table or tableSecretRef is required"))
+	}
+	if c.ConnectionStringSecretRef != nil {
+		all = append(all, validateSecretRef(c.ConnectionStringSecretRef, f.Child("connectionStringSecretRef"))...)
+	}
+	if c.TableSecretRef != nil {
+		all = append(all, validateSecretRef(c.TableSecretRef, f.Child("tableSecretRef"))...)
+	}
+	return all
+}
+
 func validateSecretRef(r *SecretRef, f *field.Path) field.ErrorList {
 	var all field.ErrorList
 	if r == nil {

api/v1/zz_generated.deepcopy.go

@@ -42,19 +42,19 @@ func main() {
 	flag.StringVar(&bindAddr, "bind-address", ":8080", "The address the GUI server binds to.")
 	flag.StringVar(&logLevel, "log-level", "info", "Log level: debug, info, warn, error")
 
-	// kubeconfig может быть зарегистрирован плагинами auth, используем переменную окружения или стандартный путь
+	// kubeconfig may be registered by auth plugins; use env var or standard path
 	flag.Parse()
 
-	// Получаем kubeconfig из переменной окружения или используем стандартный путь
+	// Get kubeconfig from env var or use standard path
 	kubeconfig := os.Getenv("KUBECONFIG")
 	if kubeconfig == "" {
-		// Попробуем получить из флага, если он был зарегистрирован плагином
+		// Try to get from flag if it was registered by a plugin
 		if f := flag.Lookup("kubeconfig"); f != nil {
 			kubeconfig = f.Value.String()
 		}
 	}
 
-	// Устанавливаем уровень логирования
+	// Set log level
 	var level zapcore.Level
 	switch logLevel {
 	case "debug":
@@ -74,23 +74,23 @@ func main() {
 	config.EncoderConfig = zap.NewProductionEncoderConfig()
 	zapLogger, err := config.Build()
 	if err != nil {
-		// Не panic — пишем в stderr и выходим с кодом 2, чтобы сообщение было видно в логах пода
+		// Don't panic — write to stderr and exit with code 2 so the message is visible in pod logs
 		os.Stderr.WriteString("gui-server: failed to create logger: " + err.Error() + "\n")
 		os.Exit(2)
 	}
 	ctrl.SetLogger(zapr.NewLogger(zapLogger))
 
 	setupLog.Info("Starting GUI server", "bind-address", bindAddr)
 
-	// Создаем GUI сервер
+	// Create GUI server
 	server, err := gui.NewServer(bindAddr, kubeconfig)
 	if err != nil {
 		setupLog.Error(err, "unable to create GUI server")
-		_ = zapLogger.Sync() // сброс буфера, чтобы сообщение об ошибке попало в логи пода
+		_ = zapLogger.Sync() // flush buffer so error message appears in pod logs
 		os.Exit(1)
 	}
 
-	// Запускаем сервер
+	// Start server
 	if err := server.Start(); err != nil {
 		setupLog.Error(err, "unable to start GUI server")
 		_ = zapLogger.Sync()