amiirkhl/telecom-cdr-pipeline-ai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(verification): require run-scoped Kafka and ClickHouse evidence

Browse Source
This commit is contained in:
Amirhossein Khalili
2026-09-19 21:30:48 +03:30
parent b5bc2d722f
commit c51fe40851
3 changed files with 675 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

745
scripts/generate_traffic.py
View File

@@ -1,59 +1,99 @@
import os
import base64
import argparse
import json
import random
import re
import sys
import time
import uuid
import urllib.error
import urllib.parse
import urllib.request
from collections import Counter
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Any, Dict, List, Tuple
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Dict, Iterable, List, Optional, Tuple
RAT_CHOICES = [1, 6, 9] # 3G, LTE_4G, 5G_NR
CLICKHOUSE_DATABASE = os.getenv("CLICKHOUSE_DATABASE", "telecom")
CLICKHOUSE_TABLE = os.getenv("CLICKHOUSE_TABLE", "cdr_records")
SERVICE_CHOICES = ["4G_DATA", "5G_DATA", "VOICE_CALL", "SMS"]
PHONE_PREFIXES = ["001", "0111", "+1"]
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@dataclass
class HttpResult:
status: int
body: bytes
elapsed_ms: float
@dataclass
class ScenarioResult:
name: str
passed: bool
detail: str
def send_http_request(
url: str,
method: str = "GET",
body: bytes = b"",
headers: Dict[str, str] = None,
timeout: float = 5.0,
) -> Tuple[int, bytes]:
req_headers = headers or {}
headers: Optional[Dict[str, str]] = None,
timeout: float = 15.0,
) -> HttpResult:
req = urllib.request.Request(
url,
data=body if method in ("POST", "PUT") else None,
headers=req_headers,
data=body if method in ("POST", "PUT", "PATCH") else None,
headers=headers or {},
method=method,
)
started = time.perf_counter()
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, resp.read()
except urllib.error.HTTPError as e:
return e.code, e.read()
except Exception as e:
return 0, str(e).encode("utf-8")
return HttpResult(
status=resp.status,
body=resp.read(),
elapsed_ms=(time.perf_counter() - started) * 1000,
)
except urllib.error.HTTPError as exc:
return HttpResult(
status=exc.code,
body=exc.read(),
elapsed_ms=(time.perf_counter() - started) * 1000,
)
except Exception as exc:
return HttpResult(
status=0,
body=str(exc).encode("utf-8", errors="replace"),
elapsed_ms=(time.perf_counter() - started) * 1000,
)
def generate_valid_record(raw_id: str) -> Dict[str, Any]:
rat_choices = [1, 6, 9] # 3G, LTE_4G, 5G_NR
svc_choices = ["4G_DATA", "5G_DATA", "VOICE_CALL", "SMS"]
rat = random.choice(rat_choices)
svc = random.choice(svc_choices)
def generate_valid_record(raw_id: str, sequence: int = 0) -> Dict[str, Any]:
"""Generate a valid record while deterministically covering all RAT/service variants."""
now = int(time.time())
start = now - random.randint(100, 1000)
end = start + random.randint(10, 600)
# Varied phone formats (00..., 011..., +...)
phone_prefix = random.choice(["001", "0111", "+1"])
phone_number = f"{phone_prefix}987555{random.randint(1000, 9999)}"
rat = RAT_CHOICES[sequence % len(RAT_CHOICES)]
svc = SERVICE_CHOICES[sequence % len(SERVICE_CHOICES)]
phone_prefix = PHONE_PREFIXES[sequence % len(PHONE_PREFIXES)]
return {
"raw_id": raw_id,
"src_ip": f"10.240.{random.randint(1, 254)}.{random.randint(1, 254)}",
"device_id": f"310410{random.randint(100000000, 999999999)}",
"caller_num": phone_number,
"caller_num": f"{phone_prefix}987555{random.randint(1000, 9999)}",
"svc": svc,
"up_bytes": random.randint(1000, 5000000),
"down_bytes": random.randint(1000, 20000000),
"up_bytes": random.randint(1000, 5_000_000),
"down_bytes": random.randint(1000, 20_000_000),
"rat": rat,
"lac": str(random.randint(1000, 9999)),
"ci": str(random.randint(1000, 9999)),
@@ -62,190 +102,469 @@ def generate_valid_record(raw_id: str) -> Dict[str, Any]:
}
def run_verification(api_url: str, clickhouse_url: str) -> None:
print("\n=======================================================")
print(" Telecom CDR Pipeline Verification & Traffic Generator")
print(f" Target Ingestion API: {api_url}")
print(f" Target ClickHouse: {clickhouse_url}")
print("=======================================================\n")
ingest_endpoint = f"{api_url}/api/v1/cdr/raw"
headers = {"Content-Type": "application/json"}
total_sent = 0
api_400_count = 0
valid_raw_ids = []
# -------------------------------------------------------------
# Scenario 1: Valid records, varied RAT & service types (1,000)
# -------------------------------------------------------------
print("[1/7] Sending 1,000 valid records with varied RAT & service types...")
for i in range(1, 1001):
raw_id = f"raw-valid-{i:05d}"
valid_raw_ids.append(raw_id)
payload = generate_valid_record(raw_id)
status, _ = send_http_request(
ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers
)
total_sent += 1
if status == 400:
api_400_count += 1
print(" -> Sent 1,000 valid records.")
# -------------------------------------------------------------
# Scenario 2: Missing required field ('caller_num') (20)
# -------------------------------------------------------------
print("[2/7] Sending 20 records missing required field 'caller_num'...")
for i in range(1, 21):
payload = generate_valid_record(f"raw-missing-{i:04d}")
del payload["caller_num"]
status, _ = send_http_request(
ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers
)
total_sent += 1
if status == 400:
api_400_count += 1
print(" -> Sent 20 records missing required field.")
# -------------------------------------------------------------
# Scenario 3: Invalid JSON body (10)
# -------------------------------------------------------------
print("[3/7] Sending 10 invalid JSON payloads (syntax errors)...")
for i in range(1, 11):
corrupted = b'{"raw_id": "bad-json-' + str(i).encode() + b'", "unclosed_string: '
status, _ = send_http_request(ingest_endpoint, "POST", corrupted, headers)
total_sent += 1
if status == 400:
api_400_count += 1
print(f" -> Sent 10 invalid JSON payloads (API 400 received: {api_400_count}).")
# -------------------------------------------------------------
# Scenario 4: Unknown RAT value (rat: 99) (10)
# -------------------------------------------------------------
print("[4/7] Sending 10 records with unknown RAT value (rat: 99)...")
for i in range(1, 11):
payload = generate_valid_record(f"raw-badrat-{i:04d}")
payload["rat"] = 99
status, _ = send_http_request(
ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers
)
total_sent += 1
if status == 400:
api_400_count += 1
print(" -> Sent 10 records with invalid RAT.")
# -------------------------------------------------------------
# Scenario 5: Inverted epoch timestamps (epoch_start > epoch_end) (10)
# -------------------------------------------------------------
print("[5/7] Sending 10 records with inverted epoch timestamps...")
for i in range(1, 11):
payload = generate_valid_record(f"raw-badepoch-{i:04d}")
payload["epoch_start"] = payload["epoch_end"] + 500
status, _ = send_http_request(
ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers
)
total_sent += 1
if status == 400:
api_400_count += 1
print(" -> Sent 10 records with invalid epoch timestamps.")
# -------------------------------------------------------------
# Scenario 6: Duplicate raw_id (20 records sent twice)
# -------------------------------------------------------------
print("[6/7] Sending 20 duplicate records (re-using existing raw_ids)...")
duplicate_count = 20
for i in range(duplicate_count):
existing_raw_id = valid_raw_ids[i]
payload = generate_valid_record(existing_raw_id)
status, _ = send_http_request(
ingest_endpoint, "POST", json.dumps(payload).encode("utf-8"), headers
)
total_sent += 1
if status == 400:
api_400_count += 1
print(" -> Sent 20 duplicate records for deduplication check.")
# -------------------------------------------------------------
# Scenario 7: High-volume burst (500 records in <5s)
# -------------------------------------------------------------
print("[7/7] Sending high-volume burst (500 records concurrently in <5s)...")
burst_payloads = [
json.dumps(generate_valid_record(f"raw-burst-{i:04d}")).encode("utf-8")
for i in range(1, 501)
]
burst_start = time.time()
burst_429_count = 0
with ThreadPoolExecutor(max_workers=25) as executor:
futures = [
executor.submit(send_http_request, ingest_endpoint, "POST", p, headers)
for p in burst_payloads
]
for f in as_completed(futures):
status, _ = f.result()
total_sent += 1
if status == 429:
burst_429_count += 1
elif status == 400:
api_400_count += 1
burst_duration = time.time() - burst_start
print(
f" -> Burst complete: 500 records dispatched in {burst_duration:.2f}s (429 rejected: {burst_429_count})."
def post_json(endpoint: str, payload: Dict[str, Any], headers: Dict[str, str]) -> HttpResult:
return send_http_request(
endpoint,
method="POST",
body=json.dumps(payload).encode("utf-8"),
headers=headers,
)
# -------------------------------------------------------------
# Wait for micro-batch buffers to flush to ClickHouse
# -------------------------------------------------------------
print("\nWaiting 4 seconds for worker micro-batches to flush to ClickHouse...")
time.sleep(4.0)
# -------------------------------------------------------------
# Query ClickHouse and API Stats for Verification
# -------------------------------------------------------------
ch_query_url = f"{clickhouse_url}/?query="
ch_records_count = 0
def status_counts(results: Iterable[HttpResult]) -> Counter:
return Counter(result.status for result in results)
def format_status_counts(counts: Counter) -> str:
if not counts:
return "none"
return ", ".join(f"{status}:{count}" for status, count in sorted(counts.items()))
def is_structured_json_error(result: HttpResult) -> bool:
if result.status < 400:
return False
try:
count_query = "SELECT count() FROM telecom.cdr_records"
status, body = send_http_request(
f"{ch_query_url}{urllib.parse.quote(count_query)}"
)
if status == 200:
ch_records_count = int(body.decode("utf-8").strip())
payload = json.loads(result.body.decode("utf-8"))
except Exception:
pass
return False
return isinstance(payload, dict) and bool(payload.get("error"))
# Query /stats endpoint
stats_status, stats_body = send_http_request(f"{api_url}/stats")
total_dlq = 0
if stats_status == 200:
try:
stats_json = json.loads(stats_body.decode("utf-8"))
total_dlq = stats_json.get("pipeline", {}).get("total_dlq", 0)
if ch_records_count == 0:
ch_records_count = stats_json.get("pipeline", {}).get(
"total_processed", 0
)
except Exception:
pass
# -------------------------------------------------------------
# Print Verification Summary
# -------------------------------------------------------------
print("\n=== Pipeline Verification Summary ===")
print(f"Records sent: {total_sent}")
print(f"Records in ClickHouse: {ch_records_count:<7} {'✅' if ch_records_count >= 1000 else '⚠️'}")
print(f"Records in DLQ: {total_dlq:<7} {'✅' if total_dlq >= 40 else '⚠️'}")
print(f"API 400 errors: {api_400_count:<7} {'✅' if api_400_count == 10 else '⚠️'}")
print(f"Duplicate raw_ids resolved: {duplicate_count:<7} ✅")
print("=====================================\n")
def fetch_json(url: str) -> Optional[Dict[str, Any]]:
result = send_http_request(url)
if result.status != 200:
return None
try:
payload = json.loads(result.body.decode("utf-8"))
return payload if isinstance(payload, dict) else None
except Exception:
return None
def stats_snapshot(stats_url: str) -> Optional[Dict[str, Any]]:
data = fetch_json(stats_url)
if data is None:
return None
pipeline = data.get("pipeline", {}) if isinstance(data.get("pipeline"), dict) else {}
if pipeline.get("worker_status") == "unreachable":
return None
breakdown = data.get("breakdown", {}) if isinstance(data.get("breakdown"), dict) else {}
dlq = breakdown.get("dlq_by_reason", {}) if isinstance(breakdown.get("dlq_by_reason"), dict) else {}
return {
"total_received": int(pipeline.get("total_received", 0) or 0),
"total_processed": int(pipeline.get("total_processed", 0) or 0),
"total_dlq": int(pipeline.get("total_dlq", 0) or 0),
"in_flight_buffer": int(pipeline.get("in_flight_buffer", 0) or 0),
"dlq_by_reason": {str(k): int(v or 0) for k, v in dlq.items()},
}
def delta(after: int, before: int) -> int:
return after - before
def dlq_reason_delta(
before: Optional[Dict[str, Any]], after: Optional[Dict[str, Any]], reason: str
) -> Optional[int]:
if before is None or after is None:
return None
before_value = before.get("dlq_by_reason", {}).get(reason, 0)
after_value = after.get("dlq_by_reason", {}).get(reason, 0)
return delta(int(after_value), int(before_value))
def clickhouse_query(clickhouse_url: str, query: str) -> Tuple[bool, str]:
parsed = urllib.parse.urlsplit(clickhouse_url)
query_params = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
query_params.append(("query", query))
url = urllib.parse.urlunsplit(parsed._replace(path=parsed.path or "/", query=urllib.parse.urlencode(query_params)))
auth = base64.b64encode(f"{os.getenv('CLICKHOUSE_USER', 'default')}:{os.getenv('CLICKHOUSE_PASSWORD', '')}".encode()).decode()
result = send_http_request(url, headers={"Authorization": f"Basic {auth}"}, timeout=10.0)
if result.status != 200:
return False, result.body.decode("utf-8", errors="replace").strip()
return True, result.body.decode("utf-8", errors="replace").strip()
def ch_count(clickhouse_url: str, where_clause: str) -> Optional[int]:
ok, body = clickhouse_query(
clickhouse_url,
f"SELECT count() FROM {CLICKHOUSE_DATABASE}.{CLICKHOUSE_TABLE} FINAL WHERE {where_clause}",
)
if not ok:
return None
try:
return int(body.splitlines()[0].strip())
except Exception:
return None
def quote_ch_string(value: str) -> str:
return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"
def verify_normalized_rows(clickhouse_url, run_prefix, payloads):
expected = {payload["raw_id"]: payload for payload in payloads}
query = (f"SELECT *, toUnixTimestamp(session_start_time) AS start_epoch, "
f"toUnixTimestamp(session_end_time) AS end_epoch "
f"FROM {CLICKHOUSE_DATABASE}.{CLICKHOUSE_TABLE} FINAL "
f"WHERE startsWith(raw_id, {quote_ch_string(run_prefix)}) FORMAT JSONEachRow")
ok, body = clickhouse_query(clickhouse_url, query)
if not ok:
return False, "ClickHouse rows unavailable"
rows = [json.loads(line) for line in body.splitlines() if line.strip()]
if len(rows) != len(expected) or {row["raw_id"] for row in rows} != set(expected):
return False, "Stored identities differ from submitted identities"
rat_map = {1: "3G", 6: "LTE_4G", 9: "5G_NR"}
for row in rows:
raw = expected[row["raw_id"]]
phone = raw["caller_num"]
phone = "+" + phone[2:] if phone.startswith("00") else "+" + phone[3:] if phone.startswith("011") else phone
strings = {"cdr_id": str(uuid.uuid5(uuid.NAMESPACE_DNS, raw["raw_id"])),
"msisdn": phone, "imsi": raw["device_id"], "service_type": raw["svc"],
"radio_access_type": rat_map[raw["rat"]], "cell_id": f"{raw['lac']}_{raw['ci']}",
"source_ip": raw.get("src_ip") or "0.0.0.0"}
numbers = {"bytes_uploaded": raw["up_bytes"], "bytes_downloaded": raw["down_bytes"],
"total_bytes": raw["up_bytes"] + raw["down_bytes"],
"duration_seconds": raw["epoch_end"] - raw["epoch_start"],
"start_epoch": raw["epoch_start"], "end_epoch": raw["epoch_end"]}
if any(row.get(key) != value for key, value in strings.items()) or any(int(row[key]) != value for key, value in numbers.items()):
return False, f"Incorrect transformation for {raw['raw_id']}"
return True, f"All {len(rows)} identities and normalized field values match"
def wait_for_pipeline(
clickhouse_url: str,
stats_url: str,
run_prefix: str,
expected_clickhouse: int,
expected_dlq_delta: int,
baseline_stats: Optional[Dict[str, Any]],
timeout_seconds: float,
) -> Tuple[Optional[int], Optional[Dict[str, Any]], float]:
started = time.monotonic()
last_count: Optional[int] = None
last_stats: Optional[Dict[str, Any]] = None
prefix_literal = quote_ch_string(run_prefix)
while True:
last_count = ch_count(clickhouse_url, f"startsWith(raw_id, {prefix_literal})")
last_stats = stats_snapshot(stats_url)
ch_ready = last_count is not None and last_count >= expected_clickhouse
if baseline_stats is not None and last_stats is not None:
dlq_ready = delta(last_stats["total_dlq"], baseline_stats["total_dlq"]) >= expected_dlq_delta
buffer_ready = last_stats.get("in_flight_buffer", 0) == 0
else:
# We still wait for ClickHouse if /stats is unavailable, then report observability separately.
dlq_ready = True
buffer_ready = True
if ch_ready and dlq_ready and buffer_ready:
break
if time.monotonic() - started >= timeout_seconds:
break
time.sleep(0.5)
return last_count, last_stats, time.monotonic() - started
def run_verification(
api_url: str,
clickhouse_url: str,
stats_url: str,
settle_timeout: float,
burst_workers: int,
kafka_brokers: str = "localhost:9092",
raw_topic: str = "cdr.raw",
dlq_topic: str = "cdr.raw.dlq",
) -> int:
run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + f"-{random.randint(1000, 9999)}"
run_prefix = f"verify-{run_id}-"
ingest_endpoint = f"{api_url.rstrip('/')}/api/v1/cdr/raw"
headers = {"Content-Type": "application/json"}
try:
from verification_kafka import KafkaEvidence
for name in (CLICKHOUSE_DATABASE, CLICKHOUSE_TABLE):
if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name):
raise ValueError(f"Invalid ClickHouse identifier: {name!r}")
evidence = KafkaEvidence(kafka_brokers, raw_topic, dlq_topic, run_prefix)
except Exception as exc:
print(f"OVERALL: FAIL - actual Kafka evidence is required: {exc}")
return 1
try:
print("\n============================================================")
print(" Telecom CDR Pipeline - Assessment Verification")
print(f" Run ID: {run_id}")
print(f" Ingestion API: {ingest_endpoint}")
print(f" Stats endpoint: {stats_url}")
print(f" ClickHouse endpoint: {clickhouse_url}")
print("============================================================\n")
baseline_stats = stats_snapshot(stats_url)
if baseline_stats is None:
print("WARN: /stats was not readable before the test; DLQ/stat deltas cannot be verified reliably.\n")
total_sent = 0
all_statuses: Counter = Counter()
scenario_http: Dict[str, List[HttpResult]] = {}
valid_payloads: List[Dict[str, Any]] = []
# 1. 1,000 valid records.
print("[1/7] 1,000 valid records (varied RAT and service type)")
valid_results: List[HttpResult] = []
for i in range(1000):
payload = generate_valid_record(f"{run_prefix}valid-{i + 1:05d}", i)
if i == 1:
# Same metadata, different identity: both must survive replacement.
payload = valid_payloads[0] | {"raw_id": payload["raw_id"]}
valid_payloads.append(payload)
result = post_json(ingest_endpoint, payload, headers)
valid_results.append(result)
scenario_http["valid"] = valid_results
counts = status_counts(valid_results)
all_statuses.update(counts)
total_sent += len(valid_results)
print(f" HTTP statuses: {format_status_counts(counts)}")
# 2. Missing required caller_num.
print("[2/7] 20 records missing required field 'caller_num'")
missing_results: List[HttpResult] = []
for i in range(20):
payload = generate_valid_record(f"{run_prefix}missing-{i + 1:04d}", i)
del payload["caller_num"]
missing_results.append(post_json(ingest_endpoint, payload, headers))
scenario_http["missing"] = missing_results
counts = status_counts(missing_results)
all_statuses.update(counts)
total_sent += len(missing_results)
print(f" HTTP statuses: {format_status_counts(counts)}")
for i in range(20):
direct = generate_valid_record(f"{run_prefix}direct-missing-{i:04d}", i)
del direct["caller_num"]
evidence.inject(direct["raw_id"], json.dumps(direct).encode(), "missing_required_field")
# 3. Invalid JSON must be rejected by the API with structured 400 errors.
print("[3/7] 10 malformed JSON requests")
malformed_results: List[HttpResult] = []
for i in range(10):
body = b'{"raw_id":"' + f"{run_prefix}bad-json-{i + 1}".encode("utf-8") + b'","broken":'
malformed_results.append(
send_http_request(ingest_endpoint, "POST", body, headers)
)
scenario_http["malformed"] = malformed_results
counts = status_counts(malformed_results)
all_statuses.update(counts)
total_sent += len(malformed_results)
structured_errors = sum(is_structured_json_error(r) for r in malformed_results)
print(f" HTTP statuses: {format_status_counts(counts)}; structured JSON errors: {structured_errors}/10")
for i in range(10):
key = f"{run_prefix}direct-json-{i:04d}"
evidence.inject(key, ('{"raw_id":"' + key + '","broken":').encode(), "malformed_json")
# 4. Unknown RAT should be accepted into the pipeline then routed to DLQ by worker.
print("[4/7] 10 records with unknown RAT (rat=99)")
bad_rat_results: List[HttpResult] = []
for i in range(10):
payload = generate_valid_record(f"{run_prefix}bad-rat-{i + 1:04d}", i)
payload["rat"] = 99
evidence.expected[payload["raw_id"]] = (json.dumps(payload).encode(), "invalid_rat_value")
bad_rat_results.append(post_json(ingest_endpoint, payload, headers))
scenario_http["bad_rat"] = bad_rat_results
counts = status_counts(bad_rat_results)
all_statuses.update(counts)
total_sent += len(bad_rat_results)
print(f" HTTP statuses: {format_status_counts(counts)}")
# 5. Inverted epoch should be routed to DLQ by worker.
print("[5/7] 10 records with epoch_start > epoch_end")
bad_epoch_results: List[HttpResult] = []
for i in range(10):
payload = generate_valid_record(f"{run_prefix}bad-epoch-{i + 1:04d}", i)
payload["epoch_start"] = payload["epoch_end"] + 500
evidence.expected[payload["raw_id"]] = (json.dumps(payload).encode(), "invalid_epoch")
bad_epoch_results.append(post_json(ingest_endpoint, payload, headers))
scenario_http["bad_epoch"] = bad_epoch_results
counts = status_counts(bad_epoch_results)
all_statuses.update(counts)
total_sent += len(bad_epoch_results)
print(f" HTTP statuses: {format_status_counts(counts)}")
# 6. Replay the exact same first 20 valid payloads, not newly-generated data.
print("[6/7] Replay 20 exact duplicate records (same raw_id and same payload)")
duplicate_results = [post_json(ingest_endpoint, payload, headers) for payload in valid_payloads[:20]]
scenario_http["duplicates"] = duplicate_results
counts = status_counts(duplicate_results)
all_statuses.update(counts)
total_sent += len(duplicate_results)
print(f" HTTP statuses: {format_status_counts(counts)}")
# 7. 500-record burst; wall-clock requirement is <5 seconds and no 429.
print(f"[7/7] 500-record concurrent burst (<5s, no 429), workers={burst_workers}")
burst_payloads = [
generate_valid_record(f"{run_prefix}burst-{i + 1:04d}", i) for i in range(500)
]
burst_started = time.perf_counter()
burst_results: List[HttpResult] = []
with ThreadPoolExecutor(max_workers=burst_workers) as executor:
futures = [executor.submit(post_json, ingest_endpoint, p, headers) for p in burst_payloads]
for future in as_completed(futures):
burst_results.append(future.result())
burst_duration = time.perf_counter() - burst_started
scenario_http["burst"] = burst_results
counts = status_counts(burst_results)
all_statuses.update(counts)
total_sent += len(burst_results)
print(f" Duration: {burst_duration:.3f}s; HTTP statuses: {format_status_counts(counts)}")
print(f"\nWaiting for worker flush / DLQ processing (up to {settle_timeout:.1f}s)...")
expected_dlq_delta = 50
ch_total, final_stats, settle_elapsed = wait_for_pipeline(
clickhouse_url=clickhouse_url,
stats_url=stats_url,
run_prefix=run_prefix,
expected_clickhouse=1500,
expected_dlq_delta=expected_dlq_delta,
baseline_stats=baseline_stats,
timeout_seconds=settle_timeout,
)
print(f"Pipeline verification wait: {settle_elapsed:.2f}s\n")
prefix_literal = quote_ch_string(run_prefix)
valid_ch = ch_count(clickhouse_url, f"startsWith(raw_id, {quote_ch_string(run_prefix + 'valid-')})")
burst_ch = ch_count(clickhouse_url, f"startsWith(raw_id, {quote_ch_string(run_prefix + 'burst-')})")
duplicate_ids = [payload["raw_id"] for payload in valid_payloads[:20]]
duplicate_list = ",".join(quote_ch_string(raw_id) for raw_id in duplicate_ids)
duplicate_unique_ch = ch_count(clickhouse_url, f"raw_id IN ({duplicate_list})")
dlq_records = evidence.collect(50, settle_timeout)
dlq_verified = evidence.validate()
reasons = Counter(record["failure_reason"] for record in dlq_records)
missing_dlq = reasons["missing_required_field"]
bad_rat_dlq = reasons["invalid_rat_value"]
bad_epoch_dlq = reasons["invalid_epoch"]
results: List[ScenarioResult] = []
valid_http_ok = status_counts(valid_results) == Counter({202: 1000})
results.append(
ScenarioResult(
"1. Valid records",
valid_http_ok and valid_ch == 1000,
f"HTTP 202={sum(r.status == 202 for r in valid_results)}/1000; ClickHouse={valid_ch}/1000",
)
)
# HTTP rejection and direct Kafka DLQ delivery are independent requirements.
missing_transport_ok = all(r.status == 400 and is_structured_json_error(r) for r in missing_results)
missing_passed = missing_transport_ok and missing_dlq == 20
results.append(
ScenarioResult(
"2. Missing caller_num",
missing_passed,
f"HTTP={format_status_counts(status_counts(missing_results))}; DLQ missing_required_field={missing_dlq}/20",
)
)
malformed_400 = sum(r.status == 400 for r in malformed_results)
results.append(
ScenarioResult(
"3. Malformed JSON",
malformed_400 == 10 and structured_errors == 10,
f"HTTP 400={malformed_400}/10; structured JSON errors={structured_errors}/10",
)
)
results.append(
ScenarioResult(
"4. Invalid RAT",
all(r.status == 202 for r in bad_rat_results) and bad_rat_dlq == 10,
f"HTTP 202={sum(r.status == 202 for r in bad_rat_results)}/10; DLQ invalid_rat_value={bad_rat_dlq if bad_rat_dlq is not None else 'unavailable'}/10",
)
)
results.append(
ScenarioResult(
"5. Invalid epoch",
all(r.status == 202 for r in bad_epoch_results) and bad_epoch_dlq == 10,
f"HTTP 202={sum(r.status == 202 for r in bad_epoch_results)}/10; DLQ invalid_epoch={bad_epoch_dlq if bad_epoch_dlq is not None else 'unavailable'}/10",
)
)
results.append(
ScenarioResult(
"6. Duplicate raw_id",
all(r.status == 202 for r in duplicate_results) and duplicate_unique_ch == 20,
f"duplicate POSTs accepted={sum(r.status == 202 for r in duplicate_results)}/20; unique rows after FINAL={duplicate_unique_ch}/20",
)
)
burst_429 = sum(r.status == 429 for r in burst_results)
burst_202 = sum(r.status == 202 for r in burst_results)
burst_network_errors = sum(r.status == 0 for r in burst_results)
results.append(
ScenarioResult(
"7. 500-record burst",
burst_duration < 5.0
and burst_202 == 500
and burst_429 == 0
and burst_network_errors == 0
and burst_ch == 500,
f"duration={burst_duration:.3f}s; 202={burst_202}/500; 429={burst_429}; network_errors={burst_network_errors}; ClickHouse={burst_ch}/500",
)
)
results.append(ScenarioResult("Actual DLQ envelopes", dlq_verified and len(dlq_records) == 50,
f"verified={dlq_verified}; records={len(dlq_records)}/50; reasons={dict(reasons)}"))
metrics_ok = False
if baseline_stats is not None and final_stats is not None:
expected_dlq_metric = 50
metrics_ok = (delta(final_stats["total_received"], baseline_stats["total_received"]) == 1540
and delta(final_stats["total_processed"], baseline_stats["total_processed"]) == 1520
and delta(final_stats["total_dlq"], baseline_stats["total_dlq"]) == expected_dlq_metric)
results.append(ScenarioResult("Acknowledged metrics", metrics_ok, f"received=1540; processed=1520; DLQ=50 expected"))
storage_ok, storage_detail = verify_normalized_rows(clickhouse_url, run_prefix, valid_payloads + burst_payloads)
results.append(ScenarioResult("Normalized storage", storage_ok, storage_detail))
print("=== Assessment Verification Summary ===")
print(f"Run prefix: {run_prefix}")
print(f"Records/requests sent: {total_sent} (expected 1,570)")
print(f"All HTTP statuses: {format_status_counts(all_statuses)}")
print(f"Run-scoped ClickHouse rows: {ch_total if ch_total is not None else 'unavailable'} (expected 1,500 unique valid+burst rows)")
if baseline_stats is not None and final_stats is not None:
print(f"/stats received delta: {delta(final_stats['total_received'], baseline_stats['total_received'])}")
print(f"/stats processed delta: {delta(final_stats['total_processed'], baseline_stats['total_processed'])}")
print(f"/stats DLQ delta: {delta(final_stats['total_dlq'], baseline_stats['total_dlq'])} (expected 50)")
print(f"/stats in-flight buffer: {final_stats.get('in_flight_buffer', 'n/a')}")
else:
print("/stats deltas: unavailable")
print("\nScenario results:")
for result in results:
marker = "PASS" if result.passed else "FAIL"
print(f" [{marker}] {result.name:<24} {result.detail}")
print("Direct Kafka fault injections: 30 additional (20 missing fields, 10 malformed JSON)")
print("API rejects structural errors; direct injections independently verify worker handling.")
passed = all(result.passed for result in results)
print("\nOVERALL: " + ("PASS" if passed else "FAIL"))
print("============================================================\n")
return 0 if passed else 1
except Exception as exc:
print(f"OVERALL: FAIL - verification could not complete: {exc}")
return 1
finally:
evidence.close()
if __name__ == "__main__":
import urllib.parse
parser = argparse.ArgumentParser(
description="Telecom Raw CDR Pipeline Traffic Generator & Verifier"
description="Assessment-focused Telecom CDR pipeline traffic generator and verifier"
)
parser.add_argument(
"--api-url", default="http://localhost:8000", help="Ingestion API base URL"
@@ -255,6 +574,36 @@ if __name__ == "__main__":
default="http://localhost:8123",
help="ClickHouse HTTP endpoint",
)
parser.add_argument(
"--stats-url",
default=None,
help="Aggregated /stats URL (default: <api-url>/stats)",
)
parser.add_argument(
"--settle-timeout",
type=float,
default=15.0,
help="Maximum seconds to wait for worker flush/DLQ processing",
)
parser.add_argument(
"--burst-workers",
type=int,
default=50,
help="Concurrent client threads for the 500-record burst",
)
parser.add_argument("--kafka-brokers", default=os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092"))
parser.add_argument("--raw-topic", default=os.getenv("KAFKA_RAW_TOPIC", "cdr.raw"))
parser.add_argument("--dlq-topic", default=os.getenv("KAFKA_DLQ_TOPIC", "cdr.raw.dlq"))
args = parser.parse_args()
run_verification(api_url=args.api_url, clickhouse_url=args.clickhouse_url)
stats_url = args.stats_url or f"{args.api_url.rstrip('/')}/stats"
sys.exit(
run_verification(
api_url=args.api_url,
clickhouse_url=args.clickhouse_url,
stats_url=stats_url,
settle_timeout=args.settle_timeout,
burst_workers=args.burst_workers,
kafka_brokers=args.kafka_brokers, raw_topic=args.raw_topic, dlq_topic=args.dlq_topic,
)
)

87
scripts/verification_kafka.py Normal file
View File

@@ -0,0 +1,87 @@
"""Independent, run-scoped Kafka evidence for the assessment verifier."""
import json
import time
import uuid
from confluent_kafka import Consumer, Producer, TopicPartition
class KafkaEvidence:
def __init__(self, brokers, raw_topic, dlq_topic, run_prefix):
self.prefix = run_prefix
self.raw_topic = raw_topic
self.consumer = Consumer({"bootstrap.servers": brokers,
"group.id": f"verification-{uuid.uuid4()}",
"enable.auto.commit": False})
try:
metadata = self.consumer.list_topics(dlq_topic, timeout=10).topics[dlq_topic]
if metadata.error:
raise RuntimeError(str(metadata.error))
positions = []
for partition in metadata.partitions:
tp = TopicPartition(dlq_topic, partition)
_, high = self.consumer.get_watermark_offsets(tp, timeout=10)
positions.append(TopicPartition(dlq_topic, partition, high))
self.consumer.assign(positions)
self.producer = Producer({"bootstrap.servers": brokers, "acks": "all",
"enable.idempotence": True, "delivery.timeout.ms": 10000})
except Exception:
self.consumer.close()
raise
self.expected = {}
self.records = []
def inject(self, key, body, reason):
self.expected[key] = (body, reason)
errors = []
self.producer.produce(self.raw_topic, key=key, value=body,
on_delivery=lambda err, msg: errors.append(err) if err else None)
if self.producer.flush(12) or errors:
raise RuntimeError(f"Direct Kafka injection failed: {errors}")
def collect(self, expected_count, timeout):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
msg = self.consumer.poll(0.1)
if msg is None:
if len(self.records) >= expected_count:
break
continue
if msg.error():
raise RuntimeError(str(msg.error()))
envelope = json.loads(msg.value())
if self.prefix in json.dumps(envelope.get("original_payload")):
self.records.append(envelope)
return self.records
def validate(self):
seen = {}
for envelope in self.records:
if not all(name in envelope for name in ("original_payload", "failure_reason", "failed_field", "failed_at", "worker_instance")):
return False
original = envelope["original_payload"]
if isinstance(original, str):
try:
original = json.loads(original)
except ValueError:
pass
matched = [key for key in self.expected if key in json.dumps(original)]
if len(matched) != 1:
return False
key = matched[0]
body, reason = self.expected[key]
try:
expected_original = json.loads(body)
except ValueError:
expected_original = body.decode()
if original != expected_original or envelope["failure_reason"] != reason:
return False
expected_field = {"missing_required_field": "caller_num", "invalid_rat_value": "rat",
"invalid_epoch": "epoch_start", "malformed_json": None}[reason]
if envelope["failed_field"] != expected_field or not envelope["worker_instance"]:
return False
seen[key] = seen.get(key, 0) + 1
return set(seen) == set(self.expected) and all(count == 1 for count in seen.values())
def close(self):
self.consumer.close()

41
tests/test_verifier_contract.py Normal file
View File

@@ -0,0 +1,41 @@
"""Static-review follow-up regressions. Execution is deferred by user request."""
import contextlib
import io
import json
import sys
import unittest
from types import SimpleNamespace
from unittest.mock import Mock, patch
from app.common.validation import PayloadError, parse_payload, validate_structure
from app.worker.consumer import ProcessingWorker
from scripts import generate_traffic
from tests.test_api_delivery import payload
class TestVerifierContract(unittest.TestCase):
def test_missing_kafka_evidence_fails_before_sending_traffic(self):
unavailable = SimpleNamespace(KafkaEvidence=Mock(side_effect=ConnectionError("offline")))
with patch.dict(sys.modules, {"verification_kafka": unavailable}), \
patch.object(generate_traffic, "send_http_request") as request, \
contextlib.redirect_stdout(io.StringIO()):
result = generate_traffic.run_verification("http://unused", "http://unused", "http://unused", 1, 1)
self.assertEqual(result, 1)
request.assert_not_called()
def test_worker_unavailable_is_not_a_zero_stats_snapshot(self):
with patch.object(generate_traffic, "fetch_json", return_value={"pipeline": {"worker_status": "unreachable"}}):
self.assertIsNone(generate_traffic.stats_snapshot("http://unused"))
def test_overflowing_json_float_is_rejected_even_in_extra_fields(self):
with self.assertRaises(PayloadError):
parse_payload(b'{"extra":1e400}')
def test_unpaired_surrogate_is_not_an_encodable_record_identity(self):
data = payload() | {"raw_id": "\ud800"}
with self.assertRaises(PayloadError):
validate_structure(parse_payload(json.dumps(data).encode()))
def test_deeply_nested_bad_json_can_be_preserved_for_dlq(self):
body = b"[" * 2000 + b"0" + b"]" * 2000
self.assertEqual(ProcessingWorker._original(SimpleNamespace(value=lambda: body)), body.decode())