[filter] connect() to support UDP

flomesh-io · Sep 5, 2022 · ee1577b · ee1577b
1 parent 58ed6cb
commit ee1577b
Show file tree

Hide file tree

Showing 6 changed files with 66 additions and 21 deletions.
diff --git a/docs/dts/Configuration.d.ts b/docs/dts/Configuration.d.ts
@@ -312,6 +312,7 @@ interface Configuration {
    *
    * @param target The target to connect to, in form of `"<host>:<port>"`, or a function that returns the target.
    * @param options Options including:
+   *   - _protocol_ - Protocol to use. Can be `"TCP"` or `"UDP"`. Default is `"TCP"`.
    *   - _bufferLimit_ - Maximum size of data allowed to stay in buffer due to slow outbound bandwidth.
    *       Can be a number in bytes or a string with a unit suffix such as `'k'`, `'m'`, `'g'` and `'t'`.
    *   - _retryCount_ - How many times it should retry connection after a failure, or -1 for the infinite retries. Defaults to 0.

diff --git a/docs/reference/api/Configuration/connect.mdx b/docs/reference/api/Configuration/connect.mdx
@@ -15,7 +15,7 @@ api: Configuration.connect
 
 A _connect_ filter does the following:
 
-- It establishes a TCP connection to a remote host specified in the _target_ parameter
+- It establishes a TCP or UDP connection to a remote host specified in the _target_ parameter
 - Input _Data_ stream to the filter is sent to that host after the connection is established
 - _Data_ stream received from that host comes out from the filter's output
 
@@ -27,7 +27,7 @@ The target address to connect to is given by the _target_ parameter. It can be a
 
 ### Outbound data buffer
 
-No limit is set to how fast input _Data_ can go into a _connect_ filter, but the outgoing TCP connection does have a limit depending on the network condition. When outgoing traffic is too slow and incoming traffic is too fast, _Data_ will be backed up in the filter"s internal buffer.
+No limit is set to how fast input _Data_ can go into a _connect_ filter, but an outgoing TCP connection does have a limit depending on the network condition. When outgoing traffic is too slow and incoming traffic is too fast, _Data_ will be backed up in the filter"s internal buffer.
 
 By default, the buffer can grow unlimitedly. You can set a limit to how much data is allowed to stay in memory by option _bufferLimit_ in the _options_ parameter. It can be a number in bytes or a string with a unit suffix such as `'k'`, `'m'`, `'g'` and `'t'`.
 

diff --git a/src/filters/connect.cpp b/src/filters/connect.cpp
@@ -28,13 +28,28 @@
 #include "utils.hpp"
 #include "log.hpp"
 
+namespace pjs {
+
+using namespace pipy;
+
+template<>
+void EnumDef<Outbound::Protocol>::init() {
+  define(Outbound::Protocol::TCP, "tcp");
+  define(Outbound::Protocol::UDP, "udp");
+}
+
+} // namespace pjs
+
 namespace pipy {
 
 //
 // Connect::Options
 //
 
 Connect::Options::Options(pjs::Object *options) {
+  Value(options, "protocol")
+    .get_enum(protocol)
+    .check_nullable();
   Value(options, "bufferLimit")
     .get_binary_size(buffer_limit)
     .check_nullable();
@@ -104,7 +119,15 @@ void Connect::process(Event *evt) {
       auto s = target.to_string();
       std::string host; int port;
       if (utils::get_host_port(s->str(), host, port)) {
-        auto outbound = new OutboundTCP(ConnectReceiver::input(), m_options);
+        Outbound *outbound = nullptr;
+        switch (m_options.protocol) {
+          case Outbound::Protocol::TCP:
+            outbound = new OutboundTCP(ConnectReceiver::input(), m_options);
+            break;
+          case Outbound::Protocol::UDP:
+            outbound = new OutboundUDP(ConnectReceiver::input(), m_options);
+            break;
+        }
         outbound->connect(host, port);
         m_outbound = outbound;
       } else {

diff --git a/src/outbound.cpp b/src/outbound.cpp
@@ -42,8 +42,8 @@ List<Outbound> Outbound::s_all_outbounds;
 //
 
 Outbound::Outbound(EventTarget::Input *output, const Options &options)
-  : m_output(output)
-  , m_options(options)
+  : m_options(options)
+  , m_output(output)
 {
   Log::debug("[outbound %p] ++", this);
   s_all_outbounds.push(this);
@@ -54,6 +54,16 @@ Outbound::~Outbound() {
   s_all_outbounds.remove(this);
 }
 
+auto Outbound::protocol_name() const -> pjs::Str* {
+  static pjs::ConstStr s_TCP("TCP");
+  static pjs::ConstStr s_UDP("UDP");
+  switch (m_options.protocol) {
+    case Protocol::TCP: return s_TCP;
+    case Protocol::UDP: return s_UDP;
+  }
+  return nullptr;
+}
+
 auto Outbound::address() -> pjs::Str* {
   if (!m_address) {
     std::string s("[");
@@ -98,10 +108,12 @@ void OutboundTCP::connect(const std::string &host, int port) {
   m_port = port;
   m_connecting = true;
 
-  auto *addr = address();
-  m_metric_traffic_out = Status::metric_outbound_out->with_labels(&addr, 1);
-  m_metric_traffic_in = Status::metric_outbound_in->with_labels(&addr, 1);
-  m_metric_conn_time = Status::metric_outbound_conn_time->with_labels(&addr, 1);
+  pjs::Str *keys[2];
+  keys[0] = protocol_name();
+  keys[1] = address();
+  m_metric_traffic_out = Status::metric_outbound_out->with_labels(keys, 2);
+  m_metric_traffic_in = Status::metric_outbound_in->with_labels(keys, 2);
+  m_metric_conn_time = Status::metric_outbound_conn_time->with_labels(keys, 2);
 
   start(0);
 }
@@ -519,10 +531,12 @@ void OutboundUDP::connect(const std::string &host, int port) {
   m_port = port;
   m_connecting = true;
 
-  auto *addr = address();
-  m_metric_traffic_out = Status::metric_outbound_out->with_labels(&addr, 1);
-  m_metric_traffic_in = Status::metric_outbound_in->with_labels(&addr, 1);
-  m_metric_conn_time = Status::metric_outbound_conn_time->with_labels(&addr, 1);
+  pjs::Str *keys[2];
+  keys[0] = protocol_name();
+  keys[1] = address();
+  m_metric_traffic_out = Status::metric_outbound_out->with_labels(keys, 2);
+  m_metric_traffic_in = Status::metric_outbound_in->with_labels(keys, 2);
+  m_metric_conn_time = Status::metric_outbound_conn_time->with_labels(keys, 2);
 
   start(0);
 }
@@ -797,13 +811,11 @@ void OutboundUDP::pump() {
 
     if (auto data = evt->as<Data>()) {
       m_socket.async_send(
-        DataChunks(m_buffer.chunks()),
+        DataChunks(data->chunks()),
         [=](const std::error_code &ec, std::size_t n) {
           if (ec != asio::error::operation_aborted) {
-            m_buffer.shift(n);
             m_metric_traffic_out->increase(n);
             Status::metric_outbound_out->increase(n);
-
             if (ec) {
               if (Log::is_enabled(Log::WARN)) {
                 char desc[200];

diff --git a/src/outbound.hpp b/src/outbound.hpp
@@ -71,6 +71,8 @@ class Outbound :
     }
   }
 
+  auto protocol() const -> Protocol { return m_options.protocol; }
+  auto protocol_name() const -> pjs::Str*;
   auto address() -> pjs::Str*;
   auto host() const -> const std::string& { return m_host; }
   auto port() const -> int { return m_port; }

diff --git a/src/status.cpp b/src/status.cpp
@@ -319,8 +319,9 @@ void Status::register_metrics() {
     label_names
   );
 
-  label_names->length(1);
-  label_names->set(0, "peer");
+  label_names->length(2);
+  label_names->set(0, "protocol");
+  label_names->set(1, "peer");
 
   stats::Gauge::make(
     pjs::Str::make("pipy_outbound_count"),
@@ -329,8 +330,10 @@ void Status::register_metrics() {
       int total = 0;
       gauge->clear();
       Outbound::for_each([&](Outbound *outbound) {
-        auto k = outbound->address();
-        auto cnt = gauge->with_labels(&k, 1);
+        pjs::Str *k[2];
+        k[0] = outbound->protocol_name();
+        k[1] = outbound->address();
+        auto cnt = gauge->with_labels(k, 2);
         cnt->increase();
         total++;
       });
@@ -497,7 +500,11 @@ void Status::dump_memory() {
   std::unordered_map<std::string, OutboundSum> outbound_sums;
   Outbound::for_each([&](Outbound *outbound) {
     char key[1000];
-    std::sprintf(key, "[%s]:%d", outbound->host().c_str(), outbound->port());
+    std::sprintf(key, "%s [%s]:%d",
+      outbound->protocol_name()->c_str(),
+      outbound->host().c_str(),
+      outbound->port()
+    );
     auto conn_time = outbound->connection_time() / (outbound->retries() + 1);
     auto &sum = outbound_sums[key];
     sum.connections++;