Load Logs into Snowflake

AISIX can write request telemetry to object storage through an observability exporter. Use this guide after configuring exporters when finance, business intelligence, or audit teams need to query gateway usage, cost, latency, cache outcomes, and guardrail outcomes in Snowflake.

The gateway writes batched NDJSON objects to a bucket you own. Snowpipe ingests those objects into a Snowflake table, and a SQL view turns each request record into columns that can be queried.

Prerequisites

Before starting, prepare the following:

• A self-hosted AISIX gateway with the admin and proxy listeners available.
• The admin key from the gateway config.yaml.
• A working model alias and caller API key that can send chat-completions requests.
• One Amazon S3 bucket or Azure Blob container for staged request telemetry.
• A Snowflake account and a role that can create storage integrations, notification integrations, stages, pipes, tables, and views.

This guide uses acme-aisix-events as the S3 bucket name and ai-gateway as the object prefix. Replace them with values for your environment.

Telemetry Ingestion Flow

The object_store exporter is an object-storage staging path. AISIX does not push request telemetry directly to Snowflake from the request path.

AISIX emits request telemetry after each supported request completes. The exporter writes gzipped NDJSON objects to Amazon S3 or Azure Blob Storage. Snowpipe loads each record into a Snowflake landing table, and a SQL view exposes request, cost, latency, cache, and guardrail columns for analysis.

Exporter delivery is metadata-oriented by default. It includes fields such as request ID, requested model alias, resolved model ID, status, token counts, cost, latency, cache status, and guardrail outcome. It does not include prompt or response text unless a supported exporter is explicitly configured for content capture.

Give each environment its own bucket prefix. AISIX partitions objects by date and hour under the configured prefix, but the object path does not add an environment segment for you.

Create an Object Store Exporter

Create the exporter first and confirm files land in object storage before wiring Snowflake. This isolates the gateway side from the warehouse ingestion side.

Amazon S3

Create an exporter that writes request telemetry to S3:

curl -sS -X POST "http://127.0.0.1:3001/admin/v1/observability_exporters" \
  -H "Authorization: Bearer YOUR_ADMIN_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "snowflake-staging",
    "kind": "object_store",
    "provider": "s3",
    "bucket": "acme-aisix-events",
    "prefix": "ai-gateway",
    "region": "us-east-1",
    "credential_ref": "acme_s3_prod"
  }'

Set the matching credentials in the gateway process environment. The suffix comes from the credential_ref value, converted to uppercase. If you add or change these variables after the gateway starts, restart the gateway before testing delivery.

export OBJSTORE_CRED_ACME_S3_PROD_AWS_ACCESS_KEY_ID="YOUR_AWS_ACCESS_KEY_ID"
export OBJSTORE_CRED_ACME_S3_PROD_AWS_SECRET_ACCESS_KEY="YOUR_AWS_SECRET_ACCESS_KEY"

Generate a few requests so the exporter has telemetry to flush:

for i in 1 2 3; do
  curl -sS -X POST "http://127.0.0.1:3000/v1/chat/completions" \
    -H "Authorization: Bearer YOUR_CALLER_API_KEY" \
    -H "Content-Type: application/json" \
    -d '{"model":"gpt-4o-prod","messages":[{"role":"user","content":"ping"}]}' > /dev/null
done

sleep 8

List the staged objects:

aws s3 ls "s3://acme-aisix-events/ai-gateway/" --recursive

The bucket should contain one or more .ndjson.gz objects under the configured prefix, date partition, and hour partition:

2026-06-09 14:03:11        742 ai-gateway/dt=2026-06-09/hh=14/9f86d081884c7d659a2feaa0c55ad015.ndjson.gz

Inspect one object before configuring Snowpipe:

aws s3 cp "s3://acme-aisix-events/ai-gateway/dt=2026-06-09/hh=14/OBJECT.ndjson.gz" - | gunzip

The output contains one request telemetry record per line. This example shows one formatted record:

{
  "schema_version": "1.0",
  "request_id": "742c6f5e-7b97-4bb1-9f5f-8cb42b4c93e1",
  "occurred_at": "2026-06-09T14:03:10Z",
  "requested_model": "gpt-4o-prod",
  "model_id": "b7c8e4f2-2e4d-4776-a8d7-09b4eb0cb2b1",
  "prompt_tokens": 8,
  "completion_tokens": 12,
  "latency_ms": 612,
  "status_code": 200,
  "cost_usd": 0.00021,
  "cache_status": "miss",
  "guardrail_blocked": false
}

Azure Blob

Create an exporter that writes request telemetry to Azure Blob:

curl -sS -X POST "http://127.0.0.1:3001/admin/v1/observability_exporters" \
  -H "Authorization: Bearer YOUR_ADMIN_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "snowflake-staging",
    "kind": "object_store",
    "provider": "azure_blob",
    "bucket": "ai-gateway",
    "prefix": "ai-gateway",
    "credential_ref": "acme_az_prod"
  }'

Set the matching credentials in the gateway process environment. If you add or change these variables after the gateway starts, restart the gateway before testing delivery.

export OBJSTORE_CRED_ACME_AZ_PROD_AZURE_ACCOUNT="YOUR_STORAGE_ACCOUNT"
export OBJSTORE_CRED_ACME_AZ_PROD_AZURE_ACCESS_KEY="YOUR_STORAGE_ACCESS_KEY"

Generate traffic and list the staged blobs:

for i in 1 2 3; do
  curl -sS -X POST "http://127.0.0.1:3000/v1/chat/completions" \
    -H "Authorization: Bearer YOUR_CALLER_API_KEY" \
    -H "Content-Type: application/json" \
    -d '{"model":"gpt-4o-prod","messages":[{"role":"user","content":"ping"}]}' > /dev/null
done

sleep 8

az storage blob list \
  --account-name "YOUR_STORAGE_ACCOUNT" \
  --account-key "YOUR_STORAGE_ACCESS_KEY" \
  --container-name "ai-gateway" \
  --prefix "ai-gateway/" \
  --query "[].name" \
  -o tsv

The container should contain gzipped NDJSON objects under the same prefix, date partition, and hour partition layout.

Create a Snowflake Landing Table

Create a landing table with a VARIANT column so Snowflake can ingest each NDJSON record without losing fields:

CREATE DATABASE IF NOT EXISTS aisix;
CREATE SCHEMA IF NOT EXISTS aisix.gateway;
USE SCHEMA aisix.gateway;

CREATE TABLE IF NOT EXISTS gateway_events (
  record VARIANT,
  source_file STRING,
  loaded_at TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

Snowflake detects gzip when COMPRESSION = AUTO is used in the pipe file format.

Connect Snowflake to S3

For S3, Snowflake reads the bucket through a storage integration and receives object-created events through the SQS queue associated with the pipe.

Create the storage integration:

CREATE STORAGE INTEGRATION aisix_s3_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'S3'
  ENABLED = TRUE
  STORAGE_AWS_ROLE_ARN = 'arn:aws:iam::123456789012:role/aisix-snowflake-read'
  STORAGE_ALLOWED_LOCATIONS = ('s3://acme-aisix-events/ai-gateway/');

DESC INTEGRATION aisix_s3_int;

Use the STORAGE_AWS_IAM_USER_ARN and STORAGE_AWS_EXTERNAL_ID values returned by DESC INTEGRATION to configure the IAM role trust policy. Grant the role permission to list the bucket and read objects under the configured prefix.

Create the stage and pipe:

CREATE STAGE aisix_stage
  URL = 's3://acme-aisix-events/ai-gateway/'
  STORAGE_INTEGRATION = aisix_s3_int;

CREATE PIPE aisix_events_pipe
  AUTO_INGEST = TRUE
  AS
    COPY INTO gateway_events (record, source_file)
      FROM (SELECT $1, METADATA$FILENAME FROM @aisix_stage)
      FILE_FORMAT = (TYPE = JSON COMPRESSION = AUTO);

SHOW PIPES;

Use the notification_channel value from SHOW PIPES to add an S3 event notification for object-created events:

aws s3api put-bucket-notification-configuration \
  --bucket "acme-aisix-events" \
  --notification-configuration '{
    "QueueConfigurations": [{
      "QueueArn": "arn:aws:sqs:us-east-1:NNNN:sf-snowpipe-example",
      "Events": ["s3:ObjectCreated:*"],
      "Filter": {"Key": {"FilterRules": [{"Name": "prefix", "Value": "ai-gateway/"}]}}
    }]
  }'

Connect Snowflake to Azure Blob

For Azure Blob, Snowflake reads the container through a storage integration and receives object-created events through a storage queue.

Create a storage queue and Event Grid subscription:

az storage queue create \
  --name "aisix-snowpipe" \
  --account-name "YOUR_STORAGE_ACCOUNT"

az eventgrid event-subscription create \
  --source-resource-id "/subscriptions/YOUR_SUBSCRIPTION_ID/resourceGroups/YOUR_RESOURCE_GROUP/providers/Microsoft.Storage/storageAccounts/YOUR_STORAGE_ACCOUNT" \
  --name "aisix-snowpipe-sub" \
  --endpoint-type storagequeue \
  --endpoint "/subscriptions/YOUR_SUBSCRIPTION_ID/resourceGroups/YOUR_RESOURCE_GROUP/providers/Microsoft.Storage/storageAccounts/YOUR_STORAGE_ACCOUNT/queueServices/default/queues/aisix-snowpipe" \
  --advanced-filter data.api stringin CopyBlob PutBlob PutBlockList FlushWithClose

Create the notification integration:

CREATE NOTIFICATION INTEGRATION aisix_az_notif
  ENABLED = TRUE
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = AZURE_STORAGE_QUEUE
  AZURE_STORAGE_QUEUE_PRIMARY_URI = 'https://YOUR_STORAGE_ACCOUNT.queue.core.windows.net/aisix-snowpipe'
  AZURE_TENANT_ID = 'YOUR_TENANT_ID';

DESC NOTIFICATION INTEGRATION aisix_az_notif;

Grant the Snowflake service principal returned by DESC NOTIFICATION INTEGRATION access to the storage queue.

Create the storage integration, stage, and pipe:

CREATE STORAGE INTEGRATION aisix_az_int
  TYPE = EXTERNAL_STAGE
  STORAGE_PROVIDER = 'AZURE'
  ENABLED = TRUE
  AZURE_TENANT_ID = 'YOUR_TENANT_ID'
  STORAGE_ALLOWED_LOCATIONS = ('azure://YOUR_STORAGE_ACCOUNT.blob.core.windows.net/ai-gateway/ai-gateway/');

DESC INTEGRATION aisix_az_int;

CREATE STAGE aisix_stage
  URL = 'azure://YOUR_STORAGE_ACCOUNT.blob.core.windows.net/ai-gateway/ai-gateway/'
  STORAGE_INTEGRATION = aisix_az_int;

CREATE PIPE aisix_events_pipe
  AUTO_INGEST = TRUE
  INTEGRATION = 'AISIX_AZ_NOTIF'
  AS
    COPY INTO gateway_events (record, source_file)
      FROM (SELECT $1, METADATA$FILENAME FROM @aisix_stage)
      FILE_FORMAT = (TYPE = JSON COMPRESSION = AUTO);

Grant the Snowflake service principal returned by DESC INTEGRATION read access to the blob container.

Query Gateway Requests

After sending more traffic through AISIX, check that Snowpipe is receiving files:

SELECT SYSTEM$PIPE_STATUS('aisix_events_pipe');

Confirm rows arrived and create a view over the raw records:

SELECT COUNT(*) FROM gateway_events;

CREATE OR REPLACE VIEW gateway_requests AS
SELECT
  record:request_id::string AS request_id,
  record:occurred_at::timestamp_tz AS occurred_at,
  record:requested_model::string AS requested_model,
  record:model_id::string AS model_id,
  record:status_code::number AS status_code,
  record:prompt_tokens::number AS prompt_tokens,
  record:completion_tokens::number AS completion_tokens,
  record:cost_usd::float AS cost_usd,
  record:latency_ms::number AS latency_ms,
  record:cache_status::string AS cache_status,
  record:guardrail_blocked::boolean AS guardrail_blocked,
  record:finish_reason::string AS finish_reason,
  source_file
FROM gateway_events;

Query recent requests:

SELECT requested_model, status_code, prompt_tokens, completion_tokens, cost_usd, cache_status
FROM gateway_requests
ORDER BY occurred_at DESC
LIMIT 10;

Run a simple cost and token summary:

SELECT
  requested_model,
  COUNT(*) AS requests,
  SUM(prompt_tokens + completion_tokens) AS total_tokens,
  ROUND(SUM(cost_usd), 5) AS total_cost_usd
FROM gateway_requests
GROUP BY requested_model
ORDER BY total_cost_usd DESC;

Cleanup

Drop the Snowflake objects when you finish testing:

DROP PIPE IF EXISTS aisix_events_pipe;
DROP STAGE IF EXISTS aisix_stage;
DROP VIEW IF EXISTS gateway_requests;
DROP TABLE IF EXISTS gateway_events;
DROP STORAGE INTEGRATION IF EXISTS aisix_s3_int;
DROP STORAGE INTEGRATION IF EXISTS aisix_az_int;
DROP NOTIFICATION INTEGRATION IF EXISTS aisix_az_notif;

Remove the bucket notification or Event Grid subscription you created, then delete the exporter:

curl -sS -X DELETE "http://127.0.0.1:3001/admin/v1/observability_exporters/YOUR_EXPORTER_ID" \
  -H "Authorization: Bearer YOUR_ADMIN_KEY"

Next Steps

You have now loaded AISIX request telemetry into Snowflake. Continue with Metrics and Logs when you need to correlate exported usage events with runtime metrics, access logs, and response headers.