TS-4042: Add feature to buffer request body before making downstream requests

ci/tsqa/tests/helpers.py

@@ -46,6 +46,7 @@ def getEnv(cls):
         ef = tsqa.environment.EnvironmentFactory(SOURCE_DIR,
                                                  os.path.join(TMP_DIR, 'base_envs'),
                                                  default_configure={'enable-experimental-plugins': None,
+                                                                    'enable-cppapi' : None,
                                                                     'enable-example-plugins': None,
                                                                     'enable-test-tools': None,
                                                                     'disable-dependency-tracking': None,

ci/tsqa/tests/test_request_buffer.py

@@ -0,0 +1,222 @@
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+import socket
+import tsqa
+import tsqa.test_cases
+import tsqa.utils
+import thread
+import SocketServer
+import time
+import requests
+unittest = tsqa.utils.import_unittest()
+import helpers
+import threading
+import logging
+
+log = logging.getLogger(__name__)
+
+class AtomicCounter:
+    def __init__(self):
+        self.lock = threading.Lock()
+        self.counter = 0
+
+    def inc(self, increment = 1):
+        self.lock.acquire()
+        self.counter += increment
+        self.lock.release()
+
+    def get(self):
+        value = 0
+        self.lock.acquire()
+        value = self.counter
+        self.lock.release()
+        return value
+
+    def reset(self):
+        self.lock.acquire()
+        self.counter = 0
+        self.lock.release()
+
+data_receive_times = AtomicCounter()
+
+class EchoReceivedLengthHandler(SocketServer.BaseRequestHandler):
+    """
+    A subclass of RequestHandler that sends back how many bytes it has received
+    """
+
+    def handle(self):
+        chunked = False
+        global data_receive_times
+
+        while True:
+            data = self.request.recv(65536)
+            log.info("server receive data length: %d" % (len(data)))
+            if len(data) > 0 and len(data) < 200:
+              log.info("server receive data: %s" % (data))
+            if not data:
+                log.info('Client disconnected')
+                break
+            if 'Transfer-Encoding: chunked' in data:
+              chunked = True
+
+            data_receive_times.inc();
+            if not chunked or '0\r\n\r\n' in data:
+              log.info('Sending data back to the client')
+              resp_str = str(len(data))
+              resp = ('HTTP/1.1 200 OK\r\n'
+                      'Content-Length: %d\r\n'
+                      'Content-Type: text/html; charset=UTF-8\r\n'
+                      'Connection: keep-alive\r\n'
+                      '\r\n%s' %(
+                        len(resp_str),
+                        resp_str
+                      ))
+              self.request.sendall(resp)
+
+            if chunked and '0\r\n\r\n' in data:
+              log.info('Client disconnected')
+              break
+
+        chunked = False
+
+class TestRequestBuffer(helpers.EnvironmentCase):
+    @classmethod
+    def setUpEnv(cls, env):
+        cls.traffic_server_host = '127.0.0.1'
+        cls.traffic_server_port = int(cls.configs['records.config']['CONFIG']['proxy.config.http.server_ports'])
+        cls.configs['records.config']['CONFIG']['proxy.config.http.chunking_enabled'] = 1
+        cls.configs['records.config']['CONFIG']['proxy.config.diags.debug.enabled'] = 1
+        cls.configs['records.config']['CONFIG']['proxy.config.diags.debug.tags'] = 'http.*'
+        # create a socket server
+        cls.port = tsqa.utils.bind_unused_port()[1]
+        cls.socket_server = tsqa.endpoint.SocketServerDaemon(EchoReceivedLengthHandler, port=cls.port)
+        cls.socket_server.start()
+        cls.socket_server.ready.wait()
+        log.info(cls.environment.layout.logdir)
+        log.info("socket_server_port = %d, cls.traffic_server_port= %d" % (cls.socket_server.port, cls.traffic_server_port))
+        cls.configs['remap.config'].add_line('map / http://127.0.0.1:{0}'.format(cls.socket_server.port))
+        cls.configs['plugin.config'].add_line('%s/RequestBufferPlugin.so' %(cls.environment.layout.plugindir))
+
+    def test_request_buffer_content_length_0(self):
+        """
+        test for sending post header with content length 0
+        """
+        global data_receive_times
+        data_receive_times.reset()
+        small_post_headers = {'Content-Length' : 0}
+        ret = requests.post(
+            'http://127.0.0.1:%d' % (self.traffic_server_port),
+            headers = small_post_headers
+        )
+
+        self.assertEqual(data_receive_times.get(), 1)
+        self.assertEqual(ret.status_code, 200)
+
+
+    def test_request_buffer_content_length_small(self):
+        """
+        test for sending post request all in once
+        """
+        global data_receive_times
+        data_receive_times.reset()
+
+        req = 'POST / HTTP/1.1\r\nConnection: keep-alive\r\nContent-Length:2\r\nHost: 127.0.0.1\r\n\r\nab'
+        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        conn.connect((self.traffic_server_host, self.traffic_server_port))
+        conn.setblocking(1)
+        conn.send(req)
+        resp = conn.recv(4096)
+
+        # data_receive_times will be 1 or 2. Request header and body may send separately
+        self.assertEqual(data_receive_times.get() == 1 or data_receive_times.get() == 2, True)
+        self.assertIn('HTTP/1.1 200 OK', resp)
+
+    def test_request_buffer_content_length_large(self):
+        """
+        test for sending large post
+        """
+        global data_receive_times
+        data_receive_times.reset()
+
+        str_length = 2000
+        send_times = 30
+        content_str = ''
+        for i in xrange(0, str_length):
+          content_str = content_str + 'a'
+
+        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        conn.connect((self.traffic_server_host, self.traffic_server_port))
+
+        hdr = 'POST / HTTP/1.1\r\nConnection: keep-alive\r\nContent-Length:%d\r\nHost: 127.0.0.1\r\n\r\n' % (str_length * send_times)
+        conn.setblocking(1)
+        conn.send(hdr)
+        for i in xrange(0, send_times):
+          time.sleep(0.1)
+          self.assertEqual(data_receive_times.get(), 0)
+          conn.send(content_str)
+
+        log.info("recv_times = %d, send_times = %d" % (data_receive_times.get(), send_times))
+        resp = conn.recv(4096)
+        self.assertIn('HTTP/1.1 200 OK', resp)
+
+    def test_request_buffer_chunked_small(self):
+        """
+        test for sending post request with chunked encoding all in once
+        """
+        global data_receive_times
+        data_receive_times.reset()
+
+        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        conn.connect((self.traffic_server_host, self.traffic_server_port))
+        req = 'POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n\r\n2\r\n12\r\n0\r\n\r\n'
+        conn.setblocking(1)
+        conn.send(req)
+
+        resp = conn.recv(4096)
+
+        log.info("recv_times = %d, send_times = 1" % (data_receive_times.get()))
+        # data_receive_times will be 1 or 2. Request header and body may send separately
+        self.assertEqual(data_receive_times.get() == 1 or data_receive_times.get() == 2, True)
+        self.assertIn('HTTP/1.1 200 OK', resp)
+
+    def test_request_buffer_chunked_large(self):
+        """
+        test for sending large post data with chunked encoding
+        """
+        global data_receive_times
+        data_receive_times.reset()
+        str_length = 2000
+        chunked_size = hex(str_length)[2:]
+        send_times = 30
+        content_str = ''
+        for i in xrange(0, str_length):
+          content_str = content_str + 'a'
+
+        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        conn.connect((self.traffic_server_host, self.traffic_server_port))
+        hdr = 'POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n\r\n'
+        conn.setblocking(1)
+        conn.send(hdr)
+
+        for i in xrange(0, send_times):
+          time.sleep(0.1)
+          self.assertEqual(data_receive_times.get(), 0)
+          conn.send('%s\r\n%s\r\n' %(chunked_size, content_str))
+
+        conn.send('0\r\n\r\n')
+        log.info("recv_times = %d, send_times = %d" % (data_receive_times.get(), send_times))
+        resp = conn.recv(4096)
+        self.assertIn('HTTP/1.1 200 OK', resp)

configure.ac

@@ -1867,6 +1867,7 @@ AC_CONFIG_FILES([
   plugins/healthchecks/Makefile
   plugins/libloader/Makefile
   plugins/regex_remap/Makefile
+  plugins/request_buffer/Makefile
   plugins/stats_over_http/Makefile
   plugins/tcpinfo/Makefile
   proxy/Makefile
@@ -1959,6 +1960,7 @@ AS_IF([test "x$enable_cppapi" = "xyes"], [
   lib/atscppapi/examples/stat_example/Makefile
   lib/atscppapi/examples/timeout_example/Makefile
   lib/atscppapi/examples/transactionhook/Makefile
+  lib/atscppapi/examples/request_buffer/Makefile
   lib/atscppapi/examples/async_http_fetch_streaming/Makefile
   lib/atscppapi/src/Makefile
 ])])

lib/atscppapi/examples/Makefile.am

@@ -19,6 +19,7 @@ SUBDIRS = \
 	helloworld \
 	globalhook \
 	transactionhook \
+	request_buffer \
 	multiple_transaction_hooks \
 	clientrequest \
 	serverresponse \

lib/atscppapi/examples/request_buffer/Makefile.am

@@ -0,0 +1,25 @@
+#
+#  Licensed to the Apache Software Foundation (ASF) under one
+#  or more contributor license agreements.  See the NOTICE file
+#  distributed with this work for additional information
+#  regarding copyright ownership.  The ASF licenses this file
+#  to you under the Apache License, Version 2.0 (the
+#  "License"); you may not use this file except in compliance
+#  with the License.  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+include $(top_srcdir)/build/plugins.mk
+
+AM_CPPFLAGS += -I$(top_srcdir)/lib/atscppapi/src/include -Wno-unused-variable
+
+target=RequestBufferPlugin.so
+pkglib_LTLIBRARIES = RequestBufferPlugin.la
+RequestBufferPlugin_la_SOURCES = RequestBufferPlugin.cc
+RequestBufferPlugin_la_LDFLAGS = -module -avoid-version -shared -L$(top_builddir)/lib/atscppapi/src/ -latscppapi  $(TS_PLUGIN_LDFLAGS)

lib/atscppapi/examples/request_buffer/RequestBufferPlugin.cc

@@ -0,0 +1,109 @@
+/**
+  Licensed to the Apache Software Foundation (ASF) under one
+  or more contributor license agreements.  See the NOTICE file
+  distributed with this work for additional information
+  regarding copyright ownership.  The ASF licenses this file
+  to you under the Apache License, Version 2.0 (the
+  "License"); you may not use this file except in compliance
+  with the License.  You may obtain a copy of the License at
+
+      http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+ */
+
+
+#include <iostream>
+#include <atscppapi/GlobalPlugin.h>
+#include <atscppapi/TransactionPlugin.h>
+#include <atscppapi/PluginInit.h>
+
+using namespace atscppapi;
+static const int MIN_BYTE_PER_SEC = 1000;
+
+class TimeRecord
+{
+public:
+  TimeRecord() { clock_gettime(CLOCK_MONOTONIC, &start_time); }
+  const timespec
+  getStartTime() const
+  {
+    return start_time;
+  }
+
+private:
+  timespec start_time;
+};
+class RequestBufferPlugin : public atscppapi::TransactionPlugin
+{
+public:
+  RequestBufferPlugin(Transaction &transaction) : TransactionPlugin(transaction)
+  {
+    // enable the request body buffering
+    transaction.configIntSet(TS_CONFIG_HTTP_REQUEST_BUFFER_ENABLED, 1);
+    // save the start time for calculating the data rate
+    timeRecord = new TimeRecord();
+
+    TransactionPlugin::registerHook(HOOK_HTTP_REQUEST_BUFFER_READ);
+    TransactionPlugin::registerHook(HOOK_HTTP_REQUEST_BUFFER_READ_COMPLETE);
+    std::cout << "Constructed!" << std::endl;
+  }
+  virtual ~RequestBufferPlugin()
+  {
+    delete timeRecord; // cleanup
+    std::cout << "Destroyed!" << std::endl;
+  }
+  void
+  handleHttpRequestBufferRead(Transaction &transaction)
+  {
+    std::cout << "request buffer read" << transaction.getClientRequestBody().size() << std::endl;
+    reached_min_speed(transaction) ? transaction.resume() : transaction.error();
+  }
+  void
+  handleHttpRequestBufferReadComplete(Transaction &transaction)
+  {
+    std::cout << "request buffer complete!" << transaction.getClientRequestBody().size() << std::endl;
+    transaction.resume();
+  }
+
+private:
+  TimeRecord *timeRecord;
+  bool
+  reached_min_speed(Transaction &transaction)
+  {
+    int64_t body_len = transaction.getClientRequestBodySize();
+    timespec now_time;
+    clock_gettime(CLOCK_MONOTONIC, &now_time);
+    double time_diff_in_sec =
+      (now_time.tv_sec - timeRecord->getStartTime().tv_sec) + 1e-9 * (now_time.tv_nsec - timeRecord->getStartTime().tv_nsec);
+    std::cout << "time_diff_in_sec = " << time_diff_in_sec << ", body_len = " << body_len
+              << ", date_rate = " << body_len / time_diff_in_sec << std::endl;
+    return body_len / time_diff_in_sec >= MIN_BYTE_PER_SEC;
+  }
+};
+
+class GlobalHookPlugin : public atscppapi::GlobalPlugin
+{
+public:
+  GlobalHookPlugin() { GlobalPlugin::registerHook(HOOK_READ_REQUEST_HEADERS); }
+  virtual void
+  handleReadRequestHeaders(Transaction &transaction)
+  {
+    std::cout << "Hello from handleReadRequestHeaders!" << std::endl;
+    if (transaction.getClientRequest().getMethod() == HTTP_METHOD_POST) {
+      transaction.addPlugin(new RequestBufferPlugin(transaction));
+    }
+    transaction.resume();
+  }
+};
+
+void
+TSPluginInit(int argc ATSCPPAPI_UNUSED, const char *argv[] ATSCPPAPI_UNUSED)
+{
+  RegisterGlobalPlugin("CPP_Example_RequestBuffer", "apache", "dev@trafficserver.apache.org");
+  new GlobalHookPlugin();
+}