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

Author: NathanFlurry

Cargo.lock

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

Cargo.toml

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

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

@@ -11,6 +11,12 @@ sources:
       - http://rivet-client:6090
     scrape_interval_secs: 15
 
+  dynamic_events_http:
+    type: http_server
+    address: 0.0.0.0:5022
+    decoding:
+      codec: json
+  
   pegboard_manager:
     type: file
     include:
@@ -92,6 +98,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-server/config.jsonc

@@ -59,6 +59,33 @@ 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
+      }
+      
+      # Merge in the column data that should be inserted
+      . = merge!(., __columns)
 
 sinks:
   prom_exporter:
@@ -118,4 +145,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

@@ -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/dev-full/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

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

@@ -253,6 +253,11 @@ impl StandaloneCtx {
 		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