MXS-173 Test for throttling filter

The test may fail if the client/maxscale/mariadb combo is too slow.
TODO, maybe: today I saw mysql_query() hang (in a poll) when maxscale
disconnected. Is there something that can be done about that?
I added a source directory, base, for stuff that should become part of a common
utility library in the future.

maxscale-system-test/CMakeLists.txt

@@ -30,13 +30,15 @@ enable_testing()
 # utilities.cmake contains all helper functions and extra tools
 include(utilities.cmake)
 
+include_directories(${CMAKE_SOURCE_DIR})
+
 # The core library
 include_directories(${CMAKE_BINARY_DIR})
 add_library(testcore SHARED testconnections.cpp nodes.cpp mariadb_nodes.cpp maxscales.cpp
   mariadb_func.cpp get_com_select_insert.cpp maxadmin_operations.cpp big_transaction.cpp
   sql_t1.cpp test_binlog_fnc.cpp get_my_ip.cpp big_load.cpp  get_com_select_insert.cpp
   different_size.cpp fw_copy_rules maxinfo_func.cpp config_operations.cpp rds_vpc.cpp execute_cmd.cpp
-  blob_test.cpp keepalived_func.cpp tcp_connection.cpp
+  blob_test.cpp keepalived_func.cpp tcp_connection.cpp base/stopwatch.cpp
   # Include the CDC connector in the core library
   ${CMAKE_SOURCE_DIR}/../connectors/cdc-connector/cdc_connector.cpp)
 target_link_libraries(testcore ${MYSQL_CLIENT} ${JANSSON_LIBRARIES} z m pthread ssl dl rt crypto crypt)
@@ -975,6 +977,9 @@ add_test_executable(mxs1628_bad_handshake.cpp mxs1628_bad_handshake replication
 # MXS-1836: MaxInfo "show eventTimes" returns garbage.
 add_test_executable(mxs1836_show_eventTimes.cpp mxs1836_show_eventTimes mxs1836_show_eventTimes LABELS REPL_BACKEND)
 
+# MXS-173 throttling filter
+add_test_executable(mxs173.cpp mxs173_throttle_filter mxs173_throttle_filter LABELS REPL_BACKEND)
+
 configure_file(templates.h.in templates.h @ONLY)
 
 include(CTest)

maxscale-system-test/base/appexception.h

@@ -0,0 +1,47 @@
+/*
+ * Copyright (c) 2018 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2020-01-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#pragma once
+
+#include <stdexcept>
+#include <sstream>
+
+namespace base
+{
+
+// TODO Add back trace.
+class AppException : public std::runtime_error
+{
+public:
+    AppException(const std::string& msg, const std::string& file,
+                 int line) :
+        std::runtime_error(msg), m_file(file), m_line(line)
+    {}
+private:
+    std::string m_file;
+    int m_line;
+};
+} //base
+
+#define DEFINE_EXCEPTION(Type) \
+    struct Type : public base::AppException { \
+        Type(const std::string& msg, const char* file, \
+            int line) : \
+            AppException(msg, file, line) {}}
+
+#define THROW(Type, msg_str) do {\
+        std::ostringstream os; \
+        os << __FILE__ << ':' << __LINE__ << '\n' << msg_str; \
+        throw Type(os.str(), __FILE__, __LINE__);} while(false)
+
+

maxscale-system-test/base/env.h

@@ -0,0 +1,46 @@
+/*
+ * Copyright (c) 2018 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2020-01-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+#pragma once
+
+#include <string>
+
+namespace base
+{
+
+/* Environment variable. Usage:
+ * Env user{"USER"};
+ * std::string home = Env{"HOME"};
+ * An environment variable can be empty() but
+ * still is_defined().
+ */
+class Env : public std::string
+{
+public:
+    Env(const std::string& name) : m_is_defined(false)
+    {
+        if (const char* var = getenv(name.c_str()))
+        {
+            m_is_defined = true;
+            this->assign(var);
+        }
+    }
+
+    bool is_defined() const
+    {
+        return m_is_defined;
+    }
+private:
+    bool m_is_defined;
+};
+
+} // base

maxscale-system-test/base/stopwatch.cpp

@@ -0,0 +1,148 @@
+/*
+ * Copyright (c) 2018 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2020-01-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+
+#include "stopwatch.h"
+
+#include <iomanip>
+#include <iostream>
+#include <sstream>
+#include <ctime>
+
+namespace base
+{
+
+StopWatch::StopWatch()
+{
+    restart();
+}
+
+Duration StopWatch::lap() const
+{
+    return {Clock::now() - m_start};
+}
+
+Duration StopWatch::restart()
+{
+    TimePoint now = Clock::now();
+    Duration lap = now - m_start;
+    m_start = now;
+    return lap;
+}
+} // base
+
+/********** OUTPUT ***********/
+namespace
+{
+using namespace base;
+struct TimeConvert
+{
+    double div;         // divide the value of the previous unit by this
+    std::string suffix; // milliseconds, hours etc.
+    double max_visual;  // threashold to switch to the next unit
+};
+// Will never get to centuries because the duration is a long carrying nanoseconds
+TimeConvert convert[]
+{
+    {1, "ns", 1000}, {1000, "us", 1000},  {1000, "ms", 1000},
+    {1000, "s", 60}, {60, "min", 60}, {60, "hours", 24},
+    {24, "days", 365.25}, {365.25, "years", 10000},
+    {100, "centuries", std::numeric_limits<double>::max()}
+};
+
+int convert_size = sizeof(convert) / sizeof(convert[0]);
+
+}
+
+namespace base
+{
+std::pair<double, std::string> dur_to_human_readable(Duration dur)
+{
+    using namespace std::chrono;
+    double time = duration_cast<nanoseconds>(dur).count();
+    bool negative = (time < 0) ? time = -time, true : false;
+
+    for (int i = 0; i <= convert_size; ++i)
+    {
+        if (i == convert_size)
+        {
+            return std::make_pair(negative ? -time : time,
+                                  convert[convert_size - 1].suffix);
+        }
+
+        time /= convert[i].div;
+
+        if (time < convert[i].max_visual)
+        {
+            return std::make_pair(negative ? -time : time, convert[i].suffix);
+        }
+    }
+
+    abort(); // should never get here
+}
+
+std::ostream& operator<<(std::ostream& os, Duration dur)
+{
+    auto p = dur_to_human_readable(dur);
+    os << p.first << p.second;
+
+    return os;
+}
+
+// TODO: this will require some thought. time_point_to_string() for a system_clock is
+// obvious, but not so for a steady_clock. Maybe TimePoint belongs to a system clock
+// and sould be called something else here, and live in a time_measuring namespace.
+std::string time_point_to_string(TimePoint tp, const std::string &fmt)
+{
+    using namespace std::chrono;
+    std::time_t timet = system_clock::to_time_t(system_clock::now()
+                                                + (tp - Clock::now()));
+    struct tm * ptm;
+    ptm = gmtime(&timet);
+    const int sz = 1024;
+    char buf[sz];
+    strftime(buf, sz, fmt.c_str(), ptm);
+    return buf;
+}
+
+std::ostream & operator<<(std::ostream & os, TimePoint tp)
+{
+    os << time_point_to_string(tp);
+    return os;
+}
+
+void test_stopwatch_output(std::ostream & os)
+{
+    long long dur[] =
+    {
+        400,    // 400ns
+        5 * 1000, // 5us
+        500 * 1000, // 500us
+        1 * 1000000, // 1ms
+        700 * 1000000LL, // 700ms
+        5 * 1000000000LL, // 5s
+        200 * 1000000000LL, // 200s
+        5 * 60 * 1000000000LL, // 5m
+        45 * 60 * 1000000000LL, // 45m
+        130 * 60 * 1000000000LL, // 130m
+        24 * 60 * 60 * 1000000000LL, // 24 hours
+        3 * 24 * 60 * 60 * 1000000000LL, // 72 hours
+        180 * 24 * 60 * 60 * 1000000000LL, // 180 days
+        1000 * 24 * 60 * 60 * 1000000000LL // 1000 days
+    };
+
+    for (unsigned i = 0; i < sizeof(dur) / sizeof(dur[0]); ++i)
+    {
+        os << Duration(dur[i]) << std::endl;
+    }
+}
+} // base

maxscale-system-test/base/stopwatch.h

@@ -0,0 +1,56 @@
+/*
+ * Copyright (c) 2018 MariaDB Corporation Ab
+ *
+ * Use of this software is governed by the Business Source License included
+ * in the LICENSE.TXT file and at www.mariadb.com/bsl11.
+ *
+ * Change Date: 2020-01-01
+ *
+ * On the date above, in accordance with the Business Source License, use
+ * of this software will be governed by version 2 or later of the General
+ * Public License.
+ */
+#pragma once
+
+#include <chrono>
+#include <iosfwd>
+#include <string>
+
+// Same StopWatch as in maxscale, but dropping support for gcc 4.4
+namespace base
+{
+using Clock = std::chrono::steady_clock;
+
+struct Duration : public Clock::duration // for ADL
+{
+    using Clock::duration::duration;
+    Duration() = default;
+    Duration(Clock::duration d) : Clock::duration(d) {}
+};
+
+using TimePoint = std::chrono::time_point<Clock, Duration>;
+
+class StopWatch
+{
+public:
+    // Starts the stopwatch, which is always running.
+    StopWatch();
+    // Get elapsed time.
+    Duration lap() const;
+    // Get elapsed time, restart StopWatch.
+    Duration restart();
+private:
+    TimePoint m_start;
+};
+
+// Returns the value as a double and string adjusted to a suffix like ms for milliseconds.
+std::pair<double, std::string> dur_to_human_readable(Duration dur);
+
+// Human readable output. No standard library for it yet.
+std::ostream& operator<<(std::ostream&, Duration dur);
+
+// TimePoint
+std::string time_point_to_string(TimePoint tp, const std::string& fmt = "%F %T");
+std::ostream& operator<<(std::ostream&, TimePoint tp);
+
+} // base

maxscale-system-test/cnf/maxscale.cnf.template.mxs173_throttle_filter

@@ -0,0 +1,116 @@
+[maxscale]
+ms_timestamp = 1
+threads = ###threads###
+
+[MySQL-Monitor]
+type = monitor
+module = mariadbmon
+servers = server1,server2,server3,server4
+user = maxskysql
+passwd = skysql
+monitor_interval = 5000
+auto_failover = 1
+
+[RW-Split-Router]
+type = service
+router = readwritesplit
+servers = server1,server2,server3,server4
+user = maxskysql
+passwd = skysql
+filters = throttle
+
+[RW-Split-Listener]
+type = listener
+service = RW-Split-Router
+protocol = mariadbclient
+port = 4006
+
+[Read-Connection-Router-Master]
+type = service
+router = readconnroute
+router_options = master
+servers = server1,server2,server3,server4
+user = maxskysql
+passwd = skysql
+
+[Read-Connection-Listener-Master]
+type = listener
+service = Read-Connection-Router-Master
+protocol = MySQLClient
+port = 4008
+
+[Read-Connection-Router-Slave]
+type = service
+router = readconnroute
+router_options = slave
+servers = server1,server2,server3,server4
+user = maxskysql
+passwd = skysql
+
+[Read-Connection-Listener-Slave]
+type = listener
+service = Read-Connection-Router-Slave
+protocol = MySQLClient
+port = 4009
+
+[throttle]
+type = filter
+module = throttlefilter
+max_qps = 1000
+throttling_duration = 10000
+sampling_duration = 250
+continuous_duration = 2000
+
+[Debug-Service]
+type = service
+router = debugcli
+
+[Debug-Listener]
+type = listener
+service = Debug-Service
+protocol = telnetd
+port = 4007
+
+[MaxAdmin-Service]
+type = service
+router = cli
+
+[MaxAdmin-Unix-Listener]
+type = listener
+service = MaxAdmin-Service
+protocol = maxscaled
+socket = default
+
+[server1]
+type = server
+address = ###node_server_IP_1###
+port = ###node_server_port_1###
+protocol = MariaDBBackend
+localhost_match_wildcard_host = true
+router_options = master
+
+[server2]
+type = server
+address = ###node_server_IP_2###
+port = ###node_server_port_2###
+protocol = MariaDBBackend
+localhost_match_wildcard_host = true
+router_options = slave
+
+[server3]
+type = server
+address = ###node_server_IP_3###
+port = ###node_server_port_3###
+protocol = MariaDBBackend
+localhost_match_wildcard_host = true
+router_options = slave
+
+[server4]
+type = server
+address = ###node_server_IP_4###
+port = ###node_server_port_4###
+protocol = MariaDBBackend
+localhost_match_wildcard_host = true
+router_options = slave
+
+

maxscale-system-test/mxs173.cpp

@@ -0,0 +1,236 @@
+/*
+ * MXS-173 throttling filter
+ *
+ */
+
+#include "testconnections.h"
+#include <base/stopwatch.h>
+#include <base/appexception.h>
+#include <iostream>
+#include <memory>
+#include <sstream>
+#include <cmath>
+
+DEFINE_EXCEPTION(Whoopsy);
+
+// TODO these should be read from maxscale.cnf. Maybe the test-lib should replace
+// any "###ENV_VAR###", with environment variables so that code and conf can share.
+constexpr int   max_qps = 1000;
+constexpr float throttling_duration = 10000 / 1000.0;
+constexpr float sampling_duration = 250 / 1000.0;
+constexpr float continuous_duration = 2000 / 1000.0;
+
+constexpr int NUM_ROWS = 100000;
+
+void create_table(MYSQL* conn)
+{
+    if (execute_query_silent(conn, "drop table if exists test.throttle;"
+                             "create table test.throttle(id int, name varchar(30),"
+                             "primary key(id));"))
+    {
+        THROW(Whoopsy, "Create table failed - could not start test");
+    }
+}
+
+void insert_rows(MYSQL* conn)
+{
+    std::ostringstream os;
+    os << "insert into throttle values\n";
+
+    for (int i = 0; i < NUM_ROWS; ++i)
+    {
+        if (i != 0)
+        {
+            os << ',';
+        }
+        os << '(' << i << ", '"
+           << std::to_string(i) << "')\n";
+    }
+
+    os << ';';
+
+    if (execute_query_silent(conn, os.str().c_str()), false)
+    {
+        THROW(Whoopsy, "Inserts failed - could not start test");
+    }
+}
+
+struct ReadSpeed
+{
+    bool error;
+    base::Duration duration;
+    float qps;
+};
+
+ReadSpeed read_rows(MYSQL* conn, int num_rows, int max_qps, bool expect_error)
+{
+    base::StopWatch sw;
+    for (int i = 0; i < num_rows; ++i)
+    {
+        int index = rand() % NUM_ROWS;
+        std::ostringstream os;
+        os << "select name from test.throttle where id=" << index;
+
+        if (mysql_query(conn, os.str().c_str()))
+        {
+            if (expect_error)
+            {
+                base::Duration dur = sw.lap();
+                std::cout << "\r             \r";
+                return {true, dur, i / std::chrono::duration<float>(dur).count()};
+            }
+            else
+            {
+                THROW(Whoopsy, "Unexpected error while reading rows.");
+            }
+        }
+
+        // Maybe a create function template make_unique_deleter. Should work for most cases.
+        // But it would be safer with one regular function per type/function pair, returning
+        // a unique pointer.
+        std::unique_ptr<MYSQL_RES, void(*)(MYSQL_RES*)> result(mysql_store_result(conn),
+                                                               mysql_free_result);
+        if (!result)
+        {
+            THROW(Whoopsy, "No resultset for index=" << index);
+        }
+
+        MYSQL_ROW row = mysql_fetch_row(&*result);
+        if (row)
+        {
+            if (std::stoi(row[0]) != index)
+            {
+                THROW(Whoopsy, "Differing values index=" << index << " name=" << row[0]);
+            }
+        }
+        else
+        {
+            THROW(Whoopsy, "Row id = " << index << " not in resultset.");
+        }
+
+
+        if ((row = mysql_fetch_row(&*result)))
+        {
+            THROW(Whoopsy, "Extra row index = " << index << " name = " << row[0] << " in resultset.");
+        }
+
+        std::cout << "\r" <<  num_rows - i - 1 << ' ';
+        if (i % 117 == 0)
+        {
+            std::cout.flush();
+        }
+    }
+
+    base::Duration dur = sw.lap();
+    std::cout << "\r             \r";
+    return ReadSpeed {false, dur, num_rows / std::chrono::duration<float>(dur).count()};
+}
+
+void gauge_raw_speed(TestConnections& test)
+{
+    const int raw_rows = NUM_ROWS / 5;
+    std::cout << "\n****\nRead " << raw_rows
+              << " rows via master readconnrouter, to gauge speed.\n";
+    auto rs = read_rows(test.maxscales->conn_master[0], raw_rows, 0, false);
+    std::cout << rs.qps << "qps " << " duration " << rs.duration << '\n';
+
+    if (rs.qps < 2 * max_qps)
+    {
+        std::ostringstream os;
+        os << "The raw speed is too slow, " << rs.qps
+           << "qps, compared to max_qps = " << max_qps << "qps for accurate testing.";
+        test.add_result(1, os.str().c_str());
+    }
+}
+
+void verify_throttling_performace(TestConnections& test)
+{
+    int three_quarter = 3 * max_qps * throttling_duration / 4;
+    std::cout << "\n****\nRead " << three_quarter
+              << " rows which should take about " << 3 * throttling_duration / 4
+              << " seconds.\nThrottling should keep qps around "
+              << max_qps << ".\n";
+    auto rs1 = read_rows(test.maxscales->conn_rwsplit[0], three_quarter, 0, false);
+    std::cout << "1: " << rs1.qps << "qps " << " duration " << rs1.duration << '\n';
+    std::cout << "Sleep for " << continuous_duration << "s (continuous_duration)\n";
+    usleep(continuous_duration * 1000000);
+    std::cout << "Run the same read again. Should be throttled, but not disconnected.\n";
+    auto rs2 = read_rows(test.maxscales->conn_rwsplit[0], three_quarter, 0, false);
+    std::cout << "2: " << rs2.qps << "qps " << " duration " << rs2.duration << '\n';
+
+    if (std::abs(rs1.qps - max_qps) > 0.1 * max_qps ||
+        std::abs(rs2.qps - max_qps) > 0.1 * max_qps)
+    {
+        std::ostringstream os;
+        os << "Throttled speed 1: " << rs1.qps << " or 2: " << rs2.qps <<
+           "differs from max_qps " << max_qps << " by more than 10%%";
+        test.add_result(1, os.str().c_str());
+    }
+}
+
+void verify_throttling_disconnect(TestConnections& test)
+{
+    int half_rows = max_qps * throttling_duration / 2;
+    std::cout << "\n****\nRead " << 3 * half_rows
+              << " rows which should cause a disconnect at a little\nbelow "
+              << half_rows << " rows to go, in about " << throttling_duration << "s.\n";
+    auto rs = read_rows(test.maxscales->conn_rwsplit[0], 3 * half_rows, 0, true);
+    std::cout << rs.qps << "qps " << " duration " << rs.duration << '\n';
+
+    if (!rs.error)
+    {
+        std::ostringstream os;
+        os << "Throttle filter did not disconnect rogue session.\n"
+           << rs.qps << "qps " << " duration " << rs.duration;
+        test.add_result(1, os.str().c_str());
+    }
+    if (std::abs(rs.qps - max_qps) > 0.1 * max_qps)
+    {
+        std::ostringstream os;
+        os << "Throttled speed " << rs.qps << " differs from max_qps " << max_qps
+           << " by more than 10%%";
+        test.add_result(1, os.str().c_str());
+    }
+}
+
+int main(int argc, char *argv[])
+{
+    srand(clock());
+    TestConnections test{argc, argv};
+
+    try
+    {
+        test.print_env();
+        test.maxscales->connect_maxscale(0);
+
+        std::cout << "Create table\n";
+        create_table(test.maxscales->conn_master[0]);
+
+        std::cout << "Insert rows\n";
+        insert_rows(test.maxscales->conn_master[0]);
+
+        gauge_raw_speed(test);
+
+        test.repl->sync_slaves();
+
+        verify_throttling_performace(test);
+
+        test.maxscales->close_maxscale_connections(0);
+        test.maxscales->connect_maxscale(0);
+
+        verify_throttling_disconnect(test);
+
+        std::cout << "\n\n";
+    }
+    catch (std::exception& ex)
+    {
+        test.add_result(1, ex.what());
+    }
+    catch (...)
+    {
+        test.add_result(1, "catch ...");
+        throw;
+    }
+
+    return test.global_result;
+}

maxscale-system-test/testconnections.cpp

@@ -504,8 +504,14 @@ void TestConnections::print_env()
     printf("Maxscale SSH key\t%s\n", maxscales->sshkey[0]);
     printf("Maxadmin password\t%s\n", maxscales->maxadmin_password[0]);
     printf("Access user\t%s\n", maxscales->access_user[0]);
+    if (repl)
+    {
         repl->print_env();
+    }
+    if (galera)
+    {
         galera->print_env();
+    }
 }
 
 const char * get_template_name(char * test_name)
@@ -686,7 +692,10 @@ int TestConnections::copy_mariadb_logs(Mariadb_nodes * repl, char * prefix)
     int exit_code;
     char str[4096];
 
-    if (repl == NULL) return local_result;
+    if (repl == NULL)
+    {
+        return local_result;
+    }
 
     sprintf(str, "mkdir -p LOGS/%s", test_name);
     system(str);
@@ -1817,7 +1826,9 @@ StringSet TestConnections::get_server_status(const char* name)
             char* p = tok;
             char *end = strchr(tok, '\n');
             if (!end)
+            {
                 end = strchr(tok, '\0');
+            }
 
             // Trim leading whitespace
             while (p < end && isspace(*p))
@@ -1913,14 +1924,14 @@ std::string dump_status(const StringSet& current, const StringSet& expected)
     std::stringstream ss;
     ss << "Current status: (";
 
-    for (const auto& a: current)
+    for (const auto& a : current)
     {
         ss << a << ",";
     }
 
     ss << ") Expected status: (";
 
-    for (const auto& a: expected)
+    for (const auto& a : expected)
     {
         ss << a << ",";
     }