Skip to main content
API7.ai Logo

Load Request Telemetry 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-attempt record into columns that can be queried.

Prerequisites​

Before starting, prepare the following:

  • One of these configuration paths:
    • AISIX Cloud with an environment, an attached gateway, and a write-scoped admin token. For On-Premises, follow the AISIX Cloud Quickstart. To request Hybrid Cloud access, contact API7.
    • An open-source AISIX gateway that loads a declarative resources.yaml file.
  • 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.

Export the gateway URL and request values used to generate test telemetry:

export AISIX_PROXY="http://127.0.0.1:3000"
export AISIX_API_KEY="YOUR_CALLER_API_KEY"
export MODEL_ALIAS="YOUR_MODEL_ALIAS"

For the AISIX Cloud examples, also export the Admin API base URL, admin token, and environment ID:

export AISIX_CP="http://localhost:8080/api"
export AISIX_TOKEN="YOUR_ADMIN_TOKEN"
export ENV_ID="YOUR_ENVIRONMENT_ID"

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 usage telemetry for each request attempt. A request that retries or fails over produces multiple records with the same request ID. 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, attempt, 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 gateway deployment its own bucket prefix. AISIX partitions objects by date and hour under the configured prefix, but the object path does not add a deployment segment for you.

Configure Object Storage Export​

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​

With AISIX Cloud, create an exporter through the Admin API:

EXPORTER_ID=$(curl -sS -X POST "$AISIX_CP/environments/$ENV_ID/observability_exporters" \
  -H "Authorization: Bearer $AISIX_TOKEN" \
  -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"
  }' | jq -r '.observability_exporter.id')

❶ Use s3 for Amazon S3 and S3-compatible storage targets.

❷ bucket and prefix define where AISIX writes Snowflake staging files. Use a distinct prefix for each gateway deployment.

❸ credential_ref maps this exporter to gateway environment variables. The suffix comes from the reference value, converted to uppercase.

For the open-source AISIX gateway, add the equivalent entry under observability_exporters in resources.yaml:

resources.yaml
_format_version: "1"

observability_exporters:
  - name: snowflake-staging
    kind: object_store
    provider: s3
    bucket: acme-aisix-events
    prefix: ai-gateway
    region: us-east-1
    credential_ref: acme_s3_prod

Configure the matching credentials in the environment of every gateway instance that sends telemetry. For a gateway launched from a shell, export these values before starting it:

# Replace with your values
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"

For the resources-file path, validate the file:

aisix validate --resources resources.yaml

After validation, start or restart the gateway with the credential variables in its process environment. Reload a running gateway only if those variables are already available to the process.

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

for i in 1 2 3; do
  curl -sS -X POST "$AISIX_PROXY/v1/chat/completions" \
    -H "Authorization: Bearer $AISIX_API_KEY" \
    -H "Content-Type: application/json" \
    -d '{
      "model": "'"${MODEL_ALIAS}"'",
      "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-attempt 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",
  "attempt_index": 0,
  "attempt_kind": "initial",
  "prompt_tokens": 8,
  "completion_tokens": 12,
  "upstream_latency_ms": 548,
  "downstream_latency_ms": 612,
  "status_code": 200,
  "cost_usd": 0.00021,
  "cache_status": "miss",
  "guardrail_blocked": false
}

Azure Blob​

With AISIX Cloud, create an exporter through the Admin API:

EXPORTER_ID=$(curl -sS -X POST "$AISIX_CP/environments/$ENV_ID/observability_exporters" \
  -H "Authorization: Bearer $AISIX_TOKEN" \
  -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"
  }' | jq -r '.observability_exporter.id')

❶ Use azure_blob when the staging destination is an Azure Blob container.

❷ For Azure Blob, bucket is the container name. The prefix controls the folder-like path under that container.

❸ credential_ref maps this exporter to gateway environment variables. The suffix comes from the reference value, converted to uppercase.

For the open-source AISIX gateway, add the equivalent entry under observability_exporters in resources.yaml:

resources.yaml
_format_version: "1"

observability_exporters:
  - name: snowflake-staging
    kind: object_store
    provider: azure_blob
    bucket: ai-gateway
    prefix: ai-gateway
    credential_ref: acme_az_prod

Configure the matching credentials in the environment of every gateway instance that sends telemetry. For a gateway launched from a shell, export these values before starting it:

# Replace with your values
export OBJSTORE_CRED_ACME_AZ_PROD_AZURE_ACCOUNT="YOUR_STORAGE_ACCOUNT"
export OBJSTORE_CRED_ACME_AZ_PROD_AZURE_ACCESS_KEY="YOUR_STORAGE_ACCESS_KEY"

For the resources-file path, validate the file:

aisix validate --resources resources.yaml

After validation, start or restart the gateway with the credential variables in its process environment. Reload a running gateway only if those variables are already available to the process.

Generate traffic and list the staged blobs:

for i in 1 2 3; do
  curl -sS -X POST "$AISIX_PROXY/v1/chat/completions" \
    -H "Authorization: Bearer $AISIX_API_KEY" \
    -H "Content-Type: application/json" \
    -d '{
      "model": "'"${MODEL_ALIAS}"'",
      "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 Request Telemetry​

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_attempts 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:attempt_index::number AS attempt_index,
  record:attempt_kind::string AS attempt_kind,
  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:upstream_latency_ms::number AS upstream_latency_ms,
  record:upstream_ttft_ms::number AS upstream_ttft_ms,
  record:downstream_latency_ms::number AS downstream_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;

Use downstream_latency_ms for the time experienced by the caller; it appears only on the terminal attempt. Use upstream_latency_ms to analyze each provider attempt, and upstream_ttft_ms for the time until that attempt streamed its first frame. The time-to-first-frame field is omitted or zero for non-streaming requests, errors, and cache hits.

Query recent attempts:

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

Run a simple cost and token summary:

SELECT
  requested_model,
  COUNT(DISTINCT request_id) AS requests,
  COUNT(*) AS attempts,
  SUM(prompt_tokens + completion_tokens) AS total_tokens,
  ROUND(SUM(cost_usd), 5) AS total_cost_usd
FROM gateway_attempts
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_attempts;
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. With AISIX Cloud, delete the exporter through the Admin API:

curl -sS -X DELETE "$AISIX_CP/environments/$ENV_ID/observability_exporters/$EXPORTER_ID" \
  -H "Authorization: Bearer $AISIX_TOKEN"

For the open-source AISIX gateway, remove the exporter entry from resources.yaml, then validate and reload the file.

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.