Update metrics for each log message similar to legacy forwarder.

3 files changed, 34 insertions, 4 deletions

diff --git a/logd/src/logd/rpc_forwarder.cpp b/logd/src/logd/rpc_forwarder.cpp
index 4f637a34836..46dd47493b5 100644
--- a/logd/src/logd/rpc_forwarder.cpp
+++ b/logd/src/logd/rpc_forwarder.cpp
@@ -1,6 +1,7 @@
 // Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include "exceptions.h"
+#include "metrics.h"
 #include "proto_converter.h"
 #include "rpc_forwarder.h"
 #include <vespa/log/exceptions.h>
@@ -16,9 +17,10 @@ using vespalib::make_string;
 
 namespace logdemon {
 
-RpcForwarder::RpcForwarder(const vespalib::string &hostname, int rpc_port,
+RpcForwarder::RpcForwarder(Metrics& metrics, const vespalib::string &hostname, int rpc_port,
                            double rpc_timeout_secs, size_t max_messages_per_request)
-    : _connection_spec(make_string("tcp/%s:%d", hostname.c_str(), rpc_port)),
+    : _metrics(metrics),
+      _connection_spec(make_string("tcp/%s:%d", hostname.c_str(), rpc_port)),
       _rpc_timeout_secs(rpc_timeout_secs),
       _max_messages_per_request(max_messages_per_request),
       _supervisor(),
@@ -73,6 +75,7 @@ RpcForwarder::forwardLine(std::string_view line)
         ++_bad_lines;
         return;
     }
+    _metrics.countLine(ns_log::Logger::logLevelNames[message.level()], message.service());
     _messages.push_back(std::move(message));
     if (_messages.size() == _max_messages_per_request) {
         flush();
diff --git a/logd/src/logd/rpc_forwarder.h b/logd/src/logd/rpc_forwarder.h
index 64ea0ebea64..dbb6e4de88c 100644
--- a/logd/src/logd/rpc_forwarder.h
+++ b/logd/src/logd/rpc_forwarder.h
@@ -10,11 +10,14 @@
 
 namespace logdemon {
 
+struct Metrics;
+
 /**
  * Implementation of the Forwarder interface that uses RPC to send protobuf encoded log messages to the logserver.
  */
 class RpcForwarder : public Forwarder {
 private:
+    Metrics& _metrics;
     vespalib::string _connection_spec;
     double _rpc_timeout_secs;
     size_t _max_messages_per_request;
@@ -24,7 +27,7 @@ private:
     int _bad_lines;
 
 public:
-    RpcForwarder(const vespalib::string& logserver_host, int logserver_rpc_port,
+    RpcForwarder(Metrics& metrics, const vespalib::string& logserver_host, int logserver_rpc_port,
                  double rpc_timeout_secs, size_t max_messages_per_request);
     ~RpcForwarder() override;
 
diff --git a/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp b/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp
index be20d715c81..97fd294cfb6 100644
--- a/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp
+++ b/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp
@@ -1,10 +1,13 @@
 // Copyright 2019 Oath Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 
 #include <logd/exceptions.h>
+#include <logd/metrics.h>
 #include <logd/rpc_forwarder.h>
 #include <vespa/vespalib/gtest/gtest.h>
+#include <vespa/vespalib/metrics/dummy_metrics_manager.h>
 
 using namespace logdemon;
+using vespalib::metrics::DummyMetricsManager;
 
 void
 encode_log_response(const ProtoConverter::ProtoLogResponse& src, FRT_Values& dst)
@@ -83,11 +86,24 @@ make_log_line(const std::string& payload)
     return "1234.5678\tmy_host\t10/20\tmy_service\tmy_component\tinfo\t" + payload;
 }
 
+struct MockMetricsManager : public DummyMetricsManager {
+    int add_count;
+    MockMetricsManager() : DummyMetricsManager(), add_count(0) {}
+    void add(Counter::Increment) override {
+        ++add_count;
+    }
+};
+
 struct RpcForwarderTest : public ::testing::Test {
     RpcServer server;
+    std::shared_ptr<MockMetricsManager> metrics_mgr;
+    Metrics metrics;
     RpcForwarder forwarder;
     RpcForwarderTest()
-        : forwarder("localhost", server.get_listen_port(), 60.0, 3)
+        : server(),
+          metrics_mgr(std::make_shared<MockMetricsManager>()),
+          metrics(metrics_mgr),
+          forwarder(metrics, "localhost", server.get_listen_port(), 60.0, 3)
     {
     }
     void forward_line(const std::string& payload) {
@@ -164,6 +180,14 @@ TEST_F(RpcForwarderTest, bad_log_lines_count_can_be_reset)
     EXPECT_EQ(0, forwarder.badLines());
 }
 
+TEST_F(RpcForwarderTest, metrics_are_updated_for_each_log_message)
+{
+    forward_line("a");
+    EXPECT_EQ(1, metrics_mgr->add_count);
+    forward_line("b");
+    EXPECT_EQ(2, metrics_mgr->add_count);
+}
+
 TEST_F(RpcForwarderTest, throws_when_rpc_reply_contains_errors)
 {
     server.reply_with_error = true;