目錄

• Envoy 源碼分析－－network L4 filter manager
  • FilterManagerImpl
    • addWriteFilter
    • addReadFilter
    • addFilter
    • initializeReadFilters
    • onRead
    • onWrite

Envoy 源碼分析－－network L4 filter manager

申明：本文的 Envoy 源碼分析基於 Envoy1.10.0。

承接Envoy 源碼分析－－network，上次 network 只分析到 L4 過濾，這次接下來分析。

L4 過濾有三個不同類型的過濾器：

• 讀過濾：當 Envoy 從下游連接接收數據時，調用讀過濾器。
• 寫過濾：當 Envoy 要發送數據到下游連接時，調用寫過濾器。
• 讀/寫過濾：當 Envoy 從下游連接接收數據和要發送數據到下游連接時，調用讀/寫過濾器。

網絡級過濾器的 API 相對簡單，因為最終過濾器只操作原始字節和少量連接事件（例如，TLS 握手完成、連接在本地或遠程斷開等）。

現在我們來分析下 L4 過濾的過濾管理，先看 UML 類圖。

FilterManagerImpl

所有的 L4 filter FilterManager 都由 FilterManagerImpl 進行管理。它提供6個接口，其中 4個接口是和 FilterManager 一樣。其余兩個則是讀和寫。

addWriteFilter

調用此接口，直接將其添加到下游過濾器。

void FilterManagerImpl::addWriteFilter(WriteFilterSharedPtr filter) {
  ASSERT(connection_.state() == Connection::State::Open);
  downstream_filters_.emplace_front(filter);
}

addReadFilter

新建 ActiveReadFilter，加入下游過濾器。

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

addFilter

加入上游過濾器和下游過濾器

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

initializeReadFilters

初始化讀過濾器，會調用各個過濾器的 onNewConnection 。

for (; entry != upstream_filters_.end(); entry++) {
  if (!(*entry)->initialized_) {
    (*entry)->initialized_ = true;
    FilterStatus status = (*entry)->filter_->onNewConnection();
    if (status == FilterStatus::StopIteration) {
      return;
    }
  }

onRead

如果沒有初始化調用 onNewConnection，然后獲取讀緩沖，對緩沖數據處理調用 onData。

  for (; entry != upstream_filters_.end(); entry++) {
    //未初始化，調用onNewConnection
    if (!(*entry)->initialized_) {
      (*entry)->initialized_ = true;
      FilterStatus status = (*entry)->filter_->onNewConnection();
      //需要過濾的數據，直接退出。
      if (status == FilterStatus::StopIteration) {
        return;
      }
    }

    BufferSource::StreamBuffer read_buffer = buffer_source_.getReadBuffer();
    if (read_buffer.buffer.length() > 0 || read_buffer.end_stream) {
      //調用onData進行處理
      FilterStatus status = (*entry)->filter_->onData(read_buffer.buffer, read_buffer.end_stream);
      //需要過濾的數據，直接退出。
      if (status == FilterStatus::StopIteration) {
        return;
      }
    }
  }

onWrite

獲取寫緩沖，過濾寫緩沖。

FilterStatus FilterManagerImpl::onWrite() {
  for (const WriteFilterSharedPtr& filter : downstream_filters_) {
    //獲取寫緩沖，調用onWrite
    BufferSource::StreamBuffer write_buffer = buffer_source_.getWriteBuffer();
    FilterStatus status = filter->onWrite(write_buffer.buffer, write_buffer.end_stream);
    if (status == FilterStatus::StopIteration) {
      return status;
    }
  }

  return FilterStatus::Continue;
}