Merge pull request #5581 from vespa-engine/vekterli/add-request-size-metric-to-operation-subset

 Add request size metric for Put, Update, Remove and Get commands

4 files changed, 163 insertions, 45 deletions

diff --git a/storage/src/tests/persistence/filestorage/filestormanagertest.cpp b/storage/src/tests/persistence/filestorage/filestormanagertest.cpp
index 3bdc622aa66..838df87662f 100644
--- a/storage/src/tests/persistence/filestorage/filestormanagertest.cpp
+++ b/storage/src/tests/persistence/filestorage/filestormanagertest.cpp
@@ -48,7 +48,11 @@ try{ \
 namespace storage {
 
 namespace {
-    spi::LoadType defaultLoadType(0, "default");
+
+spi::LoadType defaultLoadType(0, "default");
+
+struct TestFileStorComponents;
+
 }
 
 struct FileStorManagerTest : public CppUnit::TestFixture {
@@ -97,6 +101,10 @@ struct FileStorManagerTest : public CppUnit::TestFixture {
     void testStateChange();
     void testRepairNotifiesDistributorOnChange();
     void testDiskMove();
+    void put_command_size_is_added_to_metric();
+    void update_command_size_is_added_to_metric();
+    void remove_command_size_is_added_to_metric();
+    void get_command_size_is_added_to_metric();
 
     CPPUNIT_TEST_SUITE(FileStorManagerTest);
     CPPUNIT_TEST(testPut);
@@ -131,6 +139,10 @@ struct FileStorManagerTest : public CppUnit::TestFixture {
     CPPUNIT_TEST(testStateChange);
     CPPUNIT_TEST(testRepairNotifiesDistributorOnChange);
     CPPUNIT_TEST(testDiskMove);
+    CPPUNIT_TEST(put_command_size_is_added_to_metric);
+    CPPUNIT_TEST(update_command_size_is_added_to_metric);
+    CPPUNIT_TEST(remove_command_size_is_added_to_metric);
+    CPPUNIT_TEST(get_command_size_is_added_to_metric);
     CPPUNIT_TEST_SUITE_END();
 
     void createBucket(document::BucketId bid, uint16_t disk)
@@ -215,6 +227,15 @@ struct FileStorManagerTest : public CppUnit::TestFixture {
                 FileStorHandler& filestorHandler,
                 const document::BucketId& bucket,
                 uint32_t docNum);
+
+    template <typename Metric>
+    void assert_request_size_set(TestFileStorComponents& c,
+                                 std::shared_ptr<api::StorageMessage> cmd,
+                                 const Metric& metric);
+
+    auto& thread_metrics_of(FileStorManager& manager) {
+        return manager._metrics->disks[0]->threads[0];
+    }
 };
 
 CPPUNIT_TEST_SUITE_REGISTRATION(FileStorManagerTest);
@@ -2679,4 +2700,57 @@ FileStorManagerTest::testCreateBucketSetsActiveFlagInDatabaseAndReply()
     }
 }
 
+template <typename Metric>
+void FileStorManagerTest::assert_request_size_set(TestFileStorComponents& c, std::shared_ptr<api::StorageMessage> cmd, const Metric& metric) {
+    api::StorageMessageAddress address("storage", lib::NodeType::STORAGE, 3);
+    cmd->setApproxByteSize(54321);
+    cmd->setAddress(address);
+    c.top.sendDown(cmd);
+    c.top.waitForMessages(1, _waitTime);
+    CPPUNIT_ASSERT_EQUAL(static_cast<int64_t>(cmd->getApproxByteSize()), metric.request_size.getLast());
+}
+
+void FileStorManagerTest::put_command_size_is_added_to_metric() {
+    TestFileStorComponents c(*this, "put_command_size_is_added_to_metric");
+    document::BucketId bucket(16, 4000);
+    createBucket(bucket, 0);
+    auto cmd = std::make_shared<api::PutCommand>(
+            makeDocumentBucket(bucket), _node->getTestDocMan().createRandomDocument(), api::Timestamp(12345));
+
+    assert_request_size_set(c, std::move(cmd), thread_metrics_of(*c.manager)->put[defaultLoadType]);
+}
+
+void FileStorManagerTest::update_command_size_is_added_to_metric() {
+    TestFileStorComponents c(*this, "update_command_size_is_added_to_metric");
+    document::BucketId bucket(16, 4000);
+    createBucket(bucket, 0);
+    auto update = std::make_shared<document::DocumentUpdate>(
+            _node->getTestDocMan().createRandomDocument()->getType(),
+            document::DocumentId("id:foo:testdoctype1::bar"));
+    auto cmd = std::make_shared<api::UpdateCommand>(
+            makeDocumentBucket(bucket), std::move(update), api::Timestamp(123456));
+
+    assert_request_size_set(c, std::move(cmd), thread_metrics_of(*c.manager)->update[defaultLoadType]);
+}
+
+void FileStorManagerTest::remove_command_size_is_added_to_metric() {
+    TestFileStorComponents c(*this, "remove_command_size_is_added_to_metric");
+    document::BucketId bucket(16, 4000);
+    createBucket(bucket, 0);
+    auto cmd = std::make_shared<api::RemoveCommand>(
+            makeDocumentBucket(bucket), document::DocumentId("id:foo:testdoctype1::bar"), api::Timestamp(123456));
+
+    assert_request_size_set(c, std::move(cmd), thread_metrics_of(*c.manager)->remove[defaultLoadType]);
+}
+
+void FileStorManagerTest::get_command_size_is_added_to_metric() {
+    TestFileStorComponents c(*this, "get_command_size_is_added_to_metric");
+    document::BucketId bucket(16, 4000);
+    createBucket(bucket, 0);
+    auto cmd = std::make_shared<api::GetCommand>(
+            makeDocumentBucket(bucket), document::DocumentId("id:foo:testdoctype1::bar"), "[all]");
+
+    assert_request_size_set(c, std::move(cmd), thread_metrics_of(*c.manager)->get[defaultLoadType]);
+}
+
 } // storage
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
index 2efae7720e4..1293b5d1808 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
@@ -9,7 +9,7 @@ namespace storage {
 using metrics::MetricSet;
 using metrics::LoadTypeSet;
 
-FileStorThreadMetrics::Op::Op(const std::string& id, const std::string name, MetricSet* owner)
+FileStorThreadMetrics::Op::Op(const std::string& id, const std::string& name, MetricSet* owner)
     : MetricSet(id, id, name + " load in filestor thread", owner, "operationtype"),
       _name(name),
       count("count", "yamasdefault", "Number of requests processed.", this),
@@ -17,7 +17,7 @@ FileStorThreadMetrics::Op::Op(const std::string& id, const std::string name, Met
       failed("failed", "yamasdefault", "Number of failed requests.", this)
 { }
 
-FileStorThreadMetrics::Op::~Op() { }
+FileStorThreadMetrics::Op::~Op() = default;
 
 MetricSet *
 FileStorThreadMetrics::Op::clone(std::vector<Metric::UP>& ownerList,
@@ -31,13 +31,39 @@ FileStorThreadMetrics::Op::clone(std::vector<Metric::UP>& ownerList,
     return (Op*) (new Op(getName(), _name, owner))->assignValues(*this);
 }
 
-FileStorThreadMetrics::OpWithNotFound::OpWithNotFound(const std::string& id, const std::string name, MetricSet* owner)
+template <typename BaseOp>
+FileStorThreadMetrics::OpWithRequestSize<BaseOp>::OpWithRequestSize(const std::string& id, const std::string& name, MetricSet* owner)
+        : BaseOp(id, name, owner),
+          request_size("request_size", "", "Size of requests, in bytes", this)
+{
+}
+
+template <typename BaseOp>
+FileStorThreadMetrics::OpWithRequestSize<BaseOp>::~OpWithRequestSize() = default;
+
+// FIXME this has very non-intuitive semantics, ending up with copy&paste patterns
+template <typename BaseOp>
+MetricSet*
+FileStorThreadMetrics::OpWithRequestSize<BaseOp>::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<OpWithRequestSize<BaseOp>*>((new OpWithRequestSize<BaseOp>(this->getName(), this->_name, owner))
+            ->assignValues(*this));
+}
+
+FileStorThreadMetrics::OpWithNotFound::OpWithNotFound(const std::string& id, const std::string& name, MetricSet* owner)
     : Op(id, name, owner),
       notFound("not_found", "", "Number of requests that could not be "
                "completed due to source document not found.", this)
 { }
 
-FileStorThreadMetrics::OpWithNotFound::~OpWithNotFound() { }
+FileStorThreadMetrics::OpWithNotFound::~OpWithNotFound() = default;
 
 MetricSet *
 FileStorThreadMetrics::OpWithNotFound::clone(std::vector<Metric::UP>& ownerList,
@@ -54,11 +80,11 @@ FileStorThreadMetrics::OpWithNotFound::clone(std::vector<Metric::UP>& ownerList,
 }
 
 FileStorThreadMetrics::Update::Update(MetricSet* owner)
-    : OpWithNotFound("update", "Update", owner),
+    : OpWithRequestSize("update", "Update", owner),
       latencyRead("latency_read", "", "Latency of the source read in the request.", this)
 { }
 
-FileStorThreadMetrics::Update::~Update() { }
+FileStorThreadMetrics::Update::~Update() = default;
 
 MetricSet *
 FileStorThreadMetrics::Update::clone(std::vector<Metric::UP>& ownerList,
@@ -77,7 +103,7 @@ FileStorThreadMetrics::Visitor::Visitor(MetricSet* owner)
       documentsPerIterate("docs", "", "Number of entries read per iterate call", this)
 { }
 
-FileStorThreadMetrics::Visitor::~Visitor() { }
+FileStorThreadMetrics::Visitor::~Visitor() = default;
 
 MetricSet *
 FileStorThreadMetrics::Visitor::clone(std::vector<Metric::UP>& ownerList,
@@ -95,16 +121,16 @@ FileStorThreadMetrics::FileStorThreadMetrics(const std::string& name, const std:
     : MetricSet(name, "filestor partofsum thread", desc, nullptr, "thread"),
       operations("operations", "", "Number of operations processed.", this),
       failedOperations("failedoperations", "", "Number of operations throwing exceptions.", this),
-      put(lt, *&Op("put", "Put"), this),
-      get(lt, *&OpWithNotFound("get", "Get"), this),
-      remove(lt, *&OpWithNotFound("remove", "Remove"), this),
-      removeLocation(lt, *&Op("remove_location", "Remove location"), this),
-      statBucket(lt, *&Op("stat_bucket", "Stat bucket"), this),
-      update(lt, *&Update(), this),
-      revert(lt, *&OpWithNotFound("revert", "Revert"), this),
+      put(lt, OpWithRequestSize<Op>("put", "Put"), this),
+      get(lt, OpWithRequestSize<OpWithNotFound>("get", "Get"), this),
+      remove(lt, OpWithRequestSize<OpWithNotFound>("remove", "Remove"), this),
+      removeLocation(lt, Op("remove_location", "Remove location"), this),
+      statBucket(lt, Op("stat_bucket", "Stat bucket"), this),
+      update(lt, Update(), this),
+      revert(lt, OpWithNotFound("revert", "Revert"), this),
       createIterator("createiterator", "", this),
-      visit(lt, *&Visitor(), this),
-      multiOp(lt, *&Op("multioperations", "The number of multioperations that have been created"), this),
+      visit(lt, Visitor(), this),
+      multiOp(lt, Op("multioperations", "The number of multioperations that have been created"), this),
       createBuckets("createbuckets", "Number of buckets that has been created.", this),
       deleteBuckets("deletebuckets", "Number of buckets that has been deleted.", this),
       repairs("bucketverified", "Number of times buckets have been checked.", this),
@@ -245,6 +271,8 @@ template class metrics::LoadMetric<storage::FileStorThreadMetrics::Op>;
 template class metrics::LoadMetric<storage::FileStorThreadMetrics::OpWithNotFound>;
 template class metrics::LoadMetric<storage::FileStorThreadMetrics::Update>;
 template class metrics::LoadMetric<storage::FileStorThreadMetrics::Visitor>;
+template class metrics::LoadMetric<storage::FileStorThreadMetrics::OpWithRequestSize<storage::FileStorThreadMetrics::Op>>;
+template class metrics::LoadMetric<storage::FileStorThreadMetrics::OpWithRequestSize<storage::FileStorThreadMetrics::OpWithNotFound>>;
 template class metrics::SumMetric<storage::FileStorThreadMetrics::Op>;
 template class metrics::SumMetric<storage::FileStorThreadMetrics::OpWithNotFound>;
 template class metrics::SumMetric<storage::FileStorThreadMetrics::Update>;
diff --git a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
index c2fb579183b..3bc7fa33660 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
@@ -19,56 +19,64 @@ struct FileStorThreadMetrics : public metrics::MetricSet
 {
     typedef std::shared_ptr<FileStorThreadMetrics> SP;
 
-    struct Op : public metrics::MetricSet {
+    struct Op : metrics::MetricSet {
         std::string _name;
         metrics::LongCountMetric count;
         metrics::DoubleAverageMetric latency;
         metrics::LongCountMetric failed;
 
-        Op(const std::string& id, const std::string name, MetricSet* owner = 0);
-        ~Op();
+        Op(const std::string& id, const std::string& name, MetricSet* owner = nullptr);
+        ~Op() override;
 
         MetricSet * clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
                           MetricSet* owner, bool includeUnused) const override;
-        Op* operator&() { return this; }
     };
-    struct OpWithNotFound : public Op {
+
+    template <typename BaseOp>
+    struct OpWithRequestSize : BaseOp {
+        metrics::LongAverageMetric request_size;
+
+        OpWithRequestSize(const std::string& id, const std::string& name, MetricSet* owner = nullptr);
+        ~OpWithRequestSize() override;
+
+        MetricSet * clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
+                          MetricSet* owner, bool includeUnused) const override;
+    };
+
+    struct OpWithNotFound : Op {
         metrics::LongCountMetric notFound;
 
-        OpWithNotFound(const std::string& id, const std::string name, metrics::MetricSet* owner = 0);
-        ~OpWithNotFound();
+        OpWithNotFound(const std::string& id, const std::string& name, metrics::MetricSet* owner = nullptr);
+        ~OpWithNotFound() override;
         MetricSet* clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
                          MetricSet* owner, bool includeUnused) const override;
-        OpWithNotFound* operator&() { return this; }
     };
 
-    struct Update : public OpWithNotFound {
+    struct Update : OpWithRequestSize<OpWithNotFound> {
         metrics::LongAverageMetric latencyRead;
 
-        Update(MetricSet* owner = 0);
-        ~Update();
+        explicit Update(MetricSet* owner = nullptr);
+        ~Update() override;
 
         MetricSet* clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
                          MetricSet* owner, bool includeUnused) const override;
-        Update* operator&() { return this; }
     };
 
-    struct Visitor : public Op {
+    struct Visitor : Op {
         metrics::LongAverageMetric documentsPerIterate;
 
-        Visitor(MetricSet* owner = 0);
-        ~Visitor();
+        explicit Visitor(MetricSet* owner = nullptr);
+        ~Visitor() override;
 
         MetricSet * clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
                          MetricSet* owner, bool includeUnused) const override;
-        Visitor* operator&() { return this; }
     };
 
     metrics::LongCountMetric operations;
     metrics::LongCountMetric failedOperations;
-    metrics::LoadMetric<Op> put;
-    metrics::LoadMetric<OpWithNotFound> get;
-    metrics::LoadMetric<OpWithNotFound> remove;
+    metrics::LoadMetric<OpWithRequestSize<Op>> put;
+    metrics::LoadMetric<OpWithRequestSize<OpWithNotFound>> get;
+    metrics::LoadMetric<OpWithRequestSize<OpWithNotFound>> remove;
     metrics::LoadMetric<Op> removeLocation;
     metrics::LoadMetric<Op> statBucket;
     metrics::LoadMetric<Update> update;
@@ -103,7 +111,7 @@ struct FileStorThreadMetrics : public metrics::MetricSet
     metrics::LongAverageMetric batchingSize;
 
     FileStorThreadMetrics(const std::string& name, const std::string& desc, const metrics::LoadTypeSet& lt);
-    ~FileStorThreadMetrics();
+    ~FileStorThreadMetrics() override;
 };
 
 class FileStorStripeMetrics : public metrics::MetricSet
@@ -113,7 +121,7 @@ public:
     metrics::LoadMetric<metrics::DoubleAverageMetric> averageQueueWaitingTime;
     FileStorStripeMetrics(const std::string& name, const std::string& description,
                           const metrics::LoadTypeSet& loadTypes);
-    ~FileStorStripeMetrics();
+    ~FileStorStripeMetrics() override;
 };
 
 class FileStorDiskMetrics : public metrics::MetricSet
@@ -133,7 +141,7 @@ public:
 
     FileStorDiskMetrics(const std::string& name, const std::string& description,
                         const metrics::LoadTypeSet& loadTypes, MetricSet* owner);
-    ~FileStorDiskMetrics();
+    ~FileStorDiskMetrics() override;
 
     void initDiskMetrics(const metrics::LoadTypeSet& loadTypes, uint32_t numStripes, uint32_t threadsPerDisk);
 };
@@ -146,8 +154,8 @@ struct FileStorMetrics : public metrics::MetricSet
     metrics::LongCountMetric partitionEvents;
     metrics::LongCountMetric diskEvents;
 
-    FileStorMetrics(const metrics::LoadTypeSet&);
-    ~FileStorMetrics();
+    explicit FileStorMetrics(const metrics::LoadTypeSet&);
+    ~FileStorMetrics() override;
 
     void initDiskMetrics(uint16_t numDisks, const metrics::LoadTypeSet& loadTypes, uint32_t numStripes, uint32_t threadsPerDisk);
 };
diff --git a/storage/src/vespa/storage/persistence/persistencethread.cpp b/storage/src/vespa/storage/persistence/persistencethread.cpp
index fe85b55fe2b..768e104e02b 100644
--- a/storage/src/vespa/storage/persistence/persistencethread.cpp
+++ b/storage/src/vespa/storage/persistence/persistencethread.cpp
@@ -105,7 +105,9 @@ bool PersistenceThread::tasConditionMatches(const api::TestAndSetCommand & cmd,
 MessageTracker::UP
 PersistenceThread::handlePut(api::PutCommand& cmd)
 {
-    auto tracker = std::make_unique<MessageTracker>(_env._metrics.put[cmd.getLoadType()],_env._component.getClock());
+    auto& metrics = _env._metrics.put[cmd.getLoadType()];
+    auto tracker = std::make_unique<MessageTracker>(metrics, _env._component.getClock());
+    metrics.request_size.addValue(cmd.getApproxByteSize());
 
     if (tasConditionExists(cmd) && !tasConditionMatches(cmd, *tracker)) {
         return tracker;
@@ -120,7 +122,9 @@ PersistenceThread::handlePut(api::PutCommand& cmd)
 MessageTracker::UP
 PersistenceThread::handleRemove(api::RemoveCommand& cmd)
 {
-    auto tracker = std::make_unique<MessageTracker>(_env._metrics.remove[cmd.getLoadType()],_env._component.getClock());
+    auto& metrics = _env._metrics.remove[cmd.getLoadType()];
+    auto tracker = std::make_unique<MessageTracker>(metrics,_env._component.getClock());
+    metrics.request_size.addValue(cmd.getApproxByteSize());
 
     if (tasConditionExists(cmd) && !tasConditionMatches(cmd, *tracker)) {
         return tracker;
@@ -140,7 +144,9 @@ PersistenceThread::handleRemove(api::RemoveCommand& cmd)
 MessageTracker::UP
 PersistenceThread::handleUpdate(api::UpdateCommand& cmd)
 {
-    auto tracker = std::make_unique<MessageTracker>(_env._metrics.update[cmd.getLoadType()],_env._component.getClock());
+    auto& metrics = _env._metrics.update[cmd.getLoadType()];
+    auto tracker = std::make_unique<MessageTracker>(metrics, _env._component.getClock());
+    metrics.request_size.addValue(cmd.getApproxByteSize());
 
     if (tasConditionExists(cmd) && !tasConditionMatches(cmd, *tracker)) {
         return tracker;
@@ -159,7 +165,9 @@ PersistenceThread::handleUpdate(api::UpdateCommand& cmd)
 MessageTracker::UP
 PersistenceThread::handleGet(api::GetCommand& cmd)
 {
-    auto tracker = std::make_unique<MessageTracker>(_env._metrics.get[cmd.getLoadType()],_env._component.getClock());
+    auto& metrics = _env._metrics.get[cmd.getLoadType()];
+    auto tracker = std::make_unique<MessageTracker>(metrics,_env._component.getClock());
+    metrics.request_size.addValue(cmd.getApproxByteSize());
 
     document::FieldSetRepo repo;
     document::FieldSet::UP fieldSet = repo.parse(*_env._component.getTypeRepo(), cmd.getFieldSet());