listener: remove the peek from the listener filters

envoy/network/BUILD

@@ -80,6 +80,7 @@ envoy_cc_library(
     hdrs = ["filter.h"],
     deps = [
         ":listen_socket_interface",
+        ":listener_filter_buffer_interface",
         ":transport_socket_interface",
         "//envoy/buffer:buffer_interface",
         "//envoy/stream_info:stream_info_interface",
@@ -141,6 +142,14 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "listener_filter_buffer_interface",
+    hdrs = ["listener_filter_buffer.h"],
+    deps = [
+        "//envoy/buffer:buffer_interface",
+    ],
+)
+
 envoy_cc_library(
     name = "transport_socket_interface",
     hdrs = ["transport_socket.h"],

envoy/network/filter.h

@@ -4,6 +4,7 @@
 
 #include "envoy/buffer/buffer.h"
 #include "envoy/network/listen_socket.h"
+#include "envoy/network/listener_filter_buffer.h"
 #include "envoy/network/transport_socket.h"
 #include "envoy/stream_info/stream_info.h"
 #include "envoy/upstream/host_description.h"
@@ -330,6 +331,20 @@ class ListenerFilter {
    * @return status used by the filter manager to manage further filter iteration.
    */
   virtual FilterStatus onAccept(ListenerFilterCallbacks& cb) PURE;
+
+  /**
+   * Called when data read from the connection. If the filter chain doesn't get
+   * enough data, the filter chain can be stopped, then waiting for more data.
+   * @param buffer the buffer of data.
+   * @return status used by the filter manager to manage further filter iteration.
+   */
+  virtual FilterStatus onData(Network::ListenerFilterBuffer& buffer) PURE;
+
+  /**
+   * Return the size of data the filter want to inspect from the connection.
+   @return the size of data inspect from the connection. 0 means filter needn't any data.
+   */
+  virtual size_t inspectSize() const { return 0; }
 };
 
 using ListenerFilterPtr = std::unique_ptr<ListenerFilter>;

envoy/network/listener_filter_buffer.h

@@ -0,0 +1,43 @@
+#pragma once
+
+#include <cstdint>
+#include <memory>
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/common/pure.h"
+
+namespace Envoy {
+namespace Network {
+
+/**
+ * Interface for ListenerFilterBuffer
+ */
+class ListenerFilterBuffer {
+public:
+  virtual ~ListenerFilterBuffer() = default;
+
+  /**
+   * Copy the bufferred data into the specific Buffer::Instance.
+   * But it won't drain the data after copyOut, except an explicit drain method invoked.
+   * @param buffer supplies the buffer to read into.
+   * @param max_length supplies the maximum length to read.
+   * @return the length of data read into the buffer.
+   */
+  virtual uint64_t copyOut(Buffer::Instance& buffer, uint64_t length) PURE;
+
+  /**
+   * Drain the data from the beginning of the buffer.
+   * @param length the length of data to drain.
+   * @return the actual length of data drained.
+   */
+  virtual uint64_t drain(uint64_t length) PURE;
+
+  /**
+   * Return the length of data in the buffer
+   * @return length The length of data in the buffer.
+   */
+  virtual uint64_t length() const PURE;
+};
+
+} // namespace Network
+} // namespace Envoy

source/common/network/BUILD

@@ -268,6 +268,17 @@ envoy_cc_library(
     ],
 )
 
+envoy_cc_library(
+    name = "listener_filter_buffer_lib",
+    srcs = ["listener_filter_buffer_impl.cc"],
+    hdrs = ["listener_filter_buffer_impl.h"],
+    deps = [
+        "//envoy/network:io_handle_interface",
+        "//envoy/network:listener_filter_buffer_interface",
+        "//source/common/buffer:buffer_lib",
+    ],
+)
+
 envoy_cc_library(
     name = "listener_lib",
     srcs = [

source/common/network/listener_filter_buffer_impl.cc

@@ -0,0 +1,70 @@
+#include "source/common/network/listener_filter_buffer_impl.h"
+
+#include <string.h>
+
+namespace Envoy {
+namespace Network {
+
+uint64_t ListenerFilterBufferImpl::copyOut(Buffer::Instance& buffer, uint64_t length) {
+  auto size = std::min(length, data_size_);
+  if (size == 0) {
+    return 0;
+  }
+  buffer.drain(buffer.length());
+  uint8_t* raw_mem = static_cast<uint8_t*>(buffer_->frontSlice().mem_);
+  buffer.add(raw_mem + drained_size_, size);
+  return size;
+}
+
+uint64_t ListenerFilterBufferImpl::drain(uint64_t length) {
+  auto size_to_drain = std::min(length, data_size_);
+  // It doesn't drain the data from buffer directly until drain the data
+  // from actual socket.
+  drained_size_ += size_to_drain;
+  data_size_ -= size_to_drain;
+  return size_to_drain;
+}
+
+bool ListenerFilterBufferImpl::drainFromSocket() {
+  // Since we want to drain the data from the socket, so a
+  // temporary buffer need here.
+  std::unique_ptr<uint8_t[]> buf(new uint8_t[drained_size_]);
+  auto result = io_handle_.recv(buf.get(), drained_size_, 0);
+  if (!result.ok()) {
+    on_close_cb_();
+    return false;
+  }
+  buffer_->drain(drained_size_);
+  drained_size_ = 0;
+  return true;
+}
+
+PeekState ListenerFilterBufferImpl::peekFromSocket() {
+  auto raw_slice = buffer_->frontSlice();
+
+  const auto result = io_handle_.recv(raw_slice.mem_, raw_slice.len_, MSG_PEEK);
+  if (!result.ok()) {
+    if (result.err_->getErrorCode() == Api::IoError::IoErrorCode::Again) {
+      return PeekState::Again;
+    }
+    return PeekState::Error;
+  }
+  data_size_ = result.return_value_ - drained_size_;
+  return PeekState::Done;
+}
+
+void ListenerFilterBufferImpl::onFileEvent(uint32_t events) {
+  if (events & Event::FileReadyType::Closed) {
+    on_close_cb_();
+  }
+
+  auto state = peekFromSocket();
+  if (state == PeekState::Done) {
+    on_data_cb_();
+  } else if (state == PeekState::Error) {
+    on_close_cb_();
+  }
+}
+
+} // namespace Network
+} // namespace Envoy

source/common/network/listener_filter_buffer_impl.h

@@ -0,0 +1,86 @@
+#pragma once
+
+#include <bits/stdint-uintn.h>
+
+#include <atomic>
+#include <functional>
+#include <memory>
+
+#include "envoy/buffer/buffer.h"
+#include "envoy/network/io_handle.h"
+#include "envoy/network/listener_filter_buffer.h"
+
+#include "source/common/buffer/buffer_impl.h"
+
+namespace Envoy {
+namespace Network {
+
+using ListenerFilterBufferOnCloseCb = std::function<void()>;
+using ListenerFilterBufferOnDataCb = std::function<void()>;
+
+enum class PeekState {
+  // Peek data status successful.
+  Done,
+  // Need to try again.
+  Again,
+  // Error to peek data.
+  Error
+};
+
+class ListenerFilterBufferImpl : public ListenerFilterBuffer, Logger::Loggable<Logger::Id::filter> {
+public:
+  ListenerFilterBufferImpl(IoHandle& io_handle, Event::Dispatcher& dispatcher,
+                           ListenerFilterBufferOnCloseCb close_cb,
+                           ListenerFilterBufferOnDataCb on_data_cb, uint64_t buffer_size)
+      : io_handle_(io_handle), dispatcher_(dispatcher), on_close_cb_(close_cb),
+        on_data_cb_(on_data_cb), buffer_(new Buffer::OwnedImpl()) {
+    // Since we need the single slice for peek from the socket, so initialize that
+    // single slice.
+    auto reservation = buffer_->reserveSingleSlice(buffer_size);
+    reservation.commit(buffer_size);
+  }
+
+  uint64_t copyOut(Buffer::Instance& buffer, uint64_t length) override;
+  uint64_t drain(uint64_t length) override;
+  uint64_t length() const override { return buffer_->length(); }
+
+  /**
+   * trigger the data peek from the socket.
+   */
+  PeekState peekFromSocket();
+
+  /**
+   * sync the drain to the actual socket.
+   */
+  bool drainFromSocket();
+
+  void initialize() {
+    io_handle_.initializeFileEvent(
+        dispatcher_, [this](uint32_t events) { onFileEvent(events); },
+        Event::PlatformDefaultTriggerType,
+        Event::FileReadyType::Read | Event::FileReadyType::Closed);
+  }
+
+  void reset() { io_handle_.resetFileEvents(); }
+
+private:
+  void onFileEvent(uint32_t events);
+
+  IoHandle& io_handle_;
+  Event::Dispatcher& dispatcher_;
+  ListenerFilterBufferOnCloseCb on_close_cb_;
+  ListenerFilterBufferOnDataCb on_data_cb_;
+
+  // The buffer for the data peeked from the socket.
+  Buffer::InstancePtr buffer_;
+  // the size of valid data.
+  uint64_t data_size_{0};
+  // This is used for counting the drained data size before
+  // drain the data from actual socket.
+  uint64_t drained_size_{0};
+};
+
+using ListenerFilterBufferImplPtr = std::unique_ptr<ListenerFilterBufferImpl>;
+
+} // namespace Network
+} // namespace Envoy

source/extensions/filters/listener/http_inspector/http_inspector.h

@@ -70,6 +70,11 @@ class Filter : public Network::ListenerFilter, Logger::Loggable<Logger::Id::filt
   // Network::ListenerFilter
   Network::FilterStatus onAccept(Network::ListenerFilterCallbacks& cb) override;
 
+  // TODO (soulxu) implement this instead of filter peek data by itself.
+  Network::FilterStatus onData(Network::ListenerFilterBuffer&) override {
+    return Network::FilterStatus::Continue;
+  };
+
 private:
   static const absl::string_view HTTP2_CONNECTION_PREFACE;
 

source/extensions/filters/listener/original_dst/original_dst.h

@@ -26,6 +26,10 @@ class OriginalDstFilter : public Network::ListenerFilter, Logger::Loggable<Logge
   // Network::ListenerFilter
   Network::FilterStatus onAccept(Network::ListenerFilterCallbacks& cb) override;
 
+  Network::FilterStatus onData(Network::ListenerFilterBuffer&) override {
+    return Network::FilterStatus::Continue;
+  };
+
 private:
   envoy::config::core::v3::TrafficDirection traffic_direction_;
 };

source/extensions/filters/listener/original_src/original_src.h

@@ -23,6 +23,10 @@ class OriginalSrcFilter : public Network::ListenerFilter, Logger::Loggable<Logge
   // Network::ListenerFilter
   Network::FilterStatus onAccept(Network::ListenerFilterCallbacks& cb) override;
 
+  Network::FilterStatus onData(Network::ListenerFilterBuffer&) override {
+    return Network::FilterStatus::Continue;
+  };
+
 private:
   Config config_;
 };

source/extensions/filters/listener/proxy_protocol/BUILD

@@ -24,6 +24,7 @@ envoy_cc_library(
         "//envoy/network:filter_interface",
         "//envoy/network:listen_socket_interface",
         "//source/common/api:os_sys_calls_lib",
+        "//source/common/buffer:buffer_lib",
         "//source/common/common:assert_lib",
         "//source/common/common:empty_string",
         "//source/common/common:minimal_logger_lib",