add new replica-side tracing info - dma_counter and dma_size

Author: Karolina Drabik

message/messaging_service.cc

@@ -1061,14 +1061,14 @@ future<> messaging_service::send_mutation_failed(msg_addr id, unsigned shard, re
     return send_message_oneway(this, messaging_verb::MUTATION_FAILED, std::move(id), shard, std::move(response_id), num_failed, std::move(backlog));
 }
 
-void messaging_service::register_read_data(std::function<future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>> (const rpc::client_info&, rpc::opt_time_point t, query::read_command cmd, ::compat::wrapping_partition_range pr, rpc::optional<query::digest_algorithm> oda)>&& func) {
+void messaging_service::register_read_data(std::function<future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>> (const rpc::client_info&, rpc::opt_time_point t, query::read_command cmd, ::compat::wrapping_partition_range pr, rpc::optional<query::digest_algorithm> oda)>&& func) {
     register_handler(this, netw::messaging_verb::READ_DATA, std::move(func));
 }
 future<> messaging_service::unregister_read_data() {
     return unregister_handler(netw::messaging_verb::READ_DATA);
 }
-future<rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t>> messaging_service::send_read_data(msg_addr id, clock_type::time_point timeout, const query::read_command& cmd, const dht::partition_range& pr, query::digest_algorithm da) {
-    return send_message_timeout<future<rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t>>>(this, messaging_verb::READ_DATA, std::move(id), timeout, cmd, pr, da);
+future<rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>> messaging_service::send_read_data(msg_addr id, clock_type::time_point timeout, const query::read_command& cmd, const dht::partition_range& pr, query::digest_algorithm da) {
+    return send_message_timeout<future<rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>>>(this, messaging_verb::READ_DATA, std::move(id), timeout, cmd, pr, da);
 }
 
 void messaging_service::register_get_schema_version(std::function<future<frozen_schema>(unsigned, table_schema_version)>&& func) {

message/messaging_service.hh

@@ -482,9 +482,9 @@ public:
 
     // Wrapper for READ_DATA
     // Note: WTH is future<foreign_ptr<lw_shared_ptr<query::result>>
-    void register_read_data(std::function<future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>> (const rpc::client_info&, rpc::opt_time_point timeout, query::read_command cmd, ::compat::wrapping_partition_range pr, rpc::optional<query::digest_algorithm> digest)>&& func);
+    void register_read_data(std::function<future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>> (const rpc::client_info&, rpc::opt_time_point timeout, query::read_command cmd, ::compat::wrapping_partition_range pr, rpc::optional<query::digest_algorithm> digest)>&& func);
     future<> unregister_read_data();
-    future<rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t>> send_read_data(msg_addr id, clock_type::time_point timeout, const query::read_command& cmd, const dht::partition_range& pr, query::digest_algorithm da);
+    future<rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>> send_read_data(msg_addr id, clock_type::time_point timeout, const query::read_command& cmd, const dht::partition_range& pr, query::digest_algorithm da);
 
     // Wrapper for GET_SCHEMA_VERSION
     void register_get_schema_version(std::function<future<frozen_schema>(unsigned, table_schema_version)>&& func);

service/storage_proxy.cc

@@ -3549,7 +3549,7 @@ class abstract_read_executor : public enable_shared_from_this<abstract_read_exec
             });
         }
     }
-    future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>> make_data_request(gms::inet_address ep, clock_type::time_point timeout, bool want_digest) {
+    future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>> make_data_request(gms::inet_address ep, clock_type::time_point timeout, bool want_digest) {
         ++_proxy->get_stats().data_read_attempts.get_ep_stat(ep);
         auto opts = want_digest
                   ? query::result_options{query::result_request::result_and_digest, digest_algorithm(*_proxy)}
@@ -3559,11 +3559,11 @@ class abstract_read_executor : public enable_shared_from_this<abstract_read_exec
             return _proxy->query_result_local(_schema, _cmd, _partition_range, opts, _trace_state, timeout);
         } else {
             tracing::trace(_trace_state, "read_data: sending a message to /{}", ep);
-            return _proxy->_messaging.send_read_data(netw::messaging_service::msg_addr{ep, 0}, timeout, *_cmd, _partition_range, opts.digest_algo).then([this, ep](rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t> response) {
-                auto&& [result, hit_rate, cache_counter] = response;
+            return _proxy->_messaging.send_read_data(netw::messaging_service::msg_addr{ep, 0}, timeout, *_cmd, _partition_range, opts.digest_algo).then([this, ep](rpc::tuple<query::result, rpc::optional<cache_temperature>, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t> response) {
+                auto&& [result, hit_rate, cache_counter, dma_counter, dma_size] = response;
                 tracing::trace(_trace_state, "read_data: got response from /{}", ep);
                 tracing::modify_cache_counter(_trace_state, cache_counter);
-                return make_ready_future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>>(rpc::tuple(make_foreign(::make_lw_shared<query::result>(std::move(result))), hit_rate.value_or(cache_temperature::invalid()), cache_counter));
+                return make_ready_future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>>(rpc::tuple(make_foreign(::make_lw_shared<query::result>(std::move(result))), hit_rate.value_or(cache_temperature::invalid()), cache_counter, dma_counter, dma_size));
             });
         }
     }
@@ -3606,7 +3606,7 @@ class abstract_read_executor : public enable_shared_from_this<abstract_read_exec
         auto start = latency_clock::now();
         for (const gms::inet_address& ep : boost::make_iterator_range(begin, end)) {
             // Waited on indirectly, shared_from_this keeps `this` alive
-            (void)make_data_request(ep, timeout, want_digest).then_wrapped([this, resolver, ep, start, exec = shared_from_this()] (future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>> f) {
+            (void)make_data_request(ep, timeout, want_digest).then_wrapped([this, resolver, ep, start, exec = shared_from_this()] (future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>> f) {
                 try {
                     auto v = f.get0();
                     _cf->set_hit_rate(ep, std::get<1>(v));
@@ -4007,13 +4007,13 @@ ::shared_ptr<abstract_read_executor> storage_proxy::get_read_executor(lw_shared_
 
 future<rpc::tuple<query::result_digest, api::timestamp_type, cache_temperature>>
 storage_proxy::query_result_local_digest(schema_ptr s, lw_shared_ptr<query::read_command> cmd, const dht::partition_range& pr, tracing::trace_state_ptr trace_state, storage_proxy::clock_type::time_point timeout, query::digest_algorithm da) {
-    return query_result_local(std::move(s), std::move(cmd), pr, query::result_options::only_digest(da), std::move(trace_state), timeout).then([] (rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t> result_and_hit_rate) {
-        auto&& [result, hit_rate, cache_counter] = result_and_hit_rate;
+    return query_result_local(std::move(s), std::move(cmd), pr, query::result_options::only_digest(da), std::move(trace_state), timeout).then([] (rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t> result_and_hit_rate) {
+        auto&& [result, hit_rate, cache_counter, dma_counter, dma_size] = result_and_hit_rate;
         return make_ready_future<rpc::tuple<query::result_digest, api::timestamp_type, cache_temperature>>(rpc::tuple(*result->digest(), result->last_modified(), hit_rate));
     });
 }
 
-future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>>
+future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>>
 storage_proxy::query_result_local(schema_ptr s, lw_shared_ptr<query::read_command> cmd, const dht::partition_range& pr, query::result_options opts,
                                   tracing::trace_state_ptr trace_state, storage_proxy::clock_type::time_point timeout) {
     cmd->slice.options.set_if<query::partition_slice::option::with_digest>(opts.request != query::result_request::only_result);
@@ -4026,7 +4026,7 @@ storage_proxy::query_result_local(schema_ptr s, lw_shared_ptr<query::read_comman
             return db.query(gs, *cmd, opts, prv, trace_state, timeout).then([trace_state](std::tuple<lw_shared_ptr<query::result>, cache_temperature>&& f_ht) {
                 auto&& [f, ht] = f_ht;
                 tracing::trace(trace_state, "Querying is done");
-                return make_ready_future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>>(rpc::tuple(make_foreign(std::move(f)), ht, get_cache_counter(trace_state)));
+                return make_ready_future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>>(rpc::tuple(make_foreign(std::move(f)), ht, get_cache_counter(trace_state), get_dma_counter(trace_state), get_dma_size(trace_state)));
             });
         });
     } else {
@@ -4036,7 +4036,7 @@ storage_proxy::query_result_local(schema_ptr s, lw_shared_ptr<query::read_comman
                 [trace_state = std::move(trace_state)] (rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature>&& r_ht) {
             auto&& [r, ht] = r_ht;
             tracing::trace(trace_state, "Querying is done");
-            return make_ready_future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>>(rpc::tuple(std::move(r), ht, get_cache_counter(trace_state)));
+            return make_ready_future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>>(rpc::tuple(std::move(r), ht, get_cache_counter(trace_state), get_dma_counter(trace_state), get_dma_size(trace_state)));
         });
     }
 }

service/storage_proxy.hh

@@ -377,7 +377,7 @@ private:
             const inet_address_vector_replica_set& preferred_endpoints,
             bool& is_bounced_read,
             service_permit permit);
-    future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t>> query_result_local(schema_ptr, lw_shared_ptr<query::read_command> cmd, const dht::partition_range& pr,
+    future<rpc::tuple<foreign_ptr<lw_shared_ptr<query::result>>, cache_temperature, tracing::trace_state_ptr::cache_counter_t, tracing::trace_state_ptr::dma_counter_t, tracing::trace_state_ptr::dma_size_t>> query_result_local(schema_ptr, lw_shared_ptr<query::read_command> cmd, const dht::partition_range& pr,
                                                                            query::result_options opts,
                                                                            tracing::trace_state_ptr trace_state,
                                                                            clock_type::time_point timeout);

sstables/index_reader.hh

@@ -316,7 +316,7 @@ inline file make_tracked_index_file(sstable& sst, reader_permit permit, tracing:
                                     use_caching caching) {
     auto f = caching ? sst.index_file() : sst.uncached_index_file();
     f = make_tracked_file(std::move(f), std::move(permit));
-    if (!trace_state.has_tracing()) {
+    if (!trace_state.has_tracing() && !trace_state.has_opentelemetry()) {
         return f;
     }
     return tracing::make_traced_file(std::move(f), std::move(trace_state), format("{}:", sst.filename(component_type::Index)));

sstables/sstables.cc

@@ -2286,7 +2286,7 @@ input_stream<char> sstable::data_stream(uint64_t pos, size_t len, const io_prior
     options.dynamic_adjustments = std::move(history);
 
     file f = make_tracked_file(_data_file, std::move(permit));
-    if (trace_state.has_tracing()) {
+    if (trace_state.has_tracing() || trace_state.has_opentelemetry()) {
         f = tracing::make_traced_file(std::move(f), std::move(trace_state), format("{}:", get_filename()));
     }
 

tracing/trace_state.cc

@@ -365,4 +365,16 @@ void opentelemetry_state_data::serialize_cache_counter(bytes& serialized) const
     const auto *counter_ptr = reinterpret_cast<const int8_t*>(&counter);
     serialized += bytes{counter_ptr, sizeof(counter)};
 }
+
+void opentelemetry_state_data::serialize_dma_counter(bytes& serialized) const {
+    const auto counter = htonl(_dma_counter);
+    const auto *counter_ptr = reinterpret_cast<const int8_t*>(&counter);
+    serialized += bytes{counter_ptr, sizeof(counter)};
+}
+
+void opentelemetry_state_data::serialize_dma_size(bytes& serialized) const {
+    const auto size = htonl(_dma_size);
+    const auto *size_ptr = reinterpret_cast<const int8_t*>(&size);
+    serialized += bytes{size_ptr, sizeof(size)};
+}
 }

tracing/trace_state.hh

@@ -510,15 +510,23 @@ private:
 class opentelemetry_state_data final {
 public:
     using cache_counter_t = int32_t;
+    using dma_counter_t = int32_t;
+    using dma_size_t = int32_t;
 
     inet_address_vector_replica_set _replicas;
     sstring _statement_type;
     // Number of read partitions that were found in cache.
     cache_counter_t _cache_counter{0};
+    // Number of DMA reads that were executed.
+    dma_counter_t _dma_counter{0};
+    // Number of bytes read in DMA reads.
+    dma_size_t _dma_size{0};
 
     void serialize_replicas(bytes& serialized) const;
     void serialize_statement_type(bytes& serialized) const;
     void serialize_cache_counter(bytes& serialized) const;
+    void serialize_dma_counter(bytes& serialized) const;
+    void serialize_dma_size(bytes& serialized) const;
 };
 
 
@@ -538,6 +546,8 @@ private:
                                    to_reduce._replicas.begin(),
                                    to_reduce._replicas.end());
             _data._cache_counter += to_reduce._cache_counter;
+            _data._dma_counter += to_reduce._dma_counter;
+            _data._dma_size += to_reduce._dma_size;
             _data._statement_type += to_reduce._statement_type;
         }
 
@@ -594,6 +604,8 @@ public:
 
         _data->local().serialize_replicas(serialized);
         _data->local().serialize_cache_counter(serialized);
+        _data->local().serialize_dma_counter(serialized);
+        _data->local().serialize_dma_size(serialized);
         _data->local().serialize_statement_type(serialized);
 
         return serialized;
@@ -628,13 +640,49 @@ public:
         }
     }
 
+    /**
+     * Increment counter of DMA reads.
+     *
+     * @param count number of reads
+     */
+    void modify_dma_counter(opentelemetry_state_data::dma_counter_t count) {
+        if (_data->local_is_initialized()) {
+            _data->local()._dma_counter += count;
+        }
+    }
+
+    /**
+     * Increment number of bytes read in DMA reads.
+     *
+     * @param size number of bytes.
+     */
+    void modify_dma_size(opentelemetry_state_data::dma_size_t size) {
+        if (_data->local_is_initialized()) {
+            _data->local()._dma_size += size;
+        }
+    }
+
     /**
      * @return number of partitions that were found in cache
      */
     opentelemetry_state_data::cache_counter_t get_cache_counter() const {
         return _data->local()._cache_counter;
     }
 
+    /**
+     * @return number of DMA reads that were executed
+    */
+    opentelemetry_state_data::dma_counter_t get_dma_counter() const {
+        return _data->local()._dma_counter;
+    }
+
+    /**
+     * @return number of bytes read in DMA reads
+    */
+    opentelemetry_state_data::dma_size_t get_dma_size() const {
+        return _data->local()._dma_size;
+    }
+
     /**
      * @return True if OpenTelemetry trace state is stored.
      */
@@ -696,6 +744,8 @@ public:
     {}
 
     using cache_counter_t = opentelemetry_state_data::cache_counter_t;
+    using dma_counter_t = opentelemetry_state_data::dma_counter_t;
+    using dma_size_t = opentelemetry_state_data::dma_size_t;
 
     /**
      * @return True if classic trace state is stored.
@@ -983,6 +1033,18 @@ inline void modify_cache_counter(const trace_state_ptr& p, trace_state_ptr::cach
     }
 }
 
+inline void modify_dma_counter(const trace_state_ptr& p, trace_state_ptr::dma_counter_t count) {
+    if (p.has_opentelemetry()) {
+        p.get_opentelemetry_ptr()->modify_dma_counter(count);
+    }
+}
+
+inline void modify_dma_size(const trace_state_ptr& p, trace_state_ptr::dma_size_t size) {
+    if (p.has_opentelemetry()) {
+        p.get_opentelemetry_ptr()->modify_dma_size(size);
+    }
+}
+
 inline trace_state_ptr::cache_counter_t get_cache_counter(const trace_state_ptr& p) {
     if (p.has_opentelemetry()) {
         return p.get_opentelemetry_ptr()->get_cache_counter();
@@ -991,6 +1053,22 @@ inline trace_state_ptr::cache_counter_t get_cache_counter(const trace_state_ptr&
     return 0;
 }
 
+inline trace_state_ptr::dma_counter_t get_dma_counter(const trace_state_ptr& p) {
+    if (p.has_opentelemetry()) {
+        return p.get_opentelemetry_ptr()->get_dma_counter();
+    }
+
+    return 0;
+}
+
+inline trace_state_ptr::dma_size_t get_dma_size(const trace_state_ptr& p) {
+    if (p.has_opentelemetry()) {
+        return p.get_opentelemetry_ptr()->get_dma_size();
+    }
+
+    return 0;
+}
+
 inline future<> start(const trace_state_ptr& p) {
     if (p.has_opentelemetry()) {
         return p.get_opentelemetry_ptr()->start();