Пишем балансировщик нагрузки на Golang

Введение

Хотите сделать собственный балансировщик нагрузки, но не знаете как? У моего проекта больше 20 000 пользователей, и с пиковым трафиком бэкенд, развернутый на бесплатном Render, не справлялся. Интересно, что с сезонными наплывами делают в крупных компаниях вроде Flipkart и Amazon?

И вот я написал на Golang простой балансировщик нагрузки с мьютексом, семафором и другими штуками с мудреными названиями, который легко реализуется в проектах:

Реализуем логику балансировщика нагрузки с перенаправлением запроса на соответствующие бэкенды, тремя повторами при сбое запроса и обозначением бэкенда «мертвым».

Сначала простым и эффективным алгоритмом циклического перебора распределим нагрузку по бэкендам. А запрос на соответствующие сервера направим через встроенный в Golang обратный прокси-сервер.

Начинаем с создания структуры ServerNode для представления каждого сервера и его статуса:

type ServerNode struct {
 URL          *url.URL
 Alive        bool
 Mux          sync.RWMutex
 ReverseProxy *httputil.ReverseProxy
}

URL  —  это поле с URL-адресом бэкенд-сервера.

Alive  —  здесь хранится булево значение, это статус бэкенда: «мертв» он или нет.

Mux  —  это мьютекс чтения и записи из пакета sync. Им контролируется доступ к полям ServerNode, предотвращаются состояния гонок, поддерживается согласованность данных.

ReverseProxy  —  клиентские запросы перенаправляются им на этот конкретный узел сервера.

Создаем структуру ServerPool:

type ServerPool struct {
 Backends []*ServerNode
 Current  uint64
}

Она состоит из массива ServerNode, который разберем позже.

Backends  —  это массив ServerNode.

Current  —  здесь сохраняется индекс сервера, на который отправлен последний запрос.

for _, backend := range backendservers {
  log.Println("Load balancing to the backend server: ", backend)
  be, err := url.Parse(backend)
  log.Println(be)
  if err != nil {
   log.Println("Error parsing URL")
  }
  proxy := httputil.NewSingleHostReverseProxy(be)
  serverPool.Backends = append(serverPool.Backends, &lib.ServerNode{
   URL:          be,
   Alive:        true,
   ReverseProxy: proxy,
  })
}

Этой функцией в пул добавляются бэкенд-серверы, для каждого из которых настраиваются обратные прокси-сервера.

Разберем подробнее. Каждый сервер перебирается, с помощью url.Prase парсятся конечные точки.

Благодаря httputil.NewSingleHostReverseProxy создается обратный прокси-сервер, которым клиентские запросы перенаправляются на бэкенд, а ответ  —  обратно клиенту.

Заданием в Alive статуса true сервера добавляются в ServerPool.

func (s *ServerPool) NextServerIndex() int {
 nxtIndex := atomic.AddUint64(&s.Current, uint64(1)) % uint64(len(s.Backends))
 if s.Current >= math.MaxUint64-1 {
  atomic.StoreUint64(&s.Current, 0)
 }
 return int(nxtIndex)
}

Функцией NextServerIndex определяется следующий сервер, куда балансировщиком нагрузки направляется запрос. Потокобезопасность при работе в конкурентной среде обеспечивается атомарными операциями, функцией поддерживается согласованность данных.

Индекс следующего сервера получается с помощью простой логики  —  операцией по модулю.

nextIndex = current % totalLength

Функцией возвращается вычисленный nxtIndex, то есть индекс следующего бэкенд-сервера, на который балансировщиком нагрузки направляется запрос.

func lb(w http.ResponseWriter, r *http.Request) {
 peer := serverPool.GetNextPeer()
 attempts := GetAttemptsFromContext(r)
 if attempts > 3 {
  http.Error(w, "Service not available, max attempts reached", http.StatusServiceUnavailable)
  return
 }
 if peer != nil {
  peer.ReverseProxy.ServeHTTP(w, r)
  return
 }
 http.Error(w, "Service not available", http.StatusServiceUnavailable)
}

Из функции NextServerIndex этой функцией балансировки нагрузки получается следующий одноранговый узел. Из контекста получаются попытки, но об этом позже. Если попыток для бэкенда больше трех, возвращается ошибка.

В противном случае из структуры бэкенда peer вызывается функция reverseproxy.

func (s *ServerPool) MarkDownTheServer(backendUrl *url.URL, serverStatus bool) {
 for _, backend := range s.Backends {
  if backend.URL.String() == backendUrl.String() {
   backend.SetAlive(serverStatus)
   return
  }
 }
}

Заданием в Alive статуса false серверы обозначаются как «мертвые». Представьте, что вы отправились за покупками, узнали, что в магазине всего 10 ручек, и сказали об этом другу. Потом кто-то купил одну ручку, и, когда друг пришел в магазин, ручек осталось девять. Теперь он рассердится на вас, но ведь ошибки не было.

Это называется состоянием гонки или несогласованностью данных и предотвращается мьютексом:

func (b *ServerNode) IsAlive() bool {
 b.Mux.RLock()
 defer b.Mux.RUnlock()
 return b.Alive
}

func (b *ServerNode) SetAlive(alive bool) {
 b.Mux.Lock()
 defer b.Mux.Unlock()
 b.Alive = alive
}

Этими функциями определяется, «жив» бэкенд или нет, и задается его статус.

Если нужно что-то считать или записать, мы блокируем мьютекс. После использования разблокируем. Когда ответ бэкендом не выдается, повторяем.

Вот логика кода:

proxy.ErrorHandler = func(w http.ResponseWriter, r *http.Request, e error) {
   log.Printf("[%s] Request Canceled: %v\n", be.Host, r.Context().Err() == context.Canceled)
   log.Printf("[%s] %s\n", be.Host, e.Error())

   retries := GetRetryFromContext(r) // по умолчанию счетчик повторов на нуле
   log.Println("This is the retry count", retries, "of the server", serverPool.Current)

   if retries < 3 {
    time.Sleep(10 * time.Millisecond)
    ctx := context.WithValue(r.Context(), Retry, retries+1) // увеличиваем счетчик повторов и задаем его в контексте
    log.Println("check")

    proxy.ServeHTTP(w, r.WithContext(ctx))
    return
   }

   // если счетчик повторов больше трех, обозначаем сервер как «упавший»
   log.Printf("[%s] Marking server as down\n", be.Host)
   serverPool.MarkDownTheServer(be, false)

   lb(w, r) // этой функцией находится следующий «живой» сервер и перенаправляется запрос 
  }

Обработчиком регулируются ситуации со сбоями запроса к серверу бэкенда, с помощью повторов им обеспечивается надежность:

• При этом из запроса извлекается текущий контекст. В Go контекст  —  это способ переноса через границы API сроков, сигналов отмены, других связанных с запросом значений.
• Запросные данные обычно передаются контекстами между вызовами функций, особенно в HTTP-серверах  —  так контролируются тайм-ауты, распространяются метаданные.
• Этой функцией создается новый контекст, которым переносится конкретное значение. В нашем случае им добавляется ключ Retry в контекст со значением retries+1.
• Функцией contextWithValue возвращается новый контекст с соответствующей парой «ключ-значение», исходный контекст не изменяется. Здесь счетчик повторов увеличивается на единицу и сохраняется в контексте.

r.Context().Err() == context.Canceled: проверяется, не отменен ли запрос. Хостом бэкенда фиксируется, не случилась ли отмена.

GetRetryFromContext(r): из контекста запроса извлекается количество повторов. Счетчиком повторов отслеживается число повторений запроса.

if retries < 3 {
    time.Sleep(10 * time.Millisecond)
    ctx := context.WithValue(r.Context(), Retry, retries+1)
    log.Println("check")

    proxy.ServeHTTP(w, r.WithContext(ctx))
    return
}

Если счетчик повторов меньше трех, функцией выполняется повтор.

После трех повторов сервер обозначается как «упавший»:

log.Printf("[%s] Marking server as down\n", be.Host)
serverPool.MarkDownTheServer(be, false)

Подробный код  —  в репозитории.

Читайте также:

• Как реализовать в Golang двухфакторную аутентификацию с TOTP
• Go — единственный выбор для бэкенд-разработчика?
• Самый быстрый способ cоздать CRUD API в Golang

Читайте нас в Telegram, VK и Дзен

---

Перевод статьи Murugaperumal R: Load balancer — Golang — Round Robin