cel-programs

# cel-programs ## When to use Use this skill when tasks include: - creating or editing `cel.yml.hbs` agent stream templates - configuring data stream manifests for the `cel` input type - writing CEL programs with pagination, cursor management, or authentication - testing or debugging a CEL program locally with mito - setting up system tests with mock APIs for CEL-based data streams - prototyping a new CEL-based data stream's collection logic - any CEL or mito question, regardless of context ## When not to use Do not use this skill as the primary guide for: - ingest pipeline processor design (`ingest-pipelines`) - ECS field mapping (`ecs-field-mappings`) - package scaffolding (`create-integration`) - system test execution with the Elastic stack (`integration-testing` → `references/system-testing.md`) ## Mandatory workflow — mock → mito → template **This is not a suggestion. Every CEL program MUST be developed in this order.** The subagent must not write `cel.yml.hbs` until the CEL program has been validated with mito against a running mock. Skipping steps or reordering causes failures that are hard to debug. **Do NOT write more than ~10–15 new lines of CEL before running mito.** Build the program incrementally in phases (skeleton → error handling → event mapping → pagination → cursor guard), validating with mito after each phase. Writing a large program in one shot leads to cascading compilation errors that are extremely hard to debug. Follow the phased approach in `references/cel-incremental-build.md`. | Step | Action | Output | |------|--------|--------| | **1. Create the system test mock** | Write the `elastic/stream` config at `_dev/deploy/docker/files/config-<stream>.yml` with rules matching all API endpoints. Write `test-default-config.yml`. | Mock config file, docker-compose service, test config | | **2. Start the mock locally** | `stream http-server --addr=:8090 --config=...` | Running mock at `http://localhost:8090` | | **3. Create a plain `.cel` file and `state.json`** | Write the CEL program as a standalone `.cel` file. Create `state.json` with the same keys the future `state:` block will contain, but with literal test values instead of Handlebars. Point `url` at the local mock. | `program.cel`, `state.json` in `/tmp` or working dir | | **4. Run mito and iterate** | Build incrementally per `references/cel-incremental-build.md`: Phase 0 skeleton → Phase 1 error handling → Phase 2 events → Phase 3 pagination → Phase 4 cursor. Run `mito -data state.json -log_requests program.cel` after each phase. Do not proceed until mito output is correct. | Validated CEL program | | **5. ONLY THEN write `cel.yml.hbs`** | Copy the working CEL expression into `program: \|` in the Handlebars template. Replace literal test values with `{{var}}` references. Configure manifests. | Final integration template | **Step 3 detail — translating template vars to mito state:** When the future `cel.yml.hbs` will have a `state:` block like `api_key: {{api_key}}` and `batch_size: {{batch_size}}`, the `state.json` for mito testing uses the same key names with literal test values: ```json { "url": "http://localhost:8090", "api_key": "test-key", "batch_size": 50, "initial_interval": "24h" } ``` This mirrors the runtime state the CEL input would provide. Add `cursor` to test subsequent-run behavior. For the full mock-first workflow details, CLI flags, execution model, and quality standards: load `references/mito-reference.md`. --- ## cel.yml.hbs template anatomy The `cel.yml.hbs` file at `data_stream/<stream>/agent/stream/cel.yml.hbs` is a Handlebars template that renders the final CEL input configuration. It has these sections in order: ```yaml interval: {{interval}} resource.tracer: enabled: {{enable_request_tracer}} filename: "../../logs/cel/http-request-trace-*.ndjson" maxbackups: 5 {{#if proxy_url}} resource.proxy_url: {{proxy_url}} {{/if}} {{#if ssl}} resource.ssl: {{ssl}} {{/if}} {{#if http_client_timeout}} resource.timeout: {{http_client_timeout}} {{/if}} resource.url: <constructed from vars> state: <credentials and pagination config from vars> redact: fields: - <sensitive state keys> max_executions: <number, for heavy pagination> program: | <CEL expression> tags: {{#if preserve_original_event}} - preserve_original_event {{/if}} {{#each tags as |tag|}} - {{tag}} {{/each}} {{#contains "forwarded" tags}} publisher_pipeline.disable_host: true {{/contains}} {{#if processors}} processors: {{processors}} {{/if}} ``` ### Handlebars patterns | Pattern | Purpose | |---------|---------| | `{{var_name}}` | Direct variable substitution | | `{{#if var_name}}...{{/if}}` | Conditional block for optional config | | `{{#each tags as \|tag\|}}` | Iteration over list vars | | `{{#contains "forwarded" tags}}` | Check if list contains value | ### Key template fields - `resource.url` — base URL, often constructed from multiple vars (e.g., `{{url}}/api/v1/endpoint`) - `resource.headers` (ga 8.18.1) — static headers the same for every request (`Content-Type`, `Accept`, API version headers). Set here rather than in-program when headers never vary. Applied before auth headers. - `state:` — block where manifest vars are injected as CEL state; credentials and pagination settings go here - `redact.fields` — list state keys containing secrets to redact from debug logs - `max_executions` — override default 1000 for integrations with heavy pagination (e.g., 5000) - `program: |` — the CEL expression; must be a YAML literal block scalar ### Do NOT set `data_stream.dataset` in integration packages **Integration packages** (`type: integration`) must **never** include `data_stream.dataset` in `cel.yml.hbs` or define a `data_stream.dataset` manifest var. The framework automatically routes documents to the correct data stream. Setting `data_stream.dataset` overrides this routing and causes documents to land in the wrong index — typically resulting in "0 hits" during system tests. Only **input-type packages** (`type: input`) use `data_stream.dataset` because they have no predefined data streams. ## Data stream manifest configuration The data stream `manifest.yml` defines the CEL input stream and its variables. ### Standard vars every CEL stream should include | Var | Type | Purpose | |-----|------|---------| | `url` | text | API base URL | | `interval` | text | Polling interval (e.g., `5m`) | | `initial_interval` | text | Lookback window on first run (e.g., `24h`) | | `enable_request_tracer` | bool | Enable HTTP request tracing | | `http_client_timeout` | text | Request timeout (e.g., `30s`) | | `proxy_url` | text | HTTP proxy URL | | `ssl` | yaml | TLS configuration | | `tags` | text (multi) | Event tags | | `preserve_original_event` | bool | Keep original event | | `processors` | yaml | Beat processors | Auth-specific vars depend on the API (API key, OAuth client_id/secret/token_url, bearer token, etc.). Declare `enable_request_tracer` in the data stream manifest, not at the input level. Input-level tracing enables logging for all data streams in the policy. ### Package-level vs data-stream-level vars - **Package-level** vars in the root `manifest.yml` under `policy_templates[].inputs[].vars`: shared across streams (e.g., `url`, auth credentials) - **Data-stream-level** vars in `data_stream/<stream>/manifest.yml` under `streams[].vars`: stream-specific (e.g., `interval`, `batch_size`, `initial_interval`) ## Scope of the CEL program The CEL program's responsibility is **data collection only**: 1. **Fetch data** from the API endpoint(s) 2. **Handle pagination** — walk through all pages within a single polling cycle 3. **Manage cursor state** — store timestamps or page tokens in `cursor` so the next polling interval resumes where the last one left off, avoiding re-collection of already-fetched events 4. **Emit raw events** — output `{"message": e.encode_json()}` for each record The CEL program does **not** handle: - **Elasticsearch-level deduplication** — if overlapping time windows cause a few duplicate events to be collected, that is acceptable. The ingest pipeline or Elasticsearch `_id` routing handles dedup at index time, not the CEL program. - **Field mapping or transformation** — the ingest pipeline handles parsing, ECS mapping, and enrichment. - **Filtering by content** — unless the API supports server-side filtering parameters, do not filter events in the CEL program. Emit everything and let the pipeline decide. Do not search the codebase for `_id`, `document_id`, or deduplication patterns. These are not CEL concerns. ## CEL program structure patterns ### Pagination strategy selection | API behavior | Pattern | Key indicators | |---|---|---| | Returns total count + supports offset | Offset pagination | `total_count`, `offset`, `limit` in request/response | | Returns records since a timestamp | Timestamp cursor | Time-range params, no explicit page tokens | | Returns `Link` header with next URL | Link header | `Link: <url>; rel="next"` in response headers | | Returns next-page URL in response body | Next-URL | `next`, `nextLink`, `@odata.nextLink` field in JSON | | GraphQL with `pageInfo` | GraphQL cursor | `hasNextPage`, `endCursor` in `pageInfo` object | | Multi-phase subscription/content flow | Multi-step state machine | Multiple API calls with work queues in state | **Cursor timestamp selection** — use the **last** record's timestamp when the API sorts ascending; **first** when descending; `max()` with a regression guard when sort order is not guaranteed. Full code, package references, and YAML snippets for each pattern: `references/cel-pagination-patterns.md`. ## Authentication patterns Three strategies: **header** (credentials in `state:`, passed via `Header` map), **query parameter** (credentials appended to URL via `.format_query()`), **signed query** (HMAC signature computed in CEL). Config-level `auth.oauth2`/`auth.digest`/`auth.aws` applies to all requests including `.do_request()`; `auth.basic`/`auth.token` applies only to direct calls (`get()`, `post()`). Prefer input-level auth over in-program token fetching. For full code examples, optional-header syntax, and config-level auth scope details: load `references/cel-auth-patterns.md`. ## State management rules 1. **`state.url`** is populated from `resource.url` config; must be preserved in output or hardcoded 2. **`cursor`** is the only state persisted across input restarts; store pagination positions and timestamps here 3. **`events`** array is removed after each evaluation; never rely on it in subsequent runs 4. **`want_more: true`** triggers immediate re-evaluation, but only if `events` is non-empty. **Pagination continuation guardrail:** when a next-page cursor/token exists, always set `want_more: true` regardless of how many events were collected on the current page. Tying `want_more` to `size(events) > 0` stalls pagination silently — the next cursor is valid, and an empty `events` array is safe to emit. The correct pattern is `"want_more": next_cursor != ""`. 5. **All other state keys** are retained within a session but lost on restart — use `state.with()` to propagate them automatically 6. **Numbers** are serialized as floats in state JSON; cast with `int()` when using as integers 7. **Optional access** with `state.?cursor.last_timestamp.orValue(default)` prevents errors when cursor is absent 8. **Secrets** — every sensitive field in `state` must have a corresponding `redact` entry. `state.secret` is always redacted automatically. When `secret_state` is available (v8.19.14 / v9.2.8 / v9.3.3 / v9.4.0), prefer it. 9. **Cursor updates require a published event** — the input only persists cursor updates when at least one event is published. If a program updates the cursor but returns zero events, the cursor change is lost. 10. **Do not duplicate request/response handling across branches** — when an initialization branch (cursor creation, subscription, token exchange) and a steady-state branch both need the same fetch logic, consolidate it. Two approaches: split the init into a separate evaluation via `want_more: true` (Technique 6 Variant A), or use an intermediate result map to unify the branches within one evaluation (Variant B). Both are valid — see `references/cel-code-style.md` Technique 6 and the init-then-steady-state pattern in `references/cel-pagination-patterns.md`. 11. **Nesting depth** — `.as()` chain depth must not exceed 5 levels on any execution path. HTTP programs must target 2 levels inside `state.with()` (`resp` + `body`). Cursor defaults, window bounds, and page tokens must be extracted as pre-bindings *before* `state.with()`. Single-use values such as `int(state.batch_size)` must be inlined at the call site, not wrapped in `.as()`. Load `references/cel-code-style.md` for flattening techniques and before/after examples. ### Map merge and field removal `with()`, `with_replace()`, `with_update()`, and `drop()` are general-purpose map operations — they work on any map, not just state or cursor. `with()` does a **shallow merge**: nested objects are replaced entirely. This makes it a tool for cursor state transitions (omitting a sub-object removes it via clobber) as well as for building request headers, transforming response data, and constructing intermediate maps. Full semantics and examples: `references/cel-code-style.md`. ## Event output format **Events must contain ONLY `"message"`** — `{"message": e.encode_json()}`. Do not set `@timestamp` or any other field; the framework adds `@timestamp`, and duplicates cause silent document rejection in ES 9.x. See `references/cel-idioms.md` for correct/incorrect examples. ## Response handling **`resp.Body.decode_json()`** — the `bytes(resp.Body)` wrapper was required in older runtime versions but is no longer needed. Use `resp.Body.decode_json()` directly. ## Error handling Every CEL program must handle HTTP errors. Two forms: - **Single-object error (retry):** `"events": {"error": {...}}` — logs at ERROR, sets degraded status, **deletes the cursor** so the next evaluation retries. Use when data was not collected. - **Array error (advance):** `"events": [{"error": {...}}]` — cursor **is** updated. Use when the program should advance past the error. Requires a `terminate` processor in the ingest pipeline (ES 8.16.0+). Error message format: `"METHOD path: body-or-status"`. Code examples: `references/cel-idioms.md`. ### Placeholder events