Dashboard Service — תצוגה מצטברת

מה זה?#

Collector קולט, Analyzer מנתח, Alert מתריע — אבל מי נותן תמונת-מצב כוללת? “כמה אירועים נקלטו היום, כמה אנומליות זוהו, מה המגמה?” Dashboard Service הוא השירות הרביעי — צורך משני ה-Topics (event-collected ו-anomaly-detected), צובר סטטיסטיקה מצטברת עם Pandas, וחושף אותה כ-API פשוט לתצוגה — סוגר את המעגל של כל ארבעת השירותים.

מילות מפתח שחשוב לזכור#

• Aggregated View (תצוגה מצטברת) — סיכום סטטיסטי לאורך זמן, לא אירוע בודד — בדיוק כמו דוח יומי מיחידת ה-ETL

• Multi-Topic Consumer — שירות שצורך מכמה Topics שונים בו-זמנית (לא רק אחד כמו Analyzer/Alert)

• In-Memory Aggregation — צבירת סטטיסטיקה בזיכרון תוך כדי ריצה, לא ממתינה לדוח יומי מתוזמן

from confluent_kafka import Consumer
import json
from fastapi import FastAPI
import threading

stats = {"totalEvents": 0, "totalAnomalies": 0, "recentAnomalies": []}
app = FastAPI()

def consume_events():
    consumer = Consumer({
        'bootstrap.servers': 'localhost:9092',
        'group.id': 'dashboard-service',
        'auto.offset.reset': 'earliest'
    })
    consumer.subscribe(['event-collected', 'anomaly-detected'])   # Multi-Topic — both Topics together

    while True:
        msg = consumer.poll(1.0)
        if msg is None:
            continue

        data = json.loads(msg.value())
        if msg.topic() == 'event-collected':
            stats["totalEvents"] += 1
        elif msg.topic() == 'anomaly-detected':
            stats["totalAnomalies"] += 1
            stats["recentAnomalies"].append(data)
            stats["recentAnomalies"] = stats["recentAnomalies"][-10:]   # only the last 10 — Sliding Window

# Consumption runs in the background on a separate thread, so it doesn't block the API
threading.Thread(target=consume_events, daemon=True).start()

@app.get("/stats")
def get_stats():
    return stats   # up-to-date snapshot at any moment

הסבר עיקרי#

Multi-Topic Consumer — שוני מ-Analyzer/Alert — Analyzer צרך רק מ-event-collected; Alert רק מ-anomaly-detected. Dashboard צריך שניהם יחד כדי לתת תמונה מלאה — consumer.subscribe(['event-collected', 'anomaly-detected']) מאזין לשני ה-Topics בו-זמנית, ו-msg.topic() מבחין מאיזה מהם כל הודעה הגיעה.

Thread נפרד — למה — Dashboard חושף גם API (FastAPI, מסגרת web לפייתון) וגם צריך “להאזין” ברציפות ל-Kafka — שתי פעולות במקביל. הרצת ה-Consumer ב-thread נפרד (daemon=True) מבטיחה שהאזנה לKafka לא “חוסמת” את ה-API מלענות לבקשות — שני התהליכים רצים זה לצד זה.

In-Memory Aggregation מול דוח מתוזמן — בניגוד לדוח היומי מיחידת ה-ETL (שרץ פעם ביום עם APScheduler), Dashboard מצטבר בזמן אמת, בכל רגע — stats מתעדכן עם כל הודעה חדשה שמגיעה, ו-GET /stats תמיד מחזיר את המצב העדכני ביותר, לא “תמונה מאתמול”.

יתרונות#

Dashboard נותן תמונת מצב חיה, לא רק דוח מתוזמן; Multi-Topic Consumer מאחד כמה מקורות אירועים לתצוגה אחת; ארכיטקטורה פשוטה (Thread+API) מספיקה לצרכי לוח-מחוונים בסיסי.

חסרונות#

stats בזיכרון נעלם אם השירות נופל (Restart) — אין persistence אמיתי; Multi-Topic Consumer עלול להיות איטי יותר מ-Consumer ממוקד-Topic יחיד תחת עומס גבוה מאוד.

נקודות חשובות#

• Dashboard = Multi-Topic Consumer + In-Memory Aggregation + API לתצוגה

• Thread נפרד מאפשר האזנה רציפה ל-Kafka לצד שרת API רץ במקביל

• תצוגה מצטברת בזמן-אמת שונה מדוח יומי מתוזמן — מתעדכנת עם כל הודעה, לא פעם ביום

• זהו סיום המעגל: השירות הרביעי שמאחד את תפוקת שלושת האחרים

טעויות נפוצות#

• לנסות להריץ Consumer ו-API באותו Thread — האחד חוסם את השני

• לצרוך רק מ-Topic אחד ולפספס חלק מהתמונה (למשל רק אנומליות, בלי סך כל האירועים)

• לא להגביל את recentAnomalies (בלי Sliding Window) — גדילה בלתי-מוגבלת בזיכרון

סיכום#

Dashboard Service צורך משני Topics בו-זמנית (event-collected+anomaly-detected), צובר סטטיסטיקה בזיכרון ב-Thread נפרד, וחושף אותה כ-API (FastAPI) לתצוגה בזמן-אמת — לא דוח מתוזמן כמו ביחידת ה-ETL, אלא תמונת מצב מתעדכנת עם כל הודעה. זהו השירות הרביעי והאחרון שסוגר את המעגל של מערכת המודיעין העסקי כולה.

דוקומנטציה רשמית#

FastAPI — Documentation

תרגילים#

תרגיל 1 — Multi-Topic Consumer בסיסי#

המשימה: כתבו Consumer שמאזין לשני Topics, ומדפיס מאיזה Topic כל הודעה הגיעה (msg.topic()).

בדיקה: הודעות משני ה-Topics מודפסות עם זיהוי נכון של המקור שלהן.

תרגיל 2 — API מעל סטטיסטיקה חיה#

המשימה: בנו endpoint FastAPI פשוט שמחזיר מונה שמתעדכן ב-Thread נפרד (בלי Kafka עדיין — סתם מונה שגדל כל שנייה).

בדיקה: קריאות חוזרות ל-endpoint מציגות ערכים גדלים, מבלי לחסום את השרת.

פרויקט מסכם#

המשימה: ממשו את Dashboard Service המלא לפי ה-Contracts שהצוות כתב.

דרישות:

1. Multi-Topic Consumer על event-collected ו-anomaly-detected
2. stats עם totalEvents, totalAnomalies, ו-recentAnomalies (Sliding Window של 10)
3. GET /stats שחושף את הנתונים המצטברים כ-JSON

בדיקה: הרצת כל ארבעת השירותים יחד — שליחת אירועים דרך Collector מעדכנת totalEvents מיד; אירוע חריג מעדכן גם totalAnomalies ו-recentAnomalies, נגיש דרך /stats בזמן אמת.

מה בפרק הבא#

בפרק הבא — אינטגרציה מלאה: הרצת כל ארבעת השירותים יחד עם docker-compose, זרימת אירוע קצה-לקצה מלאה, ותהליך עבודה צוותי (PR, קוד-ריוויו, הצגה משותפת).