feat: get multi actors working e2e on docker compose

docker/dev-full/docker-compose.yml

@@ -357,6 +357,8 @@ services:
     volumes:
       - vector-server-data:/var/lib/vector
       - ./vector-server:/etc/vector
+    # environment:
+      # - VECTOR_LOG=debug
     networks:
       - rivet-network
 

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

@@ -21,15 +21,10 @@ sources:
     include:
       - /var/lib/rivet-client/log
 
-  pegboard_v8_isolate_runner:
-    type: file
-    include:
-      - /var/lib/rivet-client/runner/log
-
   pegboard_container_runners:
     type: file
     include:
-      - /var/lib/rivet-client/actors/*/log
+      - /var/lib/rivet-client/runners/*/log
 
 transforms:
   filter_metrics:
@@ -46,7 +41,7 @@ transforms:
       .tags.server_id = "fc67e54e-5d6a-4726-ab23-77b0e54f068f"
       .tags.datacenter_id = "f288913c-735d-4188-bf9b-2fcf6eac7b9c"
       .tags.cluster_id = "unknown"
-      .tags.pool_type = "pegboard_isolate"
+      .tags.pool_type = "pegboard"
       .tags.public_ip = "127.0.0.1"
 
   pegboard_manager_add_meta:
@@ -60,36 +55,22 @@ transforms:
       .server_id = "fc67e54e-5d6a-4726-ab23-77b0e54f068f"
       .datacenter_id = "f288913c-735d-4188-bf9b-2fcf6eac7b9c"
       .tags.cluster_id = "unknown"
-      .pool_type = "pegboard_isolate"
-      .public_ip = "127.0.0.1"
-
-  pegboard_v8_isolate_runner_add_meta:
-    type: remap
-    inputs:
-      - pegboard_v8_isolate_runner
-    source: |
-      .source = "pegboard_v8_isolate_runner"
-
-      .client_id = "fc67e54e-5d6a-4726-ab23-77b0e54f068f"
-      .server_id = "fc67e54e-5d6a-4726-ab23-77b0e54f068f"
-      .datacenter_id = "f288913c-735d-4188-bf9b-2fcf6eac7b9c"
-      .tags.cluster_id = "unknown"
-      .pool_type = "pegboard_isolate"
+      .pool_type = "pegboard"
       .public_ip = "127.0.0.1"
-
+    
   pegboard_container_runner_add_meta:
     type: remap
     inputs:
       - pegboard_container_runners
     source: |
       .source = "pegboard_container_runner"
-      .actor_id = parse_regex!(.file, r'/etc/pegboard/actors/(?P<actor_id>[0-9a-fA-F-]+)/log').actor_id
+      .runner_id = parse_regex!(.file, r'/etc/pegboard/runners/(?P<runner_id>[0-9a-fA-F-]+)/log').runner_id
 
       .client_id = "fc67e54e-5d6a-4726-ab23-77b0e54f068f"
       .server_id = "fc67e54e-5d6a-4726-ab23-77b0e54f068f"
       .datacenter_id = "f288913c-735d-4188-bf9b-2fcf6eac7b9c"
       .cluster_id = "unknown"
-      .pool_type = "pegboard_isolate"
+      .pool_type = "pegboard"
       .public_ip = "127.0.0.1"
 
 sinks:
@@ -99,7 +80,6 @@ sinks:
       - metrics_add_meta
       - dynamic_events_http
       - pegboard_manager_add_meta
-      - pegboard_v8_isolate_runner_add_meta
       - pegboard_container_runner_add_meta
     address: vector-server:6000
     healthcheck:

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

@@ -34,6 +34,15 @@ transforms:
     condition:
       type: vrl
       source: .source == "actors"
+
+  runners:
+    type: filter
+    inputs:
+      - vector
+      - tcp_json
+    condition:
+      type: vrl
+      source: .source == "runners"
 
   job_run:
     type: filter
@@ -126,6 +135,21 @@ sinks:
       password: vector
     batch:
       timeout_secs: 1.0
+
+  clickhouse_runner_logs:
+    type: clickhouse
+    inputs:
+      - runners
+    compression: gzip
+    database: db_pegboard_runner_log
+    endpoint: http://clickhouse:8123
+    table: runner_logs
+    auth:
+      strategy: basic
+      user: vector
+      password: vector
+    batch:
+      timeout_secs: 1.0
 
   clickhouse_job_run_logs:
     type: clickhouse

docker/http-debug/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install required packages
+RUN pip install flask
+
+# Create the debug server
+COPY debug_server.py .
+
+# Expose port 8080
+EXPOSE 8080
+
+# Run the server
+CMD ["python", "debug_server.py"]

docker/http-debug/debug_server.py

@@ -0,0 +1,122 @@
+from flask import Flask, request, jsonify
+import gzip
+import zlib
+from datetime import datetime
+
+app = Flask(__name__)
+
+def log_request_details():
+    """Log all request details in a formatted way"""
+    timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+    print("=" * 80)
+    print(f"[{timestamp}] NEW REQUEST")
+    print("=" * 80)
+
+    # Method and URL
+    print(f"Method: {request.method}")
+    print(f"URL: {request.url}")
+    print(f"Path: {request.path}")
+    print(f"Query String: {request.query_string.decode('utf-8')}")
+
+    # Headers
+    print("\n--- HEADERS ---")
+    for header_name, header_value in request.headers:
+        print(f"{header_name}: {header_value}")
+
+    # Query Parameters
+    if request.args:
+        print("\n--- QUERY PARAMETERS ---")
+        for key, value in request.args.items():
+            print(f"{key}: {value}")
+
+    # Form Data
+    if request.form:
+        print("\n--- FORM DATA ---")
+        for key, value in request.form.items():
+            print(f"{key}: {value}")
+
+    # Files
+    if request.files:
+        print("\n--- FILES ---")
+        for key, file in request.files.items():
+            print(f"{key}: {file.filename} (Content-Type: {file.content_type})")
+
+    # Raw Body with decompression support
+    try:
+        raw_body = request.get_data()
+        content_encoding = request.headers.get('Content-Encoding', '').lower()
+
+        print("\n--- REQUEST BODY ---")
+
+        if not raw_body:
+            print("(empty)")
+        else:
+            # Handle compressed content
+            decompressed_body = None
+
+            if content_encoding == 'gzip':
+                try:
+                    decompressed_body = gzip.decompress(raw_body).decode('utf-8')
+                    print("(Content was gzip-compressed, showing decompressed version)")
+                except Exception as e:
+                    print(f"Failed to decompress gzip content: {e}")
+            elif content_encoding == 'deflate':
+                try:
+                    decompressed_body = zlib.decompress(raw_body).decode('utf-8')
+                    print("(Content was deflate-compressed, showing decompressed version)")
+                except Exception as e:
+                    print(f"Failed to decompress deflate content: {e}")
+            elif content_encoding in ['br', 'brotli']:
+                print("(Content is brotli-compressed - brotli decompression not available)")
+                print("Raw compressed data (first 200 bytes):")
+                print(repr(raw_body[:200]))
+            elif content_encoding == 'zstd':
+                print("(Content is zstd-compressed - zstd decompression not available)")
+                print("Raw compressed data (first 200 bytes):")
+                print(repr(raw_body[:200]))
+            else:
+                # No compression or unknown compression
+                try:
+                    decompressed_body = raw_body.decode('utf-8')
+                except UnicodeDecodeError:
+                    print("(Binary content - showing first 200 bytes as hex)")
+                    print(raw_body[:200].hex())
+
+            # Display the decompressed content
+            if decompressed_body:
+                print(decompressed_body)
+
+                # Also show raw length info
+                print(f"\nRaw body length: {len(raw_body)} bytes")
+                print(f"Decompressed length: {len(decompressed_body)} bytes")
+
+    except Exception as e:
+        print(f"\n--- REQUEST BODY ---")
+        print(f"Error reading body: {e}")
+
+    print("=" * 80)
+    print()
+
+# Catch all routes for any HTTP method
+@app.route('/', defaults={'path': ''}, methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS'])
+@app.route('/<path:path>', methods=['GET', 'POST', 'PUT', 'DELETE', 'PATCH', 'HEAD', 'OPTIONS'])
+def debug_endpoint(path):
+    log_request_details()
+
+    # Return a simple response
+    response_data = {
+        "message": "Request received and logged",
+        "method": request.method,
+        "path": f"/{path}" if path else "/",
+        "timestamp": datetime.now().isoformat()
+    }
+
+    return jsonify(response_data), 200
+
+if __name__ == '__main__':
+    print("Starting HTTP Debug Server...")
+    print("All incoming requests will be logged to console")
+    print("Server listening on port 8080")
+    print("=" * 80)
+    app.run(host='0.0.0.0', port=8080, debug=False)