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

Request size is approximated by the network payload size for the commands.

4 files changed, 138 insertions, 13 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..a65ac32cd47 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.cpp
@@ -31,6 +31,34 @@ FileStorThreadMetrics::Op::clone(std::vector<Metric::UP>& ownerList,
     return (Op*) (new Op(getName(), _name, owner))->assignValues(*this);
 }
 
+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));
+}
+
+template class FileStorThreadMetrics::OpWithRequestSize<FileStorThreadMetrics::Op>;
+
 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 "
@@ -54,7 +82,7 @@ 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)
 { }
 
@@ -95,9 +123,9 @@ 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),
+      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),
@@ -245,6 +273,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..588033360ec 100644
--- a/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
+++ b/storage/src/vespa/storage/persistence/filestorage/filestormetrics.h
@@ -32,6 +32,19 @@ struct FileStorThreadMetrics : public metrics::MetricSet
                           MetricSet* owner, bool includeUnused) const override;
         Op* operator&() { return this; }
     };
+
+    template <typename BaseOp>
+    struct OpWithRequestSize : BaseOp {
+        metrics::LongAverageMetric request_size;
+
+        OpWithRequestSize(const std::string& id, const std::string& name, MetricSet* owner = 0);
+        ~OpWithRequestSize();
+
+        MetricSet * clone(std::vector<Metric::UP>& ownerList, CopyType copyType,
+                          MetricSet* owner, bool includeUnused) const override;
+        OpWithRequestSize* operator&() { return this; }
+    };
+
     struct OpWithNotFound : public Op {
         metrics::LongCountMetric notFound;
 
@@ -42,7 +55,7 @@ struct FileStorThreadMetrics : public metrics::MetricSet
         OpWithNotFound* operator&() { return this; }
     };
 
-    struct Update : public OpWithNotFound {
+    struct Update : public OpWithRequestSize<OpWithNotFound> {
         metrics::LongAverageMetric latencyRead;
 
         Update(MetricSet* owner = 0);
@@ -66,9 +79,9 @@ struct FileStorThreadMetrics : public metrics::MetricSet
 
     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;
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());