TS-4341

ci/tsqa/tests/test_origin_max_connections.py

@@ -0,0 +1,213 @@
+'''
+Test the configure entry : proxy.config.http.origin_max_connections
+'''
+
+#  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 time
+import logging
+import uuid
+import socket
+import requests
+import tsqa.test_cases
+import helpers
+import thread
+from multiprocessing import Pool
+import SocketServer
+import os
+
+log = logging.getLogger(__name__)
+
+
+# TODO: seems like a useful shared class- either add to httpbin or some shared lib
+class KAHandler(SocketServer.BaseRequestHandler):
+    '''SocketServer that returns the connection-id  as the body
+    '''
+    # class variable to set number of active sessions
+    alive_sessions = 0
+
+    def handle(self):
+        KAHandler.alive_sessions += 1
+        # Receive the data in small chunks and retransmit it
+        conn_id = uuid.uuid4().hex
+        start = time.time()
+        while True:
+            data = self.request.recv(4096).strip()
+            if data:
+                log.info('Sending data back to the client: {uid}'.format(uid=conn_id))
+            else:
+                log.info('Client disconnected: {timeout}seconds'.format(timeout=time.time() - start))
+                break
+            body = conn_id
+            if 'timeout' in data:
+                print 'sleep for a long time!'
+                time.sleep(4)
+            else:
+                time.sleep(2)
+            resp = ('HTTP/1.1 200 OK\r\n'
+                    'Content-Length: {content_length}\r\n'
+                    'Content-Type: text/html; charset=UTF-8\r\n'
+                    'Connection: keep-alive\r\n'
+                    'X-Current-Sessions: {alive_sessions}\r\n'
+                    '\r\n'
+                    '{body}'.format(content_length=len(body), alive_sessions=KAHandler.alive_sessions, body=body))
+            self.request.sendall(resp)
+        KAHandler.alive_sessions -= 1
+
+
+class TestKeepAlive_Origin_Max_connections(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.socket_server_port = int(tsqa.utils.bind_unused_port()[1])
+
+        log.info("socket_server_port = %d" % (cls.socket_server_port))
+        cls.server = tsqa.endpoint.SocketServerDaemon(KAHandler, port=cls.socket_server_port)
+        cls.server.start()
+        cls.server.ready.wait()
+
+        cls.socket_server_port2 = int(tsqa.utils.bind_unused_port()[1])
+        cls.server2 = tsqa.endpoint.SocketServerDaemon(KAHandler, port=cls.socket_server_port2)
+        cls.server2.start()
+        cls.server2.ready.wait()
+
+        queue_path = os.path.join(cls.environment.layout.sysconfdir, 'queue.conf')
+        with open(queue_path, 'w') as fh:
+            fh.write('CONFIG proxy.config.http.origin_max_connections_queue INT 2')
+
+        noqueue_path = os.path.join(cls.environment.layout.sysconfdir, 'noqueue.conf')
+        with open(noqueue_path, 'w') as fh:
+            fh.write('CONFIG proxy.config.http.origin_max_connections_queue INT 0')
+
+        cls.configs['remap.config'].add_line('map /other/queue/ http://127.0.0.1:{0} @plugin=conf_remap.so @pparam={1}'.format(cls.socket_server_port2, queue_path))
+        cls.configs['remap.config'].add_line('map /other/noqueue/ http://127.0.0.1:{0} @plugin=conf_remap.so @pparam={1}'.format(cls.socket_server_port2, noqueue_path))
+        cls.configs['remap.config'].add_line('map /other/ http://127.0.0.1:{0}'.format(cls.socket_server_port2))
+        cls.configs['remap.config'].add_line('map /queue/ http://127.0.0.1:{0} @plugin=conf_remap.so @pparam={1}'.format(cls.socket_server_port, queue_path))
+        cls.configs['remap.config'].add_line('map /noqueue/ http://127.0.0.1:{0} @plugin=conf_remap.so @pparam={1}'.format(cls.socket_server_port, noqueue_path))
+        cls.configs['remap.config'].add_line('map / http://127.0.0.1:{0}'.format(cls.socket_server_port))
+
+        cls.configs['records.config']['CONFIG'].update({
+            'proxy.config.http.origin_max_connections':  1,
+            'proxy.config.http.keep_alive_enabled_out': 1,
+            'proxy.config.http.keep_alive_no_activity_timeout_out': 1,
+            'proxy.config.http.transaction_active_timeout_out': 2,
+            'proxy.config.http.connect_attempts_timeout': 2,
+            'proxy.config.http.connect_attempts_rr_retries': 0,
+            'proxy.config.exec_thread.limit': 1,
+            'proxy.config.exec_thread.autoconfig': 0,
+        })
+
+    def _send_requests(self, total_requests, path='', other=False):
+        url = 'http://{0}:{1}/{2}'.format(self.traffic_server_host, self.traffic_server_port, path)
+        url2 = 'http://{0}:{1}/other/{2}'.format(self.traffic_server_host, self.traffic_server_port, path)
+        jobs = []
+        jobs2 = []
+        pool = Pool(processes=4)
+        for _ in xrange(0, total_requests):
+            jobs.append(pool.apply_async(requests.get, (url,)))
+            if other:
+                jobs2.append(pool.apply_async(requests.get, (url2,)))
+
+        results = []
+        results2 = []
+        for j in jobs:
+            try:
+                results.append(j.get())
+            except Exception as e:
+                results.append(e)
+
+        for j in jobs2:
+            try:
+                results2.append(j.get())
+            except Exception as e:
+                results2.append(e)
+
+        return results, results2
+
+
+    # TODO: enable after TS-4340 is merged
+    # and re-enable `other` for the remaining queueing tests
+    def tesst_origin_scoping(self):
+        '''Send 2 requests to loopback (on separate ports) and ensure that they run in parallel
+        '''
+        results, results2 = self._send_requests(1, other=True)
+
+        # TS-4340
+        # ensure that the 2 origins (2 different ports on loopback) were running in parallel
+        for i in xrange(0, REQUEST_COUNT):
+            self.assertEqual(int(results[i].get().headers['X-Current-Sessions']), 2)
+            self.assertEqual(int(results2[i].get().headers['X-Current-Sessions']), 2)
+
+    def test_origin_default_queueing(self):
+        '''By default we have no queue limit
+        '''
+        REQUEST_COUNT = 4
+        results, results2 = self._send_requests(REQUEST_COUNT)
+
+        for x in xrange(0, REQUEST_COUNT):
+            self.assertEqual(results[x].status_code, 200)
+            #self.assertEqual(results2[x].status_code, 200)
+
+    def test_origin_queueing(self):
+        '''If a queue is set, N requests are queued and the rest immediately fail
+        '''
+        REQUEST_COUNT = 4
+        results, results2 = self._send_requests(REQUEST_COUNT, path='queue/')
+
+        success = 0
+        fail = 0
+        for x in xrange(0, REQUEST_COUNT):
+            if results[x].status_code == 200:
+                success += 1
+            else:
+                fail += 1
+        self.assertEqual(success, 3)
+
+    def test_origin_queueing_timeouts(self):
+        '''Lets have some requests timeout and ensure that the queue is freed up
+        '''
+        REQUEST_COUNT = 4
+        results, results2 = self._send_requests(REQUEST_COUNT, path='queue/timeout')
+
+        success = 0
+        fail = 0
+        for x in xrange(0, REQUEST_COUNT):
+            if results[x].status_code == 200:
+                success += 1
+                print 'success', x
+            else:
+                fail += 1
+        self.assertEqual(fail, 4)
+
+        self.test_origin_queueing()
+
+    def test_origin_no_queueing(self):
+        '''If the queue is set to 0, all requests past the max immediately fail
+        '''
+        REQUEST_COUNT = 4
+        results, results2 = self._send_requests(REQUEST_COUNT, path='noqueue/')
+
+        success = 0
+        fail = 0
+        for x in xrange(0, REQUEST_COUNT):
+            if results[x].status_code == 200:
+                success += 1
+            else:
+                fail += 1
+        print 'results:', success, fail
+        self.assertEqual(success, 1)

doc/admin-guide/files/records.config.en.rst

@@ -1217,6 +1217,15 @@ Origin Server Connect Attempts
 
    Limits the number of socket connections per origin server to the value specified. To enable, set to one (``1``).
 
+.. ts:cv:: CONFIG proxy.config.http.origin_max_connections_queue INT -1
+   :reloadable:
+   :overridable:
+
+   Limits the number of requests to be queued when the :ts:cv:`proxy.config.http.origin_max_connections` is reached.
+   When disabled (``-1``) requests are will wait indefinitely for an available connection. When set to ``0`` all 
+   requests past the :ts:cv:`proxy.config.http.origin_max_connections` will immediately fail. When set to ``>0`` 
+   ATS will queue that many requests to go to the origin, any additional requests past the limit will immediately fail. 
+
 .. ts:cv:: CONFIG proxy.config.http.origin_min_keep_alive_connections INT 0
    :reloadable:
 

lib/ts/apidefs.h.in

@@ -694,6 +694,7 @@ typedef enum {
   TS_CONFIG_HTTP_CACHE_MAX_OPEN_WRITE_RETRIES,
   TS_CONFIG_HTTP_REDIRECT_USE_ORIG_CACHE_KEY,
   TS_CONFIG_HTTP_ATTACH_SERVER_SESSION_TO_CLIENT,
+  TS_CONFIG_HTTP_ORIGIN_MAX_CONNECTIONS_QUEUE,
   TS_CONFIG_LAST_ENTRY
 } TSOverridableConfigKey;
 

lib/ts/ink_inet.cc

@@ -307,6 +307,26 @@ ats_ip_hash(sockaddr const *addr)
   return zret.i;
 }
 
+uint64_t
+ats_ip_port_hash(sockaddr const *addr)
+{
+  union md5sum {
+    uint64_t i;
+    uint16_t b[4];
+    unsigned char c[16];
+  } zret;
+
+  zret.i = 0;
+  if (ats_is_ip4(addr)) {
+    zret.i = (static_cast<uint64_t>(ats_ip4_addr_cast(addr)) << 16) | (ats_ip_port_cast(addr));
+  } else if (ats_is_ip6(addr)) {
+    ink_code_md5(const_cast<uint8_t *>(ats_ip_addr8_cast(addr)), TS_IP6_SIZE, zret.c);
+    // now replace the bottom 16bits so we can account for the port.
+    zret.b[3] = ats_ip_port_cast(addr);
+  }
+  return zret.i;
+}
+
 int
 ats_ip_to_hex(sockaddr const *src, char *dst, size_t len)
 {

lib/ts/ink_inet.h

@@ -1151,6 +1151,8 @@ int ats_ip_getbestaddrinfo(char const *name, ///< [in] Address name (IPv4, IPv6,
 */
 uint32_t ats_ip_hash(sockaddr const *addr);
 
+uint64_t ats_ip_port_hash(sockaddr const *addr);
+
 /** Convert address to string as a hexidecimal value.
     The string is always nul terminated, the output string is clipped
     if @a dst is insufficient.

mgmt/RecordsConfig.cc

@@ -473,6 +473,8 @@ static const RecordElement RecordsConfig[] =
   ,
   {RECT_CONFIG, "proxy.config.http.origin_max_connections", RECD_INT, "0", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL}
   ,
+  {RECT_CONFIG, "proxy.config.http.origin_max_connections_queue", RECD_INT, "-1", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL}
+    ,
   {RECT_CONFIG, "proxy.config.http.origin_min_keep_alive_connections", RECD_INT, "0", RECU_DYNAMIC, RR_NULL, RECC_STR, "^[0-9]+$", RECA_NULL}
   ,
   {RECT_CONFIG, "proxy.config.http.attach_server_session_to_client", RECD_INT, "0", RECU_DYNAMIC, RR_NULL, RECC_INT, "[0-1]", RECA_NULL}

plugins/experimental/ts_lua/ts_lua_http_config.c

@@ -114,6 +114,7 @@ typedef enum {
   TS_LUA_CONFIG_HTTP_CACHE_MAX_OPEN_WRITE_RETRIES = TS_CONFIG_HTTP_CACHE_MAX_OPEN_WRITE_RETRIES,
   TS_LUA_CONFIG_HTTP_REDIRECT_USE_ORIG_CACHE_KEY = TS_CONFIG_HTTP_REDIRECT_USE_ORIG_CACHE_KEY,
   TS_LUA_CONFIG_HTTP_ATTACH_SERVER_SESSION_TO_CLIENT = TS_CONFIG_HTTP_ATTACH_SERVER_SESSION_TO_CLIENT,
+  TS_LUA_CONFIG_HTTP_ORIGIN_MAX_CONNECTIONS_QUEUE = TS_CONFIG_HTTP_ORIGIN_MAX_CONNECTIONS_QUEUE,
   TS_LUA_CONFIG_LAST_ENTRY = TS_CONFIG_LAST_ENTRY,
 } TSLuaOverridableConfigKey;
 
@@ -198,7 +199,9 @@ ts_lua_var_item ts_lua_http_config_vars[] = {
   TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_HTTP_ENABLE_REDIRECTION), TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_HTTP_NUMBER_OF_REDIRECTIONS),
   TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_HTTP_CACHE_MAX_OPEN_WRITE_RETRIES),
   TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_HTTP_REDIRECT_USE_ORIG_CACHE_KEY),
-  TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_HTTP_ATTACH_SERVER_SESSION_TO_CLIENT), TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_LAST_ENTRY),
+  TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_HTTP_ATTACH_SERVER_SESSION_TO_CLIENT),
+  TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_HTTP_ORIGIN_MAX_CONNECTIONS_QUEUE),
+  TS_LUA_MAKE_VAR_ITEM(TS_LUA_CONFIG_LAST_ENTRY),
 };
 
 // Needed to make sure we have the latest list of overridable http config vars when compiling

proxy/InkAPI.cc

@@ -7978,6 +7978,10 @@ _conf_to_memberp(TSOverridableConfigKey conf, OverridableHttpConfigParams *overr
     typ = OVERRIDABLE_TYPE_INT;
     ret = &overridableHttpConfig->attach_server_session_to_client;
     break;
+  case TS_CONFIG_HTTP_ORIGIN_MAX_CONNECTIONS_QUEUE:
+    typ = OVERRIDABLE_TYPE_INT;
+    ret = &overridableHttpConfig->origin_max_connections_queue;
+    break;
   // This helps avoiding compiler warnings, yet detect unhandled enum members.
   case TS_CONFIG_NULL:
   case TS_CONFIG_LAST_ENTRY:
@@ -8488,6 +8492,8 @@ TSHttpTxnConfigFind(const char *name, int length, TSOverridableConfigKey *conf,
         cnf = TS_CONFIG_HTTP_CACHE_HEURISTIC_MIN_LIFETIME;
       else if (!strncmp(name, "proxy.config.http.cache.heuristic_max_lifetime", length))
         cnf = TS_CONFIG_HTTP_CACHE_HEURISTIC_MAX_LIFETIME;
+      else if (!strncmp(name, "proxy.config.http.origin_max_connections_queue", length))
+        cnf = TS_CONFIG_HTTP_ORIGIN_MAX_CONNECTIONS_QUEUE;
       break;
     case 'r':
       if (!strncmp(name, "proxy.config.http.insert_squid_x_forwarded_for", length))

proxy/http/HttpConfig.cc

@@ -891,6 +891,7 @@ HttpConfig::startup()
   HttpEstablishStaticConfigLongLong(c.max_websocket_connections, "proxy.config.http.websocket.max_number_of_connections");
   HttpEstablishStaticConfigLongLong(c.oride.server_tcp_init_cwnd, "proxy.config.http.server_tcp_init_cwnd");
   HttpEstablishStaticConfigLongLong(c.oride.origin_max_connections, "proxy.config.http.origin_max_connections");
+  HttpEstablishStaticConfigLongLong(c.oride.origin_max_connections_queue, "proxy.config.http.origin_max_connections_queue");
   HttpEstablishStaticConfigLongLong(c.origin_min_keep_alive_connections, "proxy.config.http.origin_min_keep_alive_connections");
   HttpEstablishStaticConfigLongLong(c.oride.attach_server_session_to_client, "proxy.config.http.attach_server_session_to_client");
 
@@ -1162,6 +1163,13 @@ HttpConfig::reconfigure()
   params->max_websocket_connections = m_master.max_websocket_connections;
   params->oride.server_tcp_init_cwnd = m_master.oride.server_tcp_init_cwnd;
   params->oride.origin_max_connections = m_master.oride.origin_max_connections;
+  params->oride.origin_max_connections_queue = m_master.oride.origin_max_connections_queue;
+  // if origin_max_connections_queue is set without max_connections, it is meaningless, so we'll warn
+  if (params->oride.origin_max_connections_queue &&
+      !(params->oride.origin_max_connections || params->origin_min_keep_alive_connections)) {
+    Warning("origin_max_connections_queue is set, but neither origin_max_connections nor origin_min_keep_alive_connections are "
+            "set, please correct your records.config");
+  }
   params->origin_min_keep_alive_connections = m_master.origin_min_keep_alive_connections;
   params->oride.attach_server_session_to_client = m_master.oride.attach_server_session_to_client;
 

proxy/http/HttpConfig.h

@@ -378,7 +378,7 @@ struct OverridableHttpConfigParams {
       cache_heuristic_min_lifetime(3600), cache_heuristic_max_lifetime(86400), cache_guaranteed_min_lifetime(0),
       cache_guaranteed_max_lifetime(31536000), cache_max_stale_age(604800), keep_alive_no_activity_timeout_in(115),
       keep_alive_no_activity_timeout_out(120), transaction_no_activity_timeout_in(30), transaction_no_activity_timeout_out(30),
-      transaction_active_timeout_out(0), origin_max_connections(0), attach_server_session_to_client(0),
+      transaction_active_timeout_out(0), origin_max_connections(0), origin_max_connections_queue(0), attach_server_session_to_client(0),
       connect_attempts_max_retries(0), connect_attempts_max_retries_dead_server(3), connect_attempts_rr_retries(3),
       connect_attempts_timeout(30), post_connect_attempts_timeout(1800), down_server_timeout(300), client_abort_threshold(10),
       freshness_fuzz_time(240), freshness_fuzz_min_time(0), max_cache_open_read_retries(-1), cache_open_read_retry_time(10),
@@ -529,6 +529,7 @@ struct OverridableHttpConfigParams {
   MgmtInt transaction_no_activity_timeout_out;
   MgmtInt transaction_active_timeout_out;
   MgmtInt origin_max_connections;
+  MgmtInt origin_max_connections_queue;
 
   MgmtInt attach_server_session_to_client;
 

proxy/http/HttpConnectionCount.cc

@@ -25,3 +25,4 @@
 
 
 ConnectionCount ConnectionCount::_connectionCount;
+ConnectionCountQueue ConnectionCountQueue::_connectionCount;