북극 펭귄

1. 서론

 

이번 포스팅은 그동안 envoy-internals 시리즈의 이해를 바탕으로 사용자가 Http 전달을 요청했을 때, Envoy 내부구조를 토대로 네트워크 요청이 어떻게 흘러가는지에 대해서 전체적인 흐름을 상세히 조망해보는 시간을 가져보려 합니다. 사실 envoy에 대해서 분석했던 계기 중 하나가 도대체 어떻게 사용자의 요청이 전달되는지에 대한 궁금증에서 출발했기 때문에 이번 포스팅을 위해서 이전 시리즈의 내용이 존재했다고 생각합니다. 따라서 이번 내용은 이전 내용에 대한 이해가 선행되어야하므로 이전 시리즈 내용을 정독하고 보시는 것을 추천드립니다. 

 

---

2.  Worker 쓰레드 소켓 할당 과정

 

이전 시리즈 내용을 통해 Listener Manager에서 네트워크 요청 처리를 위해서 여러개의 Worker 쓰레드를 생성하는 것을 이해할 수 있었습니다. 이때 Worker 쓰레드 생성 갯수는 envoy 생성 당시 --concurrency 인자에 의해서 결정되는 것 또한 확인했습니다. 이번에는 Envoy 기동 과정 중 Worker 쓰레드 상호작용을 살펴보면서 Worker 쓰레드에서 어떻게 네트워크 요청을 처리하는지 살펴보겠습니다.

 

 

위 그림은 --concurrency 값이 2개이고 static_config, dynamic_config에 의해서 등록된 Listener 개수 또한 2개이면서 모두 TCP임을 가정했습니다. 이때 기동 과정을 살펴보면 다음과 같습니다.

 

1. Envoy 기동 과정에서 --concurrency 값을 살펴보고 Listener Manager에게 Worker 쓰레드 생성을 요청합니다. Listener Manager는 Worker 쓰레드를 요청만큼 생성합니다. 이 과정에서 Worker 쓰레드 내부에 Dispatcher와 Connection Handler가 생성됩니다.

 

2. Worker 쓰레드가 생성되는 과정에서 Envoy의 TLS를 관장하는 메인 쓰레드 InstanceImpl에 쓰레드를 등록합니다. 이 과정에서 InstanceImpl에서 Worker 쓰레드에 위치한 Dispatcher 정보를 registered_threads_ 에 저장할 수 있으며, 향후 Worker 쓰레드의 Dispatcher에 참조가 가능합니다.

 

3. Worker 쓰레드 생성이 마무리되면, Config 파일 파싱 도중 static configuration 정보를 등록하기 위해 Listener Manager에게 Config 정보 등록을 요청합니다. 해당 과정은 향후 LDS에 의해서도 생성될 수 있습니다. 이 과정에서 Listener Component Factory를 통해 --concurrency 만큼 Socket을 생성합니다.

 

4. Envoy의 설정이 모두 완료되면, Listener Manager에게 기동을 요청합니다.

 

5. Listener Manager에서는 기동 과정에서 등록된 Listener Config 정보를 Worker 쓰레드에 모두 Bind하기 위해 개별 Worker 쓰레드에게 Listener 생성을 요청합니다.

 

6. Worker 쓰레드내에 존재하는 Connection Handler에 Listener를 생성하기 위해 먼저 Listener로부터 자신의 Worker 쓰레드 번호에 해당하는 Socket 정보를 얻어옵니다. 그리고 Worker 쓰레드에서 외부 요청을 참조하기 위해 Dispatcher에게 Socket 정보를 전달하면서 Listener 생성을 요청합니다.

 

connection_handler_impl.cc

details->addActiveListener(
    config, address, listener_reject_fraction_, disable_listeners_,
    std::make_unique<ActiveTcpListener>(
        *this, config, runtime,
        socket_factory->getListenSocket(worker_index_.has_value() ? *worker_index_ : 0),
        address, config.connectionBalancer(*address)));

 

active_tcp_listener.cc

ActiveTcpListener::ActiveTcpListener(Network::TcpConnectionHandler& parent,
                                     Network::ListenerConfig& config, Runtime::Loader& runtime,
                                     Network::SocketSharedPtr&& socket,
                                     Network::Address::InstanceConstSharedPtr& listen_address,
                                     Network::ConnectionBalancer& connection_balancer)
    : OwnedActiveStreamListenerBase(
          parent, parent.dispatcher(),
          parent.dispatcher().createListener(std::move(socket), *this, runtime, config.bindToPort(),
                                             config.ignoreGlobalConnLimit()),
          config),
      tcp_conn_handler_(parent), connection_balancer_(connection_balancer),
      listen_address_(listen_address) {
  connection_balancer_.registerHandler(*this);
}

 

 

7. Dispatcher는 전달받은 Socket 정보를 토대로 Listener를 생성하여 반환합니다. 이때 주의해서 살펴볼 것은 사용자가 접속했을 때, 인자로 전달받은 TcpListenerCallbacks에게 Accept 요청을 수행하는데 호출되는 주체가 Connection Handler에서 전달한 ActiveTcpListener라는 것입니다. 따라서 향후 사용자가 접속하게되면, 그에 대한 Accept 처리는 ActiveTcpListener가 담당하게됩니다.

 

dispatcher_impl.cc

Network::ListenerPtr DispatcherImpl::createListener(Network::SocketSharedPtr&& socket,
                                                    Network::TcpListenerCallbacks& cb,
                                                    Runtime::Loader& runtime, bool bind_to_port,
                                                    bool ignore_global_conn_limit) {
  return std::make_unique<Network::TcpListenerImpl>(*this, random_generator_, runtime,
                                                    std::move(socket), cb, bind_to_port,
                                                    ignore_global_conn_limit);
}

 

8. 생성된 Listener에서는 소켓 정보를 libevent에 등록하여 향후 Client가 접속을 요청했을 때 libevent에 의해 요청을 전달받을 수 있도록 등록합니다.

 

tcp_listener_impl.cc

socket_->ioHandle().initializeFileEvent(
    dispatcher, [this](uint32_t events) -> void { onSocketEvent(events); },
    Event::FileTriggerType::Level, Event::FileReadyType::Read);

 

 

위와같이 8단계를 거치게되면, 생성된 모든 Worker 쓰레드에서 Listener 및 소켓을 생성하고 Client 요청을 수신받을 수 있는 상태가 완료됩니다.

 

---

3. Client Connection 연결 과정

 

지금까지 Envoy를 기동하는 과정에서 Worker 쓰레드가 생성되고, Listener 및 Socket을 생성하는 것을 살펴봤습니다.

 

tcp_listener_impl.cc

TcpListenerImpl::TcpListenerImpl(Event::DispatcherImpl& dispatcher, Random::RandomGenerator& random,
                                 Runtime::Loader& runtime, SocketSharedPtr socket,
                                 TcpListenerCallbacks& cb, bool bind_to_port,
                                 bool ignore_global_conn_limit)
    : BaseListenerImpl(dispatcher, std::move(socket)), cb_(cb), random_(random), runtime_(runtime),
      bind_to_port_(bind_to_port), reject_fraction_(0.0),
      ignore_global_conn_limit_(ignore_global_conn_limit) {
  if (bind_to_port) {
    socket_->ioHandle().initializeFileEvent(
        dispatcher, [this](uint32_t events) -> void { onSocketEvent(events); },
        Event::FileTriggerType::Level, Event::FileReadyType::Read);
  }
}

 

이때 dispatcher를 통해서 libevent에서 해당 소켓에 이벤트가 감지되면, onSocketEvent 메소드를 호출하도록 위 코드와 같이 등록됩니다. 즉 이 과정을 통해서 각각의 Worker 쓰레드에 존재하는 Listener는 dispatcher에 의해 libevent로 등록되었으므로 사용자가 OS로부터 Socket 생성을 요청했을 때, OS는 등록된 socket 중 하나를 임의로 선정하여 요청을 전달할 수 있게됩니다.

 

 

 

그렇다면 Listener 설정이 모두 완료된 이후 Client로부터 Connection 요청이 들어오면 어떠한 과정을 거치게될까요?

 

 

먼저 Socket Event가 감지되면, 위와 같이 두개의 Worker 쓰레드 중 누가 해당 요청을 처리해야하는지 선택해야합니다. 이때 Worker 쓰레드 선정에 대한 결정은 이전에 설명했듯이 전적으로 OS가 수행합니다. 따라서 가령 위와같이 Worker_0번이 선택되었으면, 해당 쓰레드의 onSocketEvent가 실행될 것입니다.

 

tcp_listener_impl.cc

void TcpListenerImpl::onSocketEvent(short flags) {
  ASSERT(bind_to_port_);
  ASSERT(flags & (Event::FileReadyType::Read));

  while (1) {
    if (!socket_->ioHandle().isOpen()) {
      PANIC(fmt::format("listener accept failure: {}", errorDetails(errno)));
    }

    sockaddr_storage remote_addr;
    socklen_t remote_addr_len = sizeof(remote_addr);

    IoHandlePtr io_handle =
        socket_->ioHandle().accept(reinterpret_cast<sockaddr*>(&remote_addr), &remote_addr_len);
    if (io_handle == nullptr) {
      break;
    }

    if (rejectCxOverGlobalLimit()) {
      io_handle->close();
      cb_.onReject(TcpListenerCallbacks::RejectCause::GlobalCxLimit);
      continue;
    } else if (random_.bernoulli(reject_fraction_)) {
      io_handle->close();
      cb_.onReject(TcpListenerCallbacks::RejectCause::OverloadAction);
      continue;
    }

    const Address::InstanceConstSharedPtr& local_address =
        local_address_ ? local_address_ : io_handle->localAddress();

    const Address::InstanceConstSharedPtr remote_address =
        (remote_addr.ss_family == AF_UNIX)
            ? io_handle->peerAddress()
            : Address::addressFromSockAddrOrThrow(remote_addr, remote_addr_len,
                                                  local_address->ip()->version() ==
                                                      Address::IpVersion::v6);
    cb_.onAccept(
        std::make_unique<AcceptedSocketImpl>(std::move(io_handle), local_address, remote_address));
  }
}

 

onSocketEvent 메소드가 호출되면, 가장 먼저 수행하는 것은 연결된 Connection 갯수가 Global 설정을 넘어섰는지 확인합니다. 이 과정에서 Global Limit이 지정되어있고 신규 연결 요청이 Limit을 넘어서게되면, 해당 소켓에 대한 연결은 Close하고 종결처리 합니다.

 

 

이때 Global Limit으로 지정될 수 있는 값은 overload.global_downstrea_max_connections에 의해서 지정될 수 있으며, 해당 값은 OverloadManager에 의해서 관리되는 값입니다. 따라서 현재 Socket에 Accepted된 개수가 해당 값을 넘었을 경우에는 Socket 연결을 해제합니다. 해당 값에 대한 자세한 설명은 envoy 공식문서를 참고 바랍니다.

 

반대로 요청이 Global Limit을 넘지 않았을 경우는 AcceptedSocket을 생성하고 Worker 쓰레드 내 Listener는 Socket 연결에 대한 Accept 처리를 위임합니다.

 

 

위 그림은 AcceptedSocket 클래스 구조를 나타냅니다. 위 내용을 통해서 우리는 AcceptedSocket을 만드는 이유에 대해서 유추해볼 수 있습니다. 코드를 살펴보면, global_accetped_socket_count_ 라는 값이 static으로 지정되어있음을 알 수 있습니다. 그리고 생성자, 소멸자 단계에서 해당 값이 증감하는 것 또한 알 수 있습니다.

 

이를 통해서 확인되는 사실은 Socket이 접속하면 현재 Accepted된 Socket 개수를 파악할 수 있습니다. 또한 새로운 Socket이 연결되었을 때 Global Limit을 넘는지 검증할 수 있는 기준을 제시합니다.

 

AcceptedSocket을 만들고나면 그 다음에는 해당 Socket을 Listener에서 Accept하는 과정이 진행됩니다. 

 

tcp_listener_impl.cc

void TcpListenerImpl::onSocketEvent(short flags) {
  ...(중략)...
  cb_.onAccept(
        std::make_unique<AcceptedSocketImpl>(std::move(io_handle), local_address, remote_address));
}

 

이전에 살펴본 onSocketEvent 메소드의 마지막 줄을 살펴보면, 저장된 Callback에서 onAccept를 수행해달라고 요청하는 것을 볼 수 있습니다. 이는 TcpListenerImpl을 생성할 때, 해당 Callback 값이 Worker에 존재하는 ActiveTcpListener 이므로 해당 ActiveTcpListener가 실질적으로 Accept를 수행함을 의미합니다.

 

active_tcp_listener.cc

void ActiveTcpListener::onAccept(Network::ConnectionSocketPtr&& socket) {
  if (listenerConnectionLimitReached()) {
    RELEASE_ASSERT(socket->connectionInfoProvider().remoteAddress() != nullptr, "");
    ENVOY_LOG(trace, "closing connection from {}: listener connection limit reached for {}",
              socket->connectionInfoProvider().remoteAddress()->asString(), config_->name());
    socket->close();
    stats_.downstream_cx_overflow_.inc();
    return;
  }

  onAcceptWorker(std::move(socket), config_->handOffRestoredDestinationConnections(), false);
}

 

ActiveTcpListener에서의 Accept 과정을 살펴보면 위 코드와 같습니다.

 

 

 

 

먼저 listenerConnectionLimitReached 메소드를 수행하면서 이를 위반할 경우 Socket을 Close하는 것을 볼 수 있습니다. 이는 이전의 Connection은 Envoy 전체의 Connection을 살펴본 것이라면, 이번에는 개별 listener 별로 Connection 제한이 있는지 검사하고 만약 지정된 값이 있을 경우 그 값을 넘어서게되면 Accept하지 않습니다. 해당 설정은 Listener에서 수행할 수 있으며 envoy 공식 문서에서 이에 대해서 소개하고 있으니 참고 바랍니다.

 

ActiveTcpListener에서는 Limit 검사만 체크하고 다시 Socket에 대한 Accept는 onAcceptWorker 메소드 호출을 통해 후속 작업을 처리합니다.

 

 

active_tcp_listener.cc

void ActiveTcpListener::onAcceptWorker(Network::ConnectionSocketPtr&& socket,
                                       bool hand_off_restored_destination_connections,
                                       bool rebalanced) {
  if (!rebalanced) {
    Network::BalancedConnectionHandler& target_handler =
        connection_balancer_.pickTargetHandler(*this);
    if (&target_handler != this) {
      target_handler.post(std::move(socket));
      return;
    }
  }

  auto active_socket = std::make_unique<ActiveTcpSocket>(*this, std::move(socket),
                                                         hand_off_restored_destination_connections);

  onSocketAccepted(std::move(active_socket));
}

 

이번에는 onAcceptWorker 메소드 내용을 살펴보겠습니다. 해당 코드를 살펴보면 connection_balancer에 의해서 TargetHandler를 구하는 것을 볼 수 있습니다.

 

그렇다면 connection_balancer는 무엇일까요?

 

static_resources:
  listeners:
  - connection_balance_config:
      extend_balance:
        name: envoy.network.connection_balance.dlb
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.network.connection_balance.dlb.v3alpha.Dlb

 

envoy에서는 쓰레드간의 효율적인 Socket 분배를 위해서 만약 하드웨어에서 DLB 지원이 된다면, 이를 사용하여 Connection을 안정적으로 분배할 수 있는 기능을 제공합니다. 이때 위와 같이 connection_balance_config를 지정하면, 해당 설정을 토대로 connection_balancer가 Target을 지정합니다.

 

해당 기술은 내부적으로는 Intel DLB hardware를 통해 구현되며, 위와 같이 Config 설정이 지정되어있다면 다른 Target으로 Load를 분산시킬 수 있습니다. 만약 지정되지 않았다면, 현재 Worker 쓰레드에서 Accept 과정이 정상적으로 진행될 것입니다. 이와 관련된 자세한 내용은 envoy 공식문서에 설명되어있으니 참고 바랍니다.

 

참고로 본 포스팅에서는 DLB 설정이 지정되어있지 않았다고 가정하므로 현재 Worker 쓰레드에서 해당 처리를 진행한다고 가정하겠습니다.

 

active_tcp_listener.cc

void ActiveTcpListener::onAcceptWorker(Network::ConnectionSocketPtr&& socket,
                                       bool hand_off_restored_destination_connections,
                                       bool rebalanced) {
  ...(중략)...

  auto active_socket = std::make_unique<ActiveTcpSocket>(*this, std::move(socket),
                                                         hand_off_restored_destination_connections);

  onSocketAccepted(std::move(active_socket));
}

 

connection_balancer에 의해서 선정된 target이 자기 자신이라면 해당 연결을 허용하기 위해 ActiveTcpSocket을 만드는 것을 볼 수 있습니다. 

 

ActiveTcpSocket까지 생성되면, 향후 Client의 요청은 해당 Socket을 통해서 모두 처리가됩니다. 

 

 

 

즉 이말은 이전에 Envoy를 처음 학습할 때 살펴봤듯이 Client가 Envoy에게 특정 API를 요청하면, 내부적으로 Listener Filters와 Filter Chains를 통과하면서 Upstream 대상을 찾고 전달한다고 했는데, 이 과정을 수행하는 주체가 해당 소켓이 됩니다. 따라서 Active Tcp Socket은 이를 지원하기 위해서 다양한 내부 프로퍼티가 있는데, 그 중 Filter와 관련된 것이 accept_filters 입니다. 해당 속성을 기반으로 향후 Listener Filters를 만들고 이를 수행하고 그 다음에는 Filter Chains를 생성하고 이를 수행하는 작업을 진행합니다. 그리고 해당 작업을 위해서 ActiveTcpSocket을 만든 이후 onSocketAccepted를 호출합니다.

 

active_stream_listener_base.h

void onSocketAccepted(std::unique_ptr<ActiveTcpSocket> active_socket) {
  // Create and run the filters
  if (config_->filterChainFactory().createListenerFilterChain(*active_socket)) {
    active_socket->startFilterChain();
  } else {
    // If create listener filter chain failed, it means the listener is missing
    // config due to the ECDS. Then close the connection directly.
    active_socket->socket().close();
    ASSERT(active_socket->isEndFilterIteration());
  }

  // Move active_socket to the sockets_ list if filter iteration needs to continue later.
  // Otherwise we let active_socket be destructed when it goes out of scope.
  if (!active_socket->isEndFilterIteration()) {
    active_socket->startTimer();
    LinkedList::moveIntoListBack(std::move(active_socket), sockets_);
  } else {
    if (!active_socket->connected()) {
      // If active_socket is about to be destructed, emit logs if a connection is not created.
      if (active_socket->streamInfo() != nullptr) {
        emitLogs(*config_, *active_socket->streamInfo());
      } else {
        // If the active_socket is not connected, this socket is not promoted to active
        // connection. Thus the stream_info_ is owned by this active socket.
        ENVOY_BUG(active_socket->streamInfo() != nullptr,
                  "the unconnected active socket must have stream info.");
      }
    }
  }
}

 

위 코드는 onSocketAccepted 메소드 내용입니다. 코드를 살펴보면, Socket이 Accepted 되면 가장 먼저 해당 소켓에 해당되는 ListenerFilterChain을 생성하고 FilterChain을 수행하는 것을 볼 수 있습니다. 즉 이 과정부터 해당 소켓은 각종 Filter들을 통과하면서 Upstream 연결이 이어지게됩니다.

 

해당 과정을 조금 더 자세히 살펴보겠습니다.

 

 

 

 

이전 내용을 토대로 ActiveTcpListener 에서 ActiveTcpSocket을 만든 것을 확인했습니다. 그리고 createListenerFactory 메소드를 호출하면서 생성한 ActiveTcpSocket을 전달했습니다.

 

그러면 내부적으로는 Listener Config에 이미 저장되어있는 listener_filter_factories 내부에 매핑된 Factory Callback을 하나씩 실행시키면서 ActiveTcpSocket의 accept_filters에 Filter를 하나씩 생성하는 과정을 거칩니다. 해당 과정을 코드로 살펴보면 다음과 같습니다.

 

 

listener_impl.cc

bool ListenerImpl::createListenerFilterChain(Network::ListenerFilterManager& manager) {
  if (Configuration::FilterChainUtility::buildFilterChain(manager, listener_filter_factories_)) {
    return true;
  } else {
    ENVOY_LOG(debug, "New connection accepted while missing configuration. "
                     "Close socket and stop the iteration onAccept.");
    missing_listener_config_stats_.extension_config_missing_.inc();
    return false;
  }
}

 

위 코드는 ActiveTcpListener 에서 ActiveTcpSocket을 생성 후 ListenerFilterChain을 생성하기 위해 createListenerFilterChain 메소드를 호출하였을 때 과정을 나타냅니다. 코드를 살펴보면, buildFilterChain 함수 호출을 통해 자신이 보유하고 있는 listener_filter_factories_ 목록을 전달하는 것을 볼 수 있습니다.

 

configuration_impl.cc

bool FilterChainUtility::buildFilterChain(Network::ListenerFilterManager& filter_manager,
                                          const Filter::ListenerFilterFactoriesList& factories) {
  for (const auto& filter_config_provider : factories) {
    auto config = filter_config_provider->config();
    if (!config.has_value()) {
      return false;
    }
    auto config_value = config.value();
    config_value(filter_manager);
  }

  return true;
}

 

buildFilterChain 내부를 살펴보면, 전달받은 factories를 순회하면서 callback 함수를 순차적으로 수행시키는 것을 볼 수 있습니다. 그리고 해당 callback을 수행할 때 전달받은 ActiveTcpSocket 정보를 다시 넘기는 것을 확인할 수 있습니다.

 

config.cc

[listener_filter_matcher, config](Network::ListenerFilterManager& filter_manager) -> void {
  filter_manager.addAcceptFilter(listener_filter_matcher, std::make_unique<Filter>(config));
};

 

위 코드는 http instpector Listener Filter의 Factory 코드이며, 개별 Listener Filter Factory의 리턴 값은 위와 같이 FilterManager 즉 ActiveTcpSocket 정보로 받아서 addAcceptFilter 메소드를 호출하고 있는 것을 볼 수 있습니다. 또한 인자를 통해서 Factory에서 보유하고 있는 Filter 정보를 새롭게 생성하여 전달하는 것을 볼 수 있습니다.

 

 

active_tcp_socket.h

void addAcceptFilter(const Network::ListenerFilterMatcherSharedPtr& listener_filter_matcher,
                     Network::ListenerFilterPtr&& filter) override {
  accept_filters_.emplace_back(
      std::make_unique<GenericListenerFilter>(listener_filter_matcher, std::move(filter)));
}

 

그리고 ActiveTcpSocket에서는 전달받은 Filter를 accept_filters에 추가함으로써 Listener Filter Chain을 완성합니다.

 

 

active_stream_listener_base.h

void onSocketAccepted(std::unique_ptr<ActiveTcpSocket> active_socket) {  
  ...(중략)...
    active_socket->startFilterChain();
  ...(후략)...
}

 

FilterChain이 완성되면, startFilterChain()을 통해 ListenerFilterChain을 차례로 수행하도록 요청합니다.

 

active_tcp_socket.h

void startFilterChain() { continueFilterChain(true); }

 

startFilterChain()은 다시 내부에 continueFilterChain 메소드에 처리를 위임합니다.

 

 

active_tcp_socket.cc

void ActiveTcpSocket::continueFilterChain(bool success) {
  if (success) {
    bool no_error = true;
    if (iter_ == accept_filters_.end()) {
      iter_ = accept_filters_.begin();
    } else {
      iter_ = std::next(iter_);
    }

    for (; iter_ != accept_filters_.end(); iter_++) {
      Network::FilterStatus status = (*iter_)->onAccept(*this);
      if (status == Network::FilterStatus::StopIteration) {        
        if (!socket().ioHandle().isOpen()) {          
          no_error = false;
          break;
        } else {
          // If the listener maxReadBytes() is 0, then it shouldn't return
          // `FilterStatus::StopIteration` from `onAccept` to wait for more data.
          ASSERT((*iter_)->maxReadBytes() != 0);
          if (listener_filter_buffer_ == nullptr) {
            if ((*iter_)->maxReadBytes() > 0) {
              createListenerFilterBuffer();
            }
          } else {
            // If the current filter expect more data than previous filters, then
            // increase the filter buffer's capacity.
            if (listener_filter_buffer_->capacity() < (*iter_)->maxReadBytes()) {
              listener_filter_buffer_->resetCapacity((*iter_)->maxReadBytes());
            }
          }
          if (listener_filter_buffer_ != nullptr) {
            listener_filter_buffer_->activateFileEvent(Event::FileReadyType::Read);
          }
          // Waiting for more data.
          return;
        }
      }
    }
    // Successfully ran all the accept filters.
    if (no_error) {
      newConnection();
    } else {
      // Signal the caller that no extra filter chain iteration is needed.
      iter_ = accept_filters_.end();
    }
  }

  // Filter execution concluded, unlink and delete this ActiveTcpSocket if it was linked.
  if (inserted()) {
    unlink();
  }
}

 

continueFilterChain 코드 내용을 자세히 살펴보면, 생성된 Listener Filters(accept_filters)를 순회하면서 onAccept를 통해 Filter로직을 수행하는 것을 볼 수 있습니다.

 

 

만약 Filter onAccept 순회 도중에 Stop을 해야될 경우가 존재한다면, 이유를 살펴보고 요청을 중지하던지 아니면 설정 변경 후 재수행을 수행하도록 작성되어있습니다.

 

그리고 모든 Filter 순회가 종료되면, 사용자 요청에 대한 Metadata 및 요청 정보를 모두 분석하였으므로 그 다음에는 newConnection() 메소드를 호출하여 본격적으로 downstream간의 연결을 위한 작업을 진행합니다.

 

active_tcp_socket.cc

void ActiveTcpSocket::newConnection() {
  connected_ = true;

  // Check if the socket may need to be redirected to another listener.
  Network::BalancedConnectionHandlerOptRef new_listener;

  if (hand_off_restored_destination_connections_ &&
      socket_->connectionInfoProvider().localAddressRestored()) {
    // Find a listener associated with the original destination address.
    new_listener =
        listener_.getBalancedHandlerByAddress(*socket_->connectionInfoProvider().localAddress());
  }

  // Reset the file events which are registered by listener filter.
  // reference https://github.com/envoyproxy/envoy/issues/8925.
  if (listener_filter_buffer_ != nullptr) {
    listener_filter_buffer_->reset();
  }

  if (new_listener.has_value()) {
    ...(중략)...
    new_listener.value().get().onAcceptWorker(std::move(socket_), false, false);
  } else {
    // Set default transport protocol if none of the listener filters did it.
    if (socket_->detectedTransportProtocol().empty()) {
      socket_->setDetectedTransportProtocol("raw_buffer");
    }
    accept_filters_.clear();
    // Create a new connection on this listener.
    listener_.newConnection(std::move(socket_), std::move(stream_info_));
  }
}

 

newConnection 메소드를 호출하면, 다른 Listener로 Redirect 해야할 필요가 있는지를 살펴보고 listener에 새로운 Connection 할당을 요청합니다.

 

active_stream_listener_base.cc

void ActiveStreamListenerBase::newConnection(Network::ConnectionSocketPtr&& socket,
                                             std::unique_ptr<StreamInfo::StreamInfo> stream_info) {
  // Find matching filter chain.
  const auto filter_chain = config_->filterChainManager().findFilterChain(*socket);
  if (filter_chain == nullptr) {
    ...(중략)...
    socket->close();
    return;
  }
 ...(후략)...
}

 

 

newConnection 메소드를 살펴보면, 먼저 Listener의 FilterChainManager로부터 해당 소켓 정보에 해당하는 Filter Chain을 찾는 과정을 수행합니다. 그리고 이 과정에서 매칭되는 Filter Chain을 찾지 못한다면, 연결된 Socket을 종료하고 마칩니다.

 

filter_chain_manager_impl.cc

const Network::FilterChain*
FilterChainManagerImpl::findFilterChain(const Network::ConnectionSocket& socket) const {
  if (matcher_) {
    return findFilterChainUsingMatcher(socket);
  }

  const auto& address = socket.connectionInfoProvider().localAddress();

  const Network::FilterChain* best_match_filter_chain = nullptr;
  // Match on destination port (only for IP addresses).
  if (address->type() == Network::Address::Type::Ip) {
    const auto port_match = destination_ports_map_.find(address->ip()->port());
    if (port_match != destination_ports_map_.end()) {
      best_match_filter_chain = findFilterChainForDestinationIP(*port_match->second.second, socket);
      if (best_match_filter_chain != nullptr) {
        return best_match_filter_chain;
      } else {
        // There is entry for specific port but none of the filter chain matches. Instead of
        // matching catch-all port 0, the fallback filter chain is returned.
        return default_filter_chain_.get();
      }
    }
  }
  // Match on catch-all port 0 if there is no specific port sub tree.
  const auto port_match = destination_ports_map_.find(0);
  if (port_match != destination_ports_map_.end()) {
    best_match_filter_chain = findFilterChainForDestinationIP(*port_match->second.second, socket);
  }
  return best_match_filter_chain != nullptr
             ? best_match_filter_chain
             // Neither exact port nor catch-all port matches. Use fallback filter chain.
             : default_filter_chain_.get();
}

 

위 코드는 Filter Chain Manager로 부터 Filter Chain을 찾는 과정을 보여줍니다.

 

Filter Chain Manager는 Listener 기동 시점에 설정 정보를 파싱하여 Filter Chain 정보를 모두 가지고 있습니다. 따라서 Socket의 Address 정보를 토대로 원하는 Filter Chain을 찾아서 반환해줍니다. 만약에 상응하는 Filter Chain 정보를 찾지 못한 경우에는 Default Filter Chain을 반환합니다.

 

 

active_stream_listener_base.cc

void ActiveStreamListenerBase::newConnection(Network::ConnectionSocketPtr&& socket,
                                             std::unique_ptr<StreamInfo::StreamInfo> stream_info) {
  ...(중략)...
  stream_info->setFilterChainName(filter_chain->name());
  auto transport_socket = filter_chain->transportSocketFactory().createDownstreamTransportSocket();
  auto server_conn_ptr = dispatcher().createServerConnection(
      std::move(socket), std::move(transport_socket), *stream_info);
  if (const auto timeout = filter_chain->transportSocketConnectTimeout();
      timeout != std::chrono::milliseconds::zero()) {
    server_conn_ptr->setTransportSocketConnectTimeout(
        timeout, stats_.downstream_cx_transport_socket_connect_timeout_);
  }
  server_conn_ptr->setBufferLimits(config_->perConnectionBufferLimitBytes());
  ...(중략)...
  const bool empty_filter_chain = !config_->filterChainFactory().createNetworkFilterChain(
      *server_conn_ptr, filter_chain->networkFilterFactories());
  if (empty_filter_chain) {
    ...(중략),,,
    server_conn_ptr->close(Network::ConnectionCloseType::NoFlush);
  }
  newActiveConnection(*filter_chain, std::move(server_conn_ptr), std::move(stream_info));
}

 

매칭되는 Filter Chain을 찾았으면, 위 코드와 같은 과정을 거칩니다. 주요 내용을 살펴보면 다음과 같습니다.

 

1. Downstream을 위한 Transport Socket을 생성합니다.

2. dispatcher에게 ServerConnection을 요청합니다. 

 

 

dispatcher_impl.cc

Network::ServerConnectionPtr
DispatcherImpl::createServerConnection(Network::ConnectionSocketPtr&& socket,
                                       Network::TransportSocketPtr&& transport_socket,
                                       StreamInfo::StreamInfo& stream_info) {
  ASSERT(isThreadSafe());
  return std::make_unique<Network::ServerConnectionImpl>(
      *this, std::move(socket), std::move(transport_socket), stream_info, true);
}

 

이때 내부적으로 dispatcher에서는 ServerConnectionImpl을 생성하는 것을 볼 수 있습니다.

 

connection_impl.cc

ServerConnectionImpl::ServerConnectionImpl(Event::Dispatcher& dispatcher,
                                           ConnectionSocketPtr&& socket,
                                           TransportSocketPtr&& transport_socket,
                                           StreamInfo::StreamInfo& stream_info, bool connected)
    : ConnectionImpl(dispatcher, std::move(socket), std::move(transport_socket), stream_info,
                     connected) {}

 

생성되는 ServerConnectionImpl의 생성자는 위와 같으며, 부호 생성자에게 전달받은 인자들을 넘깁니다.

 

connection_impl.cc

ConnectionImpl::ConnectionImpl(Event::Dispatcher& dispatcher, ConnectionSocketPtr&& socket,
                               TransportSocketPtr&& transport_socket,
                               StreamInfo::StreamInfo& stream_info, bool connected)
    : ConnectionImplBase(dispatcher, next_global_id_++),
      transport_socket_(std::move(transport_socket)), socket_(std::move(socket)),
      stream_info_(stream_info), filter_manager_(*this, *socket_),
      write_buffer_(dispatcher.getWatermarkFactory().createBuffer(
          [this]() -> void { this->onWriteBufferLowWatermark(); },
          [this]() -> void { this->onWriteBufferHighWatermark(); },
          []() -> void { /* TODO(adisuissa): Handle overflow watermark */ })),
      read_buffer_(dispatcher.getWatermarkFactory().createBuffer(
          [this]() -> void { this->onReadBufferLowWatermark(); },
          [this]() -> void { this->onReadBufferHighWatermark(); },
          []() -> void { /* TODO(adisuissa): Handle overflow watermark */ })),
      write_buffer_above_high_watermark_(false), detect_early_close_(true),
      enable_half_close_(false), read_end_stream_raised_(false), read_end_stream_(false),
      write_end_stream_(false), current_write_end_stream_(false), dispatch_buffered_data_(false),
      transport_wants_read_(false) {

  if (!connected) {
    connecting_ = true;
  }

  Event::FileTriggerType trigger = Event::PlatformDefaultTriggerType;

  // We never ask for both early close and read at the same time. If we are reading, we want to
  // consume all available data.
  socket_->ioHandle().initializeFileEvent(
      dispatcher_, [this](uint32_t events) -> void { onFileEvent(events); }, trigger,
      Event::FileReadyType::Read | Event::FileReadyType::Write);

  transport_socket_->setTransportSocketCallbacks(*this);

  // TODO(soulxu): generate the connection id inside the addressProvider directly,
  // then we don't need a setter or any of the optional stuff.
  socket_->connectionInfoProvider().setConnectionID(id());
  socket_->connectionInfoProvider().setSslConnection(transport_socket_->ssl());
}

 

 

ConnectionImpl 생성자 내부를 살펴보면, Socket 내부에 Write/Read 전용 Buffer를 설정하는 것을 볼 수 있습니다. 해당 Buffer는 사용자가 HTTP 요청을 전달했을 때 데이터를 읽는 용도 그리고 upstream으로부터 응답 데이터가 전달되었을 때 기록하는 용도로 사용됩니다.

 

또한 downstream을 위한 transport_socket 생성 및 Filter Chains 관리를 위한 filter_manager를 생성합니다.

 

중요하게 살펴볼 부분은 향후 Socket 내부에 Read/Write 이벤트가 감지되면 이를 통지받고 후속 작업을 처리하기 위해 Dispatcher에게 onFileEvent()를 등록하는 부분입니다. 이는 내부적으로 다시 libevent에게 등록이 되며, 향후 Client가 HTTP 요청을 전달하게 되면, 해당 메소드로 요청 항목이 전달되어 후속 작업을 수행할 수 있습니다.

 

 

3. Filter Chain으로부터 생성해야할 Filter 목록을 확인하여 Filter Chains(Network Filters)를 생성합니다.

 

 

ServerConnection을 생성하고 나면, 그 다음에는 Filter Chains를 생성하는 작업을 수행합니다. 이를 위해 Listener 에게 FilterChain 생성을 요청합니다. 이때 생성되는 Filter Chains는 ServerConnection 내부에 매핑되어야하기 때문에 해당 정보를 Filter Chains와 같이 전달합니다.

 

listener_impl.cc

bool ListenerImpl::createNetworkFilterChain(
    Network::Connection& connection,
    const std::vector<Network::FilterFactoryCb>& filter_factories) {
  return Configuration::FilterChainUtility::buildFilterChain(connection, filter_factories);
}

 

요청을 전달받은 Listener는 Filter Chains 생성 처리를 buildFilterChain util 함수에 위임합니다.

 

configuration_impl.cc

bool FilterChainUtility::buildFilterChain(Network::FilterManager& filter_manager,
                                          const std::vector<Network::FilterFactoryCb>& factories) {
  for (const Network::FilterFactoryCb& factory : factories) {
    factory(filter_manager);
  }

  return filter_manager.initializeReadFilters();
}

 

buildFilterChains는 전달받은 Filter를 순차적으로 순회하면서 ServerConnection 내부에 생성하도록 Callback 팩토리 메소드를 수행합니다.

 

config.cc

return [singletons, filter_config, &context,
        clear_hop_by_hop_headers](Network::FilterManager& filter_manager) -> void {
  auto hcm = std::make_shared<Http::ConnectionManagerImpl>(
      *filter_config, context.drainDecision(), context.api().randomGenerator(),
      context.httpContext(), context.runtime(), context.localInfo(), context.clusterManager(),
      context.overloadManager(), context.mainThreadDispatcher().timeSource());
  if (!clear_hop_by_hop_headers) {
    hcm->setClearHopByHopResponseHeaders(false);
  }
  filter_manager.addReadFilter(std::move(hcm));
};

 

가령 위 코드는 Http Connection Manager Filter에서 반환하는 Callback Factory 메소드가 수행되었다고 가정해봤습니다.

 

 

이때 인자로 ServerConnection을 전달받았으며, 해당 람다내에서는 HttpConnectionManager를 생성한 다음 ServerConnection 내부에 addReadFilter를 호출하여 Filter 정보를 등록하는 것을 볼 수 있습니다.

 

connection_impl.cc

void ConnectionImpl::addReadFilter(ReadFilterSharedPtr filter) {
  filter_manager_.addReadFilter(filter);
}

 

이 경우 호출되는 ServerConnection 내부에서는 filter_manager를 통해 readFilter를 등록하도록 재요청합니다. 

 

filter_manager_impl.cc

void FilterManagerImpl::addReadFilter(ReadFilterSharedPtr filter) {
  ASSERT(connection_.state() == Connection::State::Open);
  ActiveReadFilterPtr new_filter = std::make_unique<ActiveReadFilter>(*this, filter);
  filter->initializeReadFilterCallbacks(*new_filter);
  LinkedList::moveIntoListBack(std::move(new_filter), upstream_filters_);
}

 

filter_manager에서 addReadFilter가 호출되면, 생성된 Filter에 ActiveReadFilter 정보를 전달하여 해당 Filter 내에서 향후 Callback할 수 있도록 initializeReadFilterCallbacks를 등록합니다. 그리고 자신의 upstream_filters_에 Read Filter를 등록하는 것을 볼 수 있습니다.

 

conn_manager_impl.cc

void ConnectionManagerImpl::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {
  read_callbacks_ = &callbacks;
  stats_.named_.downstream_cx_total_.inc();
  stats_.named_.downstream_cx_active_.inc();
  if (read_callbacks_->connection().ssl()) {
    stats_.named_.downstream_cx_ssl_total_.inc();
    stats_.named_.downstream_cx_ssl_active_.inc();
  }

  read_callbacks_->connection().addConnectionCallbacks(*this);

  if (!read_callbacks_->connection()
           .streamInfo()
           .filterState()
           ->hasData<Network::ProxyProtocolFilterState>(Network::ProxyProtocolFilterState::key())) {
    read_callbacks_->connection().streamInfo().filterState()->setData(
        Network::ProxyProtocolFilterState::key(),
        std::make_unique<Network::ProxyProtocolFilterState>(Network::ProxyProtocolData{
            read_callbacks_->connection().connectionInfoProvider().remoteAddress(),
            read_callbacks_->connection().connectionInfoProvider().localAddress()}),
        StreamInfo::FilterState::StateType::ReadOnly,
        StreamInfo::FilterState::LifeSpan::Connection);
  }

  if (config_.idleTimeout()) {
    connection_idle_timer_ = read_callbacks_->connection().dispatcher().createScaledTimer(
        Event::ScaledTimerType::HttpDownstreamIdleConnectionTimeout,
        [this]() -> void { onIdleTimeout(); });
    connection_idle_timer_->enableTimer(config_.idleTimeout().value());
  }

  if (config_.maxConnectionDuration()) {
    connection_duration_timer_ = read_callbacks_->connection().dispatcher().createTimer(
        [this]() -> void { onConnectionDurationTimeout(); });
    connection_duration_timer_->enableTimer(config_.maxConnectionDuration().value());
  }

  read_callbacks_->connection().setDelayedCloseTimeout(config_.delayedCloseTimeout());

  read_callbacks_->connection().setConnectionStats(
      {stats_.named_.downstream_cx_rx_bytes_total_, stats_.named_.downstream_cx_rx_bytes_buffered_,
       stats_.named_.downstream_cx_tx_bytes_total_, stats_.named_.downstream_cx_tx_bytes_buffered_,
       nullptr, &stats_.named_.downstream_cx_delayed_close_timeout_});
}

 

등록된 Filter가 Http Connection Manager Filter임을 가정하였으므로, initializeReadFilter를 호출하면, 위 코드 구문이 호출될 것입니다. 해당 코드를 개략적으로 살펴보면, Callback을 자신의 프로퍼티에 등록하는 작업 외에 metric 정보를 갱신하는 것을 볼 수 있습니다. 또한 Timeout이 발생하면 이를 처리하기 위해 dispatcher에게 다양한 Timer 생성 및 Timeout 발생 시 처리 하도록 등록하는 등 Read Filter 동작과 관련된 기본 초기화 작업을 수행합니다.

 

Http Connection Manager의 경우는 Read Filter의 역할만을 수행하기 때문에 filter_manager 내부에 upstream_filters에만 filter가 추가됩니다. 하지만 FilterChain Factory 내부에는 Read Filter 뿐만 아니라 Writer Filter의 역할을 수행하는 것도 있고 Read/Write를 모두 수행하는 Filter가 존재합니다.

 

filter_manager_impl.cc

void FilterManagerImpl::addWriteFilter(WriteFilterSharedPtr filter) {
  ASSERT(connection_.state() == Connection::State::Open);
  ActiveWriteFilterPtr new_filter = std::make_unique<ActiveWriteFilter>(*this, filter);
  filter->initializeWriteFilterCallbacks(*new_filter);
  LinkedList::moveIntoList(std::move(new_filter), downstream_filters_);
}

void FilterManagerImpl::addFilter(FilterSharedPtr filter) {
  addReadFilter(filter);
  addWriteFilter(filter);
}

 

이 경우에는 람다내에서 ServerConnectionImpl에 Filter 생성을 요청할 때 addWriteFilter 혹은 addFilter 메소드를 호출합니다. 가령 addWriterFilter의 경우는 Writer Filter만의 역할을 수행하기 때문에 궁극적으로는 filter_manager의 downstream_filters_ 에 추가되면서 Read Filter 등록과 마찬가지로 ActiveWriterFilter 인스턴스를 만들어서 향후 Callback할 수 있도록 initializeWriteFilterCallbacks를 등록하는 것을 볼 수 있습니다.

 

반면 addFilter의 경우는 Read/Write를 모두 수행할 때 호출되기 때문에 addReadFilter와 addWriteFilter를 차례대로 호출합니다.

 

configuration_impl.cc

bool FilterChainUtility::buildFilterChain(Network::FilterManager& filter_manager,
                                          const std::vector<Network::FilterFactoryCb>& factories) {
  for (const Network::FilterFactoryCb& factory : factories) {
    factory(filter_manager);
  }

  return filter_manager.initializeReadFilters();
}

 

Filter Factories로 부터 Filter 생성이 모두 완료되면, ServerConnectionImpl 내부 filter_manager에는 upstream 전용 filter와 downstream 전용 filter가 모두 생성되어있습니다. 그 과정이 완료되면 ServerConnectionImpl 내부의 initializeReadFilters()를 호출 합니다.

 

connection_impl.cc

bool ConnectionImpl::initializeReadFilters() { return filter_manager_.initializeReadFilters(); }

 

그리고 해당 요청을 전달받은 ServerConnectionImpl 내부에서는 다시 filter_manager에게 해당 요청 처리를 위임합니다.

 

filter_manager_impl.cc

bool FilterManagerImpl::initializeReadFilters() {
  if (upstream_filters_.empty()) {
    return false;
  }
  onContinueReading(nullptr, connection_);
  return true;
}

void FilterManagerImpl::onContinueReading(ActiveReadFilter* filter,
                                          ReadBufferSource& buffer_source) {
  // Filter could return status == FilterStatus::StopIteration immediately, close the connection and
  // use callback to call this function.
  if (connection_.state() != Connection::State::Open) {
    return;
  }

  std::list<ActiveReadFilterPtr>::iterator entry;
  if (!filter) {
    connection_.streamInfo().addBytesReceived(buffer_source.getReadBuffer().buffer.length());
    entry = upstream_filters_.begin();
  } else {
    entry = std::next(filter->entry());
  }

  for (; entry != upstream_filters_.end(); entry++) {
    if (!(*entry)->filter_) {
      continue;
    }
    if (!(*entry)->initialized_) {
      (*entry)->initialized_ = true;
      FilterStatus status = (*entry)->filter_->onNewConnection();
      if (status == FilterStatus::StopIteration || connection_.state() != Connection::State::Open) {
        return;
      }
    }

    StreamBuffer read_buffer = buffer_source.getReadBuffer();
    if (read_buffer.buffer.length() > 0 || read_buffer.end_stream) {
      FilterStatus status = (*entry)->filter_->onData(read_buffer.buffer, read_buffer.end_stream);
      if (status == FilterStatus::StopIteration || connection_.state() != Connection::State::Open) {
        return;
      }
    }
  }
}

 

해당 처리 과정을 살펴보면, 초기에 upstream_filters_에 등록할 때 ActiveReadFilter 인스턴스를 새로 생성해서 등록했는데 초기 initialized_ 값은 기본적으로 false 입니다. 따라서 위 코드에서는 등록된 upstream_filters_ 를 순회하면서 가장 먼저 onNewConnection() 작업을 수행할 것입니다.

 

여기에는 등록된 Filter가 Http Connection Manager 하나만 존재한다고 가정하기 때문에, 해당 Filter의 onNewConnection()이 호출됩니다.

 

connection_manager_impl.cc

Network::FilterStatus ConnectionManagerImpl::onNewConnection() {
  if (!read_callbacks_->connection().streamInfo().protocol()) {
    // For Non-QUIC traffic, continue passing data to filters.
    return Network::FilterStatus::Continue;
  }
  // Only QUIC connection's stream_info_ specifies protocol.
  Buffer::OwnedImpl dummy;
  createCodec(dummy);
  ASSERT(codec_->protocol() == Protocol::Http3);
  // Stop iterating through each filters for QUIC. Currently a QUIC connection
  // only supports one filter, HCM, and bypasses the onData() interface. Because
  // QUICHE already handles de-multiplexing.
  return Network::FilterStatus::StopIteration;
}

 

해당 코드를 잠시 살펴보면, QUIC 트래픽인지 확인하고 일반 HTTP 프로토콜이라면 해당 Filter 사용이 가능하도록 구현되어있음을 확인할 수 있습니다.

 

여기까지 수행하면 Filter Chains를 생성하고 Filter의 초기화까지 수행하는 전 과정을 살펴볼 수 있습니다.

 

active_stream_listener_base.cc

void ActiveStreamListenerBase::newConnection(Network::ConnectionSocketPtr&& socket,
                                             std::unique_ptr<StreamInfo::StreamInfo> stream_info) {
  ...(중략)...
  const bool empty_filter_chain = !config_->filterChainFactory().createNetworkFilterChain(
      *server_conn_ptr, filter_chain->networkFilterFactories());
  if (empty_filter_chain) {
    ...(중략),,,
    server_conn_ptr->close(Network::ConnectionCloseType::NoFlush);
  }
  newActiveConnection(*filter_chain, std::move(server_conn_ptr), std::move(stream_info));
}

 

앞서 살펴본 긴 과정의 Filter Chains를 생성하고나면, Filter Chains가 존재하는지 여부를 bool 값으로 반환합니다. 만약에 Filter Chains 생성 과정에서 생성된 FilterChains가 전혀 존재하지 않는다면, Stream을 연결하여 작업을 이어나가는 것이 무의미하기 때문에 해당 Server Connection을 종료합니다. 그렇지 않을 경우 생성된 Filter Chains를 기반으로 ActiveConnection을 생성하는 과정을 진행합니다.

 

active_tcp_listener.cc

void ActiveTcpListener::newActiveConnection(const Network::FilterChain& filter_chain,
                                            Network::ServerConnectionPtr server_conn_ptr,
                                            std::unique_ptr<StreamInfo::StreamInfo> stream_info) {
  auto& active_connections = getOrCreateActiveConnections(filter_chain);
  auto active_connection =
      std::make_unique<ActiveTcpConnection>(active_connections, std::move(server_conn_ptr),
                                            dispatcher().timeSource(), std::move(stream_info));
  // If the connection is already closed, we can just let this connection immediately die.
  if (active_connection->connection_->state() != Network::Connection::State::Closed) {
    ENVOY_CONN_LOG(
        debug, "new connection from {}", *active_connection->connection_,
        active_connection->connection_->connectionInfoProvider().remoteAddress()->asString());
    active_connection->connection_->addConnectionCallbacks(*active_connection);
    LinkedList::moveIntoList(std::move(active_connection), active_connections.connections_);
  }
}

 

여기서 ActiveConnection이란 생성된 Filter Chains를 사용하는 Connection이 얼마나 있는지를 관리하기 위한 자료구조 입니다. 따라서 getOrCreateActiveConnection 메소드를 호출함으로써, 먼저 해당 Filter에 존재하는 ActiveConnection이 있는지를 살펴봅니다. 

 

active_stream_listener_base.cc

ActiveConnections& OwnedActiveStreamListenerBase::getOrCreateActiveConnections(
    const Network::FilterChain& filter_chain) {
  ActiveConnectionCollectionPtr& connections = connections_by_context_[&filter_chain];
  if (connections == nullptr) {
    connections = std::make_unique<ActiveConnections>(*this, filter_chain);
  }
  return *connections;
}

active_stream_listener_base.h

absl::flat_hash_map<const Network::FilterChain*, ActiveConnectionCollectionPtr>
    connections_by_context_;

 

active_stream_listener_base.h

class ActiveConnections : public Event::DeferredDeletable {
public:
  ActiveConnections(OwnedActiveStreamListenerBase& listener,
                    const Network::FilterChain& filter_chain);
  ~ActiveConnections() override;

  // listener filter chain pair is the owner of the connections
  OwnedActiveStreamListenerBase& listener_;
  const Network::FilterChain& filter_chain_;
  // Owned connections.
  std::list<std::unique_ptr<ActiveTcpConnection>> connections_;
};

 

 

이때 ActiveConnections를 가져오는 메소드는 위 코드와 같으며, 내부적으로는 FilterChain 별로 ActiveConnection 목록을 관리하는 hash map을 통해서 참조하는 것을 확인할 수 있습니다.

 

ActiveConnections를 가져오면 신규로 생성하는 ActiveConnection을 만들고 ActiveConnections에 추가하는 것으로 Client의 접속 요청 이후 Connection 할당 과정은 마무리됩니다.

 

 

---

4. Client Connection 연결 과정 정리

 

이전에 Client Connection 연결 과정을 코드를 통해서 어떻게 구성되는지 집중적으로 살펴봤습니다. 다만 지엽적인 내용까지 살펴보느라 전체적인 흐름을 이해하기 쉽지 않았을 수도 있습니다. 따라서 이번에는 큰 그림에서 컴포넌트간 메시지 교환을 중점으로 Client Connection이 어떻게 연결되는지 개략적으로 다시 살펴보도록 하겠습니다. 

 

 

 

1. Libevent 로 부터 Listener 가 Listen 하고 있는 소켓으로 사용자의 Connection 연결 요청이 접수되었을 때, 내부적으로 존재하는 Worker 쓰레드 중 하나를 선정하여 해당 Listener의 onSocketEvent 메소드를 호출합니다. 참고로 위 예시에서는 Listener 0에 요청이 전달되었을 때 Worker 0번 쓰레드가 담당한다고 가정하여 ActiveTcpListener 0이 수신받았습니다.

 

2. ActiveTcpListener에서는 전체 연결된 Connection 개수가 Global Limit을 넘었는지 확인하고 내부적으로 AcceptedSocket을 거쳐 ActiveTcpSocket을 생성하여 Connection 연결을 Accept합니다.

 

3. 생성된 Socket에서 Listener Filter Chain을 생성하기 위해 Listener Config에게 Listener Filter Chain 정보를 요청합니다.

 

4. Listener Config는 Envoy 기동 과정에서 Parsing되었거나 LDS로 부터 갱신된 Listener Filter Chains 정보를 기반으로 Chains 내부에 존재하는 Factory Callback 메소드를 순회하면서 ActiveTcpSocket 내부에 있는 accept_filters에 Listener Filter를 생성하여 바인딩합니다.

 

5. ActiveTcpSocket은 Listener Filters를 순회하면서 onAccept를 수행합니다. 

 

6. Filter Chains(Network Filters)를 생성하기 위해 ActiveTcpListener는 Listener Config 로부터 보유하고 있는 Filter Chains 정보를 요청합니다.

 

7. Listener Config 내부에 존재하는 filter_chain_manager에는 Trie, HashMap 등을 비롯하여 Filter Chain 정보를 빠르게 찾기 위한 다양한 자료구조가 존재하는데, 이를 활용하여 사용자 요청한 Filter Chain 정보를 제공합니다.

 

8. ActiveTcpListener는 ServerConnetionImpl을 생성하기 위해 Dispatcher에 요청합니다.

 

9. Dispatcher는 ServerConnectionImpl을 생성합니다.

 

10. 생성된 ServerConnectionImpl은 사용자가 접속 연결 이후 실질적으로 HTTP 요청 시, 이벤트를 수신 받아야되기 때문에 Dispatcher에게 Socket 이벤트를 전달하도록 등록 요청합니다.

 

11. Didpatcher는 Libevent에게 ServerConnectionImpl이 요청한 정보를 등록합니다. 이후 해당 Socket에 이벤트가 감지되면 ServerConnectionImpl이 등록한 onFileEvent 메소드를 호출합니다.

 

12. ActiveTcpListener는 Filter Chains를 생성하기 위해 Listner Config에게 처리를 요청합니다.

 

13. Listener Config는 Util 함수를 활용하여 Filter Chains를 생성합니다. 이때 Filter의 특성에 따라 Read Filter인 경우에는 ServerConnectionImpl 내에 있는 filter_manager의 upstream_filters_에 매핑되고 Write Filter라면 downstream_filters_에 매핑됩니다. 만약 Read/Write 둘 다 처리하는 경우에는 두 군데 모두 입력합니다.

 

14. 생성된 Filter Chain에 연결된 Server Connection 정보를 관리하기 위해 ActiveTcpSocket 내부에 있는 connections_by_contexts_에 신규로 생성된 ActiveConnection 정보를 등록합니다.

 

---

5. 마무리

 

이번 포스팅에서는 사용자가 HTTP 요청을 통해 envoy의 기능을 이용하고자 할 때 내부적으로 Listener 구성이 어떻게 되어있는지 그리고 Client Connection이 어떻게 할당되는지에 대해서 집중적으로 알아봤습니다.

 

envoy를 처음 학습하면, 이러한 과정이 마법처럼 느껴지는데 코드를 통해서 살펴보니 동작 과정에 대해서 어느정도 이해할 수 있었던 것 같습니다. 다음 포스팅에서는 연결이 완료된 이후 실제 HTTP 요청이 전달될 때 처리 과정에 대해서 살펴보겠습니다.

1. 서론

이전 시리즈 내용을 통해서 Envoy에서 가장 중요한 2가지 컴포넌트인 Listener Manager와 Cluster Manager에 대해서 살펴봤습니다.

 

 

 

 

 

다시 한번 전체 과정을 간략하게 표현해보면 다음과 같습니다.사용자의 접속 요청이 전달되면, Listener에서 요청을 전달받고 이에 부합하는 Cluster를 찾아 Downstream과 Upstream을 연결합니다. 이때 만약 사용자 요청이 Http 연결일 경우에는 Listener에서 Cluster Manager로 Cluster를 요청하는 주체가 Listener가 보유하고 있는 Network Filter 중 하나인 HttpConnectionManager입니다. 해당 컴포넌트는 Listener 내부에 존재하는 Filter이지만, 사용자 요청처리를 수행하는데 있어 중요한 역할을 수행하기 때문에. 이번 포스팅에서는 HttpConnectionManager의 기능에 대해서 살펴보겠습니다. 

---

 

 

2. HttpConnectionManager

 

 

이전 포스팅을 통해 Listener 하위에는 위 그림과 같이 Filter Chains를 관리하는 filter_chain_manager가 존재함을 확인했습니다.

 

 

 

해당 구조 내부에는 위 그림과 같이 Trie 구조가 존재하는데, 이는 사용자가 Listener를 통해서 연결을 원하는 domain을 전달했을 때, 해당 요청에 부합되는 Filter Chain을 찾기 위한 용도로 사용됩니다. 따라서 각각의 Trie 별로 매칭되는 노드에는 Filter들이 Chain 형식으로 매핑되어있습니다.

 

따라서 사용자가 접속을 요청하면, 내부적으로는 해당 Trie를 검색해서 사용자의 요청에 부합되는 항목의 Filter Chain 목록을 반환합니다. 그리고 해당 Chain 목록을 전달받으면, Chain을 탐색하면서 매핑된 Factory Callback을 실행하여 Filter Chain을 구성합니다.

 

Envoy가 제공하는 Network Filter는 굉장히 많은 종류가 있는데요. 그 중 HttpConnectionManager는 Http 요청을 처리하는 Filter로써 이에 대해서 다루어보고자 합니다.

 

 

 

 

HttpConnectionManager는 Network Filter이므로 앞서 설명한 것과 같이 사용자가 명시한 도메인에 해당되는 Network Filter 목록에 Http 처리가 매핑되어있을 경우 Filter Chain에 포함되어있을 것입니다.

 

이때 사용자가 Listener에 접속을 요청할 때마다 위 그림과 같이 내부적으로 Upstream Connection을 만듭니다. 위 그림은 현재 2개의 Connection이 생성되었음을 가정했습니다. 이때 각각의 ServerConnection은 별개의 Network Filter Chain을 가지고 있게되고 만약 2개의 Connection이 모두 Http 처리를 담당해야한다면 위 그림과 같이 2개의 별개 HttpConnectionManager가 생성될 것입니다.

 

그렇다면 HttpConnectionManager는 어떻게 구성되어있을까요?

 

 

 

생성된 HttpConnectionManager는 대략 위와 같은 모습을 구성하고 있습니다. 물론 그림으로 표현한 속성 이외에 다양한 프로퍼티가 존재하지만, 핵심이라고 생각하는 몇 개만 표현했습니다. 그렇다면 각각의 속성은 무엇이며 어떤 과정을 거쳐 생성될까요? 핵심적인 요소에 대해서 하나씩 살펴보겠습니다.

 

---

3. RDS

 

먼저 살펴볼 것은 RDS입니다. 해당 컴포넌트는 이전에 살펴본 HttpConnectionManager의 속성 중 config_와 관련이 있습니다. 해당 config안에는 HttpConnectionManager를 구성하는데 있어 필요한 속성이 지정되어있는데요. 그중 Route 관련 속성은 route_config_provider입니다.

 

 

 

이전에 살펴본 그림에서 HttpConnectionManager와 SingletonManager가 연관관계를 맺고 있는 것을 확인했는데요. SingletonManager가 가진 속성 중에서 RouteConfigProviderManager가 RDS 처리를 담당하고 있습니다.

 

그렇다면 왜 SingletonManager에 의해서 해당 속성이 관리될까요? 위 그림을 살펴보면 HttpConnectionManager는 사용자의 Connection 별로 여러개가 생성됨을 볼 수 있습니다. 하지만 RDS 처리 또한 각각의 HttpConnectionManager를 통해 관리되어야한다면, RDS 처리를 위한 overhead 또한 증가하게되고 무엇보다 동일한 정보가 중복 관리되기 때문에 관리 용이성 또한 증가합니다. 따라서 전역적으로 하나의 인스턴스만을 생성함으로써 데이터를 한 곳에서 관리하고 모든 HttpConnectionManager가 이를 공유하도록 싱글톤 패턴이 적용되어있습니다.

 

해당 내용을 코드로 살펴보면 다음과 같습니다.

 

config.cc

Utility::Singletons Utility::createSingletons(Server::Configuration::FactoryContext& context) {
  std::shared_ptr<Http::TlsCachingDateProviderImpl> date_provider =
      context.singletonManager().getTyped<Http::TlsCachingDateProviderImpl>(
          SINGLETON_MANAGER_REGISTERED_NAME(date_provider), [&context] {
            return std::make_shared<Http::TlsCachingDateProviderImpl>(
                context.mainThreadDispatcher(), context.threadLocal());
          });

  Router::RouteConfigProviderManagerSharedPtr route_config_provider_manager =
      context.singletonManager().getTyped<Router::RouteConfigProviderManager>(
          SINGLETON_MANAGER_REGISTERED_NAME(route_config_provider_manager), [&context] {
            return std::make_shared<Router::RouteConfigProviderManagerImpl>(context.admin());
          });

  Router::ScopedRoutesConfigProviderManagerSharedPtr scoped_routes_config_provider_manager =
      context.singletonManager().getTyped<Router::ScopedRoutesConfigProviderManager>(
          SINGLETON_MANAGER_REGISTERED_NAME(scoped_routes_config_provider_manager),
          [&context, route_config_provider_manager] {
            return std::make_shared<Router::ScopedRoutesConfigProviderManager>(
                context.admin(), *route_config_provider_manager);
          });

  auto http_tracer_manager = context.singletonManager().getTyped<Tracing::HttpTracerManagerImpl>(
      SINGLETON_MANAGER_REGISTERED_NAME(http_tracer_manager), [&context] {
        return std::make_shared<Tracing::HttpTracerManagerImpl>(
            std::make_unique<Tracing::TracerFactoryContextImpl>(
                context.getServerFactoryContext(), context.messageValidationVisitor()));
      });

  std::shared_ptr<Http::DownstreamFilterConfigProviderManager> filter_config_provider_manager =
      Http::FilterChainUtility::createSingletonDownstreamFilterConfigProviderManager(
          context.getServerFactoryContext());

  return {date_provider, route_config_provider_manager, scoped_routes_config_provider_manager,
          http_tracer_manager, filter_config_provider_manager};
}

 

코드를 살펴보면 위와 같이 가장 먼저 하는 것은 SingletonManager에 등록된 인스턴스 중에서  RouteConfigProviderManager, ScopedRoutesConfigProviderManager 와 더불어 다양한 Manager 등 다양한 Manager를 가져오는 작업을 선행합니다. 

 

이때 중요한 것은 앞서 언급한 2가지 Manager이며, RouteConfigProviderManager는 route_config 정보를 기반으로 RDS 혹은 StaticRouteConfig를 처리하는 인스턴스를 생성하는 역할을 수행합니다. 반면 ScopedRoutesConfigProviderManager는 Listener 설정에 scoped_routes 설정이 존재할 때 해당 scoped_routes를 처리하는 인스턴스를 생성하는 역할을 수행합니다.

 

config.cc

switch (config.route_specifier_case()) {
case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
    RouteSpecifierCase::kRds:
case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
    RouteSpecifierCase::kRouteConfig:
  route_config_provider_ = Router::RouteConfigProviderUtil::create(
      config, context_.getServerFactoryContext(), context_.messageValidationVisitor(),
      context_.initManager(), stats_prefix_, route_config_provider_manager_);
  break;
case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
    RouteSpecifierCase::kScopedRoutes:
  scoped_routes_config_provider_ = Router::ScopedRoutesConfigProviderUtil::create(
      config, context_.getServerFactoryContext(), context_.initManager(), stats_prefix_,
      scoped_routes_config_provider_manager_);
  break;
case envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
    RouteSpecifierCase::ROUTE_SPECIFIER_NOT_SET:
  PANIC_DUE_TO_CORRUPT_ENUM;
}

 

HttpConnectionManager를 구성하는 단계에서 전달받은 두 속성은 이후 route 관련 정보를 생성하는데 사용됩니다. 다만 두 속성 모두가 생성되지는 않으며, 기존에 지정된 설정내역을 살펴보고 하나를 생성합니다.

 

즉 위 코드와 같이 config에 지정된 route_specifier에 의해서 결정됩니다. 만약 route_config 설정이 지정되어있다면, 기존에 SingletonManager로부터 부여받은 route_config_provider_manager를 전달하여 route 처리를 수행할 수 있는 route_config_provider를 전달받습니다. 반면 scoped_route라면 scoped_route_config_provider_manager를 전달하여 scoped_route_config_provider를 전달받습니다.

 

본포스팅에서는 route_config만 지정되어있음을 가정하며, 따라서 위 코드에서는 route_config_provider_ 만이 생성되었음을 전제로 진행하겠습니다. 또한 route를 처리하는 방식이 static이 아닌 dynamic xDS를 활용한 동적 변경을 가정하겠습니다.

 

 

route_config_provider가 생성된 이후 RDS를 매핑하는 과정을 살펴보면 위그림과 같습니다.

 

1. HttpConnectionManager에 매핑되어있는 route_config_provider를 기반으로 route config를 분석할 수 있는 인스턴스를 생성합니다. 이후 RouteConfigProviderManager에 존재하는 dynamic_route_config_providers_ 로부터 rds 메시지 값을 해시한 결과를 기반으로 dynamic_route_config_providers_ Map에 존재하는지 살펴보고 만약 존재한다면, 해당 provider를 반환합니다.

 

2. dynamic_route_config_providers_에 존재하지 않는다면, RDS 생성을 위해 Cluster Manager로부터 Subscription factory를 요청합니다.

 

3.  xDS에 전달받기 원하는 타입 및 수신 callback을 Multiplexer에 등록합니다.

 

4. 등록이 완료된 이후 Subscription을 반환합니다.

 

5. RdsRouteConfigProviderImpl 인스턴스를 생성하고 그 안에 subscription을 바인딩합니다. 또한 dynamic_route_config_providers_ Map에 해당 인스턴스를 삽입함으로써 이후 동일한 요청이 전달되면, 새로운 provider를 생성하지 않고 매핑된 값을 반환합니다.

 

route_config_provider_manager_impl.h

auto subscription = std::make_shared<RdsRouteConfigSubscription>(
    std::move(config_update), std::move(resource_decoder), rds.config_source(),
    rds.route_config_name(), manager_identifier, factory_context,
    stat_prefix + absl::AsciiStrToLower(getRdsName()) + ".",
    absl::AsciiStrToUpper(getRdsName()), manager_);
auto provider = std::make_shared<RdsRouteConfigProviderImpl>(std::move(subscription),
                                                             factory_context);

6. 해당 provider와 subscription 정보를 반환합니다.

 

7. RdsRouteConfigProviderImpl 내부에서는 xDS API가 변경이 생겼을 때, 내부에서 ThreadLocalStorage에 존재하는 ThreadLocalConfig의 내용을 변경함으로써, 쓰레드 전체에 동일한 데이터를 공유할 수 있도록 유지합니다.

 

위와 같은 7가지 단계를 통해서 HttpConnectionManager가 생성될 때 Singleton Manager를 통해 전역적으로 RDS를 관리하는 하나의 provider를 공급받고, ThreadLocalStorage를 활용해서 모든 쓰레드에서 동일한 데이터에 대한 접근이 가능하도록 공유합니다.

 

---

 

4. Http filter

 

HttpConnectionManager는 Http 처리를 담당합니다. 이때 해당 컴포넌트 내부에는 http 처리를 위한 무수한 filter가 존재합니다. 참고로  envoy 공식 문서를 살펴보면 지정할 수 있는 filter가 여러가지 있음을 확인할 수 있습니다. 그리고 그 중에는 Routing을 담당하는 Router Filter 또한 존재합니다.

 

따라서 사용자의 요청이 전달되면 Network Filter Chain이 수행되면서 HttpConnectionManager가 실행되고 그리고 그 안에서 다시 HttpConnectionManager가 보유한 Http Filter들이 수행되면서 사용자의 요청이 처리됩니다.

 

이를 조금 더 살펴보겠습니다.

 

config.cc

Http::FilterChainHelper<Server::Configuration::FactoryContext,
                        Server::Configuration::NamedHttpFilterConfigFactory>
    helper(filter_config_provider_manager_, context_.getServerFactoryContext(), context_,
           stats_prefix_);
helper.processFilters(config.http_filters(), "http", "http", filter_factories_);

 

HttpConnectionManager를 구성하는 config 속성을 살펴보면, 해당 Config를 생성할 때 http_filters를 위한 factory를 생성하는 것을 볼 수 있습니다. 이때 이미 사용자가 지정하거나 LDS에 의해서 갱신된 Listener의 Config 정보를 살펴보면, 지정된 http filter 목록을 확인할 수 있습니다.

 

 

따라서 FilterChainHelper를 통해서 Config 정보를 토대로 filter_factories라는 Filter를 생성하는 Factory Callback의 리스트를 채우도록 처리를 위임합니다.

 

FilterChainHelper는 해당 요청을 전달받으면, DependencyManager를 통해서 전달된 Filter의 우선순위를 고려하여 정상적으로 입력이 되었는지를 검사합니다. 그리고 Filter간의 Dependency에 문제가 없게 Config가 전달되었다면, 이를 HttpConnectionManager가 보유한 filter_factories에 Filter 생성을 위한 Callback Factory 메소드를 매핑시킵니다.

 

위 과정을 거쳐 생성된 filter_factories_는 향후 사용자가 Http 요청을 전달하기 위해 Stream을 생성할 때 내부에 존재하는 filter_manager_로 해당 filter_factories_를 전달시켜 해당 filter_manager_가 http filter chain을 생성하고 수행할 수 있도록 처리를 위임합니다.

 

---

5. Codec

 

이번에 살펴볼 것은 codec입니다. HttpConnectionManager에서 codec은 사용자의 요청 정보를 분석하는 역할을 수행합니다. 사용자가 전달한 raw데이터를 지정된 프로토콜 형태로 파싱하고 분석하여 처리하는 과정에서 사용됩니다.

 

 

그림을 살펴보면, codec_은 HttpConnectionManager 내부에 있는 config에 의해서 생성할 수 있습니다. 이때 사용자의 Http 요청은 Http 1.1, Http 2.0 혹은 Http 3.0(Quic) 형태일 수 있습니다. 위 세가지 프로토콜 모두 HttpConnectionManager가 처리하는데요. 각각의 처리방식과 포맷이 다르기 때문에 사용자의 요청이 전달되었을 때, 가정 먼저 수행하는 일은 사용자의 요청이 어떤 프로토콜 형태인지를 파악하는 것입니다.

 

config.cc

Http::ServerConnectionPtr
HttpConnectionManagerConfig::createCodec(Network::Connection& connection,
                                         const Buffer::Instance& data,
                                         Http::ServerConnectionCallbacks& callbacks) {
  switch (codec_type_) {
  case CodecType::HTTP1:
    return std::make_unique<Http::Http1::ServerConnectionImpl>(
        connection, Http::Http1::CodecStats::atomicGet(http1_codec_stats_, context_.scope()),
        callbacks, http1_settings_, maxRequestHeadersKb(), maxRequestHeadersCount(),
        headersWithUnderscoresAction());
  case CodecType::HTTP2:
    return std::make_unique<Http::Http2::ServerConnectionImpl>(
        connection, callbacks,
        Http::Http2::CodecStats::atomicGet(http2_codec_stats_, context_.scope()),
        context_.api().randomGenerator(), http2_options_, maxRequestHeadersKb(),
        maxRequestHeadersCount(), headersWithUnderscoresAction());
  case CodecType::HTTP3:
#ifdef ENVOY_ENABLE_QUIC
    return std::make_unique<Quic::QuicHttpServerConnectionImpl>(
        dynamic_cast<Quic::EnvoyQuicServerSession&>(connection), callbacks,
        Http::Http3::CodecStats::atomicGet(http3_codec_stats_, context_.scope()), http3_options_,
        maxRequestHeadersKb(), maxRequestHeadersCount(), headersWithUnderscoresAction());
#else
    // Should be blocked by configuration checking at an earlier point.
    PANIC("unexpected");
#endif
  case CodecType::AUTO:
    return Http::ConnectionManagerUtility::autoCreateCodec(
        connection, data, callbacks, context_.scope(), context_.api().randomGenerator(),
        http1_codec_stats_, http2_codec_stats_, http1_settings_, http2_options_,
        maxRequestHeadersKb(), maxRequestHeadersCount(), headersWithUnderscoresAction());
  }
  PANIC_DUE_TO_CORRUPT_ENUM;
}

 

이를 Config에서 분석한 다음 Http 1.1일 경우에는 Http1::ServerConnectionImpl, Http 2.0일 경우에는 Http2:ServerConnectionImpl Http 3.0일 경우에는 QuicHttpServerConnectionImpl을 반환하여 HttpConnectionManager 내부에 존재하는 codec_에 매핑하는 작업을 수행합니다.

 

codec.h

class Connection {
public:
  virtual ~Connection() = default;

  /**
   * Dispatch incoming connection data.
   * @param data supplies the data to dispatch. The codec will drain as many bytes as it processes.
   * @return Status indicating the status of the codec. Holds any errors encountered while
   * processing the incoming data.
   */
  virtual Status dispatch(Buffer::Instance& data) PURE;

  /**
   * Indicate "go away" to the remote. No new streams can be created beyond this point.
   */
  virtual void goAway() PURE;

  /**
   * @return the protocol backing the connection. This can change if for example an HTTP/1.1
   *         connection gets an HTTP/1.0 request on it.
   */
  virtual Protocol protocol() PURE;

  /**
   * Indicate a "shutdown notice" to the remote. This is a hint that the remote should not send
   * any new streams, but if streams do arrive that will not be reset.
   */
  virtual void shutdownNotice() PURE;

  /**
   * @return bool whether the codec has data that it wants to write but cannot due to protocol
   *              reasons (e.g, needing window updates).
   */
  virtual bool wantsToWrite() PURE;

  /**
   * Called when the underlying Network::Connection goes over its high watermark.
   */
  virtual void onUnderlyingConnectionAboveWriteBufferHighWatermark() PURE;

  /**
   * Called when the underlying Network::Connection goes from over its high watermark to under its
   * low watermark.
   */
  virtual void onUnderlyingConnectionBelowWriteBufferLowWatermark() PURE;
};

 

프로토콜마다 처리 방법이 다양하지만, 모든 ServerConnectionImpl은 위와 같은 Interface 스펙을 준수합니다. 따라서 HttpConnectionManager에서는 위 interface에 정의된 메소드를 호출하여 처리를 위임할 수 있습니다. 다만 본 포스팅에서는 Http 1.1을 기준으로 처리 과정을 분석하기 때문에 Http1::ServerConnectionImpl이 반환되었다고 가정하겠습니다.

 

그렇다면 Http 1.1을 처리하기 위한 ServerConnectionImpl은 어떤 역할을 수행할까요? 해당 구조에 대해서 조금 더 자세히 살펴보겠습니다.

 

 

 

ServerConnectionImpl을 살펴보면, 여러 속성이 있지만 그 중 가장 중요한 속성은 위 2가지 입니다. 먼저 Parser_에 대해서 알아보겠습니다.

 

---

 

5-1. Parser

 

parser_의 역할은 Client로 부터 전달된 요청 내역을 Http 1.1 스펙에 맞게 분석하여 envoy가 원하는 형태로 데이터를 구성하는 작업을 담당합니다. 이때 envoy 내부에는 해당 내역을 처리하는 Parser가 2개가 존재합니다.

 

codec_impl.cc

ConnectionImpl::ConnectionImpl(Network::Connection& connection, CodecStats& stats,
                               const Http1Settings& settings, MessageType type,
                               uint32_t max_headers_kb, const uint32_t max_headers_count)
    : connection_(connection), stats_(stats), codec_settings_(settings),
      encode_only_header_key_formatter_(encodeOnlyFormatterFromSettings(settings)),
      processing_trailers_(false), handling_upgrade_(false), reset_stream_called_(false),
      deferred_end_stream_headers_(false), dispatching_(false), max_headers_kb_(max_headers_kb),
      max_headers_count_(max_headers_count) {
  if (Runtime::runtimeFeatureEnabled("envoy.reloadable_features.http1_use_balsa_parser")) {
    parser_ = std::make_unique<BalsaParser>(type, this, max_headers_kb_ * 1024, enableTrailers());
  } else {
    parser_ = std::make_unique<LegacyHttpParserImpl>(type, this);
  }
}

 

첫 번째는 LegacyHttpParserImpl로써 envoy가 기존부터 제공해온 Parser입니다. 이후 해당 Parser의 성능 개선을 위해 추가로 개발한 것이 BalsaParser입니다. 다만 BalsaParser는 아직 완전하지는 않으며, envoy에서 해당 Parser를 사용하려면

'envoy.reloadable_features.http1_use_balsa_parser' 옵션을 활성화했을 경우 사용할 수 있습니다. 참고로 본 포스팅에서는 기본으로 사용되는 LegacyHttpParserImpl에 대해서 다루어보겠습니다.

 

 

parser_가 생성되면, 추후 parser_에게 사용자 요청을 처리하도록 위임할 것입니다. 이때 Parser가 처리 중간 중간마다 특정 event 즉 header 필드명이 무엇인지 하나씩 파악했거나, header 값을 분석했을 때 이를 요청자인 ServerConnection 에게 알려줘야 해당 데이터들을 전달받아 envoy가 원하는 형태로 데이터를 가공하거나 그 이후 처리해야할 비즈니스 로직을 수행할 수 있을 것입니다.

 

이때 ServerConnectionImpl은 기본적으로 ParserCallbacks Virtual Class를 상속받았으며, 그 안에 정의된 메소드는 해당 메소드가 호출될 때 수행해야할 비즈니스 로직이 구현되어있습니다. 구현해야할 메소드는 위와 같이 총 10개이며, 각각의 의미는 다음과 같습니다.

 

메소드명	의미
onMessageBegin	Request/Response가 시작될 때 호출되는 callback으로 반환 값으로 성공/실패 메시지가 반환됨
onUrl	URL data를 Parser가 분석했을 때 호출되는 callback으로 반환 값으로 성공/실패 메시지가 반환됨
onStatus	Status data를 Parser가 분석헀을 때 호출되는 callback으로 반환 값으로 성공/실패 메시지가 반환됨
onHeaderField	header의 field 명을 수신받았을 때 호출되는 callback으로 반환 값으로 성공/실패 메시지가 반환됨
onHeaderValue	header의 value 값을 수신받았을 때 호출되는 callback으로 반환 값으로 성공/실패 메시지가 반환됨
onHeaderComplete	header 분석이 완료되었을 때 호출되는 callback으로 반환 값으로 성공/실패 메시지가 반환됨
bufferBody	body data를 분석했을 때 호출되는 callback
onMessageComplete	Parser가 HTTP 데이터를 모두 분석 완료했을 때 호출되는 callback으로 반환 값으로 성공/실패 메시지가 반환됨
onChunkHeader	chunk header를 받았을 때 호출되는 callback

 

이후 Parser에서는 처리 도중 중간 중간에 ParserCallbacks에 정의된 메소드를 호출함으로써, ServerConnectionImpl에게 Parsing 결과를 중간 중간 callback 하도록 구현되었습니다.

 

legacy_parser_impl.cc

Impl(http_parser_type type, void* data) : Impl(type) {
  parser_.data = data;
  settings_ = {
      [](http_parser* parser) -> int {
        auto* conn_impl = static_cast<ParserCallbacks*>(parser->data);
        return static_cast<int>(conn_impl->onMessageBegin());
      },
      [](http_parser* parser, const char* at, size_t length) -> int {
        auto* conn_impl = static_cast<ParserCallbacks*>(parser->data);
        return static_cast<int>(conn_impl->onUrl(at, length));
      },
      [](http_parser* parser, const char* at, size_t length) -> int {
        auto* conn_impl = static_cast<ParserCallbacks*>(parser->data);
        return static_cast<int>(conn_impl->onStatus(at, length));
      },
      [](http_parser* parser, const char* at, size_t length) -> int {
        auto* conn_impl = static_cast<ParserCallbacks*>(parser->data);
        return static_cast<int>(conn_impl->onHeaderField(at, length));
      },
      [](http_parser* parser, const char* at, size_t length) -> int {
        auto* conn_impl = static_cast<ParserCallbacks*>(parser->data);
        return static_cast<int>(conn_impl->onHeaderValue(at, length));
      },
      [](http_parser* parser) -> int {
        auto* conn_impl = static_cast<ParserCallbacks*>(parser->data);
        return static_cast<int>(conn_impl->onHeadersComplete());
      },
      [](http_parser* parser, const char* at, size_t length) -> int {
        static_cast<ParserCallbacks*>(parser->data)->bufferBody(at, length);
        return 0;
      },
      [](http_parser* parser) -> int {
        auto* conn_impl = static_cast<ParserCallbacks*>(parser->data);
        return static_cast<int>(conn_impl->onMessageComplete());
      },
      [](http_parser* parser) -> int {
        // A 0-byte chunk header is used to signal the end of the chunked body.
        // When this function is called, http-parser holds the size of the chunk in
        // parser->content_length. See
        // https://github.com/nodejs/http-parser/blob/v2.9.3/http_parser.h#L336
        const bool is_final_chunk = (parser->content_length == 0);
        static_cast<ParserCallbacks*>(parser->data)->onChunkHeader(is_final_chunk);
        return 0;
      },
      nullptr // on_chunk_complete
  };
}

 

즉 위 코드와 같이 LegacyHttpParserImpl 내부에는 Parsing 중간 중간 처리 결과를 반환할 수 있도록, settings_에 함수를 매핑했는데, 이때 ParserCallbacks에 정의된 규약에 따른 메소드를 호출하여 결과를 전달하는 것을 볼 수 있습니다.

 

 

---

 

5-2. Active Request

 

이번에는 Http1::ServerConnectionImpl이 보유하고 있는 속성 중 두번째인 active_request_ 에 대해서 알아보겠습니다.

 

codec_impl.h

  struct ActiveRequest : public Event::DeferredDeletable {
    ActiveRequest(ServerConnectionImpl& connection, StreamInfo::BytesMeterSharedPtr&& bytes_meter)
        : response_encoder_(connection, std::move(bytes_meter),
                            connection.codec_settings_.stream_error_on_invalid_http_message_) {}
    ~ActiveRequest() override = default;

    void dumpState(std::ostream& os, int indent_level) const;
    HeaderString request_url_;
    RequestDecoder* request_decoder_{};
    ResponseEncoderImpl response_encoder_;
    bool remote_complete_{};
  };

 

ActiveRequest는 위와 같이 Parser에 의해서 데이터를 Parsing 하는 과정에서 RequestDecoder, url, ResponseEncoder 등을 가지고 있는 구조체입니다. 해당 구조체를 통해서 Parsing 단계에서 connection 객체에 대한 작업 요청 및 url, encoder 지정 및 수행등을 담당합니다.

 

해당 자료구조가 사용되는 흐름을 보려면 사용자 요청을 처리하는 전단계를 살펴봐야하는데요. 이번 포스팅은 HttpConnectionManager의 특징에 대해서 살펴보기 때문에 ActiveRequest의 존재에 대해서만 이번 포스팅에서는 언급하고 해당 자료구조의 쓰임새는 다음 포스팅에서 보다 자세히 다루어보겠습니다.

 

---

6. Stream

 

이번에는 HttpConnectionManager가 관리하는 Stream 목록에 대해서 살펴보겠습니다. Stream은 Http 요청을 전달하는 하나의 흐름으로써, 하나의 Connection을 맺은 상태로 Http 요청을 전달하기 위해 여러 Stream을 생성할 수 있습니다. 따라서 이러한 개별 Stream들의 그룹을 관리하기 위해서 HttpConnectionManager 내부에는 streams_라는 List를 보유하고 있습니다.

 

 

streams_ List는 ActiveStream을 포함하고 있는데, 해당 Stream 내부에는 Http 요청에 필요한 필수적인 항목들이 포함되어있습니다. 그렇다면 ActiveStream은 언제 생성될까요?

 

 

codec_impl.cc

Status ServerConnectionImpl::onMessageBeginBase() {
  if (!resetStreamCalled()) {
    ASSERT(active_request_ == nullptr);
    active_request_ = std::make_unique<ActiveRequest>(*this, std::move(bytes_meter_before_stream_));
    if (resetStreamCalled()) {
      return codecClientError("cannot create new streams after calling reset");
    }
    active_request_->request_decoder_ = &callbacks_.newStream(active_request_->response_encoder_);

    // Check for pipelined request flood as we prepare to accept a new request.
    // Parse errors that happen prior to onMessageBegin result in stream termination, it is not
    // possible to overflow output buffers with early parse errors.
    RETURN_IF_ERROR(doFloodProtectionChecks());
  }
  return okStatus();
}

 

과정을 간략하게 살펴보면, 이전에 Parser에 대해서 살펴봤을 때, Client가 데이터 처리를 요청하면 이를 분석해서 특정 Event마다 통지한다고 설명했습니다. 이때 onMessageBegin 이벤트가 발생하면 ServerConnection 에서는 ActiveRequest를 먼저 생성하고 ActiveRequest 하위에 request_decoder_ 속성에 새로운 Stream을 만듭니다. 이때 만들어지는 Stream이 ActiveStream입니다.

 

이후 생성된 ActiveStream은 위 과정에서 표현되지는 않았지만 Parsing 과정이 진행되면서 지속적으로 참조되고 Parsing이 완료되는 시점에서 Stream 내부에 매핑된 정보에 의하여 Http 전송이 이루어지게됩니다.

 

 

이번에는 ActiveStream의 속성에 대해서 살펴보겠습니다. stream_id는 Stream 마다 생성되는 id로써, ActiveStream이 생성하는 당시에 임의의 값으로 지정됩니다. 

 

filter_manager_ 는 이전에 Http filter에 대해서 잠깐 살펴볼 때 등장한 프로퍼티명으로써, ActiveStream 내에 Http filter 생성 및 처리를 위임하는데 관여하는 프로퍼티입니다. 사용자 요청을 처리하는 과정에서 Http filter 처리가 필요한 순간에 해당 인스턴스가 사용됩니다.

 

그 다음 살펴볼 것은 timer입니다. ActiveStream 내부에는 다양한 Timer가 존재하는데, 해당 timer의 역할은 timer가 지정된 시간내에 요구하는 조건이 충족되지 않으면, 해당 연결을 해제하는 역할을 수행합니다. 이때 각각의 Timer는 Dispatcher 로부터 Timer를 생성받아 지정된 시간내에 조건이 충족되면 Timer를 Reset 하여 다시 지정된 시간만큼을 대기하며, 시간이 초괴되면 연결을 해제합니다. 개별 Timer의 역할을 살펴보면 다음과 같습니다.

 

타이머	기능
stream_idle_timer_	해당 Stream이 연결되고 어떠한 활동도 일어나지 않았을 때, 지정된 idle 시간을 초과하면 Stream 해제
request_timer_	Stream이 생성되고 난 이후 Request를 시작하고 응답이 올 때까지 대기시간을 의미하며, 지정된 시간 초과하면 Stream 해제
request_header_timer_	Downstream에서 header를 전송하고 나서 이에 대한 응답이 올 때까지의 대기시간을 의미하며, 지정된 시간 초과하면 Stream 해제
max_stream_duration_timer_	Stream이 생성되고 종료될 때까지 지속할 수 있는 시간을 의미하며, 지정된 시간을 초과하면 Stream 해제

 

상기 Timer와 관련해서는 envoy 공식 문서에서 확인할 수 있으며, 적정한 값 설정을 통해서 Stream이 생성되고 무한정 대기하지 않도록 처리할 수 있습니다.

 

---

7. Timer

 

HttpConnectionManager에서 마지막으로 살펴볼 것은 내부 프로퍼티에 존재하는 Timer입니다. 방금전에 개별 Stream 내부에서 여러 Timer 들을 살펴봤었는데, HttpConnectionManager 또한 Timer를 보유하고 있습니다.

 

 

HttpConnectionManager가 보유한 주요 Timer는 위와같이 2개입니다. 이는 HttpConnectionManager가 생성하는 단계에서 Dispatcher에게 요청하여 Timer를 생성합니다. 

 

connection_idle_timer_의 경우는 HttpConnectionManager가 생성된 이후 어떠한 Stream이 생성되지 않았을 때, idle 시간을 얼마나 부여할지를 측정하는 Timer입니다. 따라서 해당 시간 동안 Stream 연결이 이루어지지 않는다면, 연결을 해제합니다.

 

반면 connection_duration_timer_는 HttpConnectionManager가 생성되고 모든 처리가 완료될 때까지 즉 사용자의 연결 요청을 완수하는데 걸리는 데드라인을 측정하는 Timer입니다. 따라서 해당 시간 동안 모든 요청이 완료되지 않는다면, 연결을 해제합니다.

 

 

---

8. 마치며

 

이번 포스팅을 통해서 HttpConnectionManager가 보유하고 있는 속성에 대해서 알아봤습니다. 다만 이번 포스팅만으로는 HttpConnectionManager가 보유하고 있는 컴포넌트가 어떻게 상호작용하여 사용자의 요청을 처리하는지 완벽하게 이해하기는 힘듭니다. 이 부분은 이 다음 포스팅에서 진행되는 Envoy 프록시 연결 과정을 관찰하면서 조금 더 자세하게 살펴보겠습니다. 

1. 서론

 

이번 포스팅은 지난번에 이어서 Envoy의 주요 컴포넌트 중 하나인 Listener와 Listener를 관리하는 Listener Manager에 대해서 살펴보고자 합니다. 

 

---

 

 

2. Listener Manager 개요

 

Listener Manager는 단어에서 유추할 수 있듯이 Listener를 관리하는 역할을 수행합니다. 즉 Listener를 생성하고 관리합니다. 두 번째 역할을 LDS를 생성하고 관리하는 역할을 겸합니다. 이번 포스팅에서는 이에 대해서 하나씩 살펴보겠습니다. 먼저 살펴볼 것은 Listener 생성 및 관리 기능입니다.

 

Listener Manager는 Listener를 생성 후 관리한다고 설명했습니다. 여기서 꼭 기억해야할 특징은 Listener Manager가 생성한 Listener는 단일 쓰레드 기반으로 동작하지 않는다는 점입니다. 그렇다면 Envoy는 왜 Listener를 단일 쓰레드로 동작하지 않고 멀티 쓰레드로 동작시켰을까요? 이를 이해하기 위해서 사례를 통해 먼저 사용자의 요청을 처리하는 쓰레드가 하나일 경우 야기되는 문제점을 살펴보겠습니다.

 

 

 

envoy에 여러 API 요청이 들어오게되면, 그 중 하나가 쓰레드를 선점하게되고 해당 요청을 처리할 것입니다. 이때 만약 다른 API 요청이 전달된다면 어떻게 될까요? 쓰레드를 선점하고 있다면, 해당 쓰레드가 처리하는 작업 요청이 끝날 때까지 다른 요청은 대기해야할 것입니다. 따라서 이 경우 병렬성이 매우 떨어지게 됩니다. 특히나 다른 API 요청을 처리하는 Listener의 경우 미리 선점한 Listener의 작업에 의해 해당 API 요청 또한 대기하는 이슈가 발생합니다. 따라서 단일 쓰레드로 이를 처리하기에는 성능 이슈가 필연적으로 발생합니다.

 

따라서 envoy에서는 API 요청을 처리하는 쓰레드를 하나가 아니라 여러개를 구성하도록 설계되었습니다. 또한 생성된 Worker 쓰레드는 모든 Listener와 연결 되어있도록 구성되었습니다.

 

 

 

envoy는 기동 당시 --concurrency 인자를 통해 생성할 Worker 쓰레드의 개수를 지정할 수 있습니다. 만약 위 그림과 같이 3으로 지정하였다면, 내부적으로는 3개의 Worker 쓰레드를 Listener Manager가 생성합니다. 또한 만약 Config에 지정된 Listener가 2개라면 위와 같이 2개의 Listener가 생성됩니다.

 

Listener와 Worker 쓰레드가 모두 생성되고나면, 모든 Worker 쓰레드는 모든 Listener를 Listen 하도록 합니다. 그렇다면 모든 Worker 쓰레드가 모든 Listener를 Listen 하는 것은 어떤 것을 의미할까요?

 

 

 

만약 특정 Listener 별로 Worker 쓰레드를 할당할 수 있다면, 위와 같이 부하가 집중되는 Listener에 쓰레드를 많이 부여할 수 있습니다. 가령 위 그림과 같이 Listener0에 명시적으로 더 많은 쓰레드를 할당함으로써 Listener0 요청에 빠른 처리가 가능합니다.

 

하지만 만약 위와 같이 지정한 상황에서 갑자기 Listener 1쪽에 트래픽이 몰릴 경우에는 해당 Listener에 지정된 쓰레드는 하나이기 때문에 탄력적으로 대응할 수 없습니다.

 

따라서 envoy에서는 Active Listener 별로 영역을 정해서 Worker 쓰레드를 샤딩하지 않도록 구성되어있습니다. 즉 사용자 Connection이 요청되었을 때, 모든 Worker 쓰레드가 모든 Listener를 Listen하고 있기 때문에 어떠한 Worker 쓰레드에 사용자 Connection을 할당하더라도 문제가 없습니다. 또한 어떤 Worker 쓰레드에 사용자 요청을 할당하는 지에 대한 결정은 Kernel이 Socket을 Accept한 다음 결정합니다. 결국 Worker 쓰레드에 대한 Connection 할당 권한은 OS에 있습니다.

 

Connection이 Accept되어 Worker 쓰레드에 Socket이 배정되면, 해당 Socket은 이후에 Worker 쓰레드를 벗어나지 않습니다. 따라서 이후 제공되는 서비스는 최초 연결 시점에 할당된 Worker 쓰레드에서 이루어집니다.

 

지금까지 Listener Manager에서 생성되는 Worker 쓰레드에 대해서 살펴봤습니다. 그렇다면 Worker 쓰레드 생성 외에 Listener Manager는 어떤 역할을 하고 있을까요? Listener Manager가 담당하는 역할에 대해서 살펴보겠습니다.

 

---

3. Listener 생성

 

 

 

Listener Manager의 첫 번째 역할은 Listener 생성입니다. config 파일에 위치한 static_resources 정보에 포함된 Listener 정보를 살펴보고 그에따른 Listener를 생성합니다.

 

configuration_impl.cc

void MainImpl::initialize(const envoy::config::bootstrap::v3::Bootstrap& bootstrap,
                          Instance& server,
                          Upstream::ClusterManagerFactory& cluster_manager_factory) {
  ...(중략)...

  const auto& listeners = bootstrap.static_resources().listeners();
  ENVOY_LOG(info, "loading {} listener(s)", listeners.size());
  for (ssize_t i = 0; i < listeners.size(); i++) {
    ENVOY_LOG(debug, "listener #{}:", i);
    server.listenerManager().addOrUpdateListener(listeners[i], "", false);
  }

  ...(중략)...
}

 

이는 위 코드와 같이 Config 내용을 파싱한 이후 담겨있는 bootstrap 구조에서 static listener 정보를 추출한 결과를 토대로 Listener Manager에게 Listener를 생성하라고 요구하는 것을 볼 수 있습니다.

 

이때 생성되는 Listener는 어떤 과정을 거치며 내부적으로 어떤 속성을 가지고 있을까요?

 

 

Envoy 관련 첫 포스팅에서 Client 요청이 전달되면, 위 그림과 같은 트래픽 흐름이 전달된다고 설명했습니다. 이때 Listener와 연결된 Listener Filters와 Network Filters는 Listener가 관리합니다. 따라서 Listener의 역할 중 하나는 Listener Filters와 Filter Chains(Network Filters)를 생성하는 것입니다.

 

static_resources:
  listeners:
  - name: listener_https
    address:
      socket_address:
        protocol: TCP
        address: 0.0.0.0
        port_value: 80
    listener_filters:
    - name: "envoy.filters.listener.tls_inspector"
      typed_config:
        "@type": type.googleapis.com/envoy.extensions.filters.listener.tls_inspector.v3.TlsInspector
    - name: "envoy.filters.listener.http_inspector"
      typed_config:
        "@type": type.googleapis.com/envoy.extensions.filters.listener.http_inspector.v3.HttpInspector

 

 

가령 위와 같이 envoy config 파일에 listener_filters 기입되었다고 가정하면, Listener를 생성하는 단계에서 해당 필터들을 파악하고 생성하는 역할을 수행해야합니다.

 

 

 

이를 위해서 Listener 내부에는 Listener Filters를 생성하는 listener_filter_factories와 Filter Chains를 생성하는 filter_chain_manager가 존재합니다.

 

 

Listener Filters는 Worker 쓰레드에 바인딩된 소켓마다 해당 Filter들이 생성되기 때문에, Filter를 생성하여 Listener가 보관하지 않고 Filter를 생성할 수 있는 Callback 메소드들의 Pointer를 Vector Container에 저장하고 있습니다. 따라서 Client가 접속을 요청했을 경우 Worker 쓰레드에 소켓이 할당되면, 이후 listener_filter_factories에 지정된 Callback 메소드들을 순회하면서 사용자에게 적합한 Filter 목록을 만들어줍니다.

 

 

 

Filter Chains(Network Filters)는 L3/L4를 담당하는 Filter로써 Listener Config 설정에 존재하는 Filter 목록들을 참조하여 내부적으로 여러 Filter Factory를 만듭니다. 

 

    filter_chains:
    - filter_chain_match:
        server_names: ["acme.com"]
      transport_socket:
        name: envoy.transport_sockets.tls
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.transport_sockets.tls.v3.DownstreamTlsContext
          common_tls_context:
            tls_certificates:
            - certificate_chain: {filename: "certs/servercert.pem"}
              private_key: {filename: "certs/serverkey.pem"}        
      filters:
      - name: envoy.filters.network.http_connection_manager
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
          stat_prefix: ingress_http
          use_remote_address: true
          http2_protocol_options:
            max_concurrent_streams: 100
          route_config:
            name: local_route
            virtual_hosts:
            - name: local_service
              domains: ["acme.com"]
              routes:
              - match:
                  path: "/foo"
                route:
                  cluster: some_service
          http_filters:
          - name: envoy.filters.http.router
            typed_config:
              "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router

 

가령 위와같이 Listener 하위에 filter_chains를 정의했다고 가정해봅시다. 그러면 Listener를 생성하는 과정에서 해당 Config 내용을 확인하고 Transport Socket Factory와 Network Filter Factory를 구성합니다.

 

이때 filter_chain_manager의 역할은 사용자가 입력한 Filter Config 내용이 정상인지 검증을 하고 Filter 목록을 빠르게 찾기 위한 자료구조 및 Filter 생성 Callback을 생성하여 저장합니다. 이때 생성되는 Callback은 Listener와 마찬가지로 사용자가 실제 접속해서 Socket이 만들어졌을 때 해당 Socket에서 Network Filters 생성을 할때 사용됩니다.

 

 

listener_manager_impl.cc

    for (auto& address : listener.addresses()) {
      listener.addSocketFactory(std::make_unique<ListenSocketFactoryImpl>(
          factory_, address, socket_type, listener.listenSocketOptions(), listener.name(),
          listener.tcpBacklogSize(), bind_type, creation_options, server_.options().concurrency()));
    }

 

지금까지 listener 생성 과정에서 Listener Filters와 Filter Chains(Network Filters)가 생성되는 것을 살펴봤습니다. 그 밖에 중요하게 살펴봐야할 점은 Listener가 외부 요청으로부터 Accept 하기 위한 Socket을 생성한다는 점입니다. 그리고 생성되는 Socket의 개수는 --concurrency로 전달된 인자 개수 만큼 생성됩니다.

 

그렇다면 왜 --concurrency 만큼 Socket을 생성할까요?

 

 

그 이유는 향후 Listener 별로 Worker 쓰레드에게 모두 바인딩을 수행하는데, 이때 Listener에서 생성한 Socket을 개별 쓰레드에서 참조가 가능하도록 하기 위함입니다.

 

listener_impl.cc

Network::SocketSharedPtr socket = factory.createListenSocket(
    local_address_, socket_type, options_, bind_type_, socket_creation_options_, worker_index);

 

따라서 Listener 생성 과정에서 Listner Component Factory가 Socket을 생성하여 개별 Worker 쓰레드에서 향후 참조할 수 있도록 사전 작업을 수행합니다.

---

3-1 Worker 생성 및 Listener 바인드

 

listener_manager_impl.cc

  for (uint32_t i = 0; i < server.options().concurrency(); i++) {
    workers_.emplace_back(
        worker_factory.createWorker(i, server.overloadManager(), absl::StrCat("worker_", i)));
  }

 

Listener Manager가 생성될 때, 생성자 코드 내부에서는 위와 같이 전달받은 worker_factory를 통해서 Worker 쓰레드를 생성하는 것을 볼 수 있습니다. 그리고 생성하는 쓰레드의 개수는 인자로 전달된 --concurrency에 기반하고 있습니다.

 

listener_manager_impl.cc

void ListenerManagerImpl::startWorkers(GuardDog& guard_dog, std::function<void()> callback) {
  ...(중략)...
  uint32_t i = 0;

  ...(중략)...
  
  for (auto listener_it = active_listeners_.begin(); listener_it != active_listeners_.end();) {
    auto& listener = *listener_it;
    listener_it++;

   ...(중략)...
    for (const auto& worker : workers_) {
      addListenerToWorker(*worker, absl::nullopt, *listener,
                          [this, listeners_pending_init, callback]() {
                            if (--(*listeners_pending_init) == 0) {
                              stats_.workers_started_.set(1);
                              callback();
                            }
                          });
    }
  }
  for (const auto& worker : workers_) {
    ...(중략)...
    worker->start(guard_dog, worker_started_running);
    ...(중략)...
    i++;
  }

  ...(후략)...
}

 

이후 Envoy가 기동 될때, 위 코드와 같이 등록된 모든 active_listeners를 순회하면서, 모든 worker 쓰레드에 bind 하는 것을 볼 수 있습니다. 해당 내용에 대해서 조금 더 자세히 살펴보면 다음과 같습니다.

 

 

이전에 Listener 생성 과정을 살펴보면서, 위 그림과 같이 Worker 쓰레드 개수 만큼 개별 Listener에서 Socket을 생성함을 확인했습니다. 이때 생성된 Socket을 Worker 쓰레드에서 사용하기 위해서는 소켓을 각 쓰레드별로 바인딩하는 작업을 수행해야합니다.

 

 

connection_handler_impl.cc

for (auto& socket_factory : config.listenSocketFactories()) {
  auto address = socket_factory->localAddress();
  // worker_index_ doesn't have a value on the main thread for the admin server.
  details->addActiveListener(
      config, address, listener_reject_fraction_, disable_listeners_,
      std::make_unique<ActiveTcpListener>(
          *this, config, runtime,
          socket_factory->getListenSocket(worker_index_.has_value() ? *worker_index_ : 0),
          address, config.connectionBalancer(*address)));
}

 

따라서 개별 쓰레드에서는 위 코드를 수행하면서, ActiveTcpListener를 생성하는 것을 확인할 수 있습니다. 위 코드에서 주목할 부분은 socket_factory(Listener)로부터 getListenSocket 메소드를 호출하면서 인자로 Worker 쓰레드의 index 번호를 넘기는 것을 확인할 수 있습니다.

 

listener_impl.cc

Network::SocketSharedPtr ListenSocketFactoryImpl::getListenSocket(uint32_t worker_index) {  
  ASSERT(worker_index < sockets_.size() && sockets_[worker_index] != nullptr);
  return sockets_[worker_index];
}

 

이때 Listener 내부에서는 Worker 인덱스에 해당하는 Socket을 반환함으로써 Worker 쓰레드에서 해당 Listener를 참조할 수 있도록 합니다.

 

---

 

4. LDS 관리

 

Listener Manager의 두 번째 역할은 LDS를 통한 dynamic resource 처리입니다. 이때 LDS의 생성은 Listener Component Factory에서 수행하며, 생성된 LDS에서 Listener Manager를 참조하도록 하여 LDS 정보가 수신되었을 때, Listener Manager가 이를 처리할 수 있도록 연관관계가 맺어져있습니다.

 

LDS 생성 및 처리 과정을 보면 아래와 같습니다.

 

 

 

 

1. Listener Component Factory로부터 LDS를 생성합니다.

 

2. Listener Manager를 LDS에서 참조하도록 합니다.

 

3. xDS API 수행을 위해 Cluster Manager로부터 Subscription Factory 인스턴스를 요청합니다.

 

4. Subscription Factory로부터 Subscription 생성을 요청합니다.

 

5. Subscription Factory에서는 gRPC 통신이 가능하도록 설정 후 Subscription을 발급합니다.

 

6. gRPC Mux를 통해 Management Server와 통신을 수행합니다.

 

7. LDS api_type 변경 이벤트 통지 시, 해당 Subscription에 알림을 전달합니다.

 

8. Subscription내에 매핑된 LDS Callback을 호출합니다.

 

9. LDS에서는 Listener Manager를 참조하여 변경 사항에 맞추어 Listener 정보를 동기화 시킵니다.

 

위와 같은 과정을 거쳐 Listener Manager에서는 LDS 요청에 맞추어 최신 정보를 동기화시킵니다. 해당 과정의 대부분은 이전 Cluster Manager 설명 포스팅을 통해서 자세하게 소개하였으므로 해당 내용을 이해했다면, 위 과정의 흐름의 진행은 자연스러울 것입니다. 혹시 기억이 잘 안나신다면, 이전 포스팅 정독을 통해 관련 내용을 복기해보시는 것을 추천드립니다.

 

---

5. 마치며

 

지금까지 Listener Manager를 기동하면서 Worker 쓰레드 및 Listener 생성하여 이를 바인딩하는 역할과 LDS 처리역할에 대해서 살펴봤습니다. 또한 개별 Listener에서는 Configuration에 따라서 여러 Filter를 생성하기 위한 Callback 리스트를 관리하는 것을 살펴볼 수 있었습니다. 이 과정에서 생성된 정보등을 통해 향후 Client에서 Listener에 접속을 요청하면 이를 전달받아 후속 처리를 진행할 수 있습니다. 다만 이번 포스팅에서는 해당 내용에 대해서는 다루지 않으며, 이 다음 포스팅에서 조금 더 자세하게 접속 요청이 어떻게 처리되는지를 살펴보겠습니다.

1. 서론

 

이전 포스팅을 통해서 Envoy Cluster Manager의 역할에 대해서 살펴봤습니다. Cluster Manager는 Cluster 관리 뿐만 아니라 Multiplexer 및 Subscription 생성에 관여합니다. 이번 포스팅에서는 Cluster Manager에서 관리하는 Cluster 내부에서 지정할 수 있는 Load Balancer 종류와 설정 방법에 대해서 다루고자합니다. 이번 포스팅 내용은 Envoy에 대한 설명 뿐만 아니라 Load Balancer 종류에 대한 알고리즘에 대한 개념 설명 위주로 알아보면서 어떻게 envoy가 Load Balancing을 처리하는지에 대해서 살펴보고자 합니다.

 

---

2. Envoy Loadbalancer

 

 

cluster_manager_impl.cc

ClusterManagerImpl::ThreadLocalClusterManagerImpl::ClusterEntry::ClusterEntry(
    ThreadLocalClusterManagerImpl& parent, ClusterInfoConstSharedPtr cluster,
    const LoadBalancerFactorySharedPtr& lb_factory)
    : parent_(parent), lb_factory_(lb_factory), cluster_info_(cluster) {
  priority_set_.getOrCreateHostSet(0);

  // TODO(mattklein123): Consider converting other LBs over to thread local. All of them could
  // benefit given the healthy panic, locality, and priority calculations that take place.
  if (cluster->lbSubsetInfo().isEnabled()) {
    lb_ = std::make_unique<SubsetLoadBalancer>(
        cluster->lbType(), priority_set_, parent_.local_priority_set_, cluster->stats(),
        cluster->statsScope(), parent.parent_.runtime_, parent.parent_.random_,
        cluster->lbSubsetInfo(), cluster->lbRingHashConfig(), cluster->lbMaglevConfig(),
        cluster->lbRoundRobinConfig(), cluster->lbLeastRequestConfig(), cluster->lbConfig(),
        parent_.thread_local_dispatcher_.timeSource());
  } else {
    switch (cluster->lbType()) {
    case LoadBalancerType::LeastRequest: {
      ASSERT(lb_factory_ == nullptr);
      lb_ = std::make_unique<LeastRequestLoadBalancer>(
          priority_set_, parent_.local_priority_set_, cluster->stats(), parent.parent_.runtime_,
          parent.parent_.random_, cluster->lbConfig(), cluster->lbLeastRequestConfig(),
          parent.thread_local_dispatcher_.timeSource());
      break;
    }
    case LoadBalancerType::Random: {
      ASSERT(lb_factory_ == nullptr);
      lb_ = std::make_unique<RandomLoadBalancer>(priority_set_, parent_.local_priority_set_,
                                                 cluster->stats(), parent.parent_.runtime_,
                                                 parent.parent_.random_, cluster->lbConfig());
      break;
    }
    case LoadBalancerType::RoundRobin: {
      ASSERT(lb_factory_ == nullptr);
      lb_ = std::make_unique<RoundRobinLoadBalancer>(
          priority_set_, parent_.local_priority_set_, cluster->stats(), parent.parent_.runtime_,
          parent.parent_.random_, cluster->lbConfig(), cluster->lbRoundRobinConfig(),
          parent.thread_local_dispatcher_.timeSource());
      break;
    }
    case LoadBalancerType::ClusterProvided:
    case LoadBalancerType::LoadBalancingPolicyConfig:
    case LoadBalancerType::RingHash:
    case LoadBalancerType::Maglev:
    case LoadBalancerType::OriginalDst: {
      ASSERT(lb_factory_ != nullptr);
      lb_ = lb_factory_->create();
      break;
    }
    }
  }
}

 

Envoy에서 Load Balancer는 ClusterEntry 내부에 존재하는 프로퍼티로써, ClusterEntry 생성 당시에 같이 생성됩니다. 이는 위 코드에서도 확인할 수 있으며, 생성자 내부에서 지정된 LoadBalancerType에 따라서 각기 다른 LoadBalancer가 지정됨을 확인할 수 있습니다.

 

그렇다면 Load Balancer 생성 이후 해당 과정은 어느 단계에서 이루어질까요?

 

 

 

이전 포스팅에서 Router로 부터 Connection Pool 획득 과정에 대해서 다루었습니다. 이 과정에서 Connection Pool을 획득하려면 먼저 host가 지정되어야합니다. 따라서 위 단계에서 4번째 단계 즉 Router가 Cluster 정보를 얻은 다음 Connection Pool을 요청하는 단계에서 Cluster Entry 내부에서는 가장먼저 Load Balancer에게 요청하여 어떤 host에 사용자 요청을 할당할 것인지를 요청합니다. 그리고 해당 host 정보를 토대로 Cluster Manager에게 해당 host에 대한 Connection Pool 요청을 수행합니다.

 

그렇다면 envoy에서 제공하는 Load Balancer는 어떤 종류가 있으며, 각 종류별로 어떻게 동작할까요? 이에 대해서 개념적으로 살펴보겠습니다.

 

 

---

3. Latency 관점

 

 

 

 

하나의 Cluster 안에 여러개의 Host가 매핑되어있다고 가정해봅시다. 그러면 사용자가 연결 요청을 시도하면, Cluster 내부에 있는 여러 Host 중 하나를 Load Balancer가 선택할 것입니다. 이때 어떤 Host를 선택하는 것이 효율적일까요? 특정 Host에만 요청을 집중한다면, 결국 해당 Host의 처리 능력이 떨어지게되고 결국에는 해당 Host에 장애가 발생할 것입니다. 이를 처리하기 위해 응답속도 관점에서 고려할 수 있는 Load Balancing 알고리즘 종류에 대해 알아보겠습니다.

 

---

 

3-1 Random

 

첫 번째로 고려해볼 수 있는 것은 Random 알고리즘입니다. Random 알고리즘은 말 그대로 Load를 분배할 때, 임의로 아무 host에 배정함을 의미합니다. 따라서 해당 알고리즘은 특별할 것이 없으며, 굉장히 단순하고 구현하기가 쉽습니다. 즉 이말은 Load Balancing을 수행하는데 있어 overhead가 적다는 것을 의미합니다.

 

하지만 Random 알고리즘은 무작위로 부하를 분배하는 알고리즘이기 때문에 트래픽이 몰렸을 때, 완벽히 균등하게 모든 Host에 부하가 분배되지 못할 수 있습니다. 즉 운이 좋지 않으면 특정 노드에 부하가 몰릴 수 있음을 의미합니다. 따라서 해당 알고리즘은 트래픽이 적으며, Load Balancing 수행에 대한 overhead를 줄이고 싶을 때 사용하면 좋습니다.

 

envoy에서는 아래와 같이 lb_policy를 random으로 지정함으로써 Random 알고리즘을 사용하여 부하를 분산할 수 있습니다.

 

static_resources:
  clusters:
  - name: my_service
    connect_timeout: 0.25s
    type: strict_dns
    lb_policy: random
    load_assignment:
      cluster_name: my_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8080
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8081
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8082

 

 

---

 

3-2. Round Robin

 

두 번째로 살펴볼 것은 Round Robin 알고리즘입니다. 

 

 

Round Robin 알고리즘은 보편적으로 많이 사용되는 Load Balancing 알고리즘으로 부하를 모든 Host 서버를 순환하면서 균등하게 분배하는 방식을 의미합니다. 가령 위 그림과 같이 6대의 Client 요청을 분배한다고 가정해봅시다. 그러면 3대의 B/E 서버에 순차적으로 하나씩 Client 요청을 분배합니다. 따라서 첫 요청은 Host 1에게 두번째 요청은 Host 2에게 분배하는 등 요청이 들어올 때마다 균등하게 분배합니다. 해당 알고리즘은 Random과 더불어 구현이 굉장히 단순합니다. Load Balancer 입장에서는 연결된 Host 중에서 이전에 분배했던 Host가 누구였는지 기억한다면, 그 다음에 분배해야될 대상이 누군지 알 수 있기 때문입니다.

 

static_resources:
  clusters:
  - name: my_service
    connect_timeout: 0.25s
    type: strict_dns
    lb_policy: round_robin
    load_assignment:
      cluster_name: my_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8080
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8081
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8082

 

envoy에서는 위와같이 lb_policy를 round_robin으로 지정하여 Round Robin 알고리즘으로 부하 분산을 수행할 수 있습니다.

 

3-2-1 Weighted Round Robin

 

Round Robin 알고리즘은 부하가 몰릴 경우에도 순차적으로 Host에게 부하를 분배하기 때문에 효율적으로 동작합니다. 하지만 여기에는 전제조건이 붙습니다. 그것은 모든 Host가 처리 시간이 동일함을 가정하였을 경우입니다. 이 경우 모든 서버의 하드웨어 스펙이 동일하지 않는다면 어떤 일이 벌어질까요?

 

 

가령 Host 1의 하드웨어 사양이 나머지 두 개의 서버보다 2배 이상 좋다고 가정해봅시다. 이 경우 부하를 위와 같이 동일하게 분산하면, Host 1의 경우는 사용자의 요청을 다른 서버보다 빠르게 처리하여 idle 상태일 동안 나머지 서버에서는 여전히 사용자의 요청을 처리하는 경우가 발생합니다. 이 경우 요청 분배를 Host 1에 조금 더 할 수 있다면, 처리 속도를 높일 수 있음에도 불구하고 일반적인 Round Robin의 알고리즘의 경우에는 이를 처리할 수 없습니다. 따라서 이러한 하드웨어 성능 차이를 고려하기 위해 추가된 것이 가중치(Weight)입니다.

 

static_resources:
  clusters:
  - name: my_service
    connect_timeout: 0.25s
    type: strict_dns
    lb_policy: round_robin
    load_assignment:
      cluster_name: my_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8080
          load_balancing_weight: 2      
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8081
          load_balancing_weight: 1      
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8082
          load_balancing_weight: 3

 

가중치는 위와 같이 각각의 endpoint 별로 load_balancing_weight 속성을 추가할 수 있습니다. 위 예시는 해당 서버의 처리 능력을 고려하여 운영자가 임의로 매핑한 값이라고 가정해봅시다.

 

 

 

 

이 경우 Round-Robin 방식으로 동작하지만, 각각의 Server 별로 가중치가 존재하기 때문에 이를 고려하여 위와같이 부하를 분배합니다. 즉 이전과 동일하게 6개의 요청을 전달받는다고 가정한다면, Host 1은 전체 부하의 33% 정도를 할당 받으며, Host 2는 17%, Host 3의 경우는 50%의 부하를 분배받는 것을 볼 수 있습니다. 

 

---

 

3-3 Least Request

 

지금까지 Random과 Round-Robin 알고리즘에 대해 살펴봤습니다. 위 두 알고리즘은 구현이 굉장히 쉽다는 장점이 존재합니다. 특히 Round-Robin의 경우는 부하가 집중될 때도 적은 overhead를 가지면서 적절히 부하를 분배할 수 있습니다. 하지만 Round-Robin에 가중치까지 적용하였다 할지라도 여기에는 이만큼 처리할 수 있다는 가정을 기반으로 처리하는 것이지 실제 처리량을 기반으로 Load Balancer가 부하를 처리하지는 않습니다. 

 

 

가령 위 그림과 같이 Host 서버의 하드웨어 스펙이 모두 동일하다고 가정해봅시다. 또한 해당 host들이 매핑된 실제 서버에는 하나의 B/E만 존재하는 것이 아니라 여러 다른 app이 존재한다고 가정해봅시다. 이는 서버가 k8s 클러스터 등으로 구성되어있다면 충분히 가능한 시나리오일 것입니다.

 

이 경우 Round-Robin으로 분배하거나 Weighted Round-Robin을 적용하기 쉽지 않을 것입니다. 따라서 이 경우에는 실제 응답 요청을 처리하는 Connection 개수를 실시간으로 모니터링하면서 부하를 얼마나 감당할 수 있는지를 살펴보고 Load Balancer가 부하가 적은 쪽으로 전달하는 것이 효율적입니다.

 

 

 

가령 Least Request 알고리즘 환경에서는 위와 같이 현재 6개 Host에 부하를 분배한 상황에서 가령 Host 2에서 2번 요청에 대한 처리가 완료되어 Connection을 해제하였을 때, 그 다음 7번 요청을 Host 2에게 분배하는 방식입니다. 따라서 이를 지원하는 Load Balancer 타입에는 현재 연결된 active_requests가 몇 개인지를 관리하고 있습니다.

 

또한 Least Request 알고리즘을 사용하는 환경에서도 서버마다 하드웨어 스펙등이 다를 수 있기 때문에 가중치를 적용할 수 있습니다. 

 

static_resources:
  clusters:
  - name: my_service
    connect_timeout: 0.25s
    type: strict_dns
    lb_policy: least_request
    load_assignment:
      cluster_name: my_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8080
            load_balancing_weight: 2
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8081
            load_balancing_weight: 1
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8082
            load_balancing_weight: 3

 

가령 위와 같이 envoy에서 지정했다면, 내부적으로는 아래와 같은 공식을 적용하여 가중치 값이 높은 쪽으로 부하가 분산됩니다. 참고로 해당 공식은 envoy 공식 문서에서 발췌하였습니다.

 

weight = load_balancing_weight / (active_requests + 1)^active_request_bias

 

참고로 active_request_bias 값은 load balancing 과정에서 active_requests에 대하여 우선 순위를 조정할 수 있는 값으로써 envoy config를 통해 지정할 수 있으며, 기본 값은 1.0입니다. 

 

---

 

4. 가용성 관점

 

지금까지 세 가지 알고리즘에 대해서 살펴봤습니다. 세 가지 알고리즘의 특성은 구현이 용이하며, 합리적인 부하 분산을 균등하게 해준다는 점에서 이점이 존재합니다. 하지만 다음과 같은 상황에서는 어떨까요?

 

 

 

가령 위 그림과 같이 3개의 Host가 있고 각각의 Host에는 별개의 Cache 레이어가 존재하는 시스템이라고 가정해봅시다. 해당 시스템은 시스템 규모가 너무 커서 사용자별로 특정 데이터를 샤딩하여 각각의 Host 별개 Cache에 보관한다고 가정해봅시다. 이때 Cache Hit를 높이기 위해서는 특정 사용자 요청은 특정 Host로 보내져야 효율적일 것입니다.

 

하지만 Random, Round-Robin, Least Request 알고리즘은 이러한 특성을 보장해주지 않습니다. 그 이유는 해당 알고리즘은 해당 시점 전달된 트래픽 부하 분산에 초점이 맞추어져있기 때문입니다. 따라서 이전 요청 응답 전달 이후 같은 사용자 요청이 전달되었을 때, 각기 다른 Host로 전달될 가능성이 매우 높습니다. 그리고 어디로 배정될지 운영자 입장에서는 정확히 알 수 없기 때문에 운영상 불확실성을 가지고 있습니다. 이러한 불확실성은 대규모 시스템을 운영하는 입장에서는 엄청난 부담이 아닐 수 없습니다.

 

그렇다면 이러한 특성을 고려한다면, 어떻게 부하 분산하는 것이 좋을까요?

 

 

일반적으로 많이 사용하는 방법은 Hash 함수와 Modular를 활용하는 것입니다. 즉 위 그림과 같이 Host 목록에 대해서 특정 Key(사용자 Id) 등을 활용한 Hash 값을 토대로 Modular 연산을 적용하면, 특정 사용자 요청에 대해서는 매번 같은 결과 값이 나오게되니 아무리 요청을 많이 할지라도 특정 Host로 전달됨이 보장됩니다. 참고로 Modular 연산을 적용하는 이유는 아무리 Hash 값이 다르더라도 서버의 갯수만큼으로 Modular 연산을 수행하면, 서버 갯수 이내만큼의 값을 얻을 수 있기 때문입니다.

 

가령 Modular 연산 결과가 0이면 Host 1, 1이면 Host 2, 2이면 Host 3으로 할당된다고 가정했을 때, 사용자1의 Hash 값이 1이 나왔다면, 매번 Host 2로 전달이 보장됩니다.

 

위와 같이 살펴봤을 경우 Hash 와 Modular 연산 적용 또한 부하 부하를 적절히 분산하는데, 도움이 되는 것으로 보입니다. 하지만 여기에는 다음과 같은 문제가 있습니다.

 

 

가령 위와 같이 Host 1 ~ 3에 Connection과 사용자 데이터들이 Cache에 할당되어있다고 가정해봅시다. 이때 해당 비즈니스가 트래픽이 급격히 올라 Scale Out을 해야하는 상황이 발생하면 어떻게 해야할까요?

 

위 그림을 살펴보면 기존 Modular 연산은 Host 개수가 3개기 때문에 1 ~ 3 Host에만 할당할 수 있을 뿐 Host 4에는 할당할 수 없습니다. 따라서 Host 4를 수용하기 위해서는 Host 4를 Host 리스트에 포함하고 나머지 연산 또한 4로 변경해야할 것입니다.

 

하지만 이 경우 지금까지 분배되었던 Connection과 데이터 저장에 커다란 이슈가 발생합니다. 

 

 

 

기존에 Hash 값이 11인 경우에는 원래대로라면 Host 3에 지속 라우팅이 되었어야 합니다. 하지만 연산 수식이 달라졌기 때문에, Host 4가 추가된 이후에는 수식의 결과에 따라(11%4=3) Host 4에 라우팅 될 것입니다. 따라서 Scale out 이후에도 서비스 안정성을 유지하기 위해서는 기존 Host 들에 연결된 Connection과 Cache된 데이터들에 대한 재분배가 필요합니다.

 

그렇다면 어느정도 데이터가 재분배가 일어나야할까요? 위 사례를 기반으로 데이터 재분배가 일어난 다음 상황을 살펴보겠습니다. 이해를 돕기위해 데이터 재분배가 발생한 항목은 파란색 음영으로 표시하였습니다.

 

 

 

위 그림의 결과를 통해서 확인할 수 있는 사실은 Host 4 추가 이후 기존 데이터까지 포함해서 총 75%의 데이터 이동이 필요하며, 그만큼 Connection 또한 재분배가 일어난다는 점입니다. 이는 Scale out 수행해야하는 운영자 입장에서는 엄청난 부담으로 작용할 수 있습니다. 그 이유는 데이터 이동과 Connection을 분배하는 과정에서 발생하는 overhead로 인해 자칫 다른 Host에도 장애가 발생할 수 있기 때문입니다. 

 

따라서 Hash와 Modular 연산을 통해서 부하를 분산하는 환경에서는 Scale Out이 발생할 때, Connection 분배와 데이터 이동을 최소화하는 방안을 강구해야합니다. 그렇다면 어떻게 이를 해결할 수 있을까요? 

 

첫 번째로 고민해볼 수 있는 방법은 Scale Out 시에 기존 서버 대비 2배씩 확장하는 방법입니다.

 

 

가령 위 그림과 같이 기존 3대의 Host에서 Scale Out을 위해 서버를 2배로 확장했을 때 모습입니다. 이때 Connection 및 데이터 재분배 현황을 보면, 서버를 2배로 늘렸을 때 데이터 재분배 비율이 50% 정도로 낮아진 것을 볼 수 있습니다. 또한 데이터 이동 현황을 보면, 기존 Host가 보유하고 있던 데이터 중 일부가 다른 Host에게 그대로 전달된 것을 확인할 수 있습니다. 이는 데이터를 마이그레이션 해야하는 운영자 입장에서도 계획을 세우기 훨씬 수월하며, 분배에 대한 비효율도 다소 줄일 수 있습니다. 하지만 매번 Scale Out이 필요할 경우 서버를 2배씩 늘리는 것은 매우 비효율 적일 것입니다. 또한 이는 Scale Out시에는 도움이 될 수 있지만, Scale In을 수행하는 경우에는 문제가 발생합니다. Scale In을 수행할 때는 1/2씩 서버를 줄여야하는데 현실적으로 힘들기 때문입니다.

 

따라서 대규모 시스템에서 Scale in/out이 잦은 Application을 사용하는 경우 단순한 Hash 함수와 Modular 연산의 적용은 운영 안정성이 저하되는 이슈를 낳습니다.

 

---

 

4-1 Ring Hash(Consistent Hashing)

 

이전에 살펴봤듯이 Hash와 Modular 기반으로 부하를 분산시에 Scale In/Out에 탄력적으로 대응하기 위해서는 서버 추가/삭제 이후 Connection과 연관된 데이터 재분배되는 양이 적어야 합니다. 이때 Consistent Hashing 알고리즘은 이에 대한 해결책을 제시해줍니다.

 

 

 

 

Consistent Hashing 방식은 위와 같이 Host 들이 하나의 Circle 안에 배치되어있다고 가정합니다. 그리고 각각의 Host가 담당하게될 Hash 값의 데이터 범위를 할당하여 그 안에 위치한 Connection을 담당하는 구조입니다. 이때 두 Host 사이에 위치한 Hash 값을 지닌 Connection은 시계 방향에 위치한 Host에 할당되는 것이 특징입니다.

 

Host	데이터 담당 범위
Host 1	해시값 <= 10724 OR 해시값 > 12345
Host 2	10724 < 해시값 <= 11224
Host 3	11224 < 해시값 <= 12345

 

즉 위와 같이 각각의 Host 별로 Client 요청에 대한 Hash 값을 계산하였을 때, 해당 Hash 값이 담당하는 Host를 범위로써 관리합니다. 위와 같이 Ring Hash가 구축된 상태에서 Client 요청이 들어왔다고 가정해보겠습니다.

 

Case 1. 해시값 10765 요청 전달 시

 

 

 

해당 해시값이 10765이므로, 이는 Host1의 데이터 범위인 10724 보다 크고 Host 2의 해시값 범위인 11224보다 작으므로 Host 2에 할당될 것입니다. 또한 이후에 동일한 해시값이 입력된다면 지속적으로 Host 2에 할당됨이 보장됩니다.

 

Case 2. 해시값 12086 요청 전달 시

 

 

 

해시값이 12086의 요청이 전달되면, 11224보다 크고 12345보다 작으므로 Host 3에 Connection이 할당됩니다.

 

 

이후 지속적인 Connection 연결이 요청되었고 그 결과 위와 같은 모습이 되었다고 가정해봅시다. 위 그림을 살펴보면 Host 1은 해시값이 5576과, 12567에 해당하는 Connection이 연결된 상황이라고 볼 수 있습니다. 만약 이 상황에서 Host 1에 장애가 발생하여 Down되었을 경우 어떻게 될까요?

 

 

 

Ring Hash를 살펴보면, Host 1이 없어졌을 경우 그 다음 연결된 Host는 Host 2임을 알 수 있습니다. 따라서 Consistent Hashing 알고리즘에서는 Host 1에 할당되었던 기존 Connection이 Host 2로 모두 이관됨을 확인할 수 있습니다.

 

 

또한 위 결과를 통해 흥미로운 사실을 알 수 있습니다. Host 삭제에 따른 변화를 살펴보면, 기존 Host 1에 할당된 Connection만 영향을 받을 뿐 그외 나머지 Host에 연결된 Connection에는 전혀 영향을 받지 않는 것을 확인할 수 있습니다. 이를 통해 알 수 있는 사실은 Consistent Hashing 알고리즘을 사용할 경우에는 이전과 달리 Host 추가/삭제에 따른 영향도가 적어짐을 알 수 있습니다.

 

하지만 위와 같은 기본 Consistent Hashing 방식은 문제점이 존재합니다. 위 그림을 통해 살펴보면 Host 1이 Down 되었을 때 그 부하가 고스란히 Host 2에게 전달되는 것을 확인할 수 있습니다. 이는 Host 2 입장에서는 엄청난 부담으로 작용할 수 있습니다. 두 번째 문제는 Hash 값이 고르게 분포되지 않을 경우 특정 Host에 연결되는 Connection이 많을 수 있음을 의미합니다. 가령 위 예에서는 비교적 Hash 값이 고르게 분포되어 각각의 Host에서 Connection을 수행할 수 있었습니다. 하지만 상황에 따라서는 특정 Hash 범위 값이 지속 할당되면, 특정 Host에만 부하가 집중될 수 있습니다. 따라서 이를 보완하기 위해서 추가적으로 Virtual Node를 추가하는 방법을 많이 사용합니다.

 

4-2-1 Virtual Node

 

기존의 문제점이 서버 추가/제거 시 특정 Host에 Connection 재매핑이 몰린다는 특징이 있었습니다. 따라서 이러한 문제를 해결하기 위해서는 Virtual Node를 추가합니다. 이해를 돕기 위해 사례를 통해 확인해보겠습니다.

 

 

위와 같이 Host 별로 하나씩 Ring에 배치하는 것이 아니라 Ring Hash에 여러개의 Virtual Node를 생성하여 해시값을 부여하고 위 그림과 같이 배치를 수행합니다. 그러면 각각의 Node와 데이터 사이의 해시값 범위가 줄어들게 됩니다. 따라서 위 그림과 같이 Connection 별로 Hash 값을 수행한 결과가 조금 더 다양한 Host로 매핑될 수 있습니다. 

 

 

 

 

이러한 상황에서 Host 1에 Down이 발생하면 어떻게 될까요? 기존에 Host 1에 매핑되었던 Connection Hash 값 12567만 재배치를 수행하면 되기 때문에, 기존의 Consistent Hashing 방식 보다는 재배치되는 Connection 수가 많이 줄어들 수 있습니다. 

 

이렇듯 Virtual Node를 Ring Hash에 추가하면, 재배치의 이점과 Virtual Node 개수가 증가할 수록 상대적으로 촘촘하게 Ring에 배치되기 때문에 부하를 보다 고르게 분산할 수는 있습니다. 다만 이러한 방법도 완벽하게 부하를 고르게 분산할 수는 없습니다.

 

Consistent Hashing 방식의 구현체는 다양하게 존재하는데, envoy는 위와 같이 내부에 virtual node가 추가된 형태의 자체 Consistent Hashing 방식으로 구현되어있습니다.

 

이때 각각의 host 별로 Virtual Node 할당을 위해서 envoy에서는 Configuration 설정에 minimum_ring_size, maximum_ring_size, 그리고 Hash 함수 등을 설정할 수 있으며, 각각의 Host 별로도 load_balancing_weight를 지정하여 개별 Host 별로 할당받는 hash bucket 개수를 조정할 수 있습니다.

 

이를 적용한 envoy configuration 예시는 다음과 같습니다.

static_resources:
  clusters:
  - name: my_service
    connect_timeout: 0.25s
    type: strict_dns
    lb_policy: ring_hash
    ring_hash_lb_config:
      minimum_ring_size: 1024
    load_assignment:
      cluster_name: my_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8080
          load_balancing_weight: 2      
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8081
          load_balancing_weight: 1      
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8082
          load_balancing_weight: 3

 

지금까지 Consistent Hashing 방식에 대해서 살펴봤습니다. 그렇다면 Consistent Hashing 방식은 어떤 경우 사용하면 좋을까요? 이 방식은 Cluster와 Server의 변경이 잦은 경우 즉 대규모 환경에서 Reliability가 중요한 경우 Connection의 안정적인 재매핑을 위해 사용하면 좋습니다.

 

하지만 Ring Hash를 만들고 이를 유지하기 위해서는 기존에 소개한 다른 알고리즘(Random, Round-Robin, Least Request)에 비해 조금 더 많은 메모리 사용이 필요합니다. 또한 Cluster로부터 Server가 추가되거나 제거될 때 Hash Ring을 재구축 해야하기 때문에 update가 다른 알고리즘에 비해 조금 느린 특성이 있습니다. 

 

따라서 해당 알고리즘을 사용한다면 이러한 특징을 참고하여 여러 알고리즘과 비교 및 PoC를 통해 선정하는 것이 좋습니다.

 

---

 

4-2 Maglev

 

마지막으로 소개할 알고리즘은 Maglev 입니다. 해당 알고리즘은 Google에서 사용하는 Load Balancing 알고리즘으로 Consistent Hashing에 비해 부하를 보다 균등하게 분산해줄 수 있습니다. 또한 Consistent Hashing 알고리즘과 마찬가지로 서버의 삭제/수정 시에 Connection 재매핑 비율을 줄여줍니다. 해당 알고리즘 특징에 대해 살펴보면 다음과 같습니다.

 

Maglev 방식을 통해 부하를 분산하기 위해서는 2가지 자료구조가 필요합니다.

 

 

첫 번째는 Lookup table입니다. Lookup table은 추후 Load Balancer가 어떤 Host로 보낼지 라우팅 정보를 기록하는 리스트입니다. 즉 라우팅 당시 해시 값을 Lookup table의 사이즈만큼으로 나머지 연산을 했을 때 나오는 index를 기준으로 Lookup table에 매핑된 Host로 라우팅을 수행합니다. 위 예시의 경우 Lookup table의 사이즈를 임의로 7로 선정했는데, Lookup table의 사이즈는 변경이 가능하며, 사이즈가 크면 클 수록 향후 서버 추가/삭제 등의 영향을 적게 받습니다.

 

두 번째 살펴볼 것은 Permutation table입니다. 해당 자료구조는 각 Host 별로 Lookup 테이블 인덱스 어디에 매핑될 지를 희망하는 우선순위를 나타냅니다. 따라서, Lookup table의 사이즈가 결정되면, 가장 먼저 수행하는 것은 각각의 Host 별로 어디에 매핑되기를 희망하는지를 해시 연산을 통해 계산합니다. 해당 과정을 Permutation이라고 부릅니다.

 

Permutation이 끝나고 나면, 위와 같이 각 Host 별로 우선순위를 결정할 수 있습니다. 그다음에 수행할 작업은 Permutation table을 토대로 Lookup table 내용을 채우는 작업입니다.

 

 

이 과정은 Popluation 이라고 부르며, 해당 과정을 통해 Maglev 알고리즘이 균등하게 부하를 분산할 수 있도록 Lookup table을 채워줍니다. 그렇다면 Popluation은 어떻게 이루어질까요? 위 Permuation table을 기초로 Popluation 과정을 조금 더 자세히 살펴보겠습니다.

 

 

N : Host 개수

M : Lookup table 크기

next[] : Permutation table 내에서 Host 별로 다음 index를 추적을 위한 배열

entry[] : Lookup table

 

Google의 Maglev 논문을 살펴보면, 위와 같이 Popluation 하는 과정에 대한 Pseudocode를 살펴볼 수 있습니다. 해당 과정을 몇 단계만 차례차례 살펴보면서 동작 과정에 대해서 이해해보겠습니다.

 

 

Step 1.

 

 

가장 먼저 수행하는 작업은 Lookup table과 Next 배열의 초기화입니다. 이때 Lookup table의 모든 원소는 -1로 Next 배열의 경우 배열의 index를 가르켜야 하기 때문에 0으로 초기화하는 것이 특징입니다.

 

Step 2.

 

 

초기화가 완료되면, Host 별로 순차적으로 Lookup table을 갱신합니다. 먼저 0번 Host의 index가 현재 Next에서 0이므로 Permutation table의 0번 index를 살펴봅니다. 이때 Host 0은 Lookup table의 3번 index에 매핑되기 희망하므로 Lookup table의 해당 index을 탐색합니다. 이때 Lookup table에 해당 값이 -1 즉 아무 값도 매핑되지 않은 상태이므로 갱신이 가능합니다. 따라서 해당 index를 Host 0번 서버의 index인 0번으로 갱신합니다.

 

이후 Next 배열에서 그 다음 Permutation Table의 다음 참조 index를 가르키기 위해서 1증가한 값으로 갱신합니다.

 

Step 3.

 

 

Host 0번이 매핑되었으므로 그 다음은 Host 1 차례입니다. Host 1의 현재 index가 0번이므로 Permutation table에서 해당 index를 살펴봅니다. 이때 Host 1은 Lookup table의 0번 index에 매핑되기 희망하므로 Lookup table의 해당 index를 탐색합니다. 살펴본 결과, 해당 index 값이 -1 이므로 선택될 수 있습니다. 따라서 Lookup table의 0번 index를 Host 1번의 값인 1로 갱신합니다. 또한 Next 배열에서 다음 값을 참조하기 위해서 1증가한 값으로 갱신합니다.

 

Step 4.

 

 

그 다음은 Host 2번이 Lookup table에 매핑될 차례입니다. 이전과 마찬가지로 Next 배열에서 해당 Host의 Permutation table 참조 index가 0번이므로 0번을 살펴봅니다. 이때 Host 2는 Lookup table에서 3번 위치에 매핑되기를 희망합니다. 따라서 Lookup table이 3번 index를 살펴봅니다.

 

하지만 Host 0번에 기존에 선점하였기 때문에, 해당 index에 Host 2를 매핑할 수는 없습니다. 따라서 다른 Lookup table 위치를 탐색해야합니다.

 

 

이 경우에는 Permutation table의 다음 index 참조를 위해 먼저 Next 배열에 저장된 값을 1증가시킵니다. 이후 Permutation table에서 증가된 index 위치를 탐색합니다. 결과를 살펴보면, 두 번째로 선호하는 자리가 4번 index이기 때문에 Lookup table에서 해당 index를 탐색합니다. 살펴본 결과 해당 값은 -1이기 때문에 선점이 가능합니다. 따라서 Lookup table의 4번 index는 Host 2에게 배정합니다.

 

만약 또 충돌이 발생했다면, 방금전 수행했던 절차대로 Next 배열의 값을 1 증가시키고, Permutation table의 해당 위치를 탐색하여 Lookup table에 해당하는 index 값이 -1이 나올 때까지를 반복 수행하여 매핑을 수행합니다.

 

 

 

Host 2의 매핑이 완료되었으면, Next 배열의 값을 1 증가 시켜, 다음 Permutation table 참조 index를 갱신합니다.

 

Step 5.

 

Host 2까지 매핑이 모두 완료되었으면, 다시 Host 0의 차례입니다. 따라서 Step 2-4 단계를 반복하면서, 개별 Host가 선호하는 Lookup table 갱신을 수행합니다. 해당 과정은 Lookup table이 완성될 때까지 반복합니다.

 

 

 

 

Permutatiion 과정이 모두 끝나고 나면, 위 그림과 같이 Lookup table이 완성됩니다. 완성된 결과를 살펴보면, 모든 Host가 균등하게 배분되었음을 알 수 있습니다. 이는 Permutation 단계에서 각 Host 별로 돌아가면서 Lookup table을 순차적으로 갱신했기 때문입니다.

 

따라서 이후 Load Balancer는 Hash 값에 대한 Modular 연산을 통해서 Host에게 부하를 분산할 수 있게됩니다.

 

그렇다면 Maglev 알고리즘을 사용하는 상황에서 Host의 삭제가 발생했을 경우 어떻게 될까요?

 

 

 

이전과 마찬가지로 Permutation 과정을 진행합니다. 하지만 Permutation을 수행해도 기존 Host가 희망하는 해시값 기반 우선순위에는 변화가 없습니다. 따라서 Host 1이 제거될 경우 기존 table에서 Host 1 데이터들만 제거되어 갱신될 뿐, 기존 Host 들이 선호하는 값은 그대로입니다.

 

이후 Popluation을 진행하면, 위 그림의 Lookup table과 같이 변경됩니다. 이때 유심히 볼 점은 기존 Host들이 선호했던 Lookup table의 index 변화는 없는 상태에서 Host 1이 희망했던 index에만 다른 Host로 채워졌음을 알 수 있습니다. 따라서 이는 기존의 다른 Host들에게 연결된 Connection은 그대로 유지한채 기존 Host 1에 연결된 Connection만 재연결이 필요함을 의미합니다. 따라서 안정성을 확보할 수 있는 장점이 존재합니다.

 

지금까지 Maglev 방식에 대해서 살펴봤습니다. envoy에서는 다음과 같이 lb_policy를 maglev로 지정함으로써 해당 알고리즘을 사용할 수 있습니다.

 

static_resources:
  clusters:
  - name: my_service
    connect_timeout: 0.25s
    type: strict_dns
    lb_policy: maglev
    maglev_lb_config:
      table_size: 65536
    load_assignment:
      cluster_name: my_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8080
          load_balancing_weight: 2      
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8081
          load_balancing_weight: 1      
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 8082
          load_balancing_weight: 3

 

이때 위와 같이 maglev_lb_config를 통해 Lookup table의 크기를 변경할 수 있습니다. 참고로 Lookup table의 크기를 키우면 키울 수록 Host Down에 대한 Connection 재분배에 대한 영향도를 줄일 수 있으며, envoy에서는 기본값이 65536이고 최대 5000011개로 지정할 수 있습니다.

 

이때 Lookup table 사이즈를 키우게되면, Connection 재분배 비율은 줄일 수 있으나 Memory를 그만큼 사용하며, Population에 드는 시간이 증가하므로 마찬가지로 적절한 PoC를 통해서 적정 값을 도출하는 것이 좋습니다.

 

지금까지 Maglev에 대해서 살펴봤는데요. Maglev는 Consistent Hashing 방식에 비해 어떤 이점을 가지고 있을까요?

 

1. Traffic Balancing

 

Maglev의 경우는 ECMP(Equal-Cost Multipath) 라우팅을 추구합니다. Lookup table을 구성할 때 모든 Host가 동등한 비율로 배분하여 Traffic 전달 시에 보다 균등하게 부하를 분산할 수 있도록 추가합니다.

 

2. 단순한 설정

 

Maglev의 경우는 Virtual Node를 생성하지 않습니다. Consistent Hasing 방식의 경우는 Virtual Node 생성을 위한 여러 기타 설정이나 균등하게 부하를 분산하기 위해 기타 Hashing 파라미터를 조정해야할 수 있지만 Maglev는 Table 사이즈를 조정하는 정도를 통해서 목표를 달성할 수 있습니다.

 

하지만 Maglev를 사용하기 위해서는 Permutation Table, Lookup table, Next 등 다양한 자료구조가 사용되며, Host 생성/추가 발생 시에 Permutation과 Popluation 작업이 지속 발생되기 때문에 상대적으로 높은 메모리와 CPU 비용이 요구됩니다. 따라서 무조건 Maglev 방식이 좋다고 말할 수는 없으며, 언제나 비즈니스 상황과 인프라 환경에 맞추어 테스트 후 적절한 방법을 사용하는 것이 좋다고 볼 수 있습니다.

---

 

8. 마치며

 

지금까지 envoy에서 제공하는 다양한 Load Balancing 알고리즘 방법에 대해서 살펴봤습니다. 업무 환경에 적용할 때는 비즈니스 환경의 규모 및 성능을 고려하여 적절한 Load Balancing 전략을 차용하는 것이 중요합니다. 따라서 각각의 알고리즘이 어떻게 동작하며, 어떠한 특징이 있는지 잘 이해하고 있는 것이 무엇보다 중요하다고 볼 수 있겠습니다.

 

 

1. 서론

 

앞선 두 개의 포스팅을 통해 envoy 고수준 아키텍처 구조 및 쓰레딩 모델에 대해서 학습하였습니다. 이번 포스팅 부터는 Envoy의 가장 핵심이되는 주요 컴포넌트에 대해서 각 컴포넌트가 세부적으로 어떻게 구성되어있는지 그리고 컴포넌트간에 상호 작용이 어떻게 이루어지는지에 대해 자세하게 살펴보겠습니다.

 

참고로 이번 포스팅의 내용은 TCP 기반 설정이 구성되었다는 가정하에 작성하였습니다.

---

 

2. Envoy 구조

 

 

이 시리즈의 첫 번째 포스팅에서 Listener, Endpoint, Cluster 개념에 대해서 간략하게 살펴봤습니다. 그 중 Endpoint의 경우 Cluster와 종속적인 관계를 지니므로 Cluster에서 Endpoint를 관리합니다. 따라서 envoy의 핵심 컴포넌트를 꼽자면 Listener와 Cluster라고 봐도 무방합니다. 그렇다면 Listener와 Cluster는 누가 관리하고 어떻게 생성될까요? 위 그림을 보면 Listener Manager와 Cluster Manager가 해당 컴포넌트를 관리하는 것을 짐작할 수 있습니다.

 

 

 

실제로 envoy server 기동 코드를 살펴보면 config.yaml 파일을 읽어서 envoy 기동에 필요한 사용자 정의 설정을 파싱하는 작업을 선행합니다. 그 이후 주요 컴포넌트 기동에 필요한 항목등을 생성합니다.

 

이때 위 그림을보면 Listener Component Factory와 Cluster Manager Factory를 생성하는 것을 볼 수 있습니다. 여기서 Factory라는 말에서 의미가 바로 전달되듯이 해당 Factory 들은 Listener Manager와 Cluster Manager를 생성하는데 있어 중요 역할을 수행합니다.

 

지금부터는 Listener Component Factory로부터 생성되는 Listener Manager의 세부 구조와 Cluster Manager Factory로부터 생성되는 Cluster Manager에 대해서 상세하게 알아보고자 합니다. 다만 이번 포스팅에서는 Cluster Manager에 대해서 중점적으로 알아보고 Listener는 다음 포스팅에서 다루겠습니다.

 

---

 

3. Subscription Factory 관리

 

Cluster Manager는 Cluster Manager Factory로 부터 생성되는 컴포넌트로써, 이름을 통해 유추할 수 있듯이 Cluster를 관리하는 역할을 수행합니다. 따라서 이에 대한 이해를 위해서는 Cluster 관리가 어떻게 이루어지는지 먼저 학습하는 것이 좋습니다. 하지만 그 전에 Cluster Manager의 주요 역할 중 하나인 Subscription Factory 관리와 gRPC Multiplexer에 대해서 먼저 다루어보고자 합니다.

 

이를 먼저 다루는 이유는 Cluster Manager에서 관리하는 Cluster를 등록하는 방식에는 Static 방식과 Dynamic 방식이 존재하는데, Dynamic 방식의 Cluster Update 과정에 대해 이해하려면 Subscription과 gRPC Multiplexer에 대한 이해가 선행되어야하기 때문입니다.

 

방금 전 언급했듯이 Cluster Manager의 역할 중 하나는 Subscription Factory를 관리하는 것이라고 설명했습니다. 그렇다면, Subscription은 무엇이고 Subscription Factory는 어떤 역할을 수행할까요? 이 부분에 대해서 살펴보겠습니다.

 

이전에 Envoy는 xDS API를 통한 Resource(Listener, Endpoint 등)를 관리한다고 언급한바 있습니다.

 

 

 

참고로 xDS API에는 gRPC, Rest API, 파일 동기화 방법이 있지만, 여기서는 gRPC 방식을 사용한다고 가정하겠습니다.

 

이때 gRPC 통신을 위해서 내부적으로 Multiplexer를 사용하는데, 자원 동기화 요청이 Envoy로 전달되면, 이를 수신받아 해당 Resource를 처리하는 모듈(CDS, LDS, EDS 등)에게 전달 해줘야합니다. 그리고 처리하는 모듈 쪽에서는 사전에 Callback을 등록하여 응답이 전달되면 해당 Callback을 실행할 수 있도록 합니다.

 

 

이 과정에서 중간 매개체 역할을 하는 것이 Subscription입니다. 즉 Resource를 처리하는 모듈에서 Config 대상 오브젝트 타입과 Callback을 Subscription 객체로 생성하여 등록합니다. 그리고 해당 Subscription 생성은 Subscription Factory가 수행하고, 추후 gRPC Multiplexer에 등록합니다. 이후 gRPC Multiplexer로 부터 Resource 동기화 요청이 수신되면, 해당 요청에 상응하는 Subscription의 Callback이 호출되어 데이터 동기화 처리를 수행할 수 있게됩니다.

 

예를 통해서 살펴보겠습니다. 가령 Listener Manager에서 LDS를 등록한다고 가정하겠습니다. 이 경우 먼저 Cluster Manager의 Subscription Factory로부터 Subscription 생성을 요청합니다. 이때 Subscription Factory는 내부 gRPC Multiplexer에 해당 정보를 등록합니다. 이후 LDS 요청 타입의 응답이 전달되었을 경우 Callback을 통해 Listener Manager에게 응답을 전달합니다.

 

이때 Cluster Manager는 Subscription Factory를 내부 프로퍼티로 가지고 있고, 외부의 다른 컴포넌트로부터 Subscription 생성, 삭제 등의 처리를 위임받아 처리하는 역할을 수행합니다.

 

 

이를 조금 더 자세히 표현하면, 개별 xDS API들은 SubscriptionBase 클래스를 상속받고 있으며, 해당 클래스에는 각각 Config Update에 대하여 수행해야할 메소드들이 정의되어있습니다. 따라서 각각의 API는 구현이 요구되는 메소드들에 대해서 모듈 특성에 따라 처리방법이 기술되어있습니다. 이후 xDS API를 사용하기 위해서 Cluster Manager에 존재하는 Subscription Factory로부터 Subscription을 요청하는데, 이때 Subscription Factory 내부에서는 Multiplexer에 해당 Callback 정보를 등록하고 생성된 Subscription을 반환합니다.

 

---

 

4. gRPC Multiplexer

 

지금까지 xDS API 사용을 위해 Subscription Factory로부터 Subscription을 생성받는 과정에 대해서 살펴봤습니다. 이번에는 subscription 생성 이후 상호작용 수행하는 gRPC Multiplexer의 구성요소 및 동작 원리에 대해서 살펴보겠습니다. 

 

 

 

gRPC Multiplexer에는 SotW를 위한 GrpcMuxSotw, Delta XDS를 위한 GrpcMuxDelta 등 여러가지 GrpcMuxImpl을 위한 구현체가 존재합니다. 그리고 해당 구현체는 공통적으로 위 3가지 컴포넌트가 내부에 존재합니다. 해당 컴포넌트가 무엇이고 어떠한 역할을 하는지 살펴보겠습니다.

 

1. watchMap은 외부에서 생성한 Subscription을 기반으로 gRPC 통신 수행결과를 전달받아 Callback 함수를 실행시키기 위한 자료구조로써 Watch 객체를 생성하여 저장합니다. Watch 객체에는 사용자가 지정한 Callback이 매핑되어있습니다.

 

2. gRPC 통신을 위해서는 SubscriptionState가 필요합니다. 해당 객체는 SubscriptionStateFactory로 부터 생성이 가능하며, SubscriptionState에는 WatchMap에 존재하는 Watch 객체를 매핑하여, 향후 gRPC 응답이 반환되었을 때 이를 전달할 수 있도록 기능이 구현되어있습니다.

 

3. grpcStream은 gRPC 통신을 수행하는 주체로써 외부 Management Server와 통신을 담당합니다.

 

gRPC Multiplexer는 여러개가 존재할 수 있는데, 개별 Multiplexer들을 관리하기 위해서 외부에 AllMuxes가 존재합니다. AllMuxes는 생성되는 여러개의 GrpcMuxImpl의 정보를 관리하기 위한 자료구조로써 hash set 형태로 생성되는 모든 GrpcMuxImpl을 관리합니다.

 

지금까지 gRPC Multiplexer와 관련하여 컴포넌트 정보를 확인했습니다. 아직은 gRPC를 통한 동기화 과정을 살펴보지 않았기 때문에 위 컴포넌트가 어떻게 상호작용하며 어떤 역할을 수행하는지 와닿지 않을 수 있습니다. 따라서 실제 사용자가 Subscription을 요청하여 데이터가 처리되는 과정을 살펴보면서 각각의 컴포넌트가 어떻게 연계되는지를 살펴보겠습니다.

 

 

 

 

1. 이전에 설명했듯이 xDS API를 사용하려는 CDS, EDS, LDS 등은 Cluster Manager가 가지고 있는 Subscription Factory로부터 Subscription을 전달받아야합니다. 따라서 먼저 Cluster Manager로부터 Subscription Factory 정보를 얻어옵니다.

 

2. Subscription Factory는 CDS, EDS, LDS 등에 매핑되어있는 api_type(SotW, Delta 등)을 분석하여 적절한 GrpcMuxImpl 구현체를 반환하도록 요청합니다. 

 

3. api_type에 따른 GrpcMuxImpl 구현체를 생성한 이후 Subscription에 해당정보를 매핑하여 반환합니다.

 

4. Subscription을 전달받으면, 이후 통신 수행을 위하 GrpcMuxImpl에게 통신 수행을 요청합니다.

 

5. GrpcMuxImpl은 WatchMap에 Subscription에서 요구하는 type_url 정보에 해당하는 Watch 정보가 존재하는지 확인합니다. 만약 존재하지 않는다면 새로운 Watch를 생성합니다. 이때 Watch에는 Subscription에 존재하는 Callback을 저장합니다.

 

6. WatchMap에 type_url에 해당하는 Watch 정보가 없을 경우 SubscriptionStateFactory로부터 새로운 SubscriptionState를 생성합니다. 이때 생성되는 SubscriptionState에는 5번 단계에서 생성한 Watch 정보를 포함시켜, 향후 gRPC 응답이 왔을 때 해당 Watch에게 응답 수신 행위를 수행할 수 있도록 설정합니다. SubscriptionState 생성이 완료되면, 해당 정보는 GrpcMuxImpl 내부에 존재하는 subscriptions_ 자료구조에 저장됩니다.

 

7. SubscriptionState를 통해 gRPC 통신을 수행을 요청합니다.

 

8. gRPCStream은 xDS의 Management Server(ex: istio)와 연결을 수행하여 DiscoveryRequest를 요청합니다. 이후 Management Server로부터 Discovery Response를 전달하면 이를 수신받습니다.

 

9. gRPCStream은 응답 메시지의 type_url을 기준으로  subscriptions_ 자료구조에서 해당 type_url에 해당하는 Subscription을 찾아 응답 메시지를 전달합니다.

 

10. SubscriptionState는 6번 단계에서 매핑한 Watch를 통해 응답 처리를 위임합니다.

 

11. Watch는 생성당시 저장된 요청자의 Callback을 수행하여 처리를 위임합니다.

 

 

 

지금까지 gRPC Multiplexer에 대해서 살펴봤습니다. 이전 시나리오에서는 1개의 Subscription이 등록되고 gRPC 통신이 수행되는 과정에 대해서 살펴봤습니다. 실제로는 CDS, EDS, LDS 등 여러개의 Subscription이 존재하기 때문에 Subscription Factory에서 생성되는 Subscription은 다수가 될 수 있으며, 그에 따라 gRPC Mux 내부에는 Watch와 SubscriptionState 수가 다수가 될 수 있습니다. 마찬가지로 gRPC를 통해 수행되는 Mux 타입이 다수라면 해당 타입 또한 여러개가 될 수 있습니다.

 

---

 

5. Cluster 관리

 

지금까지 Cluster Manager의 기능 중 하나인 gRPC Multiplexer와 Subscription 관리에 대해서 살펴봤습니다. 이번에는 Cluster Manager에서 가장 핵심이 되는 Cluster 관리에 대해서 살펴보겠습니다.

 

Cluster Manager는 이름 그대로 Cluster를 관리하는 컴포넌트이기 때문에 가장 중요한 역할은 Cluster의 생성과 삭제 수정등을 수행하는 것 입니다.

 

 

 

Cluster Manager는 기동 당시에 Cluster Manager Factory에 의해서 생성됨을 이전에 설명했습니다. 이때 생성 과정에서 주요하게 살펴볼 부분은 Cluster Manager 생성과 동시에 CDS 설정이 존재한다면, CDS를 생성하고 config.yaml에 존재하는 Static Resource는 Parsing 정보를 토대로 읽어들어 ClusterData를 생성하는 것입니다. 또한 이후 CDS를 통해 ClusterData 정보가 업데이트되면, 해당 정보를 Cluster Manager가 전달받아 업데이트를 수행하는 것 또한 Cluster Manager의 역할입니다.

 

static_resources:
  clusters:
  - name: local_service
    connect_timeout: 0.25s
    type: STRICT_DNS
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: local_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address: { address: webservice, port_value: 90 }         
    health_checks:
      timeout: 3s
      interval: 90s
      unhealthy_threshold: 5
      healthy_threshold: 5
      no_traffic_interval: 240s      
      http_health_check:
        path: "/ping"
        event_log_path: /var/log/envoy/health_check.log
        expected_statuses:
          start: 200
          end: 201
    outlier_detection:
      consecutive_5xx: 2
      base_ejection_time: 30s
      max_ejection_percent: 40
      interval: 20s
      success_rate_minimum_hosts: 5
      success_rate_request_volume: 10

 

해당 과정에 대해서 조금 더 알아보기 위해 예를 통해서 살펴보겠습니다. 가령 위와 같이 static cluster 정보가 존재한다고 가정해봅시다. 여기서 Cluster 정보를 살펴보면, Load Balancing, Health Check, Outlier Detection 정보가 포함되어있는 것을 확인할 수 있습니다. 따라서 이를 통해 유추하자면 Envoy 내부에서 Cluster를 관리할 때 그 안에는 Load Balancing, Health Check  및 Outlier Detection을 수행하는 컴포넌트가 존재할 것이라는 것을 알 수 있습니다.

 

이번에는 Cluster Manager에서 Cluster를 만드는 과정에 대해서 살펴보겠습니다.

 

 

 

먼저 살펴볼 것은 static resource에 존재하는 cluster 정보를 Cluster Manager에서 관리하는 방법입니다. 해당 과정은 Envoy 기동 과정에서 수행되는 작업으로 생성 과정을 간략하게 살펴보면 다음과 같습니다. 먼저 envoy 기동시에 config 파일을 파싱합니다. 이후 Cluster Manager의 Cluster 저장을 위해 ClusterData 구조로 만들어서 이를 active_clusters_ 라고 불리는 map에 Cluster 정보를 저장합니다. 그렇다면 active_clusters_는 어떻게 구성되어있을까요?

 

 

해당 자료구조는 위 그림과 같이 Cluster 이름과 Cluster 정보로 이루어진 Map입니다. Cluster가 생성된 이후에 active_clusters_에 ClusterData를 추가하며, Cluster 삭제 이벤트가 발생하면 이를 삭제하는 등 해당 자료구조를 통하여 Cluster Manager가 관리하고 있는 Cluster의 종류와 갯수 현행화를 수행합니다.

 

이때 Map의 Value로 저장되는 것은 ClusterDataPtr으로써 파싱된 결과를 ClusterData 데이터 구조로 만들고 이에 대한 포인터 값을 저장하는데, ClusterData가 보유한 내부 속성 중 중요한 속성 설명 및 생성 과정에 대해 조금 더 자세히 알아보겠습니다.

 

 

 

 

 

먼저 Parsing된 Cluster 정보를 토대로 ClusterData 인스턴스 생성을 위해 ProdClusterManagerFactory에게 처리를 위임합니다. 해당 Factory 내부에서는 다시 ClusterFactoryImplBase에게 Cluster 생성 주체를 위임합니다.

 

 

cluster_factory_impl.cc

std::pair<ClusterSharedPtr, ThreadAwareLoadBalancerPtr>
ClusterFactoryImplBase::create(Server::Configuration::ServerFactoryContext& server_context,
                               const envoy::config::cluster::v3::Cluster& cluster,
                               ClusterFactoryContext& context) {
  auto stats_scope = generateStatsScope(cluster, context.stats());
  std::unique_ptr<Server::Configuration::TransportSocketFactoryContextImpl>
      transport_factory_context =
          std::make_unique<Server::Configuration::TransportSocketFactoryContextImpl>(
              server_context, context.sslContextManager(), *stats_scope, context.clusterManager(),
              context.stats(), context.messageValidationVisitor());

  std::pair<ClusterImplBaseSharedPtr, ThreadAwareLoadBalancerPtr> new_cluster_pair =
      createClusterImpl(server_context, cluster, context, *transport_factory_context,
                        std::move(stats_scope));

  if (!cluster.health_checks().empty()) {
    if (cluster.health_checks().size() != 1) {
      throw EnvoyException("Multiple health checks not supported");
    } else {
      new_cluster_pair.first->setHealthChecker(HealthCheckerFactory::create(
          cluster.health_checks()[0], *new_cluster_pair.first, context.runtime(),
          context.mainThreadDispatcher(), context.logManager(), context.messageValidationVisitor(),
          context.api()));
    }
  }

  new_cluster_pair.first->setOutlierDetector(Outlier::DetectorImplFactory::createForCluster(
      *new_cluster_pair.first, cluster, context.mainThreadDispatcher(), context.runtime(),
      context.outlierEventLogger(), context.api().randomGenerator()));

  new_cluster_pair.first->setTransportFactoryContext(std::move(transport_factory_context));
  return new_cluster_pair;
}

 

ClusterFactoryImplBase의 처리 과정을 살펴보기 위해 코드레벨로 확인해보겠습니다. ProdClusterManagerFactory로 부터 Cluster 생성 처리를 위임받은 ClusterFactoryImpl의 create 메소드내용을 살펴보면, Socket 관리, Cluster, HeathChecker, OutlierDetector를 생성하기 위해 각각의 Factory에게 처리를 위임하고 생성 결과를 전달받아 Cluster에 할당하는 것을 볼 수 있습니다. 참고로 여기서 생성되는 HealthCheker 및 OutlierDetector에 대한 자세한 설명은 조금 뒤에 다시 살펴보겠습니다.

 

위 코드에서 실제 Cluster 생성은 createClusterImpl 메소드내에서 이루어지며, 해당 메소드 내에서는 Proto 파일 정의에 따른 인스턴스 생성과 실제 Parsing 내용을 결합한 Cluster 인스턴스를 전달받습니다.

 

 

 

 

 

Cluster 생성이 완료된 이후에는 Cluster 인스턴스와 내부에 Parsing된 정보를 토대로 생성된 HealthChecker, OutlierDetector, LoadBalancer가 존재할 것입니다. 이후에는 다음과 과정을 추가로 거칩니다.

 

 

1. healthChecker가 등록되어있다면, hostCheckCompleteCallback 함수를 등록시킵니다. 해당 함수는 host가 Health Check가 실패했을 때, 해당 정보를 Cluster Manager에게 전달하고, 결과적으로는 해당 Cluster가 Connection Pool에서 해제하도록 후속 작업을 수행하기 위한 용도로 사용됩니다.

 

2. outlierDector가 등록되어있다면, changeStateCallback 함수를 등록시킵니다. 해당 함수는 만약 host가 Outlier로 판정되고, ejection 상황이라면, 해당 정보를 Cluster Manager에게 전달하고, 결과적으로는 해당 Cluster가 Connection Pool에서 해제하도록 후속 작업을 수행하기 위한 용도로 사용됩니다.

 

3. Callback이 모두 등록이 완료되면, Cluster에 설정된 Load Balancer를 생성하기 위한 메타데이터를 등록합니다. 이때 등록되는 Load Balancer 타입은 lb_policy에 명시된 종류에 따라 다르게 생성되며, Round Robin, Least Request, Hash Ring과 같은 타입들을 지원합니다.

 

4. 생성된 Cluster 정보를 Cluster Manager가 관리하는 active_clusters_ 에 추가합니다.

 

위 네가지 과정을 거치게되면 ClusterData 설정은 마무리됩니다. 지금까지 static resource로 등록된 Cluster 처리 과정에 대해서 살펴봤습니다. 이번에는 CDS를 통해 dynamic으로 등록되는 resource 처리 과정에 대해서 살펴보겠습니다.

 

 

 

 

Dynamic 설정의 경우에는 CDS를 통해 처리됨을 설명했습니다. 따라서 CDS 처리를 위해서는 Cluster Manager에서 관리하는 gRPC Multiplexer(ads_mux)와 Subscription Factory와의 상호작용이 필요합니다. 이를 토대로 CDS를 통한 동기화 과정을 살펴보면 다음과 같습니다.

 

1. CDS 등록을 위해 가장 먼저 Cluster Manager에 존재하는 Subscription Factory로부터 Subscription을 요청합니다.

 

2. Subscription Factory는 Subscription을 반환하기 CDS가 전달받기 희망하는 Resource 타입에 대한 데이터 요청 및 Callback 처리를 위해 Multiplexer에 이를 등록합니다.

 

3. CDS가 요청하는 Resource 타입 등록 및 해당 Subscription을 CDS에 반환합니다.

 

4. 외부에서 Cluster 변화(생성/삭제/수정)가 감지되어 Envoy에 통지되면 해당 내용이 Multiplexer를 지나 Callback을 통해 CDS로 전달될 것입니다. 이때 CDS 내부에서는 이를 처리하기 위해 CdsApiHelper에게 처리를 위임합니다.

 

5. CdsApiHelper에서는 해당 내용을 정제합니다. Cluster Manager에 위치한 active_clusters_에 해당 내용을 반영합니다.

만약 기존 Cluster의 업데이트라면 해당 내용을 수정하고, 신규 추가라면 Cluster 정보를 active_clusters_에 추가할 것입니다.

 

 

 

5-1. Cluster Entry

 

 

지금까지 Static 방식과 Dynamic 방식을 통해서 Cluster Manager가 Cluster를 관리하기 위해 사용되는 active_clusters_에 Cluster 정보를 현행화하는 방식에 대해서 살펴봤습니다. 하지만 여기서 active_clusters_에서 값으로 관리되는 ClusterData는 Cluster의 역할 중 핵심 기능을 포함하고 있지 않습니다. 

 

그렇다면 Cluster가 제공하는 핵심 기능은 무엇일까요?

 

 

Cluster 의 가장 중요한 기능 중 하나는 Client가 Cluster 내에 존재하는 여러 B/E host와의 연결을 시도할 때, 중간에서 Client의 연결과 B/E의 연결을 이어주는 Connection Pool 인터페이스 기능을 제공한다는 점입니다.

 

해당 기능은  active_clusters_가 관리하는 ClusterData에서 이를 제공하지 않습니다. 따라서 Cluster Manager에서는 active_clusters_ 를 Cluster 변경이 생겼을 때(CDS) 관리하는 메타데이터 용도로써 내부적으로 관리하고 외부와의 Cluster 연결 등에는 별도 자료구조(thread_local_clusters_)와 ClusterData를 확장한 ClusterEntry를 통해서 Connection Pool 관리 기능과 Load Balancer 기능을 제공합니다.

 

따라서 이번에는 Static, Dynamic 방식을 통해서 active_clusters_에 등록된 이후 진행되는 후속 과정과 이를 통해서 생성되는 ClusterEntry 및 ClusterEntry를 관리하는 thread_local_clusters_ 에 대해서 살펴보겠습니다.

 

 

 

 

Cluster Manager를 구동하는 과정에서 수행하는 작업 중 하나는 Main 쓰레드에 존재하는 Dispatcher를 통해 TLS를 할당받는 것입니다. 이때 생성하는 Slot은 ThreadLocalClustManagerImpl 로써, 외부에 존재하는 모듈이나 내부에서 Cluster 데이터 동기화를 수행할 때 해당 Slot에 존재하는 인스턴스를 활용합니다.

 

그리고 Cluster Manager 내부에는 thread_local_clusters_라고 불리는 Map이 존재하는데, 해당 Map은 Key로써는 Cluster의 이름을 Value는 ClusterData를 기반으로 만들어진 ClusterEntry를 가지고 있습니다. 즉 이전에 설명했듯이 Connection Pool 기능과 실질적인 LoadBalancer 인스턴스를 지니고 있는 ClusterEntry를 해당 자료구조가 가지고 있으며, 외부 모듈에서는 thread_local_clusters_ 접근을 통해 Cluster Manager가 보유하고 있는 Cluster 정보에 대해 참조가 가능합니다.

 

그렇다면 Static, Dynamic 등록을 통해서 active_clusters_에 생성된 Cluster 정보를 기반으로 어떤 시점에 thread_local_clusters_를 만들어낼까요? ClusterEntry 생성과 thread_local_clusters_ 삽입 과정을 통해서 이해해보겠습니다. 먼저 Static Cluster 등록부터 살펴보겠습니다.

 

 

Static Resource가 모두 등록되면 해당 ClusterData는 모두 active_clusters_에 저장되어있을 것입니다. 이후 Cluster Manager 설정이 끝나게되면, active_clusters_를 순회하면서 초기화 작업을 진행합니다.

 

이때 수행하는 작업은 크게 4단계로 이루어져있습니다.

 

1. 개별 ClusterData를 순회하면서 Cluster에 멤버가 추가되거나 우선순위가 변경되었을 때 후속 작업을 수행하기 위한 Callback을 등록합니다.

 

2. 등록되어있는 Callback의 역할은 Cluster Manager가 여러개일 경우 개별 Cluster Manager에서 보유중인 local cluster를 업데이트 하는 것입니다. 따라서 Cluster Manager 동기화를 위해 TLS Slot에 저장된 ThreadLocalClusterManagerImpl을 참조하여 후속 작업 처리를 요청합니다.

 

3. Callback 등록이 완료되면, thread_local_clusters_에 Cluster를 생성하기 위해 TLS Slot에 저장된 ThreadLocalClusterManagerImpl를 참조하여 모든 Cluster Manager에게 Cluster 생성에 대한 내용을 통지하고  이를 전달받은 Cluster Manager에서는 ClusterEntry를 생성하여 자신의 thread_local_clusters_에 저장합니다.

 

 

 

ClusterEntry는 ClusterData에 저장된 info 정보를 포함하고 있으며, 내부에는 클러스터 관리를 위한 여러 속성이 존재합니다. 그 중 몇개만 소개하자면, 먼저 Connection Pool에 접근할 수 있는 인터페이스 기능 제공을 포함합니다. 그리고 등록된 B/E의 host 및 우선순위를 관리하기 위한 PrioritySetIml이 있습니다. LoadBalancer는 해당 Cluster에 대해 외부에서 사용자 접속 요청이 들어왔을 때 실질적으로 LoadBalancer를 수행하기 위한 인스턴스가 매핑되며, AsyncClient를 통해서 Upstream과 연결을 수행합니다.

 

4. 만약 Cluster의 변경에 대하여 외부에서 통지를 받기위해 Callback을 등록했었다면, Cluster 변경에 대한 통지를 외부에게 알립니다.

 

위와 같은 과정을 거쳐 Static Resource로 등록한 자원들이 active_clusters_에 최초 저장이되고 이를 토대로 다시 전체 Cluster Manager에서 Cluster Entry를 만들고 이를 자신의 thread_local_clusters_에 저장함으로써 Cluster 관리가 이루어집니다. 또한 이후에는 외부에서 Cluster 접근을 요청할 때 thread_local_clusters_를 통해서 해당 Cluster 정보 참조가 가능해집니다.

 

 

 

 

Dynamic Resource 등록의 경우 이전에 active_clusters_ 에 추가함을 설명했는데, 이 과정에서 Cluster 초기화를 후속 진행합니다. 이때 ClusterEntry가 생성되고 해당 데이터가 thread_local_clusters_에 추가되면서 Cluster Manager에서 Cluster 동기화가 이루어집니다.

 

 

5-2. Connection Pool 관리

 

이번에는 Cluster Entry 내부에 있는 ConnectionPool 인스턴스를 통해서 Cluster Manager가 어떻게 Connection Pool을 관리하는지에 대해서 살펴보고자 합니다. 여기서 Connection Pool 관리는 Cluster Manager가 하며, 외부에서는 Cluster Entry를 통해 Cluster Manager에서 관리하는 Connection Pool 획득 및 해제등의 요청을 수행할 수 있습니다. 따라서 이 부분에 대해서 자세히 알아보도록 하겠습니다.

 

 

Cluster Manager 내부에는 Connection Pool을 관리하기 위한 map이 여러개 존재합니다. 그 중 host_http_conn_pool_map은 http 요청 관련 Connection Pool을 관리하며, 그 밖에 TCP 기반으로 Connection이 이루어지는 요청의 경우는 host_tcp_conn_pool_map을 통해 Connection Pool을 관리합니다. 본 포스팅에서는 Http 연결 요청에 대한 처리만을 다루고 있으므로 Cluster Manager에서 관리하는 host_http_conn_pool_map에 대해서만 설명하겠습니다.

 

host_http_conn_pool_map은 이름을 통해 알 수 있듯이 Map 자료구조입니다. 이는 HostPtr을 Key로 하며, Value는 Connection을 관리하기 위한 Container로 구성되어있습니다. 그리고 해당 Container 안에는 Connection Pool Map이 별도로 존재합니다.

 

따라서, 외부에서 Connection 연결을 시도하기 위해서는 먼저 host를 기반으로 host_http_conn_pool_map으로 부터 Container를 획득해야하며, 그 안에 존재하는 Connection Map을 통해 Connection 할당 및 해제를 수행할 수 있습니다.

 

그렇다면 Connection 할당 및 해제는 언제 어떻게 이루어질까요?

 

 

Envoy 관련 첫 포스팅 때, Envoy의 내부 흐름은 위와 같이 진행된다고 설명한 적이 있습니다. 여기서 Listener를 통해 Cluster에 접근하고 Load Balancing을 수행하는 과정에서 Cluster Manager를 통해 Endpoint 즉 host가 지정됩니다. 그리고 이 과정에서 Connection Pool로부터 Connection을 할당받아 요청을 전달할 수 있습니다. 정리하자면 Connection Pool로부터 Connection을 할당 받는 시점은 Load Balancing이 완료된 이후입니다.

 

위 그림으로 표시된 영역을 조금 더 자세히 확대하여 구체적으로 내부 컴포넌트가 어떻게 연계되는지 살펴보겠습니다.

 

 

1. Listener를 거쳐 Router에 도달하게되면, Client의 요청이 어떤 Cluster에게 전달되어야하는지 이미 알고 있습니다. 따라서 Router에서는 Cluster Manager에게 자신이 접근하고자 하는 Cluster 정보를 요청합니다.

 

2. Cluster Manager는 자신이 보유하고 있는 thread_local_clusters_ 에서 Router가 요청한 Cluster 정보를 찾습니다.

 

cluster_manager_impl.cc

ThreadLocalCluster* ClusterManagerImpl::getThreadLocalCluster(absl::string_view cluster) {
  ThreadLocalClusterManagerImpl& cluster_manager = *tls_;

  auto entry = cluster_manager.thread_local_clusters_.find(cluster);
  if (entry != cluster_manager.thread_local_clusters_.end()) {
    return entry->second.get();
  } else {
    return nullptr;
  }
}

 

 

참고로 이때 thread_local_clusters_에 저장된 값은 ClusterEntry이지만, ClusterEntry는 ThreadLocalCluster를 상속받기 때문에 리턴 타입은 ThreadLocalCluster로 하여 ThreadLocalCluster에서 제공되는 메소드만 호출 가능합니다.

 

3. Cluster Manager로 부터 ThreadLocalCluster를 찾아서 Router로 반환합니다.

 

4. Router에서는 Cluster내에 존재하는 host를 통해 Connection 연결을 수행해야되기 때문에, 전달받은 ThreadLocalCluster를 통해 Connection Pool 할당을 요청합니다.

 

5. 이를 전달받은 ThreadLocalCluster(ClusterEntry)는 Cluster Manager에게 요청하여 host_http_conn_pool_map 에 할당 받은 Connection Pool이 존재하는지를 확인합니다. 만약 존재한다면 해당 Connection Pool Map에 있는 Container에 접근합니다. 반면 존재하지 않는다면 새로운 Container를 생성하여 해당 Map에 추가합니다.

 

7. Container 내부에는 Cluster 별로 Connection을 관리하는 Pool을 가지고 있습니다. 그리고 여기에서 Pool 존재여부를 최종적으로 확인하고, 해당 Pool을 반환합니다. 해당 과정을 코드로 살펴보면 다음과 같습니다.

 

conn_pool_map_impl.cc

template <typename KEY_TYPE, typename POOL_TYPE>
typename ConnPoolMap<KEY_TYPE, POOL_TYPE>::PoolOptRef
ConnPoolMap<KEY_TYPE, POOL_TYPE>::getPool(const KEY_TYPE& key, const PoolFactory& factory) {
  Common::AutoDebugRecursionChecker assert_not_in(recursion_checker_);

  auto pool_iter = active_pools_.find(key);
  if (pool_iter != active_pools_.end()) {
    return std::ref(*(pool_iter->second));
  }
  ResourceLimit& connPoolResource = host_->cluster().resourceManager(priority_).connectionPools();
  // We need a new pool. Check if we have room.
  if (!connPoolResource.canCreate()) {
    // We're full. Try to free up a pool. If we can't, bail out.
    if (!freeOnePool()) {
      host_->cluster().stats().upstream_cx_pool_overflow_.inc();
      return absl::nullopt;
    }

    ...(중략)...
  }

  auto new_pool = factory();
  connPoolResource.inc();
  for (const auto& cb : cached_callbacks_) {
    new_pool->addIdleCallback(cb);
  }

  auto inserted = active_pools_.emplace(key, std::move(new_pool));
  return std::ref(*inserted.first->second);
}

 

active_pools_가 Container가 보유하고 있는 Pool을 의미합니다. 해당 과정을 살펴보면, 먼저 Container 내부에서 관리하는 active_pools_에서 해당 pool이 존재하는지를 찾습니다. 참고로 여기서 key 값은 protocol을 uint8_t로 캐스팅한 값입니다.

 

그리고 만약 값이 존재하지 않았을 때에는, 해당 Cluster가 Connection Pool을 생성이 가능한지 Resource Limit 설정 값을 살펴봅니다. 그리고 생성이 불가할 경우에는 위 코드와 같이 nullopt를 반환하는 것을 볼 수 있습니다.

 

8. 만약 Resource Limit 설정 값을 확인했을 경우 신규 Pool 생성이 가능한 경우에는 기존에 입력받은 factory 메소드로부터 신규로 Pool을 할당받고자 호출합니다.

 

cluster_manager_impl.cc

Http::ConnectionPool::InstancePtr ProdClusterManagerFactory::allocateConnPool(
    Event::Dispatcher& dispatcher, HostConstSharedPtr host, ResourcePriority priority,
    std::vector<Http::Protocol>& protocols,
    const absl::optional<envoy::config::core::v3::AlternateProtocolsCacheOptions>&
        alternate_protocol_options,
    const Network::ConnectionSocket::OptionsSharedPtr& options,
    const Network::TransportSocketOptionsConstSharedPtr& transport_socket_options,
    TimeSource& source, ClusterConnectivityState& state, Http::PersistentQuicInfoPtr& quic_info) {

  Http::HttpServerPropertiesCacheSharedPtr alternate_protocols_cache;
  if (alternate_protocol_options.has_value()) {
    // If there is configuration for an alternate protocols cache, always create one.
    alternate_protocols_cache = alternate_protocols_cache_manager_->getCache(
        alternate_protocol_options.value(), dispatcher);
  } else if (!alternate_protocol_options.has_value() &&
             (protocols.size() == 2 ||
              (protocols.size() == 1 && protocols[0] == Http::Protocol::Http2)) &&
             Runtime::runtimeFeatureEnabled(
                 "envoy.reloadable_features.allow_concurrency_for_alpn_pool")) {
    // If there is no configuration for an alternate protocols cache, still
    // create one if there's an HTTP/2 upstream (either explicitly, or for mixed
    // HTTP/1.1 and HTTP/2 pools) to track the max concurrent streams across
    // connections.
    envoy::config::core::v3::AlternateProtocolsCacheOptions default_options;
    default_options.set_name(host->cluster().name());
    alternate_protocols_cache =
        alternate_protocols_cache_manager_->getCache(default_options, dispatcher);
  }

  absl::optional<Http::HttpServerPropertiesCache::Origin> origin =
      getOrigin(transport_socket_options, host);
  if (protocols.size() == 3 &&
      context_.runtime().snapshot().featureEnabled("upstream.use_http3", 100)) {
    ASSERT(contains(protocols,
                    {Http::Protocol::Http11, Http::Protocol::Http2, Http::Protocol::Http3}));
    ASSERT(alternate_protocol_options.has_value());
    ASSERT(alternate_protocols_cache);
#ifdef ENVOY_ENABLE_QUIC
    Envoy::Http::ConnectivityGrid::ConnectivityOptions coptions{protocols};
    if (quic_info == nullptr) {
      quic_info = Quic::createPersistentQuicInfoForCluster(dispatcher, host->cluster());
    }
    return std::make_unique<Http::ConnectivityGrid>(
        dispatcher, context_.api().randomGenerator(), host, priority, options,
        transport_socket_options, state, source, alternate_protocols_cache, coptions,
        quic_stat_names_, stats_, *quic_info);
#else
    (void)quic_info;
    // Should be blocked by configuration checking at an earlier point.
    PANIC("unexpected");
#endif
  }
  if (protocols.size() >= 2) {
    if (Runtime::runtimeFeatureEnabled(
            "envoy.reloadable_features.allow_concurrency_for_alpn_pool") &&
        origin.has_value()) {
      ENVOY_BUG(origin.has_value(), "Unable to determine origin for host ");
      envoy::config::core::v3::AlternateProtocolsCacheOptions default_options;
      default_options.set_name(host->cluster().name());
      alternate_protocols_cache =
          alternate_protocols_cache_manager_->getCache(default_options, dispatcher);
    }

    ASSERT(contains(protocols, {Http::Protocol::Http11, Http::Protocol::Http2}));
    return std::make_unique<Http::HttpConnPoolImplMixed>(
        dispatcher, context_.api().randomGenerator(), host, priority, options,
        transport_socket_options, state, origin, alternate_protocols_cache);
  }
  if (protocols.size() == 1 && protocols[0] == Http::Protocol::Http2 &&
      context_.runtime().snapshot().featureEnabled("upstream.use_http2", 100)) {
    return Http::Http2::allocateConnPool(dispatcher, context_.api().randomGenerator(), host,
                                         priority, options, transport_socket_options, state, origin,
                                         alternate_protocols_cache);
  }
  if (protocols.size() == 1 && protocols[0] == Http::Protocol::Http3 &&
      context_.runtime().snapshot().featureEnabled("upstream.use_http3", 100)) {
#ifdef ENVOY_ENABLE_QUIC
    if (quic_info == nullptr) {
      quic_info = Quic::createPersistentQuicInfoForCluster(dispatcher, host->cluster());
    }
    return Http::Http3::allocateConnPool(dispatcher, context_.api().randomGenerator(), host,
                                         priority, options, transport_socket_options, state,
                                         quic_stat_names_, {}, stats_, {}, *quic_info);
#else
    UNREFERENCED_PARAMETER(source);
    // Should be blocked by configuration checking at an earlier point.
    PANIC("unexpected");
#endif
  }
  ASSERT(protocols.size() == 1 && protocols[0] == Http::Protocol::Http11);
  return Http::Http1::allocateConnPool(dispatcher, context_.api().randomGenerator(), host, priority,
                                       options, transport_socket_options, state);
}

 

여기서 할당된 factory 메소드는 ClusterManager에서 관리하는 allocateCoonPool이 최종적으로는 호출되며, Cluster Manager에서는 client가 요구하는 프로토콜이 무엇인지 확인한 다음에 해당 요청을 처리할 수 있는 Connection을 생성하여 반환합니다. 그리고 생성된 Pool을 Container가 보유하고 있는 active_pools_에 삽입합니다.

 

9. 생성된 Pool을 ThreadLocalCluster에 반환합니다.

 

10. Pool을 Router에게 반환합니다. 이후 해당 Router에서는 전달받은 Pool을 통해 Downstream과 Upstream과의 연결 작업을 요청할 수 있게됩니다.

 

위와 같이 10단계를 거치게되면, Router는 Connection Pool을 할당받아 요청을 처리할 수 있게되었음을 확인할 수 있습니다. 반면 Connection 해제는 Router로 전달되었던 Connection Pool을 통해 해제를 요청할 수 있습니다.

 

---

 

6. ADS 관리

 

dynamic_resources:
  lds_config:
    resource_api_version: V3
    ads: {}
  cds_config:
    resource_api_version: V3
    ads: {}
  ads_config:
    transport_api_version: V3
    api_type: GRPC
    grpc_services:
      envoy_grpc:
        cluster_name: envoy-control-xds

 

이전 포스팅에서 xDS API와 함께 ADS (Aggregate Discovery Service)에 대해서 설명했습니다. Cluster Manager의 또 다른 ADS를 생성하고 관리하는 것입니다. 즉 Cluster Manager에서는 xDS API를 위한 Subscription, gRPC Multiplxer 및 ADS 관리를 총체적으로 담당합니다.

---

 

7. Health Check 동작 원리

 

지금까지 Static Resource 그리고 CDS를 통한 Cluster 추가에 대해서 살펴봤습니다. 이번에는 ClusterData를 생성하는 과정에서 Health Check, Outlier Detector가 어떻게 생성되고 동작하는지 살펴보겠습니다. 먼저 살펴볼 것은 Health Check 입니다.

 

 

이전에 살펴봤듯이 ClusterData를 생성하는 과정에서 Outlier Detector, Health Chekcer, Load Balancer 등이 추가됨을 확인할 수 있습니다. 이때 Health Checker를 생성하는 역할은 HealthCheckerFactory가 담당하며, 이후 수행과정을 살펴보면 다음과 같습니다.

 

 

 

Config를 통해 전달받은 Cluster 내부에는 Http 방식 뿐만 아니라 Tcp, gRPC 혹은 Custom 방식의 Health Check를 지정할 수 있습니다. 따라서 HealthCheckerFactory에서는 사용자가 입력한 Config의 Heath Checker 방식을 확인하고 그에 걸맞는 Health Checker를 생성하도록 지정합니다. 본 포스팅에서는 Http 기반으로 생성했음을 가정하였음으로 ProdHttpHealthCheckerImpl을 생성할 것입니다.

 

 

생성된 Health Checker 인스턴스는 ClusterData의 HealthChecker에 바인딩되어 동작합니다.

 

 

cluster_manager_impl.cc

  if (new_cluster->healthChecker() != nullptr) {
    new_cluster->healthChecker()->addHostCheckCompleteCb(
        [this](HostSharedPtr host, HealthTransition changed_state) {
          if (changed_state == HealthTransition::Changed &&
              host->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC)) {
            postThreadLocalHealthFailure(host);
          }
        });
  }

 

Health Checker가 매핑되고 나면 Cluster Manager에서는 Health Checker가 정상적으로 매핑되어있는지를 확인합니다. 그리고 인스턴스가 존재할 경우에는 Health Checker에서 비정상 Health 대상으로 판단한 host에 대해서 ejection 수행을 위한 Callback을 등록합니다.

 

이때 postThreadLocalHealthFailure는 TLS를 통해서 Health가 실패했음을 전달하며, 이를 수신받은 Cluster Manager에서는 해당 host에 연결된 Connection Pool을 해제하는 작업을 수행합니다.

 

 

수행 과정을 살펴보면 위 그림과 같습니다.

 

1. Health Checker에서 주기적으로 Health Check를 수행하다가 이상이 발생하면, 내부 종료 작업을 먼저 거친 뒤에 등록된 Callback을 통해 외부로 전파합니다.

 

2. Cluster Manager에서는 기존에 등록한 Callback 함수를 수행합니다. 이때 내부적으로 ThreadLocalClusterManagerImpl을 참조하여 해당 Host 이상 여부를 전체로 전파합니다.

 

3. Dispatcher를 통해 해당 내역을 전달받은 Cluster Manager에서는 Connection을 해제하기 위하여 Connection Pool에서 해당 host 내부에 존재하는 Connection들을 순차적으로 종료하고 해당 Pool 또한 Close 시킵니다.

 

해당 과정이 정상적으로 완료되면, Health Check에 따른 연결된 Client 및 host 또한 정상적으로 종료되는 것을 이해할 수 있습니다. 그렇다면 Health Check는 어떻게 이루어질까요? Health Check Factory로부터 만들어진 ProdHttpHealthCheckerImpl 인스턴스의 구조와 수행 과정을 살펴보면서 조금 더 자세하게 알아보겠습니다.

 

 

 

1. ProdHttpHealthCheckerImpl 생성자 내부에서 Cluster  데이터변경이 발생했을 때 처리를 위한 Callback을 등록합니다.

 

2. 해당 Callback은 호출될 때 추가되는 host 정보와 삭제되는 host 정보를 같이 전달받는데, 이를 토대로 추가되는 host 정보를 active_sessions_ 에 Session을 만들어 추가합니다. 반대로 삭제되는 host 정보 또한 active_sessions_ 에서 삭제합니다.

 

3. 이때 추가되는 Session에는 Health Check를 정기적으로 수행하기 위해서 Dispatcher에 존재하는 libevent로 부터 Timer를 할당받습니다. 이때 생성되는 timer는 총 2개로 Health Check를 주기적으로 수행하기 위한 interval_timer_와 timeout을 판별하기 위한 timeout_timer_ 를 생성합니다.

 

여기서 Health Check는 Session 내에 생성되는 Timer에 의해 이루어지므로 Session 생성 과정을 보다 자세하게 살펴보겠습니다.

 

 

 

 

onInterval은 HealthCheck를 위해 주기적으로 호출되는 메소드로써, 아래와 같은 4가지 단계를 거칩니다.

 

1. client_가 존재하는지를 살펴보고 client_가 존재하지 않는다면 최초 실행임을 의미하므로 Connection을 생성합니다. 이때 생성된 Connection은 CodecClient를 통해 Wrapping 되는데, CodecClient는 HTTP Codec 타입에 따라서 생성되는 ClientConnectionImpl 인스턴스 타입입니다.

 

2. Stream을 생성합니다.

 

3. Health Check 관련 Header 설정을 진행합니다. 이때 method 및 path 등은 Health Check 설정으로 입력된 값을 따릅니다.

 

4. Upstream 설정 및 Health Check 처리를 위한 encoding 처리를 담당합니다.

 

위 과정이 마무리되면, 주기에 맞추어 Health Check 요청을 HTTP로 전달할 것입니다. 그렇다면 HTTP 요청에 대한 응답은 어떻게 처리될까요?

 

 

HttpActiveHealthCheckSession의 구조를 보면, 위와 같이 여러 Class가 상속되어있음을 확인할 수 있습니다. 그 중 요청 응답 처리는 ResponseDecoder 내부 정의된 I/F에 의해서 호출됩니다. 참고로 해당 과정에서 호출되는 메소드는 decodeHeaders와 decodeData 입니다. 따라서 HttpActiveHealthCheckSession 내부에 정의된 하기 2개의 메소드가 호출됩니다.

 

health_checker_impl.cc

void HttpHealthCheckerImpl::HttpActiveHealthCheckSession::decodeHeaders(
    Http::ResponseHeaderMapPtr&& headers, bool end_stream) {
  ASSERT(!response_headers_);
  response_headers_ = std::move(headers);
  if (end_stream) {
    onResponseComplete();
  }
}

void HttpHealthCheckerImpl::HttpActiveHealthCheckSession::decodeData(Buffer::Instance& data,
                                                                     bool end_stream) {
  if (parent_.response_buffer_size_ != 0) {
    if (!parent_.receive_bytes_.empty() &&
        response_body_->length() < parent_.response_buffer_size_) {
      response_body_->move(data, parent_.response_buffer_size_ - response_body_->length());
    }
  } else {
    if (!parent_.receive_bytes_.empty()) {
      response_body_->move(data, data.length());
    }
  }

  if (end_stream) {
    onResponseComplete();
  }
}

 

위 메소드에서 살펴볼 것은 data가 전달이 완료되지 않았을 경우에는 response_body 데이터를 지속 수신받는 것을 알 수 있고 stream이 종료되었을 경우에는 onResponseComplete() 메소드가 호출됨을 알 수 있습니다. 즉 Health Check를 결정하는 요인은 onResponseComplete() 메소드 내부에서 이루어짐을 짐작할 수 있습니다.

 

 

health_checker_impl.cc

void HttpHealthCheckerImpl::HttpActiveHealthCheckSession::onResponseComplete() {
  request_in_flight_ = false;

  switch (healthCheckResult()) {
  case HealthCheckResult::Succeeded:
    handleSuccess(false);
    break;
  case HealthCheckResult::Degraded:
    handleSuccess(true);
    break;
  case HealthCheckResult::Failed:
    handleFailure(envoy::data::core::v3::ACTIVE, /*retriable=*/false);
    break;
  case HealthCheckResult::Retriable:
    handleFailure(envoy::data::core::v3::ACTIVE, /*retriable=*/true);
    break;
  }

  if (shouldClose()) {
    client_->close();
  }

  response_headers_.reset();
  response_body_->drain(response_body_->length());
}

 

onResponseComplete() 메소드 내부를 살펴보면, healthCheck에 대한 결과를 기점으로 Switch로 분기하여 각기 다른 결과를 호출하는 것을 볼 수 있습니다. 만약 Failure가 발생한다면, 해당 timer를 disabled 시키고 종료처리할 것이며, shouldClose() 호출로 인해 client 연결 또한 종료됩니다.

 

만약 정상적으로 처리되었다면, 지속적으로 Health Check를 주기적으로 반복하여 정상여부 확인을 반복합니다.

 

---

 

8. Outlier Detection 동작 원리

 

 

 

지금까지 Health Checker의 생성 및 동작과정에 대해서 살펴봤다면, 이번에는 Outlier Detector가 생성되는 과정과 동작 방법에 대해 살펴보겠습니다.

 

 

Outlier Detector는 ClusterData 생성 당시에 DetectorImplFactory에 의해서 생성됩니다. 이때 생성되는 Outlier Detector의 인스턴스는 DetectorImpl이며 해당 인스턴스가 ClusterData의 OutlierDetector에 바인딩되어 동작합니다. 

 

해당 내용을 코드를 통해서 살펴보겠습니다.

 

outlier_detection_impl.cc

DetectorSharedPtr DetectorImplFactory::createForCluster(
    Cluster& cluster, const envoy::config::cluster::v3::Cluster& cluster_config,
    Event::Dispatcher& dispatcher, Runtime::Loader& runtime, EventLoggerSharedPtr event_logger,
    Random::RandomGenerator& random) {
  if (cluster_config.has_outlier_detection()) {

    return DetectorImpl::create(cluster, cluster_config.outlier_detection(), dispatcher, runtime,
                                dispatcher.timeSource(), std::move(event_logger), random);
  } else {
    return nullptr;
  }
}

 

다만 위 코드와 같이 모든 ClusterData에서 OutlierDetector를 생성하는 것은 아니며, Static 혹은 Dynamic Resource 내에 outlier_detection 설정이 활성화되어있는 경우에만 생성됩니다. 만약 해당 설정이 비활성화 되어있다면, OutlierDetector는 nullptr이 매핑됩니다.

 

 

cluster_manager_impl.cc

  if (new_cluster->outlierDetector() != nullptr) {
    new_cluster->outlierDetector()->addChangedStateCb([this](HostSharedPtr host) {
      if (host->healthFlagGet(Host::HealthFlag::FAILED_OUTLIER_CHECK)) {
        ENVOY_LOG_EVENT(debug, "outlier_detection_ejection",
                        "host {} in cluster {} was ejected by the outlier detector",
                        host->address()->asStringView(), host->cluster().name());
        postThreadLocalHealthFailure(host);
      }
    });
  }

 

OutlierDetector가 매핑되고 나면 Cluster Manager에서는 Outlier Detector가 정상적으로 매핑되어있는지를 확인합니다. 그리고 인스턴스가 존재할 경우에는 Outlier Detector에서 Ejection 대상으로 판단한 host에 대해서 ejection 수행을 위한 Callback을 등록합니다.

 

이때 postThreadLocalHealthFailure는 TLS를 통해서 Health가 실패했음을 전달하며, 이를 수신받은 Cluster Manager에서는 해당 host에 연결된 Connection Pool을 해제하는 작업을 수행합니다. 해당 작업은 Health Check를 수행하면서 실패했을 때 Connection 해제하는 과정과 완벽하게 동일합니다.

 

지금까지 Outlier Detector 설정 관련해서 살펴봤습니다. 이번에는 실제 Outlier Detector가 수행하는 기능에 대해서 조금 더 자세하게 살펴보기 위해 Outlier로 등록되는 DetectorImpl 구조 및 동작 원리에 대해서 살펴보겠습니다.

 

 

 

이를 위해 DetectorImpl을 생성 과정에 대해서 상세하게 짚어보겠습니다.

 

1. DetectorImpl 내부에서도 정기적으로 Cluster 내부에 존재하는 host들의 상태를 살펴보고 Outlier Detection을 지정한 임계치를 넘어가는 host를 점검하기 위해 Dispatcher로 부터 Timer를 할당받아 interval_timer_에 지정하는 작업을 합니다.

 

2. 그 다음은 외부에서 Cluster 내부 host가 추가되거나 삭제되는 등의 작업이 발생했을 때 Outlier Detector에서도 이를 감지하고 설정을 변경해야되기 때문에 ClusterEntry 내부에 존재하는 PriortySet에 Callback을 추가합니다. 해당 Callback 등록을 통해 실질적으로 Cluster 내부 host 변경이 발생했을 경우 이를 Detector가 통지받아 후속 작업을 수행할 수 있습니다. Callback을 전달받으면, 내부에서는 hosts를 관리하는 host_monitors_ Map 자료구조에 해당 hosts의 변경사항을 기록합니다. 이때 host 별로 DetectorHostMonitorImpl 인스턴스를 생성하여 Outlier Detection을 위한 기본 속성을 저장하는데, 이는 잠시 후 다시 살펴보겠습니다.

 

3. DetectorImpl 생성이 완료되면, ClusterData OutlierDetector에 매핑합니다.

 

4. ClusterData를 active_clusters_에 추가합니다.

 

5. ClusterManager에서 해당 OutlierDetector에게 Callback을 등록합니다. 이는 Outlier Detection 판정이 된 host에 대해서 Client 연결 종료 등의 후속작업을 처리하기 위함입니다.

 

위와 같은 과정을 거치고나면, Outlier Detector는 본격적으로 본인의 역할을 수행할 수 있습니다. 이번에는 초기화가 완료된 이후 후속 작업 진행 절차에 대해서 살펴보겠습니다.

 

outlier_detection_impl.cc

void DetectorImpl::armIntervalTimer() {
  interval_timer_->enableTimer(std::chrono::milliseconds(
      runtime_.snapshot().getInteger(IntervalMsRuntime, config_.intervalMs())));
}

 

먼저 초기화 작업이 끝나고나서 가장 먼저 수행하는 작업은 timer를 활성화 시키는 것입니다. 이는 위 메소드를 호출하여 실행되며, interval_timer_의 수행 주기는 위와 같이 config에 지정된 값을 활용하는 것을 볼 수 있습니다. 그렇다면, interval_timer_를 활성화 시켰을 때 어떤 작업을 수행할까요?

 

outlier_detection_impl.cc

DetectorImpl::DetectorImpl(const Cluster& cluster,
                           const envoy::config::cluster::v3::OutlierDetection& config,
                           Event::Dispatcher& dispatcher, Runtime::Loader& runtime,
                           TimeSource& time_source, EventLoggerSharedPtr event_logger,
                           Random::RandomGenerator& random)
    : config_(config), dispatcher_(dispatcher), runtime_(runtime), time_source_(time_source),
      stats_(generateStats(cluster.info()->statsScope())),
      interval_timer_(dispatcher.createTimer([this]() -> void { onIntervalTimer(); })),
      event_logger_(event_logger), random_generator_(random) {
  // Insert success rate initial numbers for each type of SR detector
  external_origin_sr_num_ = {-1, -1};
  local_origin_sr_num_ = {-1, -1};
}

 

이는 위와 같이 해당 DetectorImpl의 생성자 코드를 살펴보면 힌트를 얻을 수 있습니다. 내용을 살펴보면 dispatcher로부터 timer를 생성받고나서 해당 timer가 호출해야되는 callback이 지정된 것을 확인할 수 있습니다. 이는  위와 같이 onIntervalTimer() 메소드가 지정되어있습니다. 따라서 주기적으로 onIntervalTimer() 메소드가 호출됨으로써 Outlier Detection 작업을 수행함을 알 수 있습니다.

 

outlier_detection_impl.cc

void DetectorImpl::onIntervalTimer() {
  MonotonicTime now = time_source_.monotonicTime();

  for (auto host : host_monitors_) {
    checkHostForUneject(host.first, host.second, now);

    // Need to update the writer bucket to keep the data valid.
    host.second->updateCurrentSuccessRateBucket();
    // Refresh host success rate stat for the /clusters endpoint. If there is a new valid value, it
    // will get updated in processSuccessRateEjections().
    host.second->successRate(DetectorHostMonitor::SuccessRateMonitorType::LocalOrigin, -1);
    host.second->successRate(DetectorHostMonitor::SuccessRateMonitorType::ExternalOrigin, -1);
  }

  processSuccessRateEjections(DetectorHostMonitor::SuccessRateMonitorType::ExternalOrigin);
  processSuccessRateEjections(DetectorHostMonitor::SuccessRateMonitorType::LocalOrigin);

  armIntervalTimer();
}

 

그렇다면, 해당 메소드는 어떻게 구현되어있을까요?

위 코드와 같이 host_monitors_에 매핑된 전체 host들에 대해서 순회하면서 ejection 여부를 판단합니다. 만약 ejection이 결정되었다면, 해당 host_monitors_에 저장된 Monitor 정보를 초기화하고 ejection을 수행합니다. 또한 외부로부터 등록된 callback을 호출시켜, 후속 작업을 진행할 수 있도록 합니다.

 

 

 

참고로 host_monitors_에 매핑된 인스턴스는 DetectorHostMonitorImpl 인스턴스이며, 해당 인스턴스 내부에는 Outlier Detection을 판별하기 위한 여러가지 내부 속성 프로퍼티 값들이 존재합니다. 해당 프로퍼티 값은 Client가 Cluster Manager로부터 Connection Pool을 받아 Upstream host에 Http 요청을 완료한 이후에 전달받은 응답값을 토대로 Router가 갱신을 수행합니다.

 

따라서 Outlier Detector 내부에서 onIntervalTimer()가 수행될 때 해당 Monitor 인스턴스 내부 속성 값을 기준으로 ejection 여부를 판별할 수 있습니다.

 

 

outlier_detection_impl.cc

void DetectorImpl::armIntervalTimer() {
  interval_timer_->enableTimer(std::chrono::milliseconds(
      runtime_.snapshot().getInteger(IntervalMsRuntime, config_.intervalMs())));
}

모든 host_monitors_ 순회가 모두 끝나고 나면 다시 timer를 활성화 시켜 다음 outlier detection 처리를 반복 수행함으로써 지속적인 Outlier Detection 확인이 가능합니다.

 

---

 

9. 마치며

 

이번 포스팅에서는 Cluster Manager의 기능에 대해 일부 살펴봤습니다. Cluster Manager의 가장 중요한 것은 Cluster 관리입니다. 다만 그 외에도 ads 관리 및 subscription factory 등을 담당하는 것을 살펴봤습니다.

 

다음 포스팅에서는 Cluster에 지정되는 Load Balancer 종류에 대해 개념적으로 살펴보도록 하겠습니다. 

1. 서론

 

이전 포스팅에서는 envoy를 이해하는데 아주 기초적인 컴포넌트 종류에 대한 설명을 진행했습니다. 이번 포스팅에서는 envoy 내부 여러 쓰레드가 존재하는데, 쓰레드간에 데이터는 어떻게 이루어지는지 그리고 쓰레딩 모델은 어떻게 구성되어있는지에 대해서 살펴보겠습니다. 본격적으로 진행하기 앞서 envoy는 멀티 쓰레딩 모델이지만 내부적으로 Lock을 통해 데이터 동기화를 수행하지 않습니다. 어떻게 Lock 없이 데이터 동기화가 가능할까요? envoy 쓰레딩 모델을 학습함으로써 궁금증을 해소해보는 시간을 가져보겠습니다.

 

---

2. Threading Model

 

Envoy 내부에는 전체를 관장하는 Main 쓰레드와 Network 요청과 라우팅을 처리하기 위한 여러 Worker 쓰레드가 존재합니다. 또한 그 외에도 파일 동기화를 위해 별도 File Flush 쓰레드가 별도로 존재합니다.

 

정리하자면 envoy 내부에는 Main, Worker, File Flush 쓰레드 총 3가지로 구성되었다고 볼 수 있습니다.

 

 

 

여기서 Main 쓰레드는 envoy 내부의 xDS 정보를 관리하고, Admin Server, Health Check 등 Envoy 프로세스가 수행하는데 있어 중추적인 데이터를 관리하고 부가적인 Admin 기능을 제공합니다. 따라서 Envoy를 기동하는데 있어 주요 Metadata 정보는 Main 쓰레드에서 생성되며, 필요 시 다른 쓰레드로 해당 정보를 공유합니다.

 

그렇다면 envoy 내부에서 Main 쓰레드와 다수로 생성되는 Worker 쓰레드간의 데이터 공유가 필요할 경우에는 어떻게 해야할까요?

 

 

첫 번째로 생각해볼 수 있는 방법은 쓰레드간 공유 Data를 생성한 다음에 Main 쓰레드에서 데이터 변경이 필요한 경우 Worker 쓰레드가 해당 공유 Data 접근이 모두 끝나는 시점에 Exclusive Lock을 걸어 Worker 쓰레드의 접근을 일시적으로 차단한 다음 데이터를 변경하여 쓰레드 안정성을 확보하는 방법입니다. 하지만 위와 같은 방식은 동시성이 떨어지는 단점이 존재합니다.

 

 

두 번째 방법은 모든 쓰레드에 걸쳐 동일한 데이터를 가지고 있는 상태에서 Data 변경이 발생하면, Main 쓰레드에서 데이터를 변경한 다음 모든 Worker 쓰레드에 수정된 Data를 전달하는 것입니다. 이 경우에는 특정 Worker 쓰레드가 바라보고 있는 원본 Data에 대해 Main 쓰레드에서 수정을 가해도 Worker 쓰레드가 바라보는 원본 Data는 해당 시점에 변경되지 않기 때문에 쓰레드 안정성을 확보할 수 있습니다.

 

위와 같이 설계된 디자인에서는 쓰레드 외부간에는 데이터가 공유되지 않지만, 내부에서는 해당 데이터를 전역적으로 공유해서 사용할 수 있습니다. 이러한 기법을 Thread Local이라고 부릅니다.

 

 

다만 위 예시에서는 1개의 데이터에 대한 동기화 과정을 표현했는데, envoy 내부에는 Main과 Worker 쓰레드 간의 xDS 업데이트를 위해 사용되는 데이터 항목이 많습니다. 따라서 실제로는 위 그림과 같이 하나의 데이터 인스턴스가 아니라 여러개의 Vector 자료 구조로 개별 쓰레드 별로 관리되어야 할 것입니다.

 

위 그림과 같이 데이터를 관리하면, Lock 없이 쓰레드 안정성을 다소 확보하는 것을 확인할 수 있었습니다. 그렇다면 여기에는 비효율이 없을까요?

 

가만 생각해보면, 개별 쓰레드 별로 동일한 데이터를 모두 가지고 있기 때문에 쓰레드 안정성은 확보할 수 있으나 똑같은 데이터가 모든 쓰레드에서 독립적으로 존재하기 때문에 메모리 낭비가 커지는 단점이 존재합니다.

 

따라서 envoy에서는 이를 해결 하기 위해서 Shared Pointer와 자체 TLS(Thread Local Storage)를 구축하였습니다. 이와 관련해는 조금 뒤에 더 자세히 다루어보겠습니다.

---

3. Distpatcher

 

Envoy의 TLS(Thread Local Storage)를 살펴보기 위해서는 내부의 중추적인 역할을 수행하는 Dispatcher에 대해서 알아야합니다. 따라서 TLS를 다루기 이전에 Dispatcher에 대해서 알아보고 넘어가겠습니다.

 

이전에 Threading Model을 통해서 메인 쓰레드와 Worker 쓰레드 사이에 데이터 공유가 이루어졌음을 확인했습니다. 그렇다면 Worker 쓰레드 입장에서는 메인 쓰레드가 보낸 데이터를 어떻게 확인하고 후속 작업을 처리할까요?

 

 

envoy에서는 Dispatcher가 중간 매개체 역할을 해줍니다. 여기서 Dispatcher는 Main 쓰레드에서 보낸 데이터를 수신받으면, 이를 감지하고 Worker 쓰레드에게 통지하여 후속 작업이 진행될 수 있도록 가교 역할을 수행합니다. 

 

 

 

해당 과정을 조금 더 자세히 살펴보면 위 그림과 같습니다.

 

envoy에서 Main 쓰레드를 포함하여 모든 쓰레드에는 Dispatcher가 존재합니다. 해당 Dispatcher는 자체적인 event loop를 가지고 있어서 별도의 life cycle이 존재합니다. 

 

이 경우 가령 위 그림과 같이 Main 쓰레드에 Event가 발행할 경우 Main 쓰레드는 개별 Worker 쓰레드에 존재하는 Dispatcher에게 Event의 Callback 내용을 전달합니다. Worker 쓰레드에는 전달된 Event를 내부 Buffer에 저장하며, 저장된 Buffer에서 하나씩 Event를 추출하여 작업을 수행합니다.

 

그렇다면 Dispatcher의 내부는 어떻게 구성되어있을까요?

 

 

envoy Dispatcher 내부에는 자체적인 event loop가 있다고 했는데, 해당 기능은 자체 구현하지 않고 C언어의 라이브러리인 libevent를 사용합니다. 다만 libevent가 C언어로 구성되어있기 때문에 메모리 관리를 위해 libevent에서 제공하는 자료구조를 unique_ptr 스마트 포인터로 감싸서 자동으로 관리할 수 있도록 Wrapping 하였습니다. 따라서 Dispatcher의 핵심은 libevent라고 볼 수 있습니다.

 

libevent 라이브러리는 다양한 기능을 제공하는데, 핵심 기능은 event가 발생되었을 때 이를 감지하고 알림을 전달해주는 역할을 수행합니다. libevent 공식 홈페이지를 보면 제공하는 기능에 대해서 보다 상세하게 소개되어있으며, 이에 대해서는 아래 공식 홈페이지를 참조하시기 바랍니다.

 

https://libevent.org/

 

libevent

The libevent API provides a mechanism to execute a callback function when a specific event occurs on a file descriptor or after a timeout has been reached. Furthermore, libevent also support callbacks due to signals or regular timeouts. libevent is meant t

libevent.org

 

 

지금까지 Dispatcher를 통한 Event 전달 과정과 해당 기능을 수행하는 libevent에 대해서 알아봤습니다. 이번에는 Dispatcher에서 Event를 전달하는 부분을 코드를 살펴보면서, 조금 더 수행 과정에 대해서 들여다보겠습니다.

 

 

 

 

 

먼저 쓰레드간에 데이터 공유 시, 쓰레드 별로 작업 부하가 다를 수 있기 때문에 전달된 Event가 즉시 처리될 수 없습니다. 따라서 Dispatcher 내부에는 Buffer를 위해 위와 같이 post_callbacks_ 라는 자료 구조에 Event Callback을 저장합니다.

 

 

위 코드를 통해서 방금 전 설명한 내용처럼 Dispatcher의 post를 통해서 Callback Event가 전달되면, 바로 처리하지 않고 내부에 존재하는 post_callbacks_ 자료구조에 해당 callback 함수를 저장하는 것을 확인할 수 있습니다.

 

그렇다면, 해당 callback은 언제 실행될까요?

 

 

위 코드는 Dispatcher의 생성자 부분입니다. 여기서 Dispatcher를 구성하기 위해 다양한 속성들이 Member Initializer에 의해서 매핑되고 있지만, Event 전달 관점에서 주목할 부분은 빨간색으로 밑줄 친 영역입니다. 

 

Dispatcher 내부에는 Scheduler가 존재하는데, post callback을 처리하기 위해서 별도 Scheduler를 통해 수행할 수 있습니다. 이때 위와 같이 runPostCallbacks()라는 메소드를 호출하는 것을 확인할 수 있습니다.

 

 

해당 메소드의 역할을 살펴보면, post_callbacks_에 저장된 내용이 하나도 없을 때까지 Event 내용을 하나씩 꺼내서 실행시키는 역할을 수행합니다. 이후 모든 작업이 완료되면, 향후 Post 메소드를 통해 Event가 유입될 때까지 작업을 진행하지 않습니다.

 

지금까지 Dispatcher 내부 동작 과정에 대해서 살펴봤습니다. 이번에는 Dispatcher가 취급하는 Event 종류에 대해서 알아보겠습니다.

 

 

 

 

Dispatcher를 통해 전달되는 Event는 모두 위 클래스를 상속받습니다. 해당 클래스 내용을 잠시 살펴보면, event 타입의 프로퍼티를 기본적으로 가지고 있으며, 해당 클래스가 소멸할 때 event를 release 하도록 되어있습니다.

 

 

해당 event를 상속하는 Event 타입은 위 그림과 같이 크게 3가지로 나뉩니다.

 

첫 번째는 Signal Event입니다. 이는 생성 즉시 Event를 발행할 때 사용되는 Event 타입입니다.

 

두 번째는 Timer Event입니다. 해당 Event는 내부 속성에 Timer가 존재하여, 지정된 시간이 지난 이후 Event를 발행할 수 있으며, Timer를 활성화하거나 비 활성화 등을 통해서 Timer를 조정할 수 있습니다.

 

세 번째는 File Event입니다. 해당 Event 타입은 File에 읽기 혹은 쓰기등의 내부 Event가 발생했을 때, 해당 Event를 전달하는 것을 목적으로 합니다. 또한 Linux의 경우 Socket은 File로 취급됩니다. 따라서 Socket을 쓰거나 읽을 때, 해당 Event를 감지하여 전달할 수 있습니다.

 

envoy에서는 위 세가지 Event 타입을 폭넓게 활용하여 여러 쓰레드간에 데이터 공유에 활용합니다. 

 

 

---

4. TLS(Thread Local Storage)

 

Threading Model을 설명하면서, 쓰레드마다 동일한 데이터를 보관할 경우 메모리 낭비로 인해 비효율이 발생함을 설명하고 이를 위해서 envoy에서는 TLS를 사용한다고 언급했습니다. envoy의 TLS에 대해서 알아보면서, envoy가 어떻게 메모리 낭비를 줄이기 위해 최적화 기법을 사용했는지를 살펴보겠습니다.

 

 

TLS는 말 그대로 Thread Local 객체들을 저장하는 저장소이므로 가장 먼저 해야할 일은 TLS에 저장되는 타입에 대한 정의가 필요합니다.

 

 

 

그에 따라 위와 같이 TLS에 저장하려는 데이터 타입에는 envoy에서 정의한 ThreadLocalObject 타입을 필수적으로 상속 받아야합니다. 그렇다면 ThradLocalObject 클래스는 어떻게 정의되어있을까요?

 

 

ThreadLocalObject 클래스를 살펴보면, 그 안에 어떠한 속성도 정의되어있지 않으며 asType 메소드가 존재하여 지정된 타입으로 캐스팅하여 반환하는 기능만을 제공하는 것을 확인할 수 있습니다.

 

또한 위 코드에서 주목할 부분은 ThreadLocalObject를 Shared Pointer로 감싼 타입을 새로 정의했다는 것입니다.

 

 

 

위 코드는 각 쓰레드별로 저장되는 TLS 구조의 모습입니다. 해당 구조체에는 각 쓰레드별로 구동되는 Dispatcher 정보를 저장할 수 있는 속성과, 이전 코드에서 확인한 ThreadLocalObject를 Shared Pointer로 감싼 타입을 Vector Container로 가지고 있는 것을 확인할 수 있습니다.

 

그렇다면 위와 같이 Shared Pointer 형태로 데이터를 관리하면 어떤 이점이 있을까요?

 

 

 

첫 번째로 얻을 수 있는 이점은 데이터 자체를 저장하는 것이 아니라 Pointer를 지니고 있으므로 데이터 중복을 줄일 수 있으며, 메모리 주소 값을 저장하므로 비용이 많이 절감됩니다.

 

두 번째는 Shared Pointer로 데이터를 가르키는 것이기 때문에 메모리 관리를 자동으로 수행할 수 있습니다. 

 

 

가령 위와 같이 Shared Pointer로 가르키는 오브젝트에 대한 링크가 모두 해제되었을 경우 참조 카운트가 0이므로 자동으로 메모리를 해제할 수 있습니다.

 

그렇다면 위와 같이 Shared Pointer를 사용하게 되면 모든 문제가 해결될까요? 정답은 그렇지 않습니다. 위와 같이 구성하게 되면 결국 처음에 야기되었던 문제점이 다시 발생합니다. 그 이유는 결국 공유 데이터를 가르키는 것이기 때문에 결론적으로는 쓰레드 안정성에 대한 이슈가 다시 불거지기 때문입니다.

 

따라서 envoy에서는 기존에 살펴본 Dispatcher와 각 쓰레드별로 Shared Pointer를 지정하도록 구성함으로써 이러한 문제를 해결하고자 했습니다. 이와 관련해서는 다음 시나리오를 통해서 확인해보겠습니다.

 

 

위와 같이 두 개의 쓰레드에서 서로 동일한 Thread Local Object를 가르키고 있다고 가정해봅시다. 이러한 상황에서 Thread Local Object의 업데이트가 필요할 경우 다음과 같은 과정을 수행합니다.

 

 

 

 

1. 변경을 수행해야하는 쓰레드에서 Thread Local Object를 새로 만들고 기존 Thread Local Object를 가르키던 포인터를 신규 Object를 가르키도록 변경합니다.

 

2. 변경을 수행한 쓰레드에서 다른 쓰레드에게 변경 내용을 전파합니다. 따라서 해당 쓰레드의 Dispatcher에게 post 요청을 통해 변경 Event를 전달합니다.

 

3. Dispatcher는 전달된 Event 내역을 자신의 buffer에 저장하였다가 쓰레드에게 데이터 변경 내용을 전달합니다.

 

 

4. Event를 전달받은 쓰레드에서는 Event를 통해 전달된 Callback을 수행하면서 자신이 가르키고 있는 Thread Local Object를 신규로 변경합니다. 이 과정을 통해 모든 쓰레드에서 새로운 Thread Local Object로 변경이 완료되면, 기존의 Thread Local Object의 참조 카운트는 0이 되므로 삭제할 수 있습니다.

 

위와 같이 4가지 과정을 수행하면, 쓰레드 안정성을 확보할 수 있습니다. 그 이유는 첫 번째 단계를 살펴보면, 두 개의 쓰레드가 공유 데이터를 바라보고 있다가 특정 쓰레드에서 데이터 변경이 수행되었을 경우 다른 쓰레드에 영향을 미치지 않기 때문입니다. 이후 다른 쓰레드의 event loop 수행 과정에서 변경 내용을 통지받으면 그제서야 변경된 데이터를 바라볼 수 있도록 적용됩니다.

 

이는 데이터 변경이 발생되었을 때 동일 시점에 모든 쓰레드에서 공유 데이터를 바라볼 수 있는 Strict Consistency는 포기하더라도 Eventually Consistency를 적용함으로써 쓰레드 안정성, 메모리 효율성과 성능을 얻는 것이 이점이 크기 때문입니다.

 

지금까지 TLS에 대해서 기본적인 동작 과정에 대해서 개략적으로 살펴봤습니다. 이번에는 조금 더 구체적으로 Thread Local Data를 활용해서 Envoy내 쓰레드 끼리 어떻게 동기화를 수행하는지에 대해서 살펴보겠습니다.

 

---

4-1 Envoy TLS 동작 과정

 

Envoy 내부에는 TLS를 관장하기 위해 InstanceImpl 인스턴스가 존재합니다. 이를 위해 해당 클래스 내부에는 TLS와 쓰레드간 상호작용을 위한 여러 메소드와 속성이 포함되어있습니다. 코드를 토대로 해당 클래스에 대해서 자세하게 살펴보겠습니다.

 

 

위 코드에서 먼저 눈에 띄는 것은 TLS를 위한 자료구조인 ThreadLocalData 구조체가 해당 클래스내에 포함된 것을 확인할 수 있습니다. 또한 ThreadLocalData는 C++의 thread_local 키워드가 붙어있어 쓰레드내 전역적으로 공유되는 데이터임을 알 수 있습니다.

 

 

 

그 다음 눈 여겨 볼 점은 dispatcher를 저장하는 점과 registered_threads_ 속성을 통해 해당 쓰레드와 연결되는 쓰레드들을 관리한다는 점입니다.

 

void InstanceImpl::registerThread(Event::Dispatcher& dispatcher, bool main_thread) {
  ASSERT_IS_MAIN_OR_TEST_THREAD();
  ASSERT(!shutdown_);

  if (main_thread) {
    main_thread_dispatcher_ = &dispatcher;
    thread_local_data_.dispatcher_ = &dispatcher;
  } else {
    ASSERT(!containsReference(registered_threads_, dispatcher));
    registered_threads_.push_back(dispatcher);
    dispatcher.post([&dispatcher] { thread_local_data_.dispatcher_ = &dispatcher; });
  }
}

 

실제 Thread 등록은 Worker 쓰레드가 생성되는 시점에 registerThread 메소드를 호출하여 InstanceImpl에 등록됩니다. 이때 호출되는 코드를 살펴보면, 등록되는 Thread가 Main 쓰레드인 경우와 그렇지 않은 경우가 분리되어서 저장되는 것을 확인할 수 있습니다.

 

 

그밖에 InstanceImpl 코드를 더 살펴보면,  그 전에는 보지 못했던 Slot 구조체와 이를 저장하는 slots_ Vector Container가 존재하는 것을 알 수 있습니다. Slot은 무엇이고 어떤 역할을 수행할까요?

 

 

InstainceImpl은 ThreadLocalData를  thread_local 속성으로 가지고 있지만, 사용자가 ThreadLocalData에 데이터를 할당하고자 할때 직접 할당하는 것을 방지하도록 설계되어있습니다.

 

대신 Client는 ThreadLocalData에 대한 추가,수정,삭제 작업을 희망할 경우 InstanceImpl로 부터 Slot을 할당 받아야합니다. 여기서 Slot은 ThreadLocalData에 작업 수행 및 ThreadLocalData를 공유하는 모든 쓰레드에게 명령을 요청하기 위한 일종의 interface라고 볼 수 있습니다.

 

SlotPtr InstanceImpl::allocateSlot() {
  ASSERT_IS_MAIN_OR_TEST_THREAD();
  ASSERT(!shutdown_);

  if (free_slot_indexes_.empty()) {
    SlotPtr slot = std::make_unique<SlotImpl>(*this, slots_.size());
    slots_.push_back(slot.get());
    return slot;
  }
  const uint32_t idx = free_slot_indexes_.front();
  free_slot_indexes_.pop_front();
  ASSERT(idx < slots_.size());
  SlotPtr slot = std::make_unique<SlotImpl>(*this, idx);
  slots_[idx] = slot.get();
  return slot;
}

 

Slot을 할당받기 위해서는 Client는 InstanceImpl에게 allocateSlot() 메소드를 호출할 수 있습니다. 이때 내부적으로는 비어있는 Slot을 관리하는 free_slot_indexes가 존재하는데, free한 Slot이 비었을 경우 Slot을 추가로 생성해서 반환하는 것을 볼 수 있습니다. 반면 비어있는 Slot이 존재한다면, 해당 Slot의 index를 기반으로 Slot을 반환합니다.

 

 

위 코드에서 주목할 부부은 Slot을 생성할 때, 두 번째 인자에 비어있는 Slot의 index를 지정한다는 점입니다. 따라서 Client 입장에서 전달받은 Slot에서는 Instance 내부에 존재하는 slots_  list에서 몇 번째 index에 Slot이 생성되었는지 알 수 있습니다.

 

void InstanceImpl::SlotImpl::set(InitializeCb cb) {
  ASSERT_IS_MAIN_OR_TEST_THREAD();
  ASSERT(!parent_.shutdown_);

  for (Event::Dispatcher& dispatcher : parent_.registered_threads_) {
    // See the header file comments for still_alive_guard_ for why we capture index_.
    dispatcher.post(wrapCallback(
        [index = index_, cb, &dispatcher]() -> void { setThreadLocal(index, cb(dispatcher)); }));
  }

  // Handle main thread.
  setThreadLocal(index_, cb(*parent_.main_thread_dispatcher_));
}

 

Client는 Slot을 할당 받으면, Thread Local에 저장하기 위한 Callback Event를 등록할 수 있습니다. 이때 위 코드와 같이 set 메소드를 통해 저장할 수 있으며, 코드 내용을 자세히 살펴보면, 등록된 Thread를 모두 순회하면서 dispatcher에게 post 메소드를 호출하여, 각자 쓰레드가 보유한 setThreadLocal을 호출하도록 요구하는 것을 볼 수 있습니다.

 

해당 과정을 통해서 쓰레드 내의 변경 사항을 다른 쓰레드에 전파하게 됩니다.

 

void InstanceImpl::setThreadLocal(uint32_t index, ThreadLocalObjectSharedPtr object) {
  if (thread_local_data_.data_.size() <= index) {
    thread_local_data_.data_.resize(index + 1);
  }

  thread_local_data_.data_[index] = object;
}

 

setThreadLocal 코드는 위와 같습니다. 해당 코드 내용을 보면 ThreadLocalData 구조체 내에있는 data_에 접근하여 Slot에 지정된 index에 해당하는 위치에 Callback을 저장하는 것을 볼 수 있습니다.

 

 

 

Client에서 수행할 수 있는 두 번째 요청은 runOnAllThreads 입니다. InstanceImpl로 부터 전달받은 Slot에는 index 값과 더불어 InstanceImpl에 대한 parent_ 속성에 저장되어있습니다.

 

void InstanceImpl::SlotImpl::runOnAllThreads(const UpdateCb& cb) {
  parent_.runOnAllThreads(dataCallback(cb));
}

 

따라서 Client 입장에서 전달받은 Slot에 runOnAllThreads 명령을 요청하면, 내부적으로는 InstanceImpl에게 runOnAllThreads를 실행시키도록 위임합니다.

 

void InstanceImpl::runOnAllThreads(Event::PostCb cb) {
  ASSERT_IS_MAIN_OR_TEST_THREAD();
  ASSERT(!shutdown_);

  for (Event::Dispatcher& dispatcher : registered_threads_) {
    dispatcher.post(cb);
  }

  // Handle main thread.
  cb();
}

 

요청을 전달받은 InstanceImpl에서는 이전과 마찬가지로 registered_threads_에 등록된 dispatcher 들을 순회하면서 사용자가 전달한 Callback을 전달하고 Main 쓰레드 내에서 해당 Callback 또한 실행시킴으로써 연관된 모든 쓰레드에서 사용자 요청을 처리하도록 작업을 수행합니다.

 

지금까지 InstanceImpl을 통해 쓰레드를 등록, Thread Local Object 할당 및 Dispatcher를 활용한 데이터 공유 과정에 대해서 살펴봤습니다. 이번에는 Envoy가 종료될 때 TLS 초기화가 어떻게 이루어지는지 살펴보겠습니다.

 

 

void InstanceImpl::shutdownGlobalThreading() {
  ASSERT_IS_MAIN_OR_TEST_THREAD();
  ASSERT(!shutdown_);
  shutdown_ = true;
}

 

Envoy가 종료될 때, InstanceImpl에 suhtdownGlobalThreading() 메소드를 호출하여 종료를 요청합니다. 다만 해당 메소드의 역할은 InstanceImpl 내부에 존재하는 shutdown_ bool 값을 true로 변경하는 것입니다.

 

void InstanceImpl::shutdownThread() {
  ASSERT(shutdown_);
  for (auto it = thread_local_data_.data_.rbegin(); it != thread_local_data_.data_.rend(); ++it) {
    it->reset();
  }
  thread_local_data_.data_.clear();
}

 

이후 개별적인 Worker 쓰레드는 shutdownThread() 메소드를 호출함으로써 쓰레드 별로 종료 과정이 이루어집니다. 이때 위 코드에서 주목할 부분은 ThreadLocalData 내부 데이터를 종료할 때, 뒤에서부터 reset 과정이 진행되는 점입니다.

 

그 이유는 ThreadLocalData를 구성할 때 가장 먼저 적재되는 데이터는 ClusterManager에서 생성되는 ThreadLocalObject이고 해당 데이터들은 종료될 때까지 수정이 이루어지지 않는 데이터이면서, 여러 Object에서 참조될 수 있는 가장 중요한 데이터일 수 있습니다.

 

따라서 중요도가 가장 낮은 Object 부터 종료를 시작해서 가장 높은 Object를 나중에 종료시킴으로써 의존성이 강한 Object가 먼저 삭제되어 발생될 수 있는 문제를 예방합니다.

 

---

5. 마치며

 

지금까지 Envoy 쓰레드 모델과, TLS에 대해서 살펴봤습니다. TLS를 이해하기 위해서는 Dispatcher의 기능 중 일부에 이해해야하기 때문에 이 부분에 대해서도 다루어봤습니다.

 

또한 코드를 살펴보면서 Envoy에서 Lock을 사용하지 않고도 공유 데이터를 사용하기 위해 많은 고민이 반영되었음을 확인할 수 있었습니다.

 

Envoy의 쓰레딩 모델과 TLS를 이해하는 것은 istio를 이해하는데 있어서도 중요합니다. Envoy 내부에서 xDS API 업데이트가 발생되었을 때 내부적으로 TLS를 통해서 데이터 변경을 전파하고 반영하기 때문입니다. 그러한 측면에서 이번 포스팅 내용은 어렵지만 중요하다고 볼 수 있습니다.

1. 서론

 

istio를 처음 사용하면, 이전과 달리 설치가 및 설정 방법이 편해져서 사용하기 쉽습니다. 그럼에도 불구하고 istio를 운영하는 것은 여전히 어려운 일입니다. 그 이유는 istio에 문제가 생겼을 때 이를 해결하기 위해서는 istio 내부 아키텍처에 대한 이해가 필수적이기 때문입니다.

 

istio는 엄밀히 말해서 envoy proxy를 관리하는 컨트롤러라고 볼 수 있습니다. 따라서 istio 뿐만 아니라 envoy proxy의 내부 구조에 대해서도 잘 알고 있어야합니다. 이번 포스팅부터 진행되는 envoy internals 시리즈는 envoy 내부 구조와 흐름에 대해서 살펴보면서, 트러블 슈팅을 위한 인사이트를 갖는 것을 목적으로 하고 있습니다. 다만 모든 내용을 다루지는 않으며, istio를 이해하는데 있어 필수적인 부분에 대해서만 살펴보겠습니다. 이번 포스팅은 envoy 관련 첫번째 포스팅으로 envoy 구조와 내부 컴포넌트 동작 원리에 대해서 알아보도록 하겠습니다.

 

---

 

2. Envoy 컴포넌트

 

 

Envoy는 Proxy 프로그램으로 사용자와 Service 중간에서 Proxy 역할을 수행합니다. 이때 Client로부터 전달받는 트래픽은 Downstream, 전달받은 요청을 Service로 전달하는 트래픽을 Upstream이라고 부릅니다. 즉 Envoy를 기준으로 상위 서비스 전달은 Upstream Envoy로 흘러 들어오는 스트림은 Downstream입니다.

 

그렇다면 Downstream을 통해 전달되는 트래픽이 Envoy의 어떤 과정을 거쳐서 Upstream으로 Service에 전달될까요?

먼저 Envoy의 주요 컴포넌트에 대해서 먼저 알아보겠습니다.

 

 

 

Envoy 내부에서 가장 핵심이 되는 컴포넌트는 위 그림과 같습니다. 도식화된 그림을 살펴보면 여러개 Listener 그리고 요청을 전달하기 위한 Route 과정 그리고 해당 Route 과정을 통해서 Cluster에 전달되고 Cluster는 Load Balancing 정책에 따라서 자신이 보유하고 있는 Endpoint 중 하나를 선정합니다. 결과적으로는 해당 Endpoint에 매칭되는 Service에 트래픽이 전달됩니다. 

 

갑자기 여러가지 컴포넌트가 등장했는데요. 지금부터 하나하나씩 살펴보겠습니다.

 

---

2-1 Endpoint

 

 

먼저 살펴볼 것은 Endpoint입니다. Endpoint는 위 그림과 같이 Proxy를 통해 연결해야하는 최종 목적지 주소와 Port 번호를 의미합니다. 위와 같이 address는 IP일 수도 있고 혹은 도메인 네임일 수도 있습니다. 그 밖에 health check를 위한 설정 및 Load Balancing을 수행할 때 가중치 와 우선순위 등을 설정할 수 있습니다. 자세한 내용은 Envoy 공식 문서를 통해서 확인하시기 바랍니다.

 

---

 

2-2 Cluster

 

 

Service는 가용성 혹은 성능 향상의 목적으로 여러 서버에 동일한 Service를 배포합니다. 따라서 단일 Endpoint를 통해 여러개 Service를 관리할 수 있는 논리적인 집합 단위가 필요합니다.

 

Envoy Proxy에서는 Cluster 컴포넌트를 통해 Endpoint들을 그룹핑하여 관리할 수 있습니다. 해당 컴포넌트를 통해서 Endpoint 중에서 어디로 트래픽을 보낼지 결정하는 Load Balancing을 결정할 수 있습니다.

 

  clusters:
  - name: some_service
    connect_timeout: 0.25s
    type: STATIC
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: some_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 1234

 

위 설정은 Envoy proxy 공식 문서의 설정 예시입니다. Cluster 하위에 Endpoint 목록을 지정하도록 되어 있는 것을 확인할 수 있습니다. 달리 말하면 Cluster와 Endpoint 간에는 의존성이 존재함을 확인할 수 있습니다. 위 설정에는 Endpoint 지정외에도 다양한 설정을 지정할 수 있습니다. 가령 Circuit Breaker 설정 HTTP Connection과 DNS refresh와 Resolving 정책 등의 부가적인 옵션을 해당 컴포넌트를 통해서 설정할 수 있습니다. 자세한 옵션 설정은 Envoy 공식 문서를 통해서 확인하시기 바랍니다.

 

---

 

2-3 Listener

 

 

Listener는 Envoy Proxy가 어떤 address의 어떤 port로 접속하는 요청에 대해서 Proxy 처리를 수행할 것인지를 설정합니다. 가령 위와같이 지정했다면, 해당 Envoy Proxy가 위치한 서버로 접속하는 80 Port 요청에 대해서 Envoy Proxy가 트래픽을 전달받고 후속 작업을 처리하게됩니다. 즉 Listener는 Envoy Proxy로 흐름을 전달하는 문지기 역할이라고 볼 수 있습니다.

 

 

Listener를 통해 트래픽이 전달되었다면, 이를 Cluster에 전달하기 위해서는 여러 내부 과정을 거쳐야합니다. Listener에는 Listener Filters와 Filter Chains(Network Filters)를 지니고 있습니다. 따라서 실제 트래픽이 전달되었을 때 Listener 내부에 위치한 Filter들을 통과하면서 사용자 요청을 분석하고 어떤 Cluster로 전달해야할지 등을 결정합니다. 그렇다면 Listener Filter와 Filter Chains는 무엇이 다를까요?

 

Listener Filters는 Connection에 대한 Metadata를 조작하거나 추가하는 데 사용되며, 변경된 정보를 토대로 Filter Chains에 존재하는 무수한 Filter 중에서 해당 요청을 처리하는데 적합한 Filter를 선정하는데 사용됩니다. 즉 실제 사용자 요청을 처리하는 것은 FIlter Chains에 존재하는 Filter이지만 해당 Filter를 사용하기 위해서 사전에 보조적인 작업을 담당하는 것이 Listener FIlter라고 볼 수 있습니다.

 

참고로 Envoy에서 제공하는 Listener Filter 목록은 위와 같으며, 개별 Filter의 역할은 다음과 같습니다.

 

Filter	역할
HTTP Inspector	Application에서 전달한 Traffic을 분석하여 해당 네트워크 요청이 HTTP인지 확인합니다. 또한 HTTP 요청이 맞다면, HTTP 1.1 요청인지 혹은 HTTP 2 요청인지를 분석합니다. 이를 토대로 추후 네트워크 요청에 적합한 Filter Chain을  찾는데 사용됩니다.
Original Destination	IpTable에 의해서 redirect된 소켓의 원래 목적지 값 주소를 알기 위해 SO_ORIGINAL_DST 값을 읽는 역할을 수행합니다.  해당 값은 Envoy 처리 이후 Connection Local address로 설정하는데 사용됩니다.
Original Source	Client가 Envoy의 주소로 Downstream 연결을 시도하면,  Envoy는 Upstream과 통신을 위해서는  Source IP를 Envoy의 주소로 변경이 필요합니다. 따라서 해당 과정을 통해 연결의 목적지 주소를 Source 주소로 복제하는데 사용합니다.
Proxy Protocol	해당 Listener Filter는 HA Proxy Protocol을 지원하기 위한 역할을 수행합니다.
TLS Inspector	HTTP Inspector와 유사하게 해당 요청이 TLS 요청인지를 확인합니다. 이를 토대로 추후 네트워크 요청에 적합한 Filter Chain을 찾는데 사용됩니다.

 

Listener Filter를 통해서 Connection 요청 Metadata를 조작하거나 어떤 요청인지를 분석하고난 이후에는 Filter Chains(Network Filters)에 위치한 FIlter들을 통과하면서 사용자의 요청에 적합한 Filter를 찾아서 수행합니다.

 

 

기본적으로 제공하는 FIlter Chains 목록은 위와 같습니다. 해당 항목들은 L3/L4 Filter 기능을 담당하며, 사용자 요청에 적합한 Filter를 찾아서 처리하고 목적지로 전달하는데 사용합니다. 만약 위 Filters 중 사용자 요청을 처리할 수 없는 경우에는 Envoy에서 제공하는 Default Chain이 사용되며, 만약 Default Chain 설정을 하지 않았다면 해당 Connection은 종료됩니다. 참고로 envoy proxy 기동시 위와 같이 모든 Filter가 등록되는 것은 아니며 순서 또한 다를 수 있습니다. 위 Filter 목록 중 필요한 Filter만 선별적으로 등록 가능합니다. 이에 대해서는 추후 살펴보겠습니다.

 

개별 FIlter의 기능 설명은 공식 문서를 참고하기 바라며, 여기서는 HTTP connection manager에 대해 중점으로 다루어 보고자 합니다.

 

---

2-3-1 HTTP Connection manager

Envoy에서 HTTP를 담당하는 Network Level의 Filter로써 Filter Chain의 가장 마지막에 위치합니다. 해당 Filter의 주요 기능은 raw byte를 HTTP 메시지 convert를 담당하며, 내부적으로 HTTP L7 Filter들이 존재하여 부가적인 작업을 수행합니다. 따라서 HTTP 요청이 전달되었을 때, Listener Filters 이후 Filter Chain을 수행하는 과정에서 HTTP Connection manager가 요청을 처리할 경우 내부적으로 sub filters를 적용하여 부가작업을 수행합니다.

 

HTTP Connection manager가 담당하는 기능에 대해서 몇 가지 살펴보도록 하겠습니다.

 

1) HTTP header 조작

 - 여러가지 Security 이유로 인해 Envoy를 통해서 특정 header를 삭제하거나 값을 변경할 수 있습니다. 가령 use_remote_address 옵션을 true로 변경했을 경우 connection manager는 실제 전달되는 remote address를 x-forwared-for http header에 사용합니다. 그밖에 다양한 header에 대해서 조작이 가능합니다. 

 

2) Retry 설정

- HTTP 요청에 대해서 내부적으로 연결이 실패했을 때 얼만큼 Retry할 것인지를 설정할 수 있습니다.

 

3) Redirect

- 요청 서비스에서 Redirect 응답이 왔을 때 Proxy 내부에서 해당 3xx 응답을 기반으로 Redirect를 수행할 수 있습니다.

 

4) Timeout

- HTTP 요청에 대해서 응답이 지정 시간동안 없을 경우 요청을 취소할 수 있습니다.

 

위와 같은 기능 외 HTTP Connection manager에는 L7 Filters들이 있어서 해당 Filters를 통해 부가적인 작업을 수행할 수 있다고 말했습니다. 여기에는 다음과 같은 Filter들이 해당됩니다.

 

 

각 Filter에 대한 설명은 Envoy 공식 문서를 참고하시기 바라며, 참고로 위 Filter 중에 Router Filter의 경우는 사용자가 지정한 router 규칙에 일치하는 URL로 접근하였을 경우 지정된 Cluster로 Forwarding을 담당하는 Filter입니다.

 

그렇다면, 지금까지 설명을 바탕으로 HTTP 요청에 대한 Envoy Proxy 내부의 네트워크 흐름을 다시 한번 짚어보겠습니다.

 

1) Listener에 구성된 address와 port에 상응하는 요청이 들어오면, Listener 내부적으로 Listener Filters로 전달합니다.

2) Listener Filters를 순회하면서 Connection Metadata를 조작하고 이후 Network Filters로 전달합니다.

3) Filter Chains(Network Filters)를 순회하면서 사용자 요청을 처리하는데 적합한 Filter를 찾고 해당 Filter가 요청을 처리합니다.

(※ 위 예제에서는 HTTP 요청이 들어왔음을 가정했으므로 HTTP Connection Manager가 이를 담당합니다.)

4) HTTP Connection Manager 내부에 있는 Sub Filters를 순회합니다.

5) L7 필터 마지막에 위치한 Router Filter는 사용자의 Routing Path 요청과 적합한 Cluster를 찾고 해당 Cluster에게 트래픽을 Forwarding 하는 역할을 수행합니다. 

6) Cluster는 내부에 설정된 로드밸런싱 정책등을 고려하여 적합한 Endpoint를 선정하며, 해당 Endpoint에 매칭되는 Service로 트래픽이 전달됩니다.

 

---

 

2-4 Envoy 컴포넌트 등록 방법

 

 

지금까지 Envoy 내부에 존재하는 컴포넌트를 기반으로 Envoy 네트워크 흐름에 대해서 살펴봤습니다. 그렇다면 Cluster, Listener, Endpoint 등은 어떻게 등록할 수 있을까요?

 

Envoy에서는 2가지 방식으로 컴포넌트 정보를 등록하거나 수정할 수 있습니다. 첫 번째 방식은 Static 방식이고 두 번째 방식은 Dynamic 방식입니다. 

 

---

2-4-1 Static 등록 방식

 

Static 방식은 말 그대로 Envoy 기동 시점에 사용자가 지정한 Config 파일 정보를 기반으로 내부 컴포넌트를 등록하는 방식입니다. 따라서 Envoy가 이해할 수 있는 형태로 내부 컴포넌트 설정을 기술해야합니다. Envoy에서는 YAML 형태로 이를 지정할 수 있으며, 아래 예제를 통해서 간단히 살펴보겠습니다.

 

static_resources:
  listeners:
  - name: listener_0
    address:
      socket_address: { address: 127.0.0.1, port_value: 10000 }
    filter_chains:
    - filters:
      - name: envoy.filters.network.http_connection_manager
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
          stat_prefix: ingress_http
          codec_type: AUTO
          route_config:
            name: local_route
            virtual_hosts:
            - name: local_service
              domains: ["*"]
              routes:
              - match: { prefix: "/" }
                route: { cluster: some_service }
          http_filters:
          - name: envoy.filters.http.router
            typed_config:
              "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router
  clusters:
  - name: some_service
    connect_timeout: 0.25s
    type: STATIC
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: some_service
      endpoints:
      - lb_endpoints:
        - endpoint:
            address:
              socket_address:
                address: 127.0.0.1
                port_value: 1234

 

해당 설정은 Envoy 공식 문서에 존재하는 예제입니다. 지금까지 살펴본 Envoy 컴포넌트에 대한 개념을 잘 숙지했다면, 위 예제의 큰 구조를 쉽게 파악할 수 있습니다. 

 

먼저 Listener를 보면 listener_0이라는 이름으로 1개의 Listener가 등록되어있고, 내부에는 filter_chains 항목을 통해 Network Filter인 HTTP Connection Manager를 등록하는 것을 확인할 수 있습니다. 또한 Routes 설정을 통해 /으로 들어오는 모든 prefix에 대해서 some_service라는 Cluster로 전달하도록 지정했음을 확인할 수 있습니다. 이를 토대로 Listener로 들어오는 모든 HTTP 설정에 대해서 some_service Cluster로 트래픽이 전달되도록 Router Filter가 지정될 것임을 알 수 있습니다.

 

Clusters 항목을 보면, 이전에 Route에서 지정한 some_service 이름으로 등록되어있음을 확인할 수 있고 내부적으로 endpoint를 등록하여 Cluster로 트래픽이 전달되었을 때 내부적으로 어떤 endpoint로 전달할 수 있는지를 알 수 있습니다. 추가적으로 ROUND_ROBIN 정책을 적용하여 트래픽 부하를 고르게 분산하도록 지정되어있음을 확인할 수 있습니다.

 

---

 

2-4-2 Dynamic 등록 방식

 

Static 등록 방식은 Config 파일에 직접 기술하여 해당 정보를 토대로 envoy proxy를 구성하는 방법입니다. 하지만 Listener, Endpoint와 Cluster 정보가 수시로 바뀌는 상황에서는 Static 등록 방식을 사용하기에 적합하지 않습니다. 그 이유는 변경할 때마다 해당 Config 파일 수정이 필요하며, envoy proxy 또한 reload 해야하기 때문입니다.

 

따라서 envoy proxy에서는 Static 등록 방식 이외에 Dynamic 방식을 제공하여 envoy proxy 기동 중에도 내부 컴포넌트 설정을 변경할 수 있도록 인터페이스를 제공하였습니다. 이를 xDS API라고 부르며, FIle 동기화, REST 혹은 gRPC  방식이 존재합니다. 참고로 아직 살펴보지는 않았지만 istio에서는 gRPC 통신을 이용하여 envoy proxy의 정보를 동적으로 변경합니다.

 

그렇다면 xDS는 어떠한 종류가 있으며, 어떤 컴포넌트와 매칭되어 변경을 수행할 수 있을까요? 이에 대해서 간단히 알아보도록 하겠습니다.

 

 

 

이전에 살펴봤던 주요 컴포넌트 Listener, Cluster, Endpoint를 갱신할 수 있는 Discovery Service가 제공됩니다. 따라서 envoy proxy에서 요구하는 spec에 맞게 gRPC 혹은 REST 호출을 보내면 개별 컴포넌트에 해당하는 Discovery Service를 통해서 내부 컴포넌트 설정을 변경할 수 있습니다. 

 

 

 

참고로 istio에서는 위와 같이 중앙에 envoy를 관리하는 Management Server가 있으며, Envoy의 xDS를 통해서 중앙에서 Configuration을 등록과 수정등을 자유롭게 할 수 있습니다.

 

위와 같은 Discovery Service외에도 VHDS, SRDS, LDS, SDS, RTDS, ECDS등 다양한 Discovery Service가 존재하며, 이는 envoy 공식 문서를 통해서 참고하시기 바랍니다.

 

---

 

ADS

 

이번에는 xDS 관련하여 또 하나 살펴볼 주요 내용 중 하나인 ADS(Aggregated xDS)에 대해서 살펴보겠습니다. 먼저 ADS는 무엇이고 왜 사용할까요?

 

Envoy 내부 동작 구조에 대해서 자세히 살펴보지 않았지만, 우선 가볍게 짚고 넘어가자면 Envoy 내부의 쓰레딩 모델은 기본적으로 Lock 없이 데이터를 주고 받으며, 데이터 동기화는 Eventually Consistency를 전제로 설계되어있습니다. 즉 이말은 데이터를 전달한다고 해서 바로 반영하는 것은 아니고 완벽한 동기화가 일시에 이루어지지 않음을 의미합니다.

 

그리고 이러한 구조는 다음과 같은 상황을 맞이할 수 있습니다.

 

 

 

이전에 살펴봤듯이 Endpoint와 Cluster는 의존 관계를 맺고 있음을 확인했습니다. 즉 Cluster는 Endpoint의 논리적 집합이었음을 이전 내용을 통해 확인했습니다.

 

그리고 이러한 상황에서 만약 위와 같이 Endpoint와 Cluster가 추가되어서 이를 갱신하는 상황이 발생한다고 가정했을 때, EDS를 통한 갱신이 CDS보다 먼저 이루어진다면 어떻게 될까요?

 

Envoy 입장에서는 EDS를 통해서 전달된 Endpoint의 대상이 Cluster-A라고 전달받았지만, 아직 CDS를 통해 Cluster 정보를 전달받지 못한 상황이므로 일정 기간 동안에는 Cluster 정보에 대한 동기화가 진행되지 않는 이상현상이 발생됩니다. 

 

따라서 이러한 이슈를 해결하고자 등장한 것이 Aggregated Discovery Service입니다.

 

 

ADS는 Single gRPC 스트림으로 구성된 서비스로써 envoy에 전달되는 resource의 순서를 적용하려는 사용자를 위해 집계된 xDS를 단일 gRPC 스트림으로 전달할 수 있습니다. 따라서 위의 경우 ADS와 CDS의 의존 관계에 있을 때에도 이를 집계한 결과를 단일 스트림 형태로 envoy 전달하기 때문에 일관성을 달성할 수 있는 특징을 지니고 있습니다.

 

  "dynamic_resources": {
    "lds_config": {
      "ads": {},
      "initial_fetch_timeout": "0s",
      "resource_api_version": "V3"
    },
    "cds_config": {
      "ads": {},
      "initial_fetch_timeout": "0s",
      "resource_api_version": "V3"
    },
    "ads_config": {
      "api_type": "GRPC",
      "set_node_on_first_message_only": true,
      "transport_api_version": "V3",
      "grpc_services": [
        {
          "envoy_grpc": {
            "cluster_name": "xds-grpc"
          }
        }
      ]
    }
  },

 

dynamic_resources는 위와 같이 기존 Config yaml에 위와 같이 설정할 수 있습니다. 예를들어 위와 같이 설정을 지정할 수 있습니다. 내용을 살펴보면 lds와 cds는 ads를 통해서 해당 내용을 전달받을 수 있으며, ads는 xds-grpc 클러스터를 통해서 해당 정보를 가져오도록 지정되어 있습니다.

 

참고로 위 설정은 istio를 통해 배포된 Pod에 속한 envoy proxy의 설정 파일 중 일부를 발췌한 내용이며, 해당 설정에 대한 자세한 내역은 차후 포스팅을 통해 다루어보겠습니다.

 

---

3. xDS API 

 

xDS API는 istio와 연계하여 Service Discovery를 수행하는데 있어 주요하게 사용됩니다. 따라서 xDS API의 종류와 동작 방법에 대해서 보다 자세히 살펴보고자 합니다.

 

먼저 xDS 지원 방식 부터 살펴보겠습니다. xDS 지원 방식은 총 3가지(File 동기화, HTTP, gRPC)입니다. 

 

 

File 동기화 방식은 위와 같이 envoy에서 File의 상태를 관찰하고 설정이 적용된 File에서 변경이 일어났을 경우 envoy에서 이를 인지하여 내부 컴포넌트의 설정을 동기화하는 방식입니다. 

 

 

반면 gRPC와 HTTP 방식은 config 정보를 전달하는 Management Server가 중앙에 존재합니다. 따라서 envoy에서는 Management Server에 요청하여 config 정보를 전달받고 전달받은 정보를 토대로 자신의 내부 컴포넌트 설정을 동기화합니다.

 

HTTP 방식은 주기적인 polling을 통해서 Management Server로부터 변경된 데이터 항목을 전달받아 갱신합니다. 반면 gRPC는 bidirectional streaming 통신을 통해 데이터를 주고 받는 차이점이 존재합니다. 해당 통신 방법에 대해서 궁금하신 분은 제 블로그 아래 내용을 참고 부탁드립니다.

 

https://cla9.tistory.com/177?category=993774 

 

3. gRPC는 왜 빠를까? (통신 방식) - 2

서론 지난 포스팅에서는 gRPC에서 사용되는 protobuf와 REST 통신에서 사용되는 JSON 크기와 Serialization/Deserialization 관점에서 성능을 비교해봤습니다. 이번에는 gRPC에서 제공하는 통신 방법에 대해서

cla9.tistory.com

 

envoy에서 널리 사용하는 방식은 gRPC 방식이며, istio 또한 gRPC 방식을 통해 xDS 정보를 전달받습니다. 따라서 본 포스팅에서는 gRPC 방식에 보다 초점을 맞추어 살펴보겠습니다.

 

---

 

3-1 xDS gRPC 

 

 

 

앞서 envoy에서 gRPC를 활용한 xDS 방식은 envoy와 config를 전달하는 Management Server 사이에 bidirectional streaming 통신을 사용한다고 설명했습니다. 따라서 해당 방식은 Connection이 끊기지 않고 지속 연결된 상태라고 봐도 무방합니다.

 

이러한 상황에서 처음 Management Server에 envoy가 연결되면, 위와 같이 Discovery Request를 Management Server에 전달합니다. 그러면 Management Server에서는 envoy가 요청하는 Config에 대해서 전체 목록을 전달하게되고, envoy는 해당 설정을 전달받아 Config 업데이트를 수행합니다.

 

 

 

이후 Config 업데이트가 완료되면, envoy는 이전에 Management Server로부터 전달받은 Config 항목에 대한 응답을 전달합니다. 만약 Config 내용을 정상적으로 업데이트를 수행했을 경우에는 ACK를 응답하고 그렇지 않을 경우에는 NACK를 응답합니다. (※  ACK와 NACK의 구조와 동작방식에 대해서는 envoy 공식 문서에서 자세히 다루고 있으니 참고바랍니다.)

 

이때 응답 메시지는 별개의 포맷을 활용하지 않고 다음 Discovery Request를 전달할 때, 응답을 포함하여 전달됩니다. 해당 Discovery Request 메시지는 Config 업데이트 완료 이후 Management Server에 Config가 변경되었을 때, 동기화된 Config 내역을 다시 전달받기 위해 요청하는 메시지입니다. 따라서 Management Server에 Config가 변경되거나 새로운 Resource가 추가되었을 경우 envoy에게 Discovery Response를 전달함으로써 동기화 작업이 이루어집니다.

 

---

 

3-2 SotW(State of the world), Delta xDS

 

 

지금까지 envoy gRPC 동작 과정에 대해서 가볍게 살펴봤는데, 이번에는 Management Server에서 Discovery Response를 응답 메시지를 전달하는데 있어서 선택할 수 있는 2가지 방법에 대해서 살펴보겠습니다.

 

첫 번째 방법은 SotW(State of the world) 방식입니다.

 

 

가령 envoy가 Cluster 정보를 동기화 하기 위해 xDS로 연결되었으며 이미 한차례 Cluster 동기화되었다고 가정해봅시다. 이때 SotW 방식은 동기화된 Cluster 정보 중 하나가 변경되었을 경우에 전체 Cluster 정보를 전달하는 방식을 의미합니다. 해당 방식은 구현이 간단한 반면에 전체 데이터 중 일부만 변경되었음에도 불구하고 전체 정보를 전달하기 때문에 많은 Network overhead가 발생할 수 있습니다.

 

 

참고로 이때 DiscoveryResponse 응답 포맷은 위와같으며, 여기서 resources를 통해서 전체 resource 정보가 전달됩니다.

 

 

두 번째 방법은 Delta 방식입니다.

 

 

해당 방식은 변경된 Config 정보만을 선별하여 전체를 전달하는 것이 아니라 변경분 (Delta 혹은 incremental)만 전달하는 방식을 의미합니다. 

 

 

이때 Delta Discovery Response 응답 포맷은 위와 같으며, resources 항목은 변경된 항목만 추가됩니다. 또한 기존에 존재하는 Resource가 삭제되었을 경우에는 removed_resources를 통해서 이를 envoy에게 전달합니다.

 

지금까지 SotW, Delta 두 가지 방식에 대해서 살펴봤습니다. 그렇다면 두 방식은 어떻게 지정할 수 있을까요?

 

 

위 설정과 같이 envoy configuration에서 api_type을 지정할 때, GRPC로 지정 혹은 DELTA_GRPC로 지정하면 입력 값에 따라서 동작 방식이 결정됩니다.

 

 

 

지금까지 gRPC 방식에서 Config 정보를 전달하는 두 가지 방법에 대해서 살펴봤습니다. 그렇다면 istio에서는 어떤 방식을 기본적으로 사용할까요? istio에서는 기본적으로는 SotW 방식을 사용하고 있으며, 사이드카 컨테이너를 주입할 때 사용자가 지정한 ISTIO_DELTA_XDS 값을 통해서 Delta 방식으로 변경할 수 있습니다. 하지만 현재는 값을 변경한다고 할지라도 실제 데이터를 전달할 때 Delta 값만을 전달하지 않습니다. 

 

---

 

4. 마무리

 

지금까지 Envoy 내부 주요 컴포넌트 및 설정 방식에 대해 살펴봤습니다. istio는 envoy를 얼만큼 잘 이해하고 있느냐에 따라서 istio에 대한 전문성을 확보한다고 생각합니다. 또한 istio가 문제가 생겼을 때 트러블 슈팅할 때는 envoy 설정에 대한 이해를 기반으로 진행되어야합니다. 따라서 이번 포스팅의 내용은 간략하지만 중요한 부분을 다루고 있습니다.

 

다음 포스팅에서는 Envoy 내부 구조에 대해서 조금 더 자세히 알아보도록 하겠습니다.