feat: add vector http source (#800)

<!-- Please make sure there is an issue that this PR is correlated to. -->

## Changes

<!-- If there are frontend changes, please include screenshots. -->

infra/tf/k8s_infra/traefik_tunnel.tf

@@ -18,19 +18,19 @@ locals {
 				service = "nomad-server"
 				service_namespace = kubernetes_namespace.nomad.0.metadata[0].name
 				service_port = 4647
-			}
+			},
 
 			# Addresses specific Nomad servers.
 			"nomad-server-0" = {
 				service = "nomad-server-0"
 				service_namespace = kubernetes_namespace.nomad.0.metadata[0].name
 				service_port = 4647
-			}
+			},
 			"nomad-server-1" = {
 				service = "nomad-server-1"
 				service_namespace = kubernetes_namespace.nomad.0.metadata[0].name
 				service_port = 4647
-			}
+			},
 			"nomad-server-2" = {
 				service = "nomad-server-2"
 				service_namespace = kubernetes_namespace.nomad.0.metadata[0].name
@@ -42,7 +42,7 @@ locals {
 				service = "vector"
 				service_namespace = kubernetes_namespace.vector.0.metadata[0].name
 				service_port = 6000
-			}
+			},
 			"vector-tcp-json" = {
 				service = "vector"
 				service_namespace = kubernetes_namespace.vector.0.metadata[0].name
@@ -261,9 +261,9 @@ data "kubernetes_service" "traefik_tunnel" {
 	}
 }
 
-resource "kubectl_manifest" "traefik_nomad_router" {
+resource "kubectl_manifest" "traefik_router" {
 	depends_on = [helm_release.traefik_tunnel]
-	for_each = var.edge_enabled ? local.tunnel_services : {}
+	for_each = local.tunnel_services
 
 	yaml_body = yamlencode({
 		apiVersion = "traefik.io/v1alpha1"
@@ -299,7 +299,6 @@ resource "kubectl_manifest" "traefik_nomad_router" {
 					name = "ingress-tunnel",
 					namespace = "traefik-tunnel"
 				}
-
 			}
 		}
 	})

infra/tf/k8s_infra/vector.tf

@@ -5,3 +5,93 @@ resource "kubernetes_namespace" "vector" {
 		name = "vector"
 	}
 }
+
+module "vector_secrets" {
+	source = "../modules/secrets"
+
+	keys = [
+		"vector/http/username",
+		"vector/http/password",
+	]
+}
+
+resource "kubectl_manifest" "vector_ingress_route" {
+	for_each = var.prometheus_enabled ? local.entrypoints : {}
+
+	depends_on = [null_resource.daemons, kubectl_manifest.vector_basic_auth]
+
+	yaml_body = yamlencode({
+		apiVersion = "traefik.io/v1alpha1"
+		kind = "IngressRoute"
+
+		metadata = {
+			name = "vector-${each.key}"
+			namespace = kubernetes_namespace.vector.0.metadata.0.name
+			labels = {
+				"traefik-instance" = "main"
+			}
+		}
+
+		spec = {
+			entryPoints = [ each.key ]
+
+			routes = [
+				{
+					kind  = "Rule"
+					match = "Host(`vector.${var.domain_main}`)"
+					priority = 50
+					middlewares = [{
+						name = "vector-basic-auth"
+						namespace = kubernetes_namespace.vector.0.metadata.0.name
+					}]
+					services = [{
+						name = "vector"
+						port = 6200
+					}]
+				}
+			]
+
+			tls = lookup(each.value, "tls", null)
+		}
+	})
+}
+
+resource "kubernetes_secret" "vector_basic_auth_secret" {
+	count = var.prometheus_enabled ? 1 : 0
+	type = "kubernetes.io/basic-auth"
+
+	metadata {
+		name = "vector-route-basic-auth"
+		namespace = kubernetes_namespace.vector.0.metadata.0.name
+	}
+
+	data = {
+		username = module.vector_secrets.values["vector/http/username"]
+		password = module.vector_secrets.values["vector/http/password"]
+	}
+}
+
+# MARK: Middleware
+resource "kubectl_manifest" "vector_basic_auth" {
+	count = var.prometheus_enabled ? 1 : 0
+	depends_on = [helm_release.traefik]
+
+	yaml_body = yamlencode({
+		apiVersion = "traefik.io/v1alpha1"
+		kind = "Middleware"
+
+		metadata = {
+			name = "vector-basic-auth"
+			namespace = kubernetes_namespace.vector.0.metadata.0.name
+			labels = {
+				"traefik-instance" = "main"
+			}
+		}
+
+		spec = {
+			basicAuth = {
+				secret = kubernetes_secret.vector_basic_auth_secret.0.metadata.0.name
+			}
+		}
+	})
+}

infra/tf/vector/vector.tf

@@ -62,6 +62,12 @@ resource "helm_release" "vector" {
 					address = "0.0.0.0:6100"
 					decoding = { codec = "json" }
 				}
+
+				http_json = {
+					type = "http_server"
+					address = "0.0.0.0:6200"
+					decoding = { codec = "json" }
+				}
 
 				vector_metrics = {
 					type = "internal_metrics"
@@ -109,6 +115,27 @@ resource "helm_release" "vector" {
 					}
 				}
 
+				clickhouse_cf_logs = {
+					type = "clickhouse"
+					inputs = ["http_json"]
+					compression = "gzip"
+					database = "db_cf_log"
+					endpoint = "https://${var.clickhouse_host}:${var.clickhouse_port_https}"
+					table = "cf_tail_events"
+					auth = {
+						strategy = "basic"
+						user = "vector"
+						# Escape values for Vector
+						password = replace(module.secrets.values["clickhouse/users/vector/password"], "$", "$$")
+					}
+					tls = local.clickhouse_k8s ? {
+						ca_file = "/usr/local/share/ca-certificates/clickhouse-ca.crt"
+					} : {}
+					batch = {
+						timeout_secs = 5.0
+					}
+				}
+
 				console = {
 					type = "console"
 					inputs = ["vector_logs"]

lib/bolt/core/src/dep/terraform/gen.rs

@@ -347,6 +347,14 @@ async fn vars(ctx: &ProjectContext) {
 			}));
 		}
 
+		// Add vector
+		if config.prometheus.is_some() {
+			extra_dns.push(json!({
+				"zone_name": "main",
+				"name": format!("vector.{}", domain_main),
+			}));
+		}
+
 		vars.insert("extra_dns".into(), json!(extra_dns));
 	}
 

lib/bolt/core/src/dep/terraform/remote_states.rs

@@ -40,7 +40,7 @@ pub fn dependency_graph(_ctx: &ProjectContext) -> HashMap<&'static str, Vec<Remo
 			RemoteStateBuilder::default().plan_id("cockroachdb_managed").build().unwrap(),
 		],
 		"opengb" => vec![
-			RemoteStateBuilder::default().plan_id("dns").build().unwrap()
+			RemoteStateBuilder::default().plan_id("dns").build().unwrap(),
 		],
 	}
 }

lib/bolt/core/src/tasks/config/generate.rs

@@ -233,6 +233,18 @@ pub async fn generate(project_path: &Path, ns_id: &str) -> Result<()> {
 			.await?;
 	}
 
+	// MARK: Vector
+	generator
+		.generate_secret(&["vector", "http", "username"], || async {
+			Ok(value("rivet"))
+		})
+		.await?;
+	generator
+		.generate_secret(&["vector", "http", "password"], || async {
+			Ok(value(generate_password(64)))
+		})
+		.await?;
+
 	// MARK: Minio
 	if generator.ns["s3"].get("minio").is_some() {
 		let root_pass = generate_password(32);

svc/pkg/cf/db/log/Service.toml

@@ -0,0 +1,7 @@
+[service]
+name = "db-cf-log"
+
+[runtime]
+kind = "clickhouse"
+
+[database]

svc/pkg/cf/db/log/migrations/20240517214700_init.up.sql

@@ -0,0 +1,15 @@
+SET allow_experimental_object_type = 1;
+
+CREATE TABLE IF NOT EXISTS cf_tail_events
+(
+	script_name String,
+	ts DateTime64(3),  -- This is tail_event.eventTimestamp, DateTime64(3) is milliseconds
+	ray_id UUID,  -- Extracted by tail worker
+	tail_event JSON,  -- TODO: might need to do the trickery used in analytics event
+	INDEX idx_ray_id (ray_id) TYPE bloom_filter() GRANULARITY 4
+)
+ENGINE = ReplicatedMergeTree()
+PARTITION BY toStartOfHour(ts)
+ORDER BY (script_name, ts, ray_id)
+TTL toDate(ts + toIntervalDay(3))
+SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1;