streaming: Fix stream_manager progress api

For each stream_session, we pretend we are sending/receiving one file,
to make it compatible with nodetool. For receiving_files, the file name
is "rxnofile". For sending_files, the file name is "txnofile".

stream_manager::update_all_progress_info is introduced to update the
progress info of all the stream_sessions in the node. We need this
because streaming mutations are received on all the cores, but the
stream_session object is only on one of the cores. It adds overhead if
we update progress info in stream_session object whenever we receive a
streaming mutation. So, what we do now is when we really need the
progress info, we update the progress info in stream_session object.

With http://127.0.0.$i:10000/stream_manager/, it looks like below when
decommission node 3 in a 3 nodes cluster.

=========== GET NODE 1
[{"plan_id": "935a2cc0-dc6b-11e5-bdbf-000000000000", "description":
"Unbootstrap", "sessions": [{"receiving_files": [{"value": {"direction":
"IN", "file_name": "rxnofile", "session_index": 0, "total_bytes":
16876296, "peer": "127.0.0.3", "current_bytes": 16876296}, "key":
"rxnofile"}], "receiving_summaries": [{"files": 1, "total_size": 0,
"cf_id": "869d8630-dc6b-11e5-bdbf-000000000000"}], "session_index": 0,
"state": "PREPARING", "connecting": "127.0.0.3", "peer": "127.0.0.3"}]}]

=========== GET NODE 2

[{"plan_id": "935a2cc0-dc6b-11e5-bdbf-000000000000", "description":
"Unbootstrap", "sessions": [{"receiving_files": [{"value": {"direction":
"IN", "file_name": "rxnofile", "session_index": 0, "total_bytes":
16755552, "peer": "127.0.0.3", "current_bytes": 16755552}, "key":
"rxnofile"}], "receiving_summaries": [{"files": 1, "total_size": 0,
"cf_id": "869d8630-dc6b-11e5-bdbf-000000000000"}], "session_index": 0,
"state": "PREPARING", "connecting": "127.0.0.3", "peer": "127.0.0.3"}]}]

=========== GET NODE 3
[{"plan_id": "935a2cc0-dc6b-11e5-bdbf-000000000000", "description":
"Unbootstrap", "sessions": [{"sending_files": [{"value": {"direction":
"OUT", "file_name": "txnofile", "session_index": 0, "total_bytes":
16876296, "peer": "127.0.0.1", "current_bytes": 16876296}, "key":
"txnofile"}], "sending_summaries": [{"files": 1, "total_size": 0,
"cf_id": "869d8630-dc6b-11e5-bdbf-000000000000"}], "session_index": 0,
"state": "PREPARING", "connecting": "127.0.0.1", "peer":
"127.0.0.1"},{"sending_files": [{"value": {"direction": "OUT",
"file_name": "txnofile", "session_index": 0, "total_bytes": 16755552,
"peer": "127.0.0.2", "current_bytes": 16755552}, "key": "txnofile"}],
"sending_summaries": [{"files": 1, "total_size": 0, "cf_id":
"869d8630-dc6b-11e5-bdbf-000000000000"}], "session_index": 0, "state":
"PREPARING", "connecting": "127.0.0.2", "peer": "127.0.0.2"}]}]

api/stream_manager.cc

@@ -32,11 +32,16 @@ namespace hs = httpd::stream_manager_json;
 
 static void set_summaries(const std::vector<streaming::stream_summary>& from,
         json::json_list<hs::stream_summary>& to) {
-    for (auto sum : from) {
+    if (!from.empty()) {
         hs::stream_summary res;
-        res.cf_id = boost::lexical_cast<std::string>(sum.cf_id);
-        res.files = sum.files;
-        res.total_size = sum.total_size;
+        res.cf_id = boost::lexical_cast<std::string>(from.front().cf_id);
+        // For each stream_session, we pretend we are sending/receiving one
+        // file, to make it compatible with nodetool.
+        res.files = 1;
+        // We can not estimate total number of bytes the stream_session will
+        // send or recvieve since we don't know the size of the frozen_mutation
+        // until we read it.
+        res.total_size = 0;
         to.push(res);
     }
 }
@@ -85,18 +90,22 @@ static hs::stream_state get_state(
 void set_stream_manager(http_context& ctx, routes& r) {
     hs::get_current_streams.set(r,
             [] (std::unique_ptr<request> req) {
-                return streaming::get_stream_manager().map_reduce0([](streaming::stream_manager& stream) {
-                    std::vector<hs::stream_state> res;
-                    for (auto i : stream.get_initiated_streams()) {
-                        res.push_back(get_state(*i.second.get()));
-                    }
-                    for (auto i : stream.get_receiving_streams()) {
-                        res.push_back(get_state(*i.second.get()));
-                    }
-                    return res;
-                }, std::vector<hs::stream_state>(),concat<hs::stream_state>).
-                then([](const std::vector<hs::stream_state>& res) {
-                    return make_ready_future<json::json_return_type>(res);
+                return streaming::get_stream_manager().invoke_on_all([] (auto& sm) {
+                    return sm.update_all_progress_info();
+                }).then([] {
+                    return streaming::get_stream_manager().map_reduce0([](streaming::stream_manager& stream) {
+                        std::vector<hs::stream_state> res;
+                        for (auto i : stream.get_initiated_streams()) {
+                            res.push_back(get_state(*i.second.get()));
+                        }
+                        for (auto i : stream.get_receiving_streams()) {
+                            res.push_back(get_state(*i.second.get()));
+                        }
+                        return res;
+                    }, std::vector<hs::stream_state>(),concat<hs::stream_state>).
+                    then([](const std::vector<hs::stream_state>& res) {
+                        return make_ready_future<json::json_return_type>(res);
+                    });
                 });
             });
 

streaming/stream_manager.cc

@@ -132,6 +132,16 @@ void stream_manager::update_progress(UUID cf_id, gms::inet_address peer, progres
     }
 }
 
+future<> stream_manager::update_all_progress_info() {
+    return seastar::async([this] {
+        for (auto sr: get_all_streams()) {
+            for (auto session : sr->get_coordinator()->get_all_stream_sessions()) {
+                session->update_progress().get();
+            }
+        }
+    });
+}
+
 void stream_manager::remove_progress(UUID plan_id) {
     _stream_bytes.erase(plan_id);
 }

streaming/stream_manager.hh

@@ -117,6 +117,7 @@ public:
     }
 
     void update_progress(UUID cf_id, gms::inet_address peer, progress_info::direction dir, size_t fm_size);
+    future<> update_all_progress_info();
 
     void remove_progress(UUID plan_id);
 

streaming/stream_session.cc

@@ -534,4 +534,19 @@ sstring stream_session::description() {
     return _stream_result  ? _stream_result->description : "";
 }
 
+future<> stream_session::update_progress() {
+    return get_local_stream_manager().get_progress_on_all_shards(plan_id(), peer).then([this] (auto sbytes) {
+        auto bytes_sent = sbytes.bytes_sent;
+        if (bytes_sent > 0) {
+            auto tx = progress_info(this->peer, "txnofile", progress_info::direction::OUT, bytes_sent, bytes_sent);
+            _session_info.update_progress(std::move(tx));
+        }
+        auto bytes_received = sbytes.bytes_received;
+        if (bytes_received > 0) {
+            auto rx = progress_info(this->peer, "rxnofile", progress_info::direction::IN, bytes_received, bytes_received);
+            _session_info.update_progress(std::move(rx));
+        }
+    });
+}
+
 } // namespace streaming

streaming/stream_session.hh

@@ -329,9 +329,7 @@ public:
         return _session_info;
     }
 
-    void update_progress(const progress_info& info) {
-        _session_info.update_progress(info);
-    }
+    future<> update_progress();
 
     void receive_task_completed(UUID cf_id);
     void transfer_task_completed(UUID cf_id);