בניית מערכת לשליחת Webhooks באמצעות Kafka, SQS & S3

זמן קריאה: 6 דקות

אין זה סוד כי העולם הופך להיות יותר ויותר מחובר, ומערכות תוכנה ואפליקציות אינן יוצאות מן הכלל. למערכות אלו נדרשת היכולת לתקשר עם מערכות אחרות הנפרדות מהן לחלוטין ולשתף ביניהן מידע.

Webhooks הם הודעות אוטומטיות שנשלחות ממערכות ואפליקציות כאשר אירוע כלשהו קרה כחלק מפעולתן. הודעות אלו מורכבות מ-payload (מטען בעברית, אך תסלחו לי אם אבחר להשתמש במונח הלועזי) והן נשלחות לכתובת URL ספציפית וייחודית. Webhooks כמעט תמיד יהיו מהירים יותר מ-Polling, מה שהופך אותם לפתרון אידיאלי לשליחת (או בעצם ״דחיפת״) אירועים ממערכת לעולם החיצון אליה.

Webhooks נמצאים בשימוש נרחב בענקיות התעשייה כגון Shopify, Stripe, Twitter & Twilio. אם ניקח לדוגמה את PayPal, אז על ידי שימוש ב-Webhooks הם מודיעים לאפליקצית ה-eCommerce שלך שהלקוחות ביצעו את התשלום.

במאמר אציג פתרון לבניית מערכת לשליחת Webhooks באמצעות שימוש ב- Apache Kafka, AWS SQS & S3. בחברה בה אני עובד היו מספר מימושים שונים לשליחת Webhooks במוצרים שונים של החברה. לכן, החלטנו למזג את כל הפתרונות הקיימים ולהציג פתרון מאוחד לכל המוצרים שלנו. הפתרון המוצג פה מבוסס על הפתרון שמימשנו.

אז, אם אתם מעוניינים לבנות מערכת לשליחת Webhooks משלכם, המאמר הבא הוא בשבילכם!

הגדרת הדרישות

בואו נסתכל על מה רצינו להשיג מהמערכת החדשה, ומדוע:

סקיילביליות – המערכת תידרש להסתגל לשינויים ועומסי עבודה שונים בעזרת horizontal scalability
תמיכה בניסיונות חוזרים – שליחת Webhooks יכולה לעיתים להיכשל, עקב בעיה בצד המקבל (למשל בעקבות באג, מערכת שאינה זמינה וכו׳). היינו יכולים פשוט לזרוק הצידה את ההודעות שנכשלו, אך זה יוביל לאובדן של מידע בצד המקבל. אנחנו זקוקים למנגנון לבצע שוב את השליחה עם exponential backoff ועד 24 שעות
תמיכה ב-Payload גדול – תוכן האירועים שנשלחים יכול להיות גדול מאוד, עד כמה מגה בייט של מידע
אגנוסטיות ל-Payload – הבנה של תוכן ההודעות איננו נדרש (ואף לא אפשרי). ההודעות נשלחות כפי שהגיעו למערכת.
מנגנון Deduplication של הודעות – כל האירועים יגיעו בסופו של דבר ליעדם, אך ייתכן כי יותר מפעם אחת. לכל אירוע יהיה מזהה ייחודי שיאפשר לצד המקבל להימנע מעיבוד כפול של הודעות.
ממשק API פומבי – הכרחי על מנת להירשם לאירועים.

הרשנו לעצמנו מעט פשרות:

אין הבטחה לסדר – איננו מבטחים סדר בין ההודעות שנשלחות
תמיכה בתקשורת מסוג HTTP/S בלבד – שליחת Webhhoks בכל פרוטוקול או צורת תקשורת אחרת (כגון Apache Kafka, TCP ועוד) איננו ייתמך.

בניית הפונקצינאליות

כבר מההתחלה היה לנו ברור כי המערכת תהיה מורכבת מ-2 שכבות: ה-API (שישמש לביצוע קונפיגורציה) וה-מוח (שיבצע את שליחת ההודעות בפועל). בואו נסתכל על כל אחד מהם בנפרד…

שכבת ה-API

משמשת לרישום והגדרת Webhooks. החלטנו ללכת על הפתרון הפשוט ביותר והנפוץ ביותר ולכן השתמשנו ב- REST API. ה-API ניתן לשימוש ע״י משתמש הקצה (מקבל ההודעות) או ע״י אפליקציה/מערכת אחרת.

ה-URL של נקודת הקצה בנוי בצורה הבאה:

[base-path]/webhooks/

נגדיר 3 סוגי אובייקטים:

1. אובייקט Webhook Target – ישות שמכמיסה בתוכה את URL היעד של ה-Webhook

				
					
{
    "id": 111111111,
    "url": "http://www.dummy.com",
    "createdAt": "2021-10-03T17:14:23Z",
    "updatedAt": "2021-10-03T17:14:23Z"
}

2. אובייקט Webhook Filter – ישות המכילה רשימה של סוגי אירועים שהמשתמש מעוניין להאזין להם (סוג של קבוצת אירועים)

				
					
{
    "id": 222222222,
    "events": [
        "core.orders.created.v1",
        "core.orders.updated.v2"
    ],
    "createdAt": "2021-10-03T17:14:23Z",
    "updatedAt": "2021-10-03T17:14:23Z"
}

3. אובייקט Webhook Subscription – ישות שמחזיקה שילוב של webhook target ו-webhook filter. הודעות ישלחו רק ל-subscriptions אקטיבים

				
					
{
    "id": 123456789,
    "targetId": 111111111,
    "filterId": 222222222,
    "active": true,
    "createdAt": "2021-10-03T17:14:23Z",
    "updatedAt": "2021-10-03T17:14:23Z"
}

ישנה תמיכה לכל ישות בפעולות הבאות:

1. שליפה לפי id

GET /webhooks/<entity>/<entity_id>

2. שליפה של הכל

GET /webhooks/<entity>

3. יצירה

POST /webhooks/<entity>

4. עדכון

PATCH /webhooks/<entity>/<entity_id>

5. מחיקה לפי id

DELETE /webhooks/<entity>/<entity_id>

התהליך ליצירת Webhook מורכב מ-3 שלבים:

יצירה של webhook target שיצביע למיקום שאליו נדרש ה-webhook להישלח.
יצירה של webhook filter שיאמר אילו אירועים נדרשים להישלח.
יצירה של webhook subscription לבצע קישור של target ושל filter.

שכבת ה-״מוח״ (Brain)

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

דבר ראשון, נגדיר טרמינולוגיה פשוטה:

המונח Webhook Event מתייחס לאירוע שהתקבל מאפליקציות/מערכות אחרות
המונח Webhook Delivery מתייחס לקומבינציה של Webhook Target ושל Webhook Event. מכיוון שאירוע מסוים יכול לעניין מספר רב של לקוחות, הוא ישוכפל כמספר היעדים שמעוניינים בו.
המונח Webhook Message מתאר את המידע עצמו שנשלח ל-URL היעד.

ה-Consumer

רכיב זה אחראי על טיפול בהודעות שמגיעות מ-Kafka ומחליט (על בסיס ה-subscription data) אילו יעדים רלוונטים לאירוע שהתקבל. מרגע ההחלטה, הוא מייצר webhook delivery לכל יעד שנדרש לקבל את המידע על האירוע, ומפרסם (publish) הודעה מקבילה ל-AWS SQS queue.

מכיוון של-AWS SQS יש מגבלה של 256KB לכל הודעה, וישנם הודעות במערכת הגדולות מכך, אנחנו מאחסנים את תוכן ההודעה (ה-Payload) ב-AWS S3 bucket לשליפה עתידית.

ה-Cache (זיכרון מטמון)

ישנם המון אירועים שקורים במערכת, ולכל אחד מהם אנו נדרשים להתאים את כל ה-subscriptions הרלוונטים המוגדרים ב-subscriptions data DB. על מנת לא להעמיס על על ה-DB עם קריאות תכופות, ובשילוב עם העובדה שה-subscriptions data לא משתנה לעיתים תכופות, אנחנו מחזיקים זיכרון (מטמון) מקומי עם הגדרות ה-subscriptions.

ה-Dispatcher

רכיב זה הינו החלק במערכת שמבצע את השליחה בפועל של ה-Payload של האירועים כ-Webhooks Messages ל-URL היעד. הוא מושך את ה-Webhook Delivery שנוצר ע״י ה-Consumer מה-AWS SQS Queue, ומבצע קרֿיאת HTTP POST ל-URL הנדרש עם ה-payload בגוף הקריאה.

ה-Retry Manager

כפי שצויין בשלב הדרישות, שליחת Webhooks יכולה להיכשל, ולכן אנו זקוקים למנגנון לביצוע ניסיונות חוזרים (עם exponential backoff ועד ל-24 שעות). למרות מאמצים למצוא פתרונות קיימים שיעמדו בדרישות שלנו, לא הצלחנו למצוא פתרון שכזה ועל כן החלטנו לממש כזה בעצמנו, תוך שימוש ביכולת השהיית הודעות (delayed messages) של SQS וב-S3 לשמירת ה-Payload של הודעות. במידה והנכם מעוניינים בפירוט נוסף כיצד נבנה הפתרון, אנא המתינו למאמר הבא שלי בנושא.

איך הכל מתחבר

דרך פעולת המערכת

באמצעות ה-API, נגדיר Webhook חדש (כלומר target, filter & subscription) שישמר ב- subscription data DB. שלב זה יבוצע פעם אחת ומרגע זה ואילך אירועים רלוונטים ישלחו כהודעות Webhooks.
ברגע שאירוע כלשהו יתרחש באפליקציה/מערכת המשתמשת במערכת לשליחת Webhooks שבנינו (מסומן כ-App1 בדיאגרמה 2), על האפליקציה לשלוח הודעת Kafka ל-Topic יעודי המכילה את תוכן האירוע (ה-Payload) ואת סוג האירוע שקרה. להלן סכימה של הודעה כזו:

				
					
{
  "type": "record",
  "name": "WebhookEvent",
  "namespace": "com.mycompany.webhookdelivery.messages",
  "fields": [
    {
      "name": "event_type",
      "type": "string"
    },
    {
      "name": "payload",
      "type": "string"
    }
  ]
}

ה-Consumer מושך את ההודעה מ- Kafka Topic יעודי ומשתמש בהגדרות שבוצעו קודם לכן ב-API אשר שמורות בזיכרון המטמון המקומי. הוא מחליט האם קיים Subscription שמשתמש ב-Filter המוגדר בו אירוע מאותו הסוג שקרה. במידה וכן, הוא אוסף את כל ה-subscriptions שנמצאו מתאימים ומייצר לכל URL יעד בתוכם Webhook Delivery, ואלו בתורם נשלחים ל-AWS SQS queue.
ה-Dispatcher מושך את ה-Webhook Deliveries מה-AWS SQS queue ושולח אותם ל-URL היעד. אם ה-Dispatcher מקבל בתור תוכן ההודעה לינק ל-S3, הוא מביא את תוכנו ומחליף אותו בתור ה-Payload של ההודעה ל-URL היעד.
במידה והשליחה נכשלה, ה-Delivery מועבר ל-Retry Manager אשר אחראי ״לדחוף״ את ההודעה שוב אל ה-Dispatcher, שוב ושוב עד אשר השליחה תצליח או עד אשר מספר הניסיונות יעבור את זה המוגדר במערכת.

מחשבות לסיום

המאמר הנוכחי רק מזכיר נושאים כגון איך פתרנו בעיות העולם האמיתי שקורות (למשל מנגנון ה-Retry) או שיפורי ביצועים (זיכרון מטמון מקומי ופעולות I/O).

בנוסף, נושאים מעניינים נוספים שיש לתת עליהם את הדעת בעת תכנון מערכת שכזו הם השימוש בה ואינטגרציה שלה עם מערכת multi-tenant SaaS. ישנם נושאים שיש לטפל בהם, כגון הרעבה (Starvation) ו-performance interference, סקיוריטי (CRC, Signature Header Validation), הפרדה בין Data ועוד.

לכל הנושאים הללו אני מזמין אתכם לקרוא את המאמר הבא שלי – פתרון בעיות במערכת לשליחת Webhooks במערכת multi-tenant SaaS.

CodeTime