Merge pull request #19643 from vespa-engine/toregge/update-merge-latency-metrics-from-operation-complete-callbacks

Update merge handler put/remove latency metrics from operation complete callback.

6 files changed, 24 insertions, 23 deletions

diff --git a/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp b/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp
index 188b541c1b8..6e1e62d2080 100644
--- a/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp
+++ b/storage/src/tests/persistence/apply_bucket_diff_state_test.cpp
@@ -21,7 +21,6 @@ spi::Result spi_result_ok;
 spi::Result spi_result_fail(spi::Result::ErrorType::RESOURCE_EXHAUSTED, "write blocked");
 document::BucketIdFactory bucket_id_factory;
 const char *test_op = "put";
-metrics::DoubleAverageMetric dummy_metric("dummy", metrics::DoubleAverageMetric::Tags(), "dummy desc");
 document::Bucket dummy_document_bucket(makeDocumentBucket(document::BucketId(0, 16)));
 
 class DummyMergeBucketInfoSyncer : public MergeBucketInfoSyncer
@@ -42,10 +41,10 @@ public:
 ApplyBucketDiffEntryResult
 make_result(spi::Result &spi_result, const DocumentId &doc_id)
 {
-    std::promise<std::pair<std::unique_ptr<spi::Result>, double>> result_promise;
-    result_promise.set_value(std::make_pair(std::make_unique<spi::Result>(spi_result), 0.1));
+    std::promise<std::unique_ptr<spi::Result>> result_promise;
+    result_promise.set_value(std::make_unique<spi::Result>(spi_result));
     spi::Bucket bucket(makeDocumentBucket(bucket_id_factory.getBucketId(doc_id)));
-    return ApplyBucketDiffEntryResult(result_promise.get_future(), bucket, doc_id, test_op, dummy_metric);
+    return ApplyBucketDiffEntryResult(result_promise.get_future(), bucket, doc_id, test_op);
 }
 
 void push_ok(ApplyBucketDiffState &state)
diff --git a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.cpp b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.cpp
index 6d6d56351ec..034089d4eca 100644
--- a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.cpp
+++ b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.cpp
@@ -6,10 +6,11 @@
 
 namespace storage {
 
-ApplyBucketDiffEntryComplete::ApplyBucketDiffEntryComplete(ResultPromise result_promise, const framework::Clock& clock)
+ApplyBucketDiffEntryComplete::ApplyBucketDiffEntryComplete(ResultPromise result_promise, const framework::Clock& clock, metrics::DoubleAverageMetric& latency_metric)
     : _result_handler(nullptr),
       _result_promise(std::move(result_promise)),
-      _start_time(clock)
+      _start_time(clock),
+      _latency_metric(latency_metric)
 {
 }
 
@@ -21,7 +22,9 @@ ApplyBucketDiffEntryComplete::onComplete(std::unique_ptr<spi::Result> result)
     if (_result_handler != nullptr) {
         _result_handler->handle(*result);
     }
-    _result_promise.set_value(std::make_pair(std::move(result), _start_time.getElapsedTimeAsDouble()));
+    double elapsed = _start_time.getElapsedTimeAsDouble();
+    _latency_metric.addValue(elapsed);
+    _result_promise.set_value(std::move(result));
 }
 
 void
diff --git a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.h b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.h
index d1794666e54..6f05ed82fa0 100644
--- a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.h
+++ b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_complete.h
@@ -2,6 +2,7 @@
 
 #pragma once
 
+#include <vespa/metrics/valuemetric.h>
 #include <vespa/persistence/spi/operationcomplete.h>
 #include <vespa/storageframework/generic/clock/timer.h>
 #include <future>
@@ -14,12 +15,13 @@ namespace storage {
  */
 class ApplyBucketDiffEntryComplete : public spi::OperationComplete
 {
-    using ResultPromise = std::promise<std::pair<std::unique_ptr<spi::Result>, double>>;
+    using ResultPromise = std::promise<std::unique_ptr<spi::Result>>;
     const spi::ResultHandler*     _result_handler;
     ResultPromise                 _result_promise;
     framework::MilliSecTimer      _start_time;
+    metrics::DoubleAverageMetric& _latency_metric;
 public:
-    ApplyBucketDiffEntryComplete(ResultPromise result_promise, const framework::Clock& clock);
+    ApplyBucketDiffEntryComplete(ResultPromise result_promise, const framework::Clock& clock, metrics::DoubleAverageMetric& latency_metric);
     ~ApplyBucketDiffEntryComplete();
     void onComplete(std::unique_ptr<spi::Result> result) override;
     void addResultHandler(const spi::ResultHandler* resultHandler) override;
diff --git a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.cpp b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.cpp
index abd09d647a5..9cb5bb1bfd0 100644
--- a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.cpp
+++ b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.cpp
@@ -7,12 +7,11 @@
 
 namespace storage {
 
-ApplyBucketDiffEntryResult::ApplyBucketDiffEntryResult(FutureResult future_result, spi::Bucket bucket, document::DocumentId doc_id, const char *op, metrics::DoubleAverageMetric& latency_metric)
+ApplyBucketDiffEntryResult::ApplyBucketDiffEntryResult(FutureResult future_result, spi::Bucket bucket, document::DocumentId doc_id, const char *op)
     : _future_result(std::move(future_result)),
       _bucket(bucket),
       _doc_id(std::move(doc_id)),
-      _op(op),
-      _latency_metric(latency_metric)
+      _op(op)
 {
 }
 
@@ -32,15 +31,14 @@ ApplyBucketDiffEntryResult::check_result()
 {
     assert(_future_result.valid());
     auto result = _future_result.get();
-    if (result.first->hasError()) {
+    if (result->hasError()) {
         vespalib::asciistream ss;
         ss << "Failed " << _op
            << " for " << _doc_id.toString()
            << " in " << _bucket
-           << ": " << result.first->toString();
+           << ": " << result->toString();
         throw std::runtime_error(ss.str());
     }
-    _latency_metric.addValue(result.second);
 }
 
 }
diff --git a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.h b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.h
index f2213070831..662656f2dbc 100644
--- a/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.h
+++ b/storage/src/vespa/storage/persistence/apply_bucket_diff_entry_result.h
@@ -15,15 +15,14 @@ namespace storage {
  * Result of a bucket diff entry spi operation (putAsync or removeAsync)
  */
 class ApplyBucketDiffEntryResult {
-    using FutureResult = std::future<std::pair<std::unique_ptr<spi::Result>, double>>;
+    using FutureResult = std::future<std::unique_ptr<spi::Result>>;
     FutureResult         _future_result;
     spi::Bucket          _bucket;
     document::DocumentId _doc_id;
     const char*          _op;
-    metrics::DoubleAverageMetric& _latency_metric;
 
 public:
-    ApplyBucketDiffEntryResult(FutureResult future_result, spi::Bucket bucket, document::DocumentId doc_id, const char *op, metrics::DoubleAverageMetric& latency_metric);
+    ApplyBucketDiffEntryResult(FutureResult future_result, spi::Bucket bucket, document::DocumentId doc_id, const char *op);
     ApplyBucketDiffEntryResult(ApplyBucketDiffEntryResult &&rhs);
     ~ApplyBucketDiffEntryResult();
     void wait();
diff --git a/storage/src/vespa/storage/persistence/mergehandler.cpp b/storage/src/vespa/storage/persistence/mergehandler.cpp
index 963fddd9fb5..274aba303fb 100644
--- a/storage/src/vespa/storage/persistence/mergehandler.cpp
+++ b/storage/src/vespa/storage/persistence/mergehandler.cpp
@@ -489,21 +489,21 @@ MergeHandler::applyDiffEntry(const spi::Bucket& bucket,
                              spi::Context& context,
                              const document::DocumentTypeRepo& repo) const
 {
-    std::promise<std::pair<std::unique_ptr<spi::Result>, double>> result_promise;
+    std::promise<std::unique_ptr<spi::Result>> result_promise;
     auto future_result = result_promise.get_future();
     spi::Timestamp timestamp(e._entry._timestamp);
     if (!(e._entry._flags & (DELETED | DELETED_IN_PLACE))) {
         // Regular put entry
         Document::SP doc(deserializeDiffDocument(e, repo));
         DocumentId docId = doc->getId();
-        auto complete = std::make_unique<ApplyBucketDiffEntryComplete>(std::move(result_promise), _clock);
+        auto complete = std::make_unique<ApplyBucketDiffEntryComplete>(std::move(result_promise), _clock, _env._metrics.merge_handler_metrics.put_latency);
         _spi.putAsync(bucket, timestamp, std::move(doc), context, std::move(complete));
-        return ApplyBucketDiffEntryResult(std::move(future_result), bucket, std::move(docId), "put", _env._metrics.merge_handler_metrics.put_latency);
+        return ApplyBucketDiffEntryResult(std::move(future_result), bucket, std::move(docId), "put");
     } else {
         DocumentId docId(e._docName);
-        auto complete = std::make_unique<ApplyBucketDiffEntryComplete>(std::move(result_promise), _clock);
+        auto complete = std::make_unique<ApplyBucketDiffEntryComplete>(std::move(result_promise), _clock, _env._metrics.merge_handler_metrics.remove_latency);
         _spi.removeAsync(bucket, timestamp, docId, context, std::move(complete));
-        return ApplyBucketDiffEntryResult(std::move(future_result), bucket, std::move(docId), "remove", _env._metrics.merge_handler_metrics.remove_latency);
+        return ApplyBucketDiffEntryResult(std::move(future_result), bucket, std::move(docId), "remove");
     }
 }