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

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

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

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

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

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

@@ -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/vector-server/vector.yaml

@@ -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/monolith/vector-server/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

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

@@ -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

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

@@ -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/operation.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/standalone.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/test.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/worker/src/test.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
+	}
 }

packages/common/clickhouse-inserter/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "clickhouse-inserter"
+version.workspace = true
+authors.workspace = true
+license.workspace = true
+edition.workspace = true
+
+[dependencies]
+global-error.workspace = true
+tokio.workspace = true
+reqwest = { version = "0.11", features = ["json"] }
+tracing.workspace = true
+serde = { version = "1.0", features = ["derive"] }
+serde_json = { version = "1.0", features = ["raw_value"] }
+thiserror = "1.0"
+tokio-util = "0.7"
+futures = "0.3"
+async-channel = "2.1.1"
+anyhow.workspace = true

packages/common/clickhouse-inserter/src/error.rs

@@ -0,0 +1,16 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("failed to send event to ClickHouse inserter")]
+    ChannelSendError,
+
+    #[error("serialization error: {0}")]
+    SerializationError(#[source] serde_json::Error),
+
+    #[error("failed to build reqwest client: {0}")]
+    ReqwestBuildError(#[source] reqwest::Error),
+
+    #[error("failed to spawn background task")]
+    TaskSpawnError,
+}