Filter log messages based on log level, similar as in legacy forwarder.

5 files changed, 52 insertions, 13 deletions

diff --git a/logd/src/logd/forwarder.h b/logd/src/logd/forwarder.h
index b26abf01658..93cfdb3de9f 100644
--- a/logd/src/logd/forwarder.h
+++ b/logd/src/logd/forwarder.h
@@ -2,10 +2,15 @@
 
 #pragma once
 
+#include <vespa/log/log.h>
+#include <map>
 #include <string_view>
 
 namespace logdemon {
 
+// Mapping saying if a level should be forwarded or not
+using ForwardMap = std::map<ns_log::Logger::LogLevel, bool>;
+
 /**
  * Interface used to forward log lines to something.
  */
diff --git a/logd/src/logd/legacy_forwarder.h b/logd/src/logd/legacy_forwarder.h
index 7adeabef29e..db3bf84fd4f 100644
--- a/logd/src/logd/legacy_forwarder.h
+++ b/logd/src/logd/legacy_forwarder.h
@@ -2,15 +2,9 @@
 #pragma once
 
 #include "forwarder.h"
-#include <vespa/log/log.h>
-#include <map>
-#include <unordered_set>
 
 namespace logdemon {
 
-// Mapping saying if a level should be forwarded or not
-using ForwardMap = std::map<ns_log::Logger::LogLevel, bool>;
-
 struct Metrics;
 
 /**
diff --git a/logd/src/logd/rpc_forwarder.cpp b/logd/src/logd/rpc_forwarder.cpp
index 46dd47493b5..54ddcbb7c81 100644
--- a/logd/src/logd/rpc_forwarder.cpp
+++ b/logd/src/logd/rpc_forwarder.cpp
@@ -26,7 +26,8 @@ RpcForwarder::RpcForwarder(Metrics& metrics, const vespalib::string &hostname, i
       _supervisor(),
       _target(),
       _messages(),
-      _bad_lines(0)
+      _bad_lines(0),
+      _forward_filter()
 {
     _supervisor.Start();
     _target = _supervisor.GetTarget(_connection_spec.c_str());
@@ -62,6 +63,16 @@ decode_log_response(FRT_RPCRequest& src, ProtoConverter::ProtoLogResponse& dst)
     return dst.ParseFromArray(values[2]._data._buf, values[2]._data._len);
 }
 
+bool
+should_forward_log_message(const LogMessage& message, const ForwardMap& filter)
+{
+    auto found = filter.find(message.level());
+    if (found != filter.end()) {
+        return found->second;
+    }
+    return false;
+}
+
 }
 
 void
@@ -76,9 +87,11 @@ RpcForwarder::forwardLine(std::string_view line)
         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();
+    if (should_forward_log_message(message, _forward_filter)) {
+        _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 dbb6e4de88c..79ffb2154bc 100644
--- a/logd/src/logd/rpc_forwarder.h
+++ b/logd/src/logd/rpc_forwarder.h
@@ -25,11 +25,13 @@ private:
     FRT_Target* _target;
     std::vector<ns_log::LogMessage> _messages;
     int _bad_lines;
+    ForwardMap _forward_filter;
 
 public:
     RpcForwarder(Metrics& metrics, const vespalib::string& logserver_host, int logserver_rpc_port,
                  double rpc_timeout_secs, size_t max_messages_per_request);
     ~RpcForwarder() override;
+    void set_forward_filter(const ForwardMap& forward_filter) { _forward_filter = forward_filter; }
 
     // Implements Forwarder
     void sendMode() override {}
diff --git a/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp b/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp
index 97fd294cfb6..d081740b0d5 100644
--- a/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp
+++ b/logd/src/tests/rpc_forwarder/rpc_forwarder_test.cpp
@@ -81,9 +81,9 @@ public:
 };
 
 std::string
-make_log_line(const std::string& payload)
+make_log_line(const std::string& level, const std::string& payload)
 {
-    return "1234.5678\tmy_host\t10/20\tmy_service\tmy_component\tinfo\t" + payload;
+    return "1234.5678\tmy_host\t10/20\tmy_service\tmy_component\t" + level + "\t" + payload;
 }
 
 struct MockMetricsManager : public DummyMetricsManager {
@@ -105,9 +105,18 @@ struct RpcForwarderTest : public ::testing::Test {
           metrics(metrics_mgr),
           forwarder(metrics, "localhost", server.get_listen_port(), 60.0, 3)
     {
+        ForwardMap forward_filter;
+        forward_filter[ns_log::Logger::error] = true;
+        forward_filter[ns_log::Logger::warning] = false;
+        forward_filter[ns_log::Logger::info] = true;
+        // all other log levels are implicit false
+        forwarder.set_forward_filter(forward_filter);
     }
     void forward_line(const std::string& payload) {
-        forwarder.forwardLine(make_log_line(payload));
+        forwarder.forwardLine(make_log_line("info", payload));
+    }
+    void forward_line(const std::string& level, const std::string& payload) {
+        forwarder.forwardLine(make_log_line(level, payload));
     }
     void forward_bad_line() {
         forwarder.forwardLine("badline");
@@ -188,6 +197,22 @@ TEST_F(RpcForwarderTest, metrics_are_updated_for_each_log_message)
     EXPECT_EQ(2, metrics_mgr->add_count);
 }
 
+TEST_F(RpcForwarderTest, log_messages_are_filtered_on_log_level)
+{
+    forward_line("fatal",   "a");
+    forward_line("error",   "b");
+    forward_line("warning", "c");
+    forward_line("config",  "d");
+    forward_line("info",    "e");
+    forward_line("event",   "f");
+    forward_line("debug",   "g");
+    forward_line("spam",    "h");
+    forward_line("null",    "i");
+    flush();
+    expect_messages(1, {"b", "e"});
+    EXPECT_EQ(9, metrics_mgr->add_count);
+}
+
 TEST_F(RpcForwarderTest, throws_when_rpc_reply_contains_errors)
 {
     server.reply_with_error = true;