Merge pull request #144 from Synthetixio/feat/clickhouse

ClickHouse Support

docker-compose.yml

@@ -1,4 +1,25 @@
 services:
+  clickhouse:
+    image: clickhouse/clickhouse-server:latest
+    restart: always
+    networks:
+      - data
+    ulimits:
+      nofile:
+        soft: 262144
+        hard: 262144
+    volumes:
+      - clickhouse_data:/var/lib/clickhouse
+      - ./parquet-data:/var/lib/clickhouse/user_files/parquet-data
+    ports:
+      - 8123:8123
+      - 9000:9000
+      - 9009:9009
+    deploy:
+      resources:
+        limits:
+          cpus: "4.0"
+          memory: 8192M
   db:
     image: ghcr.io/synthetixio/data/postgres:${VERSION}
     restart: always
@@ -36,6 +57,18 @@ services:
     volumes:
       - ./parquet-data:/parquet-data
 
+  indexer-listener:
+    build:
+      context: ./indexers
+      dockerfile: Dockerfile.listener
+    networks:
+      - data
+    restart: always
+    environment:
+      PYTHONUNBUFFERED: 1
+    volumes:
+      - ./parquet-data:/parquet-data
+
   transformer:
     image: ghcr.io/synthetixio/data/transformer:${VERSION}
     build:
@@ -85,3 +118,7 @@ services:
 networks:
   data:
     driver: bridge
+
+volumes:
+  clickhouse_data:
+    driver: local

indexers/Dockerfile

@@ -1,19 +1,15 @@
-FROM node:16-alpine
+FROM node:16-bullseye
 COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
 
 WORKDIR /app
 
 COPY package*.json ./
 COPY patches/ ./patches/
 
-RUN apk add --no-cache python3 python3-dev py3-pip clang cmake build-base git
-
-RUN npm ci
+RUN apt-get update && apt-get install -y build-essential && npm ci && apt-get remove -y build-essential
 
 COPY pyproject.toml uv.lock ./
-RUN uv sync --frozen
-
-RUN apk del build-base
+RUN uv sync --frozen --no-dev
 
 COPY . .
 

indexers/Dockerfile.listener

@@ -0,0 +1,16 @@
+FROM python:3.11-slim
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y \
+    gcc \
+    python3-dev \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY pyproject.toml uv.lock ./
+RUN uv sync --frozen --only-dev
+
+COPY scripts/listener.py .
+
+CMD ["uv", "run", "python", "listener.py"]

indexers/entrypoint.sh

@@ -3,14 +3,23 @@
 # Exit on error
 set -e
 
-# Get contract data from SDK and generate squidgen.yaml and squid.yaml
-uv run main.py --network_name $NETWORK_NAME --protocol_name $PROTOCOL_NAME "$@"
+COMMAND="uv run main.py --network_name $NETWORK_NAME --protocol_name $PROTOCOL_NAME"
+
+if [ ! -z "$BLOCK_FROM" ]; then
+    COMMAND="$COMMAND --block_from $BLOCK_FROM"
+fi
+
+if [ ! -z "$BLOCK_TO" ]; then
+    COMMAND="$COMMAND --block_to $BLOCK_TO"
+fi
+
+$COMMAND
 
 # Generate squid processor
 npm run generate:processor
 
 # Build squid processor
 npm run build
 
-# Start the squid processor
-npm run start
+# Start squid processor
+npm run start

indexers/main.py

@@ -3,11 +3,14 @@
 import argparse
 from dotenv import load_dotenv
 import yaml
+import clickhouse_connect
 from synthetix import Synthetix
 
 # load environment variables
 load_dotenv()
 
+RAW_DATA_PATH = "/parquet-data/indexers/raw"
+
 
 def save_abi(abi, contract_name):
     os.makedirs("abi", exist_ok=True)
@@ -19,23 +22,23 @@ def create_squidgen_config(
     rpc_url,
     archive_url,
     network_name,
-    contracts_info,
+    contracts,
     block_range,
     protocol_name,
     rate_limit=10,
 ):
     config = {
         "archive": archive_url,
-        "finalityConfirmation": 1,
         "chain": {"url": rpc_url, "rateLimit": rate_limit},
+        "finalityConfirmation": 1,
         "target": {
             "type": "parquet",
-            "path": f"/parquet-data/indexers/raw/{network_name}/{protocol_name}",
+            "path": f"{RAW_DATA_PATH}/{network_name}/{protocol_name}",
         },
         "contracts": [],
     }
 
-    for contract in contracts_info:
+    for contract in contracts:
         name = contract["name"]
         address = contract["address"]
         contract_config = {
@@ -51,21 +54,6 @@ def create_squidgen_config(
     return config
 
 
-def create_squid_config(network_name):
-    squid_config = {
-        "manifestVersion": "subsquid.io/v0.1",
-        "name": network_name,
-        "version": 1,
-        "description": "A squid indexer generated from an ABI template",
-        "build": None,
-        "deploy": {
-            "processor": {"cmd": ["node", "lib/main"]},
-        },
-    }
-
-    return squid_config
-
-
 def write_yaml(config, filename):
     with open(filename, "w") as file:
         yaml.dump(config, file, default_flow_style=False)
@@ -93,6 +81,18 @@ def load_network_config(path):
         type=str,
         help="Comma-separated list of contract names to index.",
     )
+    parser.add_argument(
+        "--block_from",
+        type=int,
+        help="Block number to start indexing from",
+        required=False,
+    )
+    parser.add_argument(
+        "--block_to",
+        type=int,
+        help="Block number to end indexing at",
+        required=False,
+    )
     args = parser.parse_args()
 
     network_name = args.network_name
@@ -133,14 +133,15 @@ def load_network_config(path):
         )
 
     # Set block range based on config.
-    # If "to" is "latest", use the latest block from the RPC endpoint.
     block_range = {}
-    block_range["from"] = custom_config["range"].get("from", 0)
-    if "to" in custom_config["range"]:
-        if custom_config["range"]["to"] == "latest":
-            block_range["to"] = snx.web3.eth.block_number
-        else:
-            block_range["to"] = custom_config["range"]["to"]
+    if args.block_from is not None:
+        block_range["from"] = args.block_from
+    else:
+        block_range["from"] = custom_config["range"].get("from", 0)
+    if args.block_to is not None:
+        block_range["to"] = args.block_to
+    elif "to" in custom_config["range"]:
+        block_range["to"] = custom_config["range"]["to"]
 
     # Get contracts from SDK or ABI files
     contracts = []
@@ -170,19 +171,22 @@ def load_network_config(path):
     # Create squidgen generator config
     rate_limit = custom_config.get("rate_limit", 10)
     squidgen_config = create_squidgen_config(
-        rpc_endpoint,
-        archive_url,
-        network_name,
-        contracts,
-        block_range,
-        protocol_name,
-        rate_limit,
+        rpc_url=rpc_endpoint,
+        archive_url=archive_url,
+        network_name=network_name,
+        contracts=contracts,
+        block_range=block_range,
+        protocol_name=protocol_name,
+        rate_limit=rate_limit,
     )
     write_yaml(squidgen_config, "squidgen.yaml")
 
-    squid_config = create_squid_config(args.network_name)
-    write_yaml(squid_config, "squid.yaml")
-
     snx.logger.info(
-        f"squidgen.yaml, squid.yaml, and ABI files have been generated for {args.network_name}"
+        f"squidgen.yaml and ABI files have been generated for {args.network_name}"
     )
+
+    # Create database in ClickHouse
+    client = clickhouse_connect.get_client(host="clickhouse", port=8123, user="default")
+    client.command(f"create database if not exists {network_name}")
+    client.close()
+    snx.logger.info(f"Database '{network_name}' has been created in ClickHouse")

indexers/networks/arbitrum_mainnet/network_config.yaml

@@ -7,7 +7,6 @@ configs:
   synthetix:
     range:
       from: 218000000
-      to: latest
     contracts_from_sdk:
       - name: CoreProxy
         package: system
@@ -23,7 +22,6 @@ configs:
   curve:
     range:
       from: 236000000
-      to: latest
     contracts_from_abi:
       - name: CurveUsdx
         address: '0x096A8865367686290639bc50bF8D85C0110d9Fea'

indexers/networks/base_mainnet/network_config.yaml

@@ -7,7 +7,6 @@ configs:
   synthetix:
     range:
       from: 7500000
-      to: latest
     contracts_from_sdk:
       - name: CoreProxy
         package: system