Security Insights سرویس ارائه‌دهنده توصیه‌های عملیاتی امنیتی برای هر حساب Cloudflare است. برای پیدا کردن این insights، ما به‌صورت منظم همه حساب‌ها، zones و رکوردهای DNS را اسکن می‌کنیم تا خطرات و misconfigurationهای احتمالی را کشف کنیم. اما دو مشکل کلیدی پیدا شد: اول اینکه اسکن‌ها خیلی کم‌فواصل بودند — معمولاً هفتگی یا دو هفته‌ای — و در نتیجه خطرهای جدید ممکن بود تا دو هفته ناشناخته بمانند. دوم اینکه اسکن خودکار برای بسیاری از حساب‌های پلن رایگان opt-in بود، یعنی تعداد زیادی از حساب‌ها اصلاً اسکن نمی‌شدند ⚠️

با توجه به افزایش حملات خودکار، پنجره زمانی کشف misconfigurationها دارد کوچک‌تر می‌شود؛ بنابراین یافتن این مشکلات برای همه مشتریان‌مان حیاتی است تا به هدف‌مان در ساختن یک اینترنت بهتر برای همه نزدیک شویم. ما محاسبه کردیم که برای افزایش فرکانس اسکن و فعال‌سازی اسکن خودکار برای همه حساب‌ها باید throughput اسکن را به‌طور میانگین حدود 10x افزایش دهیم — از 10 اسکن در ثانیه به حدود 100 اسکن در ثانیه. اما سیستم ما قبلاً تحت فشار بود: میلیون‌ها event در backlog منتظر پردازش بودند، API ما مرتباً timeout می‌داد و فرآیندها crash می‌کردند. باید سیستم را اصلاح و مقیاس‌پذیر می‌کردیم 🚀 این داستان نحوه افزایش بیش از 10x در throughput اسکن برای Security Insights، فعال‌سازی insights برای میلیون‌ها مشتری و دو برابر کردن فرکانس اسکن برای همه مشتریان است. ادامه دهید تا ببینید چگونه این بهبودها را انجام دادیم.

How we scan for security insights — در سطح بالا، اسکن‌های خودکار امنیتی ما توسط یک scheduler راه‌اندازی می‌شوند. وقتی یک account یا zone وقت اسکن شد، scheduler یک یا چند پیام به Apache Kafka، پلتفرم open-source برای event streaming توزیع‌شده، منتشر می‌کند. این پیام‌ها به تعداد زیادی checker می‌رسند: میکروسرویس‌های تخصصی نوشته‌شده به زبان Go که دارایی‌ها یا پیکربندی‌های خاص را اسکن می‌کنند.

برای هر پیام، هر checker نتایج خودش (یعنی security insights پیدا شده) را به API داخلی ما می‌فرستد و آن API این نتایج را در یک پایگاه‌داده Postgres پایدار می‌کند.

Making it scale — Scaling Kafka: Apache Kafka الزاماً یک queue خالص نیست؛ بلکه یک event stream partitioned است (اگرچه اخیراً semanticsِ queue هم در آن اضافه شده). درون هر partition پیام‌ها باید به ترتیب مصرف و پردازش شوند؛ این با صف‌های متداول فرق دارد که ممکن است مصرف به ترتیب باشد اما پردازش می‌تواند out-of-order باشد. نتیجه اینکه در یک consumer group فقط یک consumer فعال می‌تواند برای هر partition وجود داشته باشد.

این دو پیامد برای ما داشت: پیام‌هایی که پردازششان کند است، consumer را از حرکت به پیام بعدی باز می‌دارند؛ و برای هر checker، تعداد consumerها محدود به تعداد partitionها است (هر checker consumer group خودش را دارد). می‌توانستیم با افزایش تعداد partitionها مقیاس‌پذیری را بالا ببریم، اما این کار بار بیشتری روی خودِ Kafka broker می‌گذاشت که بین سرویس‌های زیادی مشترک است؛ بنابراین این راه‌حل را به‌عنوان آخرین گزینه نگه داشتیم و ابتدا به بهبود کد و معماری تمرکز کردیم.

Introducing parallel processing — اگرچه مصرف پیام‌ها باید ordered باشد، هیچ مانعی برای مصرف چند پیام به‌صورت همزمان وجود ندارد. ما checkers را تغییر دادیم تا پیام‌ها را در batch مصرف کنند و هر پیام را در یک goroutine جدا پردازش کنند. معامله‌گری‌ها این بود که اگر پروسس وسط یک batch کرش کند، کار بیشتری باید دوباره انجام شود و میزان استفاده حافظه کمی بیشتر می‌شود؛ اما در عمل این هزینه‌ها قابل قبول بودند و throughput به‌طرز قابل‌ملاحظه‌ای افزایش یافت.

Avoiding head-of-line blocking — بعضی پیام‌ها برای چند تا از checkers ما عملاً بسیار زمان‌بر بودند؛ مثلاً یک account/zone ممکن است دارایی‌های خیلی بیشتری نسبت به دیگری داشته باشد. در بدترین حالت، پردازش چنین پیام‌هایی می‌تواند دقیقه‌ها یا حتی ساعت‌ها طول بکشد در حالی که میانگین پردازش پیام معمولاً ثانیه‌ها یا میلی‌ثانیه‌هاست. راه‌حل ساده‌ای انتخاب کردیم: تقسیم consumer groupها و checkers به دو مسیر — «slow lane» و «fast lane». با معیارهای ساده‌ای می‌توانستیم سریع تشخیص دهیم که آیا یک پیام slow‌ است یا fast؛ اگر fast-lane با یک پیام slow مواجه می‌شد، از آن عبور می‌کرد. این کار مشکل را حل کرد: پیام‌های کند منابع و زمان اختصاصی پیدا کردند و پیام‌های سریع با سرعت معمول جلو رفتند ✅

Optimizing our database queries — هر insight که پیدا می‌کنیم باید در Postgres نوشته شود. این کار توسط یک endpoint واحد API انجام می‌شود که checkers با لیستی از insights آن را فراخوانی می‌کنند. پیاده‌سازی اولیه شبیه این بود:

for _, issue := range issues {
    _, err = tx.Exec(ctx, `INSERT INTO table ... VALUES ($1, $2, ...) ON CONFLICT DO UPDATE ...`, ...)
    if err != nil {
        return err
    }
}

خواننده دقیق متوجه می‌شود که برای مجموعه‌های بزرگ از insights، این کد برای هر insight یک round trip به دیتابیس انجام می‌دهد. با حداکثر اندازه مشاهده‌شده 500,000، این معادل نیم میلیون round trip، query و transaction در یک فراخوانی API بود — کاملاً غیرقابل‌قبول. ابتدا استاندارد طلایی bulk insert در Postgres یعنی COPY به یک temporary table را امتحان کردیم، اما این باعث bloat در system tables شد. در نهایت روی یک روش هیبرید توافق کردیم: از UNNEST وقتی تعداد issues زیر یک آستانه بود و از COPY وقتی تعداد از آن آستانه فراتر می‌رفت. این روش ترکیبی بهترین نتیجه را داد: درج‌های بسیار بزرگ در ثانیه انجام می‌شد و درج‌های کوچک در میلی‌ثانیه.

Investigating our API timeouts — در مسیر مقیاس‌بندی، چند رفتار عجیب در API داخلی‌مان دیدیم: تعداد زیادی از درخواست‌ها با client-side timeout مواجه می‌شدند؛ بسیاری از checkers بین 20 تا 90 درصد زمان پردازش خود را صرف یک فراخوانی API می‌کردند؛ و وقتی حجم زیادی از اسکن‌ها را trigger می‌کردیم، throughput در ابتدا بالا بود اما به‌تدریج افت می‌کرد. همۀ این‌ها یک ریشه مشترک داشتند: latency. پایگاه‌داده اصلی ما در Portland، Oregon بود، در حالی که API ما active-active در Portland و Amsterdam اجرا می‌شد. حتی با سرعت نور، round-trip latency بین Portland و Amsterdam حدود 50 ms است. در نتیجه، queryهای ارسال‌شده از instanceهای Amsterdam زمان بیشتری می‌گرفتند و کانکشن‌ها در connection pool کلاینت برای مدت طولانی‌تری اشغال می‌شدند. با حجم بالای درخواست‌ها، pool به سرعت تخلیه می‌شد و timeoutها رخ می‌دادند. میانگین زمان یک فراخوانی API در Portland حدود 10 ms و در Amsterdam نزدیک به 3 ثانیه بود!

اما چرا throughput پیام‌ها افت می‌کرد؟ هر پروسس checker مجموعه‌ای از partitionها را برای مصرف دریافت می‌کند. API ما لودبالانس شده بود و چون ما کانکشن را در طول عمر پروسس باز نگه می‌داشتیم، بعضی پروسس‌ها کانکشن‌شان به Amsterdam و بقیه به Portland متصل می‌شدند. partitionهای متصل به Portland سریع‌تر پردازش می‌شدند و آن‌هایی که به Amsterdam وابسته بودند عقب ماندند — نتیجه افزایش Kafka lag برای بعضی partitionها. راه‌حل ساده بود: API را از حالت active-active به active-passive تغییر دادیم و اطمینان حاصل کردیم که endpoint فعال همواره دنبال primary database برود. مشکل latency تقریباً یک‌شبه از بین رفت ✨

Rethinking the scheduler — Kafka را مقیاس کردیم، queryها را بهینه کردیم و API را اصلاح کردیم، اما هنوز مشکل دیگری وجود داشت: باید مطمئن می‌شدیم اسکن‌ها به‌صورت تقریباً یکنواخت در زمان پخش شوند. قرار دادن همه اسکن‌ها در صف در یک زمان ممکن نبود چون topicِ Kafka ما retention مبتنی بر زمان دارد؛ در نتیجه اسکن‌ها در Kafka تجمع می‌کردند و قبل از پردازش حذف می‌شدند. scheduler ما هم یکنواختی خوبی نداشت و تعداد اسکن‌های trigger شده در نقاطی از هفته به‌شدت spike می‌زد — گاهی صدها هزار اسکن در عرض چند دقیقه. علت این بود که scheduler اسکن‌ها را در بازه‌های ثابت و تکرارشونده trigger می‌کرد. در pseudocode به‌صورت خلاصه شبیه این بود:

Loop forever:
    Find accounts where last_scheduled_at + scanning_frequency