MXS-1980 Check that Galera replication works

It should be possible to START SLAVE and STOP SLAVE irrespective
of which Galera node updates are mode to.

That will be the case if @@log_slave_updates is on and each node
in the Galera cluster have the same server id. Otherwise it will
fail with the current incarnation of BLR.

maxscale-system-test/CMakeLists.txt

@@ -1068,6 +1068,10 @@ add_test_executable(mxs1113_schemarouter_ps.cpp mxs1113_schemarouter_ps mxs1113_
 # https://jira.mariadb.org/browse/MXS-2037
 add_test_executable(mxs2037_namedserver_wildcards.cpp mxs2037_namedserver_wildcards mxs2037_namedserver_wildcards LABELS namedserverfilter LIGHT REPL_BACKEND)
 
+# MXS-1980: Support Galera cluster nodes as masters for Binlog Router
+# https://jira.mariadb.org/browse/MXS-1980
+add_test_executable(mxs1980_blr_galera_server_ids.cpp mxs1980_blr_galera_server_ids mxs1980_blr_galera_server_ids LABELS binlogrouter GALERA_BACKEND)
+
 # MXS-701: Binlog filtering
 # https://jira.mariadb.org/browse/MXS-701
 add_test_executable(mxs701_binlog_filter.cpp mxs701_binlog_filter mxs701_binlog_filter LABELS bilogrouter REPL_BACKEND)

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

@@ -0,0 +1,26 @@
+[BLR]
+type=service
+router=binlogrouter
+user=repl
+password=repl
+version_string=5.6.15-log
+master_id=5
+server_id=4711
+mariadb10_master_gtid=On
+
+[BLR-Listener]
+type=listener
+service=BLR
+protocol=MariaDBClient
+port=5306
+
+[CLI]
+type=service
+router=cli
+
+[CLI Listener]
+type=listener
+service=CLI
+protocol=maxscaled
+#address=localhost
+socket=default

maxscale-system-test/mxs1980_blr_galera_server_ids.cpp

@@ -0,0 +1,366 @@
+/*
+ * Copyright (c) 2016 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: 2022-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 <algorithm>
+#include <map>
+#include <iostream>
+#include <iterator>
+#include <sstream>
+#include <maxbase/assert.h>
+#include <maxbase/log.hh>
+#include "testconnections.h"
+
+using namespace std;
+
+// This test checks that BLR replication from a Galera cluster works if
+// - all servers in the Galera cluster have @@log_slave_updates on,
+// - all servers in the Galera cluster have the same server id, and
+// - even if updates are made in every node of the cluster.
+//
+//
+// By default that will not work as BLR stores the binlog file in a directory
+// named according to the server id *and* later assumes that the directory
+// can be deduced from the GTID. That is an erroneous assumption, as the GTID
+// of events generated in a Galera cluster contain the server id of the node
+// where the write was generated, not the server id of the node from which
+// BLR replicates.
+
+namespace
+{
+
+// The amount of time slept between various operations that are
+// expected to take some time before becoming visible.
+
+const int REPLICATION_SLEEP = 5; // Seconds
+
+string get_gtid_current_pos(TestConnections& test, MYSQL* pMysql)
+{
+    std::vector<string> row = get_row(pMysql, "SELECT @@gtid_current_pos");
+
+    test.expect(row.size() == 1, "Did not get @@gtid_current_pos");
+
+    return row.empty() ? "" : row[0];
+}
+
+string get_server_id(TestConnections& test, MYSQL* pMysql)
+{
+    std::vector<string> row = get_row(pMysql, "SELECT @@server_id");
+
+    test.expect(row.size() == 1, "Did not get @@server_id");
+
+    return row.empty() ? "" : row[0];
+}
+
+// Setup BLR to replicate from galera_000.
+bool setup_blr(TestConnections& test, const string& gtid, const char* zHost, int port)
+{
+    test.tprintf("Connecting to BLR at %s:%d", zHost, port);
+
+    MYSQL* pMysql = open_conn_no_db(port, zHost, "repl", "repl");
+    test.expect(pMysql, "Could not open connection to %s.", zHost);
+
+    if (pMysql)
+    {
+        test.try_query(pMysql, "STOP SLAVE");
+        test.try_query(pMysql, "SET @@global.gtid_slave_pos='%s'", gtid.c_str());
+
+        mxb_assert(test.galera);
+        Mariadb_nodes& gc = *test.galera;
+
+        stringstream ss;
+
+        ss << "CHANGE MASTER ";
+        ss << "TO MASTER_HOST='" << gc.IP[0] << "', ";
+        ss << "MASTER_PORT=" << gc.port[0] << ", ";
+        ss << "MASTER_USER='repl', MASTER_PASSWORD='repl', MASTER_USE_GTID=Slave_pos";
+
+        string stmt = ss.str();
+
+        cout << stmt << endl;
+
+        test.try_query(pMysql, "%s", stmt.c_str());
+        test.try_query(pMysql, "START SLAVE");
+
+        mysql_close(pMysql);
+    }
+
+    return test.global_result == 0;
+}
+
+// Setup slave to replicate from BLR.
+bool setup_slave(TestConnections& test,
+                 const string& gtid,
+                 MYSQL* pSlave,
+                 const char* zMaxscale_host,
+                 int maxscale_port)
+{
+    test.try_query(pSlave, "STOP SLAVE");
+    test.try_query(pSlave, "RESET SLAVE");
+    test.try_query(pSlave, "DROP TABLE IF EXISTS test.MXS2011");
+    test.try_query(pSlave, "SET @@global.gtid_slave_pos='%s'", gtid.c_str());
+
+    stringstream ss;
+
+    ss << "CHANGE MASTER TO ";
+    ss << "MASTER_HOST='" << zMaxscale_host << "', ";
+    ss << "MASTER_PORT=" << maxscale_port << ", ";
+    ss << "MASTER_USER='repl', MASTER_PASSWORD='repl', MASTER_USE_GTID=Slave_pos";
+
+    string stmt = ss.str();
+
+    cout << stmt << endl;
+
+    test.try_query(pSlave, "%s", stmt.c_str());
+    test.try_query(pSlave, "START SLAVE");
+
+    return test.global_result == 0;
+}
+
+bool setup_schema(TestConnections& test, MYSQL* pServer)
+{
+    test.try_query(pServer, "DROP TABLE IF EXISTS test.MXS2011");
+    test.try_query(pServer, "CREATE TABLE test.MXS2011 (i INT)");
+
+    return test.global_result == 0;
+}
+
+unsigned count = 0;
+
+void insert(TestConnections& test, MYSQL* pMaster)
+{
+    stringstream ss;
+    ss << "INSERT INTO test.MXS2011 VALUES (" << ++count << ")";
+
+    string stmt = ss.str();
+
+    cout << stmt.c_str() << endl;
+
+    test.try_query(pMaster, "%s", stmt.c_str());
+}
+
+void select(TestConnections& test, MYSQL* pSlave)
+{
+    my_ulonglong nRows;
+    unsigned long long nResult_sets;
+
+    int rc = execute_query_num_of_rows(pSlave, "SELECT * FROM test.MXS2011", &nRows, &nResult_sets);
+    test.expect(rc == 0, "Execution of SELECT failed.");
+
+    if (rc == 0)
+    {
+        mxb_assert(nResult_sets == 1);
+
+        test.expect(nRows == count, "Expected %d rows, got %d.", count, (int)nRows);
+    }
+}
+
+bool insert_select(TestConnections& test, MYSQL* pSlave, MYSQL* pMaster)
+{
+    insert(test, pMaster);
+    sleep(REPLICATION_SLEEP); // To ensure that the insert reaches the slave.
+    select(test, pSlave);
+
+    return test.global_result == 0;
+}
+
+bool insert_select(TestConnections& test, MYSQL* pSlave)
+{
+    Mariadb_nodes& gc = *test.galera;
+
+    for (int i = 0; i < gc.N; ++i)
+    {
+        MYSQL* pMaster = gc.nodes[i];
+
+        insert_select(test, pSlave, pMaster);
+    }
+
+    return test.global_result == 0;
+}
+
+void reset_galera(TestConnections& test)
+{
+    Mariadb_nodes& gc = *test.galera;
+
+    for (int i = 0; i < gc.N; ++i)
+    {
+        test.try_query(gc.nodes[i], "RESET MASTER");
+    }
+}
+
+// Ensure log_slave_updates is on.
+void setup_galera(TestConnections& test)
+{
+    Mariadb_nodes& gc = *test.galera;
+
+    for (int i = 0; i < gc.N; ++i)
+    {
+        gc.stash_server_settings(i);
+        gc.add_server_setting(i, "log_slave_updates=1");
+    }
+}
+
+// Restore log_slave_updates as it was.
+void restore_galera(TestConnections& test)
+{
+    Mariadb_nodes& gc = *test.galera;
+
+    for (int i = 0; i < gc.N; ++i)
+    {
+        gc.restore_server_settings(i);
+    }
+
+    int rc = gc.start_replication();
+    test.expect(rc == 0, "Could not start Galera cluster.");
+}
+
+bool setup_server_ids(TestConnections& test, map<int, string>* pServer_ids_by_index)
+{
+    Mariadb_nodes& gc = *test.galera;
+
+    string common_server_id = get_server_id(test, gc.nodes[0]);
+
+    if (!common_server_id.empty())
+    {
+        test.tprintf("Setting server_id for all servers to %s.", common_server_id.c_str());
+
+        for (int i = 1; i < gc.N; ++i)
+        {
+            string server_id = get_server_id(test, gc.nodes[i]);
+
+            if (!server_id.empty())
+            {
+                test.tprintf("Changing id from %s to %s.", server_id.c_str(), common_server_id.c_str());
+                test.try_query(gc.nodes[i], "set GLOBAL server_id=%s", common_server_id.c_str());
+                pServer_ids_by_index->insert(std::make_pair(i, server_id));
+            }
+        }
+    }
+
+    return test.global_result != 0;
+}
+
+void restore_server_ids(TestConnections& test, const map<int, string>& server_ids_by_index)
+{
+    for_each(server_ids_by_index.begin(),
+             server_ids_by_index.end(),
+             [&test] (const pair<int, string>& server_id_by_index) {
+                 test.try_query(test.galera->nodes[server_id_by_index.first],
+                                "set GLOBAL server_id=%s", server_id_by_index.second.c_str());
+             });
+}
+
+// STOP SLAVE; START SLAVE cycle.
+void restart_slave(TestConnections& test, MYSQL* pSlave)
+{
+    Row row;
+
+    auto replication_failed = [] (const std::string& column)
+        {
+            return column.find("Got fatal error") != string::npos;
+        };
+
+    test.try_query(pSlave, "STOP SLAVE");
+
+    row = get_row(pSlave, "SHOW SLAVE STATUS");
+    auto it1 = std::find_if(row.begin(), row.end(), replication_failed);
+    test.expect(it1 == row.end(), "Replication failed.");
+
+    test.try_query(pSlave, "START SLAVE");
+
+    sleep(REPLICATION_SLEEP);
+
+    // With the correct setup:
+    // - log_slave_updates is on,
+    // - all Galera nodes have the same server id,
+    // this should work.
+
+    row = get_row(pSlave, "SHOW SLAVE STATUS");
+    auto it2 = std::find_if(row.begin(), row.end(), replication_failed);
+    test.expect(it2 == row.end(), "START SLAVE failed.");
+}
+
+}
+
+int main(int argc, char* argv[])
+{
+    mxb::Log log(MXB_LOG_TARGET_STDOUT);
+
+    TestConnections::skip_maxscale_start(true);
+    TestConnections test(argc, argv);
+
+    test.maxscales->ssh_node(0, "rm -f /var/lib/maxscale/master.ini", true);
+    test.maxscales->ssh_node(0, "rm -f /var/lib/maxscale/gtid_maps.db", true);
+    test.maxscales->ssh_node(0, "rm -rf /var/lib/maxscale/0", true);
+
+    test.start_maxscale(0);
+
+    setup_galera(test);
+    test.galera->start_replication(); // Causes restart.
+
+    Mariadb_nodes& gc = *test.galera;
+    gc.connect();
+
+    reset_galera(test);
+
+    string gtid = get_gtid_current_pos(test, gc.nodes[0]);
+
+    cout << "GTID: " << gtid << endl;
+
+    const char* zValue;
+
+    // Env-vars for debugging.
+    zValue = getenv("MXS2047_BLR_HOST");
+    const char* zMaxscale_host = (zValue ? zValue : test.maxscales->IP[0]);
+    cout << "MaxScale host: " << zMaxscale_host << endl;
+
+    zValue = getenv("MXS2047_BLR_PORT");
+    int maxscale_port = (zValue ? atoi(zValue) : test.maxscales->binlog_port[0]);
+    cout << "MaxScale port: " << maxscale_port << endl;
+
+    map<int, string> server_ids_by_index;
+
+    if (setup_server_ids(test, &server_ids_by_index))
+    {
+        if (setup_blr(test, gtid, zMaxscale_host, maxscale_port))
+        {
+            int slave_index = test.repl->N - 1; // We use the last slave.
+
+            Mariadb_nodes& ms = *test.repl;
+            ms.connect(slave_index);
+
+            MYSQL* pSlave = ms.nodes[slave_index];
+            mxb_assert(pSlave);
+
+            if (setup_slave(test, gtid, pSlave, zMaxscale_host, maxscale_port))
+            {
+                if (setup_schema(test, gc.nodes[0]))
+                {
+                    sleep(REPLICATION_SLEEP);
+
+                    if (insert_select(test, pSlave))
+                    {
+                        restart_slave(test, pSlave);
+                    }
+                }
+            }
+        }
+    }
+
+    // Since setting the server ids can fail half-way, we run this irrespective
+    // of what setup_server_ids() returns.
+    restore_server_ids(test, server_ids_by_index);
+
+    restore_galera(test);
+
+    return test.global_result;
+}