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

Author: NathanFlurry

Cargo.lock

@@ -11,6 +11,12 @@ sources:
       - http://rivet-client:6090
     scrape_interval_secs: 15
 
+  http_events:
+    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
+      - http_events
       - pegboard_manager_add_meta
       - pegboard_v8_isolate_runner_add_meta
       - pegboard_container_runner_add_meta

Cargo.toml

@@ -19,6 +19,14 @@ sources:
     decoding:
       codec: json
 
+  http_json:
+    type: http_server
+    address: 0.0.0.0:6200
+    decoding:
+      codec: json
+    path: ""
+    strict_path: false
+  
   vector_metrics:
     type: internal_metrics
 
@@ -59,6 +67,32 @@ transforms:
     condition:
       type: vrl
       source: .source == "pegboard_container_runner"
+      
+  clickhouse_events:
+    type: filter
+    inputs:
+      - http_json
+    condition:
+      type: vrl
+      source: .source == "clickhouse"
+  
+  clickhouse_events_transform:
+    type: remap
+    inputs:
+      - clickhouse_events
+    source: |
+      # Extract and store metadata
+      __database = .database ?? "custom_events"
+      __table = .table
+      
+      # 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 +152,19 @@ sinks:
     path: "/var/log/vector/pegboard_container_runner/%Y-%m-%d.log"
     encoding:
       codec: "text"
+      
+  clickhouse_dynamic_events:
+    type: clickhouse
+    inputs:
+      - clickhouse_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

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

@@ -296,6 +296,11 @@ impl TestCtx {
 		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, fields(%workflow_id))]
 	pub async fn sqlite_for_workflow(&self, workflow_id: Uuid) -> GlobalResult<SqlitePool> {
 		common::sqlite_for_workflow(&self.db, &self.conn, workflow_id, true)

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

@@ -84,4 +84,8 @@ impl TestCtx {
 	pub async fn clickhouse(&self) -> GlobalResult<ClickHousePool> {
 		self.op_ctx.clickhouse().await
 	}
+
+	pub async fn clickhouse_inserter(&self) -> GlobalResult<ClickHouseInserterHandle> {
+		self.op_ctx.clickhouse_inserter().await
+	}
 }