✨ feat(sink): postgres and kafka export backends

Add first-class postgres upsert and kafka event sinks per ADR-0025 with
testcontainers integration tests; drop invalid prometheus sink stub.

Author: konih

api/v1alpha1/kollectsink_types.go

@@ -7,13 +7,10 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
-// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
-
 // KollectSinkSpec defines the desired state of KollectSink.
 type KollectSinkSpec struct {
 	// type selects the sink backend implementation.
-	// +kubebuilder:validation:Enum=git;gitlab;s3;gcs;prometheus
+	// +kubebuilder:validation:Enum=git;gitlab;s3;gcs;postgres;kafka
 	// +required
 	Type string `json:"type"`
 
@@ -33,6 +30,49 @@ type KollectSinkSpec struct {
 	// The annotation kollect.dev/test-connection=true has the same effect.
 	// +optional
 	ConnectionTest bool `json:"connectionTest,omitempty"`
+
+	// cluster labels exported inventory in multi-cluster installs.
+	// +optional
+	Cluster string `json:"cluster,omitempty"`
+
+	// postgres configures a PostgreSQL database sink.
+	// +optional
+	Postgres *PostgresSpec `json:"postgres,omitempty"`
+
+	// kafka configures a Kafka or Redpanda event sink.
+	// +optional
+	Kafka *KafkaSpec `json:"kafka,omitempty"`
+}
+
+// PostgresSpec configures PostgreSQL upsert export.
+type PostgresSpec struct {
+	// databaseRef references a Secret containing the connection string (key dsn or url).
+	// +required
+	DatabaseRef *SecretReference `json:"databaseRef"`
+
+	// table is the destination table name.
+	// +required
+	Table string `json:"table"`
+
+	// schema is the PostgreSQL schema (default public).
+	// +optional
+	Schema string `json:"schema,omitempty"`
+}
+
+// KafkaSpec configures Kafka inventory change events.
+type KafkaSpec struct {
+	// brokers lists Kafka bootstrap addresses (host:port).
+	// +required
+	// +listType=atomic
+	Brokers []string `json:"brokers"`
+
+	// topic is the destination topic for inventory events.
+	// +required
+	Topic string `json:"topic"`
+
+	// secretRef references a Secret with optional SASL/TLS credentials.
+	// +optional
+	SecretRef *SecretReference `json:"secretRef,omitempty"`
 }
 
 // TLSSpec configures custom CA trust for sink endpoints.
@@ -64,20 +104,7 @@ type SecretReference struct {
 
 // KollectSinkStatus defines the observed state of KollectSink.
 type KollectSinkStatus struct {
-	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster
-	// Important: Run "make" to regenerate code after modifying this file
-
-	// See Kubernetes API conventions for typical status properties.
-
 	// conditions represent the current state of the KollectSink resource.
-	// Each condition has a unique type and reflects the status of a specific aspect of the resource.
-	//
-	// Standard condition types include:
-	// - "Available": the resource is fully functional
-	// - "Progressing": the resource is being created or updated
-	// - "Degraded": the resource failed to reach or maintain its desired state
-	//
-	// The status of each condition is one of True, False, or Unknown.
 	// +listType=map
 	// +listMapKey=type
 	// +optional

api/v1alpha1/zz_generated.deepcopy.go

@@ -41,6 +41,9 @@ spec:
           spec:
             description: spec defines the desired state of KollectSink
             properties:
+              cluster:
+                description: cluster labels exported inventory in multi-cluster installs.
+                type: string
               connectionTest:
                 description: |-
                   connectionTest requests a connectivity check on create/update when true.
@@ -50,6 +53,63 @@ spec:
                 description: endpoint is the backend-specific destination (URL, bucket,
                   and so on).
                 type: string
+              kafka:
+                description: kafka configures a Kafka or Redpanda event sink.
+                properties:
+                  brokers:
+                    description: brokers lists Kafka bootstrap addresses (host:port).
+                    items:
+                      type: string
+                    type: array
+                    x-kubernetes-list-type: atomic
+                  secretRef:
+                    description: secretRef references a Secret with optional SASL/TLS
+                      credentials.
+                    properties:
+                      name:
+                        description: name is the name of the referenced Secret.
+                        type: string
+                      namespace:
+                        description: namespace is the namespace of the referenced
+                          Secret.
+                        type: string
+                    required:
+                    - name
+                    type: object
+                  topic:
+                    description: topic is the destination topic for inventory events.
+                    type: string
+                required:
+                - brokers
+                - topic
+                type: object
+              postgres:
+                description: postgres configures a PostgreSQL database sink.
+                properties:
+                  databaseRef:
+                    description: databaseRef references a Secret containing the connection
+                      string (key dsn or url).
+                    properties:
+                      name:
+                        description: name is the name of the referenced Secret.
+                        type: string
+                      namespace:
+                        description: namespace is the namespace of the referenced
+                          Secret.
+                        type: string
+                    required:
+                    - name
+                    type: object
+                  schema:
+                    description: schema is the PostgreSQL schema (default public).
+                    type: string
+                  table:
+                    description: table is the destination table name.
+                    type: string
+                required:
+                - databaseRef
+                - table
+                type: object
               secretRef:
                 description: secretRef references a Secret holding credentials for
                   the sink.
@@ -99,7 +159,8 @@ spec:
                 - gitlab
                 - s3
                 - gcs
-                - prometheus
+                - postgres
+                - kafka
                 type: string
             required:
             - type
@@ -108,16 +169,8 @@ spec:
             description: status defines the observed state of KollectSink
             properties:
               conditions:
-                description: |-
-                  conditions represent the current state of the KollectSink resource.
-                  Each condition has a unique type and reflects the status of a specific aspect of the resource.
-
-                  Standard condition types include:
-                  - "Available": the resource is fully functional
-                  - "Progressing": the resource is being created or updated
-                  - "Degraded": the resource failed to reach or maintain its desired state
-
-                  The status of each condition is one of True, False, or Unknown.
+                description: conditions represent the current state of the KollectSink
+                  resource.
                 items:
                   description: Condition contains details for one aspect of the current
                     state of this API Resource.

config/crd/bases/kollect.dev_kollectsinks.yaml

@@ -0,0 +1,27 @@
+apiVersion: kollect.dev/v1alpha1
+kind: KollectSink
+metadata:
+  name: kafka-inventory-demo
+  labels:
+    app.kubernetes.io/name: kollect
+    app.kubernetes.io/managed-by: kustomize
+spec:
+  type: kafka
+  cluster: kind-kollect-dev
+  connectionTest: true
+  kafka:
+    brokers:
+      - redpanda.kollect-system.svc:9092
+    topic: kollect-inventory-events
+    secretRef:
+      name: kafka-sasl-credentials
+      namespace: kollect-system
+  # Secret example (optional SASL; create separately):
+  # apiVersion: v1
+  # kind: Secret
+  # metadata:
+  #   name: kafka-sasl-credentials
+  #   namespace: kollect-system
+  # stringData:
+  #   username: kollect
+  #   password: example

config/samples/kollect_v1alpha1_kollectsink_kafka.yaml

@@ -0,0 +1,25 @@
+apiVersion: kollect.dev/v1alpha1
+kind: KollectSink
+metadata:
+  name: postgres-inventory-demo
+  labels:
+    app.kubernetes.io/name: kollect
+    app.kubernetes.io/managed-by: kustomize
+spec:
+  type: postgres
+  cluster: kind-kollect-dev
+  connectionTest: true
+  postgres:
+    databaseRef:
+      name: inventory-postgres-dsn
+      namespace: kollect-system
+    schema: public
+    table: inventory_items
+  # Secret example (create separately; never commit real credentials):
+  # apiVersion: v1
+  # kind: Secret
+  # metadata:
+  #   name: inventory-postgres-dsn
+  #   namespace: kollect-system
+  # stringData:
+  #   dsn: postgres://kollect:example@postgres.kollect-system.svc:5432/inventory?sslmode=disable

config/samples/kollect_v1alpha1_kollectsink_postgres.yaml

@@ -4,6 +4,8 @@ resources:
 - kollect_v1alpha1_kollectprofile_service-endpoints.yaml
 - kollect_v1alpha1_kollectprofile_ingress-hosts.yaml
 - kollect_v1alpha1_kollectsink.yaml
+- kollect_v1alpha1_kollectsink_postgres.yaml
+- kollect_v1alpha1_kollectsink_kafka.yaml
 - kollect_v1alpha1_kollecttarget.yaml
 - kollect_v1alpha1_kollectinventory.yaml
 # +kubebuilder:scaffold:manifestskustomizesamples

config/samples/kustomization.yaml

@@ -10,12 +10,16 @@ require (
 	github.com/go-git/go-billy/v5 v5.9.0
 	github.com/go-git/go-git/v5 v5.19.1
 	github.com/google/cel-go v0.26.0
+	github.com/jackc/pgx/v5 v5.10.0
 	github.com/onsi/ginkgo/v2 v2.27.2
 	github.com/onsi/gomega v1.38.2
 	github.com/prometheus/client_golang v1.23.2
 	github.com/redis/go-redis/v9 v9.20.0
+	github.com/segmentio/kafka-go v0.4.51
 	github.com/testcontainers/testcontainers-go/modules/minio v0.42.0
+	github.com/testcontainers/testcontainers-go/modules/postgres v0.42.0
 	github.com/testcontainers/testcontainers-go/modules/redis v0.42.0
+	github.com/testcontainers/testcontainers-go/modules/redpanda v0.42.0
 	k8s.io/api v0.35.0
 	k8s.io/apimachinery v0.35.0
 	k8s.io/client-go v0.35.0
@@ -83,6 +87,9 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
@@ -108,6 +115,7 @@ require (
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.1 // indirect
+	github.com/pierrec/lz4/v4 v4.1.19 // indirect
 	github.com/pjbgf/sha1cd v0.6.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect
@@ -127,6 +135,9 @@ require (
 	github.com/tklauser/numcpus v0.11.0 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/xanzy/ssh-agent v0.3.3 // indirect
+	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect