چطور در آروان‌کلاد با ساختمان‌داده‌ و الگوریتم‌های بهینه مشکل حجم زیاد اطلاعات را حل کردیم

در ۸ بهمن‌ماه ۱۴۰۳، آروان‌کلاد اختلالی را در محصول متریک‌اکسپورتر تجربه کرد که بر اکثر مشتریان تأثیر گذاشت. اشکال از این قرار بود که کاربران امکان دریافت متریک‌های دامنه‌ی خود را نداشتند و با HTTP Timeout مواجه میشدند.

در این پست توضیح میدهیم چه اتفاقی افتاد و برای جلوگیری از این واقعه چه اقداماتی انجام دادیم. همینطور امیدوارم اطلاعات ارایه‌ شده در این پست برای سایر مهندسین (حتی اگر از این سرویس استفاده نمیکنند) مفید واقع شود.

پیش‌زمینه

شبکه آروان‌کلاد یک سیستم توزیع شده در سطح جهانی است و خدمات مختلفی را ارایه میدهد. هر قسمت از هر بخش از این سیستم Event Log هایی تولید میکند که جزییات اتفاقاتی که در هر بخش از سیستم رخ داده را شامل میشود، برای نمونه لاگی که به ازای هر ریکوئست ایجاد میشود.

سی‌دی‌ان آروان‌کلاد روزانه نزدیک به یک میلیون لاگ ایجاد، ارسال و پردازش و متریک‌های دامنه را تولید میکند، که به همین دلیل چالش‌های متنوعی نیز در این بین گریبانگیر توسعه‌دهندگان خواهد بود.

معماری

سیستم لاگ CDN (متریک‌ اکسپورتر) متشکل از یک پایپ‌لاین با ۴ استیج بوده که هر جز نقش ویژه‌ای دارند. استیج اول (A) لاگ‌های تولید‌شده توسط پروکسی CDN را دریافت میکند و روی کافکا (به عنوان مسیج بروکر) میریزد. استیج دوم (B) لاگ‌ها را از کافکا خوانده، بر اساس الگو‌هایی که به آن داده شده تغییراتی روی آنها بوجود می‌آورد تا سایر بخش‌ها بتوانند از لاگ‌ها بهره‌برداری کنند.

استیج سوم که به آن LogForwarder هم میگوییم وظیفه‌ی اساسی ارسال لاگ را برعهده دارد و چهارمین استیج با نام Metric Exporter لاگ‌هایی که استیج سوم آماده‌سازی کرده (در کافکا) را برداشته، تجمیع و پروسس‌های مورد نظر را روی آن انجام داده و کاربر در نهایت با صدا زدن اندپویت HTTP متریک‌هایش را دریافت میکند.

شایان ذکر است که کامپوننت Metric Exporter هیچ External Storage ی ندارد و اطلاعات را در مموری نگه میدارد تا بیشترین پرفورمنس برای آپدیت، حذف، اضافه و تحویل متریک‌ها به مشتریان وجود داشته باشد. همینطور تنها کانسیومر Stage B لاگ‌فرورادر نیست و کانسیومرهای دیگری بر حسب نیاز مسیج‌های این تاپیک را میخوانند. به همین ترتیب LogForwarder علاوه بر ارسال لاگ به مقاصد مختلف مانند SysLog Server، انواع S3 ها و ... ، یکی از مقاصدش کافکا و تاپیک مورد نظر Metric Exporter است.

چه اتفاقی افتاد

در تاریخ ۷ بهمن نسخه جدیدی منتشر کردیم که امکان اسنپ‌شات گرفتن از آخرین وضعیت متریک‌ها را ایجاد کند. در تاریخ ۸ بهمن متوجه شدیم تعداد زیادی از درخواست‌های کاربران Timeout شده و سیستم دچار اختلال گردیده است. اختلال در واقع افزایش مموری (RAM) تا حدی بود که اپ توان پردازش را از دست میداد. طبیعتا اولین حدسمان این بود که مشکل از آخرین فیچری است که به برنامه اضافه کردیم، اما بعد از کالبدشکافی‌های مفصل در محیط استیجینگ متوجه شدیم قابلیت Persistency که اضافه شده بود در واقع کمک میکرد تا مشکل اصلی زودتر از موعد خودش را نشان دهد.

ریشه‌یابی

به کمک محیط استیجینگ و شبیه‌سازی حالت رخ داده، مشکل اصلی را در استفاده نادرست از ساختمان‌داده‌ و الگوریتم‌های غیر بهینه برای نگه‌داری مقادیر مختلف، تشخیص دادیم. ساختار به این شکل بود که متریک‌ها تا بینهایت (تعداد بسیار بسیار زیاد) میتوانستند تولید شوند و این امر خارج از کنترل ما بود. همین موضوع باعث میشد رفتار سیستم پیش‌بینی‌پذیر نباشد و اختلالات مشابه رخ بدهد.

چه کردیم

مهترین قدم تغییر ساختمان‌های داده و الگوریتم‌های مربوط به آنها به انواع بهینه‌تر است به طوری که رفتار برنامه‌ را پیش‌بینی‌پذیر کنیم. برای نمونه استفاده از الگوریتم SpaceSaving که یک روش کارآمد برای تخمین فرکانس پرتکرارترین اقلام در یک جریان داده (Stream) است. این الگوریتم از یک تعداد محدود شمارنده (Counters) استفاده می‌کند و مقدار تقریبی تکرار اقلام را نگه می‌دارد.

نمونه‌ کدی که برای پیاده‌سازی الگوریتم SpaceSaving نوشتیم:

type Entry struct {
   Value string // The value of the element
   Count int    // The count of the element
   index int    // The index of the element in the heap
}

// PriorityQueue implements heap.Interface and holds Entries
type PriorityQueue []*Entry

func (pq PriorityQueue) Len() int { return len(pq) }

func (pq PriorityQueue) Less(i, j int) bool {
   // we want Pop to give us the lowest count, so less means more here.
   return pq[i].Count < pq[j].Count
}

func (pq PriorityQueue) Swap(i, j int) {
   pq[i], pq[j] = pq[j], pq[i]
   pq[i].index = i
   pq[j].index = j
}

func (pq *PriorityQueue) Push(x interface{}) {
   n := len(*pq)
   entry := x.(*Entry)
   entry.index = n
   *pq = append(*pq, entry)
}

func (pq *PriorityQueue) Pop() interface{} {
   old := *pq
   n := len(old)
   entry := old[n-1]
   entry.index = -1 // for safety
   *pq = old[0 : n-1]
   return entry
}

// SpaceSaving tracks top-k elements
type SpaceSaving struct {
   k        int
   elements map[string]*Entry
   pq       PriorityQueue
}

func newSpaceSaving(k int) *SpaceSaving {
   return &SpaceSaving{
      k: k,
      elements: make(map[string]*Entry),
      pq: make(PriorityQueue, 0, k),
   }
}

func (ss *SpaceSaving) Add(value string) {
   if entry, exists := ss.elements[value]; exists {
      entry.Count++
      heap.Fix(&ss.pq, entry.index)
   } else {
      if ss.pq.Len() < ss.k {
         entry := &Entry{
            Value: value,
            Count: 1,
         }
         heap.Push(&ss.pq, entry)
         ss.elements[value] = entry
      } else {
         min := heap.Pop(&ss.pq).(*Entry)
         delete(ss.elements, min.Value)
         min.Value = value
         min.Count++
         heap.Push(&ss.pq, min)
         ss.elements[value] = min
      }
   }
}

func (ss *SpaceSaving) TopK() []*Entry {
   entries := make([]*Entry, len(ss.pq))
   copy(entries, ss.pq)

   // sort the slice based on count
   sort.Slice(entries, func(i, j int) bool {
      return entries[i].Count > entries[j].Count
   })

   return entries
}

از سایر اقدامات میتوان به بهبود سیستم مانیتورینگ اشاره کرد. همینطور بهبود اساسی محیط استیجینگ و تست‌های با لود بالا در این محیط برای اطمینان از عملکرد صحیح برنامه.

روبه‌جلو

در صورتی که SpaceSaving به طور کامل جواب کار مارا ندهد روش‌های دیگری هم برای بهینه سازی داریم از جمله Fixed-size Time Buckets و Circular Buffer. همینطور افزایش متریک‌های داخلی اپ به منظور مانیتور بهتر آن و الرت‌های دقیق‌ برای همین موضوع باید در دستور کار قرار بگیرد.