MXS-1632: Add per-server read-write statistics

The read-write distribution in readwritesplit is now stored in a map
partitioned by the servers that the router has used. Currently, the
statistics for removed servers aren't dropped so some filtering still
needs to be added.

include/maxscale/routingworker.hh

@@ -490,6 +490,12 @@ public:
         return get_local_value();
     }
 
+    // Dereference operator
+    T& operator*()
+    {
+        return *get_local_value();
+    }
+
     /**
      * Assign a value
      *

server/modules/routing/readwritesplit/readwritesplit.cc

@@ -107,6 +107,27 @@ const Stats& RWSplit::stats() const
 {
     return m_stats;
 }
+
+ServerStats& RWSplit::server_stats(SERVER* server)
+{
+    return (*m_server_stats)[server];
+}
+
+RWSplit::SrvStatMap RWSplit::all_server_stats() const
+{
+    SrvStatMap stats;
+
+    for (const auto& a : m_server_stats.values())
+    {
+        for (const auto& b : a)
+        {
+            stats[b.first] += b.second;
+        }
+    }
+
+    return stats;
+}
+
 int RWSplit::max_slave_count() const
 {
     int router_nservers = m_service->n_dbref;
@@ -354,6 +375,22 @@ void RWSplit::diagnostics(DCB* dcb)
                        ref->server->stats.n_current_ops);
         }
     }
+
+    auto srv_stats = all_server_stats();
+
+    if (!srv_stats.empty())
+    {
+        dcb_printf(dcb, "    Server    Total    Read    Write\n");
+        for (const auto& s : srv_stats)
+        {
+            dcb_printf(dcb,
+                       "    %s %10lu %10lu %10lu\n",
+                       s.first->name,
+                       s.second.total,
+                       s.second.read,
+                       s.second.write);
+        }
+    }
 }
 
 json_t* RWSplit::diagnostics_json() const
@@ -377,6 +414,25 @@ json_t* RWSplit::diagnostics_json() const
         json_object_set_new(rval, "weightby", json_string(weightby));
     }
 
+    auto srv_stats = all_server_stats();
+
+    if (!srv_stats.empty())
+    {
+        json_t* arr = json_array();
+
+        for (const auto& a : srv_stats)
+        {
+            json_t* obj = json_object();
+            json_object_set_new(obj, "id", json_string(a.first->name));
+            json_object_set_new(obj, "total", json_integer(a.second.total));
+            json_object_set_new(obj, "read", json_integer(a.second.read));
+            json_object_set_new(obj, "write", json_integer(a.second.write));
+            json_array_append_new(arr, obj);
+        }
+
+        json_object_set_new(rval, "server_query_statistics", arr);
+    }
+
     return rval;
 }
 

server/modules/routing/readwritesplit/readwritesplit.hh

@@ -219,6 +219,21 @@ struct Stats
     uint64_t n_rw_trx = 0;          /**< Read-write transaction count */
 };
 
+// Statistics for one server
+struct ServerStats
+{
+    uint64_t total = 0;     // Sum of master + slave + all
+    uint64_t read = 0;      // Write queries
+    uint64_t write = 0;     // Read queries
+
+    void operator+=(const ServerStats& rhs)
+    {
+        total += rhs.total;
+        read += rhs.read;
+        write += rhs.write;
+    }
+};
+
 class RWSplitSession;
 
 /**
@@ -230,6 +245,9 @@ class RWSplit : public mxs::Router<RWSplit, RWSplitSession>
     RWSplit& operator=(const RWSplit&);
 
 public:
+
+    using SrvStatMap = std::map<SERVER*, ServerStats>;
+
     RWSplit(SERVICE* service, const Config& config);
     ~RWSplit();
 
@@ -237,14 +255,17 @@ public:
     const Config& config() const;
     Stats&        stats();
     const Stats&  stats() const;
-    int           max_slave_count() const;
+    ServerStats&  server_stats(SERVER* server);
-    bool          have_enough_servers() const;
+    SrvStatMap    all_server_stats() const;
-    bool          select_connect_backend_servers(MXS_SESSION*         session,
+
-                                                 mxs::SRWBackendList& backends,
+    int  max_slave_count() const;
-                                                 mxs::SRWBackend& current_master,
+    bool have_enough_servers() const;
-                                                 mxs::SessionCommandList* sescmd_list,
+    bool select_connect_backend_servers(MXS_SESSION* session,
-                                                 int* expected_responses,
+                                        mxs::SRWBackendList& backends,
-                                                 connection_type type);
+                                        mxs::SRWBackend& current_master,
+                                        mxs::SessionCommandList* sescmd_list,
+                                        int* expected_responses,
+                                        connection_type type);
     // API functions
 
     /**
@@ -311,9 +332,10 @@ private:
     // Called when worker local data needs to be updated
     static void update_config(void* data);
 
-    SERVICE*                   m_service;   /**< Service where the router belongs*/
+    SERVICE*                       m_service;   /**< Service where the router belongs*/
-    mxs::rworker_local<Config> m_config;
+    mxs::rworker_local<Config>     m_config;
-    Stats                      m_stats;
+    Stats                          m_stats;
+    mxs::rworker_local<SrvStatMap> m_server_stats;
 };
 
 static inline const char* select_criteria_to_str(select_criteria_t type)

server/modules/routing/readwritesplit/rwsplit_route_stmt.cc

@@ -439,6 +439,8 @@ bool RWSplitSession::route_session_write(GWBUF* querybuf, uint8_t command, uint3
             {
                 nsucc += 1;
                 atomic_add_uint64(&backend->server()->stats.packets, 1);
+                m_router->server_stats(backend->server()).total++;
+                m_router->server_stats(backend->server()).read++;
 
                 if (expecting_response)
                 {
@@ -804,6 +806,7 @@ SRWBackend RWSplitSession::handle_slave_is_target(uint8_t cmd, uint32_t stmt_id)
     if (target)
     {
         atomic_add_uint64(&m_router->stats().n_slave, 1);
+        m_router->server_stats(target->server()).read++;
         mxb_assert(target->in_use() || target->can_connect());
     }
     else
@@ -936,6 +939,7 @@ bool RWSplitSession::handle_master_is_target(SRWBackend* dest)
     if (target && target == m_current_master)
     {
         atomic_add_uint64(&m_router->stats().n_master, 1);
+        m_router->server_stats(target->server()).write++;
     }
     else
     {
@@ -1090,6 +1094,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);
+        m_router->server_stats(target->server()).total++;
 
         if (!m_qc.large_query())
         {