Merge branch '2.2' into develop

include/maxscale/protocol/mysql.h

@@ -532,6 +532,9 @@ mxs_auth_state_t gw_send_backend_auth(DCB *dcb);
 /** Sends a response for an AuthSwitchRequest to the default auth plugin */
 int send_mysql_native_password_response(DCB* dcb);
 
+/** Sends an AuthSwitchRequest packet with the default auth plugin to the DCB */
+bool send_auth_switch_request_packet(DCB* dcb);
+
 /** Write an OK packet to a DCB */
 int mxs_mysql_send_ok(DCB *dcb, int sequence, uint8_t affected_rows, const char* message);
 

include/maxscale/server.h

@@ -63,6 +63,7 @@ typedef struct
     int n_persistent;     /**< Current persistent pool */
     uint64_t n_new_conn;  /**< Times the current pool was empty */
     uint64_t n_from_pool; /**< Times when a connection was available from the pool */
+    uint64_t packets;     /**< Number of packets routed to this server */
 } SERVER_STATS;
 
 /**
@@ -321,6 +322,15 @@ void server_add_parameter(SERVER *server, const char *name, const char *value);
  */
 bool server_remove_parameter(SERVER *server, const char *name);
 
+/**
+ * @brief Update server parameter
+ *
+ * @param server Server to update
+ * @param name   Parameter to update
+ * @param value  New value of parameter
+ */
+void server_update_parameter(SERVER *server, const char *name, const char *value);
+
 /**
  * @brief Check if a server points to a local MaxScale service
  *
@@ -358,7 +368,7 @@ extern void server_set_status_nolock(SERVER *server, uint64_t bit);
 extern void server_clear_status_nolock(SERVER *server, uint64_t bit);
 extern void server_transfer_status(SERVER *dest_server, const SERVER *source_server);
 extern void server_add_mon_user(SERVER *server, const char *user, const char *passwd);
-extern const char *server_get_parameter(const SERVER *server, const char *name);
+extern bool server_get_parameter(const SERVER *server, const char *name, char* out, size_t size);
 extern void server_update_credentials(SERVER *server, const char *user, const char *passwd);
 extern DCB* server_get_persistent(SERVER *server, const char *user, const char* ip, const char *protocol, int id);
 extern void server_update_address(SERVER *server, const char *address);

maxctrl/lib/api.js

@@ -17,6 +17,13 @@ exports.desc = 'Raw REST API access'
 exports.handler = function() {}
 exports.builder = function(yargs) {
     yargs
+        .group(['sum'], 'API options:')
+        .option('sum', {
+            describe: 'Calculate sum of API result. Only works for arrays of numbers ' +
+                'e.g. `api get --sum servers data[].attributes.statistics.connections`.',
+            type: 'boolean',
+            default: false
+        })
         .command('get <resource> [path]', 'Get raw JSON', function(yargs) {
             return yargs.epilog('Perform a raw REST API call. ' +
                                 'The path definition uses JavaScript syntax to extract values. ' +
@@ -27,7 +34,11 @@ exports.builder = function(yargs) {
             maxctrl(argv, function(host) {
                 return doRequest(host, argv.resource, (res) => {
                     if (argv.path) {
-                        res = _.getPath(res, argv.path, '')
+                        res = _.getPath(res, argv.path)
+                    }
+
+                    if (argv.sum && Array.isArray(res) && typeof(res[0]) == 'number') {
+                        res = res.reduce((sum, value) => value ? sum + value : sum)
                     }
 
                     return JSON.stringify(res)

server/core/config_runtime.cc

@@ -385,14 +385,14 @@ bool runtime_alter_server(SERVER *server, const char *key, const char *value)
     }
     else
     {
-        if (!server_remove_parameter(server, key) && !value[0])
+        if (!value[0] && !server_remove_parameter(server, key))
         {
             // Not a valid parameter
         }
         else if (value[0])
         {
             valid = true;
-            server_add_parameter(server, key, value);
+            server_update_parameter(server, key, value);
 
             /**
              * It's likely that this parameter is used as a weighting parameter.

server/core/server.cc

@@ -566,6 +566,7 @@ dprintServer(DCB *dcb, const SERVER *server)
     dcb_printf(dcb, "\tNumber of connections:               %d\n", server->stats.n_connections);
     dcb_printf(dcb, "\tCurrent no. of conns:                %d\n", server->stats.n_current);
     dcb_printf(dcb, "\tCurrent no. of operations:           %d\n", server->stats.n_current_ops);
+    dcb_printf(dcb, "\tNumber of routed packets:            %lu\n", server->stats.packets);
     if (server->persistpoolmax)
     {
         dcb_printf(dcb, "\tPersistent pool size:                %d\n", server->stats.n_persistent);
@@ -847,6 +848,27 @@ server_update_credentials(SERVER *server, const char *user, const char *passwd)
     }
 }
 
+static SERVER_PARAM* allocate_parameter(const char* name, const char* value)
+{
+    char *my_name = MXS_STRDUP(name);
+    char *my_value = MXS_STRDUP(value);
+
+    SERVER_PARAM *param = (SERVER_PARAM *)MXS_MALLOC(sizeof(SERVER_PARAM));
+
+    if (!my_name || !my_value || !param)
+    {
+        MXS_FREE(my_name);
+        MXS_FREE(my_value);
+        MXS_FREE(param);
+        return NULL;
+    }
+
+    param->active = true;
+    param->name = my_name;
+    param->value = my_value;
+
+    return param;
+}
 
 /**
  * Add a server parameter to a server.
@@ -860,27 +882,15 @@ server_update_credentials(SERVER *server, const char *user, const char *passwd)
  */
 void server_add_parameter(SERVER *server, const char *name, const char *value)
 {
-    char *my_name = MXS_STRDUP(name);
+    SERVER_PARAM* param = allocate_parameter(name, value);
-    char *my_value = MXS_STRDUP(value);
 
-    SERVER_PARAM *param = (SERVER_PARAM *)MXS_MALLOC(sizeof(SERVER_PARAM));
+    if (param)
-
-    if (!my_name || !my_value || !param)
     {
-        MXS_FREE(my_name);
+        spinlock_acquire(&server->lock);
-        MXS_FREE(my_value);
+        param->next = server->parameters;
-        MXS_FREE(param);
+        server->parameters = param;
-        return;
+        spinlock_release(&server->lock);
     }
-
-    param->active = true;
-    param->name = my_name;
-    param->value = my_value;
-
-    spinlock_acquire(&server->lock);
-    param->next = server->parameters;
-    server->parameters = param;
-    spinlock_release(&server->lock);
 }
 
 bool server_remove_parameter(SERVER *server, const char *name)
@@ -902,6 +912,32 @@ bool server_remove_parameter(SERVER *server, const char *name)
     return rval;
 }
 
+void server_update_parameter(SERVER *server, const char *name, const char *value)
+{
+    SERVER_PARAM* param = allocate_parameter(name, value);
+
+    if (param)
+    {
+        spinlock_acquire(&server->lock);
+
+        // Insert new value
+        param->next = server->parameters;
+        server->parameters = param;
+
+        // Mark old value, if found, as inactive
+        for (SERVER_PARAM *p = server->parameters->next; p; p = p->next)
+        {
+            if (strcmp(p->name, name) == 0 && p->active)
+            {
+                p->active = false;
+                break;
+            }
+        }
+
+        spinlock_release(&server->lock);
+    }
+}
+
 /**
  * Free a list of server parameters
  * @param tofree Parameter list to free
@@ -923,24 +959,35 @@ static void server_parameter_free(SERVER_PARAM *tofree)
 /**
  * Retrieve a parameter value from a server
  *
- * @param server        The server we are looking for a parameter of
+ * @param server The server we are looking for a parameter of
- * @param name          The name of the parameter we require
+ * @param name   The name of the parameter we require
- * @return      The parameter value or NULL if not found
+ * @param out    Buffer where value is stored, use NULL to check if the parameter exists
+ * @param size   Size of @c out, ignored if @c out is NULL
+ *
+ * @return True if parameter was found
  */
-const char *
+bool server_get_parameter(const SERVER *server, const char *name, char* out, size_t size)
-server_get_parameter(const SERVER *server, const char *name)
 {
+    bool found = false;
     SERVER_PARAM *param = server->parameters;
+    spinlock_acquire(&server->lock);
 
     while (param)
     {
         if (strcmp(param->name, name) == 0 && param->active)
         {
-            return param->value;
+            if (out)
+            {
+                snprintf(out, size, "%s", param->value);
+            }
+            found = true;
+            break;
         }
         param = param->next;
     }
-    return NULL;
+
+    spinlock_release(&server->lock);
+    return found;
 }
 
 /**
@@ -1459,6 +1506,7 @@ static json_t* server_json_attributes(const SERVER* server)
     json_object_set_new(stats, "connections", json_integer(server->stats.n_current));
     json_object_set_new(stats, "total_connections", json_integer(server->stats.n_connections));
     json_object_set_new(stats, "active_operations", json_integer(server->stats.n_current_ops));
+    json_object_set_new(stats, "routed_packets", json_integer(server->stats.packets));
 
     json_object_set_new(attr, "statistics", stats);
 

server/core/service.cc

@@ -2247,6 +2247,7 @@ static void service_calculate_weights(SERVICE *service)
 
     if (*weightby && service->dbref)
     {
+        char buf[50]; // Enough to hold most numbers
         /** Service has a weighting parameter and at least one server */
         int total = 0;
 
@@ -2254,10 +2255,10 @@ static void service_calculate_weights(SERVICE *service)
         for (SERVER_REF *server = service->dbref; server; server = server->next)
         {
             server->weight = SERVICE_BASE_SERVER_WEIGHT;
-            const char *param = server_get_parameter(server->server, weightby);
+
-            if (param)
+            if (server_get_parameter(server->server, weightby, buf, sizeof(buf)))
             {
-                total += atoi(param);
+                total += atoi(buf);
             }
         }
 
@@ -2278,10 +2279,9 @@ static void service_calculate_weights(SERVICE *service)
             /** Calculate the relative weight of the servers */
             for (SERVER_REF *server = service->dbref; server; server = server->next)
             {
-                const char *param = server_get_parameter(server->server, weightby);
+                if (server_get_parameter(server->server, weightby, buf, sizeof(buf)))
-                if (param)
                 {
-                    int wght = atoi(param);
+                    int wght = atoi(buf);
                     int perc = (wght * SERVICE_BASE_SERVER_WEIGHT) / total;
 
                     if (perc == 0)

server/core/test/test_server.cc

@@ -63,12 +63,13 @@ test1()
     //ss_info_dassert(NULL != service, "New server with valid protocol and port must not be null");
     //ss_info_dassert(0 != service_isvalid(service), "Service must be valid after creation");
 
+    char buf[120];
     ss_dfprintf(stderr, "\t..done\nTest Parameter for Server.");
-    ss_info_dassert(NULL == server_get_parameter(server, (char*)"name"), "Parameter should be null when not set");
+    ss_info_dassert(!server_get_parameter(server, "name", buf, sizeof(buf)), "Parameter should be null when not set");
     server_add_parameter(server, "name", "value");
     mxs_log_flush_sync();
-    ss_info_dassert(0 == strcmp("value", server_get_parameter(server, (char*)"name")),
+    ss_dassert(server_get_parameter(server, "name", buf, sizeof(buf)));
-                    "Parameter should be returned correctly");
+    ss_info_dassert(strcmp("value", buf) == 0, "Parameter should be returned correctly");
     ss_dfprintf(stderr, "\t..done\nTesting Unique Name for Server.");
     ss_info_dassert(NULL == server_find_by_unique_name("non-existent"),
                     "Should not find non-existent unique name.");

server/modules/monitor/galeramon/galeramon.cc

@@ -693,7 +693,6 @@ static MXS_MONITORED_SERVER *get_candidate_master(MXS_MONITOR* mon)
     long min_id = -1;
     int minval = INT_MAX;
     int currval;
-    const char* value;
     /* set min_id to the lowest value of moitor_servers->server->node_id */
     while (moitor_servers)
     {
@@ -701,11 +700,11 @@ static MXS_MONITORED_SERVER *get_candidate_master(MXS_MONITOR* mon)
         {
 
             moitor_servers->server->depth = 0;
-
+            char buf[50]; // Enough to hold most numbers
-            if (handle->use_priority && (value = server_get_parameter(moitor_servers->server, "priority")) != NULL)
+            if (handle->use_priority && server_get_parameter(moitor_servers->server, "priority", buf, sizeof(buf)))
             {
                 /** The server has a priority  */
-                if ((currval = atoi(value)) > 0)
+                if ((currval = atoi(buf)) > 0)
                 {
                     /** The priority is valid */
                     if (currval < minval && currval > 0)
@@ -718,7 +717,8 @@ static MXS_MONITORED_SERVER *get_candidate_master(MXS_MONITOR* mon)
             else if (moitor_servers->server->node_id >= 0 &&
                      (!handle->use_priority || /** Server priority disabled*/
                       candidate_master == NULL || /** No candidate chosen */
-                      server_get_parameter(candidate_master->server, "priority") == NULL)) /** Candidate has no priority */
+                       /** Candidate has no priority */
+                      !server_get_parameter(moitor_servers->server, "priority", buf, sizeof(buf))))
             {
                 if (min_id < 0 || moitor_servers->server->node_id < min_id)
                 {
@@ -852,8 +852,9 @@ static void update_sst_donor_nodes(MXS_MONITOR *mon, int is_cluster)
              * If no server has "priority" set, then
              * the server list will be order by default method.
              */
+
             if (handle->use_priority &&
-                server_get_parameter(ptr->server, "priority"))
+                server_get_parameter(ptr->server, "priority", NULL, 0))
             {
                 ignore_priority = false;
             }
@@ -986,28 +987,29 @@ static int compare_node_priority (const void *a, const void *b)
 {
     const MXS_MONITORED_SERVER *s_a = *(MXS_MONITORED_SERVER * const *)a;
     const MXS_MONITORED_SERVER *s_b = *(MXS_MONITORED_SERVER * const *)b;
-
+    char pri_a[50];
-    const char *pri_a = server_get_parameter(s_a->server, "priority");
+    char pri_b[50];
-    const char *pri_b = server_get_parameter(s_b->server, "priority");
+    bool have_a = server_get_parameter(s_a->server, "priority", pri_a, sizeof(pri_a));
+    bool have_b = server_get_parameter(s_b->server, "priority", pri_b, sizeof(pri_b));
 
     /**
      * Check priority parameter:
      *
      * Return a - b in case of issues
      */
-    if (!pri_a && pri_b)
+    if (!have_a && have_b)
     {
         MXS_DEBUG("Server %s has no given priority. It will be at the beginning of the list",
                   s_a->server->name);
         return -(INT_MAX - 1);
     }
-    else if (pri_a && !pri_b)
+    else if (have_a && !have_b)
     {
         MXS_DEBUG("Server %s has no given priority. It will be at the beginning of the list",
                   s_b->server->name);
         return INT_MAX - 1;
     }
-    else if (!pri_a && !pri_b)
+    else if (!have_a && !have_b)
     {
         MXS_DEBUG("Servers %s and %s have no given priority. They be at the beginning of the list",
                   s_a->server->name,

server/modules/protocol/MySQL/mariadbclient/mysql_client.cc

@@ -1043,7 +1043,14 @@ gw_read_normal_data(DCB *dcb, GWBUF *read_buffer, int nbytes_read)
      * the smallest version number. */
     qc_set_server_version(service_get_version(session->service, SERVICE_VERSION_MIN));
 
-    spec_com_res_t res = process_special_commands(dcb, read_buffer, nbytes_read);
+    spec_com_res_t res = RES_CONTINUE;
+    MySQLProtocol* proto = static_cast<MySQLProtocol*>(dcb->protocol);
+
+    if (!proto->changing_user)
+    {
+        res = process_special_commands(dcb, read_buffer, nbytes_read);
+    }
+
     int rval = 1;
     switch (res)
     {
@@ -1505,6 +1512,37 @@ void update_current_command(DCB* dcb, GWBUF* buffer)
     check_pool_candidate(dcb);
 }
 
+/**
+ * Perform re-authentication of the client
+ *
+ * @param session   Client session
+ * @param packetbuf Client's response to the AuthSwitchRequest
+ *
+ * @return True if the user is allowed access
+ */
+static bool reauthenticate_client(MXS_SESSION* session, GWBUF* packetbuf)
+{
+    bool rval = false;
+
+    if (session->client_dcb->authfunc.reauthenticate)
+    {
+        MySQLProtocol* proto = (MySQLProtocol*)session->client_dcb->protocol;
+        MYSQL_session* data = (MYSQL_session*)session->client_dcb->data;
+        uint8_t client_sha1[MYSQL_SCRAMBLE_LEN] = {};
+        uint8_t payload[gwbuf_length(packetbuf) - MYSQL_HEADER_LEN];
+        gwbuf_copy_data(packetbuf, MYSQL_HEADER_LEN, sizeof(payload), payload);
+
+        int rc = session->client_dcb->authfunc.reauthenticate(session->client_dcb, data->user,
+                                                              payload, sizeof(payload),
+                                                              proto->scramble, sizeof(proto->scramble),
+                                                              client_sha1, sizeof(client_sha1));
+
+        rval = rc == MXS_AUTH_SUCCEEDED;
+    }
+
+    return rval;
+}
+
 /**
  * Detect if buffer includes partial mysql packet or multiple packets.
  * Store partial packet to dcb_readqueue. Send complete packets one by one
@@ -1536,11 +1574,15 @@ static int route_by_statement(MXS_SESSION* session, uint64_t capabilities, GWBUF
         if (packetbuf != NULL)
         {
             CHK_GWBUF(packetbuf);
+            MySQLProtocol* proto = (MySQLProtocol*)session->client_dcb->protocol;
 
             /**
              * Update the currently command being executed.
              */
-            update_current_command(session->client_dcb, packetbuf);
+            if (!proto->changing_user)
+            {
+                update_current_command(session->client_dcb, packetbuf);
+            }
 
             if (rcap_type_required(capabilities, RCAP_TYPE_CONTIGUOUS_INPUT))
             {
@@ -1615,8 +1657,26 @@ static int route_by_statement(MXS_SESSION* session, uint64_t capabilities, GWBUF
                 }
             }
 
-            /** Route query */
+            bool changed_user = false;
-            rc = MXS_SESSION_ROUTE_QUERY(session, packetbuf);
+
+            if (!proto->changing_user && proto->current_command == MXS_COM_CHANGE_USER)
+            {
+                changed_user = true;
+                send_auth_switch_request_packet(session->client_dcb);
+            }
+
+            if (proto->changing_user)
+            {
+                rc = reauthenticate_client(session, packetbuf) ? 1 : 0;
+                gwbuf_free(packetbuf);
+            }
+            else
+            {
+                /** Route query */
+                rc = MXS_SESSION_ROUTE_QUERY(session, packetbuf);
+            }
+
+            proto->changing_user = changed_user;
         }
         else
         {

server/modules/protocol/MySQL/mysql_common.cc

@@ -316,7 +316,7 @@ mysql_create_standard_error(int packet_number,
     gw_mysql_set_byte3(mysql_packet_header, mysql_payload_size);
 
     // write packet number, now is 0
-    mysql_packet_header[3] = 0;
+    mysql_packet_header[3] = packet_number;
     memcpy(outbuf, mysql_packet_header, sizeof(mysql_packet_header));
 
     // current buffer pointer
@@ -1410,6 +1410,23 @@ int send_mysql_native_password_response(DCB* dcb)
     return dcb_write(dcb, buffer);
 }
 
+bool send_auth_switch_request_packet(DCB* dcb)
+{
+    MySQLProtocol* proto = (MySQLProtocol*) dcb->protocol;
+    const char plugin[] = DEFAULT_MYSQL_AUTH_PLUGIN;
+    uint32_t len = 1 + sizeof(plugin) + GW_MYSQL_SCRAMBLE_SIZE;
+    GWBUF* buffer = gwbuf_alloc(MYSQL_HEADER_LEN + len);
+
+    uint8_t* data = GWBUF_DATA(buffer);
+    gw_mysql_set_byte3(data, len);
+    data[3] = 1; // First response to the COM_CHANGE_USER
+    data[MYSQL_HEADER_LEN] = MYSQL_REPLY_AUTHSWITCHREQUEST;
+    memcpy(data + MYSQL_HEADER_LEN + 1, plugin, sizeof(plugin));
+    memcpy(data + MYSQL_HEADER_LEN + 1 + sizeof(plugin), proto->scramble, GW_MYSQL_SCRAMBLE_SIZE);
+
+    return dcb_write(dcb, buffer) != 0;
+}
+
 /**
  * Decode mysql server handshake
  *

server/modules/routing/readconnroute/readconnroute.c

@@ -589,6 +589,9 @@ routeQuery(MXS_ROUTER *instance, MXS_ROUTER_SESSION *router_session, GWBUF *queu
 
     inst->stats.n_queries++;
 
+    // Due to the streaming nature of readconnroute, this is not accurate
+    atomic_add_uint64(&router_cli_ses->backend->server->stats.packets, 1);
+
     /** Dirty read for quick check if router is closed. */
     if (router_cli_ses->rses_closed)
     {

server/modules/routing/readwritesplit/rwsplit_route_stmt.cc

@@ -377,6 +377,7 @@ bool RWSplitSession::route_session_write(GWBUF *querybuf, uint8_t command, uint3
             if (backend->execute_session_command())
             {
                 nsucc += 1;
+                atomic_add_uint64(&backend->server()->stats.packets, 1);
 
                 if (expecting_response)
                 {
@@ -973,6 +974,7 @@ bool RWSplitSession::handle_got_target(GWBUF* querybuf, SRWBackend& target, bool
         }
 
         atomic_add_uint64(&m_router->stats().n_queries, 1);
+        atomic_add_uint64(&target->server()->stats.packets, 1);
 
         if (!large_query && response == mxs::Backend::EXPECT_RESPONSE)
         {

server/modules/routing/schemarouter/schemaroutersession.cc

@@ -443,6 +443,7 @@ int32_t SchemaRouterSession::routeQuery(GWBUF* pPacket)
         {
             /** Add one query response waiter to backend reference */
             atomic_add(&m_router->m_stats.n_queries, 1);
+            atomic_add_uint64(&bref->server()->stats.packets, 1);
             ret = 1;
         }
         else
@@ -743,6 +744,7 @@ bool SchemaRouterSession::route_session_write(GWBUF* querybuf, uint8_t command)
                 if ((*it)->execute_session_command())
                 {
                     succp = true;
+                    atomic_add_uint64(&(*it)->server()->stats.packets, 1);
                 }
                 else
                 {