Add dedicated condition probe metrics for `PutOperation`/`RemoveOperation`

Follows the same pattern as that used for sub-operation metrics for
write-repair during Update processing.

16 files changed, 74 insertions, 38 deletions

diff --git a/storage/src/tests/distributor/check_condition_test.cpp b/storage/src/tests/distributor/check_condition_test.cpp
index 1b5cede8af6..ee8c9b888bb 100644
--- a/storage/src/tests/distributor/check_condition_test.cpp
+++ b/storage/src/tests/distributor/check_condition_test.cpp
@@ -253,4 +253,20 @@ TEST_F(CheckConditionTest, nested_get_traces_are_propagated_to_outcome) {
     });
 }
 
+TEST_F(CheckConditionTest, condition_evaluation_increments_probe_latency_metrics) {
+    getClock().setAbsoluteTimeInSeconds(1);
+    EXPECT_EQ(_metrics.latency.getLongValue("count"), 0);
+    EXPECT_EQ(_metrics.ok.getLongValue("last"), 0);
+    test_cond_with_2_gets_sent([&](auto& cond) {
+        cond.handle_reply(_sender, make_matched_reply(0));
+        getClock().setAbsoluteTimeInSeconds(3);
+        cond.handle_reply(_sender, make_matched_reply(1));
+    }, [&](auto& outcome) noexcept {
+        (void)outcome;
+    });
+    EXPECT_EQ(_metrics.latency.getLongValue("count"), 1);
+    EXPECT_EQ(_metrics.ok.getLongValue("last"), 1);
+    EXPECT_DOUBLE_EQ(_metrics.latency.getLast(), 2'000.0); // in millis
+}
+
 }
diff --git a/storage/src/tests/distributor/putoperationtest.cpp b/storage/src/tests/distributor/putoperationtest.cpp
index ff375e5b902..76b6741442e 100644
--- a/storage/src/tests/distributor/putoperationtest.cpp
+++ b/storage/src/tests/distributor/putoperationtest.cpp
@@ -73,7 +73,8 @@ public:
                                             operation_context(),
                                             getDistributorBucketSpace(),
                                             msg,
-                                            metrics().puts);
+                                            metrics().puts,
+                                            metrics().put_condition_probes);
         op->start(_sender);
     }
 
diff --git a/storage/src/tests/distributor/removeoperationtest.cpp b/storage/src/tests/distributor/removeoperationtest.cpp
index d352d23bb8c..d169c80a95d 100644
--- a/storage/src/tests/distributor/removeoperationtest.cpp
+++ b/storage/src/tests/distributor/removeoperationtest.cpp
@@ -41,7 +41,8 @@ struct RemoveOperationTest : Test, DistributorStripeTestUtil {
                 operation_context(),
                 getDistributorBucketSpace(),
                 msg,
-                metrics().removes);
+                metrics().removes,
+                metrics().remove_condition_probes);
 
         op->start(_sender);
     }
diff --git a/storage/src/vespa/storage/distributor/distributormetricsset.cpp b/storage/src/vespa/storage/distributor/distributormetricsset.cpp
index fad44782dd4..cbc0e6f6eef 100644
--- a/storage/src/vespa/storage/distributor/distributormetricsset.cpp
+++ b/storage/src/vespa/storage/distributor/distributormetricsset.cpp
@@ -16,11 +16,13 @@ BucketDbMetrics::~BucketDbMetrics() = default;
 DistributorMetricSet::DistributorMetricSet()
     : MetricSet("distributor", {{"distributor"}}, ""),
       puts("puts", this),
+      put_condition_probes("put_condition_probes", this),
       updates(this),
       update_puts("update_puts", this),
       update_gets("update_gets", this),
       update_metadata_gets("update_metadata_gets", this),
       removes("removes", this),
+      remove_condition_probes("remove_condition_probes", this),
       removelocations("removelocations", this),
       gets("gets", this),
       stats("stats", this),
diff --git a/storage/src/vespa/storage/distributor/distributormetricsset.h b/storage/src/vespa/storage/distributor/distributormetricsset.h
index ac140b85282..739e84759f1 100644
--- a/storage/src/vespa/storage/distributor/distributormetricsset.h
+++ b/storage/src/vespa/storage/distributor/distributormetricsset.h
@@ -20,24 +20,26 @@ struct BucketDbMetrics : metrics::MetricSet {
 class DistributorMetricSet : public metrics::MetricSet {
 public:
     PersistenceOperationMetricSet puts;
-    UpdateMetricSet updates;
+    PersistenceOperationMetricSet put_condition_probes;
+    UpdateMetricSet               updates;
     PersistenceOperationMetricSet update_puts;
     PersistenceOperationMetricSet update_gets;
     PersistenceOperationMetricSet update_metadata_gets;
     PersistenceOperationMetricSet removes;
+    PersistenceOperationMetricSet remove_condition_probes;
     PersistenceOperationMetricSet removelocations;
     PersistenceOperationMetricSet gets;
     PersistenceOperationMetricSet stats;
     PersistenceOperationMetricSet getbucketlists;
-    VisitorMetricSet visits;
-    metrics::DoubleAverageMetric stateTransitionTime;
-    metrics::DoubleAverageMetric set_cluster_state_processing_time;
-    metrics::DoubleAverageMetric activate_cluster_state_processing_time;
-    metrics::DoubleAverageMetric recoveryModeTime;
-    metrics::LongValueMetric docsStored;
-    metrics::LongValueMetric bytesStored;
-    BucketDbMetrics mutable_dbs;
-    BucketDbMetrics read_only_dbs;
+    VisitorMetricSet              visits;
+    metrics::DoubleAverageMetric  stateTransitionTime;
+    metrics::DoubleAverageMetric  set_cluster_state_processing_time;
+    metrics::DoubleAverageMetric  activate_cluster_state_processing_time;
+    metrics::DoubleAverageMetric  recoveryModeTime;
+    metrics::LongValueMetric      docsStored;
+    metrics::LongValueMetric      bytesStored;
+    BucketDbMetrics               mutable_dbs;
+    BucketDbMetrics               read_only_dbs;
 
     explicit DistributorMetricSet();
     ~DistributorMetricSet() override;
diff --git a/storage/src/vespa/storage/distributor/externaloperationhandler.cpp b/storage/src/vespa/storage/distributor/externaloperationhandler.cpp
index 6cb404aaa0a..d6bb5562a07 100644
--- a/storage/src/vespa/storage/distributor/externaloperationhandler.cpp
+++ b/storage/src/vespa/storage/distributor/externaloperationhandler.cpp
@@ -332,7 +332,9 @@ bool ExternalOperationHandler::onPut(const std::shared_ptr<api::PutCommand>& cmd
     if (allow) {
         _op = std::make_shared<PutOperation>(_node_ctx, _op_ctx,
                                              _op_ctx.bucket_space_repo().get(bucket_space),
-                                             std::move(cmd), getMetrics().puts, std::move(handle));
+                                             std::move(cmd),
+                                             getMetrics().puts, getMetrics().put_condition_probes,
+                                             std::move(handle));
     } else {
         _msg_sender.sendUp(makeConcurrentMutationRejectionReply(*cmd, cmd->getDocumentId(), metrics));
     }
@@ -386,7 +388,8 @@ bool ExternalOperationHandler::onRemove(const std::shared_ptr<api::RemoveCommand
         auto &distributorBucketSpace(_op_ctx.bucket_space_repo().get(bucket_space));
 
         _op = std::make_shared<RemoveOperation>(_node_ctx, _op_ctx, distributorBucketSpace, std::move(cmd),
-                                                getMetrics().removes, std::move(handle));
+                                                getMetrics().removes, getMetrics().remove_condition_probes,
+                                                std::move(handle));
     } else {
         _msg_sender.sendUp(makeConcurrentMutationRejectionReply(*cmd, cmd->getDocumentId(), metrics));
     }
diff --git a/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp b/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp
index 9f7dbcaa132..fc619d9eb23 100644
--- a/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/check_condition.cpp
@@ -58,7 +58,7 @@ CheckCondition::CheckCondition(const document::Bucket& bucket,
                                const documentapi::TestAndSetCondition& tas_condition,
                                const DistributorBucketSpace& bucket_space,
                                const DistributorNodeContext& node_ctx,
-                               PersistenceOperationMetricSet& metric,
+                               PersistenceOperationMetricSet& condition_probe_metrics,
                                uint32_t trace_level,
                                private_ctor_tag)
     : _doc_id_bucket(bucket),
@@ -66,7 +66,8 @@ CheckCondition::CheckCondition(const document::Bucket& bucket,
       _node_ctx(node_ctx),
       _cluster_state_version_at_creation_time(_bucket_space.getClusterState().getVersion()),
       _cond_get_op(),
-      _sent_message_map()
+      _sent_message_map(),
+      _outcome()
 {
     // Condition checks only return metadata back to the distributor and thus have an empty fieldset.
     // Side note: the BucketId provided to the GetCommand is ignored; GetOperation computes explicitly from the doc ID.
@@ -75,8 +76,8 @@ CheckCondition::CheckCondition(const document::Bucket& bucket,
     get_cmd->getTrace().setLevel(trace_level);
     _cond_get_op = std::make_shared<GetOperation>(_node_ctx, _bucket_space,
                                                   _bucket_space.getBucketDatabase().acquire_read_guard(),
-                                                  std::move(get_cmd),
-                                                  metric, api::InternalReadConsistency::Strong);
+                                                  std::move(get_cmd), condition_probe_metrics,
+                                                  api::InternalReadConsistency::Strong);
 }
 
 CheckCondition::~CheckCondition() = default;
@@ -220,7 +221,7 @@ CheckCondition::create_if_inconsistent_replicas(const document::Bucket& bucket,
                                                 const documentapi::TestAndSetCondition& tas_condition,
                                                 const DistributorNodeContext& node_ctx,
                                                 const DistributorStripeOperationContext& op_ctx,
-                                                PersistenceOperationMetricSet& metric,
+                                                PersistenceOperationMetricSet& condition_probe_metrics,
                                                 uint32_t trace_level)
 {
     // TODO move this check to the caller?
@@ -237,8 +238,8 @@ CheckCondition::create_if_inconsistent_replicas(const document::Bucket& bucket,
     if (!all_nodes_support_document_condition_probe(entries, op_ctx)) {
         return {}; // Want write-repair, but one or more nodes are too old to use the feature
     }
-    return std::make_shared<CheckCondition>(bucket, doc_id, tas_condition, bucket_space,
-                                            node_ctx, metric, trace_level, private_ctor_tag{});
+    return std::make_shared<CheckCondition>(bucket, doc_id, tas_condition, bucket_space, node_ctx,
+                                            condition_probe_metrics, trace_level, private_ctor_tag{});
 }
 
 }
diff --git a/storage/src/vespa/storage/distributor/operations/external/check_condition.h b/storage/src/vespa/storage/distributor/operations/external/check_condition.h
index 062c9bb831d..382aec6242c 100644
--- a/storage/src/vespa/storage/distributor/operations/external/check_condition.h
+++ b/storage/src/vespa/storage/distributor/operations/external/check_condition.h
@@ -114,7 +114,7 @@ public:
                    const documentapi::TestAndSetCondition& tas_condition,
                    const DistributorBucketSpace& bucket_space,
                    const DistributorNodeContext& node_ctx,
-                   PersistenceOperationMetricSet& metric,
+                   PersistenceOperationMetricSet& condition_probe_metrics,
                    uint32_t trace_level,
                    private_ctor_tag);
     ~CheckCondition();
@@ -135,7 +135,7 @@ public:
             const documentapi::TestAndSetCondition& tas_condition,
             const DistributorNodeContext& node_ctx,
             const DistributorStripeOperationContext& op_ctx,
-            PersistenceOperationMetricSet& metric,
+            PersistenceOperationMetricSet& condition_probe_metrics,
             uint32_t trace_level);
 private:
     [[nodiscard]] bool replica_set_changed_after_get_operation() const;
diff --git a/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp
index 952aeff0800..8c6fdb314f3 100644
--- a/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/putoperation.cpp
@@ -26,6 +26,7 @@ PutOperation::PutOperation(const DistributorNodeContext& node_ctx,
                            DistributorBucketSpace& bucket_space,
                            std::shared_ptr<api::PutCommand> msg,
                            PersistenceOperationMetricSet& metric,
+                           PersistenceOperationMetricSet& condition_probe_metrics,
                            SequencingHandle sequencing_handle)
     : SequencedOperation(std::move(sequencing_handle)),
       _tracker_instance(metric, std::make_shared<api::PutReply>(*msg), node_ctx, op_ctx, msg->getTimestamp()),
@@ -34,7 +35,7 @@ PutOperation::PutOperation(const DistributorNodeContext& node_ctx,
       _doc_id_bucket_id(document::BucketIdFactory{}.getBucketId(_msg->getDocumentId())),
       _node_ctx(node_ctx),
       _op_ctx(op_ctx),
-      _temp_metric(metric), // TODO
+      _condition_probe_metrics(condition_probe_metrics),
       _bucket_space(bucket_space)
 {
 }
@@ -156,7 +157,7 @@ void PutOperation::start_conditional_put(DistributorStripeMessageSender& sender)
     document::Bucket bucket(_msg->getBucket().getBucketSpace(), _doc_id_bucket_id);
     _check_condition = CheckCondition::create_if_inconsistent_replicas(bucket, _bucket_space, _msg->getDocumentId(),
                                                                        _msg->getCondition(), _node_ctx, _op_ctx,
-                                                                       _temp_metric, _msg->getTrace().getLevel());
+                                                                       _condition_probe_metrics, _msg->getTrace().getLevel());
     if (!_check_condition) {
         start_direct_put_dispatch(sender);
     } else {
diff --git a/storage/src/vespa/storage/distributor/operations/external/putoperation.h b/storage/src/vespa/storage/distributor/operations/external/putoperation.h
index 6befb8d3e38..635accc1865 100644
--- a/storage/src/vespa/storage/distributor/operations/external/putoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/external/putoperation.h
@@ -28,6 +28,7 @@ public:
                  DistributorBucketSpace& bucketSpace,
                  std::shared_ptr<api::PutCommand> msg,
                  PersistenceOperationMetricSet& metric,
+                 PersistenceOperationMetricSet& condition_probe_metrics,
                  SequencingHandle sequencingHandle = SequencingHandle());
     ~PutOperation() override;
 
@@ -44,7 +45,7 @@ private:
     document::BucketId                 _doc_id_bucket_id;
     const DistributorNodeContext&      _node_ctx;
     DistributorStripeOperationContext& _op_ctx;
-    PersistenceOperationMetricSet&     _temp_metric;
+    PersistenceOperationMetricSet&     _condition_probe_metrics;
     DistributorBucketSpace&            _bucket_space;
     std::shared_ptr<CheckCondition>    _check_condition;
 
diff --git a/storage/src/vespa/storage/distributor/operations/external/removeoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/removeoperation.cpp
index 59ae4120fd6..96182b0744f 100644
--- a/storage/src/vespa/storage/distributor/operations/external/removeoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/removeoperation.cpp
@@ -16,6 +16,7 @@ RemoveOperation::RemoveOperation(const DistributorNodeContext& node_ctx,
                                  DistributorBucketSpace& bucketSpace,
                                  std::shared_ptr<api::RemoveCommand> msg,
                                  PersistenceOperationMetricSet& metric,
+                                 PersistenceOperationMetricSet& condition_probe_metrics,
                                  SequencingHandle sequencingHandle)
     : SequencedOperation(std::move(sequencingHandle)),
       _tracker_instance(metric,
@@ -26,7 +27,7 @@ RemoveOperation::RemoveOperation(const DistributorNodeContext& node_ctx,
       _doc_id_bucket_id(document::BucketIdFactory{}.getBucketId(_msg->getDocumentId())),
       _node_ctx(node_ctx),
       _op_ctx(op_ctx),
-      _temp_metric(metric), // TODO
+      _condition_probe_metrics(condition_probe_metrics),
       _bucket_space(bucketSpace),
       _check_condition()
 {
@@ -48,7 +49,7 @@ void RemoveOperation::start_conditional_remove(DistributorStripeMessageSender& s
     document::Bucket bucket(_msg->getBucket().getBucketSpace(), _doc_id_bucket_id);
     _check_condition = CheckCondition::create_if_inconsistent_replicas(bucket, _bucket_space, _msg->getDocumentId(),
                                                                        _msg->getCondition(), _node_ctx, _op_ctx,
-                                                                       _temp_metric, _msg->getTrace().getLevel());
+                                                                       _condition_probe_metrics, _msg->getTrace().getLevel());
     if (!_check_condition) {
         start_direct_remove_dispatch(sender);
     } else {
diff --git a/storage/src/vespa/storage/distributor/operations/external/removeoperation.h b/storage/src/vespa/storage/distributor/operations/external/removeoperation.h
index 349a6182937..9f3a98294ea 100644
--- a/storage/src/vespa/storage/distributor/operations/external/removeoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/external/removeoperation.h
@@ -19,6 +19,7 @@ public:
                     DistributorBucketSpace& bucketSpace,
                     std::shared_ptr<api::RemoveCommand> msg,
                     PersistenceOperationMetricSet& metric,
+                    PersistenceOperationMetricSet& condition_probe_metrics,
                     SequencingHandle sequencingHandle = SequencingHandle());
     ~RemoveOperation() override;
 
@@ -36,7 +37,7 @@ private:
     document::BucketId                  _doc_id_bucket_id;
     const DistributorNodeContext&       _node_ctx;
     DistributorStripeOperationContext&  _op_ctx;
-    PersistenceOperationMetricSet&      _temp_metric;
+    PersistenceOperationMetricSet&      _condition_probe_metrics;
     DistributorBucketSpace&             _bucket_space;
     std::shared_ptr<CheckCondition>     _check_condition;
 
diff --git a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp
index 0cb4b223c11..73c65f54b21 100644
--- a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.cpp
@@ -34,6 +34,7 @@ TwoPhaseUpdateOperation::TwoPhaseUpdateOperation(
     : SequencedOperation(std::move(sequencingHandle)),
       _updateMetric(metrics.updates),
       _putMetric(metrics.update_puts),
+      _put_condition_probe_metrics(metrics.put_condition_probes), // Updates never trigger put write repair, so we sneakily use a ref to someone else
       _getMetric(metrics.update_gets),
       _metadata_get_metrics(metrics.update_metadata_gets),
       _updateCmd(std::move(msg)),
@@ -263,7 +264,7 @@ TwoPhaseUpdateOperation::schedulePutsWithUpdatedDocument(std::shared_ptr<documen
     document::Bucket bucket(_updateCmd->getBucket().getBucketSpace(), document::BucketId(0));
     auto put = std::make_shared<api::PutCommand>(bucket, doc, putTimestamp);
     copyMessageSettings(*_updateCmd, *put);
-    auto putOperation = std::make_shared<PutOperation>(_node_ctx, _op_ctx, _bucketSpace, std::move(put), _putMetric);
+    auto putOperation = std::make_shared<PutOperation>(_node_ctx, _op_ctx, _bucketSpace, std::move(put), _putMetric, _put_condition_probe_metrics);
     PutOperation & op = *putOperation;
     IntermediateMessageSender intermediate(_sentMessageMap, std::move(putOperation), sender);
     op.start(intermediate, _node_ctx.clock().getSystemTime());
diff --git a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
index 486ed766510..d2ad5359fa6 100644
--- a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
@@ -139,6 +139,7 @@ private:
 
     UpdateMetricSet&                    _updateMetric;
     PersistenceOperationMetricSet&      _putMetric;
+    PersistenceOperationMetricSet&      _put_condition_probe_metrics;
     PersistenceOperationMetricSet&      _getMetric;
     PersistenceOperationMetricSet&      _metadata_get_metrics;
     std::shared_ptr<api::UpdateCommand> _updateCmd;
diff --git a/storage/src/vespa/storage/distributor/persistence_operation_metric_set.cpp b/storage/src/vespa/storage/distributor/persistence_operation_metric_set.cpp
index 944b4bafa0a..e66884c4060 100644
--- a/storage/src/vespa/storage/distributor/persistence_operation_metric_set.cpp
+++ b/storage/src/vespa/storage/distributor/persistence_operation_metric_set.cpp
@@ -58,8 +58,8 @@ PersistenceFailuresMetricSet::clone(std::vector<Metric::UP>& ownerList, CopyType
     if (copyType == INACTIVE) {
         return MetricSet::clone(ownerList, INACTIVE, owner, includeUnused);
     }
-    return (PersistenceFailuresMetricSet*)
-            (new PersistenceFailuresMetricSet(owner))->assignValues(*this);
+    return dynamic_cast<PersistenceFailuresMetricSet*>(
+            (new PersistenceFailuresMetricSet(owner))->assignValues(*this));
 }
 
 PersistenceOperationMetricSet::PersistenceOperationMetricSet(const std::string& name, MetricSet* owner)
@@ -69,6 +69,11 @@ PersistenceOperationMetricSet::PersistenceOperationMetricSet(const std::string&
       failures(this)
 { }
 
+PersistenceOperationMetricSet::PersistenceOperationMetricSet(const std::string& name)
+    : PersistenceOperationMetricSet(name, nullptr)
+{
+}
+
 PersistenceOperationMetricSet::~PersistenceOperationMetricSet() = default;
 
 MetricSet *
@@ -78,9 +83,8 @@ PersistenceOperationMetricSet::clone(std::vector<Metric::UP>& ownerList, CopyTyp
     if (copyType == INACTIVE) {
         return MetricSet::clone(ownerList, INACTIVE, owner, includeUnused);
     }
-    return (PersistenceOperationMetricSet*)
-            (new PersistenceOperationMetricSet(getName(), owner))
-                ->assignValues(*this);
+    return dynamic_cast<PersistenceOperationMetricSet*>(
+            (new PersistenceOperationMetricSet(getName(), owner))->assignValues(*this));
 }
 
 void
diff --git a/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h b/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h
index b818d1bdd9f..eb1c3f57252 100644
--- a/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h
+++ b/storage/src/vespa/storage/distributor/persistence_operation_metric_set.h
@@ -40,10 +40,11 @@ class PersistenceOperationMetricSet : public metrics::MetricSet
     mutable std::mutex _mutex;
 public:
     metrics::DoubleAverageMetric latency;
-    metrics::LongCountMetric ok;
+    metrics::LongCountMetric     ok;
     PersistenceFailuresMetricSet failures;
 
-    PersistenceOperationMetricSet(const std::string& name, metrics::MetricSet* owner = nullptr);
+    PersistenceOperationMetricSet(const std::string& name, metrics::MetricSet* owner);
+    explicit PersistenceOperationMetricSet(const std::string& name);
     ~PersistenceOperationMetricSet() override;
 
     MetricSet * clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
@@ -57,7 +58,6 @@ public:
      */
     void updateFromResult(const api::ReturnCode& result);
 
-    friend class LockWrapper;
     class LockWrapper {
         std::unique_lock<std::mutex> _lock;
         PersistenceOperationMetricSet& _self;