diff --git a/scripts/generate_traffic.py b/scripts/generate_traffic.py index 7639beb..fe3acc5 100644 --- a/scripts/generate_traffic.py +++ b/scripts/generate_traffic.py @@ -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: /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, + ) + ) diff --git a/scripts/verification_kafka.py b/scripts/verification_kafka.py new file mode 100644 index 0000000..f2902a6 --- /dev/null +++ b/scripts/verification_kafka.py @@ -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() diff --git a/tests/test_verifier_contract.py b/tests/test_verifier_contract.py new file mode 100644 index 0000000..9f8784a --- /dev/null +++ b/tests/test_verifier_contract.py @@ -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())