122 lines
7.9 KiB
Markdown
122 lines
7.9 KiB
Markdown
## Purpose
|
|
|
|
Defines stream consumption, safe parsing, schema normalization, micro-batching into ClickHouse, and at-least-once offset management for raw telecom CDRs.
|
|
|
|
## ADDED Requirements
|
|
|
|
### Requirement: Safe Parsing and Fault Tolerance
|
|
The Processing Worker SHALL consume messages from `cdr.raw` and process them without crashing, routing malformed or unprocessable payloads to `cdr.raw.dlq`.
|
|
|
|
#### Scenario: Malformed or unparseable JSON payload
|
|
- **WHEN** a message consumed from `cdr.raw` cannot be parsed as valid JSON
|
|
- **THEN** the worker wraps the raw message in a DLQ envelope with `failure_reason: "malformed_json"` and publishes it to `cdr.raw.dlq` without terminating the process
|
|
|
|
#### Scenario: Missing required field
|
|
- **WHEN** a consumed record is missing any required field (`raw_id`, `device_id`, `caller_num`, `svc`, `rat`, `epoch_start`, `epoch_end`, `up_bytes`, `down_bytes`, `lac`, `ci`) or contains null for a required field
|
|
- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "missing_required_field"` and `failed_field` set to the missing field name
|
|
|
|
#### Scenario: Inverted epoch timestamps
|
|
- **WHEN** a consumed record has `epoch_start > epoch_end`
|
|
- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "invalid_epoch"` and `failed_field: "epoch_start"`
|
|
|
|
#### Scenario: Negative byte counts
|
|
- **WHEN** a consumed record has `up_bytes < 0` or `down_bytes < 0`
|
|
- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "normalization_error"` and `failed_field` set to the negative byte field
|
|
|
|
### Requirement: E.164 Phone Number Normalization
|
|
The Processing Worker SHALL normalize `caller_num` into a valid E.164 format `msisdn` using the `phonenumbers` library.
|
|
|
|
#### Scenario: Leading international prefix normalization
|
|
- **WHEN** a valid record contains `caller_num` starting with `00` or `011`
|
|
- **THEN** the worker replaces the prefix with `+`, parses the number using `phonenumbers`, and sets `msisdn` to the standardized E.164 string
|
|
|
|
#### Scenario: Existing E.164 number preserved
|
|
- **WHEN** a record contains `caller_num` already formatted with a leading `+`
|
|
- **THEN** the worker parses and validates the number using `phonenumbers` and outputs the verified E.164 string
|
|
|
|
#### Scenario: Unparseable phone number routed to DLQ
|
|
- **WHEN** `caller_num` cannot be normalized or validated by `phonenumbers`
|
|
- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "normalization_error"` and `failed_field: "caller_num"`
|
|
|
|
### Requirement: Radio Access Type Mapping
|
|
The Processing Worker SHALL map the integer `rat` field to standard string representations.
|
|
|
|
#### Scenario: Supported RAT values mapped
|
|
- **WHEN** a record has `rat` equal to `1`, `6`, or `9`
|
|
- **THEN** the worker maps `radio_access_type` to `"3G"`, `"LTE_4G"`, or `"5G_NR"` respectively
|
|
|
|
#### Scenario: Unsupported RAT value routed to DLQ
|
|
- **WHEN** a record has `rat` not equal to 1, 6, or 9
|
|
- **THEN** the worker routes the record to `cdr.raw.dlq` with `failure_reason: "invalid_rat_value"` and `failed_field: "rat"`
|
|
|
|
### Requirement: Schema Transformation and Deterministic Identity
|
|
The Processing Worker SHALL transform valid raw CDRs into the normalized schema with deterministic identity derivation.
|
|
|
|
#### Scenario: Deterministic UUIDv5 cdr_id generation
|
|
- **WHEN** a raw record with `raw_id` is normalized
|
|
- **THEN** the worker computes `cdr_id` as `uuid.uuid5(uuid.NAMESPACE_DNS, raw_id)` ensuring identical IDs for reprocessed duplicate records
|
|
|
|
#### Scenario: Computed metrics and cell identification
|
|
- **WHEN** raw fields are mapped to the normalized schema
|
|
- **THEN** `cell_id` is set to `lac + "_" + ci`, `total_bytes` is computed as `up_bytes + down_bytes`, `duration_seconds` is computed as `epoch_end - epoch_start`, `session_start_time` and `session_end_time` are converted to UTC DateTimes, and `source_ip` defaults to `"0.0.0.0"` if absent
|
|
|
|
### Requirement: Micro-Batching Buffer Strategy
|
|
The Processing Worker SHALL buffer normalized records in memory and flush batches to ClickHouse based on size or elapsed time thresholds.
|
|
|
|
#### Scenario: Batch flushed on count threshold
|
|
- **WHEN** the in-memory normalized record buffer reaches `BATCH_SIZE` (default 1,000 records)
|
|
- **THEN** the worker immediately flushes the batch to ClickHouse
|
|
|
|
#### Scenario: Batch flushed on time interval threshold
|
|
- **WHEN** `BATCH_FLUSH_INTERVAL_SECONDS` (default 2 seconds) elapses since the last flush and the buffer contains at least one record
|
|
- **THEN** the worker flushes the accumulated buffer to ClickHouse
|
|
|
|
### Requirement: At-Least-Once Delivery and Offset Management
|
|
The Processing Worker SHALL commit Kafka consumer offsets only after ClickHouse successfully acknowledges the batch write.
|
|
|
|
#### Scenario: Offsets committed after successful write
|
|
- **WHEN** a micro-batch is written and acknowledged by ClickHouse
|
|
- **THEN** the worker commits the highest offsets corresponding to the records in that batch
|
|
|
|
#### Scenario: ClickHouse write timeout retry with exponential backoff
|
|
- **WHEN** a batch write to ClickHouse fails or times out
|
|
- **THEN** the worker retries the write up to `CH_RETRY_MAX_ATTEMPTS` (default 3) with exponential backoff (`CH_RETRY_INITIAL_DELAY_MS` multiplied by `CH_RETRY_BACKOFF_FACTOR`); if all attempts fail, the worker routes all records in the batch to `cdr.raw.dlq` and commits offsets to avoid pipeline deadlock
|
|
|
|
### Requirement: Internal Runtime Statistics Endpoint
|
|
The Processing Worker SHALL expose a lightweight internal HTTP endpoint (e.g., `GET /internal/stats`) to provide its runtime statistics to the Ingestion API.
|
|
|
|
#### Scenario: Internal stats queried
|
|
- **WHEN** an HTTP request is made to the worker's internal stats endpoint
|
|
- **THEN** the worker returns HTTP status `200 OK` with JSON containing `total_processed`, `total_dlq`, `in_flight_buffer`, `last_flush_at`, and breakdown counts by service type, RAT, and DLQ reason
|
|
|
|
### Requirement: Graceful Shutdown Without Data Loss
|
|
The Processing Worker SHALL handle `SIGTERM` and `SIGINT` signals to flush buffered records before terminating.
|
|
|
|
#### Scenario: SIGTERM received during operation
|
|
- **WHEN** a `SIGTERM` signal is received by the worker process
|
|
- **THEN** the worker stops consuming new messages, flushes any remaining in-memory buffered records to ClickHouse, commits the associated Kafka offsets, and exits cleanly with exit code 0
|
|
|
|
### Requirement: Safe partition progress and confirmed DLQ
|
|
The worker SHALL disable automatic offset storage, commit explicit next offsets, and await all destination acknowledgements.
|
|
|
|
#### Scenario: Valid record precedes an invalid record
|
|
- **WHEN** an invalid record follows buffered valid records
|
|
- **THEN** flush and commit those valid records before acknowledging and committing the invalid record to DLQ
|
|
- **THEN** route the invalid record to DLQ and defer committing its offset until buffered valid records are flushed, or commit immediately if no preceding uncommitted records exist on that partition
|
|
|
|
#### Scenario: Destination or commit failure
|
|
- **WHEN** DLQ delivery or any partition commit fails
|
|
- **THEN** stop nonzero without advancing unsafe offsets or reclassifying the error as a ClickHouse failure; fatal cleanup does not flush the buffer
|
|
|
|
#### Scenario: Exhausted ClickHouse retries
|
|
- **WHEN** three configured insert attempts fail
|
|
- **THEN** publish original vendor payloads with `normalization_error` and commit only after all DLQ deliveries succeed
|
|
- **THEN** publish original vendor payloads with normalization_error and commit only after all DLQ deliveries succeed
|
|
|
|
### Requirement: Storage-safe normalization
|
|
The worker SHALL reject numeric booleans, fractional numeric fields, out-of-range UInt64 and UInt32 values including total bytes, invalid IPv4, and timestamps outside DateTime range without crashing.
|
|
|
|
#### Scenario: Invalid scalar or conversion
|
|
- **WHEN** any field fails its type, range or conversion check
|
|
- **THEN** route only that record to DLQ with a field-specific reason and preserve other valid records
|