# Examples Reference > Auto-generated from docs.expanso.io documentation ## Pipeline Examples # Pipeline Examples Browse simple, practical pipeline examples to help you get started with Expanso Edge. These examples demonstrate common patterns and best practices. ## Quick Start with Examples All pipeline examples are available as downloadable YAML files. You can use them directly with `curl`: ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/basic-pipeline.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: basic-pipeline\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. Each example page below includes specific download and usage instructions. --- ## Getting Started Examples Perfect for learning the basics: ### [Basic Pipeline](/examples/basic-pipeline) A simple log processing pipeline that reads from a file, parses JSON, and outputs to stdout. **Start here** if you're new to Expanso. **What you'll learn:** - Basic pipeline structure - Reading from files - JSON parsing - Simple transformations --- ### [HTTP to File](/examples/http-to-file) Receive HTTP webhooks and write them to files. Shows how to set up an HTTP server input and file output. **What you'll learn:** - HTTP server setup - Receiving webhooks - Writing to files - Path templating with timestamps --- ### [Data Transformer](/examples/data-transformer) Transform JSON data using Bloblang. Demonstrates field mapping, filtering, and data enrichment. **What you'll learn:** - Bloblang transformation language - Field mapping and renaming - Filtering and conditional logic - Data enrichment patterns --- ### [Metrics Collector](/examples/metrics-collector) Collect system metrics and send them to a monitoring service. Shows time-series data handling. **What you'll learn:** - Periodic data generation - Metrics collection - Batching for performance - Sending to monitoring services --- ## Advanced Pattern Examples ### [Log Processing Patterns](/examples/log-processing) Multiple examples showing log aggregation, filtering, routing, and analysis patterns. **Patterns covered:** - Multi-source log aggregation - Severity-based routing - PII masking and redaction - Log enrichment with metadata --- ### [Data Routing](/examples/data-routing) Examples of routing data to multiple destinations based on content, patterns, and conditions. **Patterns covered:** - Content-based routing - Fan-out to multiple destinations - Switch/case routing logic - Fallback and retry patterns --- ### [Data Security](/examples/data-security) Security-focused examples including encryption, PII masking, and data redaction. **Patterns covered:** - PII detection and masking - Field-level encryption - Sensitive data redaction - Compliance patterns --- ## Need More Complex Examples? These documentation examples are designed to be simple and easy to understand. **For production-ready, complex examples** with complete deployment configurations, visit:

πŸš€ Production Examples & Templates

Explore complete, production-ready pipeline examples with deployment configurations, monitoring setup, and best practices.

Visit examples.expanso.io β†’

Includes: Kubernetes deployments, CI/CD pipelines, multi-cloud examples, and more

--- ## What You'll Find at examples.expanso.io **Production Deployments:** - Complete Kubernetes manifests - Helm charts and operators - Docker Compose configurations - Terraform/CloudFormation templates **Complex Integrations:** - Multi-stage data processing pipelines - Microservices communication patterns - Event-driven architectures - Real-time analytics systems - Splunk edge processing (HEC routing, cost optimization) **Advanced Use Cases:** - MLOps and model deployment - Time-series data processing - Real-time fraud detection - IoT device management at scale - Multi-region data synchronization - O-RAN Telco telemetry pipelines - Enterprise data migration (DB2, DataStage replacement) **DevOps & Operations:** - CI/CD pipeline examples - Monitoring and alerting setup - High-availability configurations - Disaster recovery patterns --- ## Example Categories Browse by category to find examples relevant to your use case:

πŸ“Š Data Processing

πŸ”Œ Integrations

πŸ›‘οΈ Security & Compliance

πŸ“‘ IoT & Edge

--- ## How to Use These Examples 1. **Start Simple**: Begin with [Basic Pipeline](/examples/basic-pipeline) to understand the fundamentals 2. **Learn Patterns**: Explore specific pattern examples for your use case 3. **Adapt**: Copy and modify examples for your needs 4. **Go Production**: Visit [examples.expanso.io](https://examples.expanso.io) for production-ready configurations --- ## Community Examples Have an example to share? We'd love to feature it! Contribute your examples to help the community: - Email your example to [support@expanso.io](mailto:support@expanso.io) - Share in our [community Slack](https://exso.cloud/slack) - Tag us on X [@ExpansoIO](https://x.com/ExpansoIO) --- ## Next Steps - **[Quick Start Guide](/getting-started/quickstart/)**: Build your first pipeline - **[Components Reference](/components)**: Explore all available components - **[Bloblang Guide](/guides/bloblang)**: Learn the transformation language - **[Use Cases](/use-cases)**: Real-world implementation patterns ## Basic Pipeline Example # Basic Log Processing Pipeline This example demonstrates a simple log processing pipeline that reads from a file, parses JSON logs, and outputs to stdout. It's a great starting point for understanding how Expanso pipelines work. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/basic-pipeline.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: basic-pipeline\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`basic-pipeline.yaml`](pathname:///examples/basic-pipeline.yaml)
## What This Pipeline Does 1. **Reads** log lines from a file 2. **Parses** each line as JSON 3. **Extracts** the timestamp field 4. **Outputs** the processed logs to stdout ## Complete Pipeline ```yaml input: file: paths: [/var/log/app.log] codec: lines pipeline: processors: - mapping: | # Parse the log line as JSON root = this.parse_json() - mapping: | # Parse timestamp field if it exists root.timestamp = this.timestamp.parse_timestamp("2006-01-02T15:04:05Z") output: stdout: codec: lines ``` ## Configuration Breakdown ### Input: File ```yaml input: file: paths: [/var/log/app.log] codec: lines ``` The `file` input reads from `/var/log/app.log`, processing each line individually. The `lines` codec tells Expanso to treat each line as a separate message. **See:** [File Input Documentation](/components/inputs/file) ### Processors: Parse & Transform ```yaml pipeline: processors: - mapping: | # Parse the log line as JSON root = this.parse_json() - mapping: | # Parse timestamp field if it exists root.timestamp = this.timestamp.parse_timestamp("2006-01-02T15:04:05Z") ``` The pipeline uses two `mapping` processors (which use the Bloblang language): 1. **JSON Parsing**: Converts each log line from a string to a JSON object 2. **Timestamp Parsing**: Converts the timestamp string to a proper timestamp type **See:** [Mapping Processor](/components/processors/mapping) | [Bloblang Guide](/guides/bloblang) ### Output: Stdout ```yaml output: stdout: codec: lines ``` Outputs each processed message to stdout in JSON format. Useful for testing and debugging. **See:** [Stdout Output Documentation](/components/outputs/stdout) ## Example Input/Output **Input Log Line:** ```json {"timestamp":"2024-01-15T10:30:00Z","level":"INFO","message":"User login successful","user_id":"12345"} ``` **Output:** ```json { "timestamp": "2024-01-15T10:30:00Z", "level": "INFO", "message": "User login successful", "user_id": "12345" } ``` ## Common Variations ### Add Filtering Filter out debug-level logs: ```yaml pipeline: processors: - mapping: | root = this.parse_json() # Drop debug logs root = if this.level == "DEBUG" { deleted() } ``` ### Output to File Change the output to write to a file instead of stdout: ```yaml output: file: path: /var/log/processed-${! timestamp_unix() }.json codec: lines ``` ### Add Timestamps Add a processing timestamp: ```yaml pipeline: processors: - mapping: | root = this.parse_json() root.processed_at = now() ``` ## Next Steps - **[File Input](/components/inputs/file)** - Read from files and streams - **[Mapping Processor](/components/processors/mapping)** - Transform data with Bloblang - **[Bloblang Guide](/guides/bloblang)** - Data transformation reference - **[Log Processing Example](/examples/log-processing)** - Advanced log processing patterns ## Data Routing & Distribution # Data Routing & Distribution Route data intelligently based on content, priority, or destination requirements. These guides show you how to send data to multiple systems, implement priority queues, protect downstream services, and split messages for different processing paths. ## Guides in This Category ### [Send Data to Multiple Destinations](https://examples.expanso.io/data-routing/fan-out-pattern/) **Problem:** You need to send the same data to multiple systems simultaneouslyβ€”real-time processing AND long-term storage. **Solution:** Use broker output with fan_out pattern to send data to Kafka, S3, and Elasticsearch with independent retry logic per destination. **Key Benefits:** Real-time + batch processing, independent failure handling, parallel delivery **Read the guide:** [Send Data to Multiple Destinations β†’](https://examples.expanso.io/data-routing/fan-out-pattern/) --- ### [Route Data Based on Content](https://examples.expanso.io/data-routing/content-routing/) **Problem:** Different types of data need different handlingβ€”critical errors to one destination, info logs to another. **Solution:** Use switch processor for conditional routing based on severity, region, event type, or any field value. **Key Benefits:** 66% bandwidth reduction, intelligent filtering, priority handling **Read the guide:** [Route Data Based on Content β†’](https://examples.expanso.io/data-routing/content-routing/) --- ### [Implement Content-Based Splitting](https://examples.expanso.io/data-routing/content-splitting/) **Problem:** Batch messages or arrays need to be split into individual events for per-message processing. **Solution:** Use split processor to break JSON arrays, CSV batches, or multiline logs into individual messages. **Key Benefits:** Enable per-message routing, parallel processing, granular error handling **Read the guide:** [Implement Content-Based Splitting β†’](https://examples.expanso.io/data-routing/content-splitting/) --- ### [Create Priority Queues](https://examples.expanso.io/data-routing/priority-queues/) **Problem:** Critical messages get stuck behind low-priority data, violating SLAs. **Solution:** Route messages to separate queues based on priority levels. **Key Benefits:** SLA compliance, fair scheduling, prevent starvation, 93% cost reduction for high-priority data **Read the guide:** [Create Priority Queues β†’](https://examples.expanso.io/data-routing/priority-queues/) --- ### [Implement Circuit Breakers](https://examples.expanso.io/data-routing/circuit-breakers/) **Problem:** Failing downstream systems cause cascading failures and resource exhaustion. **Solution:** Implement circuit breaker pattern with retry limits, fallback destinations, and automatic recovery. **Key Benefits:** Prevent cascading failures, automatic recovery, resource protection, graceful degradation **Read the guide:** [Implement Circuit Breakers β†’](https://examples.expanso.io/data-routing/circuit-breakers/) --- ## Common Routing Patterns ### Pattern 1: Fan-Out (One-to-Many) ``` β”Œβ”€β†’ Kafka (real-time) Input ───┼─→ S3 (archive) └─→ Elasticsearch (search) ``` **Use when:** Same data needs multiple destinations **Guide:** [Send Data to Multiple Destinations](https://examples.expanso.io/data-routing/fan-out-pattern/) --- ### Pattern 2: Content-Based Routing ``` β”Œβ”€β†’ [severity=error] β†’ Kafka Topic: errors Input ───┼─→ [severity=warn] β†’ Kafka Topic: warnings └─→ [severity=info] β†’ Local File ``` **Use when:** Different data types need different handling **Guide:** [Route Data Based on Content](https://examples.expanso.io/data-routing/content-routing/) --- ### Pattern 3: Priority Queuing ``` β”Œβ”€β†’ [priority=critical] β†’ Fast Queue (10 workers) Input ───┼─→ [priority=high] β†’ Medium Queue (5 workers) └─→ [priority=normal] β†’ Slow Queue (2 workers) ``` **Use when:** SLA requirements vary by message type **Guide:** [Create Priority Queues](https://examples.expanso.io/data-routing/priority-queues/) --- ### Pattern 4: Circuit Breaker with Fallback ``` Input ─→ Circuit Breaker ─┬─→ [Closed] β†’ Destination └─→ [Open] β†’ Dead Letter Queue ``` **Use when:** Downstream systems have availability issues **Guide:** [Implement Circuit Breakers](https://examples.expanso.io/data-routing/circuit-breakers/) --- ### Pattern 5: Split and Route ``` Batch Input ─→ Split ─┬─→ [type=A] β†’ Destination 1 β”œβ”€β†’ [type=B] β†’ Destination 2 └─→ [type=C] β†’ Destination 3 ``` **Use when:** Batches contain different message types **Guide:** [Implement Content-Based Splitting](https://examples.expanso.io/data-routing/content-splitting/) --- ## Data Security & Compliance # Data Security & Compliance Process data securely at the edge while meeting regulatory requirements like GDPR, CCPA, HIPAA, and PCI-DSS. These guides show you how to remove sensitive information, validate data quality, and encrypt fields before data leaves your edge nodes. ## Why Security at the Edge? Processing data at the edge offers unique security advantages: - **Minimize Data Exposure** - PII never leaves the edge network - **Reduce Attack Surface** - Fewer systems handle sensitive data - **Compliance by Design** - Meet regulations before cloud upload - **Audit Trail** - Track data transformations locally - **Bandwidth Savings** - Remove sensitive fields reduces payload size --- ## Guides in This Category ### [Remove PII from Data Streams](https://examples.expanso.io/data-security/remove-pii/) **Problem:** You need to comply with GDPR, CCPA, or other privacy regulations that restrict collecting or transmitting personally identifiable information (PII). **Solution:** Hash IP addresses with SHA-256, anonymize email addresses, remove names and sensitive fields at the edge before data reaches the cloud. **Key Benefits:** - βœ… GDPR/CCPA compliance without cloud processing - βœ… 28% bandwidth savings by removing PII fields - βœ… Irreversible anonymization with cryptographic hashing - βœ… Audit logging for compliance verification **Best For:** Customer data, web analytics, application logs, user behavior tracking **Read the guide:** [Remove PII from Data Streams β†’](https://examples.expanso.io/data-security/remove-pii/) --- ### [Enforce Data Schema Validation](https://examples.expanso.io/data-security/enforce-schema/) **Problem:** Invalid or malformed data breaks downstream systems, causing pipeline failures, analytics errors, and data quality issues. **Solution:** Validate data against JSON Schema at the edge, route failures to a dead letter queue, and maintain schema versions in version control. **Key Benefits:** - βœ… Fail fast at the edge before cloud processing - βœ… Prevent bad data from reaching analytics systems - βœ… Schema evolution with backward compatibility - βœ… Automated testing in CI/CD pipelines **Best For:** API ingestion, IoT sensor data, event streams, third-party integrations **Read the guide:** [Enforce Data Schema Validation β†’](https://examples.expanso.io/data-security/enforce-schema/) --- ### [Encrypt Sensitive Data in Transit](https://examples.expanso.io/data-security/encrypt-data/) **Problem:** Sensitive data like credit cards, SSNs, or health records must be encrypted before transmission to meet compliance requirements. **Solution:** Implement field-level AES-256-GCM encryption at the edge with key rotation, HSM integration, and HMAC signatures for webhooks. **Key Benefits:** - βœ… Field-level encryption protects specific data - βœ… Key rotation with versioning for security - βœ… HSM integration (AWS KMS, Vault) for key management - βœ… HMAC signatures verify webhook authenticity **Best For:** Payment processing, healthcare data (HIPAA), financial services, PCI-DSS compliance **Read the guide:** [Encrypt Sensitive Data in Transit β†’](https://examples.expanso.io/data-security/encrypt-data/) --- ### [Cross-Border GDPR Compliance](https://examples.expanso.io/data-security/cross-border-gdpr/) **Problem:** Your organization needs to transfer EU customer data to global analytics systems, but GDPR Article 44+ restricts personal data transfers outside the EU/EEA. **Solution:** Fully anonymize data at the EU edge before cross-border transfer using tiered treatmentβ€”delete direct identifiers, hash pseudonymous IDs, and generalize quasi-identifiers to achieve k-anonymity. **Key Benefits:** - βœ… GDPR Article 44 compliance for cross-border transfers - βœ… Tiered anonymization (delete/hash/generalize) - βœ… Dual output: anonymized to global, full data stays in-region - βœ… Compliance audit trail with verification metadata **Best For:** Global analytics platforms, EU financial data, multi-region data architectures, data sovereignty requirements **Read the guide:** [Cross-Border GDPR Compliance β†’](https://examples.expanso.io/data-security/cross-border-gdpr/) --- ## Common Security Patterns ### Pattern 1: Remove PII Before Cloud Upload ``` Edge Node: β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ HTTP Input β†’ Remove PII β†’ Hash IPs β†’ Cloud β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` **Use when:** Collecting analytics data with user information **Compliance:** GDPR Article 25 (Privacy by Design), CCPA **Guide:** [Remove PII from Data Streams](https://examples.expanso.io/data-security/remove-pii/) --- ### Pattern 2: Validate Before Processing ``` Edge Node: β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Input β†’ Schema Validation ──┬─→ Valid β†’ Process β”‚ β”‚ └─→ Invalid β†’ DLQ β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` **Use when:** Accepting data from untrusted sources **Compliance:** Data Quality Management, SOC 2 **Guide:** [Enforce Data Schema Validation](https://examples.expanso.io/data-security/enforce-schema/) --- ### Pattern 3: Encrypt Sensitive Fields ``` Edge Node: β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”‚ Input β†’ Encrypt(credit_card, ssn) β†’ Cloud β”‚ β”‚ ↓ β”‚ β”‚ Decrypt only in secure processing zone β”‚ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ ``` **Use when:** Handling payment or health data **Compliance:** PCI-DSS, HIPAA, SOC 2 **Guide:** [Encrypt Sensitive Data in Transit](https://examples.expanso.io/data-security/encrypt-data/) --- ## Regulatory Compliance Reference ### GDPR (General Data Protection Regulation) **Relevant Guides:** - [Remove PII from Data Streams](https://examples.expanso.io/data-security/remove-pii/) - Right to anonymization (Article 17) - [Cross-Border GDPR Compliance](https://examples.expanso.io/data-security/cross-border-gdpr/) - Data transfers outside EU/EEA (Article 44) - [Enforce Data Schema Validation](https://examples.expanso.io/data-security/enforce-schema/) - Data accuracy requirements (Article 5) **Key Requirements:** - Minimize data collection (Article 5.1c) - Anonymize when possible (Recital 26) - Data protection by design (Article 25) - Right to erasure (Article 17) - Cross-border transfer restrictions (Article 44-49) --- ### CCPA (California Consumer Privacy Act) **Relevant Guides:** - [Remove PII from Data Streams](https://examples.expanso.io/data-security/remove-pii/) - De-identification requirements - [Encrypt Sensitive Data in Transit](https://examples.expanso.io/data-security/encrypt-data/) - Reasonable security measures **Key Requirements:** - Consumer data rights - De-identification of personal information - Reasonable security procedures - Privacy policy requirements --- ### HIPAA (Health Insurance Portability and Accountability Act) **Relevant Guides:** - [Encrypt Sensitive Data in Transit](https://examples.expanso.io/data-security/encrypt-data/) - Encryption requirements - [Enforce Data Schema Validation](https://examples.expanso.io/data-security/enforce-schema/) - Data integrity **Key Requirements:** - Encryption of ePHI in transit (Β§164.312(e)(1)) - Integrity controls (Β§164.312(c)(1)) - Access controls (Β§164.312(a)(1)) - Audit controls (Β§164.312(b)) --- ### PCI-DSS (Payment Card Industry Data Security Standard) **Relevant Guides:** - [Encrypt Sensitive Data in Transit](https://examples.expanso.io/data-security/encrypt-data/) - Cardholder data encryption - [Remove PII from Data Streams](https://examples.expanso.io/data-security/remove-pii/) - Data minimization **Key Requirements:** - Protect cardholder data (Requirement 3) - Encrypt transmission of cardholder data (Requirement 4) - Restrict access to cardholder data (Requirement 7) - Maintain audit trails (Requirement 10) --- ## Security Best Practices ### 1. Process at the Edge Remove, anonymize, or encrypt sensitive data at the edge **before** it reaches the cloud. This minimizes the attack surface and simplifies compliance. ### 2. Use Cryptographic Hashing For anonymization, use SHA-256 or stronger hashing algorithms. Never use reversible encoding like Base64. ### 3. Implement Schema Validation Validate all incoming data against a schema to prevent injection attacks and ensure data quality. ### 4. Rotate Encryption Keys Implement key rotation policies and version your encryption keys for forward secrecy. ### 5. Monitor Security Events Track validation failures, encryption errors, and PII detection in your monitoring system. ### 6. Audit All Transformations Log all security-related operations (PII removal, encryption, validation) for compliance audits. --- ## Data Generation & Transformation # Data Generation & Transformation This example demonstrates using the `generate` input to create synthetic data for testing pipelines. It's useful for development, testing transformations, and simulating data sources without external dependencies. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/data-transformer.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: data-transformer\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`data-transformer.yaml`](pathname:///examples/data-transformer.yaml)
## What This Pipeline Does 1. **Generates** messages every 5 seconds with synthetic data 2. **Creates** unique IDs, timestamps, and counter values 3. **Enriches** with processing metadata 4. **Outputs** to stdout for inspection ## Complete Pipeline ```yaml input: generate: interval: 5s mapping: | root.id = uuid_v4() root.message = "Hello from Expanso" root.timestamp = now() root.count = counter() pipeline: processors: - mapping: | root = this root.processed_at = now() output: stdout: codec: lines ``` ## Try Locally This pipeline needs nothing external, so it runs on a local agent with no account. Start one as in Step 3 of [Try It Without an Account](/getting-started/try-locally#step-3-start-a-local-agent), then run this in a second terminal, from the same folder: ```bash curl -fsSL -o data-transformer.yaml \ https://docs.expanso.io/examples/data-transformer.yaml expanso-edge validate data-transformer.yaml { printf 'name: data-transformer\ntype: pipeline\nconfig:\n' sed 's/^/ /' data-transformer.yaml; } > data-transformer-job.yaml export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" expanso-cli job deploy data-transformer-job.yaml \ --endpoint http://localhost:9010 ``` The `printf` and `sed` lines wrap the pipeline in the job fields (`name`, `type`, `config`) that `job deploy` expects. `EXPANSO_CLI_CONFIG_DIR` and `--endpoint` keep the CLI on the local agent, away from any saved profile or Expanso Cloud workspace. Every 5 seconds a message like this appears in the agent's terminal (IDs and times vary): ```json {"count":1,"id":"...","message":"Hello from Expanso","processed_at":"...","timestamp":"..."} ``` The job keeps running until you stop it: ```bash expanso-cli job stop data-transformer --force \ --endpoint http://localhost:9010 ``` On a local agent the pipeline can keep producing output for a short while after the stop (22 seconds in one test), then halts. Stopping the agent with `Ctrl+C` stops everything at once. ## Configuration Breakdown ### Input: Generate ```yaml input: generate: interval: 5s mapping: | root.id = uuid_v4() root.message = "Hello from Expanso" root.timestamp = now() root.count = counter() ``` The `generate` input creates messages at a fixed interval using a Bloblang mapping. Each message gets: - `id`: Unique UUID (e.g., `550e8400-e29b-41d4-a716-446655440000`) - `message`: Static string - `timestamp`: Current timestamp in RFC3339 format - `count`: Incrementing counter starting from 1 **See:** [Generate Input](/components/inputs/generate) ### Processor: Add Metadata ```yaml pipeline: processors: - mapping: | root = this root.processed_at = now() ``` Copies the entire input message and adds a `processed_at` timestamp showing when the message was processed. ### Output: Stdout ```yaml output: stdout: codec: lines ``` Prints each message to stdout as a single line of JSON. **See:** [Stdout Output](/components/outputs/stdout) ## Example Output Each message will look like this: ```json {"id":"a1b2c3d4-e5f6-7890-abcd-ef1234567890","message":"Hello from Expanso","timestamp":"2024-01-15T10:30:00Z","count":1,"processed_at":"2024-01-15T10:30:00.123Z"} {"id":"b2c3d4e5-f6a7-8901-bcde-f12345678901","message":"Hello from Expanso","timestamp":"2024-01-15T10:30:05Z","count":2,"processed_at":"2024-01-15T10:30:05.456Z"} {"id":"c3d4e5f6-a7b8-9012-cdef-012345678902","message":"Hello from Expanso","timestamp":"2024-01-15T10:30:10Z","count":3,"processed_at":"2024-01-15T10:30:10.789Z"} ``` ## Common Variations ### Generate Realistic Test Data Create more complex synthetic data: ```yaml input: generate: interval: 1s mapping: | root.user_id = "user-" + counter().string() root.event_type = ["login", "logout", "purchase", "view"].index(counter() % 4) root.timestamp = now() root.value = random_int(min: 1, max: 100) root.metadata.ip = "192.168.1." + (random_int(min: 1, max: 254)).string() root.metadata.user_agent = "TestAgent/1.0" ``` ### Generate Batch Test Data Generate a specific number of messages as fast as possible: ```yaml input: generate: interval: "" # Empty interval = as fast as possible count: 1000 # Generate exactly 1000 messages mapping: | root.id = uuid_v4() root.batch_num = counter() root.data = "test-data-" + counter().string() ``` ### Scheduled Processing Use generate to trigger scheduled processing (e.g., every 5 minutes): ```yaml input: generate: interval: '@every 5m' mapping: 'root = {}' pipeline: processors: # Add your scheduled processing here - mapping: | root.scheduled_run = now() root.task = "periodic-cleanup" ``` ### Simulate Streaming Data Generate continuous events with varying data: ```yaml input: generate: interval: 100ms # 10 messages per second mapping: | root.sensor_id = "sensor-" + (random_int(min: 1, max: 10)).string() root.temperature = random_int(min: 15, max: 30) root.humidity = random_int(min: 30, max: 80) root.timestamp = now() ``` ### Transform Generated Data Apply complex transformations to synthetic data: ```yaml input: generate: interval: 2s mapping: | root.raw_value = random_int(min: 0, max: 1000) root.timestamp = now() pipeline: processors: - mapping: | # Calculate derived values root.value = this.raw_value root.category = if this.raw_value < 300 { "low" } else if this.raw_value < 700 { "medium" } else { "high" } root.normalized = this.raw_value / 1000.0 root.processed_at = now() ``` ## Use Cases ### Development & Testing Generate test data without external dependencies: - Test transformations before connecting to real data sources - Validate pipeline configurations - Benchmark processing performance - Debug routing logic ### Scheduled Tasks Trigger periodic processing: - Run cleanup tasks every hour - Fetch data from APIs on schedule - Generate periodic reports - Trigger batch jobs ### Load Testing Simulate high-volume data sources: - Test pipeline throughput - Validate error handling under load - Measure resource usage - Identify bottlenecks ### Data Simulation Create realistic test scenarios: - Simulate sensor data for IoT testing - Generate user events for analytics testing - Create transaction data for fraud detection testing - Mock API responses ## Important Notes ### Interval Formats The `interval` field supports several formats: ```yaml interval: 1s # Every second interval: 500ms # Every 500 milliseconds interval: 1m # Every minute interval: '@every 5m' # Every 5 minutes (cron-style) interval: "" # As fast as possible (use with count) ``` ### Counter Behavior The `counter()` function maintains state across messages: - Starts at 1 - Increments by 1 for each message - Persists across the lifetime of the input - Resets if the pipeline restarts ### Performance Considerations For high-volume generation: - Use shorter intervals (e.g., `100ms` for 10 msgs/sec) - Set `count` to limit total messages - Use `batch_size` to group messages - Monitor resource usage ### Memory Management When generating large volumes: - Set explicit `count` to prevent infinite generation - Use appropriate `interval` to control rate - Consider downstream processing capacity - Monitor memory usage ## Next Steps - **[Generate Input](/components/inputs/generate)** - Full configuration options - **[Mapping Processor](/components/processors/mapping)** - Transform data with Bloblang - **[Bloblang Guide](/guides/bloblang)** - Data transformation reference ## Deadband Filtering for Numeric Streams # Deadband Filtering for Numeric Streams This page collects five end-to-end pipelines that use the [`deadband`](/components/processors/deadband) processor to drop noisy or unchanged numeric readings *before* they leave the edge. Each example is a working pipeline you can run as-is; the surrounding inputs and outputs are kept minimal so the focus stays on the deadband configuration. For the full configuration surface, see the [`deadband` processor reference](/components/processors/deadband). ## 1. Simple absolute band with heartbeat A SCADA tag streams a reading every second. We only want to forward a reading when the value has moved by at least 0.5 units, **or** once a minute regardless so downstream can tell *unchanged* from *dead*. ```yaml input: generate: interval: 1s mapping: | root.value = random_int(min: 200, max: 220) * 0.1 meta node_id = "sensor-1" pipeline: processors: - deadband: single: value_field: value # body.value tag_source: meta tag_field: node_id # metadata.node_id default: mode: absolute threshold: 0.5 heartbeat_interval: 60s output: stdout: {} ``` **What survives.** With a synthetic stream centered around 21.0 Β± 1.0, you will see one reading at startup (the first reading is always forwarded), then a forward each time the value drifts 0.5 units or more from the last forwarded baseline, plus one keep-alive every 60 seconds even when nothing has moved. **Why `tag_source: meta`.** Tag ids do not change message-to-message for a given sensor β€” they belong on the message envelope, not in the body. Reading from metadata also lets you set the tag id with a single `mapping` step at ingest and never repeat it in the payload. ### Try this one locally This first pipeline needs nothing external, so it runs on a local agent with no account. Start one as in Step 3 of [Try It Without an Account](/getting-started/try-locally#step-3-start-a-local-agent), then run this in a second terminal, from the same folder: ```bash curl -fsSL -o deadband-filtering.yaml \ https://docs.expanso.io/examples/deadband-filtering.yaml expanso-edge validate deadband-filtering.yaml { printf 'name: deadband-filtering\ntype: pipeline\nconfig:\n' sed 's/^/ /' deadband-filtering.yaml; } > deadband-job.yaml export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" expanso-cli job deploy deadband-job.yaml \ --endpoint http://localhost:9010 ``` The `printf` and `sed` lines wrap the pipeline in the job fields (`name`, `type`, `config`) that `job deploy` expects. `EXPANSO_CLI_CONFIG_DIR` and `--endpoint` keep the CLI on the local agent, away from any saved profile or Expanso Cloud workspace. Readings appear in the agent's terminal. In one 12-second run it forwarded these 9 readings: ```json {"value":21.1} {"value":20.3} {"value":21} {"value":20.5} {"value":21.5} {"value":20} {"value":21.5} {"value":20.6} {"value":21.5} ``` Each forwarded value is at least 0.5 away from the one before it. The job keeps running until you stop it: ```bash expanso-cli job stop deadband-filtering --force \ --endpoint http://localhost:9010 ``` On a local agent the pipeline can keep producing output for a short while after the stop (22 seconds in one test), then halts. Stopping the agent with `Ctrl+C` stops everything at once. ## 2. Sparkplug-style metrics array Many industrial protocols ship a *batch* of metrics in a single message β€” a Sparkplug B device data event, an OpenTelemetry metric snapshot, an MQTT payload with several gauges. The `metrics` source shape deadbands each metric independently and rewrites the array with the sub-threshold metrics removed. ```yaml input: generate: interval: 5s mapping: | root.device = "plc-12" root.metrics = [ { "name": "rpm", "value": 1500 + (random_int(min: 0, max: 4) - 2) }, { "name": "temp", "value": 40.0 + (random_int(min: 0, max: 2) - 1) * 0.05 }, { "name": "psi", "value": 75.0 } ] pipeline: processors: - deadband: metrics: path: metrics # body.metrics name_field: name # per-element tag id value_field: value default: mode: absolute threshold: 1.0 output: stdout: {} ``` **What survives.** `rpm` swings by 2 units between messages, so it crosses the band on most ticks. `temp` moves by 0.05 each tick, well under the band, so it is filtered out of the array. `psi` is constant, so after the first message it disappears from the output array entirely. When *every* metric falls within the band on a given tick, the whole message is dropped; when *some* survive, the output JSON keeps the surrounding `device` field and contains only the surviving metrics under `metrics`. **Byte-stable passthrough.** If no metric is filtered on a given message β€” every tag crossed its band that tick β€” the processor forwards the **original bytes untouched**, not a re-marshaled copy. This matters if you sign messages downstream: a re-marshal would reorder JSON keys and break the signature. If you sign after filtering, place the [`signature`](/components/processors/signature) processor **after** `deadband` so it operates on the canonical filtered form. ## 3. Percent band for wide-magnitude signals A flow-rate sensor reads anywhere from 5 to 5000 units per second depending on plant state. A fixed absolute band has to be tuned to the *peak* magnitude, which then suppresses meaningful changes at the low end. A percent band scales naturally: a 5% move is meaningful at both 50 and 5000. ```yaml input: generate: interval: 1s mapping: | root.reading = random_int(min: 100, max: 5000) meta tag = "flow-3" pipeline: processors: - deadband: single: value_field: reading tag_source: meta tag_field: tag default: mode: percent threshold: 5 # forward on a 5% move output: stdout: {} ``` :::caution[Percent thresholds are 0–100, not 0–1] The submission validator emits a warning when a percent threshold is `> 100`, because it almost always means the author wrote a fraction. `0.5` here would mean **0.5%**, not 50%. To get 50%, write `50`. ::: ## 4. Catch fast ramps with `max_slew` A pressure signal sits at 75 PSI with Β±1 PSI jitter, which an absolute band of 5 PSI happily suppresses. But when a leak develops the value ramps from 75 to 60 PSI over 4 seconds β€” by the time the band fires you have already missed the early part of the event. Adding a `max_slew` trigger forwards a reading as soon as the **rate of change** crosses a threshold, even if the band would still suppress it. ```yaml input: generate: interval: 1s mapping: | # 0.8 chance of jitter, 0.2 chance of a step toward 60. root.value = if random_int(min: 0, max: 9) < 8 { 75.0 + ((random_int(min: 0, max: 20) - 10) * 0.1) } else { 75.0 - random_int(min: 0, max: 15) } meta tag_id = "pressure-1" pipeline: processors: - deadband: single: value_field: value tag_source: meta tag_field: tag_id default: mode: absolute threshold: 5 # noise floor β€” wide enough to swallow jitter max_slew: 2.0 # but forward when |Ξ”| / Ξ”t β‰₯ 2 PSI/s output: stdout: {} ``` `max_slew` is measured in **value units per second**, where seconds are wall-clock time since the last forward for that tag. The band and the slew triggers are independent: a forward fires when **any** of `first-seen`, `heartbeat`, `slew`, or `band` is true. A `dt` of zero or negative (a backward clock step) is skipped safely β€” only the band test applies until the clock recovers. ## 5. SCADA-shaped pipeline with per-tag overrides A real SCADA / IIoT deployment runs a heterogeneous tag set on one pipeline: a noisy vibration sensor, a slow temperature signal, a high-rate-of-change pressure, and a low-cardinality counter. Each needs its own tuning. The `default` block sets the floor; `tag_overrides` patches individual tags. ```yaml input: kafka: addresses: [ "broker:9092" ] topics: [ "scada.tags" ] consumer_group: "deadband-edge" pipeline: processors: # Earlier steps will have set metadata.sensor_id and shaped the body to # { reading: }. The deadband only forwards a meaningful change # per sensor β€” every other reading is dropped here, at the edge. - deadband: single: value_source: body value_field: reading tag_source: meta tag_field: sensor_id default: mode: absolute threshold: 0.5 heartbeat_interval: 60s # one keep-alive per tag per minute tag_overrides: vibration-1: # noisy mechanical signal: wider band, faster heartbeat threshold: 5.0 heartbeat_interval: 30s pressure-9: # mostly stable but occasional fast ramps threshold: 2.0 max_slew: 1.5 counter-7: # high-magnitude monotonic counter β€” use percent mode: percent threshold: 1 on_error: drop # discard malformed readings rather than passing them through max_tags: 50000 # bound state when sensor_id comes from message content output: kafka: addresses: [ "broker:9092" ] topic: "scada.tags.deadbanded" ``` **Why these settings.** - **`tag_source: meta`** keeps the tag id on the envelope so a content-driven payload cannot create a new tracked tag β€” `max_tags` is then a memory backstop, not a routine control. - **Default `heartbeat_interval: 60s`** preserves a keep-alive signal per tag even when nothing has moved, so downstream consumers can tell *quiet* from *dead*. - **Default `threshold: 0.5`** is the noise floor for the majority of tags; only the outliers (`vibration-1`, `pressure-9`, `counter-7`) need overrides. - **`on_error: drop`** is appropriate when malformed readings are non-recoverable noise. Use `fail` instead to route them through [error handling](/guides/pipelines/error_handling) into a dead-letter topic. **Compounding with downstream signing.** Adding a [`signature`](/components/processors/signature) processor after `deadband` produces signed, deadbanded events: downstream consumers can verify both that the reading came from the producing edge node and that the bytes were not altered after filtering. The signature covers the filtered output, not the dropped originals β€” which is the right behavior for audit and compliance use cases. ## Common variations ### Read the value from metadata too When an upstream step has already pulled the numeric value into metadata (a parser, an enrichment step, or an OPC-UA source), point `value_source` at `meta`: ```yaml - deadband: single: value_source: meta value_field: reading # metadata.reading tag_source: meta tag_field: sensor_id default: mode: absolute threshold: 0.5 ``` Metadata values are parsed with `strconv.ParseFloat`. `NaN`, `+Inf`, `-Inf`, and overflowing values are rejected (they would otherwise silently disable deadbanding for the tag) and route through `on_error`. ### Forward on any non-zero change (deduplication only) A `threshold: 0` band forwards on any change β€” useful for deduplicating identical consecutive readings without further filtering. Combine with `heartbeat_interval` to get a "one reading per change, plus a keep-alive" stream. ```yaml default: mode: absolute threshold: 0 heartbeat_interval: 5m ``` ### Inherit only the parts you want to override `tag_overrides` entries are partial; any sub-field you do not set inherits from `default`. To run one tag in `percent` mode while everything else stays `absolute`, set only `mode` and `threshold` on the override: ```yaml default: mode: absolute threshold: 0.5 heartbeat_interval: 60s # inherited by every override below tag_overrides: counter-7: mode: percent threshold: 1 # 1%; heartbeat_interval still 60s vibration-1: threshold: 5.0 # absolute mode (inherited); just a wider band ``` ## Next Steps - **[`deadband` processor reference](/components/processors/deadband)** β€” every field, every default, the decision logic, error postures, and state bounds. - **[`signature` processor](/components/processors/signature)** β€” sign the deadbanded output for tamper-evident audit trails. - **[Pipeline error handling](/guides/pipelines/error_handling)** β€” route messages that fail validation to a dead-letter sink. - **[Quick Start](/getting-started/quickstart/)** β€” how to actually deploy and run these pipelines on an edge node. ## HTTP to File Pipeline # HTTP to File Pipeline This example demonstrates receiving data via HTTP POST requests, applying transformations, batching events, and writing them to files. It's useful for receiving webhook events, logs, or metrics from external systems. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/http-to-file.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: http-to-file\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`http-to-file.yaml`](pathname:///examples/http-to-file.yaml)
## What This Pipeline Does 1. **Receives** HTTP POST requests at `/events` endpoint 2. **Validates** and parses JSON payloads 3. **Enriches** with metadata (timestamp, node ID, processing ID) 4. **Batches** events into JSON arrays 5. **Writes** batches to timestamped files ## Complete Pipeline ```yaml input: http_server: address: 0.0.0.0:8082 path: /events allowed_verbs: - POST sync_response: status: "202" headers: Content-Type: application/json pipeline: processors: # Validate and parse JSON - mapping: | root = this.parse_json().catch(deleted()) # Add metadata - mapping: | root.received_at = now() root.node_id = env("NODE_ID").or("unknown") root.processing_id = uuid_v4() # Group into batches - archive: format: json_array # Add batch metadata - mapping: | root.batch_id = uuid_v4() root.batch_size = this.length() root.events = this output: file: path: /var/log/expanso/events/${!timestamp_unix()}.json ``` ## Configuration Breakdown ### Input: HTTP Server ```yaml input: http_server: address: 0.0.0.0:8082 path: /events allowed_verbs: - POST sync_response: status: "202" headers: Content-Type: application/json ``` The `http_server` input listens on port 8082 and accepts POST requests at `/events`. It responds immediately with HTTP 202 (Accepted) to acknowledge receipt without waiting for processing to complete. **Note:** The `sync_response.status` must be a string, not a number. **See:** [HTTP Server Input](/components/inputs/http_server) ### Processors: Validate, Enrich, Batch #### 1. Parse and Validate JSON ```yaml - mapping: | root = this.parse_json().catch(deleted()) ``` Attempts to parse the incoming payload as JSON. If parsing fails, the message is deleted (dropped). This prevents invalid data from proceeding through the pipeline. #### 2. Add Metadata ```yaml - mapping: | root.received_at = now() root.node_id = env("NODE_ID").or("unknown") root.processing_id = uuid_v4() ``` Enriches each event with: - `received_at`: Current timestamp - `node_id`: From NODE_ID environment variable, defaults to "unknown" - `processing_id`: Unique UUID for tracking #### 3. Batching with Archive ```yaml - archive: format: json_array ``` Groups individual events into JSON arrays. This creates batches rather than writing one file per event. **See:** [Archive Processor](/components/processors/archive) #### 4. Batch Metadata ```yaml - mapping: | root.batch_id = uuid_v4() root.batch_size = this.length() root.events = this ``` Wraps the batched events array with metadata: - `batch_id`: Unique identifier for this batch - `batch_size`: Number of events in the batch - `events`: The array of events ### Output: File ```yaml output: file: path: /var/log/expanso/events/${!timestamp_unix()}.json ``` Writes each batch to a file with a Unix timestamp in the filename. The `${!timestamp_unix()}` function interpolation generates unique filenames. **Example filename:** `/var/log/expanso/events/1706191234.json` **See:** [File Output](/components/outputs/file) ## Example Usage ### Send a Test Event ```bash curl -X POST http://localhost:8082/events \ -H "Content-Type: application/json" \ -d '{"event":"user.signup","user_id":"12345","timestamp":"2024-01-15T10:30:00Z"}' ``` ### Expected Response ``` HTTP/1.1 202 Accepted Content-Type: application/json ``` ### Resulting File Content After batching, the output file might look like: ```json { "batch_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890", "batch_size": 1, "events": [ { "event": "user.signup", "user_id": "12345", "timestamp": "2024-01-15T10:30:00Z", "received_at": "2024-01-15T10:30:01.234Z", "node_id": "edge-001", "processing_id": "b2c3d4e5-f6g7-8901-bcde-f12345678901" } ] } ``` ## Common Variations ### Filter Specific Events Only process certain event types: ```yaml pipeline: processors: - mapping: | root = this.parse_json().catch(deleted()) # Filter to specific event types - mapping: | root = if ["user.signup", "user.login"].contains(this.event) { this } else { deleted() } ``` ### Route to Multiple Outputs Send different event types to different files: ```yaml output: switch: cases: - check: this.events.0.event.has_prefix("user.") output: file: path: /var/log/expanso/user-events/${!timestamp_unix()}.json - check: this.events.0.event.has_prefix("system.") output: file: path: /var/log/expanso/system-events/${!timestamp_unix()}.json - output: file: path: /var/log/expanso/other-events/${!timestamp_unix()}.json ``` ### Adjust Batch Size Control how many events are batched together: ```yaml pipeline: processors: # ... other processors ... # Batch events - archive: format: json_array - mapping: | root.batch_id = uuid_v4() root.batch_size = this.length() root.events = this output: file: path: /var/log/expanso/events/${!timestamp_unix()}.json ``` ## Next Steps - **[HTTP Server Input](/components/inputs/http_server)** - Full configuration options - **[Archive Processor](/components/processors/archive)** - Batching patterns - **[Bloblang Guide](/guides/bloblang)** - Data transformation reference ## HTTP Webhook Receiver # HTTP Webhook Receiver This example demonstrates receiving HTTP webhooks from external services, validating their signatures, and routing them to different destinations based on content. It's useful for integrating with services like GitHub, Stripe, Slack, or any system that sends webhook notifications. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/http-webhook.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: http-webhook\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`http-webhook.yaml`](pathname:///examples/http-webhook.yaml)
## What This Pipeline Does 1. **Receives** HTTP POST requests at multiple webhook endpoints 2. **Validates** webhook signatures using HMAC-SHA256 3. **Parses** JSON payloads and extracts event type 4. **Routes** events to different outputs based on type 5. **Responds** with appropriate HTTP status codes ## Complete Pipeline ```yaml input: http_server: address: 0.0.0.0:8080 path: /webhooks/{source} allowed_verbs: - POST timeout: 30s sync_response: status: "${! json(\"status\").or(\"200\") }" headers: Content-Type: application/json pipeline: processors: # Extract webhook source from path parameter - mapping: | meta source = @http_server_request_path.trim_prefix("/webhooks/") root = this # Parse JSON payload - mapping: | root = this.parse_json() root.webhook_source = @source root.received_at = now() # Validate signature (for sources that provide one) - branch: request_map: | root.payload = content() root.signature = @"X-Hub-Signature-256".or(@"X-Signature").or("") root.secret = env("WEBHOOK_SECRET").or("") processors: - mapping: | let expected = "sha256=" + this.payload.hash("hmac_sha256", this.secret).encode("hex") root.valid = if this.signature == "" { true # No signature provided, skip validation } else { this.signature == expected } result_map: | root.signature_valid = this.valid # Drop invalid signatures and return 401 - mapping: | root = this root.status = if this.signature_valid == false { "401" } else { "200" } root.error = if this.signature_valid == false { "Invalid signature" } else { deleted() } # Log the webhook - log: level: INFO message: "Received webhook" fields_mapping: | root.source = this.webhook_source root.event = this.event.or(this.type).or("unknown") output: switch: cases: # Invalid signature - respond with error - check: this.status == "401" output: sync_response: {} # Route based on webhook source or event type - check: this.webhook_source == "github" output: broker: outputs: - sync_response: {} - http_client: url: ${GITHUB_EVENTS_URL} verb: POST headers: Content-Type: application/json - check: this.webhook_source == "stripe" output: broker: outputs: - sync_response: {} - http_client: url: ${STRIPE_EVENTS_URL} verb: POST headers: Content-Type: application/json # Default: acknowledge and log - output: broker: outputs: - sync_response: {} - stdout: codec: lines ``` ## Configuration Breakdown ### Input: HTTP Server with Path Parameters ```yaml input: http_server: address: 0.0.0.0:8080 path: /webhooks/{source} allowed_verbs: - POST timeout: 30s sync_response: status: "${! json(\"status\").or(\"200\") }" headers: Content-Type: application/json ``` The `http_server` input listens on port 8080. The `{source}` path parameter captures the webhook source (e.g., `/webhooks/github`, `/webhooks/stripe`). The dynamic `sync_response.status` allows processors to control the HTTP response code. **See:** [HTTP Server Input](/components/inputs/http_server) ### Processors: Validate and Route #### 1. Extract Source from Path ```yaml - mapping: | meta source = @http_server_request_path.trim_prefix("/webhooks/") root = this ``` Extracts the webhook source from the URL path and stores it as metadata for later use. #### 2. Parse Payload and Add Metadata ```yaml - mapping: | root = this.parse_json() root.webhook_source = @source root.received_at = now() ``` Parses the JSON payload and enriches it with the source identifier and timestamp. #### 3. Validate Webhook Signature ```yaml - branch: request_map: | root.payload = content() root.signature = @"X-Hub-Signature-256".or(@"X-Signature").or("") root.secret = env("WEBHOOK_SECRET").or("") processors: - mapping: | let expected = "sha256=" + this.payload.hash("hmac_sha256", this.secret).encode("hex") root.valid = if this.signature == "" { true # No signature provided, skip validation } else { this.signature == expected } result_map: | root.signature_valid = this.valid ``` Uses a `branch` processor to validate HMAC-SHA256 signatures. This is compatible with GitHub's `X-Hub-Signature-256` header and similar patterns used by other services. **See:** [Branch Processor](/components/processors/branch) ### Output: Route by Source ```yaml output: switch: cases: - check: this.webhook_source == "github" output: broker: outputs: - sync_response: {} - http_client: url: ${GITHUB_EVENTS_URL} verb: POST ``` Routes webhooks to different destinations based on their source. The `broker` sends both a synchronous response back to the sender and forwards the event to downstream services. **See:** [Switch Output](/components/outputs/switch) | [Broker Output](/components/outputs/broker) ## Try It Locally ### 1. Start the Pipeline ```bash # Set optional environment variables in the agent's terminal export WEBHOOK_SECRET="your-secret-here" export GITHUB_EVENTS_URL="http://localhost:9000/events" export STRIPE_EVENTS_URL="http://localhost:9001/events" # Start a local edge agent in the same terminal expanso-edge run --local --data-dir ./.edge ``` Then deploy the pipeline to the agent from a second terminal, as in step 3 of [Download & Run](#download--run). ### 2. Send Test Webhooks **Simple webhook (no signature):** ```bash curl -X POST http://localhost:8080/webhooks/test \ -H "Content-Type: application/json" \ -d '{"event": "test.event", "data": {"id": "123"}}' ``` **GitHub-style webhook with signature:** ```bash # Generate signature PAYLOAD='{"event": "push", "repository": "my-repo"}' SIGNATURE="sha256=$(echo -n "$PAYLOAD" | openssl dgst -sha256 -hmac "your-secret-here" | cut -d' ' -f2)" curl -X POST http://localhost:8080/webhooks/github \ -H "Content-Type: application/json" \ -H "X-Hub-Signature-256: $SIGNATURE" \ -d "$PAYLOAD" ``` ### 3. Expected Responses **Valid webhook:** ``` HTTP/1.1 200 OK Content-Type: application/json ``` **Invalid signature:** ``` HTTP/1.1 401 Unauthorized Content-Type: application/json {"error": "Invalid signature"} ``` ## Common Variations ### Filter by Event Type Only process specific event types: ```yaml pipeline: processors: - mapping: | root = this.parse_json() # Only process specific events - mapping: | root = if ["push", "pull_request", "release"].contains(this.event.or(this.action)) { this } else { deleted() } ``` ### Add Rate Limiting Protect against webhook floods: ```yaml rate_limit_resources: - label: webhook_limit local: count: 100 interval: 1m input: http_server: path: /webhooks/{source} rate_limit: webhook_limit ``` ### Store Webhooks for Replay Save webhooks to a file for debugging or replay: ```yaml output: broker: pattern: fan_out outputs: - sync_response: {} - file: path: /var/log/webhooks/${!@source}/${!timestamp_unix()}.json codec: lines ``` ### Multiple Signature Formats Support different signature header formats: ```yaml - mapping: | # Try different header names let sig = @"X-Hub-Signature-256".or( @"X-Signature-256" ).or( @"Stripe-Signature" ).or("") # Handle different formats let clean_sig = if sig.has_prefix("sha256=") { sig } else if sig.has_prefix("t=") { # Stripe format: extract the signature sig.re_find_object("v1=(?P[a-f0-9]+)").sig.or("") } else { sig } meta signature = clean_sig ``` ## Next Steps - **[HTTP Server Input](/components/inputs/http_server)** - Full configuration options - **[Branch Processor](/components/processors/branch)** - Conditional processing - **[Switch Output](/components/outputs/switch)** - Content-based routing - **[HTTP Client Output](/components/outputs/http_client)** - Forward to HTTP endpoints - **[Bloblang Guide](/guides/bloblang)** - Data transformation reference ## Kafka to S3 Pipeline # Kafka to S3 Pipeline This example demonstrates consuming messages from Apache Kafka and writing them to Amazon S3 as batched JSON files. Useful for data lake ingestion, event stream archiving, or ETL workflows. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/kafka-to-s3.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: kafka-to-s3\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`kafka-to-s3.yaml`](pathname:///examples/kafka-to-s3.yaml)
## What This Pipeline Does 1. **Consumes** messages from a Kafka topic 2. **Parses** and validates JSON payloads 3. **Enriches** messages with metadata (Kafka partition, offset, timestamp) 4. **Batches** messages for efficient S3 uploads 5. **Writes** batched data to S3 with partitioned paths ## Complete Pipeline Configure your Kafka brokers and S3 bucket, then run: ```yaml input: kafka: addresses: - localhost:9092 topics: - events consumer_group: expanso-s3-archiver start_from_oldest: true pipeline: processors: # Parse JSON payload - mapping: | root = this.parse_json().catch(deleted()) # Add Kafka metadata for traceability - mapping: | root.kafka_topic = @kafka_topic root.kafka_partition = @kafka_partition root.kafka_offset = @kafka_offset root.processed_at = now() # Batch messages into JSON array - archive: format: json_array output: aws_s3: bucket: my-data-lake path: events/${!now().ts_format("2006/01/02/15")}/batch-${!timestamp_unix_nano()}.json content_type: application/json region: us-east-1 batching: count: 100 period: 30s ``` ## Configuration Breakdown ### Input: Kafka ```yaml input: kafka: addresses: - localhost:9092 topics: - events consumer_group: expanso-s3-archiver start_from_oldest: true ``` The `kafka` input connects to your Kafka cluster and consumes messages from the specified topic. Parallelism is controlled by `checkpoint_limit`, and offsets are committed periodically to track progress. **Key settings:** - `addresses`: Your Kafka broker addresses - `topics`: The topic(s) to consume from - `consumer_group`: Groups consumers for offset tracking and load balancing - `start_from_oldest`: Start from earliest messages on first run **See:** [Kafka Input](/components/inputs/kafka) ### Processors: Parse, Enrich, Batch #### 1. Parse JSON ```yaml - mapping: | root = this.parse_json().catch(deleted()) ``` Parses each Kafka message as JSON. Invalid messages are dropped to prevent pipeline errors. #### 2. Add Metadata ```yaml - mapping: | root.kafka_topic = @kafka_topic root.kafka_partition = @kafka_partition root.kafka_offset = @kafka_offset root.processed_at = now() ``` Enriches each message with Kafka metadata for traceability: - `kafka_topic`: Source topic - `kafka_partition`: Partition number - `kafka_offset`: Message offset - Processing timestamp #### 3. Batch Messages ```yaml - archive: format: json_array ``` Uses the archive processor to group messages into JSON arrays. Combined with the output batching config, this creates efficient batches for S3 uploads. **See:** [Archive Processor](/components/processors/archive) ### Output: AWS S3 ```yaml output: aws_s3: bucket: my-data-lake path: events/${!now().ts_format("2006/01/02/15")}/batch-${!timestamp_unix_nano()}.json content_type: application/json region: us-east-1 batching: count: 100 period: 30s ``` Writes batched messages to S3 with a time-partitioned path structure. The path uses interpolation functions to organize files by date and ensure unique filenames. **Path structure:** `events/YYYY/MM/DD/HH/batch-timestamp.json` This enables efficient querying with tools like Athena, Spark, or Presto that support partition pruning. **See:** [AWS S3 Output](/components/outputs/aws_s3) ## Common Variations ### Add Compression Compress files before upload to reduce storage costs: ```yaml output: aws_s3: bucket: my-data-lake path: events/${!now().ts_format("2006/01/02/15")}/batch-${!timestamp_unix_nano()}.json.gz content_type: application/json content_encoding: gzip region: us-east-1 batching: count: 100 period: 30s processors: - compress: algorithm: gzip ``` ### Route by Topic When consuming from multiple topics, route to different S3 prefixes: ```yaml input: kafka: addresses: - localhost:9092 topics: - orders - users - events consumer_group: expanso-s3-archiver output: aws_s3: bucket: my-data-lake path: ${!@kafka_topic}/${!now().ts_format("2006/01/02/15")}/batch-${!timestamp_unix_nano()}.json content_type: application/json region: us-east-1 batching: count: 100 period: 30s ``` ### Add Error Handling Send failed messages to a dead letter topic instead of dropping them: ```yaml pipeline: processors: - try: - mapping: | root = this.parse_json() - mapping: | root.kafka_topic = @kafka_topic root.kafka_partition = @kafka_partition root.kafka_offset = @kafka_offset root.processed_at = now() - catch: - mapping: | root.error = error() root.raw_message = content().string() root.failed_at = now() root.dead_letter = true output: switch: cases: - check: this.dead_letter == true output: kafka: addresses: - localhost:9092 topic: dead-letter-events - output: aws_s3: bucket: my-data-lake path: events/${!now().ts_format("2006/01/02/15")}/batch-${!timestamp_unix_nano()}.json ``` ## Next Steps - **[Kafka Input](/components/inputs/kafka)** - Full Kafka configuration options - **[AWS S3 Output](/components/outputs/aws_s3)** - S3 configuration and batching - **[Archive Processor](/components/processors/archive)** - Batching patterns - **[AWS Guide](/guides/cloud/aws)** - AWS credentials and configuration ## Log Processing # Log Processing Examples Process logs at the edge to reduce bandwidth costs, improve performance, and ensure compliance. These examples demonstrate different approaches to log processing, from simple filtering to production-ready pipelines. ## Available Examples ### Filter by Severity Basic log filtering that routes ERROR and WARN messages to separate destinations. Perfect for understanding the fundamentals of log routing. [View Example β†’](https://examples.expanso.io/log-processing/filter-severity/) ### Enrich & Export to S3 Add metadata and lineage tracking to logs before batching and exporting to Amazon S3. Demonstrates enrichment patterns and efficient cloud uploads. [View Example β†’](https://examples.expanso.io/log-processing/enrich-export/) ### Production Pipeline Complete production-ready pipeline with HTTP input, JSON parsing, validation, enrichment, PII redaction, and multi-destination routing (Elasticsearch, S3, local backup). [View Example β†’](https://examples.expanso.io/log-processing/production-pipeline/) ## Common Patterns These examples demonstrate key patterns used in log processing: - **Severity filtering** - Route based on log level (ERROR, WARN, INFO, DEBUG) - **Metadata enrichment** - Add node ID, region, pipeline info for traceability - **Multi-destination routing** - Send to real-time search, archival storage, and backups - **Batching** - Optimize bandwidth and API costs by batching uploads - **PII redaction** - Remove sensitive data before logs leave the edge ## Related Content - **[Parse Log Formats](https://examples.expanso.io/data-transformation/parse-logs/)** - Parse JSON, CSV, Apache, Syslog formats - **[Remove PII](https://examples.expanso.io/data-security/remove-pii/)** - Redact sensitive information - **[Content Routing](https://examples.expanso.io/data-routing/content-routing/)** - Route data based on content - **[Bloblang Guide](/guides/bloblang)** - Transformation language reference ## Metadata Enrichment for Lineage # Metadata Enrichment for Lineage This example demonstrates how to stamp every message with pipeline provenance, node identity, and custom compliance tags using the [`metadata`](/components/processors/metadata) processor. Use it as a starting point for building OpenLineage-style lineage events or compliance audit trails. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/metadata-enrichment.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: metadata-enrichment\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`metadata-enrichment.yaml`](pathname:///examples/metadata-enrichment.yaml)
## What This Pipeline Does 1. **Generates** a synthetic event every 5 seconds (no external dependencies) 2. **Enriches** each event with pipeline, node, and OpenLineage-aligned identity fields 3. **Tags** events with custom compliance metadata (owner, compliance tier) 4. **Outputs** the enriched JSON to stdout so you can inspect the result ## Complete Pipeline ```yaml input: generate: mapping: | root = { "event_id": uuid_v4(), "level": "info", "message": "user signup" } interval: 5s pipeline: processors: - metadata: include: [core, pipeline, node] custom: pipeline_owner: alice@example.com compliance_tier: pii target: body format: nested body_key: lineage output: stdout: codec: lines ``` ## Try Locally This pipeline needs nothing external, so it runs on a local agent with no account. Start one as in Step 3 of [Try It Without an Account](/getting-started/try-locally#step-3-start-a-local-agent), then run this in a second terminal, from the same folder: ```bash curl -fsSL -o metadata-enrichment.yaml \ https://docs.expanso.io/examples/metadata-enrichment.yaml expanso-edge validate metadata-enrichment.yaml { printf 'name: metadata-enrichment\ntype: pipeline\nconfig:\n' sed 's/^/ /' metadata-enrichment.yaml; } > metadata-job.yaml export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" expanso-cli job deploy metadata-job.yaml \ --endpoint http://localhost:9010 ``` The `printf` and `sed` lines wrap the pipeline in the job fields (`name`, `type`, `config`) that `job deploy` expects. `EXPANSO_CLI_CONFIG_DIR` and `--endpoint` keep the CLI on the local agent, away from any saved profile or Expanso Cloud workspace. Every 5 seconds an event appears in the agent's terminal as one JSON line. Formatted for reading, it looks like this (IDs and times vary): ```json { "event_id": "...", "level": "info", "lineage": { "agent_version": "", "cluster_name": "", "compliance_tier": "pii", "environment": "", "event_time": "...", "git_branch": "", "git_commit_sha": "", "git_repo_url": "", "hostname": "", "job_name": "metadata-enrichment", "job_namespace": "default", "node_id": "...", "pipeline_name": "metadata-enrichment", "pipeline_owner": "alice@example.com", "pipeline_version": 1, "producer": "https://expanso.io/edge", "region": "", "run_id": "e-..." }, "message": "user signup" } ``` On a local agent, the node fields such as `hostname`, `region`, and `agent_version` are empty, and `node_id` is derived from your machine's host name. The job keeps running until you stop it: ```bash expanso-cli job stop metadata-enrichment --force \ --endpoint http://localhost:9010 ``` On a local agent the pipeline can keep producing output for a short while after the stop (22 seconds in one test), then halts. Stopping the agent with `Ctrl+C` stops everything at once. ## Configuration Breakdown ### Input: Generate ```yaml input: generate: mapping: | root = { "event_id": uuid_v4(), "level": "info", "message": "user signup" } interval: 5s ``` The `generate` input emits a synthetic message every 5 seconds, so the pipeline runs anywhere without external dependencies. Replace it with `kafka`, `file`, or any other input when wiring up to real data. **See:** [Generate Input](/components/inputs/generate) ### Processor: Metadata Enrichment ```yaml - metadata: include: [core, pipeline, node] custom: pipeline_owner: alice@example.com compliance_tier: pii target: body format: nested body_key: lineage ``` The `metadata` processor splices three categories of runtime metadata under a `lineage` key: - `core` β€” OpenLineage-aligned identity (`run_id`, `job_name`, `job_namespace`, `event_time`, `producer`) - `pipeline` β€” pipeline name and version - `node` β€” node ID, region, environment, cluster, hostname, agent version The `custom:` block adds two static fields that travel with every event for ownership and compliance tracking. **See:** [Metadata Processor Reference](/components/processors/metadata) ### Output: Stdout ```yaml output: stdout: codec: lines ``` Each enriched event is printed as a single JSON line. Swap this for `kafka`, `aws_s3`, or `http_client` when you're ready to send the data downstream. ## Example Output A single emitted event looks like this: ```json { "event_id": "9f3e1b8a-2c5d-4e7f-9a01-3b2c4d5e6f7a", "level": "info", "message": "user signup", "lineage": { "run_id": "01HF7K8M9N0P1Q2R3S4T5V6W7X", "job_name": "lineage-demo", "job_namespace": "default", "event_time": "2026-04-30T12:34:56.789Z", "producer": "https://expanso.io/edge/v1.4.2", "pipeline_name": "lineage-demo", "pipeline_version": "v1", "git_commit_sha": "", "git_repo_url": "", "git_branch": "", "node_id": "node-eu-west-1-a", "hostname": "edge-01", "region": "eu-west-1", "environment": "production", "cluster_name": "edge-eu", "agent_version": "1.4.2", "pipeline_owner": "alice@example.com", "compliance_tier": "pii" } } ``` ## Common Variations **Attach to message metadata instead of the body** Use `target: meta` to write to message metadata, then reference fields downstream via interpolation (`${! metadata("region") }`). ```yaml - metadata: include: [core, pipeline, node] target: meta ``` **Surface runtime counters** Opt in to the `runtime` category to attach `records_in`, `bytes_out`, `error_count`, and `duration_ms`: ```yaml - metadata: include: [pipeline, runtime] target: body format: nested body_key: stats ``` **Flat-merge into the body** If your downstream consumer expects a flat JSON object, use `format: flat`. Note that metadata keys overwrite body keys on collision β€” use `exclude` to drop fields that would clash. ```yaml - metadata: include: [pipeline] exclude: [git_commit_sha, git_repo_url, git_branch] target: body format: flat ``` **Promote inbound metadata (e.g. Sparkplug B) into the body** For SCADA / IIoT inputs that stamp domain metadata on every message (e.g. Sparkplug B's `spb_*` keys, Kafka headers, MQTT topic parts) and emit a scalar metric value as the body, use the `promote` block to land those metadata keys as **columns** without per-pipeline Bloblang glue: ```yaml - metadata: target: body format: flat promote: match: "spb_" rename: { spb_metric_name: metric, spb_datatype: datatype } cast: { spb_timestamp: int, spb_is_historical: bool } wrap_key: value # wraps a scalar body so promoted fields can attach ``` See the dedicated [Promote Inbound Metadata example](/examples/promote-inbound-metadata) for the full SCADA-shaped walkthrough, and [Promoting inbound metadata into the body](/components/processors/metadata#promoting-inbound-metadata-into-the-body) for the field reference. ## Next Steps - **[Metadata Processor Reference](/components/processors/metadata)** β€” full configuration reference, all field categories, validation rules - **[Promote Inbound Metadata example](/examples/promote-inbound-metadata)** β€” copy input-supplied metadata into the body as columns (Sparkplug B, Kafka headers, MQTT topics) - **[Pipeline Metadata Guide](/guides/pipelines/metadata)** β€” the implicit `@pipeline_id`, `@node_id`, and `@node_label_*` keys - **[Quick Start Guide](/getting-started/quickstart/)** β€” deploying pipelines to a real edge node ## Metrics Collection & Monitoring # Metrics Collection & Monitoring This example demonstrates collecting metrics on a schedule and routing them to multiple destinations using the broker output. It shows how to generate structured metrics data, calculate derived values, and fan out to multiple outputs. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/metrics-collector.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: metrics-collector\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`metrics-collector.yaml`](pathname:///examples/metrics-collector.yaml)
## What This Pipeline Does 1. **Collects** metrics every 30 seconds 2. **Enriches** with metadata (node ID, version) 3. **Calculates** health scores based on thresholds 4. **Routes** to multiple outputs (stdout and file) ## Complete Pipeline ```yaml input: generate: interval: 30s mapping: | # Collect system metrics root.timestamp = now() root.host = env("HOSTNAME").or("localhost") root.metrics.cpu_usage = random_int(min: 0, max: 100) root.metrics.memory_usage = random_int(min: 0, max: 100) root.metrics.disk_usage = random_int(min: 0, max: 100) pipeline: processors: - mapping: | # Add metadata root.node_id = env("NODE_ID").or("unknown") root.edge_version = env("EDGE_VERSION").or("dev") # Calculate health score let cpu = this.metrics.cpu_usage let mem = this.metrics.memory_usage let disk = this.metrics.disk_usage root.health_score = if cpu > 90 || mem > 90 || disk > 90 { "critical" } else if cpu > 70 || mem > 70 || disk > 70 { "warning" } else { "healthy" } output: broker: pattern: fan_out outputs: - stdout: codec: lines - file: path: /var/log/expanso/metrics.json codec: lines ``` ## Configuration Breakdown ### Input: Generate Metrics ```yaml input: generate: interval: 30s mapping: | root.timestamp = now() root.host = env("HOSTNAME").or("localhost") root.metrics.cpu_usage = random_int(min: 0, max: 100) root.metrics.memory_usage = random_int(min: 0, max: 100) root.metrics.disk_usage = random_int(min: 0, max: 100) ``` Generates metrics every 30 seconds with: - `timestamp`: Current time in RFC3339 format - `host`: Hostname from `HOSTNAME` environment variable - `metrics`: Simulated CPU, memory, and disk usage percentages **Note:** This example uses `random_int()` to simulate metrics. For real system metrics, you would integrate with actual monitoring tools or use the `http_client` input to poll metrics endpoints. **See:** [Generate Input](/components/inputs/generate) ### Processor: Enrich & Score ```yaml pipeline: processors: - mapping: | # Add metadata root.node_id = env("NODE_ID").or("unknown") root.edge_version = env("EDGE_VERSION").or("dev") # Calculate health score let cpu = this.metrics.cpu_usage let mem = this.metrics.memory_usage let disk = this.metrics.disk_usage root.health_score = if cpu > 90 || mem > 90 || disk > 90 { "critical" } else if cpu > 70 || mem > 70 || disk > 70 { "warning" } else { "healthy" } ``` This processor: 1. **Adds metadata**: Node ID and version from environment variables 2. **Calculates health score** based on thresholds: - `critical`: Any metric > 90% - `warning`: Any metric > 70% - `healthy`: All metrics ≀ 70% **See:** [Mapping Processor](/components/processors/mapping) | [Bloblang Guide](/guides/bloblang) ### Output: Fan Out to Multiple Destinations ```yaml output: broker: pattern: fan_out outputs: - stdout: codec: lines - file: path: /var/log/expanso/metrics.json codec: lines ``` The `broker` output with `fan_out` pattern sends each message to **all** outputs: - **stdout**: For real-time monitoring/debugging - **file**: For persistent storage and later analysis **See:** [Broker Output](/components/outputs/broker) | [File Output](/components/outputs/file) ## Example Output Each metric message looks like: ```json { "timestamp": "2024-01-15T10:30:00Z", "host": "edge-node-01", "metrics": { "cpu_usage": 45, "memory_usage": 62, "disk_usage": 38 }, "node_id": "edge-001", "edge_version": "1.2.0", "health_score": "healthy" } ``` Critical alert example: ```json { "timestamp": "2024-01-15T10:35:00Z", "host": "edge-node-01", "metrics": { "cpu_usage": 95, "memory_usage": 88, "disk_usage": 42 }, "node_id": "edge-001", "edge_version": "1.2.0", "health_score": "critical" } ``` ## Common Variations ### Fetch Real Metrics from API Replace simulated data with actual metrics: ```yaml input: http_client: url: http://localhost:9090/metrics verb: GET pipeline: processors: - mapping: | # Parse Prometheus or other metrics format root = this.parse_json() root.node_id = env("NODE_ID").or("unknown") root.collected_at = now() ``` ### Alert on Critical Metrics Send critical metrics to an alerting endpoint: ```yaml output: broker: pattern: fan_out outputs: # Always write to file - file: path: /var/log/expanso/metrics.json # Alert on critical health - switch: cases: - check: this.health_score == "critical" output: http_client: url: http://alerts.example.com/webhook verb: POST headers: Content-Type: application/json ``` ### Different Metrics Per Environment Adjust collection based on environment: ```yaml input: generate: interval: '@every 1m' mapping: | root.timestamp = now() root.environment = env("ENVIRONMENT").or("production") # More detailed metrics in production root.metrics = if env("ENVIRONMENT") == "production" { { "cpu": random_int(min: 0, max: 100), "memory": random_int(min: 0, max: 100), "disk": random_int(min: 0, max: 100), "network_in": random_int(min: 0, max: 1000), "network_out": random_int(min: 0, max: 1000), "active_connections": random_int(min: 0, max: 500) } } else { { "cpu": random_int(min: 0, max: 100), "memory": random_int(min: 0, max: 100) } } ``` ### Aggregate Metrics Over Time Calculate averages or aggregates: ```yaml pipeline: processors: - mapping: | root = this root.collected_at = now() # Buffer and aggregate (requires additional configuration) - metric: type: gauge name: cpu_usage labels: host: ${! this.host } value: ${! this.metrics.cpu_usage } - metric: type: gauge name: memory_usage labels: host: ${! this.host } value: ${! this.metrics.memory_usage } ``` ### Route by Health Score Send different health scores to different destinations: ```yaml output: switch: cases: # Critical metrics go to PagerDuty - check: this.health_score == "critical" output: http_client: url: https://events.pagerduty.com/v2/enqueue verb: POST # Warning metrics go to Slack - check: this.health_score == "warning" output: http_client: url: https://hooks.slack.com/services/YOUR/WEBHOOK/URL verb: POST # Healthy metrics just get logged - output: file: path: /var/log/expanso/healthy-metrics.json ``` ### Add Statistical Analysis Calculate moving averages or percentiles: ```yaml pipeline: processors: - mapping: | # Store metrics root = this # Calculate average across all metrics let metrics_list = [ this.metrics.cpu_usage, this.metrics.memory_usage, this.metrics.disk_usage ] root.avg_utilization = metrics_list.sum() / metrics_list.length() root.max_utilization = metrics_list.max() root.min_utilization = metrics_list.min() ``` ## Real-World Integration ### Prometheus Metrics For production monitoring, integrate with Prometheus: ```yaml input: http_client: url: http://prometheus:9090/api/v1/query verb: GET headers: Accept: application/json payload: 'query=up' pipeline: processors: - mapping: | root = this.data.result.map_each(r -> { "metric": r.metric, "value": r.value.1.number(), "timestamp": now() }) ``` ### Datadog Integration Send metrics to Datadog: ```yaml output: http_client: url: https://api.datadoghq.com/api/v2/series verb: POST headers: DD-API-KEY: ${DD_API_KEY} Content-Type: application/json batching: count: 10 period: 10s ``` ### InfluxDB Time-Series Storage Store metrics in InfluxDB: ```yaml output: http_client: url: http://influxdb:8086/write?db=metrics verb: POST headers: Content-Type: text/plain batching: count: 100 period: 10s ``` ## Next Steps - **[Generate Input](/components/inputs/generate)** - Scheduled data generation - **[Broker Output](/components/outputs/broker)** - Multi-destination routing - **[HTTP Client Input](/components/inputs/http_client)** - Fetch metrics from APIs - **[Bloblang Guide](/guides/bloblang)** - Data transformation and scoring logic ## Promote Inbound Metadata into the Body # Promote Inbound Metadata into the Body This page collects four worked pipelines that use the [`metadata` processor's `promote` block](/components/processors/metadata#promoting-inbound-metadata-into-the-body) to copy **input-supplied message metadata** into the JSON body as fields. The driving use case is SCADA / IIoT β€” a Sparkplug B device stamps `spb_metric_name`, `spb_datatype`, `spb_timestamp`, … on every metric and emits a scalar value as the body. Without `promote`, every pipeline that lands these to a columnar store has to repeat the same `mapping` block per field. With `promote`, it's one declarative section. The feature is fully general: anywhere an input stamps metadata you want as **columns**, `promote` removes the boilerplate. For the full configuration surface, see the [`promote` field reference](/components/processors/metadata#promoting-inbound-metadata-into-the-body). ## 1. Sparkplug B β€” scalar body, ~20 spb_* keys The canonical SCADA shape. A Sparkplug B input emits each metric as a scalar value (e.g. `23.5`) and stamps the metric name, datatype, timestamp, sequence, alias, quality, source device, and a dozen other fields as `spb_*` metadata. Promote them all in one section, rename a few for column-friendly names, cast the numeric and boolean ones, and stamp Expanso provenance on top. ```yaml pipeline: processors: - metadata: include: [core, pipeline, node] target: body format: flat promote: match: "spb_" # every spb_* key the input stamped rename: spb_metric_name: metric spb_datatype: datatype spb_timestamp: timestamp spb_sequence: sequence spb_is_historical: is_historical spb_alias: alias spb_device_id: device_id cast: spb_timestamp: int # epoch millis -> int64 spb_sequence: int # int alias counter spb_is_historical: bool # "true"/"false" -> bool wrap_key: value # 23.5 -> { "value": 23.5, ... } output: stdout: codec: lines ``` **Input message** (the body is the scalar `23.5`; metadata is set by the input): ```text body: 23.5 metadata: spb_metric_name: "Temperature" spb_datatype: "Float" spb_timestamp: "1730000000000" spb_sequence: "42" spb_is_historical: "false" spb_alias: "17" spb_device_id: "plc-12" ``` **Output**: ```json { "value": 23.5, "metric": "Temperature", "datatype": "Float", "timestamp": 1730000000000, "sequence": 42, "is_historical": false, "alias": "17", "device_id": "plc-12", "run_id": "01HF...", "job_name": "scada-ingest", "job_namespace": "default", "event_time": "2026-04-30T12:34:56.789Z", "producer": "https://expanso.io/edge/v1.4.2", "pipeline_name": "scada-ingest", "pipeline_version": "v3", "node_id": "edge-eu-west-1-a", "hostname": "edge-01", "region": "eu-west-1", "environment": "production", "cluster_name": "edge-eu", "agent_version": "1.4.2" } ``` **Walking through the configuration:** - **`target: body, format: flat`** β€” `promote` requires `target: body`. Flat puts every field at the JSON root so downstream Parquet / Delta sees them as top-level columns. Switch to `format: nested` (with `body_key`) if you want a nested structure. - **`match: "spb_"`** β€” a literal prefix match. `match` matches by prefix unless wrapped in slashes (`/regex/`). - **`rename:`** β€” keyed by the **inbound** metadata key, even when a `cast` for the same key is also configured. - **`cast:`** β€” `int`, `float`, or `bool`. A cast that fails to parse (or, for `float`, `NaN` / `Β±Inf`) falls back to the string value and **does not** increment `error_count` β€” a recoverable cast fallback is not a pipeline error. - **`wrap_key: value`** β€” the body `23.5` is not an object, so `promote` wraps it as `{ "value": 23.5 }` before adding fields. Without `promote`, a non-object body is passed through unchanged with a warning β€” wrapping only happens when `promote` is active. - **Precedence:** original body < promoted fields < resolved Expanso provenance. The resolved `run_id`, `node_id`, `pipeline_name`, … always win, so an inbound `spb_node_id` (if some input stamps one) cannot spoof Expanso's own provenance. ## 2. Kafka headers β€” promote selected headers as columns A Kafka input stamps every header as `kafka_` metadata. Promote only the ones you want as columns, leave the noisy ones alone, and cast the timestamp. ```yaml input: kafka: addresses: [ "broker:9092" ] topics: [ "orders.events" ] consumer_group: "edge-orders" pipeline: processors: - metadata: target: body format: flat promote: names: - kafka_tenant_id - kafka_event_type - kafka_correlation_id - kafka_ingest_timestamp rename: kafka_tenant_id: tenant kafka_event_type: event_type kafka_correlation_id: correlation_id kafka_ingest_timestamp: ingest_ts cast: kafka_ingest_timestamp: int output: aws_s3: bucket: orders-events path: ${! @tenant }/${! timestamp_unix() }.json ``` **Why `names` and not `match`.** Kafka stamps a lot of internal `kafka_*` keys (`kafka_topic`, `kafka_partition`, `kafka_offset`, …). A broad `match: "kafka_"` would pull all of them into the body. Listing the specific headers via `names` is more deliberate and keeps the columnar schema tight. ## 3. MQTT topic parts via regex An MQTT input stamps the full topic as `mqtt_topic`. To break a topic like `building/floor-3/zone-12/temperature` into `building`, `floor`, `zone`, and `metric` columns, use a `mapping` step to split the topic into per-part metadata keys and then promote them with a regex match: ```yaml input: mqtt: urls: [ "tcp://broker:1883" ] topics: [ "building/+/+/+" ] pipeline: processors: # First, split the topic into per-part metadata keys. - mapping: | let parts = metadata("mqtt_topic").split("/") meta topic_building = $parts.index(0).or("") meta topic_floor = $parts.index(1).or("") meta topic_zone = $parts.index(2).or("") meta topic_metric = $parts.index(3).or("") # Then promote everything that starts with topic_ into the body. - metadata: target: body format: flat promote: match: "/^topic_/" # regex: leading slashes are the regex marker rename: topic_building: building topic_floor: floor topic_zone: zone topic_metric: metric wrap_key: payload # MQTT body is opaque bytes β€” wrap as "payload" output: stdout: {} ``` To persist the promoted rows as Parquet, add a `parquet_encode` processor after the promote step and write the encoded output with a `file` or `aws_s3` output. `/^topic_/` is a regex because it is wrapped in slashes. A bare `topic_` would be a literal prefix; both forms work here, but regex gives you the full power of Go's `regexp` package when you need it (lookahead, character classes, alternation). The empty regex `//` is rejected at submission β€” it would match every metadata key. ## 4. Promote `file_path` from the `tail` input The [`tail` input](/components/inputs/tail) stamps every line with `file_path` (absolute) and `file_name` (basename) metadata. Promote them into the body so downstream stores see them as columns, and use the `promote` block alongside lineage stamping: ```yaml input: tail: paths: [ /var/log/myapp/*.log ] pipeline: processors: - mapping: | # Try to parse JSON logs, fall back to raw text. let parsed = this.parse_json().catch(null) root = if $parsed != null { $parsed } else { { "message": this } } - metadata: include: [core, pipeline, node] # also stamp Expanso provenance target: body format: nested body_key: provenance promote: names: [file_path, file_name] rename: file_path: source_file file_name: source_basename output: http_client: url: https://ingest.example.com/v1/logs verb: POST ``` **What you get.** Every emitted message has `source_file` and `source_basename` at the JSON root (alongside whatever the upstream `mapping` step produced), plus a `provenance` object carrying `run_id`, `node_id`, `pipeline_name`, etc. The nested-format precedence rule still applies inside `provenance` (promoted < resolved); the root-level fields from `promote` sit outside the nested object. ## Common variations ### Promote everything except reserved fields `match: "/.+/"` selects every metadata key. The runtime drops any key whose source name or rename target is reserved (`node_id`, `pipeline_id`, `node_label_*`, …) β€” see the [reserved-name guard](/components/processors/metadata#reserved-name-guard) β€” so a permissive match cannot leak Expanso's own provenance. ```yaml promote: match: "/.+/" ``` A bare empty regex (`match: "//"`) is rejected at submission instead of silently being a "match-everything" pattern β€” that ambiguity was deliberately closed in the schema. ### Promote with format: nested If your downstream consumer expects a nested structure rather than flat columns, set `format: nested` and `body_key`. Promoted fields go inside the nested object alongside the resolved Expanso fields: ```yaml - metadata: include: [core, pipeline] target: body format: nested body_key: scada promote: match: "spb_" rename: { spb_metric_name: metric } ``` A scalar body `23.5` becomes: ```json { "value": 23.5, "scada": { "metric": "Temperature", "datatype": "Float", "run_id": "01HF...", "node_id": "edge-eu-west-1-a" } } ``` `wrap_key` defaults to `value`. In nested format, `wrap_key` must not equal `body_key` β€” the nested metadata object would otherwise clobber the wrapped body value, and the validator rejects this at submission. ### Cast a metadata value to float, with NaN-safe fallback A device that occasionally emits `"NaN"` or `"-"` for an unset reading would otherwise drop the entire record (a non-finite float fails the body marshal). The `float` cast deliberately treats `NaN` / `Β±Inf` / out-of-range as a parse failure and falls back to the original string: ```yaml promote: match: "spb_" cast: spb_value: float ``` A reading of `"23.5"` becomes the float `23.5`; a reading of `"NaN"` is kept as the string `"NaN"`. Cast fallbacks are logged at debug only β€” they do **not** increment `error_count`, so the false-positive alert risk on a noisy source is zero. ## Next Steps - **[`metadata` processor reference](/components/processors/metadata)** β€” full configuration reference, including the `promote` field reference, validation rules, and the reserved-name guard. - **[Metadata Enrichment for Lineage example](/examples/metadata-enrichment)** β€” the lineage-focused companion example for the same processor. - **[`tail` input](/components/inputs/tail)** β€” the log-tailing input that stamps `file_path` / `file_name` metadata used in example 4. - **[Pipeline metadata guide](/guides/pipelines/metadata)** β€” the implicit `@pipeline_id`, `@node_id`, and `@node_label_*` keys you do not need `promote` for. - **[Quick Start](/getting-started/quickstart/)** β€” how to actually deploy and run these pipelines on an edge node. ## S3 Intelligent Tiering # S3 Intelligent Tiering Route data to different S3 storage classes based on content, reducing storage costs by 60-80% while keeping critical data instantly accessible. ## The Problem Organizations storing logs, events, or telemetry in S3 often put everything into S3 Standard β€” even data that's rarely accessed. With 5-year retention requirements common for audit and compliance, this adds up fast: | Storage Class | Cost (per TB/month) | Access Time | |---|---|---| | S3 Standard | ~$23.00 | Instant | | S3 Standard-IA | ~$12.50 | Instant | | S3 Glacier Instant Retrieval | ~$4.00 | Milliseconds | | S3 Glacier Deep Archive | ~$1.00 | 12-48 hours | Storing 100 TB for 5 years in Standard costs **~$138,000**. The same data in Glacier Deep Archive costs **~$6,000**. The challenge is routing the right data to the right tier β€” automatically, in real-time. ## How Expanso Solves This Instead of relying on S3 Lifecycle policies (which move data days or weeks after creation), Expanso Edge routes data to the correct storage class **at write time**. Your pipeline inspects each event and decides immediately: does this go to hot storage or cold archive? ```mermaid graph LR A[Log Source] --> B[Expanso Edge] B -->|Errors & Alerts| C[S3 Standard
Instant Access] B -->|Normal Logs| D[S3 Standard-IA
Infrequent Access] B -->|Raw Archive| E[S3 Glacier
Deep Archive] ``` ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/s3-intelligent-tiering.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: s3-intelligent-tiering\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`s3-intelligent-tiering.yaml`](pathname:///examples/s3-intelligent-tiering.yaml)
## Complete Pipeline This pipeline reads log events from Kafka, classifies them by severity, and routes each to the appropriate S3 storage class: ```yaml input: kafka: addresses: - ${KAFKA_BROKERS:localhost:9092} topics: - application-logs consumer_group: expanso-s3-tiering pipeline: processors: # Parse the log event - mapping: | root = this.parse_json() # Classify severity for tiering - mapping: | let level = this.level.lowercase().or("info") root = this root.storage_tier = match $level { "error" => "hot", "fatal" => "hot", "warn" => "warm", _ => "cold" } output: switch: cases: # Errors and fatals β†’ S3 Standard (instant access) - check: this.storage_tier == "hot" output: aws_s3: bucket: ${S3_BUCKET:my-logs} path: hot/${!now().ts_format("2006/01/02/15")}/${!uuid_v4()}.json storage_class: STANDARD region: ${AWS_REGION:us-east-1} batching: count: 50 period: 10s # Warnings β†’ S3 Standard-IA (infrequent access) - check: this.storage_tier == "warm" output: aws_s3: bucket: ${S3_BUCKET:my-logs} path: warm/${!now().ts_format("2006/01/02")}/${!uuid_v4()}.json storage_class: STANDARD_IA region: ${AWS_REGION:us-east-1} batching: count: 500 period: 60s # Everything else β†’ Glacier Deep Archive - check: this.storage_tier == "cold" output: aws_s3: bucket: ${S3_BUCKET:my-logs} path: archive/${!now().ts_format("2006/01/02")}/${!uuid_v4()}.json storage_class: DEEP_ARCHIVE region: ${AWS_REGION:us-east-1} batching: count: 2000 period: 300s ``` ## Configuration Breakdown ### Classification Logic ```yaml - mapping: | let level = this.level.lowercase().or("info") root = this root.storage_tier = match $level { "error" => "hot", "fatal" => "hot", "warn" => "warm", _ => "cold" } ``` This is where the cost savings happen. Each event is tagged with a `storage_tier` based on its severity. You control the rules β€” severity is just one approach. You could also classify by: - **Source application** (billing logs stay hot, debug logs go cold) - **Age of data** (last 24h hot, everything else archive) - **Content patterns** (events matching a regex stay accessible) ### Storage Class Routing The `switch` output routes each event to the correct S3 destination. Notice how batching differs by tier: - **Hot (Standard):** Small batches, frequent writes β€” you need this data fast - **Warm (Standard-IA):** Medium batches β€” balances cost and accessibility - **Cold (Deep Archive):** Large batches, infrequent writes β€” maximizes cost savings The `storage_class` field on the S3 output tells AWS which tier to use at upload time β€” no lifecycle policies needed. **See:** [AWS S3 Output](/components/outputs/aws_s3) for all configuration options. ### Batching for Cost Efficiency Larger batches mean fewer S3 PUT requests and lower costs. Deep Archive data is batched aggressively (2000 events or 5 minutes) since you won't need immediate access: ```yaml batching: count: 2000 # Batch up to 2000 events period: 300s # Or flush every 5 minutes ``` **See:** [Batching Guide](/guides/pipelines/batching) for advanced batching patterns. ## Cost Savings Example Consider an organization processing **10 TB/month** of logs with 5-year retention: | Approach | Monthly Cost | 5-Year Cost | |---|---|---| | All S3 Standard | $230/TB Γ— 10 TB = **$2,300** | **$138,000** | | With intelligent tiering (80% cold, 15% warm, 5% hot) | **$225** | **$13,500** | That's a **~90% reduction** in storage costs β€” just by routing data to the right tier at write time. :::note[Real-world results vary] Your savings depend on the ratio of hot/warm/cold data. Most organizations find that 70-90% of log data is rarely accessed after the first 24 hours, making this pattern highly effective. ::: ## Common Variations ### Add Compression Before Archiving Stack gzip compression with cold storage for maximum savings: ```yaml # In the cold storage output output: aws_s3: bucket: ${S3_BUCKET:my-logs} path: archive/${!now().ts_format("2006/01/02")}/${!uuid_v4()}.json.gz storage_class: DEEP_ARCHIVE content_encoding: gzip region: ${AWS_REGION:us-east-1} batching: count: 2000 period: 300s processors: - compress: algorithm: gzip ``` ### Dual-Write: Hot Summary + Cold Archive Keep a summarized version in Standard for dashboards while archiving the full raw data: ```yaml output: broker: outputs: # Full raw data β†’ Deep Archive - aws_s3: bucket: ${S3_BUCKET:my-logs} path: raw/${!now().ts_format("2006/01/02")}/${!uuid_v4()}.json storage_class: DEEP_ARCHIVE region: ${AWS_REGION:us-east-1} batching: count: 2000 period: 300s # Summarized data β†’ Standard (for dashboards) - aws_s3: bucket: ${S3_BUCKET:my-logs} path: summary/${!now().ts_format("2006/01/02/15")}/${!uuid_v4()}.json storage_class: STANDARD region: ${AWS_REGION:us-east-1} batching: count: 100 period: 30s # Shape the summary at flush time, on this branch only processors: - mapping: | root.timestamp = this.timestamp root.level = this.level root.service = this.service root.message = this.message.slice(0, 200) ``` ### Use Kinesis Instead of Kafka If your logs already flow through AWS Kinesis (common in multi-account AWS setups), swap the input: ```yaml input: aws_kinesis: streams: - arn:aws:kinesis:us-east-1:123456789:stream/application-logs dynamodb: table: expanso-kinesis-checkpoints ``` The rest of the pipeline stays the same. **See:** [Kinesis Input](/components/inputs/aws_kinesis) ## Next Steps - **[AWS S3 Output](/components/outputs/aws_s3)** β€” Full S3 configuration options including `storage_class` - **[Kafka to S3](/examples/kafka-to-s3)** β€” Simpler Kafka-to-S3 pipeline without tiering - **[Batching Guide](/guides/pipelines/batching)** β€” Optimize batch sizes for cost and throughput - **[Log Processing Use Case](/use-cases/log-processing)** β€” Broader log processing patterns - **[AWS Guide](/guides/cloud/aws)** β€” AWS credentials and regional configuration ## Sign Pipeline Messages # Sign Pipeline Messages This example demonstrates how to sign every pipeline message with the edge node's Ed25519 identity and fan it out to two outputs at once β€” an HTTP endpoint that carries the signature as transport headers, and a file sink that carries the signature inside the body. It uses a single [`signature`](/components/processors/signature) processor in `target: both` mode plus a [`switch`](/components/outputs/switch) output. ## Download & Run
**Quick Start:** ```bash # 1. Download the pipeline and check it offline curl -sSLo pipeline.yaml \ https://docs.expanso.io/examples/signing-and-verifying.yaml expanso-edge validate pipeline.yaml # 2. Start a local edge agent (its state stays in ./.edge) expanso-edge run --local --data-dir ./.edge # 3. In a second terminal, deploy the pipeline to it as a job export EXPANSO_CLI_CONFIG_DIR="$PWD/.expanso-cli" { printf 'name: signing-and-verifying\ntype: pipeline\nconfig:\n' sed 's/^/ /' pipeline.yaml; } | expanso-cli --endpoint http://localhost:9010 job deploy - ``` This runs on your machine with no account. Step 3 wraps the pipeline in the `name`, `type` and `config` fields that a job needs, and a `stdout` output writes to the agent's terminal. If the pipeline reads environment variables, export them in the terminal where you start the agent. See [Local Mode](/getting-started/local-mode) for more, or [Deploy to Cloud](/getting-started/deploy-to-cloud) to run it on managed nodes. **Download:** [`signing-and-verifying.yaml`](pathname:///examples/signing-and-verifying.yaml)
## What This Pipeline Does 1. **Generates** a synthetic event every second (no external dependencies) 2. **Signs** each event with the edge node's Ed25519 identity, writing the signature to both metadata and the JSON body 3. **Routes** each event by its `destination` field via a `switch` output 4. **Delivers** matching events to an HTTP endpoint, propagating the signature as headers 5. **Persists** matching events to a JSON-Lines file, where the signature lives inside the body and the record is verifiable on its own ## Complete Pipeline ```yaml input: generate: interval: 1s mapping: | root.temp_c = 21.5 root.sensor_id = 42 root.marker = "sig-example" root.destination = if random_int(min: 0, max: 1) == 0 { "http" } else { "file" } pipeline: processors: - signature: target: both body_format: nested body_key: _signature on_signing_error: fail output: switch: cases: - check: this.destination == "http" output: http_client: url: https://example.com/ingest verb: POST metadata: include_patterns: - "^expanso_" - check: this.destination == "file" output: file: path: /var/log/signed-events.jsonl codec: lines ``` ## Configuration Breakdown ### Input: Generate ```yaml input: generate: interval: 1s mapping: | root.temp_c = 21.5 root.sensor_id = 42 root.marker = "sig-example" root.destination = if random_int(min: 0, max: 1) == 0 { "http" } else { "file" } ``` A self-contained synthetic event so the pipeline runs out of the box. The `destination` field exists only to demonstrate the `switch` output β€” in a real deployment your routing logic would be driven by upstream data. **See:** [Generate Input](/components/inputs/generate) | [Bloblang Guide](/guides/bloblang) ### Signing: `signature` with `target: both` ```yaml pipeline: processors: - signature: target: both body_format: nested body_key: _signature on_signing_error: fail ``` A single signing step covers both sinks: - `target: both` writes the signature to **message metadata** (three `expanso_*` keys) **and** splices it into the JSON body under `_signature`. Each sink picks up whichever form it can carry. - `body_format: nested` puts the signature material under a single root key (`_signature`) holding `alg`, `kid`, and `sig`. The other layout β€” `flat` β€” would put three keys at the JSON root instead. - `on_signing_error: fail` makes the pipeline reject unsigned messages rather than letting them through. Combined with your standard [error handling](/guides/pipelines/error_handling), failing messages can be dead-lettered or retried. Placement matters. The `signature` processor sits at the **end** of the pipeline, so the signature attests to the message as it leaves the node. Moving it before any further transformation would attest to the raw ingest payload instead. See [Pipeline position is the signature's meaning](/components/processors/signature#pipeline-position-is-the-signatures-meaning). **See:** [`signature` processor reference](/components/processors/signature) ### Output branch: HTTP with header propagation ```yaml - check: this.destination == "http" output: http_client: url: https://example.com/ingest verb: POST metadata: include_patterns: - "^expanso_" ``` The `http_client` output ships the body as POST data and exposes selected metadata as transport headers. The `^expanso_` regex matches the three signature keys (`expanso_signature`, `expanso_keyid`, `expanso_signature_alg`) and propagates them as headers β€” the receiver sees the unmodified ingest body plus three signature headers (HTTP headers are case-insensitive, so they may arrive as `expanso_signature` or `Expanso-Signature` depending on the receiver). This branch demonstrates **why metadata mode exists**: the body on the wire is the original ingest bytes, and a verifier reconstructs the canonical form on its own by parsing and re-marshalling with sorted keys before calling Ed25519.Verify. :::warning[HTTP header names with underscores] Some HTTP servers and proxies (nginx with the default `underscores_in_headers off`, AWS ALB, parts of the CDN ecosystem) silently strip or rewrite request headers whose names contain underscores. If you ship behind any such infrastructure, rename the keys to use hyphens β€” for example, `metadata_signature_key: x-expanso-signature` β€” and configure your verifier to match. ::: **See:** [HTTP Client Output](/components/outputs/http_client) ### Output branch: File sink with body-embedded signature ```yaml - check: this.destination == "file" output: file: path: /var/log/signed-events.jsonl codec: lines ``` The `file` output writes one JSON object per line. File records have no headers, so the metadata-only signature would be lost. Because we used `target: both`, the signature also lives **inside** the body as the `_signature` key, and a downstream tool reading the JSON-Lines file can verify each record on its own with nothing but the file contents and the producing node's public key. This branch demonstrates **why body mode exists**: any consumer that does not carry transport headers β€” file, object store, archives, replay logs β€” needs the signature inside the record. **See:** [File Output](/components/outputs/file) | [Switch Output](/components/outputs/switch) ## What a Consumer Sees For an input event of: ```json {"temp_c":21.5,"sensor_id":42,"marker":"sig-example","destination":"http"} ``` The **HTTP receiver** gets an unchanged body plus three headers: ```http POST /ingest HTTP/1.1 Content-Type: application/json Expanso_signature_alg: Ed25519 Expanso_keyid: node:a1b2c3d4-5e6f-7890-abcd-ef1234567890#sha256:EftgCusZmspbrXBf1Gx8jnomxQdxTr0O6pI9jhqex3w Expanso_signature: uSRm83nRNPKfOG/8P4WS/Vwb9Cnty9OF1Kgk7cxgZZZyCITfRwUJXEkOXMRnrV+yO/hlGhbroF1RQzRbPVvLAw== {"temp_c":21.5,"sensor_id":42,"marker":"sig-example","destination":"http"} ``` The **file consumer** gets self-contained JSON-Lines records: ```json {"_signature":{"alg":"Ed25519","kid":"node:a1b2c3d4-...#sha256:nVaBCZ...quU","sig":"Vb0/zoMrfNlG...mzRm1FnBw=="},"destination":"file","marker":"sig-example","sensor_id":42,"temp_c":21.5} ``` Notice that JSON object keys appear in **byte-wise lexicographic order** in the canonical form a verifier reconstructs (the order Go's `encoding/json` produces). This is intentional β€” the producer round-trips through a sorted-keys representation before signing, so a verifier that does the same round-trip will recover identical bytes regardless of the original field order. ## Common Variations ### Use `target: meta` only When all of your outputs propagate metadata as headers (HTTP, Kafka, NATS), drop the body splice entirely: ```yaml pipeline: processors: - signature: {} # target: meta is the default ``` ### Use `target: body` only When **no** sink in your fan-out carries transport headers β€” e.g. file, S3, GCS β€” keep the signature inside the body: ```yaml pipeline: processors: - signature: target: body ``` ### Sign the raw ingest payload instead of the enriched output Move the `signature` step **before** any enrichment. The signature then attests to exactly what the upstream producer emitted, which is what regulated workloads usually want: ```yaml pipeline: processors: - signature: {} - mapping: | root.enriched_at = now() root.region = "eu-west-1" ``` ### Rename the metadata keys If your payload already uses one of the default key names, rename them β€” but remember the verifier must use the same names: ```yaml pipeline: processors: - signature: target: both metadata_signature_key: x_sensor_signature metadata_keyid_key: x_sensor_keyid metadata_algorithm_key: x_sensor_alg body_key: _attestation ``` ## Verifying Signatures Downstream A consumer that wants to validate signed messages needs to apply the same canonicalisation rule the producer used, then call any Ed25519 verifier. Pseudocode: ```text 1. Recover canonical bytes: - If signature was in metadata: parse body as JSON with UseNumber, re-marshal with sorted keys. - If signature was in body (nested): parse, extract alg/kid/sig from `_signature`, delete that root key, re-marshal the remaining map with sorted keys. 2. Resolve the public key from `kid` (format: node:#sha256:). 3. ok := ed25519.Verify(publicKey, canonicalBytes, base64Decode(sig)) ``` The single most common failure mode is canonicalisation drift β€” verifiers that don't sort keys, that lose `int64` precision through a `float64` conversion, or that don't delete the body-embedded signature before re-marshalling. See the [signature reference's Verification section](/components/processors/signature#verification) for the complete protocol. ## Next Steps - **[`signature` processor reference](/components/processors/signature)** β€” every field, validator rule, and the full Verification protocol. - **[Quick Start](/getting-started/quickstart/)** β€” how to actually deploy and run this pipeline on an edge node. - **[`metadata` processor](/components/processors/metadata)** β€” chain with `signature` so signatures cover enriched lineage events. - **[Bloblang Guide](/guides/bloblang)** β€” when you need to compute fields before signing. ## Tail Log Files With Rotation and Restart-Safe Checkpoints # Tail Log Files With Rotation and Restart-Safe Checkpoints This page collects five end-to-end pipelines that use the [`tail`](/components/inputs/tail) input to follow local log files. Each pipeline is a working configuration you can drop into a job spec; the surrounding processors and outputs are kept minimal so the focus stays on the tailing configuration. For the full configuration surface, see the [`tail` input reference](/components/inputs/tail). ## 1. Basic application log tail The simplest possible deployment: follow every `.log` file under one directory, forward each line unchanged to stdout. On the first run, only newly appended lines are forwarded (`start_at: end` is the default). Every subsequent restart resumes from the persisted checkpoint, so the edge can be patched, restarted, or crash-rebooted with no data loss past whatever is already in flight. ```yaml input: tail: paths: [ /var/log/myapp/*.log ] pipeline: processors: [] output: stdout: {} ``` **What you get.** One message per appended line. Each message carries `file_path` and `file_name` metadata identifying the source file, so a multi-file glob can be routed downstream by source. The checkpoint lives at `/executions//state/tail//`, anchored to the pipeline so re-deploying under a new pipeline ID starts fresh by design. **On rotation.** When `logrotate` renames `app.log β†’ app.log.1` and creates a fresh `app.log`, the existing file is finished cleanly before its state is dropped, and the new file is picked up at offset 0. Up to three generations of rotated files are tracked concurrently, so a fast-rotating tail (size-based rotation under heavy write) never silently misses lines. ## 2. Multi-line Java stack traces Java logs ship every stack trace as a header line plus dozens of indented frame lines. Treating each line as a message scatters a single error across many records and breaks downstream parsing. Use `multiline` to reassemble them. ```yaml input: tail: paths: [ /var/log/myapp/*.log ] multiline: line_start_pattern: '^\d{4}-\d{2}-\d{2}' pipeline: processors: - mapping: | root.timestamp = this.re_find_all_object("^(?P\\S+ \\S+)").index(0).ts.or("") root.level = this.re_find_all_object("\\] (?P[A-Z]+) ").index(0).lvl.or("INFO") root.message = this root.source_file = metadata("file_path") output: stdout: {} ``` **What `line_start_pattern` does.** Every line that does **not** match `^\d{4}-\d{2}-\d{2}` (a `YYYY-MM-DD` timestamp at column zero) is appended to the previous entry. So: ``` 2026-05-22 14:23:11 [main] ERROR Could not connect java.net.ConnectException: Connection refused at sun.nio.ch.SocketChannelImpl... at java.base/java.net... 2026-05-22 14:23:12 [main] INFO Retry in 5s ``` becomes two messages: the ERROR (header + 3 stack frames) and the INFO. A trailing partial entry without a following match is held until the next match arrives, the file rotates, or the input shuts down. **`start_at` and partial first entries.** Default `start_at: end` skips existing content; on first run the first multiline entry may be partial if the input attaches mid-message. Set `start_at: beginning` when capturing the whole file matters more than skipping the existing tail. ## 3. JSON logs with a text fallback Many applications emit JSON logs in production but fall back to plain text in development or when something goes wrong before the logger initializes. Parse the JSON when present; preserve the text otherwise. ```yaml input: tail: paths: [ /var/log/myapp/*.log ] pipeline: processors: - mapping: | # Try to parse JSON; fall back to wrapping the raw line as { "message": ... }. let parsed = this.parse_json().catch(null) root = if $parsed != null { $parsed } else { { "message": this } } # Add tail-supplied identity to every record. root.ingested_at = now() root.source_file = metadata("file_path") - mapping: | # Drop empty / whitespace-only lines (common around log restarts). root = if this.message.or("").trim() == "" { deleted() } output: stdout: {} ``` **Drop ack semantics.** Lines dropped with `deleted()` in the `mapping` processor still **ack** the input, so `tail` correctly advances its checkpoint past them. Only a true downstream nack (an output error, a `mapping` processor returning a non-nil error, etc.) holds the watermark. This is the behavior most pipelines want: dropping is not data loss. ## 4. Multiple log sources, routed by file_path A single edge node often tails several applications at once. Globs and explicit paths can be combined in one `paths` list; downstream can discriminate by the per-message `file_path` and `file_name` metadata. ```yaml input: tail: paths: - /var/log/nginx/*.log - /var/log/myapp/server.log - /var/log/myapp/audit-*.log exclude: - /var/log/nginx/*.gz # skip rotated archives pipeline: processors: - mapping: | root.line = this root.source_file = metadata("file_path") root.app = if metadata("file_path").contains("/nginx/") { "nginx" } else if metadata("file_path").contains("/audit-") { "audit" } else { "server" } output: switch: cases: - check: this.app == "nginx" output: aws_s3: bucket: prod-edge-logs path: nginx/${! @file_name }-${! timestamp_unix() }.jsonl - check: this.app == "audit" output: file: path: /var/lib/expanso/audit-trail.jsonl codec: lines - check: this.app == "server" output: kafka: addresses: [ "broker:9092" ] topic: edge.server.logs ``` **Why one input, not three.** A single `tail` shares one polling loop, one in-memory queue, and one checkpoint key per (sorted) path set β€” lower memory, fewer goroutines, and a single restart resume point. Use multiple `tail` inputs only when you need genuinely independent checkpoint identities (different retention policies, separately resettable state). ## 5. Hardened production setup with explicit checkpoint identity When the same log set is consumed by long-lived pipelines, set an explicit `checkpoint_id` so the checkpoint key is **stable across config edits**. Adding or removing files from `paths` then no longer invalidates the checkpoint, and you can bump the id deliberately to start fresh. ```yaml input: tail: paths: - /var/log/myapp/*.log - /var/log/myapp/sidecar-*.log # Explicit identity. Path edits keep the same checkpoint; bumping this # value (e.g. "myapp-logs-v1" β†’ "myapp-logs-v2") starts fresh. checkpoint_id: myapp-logs-v1 start_at: end poll_interval: 200ms encoding: utf-8 max_log_size: 10MiB # raise for known-large JSON blobs multiline: line_start_pattern: '^\d{4}-\d{2}-\d{2}' # Default: re-deliver nacked batches automatically. Set to false to make # a permanent downstream rejection visibly stall the input (no data loss, # but the pipeline halts until you fix the rejection). auto_replay_nacks: true pipeline: processors: - mapping: | # Build a deterministic event id so a downstream replay can dedupe. root.event_id = (metadata("file_path") + ":" + content().hash("sha256").encode("hex")).slice(0, 40) root.line = this root.source = { "file": metadata("file_path"), "name": metadata("file_name") } root.ingested_at = now() - signature: {} # sign every line with the edge node's identity output: http_client: url: https://ingest.example.com/v1/logs verb: POST metadata: include_patterns: - "^expanso_" # propagate the signature as transport headers ``` **Why each piece matters.** - **`checkpoint_id: myapp-logs-v1`** β€” survives edits to `paths`. Without it, adding `sidecar-*.log` would change the implicit key (it's a hash of the sorted paths) and the input would re-tail from `start_at`. - **`max_log_size: 10MiB`** β€” raised from the 1 MiB default for known-large JSON-blob log lines. Entries larger than this are split at the boundary rather than dropped. - **`auto_replay_nacks: true`** β€” the default; explicit here for visibility. With it off, a permanent downstream rejection (e.g. a malformed message that always fails validation) would stall the checkpoint rather than silently dropping the data β€” a deliberate trade-off that surfaces in monitoring instead of losing lines. - **Deterministic `event_id` from `file_path` + content hash** β€” paired with a downstream store keyed by this id, this gives effective deduplication on top of `tail`'s at-least-once delivery. A crash-induced re-delivery of an already-stored line is a no-op. - **`signature: {}` after the mapping step** β€” signs every line with the edge node's Ed25519 identity. Downstream verifies that the line came from the producing edge and was not altered after `tail` emitted it. See [Sign Pipeline Messages](/examples/signing-and-verifying). ## Common variations ### Read each log file from the beginning on first run The default `start_at: end` skips existing content. To capture the whole file (e.g. for one-time ingest of an existing log directory): ```yaml input: tail: paths: [ /var/log/myapp/*.log ] start_at: beginning ``` Once a checkpoint exists, it always overrides `start_at`. Restarting with `start_at: beginning` still resumes from the last delivered byte β€” you cannot accidentally re-read a file just by toggling the field. ### Tune the poll interval for high-throughput logs The default `poll_interval: 200ms` is a fine middle ground. For latency-sensitive feeds (live alerting), drop it to `50ms`; for chatty file systems where the polling cost matters (many `tail` inputs on one node), raise to `1s`: ```yaml input: tail: paths: [ /var/log/critical-app/*.log ] poll_interval: 50ms ``` ### Read non-UTF-8 files Legacy Windows logs and some European-locale systems still emit `windows-1252` or `latin-1`. The encoding is decoded at the byte boundary before line splitting, so multibyte characters that straddle reads are handled correctly: ```yaml input: tail: paths: [ /var/log/legacy/*.log ] encoding: windows-1252 ``` An unsupported encoding name fails at submit, not at pipeline start. ### Group with a terminator instead of a header For log formats where each entry ends with a known marker: ```yaml input: tail: paths: [ /var/log/myapp/transactions.log ] multiline: line_end_pattern: '^---END---$' ``` Only one of `line_start_pattern` or `line_end_pattern` may be set; the validator rejects configs that set both. ## Next Steps - **[`tail` input reference](/components/inputs/tail)** β€” every field, default, validation rule, and the full at-least-once delivery / checkpoint contract. - **[`signature` processor](/components/processors/signature)** β€” sign tailed log lines so downstream can verify the producing edge. - **[Bloblang guide](/guides/bloblang)** β€” parse, filter, and reshape log lines. - **[Pipeline error handling](/guides/pipelines/error_handling)** β€” route lines that fail parsing to a dead-letter sink. - **[Quick Start](/getting-started/quickstart/)** β€” how to actually deploy and run these pipelines on an edge node. ## Verify Lineage Event Signatures # Verify Lineage Event Signatures Every [OpenLineage event emitted by Expanso Edge](/guides/pipelines/lineage) carries an Ed25519 signature under `run.facets.signature`. The signature proves which edge node produced the event and that the payload was not modified in transit. This page shows how to verify that signature in your downstream consumer. The verification protocol is byte-precise. Most verifier bugs come from drift between the signer's canonical form and the verifier's reconstruction β€” wrong number type, wrong key ordering, wrong escape rule for `<`, `>`, or `&`. The reference implementations below match the signer byte-for-byte; deviate from them carefully. ## What a signed event looks like A `START` event with the signature facet expanded: ```json { "eventType": "START", "eventTime": "2026-05-22T14:23:11.123456789Z", "producer": "https://expanso.io/edge/v1.2.3", "schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json", "run": { "runId": "abcdef00-1111-2222-3333-444455556666", "facets": { "nominalTime": { "_producer": "...", "_schemaURL": "...", "nominalStartTime": "..." }, "expanso_run": { "_producer": "...", "_schemaURL": "...", "fields": { "...": "..." } }, "signature": { "_producer": "https://expanso.io/lineage/signature-facet/v1", "_schemaURL": "https://expanso.io/spec/lineage/signature.json", "algorithm": "Ed25519", "keyId": "node:7f769f72-a4d4-4a05-8082-d63262957a6f#sha256:nVaBCZ-XWaG8mO0c9VgvVx_quU", "payloadHash": "sha256:8a7b6c5d4e3f2g1h0i9j8k7l6m5n4o3p2q1r0s9t8u7v6w5x4y3z2", "signature": "Vb0/zoMrfNlG7q4mzRm1FnBwSCAJUq+wzD2Te6CrgZJrjMHC7nKHSqMOj6PgPGGevtmzRm1FnBw==" } } }, "job": { "namespace": "production", "name": "log_processor" }, "inputs": [ { "namespace": "kafka://broker:9092", "name": "events.in" } ], "outputs": [ { "namespace": "s3://my-bucket", "name": "logs/${!timestamp_unix()}.json" } ] } ``` ## The verification protocol To verify the signature, reconstruct the **canonical bytes** that the signer signed and pass them to `ed25519.Verify` along with the decoded signature. The signer signs the result of `json.Marshal` applied to the event-as-a-map **with the `signature` facet removed**. To reproduce that exactly: 1. **Parse the event JSON into a map**, preserving integer precision. In Go, use `json.Decoder` with `UseNumber()`. Without this, integer fields like byte counts above 2^53 lose precision through `float64` and the canonical bytes diverge β€” verification will fail for any event with a large counter. 2. **Delete `map["run"]["facets"]["signature"]`** from the map. The signature facet is excluded from the bytes being signed. 3. **Re-marshal the map to JSON** with sorted keys and Go-style HTML escaping. Go's `encoding/json` sorts map keys automatically; you only need to match its escape rules. The signer escapes `<` as `<`, `>` as `>`, `&` as `&`, U+2028 as `
`, and U+2029 as `
`. All other non-ASCII characters pass through as UTF-8 bytes. 4. **Base64-decode the signature value** with **standard** base64 (with padding β€” `base64.StdEncoding` in Go, the default in most languages). 5. **Verify** with the producing node's public key: `ed25519.Verify(pub, canonicalBytes, signature)`. The `payloadHash` field on the signature facet is an audit aid only. It records `sha256(canonicalBytes)` so an auditor reviewing the wire payload can recompute the hash for cross-referencing. Verifiers do **not** use it as the signature input β€” the input is always the canonical bytes themselves. ### Field encodings | Field | Encoding | |-------|----------| | `signature` | `base64.StdEncoding` (with `=` padding) | | `payloadHash` | `sha256:` prefix followed by `base64.RawURLEncoding` (URL-safe alphabet, no padding) | | `keyId` | `node:#sha256:` where the thumbprint is `base64.RawURLEncoding` of `SHA-256(canonical-JWK)` | ## Go reference verifier ```go package main import ( "bytes" "crypto/ed25519" "encoding/base64" "encoding/json" "errors" "fmt" "io" ) // VerifyLineageEvent verifies the Ed25519 signature on an OpenLineage event // emitted by Expanso Edge. eventJSON is the raw JSON bytes as received from // the wire. pubKey is the producing edge node's Ed25519 public key, resolved // out-of-band from the signature facet's keyId. func VerifyLineageEvent(eventJSON []byte, pubKey ed25519.PublicKey) error { // 1. Parse into a map, preserving integer precision via UseNumber. dec := json.NewDecoder(bytes.NewReader(eventJSON)) dec.UseNumber() var event map[string]any if err := dec.Decode(&event); err != nil { return fmt.Errorf("parse event: %w", err) } // Reject trailing junk that would slip past Decode silently. var rest any if err := dec.Decode(&rest); err != io.EOF { return errors.New("event has trailing content after JSON value") } // 2. Extract the signature facet and remove it from the map. run, ok := event["run"].(map[string]any) if !ok { return errors.New("event missing run object") } facets, ok := run["facets"].(map[string]any) if !ok { return errors.New("event missing run.facets") } sigFacet, ok := facets["signature"].(map[string]any) if !ok { return errors.New("event missing signature facet") } delete(facets, "signature") // 3. Re-marshal the map. json.Marshal on a map sorts keys and applies // Go's default HTML escaping β€” exactly what the signer did. canonical, err := json.Marshal(event) if err != nil { return fmt.Errorf("marshal canonical bytes: %w", err) } // 4. Decode the base64 signature with StdEncoding (padded). sigB64, ok := sigFacet["signature"].(string) if !ok { return errors.New("signature facet missing signature field") } sig, err := base64.StdEncoding.DecodeString(sigB64) if err != nil { return fmt.Errorf("decode signature: %w", err) } // 5. Verify. if !ed25519.Verify(pubKey, canonical, sig) { return errors.New("ed25519 signature does not verify") } return nil } ``` The Go verifier matches the signer byte-for-byte because both call the same `json.Marshal` on a `map[string]any`. There is no encoding subtlety to manage. ## Python reference verifier Python verification is more delicate because Python's `json.dumps` does not match Go's `encoding/json.Marshal` out of the box. Two differences must be corrected: - Python's `json.dumps` does **not** escape `<`, `>`, or `&` by default. Go does. Apply the substitution after dumping. - Python's default `ensure_ascii=True` escapes **all** non-ASCII characters as `\uXXXX`. Go passes most non-ASCII through as raw UTF-8. Disable with `ensure_ascii=False`. The verifier below applies the corrections in the order Go applies them. ```python import base64 import json from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey from cryptography.exceptions import InvalidSignature def _go_escape(payload: bytes) -> bytes: """ Apply Go encoding/json's default HTML/JSONP escapes to JSON bytes that were produced with ensure_ascii=False. Go escapes the five characters below; Python does not. """ return ( payload .replace(b"<", b"\\u003c") .replace(b">", b"\\u003e") .replace(b"&", b"\\u0026") .replace(b"\xe2\x80\xa8", b"\\u2028") # U+2028 LINE SEPARATOR .replace(b"\xe2\x80\xa9", b"\\u2029") # U+2029 PARAGRAPH SEPARATOR ) def verify_lineage_event(event_json: bytes, public_key: Ed25519PublicKey) -> None: """ Verify the Ed25519 signature on an Expanso Edge OpenLineage event. Raises InvalidSignature on verification failure. """ # 1. Parse. Python ints are arbitrary precision, so no UseNumber-equivalent # is needed β€” but we must reject trailing junk to match the Go contract. event = json.loads(event_json) if not isinstance(event, dict): raise ValueError("event is not a JSON object") # 2. Extract and remove the signature facet. sig_facet = event["run"]["facets"].pop("signature") # 3. Re-marshal with sorted keys, no whitespace, raw UTF-8 β€” then apply # Go's HTML/JSONP escape pass to produce byte-identical canonical # bytes. canonical = json.dumps( event, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False, ).encode("utf-8") canonical = _go_escape(canonical) # 4. Decode the signature (standard base64, with padding). sig = base64.b64decode(sig_facet["signature"]) # 5. Verify. cryptography raises InvalidSignature on failure. public_key.verify(sig, canonical) ``` ## Test your verifier against a real event Before deploying a verifier, prove byte-for-byte agreement with the signer using a real event. The single highest-value test is: 1. Capture one signed event from a running edge β€” write to a file with the file transport, then read one line. 2. Run both verifiers (or your verifier and the Go reference) over the same bytes. 3. Confirm both reconstruct identical canonical bytes β€” `sha256(canonical)` must match between languages, and both must match the `payloadHash` field on the signature facet (minus the `sha256:` prefix and after re-encoding with standard base64). 4. Confirm both verifiers return success against the signature. If `sha256(canonical)` matches `payloadHash` but verification fails, the signature or public key is wrong. If `sha256(canonical)` does not match `payloadHash`, the canonicalization is wrong β€” re-check the escape rules and key sorting in step 3. ## Discovering the public key The `keyId` field on the signature facet identifies the signing node: ``` node:7f769f72-a4d4-4a05-8082-d63262957a6f#sha256:nVaBCZ-XWaG8mO0c9VgvVx_quU ``` The nodeID is between `node:` and `#`. Resolve the public key out of band β€” for example, by fetching the node's published JWK from a registry, or by trust-on-first-use against a key set maintained by your operator. Key distribution is a deployment concern, not part of the signature protocol itself. The same `keyId` format and signing core are used by the [`signature` processor](/components/processors/signature). The [Sign Pipeline Messages](/examples/signing-and-verifying) example covers the same key-resolution problem in the per-message context. ## Common failure modes - **`float64` precision loss.** Verifier uses a default JSON decoder, large byte counters in `expanso_run.fields` parse as `float64`, canonical bytes diverge. In Go, set `UseNumber()`. In Python, use `json.loads` (which preserves int precision natively). - **Unsorted keys.** Verifier marshals through a hash map that does not sort, or builds JSON manually. Sort every nested object's keys recursively. Go's `json.Marshal` on a `map[string]any` does this for you. - **HTML-escape drift in Python.** Verifier uses `json.dumps` defaults and gets unescaped `<`, `>`, `&`. Apply the `_go_escape` substitution shown above. - **`ensure_ascii` drift in Python.** Verifier uses `json.dumps` with `ensure_ascii=True` (default) and escapes all non-ASCII; Go does not. Set `ensure_ascii=False`. - **Wrong base64 alphabet.** Verifier decodes `signature` with URL-safe base64. The signature uses **standard** base64 with padding (`base64.StdEncoding`). Only `payloadHash` uses URL-safe base64, and verifiers do not need to decode it. - **Failing to remove the signature facet before re-marshaling.** The signer signs bytes that do not include the facet; the verifier must reconstruct the same bytes. ## Related - [OpenLineage Emission how-to](/guides/pipelines/lineage) β€” enable the emitter on the edge. - [Lineage Configuration](/references/configuration/edge-config) β€” full config reference. - [`signature` processor](/components/processors/signature) β€” per-message signing using the same Ed25519 identity and `keyId` format. - [Sign Pipeline Messages](/examples/signing-and-verifying) β€” sign and verify message bodies, with the same canonicalization concerns. ## Full Reference See https://docs.expanso.io/examples for complete documentation.