Skip to content

I
ids-project
Commit 38c0d8ce authored by AreejMh57's avatar AreejMh57
Browse files
Options

Update ids_streaming.py with blocked-IP filter; clean up spark experiment files

parent 649c43ed
Show whitespace changes
Inline Side-by-side
Showing with 623 additions and 1 deletion
ids-project-k1/spark/ids_streaming.py
View file @ 38c0d8ce
...@@ -69,6 +69,28 @@ index_to_label_udf = udf(index_to_label, StringType()) ...@@ -69,6 +69,28 @@ index_to_label_udf = udf(index_to_label, StringType())
# (matched on src_ip + timestamp) to build a labeled retraining set. # (matched on src_ip + timestamp) to build a labeled retraining set.
HDFS_RESULTS_PATH = "hdfs://namenode:9000/ids/model_results" HDFS_RESULTS_PATH = "hdfs://namenode:9000/ids/model_results"
# ── NEW: Blocked IPs Filter ──
# Before analyzing any batch, Spark asks FastAPI which IPs are
# currently blocked, and drops their logs BEFORE running them through
# the model. This avoids wasting inference cycles on traffic that's
# already handled, and avoids spamming duplicate alerts for a source
# the admin already blocked.
BLOCKED_IPS_ENDPOINT = "http://10.0.0.124:8000/api/blocked-ips"
def get_blocked_ips():
try:
resp = requests.get(BLOCKED_IPS_ENDPOINT, timeout=3)
if resp.status_code == 200:
data = resp.json()
# Adjust parsing here if the FastAPI response shape differs,
# e.g. {"blocked_ips": [...]} vs a plain list [...]
if isinstance(data, dict):
return set(data.get("blocked_ips", []))
return set(data)
except Exception as ex:
print("Could not fetch blocked IPs, proceeding without filter: " + str(ex))
return set()
print("Connecting to Kafka...") print("Connecting to Kafka...")
raw_stream = spark.readStream \ raw_stream = spark.readStream \
.format("kafka") \ .format("kafka") \
...@@ -86,6 +108,19 @@ def process_batch(batch_df, batch_id): ...@@ -86,6 +108,19 @@ def process_batch(batch_df, batch_id):
return return
count = batch_df.count() count = batch_df.count()
print("=== Batch " + str(batch_id) + " --- " + str(count) + " logs ===") print("=== Batch " + str(batch_id) + " --- " + str(count) + " logs ===")
# ── NEW: filter out already-blocked source IPs before analysis ──
blocked_ips = get_blocked_ips()
if blocked_ips:
before_count = count
batch_df = batch_df.filter(~col("src_ip").isin(list(blocked_ips)))
count = batch_df.count()
print("Filtered " + str(before_count - count) + " logs from " +
str(len(blocked_ips)) + " blocked IP(s). Remaining: " + str(count))
if batch_df.isEmpty():
print("All logs in this batch were from blocked IPs — skipping batch.")
return
try: try:
featured = assembler.transform(batch_df) featured = assembler.transform(batch_df)
s1 = model_stage1.transform(featured) \ s1 = model_stage1.transform(featured) \
......
ids-project-k2/test/detection_latency_test.py 0 → 100644
View file @ 38c0d8ce
import json
import time
from datetime import datetime
from kafka import KafkaProducer
import psycopg2
producer = KafkaProducer(
bootstrap_servers=['10.0.0.15:9093'],
value_serializer=lambda x: json.dumps(x).encode('utf-8')
)
known_attack_ip = "203.0.113.99"
# ── سجل حقيقي 100% من ملف 02-21-2018.csv (DDOS attack-HOIC) ──
attack_record = {
"timestamp": datetime.now().isoformat(),
"src_ip": known_attack_ip,
"Dst Port": 880.0,
"Protocol": 6.0,
"Flow Duration": 1.0,
"Tot Fwd Pkts": 3.0,
"Tot Bwd Pkts": 4.0,
"TotLen Fwd Pkts": 316.0,
"TotLen Bwd Pkts": 935.0,
"Fwd Pkt Len Max": 316.0,
"Fwd Pkt Len Min": 0.0,
"Bwd Pkt Len Max": 935.0,
"Bwd Pkt Len Min": 0.0,
"Flow Pkts/s": 6.1001487,
"Flow Byts/s": 15.6123,
"Fwd Header Len": 7.0,
"Bwd Header Len": 9.0,
"Fwd Pkts/s": 97.4714923,
"Bwd Pkts/s": 36.6286564,
"Pkt Len Min": 8.0,
"Pkt Len Max": 95.0,
"SYN Flag Cnt": 7.0,
"RST Flag Cnt": 3.0,
"PSH Flag Cnt": 2.0,
"ACK Flag Cnt": 1.0,
"label": "DDOS attack-HOIC"
}
print(f"[T=0] Sending REAL DDOS attack-HOIC record (src_ip={known_attack_ip}) to Kafka...")
attack_start_time = time.time()
producer.send('network-logs', value=attack_record)
producer.flush()
print(f"Sent at: {attack_record['timestamp']}")
conn = psycopg2.connect(
host="10.0.0.18", dbname="ids_db",
user="ids_user", password="ids_pass123"
)
cur = conn.cursor()
print("Monitoring PostgreSQL for detection (polling every 0.5s)...")
timeout_seconds = 60
detected = False
while (time.time() - attack_start_time) < timeout_seconds:
cur.execute("""
SELECT id, attack_type, occurrence_count, created_at FROM alerts
WHERE src_ip = %s ORDER BY created_at DESC LIMIT 1
""", (known_attack_ip,))
result = cur.fetchone()
if result:
detection_latency = time.time() - attack_start_time
print(f"\nDETECTED")
print(f"Attack type: {result[1]}")
print(f"Detection latency: {detection_latency:.2f} seconds")
detected = True
break
time.sleep(0.5)
if not detected:
print(f"\nNo detection within {timeout_seconds}s timeout")
cur.close()
conn.close()
ids-project-k2/test/full_system_load_test.py 0 → 100644
View file @ 38c0d8ce
import json
import time
import random
from datetime import datetime
from kafka import KafkaProducer
import psycopg2
# ============================================================
# CONFIGURATION
# ============================================================
TOTAL_RECORDS = 200 # إجمالي عدد السجلات المُرسَلة
ATTACK_RATIO = 0.10 # نسبة الهجوم (10%)
KAFKA_BROKER = '10.0.0.15:9093'
DB_HOST = "10.0.0.18"
TIMEOUT_SECONDS = 120 # مدة انتظار الكشف بعد إرسال آخر سجل
# ============================================================
# TEMPLATES (from real CIC-IDS2018 records)
# ============================================================
BENIGN_TEMPLATE = {
"Dst Port": 80.0, "Protocol": 6.0, "Flow Duration": 10085.0,
"Tot Fwd Pkts": 3.0, "Tot Bwd Pkts": 4.0,
"TotLen Fwd Pkts": 316.0, "TotLen Bwd Pkts": 935.0,
"Fwd Pkt Len Max": 316.0, "Fwd Pkt Len Min": 0.0,
"Bwd Pkt Len Max": 935.0, "Bwd Pkt Len Min": 0.0,
"Flow Pkts/s": 694.1001487, "Flow Byts/s": 124045.6123,
"Fwd Header Len": 72.0, "Bwd Header Len": 92.0,
"Fwd Pkts/s": 297.4714923, "Bwd Pkts/s": 396.6286564,
"Pkt Len Min": 0.0, "Pkt Len Max": 935.0,
"SYN Flag Cnt": 0.0, "RST Flag Cnt": 1.0,
"PSH Flag Cnt": 1.0, "ACK Flag Cnt": 0.0,
}
ATTACK_TEMPLATE = {
"Dst Port": 880.0, "Protocol": 6.0, "Flow Duration": 1.0,
"Tot Fwd Pkts": 3.0, "Tot Bwd Pkts": 4.0,
"TotLen Fwd Pkts": 316.0, "TotLen Bwd Pkts": 935.0,
"Fwd Pkt Len Max": 316.0, "Fwd Pkt Len Min": 0.0,
"Bwd Pkt Len Max": 935.0, "Bwd Pkt Len Min": 0.0,
"Flow Pkts/s": 6.1001487, "Flow Byts/s": 15.6123,
"Fwd Header Len": 7.0, "Bwd Header Len": 9.0,
"Fwd Pkts/s": 97.4714923, "Bwd Pkts/s": 36.6286564,
"Pkt Len Min": 8.0, "Pkt Len Max": 95.0,
"SYN Flag Cnt": 7.0, "RST Flag Cnt": 3.0,
"PSH Flag Cnt": 2.0, "ACK Flag Cnt": 1.0,
}
# ============================================================
# GENERATE UNIQUE IP POOLS (test-only ranges, won't collide)
# ============================================================
def gen_attack_ip(i):
return f"203.0.113.{i % 254 + 1}"
def gen_benign_ip(i):
return f"198.51.100.{i % 254 + 1}"
# ============================================================
# BUILD MIXED RECORD SET
# ============================================================
n_attacks = int(TOTAL_RECORDS * ATTACK_RATIO)
n_benign = TOTAL_RECORDS - n_attacks
records = []
for i in range(n_benign):
r = dict(BENIGN_TEMPLATE)
r["src_ip"] = gen_benign_ip(i)
r["label"] = "Benign"
r["_expected"] = "Benign"
records.append(r)
for i in range(n_attacks):
r = dict(ATTACK_TEMPLATE)
r["src_ip"] = gen_attack_ip(i)
r["label"] = "DDOS attack-HOIC"
r["_expected"] = "Attack"
records.append(r)
random.shuffle(records) # نخلطهم عشوائياً قبل الإرسال
print(f"Prepared {len(records)} records total: {n_benign} Benign + {n_attacks} Attack ({ATTACK_RATIO*100:.0f}%)")
# ============================================================
# SEND BURST TO KAFKA
# ============================================================
producer = KafkaProducer(
bootstrap_servers=[KAFKA_BROKER],
value_serializer=lambda x: json.dumps(x).encode('utf-8')
)
attack_ips_sent = set()
sent_timestamps = {}
print(f"\n[T=0] Sending burst of {len(records)} records to Kafka...")
burst_start_time = time.time()
for r in records:
r["timestamp"] = datetime.now().isoformat()
expected = r.pop("_expected")
if expected == "Attack":
attack_ips_sent.add(r["src_ip"])
sent_timestamps[r["src_ip"]] = time.time()
producer.send('network-logs', value=r)
producer.flush()
burst_send_duration = time.time() - burst_start_time
print(f"Burst sent in {burst_send_duration:.2f} seconds")
print(f"Attack IPs sent ({len(attack_ips_sent)}): {sorted(attack_ips_sent)}")
# ============================================================
# MONITOR POSTGRESQL
# ============================================================
conn = psycopg2.connect(host=DB_HOST, dbname="ids_db",
user="ids_user", password="ids_pass123")
cur = conn.cursor()
detected = {}
false_positives = []
print(f"\nMonitoring PostgreSQL for up to {TIMEOUT_SECONDS}s...")
monitor_start = time.time()
while (time.time() - monitor_start) < TIMEOUT_SECONDS:
# تحقق من التنبيهات المرتبطة بـ IPs الهجوم
for ip in attack_ips_sent:
if ip in detected:
continue
cur.execute("""
SELECT attack_type, created_at FROM alerts
WHERE src_ip = %s ORDER BY created_at DESC LIMIT 1
""", (ip,))
result = cur.fetchone()
if result:
latency = time.time() - sent_timestamps[ip]
detected[ip] = {"attack_type": result[0], "latency": latency}
# تحقق من false positives (IPs طبيعية ظهرت كتنبيه)
cur.execute("""
SELECT DISTINCT src_ip, attack_type FROM alerts
WHERE src_ip LIKE '198.51.100.%'
""")
fps = cur.fetchall()
for ip, atype in fps:
if ip not in [f["ip"] for f in false_positives]:
false_positives.append({"ip": ip, "attack_type": atype})
if len(detected) == len(attack_ips_sent):
break
time.sleep(0.5)
cur.close()
conn.close()
# ============================================================
# RESULTS
# ============================================================
print("\n" + "="*60)
print("RESULTS")
print("="*60)
print(f"Total records sent: {len(records)} ({n_benign} Benign + {n_attacks} Attack)")
print(f"Attacks detected: {len(detected)} / {len(attack_ips_sent)}")
print(f"Detection Rate: {len(detected)/len(attack_ips_sent)*100:.1f}%")
if detected:
latencies = [d["latency"] for d in detected.values()]
print(f"\nDetection Latency (seconds):")
print(f" Min: {min(latencies):.2f}")
print(f" Max: {max(latencies):.2f}")
print(f" Avg: {sum(latencies)/len(latencies):.2f}")
print(f"\nFalse Positives (Benign flagged as Attack): {len(false_positives)}")
for fp in false_positives:
print(f" {fp['ip']} -> {fp['attack_type']}")
undetected = attack_ips_sent - set(detected.keys())
if undetected:
print(f"\nUndetected attacks (missed): {len(undetected)}")
for ip in undetected:
print(f" {ip}")
ids-project-k2/test/full_system_load_test_1000.py 0 → 100644
View file @ 38c0d8ce
import json
import time
import random
from datetime import datetime
from kafka import KafkaProducer
import psycopg2
# ============================================================
# CONFIGURATION
# ============================================================
TOTAL_RECORDS = 1000 # إجمالي عدد السجلات المُرسَلة
ATTACK_RATIO = 0.10 # نسبة الهجوم (10%)
KAFKA_BROKER = '10.0.0.15:9093'
DB_HOST = "10.0.0.18"
TIMEOUT_SECONDS = 180 # مدة انتظار الكشف بعد إرسال آخر سجل
# ============================================================
# TEMPLATES (from real CIC-IDS2018 records)
# ============================================================
BENIGN_TEMPLATE = {
"Dst Port": 80.0, "Protocol": 6.0, "Flow Duration": 10085.0,
"Tot Fwd Pkts": 3.0, "Tot Bwd Pkts": 4.0,
"TotLen Fwd Pkts": 316.0, "TotLen Bwd Pkts": 935.0,
"Fwd Pkt Len Max": 316.0, "Fwd Pkt Len Min": 0.0,
"Bwd Pkt Len Max": 935.0, "Bwd Pkt Len Min": 0.0,
"Flow Pkts/s": 694.1001487, "Flow Byts/s": 124045.6123,
"Fwd Header Len": 72.0, "Bwd Header Len": 92.0,
"Fwd Pkts/s": 297.4714923, "Bwd Pkts/s": 396.6286564,
"Pkt Len Min": 0.0, "Pkt Len Max": 935.0,
"SYN Flag Cnt": 0.0, "RST Flag Cnt": 1.0,
"PSH Flag Cnt": 1.0, "ACK Flag Cnt": 0.0,
}
ATTACK_TEMPLATE = {
"Dst Port": 880.0, "Protocol": 6.0, "Flow Duration": 1.0,
"Tot Fwd Pkts": 3.0, "Tot Bwd Pkts": 4.0,
"TotLen Fwd Pkts": 316.0, "TotLen Bwd Pkts": 935.0,
"Fwd Pkt Len Max": 316.0, "Fwd Pkt Len Min": 0.0,
"Bwd Pkt Len Max": 935.0, "Bwd Pkt Len Min": 0.0,
"Flow Pkts/s": 6.1001487, "Flow Byts/s": 15.6123,
"Fwd Header Len": 7.0, "Bwd Header Len": 9.0,
"Fwd Pkts/s": 97.4714923, "Bwd Pkts/s": 36.6286564,
"Pkt Len Min": 8.0, "Pkt Len Max": 95.0,
"SYN Flag Cnt": 7.0, "RST Flag Cnt": 3.0,
"PSH Flag Cnt": 2.0, "ACK Flag Cnt": 1.0,
}
# ============================================================
# GENERATE UNIQUE IP POOLS (test-only ranges, won't collide)
# ============================================================
def gen_attack_ip(i):
return f"203.0.113.{i % 254 + 1}"
def gen_benign_ip(i):
return f"198.51.100.{i % 254 + 1}"
# ============================================================
# BUILD MIXED RECORD SET
# ============================================================
n_attacks = int(TOTAL_RECORDS * ATTACK_RATIO)
n_benign = TOTAL_RECORDS - n_attacks
records = []
for i in range(n_benign):
r = dict(BENIGN_TEMPLATE)
r["src_ip"] = gen_benign_ip(i)
r["label"] = "Benign"
r["_expected"] = "Benign"
records.append(r)
for i in range(n_attacks):
r = dict(ATTACK_TEMPLATE)
r["src_ip"] = gen_attack_ip(i)
r["label"] = "DDOS attack-HOIC"
r["_expected"] = "Attack"
records.append(r)
random.shuffle(records) # نخلطهم عشوائياً قبل الإرسال
print(f"Prepared {len(records)} records total: {n_benign} Benign + {n_attacks} Attack ({ATTACK_RATIO*100:.0f}%)")
# ============================================================
# SEND BURST TO KAFKA
# ============================================================
producer = KafkaProducer(
bootstrap_servers=[KAFKA_BROKER],
value_serializer=lambda x: json.dumps(x).encode('utf-8')
)
attack_ips_sent = set()
sent_timestamps = {}
print(f"\n[T=0] Sending burst of {len(records)} records to Kafka...")
burst_start_time = time.time()
for r in records:
r["timestamp"] = datetime.now().isoformat()
expected = r.pop("_expected")
if expected == "Attack":
attack_ips_sent.add(r["src_ip"])
sent_timestamps[r["src_ip"]] = time.time()
producer.send('network-logs', value=r)
producer.flush()
burst_send_duration = time.time() - burst_start_time
print(f"Burst sent in {burst_send_duration:.2f} seconds")
print(f"Attack IPs sent ({len(attack_ips_sent)}): {sorted(attack_ips_sent)}")
# ============================================================
# MONITOR POSTGRESQL
# ============================================================
conn = psycopg2.connect(host=DB_HOST, dbname="ids_db",
user="ids_user", password="ids_pass123")
cur = conn.cursor()
detected = {}
false_positives = []
print(f"\nMonitoring PostgreSQL for up to {TIMEOUT_SECONDS}s...")
monitor_start = time.time()
while (time.time() - monitor_start) < TIMEOUT_SECONDS:
# تحقق من التنبيهات المرتبطة بـ IPs الهجوم
for ip in attack_ips_sent:
if ip in detected:
continue
cur.execute("""
SELECT attack_type, created_at FROM alerts
WHERE src_ip = %s ORDER BY created_at DESC LIMIT 1
""", (ip,))
result = cur.fetchone()
if result:
latency = time.time() - sent_timestamps[ip]
detected[ip] = {"attack_type": result[0], "latency": latency}
# تحقق من false positives (IPs طبيعية ظهرت كتنبيه)
cur.execute("""
SELECT DISTINCT src_ip, attack_type FROM alerts
WHERE src_ip LIKE '198.51.100.%'
""")
fps = cur.fetchall()
for ip, atype in fps:
if ip not in [f["ip"] for f in false_positives]:
false_positives.append({"ip": ip, "attack_type": atype})
if len(detected) == len(attack_ips_sent):
break
time.sleep(0.5)
cur.close()
conn.close()
# ============================================================
# RESULTS
# ============================================================
print("\n" + "="*60)
print("RESULTS")
print("="*60)
print(f"Total records sent: {len(records)} ({n_benign} Benign + {n_attacks} Attack)")
print(f"Attacks detected: {len(detected)} / {len(attack_ips_sent)}")
print(f"Detection Rate: {len(detected)/len(attack_ips_sent)*100:.1f}%")
if detected:
latencies = [d["latency"] for d in detected.values()]
print(f"\nDetection Latency (seconds):")
print(f" Min: {min(latencies):.2f}")
print(f" Max: {max(latencies):.2f}")
print(f" Avg: {sum(latencies)/len(latencies):.2f}")
print(f"\nFalse Positives (Benign flagged as Attack): {len(false_positives)}")
for fp in false_positives:
print(f" {fp['ip']} -> {fp['attack_type']}")
undetected = attack_ips_sent - set(detected.keys())
if undetected:
print(f"\nUndetected attacks (missed): {len(undetected)}")
for ip in undetected:
print(f" {ip}")
ids-project-k2/test/spark_burst_simulation.py 0 → 100644
View file @ 38c0d8ce
import requests
import time
import json
FASTAPI_URL = "http://localhost:8000/api/alerts" # عدّليها لـ 10.0.0.124:8000 لو بتشغليها من مكان تاني غير k2
def simulate_spark_alert_burst(n_unique_ips, base_ip_prefix="203.0.113"):
results = {"success": 0, "timeout": 0, "other_error": 0, "latencies": []}
print(f"\n{'='*60}")
print(f"Simulating Spark sending {n_unique_ips} sequential alerts (timeout=3s each)...")
print(f"{'='*60}")
batch_start = time.time()
for i in range(n_unique_ips):
alert = {
"attack_type": "DDoS",
"src_ip": f"{base_ip_prefix}.{i % 254 + 1}",
"severity": "HIGH",
"occurrence_count": 1,
"sample_features": {}
}
start = time.time()
try:
resp = requests.post(FASTAPI_URL, json=alert, timeout=3)
elapsed = time.time() - start
results["latencies"].append(elapsed)
if resp.status_code == 200:
results["success"] += 1
else:
print(f" [UNEXPECTED STATUS] Alert #{i+1}: {resp.status_code}")
except requests.exceptions.Timeout:
results["timeout"] += 1
print(f" [LOST] Alert #{i+1} DROPPED (timeout > 3s)")
except Exception as ex:
results["other_error"] += 1
print(f" [ERROR] Alert #{i+1}: {ex}")
total_batch_time = time.time() - batch_start
print(f"\n--- Results for {n_unique_ips} sequential alerts ---")
print(f"Successful: {results['success']}/{n_unique_ips}")
print(f"LOST (timeout): {results['timeout']}/{n_unique_ips} <-- critical metric")
print(f"Other errors: {results['other_error']}/{n_unique_ips}")
if results["latencies"]:
avg_lat = sum(results["latencies"]) / len(results["latencies"])
print(f"Avg latency (successful only): {avg_lat:.3f}s")
print(f"Max latency (successful only): {max(results['latencies']):.3f}s")
print(f"Total time for Spark to send this batch: {total_batch_time:.2f}s")
print(f"(Spark's next trigger fires every 15s — if this exceeds ~15s, batches back up)")
return {
"n": n_unique_ips,
"success": results["success"],
"lost": results["timeout"],
"errors": results["other_error"],
"total_time": total_batch_time
}
if __name__ == "__main__":
summary = []
for n in [10, 30, 60, 100]:
r = simulate_spark_alert_burst(n)
summary.append(r)
time.sleep(3) # فترة راحة بين كل حجم دفعة
print(f"\n{'='*60}")
print("FINAL SUMMARY")
print(f"{'='*60}")
print(f"{'Batch Size':<12}{'Success':<10}{'Lost':<10}{'Errors':<10}{'Total Time (s)':<15}")
for r in summary:
print(f"{r['n']:<12}{r['success']:<10}{r['lost']:<10}{r['errors']:<10}{r['total_time']:<15.2f}")
ids-project-k2/test/spark_burst_simulation_1000.py 0 → 100644
View file @ 38c0d8ce
import requests
import time
import json
FASTAPI_URL = "http://localhost:8000/api/alerts" # عدّليها لـ 10.0.0.124:8000 لو بتشغليها من مكان تاني غير k2
def simulate_spark_alert_burst(n_unique_ips, base_ip_prefix="203.0.113"):
results = {"success": 0, "timeout": 0, "other_error": 0, "latencies": []}
print(f"\n{'='*60}")
print(f"Simulating Spark sending {n_unique_ips} sequential alerts (timeout=3s each)...")
print(f"{'='*60}")
batch_start = time.time()
for i in range(n_unique_ips):
alert = {
"attack_type": "DDoS",
"src_ip": f"{base_ip_prefix}.{i % 254 + 1}",
"severity": "HIGH",
"occurrence_count": 1,
"sample_features": {}
}
start = time.time()
try:
resp = requests.post(FASTAPI_URL, json=alert, timeout=3)
elapsed = time.time() - start
results["latencies"].append(elapsed)
if resp.status_code == 200:
results["success"] += 1
else:
print(f" [UNEXPECTED STATUS] Alert #{i+1}: {resp.status_code}")
except requests.exceptions.Timeout:
results["timeout"] += 1
print(f" [LOST] Alert #{i+1} DROPPED (timeout > 3s)")
except Exception as ex:
results["other_error"] += 1
print(f" [ERROR] Alert #{i+1}: {ex}")
total_batch_time = time.time() - batch_start
print(f"\n--- Results for {n_unique_ips} sequential alerts ---")
print(f"Successful: {results['success']}/{n_unique_ips}")
print(f"LOST (timeout): {results['timeout']}/{n_unique_ips} <-- critical metric")
print(f"Other errors: {results['other_error']}/{n_unique_ips}")
if results["latencies"]:
avg_lat = sum(results["latencies"]) / len(results["latencies"])
print(f"Avg latency (successful only): {avg_lat:.3f}s")
print(f"Max latency (successful only): {max(results['latencies']):.3f}s")
print(f"Total time for Spark to send this batch: {total_batch_time:.2f}s")
print(f"(Spark's next trigger fires every 15s — if this exceeds ~15s, batches back up)")
return {
"n": n_unique_ips,
"success": results["success"],
"lost": results["timeout"],
"errors": results["other_error"],
"total_time": total_batch_time
}
if __name__ == "__main__":
summary = []
for n in [100, 300, 600, 1000]:
r = simulate_spark_alert_burst(n)
summary.append(r)
time.sleep(3) # فترة راحة بين كل حجم دفعة
print(f"\n{'='*60}")
print("FINAL SUMMARY")
print(f"{'='*60}")
print(f"{'Batch Size':<12}{'Success':<10}{'Lost':<10}{'Errors':<10}{'Total Time (s)':<15}")
for r in summary:
print(f"{r['n']:<12}{r['success']:<10}{r['lost']:<10}{r['errors']:<10}{r['total_time']:<15.2f}")
report/report-areej.pdf
View file @ 38c0d8ce
No preview for this file type
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment

HIAST - GitLab