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

NathanFlurry · 2025-06-03T06:32:35Z

Changes

cloudflare-workers-and-pages · 2025-06-03T06:32:37Z

Deploying rivet-studio with Cloudflare Pages

Latest commit:	`526f37e`
Status:	✅ Deploy successful!
Preview URL:	https://124bf5a3.rivet-studio.pages.dev
Branch Preview URL:	https://06-02-chore-vector-add-vecto.rivet-studio.pages.dev

View logs

NathanFlurry · 2025-06-03T06:32:51Z

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

How to use the Graphite Merge Queue

Add the label merge-queue to this PR to add it to the merge queue.

You must have a Graphite account in order to use the merge queue. Sign up using this link.

_{An organization admin has enabled the Graphite Merge Queue in this repository.} _{Please do not merge from GitHub as this will restart CI on PRs being processed by the merge queue.}

This stack of pull requests is managed by Graphite. Learn more about stacking.

cloudflare-workers-and-pages · 2025-06-03T06:36:12Z

Deploying rivet with Cloudflare Pages

Latest commit:	`526f37e`
Status:	✅ Deploy successful!
Preview URL:	https://4f91f438.rivet.pages.dev
Branch Preview URL:	https://06-02-chore-vector-add-vecto.rivet.pages.dev

View logs

cloudflare-workers-and-pages · 2025-06-03T06:36:14Z

Deploying rivet-hub with Cloudflare Pages

Latest commit:	`526f37e`
Status:	✅ Deploy successful!
Preview URL:	https://04f65c11.rivet-hub-7jb.pages.dev
Branch Preview URL:	https://06-02-chore-vector-add-vecto.rivet-hub-7jb.pages.dev

View logs

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

MasterPtato · 2025-06-04T19:25:42Z

packages/common/pools/src/pools.rs

@@ -40,12 +42,24 @@ impl Pools {
 		let clickhouse = crate::db::clickhouse::setup(config.clone())?;
 		let sqlite = SqlitePoolManager::new(fdb.clone()).await?;

+		// Create the ClickHouse inserter if ClickHouse is enabled


Suggested change

// Create the ClickHouse inserter if ClickHouse is enabled

// Create the ClickHouse inserter if vector is enabled

MasterPtato · 2025-06-04T19:29:26Z

packages/edge/infra/guard/core/src/proxy_service.rs

 	) -> GlobalResult<Response<Full<Bytes>>> {
+		let _request_start = Instant::now();


MasterPtato · 2025-06-04T19:34:13Z

packages/edge/infra/guard/core/src/proxy_service.rs

+		let request_id = Uuid::new_v4();
+		let mut request_context =
+			RequestContext::new_with_request_id(request_id, self.state.clickhouse_inserter.clone());


Uses new_with_request_id with a new uuid, same as new

Suggested change

let request_id = Uuid::new_v4();

let mut request_context =

RequestContext::new_with_request_id(request_id, self.state.clickhouse_inserter.clone());

let mut request_context =

RequestContext::new(self.state.clickhouse_inserter.clone());

MasterPtato · 2025-06-04T19:42:40Z

packages/edge/services/pegboard/src/workflows/actor/analytics.rs

+	network_ports_ingress: HashMap<String, ActorClickHouseRowPortIngress>,
+	network_ports_host: HashMap<String, ActorClickHouseRowPortHost>,
+	network_ports_proxied: HashMap<String, ActorClickHouseRowPortProxied>,
+	client_id: Uuid,


Should we add runner_id now or later?

packages/edge/services/pegboard/src/workflows/actor/analytics.rs

MasterPtato · 2025-06-04T19:49:19Z

packages/edge/services/pegboard/src/workflows/actor/destroy.rs

+	if let Some(actor_input) = &input.actor_input {
+		// Get metadata
+		let meta = ctx
+			.activity(GetMetaInput {


Shouldn't this also be .v(2)?

MasterPtato · 2025-06-04T19:54:56Z

packages/edge/services/pegboard/src/workflows/actor/mod.rs

@@ -261,6 +267,12 @@ pub async fn pegboard_actor(ctx: &mut WorkflowCtx, input: &Input) -> GlobalResul
 										})
 										.await?;

+									ctx.v(2).activity(InsertClickHouseInput {
+										input: input.clone(),
+										meta,


This metadata includes build data which changes if the actor's build is changed. the metadata is re-fetched on reschedule in reschedule_actor as actor_setup.meta

greptile-apps

PR Summary

This PR adds a Vector pipeline to ship ClickHouse events from edge services, with significant changes across the codebase. Here are the key points:

Introduces new clickhouse-inserter package for batched event insertion with configurable batch sizes and intervals
Adds http_requests table in ClickHouse for guard analytics with 30-day TTL and comprehensive request metadata
Creates new actor_logs2 table with 14-day TTL and materialized view actor_logs2_with_metadata for enhanced actor logging
Implements request context tracking in guard service with detailed metrics collection and WebSocket support
Adds non-interactive route management capabilities to CLI with auto-create and auto-sync options

_{58 file(s) reviewed, 14 comment(s)}
_{Edit PR Review Bot Settings | Greptile}

greptile-apps · 2025-06-05T02:12:32Z

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

+  dynamic_events_http:
+    type: http_server
+    address: 0.0.0.0:5022
+    encoding: ndjson


logic: No transform is defined for dynamic_events_http before sending to sink. Consider adding a transform to ensure consistent metadata tagging like other sources.

greptile-apps · 2025-06-05T02:15:54Z

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

+					if events.len() >= BATCH_SIZE {
+						if let Err(e) = self.send_events(&events).await {
+							tracing::error!(?e, "failed to send events to Vector");
+						}
+						events.clear();
+					}


logic: Events are cleared even if send_events fails. This could result in data loss. Consider only clearing after successful send or implementing a retry mechanism.

Suggested change

if events.len() >= BATCH_SIZE {

if let Err(e) = self.send_events(&events).await {

tracing::error!(?e, "failed to send events to Vector");

}

events.clear();

}

if events.len() >= BATCH_SIZE {

if let Err(e) = self.send_events(&events).await {

tracing::error!(?e, "failed to send events to Vector");

} else {

events.clear();

}

}

greptile-apps · 2025-06-05T02:19:58Z

packages/edge/infra/guard/core/src/request_context.rs

+use global_error::GlobalResult;
+use lazy_static::lazy_static;
+use std::{net::IpAddr, time::SystemTime};
+use tracing::warn;


style: tracing::warn is imported but never used in the code

Suggested change

use tracing::warn;

use global_error::GlobalResult;

use lazy_static::lazy_static;

use std::{net::IpAddr, time::SystemTime};

use uuid::Uuid;

greptile-apps · 2025-06-05T02:20:28Z

packages/edge/infra/guard/core/src/proxy_service.rs

+				let request_send_start = Instant::now();
 				match timeout(timeout_duration, self.client.request(proxied_req)).await {


logic: request_send_start is captured but response_receive_time is calculated and never used. Consider either removing the unused timing or using it for metrics/logging.

greptile-apps · 2025-06-05T02:21:04Z

packages/edge/services/pegboard/db/actor-log/migrations/20250604022411_actor_log2.up.sql

+    ts DateTime64 (9),
+    stream_type UInt8, -- pegboard::types::LogsStreamType
+    message String
+) ENGINE = ReplicatedMergeTree ()


logic: ReplicatedMergeTree() missing required parameters for zookeeper path and replica name

greptile-apps · 2025-06-05T02:23:47Z

packages/edge/services/pegboard/src/workflows/actor/migrations.rs

+        ALTER TABLE state ADD project_id BLOB DEFAULT X'00000000000000000000000000000000';  -- UUID
+        ALTER TABLE state ADD root_user_enabled INT DEFAULT false;
+        ALTER TABLE state ADD build_kind INT DEFAULT -1;
+        ALTER TABLE state ADD build_compression INT DEFAULT -1;


logic: The default values for build_kind and build_compression are set to -1, which may be invalid enum values. Consider using 0 or NULL if these are meant to represent unset states.

greptile-apps · 2025-06-05T02:23:52Z

packages/edge/services/pegboard/src/workflows/actor/analytics.rs

+	let state_row = sql_fetch_one!(
+		[ctx, StateRow, &pool]
+		"
+		SELECT 
+			project_id,
+			env_id,
+			json(tags) AS tags,
+			resources_cpu_millicores,
+			resources_memory_mib,
+			selected_resources_cpu_millicores,
+			selected_resources_memory_mib,
+			client_id,
+			client_workflow_id,
+			client_wan_hostname,
+			lifecycle_kill_timeout_ms,
+			lifecycle_durable,
+			create_ts,
+			start_ts,
+			connectable_ts,
+			finish_ts,
+			destroy_ts,
+			image_id,
+			build_kind,
+			build_compression,
+			root_user_enabled,
+			json(args) AS args,
+			network_mode,
+			json(environment) AS environment
+		FROM state
+		",
+	)
+	.await?;


logic: SQL query lacks a WHERE clause to filter by actor_id, which could return incorrect data if multiple actors exist.

Suggested change

let state_row = sql_fetch_one!(

[ctx, StateRow, &pool]

"

SELECT

project_id,

env_id,

json(tags) AS tags,

resources_cpu_millicores,

resources_memory_mib,

selected_resources_cpu_millicores,

selected_resources_memory_mib,

client_id,

client_workflow_id,

client_wan_hostname,

lifecycle_kill_timeout_ms,

lifecycle_durable,

create_ts,

start_ts,

connectable_ts,

finish_ts,

destroy_ts,

image_id,

build_kind,

build_compression,

root_user_enabled,

json(args) AS args,

network_mode,

json(environment) AS environment

FROM state

",

)

.await?;

let state_row = sql_fetch_one!(

[ctx, StateRow, &pool]

"

SELECT

project_id,

env_id,

json(tags) AS tags,

resources_cpu_millicores,

resources_memory_mib,

selected_resources_cpu_millicores,

selected_resources_memory_mib,

client_id,

client_workflow_id,

client_wan_hostname,

lifecycle_kill_timeout_ms,

lifecycle_durable,

create_ts,

start_ts,

connectable_ts,

finish_ts,

destroy_ts,

image_id,

build_kind,

build_compression,

root_user_enabled,

json(args) AS args,

network_mode,

json(environment) AS environment

FROM state

WHERE actor_id = ?

",

input.actor_id

)

.await?;

greptile-apps · 2025-06-05T02:24:09Z

packages/edge/services/pegboard/src/workflows/actor/setup.rs

+	sql_execute!(
+		[ctx, pool]
+		"
+		UPDATE state
+		SET
+			project_id = ?,
+			build_kind = ?,
+			build_compression = ?,
+			root_user_enabled = ?
+		",
+		input.meta.project_id,
+		input.meta.build_kind as i64,
+		input.meta.build_compression as i64,
+		input.root_user_enabled,
+	)
+	.await?;


logic: SQL update lacks a WHERE clause - could update all state records unintentionally

greptile-apps · 2025-06-05T02:25:00Z

packages/toolchain/cli/src/commands/route/endpoint.rs

+		let result = apis::routes_api::routes_update(
+			&ctx.openapi_config_cloud,
+			&self.name,
+			update_route_body.clone(),


style: Unnecessary clone() of update_route_body since it's not used after the API call

Suggested change

update_route_body.clone(),

update_route_body,

greptile-apps · 2025-06-05T02:25:01Z

packages/toolchain/cli/src/commands/route/endpoint.rs

+	let matching_route = routes_response
+		.routes
+		.iter()
+		.find(|route| route.id == *route_id)


logic: Route lookup uses ID equality but name field may not be an ID. Consider documenting the expected format or validating input.

… edge

graphite-app · 2025-06-09T16:47:03Z

Merge activity

Jun 9, 4:47 PM UTC: MasterPtato added this pull request to the Graphite merge queue.
Jun 9, 4:48 PM UTC: CI is running for this pull request on a draft pull request (#2574) due to your merge queue CI optimization settings.
Jun 9, 4:49 PM UTC: Merged by the Graphite merge queue via draft PR: #2574.

… edge (#2526)  ## Changes

NathanFlurry mentioned this pull request Jun 3, 2025

chore(examples): multitenant deploys example #2527

Closed

NathanFlurry force-pushed the 06-02-chore_vector_add_vector_pipeline_to_ship_clickhouse_events_from_the_edge branch from b9cc460 to 8d1f2c3 Compare June 3, 2025 06:34

NathanFlurry force-pushed the 06-02-chore_vector_add_vector_pipeline_to_ship_clickhouse_events_from_the_edge branch 2 times, most recently from 71023be to 58a5eca Compare June 4, 2025 01:42

NathanFlurry requested a review from MasterPtato June 4, 2025 01:47

MasterPtato requested changes Jun 4, 2025

View reviewed changes

NathanFlurry force-pushed the 06-02-chore_vector_add_vector_pipeline_to_ship_clickhouse_events_from_the_edge branch 2 times, most recently from 43b5230 to 64194de Compare June 4, 2025 22:01

MasterPtato changed the base branch from main to graphite-base/2526 June 4, 2025 23:46

MasterPtato force-pushed the 06-02-chore_vector_add_vector_pipeline_to_ship_clickhouse_events_from_the_edge branch from 64194de to e120ace Compare June 4, 2025 23:46

MasterPtato changed the base branch from graphite-base/2526 to 06-04-increase-actor-log-ttl_ June 4, 2025 23:46

MasterPtato marked this pull request as ready for review June 5, 2025 02:11

This was referenced Jun 5, 2025

chore: fix dc wf #2552

Closed

fix: add vector http config to edge configs #2553

Closed

greptile-apps bot reviewed Jun 5, 2025

View reviewed changes

MasterPtato force-pushed the 06-02-chore_vector_add_vector_pipeline_to_ship_clickhouse_events_from_the_edge branch from b43c653 to a32e202 Compare June 6, 2025 01:36

MasterPtato force-pushed the 06-04-increase-actor-log-ttl_ branch from 9bc9e25 to 21b6b12 Compare June 6, 2025 01:36

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

526f37e

… edge

MasterPtato force-pushed the 06-04-increase-actor-log-ttl_ branch from 21b6b12 to ec7ede8 Compare June 6, 2025 18:29

MasterPtato force-pushed the 06-02-chore_vector_add_vector_pipeline_to_ship_clickhouse_events_from_the_edge branch from a32e202 to 526f37e Compare June 6, 2025 18:29

This was referenced Jun 6, 2025

fix(pegboard): fix allocation metrics #2568

Closed

fix: dont include installing servers in service discovery #2569

Closed

feat: db sh for workflows #2570

Closed

MasterPtato approved these changes Jun 9, 2025

View reviewed changes

graphite-app bot closed this Jun 9, 2025

graphite-app bot deleted the 06-02-chore_vector_add_vector_pipeline_to_ship_clickhouse_events_from_the_edge branch June 9, 2025 16:49

github-actions bot mentioned this pull request Jun 9, 2025

chore(main): release 25.5.1 #2423

Merged

MasterPtato mentioned this pull request Jun 9, 2025

fix(pegboard): fix container runner orphaning #2575

Closed

	// Create the ClickHouse inserter if ClickHouse is enabled
	// Create the ClickHouse inserter if vector is enabled

		) -> GlobalResult<Response<Full<Bytes>>> {
		let _request_start = Instant::now();

		let request_send_start = Instant::now();
		match timeout(timeout_duration, self.client.request(proxied_req)).await {

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

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

Uh oh!

Conversation

NathanFlurry commented Jun 3, 2025

Changes

Uh oh!

cloudflare-workers-and-pages bot commented Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Deploying rivet-studio with Cloudflare Pages

Uh oh!

NathanFlurry commented Jun 3, 2025 • edited by MasterPtato Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

How to use the Graphite Merge Queue

Uh oh!

cloudflare-workers-and-pages bot commented Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Deploying rivet with Cloudflare Pages

Uh oh!

cloudflare-workers-and-pages bot commented Jun 3, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Deploying rivet-hub with Cloudflare Pages

Uh oh!

Uh oh!

MasterPtato Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

MasterPtato Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

MasterPtato Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

MasterPtato Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

NathanFlurry Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

MasterPtato Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

MasterPtato Jun 4, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot left a comment

Choose a reason for hiding this comment

PR Summary

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

greptile-apps bot Jun 5, 2025

Choose a reason for hiding this comment

Uh oh!

cloudflare-workers-and-pages bot commented Jun 3, 2025 •

edited

Loading

NathanFlurry commented Jun 3, 2025 •

edited by MasterPtato

Loading

cloudflare-workers-and-pages bot commented Jun 3, 2025 •

edited

Loading

cloudflare-workers-and-pages bot commented Jun 3, 2025 •

edited

Loading

graphite-app bot commented Jun 9, 2025 •

edited

Loading