Collector Service — REST + Kafka Producer

מה זה?#

Collector הוא “שער הכניסה” למערכת המודיעין — מקבל אירועים גולמיים (למשל מדדי חיישן, אירועי משתמש) דרך REST API רגיל (בדיוק כמו כל Controller שכבר נבנה בקורס), ומפרסם אותם ל-Kafka כדי שהם יהיו זמינים לכל שירות אחר שמעוניין (Analyzer, בשיעור הבא) — בלי ש-Collector בכלל צריך לדעת מי מאזין.

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

• Collector Pattern — שירות שתפקידו היחיד הוא לקלוט נתונים גולמיים ולהעביר הלאה, בלי לוגיקה עסקית כבדה (SRP)

• Ingestion Endpoint — ה-endpoint הספציפי שמקבל את הנתונים הנכנסים (POST /events)

• Fire-and-Forget (בהקשר הזה) — Collector מפרסם ל-Kafka ולא ממתין לתגובה מה-Consumers, בדיוק כמו שנלמד ביחידת NoSQL

// EventsController — Ingestion Endpoint, built exactly like the Controllers from the ASP.NET Core unit
[ApiController]
[Route("api/[controller]")]
public class EventsController : ControllerBase
{
    private readonly IProducer<Null, string> _producer;

    public EventsController(IProducer<Null, string> producer) => _producer = producer;   // DI, from the ASP.NET Core unit

    [HttpPost]
    public async Task<IActionResult> Collect([FromBody] EventDto eventDto)
    {
        // Matches exactly the Interface Contract written in the design lesson
        var message = JsonSerializer.Serialize(new
        {
            eventId = Guid.NewGuid().ToString(),
            eventType = eventDto.EventType,
            value = eventDto.Value,
            timestamp = DateTime.UtcNow.ToString("o"),
            source = eventDto.Source
        });

        await _producer.ProduceAsync("event-collected", new Message<Null, string> { Value = message });
        return Accepted();   // 202 — accepted, processed in the background
    }
}

// DataAnnotations on the DTO — validation, from the ASP.NET Core unit
public class EventDto
{
    [Required]
    public string EventType { get; set; }

    [Range(double.MinValue, double.MaxValue)]
    public double Value { get; set; }

    [Required]
    public string Source { get; set; }
}

הסבר עיקרי#

Collector לא מכיל שום לוגיקה עסקית — בכוונה — SRP במימוש: Collector “אחריות” יחידה היא לקלוט ולהעביר הלאה. הוא לא מחשב אנומליות (זה Analyzer), לא שולח התראות (זה Alert). זה מה שהופך אותו לפשוט, יציב, וקל להחליף/להרחיב בלי לגעת בשאר המערכת.

202 Accepted, לא 201 Created — מיחידת ה-HTTP: 201 אומר “נוצר משאב, הנה הוא”; 202 אומר “התקבל, מעובד — לא באופן מיידי/סינכרוני”. מכיוון שהעיבוד האמיתי (ניתוח, התראות) קורה אסינכרונית דרך Kafka, 202 הוא הביטוי המדויק יותר של “קיבלתי, אני לא מבטיח מתי זה יעובד”.

Message Schema תואם ל-Interface Contract — שימו לב שה-JSON שנשלח ל-Kafka תואם בדיוק למה שהוסכם בשיעור התכנון (eventId, eventType, value, timestamp, source) — זה לא מקרי: זו בדיוק המשמעות המעשית של “לכתוב חוזה לפני קוד” — מי שכותב את Analyzer (בשיעור הבא) יכול לסמוך על המבנה הזה בלי לתאם מחדש.

יתרונות#

Collector פשוט וממוקד (SRP) — קל לבדוק, לתחזק, ולהחליף; 202 Accepted מתקשר במדויק את אופי העיבוד האסינכרוני ללקוח; DataAnnotations תופסות קלט לא-תקין לפני שהוא בכלל מגיע ל-Kafka.

חסרונות#

Collector לא יודע (ולא יכול לדעת) אם העיבוד בפועל הצליח — זו תוצאה מכוונת של async, אך דורשת ניטור נפרד (Alert Service, בשיעורים הבאים) כדי לגלות כשלים; אין ערבות סדר מלאה בין הודעות מ-Producers שונים.

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

• Collector = SRP במימוש: קליטה והעברה בלבד, בלי לוגיקה עסקית

• 202 Accepted מתאים לעיבוד אסינכרוני; 201 Created לעיבוד סינכרוני שהושלם

• Message Schema חייב להתאים בדיוק ל-Interface Contract שהוסכם מראש

• DTO עם DataAnnotations תופס קלט שגוי לפני שהוא מגיע ל-Kafka בכלל

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

• להוסיף לוגיקה עסקית (חישובים, החלטות) ל-Collector — מפר SRP ומטשטש את גבול האחריות

• לסטות ממבנה ה-JSON המוסכם ב-Contract “כי זה נוח יותר” — שובר את Analyzer בצד השני

• להחזיר 201/200 במקום 202 על עיבוד שבפועל קורה אסינכרונית — מטעה לגבי הבטחת ההשלמה

סיכום#

Collector Service קולט אירועים דרך REST endpoint (עם ולידציה מ-DataAnnotations), ומפרסם אותם ל-Kafka בפורמט שתואם בדיוק ל-Interface Contract שהוסכם בשיעור התכנון. הוא נשאר ממוקד (SRP) — קליטה והעברה, בלי לוגיקה עסקית — ומחזיר 202 Accepted שמבטא נכון את האופי האסינכרוני של העיבוד שיקרה בהמשך אצל שירותים אחרים.

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

Microsoft Learn — Controller-based APIs

Confluent — .NET Producer

תרגילים#

תרגיל 1 — Endpoint בסיסי#

המשימה: כתבו EventsController עם POST /api/events שמקבל EventDto ומדפיס אותו (עדיין בלי Kafka), עם DataAnnotations.

בדיקה: בקשה תקינה מודפסת נכון; בקשה עם שדה חובה חסר מקבלת 400 אוטומטית.

תרגיל 2 — פרסום ל-Kafka#

המשימה: הוסיפו את חלק ה-Kafka Producer ל-Controller מתרגיל 1, ואמתו עם Consumer בסיסי (מיחידת NoSQL) שההודעה מתקבלת בפורמט הנכון.

בדיקה: ה-Consumer מקבל הודעה תואמת בדיוק ל-Message Schema שהוגדר.

פרויקט מסכם#

המשימה: ממשו את Collector Service המלא לפי ה-Interface Contract שהצוות כתב בשיעור התכנון.

דרישות:

1. EventsController עם POST /api/events, DataAnnotations מלאות
2. פרסום ל-Kafka בפורמט תואם-Contract במדויק
3. 202 Accepted בתשובה מוצלחת; 400 על קלט לא-תקין

בדיקה: בקשה תקינה מתקבלת ומפורסמת ל-Kafka בפורמט הנכון, ניתן לאימות עם Consumer בדיקה; בקשה לא-תקינה נדחית לפני שהיא מגיעה ל-Kafka בכלל.

מה בפרק הבא#

בפרק הבא — Analyzer Service: השירות השני — צורך אירועים מ-Kafka ב-Python, מעבד עם Pandas, ומזהה אנומליות.