End-to-End Examples

Complete pipeline configurations demonstrating edx_code in real-world scenarios.

  3 minute read  

Complete Pipeline Example (Code Processor)

This configuration demonstrates a complete pipeline that parses JSON logs, uses the Code Processor for complex transformations, and forwards the results to a destination.

Note: While you write multi-line JavaScript in the Code Processor UI editor, it gets serialized as a single-line edx_code() statement in YAML:

- name: kubernetes_input_busy
  type: kubernetes_input
  user_description: Application Logs
  include:
  - k8s.namespace.name=production
  log_parsing_mode: full

- name: transform_sequence
  type: sequence
  user_description: Data Transformation Pipeline
  processors:
  - type: ottl_transform
    metadata: '{"id":"data-prep","type":"ottl_transform","name":"Extract Fields"}'
    data_types:
    - log
    statements: |
      // Extract fields from parsed JSON body
      set(attributes["user_id"], body["user"]["id"])
      set(attributes["event_type"], body["event_type"])
      set(attributes["score"], body["score"])
      set(attributes["level"], body["level"])

      // Type conversion for numeric operations
      set(attributes["score"], Int(attributes["score"]))      

  - type: ottl_transform
    metadata: '{"id":"transform","type":"ottl_transform","name":"Advanced Transforms (Code Processor)"}'
    data_types:
    - log
    statements: edx_code("item['attributes']['score_category'] = item['attributes']['score'] > 80 ? 'excellent' : item['attributes']['score'] > 50 ? 'good' : 'needs_improvement'; if (item['attributes']['level'] === 'ERROR' || item['attributes']['level'] === 'CRITICAL') { if (!item['attributes']['alert']) item['attributes']['alert'] = {}; item['attributes']['alert']['user_id'] = item['attributes']['user_id']; item['attributes']['alert']['severity'] = item['attributes']['level']; item['attributes']['alert']['timestamp'] = new Date().toISOString(); } item['attributes']['summary'] = `User ${item['attributes']['user_id']}: ${item['attributes']['event_type']} - ${item['attributes']['score_category']}`;")

- name: http_output
  type: http_output
  user_description: Forward to Analytics
  endpoint: https://analytics.example.com/logs
  use_legacy_formatting: false

In the Code Processor UI, you would write:

// Categorize score
item['attributes']['score_category'] = item['attributes']['score'] > 80 ? 'excellent' : item['attributes']['score'] > 50 ? 'good' : 'needs_improvement';

// Create alert object if critical
if (item['attributes']['level'] === 'ERROR' || item['attributes']['level'] === 'CRITICAL') {
  if (!item['attributes']['alert']) item['attributes']['alert'] = {};
  item['attributes']['alert']['user_id'] = item['attributes']['user_id'];
  item['attributes']['alert']['severity'] = item['attributes']['level'];
  item['attributes']['alert']['timestamp'] = new Date().toISOString();
}

// Generate summary
item['attributes']['summary'] = `User ${item['attributes']['user_id']}: ${item['attributes']['event_type']} - ${item['attributes']['score_category']}`;

Input Sample

{
  "user": {
    "id": "user-123",
    "email": "test@example.com"
  },
  "event_type": "login",
  "score": 85,
  "level": "INFO"
}

Output Sample

{
  "user_id": "user-123",
  "event_type": "login",
  "score": 85,
  "level": "INFO",
  "score_category": "excellent",
  "summary": "User user-123: login - excellent"
}

If the level were “ERROR”:

{
  "user_id": "user-123",
  "event_type": "login",
  "score": 85,
  "level": "ERROR",
  "score_category": "excellent",
  "alert": {
    "user_id": "user-123",
    "severity": "ERROR",
    "timestamp": "2025-09-28T14:23:45.678Z"
  },
  "summary": "User user-123: login - excellent"
}

Use Case: Log Enrichment

Enrich logs with derived fields for better analysis:

// Extract user tier from score
item['attributes']['user_tier'] = item['attributes']['score'] > 90 ? 'premium' : item['attributes']['score'] > 60 ? 'standard' : 'basic';

// Add timestamp
item['attributes']['enriched_at'] = new Date().toISOString();

// Create searchable tags
item['attributes']['tags'] = [item['attributes']['user_tier'], item['attributes']['level'], item['attributes']['event_type']];

Use Case: Data Normalization

Normalize inconsistent data formats:

// Normalize level field
const level = item['attributes']['level'].toUpperCase();
if (level === 'ERR' || level === 'ERROR') {
  item['attributes']['normalized_level'] = 'ERROR';
} else if (level === 'WARN' || level === 'WARNING') {
  item['attributes']['normalized_level'] = 'WARN';
} else if (level === 'INFO' || level === 'INFORMATION') {
  item['attributes']['normalized_level'] = 'INFO';
} else {
  item['attributes']['normalized_level'] = 'DEBUG';
}

Use Case: Alert Generation

Generate alerts based on complex conditions:

if (item['attributes']['level'] === 'ERROR' && item['attributes']['retry_count'] > 3) {
  if (!item['attributes']['alert']) item['attributes']['alert'] = {};
  item['attributes']['alert']['type'] = 'critical';
  item['attributes']['alert']['message'] = `Repeated errors detected: ${item['attributes']['error_message']}`;
  item['attributes']['alert']['retry_count'] = item['attributes']['retry_count'];
  item['attributes']['alert']['requires_action'] = true;
}

Alternative: Using Custom Processor with OTTL

If you need to combine edx_code with other OTTL statements in the same processor, you can use the Custom (OTTL Transform) processor. This approach is useful when mixing JavaScript transformations with native OTTL functions:

processors:
  - type: ottl_transform
    data_types:
    - log
    statements: |
      // First, use native OTTL to extract and convert fields
      set(attributes["score"], Int(body["score"]))
      set(attributes["level"], body["level"])

      // Then, use edx_code for complex JavaScript logic
      edx_code("item['attributes']['score_category'] = item['attributes']['score'] > 80 ? 'excellent' : item['attributes']['score'] > 50 ? 'good' : 'needs_improvement';")

      // Mix with more OTTL functions
      set(attributes["processed_at"], Now())