Optimize ingest indexing

docker-compose.signoz.yml

@@ -0,0 +1,120 @@
+# SigNoz Observability Stack
+# Usage: docker compose -f docker-compose.yml -f docker-compose.override.yml -f docker-compose.signoz.yml up -d
+#
+# After first run, execute schema migrations:
+#   docker run --rm --network conserver signoz/signoz-schema-migrator:latest sync --dsn='tcp://signoz-clickhouse:9000'
+#
+# Access UI at: http://localhost:3301
+
+networks:
+  conserver:
+    external: true
+
+volumes:
+  signoz_clickhouse_data:
+  signoz_zookeeper_data:
+  signoz_zookeeper_log:
+  signoz_data:
+
+services:
+  signoz-zookeeper:
+    image: zookeeper:3.9
+    container_name: signoz-zookeeper
+    hostname: signoz-zookeeper
+    environment:
+      - ZOO_AUTOPURGE_PURGEINTERVAL=1
+      - ZOO_4LW_COMMANDS_WHITELIST=mntr,ruok,stat
+    volumes:
+      - signoz_zookeeper_data:/data
+      - signoz_zookeeper_log:/datalog
+    networks:
+      - conserver
+    healthcheck:
+      test: ["CMD-SHELL", "echo ruok | nc localhost 2181 | grep imok"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+    restart: unless-stopped
+
+  signoz-clickhouse:
+    image: clickhouse/clickhouse-server:24.1.2-alpine
+    container_name: signoz-clickhouse
+    hostname: signoz-clickhouse
+    tty: true
+    depends_on:
+      signoz-zookeeper:
+        condition: service_healthy
+    volumes:
+      - signoz_clickhouse_data:/var/lib/clickhouse
+      - ./signoz/zz-clickhouse-config.xml:/etc/clickhouse-server/config.d/zz-clickhouse-config.xml:ro
+      - ./signoz/clickhouse-users.xml:/etc/clickhouse-server/users.d/users.xml:ro
+    environment:
+      - CLICKHOUSE_DB=signoz_traces
+      - CLICKHOUSE_USER=default
+      - CLICKHOUSE_PASSWORD=
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
+    networks:
+      - conserver
+    healthcheck:
+      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://127.0.0.1:8123/ping"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+    restart: unless-stopped
+
+  signoz-otel-collector:
+    image: signoz/signoz-otel-collector:latest
+    container_name: signoz-otel-collector
+    hostname: signoz-otel-collector
+    command:
+      - "--config=/etc/otel-collector-config.yaml"
+    depends_on:
+      signoz-clickhouse:
+        condition: service_healthy
+    environment:
+      - OTEL_RESOURCE_ATTRIBUTES=host.name=signoz-host,os.type=linux
+    volumes:
+      - ./signoz/otel-collector-config.yaml:/etc/otel-collector-config.yaml:ro
+    ports:
+      - "4317:4317"   # OTLP gRPC receiver
+      - "4318:4318"   # OTLP HTTP receiver
+    networks:
+      - conserver
+    healthcheck:
+      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://127.0.0.1:13133/"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+    restart: unless-stopped
+
+  signoz:
+    image: signoz/query-service:latest
+    container_name: signoz
+    hostname: signoz
+    depends_on:
+      signoz-clickhouse:
+        condition: service_healthy
+    environment:
+      - ClickHouseUrl=tcp://signoz-clickhouse:9000
+      - SIGNOZ_LOCAL_DB_PATH=/var/lib/signoz/signoz.db
+      - DASHBOARDS_PATH=/root/config/dashboards
+      - STORAGE=clickhouse
+      - GODEBUG=netdns=go
+      - TELEMETRY_ENABLED=true
+      - DEPLOYMENT_TYPE=docker-standalone
+    volumes:
+      - signoz_data:/var/lib/signoz
+      - ./signoz/dashboards:/root/config/dashboards
+    ports:
+      - "3301:8080"   # Web UI
+    networks:
+      - conserver
+    healthcheck:
+      test: ["CMD", "wget", "--no-verbose", "--tries=1", "--spider", "http://127.0.0.1:8080/api/v1/health"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+    restart: unless-stopped

docker/Dockerfile

@@ -10,6 +10,9 @@ ENV VCON_SERVER_VERSION=${VCON_SERVER_VERSION}
 ENV VCON_SERVER_GIT_COMMIT=${VCON_SERVER_GIT_COMMIT}
 ENV VCON_SERVER_BUILD_TIME=${VCON_SERVER_BUILD_TIME}
 
+# Configure apt to use HTTPS sources (required when HTTP port 80 is blocked)
+RUN sed -i 's|http://deb.debian.org|https://deb.debian.org|g' /etc/apt/sources.list.d/debian.sources
+
 RUN apt-get update && \
     apt-get install -y libavdevice-dev ffmpeg
 

server/api.py

@@ -260,6 +260,24 @@ async def health_check() -> JSONResponse:
     })
 
 
+@app.get(
+    "/stats/queue",
+    summary="Get queue depth",
+    description="Returns the number of items in a Redis list (queue)",
+    tags=["system"],
+)
+async def get_queue_depth(
+    list_name: str = Query(..., description="Name of the Redis list to measure")
+) -> JSONResponse:
+    """Get the current depth of a Redis list. Public endpoint (no auth) for monitoring and backpressure."""
+    try:
+        depth = await redis_async.llen(list_name)
+        return JSONResponse(content={"list_name": list_name, "depth": depth})
+    except Exception as e:
+        logger.error(f"Error getting queue depth for '{list_name}': {str(e)}")
+        raise HTTPException(status_code=500, detail="Failed to get queue depth")
+
+
 class Vcon(BaseModel):
     """Pydantic model representing a vCon (Voice Conversation) record.
 
@@ -659,7 +677,7 @@ async def post_vcon(
         await add_vcon_to_set(key, timestamp)
 
         logger.debug(f"Indexing vCon {inbound_vcon.uuid}")
-        await index_vcon(inbound_vcon.uuid)
+        await index_vcon_parties(str(inbound_vcon.uuid), dict_vcon["parties"])
 
         # Add to ingress lists if specified
         if ingress_lists:
@@ -754,7 +772,7 @@ async def external_ingress_vcon(
         await add_vcon_to_set(key, timestamp)
 
         logger.debug(f"Indexing vCon {inbound_vcon.uuid}")
-        await index_vcon(inbound_vcon.uuid)
+        await index_vcon_parties(str(inbound_vcon.uuid), dict_vcon["parties"])
 
         # Always add to the specified ingress list (required for this endpoint)
         vcon_uuid_str = str(inbound_vcon.uuid)
@@ -1042,25 +1060,17 @@ async def get_dlq_vcons(
         raise HTTPException(status_code=500, detail="Failed to read DLQ")
 
 
-async def index_vcon(uuid: UUID) -> None:
-    """Index a vCon for searching.
+async def index_vcon_parties(vcon_uuid: str, parties: list) -> None:
+    """Index a vCon's parties for searching.
 
-    Adds the vCon to the sorted set and indexes it by party information
-    (tel, mailto, name) for searching. All indexed keys will expire after
-    VCON_INDEX_EXPIRY seconds.
+    Indexes by party information (tel, mailto, name). All indexed keys
+    will expire after VCON_INDEX_EXPIRY seconds.
 
     Args:
-        uuid: UUID of the vCon to index
+        vcon_uuid: UUID string of the vCon
+        parties: List of party dicts from the vCon
     """
-    key = f"vcon:{uuid}"
-    vcon = await redis_async.json().get(key)
-    created_at = datetime.fromisoformat(vcon["created_at"])
-    timestamp = int(created_at.timestamp())
-    vcon_uuid = vcon["uuid"]
-    await add_vcon_to_set(key, timestamp)
-
-    # Index by party information with expiration
-    for party in vcon["parties"]:
+    for party in parties:
         if party.get("tel"):
             tel_key = f"tel:{party['tel']}"
             await redis_async.sadd(tel_key, vcon_uuid)
@@ -1075,6 +1085,25 @@ async def index_vcon(uuid: UUID) -> None:
             await redis_async.expire(name_key, VCON_INDEX_EXPIRY)
 
 
+async def index_vcon(uuid: UUID) -> None:
+    """Index a vCon for searching (reads from Redis).
+
+    Reads the vCon from Redis, adds it to the sorted set, and indexes
+    by party information. Used for bulk re-indexing. For the ingest path,
+    use index_vcon_parties() directly to avoid redundant Redis reads.
+
+    Args:
+        uuid: UUID of the vCon to index
+    """
+    key = f"vcon:{uuid}"
+    vcon = await redis_async.json().get(key)
+    created_at = datetime.fromisoformat(vcon["created_at"])
+    timestamp = int(created_at.timestamp())
+    vcon_uuid = vcon["uuid"]
+    await add_vcon_to_set(key, timestamp)
+    await index_vcon_parties(vcon_uuid, vcon["parties"])
+
+
 @api_router.get(
     "/index_vcons",
     status_code=200,