[Feature][Core] Support Traffic Dyeing (Sampling) and Context-Aware Metrics

[corgy-w]

Search before asking

• I had searched in the feature and found no similar feature requirement.

Description

Currently, SeaTunnel Metrics are generally global. We cannot distinguish metrics for a specific subset of data (e.g., sampled traffic, heartbeat packets) without aggressively modifying the connector/transform code to add tags manually.

This proposal aims to introduce a System-Level Traffic Dyeing (Sampling) mechanism. It allows:

1. Marking specific SeaTunnelRows as "Sampled" or "Dyed" at the Source.
2. Propagating this "Color" context implicitly throughout the execution engine (Source -> Transform -> Sink).
3. Automatically routing metrics to different counters (e.g., sink_write_count vs sink_write_count_sampled) based on the context, without changing existing Connector code.

Proposed Solution / Architecture:

1. Data Protocol: Add a long flags field to SeaTunnelRow (Core Data Structure). This uses a bitmask to carry system signals (e.g., IS_SAMPLED, IS_HEARTBEAT) with minimal serialization overhead compared to the existing options Map.
2. Context Propagation: Introduce MetricTraceContext (based on ThreadLocal) in seatunnel-engine.
   • When SeaTunnelSourceCollector receives a row, it checks the flags and sets the MetricTraceContext.
   • The context is cleared after the row is processed to ensure safety.
   • For async operations, the context must be captured and replayed.
3. Metrics Integration: Enhance MetricsContext (or AbstractMetricsContext) to support "Context-Aware" or "Routing" metrics.
   • The Metric object (e.g., Counter) acts as a proxy.
   • It checks MetricTraceContext on every update (inc()) and routes the value to the appropriate underlying counter (e.g., standard vs. sampled).

Usage Scenario

1. Sampling Observation: In high-throughput scenarios (e.g., 1M QPS), calculating precise latency/success-rate for every record is expensive. Users can configure the Source to "sample" 1% of traffic (set the flag). The Metrics system will then automatically track metrics for this 1% separately, providing health visibility with low overhead.
2. Heartbeat Monitoring: Distinguish "Heartbeat" rows (synthetic data for keep-alive) from actual business data in metrics.
3. Trace/Debug: "Dye" specific rows to trace their flow and performance through the pipeline without mixing them with normal traffic stats.

Related issues

None.

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[davidzollo]

I'll have a try to do this feature

[davidzollo]

A framework-level data tracing system that tracks sampled records through the entire pipeline (Source → Queue → Transform → Sink) with minimal overhead, enabling precise performance bottleneck identification without requiring any connector modifications.

---

Trace Flow Diagram

sequenceDiagram
    participant Connector as Source Connector
    participant SourceTask as SeaTunnelSourceCollector
    participant Queue as IntermediateQueue
    participant Transform as TransformFlowLifeCycle
    participant Sink as SinkFlowLifeCycle
    participant EventService as EventService
    participant Master as Master Node
    participant Collector as Trace Collector

    Note over SourceTask: Sampling Decision<br/>(1 in 100k records)
    
    Connector->>SourceTask: emit(record)
    SourceTask->>SourceTask: Check sampling<br/>(seq % sampleRate == 0)
    SourceTask->>SourceTask: Create trace payload<br/>traceId, startTs
    SourceTask->>SourceTask: Append SOURCE_EMIT<br/>(taskId, timestamp)
    SourceTask->>Queue: collect(record + payload)
    
    Note over Queue: Record enters queue
    Queue->>Queue: Append QUEUE_IN<br/>(taskId, timestamp)
    Queue->>Queue: Store in buffer
    
    Note over Queue: Record dequeued
    Queue->>Queue: Append QUEUE_OUT<br/>(taskId, timestamp)
    Queue->>Transform: collect(record + payload)
    
    Note over Transform: Transform processing
    Transform->>Transform: Append TRANSFORM_IN<br/>(taskId, timestamp)
    Transform->>Transform: Execute transform logic
    Transform->>Transform: Append TRANSFORM_OUT<br/>(taskId, timestamp)
    Transform->>Sink: collect(output + payload)
    
    Note over Sink: Sink write
    Sink->>Sink: Append SINK_WRITE_DONE<br/>(taskId, timestamp)
    Sink->>Sink: writer.write(record)
    Sink->>Sink: Create StainTraceEvent<br/>(traceId, payload, taskId, tableId)
    Sink->>EventService: reportEvent(event)
    
    Note over EventService: Async batching
    EventService->>Master: BatchedEvents via Hazelcast
    
    Note over Master: HTTP reporting
    Master->>Master: Convert to JSON
    Master->>Collector: POST /ingest<br/>[{traceId, payload, ...}]
    
    Note over Collector: Parse & Store
    Collector->>Collector: Decode binary payload
    Collector->>Collector: Extract timing entries
    Collector->>Collector: Calculate metrics<br/>(e2e, queue_wait, etc.)
    Collector->>Collector: Store in DB<br/>(PostgreSQL/MySQL/ClickHouse)
    Collector-->>Master: 200 OK

Loading

Key Points:

1. Single-record flow: One sampled record = one complete trace
2. Zero events during flow: No reporting until Sink (avoids event storm)
3. Binary payload: Compact, efficient cross-node transmission
4. Async reporting: Non-blocking event submission
5. Centralized collection: Trace Collector decodes and stores for analysis

---

Architecture Overview

Core Concept: Sample-Based One-Shot Reporting

Instead of reporting at every stage (which causes event storms), we:

1. Mark a small fraction of records at the Source with a trace payload
2. Propagate the payload through the Engine's internal flow
3. Report once at the Sink with complete timing breakdown

This ensures reporting volume is strictly proportional to sampling rate, with each sampled record generating exactly 1 event.

Trace Stages (Actual Implementation)

Stage	Code	Meaning	Framework Location
SOURCE_EMIT	1	Source emits record	SeaTunnelSourceCollector.collect()
QUEUE_IN	2	Queue enqueue complete	IntermediateBlockingQueue.received() / RecordEventProducer
QUEUE_OUT	3	Queue dequeue start	IntermediateBlockingQueue.collect() / RecordEventHandler
TRANSFORM_IN	4	Transform receives record	TransformFlowLifeCycle.received()
TRANSFORM_OUT	5	Transform outputs record	Before collector.collect(output) in TransformFlowLifeCycle
SINK_WRITE_DONE	6	Sink write complete	After writer.write() in SinkFlowLifeCycle

---

Implementation Details

1. Trace Payload Protocol (Cross-Node Transmission)

We extend RecordSerializer to transmit a compact binary payload stored in SeaTunnelRow.options["__st_trace_payload"]:

Binary Format:

MAGIC(4)  = 0x53545452  // 'STTR'
VER(2)    = 1
TRACE_ID(8)
START_TS_MS(8)
COUNT(2)  // number of entries
ENTRIES:
  repeat COUNT times:
    STAGE(1)     // 1-6 (SOURCE_EMIT to SINK_WRITE_DONE)
    TASK_ID(8)   // TaskLocation.getTaskID()
    TS_MS(8)     // System.currentTimeMillis()

Key Classes:

• StainTracePayload: Encode/decode binary payload
• StainTraceStage: Enum defining 6 trace stages
• StainTraceUtils: Helper methods for payload manipulation

Size Control:

• Configurable max entries per trace (default: 32)
• When limit reached, stop appending but continue propagating

2. Sampling & Budget Control

Deterministic Sampling (Source only):

• Implemented in StainTraceSampler
• Each Source subtask maintains sequence counter
• Sample when seq % sampleRate == 0
• No random number generation overhead

Strict Budget Control (per Worker):

• Implemented in StainTraceBudget
• Config: stain-trace-max-traces-per-second-per-worker
• When limit reached: stop creating new samples
• Existing samples continue propagating
• Guarantees upper bound on event volume

3. Framework Modifications (All in Engine Layer)

Configuration:

• EngineConfig: Added stain-trace config fields
• ServerConfigOptions: Define config keys
• YamlSeaTunnelDomConfigProcessor: Parse YAML config

Serialization:

• RecordSerializer: Extended to serialize/deserialize trace payload
• Backward compatible: old format data still readable
• Only serializes __st_trace_payload, not entire options map

Framework Integration Points:

Component	File	Implementation
Source	SeaTunnelSourceCollector	Create payload, append SOURCE_EMIT
Queue	IntermediateQueueFlowLifeCycle IntermediateBlockingQueue RecordEventProducer/Handler	Append QUEUE_IN/QUEUE_OUT
Transform	TransformFlowLifeCycle	Append TRANSFORM_IN/OUT Handle 1-to-N (only first output inherits)
Sink	SinkFlowLifeCycle	Append SINK_WRITE_DONE, report event

Event Reporting:

• StainTraceEvent: New event type with fields (traceId, payload, sinkTaskId, tableId, jobId)
• Reuses existing JobEventHttpReportHandler infrastructure
• HTTP POST JSON to collector service

4. Task Mapping

Problem: TaskID (long) is not human-readable
Solution: TaskMappingBuilder + REST API

• GetJobTaskMappingOperation: Hazelcast operation to fetch task mapping
• TraceTaskMappingService: REST endpoint /hazelcast/rest/maps/trace-task-mapping/{jobId}
• Maps taskId → readable name (e.g., "Source[0]-Kafka-Topic-A")

---

Trace Collector Service

Standalone HTTP service for collecting and querying traces:

Supported Databases:

• PostgreSQL (PostgresTraceRepository)
• MySQL (MySqlTraceRepository)
• ClickHouse (ClickHouseTraceRepository)

REST APIs:

• POST /ingest - Receive StainTraceEvent array from Engine
• GET /traces?trace_id=X - Query by trace ID
• GET /traces?job_id=X - Query by job ID
• GET /trace/{id} - Get trace detail
• GET /health - Health check
• GET /metrics - Prometheus metrics

Features:

• Task mapping cache (TaskMappingCache): Auto-fetch task names from Engine
• Built-in metrics (TraceCollectorMetrics): Ingestion rate, errors, query latency
• Authentication (TraceAuth): Optional bearer token auth
• Payload decoding (SttrPayloadDecoder): Parse binary payload to readable entries

Deployment:

• Standalone Java application
• Config: trace-collector.properties
• Database init scripts: init-mysql.sql, etc.

Quick Start: seatunnel-trace/STAIN_TRACE_QUICKSTART.md

---

Configuration

Enable in seatunnel.yaml:

seatunnel:
  engine:
    stain-trace-enabled: true
    stain-trace-sample-rate: 100000        # 1 in 100k records
    stain-trace-max-traces-per-second-per-worker: 50
    stain-trace-max-entries-per-trace: 32

Production-Safe Defaults:

• enabled: false (opt-in, zero overhead when disabled)
• Single boolean check in hot path
• ~0.1-1% overhead with 1% sampling when enabled

---

Performance Analysis

Metrics Calculation

From trace payload entries, calculate:

# End-to-end latency
e2e_latency = SINK_WRITE_DONE.ts - SOURCE_EMIT.ts

# Queue wait time
queue_wait = QUEUE_OUT.ts - QUEUE_IN.ts

# Transform processing
transform_time = TRANSFORM_OUT.ts - TRANSFORM_IN.ts

# Sink write time
sink_write = SINK_WRITE_DONE.ts - TRANSFORM_OUT.ts

Bottleneck Identification

Aggregate across samples:

• P95 e2e latency: Identify slow records
• P95 queue wait: Detect backpressure
• P95 sink write: Find storage bottlenecks
• P95 transform time: Identify expensive transformations

Example OLAP Query (ClickHouse)

CREATE TABLE trace_records (
  job_id String,
  trace_id UInt64,
  start_ts DateTime,
  end_ts DateTime,
  e2e_ms UInt32,
  queue_wait_ms UInt32,
  transform_ms UInt32,
  sink_write_ms UInt32,
  entries Array(Tuple(stage UInt8, task_id UInt64, ts_ms UInt64))
) ENGINE = MergeTree()
ORDER BY (job_id, start_ts);

-- Analyze bottlenecks
SELECT 
  job_id,
  quantile(0.95)(e2e_ms) as p95_e2e,
  quantile(0.95)(queue_wait_ms) as p95_queue,
  quantile(0.95)(sink_write_ms) as p95_sink
FROM trace_records
WHERE job_id = '123456' AND start_ts > now() - INTERVAL 1 HOUR
GROUP BY job_id;

---

Testing & Validation

Unit Tests

• StainTracePayloadTest: Binary encoding/decoding
• StainTraceSamplerTest: Sampling logic, budget control
• StainTraceUtilsTest: Utility methods
• RecordSerializerTest: Backward compatibility (old → new), payload serialization
• SttrPayloadDecoderTest: Trace collector payload parsing

Integration Tests

• StainTraceFlowIT: End-to-end trace through Source → Transform → Sink
• TransformFlowLifeCycleStainTraceTest:
  • 1-to-1 propagation
  • 1-to-N (FlatMap): only first output inherits payload
  • 1-to-0 (Filter): no event reported
• JobEventHttpReportHandlerTest: Event reporting

Performance Regression

• Baseline with tracing disabled
• Verify negligible impact with production sampling rate

---

Deployment & Compatibility

Non-Rolling Upgrade (Required)

1. Stop all jobs and Engine nodes
2. Replace jars (includes serialization protocol change)
3. Start Master → Workers
4. Enable on canary job first
5. Validate events, expand to production

Backward Compatibility

• New reads old: ✅ New version can read old serialization format

Clock Synchronization (Critical)

• All nodes must have NTP/chrony enabled
• Clock skew should be ≤ 50ms
• High skew causes incorrect timing analysis

---

Web UI Integration

New trace visualization page in SeaTunnel Engine UI:

Features:

• Query traces by trace_id or job_id
• Display timing breakdown per stage
• Visualize bottlenecks with bar charts
• Export trace data

Files:

• src/views/trace/index.tsx: Main trace page
• src/service/trace/: API client
• Updated sidebar menu with trace entry

---

Comparison with Original Proposal

This issue proposed:

• Flags in SeaTunnelRow: We use dedicated binary payload (cleaner serialization)
• ThreadLocal context: We serialize payload with record (simpler, no async complexity)
• Metrics integration: We focus on tracing events for external analysis

Our approach prioritizes:

• End-to-end visibility over per-stage metrics aggregation
• Minimal serialization overhead over ThreadLocal management
• External OLAP analysis over in-process metric splitting
• Zero connector changes - all connectors get tracing automatically

---

Resources

• PR [Feature] Add stain trace for data lineage and performance analysis #10434: [Feature] Add stain trace for data lineage and performance analysis #10434
• Quick Start: seatunnel-trace/STAIN_TRACE_QUICKSTART.md (in PR)
• Database Init: seatunnel-trace/seatunnel-trace-collector/src/main/resources/init-mysql.sql

---

Discussion: Metrics Segmentation

This implementation focuses on performance tracing rather than metrics segmentation. However, the infrastructure could be extended for the original use case:

Potential Extension (if metrics segmentation is needed):

1. Reuse StainTracePayload for flag propagation
2. Add metrics proxy layer that reads payload flags
3. Route metrics to separate counters (e.g., records_sampled, records_heartbeat)

Trade-offs:

• Current: Low overhead, external analysis, rich timing data
• Extension: In-process metrics, real-time dashboards, less detailed timing

Would love feedback on:

• Does tracing-first approach meet your needs?
• Should we add metrics segmentation on top of this?
• Any other use cases we should consider?