chore(vector): add vector pipeline to ship clickhouse events from the edge

Author: NathanFlurry

Cargo.lock

@@ -68,6 +68,10 @@
 				}
 			}
 		},
+		"vector_http": {
+			"host": "vector-client",
+			"port": 5022
+		},
 		"prometheus": {
 			"url": "http://prometheus:9090"
 		},

Cargo.toml

@@ -80,6 +80,10 @@
 				}
 			}
 		},
+		"vector_http": {
+			"host": "vector-client",
+			"port": 5022
+		},
 		"prometheus": {
 			"url": "http://prometheus:9090"
 		},

docker/dev-full/rivet-edge-server/config.jsonc

@@ -11,6 +11,11 @@ sources:
       - http://rivet-client:6090
     scrape_interval_secs: 15
 
+  dynamic_events_http:
+    type: http_server
+    address: 0.0.0.0:5022
+    encoding: ndjson
+  
   pegboard_manager:
     type: file
     include:
@@ -92,6 +97,7 @@ sinks:
     type: vector
     inputs:
       - metrics_add_meta
+      - dynamic_events_http
       - pegboard_manager_add_meta
       - pegboard_v8_isolate_runner_add_meta
       - pegboard_container_runner_add_meta

docker/dev-full/rivet-guard/config.jsonc

@@ -59,6 +59,35 @@ transforms:
     condition:
       type: vrl
       source: .source == "pegboard_container_runner"
+      
+  clickhouse_dynamic_events_filter:
+    type: filter
+    inputs:
+      - vector
+    condition:
+      type: vrl
+      source: .source == "clickhouse"
+  
+  clickhouse_dynamic_events_transform:
+    type: remap
+    inputs:
+      - clickhouse_dynamic_events_filter
+    source: |
+      # Extract and store metadata
+      __database = .database
+      __table = .table
+      __columns = .columns
+      
+      # Create a new object with just the columns data
+      . = {
+        "__database": __database,
+        "__table": __table,
+        # By default insert namespace column since most tables include this
+        "namespace": "rivet"
+      }
+      
+      # Merge in the column data that should be inserted
+      . = merge!(., __columns)
 
 sinks:
   prom_exporter:
@@ -82,7 +111,7 @@ sinks:
     compression: gzip
     database: db_pegboard_actor_log
     endpoint: http://clickhouse:8123
-    table: actor_logs
+    table: actor_logs2
     auth:
       strategy: basic
       user: vector
@@ -118,4 +147,19 @@ sinks:
     path: "/var/log/vector/pegboard_container_runner/%Y-%m-%d.log"
     encoding:
       codec: "text"
+      
+  clickhouse_dynamic_events:
+    type: clickhouse
+    inputs:
+      - clickhouse_dynamic_events_transform
+    compression: gzip
+    endpoint: http://clickhouse:8123
+    database: "{{ __database }}"
+    table: "{{ __table }}"
+    auth:
+      strategy: basic
+      user: vector
+      password: vector
+    batch:
+      timeout_secs: 1.0
 

docker/dev-full/vector-client/vector.yaml

@@ -82,7 +82,7 @@ sinks:
     compression: gzip
     endpoint: http://clickhouse:9300
     database: db_pegboard_actor_log
-    table: actor_logs
+    table: actor_logs2
     auth:
       strategy: basic
       user: vector

docker/dev-full/vector-server/vector.yaml

@@ -238,6 +238,11 @@ impl ActivityCtx {
 		self.conn.clickhouse().await
 	}
 
+	#[tracing::instrument(skip_all)]
+	pub async fn clickhouse_inserter(&self) -> GlobalResult<ClickHouseInserterHandle> {
+		self.conn.clickhouse_inserter().await
+	}
+
 	#[tracing::instrument(skip_all)]
 	pub async fn fdb(&self) -> Result<FdbPool, rivet_pools::Error> {
 		self.conn.fdb().await

docker/monolith/vector-server/vector.yaml

@@ -242,6 +242,11 @@ impl ApiCtx {
 		self.conn.clickhouse().await
 	}
 
+	#[tracing::instrument(skip_all)]
+	pub async fn clickhouse_inserter(&self) -> GlobalResult<ClickHouseInserterHandle> {
+		self.conn.clickhouse_inserter().await
+	}
+
 	#[tracing::instrument(skip_all)]
 	pub async fn fdb(&self) -> Result<FdbPool, rivet_pools::Error> {
 		self.conn.fdb().await

packages/common/chirp-workflow/core/src/ctx/activity.rs

@@ -266,6 +266,11 @@ impl OperationCtx {
 		self.conn.clickhouse().await
 	}
 
+	#[tracing::instrument(skip_all)]
+	pub async fn clickhouse_inserter(&self) -> GlobalResult<ClickHouseInserterHandle> {
+		self.conn.clickhouse_inserter().await
+	}
+
 	#[tracing::instrument(skip_all)]
 	pub async fn fdb(&self) -> Result<FdbPool, rivet_pools::Error> {
 		self.conn.fdb().await