Remove old config

Author: ilyario

CHANGELOG.md

@@ -2,6 +2,29 @@
 
 ## [1.0.4](https://github.com/dataflow-operator/dataflow/compare/v1.0.3...v1.0.4) (2026-03-15)
 
+### BREAKING CHANGE
+
+Удалена поддержка legacy-формата конфигурации. Source, sink и transformations теперь используют только формат `type` + `config`.
+
+**Миграция:**
+```yaml
+# Было (legacy):
+source:
+  type: kafka
+  kafka:
+    brokers: [localhost:9092]
+    topic: my-topic
+
+# Стало:
+source:
+  type: kafka
+  config:
+    brokers: [localhost:9092]
+    topic: my-topic
+```
+
+Структура внутри `config` совпадает со структурой внутри `kafka`/`postgresql`/`clickhouse`/`trino` — меняется только ключ верхнего уровня.
+
 ## [1.0.3](https://github.com/dataflow-operator/dataflow/compare/v1.0.2...v1.0.3) (2026-03-09)
 
 ## [1.0.2](https://github.com/dataflow-operator/dataflow/compare/v1.0.1...v1.0.2) (2026-03-09)

api/v1/dataflow_types.go

@@ -78,7 +78,6 @@ type DataFlowSpec struct {
 }
 
 // SourceSpec defines the source configuration (type + config).
-// Supports both generic format (type+config) and legacy format (type+kafka/postgresql/etc) when unmarshaling.
 // +kubebuilder:pruning:PreserveUnknownFields
 type SourceSpec struct {
 	// Type of source: kafka, postgresql, trino, clickhouse, nessie, or plugin type
@@ -91,50 +90,6 @@ type SourceSpec struct {
 	Config *runtime.RawExtension `json:"config,omitempty"`
 }
 
-// sourceSpecRaw is used for unmarshaling legacy format (type + kafka/postgresql/etc).
-type sourceSpecRaw struct {
-	Type       string                `json:"type"`
-	Config     *runtime.RawExtension `json:"config,omitempty"`
-	Kafka      *KafkaSourceSpec      `json:"kafka,omitempty"`
-	PostgreSQL *PostgreSQLSourceSpec `json:"postgresql,omitempty"`
-	Trino      *TrinoSourceSpec      `json:"trino,omitempty"`
-	ClickHouse *ClickHouseSourceSpec `json:"clickhouse,omitempty"`
-	Nessie     *NessieSourceSpec     `json:"nessie,omitempty"`
-}
-
-// UnmarshalJSON supports both generic (type+config) and legacy (type+kafka/etc) formats.
-func (s *SourceSpec) UnmarshalJSON(data []byte) error {
-	var r sourceSpecRaw
-	if err := json.Unmarshal(data, &r); err != nil {
-		return err
-	}
-	s.Type = r.Type
-	s.Config = r.Config
-	if s.Config == nil || len(s.Config.Raw) == 0 {
-		var cfg interface{}
-		switch r.Type {
-		case "kafka":
-			cfg = r.Kafka
-		case "postgresql":
-			cfg = r.PostgreSQL
-		case "trino":
-			cfg = r.Trino
-		case "clickhouse":
-			cfg = r.ClickHouse
-		case "nessie":
-			cfg = r.Nessie
-		}
-		if cfg != nil {
-			b, err := json.Marshal(cfg)
-			if err != nil {
-				return err
-			}
-			s.Config = &runtime.RawExtension{Raw: b}
-		}
-	}
-	return nil
-}
-
 // GetKafkaConfig returns Kafka config from Config.
 func (s *SourceSpec) GetKafkaConfig() (*KafkaSourceSpec, error) {
 	if s.Config == nil || len(s.Config.Raw) == 0 {
@@ -505,7 +460,6 @@ type KeycloakConfig struct {
 }
 
 // SinkSpec defines the sink configuration (type + config).
-// Supports both generic format (type+config) and legacy format (type+kafka/postgresql/etc) when unmarshaling.
 // +kubebuilder:pruning:PreserveUnknownFields
 type SinkSpec struct {
 	// Type of sink: kafka, postgresql, trino, clickhouse, nessie, or plugin type
@@ -517,50 +471,6 @@ type SinkSpec struct {
 	Config *runtime.RawExtension `json:"config,omitempty"`
 }
 
-// sinkSpecRaw is used for unmarshaling legacy format.
-type sinkSpecRaw struct {
-	Type       string                `json:"type"`
-	Config     *runtime.RawExtension `json:"config,omitempty"`
-	Kafka      *KafkaSinkSpec        `json:"kafka,omitempty"`
-	PostgreSQL *PostgreSQLSinkSpec   `json:"postgresql,omitempty"`
-	Trino      *TrinoSinkSpec        `json:"trino,omitempty"`
-	ClickHouse *ClickHouseSinkSpec   `json:"clickhouse,omitempty"`
-	Nessie     *NessieSinkSpec       `json:"nessie,omitempty"`
-}
-
-// UnmarshalJSON supports both generic and legacy formats.
-func (s *SinkSpec) UnmarshalJSON(data []byte) error {
-	var r sinkSpecRaw
-	if err := json.Unmarshal(data, &r); err != nil {
-		return err
-	}
-	s.Type = r.Type
-	s.Config = r.Config
-	if s.Config == nil || len(s.Config.Raw) == 0 {
-		var cfg interface{}
-		switch r.Type {
-		case "kafka":
-			cfg = r.Kafka
-		case "postgresql":
-			cfg = r.PostgreSQL
-		case "trino":
-			cfg = r.Trino
-		case "clickhouse":
-			cfg = r.ClickHouse
-		case "nessie":
-			cfg = r.Nessie
-		}
-		if cfg != nil {
-			b, err := json.Marshal(cfg)
-			if err != nil {
-				return err
-			}
-			s.Config = &runtime.RawExtension{Raw: b}
-		}
-	}
-	return nil
-}
-
 // GetKafkaConfig returns Kafka sink config.
 func (s *SinkSpec) GetKafkaConfig() (*KafkaSinkSpec, error) {
 	if s.Config == nil || len(s.Config.Raw) == 0 {
@@ -905,7 +815,6 @@ type SASLConfig struct {
 }
 
 // TransformationSpec defines a transformation to apply (type + config).
-// Supports both generic format (type+config) and legacy format (type+timestamp/flatten/etc) when unmarshaling.
 // +kubebuilder:pruning:PreserveUnknownFields
 type TransformationSpec struct {
 	// Type of transformation: timestamp, flatten, filter, mask, router, select, remove, snakeCase, camelCase
@@ -917,62 +826,6 @@ type TransformationSpec struct {
 	Config *runtime.RawExtension `json:"config,omitempty"`
 }
 
-// transformationSpecRaw is used for unmarshaling legacy format.
-type transformationSpecRaw struct {
-	Type      string                   `json:"type"`
-	Config    *runtime.RawExtension    `json:"config,omitempty"`
-	Timestamp *TimestampTransformation `json:"timestamp,omitempty"`
-	Flatten   *FlattenTransformation   `json:"flatten,omitempty"`
-	Filter    *FilterTransformation    `json:"filter,omitempty"`
-	Mask      *MaskTransformation      `json:"mask,omitempty"`
-	Router    *RouterTransformation    `json:"router,omitempty"`
-	Select    *SelectTransformation    `json:"select,omitempty"`
-	Remove    *RemoveTransformation    `json:"remove,omitempty"`
-	SnakeCase *SnakeCaseTransformation `json:"snakeCase,omitempty"`
-	CamelCase *CamelCaseTransformation `json:"camelCase,omitempty"`
-}
-
-// UnmarshalJSON supports both generic and legacy formats.
-func (t *TransformationSpec) UnmarshalJSON(data []byte) error {
-	var r transformationSpecRaw
-	if err := json.Unmarshal(data, &r); err != nil {
-		return err
-	}
-	t.Type = r.Type
-	t.Config = r.Config
-	if t.Config == nil || len(t.Config.Raw) == 0 {
-		var cfg interface{}
-		switch r.Type {
-		case "timestamp":
-			cfg = r.Timestamp
-		case "flatten":
-			cfg = r.Flatten
-		case "filter":
-			cfg = r.Filter
-		case "mask":
-			cfg = r.Mask
-		case "router":
-			cfg = r.Router
-		case "select":
-			cfg = r.Select
-		case "remove":
-			cfg = r.Remove
-		case "snakeCase":
-			cfg = r.SnakeCase
-		case "camelCase":
-			cfg = r.CamelCase
-		}
-		if cfg != nil {
-			b, err := json.Marshal(cfg)
-			if err != nil {
-				return err
-			}
-			t.Config = &runtime.RawExtension{Raw: b}
-		}
-	}
-	return nil
-}
-
 // GetTimestampConfig returns Timestamp transformation config.
 func (t *TransformationSpec) GetTimestampConfig() (*TimestampTransformation, error) {
 	if t.Config == nil || len(t.Config.Raw) == 0 {

config/crd/bases/dataflow.dataflow.io_dataflows.yaml

@@ -1149,9 +1149,8 @@ spec:
                 description: Transformations is a list of transformations to apply
                   to messages
                 items:
-                  description: |-
-                    TransformationSpec defines a transformation to apply (type + config).
-                    Supports both generic format (type+config) and legacy format (type+timestamp/flatten/etc) when unmarshaling.
+                  description: TransformationSpec defines a transformation to apply
+                    (type + config).
                   properties:
                     config:
                       description: Config holds transformation configuration. Structure
@@ -1171,7 +1170,6 @@ spec:
             - sink
             - source
             type: object
-            x-kubernetes-preserve-unknown-fields: true
           status:
             description: DataFlowStatus defines the observed state of DataFlow
             properties:

config/samples/clickhouse-to-clickhouse.yaml

@@ -5,13 +5,13 @@ metadata:
 spec:
   source:
     type: clickhouse
-    clickhouse:
+    config:
       connectionString: "clickhouse://dataflow:dataflow@localhost:9000/dataflow?dial_timeout=10s"
       table: products
       pollInterval: 5
   sink:
     type: clickhouse
-    clickhouse:
+    config:
       connectionString: "clickhouse://dataflow:dataflow@localhost:9000/dataflow?dial_timeout=10s"
       table: products_clone
       batchSize: 100

config/samples/clickhouse-to-clickhouse2.yaml

@@ -5,13 +5,13 @@ metadata:
 spec:
   source:
     type: clickhouse
-    clickhouse:
+    config:
       connectionString: "clickhouse://dataflow:dataflow@localhost:9000/dataflow?dial_timeout=10s"
       table: products
       pollInterval: 5
   sink:
     type: clickhouse
-    clickhouse:
+    config:
       connectionString: "clickhouse://dataflow:dataflow@localhost:9000/dataflow?dial_timeout=10s"
       table: products_row
       batchSize: 100

config/samples/flatten-example.yaml

@@ -5,22 +5,21 @@ metadata:
 spec:
   source:
     type: kafka
-    kafka:
+    config:
       brokers:
         - localhost:9092
       topic: stock-topic
       consumerGroup: dataflow-group
   sink:
     type: postgresql
-    postgresql:
+    config:
       connectionString: "postgres://dataflow:dataflow@postgres:5432/dataflow?sslmode=disable"
       table: stock_items
       autoCreateTable: true
   transformations:
     - type: flatten
-      flatten:
+      config:
         field: rowsStock
     - type: timestamp
-      timestamp:
+      config:
         fieldName: created_at
-

config/samples/kafka-to-clickhouse.yaml

@@ -5,14 +5,14 @@ metadata:
 spec:
   source:
     type: kafka
-    kafka:
+    config:
       brokers:
         - localhost:9092
       topic: input-topic
       consumerGroup: dataflow-group
   sink:
     type: clickhouse
-    clickhouse:
+    config:
       connectionString: "clickhouse://default@clickhouse:9000/default?dial_timeout=10s"
       table: output_table
       batchSize: 100

config/samples/kafka-to-postgres-secrets.yaml

@@ -28,7 +28,7 @@ metadata:
 spec:
   source:
     type: kafka
-    kafka:
+    config:
       brokersSecretRef:
         name: kafka-credentials
         key: brokers
@@ -48,11 +48,10 @@ spec:
           key: password
   sink:
     type: postgresql
-    postgresql:
+    config:
       connectionStringSecretRef:
         name: postgres-credentials
         key: connectionString
       tableSecretRef:
         name: postgres-credentials
         key: table
-

config/samples/kafka-to-postgres-with-errors.yaml

@@ -5,19 +5,19 @@ metadata:
 spec:
   source:
     type: kafka
-    kafka:
+    config:
       brokers:
         - localhost:9092
       topic: input-topic
       consumerGroup: dataflow-group
   sink:
     type: postgresql
-    postgresql:
+    config:
       connectionString: "postgres://dataflow:dataflow@postgres:5432/dataflow?sslmode=disable"
       table: output_table
   errors:
     type: kafka
-    kafka:
+    config:
       brokers:
         - localhost:9092
       topic: error-topic

config/samples/kafka-to-postgres-with-resources.yaml

@@ -5,14 +5,14 @@ metadata:
 spec:
   source:
     type: kafka
-    kafka:
+    config:
       brokers:
         - localhost:9092
       topic: input-topic
       consumerGroup: dataflow-group
   sink:
     type: postgresql
-    postgresql:
+    config:
       connectionString: "postgres://dataflow:dataflow@postgres:5432/dataflow?sslmode=disable"
       table: output_table
   # Настройка ресурсов для пода процессора