--- name: cribl-stream description: "Build MCP servers, author routes/pipelines/functions, and manage all Cribl Stream artifacts (Stream, Edge, Search, Lake) via REST API and YAML configs. Use when user mentions Cribl, Cribl Stream/Edge/Search/Lake, routes, pipelines, sources, destinations, packs, or observability pipeline management." metadata: author: kundeng version: "1.0.0" source: "Cribl Stream Documentation Manual v4.17 (generated 2026-04-22)" --- # Cribl Stream Skill You are an expert in Cribl Stream — the observability pipeline platform for routing, transforming, and delivering machine data. You help users build MCP servers that manage Cribl Stream artifacts and author configurations programmatically. ## Online Documentation — Fetch When Needed This skill captures the v4.17.1 docs offline, but versions, endpoint shapes, and new artifact types evolve. **When the local reference here is incomplete, ambiguous, or you suspect it's stale, fetch the canonical online docs.** ### Authoritative URLs (use WebFetch or the docs MCP) | Resource | URL | |----------|-----| | Cribl Stream docs (current) | https://docs.cribl.io/stream/ | | Cribl Stream API Reference (Swagger) | https://docs.cribl.io/api-reference/ | | Cribl Stream OpenAPI spec | https://docs.cribl.io/stream/api-reference/ | | Cribl Edge docs | https://docs.cribl.io/edge/ | | Cribl Search docs | https://docs.cribl.io/search/ | | Cribl Lake docs | https://docs.cribl.io/lake/ | | Cribl.Cloud docs | https://docs.cribl.io/cloud/ | | Routes deep-dive | https://docs.cribl.io/stream/routes/ | | Pipelines deep-dive | https://docs.cribl.io/stream/pipelines/ | | Functions reference | https://docs.cribl.io/stream/functions/ | | Sources reference | https://docs.cribl.io/stream/sources/ | | Destinations reference | https://docs.cribl.io/stream/destinations/ | | Packs Dispensary | https://packs.cribl.io/ | | GitOps & Version Control | https://docs.cribl.io/stream/gitops/ | | REST/API Endpoint Collector | https://docs.cribl.io/stream/collectors-rest/ | | CLI reference | https://docs.cribl.io/stream/cli-reference/ | | Config files reference | https://docs.cribl.io/stream/config-files/ | | Cribl Expressions (C.*) | https://docs.cribl.io/stream/cribl-expressions/ | | Knowledge Objects | https://docs.cribl.io/stream/knowledge-objects/ | | Worker Groups | https://docs.cribl.io/stream/worker-groups/ | | Notifications | https://docs.cribl.io/stream/notifications/ | | Community Slack | https://cribl.io/community/ | | GitHub (Cribl org) | https://github.com/criblio | | Cribl Outpost | https://docs.cribl.io/stream/outpost/ | ### When to fetch online - **Building/calling endpoints not listed below** → fetch the API Reference for exact request/response shape - **Configuring a Source/Destination type with vendor-specific fields** (e.g., a new SaaS connector, OAuth quirks) → fetch the per-destination docs - **User is on a different Cribl version** (3.x, newer than 4.17, or Cribl.Cloud) → confirm endpoint paths haven't changed - **A YAML field name doesn't match what you remember** → fetch the matching config file reference page - **Building a Pack for the Dispensary** → fetch packs.cribl.io for current publication standards Prefer Context7 MCP (`resolve-library-id` for `cribl`) when available, otherwise use `WebFetch` directly. Don't guess endpoint paths or field names — fetch and confirm. ## When to Use This Skill - User wants to build an MCP server for Cribl Stream management - User needs to author/modify routes, pipelines, functions, sources, destinations, packs - User asks about the Cribl Stream REST API - User needs to automate Cribl Stream configuration via code - User asks about Cribl config file formats (YAML) --- ## Cribl Stream Architecture Overview Cribl Stream processes machine data in real time. The data flow is: ``` Sources → [Pre-Processing Pipeline] → Routes → Processing Pipelines → [Post-Processing Pipeline] → Destinations ``` ### Core Concepts - **Sources** (inputs.yml): Data ingestion points (Syslog, Kafka, S3, Splunk HEC, HTTP, etc.) - **Routes** (pipelines/route.yml): Filter expressions that select events and direct them to Pipelines. Evaluated top-down. Each route has a filter (JS expression), a pipeline, a destination, and a Final toggle. - **Pipelines** (pipelines//conf.yml): Ordered sequences of Functions that transform events. - **Functions**: Individual processing steps within pipelines (Eval, Mask, Drop, Lookup, Parser, Regex Extract, etc.) - **Destinations** (outputs.yml): Where processed data is sent (Splunk, S3, Elasticsearch, Kafka, Syslog, etc.) - **Packs**: Self-contained bundles of routes, pipelines, sources, destinations, and knowledge objects. - **Knowledge Objects**: Reusable config — Variables, Lookups, Event Breakers, Regexes, Parsers, Schemas, Grok Patterns. - **Worker Groups**: In distributed deployments, groups of worker nodes with shared configuration. --- ## Cribl Stream REST API Reference ### Authentication ```bash # Login — returns a Bearer token POST /api/v1/auth/login Content-Type: application/json Body: {"username": "", "password": ""} # Response: {"token": "..."} # Use the token in subsequent requests: Authorization: Bearer ``` The API server listens on port **9000** by default (configurable in cribl.yml). Token TTL defaults to 3600 seconds (configurable via `api.authTokenTTL` in cribl.yml). ### API URL Patterns Base URL: `http://:9000/api/v1` For **distributed deployments** (Worker Groups): - `GET /api/v1/m//routes` — routes for a specific worker group - `GET /api/v1/m//pipelines` — pipelines for a group - `GET /api/v1/m//inputs` — sources for a group - `GET /api/v1/m//outputs` — destinations for a group For **single-instance** deployments, omit the `/m/` segment. ### Key API Endpoints #### Routes ``` GET /api/v1/routes — List all routes GET /api/v1/routes/ — Get a specific route POST /api/v1/routes — Create a route PATCH /api/v1/routes/ — Update a route DELETE /api/v1/routes/ — Delete a route ``` #### Pipelines ``` GET /api/v1/pipelines — List all pipelines GET /api/v1/pipelines/ — Get a specific pipeline POST /api/v1/pipelines — Create a pipeline PATCH /api/v1/pipelines/ — Update a pipeline DELETE /api/v1/pipelines/ — Delete a pipeline ``` #### Sources (Inputs) ``` GET /api/v1/inputs — List all sources GET /api/v1/inputs/ — Get a specific source POST /api/v1/inputs — Create a source PATCH /api/v1/inputs/ — Update a source DELETE /api/v1/inputs/ — Delete a source ``` #### Destinations (Outputs) ``` GET /api/v1/outputs — List all destinations GET /api/v1/outputs/ — Get a specific destination POST /api/v1/outputs — Create a destination PATCH /api/v1/outputs/ — Update a destination DELETE /api/v1/outputs/ — Delete a destination ``` #### Packs ``` GET /api/v1/packs — List all packs POST /api/v1/packs — Install a pack DELETE /api/v1/packs/ — Remove a pack ``` #### Knowledge Objects ``` GET /api/v1/lib/vars — List global variables POST /api/v1/lib/vars — Create a variable PATCH /api/v1/lib/vars/ — Update a variable DELETE /api/v1/lib/vars/ — Delete a variable GET /api/v1/lib/lookups — List lookups GET /api/v1/lib/regexes — List regexes GET /api/v1/lib/breakers — List event breakers GET /api/v1/lib/parsers — List parsers GET /api/v1/lib/schemas — List JSON schemas GET /api/v1/lib/parquet-schemas — List Parquet schemas ``` #### Version Control & Deployment ``` POST /api/v1/version/commit — Commit changes POST /api/v1/version/deploy — Deploy committed changes POST /api/v1/version/sync — Sync from Git remote (GitOps) GET /api/v1/version/info — Get version info ``` #### System ``` GET /api/v1/system/info — System information GET /api/v1/system/metrics — System metrics GET /api/v1/system/logs — System logs POST /api/v1/system/restart — Restart Cribl Stream POST /api/v1/system/reload — Reload configuration GET /api/v1/health — Health check ``` #### Workers (Distributed) ``` GET /api/v1/master/groups — List worker groups GET /api/v1/master/workers — List all workers GET /api/v1/master/groups//workers — Workers in a group ``` #### Jobs (Collectors) ``` GET /api/v1/jobs — List jobs POST /api/v1/jobs — Create/run a job GET /api/v1/jobs/ — Get job status DELETE /api/v1/jobs/ — Cancel a job ``` #### Notifications ``` GET /api/v1/notifications — List notifications POST /api/v1/notifications — Create a notification ``` #### Secrets ``` GET /api/v1/secrets — List secrets POST /api/v1/secrets — Create a secret ``` --- ## Configuration File Formats All config files are YAML. Located at `$CRIBL_HOME/local/cribl/` (single-instance) or `$CRIBL_HOME/groups//local/cribl/` (distributed). ### Route Configuration (pipelines/route.yml) ```yaml id: default groups: {} comments: [] routes: - id: my_route_name name: my_route_name filter: "source=='syslog' && severity<=3" pipeline: my_pipeline output: splunk_hec final: true disabled: false description: "Route critical syslog to Splunk" - id: catch_all name: catch_all filter: "true" pipeline: passthru output: default final: true ``` **Route fields:** - `id` / `name`: Unique identifier (case-sensitive) - `filter`: JavaScript expression evaluated per event. `true` matches all events. - `pipeline`: Pipeline ID or Pack reference to process matched events - `output`: Destination ID to send processed events to - `final`: When true (default), matched events stop at this route. When false, events continue to next route (cloning behavior). - `disabled`: Toggle route on/off - `description`: Optional description - `cloneFields`: When final=false, additional fields to add to cloned events ### Pipeline Configuration (pipelines//conf.yml) ```yaml output: default groups: {} asyncFuncTimeout: 1000 functions: - id: eval filter: "true" disabled: false conf: add: - name: datacenter value: "'us-east-1'" remove: - date_* - id: mask filter: "sourcetype=='access_combined'" disabled: false conf: rules: - matchRegex: "/(\\d{3}-\\d{2}-)\\d{4}/g" replaceExpr: "`${g1}XXXX`" fields: - _raw - id: drop filter: "severity > 6" disabled: false final: true ``` ### Source Configuration (inputs.yml) ```yaml inputs: syslog_input: type: syslog host: "0.0.0.0" port: 514 protocol: udp sendToRoutes: true pipeline: pre_process_syslog disabled: false metadata: - name: source value: "'syslog'" breakerRulesets: - Syslog splunk_hec_input: type: splunk_hec host: "0.0.0.0" port: 8088 token: "my-hec-token" sendToRoutes: true disabled: false http_input: type: http host: "0.0.0.0" port: 10080 sendToRoutes: true ``` ### Destination Configuration (outputs.yml) ```yaml outputs: splunk_hec: type: splunk_hec url: "https://splunk.example.com:8088/services/collector/event" token: "my-hec-token" compress: true onBackpressure: block pipeline: post_process_splunk s3_archive: type: s3 bucket: "my-archive-bucket" region: "us-east-1" awsAuthenticationMethod: auto stagePath: "/tmp/cribl/s3" compress: gzip partitionExpr: "`${sourcetype}/${_time.strftime('%Y/%m/%d')}`" devnull: type: devnull default: type: default defaultId: splunk_hec webhook_out: type: webhook url: "https://webhook.example.com/events" method: POST format: json authType: token token: "bearer-token-here" syslog_out: type: syslog host: "syslog.example.com" port: 514 protocol: tcp facility: 1 severity: 6 messageFormat: rfc5424 elasticsearch_out: type: elastic url: "https://es.example.com:9200" index: "logs-cribl" docTypeExpr: "'_doc'" ``` ### Global Variables (vars.yml) ```yaml variables: my_threshold: type: number value: 42 description: "Event threshold for alerting" api_key: type: encrypted_string value: "encrypted_value_here" description: "External API key" blocked_hosts: type: array value: '["bad-host-1", "test-server-99"]' ``` Reference in expressions: `C.vars.my_threshold`, `C.vars.blocked_hosts.includes(host)` ### Event Breakers (breakers.yml) ```yaml id: my_breaker_ruleset lib: custom description: "Custom event breakers" rules: - name: json_events condition: "/^\\{/" type: json timestampAnchorRegex: "/\"timestamp\"\\s*:\\s*/" timestamp: type: auto - name: syslog_events condition: "/^<\\d+>/" type: regex timestampAnchorRegex: "/^/" eventBreakerRegex: "/[\\r\\n]+/" ``` --- ## Available Functions (Pipeline Processing Steps) | Function | Purpose | |----------|---------| | **Eval** | Add, modify, or remove event fields | | **Mask** | Regex-based pattern masking/redaction (supports C.Mask.random, C.Mask.md5, C.Mask.sha256, C.Mask.REDACTED, etc.) | | **Drop** | Drop events matching a filter | | **Sampling** | Sample events at a configurable rate | | **Dynamic Sampling** | Sample based on field cardinality | | **Regex Extract** | Extract fields using regex capture groups | | **Regex Filter** | Filter events based on regex patterns | | **Parser** | Parse structured data (JSON, KV, CSV, etc.) | | **Serialize** | Serialize events to a specific format | | **Lookup** | Enrich events from lookup tables | | **GeoIP** | Add geographic data from IP addresses | | **DNS Lookup** | Resolve hostnames/IPs | | **Clone** | Duplicate events with modifications | | **Drop Dimensions** | Remove metric dimensions to reduce cardinality | | **Flatten** | Flatten nested objects | | **Fold Keys** | Fold object keys matching patterns | | **Rename** | Rename fields | | **Unroll** | Expand array fields into separate events | | **JSON Unroll** | Unroll JSON arrays | | **XML Unroll** | Unroll XML elements | | **Aggregations** | Aggregate events over time windows | | **Aggregate Metrics** | Aggregate metric events | | **Publish Metrics** | Convert events to metrics format | | **Rollup Metrics** | Roll up metric granularity | | **Suppress** | Suppress duplicate events | | **Chain** | Send events to another pipeline | | **Code** | Run custom JavaScript code on events | | **Tee** | Copy events to a secondary destination | | **Auto Timestamp** | Automatically detect and normalize timestamps | | **CEF Serializer** | Serialize events to CEF format | | **Grok** | Parse events using Grok patterns | | **Guard** | Apply Cribl Guard (DLP) rules | | **Numerify** | Convert string fields to numbers | | **SNMP Trap Serialize** | Serialize SNMP trap events | | **OTLP Logs/Metrics/Traces** | Convert to/from OpenTelemetry format | | **Event Breaker** | Re-break events within a pipeline | | **Comment** | Add documentation comments (no processing) | ### Function Configuration Pattern Every function has: - `id`: Function type (e.g., "eval", "mask", "drop") - `filter`: JS expression determining which events this function processes (default: `"true"`) - `disabled`: Boolean toggle - `final`: When true, matched events don't continue to next function - `conf`: Function-specific configuration --- ## Cribl Expressions (C.* helpers) Use in any JavaScript expression field: ```javascript // Variables C.vars.myVariable C.vars.epochTime() // expression-type variable // Lookups C.Lookup('lookup_file.csv', 'key_field') C.Lookup('lookup_file.csv', 'key_field').match(source_ip, 'result_field') // Masking C.Mask.random(length) // random alphanumeric C.Mask.repeat(length, char) // repeating character C.Mask.md5(value) // MD5 hash C.Mask.sha1(value) // SHA1 hash C.Mask.sha256(value) // SHA256 hash C.Mask.REDACTED // literal "REDACTED" // Encryption C.Crypto.encrypt(value, keyId, algorithm) C.Crypto.decrypt(value, keyId) // Encoding C.Encode.uri(value) C.Encode.base64(value) C.Decode.base64(value) C.Decode.uri(value) // Secrets C.Secret('secretName', 'type').value C.Secret('api_key', 'keypair').apiKey // Time C.Time.now() C.Time.strftime(epoch, format) C.Time.strptime(string, format) // Network C.Net.cidrMatch(ip, cidr) // Text C.Text.hashCode(value) ``` --- ## Building an MCP Server for Cribl Stream When building an MCP server, implement tools for each artifact type. Here is the recommended tool structure: ### Required MCP Tools ```typescript // Authentication cribl_login(host, username, password) → token // Routes cribl_list_routes(group?) cribl_get_route(id, group?) cribl_create_route(id, filter, pipeline, output, final?, group?) cribl_update_route(id, updates, group?) cribl_delete_route(id, group?) cribl_reorder_routes(routeIds[], group?) // Pipelines cribl_list_pipelines(group?) cribl_get_pipeline(id, group?) cribl_create_pipeline(id, functions[], group?) cribl_update_pipeline(id, updates, group?) cribl_delete_pipeline(id, group?) cribl_add_function(pipelineId, function, position?, group?) // Sources cribl_list_sources(group?) cribl_get_source(id, group?) cribl_create_source(id, type, config, group?) cribl_update_source(id, updates, group?) cribl_delete_source(id, group?) // Destinations cribl_list_destinations(group?) cribl_get_destination(id, group?) cribl_create_destination(id, type, config, group?) cribl_update_destination(id, updates, group?) cribl_delete_destination(id, group?) // Knowledge Objects cribl_list_variables(group?) cribl_create_variable(id, type, value, description?, group?) cribl_list_lookups(group?) cribl_update_lookup(id, file_content, group?) cribl_list_event_breakers(group?) // Packs cribl_list_packs(group?) cribl_install_pack(source, group?) cribl_export_pack(id, group?) // Version Control cribl_commit(message, group?) cribl_deploy(group?) cribl_sync() // System cribl_health() cribl_system_info() cribl_system_metrics() cribl_restart(group?) cribl_reload(group?) // Workers (distributed) cribl_list_groups() cribl_list_workers(group?) // Jobs cribl_list_jobs(group?) cribl_run_job(collectorId, config?, group?) cribl_get_job_status(jobId, group?) ``` ### MCP Server Implementation Pattern (TypeScript) ```typescript import { McpServer } from "@anthropic-ai/mcp"; const CRIBL_HOST = process.env.CRIBL_HOST || "http://localhost:9000"; let authToken: string | null = null; async function criblApi(method: string, path: string, body?: any) { const url = `${CRIBL_HOST}/api/v1${path}`; const res = await fetch(url, { method, headers: { "Content-Type": "application/json", ...(authToken ? { Authorization: `Bearer ${authToken}` } : {}), }, body: body ? JSON.stringify(body) : undefined, }); if (!res.ok) throw new Error(`Cribl API ${method} ${path}: ${res.status} ${await res.text()}`); return res.json(); } async function login(username: string, password: string) { const result = await criblApi("POST", "/auth/login", { username, password }); authToken = result.token; return result; } function groupPrefix(group?: string) { return group ? `/m/${group}` : ""; } // Example tool implementations: // List routes async function listRoutes(group?: string) { return criblApi("GET", `${groupPrefix(group)}/routes`); } // Create a route async function createRoute(route: { id: string; filter: string; pipeline: string; output: string; final?: boolean; description?: string; }, group?: string) { return criblApi("POST", `${groupPrefix(group)}/routes`, { ...route, final: route.final ?? true, }); } // Create a pipeline with functions async function createPipeline(id: string, functions: any[], group?: string) { return criblApi("POST", `${groupPrefix(group)}/pipelines`, { id, conf: { functions }, }); } // Commit and deploy async function commitAndDeploy(message: string, group?: string) { await criblApi("POST", `${groupPrefix(group)}/version/commit`, { message }); await criblApi("POST", `${groupPrefix(group)}/version/deploy`); } ``` ### Environment Variables ```bash CRIBL_HOST=http://localhost:9000 # API host CRIBL_USERNAME=admin # Login username CRIBL_PASSWORD= # Login password CRIBL_API_PORT=9000 # API port (default 9000) ``` --- ## Common Source Types | Type | `type` value | Key Config | |------|-------------|------------| | Syslog | `syslog` | host, port, protocol (tcp/udp) | | Splunk HEC | `splunk_hec` | host, port, token | | Splunk S2S | `splunk` | host, port | | HTTP/S | `http` | host, port | | Kafka | `kafka` | brokers[], topics[], groupId | | Amazon S3 | `s3` | bucket, region, queueName | | Amazon SQS | `sqs` | queueName, region | | Amazon Kinesis | `kinesis` | streamName, region | | Azure Event Hubs | `azure_eh` | connectionString, consumerGroup | | Google Pub/Sub | `google_pubsub` | projectId, subscriptionName | | TCP JSON | `tcp_json` | host, port | | Raw TCP | `raw_tcp` | host, port | | Raw UDP | `raw_udp` | host, port | | File Monitor | `file` | path, mode | | Windows Events | `windows_event_logs` | channels[] | | Elasticsearch API | `elastic` | url, index | | Prometheus Remote Write | `prometheus_rw` | host, port | | Grafana | `grafana` | host, port | | Office 365 | `office365_activity` | tenantId, clientId | | REST Collector | `collection` | various | | Script Collector | `script` | command, args | ## Common Destination Types | Type | `type` value | Key Config | |------|-------------|------------| | Splunk HEC | `splunk_hec` | url, token | | Splunk S2S | `splunk` | host, port | | Splunk Load Balanced | `splunk_lb` | hosts[], port | | Amazon S3 | `s3` | bucket, region, stagePath | | Elasticsearch | `elastic` | url, index | | Kafka | `kafka` | brokers[], topic | | Syslog | `syslog` | host, port, protocol, facility | | Webhook | `webhook` | url, method, format | | Azure Sentinel | `sentinel` | loginUrl, client_id | | Google Cloud Storage | `google_cloud_storage` | bucket, region | | Datadog | `datadog` | url, apiKey | | New Relic | `newrelic` | url, licenseKey | | Cribl HTTP | `cribl_http` | url | | Cribl TCP | `cribl_tcp` | host, port | | DevNull | `devnull` | (none) | | Default | `default` | defaultId | | Output Router | `output_router` | rules[] | | Filesystem/NFS | `filesystem` | path | | InfluxDB | `influxdb` | url, bucket, org, token | | Prometheus | `prometheus` | url | | Loki | `loki` | url | | MinIO | `minio` | bucket, endpoint | --- ## Important Operational Notes ### Changes That Require Reload (fast, no data loss) - Functions, Pipelines, Packs, Routes - Lookups, Parquet schemas, Global variables - Group Settings > Limits, Logging levels ### Changes That Require Restart (brief interruption) - Sources, Destinations - Event Breakers, QuickConnect configs - Distributed mode changes, Worker Group assignment - TLS settings, Worker process count ### Route Evaluation Rules 1. Routes evaluate **top-down** in display order 2. **Final=true** (default): matched events stop, non-matched continue 3. **Final=false**: ALL events continue (matched events become clones) 4. Always use a catch-all route at the bottom (`filter: "true"`) 5. Design for early reduction — broad filters first to minimize processing ### Pipeline Best Practices - Use function filters to narrow scope (`source=='foo'`) - Don't overload pipelines — separate pre/post-processing - Use Eval for field manipulation, Parser for full parsing - Use `__e['field-name']` for fields with special characters ### Filter Expression Examples ```javascript true // match all source=='syslog' // exact match sourcetype=='access_combined' && status>=400 // compound host.startsWith('web-') // string method C.Net.cidrMatch(src_ip, '10.0.0.0/8') // CIDR match severity <= 3 // numeric comparison C.vars.blocked_hosts.includes(host) // array lookup /error|warn/i.test(_raw) // regex test __e['user-agent'].includes('bot') // special char fields Math.random() < 0.1 // 10% sampling ``` --- ## CLI Reference (Key Commands) ```bash cribl start # Start Cribl Stream cribl stop # Stop cribl restart # Restart cribl reload # Reload config (no restart) cribl status # Show status cribl version # Show version cribl mode-single # Switch to single-instance mode cribl mode-master # Switch to leader mode cribl mode-worker # Switch to worker mode # Auth cribl auth login -H http://host:9000 -u admin -p pass # Variables cribl vars add -i myVar -v 42 -t number -d "description" cribl vars get cribl vars update -i myVar -v 100 cribl vars remove -i myVar # Pipelines cat data.log | cribl pipe -p pipelineName # Packs cribl pack install cribl pack list cribl pack export ```