Measure hardware IO for update operations (#5922)

* Measure update operations hardware IO

* Add support for distributed setups

* also measure update_local

* Add consensus tests for HW metrics of update operations

* add test for upserting without waiting

* Disable HW usage reporting when not waiting for update API

* Review remarks

* Fix resharding collecting hw measurements

* Fix metric type

* New struct HardwareData for better accumulation

* Ensure we always apply CPU multiplier

* Apply suggestions from code review

* Update src/actix/api/update_api.rs

Co-authored-by: Tim Visée <tim+github@visee.me>

* Fix assert_with_upper_bound_error threshold calculation.

* Clarifying why we don't measure shard cleanup

---------

Co-authored-by: Tim Visée <tim+github@visee.me>

docs/grpc/docs.md

@@ -3501,6 +3501,7 @@ Additionally, the first and last points of each GeoLineString must be the same.
 | ----- | ---- | ----- | ----------- |
 | result | [UpdateResult](#qdrant-UpdateResult) |  |  |
 | time | [double](#double) |  | Time spent to process |
+| usage | [HardwareUsage](#qdrant-HardwareUsage) | optional |  |
 
 
 

lib/api/src/grpc/conversions.rs

@@ -4,6 +4,7 @@ use std::time::Instant;
 
 use chrono::{NaiveDateTime, Timelike};
 use common::counter::hardware_accumulator::HwMeasurementAcc;
+use common::counter::hardware_data::{HardwareData, RealCpuMeasurement};
 use itertools::Itertools;
 use segment::common::operation_error::OperationError;
 use segment::data_types::index::{
@@ -2124,10 +2125,15 @@ pub fn into_named_vector_struct(
 
 impl From<PointsOperationResponseInternal> for PointsOperationResponse {
     fn from(resp: PointsOperationResponseInternal) -> Self {
-        let PointsOperationResponseInternal { result, time } = resp;
+        let PointsOperationResponseInternal {
+            result,
+            time,
+            usage,
+        } = resp;
         Self {
             result: result.map(Into::into),
             time,
+            usage,
         }
     }
 }
@@ -2135,10 +2141,15 @@ impl From<PointsOperationResponseInternal> for PointsOperationResponse {
 // TODO: Make it explicit `from_operations_response` method instead of `impl From<PointsOperationResponse>`?
 impl From<PointsOperationResponse> for PointsOperationResponseInternal {
     fn from(resp: PointsOperationResponse) -> Self {
-        let PointsOperationResponse { result, time } = resp;
+        let PointsOperationResponse {
+            result,
+            time,
+            usage,
+        } = resp;
         Self {
             result: result.map(Into::into),
             time,
+            usage,
         }
     }
 }
@@ -2603,6 +2614,26 @@ impl From<HwMeasurementAcc> for HardwareUsage {
     }
 }
 
+impl From<HardwareUsage> for HardwareData {
+    fn from(value: HardwareUsage) -> Self {
+        let HardwareUsage {
+            cpu,
+            payload_io_read,
+            payload_io_write,
+            vector_io_read,
+            vector_io_write,
+        } = value;
+
+        HardwareData {
+            cpu: RealCpuMeasurement::new(cpu as usize, 1), // Multiplier in API already applied.
+            payload_io_read: payload_io_read as usize,
+            payload_io_write: payload_io_write as usize,
+            vector_io_read: vector_io_read as usize,
+            vector_io_write: vector_io_write as usize,
+        }
+    }
+}
+
 impl Formula {
     /// This implementation is only used to forward a request to remote shards.
     ///

lib/api/src/grpc/proto/points.proto

@@ -810,6 +810,7 @@ message UpdateBatchPoints {
 message PointsOperationResponse {
   UpdateResult result = 1;
   double time = 2; // Time spent to process
+  optional HardwareUsage usage = 3;
 }
 
 message UpdateResult {

lib/api/src/grpc/proto/points_internal_service.proto

@@ -100,6 +100,7 @@ message DeleteFieldIndexCollectionInternal {
 message PointsOperationResponseInternal {
   UpdateResultInternal result = 1;
   double time = 2; // Time spent to process
+  optional HardwareUsage usage = 3;
 }
 
 // Has to be backward compatible with `UpdateResult`!
@@ -318,4 +319,5 @@ message FacetHitInternal {
 message FacetResponseInternal {
     repeated FacetHitInternal hits = 1;
     double time = 2; // Time spent to process
+    optional HardwareUsage usage = 3;
 }

lib/api/src/grpc/qdrant.rs

@@ -5795,6 +5795,8 @@ pub struct PointsOperationResponse {
     /// Time spent to process
     #[prost(double, tag = "2")]
     pub time: f64,
+    #[prost(message, optional, tag = "3")]
+    pub usage: ::core::option::Option<HardwareUsage>,
 }
 #[derive(serde::Serialize)]
 #[allow(clippy::derive_partial_eq_without_eq)]
@@ -9298,6 +9300,8 @@ pub struct PointsOperationResponseInternal {
     /// Time spent to process
     #[prost(double, tag = "2")]
     pub time: f64,
+    #[prost(message, optional, tag = "3")]
+    pub usage: ::core::option::Option<HardwareUsage>,
 }
 /// Has to be backward compatible with `UpdateResult`!
 #[derive(serde::Serialize)]
@@ -9786,6 +9790,8 @@ pub struct FacetResponseInternal {
     /// Time spent to process
     #[prost(double, tag = "2")]
     pub time: f64,
+    #[prost(message, optional, tag = "3")]
+    pub usage: ::core::option::Option<HardwareUsage>,
 }
 /// Generated client implementations.
 pub mod points_internal_client {

lib/collection/src/collection/clean.rs

@@ -257,6 +257,9 @@ async fn clean_task(
     shard_id: ShardId,
     sender: Sender<ShardCleanStatus>,
 ) -> Result<(), CollectionError> {
+    // Do not measure the hardware usage of these deletes as clean the shard is always considered an internal operation
+    // users should not be billed for.
+
     let mut offset = None;
     let mut deleted_points = 0;
 
@@ -332,7 +335,10 @@ async fn clean_task(
             OperationWithClockTag::from(CollectionUpdateOperations::PointOperation(
                 crate::operations::point_ops::PointOperations::DeletePoints { ids },
             ));
-        if let Err(err) = shard.update_local(delete_operation, last_batch).await {
+        if let Err(err) = shard
+            .update_local(delete_operation, last_batch, HwMeasurementAcc::disposable())
+            .await
+        {
             return Err(CollectionError::service_error(format!(
                 "Failed to delete points from shard: {err}",
             )));

lib/collection/src/collection/payload_index_schema.rs

@@ -1,6 +1,7 @@
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 
+use common::counter::hardware_accumulator::HwMeasurementAcc;
 use segment::json_path::JsonPath;
 use segment::problems::unindexed_field;
 use segment::types::{Filter, PayloadFieldSchema, PayloadKeyType};
@@ -66,7 +67,8 @@ impl Collection {
             }),
         );
 
-        self.update_all_local(create_index_operation, wait).await
+        self.update_all_local(create_index_operation, wait, HwMeasurementAcc::disposable()) // Unmeasured API
+            .await
     }
 
     pub async fn drop_payload_index(
@@ -81,7 +83,13 @@ impl Collection {
             FieldIndexOperations::DeleteIndex(field_name),
         );
 
-        let result = self.update_all_local(delete_index_operation, false).await?;
+        let result = self
+            .update_all_local(
+                delete_index_operation,
+                false,
+                HwMeasurementAcc::disposable(), // Unmeasured API
+            )
+            .await?;
 
         Ok(result)
     }

lib/collection/src/collection/point_ops.rs

@@ -28,6 +28,7 @@ impl Collection {
         &self,
         operation: CollectionUpdateOperations,
         wait: bool,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<Option<UpdateResult>> {
         let update_lock = self.updates_lock.clone().read_owned().await;
         let shard_holder = self.shards_holder.clone().read_owned().await;
@@ -49,7 +50,11 @@ impl Collection {
                     //
                     // We update *all* shards with a single operation, but each shard has it's own clock,
                     // so it's *impossible* to assign any single clock tag to this operation.
-                    shard.update_local(OperationWithClockTag::from(operation.clone()), wait)
+                    shard.update_local(
+                        OperationWithClockTag::from(operation.clone()),
+                        wait,
+                        hw_measurement_acc.clone(),
+                    )
                 })
                 .collect();
 
@@ -85,6 +90,7 @@ impl Collection {
         shard_selection: ShardId,
         wait: bool,
         ordering: WriteOrdering,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         let update_lock = self.updates_lock.clone().read_owned().await;
         let shard_holder = self.shards_holder.clone().read_owned().await;
@@ -97,7 +103,7 @@ impl Collection {
             };
 
             match ordering {
-                WriteOrdering::Weak => shard.update_local(operation, wait).await,
+                WriteOrdering::Weak => shard.update_local(operation, wait, hw_measurement_acc.clone()).await,
                 WriteOrdering::Medium | WriteOrdering::Strong => {
                     if let Some(clock_tag) = operation.clock_tag {
                         log::warn!(
@@ -108,7 +114,7 @@ impl Collection {
                     }
 
                     shard
-                        .update_with_consistency(operation.operation, wait, ordering, false)
+                        .update_with_consistency(operation.operation, wait, ordering, false, hw_measurement_acc)
                         .await
                         .map(Some)
                 }
@@ -136,6 +142,7 @@ impl Collection {
         wait: bool,
         ordering: WriteOrdering,
         shard_keys_selection: Option<ShardKey>,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         let update_lock = self.updates_lock.clone().read_owned().await;
         let shard_holder = self.shards_holder.clone().read_owned().await;
@@ -149,6 +156,7 @@ impl Collection {
             for (shard, operation) in operations {
                 let operation = shard_holder.split_by_mode(shard.shard_id, operation);
 
+                let hw_acc = hw_measurement_acc.clone();
                 updates.push(async move {
                     let mut result = UpdateResult {
                         operation_id: None,
@@ -158,13 +166,25 @@ impl Collection {
 
                     for operation in operation.update_all {
                         result = shard
-                            .update_with_consistency(operation, wait, ordering, false)
+                            .update_with_consistency(
+                                operation,
+                                wait,
+                                ordering,
+                                false,
+                                hw_acc.clone(),
+                            )
                             .await?;
                     }
 
                     for operation in operation.update_only_existing {
                         let res = shard
-                            .update_with_consistency(operation, wait, ordering, true)
+                            .update_with_consistency(
+                                operation,
+                                wait,
+                                ordering,
+                                true,
+                                hw_acc.clone(),
+                            )
                             .await;
 
                         if let Err(err) = &res {
@@ -228,8 +248,9 @@ impl Collection {
         operation: CollectionUpdateOperations,
         wait: bool,
         ordering: WriteOrdering,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
-        self.update_from_client(operation, wait, ordering, None)
+        self.update_from_client(operation, wait, ordering, None, hw_measurement_acc)
             .await
     }
 

lib/collection/src/collection/sharding_keys.rs

@@ -1,5 +1,6 @@
 use std::collections::HashSet;
 
+use common::counter::hardware_accumulator::HwMeasurementAcc;
 use segment::types::ShardKey;
 
 use crate::collection::Collection;
@@ -60,6 +61,8 @@ impl Collection {
         shard_key: ShardKey,
         placement: ShardsPlacement,
     ) -> Result<(), CollectionError> {
+        let hw_counter = HwMeasurementAcc::disposable(); // TODO(io_measurement): propagate value
+
         let state = self.state().await;
         match state.config.params.sharding_method.unwrap_or_default() {
             ShardingMethod::Auto => {
@@ -121,7 +124,11 @@ impl Collection {
                 );
 
                 replica_set
-                    .update_local(OperationWithClockTag::from(create_index_op), true) // TODO: Assign clock tag!? 🤔
+                    .update_local(
+                        OperationWithClockTag::from(create_index_op),
+                        true,
+                        hw_counter.clone(),
+                    ) // TODO: Assign clock tag!? 🤔
                     .await?;
             }
 

lib/collection/src/shards/local_shard/shard_ops.rs

@@ -38,7 +38,7 @@ impl ShardOperation for LocalShard {
         &self,
         mut operation: OperationWithClockTag,
         wait: bool,
-        _hw_measurement_acc: HwMeasurementAcc, // TODO(io_measurement): propagate value!
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         // `LocalShard::update` only has a single cancel safe `await`, WAL operations are blocking,
         // and update is applied by a separate task, so, surprisingly, this method is cancel safe. :D
@@ -88,6 +88,7 @@ impl ShardOperation for LocalShard {
                 operation: operation.operation,
                 sender: callback_sender,
                 wait,
+                hw_measurements: hw_measurement_acc,
             }));
 
             operation_id

lib/collection/src/shards/queue_proxy_shard.rs

@@ -502,7 +502,8 @@ impl Inner {
         // Transfer batch with retries and store last transferred ID
         let last_idx = batch.last().map(|(idx, _)| *idx);
         for remaining_attempts in (0..BATCH_RETRIES).rev() {
-            match transfer_operations_batch(&batch, &self.remote_shard).await {
+            let disposed_hw = HwMeasurementAcc::disposable(); // Internal operation
+            match transfer_operations_batch(&batch, &self.remote_shard, disposed_hw).await {
                 Ok(()) => {
                     if let Some(idx) = last_idx {
                         self.transfer_from.store(idx + 1, Ordering::Relaxed);
@@ -688,6 +689,7 @@ impl ShardOperation for Inner {
 async fn transfer_operations_batch(
     batch: &[(u64, OperationWithClockTag)],
     remote_shard: &RemoteShard,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> CollectionResult<()> {
     // TODO: naive transfer approach, transfer batch of points instead
     for (_idx, operation) in batch {
@@ -700,7 +702,12 @@ async fn transfer_operations_batch(
         }
 
         remote_shard
-            .forward_update(operation, true, WriteOrdering::Weak)
+            .forward_update(
+                operation,
+                true,
+                WriteOrdering::Weak,
+                hw_measurement_acc.clone(),
+            )
             .await?;
     }
     Ok(())

lib/collection/src/shards/remote_shard.rs

@@ -223,6 +223,7 @@ impl RemoteShard {
         operation: OperationWithClockTag,
         wait: bool,
         ordering: WriteOrdering,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         // `RemoteShard::execute_update_operation` is cancel safe, so this method is cancel safe.
 
@@ -232,6 +233,7 @@ impl RemoteShard {
             operation,
             wait,
             Some(ordering),
+            hw_measurement_acc,
         )
         .await
     }
@@ -246,6 +248,7 @@ impl RemoteShard {
         operation: OperationWithClockTag,
         wait: bool,
         ordering: Option<WriteOrdering>,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         // Cancelling remote request should always be safe on the client side and update API
         // *should be* cancel safe on the server side, so this method is cancel safe.
@@ -496,6 +499,11 @@ impl RemoteShard {
                 }
             },
         };
+
+        if let Some(hw_usage) = point_operation_response.usage {
+            hw_measurement_acc.accumulate_request(hw_usage);
+        }
+
         match point_operation_response.result {
             None => Err(CollectionError::service_error(
                 "Malformed UpdateResult type".to_string(),
@@ -665,14 +673,21 @@ impl ShardOperation for RemoteShard {
         &self,
         operation: OperationWithClockTag,
         wait: bool,
-        _hw_measurement_acc: HwMeasurementAcc, // TODO(io_measurement) fill this with response data
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         // `RemoteShard::execute_update_operation` is cancel safe, so this method is cancel safe.
 
         // targets the shard explicitly
         let shard_id = Some(self.id);
-        self.execute_update_operation(shard_id, self.collection_id.clone(), operation, wait, None)
-            .await
+        self.execute_update_operation(
+            shard_id,
+            self.collection_id.clone(),
+            operation,
+            wait,
+            None,
+            hw_measurement_acc,
+        )
+        .await
     }
 
     async fn scroll_by(
@@ -719,14 +734,8 @@ impl ShardOperation for RemoteShard {
         // We need the `____ordered_with____` value even if the user didn't request payload
         let parse_payload = with_payload_interface.is_required() || order_by.is_some();
 
-        if let Some(usage) = scroll_response.usage {
-            hw_measurement_acc.accumulate_request(
-                usage.cpu as usize,
-                usage.payload_io_read as usize,
-                usage.payload_io_write as usize,
-                usage.vector_io_read as usize,
-                usage.vector_io_write as usize,
-            );
+        if let Some(hw_usage) = scroll_response.usage {
+            hw_measurement_acc.accumulate_request(hw_usage);
         }
 
         let result: Result<Vec<RecordInternal>, Status> = scroll_response
@@ -756,6 +765,7 @@ impl ShardOperation for RemoteShard {
         let result: Result<CollectionInfo, Status> = get_collection_response.try_into();
         result.map_err(|e| e.into())
     }
+
     async fn core_search(
         &self,
         batch_request: Arc<CoreSearchRequestBatch>,
@@ -798,14 +808,8 @@ impl ShardOperation for RemoteShard {
             usage,
         } = search_batch_response;
 
-        if let Some(usage) = usage {
-            hw_measurement_acc.accumulate_request(
-                usage.cpu as usize,
-                usage.payload_io_read as usize,
-                usage.payload_io_write as usize,
-                usage.vector_io_read as usize,
-                usage.vector_io_write as usize,
-            );
+        if let Some(hw_usage) = usage {
+            hw_measurement_acc.accumulate_request(hw_usage);
         }
 
         let result: Result<Vec<Vec<ScoredPoint>>, Status> = result
@@ -869,14 +873,8 @@ impl ShardOperation for RemoteShard {
             usage,
         } = count_response;
 
-        if let Some(usage) = usage {
-            hw_measurement_acc.accumulate_request(
-                usage.cpu as usize,
-                usage.payload_io_read as usize,
-                usage.payload_io_write as usize,
-                usage.vector_io_read as usize,
-                usage.vector_io_write as usize,
-            );
+        if let Some(hw_usage) = usage {
+            hw_measurement_acc.accumulate_request(hw_usage);
         }
 
         result.map_or_else(
@@ -924,14 +922,8 @@ impl ShardOperation for RemoteShard {
             .await?
             .into_inner();
 
-        if let Some(usage) = get_response.usage {
-            hw_measurement_acc.accumulate_request(
-                usage.cpu as usize,
-                usage.payload_io_read as usize,
-                usage.payload_io_write as usize,
-                usage.vector_io_read as usize,
-                usage.vector_io_write as usize,
-            );
+        if let Some(hw_usage) = get_response.usage {
+            hw_measurement_acc.accumulate_request(hw_usage);
         }
 
         let result: Result<Vec<RecordInternal>, Status> = get_response
@@ -987,14 +979,8 @@ impl ShardOperation for RemoteShard {
             usage,
         } = batch_response;
 
-        if let Some(usage) = usage {
-            hw_measurement_acc.accumulate_request(
-                usage.cpu as usize,
-                usage.payload_io_read as usize,
-                usage.payload_io_write as usize,
-                usage.vector_io_read as usize,
-                usage.vector_io_write as usize,
-            );
+        if let Some(hw_usage) = usage {
+            hw_measurement_acc.accumulate_request(hw_usage);
         }
 
         let result = results
@@ -1027,7 +1013,7 @@ impl ShardOperation for RemoteShard {
         request: Arc<FacetParams>,
         _search_runtime_handle: &Handle,
         timeout: Option<Duration>,
-        _hw_measurement_acc: HwMeasurementAcc,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<FacetResponse> {
         let processed_timeout = Self::process_read_timeout(timeout, "facet")?;
         let mut timer = ScopeDurationMeasurer::new(&self.telemetry_search_durations);
@@ -1063,7 +1049,9 @@ impl ShardOperation for RemoteShard {
             .await?
             .into_inner();
 
-        // TODO(io_measurement): measure remote io usage here!
+        if let Some(hw_usage) = response.usage {
+            hw_measurement_acc.accumulate_request(hw_usage);
+        }
 
         let hits = response
             .hits

lib/collection/src/shards/replica_set/mod.rs

@@ -945,12 +945,15 @@ impl ShardReplicaSet {
             });
 
         // TODO(resharding): Assign clock tag to the operation!? 🤔
-        let result = self.update_local(op.into(), true).await?.ok_or_else(|| {
-            CollectionError::bad_request(format!(
-                "local shard {}:{} does not exist or is unavailable",
-                self.collection_id, self.shard_id,
-            ))
-        })?;
+        let result = self
+            .update_local(op.into(), true, hw_measurement_acc)
+            .await?
+            .ok_or_else(|| {
+                CollectionError::bad_request(format!(
+                    "local shard {}:{} does not exist or is unavailable",
+                    self.collection_id, self.shard_id,
+                ))
+            })?;
 
         Ok(result)
     }
@@ -1285,6 +1288,21 @@ impl ReplicaState {
             | ReplicaState::Listener => false,
         }
     }
+
+    /// Returns `true` if the replica state is resharding, either up or down.
+    pub fn is_resharding(&self) -> bool {
+        match self {
+            ReplicaState::Resharding | ReplicaState::ReshardingScaleDown => true,
+
+            ReplicaState::Partial
+            | ReplicaState::PartialSnapshot
+            | ReplicaState::Recovery
+            | ReplicaState::Active
+            | ReplicaState::Dead
+            | ReplicaState::Initializing
+            | ReplicaState::Listener => false,
+        }
+    }
 }
 
 /// Represents a change in replica set, due to scaling of `replication_factor`

lib/collection/src/shards/replica_set/update.rs

@@ -31,6 +31,7 @@ impl ShardReplicaSet {
         &self,
         operation: OperationWithClockTag,
         wait: bool,
+        mut hw_measurement: HwMeasurementAcc,
     ) -> CollectionResult<Option<UpdateResult>> {
         // `ShardOperations::update` is not guaranteed to be cancel safe, so this method is not
         // cancel safe.
@@ -45,8 +46,10 @@ impl ShardReplicaSet {
             return Ok(None);
         };
 
-        // TODO(io_measurement): Propagate measurements to caller
-        let hw_measurement = HwMeasurementAcc::disposable();
+        // Don't measure hw when resharding
+        if state.is_resharding() {
+            hw_measurement = HwMeasurementAcc::disposable();
+        }
 
         let result = match state {
             ReplicaState::Active => {
@@ -104,6 +107,7 @@ impl ShardReplicaSet {
         wait: bool,
         ordering: WriteOrdering,
         update_only_existing: bool,
+        mut hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         // `ShardReplicaSet::update` is not cancel safe, so this method is not cancel safe.
 
@@ -114,6 +118,12 @@ impl ShardReplicaSet {
             )));
         };
 
+        // Don't measure hw when resharding
+        let peer_state = self.peer_state(leader_peer);
+        if peer_state.is_some_and(|state| state.is_resharding()) {
+            hw_measurement_acc = HwMeasurementAcc::disposable();
+        }
+
         // If we are the leader, run the update from this replica set
         if leader_peer == self.this_peer_id() {
             // Lock updates if ordering is strong or medium
@@ -124,10 +134,11 @@ impl ShardReplicaSet {
                 WriteOrdering::Weak => None,
             };
 
-            self.update(operation, wait, update_only_existing).await
+            self.update(operation, wait, update_only_existing, hw_measurement_acc)
+                .await
         } else {
             // Forward the update to the designated leader
-            self.forward_update(leader_peer, operation, wait, ordering)
+            self.forward_update(leader_peer, operation, wait, ordering, hw_measurement_acc)
                 .await
                 .map_err(|err| {
                     if err.is_transient() {
@@ -179,6 +190,7 @@ impl ShardReplicaSet {
         operation: CollectionUpdateOperations,
         wait: bool,
         update_only_existing: bool,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         // `ShardRepilcaSet::update_impl` is not cancel safe, so this method is not cancel safe.
 
@@ -194,7 +206,13 @@ impl ShardReplicaSet {
             let is_non_zero_tick = clock.current_tick().is_some();
 
             let res = self
-                .update_impl(operation.clone(), wait, &mut clock, update_only_existing)
+                .update_impl(
+                    operation.clone(),
+                    wait,
+                    &mut clock,
+                    update_only_existing,
+                    hw_measurement_acc.clone(),
+                )
                 .await?;
 
             if let Some(res) = res {
@@ -234,13 +252,12 @@ impl ShardReplicaSet {
         wait: bool,
         clock: &mut clock_set::ClockGuard,
         update_only_existing: bool,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<Option<UpdateResult>> {
         // `LocalShard::update` is not guaranteed to be cancel safe and it's impossible to cancel
         // multiple parallel updates in a way that is *guaranteed* not to introduce inconsistencies
         // between nodes, so this method is not cancel safe.
 
-        let hw_measurement_acc = HwMeasurementAcc::disposable(); // TODO(io_measurement) propagate values!
-
         let remotes = self.remotes.read().await;
         let local = self.local.read().await;
         let replica_count = usize::from(local.is_some()) + remotes.len();
@@ -632,6 +649,7 @@ impl ShardReplicaSet {
         operation: CollectionUpdateOperations,
         wait: bool,
         ordering: WriteOrdering,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> CollectionResult<UpdateResult> {
         // `RemoteShard::forward_update` is cancel safe, so this method is cancel safe.
 
@@ -645,7 +663,12 @@ impl ShardReplicaSet {
         };
 
         remote_leader
-            .forward_update(OperationWithClockTag::from(operation), wait, ordering) // `clock_tag` *have to* be `None`!
+            .forward_update(
+                OperationWithClockTag::from(operation),
+                wait,
+                ordering,
+                hw_measurement_acc,
+            ) // `clock_tag` *has to* be `None`!
             .await
     }
 }

lib/collection/src/tests/points_dedup.rs

@@ -132,7 +132,7 @@ async fn fixture() -> Collection {
             ])),
         ));
         shard
-            .update_local(op, true)
+            .update_local(op, true, HwMeasurementAcc::new())
             .await
             .expect("failed to insert points");
     }

lib/collection/src/update_handler.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 
 use common::budget::ResourceBudget;
+use common::counter::hardware_accumulator::HwMeasurementAcc;
 use common::counter::hardware_counter::HardwareCounterCell;
 use common::panic;
 use itertools::Itertools;
@@ -55,6 +56,7 @@ pub struct OperationData {
     pub wait: bool,
     /// Callback notification channel
     pub sender: Option<oneshot::Sender<CollectionResult<usize>>>,
+    pub hw_measurements: HwMeasurementAcc,
 }
 
 /// Signal, used to inform Updater process
@@ -684,6 +686,7 @@ impl UpdateHandler {
                     operation,
                     sender,
                     wait,
+                    hw_measurements,
                 }) => {
                     let flush_res = if wait {
                         wal.lock().flush().map_err(|err| {
@@ -695,10 +698,13 @@ impl UpdateHandler {
                         Ok(())
                     };
 
-                    let hw_counter = HardwareCounterCell::disposable(); // TODO(io_measurement): implement!!
-
                     let operation_result = flush_res.and_then(|_| {
-                        CollectionUpdater::update(&segments, op_num, operation, &hw_counter)
+                        CollectionUpdater::update(
+                            &segments,
+                            op_num,
+                            operation,
+                            &hw_measurements.get_counter_cell(),
+                        )
                     });
 
                     let res = match operation_result {

lib/collection/tests/integration/collection_restore_test.rs

@@ -44,8 +44,9 @@ async fn test_collection_reloading_with_shards(shard_number: u32) {
                     payloads: None,
                 }),
             ));
+        let hw_counter = HwMeasurementAcc::new();
         collection
-            .update_from_client_simple(insert_points, true, WriteOrdering::default())
+            .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
             .await
             .unwrap();
     }
@@ -88,8 +89,9 @@ async fn test_collection_payload_reloading_with_shards(shard_number: u32) {
                     payloads: serde_json::from_str(r#"[{ "k": "v1" } , { "k": "v2"}]"#).unwrap(),
                 }),
             ));
+        let hw_counter = HwMeasurementAcc::new();
         collection
-            .update_from_client_simple(insert_points, true, WriteOrdering::default())
+            .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
             .await
             .unwrap();
     }
@@ -168,8 +170,9 @@ async fn test_collection_payload_custom_payload_with_shards(shard_number: u32) {
                     .unwrap(),
                 }),
             ));
+        let hw_counter = HwMeasurementAcc::new();
         collection
-            .update_from_client_simple(insert_points, true, WriteOrdering::default())
+            .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
             .await
             .unwrap();
     }

lib/collection/tests/integration/collection_test.rs

@@ -58,8 +58,9 @@ async fn test_collection_updater_with_shards(shard_number: u32) {
         PointInsertOperationsInternal::from(batch),
     ));
 
+    let hw_counter = HwMeasurementAcc::new();
     let insert_result = collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
         .await;
 
     match insert_result {
@@ -128,8 +129,9 @@ async fn test_collection_search_with_payload_and_vector_with_shards(shard_number
         PointInsertOperationsInternal::from(batch),
     ));
 
+    let hw_counter = HwMeasurementAcc::new();
     let insert_result = collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
         .await;
 
     match insert_result {
@@ -231,8 +233,9 @@ async fn test_collection_loading_with_shards(shard_number: u32) {
             PointOperations::UpsertPoints(PointInsertOperationsInternal::from(batch)),
         );
 
+        let hw_counter = HwMeasurementAcc::new();
         collection
-            .update_from_client_simple(insert_points, true, WriteOrdering::default())
+            .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
             .await
             .unwrap();
 
@@ -246,8 +249,9 @@ async fn test_collection_loading_with_shards(shard_number: u32) {
                 key: None,
             }));
 
+        let hw_counter = HwMeasurementAcc::new();
         collection
-            .update_from_client_simple(assign_payload, true, WriteOrdering::default())
+            .update_from_client_simple(assign_payload, true, WriteOrdering::default(), hw_counter)
             .await
             .unwrap();
     }
@@ -374,7 +378,12 @@ async fn test_recommendation_api_with_shards(shard_number: u32) {
 
     let hw_acc = HwMeasurementAcc::new();
     collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(
+            insert_points,
+            true,
+            WriteOrdering::default(),
+            hw_acc.clone(),
+        )
         .await
         .unwrap();
     let result = recommend_by(
@@ -432,8 +441,9 @@ async fn test_read_api_with_shards(shard_number: u32) {
         PointInsertOperationsInternal::from(batch),
     ));
 
+    let hw_counter = HwMeasurementAcc::new();
     collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
         .await
         .unwrap();
 
@@ -529,8 +539,9 @@ async fn test_ordered_scroll_api_with_shards(shard_number: u32) {
         PointInsertOperationsInternal::from(batch),
     ));
 
+    let hw_counter = HwMeasurementAcc::new();
     collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
         .await
         .unwrap();
 
@@ -782,8 +793,14 @@ async fn test_collection_delete_points_by_filter_with_shards(shard_number: u32)
         PointInsertOperationsInternal::from(batch),
     ));
 
+    let hw_counter = HwMeasurementAcc::new();
     let insert_result = collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(
+            insert_points,
+            true,
+            WriteOrdering::default(),
+            hw_counter.clone(),
+        )
         .await;
 
     match insert_result {
@@ -803,7 +820,7 @@ async fn test_collection_delete_points_by_filter_with_shards(shard_number: u32)
     );
 
     let delete_result = collection
-        .update_from_client_simple(delete_points, true, WriteOrdering::default())
+        .update_from_client_simple(delete_points, true, WriteOrdering::default(), hw_counter)
         .await;
 
     match delete_result {

lib/collection/tests/integration/distance_matrix_test.rs

@@ -66,8 +66,9 @@ async fn distance_matrix_anonymous_vector() {
         ),
     );
 
+    let hw_counter = HwMeasurementAcc::new();
     collection
-        .update_from_client_simple(upsert_points, true, WriteOrdering::default())
+        .update_from_client_simple(upsert_points, true, WriteOrdering::default(), hw_counter)
         .await
         .unwrap();
 

lib/collection/tests/integration/grouping_test.rs

@@ -80,8 +80,14 @@ mod group_by {
             PointOperations::UpsertPoints(PointInsertOperationsInternal::from(batch)),
         );
 
+        let hw_counter = HwMeasurementAcc::new();
         let insert_result = collection
-            .update_from_client_simple(insert_points, true, WriteOrdering::default())
+            .update_from_client_simple(
+                insert_points,
+                true,
+                WriteOrdering::default(),
+                hw_counter.clone(),
+            )
             .await
             .expect("insert failed");
 
@@ -514,6 +520,8 @@ mod group_by_builder {
         let collection_dir = tempfile::Builder::new().prefix("chunks").tempdir().unwrap();
         let collection = simple_collection_fixture(collection_dir.path(), 1).await;
 
+        let hw_counter = HwMeasurementAcc::new();
+
         // insert chunk points
         {
             let batch = BatchPersisted {
@@ -536,7 +544,12 @@ mod group_by_builder {
             );
 
             let insert_result = collection
-                .update_from_client_simple(insert_points, true, WriteOrdering::default())
+                .update_from_client_simple(
+                    insert_points,
+                    true,
+                    WriteOrdering::default(),
+                    hw_counter.clone(),
+                )
                 .await
                 .expect("insert failed");
 
@@ -565,7 +578,12 @@ mod group_by_builder {
                 PointOperations::UpsertPoints(PointInsertOperationsInternal::from(batch)),
             );
             let insert_result = lookup_collection
-                .update_from_client_simple(insert_points, true, WriteOrdering::default())
+                .update_from_client_simple(
+                    insert_points,
+                    true,
+                    WriteOrdering::default(),
+                    hw_counter.clone(),
+                )
                 .await
                 .expect("insert failed");
 

lib/collection/tests/integration/lookup_test.rs

@@ -69,8 +69,10 @@ async fn setup() -> Resources {
         PointOperations::UpsertPoints(PointInsertOperationsInternal::from(batch)),
     );
 
+    let hw_counter = HwMeasurementAcc::new();
+
     collection
-        .update_from_client_simple(upsert_points, true, WriteOrdering::default())
+        .update_from_client_simple(upsert_points, true, WriteOrdering::default(), hw_counter)
         .await
         .unwrap();
 

lib/collection/tests/integration/multi_vec_test.rs

@@ -107,8 +107,9 @@ async fn test_multi_vec_with_shards(shard_number: u32) {
     let insert_points = CollectionUpdateOperations::PointOperation(PointOperations::UpsertPoints(
         PointInsertOperationsInternal::PointsList(points),
     ));
+    let hw_counter = HwMeasurementAcc::new();
     collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
         .await
         .unwrap();
 

lib/collection/tests/integration/pagination_test.rs

@@ -37,8 +37,9 @@ async fn test_collection_paginated_search_with_shards(shard_number: u32) {
     let insert_points = CollectionUpdateOperations::PointOperation(PointOperations::UpsertPoints(
         PointInsertOperationsInternal::PointsList(points),
     ));
+    let hw_counter = HwMeasurementAcc::new();
     collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
         .await
         .unwrap();
 

lib/collection/tests/integration/snapshot_recovery_test.rs

@@ -106,8 +106,9 @@ async fn _test_snapshot_and_recover_collection(node_type: NodeType) {
     let insert_points = CollectionUpdateOperations::PointOperation(PointOperations::UpsertPoints(
         PointInsertOperationsInternal::PointsList(points),
     ));
+    let hw_counter = HwMeasurementAcc::new();
     collection
-        .update_from_client_simple(insert_points, true, WriteOrdering::default())
+        .update_from_client_simple(insert_points, true, WriteOrdering::default(), hw_counter)
         .await
         .unwrap();
 

lib/common/common/src/counter/hardware_accumulator.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
 
 use super::hardware_counter::HardwareCounterCell;
+use super::hardware_data::HardwareData;
 
 /// Data structure, that routes hardware measurement counters to specific location.
 /// Shared drain MUST NOT create its own counters, but only hold a reference to the existing one,
@@ -51,6 +52,23 @@ impl HwSharedDrain {
             vector_io_write_counter: vector_io_write,
         }
     }
+
+    /// Accumulates all values from `src` into this HwSharedDrain.
+    fn accumulate_from_hw_data(&self, src: HardwareData) {
+        let HwSharedDrain {
+            cpu_counter,
+            payload_io_read_counter,
+            payload_io_write_counter,
+            vector_io_read_counter,
+            vector_io_write_counter,
+        } = self;
+
+        cpu_counter.fetch_add(src.cpu.get(), Ordering::Relaxed);
+        payload_io_read_counter.fetch_add(src.payload_io_read, Ordering::Relaxed);
+        payload_io_write_counter.fetch_add(src.payload_io_write, Ordering::Relaxed);
+        vector_io_read_counter.fetch_add(src.vector_io_read, Ordering::Relaxed);
+        vector_io_write_counter.fetch_add(src.vector_io_write, Ordering::Relaxed);
+    }
 }
 
 impl Clone for HwSharedDrain {
@@ -115,59 +133,18 @@ impl HwMeasurementAcc {
         }
     }
 
-    pub fn accumulate(
-        &self,
-        cpu: usize,
-        payload_io_read: usize,
-        payload_io_write: usize,
-        vector_io_read: usize,
-        vector_io_write: usize,
-    ) {
-        self.accumulate_request(
-            cpu,
-            payload_io_read,
-            payload_io_write,
-            vector_io_read,
-            vector_io_write,
-        );
-
-        let HwSharedDrain {
-            cpu_counter,
-            payload_io_read_counter,
-            payload_io_write_counter,
-            vector_io_read_counter,
-            vector_io_write_counter,
-        } = &self.metrics_drain;
-        cpu_counter.fetch_add(cpu, Ordering::Relaxed);
-        payload_io_read_counter.fetch_add(payload_io_read, Ordering::Relaxed);
-        payload_io_write_counter.fetch_add(payload_io_write, Ordering::Relaxed);
-        vector_io_read_counter.fetch_add(vector_io_read, Ordering::Relaxed);
-        vector_io_write_counter.fetch_add(vector_io_write, Ordering::Relaxed);
+    pub fn accumulate<T: Into<HardwareData>>(&self, src: T) {
+        let src = src.into();
+        self.request_drain.accumulate_from_hw_data(src);
+        self.metrics_drain.accumulate_from_hw_data(src);
     }
 
-    /// Accumulate usage values for request drain only
+    /// Accumulate usage values for request drain only.
     /// This is useful if we want to report usage, which happened on another machine
     /// So we don't want to accumulate the same usage on the current machine second time
-    pub fn accumulate_request(
-        &self,
-        cpu: usize,
-        payload_io_read: usize,
-        payload_io_write: usize,
-        vector_io_read: usize,
-        vector_io_write: usize,
-    ) {
-        let HwSharedDrain {
-            cpu_counter,
-            payload_io_read_counter,
-            payload_io_write_counter,
-            vector_io_read_counter,
-            vector_io_write_counter,
-        } = &self.request_drain;
-        cpu_counter.fetch_add(cpu, Ordering::Relaxed);
-        payload_io_read_counter.fetch_add(payload_io_read, Ordering::Relaxed);
-        payload_io_write_counter.fetch_add(payload_io_write, Ordering::Relaxed);
-        vector_io_read_counter.fetch_add(vector_io_read, Ordering::Relaxed);
-        vector_io_write_counter.fetch_add(vector_io_write, Ordering::Relaxed);
+    pub fn accumulate_request<T: Into<HardwareData>>(&self, src: T) {
+        let src = src.into();
+        self.request_drain.accumulate_from_hw_data(src);
     }
 
     pub fn get_cpu(&self) -> usize {

lib/common/common/src/counter/hardware_counter.rs

@@ -1,5 +1,6 @@
 use super::counter_cell::CounterCell;
 use super::hardware_accumulator::HwMeasurementAcc;
+use super::hardware_data::{HardwareData, RealCpuMeasurement};
 
 /// Collection of different types of hardware measurements.
 ///
@@ -9,7 +10,7 @@ use super::hardware_accumulator::HwMeasurementAcc;
 #[derive(Debug)]
 pub struct HardwareCounterCell {
     cpu_multiplier: usize,
-    pub(super) cpu_counter: CounterCell,
+    cpu_counter: CounterCell,
     pub(super) payload_io_read_counter: CounterCell,
     pub(super) payload_io_write_counter: CounterCell,
     pub(super) vector_io_read_counter: CounterCell,
@@ -96,11 +97,20 @@ impl HardwareCounterCell {
         self.cpu_multiplier = multiplier;
     }
 
+    /// Returns the real cpu value with multiplier applied.
+    pub fn get_cpu(&self) -> RealCpuMeasurement {
+        RealCpuMeasurement::new(self.cpu_counter.get(), self.cpu_multiplier)
+    }
+
+    /// Returns the CPU counter that can be used for counting.
+    /// Should *never* be used for reading CPU measurements! Use `.get_cpu()` for this.
     #[inline]
     pub fn cpu_counter(&self) -> &CounterCell {
         &self.cpu_counter
     }
 
+    /// Returns the CPU counter that can be used for counting.
+    /// Should *never* be used for reading CPU measurements! Use `.get_cpu()` for this.
     #[inline]
     pub fn cpu_counter_mut(&mut self) -> &mut CounterCell {
         &mut self.cpu_counter
@@ -146,25 +156,29 @@ impl HardwareCounterCell {
         &mut self.vector_io_write_counter
     }
 
-    fn merge_to_accumulator(&self) {
+    /// Returns a copy of the current measurements made by this counter. Ignores all values from the parent accumulator.
+    pub fn get_hw_data(&self) -> HardwareData {
         let HardwareCounterCell {
-            cpu_multiplier,
-            cpu_counter,
+            cpu_multiplier: _,
+            cpu_counter: _, // We use .get_cpu() to calculate the real CPU value.
             payload_io_read_counter,
             payload_io_write_counter,
             vector_io_read_counter,
             vector_io_write_counter,
-            accumulator,
+            accumulator: _,
         } = self;
 
-        let cpu_value = cpu_counter.get() * cpu_multiplier;
-        accumulator.accumulate(
-            cpu_value,
-            payload_io_read_counter.get(),
-            payload_io_write_counter.get(),
-            vector_io_read_counter.get(),
-            vector_io_write_counter.get(),
-        );
+        HardwareData {
+            cpu: self.get_cpu(),
+            payload_io_read: payload_io_read_counter.get(),
+            payload_io_write: payload_io_write_counter.get(),
+            vector_io_read: vector_io_read_counter.get(),
+            vector_io_write: vector_io_write_counter.get(),
+        }
+    }
+
+    fn merge_to_accumulator(&self) {
+        self.accumulator.accumulate(self.get_hw_data());
     }
 }
 

lib/common/common/src/counter/hardware_data.rs

@@ -0,0 +1,25 @@
+/// Contains all hardware metrics. Only serves as value holding structure without any semantics.
+#[derive(Copy, Clone)]
+pub struct HardwareData {
+    pub cpu: RealCpuMeasurement,
+    pub payload_io_read: usize,
+    pub payload_io_write: usize,
+    pub vector_io_read: usize,
+    pub vector_io_write: usize,
+}
+
+/// Wrapper type to ensure we properly apply cpu-multiplier everywhere.
+#[derive(Copy, Clone)]
+pub struct RealCpuMeasurement(usize);
+
+impl RealCpuMeasurement {
+    pub fn new(cpu: usize, multiplier: usize) -> Self {
+        Self(cpu * multiplier)
+    }
+
+    /// Returns the real CPU value.
+    #[inline]
+    pub fn get(self) -> usize {
+        self.0
+    }
+}

lib/common/common/src/counter/mod.rs

@@ -1,3 +1,4 @@
 pub mod counter_cell;
 pub mod hardware_accumulator;
 pub mod hardware_counter;
+pub mod hardware_data;

lib/storage/src/content_manager/data_transfer.rs

@@ -131,8 +131,9 @@ async fn replicate_shard_data(
         let target_collection =
             handle_get_collection(collections_read.get(target_collection_name))?;
 
+        let hw_counter = HwMeasurementAcc::disposable(); // Internal operation
         target_collection
-            .update_from_client_simple(upsert_request, false, WriteOrdering::default())
+            .update_from_client_simple(upsert_request, false, WriteOrdering::default(), hw_counter)
             .await?;
 
         if offset.is_none() {
@@ -233,8 +234,9 @@ pub async fn transfer_indexes(
                 field_schema: Some(schema.try_into()?),
             }),
         );
+        let hw_counter = HwMeasurementAcc::disposable(); // Internal operation
         target_collection
-            .update_from_client_simple(request, false, WriteOrdering::default())
+            .update_from_client_simple(request, false, WriteOrdering::default(), hw_counter)
             .await?;
     }
 

lib/storage/src/content_manager/toc/point_ops.rs

@@ -445,13 +445,20 @@ impl TableOfContent {
         operation: CollectionUpdateOperations,
         wait: bool,
         ordering: WriteOrdering,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> StorageResult<UpdateResult> {
         // `Collection::update_from_client` is cancel safe, so this method is cancel safe.
 
         let updates: FuturesUnordered<_> = shard_keys
             .into_iter()
             .map(|shard_key| {
-                collection.update_from_client(operation.clone(), wait, ordering, Some(shard_key))
+                collection.update_from_client(
+                    operation.clone(),
+                    wait,
+                    ordering,
+                    Some(shard_key),
+                    hw_measurement_acc.clone(),
+                )
             })
             .collect();
 
@@ -467,6 +474,7 @@ impl TableOfContent {
     /// # Cancel safety
     ///
     /// This method is cancel safe.
+    #[allow(clippy::too_many_arguments)]
     pub async fn update(
         &self,
         collection_name: &str,
@@ -475,6 +483,7 @@ impl TableOfContent {
         ordering: WriteOrdering,
         shard_selector: ShardSelectorInternal,
         access: Access,
+        hw_measurement_acc: HwMeasurementAcc,
     ) -> StorageResult<UpdateResult> {
         let collection_pass = access.check_point_op(collection_name, &mut operation.operation)?;
 
@@ -532,7 +541,13 @@ impl TableOfContent {
         let res = match shard_selector {
             ShardSelectorInternal::Empty => {
                 collection
-                    .update_from_client(operation.operation, wait, ordering, None)
+                    .update_from_client(
+                        operation.operation,
+                        wait,
+                        ordering,
+                        None,
+                        hw_measurement_acc.clone(),
+                    )
                     .await?
             }
 
@@ -540,7 +555,13 @@ impl TableOfContent {
                 let shard_keys = collection.get_shard_keys().await;
                 if shard_keys.is_empty() {
                     collection
-                        .update_from_client(operation.operation, wait, ordering, None)
+                        .update_from_client(
+                            operation.operation,
+                            wait,
+                            ordering,
+                            None,
+                            hw_measurement_acc.clone(),
+                        )
                         .await?
                 } else {
                     Self::_update_shard_keys(
@@ -549,6 +570,7 @@ impl TableOfContent {
                         operation.operation,
                         wait,
                         ordering,
+                        hw_measurement_acc.clone(),
                     )
                     .await?
                 }
@@ -556,7 +578,13 @@ impl TableOfContent {
 
             ShardSelectorInternal::ShardKey(shard_key) => {
                 collection
-                    .update_from_client(operation.operation, wait, ordering, Some(shard_key))
+                    .update_from_client(
+                        operation.operation,
+                        wait,
+                        ordering,
+                        Some(shard_key),
+                        hw_measurement_acc.clone(),
+                    )
                     .await?
             }
 
@@ -567,13 +595,20 @@ impl TableOfContent {
                     operation.operation,
                     wait,
                     ordering,
+                    hw_measurement_acc.clone(),
                 )
                 .await?
             }
 
             ShardSelectorInternal::ShardId(shard_selection) => {
                 collection
-                    .update_from_peer(operation, shard_selection, wait, ordering)
+                    .update_from_peer(
+                        operation,
+                        shard_selection,
+                        wait,
+                        ordering,
+                        hw_measurement_acc.clone(),
+                    )
                     .await?
             }
         };

lib/storage/src/content_manager/toc/request_hw_counter.rs

@@ -11,6 +11,7 @@ impl TableOfContent {
     }
 }
 
+#[derive(Clone)]
 pub struct RequestHwCounter {
     counter: HwMeasurementAcc,
     /// If this flag is set, RequestHwCounter will be converted into non-None API representation.

src/actix/api/count_api.rs

@@ -49,6 +49,7 @@ async fn count_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
 
     let timing = Instant::now();

src/actix/api/discovery_api.rs

@@ -52,6 +52,7 @@ async fn discover_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
 
     let timing = Instant::now();
@@ -106,6 +107,7 @@ async fn discover_batch_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 

src/actix/api/facet_api.rs

@@ -54,6 +54,7 @@ async fn facet(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
 
     let response = dispatcher

src/actix/api/local_shard_api.rs

@@ -50,6 +50,7 @@ async fn get_points(
         &dispatcher,
         path.collection.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -107,6 +108,7 @@ async fn scroll_points(
         &dispatcher,
         path.collection.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -180,6 +182,7 @@ async fn count_points(
         &dispatcher,
         path.collection.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
     let hw_measurement_acc = request_hw_counter.get_counter();

src/actix/api/query_api.rs

@@ -53,6 +53,7 @@ async fn query_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -119,6 +120,7 @@ async fn query_points_batch(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
     let hw_measurement_acc = request_hw_counter.get_counter();
@@ -198,6 +200,7 @@ async fn query_points_groups(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
     let hw_measurement_acc = request_hw_counter.get_counter();

src/actix/api/recommend_api.rs

@@ -61,6 +61,7 @@ async fn recommend_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
 
     let timing = Instant::now();
@@ -146,6 +147,7 @@ async fn recommend_batch_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -210,6 +212,7 @@ async fn recommend_point_groups(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 

src/actix/api/retrieve_api.rs

@@ -100,6 +100,7 @@ async fn get_point(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -158,6 +159,7 @@ async fn get_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -218,6 +220,7 @@ async fn scroll_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 

src/actix/api/search_api.rs

@@ -60,6 +60,7 @@ async fn search_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
 
     let timing = Instant::now();
@@ -130,6 +131,7 @@ async fn batch_search_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
 
     let timing = Instant::now();
@@ -195,6 +197,7 @@ async fn search_point_groups(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -249,6 +252,7 @@ async fn search_points_matrix_pairs(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 
@@ -304,6 +308,7 @@ async fn search_points_matrix_offsets(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        None,
     );
     let timing = Instant::now();
 

src/actix/api/update_api.rs

@@ -7,6 +7,7 @@ use collection::operations::payload_ops::{DeletePayload, SetPayload};
 use collection::operations::point_ops::PointsSelector;
 use collection::operations::types::UpdateResult;
 use collection::operations::vector_ops::DeleteVectors;
+use common::counter::hardware_accumulator::HwMeasurementAcc;
 use segment::json_path::JsonPath;
 use serde::Deserialize;
 use storage::content_manager::collection_verification::check_strict_mode;
@@ -50,10 +51,10 @@ async fn upsert_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure upsertion io
     let res = do_upsert_points(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -62,6 +63,7 @@ async fn upsert_points(
         params.into_inner(),
         access,
         inference_token,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -89,10 +91,10 @@ async fn delete_points(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure io of deletion
     let res = do_delete_points(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -101,6 +103,7 @@ async fn delete_points(
         params.into_inner(),
         access,
         inference_token,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -129,10 +132,10 @@ async fn update_vectors(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure update io
     let res = do_update_vectors(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -141,6 +144,7 @@ async fn update_vectors(
         params.into_inner(),
         access,
         inference_token,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -169,10 +173,10 @@ async fn delete_vectors(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure vector deletion io
     let response = do_delete_vectors(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -180,6 +184,7 @@ async fn delete_vectors(
         InternalUpdateParams::default(),
         params.into_inner(),
         access,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -207,10 +212,10 @@ async fn set_payload(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure upsertion io
     let res = do_set_payload(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -218,6 +223,7 @@ async fn set_payload(
         InternalUpdateParams::default(),
         params.into_inner(),
         access,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -244,10 +250,10 @@ async fn overwrite_payload(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure upsertion io
     let res = do_overwrite_payload(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -255,6 +261,7 @@ async fn overwrite_payload(
         InternalUpdateParams::default(),
         params.into_inner(),
         access,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -281,10 +288,10 @@ async fn delete_payload(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure upsertion io
     let res = do_delete_payload(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -292,6 +299,7 @@ async fn delete_payload(
         InternalUpdateParams::default(),
         params.into_inner(),
         access,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -318,10 +326,10 @@ async fn clear_payload(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
     let timing = Instant::now();
 
-    // TODO(io_measurement): Measure upsertion io
     let res = do_clear_payload(
         dispatcher.toc(&access, &pass).clone(),
         collection.into_inner().name,
@@ -329,6 +337,7 @@ async fn clear_payload(
         InternalUpdateParams::default(),
         params.into_inner(),
         access,
+        request_hw_counter.get_counter(),
     )
     .await;
 
@@ -368,6 +377,7 @@ async fn update_batch(
         &dispatcher,
         collection.name.clone(),
         service_config.hardware_reporting(),
+        Some(params.wait),
     );
 
     let timing = Instant::now();
@@ -381,6 +391,7 @@ async fn update_batch(
         params.into_inner(),
         access,
         inference_token,
+        request_hw_counter.get_counter(),
     )
     .await;
     process_response(response, timing, request_hw_counter.to_rest_api())
@@ -404,6 +415,7 @@ async fn create_field_index(
         InternalUpdateParams::default(),
         params.into_inner(),
         access,
+        HwMeasurementAcc::disposable(), // TODO(io_measurement): measure payload index creation?
     )
     .await;
     process_response(response, timing, None)
@@ -426,6 +438,7 @@ async fn delete_field_index(
         InternalUpdateParams::default(),
         params.into_inner(),
         access,
+        HwMeasurementAcc::disposable(), // API unmeasured
     )
     .await;
     process_response(response, timing, None)

src/actix/helpers.rs

@@ -17,7 +17,9 @@ pub fn get_request_hardware_counter(
     dispatcher: &Dispatcher,
     collection_name: String,
     report_to_api: bool,
+    wait: Option<bool>,
 ) -> RequestHwCounter {
+    let report_to_api = report_to_api && wait != Some(false);
     RequestHwCounter::new(
         HwMeasurementAcc::new_with_metrics_drain(
             dispatcher.get_collection_hw_metrics(collection_name),

src/common/metrics.rs

@@ -338,36 +338,36 @@ impl MetricsProvider for HardwareTelemetry {
             metrics.push(metric_family(
                 "collection_hardware_metric_cpu",
                 "CPU measurements of a collection",
-                MetricType::GAUGE,
-                vec![gauge(*cpu as f64, &[("id", collection)])],
+                MetricType::COUNTER,
+                vec![counter(*cpu as f64, &[("id", collection)])],
             ));
 
             metrics.push(metric_family(
                 "collection_hardware_metric_payload_io_read",
                 "Total IO payload read metrics of a collection",
-                MetricType::GAUGE,
-                vec![gauge(*payload_io_read as f64, &[("id", collection)])],
+                MetricType::COUNTER,
+                vec![counter(*payload_io_read as f64, &[("id", collection)])],
             ));
 
             metrics.push(metric_family(
                 "collection_hardware_metric_payload_io_write",
                 "Total IO payload write metrics of a collection",
-                MetricType::GAUGE,
-                vec![gauge(*payload_io_write as f64, &[("id", collection)])],
+                MetricType::COUNTER,
+                vec![counter(*payload_io_write as f64, &[("id", collection)])],
             ));
 
             metrics.push(metric_family(
                 "collection_hardware_metric_vector_io_read",
                 "Total IO vector read metrics of a collection",
-                MetricType::GAUGE,
-                vec![gauge(*vector_io_read as f64, &[("id", collection)])],
+                MetricType::COUNTER,
+                vec![counter(*vector_io_read as f64, &[("id", collection)])],
             ));
 
             metrics.push(metric_family(
                 "collection_hardware_metric_vector_io_write",
                 "Total IO vector write metrics of a collection",
-                MetricType::GAUGE,
-                vec![gauge(*vector_io_write as f64, &[("id", collection)])],
+                MetricType::COUNTER,
+                vec![counter(*vector_io_write as f64, &[("id", collection)])],
             ));
         }
     }

src/common/update.rs

@@ -11,6 +11,7 @@ use collection::operations::vector_ops::*;
 use collection::operations::verification::*;
 use collection::operations::*;
 use collection::shards::shard::ShardId;
+use common::counter::hardware_accumulator::HwMeasurementAcc;
 use schemars::JsonSchema;
 use segment::json_path::JsonPath;
 use segment::types::{PayloadFieldSchema, PayloadKeyType, StrictModeConfig};
@@ -226,6 +227,7 @@ pub struct CreateFieldIndex {
     pub field_schema: Option<PayloadFieldSchema>,
 }
 
+#[expect(clippy::too_many_arguments)]
 pub async fn do_upsert_points(
     toc: Arc<TableOfContent>,
     collection_name: String,
@@ -234,6 +236,7 @@ pub async fn do_upsert_points(
     params: UpdateParams,
     access: Access,
     inference_token: InferenceToken,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let (shard_key, operation) = match operation {
         PointInsertOperations::PointsBatch(PointsBatch { batch, shard_key }) => (
@@ -261,10 +264,12 @@ pub async fn do_upsert_points(
         params,
         shard_key,
         access,
+        hw_measurement_acc,
     )
     .await
 }
 
+#[expect(clippy::too_many_arguments)]
 pub async fn do_delete_points(
     toc: Arc<TableOfContent>,
     collection_name: String,
@@ -273,6 +278,7 @@ pub async fn do_delete_points(
     params: UpdateParams,
     access: Access,
     _inference_token: InferenceToken,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let (point_operation, shard_key) = match points {
         PointsSelector::PointIdsSelector(PointIdsList { points, shard_key }) => {
@@ -293,10 +299,12 @@ pub async fn do_delete_points(
         params,
         shard_key,
         access,
+        hw_measurement_acc,
     )
     .await
 }
 
+#[expect(clippy::too_many_arguments)]
 pub async fn do_update_vectors(
     toc: Arc<TableOfContent>,
     collection_name: String,
@@ -305,6 +313,7 @@ pub async fn do_update_vectors(
     params: UpdateParams,
     access: Access,
     inference_token: InferenceToken,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let UpdateVectors { points, shard_key } = operation;
 
@@ -325,6 +334,7 @@ pub async fn do_update_vectors(
         params,
         shard_key,
         access,
+        hw_measurement_acc,
     )
     .await
 }
@@ -336,6 +346,7 @@ pub async fn do_delete_vectors(
     internal_params: InternalUpdateParams,
     params: UpdateParams,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     // TODO: Is this cancel safe!?
 
@@ -365,6 +376,7 @@ pub async fn do_delete_vectors(
                 params,
                 shard_key.clone(),
                 access.clone(),
+                hw_measurement_acc.clone(),
             )
             .await?,
         );
@@ -383,6 +395,7 @@ pub async fn do_delete_vectors(
                 params,
                 shard_key,
                 access,
+                hw_measurement_acc,
             )
             .await?,
         );
@@ -398,6 +411,7 @@ pub async fn do_set_payload(
     internal_params: InternalUpdateParams,
     params: UpdateParams,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let SetPayload {
         points,
@@ -423,6 +437,7 @@ pub async fn do_set_payload(
         params,
         shard_key,
         access,
+        hw_measurement_acc,
     )
     .await
 }
@@ -434,6 +449,7 @@ pub async fn do_overwrite_payload(
     internal_params: InternalUpdateParams,
     params: UpdateParams,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let SetPayload {
         points,
@@ -460,6 +476,7 @@ pub async fn do_overwrite_payload(
         params,
         shard_key,
         access,
+        hw_measurement_acc,
     )
     .await
 }
@@ -471,6 +488,7 @@ pub async fn do_delete_payload(
     internal_params: InternalUpdateParams,
     params: UpdateParams,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let DeletePayload {
         keys,
@@ -494,6 +512,7 @@ pub async fn do_delete_payload(
         params,
         shard_key,
         access,
+        hw_measurement_acc,
     )
     .await
 }
@@ -505,6 +524,7 @@ pub async fn do_clear_payload(
     internal_params: InternalUpdateParams,
     params: UpdateParams,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let (point_operation, shard_key) = match points {
         PointsSelector::PointIdsSelector(PointIdsList { points, shard_key }) => {
@@ -525,10 +545,12 @@ pub async fn do_clear_payload(
         params,
         shard_key,
         access,
+        hw_measurement_acc,
     )
     .await
 }
 
+#[expect(clippy::too_many_arguments)]
 pub async fn do_batch_update_points(
     toc: Arc<TableOfContent>,
     collection_name: String,
@@ -537,6 +559,7 @@ pub async fn do_batch_update_points(
     params: UpdateParams,
     access: Access,
     inference_token: InferenceToken,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<Vec<UpdateResult>, StorageError> {
     let mut results = Vec::with_capacity(operations.len());
     for operation in operations {
@@ -550,6 +573,7 @@ pub async fn do_batch_update_points(
                     params,
                     access.clone(),
                     inference_token.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -562,6 +586,7 @@ pub async fn do_batch_update_points(
                     params,
                     access.clone(),
                     inference_token.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -573,6 +598,7 @@ pub async fn do_batch_update_points(
                     internal_params,
                     params,
                     access.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -584,6 +610,7 @@ pub async fn do_batch_update_points(
                     internal_params,
                     params,
                     access.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -595,6 +622,7 @@ pub async fn do_batch_update_points(
                     internal_params,
                     params,
                     access.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -606,6 +634,7 @@ pub async fn do_batch_update_points(
                     internal_params,
                     params,
                     access.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -618,6 +647,7 @@ pub async fn do_batch_update_points(
                     params,
                     access.clone(),
                     inference_token.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -629,6 +659,7 @@ pub async fn do_batch_update_points(
                     internal_params,
                     params,
                     access.clone(),
+                    hw_measurement_acc.clone(),
                 )
                 .await
             }
@@ -645,6 +676,7 @@ pub async fn do_create_index(
     internal_params: InternalUpdateParams,
     params: UpdateParams,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     // TODO: Is this cancel safe!?
 
@@ -684,6 +716,7 @@ pub async fn do_create_index(
         Some(field_schema),
         internal_params,
         params,
+        hw_measurement_acc,
     )
     .await
 }
@@ -695,6 +728,7 @@ pub async fn do_create_index_internal(
     field_schema: Option<PayloadFieldSchema>,
     internal_params: InternalUpdateParams,
     params: UpdateParams,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let operation = CollectionUpdateOperations::FieldIndexOperation(
         FieldIndexOperations::CreateIndex(CreateIndex {
@@ -711,6 +745,7 @@ pub async fn do_create_index_internal(
         params,
         None,
         Access::full("Internal API"),
+        hw_measurement_acc,
     )
     .await
 }
@@ -722,6 +757,7 @@ pub async fn do_delete_index(
     internal_params: InternalUpdateParams,
     params: UpdateParams,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     // TODO: Is this cancel safe!?
 
@@ -743,7 +779,15 @@ pub async fn do_delete_index(
         .submit_collection_meta_op(consensus_op, access, wait_timeout)
         .await?;
 
-    do_delete_index_internal(toc, collection_name, index_name, internal_params, params).await
+    do_delete_index_internal(
+        toc,
+        collection_name,
+        index_name,
+        internal_params,
+        params,
+        hw_measurement_acc,
+    )
+    .await
 }
 
 pub async fn do_delete_index_internal(
@@ -752,6 +796,7 @@ pub async fn do_delete_index_internal(
     index_name: JsonPath,
     internal_params: InternalUpdateParams,
     params: UpdateParams,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let operation = CollectionUpdateOperations::FieldIndexOperation(
         FieldIndexOperations::DeleteIndex(index_name),
@@ -765,10 +810,12 @@ pub async fn do_delete_index_internal(
         params,
         None,
         Access::full("Internal API"),
+        hw_measurement_acc,
     )
     .await
 }
 
+#[expect(clippy::too_many_arguments)]
 pub async fn update(
     toc: &TableOfContent,
     collection_name: &str,
@@ -777,6 +824,7 @@ pub async fn update(
     params: UpdateParams,
     shard_key: Option<ShardKeySelector>,
     access: Access,
+    hw_measurement_acc: HwMeasurementAcc,
 ) -> Result<UpdateResult, StorageError> {
     let InternalUpdateParams {
         shard_id,
@@ -823,6 +871,7 @@ pub async fn update(
         ordering,
         shard_selector,
         access,
+        hw_measurement_acc,
     )
     .await
 }

src/tonic/api/points_api.rs

@@ -43,12 +43,17 @@ impl PointsService {
         }
     }
 
-    fn get_request_collection_hw_usage_counter(&self, collection_name: String) -> RequestHwCounter {
+    fn get_request_collection_hw_usage_counter(
+        &self,
+        collection_name: String,
+        wait: Option<bool>,
+    ) -> RequestHwCounter {
         let counter = HwMeasurementAcc::new_with_metrics_drain(
             self.dispatcher.get_collection_hw_metrics(collection_name),
         );
 
-        RequestHwCounter::new(counter, self.service_config.hardware_reporting())
+        let waiting = wait != Some(false);
+        RequestHwCounter::new(counter, self.service_config.hardware_reporting() && waiting)
     }
 }
 
@@ -63,12 +68,17 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let inference_token = extract_token(&request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         upsert(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
             inference_token,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -83,12 +93,17 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let inference_token = extract_token(&request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         delete(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
             inference_token,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -101,8 +116,8 @@ impl Points for PointsService {
 
         let inner_request = request.into_inner();
 
-        let hw_metrics =
-            self.get_request_collection_hw_usage_counter(inner_request.collection_name.clone());
+        let hw_metrics = self
+            .get_request_collection_hw_usage_counter(inner_request.collection_name.clone(), None);
 
         get(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -125,12 +140,17 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let inference_token = extract_token(&request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         update_vectors(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
             inference_token,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -144,11 +164,17 @@ impl Points for PointsService {
 
         let access = extract_access(&mut request);
 
+        let hw_metrics = self.get_request_collection_hw_usage_counter(
+            request.get_ref().collection_name.clone(),
+            None,
+        );
+
         delete_vectors(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -162,11 +188,16 @@ impl Points for PointsService {
 
         let access = extract_access(&mut request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         set_payload(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -180,11 +211,16 @@ impl Points for PointsService {
 
         let access = extract_access(&mut request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         overwrite_payload(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -198,11 +234,16 @@ impl Points for PointsService {
 
         let access = extract_access(&mut request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         delete_payload(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -216,11 +257,16 @@ impl Points for PointsService {
 
         let access = extract_access(&mut request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         clear_payload(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -235,12 +281,17 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let inference_token = extract_token(&request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         update_batch(
             &self.dispatcher,
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
             inference_token,
+            hw_metrics,
         )
         .await
     }
@@ -253,11 +304,16 @@ impl Points for PointsService {
 
         let access = extract_access(&mut request);
 
+        let collection_name = request.get_ref().collection_name.clone();
+        let wait = Some(request.get_ref().wait.unwrap_or(false));
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, wait);
+
         create_field_index(
             self.dispatcher.clone(),
             request.into_inner(),
             InternalUpdateParams::default(),
             access,
+            hw_metrics,
         )
         .await
         .map(|resp| resp.map(Into::into))
@@ -287,8 +343,9 @@ impl Points for PointsService {
     ) -> Result<Response<SearchResponse>, Status> {
         validate(request.get_ref())?;
         let access = extract_access(&mut request);
+
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
 
         let res = search(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -330,7 +387,8 @@ impl Points for PointsService {
             requests.push((core_search_request, shard_selector));
         }
 
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name.clone());
+        let hw_metrics =
+            self.get_request_collection_hw_usage_counter(collection_name.clone(), None);
 
         let res = core_search_batch(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -353,7 +411,7 @@ impl Points for PointsService {
         validate(request.get_ref())?;
         let access = extract_access(&mut request);
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
         let res = search_groups(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
@@ -376,8 +434,8 @@ impl Points for PointsService {
 
         let inner_request = request.into_inner();
 
-        let hw_metrics =
-            self.get_request_collection_hw_usage_counter(inner_request.collection_name.clone());
+        let hw_metrics = self
+            .get_request_collection_hw_usage_counter(inner_request.collection_name.clone(), None);
 
         scroll(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -396,7 +454,7 @@ impl Points for PointsService {
         validate(request.get_ref())?;
         let access = extract_access(&mut request);
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
         let res = recommend(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
@@ -421,7 +479,8 @@ impl Points for PointsService {
             timeout,
         } = request.into_inner();
 
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name.clone());
+        let hw_metrics =
+            self.get_request_collection_hw_usage_counter(collection_name.clone(), None);
 
         let res = recommend_batch(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -444,7 +503,7 @@ impl Points for PointsService {
         validate(request.get_ref())?;
         let access = extract_access(&mut request);
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
 
         let res = recommend_groups(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -465,7 +524,7 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let collection_name = request.get_ref().collection_name.clone();
 
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
         let res = discover(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
@@ -490,7 +549,8 @@ impl Points for PointsService {
             timeout,
         } = request.into_inner();
 
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name.clone());
+        let hw_metrics =
+            self.get_request_collection_hw_usage_counter(collection_name.clone(), None);
         let res = discover_batch(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             &collection_name,
@@ -513,7 +573,7 @@ impl Points for PointsService {
 
         let access = extract_access(&mut request);
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
         let res = count(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
@@ -534,7 +594,7 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let inference_token = extract_token(&request);
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
 
         let res = query(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -564,7 +624,8 @@ impl Points for PointsService {
             timeout,
         } = request;
         let timeout = timeout.map(Duration::from_secs);
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name.clone());
+        let hw_metrics =
+            self.get_request_collection_hw_usage_counter(collection_name.clone(), None);
         let res = query_batch(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             &collection_name,
@@ -587,7 +648,7 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let inference_token = extract_token(&request);
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
 
         let res = query_groups(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
@@ -607,8 +668,10 @@ impl Points for PointsService {
     ) -> Result<Response<FacetResponse>, Status> {
         validate(request.get_ref())?;
         let access = extract_access(&mut request);
-        let hw_metrics =
-            self.get_request_collection_hw_usage_counter(request.get_ref().collection_name.clone());
+        let hw_metrics = self.get_request_collection_hw_usage_counter(
+            request.get_ref().collection_name.clone(),
+            None,
+        );
         facet(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
@@ -626,7 +689,7 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let timing = Instant::now();
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
         let search_matrix_response = search_points_matrix(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),
@@ -652,7 +715,7 @@ impl Points for PointsService {
         let access = extract_access(&mut request);
         let timing = Instant::now();
         let collection_name = request.get_ref().collection_name.clone();
-        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name);
+        let hw_metrics = self.get_request_collection_hw_usage_counter(collection_name, None);
         let search_matrix_response = search_points_matrix(
             StrictModeCheckedTocProvider::new(&self.dispatcher),
             request.into_inner(),

src/tonic/api/points_internal_api.rs

@@ -148,7 +148,7 @@ async fn facet_counts_internal(
     let response = FacetResponseInternal {
         hits: hits.into_iter().map(From::from).collect_vec(),
         time: timing.elapsed().as_secs_f64(),
-        // TODO(io_measurement): add hw data
+        usage: request_hw_data.to_grpc_api(),
     };
 
     Ok(Response::new(response))
@@ -188,12 +188,19 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let upsert_points = extract_internal_request(upsert_points)?;
+
+        let hw_data = self.get_request_collection_hw_usage_counter_for_internal(
+            upsert_points.collection_name.clone(),
+        );
+
         upsert(
             StrictModeCheckedInternalTocProvider::new(&self.toc),
-            extract_internal_request(upsert_points)?,
+            upsert_points,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
             inference_token,
+            hw_data,
         )
         .await
     }
@@ -212,12 +219,19 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let delete_points = extract_internal_request(delete_points)?;
+
+        let hw_metrics = self.get_request_collection_hw_usage_counter_for_internal(
+            delete_points.collection_name.clone(),
+        );
+
         delete(
             UncheckedTocProvider::new_unchecked(&self.toc),
-            extract_internal_request(delete_points)?,
+            delete_points,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
             inference_token,
+            hw_metrics,
         )
         .await
     }
@@ -236,12 +250,19 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let update_point_vectors = extract_internal_request(update_vectors_req)?;
+
+        let hw_metrics = self.get_request_collection_hw_usage_counter_for_internal(
+            update_point_vectors.collection_name.clone(),
+        );
+
         update_vectors(
             StrictModeCheckedInternalTocProvider::new(&self.toc),
-            extract_internal_request(update_vectors_req)?,
+            update_point_vectors,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
             inference_token,
+            hw_metrics,
         )
         .await
     }
@@ -258,11 +279,18 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let delete_point_vectors = extract_internal_request(delete_vectors_req)?;
+
+        let hw_metrics = self.get_request_collection_hw_usage_counter_for_internal(
+            delete_point_vectors.collection_name.clone(),
+        );
+
         delete_vectors(
             UncheckedTocProvider::new_unchecked(&self.toc),
-            extract_internal_request(delete_vectors_req)?,
+            delete_point_vectors,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
+            hw_metrics,
         )
         .await
     }
@@ -279,11 +307,18 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let set_payload_points = extract_internal_request(set_payload_points)?;
+
+        let hw_metrics = self.get_request_collection_hw_usage_counter_for_internal(
+            set_payload_points.collection_name.clone(),
+        );
+
         set_payload(
             StrictModeCheckedInternalTocProvider::new(&self.toc),
-            extract_internal_request(set_payload_points)?,
+            set_payload_points,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
+            hw_metrics,
         )
         .await
     }
@@ -300,11 +335,18 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let set_payload_points = extract_internal_request(set_payload_points)?;
+
+        let hw_metrics = self.get_request_collection_hw_usage_counter_for_internal(
+            set_payload_points.collection_name.clone(),
+        );
+
         overwrite_payload(
             StrictModeCheckedInternalTocProvider::new(&self.toc),
-            extract_internal_request(set_payload_points)?,
+            set_payload_points,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
+            hw_metrics,
         )
         .await
     }
@@ -321,11 +363,18 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let delete_payload_points = extract_internal_request(delete_payload_points)?;
+
+        let hw_metrics = self.get_request_collection_hw_usage_counter_for_internal(
+            delete_payload_points.collection_name.clone(),
+        );
+
         delete_payload(
             UncheckedTocProvider::new_unchecked(&self.toc),
-            extract_internal_request(delete_payload_points)?,
+            delete_payload_points,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
+            hw_metrics,
         )
         .await
     }
@@ -342,11 +391,18 @@ impl PointsInternal for PointsInternalService {
             clock_tag,
         } = request.into_inner();
 
+        let clear_payload_points = extract_internal_request(clear_payload_points)?;
+
+        let hw_metrics = self.get_request_collection_hw_usage_counter_for_internal(
+            clear_payload_points.collection_name.clone(),
+        );
+
         clear_payload(
             UncheckedTocProvider::new_unchecked(&self.toc),
-            extract_internal_request(clear_payload_points)?,
+            clear_payload_points,
             InternalUpdateParams::from_grpc(shard_id, clock_tag),
             FULL_ACCESS.clone(),
+            hw_metrics,
         )
         .await
     }

src/tonic/api/update_common.rs

@@ -2,6 +2,7 @@ use std::sync::Arc;
 use std::time::Instant;
 
 use api::conversions::json::{json_path_from_proto, proto_to_payloads};
+use api::grpc::HardwareUsage;
 use api::grpc::qdrant::payload_index_params::IndexParams;
 use api::grpc::qdrant::points_update_operation::{ClearPayload, Operation, PointStructList};
 use api::grpc::qdrant::{
@@ -18,11 +19,13 @@ use collection::operations::conversions::try_points_selector_from_grpc;
 use collection::operations::payload_ops::DeletePayload;
 use collection::operations::point_ops::{self, PointOperations, PointSyncOperation};
 use collection::operations::vector_ops::DeleteVectors;
+use common::counter::hardware_accumulator::HwMeasurementAcc;
 use itertools::Itertools;
 use segment::types::{
     ExtendedPointId, Filter, PayloadFieldSchema, PayloadSchemaParams, PayloadSchemaType,
 };
 use storage::content_manager::toc::TableOfContent;
+use storage::content_manager::toc::request_hw_counter::RequestHwCounter;
 use storage::dispatcher::Dispatcher;
 use storage::rbac::Access;
 use tonic::{Response, Status};
@@ -39,6 +42,7 @@ pub async fn upsert(
     internal_params: InternalUpdateParams,
     access: Access,
     inference_token: InferenceToken,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let UpsertPoints {
         collection_name,
@@ -68,10 +72,12 @@ pub async fn upsert(
         UpdateParams::from_grpc(wait, ordering)?,
         access,
         inference_token,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -81,6 +87,7 @@ pub async fn delete(
     internal_params: InternalUpdateParams,
     access: Access,
     inference_token: InferenceToken,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let DeletePoints {
         collection_name,
@@ -108,10 +115,12 @@ pub async fn delete(
         UpdateParams::from_grpc(wait, ordering)?,
         access,
         inference_token,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -121,6 +130,7 @@ pub async fn update_vectors(
     internal_params: InternalUpdateParams,
     access: Access,
     inference_token: InferenceToken,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let UpdatePointVectors {
         collection_name,
@@ -162,10 +172,12 @@ pub async fn update_vectors(
         UpdateParams::from_grpc(wait, ordering)?,
         access,
         inference_token,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -174,6 +186,7 @@ pub async fn delete_vectors(
     delete_point_vectors: DeletePointVectors,
     internal_params: InternalUpdateParams,
     access: Access,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let DeletePointVectors {
         collection_name,
@@ -209,10 +222,12 @@ pub async fn delete_vectors(
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
         access,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -221,6 +236,7 @@ pub async fn set_payload(
     set_payload_points: SetPayloadPoints,
     internal_params: InternalUpdateParams,
     access: Access,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let SetPayloadPoints {
         collection_name,
@@ -254,10 +270,12 @@ pub async fn set_payload(
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
         access,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -266,6 +284,7 @@ pub async fn overwrite_payload(
     set_payload_points: SetPayloadPoints,
     internal_params: InternalUpdateParams,
     access: Access,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let SetPayloadPoints {
         collection_name,
@@ -299,10 +318,12 @@ pub async fn overwrite_payload(
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
         access,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -311,6 +332,7 @@ pub async fn delete_payload(
     delete_payload_points: DeletePayloadPoints,
     internal_params: InternalUpdateParams,
     access: Access,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let DeletePayloadPoints {
         collection_name,
@@ -342,10 +364,12 @@ pub async fn delete_payload(
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
         access,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -354,6 +378,7 @@ pub async fn clear_payload(
     clear_payload_points: ClearPayloadPoints,
     internal_params: InternalUpdateParams,
     access: Access,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let ClearPayloadPoints {
         collection_name,
@@ -380,10 +405,12 @@ pub async fn clear_payload(
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
         access,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -393,6 +420,7 @@ pub async fn update_batch(
     internal_params: InternalUpdateParams,
     access: Access,
     inference_token: InferenceToken,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<UpdateBatchResponse>, Status> {
     let UpdateBatchPoints {
         collection_name,
@@ -426,6 +454,7 @@ pub async fn update_batch(
                     internal_params,
                     access.clone(),
                     inference_token.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -442,6 +471,7 @@ pub async fn update_batch(
                     internal_params,
                     access.clone(),
                     inference_token.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -466,6 +496,7 @@ pub async fn update_batch(
                     },
                     internal_params,
                     access.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -491,6 +522,7 @@ pub async fn update_batch(
                     },
                     internal_params,
                     access.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -513,6 +545,7 @@ pub async fn update_batch(
                     },
                     internal_params,
                     access.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -531,6 +564,7 @@ pub async fn update_batch(
                     },
                     internal_params,
                     access.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -552,6 +586,7 @@ pub async fn update_batch(
                     internal_params,
                     access.clone(),
                     inference_token.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -574,6 +609,7 @@ pub async fn update_batch(
                     },
                     internal_params,
                     access.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -589,6 +625,7 @@ pub async fn update_batch(
                     },
                     internal_params,
                     access.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -608,6 +645,7 @@ pub async fn update_batch(
                     internal_params,
                     access.clone(),
                     inference_token.clone(),
+                    request_hw_counter.clone(),
                 )
                 .await
             }
@@ -628,6 +666,7 @@ pub async fn create_field_index(
     create_field_index_collection: CreateFieldIndexCollection,
     internal_params: InternalUpdateParams,
     access: Access,
+    request_hw_counter: RequestHwCounter,
 ) -> Result<Response<PointsOperationResponseInternal>, Status> {
     let CreateFieldIndexCollection {
         collection_name,
@@ -654,10 +693,12 @@ pub async fn create_field_index(
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
         access,
+        request_hw_counter.get_counter(),
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response =
+        points_operation_response_internal(timing, result, request_hw_counter.to_grpc_api());
     Ok(Response::new(response))
 }
 
@@ -686,10 +727,11 @@ pub async fn create_field_index_internal(
         field_schema,
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
+        HwMeasurementAcc::disposable(), // API unmeasured
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response = points_operation_response_internal(timing, result, None);
     Ok(Response::new(response))
 }
 
@@ -716,10 +758,11 @@ pub async fn delete_field_index(
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
         access,
+        HwMeasurementAcc::disposable(), // API unmeasured
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response = points_operation_response_internal(timing, result, None);
     Ok(Response::new(response))
 }
 
@@ -744,10 +787,11 @@ pub async fn delete_field_index_internal(
         field_name,
         internal_params,
         UpdateParams::from_grpc(wait, ordering)?,
+        HwMeasurementAcc::disposable(), // API unmeasured
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response = points_operation_response_internal(timing, result, None);
     Ok(Response::new(response))
 }
 
@@ -793,20 +837,24 @@ pub async fn sync(
         UpdateParams::from_grpc(wait, ordering)?,
         None,
         access,
+        HwMeasurementAcc::disposable(), // API unmeasured
     )
     .await?;
 
-    let response = points_operation_response_internal(timing, result);
+    let response = points_operation_response_internal(timing, result, None);
     Ok(Response::new(response))
 }
 
 pub fn points_operation_response_internal(
     timing: Instant,
     update_result: collection::operations::types::UpdateResult,
+    usage: Option<HardwareUsage>,
 ) -> PointsOperationResponseInternal {
     PointsOperationResponseInternal {
         result: Some(update_result.into()),
         time: timing.elapsed().as_secs_f64(),
+        usage,
+        // usage: Some(hw_measurement_acc.api)
     }
 }
 

tests/consensus_tests/assertions.py

@@ -1,3 +1,5 @@
+import itertools
+
 import requests
 
 
@@ -9,3 +11,18 @@ def assert_http_ok(response: requests.Response):
         else:
             raise Exception(
                 f"{base_msg} with response body:\n{response.json()}")
+
+
+def assert_hw_measurements_equal(left: dict[str, int], right: dict[str, int]):
+    keys = set([key for key in itertools.chain(left.keys(), right.keys())])
+    for key in keys:
+        if key in left and left[key] > 0:
+            assert right.get(key) == left[key]
+
+        if key in right and right[key] > 0:
+            assert left.get(key) == right[key]
+
+
+def assert_hw_measurements_equal_many(left_list: list[dict[str, int]], right_list: list[dict[str, int]]):
+    for left,right in zip(left_list, right_list):
+        assert_hw_measurements_equal(left, right)

tests/consensus_tests/fixtures.py

@@ -47,6 +47,23 @@ def upsert_points(
     )
 
 
+def update_points_payload(
+        peer_url,
+        points,
+        collection_name="test_collection",
+        wait="true",
+):
+    r_batch = requests.post(
+        f"{peer_url}/collections/{collection_name}/points/payload?wait={wait}",
+        json={
+            "points": points,
+            "payload": {"city": random.choice(CITIES)},
+        },
+    )
+    assert_http_ok(r_batch)
+    return r_batch.json()
+
+
 def upsert_random_points(
     peer_url,
     num,
@@ -187,3 +204,11 @@ def set_strict_mode(peer_id, collection_name, strict_mode_config):
             "strict_mode_config": strict_mode_config,
         },
     ).raise_for_status()
+
+
+def get_telemetry_hw_info(peer_url, collection):
+    r_search = requests.get(
+        f"{peer_url}/telemetry", params="details_level=3"
+    )
+    assert_http_ok(r_search)
+    return r_search.json()["result"]["hardware"]["collection_data"][collection]

tests/consensus_tests/test_hw_measurement.py

@@ -0,0 +1,100 @@
+import logging
+import pathlib
+
+from .fixtures import create_collection, upsert_random_points, get_telemetry_hw_info, update_points_payload
+from .utils import *
+from math import ceil
+
+logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger()
+
+N_PEERS = 4
+N_SHARDS = N_PEERS
+N_REPLICAS = 1
+COLLECTION_NAME = "test_collection_hw_counting"
+
+
+def test_measuring_hw_for_updates(tmp_path: pathlib.Path):
+    peer_urls, peer_dirs, bootstrap_url = start_cluster(tmp_path, N_PEERS)
+    create_collection(peer_urls[0], collection=COLLECTION_NAME, shard_number=N_SHARDS, replication_factor=N_REPLICAS)
+    wait_collection_exists_and_active_on_all_peers(collection_name=COLLECTION_NAME, peer_api_uris=peer_urls)
+    upsert_random_points(peer_urls[0], 200, collection_name=COLLECTION_NAME)
+
+    peer_hw_infos = [get_telemetry_hw_info(x, COLLECTION_NAME) for x in peer_urls]
+
+    # Check initial insertion IO measurements are reported in telemetry
+    for i in peer_hw_infos:
+        assert i["payload_io_write"] > 0
+        # TODO: Add tests for vector writes
+
+    # Upsert ~20 vectors into each shard
+    upsert_random_points(peer_urls[0], N_SHARDS * 20, collection_name=COLLECTION_NAME)
+
+    # Check upsert
+    for peer_idx in range(N_PEERS):
+        peer_url = peer_urls[peer_idx]
+        peer_hw = get_telemetry_hw_info(peer_url, COLLECTION_NAME)
+
+        # Assert that each nodes telemetry has been increased by some bytes
+        expected_delta = (19 * 5)  # ~20 vectors times avg. 5 bytes payload
+        assert abs(peer_hw["payload_io_write"] - peer_hw_infos[peer_idx]["payload_io_write"]) >= expected_delta
+        # TODO: also test for written vectors when implemented!
+
+    peer_hw_infos = [get_telemetry_hw_info(x, COLLECTION_NAME) for x in peer_urls]
+
+    total_payload_io_write_old = sum([x["payload_io_write"] for x in peer_hw_infos])
+
+    # Update 20 points
+    update_hw_data = update_points_payload(peer_urls[0], collection_name=COLLECTION_NAME, points=[x for x in range(N_PEERS*20)])["usage"]
+
+    total_payload_io_write = 0
+
+    # Check payload update
+    for peer_idx in range(N_PEERS):
+        peer_url = peer_urls[peer_idx]
+        peer_hw = get_telemetry_hw_info(peer_url, COLLECTION_NAME)
+
+        total_payload_io_write += peer_hw["payload_io_write"]
+
+        # Assert that each nodes telemetry has been increased by some bytes
+        assert abs(peer_hw["payload_io_write"] - peer_hw_infos[peer_idx]["payload_io_write"]) >= (19 * 5)
+
+    # Check that API response hardware data is equal to the data reported in telemetry!
+    assert update_hw_data['payload_io_write'] == total_payload_io_write - total_payload_io_write_old
+    # TODO: also test vector updates when implemented
+
+
+def test_measuring_hw_for_updates_without_waiting(tmp_path: pathlib.Path):
+    peer_urls, peer_dirs, bootstrap_url = start_cluster(tmp_path, N_PEERS)
+    create_collection(peer_urls[0], collection=COLLECTION_NAME, shard_number=N_SHARDS, replication_factor=N_REPLICAS)
+    wait_collection_exists_and_active_on_all_peers(collection_name=COLLECTION_NAME, peer_api_uris=peer_urls)
+
+    check_collection_points_count(peer_urls[0], COLLECTION_NAME, 0)
+
+    upsert_vectors = 50
+    total_vectors = upsert_vectors * N_PEERS  # 200 vectors
+
+    # Upsert 200 points without waiting
+    upsert_random_points(peer_urls[0], total_vectors, collection_name=COLLECTION_NAME, wait="false")
+
+    wait_collection_points_count(peer_urls[0], COLLECTION_NAME, total_vectors)
+
+    # Check metrics getting collected on each node, despite `wait=false`.
+    for peer_idx in range(N_PEERS):
+        peer_url = peer_urls[peer_idx]
+        peer_hw = get_telemetry_hw_info(peer_url, COLLECTION_NAME)
+
+        # Assert that each nodes telemetry has been increased by some bytes
+        assert peer_hw["payload_io_write"] >= upsert_vectors * 5  # 50 vectors on this node with payload of ~5 bytes
+
+    # TODO: also test vector updates when implemented
+
+def assert_with_upper_bound_error(inp: int, min_value: int, upper_bound_error_percent: float = 0.05):
+    """Asserts `inp` being equal to `min_value` with a max upperbound error given in percent."""
+    if inp < min_value:
+        assert False, f"Assertion {inp} >= {min_value} failed"
+
+    upper_bound = ceil(float(min_value) + float(min_value) * upper_bound_error_percent)
+
+    if inp > upper_bound:
+        assert False, f"Assertion {inp} being below upperbound error of {upper_bound_error_percent}(={upper_bound}) failed."

tests/consensus_tests/test_resharding.py

@@ -2,7 +2,8 @@ import pathlib
 from time import sleep
 from typing import Any, Literal
 
-from .fixtures import upsert_random_points, create_collection
+from .assertions import assert_hw_measurements_equal_many
+from .fixtures import upsert_random_points, create_collection, get_telemetry_hw_info
 from .utils import *
 
 
@@ -237,9 +238,16 @@ def test_resharding_transfer(tmp_path: pathlib.Path, direction: Literal["up", "d
         # Find replica of selected shard
         peer_id, peer_uri = find_replica(shard_id, info, peer_uris, peer_ids)
 
+        # Collect all nodes hardware measurements before transferring
+        hw = [get_telemetry_hw_info(uri, COLLECTION_NAME) for uri in peer_uris]
+
         # Migrate resharding points
         migrate_points(peer_uris[0], peer_id, shard_id, target_peer_id, target_shard_id, direction)
 
+        # Assert that no hardware measurements have been measured for the transfer
+        new_hw = [get_telemetry_hw_info(uri, COLLECTION_NAME) for uri in peer_uris]
+        assert_hw_measurements_equal_many(hw, new_hw)
+
         # Assert that all points were correctly migrated
         assert_resharding_points(peer_uri, shard_id, target_peer_uri, target_shard_id)
 

tests/consensus_tests/utils.py

@@ -79,6 +79,7 @@ def get_env(p2p_port: int, grpc_port: int, http_port: int) -> Dict[str, str]:
     env["QDRANT__SERVICE__HTTP_PORT"] = str(http_port)
     env["QDRANT__SERVICE__GRPC_PORT"] = str(grpc_port)
     env["QDRANT__LOG_LEVEL"] = "DEBUG,raft::raft=info"
+    env["QDRANT__SERVICE__HARDWARE_REPORTING"] = "true"
     return env