fix(pegboard): convert ws manager socket to unix socket

Author: MasterPtato

Cargo.toml

@@ -14,4 +14,4 @@ scrape_configs:
   - job_name: rivet-client
     static_configs:
       - targets:
-        - rivet-client:6090
+        - rivet-client:8091

docker/dev-full/prometheus/prometheus.yml

@@ -8,7 +8,7 @@ sources:
   prometheus_pegboard:
     type: prometheus_scrape
     endpoints:
-      - http://rivet-client:6090
+      - http://rivet-client:8091
     scrape_interval_secs: 15
 
   dynamic_events_http:

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

@@ -103,7 +103,7 @@ const services: Service[] = [
 		rootUser: true,
 		ports: {
 			runner: 6080,
-			metrics: 6090,
+			metrics: 8091,
 		},
 	},
 

docker/monolith/build-scripts/setup_s6.ts

@@ -12,7 +12,7 @@ sources:
   prometheus_pegboard:
     type: prometheus_scrape
     endpoints:
-      - http://rivet-client:6090
+      - http://rivet-client:8091
     scrape_interval_secs: 15
 
   pegboard_manager:

docker/monolith/vector-client/vector.yaml

@@ -83,6 +83,14 @@ impl Worker {
 		loop {
 			tokio::select! {
 				_ = tick_interval.tick() => {},
+				res = wake_sub.next() => {
+					if res.is_none() {
+						return Err(WorkflowError::SubscriptionUnsubscribed.into());
+					}
+
+					tick_interval.reset();
+				},
+
 				res = &mut gc_handle => {
 					tracing::error!(?res, "metrics task unexpectedly stopped");
 					break;
@@ -91,13 +99,6 @@ impl Worker {
 					tracing::error!(?res, "metrics task unexpectedly stopped");
 					break;
 				},
-				res = wake_sub.next() => {
-					if res.is_none() {
-						return Err(WorkflowError::SubscriptionUnsubscribed.into());
-					}
-
-					tick_interval.reset();
-				},
 				_ = ctrl_c() => break,
 				_ = sigterm.recv() => break,
 			}

packages/common/chirp-workflow/core/src/worker.rs

@@ -1,11 +1,12 @@
+use std::net::SocketAddr;
+
 use global_error::prelude::*;
 use hyper::{
 	header::CONTENT_TYPE,
 	service::{make_service_fn, service_fn},
 	Body, Request, Response, Server,
 };
 use prometheus::{Encoder, TextEncoder};
-use std::net::SocketAddr;
 
 // TODO: Record extra labels
 

packages/common/metrics/src/server.rs

@@ -36,6 +36,8 @@ pub enum Id {
 	V1([u8; 18]),
 }
 
+
+
 impl Id {
 	/// Construct V0 from uuid.
 	pub fn new_v0() -> Self {
@@ -313,6 +315,12 @@ impl sqlx::postgres::PgHasArrayType for Id {
 	}
 }
 
+impl Default for Id {
+	fn default() -> Self {
+		Id::V0(Uuid::new_v4())
+	}
+}
+
 /// Decode a base36 string into a fixed-size byte array.
 fn decode_base36<const N: usize>(s: &str) -> Result<[u8; N], IdError> {
 	let mut data = [0u8; N];

packages/common/util/core/src/id.rs

@@ -81,7 +81,7 @@ pub fn configure(namespace: &str, config: &Config, pool_type: PoolType) -> Globa
 		}
 	});
 
-	// Add pegboard manager and runner logs
+	// Add pegboard manager and actor logs
 	match pool_type {
 		PoolType::Pegboard | PoolType::PegboardIsolate => {
 			config_json["sources"]["pegboard_manager"] = json!({

packages/core/services/cluster/src/workflows/server/install/install_scripts/components/vector.rs

@@ -179,7 +179,7 @@ pub async fn gen_initialize(
 				"pegboard".into(),
 				components::vector::PrometheusTarget {
 					// Should match port from pb manager config
-					endpoint: "http://127.0.0.1:6090".into(),
+					endpoint: "http://127.0.0.1:8091".into(),
 					scrape_interval: 15,
 				},
 			);
@@ -194,7 +194,7 @@ pub async fn gen_initialize(
 				"pegboard".into(),
 				components::vector::PrometheusTarget {
 					// Should match port from pb manager config
-					endpoint: "http://127.0.0.1:6090".into(),
+					endpoint: "http://127.0.0.1:8091".into(),
 					scrape_interval: 15,
 				},
 			);