OpenAI with Cribl Stream

What the source collects

The OpenAI Source in Cribl Stream, described in the OpenAI Source documentation, is a pull-based collector. The page says it "collects model invocation logs, audit logs, and other organization-level telemetry from the OpenAI platform". The page's info box adds that the type is Pull, TLS support is Yes, and Event Breaker support is No.

When you enable specific content types, the page says the Source polls each enabled OpenAI endpoint on its own schedule, uses a shared OpenAI organization auth token for all endpoint calls, tracks state independently for each endpoint, and emits events into Cribl Stream.

The Source "exposes up to 10 OpenAI endpoints as individually toggleable content types": Audit Logs, Costs, Users, Projects, Completions Usage Details, Embeddings Usage Details, Moderations Usage Details, Images Usage Details, Audio Speeches Usage Details, and Audio Transcriptions Usage Details. Every content type uses the same controls: an enable toggle, query parameter details, pagination, schedule, state tracking, and fixed Fields attached to every event from that endpoint.

Setting up the source

Before configuring one, the page lists prerequisites: a Cribl Stream deployment (Cloud or self-hosted), an OpenAI organization with access to the relevant organization-level APIs (such as usage and audit logs), an OpenAI organization admin key with permissions to call the endpoints you plan to enable, created on OpenAI's organization admin keys page, and HTTPS connectivity from your Cribl Workers to the OpenAI API, direct or via an HTTP/S proxy. For proxying, the page points to the System Proxy Configuration documentation.

From the top nav, click Manage and select a Worker Group. Via QuickConnect, click Routing then QuickConnect, then Add Source, select the OpenAI tile, and click Add Source or (if displayed) Select Existing. Via the Routing UI, click Data then Sources, select OpenAI from the tiles or the Sources left nav, then Add Source to open the modal.

Under General Settings, the modal documents these fields:

• Input ID: a unique name identifying this OpenAI Source definition. If you clone this Source, Cribl Stream adds -CLONE to the original Input ID.
• Description: optional.
• API Key: select or create a stored API key.
• OpenAI Organization: optional, the OpenAI-Organization request header value, typically org-xxxxxxxxxxxxxxxxxxxxxxxx.
• OpenAI Project: optional, the OpenAI-Project request header value, typically proj_xxxxxxxxxxxxxxxxxxxxxxxx.
• Tags: optional tags for filtering and grouping Sources in the UI; they aren't added to processed events, and names are separated by a tab or hard return.

Then enable the endpoints you want under Content Types, optionally configure Processing, Retries, and Advanced Settings, and click Save then Commit & Deploy.

For each enabled content type, the Cron schedule configures when collection requests are made. Job timeout caps how long a job may run; units are seconds if not specified (samples 30, 45s, 15m), the minimum granularity is 10 seconds so 45s rounds up to a 50-second timeout, and the default 0 means unlimited. Log Level sets task logging verbosity.

The Earliest time and Latest time fields define the time range of events to collect, based on the _time field. Their syntax, copied from the Scheduling section of the OpenAI Source page:

[+|-]<time_integer><time_unit>@<snap-to_time_unit>

- marks times in the past, + times in the future, or omit both with now. The @ modifier always snaps down (backwards), even with a future + offset: @d to the start of today, @w or @w7 to the preceding Sunday, w1 through w6 for the other weekdays, @mon to the 1st of a month, @q to the most recent quarter start, and @y to Jan. 1.

Under State tracking, the State update expression is a JavaScript expression that defines how to update state from an event, and the State merge expression defines which state to keep when a task's newly reported state is merged with the previously saved state. The defaults, copied from the State tracking section, track the latest _time in events from a collection run:

__timestampExtracted !== false && {latestTime: (state.latestTime || 0) > _time ? state.latestTime : _time}

prevState.latestTime > newState.latestTime ? prevState : newState

Per the page, __timestampExtracted is false when the Event Breaker could not parse an event's time, and the update expression then short-circuits without updating state. State values must resolve to an object, such as { "latestTime": 17122806161 }, or Cribl Stream ignores the result; the merge expression keeps whichever of prevState and newState has the greatest latestTime. The page frames these controls as a defense against overlaps between jobs and gaps in data, since a run can pick up from where the last run ended.

Retries adjust the defaults for failed collect jobs. Retry type is the HTTP retry algorithm: Backoff (default), Static, or Disabled. The Initial retry interval (ms) (maximum 20,000 ms; 0 retries immediately) spaces the first retry; the Retry limit defaults to 5 (maximum 20; 0 retries not at all). The Backoff multiplier defaults to 2, so Cribl Stream retries after 2 seconds, then 4, then 8, and so forth. Retry HTTP codes default to 429 and 503, and codes in the 200 series are never retried.

Honor Retry-After header is on by default and processes delays up to a maximum of 20 seconds; toggled off, retry-after headers are ignored entirely. Separate toggles retry a connection after a timeout (ETIMEDOUT) or a peer reset (ECONNRESET). Processing Settings add Fields to each event using Eval-like functionality: a Name and a Value that is a JavaScript expression, which can be a constant; Fields-tab fields normally override same-named event fields, but you can make event fields win instead. Pre-Processing selects a Pipeline to run before routes, available only when Send to Routes is toggled on.

Advanced Settings cover Request timeout (seconds) (default 300, with 0 waiting indefinitely), Time to live (seconds) (default 4h), and Environment (if you are using GitOps, it scopes the config to one Git branch; if empty, the config is enabled everywhere). Under Connected Destinations, Send to Routes enables conditional routing, filtering, and cloning via the Routing table, while QuickConnect sends data to one or more Destinations over direct connections.

What the events look like

The OpenAI Source page does not publish a sample event or a field map for any content type. What it documents are two Cribl Stream internal fields for this Source: __collectible, which "Contains metadata about each collection job", and __collectStats, which "Contains per-request metadata".

These are "meta" fields: not part of an event, but accessible. Every OpenAI API response also includes rate-limit headers describing current limits and remaining quota: x-ratelimit-limit-requests / x-ratelimit-remaining-requests, x-ratelimit-limit-tokens / x-ratelimit-remaining-tokens, and x-ratelimit-reset-requests / x-ratelimit-reset-tokens.

What the docs flag

OpenAI enforces per-organization and per-project rate limits on both the number of requests and the number of tokens over a given interval. If your schedules or enabled content types exceed these limits, OpenAI returns 429 responses. The Source relies on its retry settings and your polling schedule to avoid sustained rate-limit errors, and if repeated 429s appear in job logs, the page directs you to reduce the polling frequency or disable nonessential endpoints.

The Source treats all non-200 responses from configured endpoints as errors, including 1xx, 3xx, 4xx, and 5xx responses. Two cases are non-fatal: where a collect job launches multiple tasks and only a subset fails, Cribl Stream places the job in failed status but does not retry the failed tasks; and where a job receives a 3xx redirection error code, it follows the underlying library's treatment of the error and does not necessarily treat it as fatal.

Smaller flags: Tags are not added to processed events, cloning adds -CLONE to the original Input ID, and every content type arrives with Scheduling and Query Parameters already configured. The configuration modal also has Live Data and Logs tabs for troubleshooting.

Frequently asked questions

What OpenAI data types does the Cribl Stream OpenAI Source collect?

It collects model invocation logs, audit logs, and other organization-level telemetry from the OpenAI platform, exposing up to 10 endpoints as individually toggleable content types. The types include Audit Logs, Costs, Users, Projects, and Usage Details endpoints for completions, embeddings, moderations, images, audio speeches, and audio transcriptions, and each enabled endpoint is polled on its own schedule.

How do you authenticate the Cribl Stream OpenAI Source?

You provide an OpenAI organization admin key with permissions to call the endpoints you plan to enable, created on OpenAI's organization admin keys page, and select or create a stored API Key in the Source modal. A single shared OpenAI organization auth token is used for all endpoint calls.

How does the OpenAI Source avoid overlapping or missing events between polls?

It tracks state independently for each endpoint, using State update and State merge expressions that default to tracking the latest _time field found in events gathered in a collection run. The page frames this as a defense against overlaps between jobs and gaps in data, since a run can pick up from where the last run ended.

What should you do when the OpenAI Source hits OpenAI rate limits?

OpenAI returns 429 responses when your schedules or enabled content types exceed its per-organization and per-project limits, and the Source relies on its retry settings and your polling schedule to avoid sustained rate-limit errors. If repeated 429s appear in job logs, reduce the polling frequency or disable nonessential endpoints.