Add explicit metric for diverging timestamps on updated documents

Easier to track across nodes than hunting for log warnings.

11 files changed, 95 insertions, 13 deletions

diff --git a/storage/src/tests/distributor/externaloperationhandlertest.cpp b/storage/src/tests/distributor/externaloperationhandlertest.cpp
index 54aca78d13d..a9956712679 100644
--- a/storage/src/tests/distributor/externaloperationhandlertest.cpp
+++ b/storage/src/tests/distributor/externaloperationhandlertest.cpp
@@ -62,12 +62,20 @@ class ExternalOperationHandlerTest : public CppUnit::TestFixture,
                 .safe_time_not_reached.getLongValue("count");
     }
 
+    int64_t safe_time_not_reached_metric_count(const metrics::LoadMetric<UpdateMetricSet>& metrics) const {
+        return metrics[documentapi::LoadType::DEFAULT].failures.safe_time_not_reached.getLongValue("count");
+    }
+
     int64_t concurrent_mutatations_metric_count(
             const metrics::LoadMetric<PersistenceOperationMetricSet>& metrics) const {
         return metrics[documentapi::LoadType::DEFAULT].failures
                 .concurrent_mutations.getLongValue("count");
     }
 
+    int64_t concurrent_mutatations_metric_count(const metrics::LoadMetric<UpdateMetricSet>& metrics) const {
+        return metrics[documentapi::LoadType::DEFAULT].failures.concurrent_mutations.getLongValue("count");
+    }
+
     void set_up_distributor_for_sequencing_test();
 
     const vespalib::string _dummy_id{"id:foo:testdoctype1::bar"};
diff --git a/storage/src/tests/distributor/twophaseupdateoperationtest.cpp b/storage/src/tests/distributor/twophaseupdateoperationtest.cpp
index f4b80f7961c..3c82931467e 100644
--- a/storage/src/tests/distributor/twophaseupdateoperationtest.cpp
+++ b/storage/src/tests/distributor/twophaseupdateoperationtest.cpp
@@ -204,7 +204,7 @@ TwoPhaseUpdateOperationTest::replyToMessage(
         api::ReturnCode::Result result)
 {
     std::shared_ptr<api::StorageMessage> msg2 = sender.commands.at(index);
-    UpdateCommand& updatec = dynamic_cast<UpdateCommand&>(*msg2);
+    auto& updatec = dynamic_cast<UpdateCommand&>(*msg2);
     std::unique_ptr<api::StorageReply> reply(updatec.makeReply());
     static_cast<api::UpdateReply*>(reply.get())->setOldTimestamp(oldTimestamp);
     reply->setResult(api::ReturnCode(result, ""));
@@ -222,7 +222,7 @@ TwoPhaseUpdateOperationTest::replyToPut(
         const std::string& traceMsg)
 {
     std::shared_ptr<api::StorageMessage> msg2 = sender.commands.at(index);
-    PutCommand& putc = dynamic_cast<PutCommand&>(*msg2);
+    auto& putc = dynamic_cast<PutCommand&>(*msg2);
     std::unique_ptr<api::StorageReply> reply(putc.makeReply());
     reply->setResult(api::ReturnCode(result, ""));
     if (!traceMsg.empty()) {
@@ -240,7 +240,7 @@ TwoPhaseUpdateOperationTest::replyToCreateBucket(
         api::ReturnCode::Result result)
 {
     std::shared_ptr<api::StorageMessage> msg2 = sender.commands.at(index);
-    CreateBucketCommand& putc = dynamic_cast<CreateBucketCommand&>(*msg2);
+    auto& putc = dynamic_cast<CreateBucketCommand&>(*msg2);
     std::unique_ptr<api::StorageReply> reply(putc.makeReply());
     reply->setResult(api::ReturnCode(result, ""));
     callback.receive(sender,
@@ -257,8 +257,7 @@ TwoPhaseUpdateOperationTest::replyToGet(
         api::ReturnCode::Result result,
         const std::string& traceMsg)
 {
-    const api::GetCommand& get(
-            static_cast<const api::GetCommand&>(*sender.commands.at(index)));
+    auto& get = static_cast<const api::GetCommand&>(*sender.commands.at(index));
     std::shared_ptr<api::StorageReply> reply;
 
     if (haveDocument) {
diff --git a/storage/src/tests/distributor/updateoperationtest.cpp b/storage/src/tests/distributor/updateoperationtest.cpp
index 9ce862f5db8..63e2f4f00c1 100644
--- a/storage/src/tests/distributor/updateoperationtest.cpp
+++ b/storage/src/tests/distributor/updateoperationtest.cpp
@@ -61,7 +61,7 @@ public:
     }
 
     void replyToMessage(UpdateOperation& callback, MessageSenderStub& sender, uint32_t index,
-                        uint64_t oldTimestamp, api::BucketInfo info = api::BucketInfo(2,4,6));
+                        uint64_t oldTimestamp, const api::BucketInfo& info = api::BucketInfo(2,4,6));
 
     std::shared_ptr<UpdateOperation>
     sendUpdate(const std::string& bucketState);
@@ -94,7 +94,7 @@ UpdateOperation_Test::sendUpdate(const std::string& bucketState)
 
 void
 UpdateOperation_Test::replyToMessage(UpdateOperation& callback, MessageSenderStub& sender, uint32_t index,
-                                     uint64_t oldTimestamp, api::BucketInfo info)
+                                     uint64_t oldTimestamp, const api::BucketInfo& info)
 {
     std::shared_ptr<api::StorageMessage> msg2  = sender.commands[index];
     UpdateCommand* updatec = dynamic_cast<UpdateCommand*>(msg2.get());
@@ -123,6 +123,9 @@ UpdateOperation_Test::testSimple()
             std::string("UpdateReply(doc:test:test, BucketId(0x0000000000000000), "
                         "timestamp 100, timestamp of updated doc: 90) ReturnCode(NONE)"),
             sender.getLastReply(true));
+
+    auto& metrics = getDistributor().getMetrics().updates[documentapi::LoadType::DEFAULT];
+    CPPUNIT_ASSERT_EQUAL(0UL, metrics.diverging_timestamp_updates.getValue());
 }
 
 void
@@ -168,6 +171,9 @@ UpdateOperation_Test::testMultiNode()
                     "node(idx=1,crc=0x2,docs=4/4,bytes=6/6,trusted=true,active=false,ready=false), "
                     "node(idx=0,crc=0x2,docs=4/4,bytes=6/6,trusted=true,active=false,ready=false)"),
             dumpBucket(_bId));
+
+    auto& metrics = getDistributor().getMetrics().updates[documentapi::LoadType::DEFAULT];
+    CPPUNIT_ASSERT_EQUAL(0UL, metrics.diverging_timestamp_updates.getValue());
 }
 
 void
@@ -188,5 +194,8 @@ UpdateOperation_Test::testMultiNodeInconsistentTimestamp()
                         "timestamp 100, timestamp of updated doc: 120 Was inconsistent "
                         "(best node 1)) ReturnCode(NONE)"),
             sender.getLastReply(true));
+
+    auto& metrics = getDistributor().getMetrics().updates[documentapi::LoadType::DEFAULT];
+    CPPUNIT_ASSERT_EQUAL(1UL, metrics.diverging_timestamp_updates.getValue());
 }
 
diff --git a/storage/src/vespa/storage/distributor/CMakeLists.txt b/storage/src/vespa/storage/distributor/CMakeLists.txt
index 2fd51433306..bd82c0e4d6e 100644
--- a/storage/src/vespa/storage/distributor/CMakeLists.txt
+++ b/storage/src/vespa/storage/distributor/CMakeLists.txt
@@ -34,6 +34,7 @@ vespa_add_library(storage_distributor
     statecheckers.cpp
     statusreporterdelegate.cpp
     throttlingoperationstarter.cpp
+    update_metric_set.cpp
     visitormetricsset.cpp
     $<TARGET_OBJECTS:storage_distributoroperation>
     $<TARGET_OBJECTS:storage_distributoroperationexternal>
diff --git a/storage/src/vespa/storage/distributor/distributormetricsset.cpp b/storage/src/vespa/storage/distributor/distributormetricsset.cpp
index b7725559b1d..927dc06182d 100644
--- a/storage/src/vespa/storage/distributor/distributormetricsset.cpp
+++ b/storage/src/vespa/storage/distributor/distributormetricsset.cpp
@@ -10,7 +10,7 @@ using metrics::MetricSet;
 DistributorMetricSet::DistributorMetricSet(const metrics::LoadTypeSet& lt)
     : MetricSet("distributor", {{"distributor"}}, ""),
       puts(lt, PersistenceOperationMetricSet("puts"), this),
-      updates(lt, PersistenceOperationMetricSet("updates"), this),
+      updates(lt, UpdateMetricSet(), this),
       update_puts(lt, PersistenceOperationMetricSet("update_puts"), this),
       update_gets(lt, PersistenceOperationMetricSet("update_gets"), this),
       removes(lt, PersistenceOperationMetricSet("removes"), this),
diff --git a/storage/src/vespa/storage/distributor/distributormetricsset.h b/storage/src/vespa/storage/distributor/distributormetricsset.h
index b6a429761a0..5a64027f500 100644
--- a/storage/src/vespa/storage/distributor/distributormetricsset.h
+++ b/storage/src/vespa/storage/distributor/distributormetricsset.h
@@ -2,6 +2,7 @@
 #pragma once
 
 #include "persistence_operation_metric_set.h"
+#include "update_metric_set.h"
 #include "visitormetricsset.h"
 #include <vespa/metrics/metrics.h>
 #include <vespa/documentapi/loadtypes/loadtypeset.h>
@@ -12,7 +13,7 @@ class DistributorMetricSet : public metrics::MetricSet
 {
 public:
     metrics::LoadMetric<PersistenceOperationMetricSet> puts;
-    metrics::LoadMetric<PersistenceOperationMetricSet> updates;
+    metrics::LoadMetric<UpdateMetricSet> updates;
     metrics::LoadMetric<PersistenceOperationMetricSet> update_puts;
     metrics::LoadMetric<PersistenceOperationMetricSet> update_gets;
     metrics::LoadMetric<PersistenceOperationMetricSet> removes;
diff --git a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
index 6efce913e70..f9787141a19 100644
--- a/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/external/twophaseupdateoperation.h
@@ -18,6 +18,8 @@ class UpdateCommand;
 class CreateBucketReply;
 }
 
+class UpdateMetricSet;
+
 namespace distributor {
 
 class DistributorBucketSpace;
@@ -120,7 +122,7 @@ private:
     void replyWithTasFailure(DistributorMessageSender& sender,
                              vespalib::stringref message);
 
-    PersistenceOperationMetricSet& _updateMetric;
+    UpdateMetricSet& _updateMetric;
     PersistenceOperationMetricSet& _putMetric;
     PersistenceOperationMetricSet& _getMetric;
     std::shared_ptr<api::UpdateCommand> _updateCmd;
diff --git a/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp b/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp
index f5e67708fe5..c8f28391ec0 100644
--- a/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp
+++ b/storage/src/vespa/storage/distributor/operations/external/updateoperation.cpp
@@ -20,14 +20,15 @@ using document::BucketSpace;
 UpdateOperation::UpdateOperation(DistributorComponent& manager,
                                  DistributorBucketSpace &bucketSpace,
                                  const std::shared_ptr<api::UpdateCommand> & msg,
-                                 PersistenceOperationMetricSet& metric)
+                                 UpdateMetricSet& metric)
     : Operation(),
       _trackerInstance(metric, std::make_shared<api::UpdateReply>(*msg),
                        manager, msg->getTimestamp()),
       _tracker(_trackerInstance),
       _msg(msg),
       _manager(manager),
-      _bucketSpace(bucketSpace)
+      _bucketSpace(bucketSpace),
+      _metrics(metric)
 {
 }
 
@@ -149,6 +150,7 @@ UpdateOperation::onReceive(DistributorMessageSender& sender,
                                      reply.getBucket().toString().c_str(),
                                      _results[i].oldTs, _results[i].nodeId,
                                      _results[goodNode].oldTs, _results[goodNode].nodeId);
+                        _metrics.diverging_timestamp_updates.inc();
 
                         replyToSend.setNodeWithNewestTimestamp(_results[goodNode].nodeId);
                         _newestTimestampLocation.first  = _results[goodNode].bucketId;
diff --git a/storage/src/vespa/storage/distributor/operations/external/updateoperation.h b/storage/src/vespa/storage/distributor/operations/external/updateoperation.h
index a23fd2ab876..cf432fa2305 100644
--- a/storage/src/vespa/storage/distributor/operations/external/updateoperation.h
+++ b/storage/src/vespa/storage/distributor/operations/external/updateoperation.h
@@ -15,6 +15,8 @@ class UpdateCommand;
 class CreateBucketReply;
 }
 
+class UpdateMetricSet;
+
 namespace distributor {
 
 class DistributorBucketSpace;
@@ -25,7 +27,7 @@ public:
     UpdateOperation(DistributorComponent& manager,
                     DistributorBucketSpace &bucketSpace,
                     const std::shared_ptr<api::UpdateCommand> & msg,
-                    PersistenceOperationMetricSet& metric);
+                    UpdateMetricSet& metric);
 
     void onStart(DistributorMessageSender& sender) override;
     const char* getName() const override { return "update"; };
@@ -59,6 +61,7 @@ private:
     };
 
     std::vector<OldTimestamp> _results;
+    UpdateMetricSet& _metrics;
 };
 
 }
diff --git a/storage/src/vespa/storage/distributor/update_metric_set.cpp b/storage/src/vespa/storage/distributor/update_metric_set.cpp
new file mode 100644
index 00000000000..5c343b395e6
--- /dev/null
+++ b/storage/src/vespa/storage/distributor/update_metric_set.cpp
@@ -0,0 +1,34 @@
+// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+
+#include "update_metric_set.h"
+#include <vespa/metrics/loadmetric.hpp>
+#include <vespa/metrics/summetric.hpp>
+
+namespace storage {
+
+using metrics::MetricSet;
+
+UpdateMetricSet::UpdateMetricSet(MetricSet* owner)
+    : PersistenceOperationMetricSet("updates", owner),
+      diverging_timestamp_updates("diverging_timestamp_updates", {},
+                                  "Number of updates that report they were performed against "
+                                  "divergent version timestamps on different replicas", this)
+{
+}
+
+UpdateMetricSet::~UpdateMetricSet() = default;
+
+MetricSet *
+UpdateMetricSet::clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
+                        MetricSet* owner, bool includeUnused) const
+{
+    if (copyType == INACTIVE) {
+        return MetricSet::clone(ownerList, INACTIVE, owner, includeUnused);
+    }
+    return static_cast<MetricSet*>((new UpdateMetricSet(owner))->assignValues(*this));
+}
+
+}
+
+template class metrics::LoadMetric<storage::UpdateMetricSet>;
+template class metrics::SumMetric<storage::UpdateMetricSet>;
diff --git a/storage/src/vespa/storage/distributor/update_metric_set.h b/storage/src/vespa/storage/distributor/update_metric_set.h
new file mode 100644
index 00000000000..de8474c949c
--- /dev/null
+++ b/storage/src/vespa/storage/distributor/update_metric_set.h
@@ -0,0 +1,23 @@
+// Copyright 2018 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+#pragma once
+
+#include "persistence_operation_metric_set.h"
+#include <vespa/documentapi/loadtypes/loadtypeset.h>
+#include <vespa/metrics/metrics.h>
+
+namespace storage {
+
+class UpdateMetricSet : public PersistenceOperationMetricSet {
+public:
+    metrics::LongCountMetric diverging_timestamp_updates;
+
+    explicit UpdateMetricSet(MetricSet* owner = nullptr);
+    ~UpdateMetricSet() override;
+
+    MetricSet* clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
+                     MetricSet* owner, bool includeUnused) const override;
+};
+
+} // storage
+
+