[demux] add control-pattern config

docs/schemas/config-v1.schema.json

@@ -910,10 +910,20 @@
                             "title": "/log/demux/<name>",
                             "type": "object",
                             "properties": {
+                                "enabled": {
+                                    "title": "/log/demux/<name>/enabled",
+                                    "description": "Indicates whether this demuxer will be used at the demuxing stage",
+                                    "type": "boolean"
+                                },
                                 "pattern": {
                                     "title": "/log/demux/<name>/pattern",
                                     "description": "A regular expression to match a line in a multiplexed file",
                                     "type": "string"
+                                },
+                                "control-pattern": {
+                                    "title": "/log/demux/<name>/control-pattern",
+                                    "description": "A regular expression to match a control line in a multiplexed file",
+                                    "type": "string"
                                 }
                             },
                             "additionalProperties": false

src/base/piper.file.cc

@@ -79,8 +79,8 @@ read_header(int fd, const char* first8)
     return meta_buf;
 }
 
-std::optional<std::string>
-multiplex_id_for_line(string_fragment line)
+multiplex_matcher::match_result
+multiplex_matcher::match(const string_fragment& line)
 {
     const auto& cfg = injector::get<const config&>();
     auto md = lnav::pcre2pp::match_data::unitialized();
@@ -88,32 +88,55 @@ multiplex_id_for_line(string_fragment line)
     for (const auto& demux_pair : cfg.c_demux_definitions) {
         const auto& df = demux_pair.second;
 
-        if (!df.dd_enabled) {
+        if (!df.dd_valid || !df.dd_enabled) {
+            continue;
+        }
+
+        if (!this->mm_partial_match_ids.empty()
+            && this->mm_partial_match_ids.count(demux_pair.first) == 0)
+        {
             continue;
         }
 
         log_info("attempting to demux using: %s", demux_pair.first.c_str());
-        md = df.dd_pattern.pp_value->create_match_data();
-        if (df.dd_pattern.pp_value->capture_from(line)
-                .into(md)
-                .matches()
-                .ignore_error())
         {
-            log_info("  demuxer pattern matched");
-            if (!md[df.dd_muxid_capture_index].has_value()) {
-                log_info("    however, mux_id was not captured");
-                continue;
+            md = df.dd_pattern.pp_value->create_match_data();
+            if (df.dd_pattern.pp_value->capture_from(line)
+                    .into(md)
+                    .matches()
+                    .ignore_error())
+            {
+                log_info("  demuxer pattern matched");
+                if (!md[df.dd_muxid_capture_index].has_value()) {
+                    log_info("    however, mux_id was not captured");
+                    continue;
+                }
+                if (!md[df.dd_body_capture_index].has_value()) {
+                    log_info("    however, body was not captured");
+                    continue;
+                }
+                log_info("  and required captures were found, using demuxer");
+                return found{demux_pair.first};
             }
-            if (!md[df.dd_body_capture_index].has_value()) {
-                log_info("    however, body was not captured");
-                continue;
+        }
+        if (df.dd_control_pattern.pp_value) {
+            md = df.dd_control_pattern.pp_value->create_match_data();
+            if (df.dd_control_pattern.pp_value->capture_from(line)
+                    .into(md)
+                    .matches()
+                    .ignore_error())
+            {
+                log_info("  demuxer control pattern matched");
+                this->mm_partial_match_ids.emplace(demux_pair.first);
             }
-            log_info("  and required captures were found, using demuxer");
-            return demux_pair.first;
         }
     }
 
-    return std::nullopt;
+    if (this->mm_partial_match_ids.empty()) {
+        return not_found{};
+    }
+
+    return partial{};
 }
 
 }  // namespace piper

src/base/piper.file.hh

@@ -32,6 +32,7 @@
 
 #include <map>
 #include <optional>
+#include <set>
 #include <string>
 
 #include <sys/time.h>
@@ -39,6 +40,7 @@
 #include "auto_mem.hh"
 #include "base/intern_string.hh"
 #include "ghc/filesystem.hpp"
+#include "mapbox/variant_io.hpp"
 #include "time_util.hh"
 
 namespace lnav {
@@ -81,7 +83,22 @@ extern const char HEADER_MAGIC[4];
 
 std::optional<auto_buffer> read_header(int fd, const char* first8);
 
-std::optional<std::string> multiplex_id_for_line(string_fragment line);
+class multiplex_matcher {
+public:
+    struct found {
+        std::string f_id;
+    };
+
+    struct partial {};
+    struct not_found {};
+
+    using match_result = mapbox::util::variant<found, partial, not_found>;
+
+    match_result match(const string_fragment& line);
+
+private:
+    std::set<std::string> mm_partial_match_ids;
+};
 
 }  // namespace piper
 }  // namespace lnav

src/file_format.cc

@@ -72,26 +72,59 @@ detect_file_format(const ghc::filesystem::path& filename)
                 log_info("%s: appears to be a SQLite DB", filename.c_str());
                 retval = file_format_t::SQLITE_DB;
             } else {
+                auto looping = true;
+                lnav::piper::multiplex_matcher mm;
                 file_range next_range;
                 line_buffer lb;
                 lb.set_fd(fd);
 
-                auto load_res = lb.load_next_line(next_range);
-                if (load_res.isOk() && lb.is_header_utf8() && !lb.is_piper()) {
+                while (looping) {
+                    auto load_res = lb.load_next_line(next_range);
+                    if (load_res.isErr()) {
+                        log_error(
+                            "unable to load line for demux matching: %s -- %s",
+                            filename.c_str(),
+                            load_res.unwrapErr().c_str());
+                        break;
+                    }
+                    if (!lb.is_header_utf8()) {
+                        log_info("file is not UTF-8: %s", filename.c_str());
+                        break;
+                    }
+                    if (lb.is_piper()) {
+                        log_info("skipping demux match for piper file: %s",
+                                 filename.c_str());
+                        break;
+                    }
                     auto li = load_res.unwrap();
                     auto read_res = lb.read_range(li.li_file_range);
-                    if (read_res.isOk()) {
-                        auto sbr = read_res.unwrap();
-                        auto demux_id_opt = lnav::piper::multiplex_id_for_line(
-                            sbr.to_string_fragment());
+                    if (read_res.isErr()) {
+                        log_error(
+                            "unable to read line for demux matching: %s -- %s",
+                            filename.c_str(),
+                            read_res.unwrapErr().c_str());
+                        break;
+                    }
+                    auto sbr = read_res.unwrap();
+                    auto match_res = mm.match(sbr.to_string_fragment());
 
-                        if (demux_id_opt) {
+                    looping = match_res.match(
+                        [&retval,
+                         &filename](lnav::piper::multiplex_matcher::found f) {
                             log_info("%s: is multiplexed using %s",
                                      filename.c_str(),
-                                     demux_id_opt.value().c_str());
-                            return file_format_t::MULTIPLEXED;
-                        }
-                    }
+                                     f.f_id.c_str());
+                            retval = file_format_t::MULTIPLEXED;
+                            return false;
+                        },
+                        [](lnav::piper::multiplex_matcher::not_found nf) {
+                            return false;
+                        },
+                        [](lnav::piper::multiplex_matcher::partial p) {
+                            return true;
+                        });
+
+                    next_range = li.li_file_range;
                 }
             }
         }

src/lnav_config.cc

@@ -384,11 +384,13 @@ update_installs_from_git()
                                         git_dir.string());
             int ret = system(pull_cmd.c_str());
             if (ret == -1) {
-                std::cerr << "Failed to spawn command " << "\"" << pull_cmd
-                          << "\": " << strerror(errno) << std::endl;
+                std::cerr << "Failed to spawn command "
+                          << "\"" << pull_cmd << "\": " << strerror(errno)
+                          << std::endl;
                 retval = false;
             } else if (ret > 0) {
-                std::cerr << "Command " << "\"" << pull_cmd
+                std::cerr << "Command "
+                          << "\"" << pull_cmd
                           << "\" failed: " << strerror(errno) << std::endl;
                 retval = false;
             }
@@ -1198,11 +1200,21 @@ static const struct json_path_container archive_handlers = {
 
 static const struct typed_json_path_container<lnav::piper::demux_def>
     demux_def_handlers = {
-    yajlpp::property_handler("pattern")
-        .with_synopsis("<regex>")
+    yajlpp::property_handler("enabled")
         .with_description(
-            "A regular expression to match a line in a multiplexed file")
-        .for_field(&lnav::piper::demux_def::dd_pattern),
+            "Indicates whether this demuxer will be used at the demuxing stage")
+        .for_field(&lnav::piper::demux_def::dd_enabled),
+        yajlpp::property_handler("pattern")
+            .with_synopsis("<regex>")
+            .with_description(
+                "A regular expression to match a line in a multiplexed file")
+            .for_field(&lnav::piper::demux_def::dd_pattern),
+        yajlpp::property_handler("control-pattern")
+            .with_synopsis("<regex>")
+            .with_description(
+                "A regular expression to match a control line in a multiplexed "
+                "file")
+            .for_field(&lnav::piper::demux_def::dd_control_pattern),
 };
 
 static const struct json_path_container demux_defs_handlers = {

src/piper.looper.cc

@@ -120,7 +120,7 @@ public:
                     = ncap.get_index();
             }
 
-            dd.dd_enabled = true;
+            dd.dd_valid = true;
         }
     }
 };
@@ -267,6 +267,7 @@ looper::loop()
     struct timeval line_tv;
     struct exttm line_tm;
     auto file_timeout = 0ms;
+    multiplex_matcher mmatcher;
 
     log_info("starting loop to capture: %s (%d %d)",
              this->l_name.c_str(),
@@ -464,22 +465,25 @@ looper::loop()
                 auto body_sf = sbr.to_string_fragment();
                 auto ts_sf = string_fragment{};
                 if (!curr_demux_def && !demux_attempted) {
-                    log_trace("first input line: %s",
+                    log_trace("demux input line: %s",
                               fmt::format(FMT_STRING("{:?}"), body_sf).c_str());
 
-                    auto demux_id_opt = multiplex_id_for_line(body_sf);
-                    if (demux_id_opt) {
-                        curr_demux_def = cfg.c_demux_definitions
-                                             .find(demux_id_opt.value())
-                                             ->second;
-                        {
-                            safe::WriteAccess<safe_demux_id> di(
-                                this->l_demux_id);
-
-                            di->assign(demux_id_opt.value());
-                        }
-                    }
-                    demux_attempted = true;
+                    auto match_res = mmatcher.match(body_sf);
+                    demux_attempted = match_res.match(
+                        [this, &curr_demux_def, &cfg](
+                            multiplex_matcher::found f) {
+                            curr_demux_def
+                                = cfg.c_demux_definitions.find(f.f_id)->second;
+                            {
+                                safe::WriteAccess<safe_demux_id> di(
+                                    this->l_demux_id);
+
+                                di->assign(f.f_id);
+                            }
+                            return true;
+                        },
+                        [](multiplex_matcher::not_found nf) { return true; },
+                        [](multiplex_matcher::partial p) { return false; });
                 }
                 std::optional<log_level_t> demux_level;
                 if (curr_demux_def

src/piper.looper.cfg.hh

@@ -42,8 +42,10 @@ namespace lnav {
 namespace piper {
 
 struct demux_def {
-    bool dd_enabled{false};
-    factory_container<lnav::pcre2pp::code> dd_pattern;
+    bool dd_enabled{true};
+    bool dd_valid{false};
+    factory_container<pcre2pp::code> dd_control_pattern;
+    factory_container<pcre2pp::code> dd_pattern;
     int dd_timestamp_capture_index{-1};
     int dd_muxid_capture_index{-1};
     int dd_body_capture_index{-1};

src/root-config.json

@@ -29,6 +29,7 @@
                 "pattern": "^(?:\\x1b\\[\\d*K)?(?<mux_id>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*)\\s+\\| (?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{9}Z )?(?<body>.*)"
             },
             "recv-with-pod": {
+                "control-pattern": "^===== (?:START|END) =====$",
                 "pattern": "^(?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}(?:Z|[+\\-]\\d{2}:\\d{2})) source=[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]* (?<body>.*) kubernetes_host=(?<k8s_host>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*) kubernetes_pod_name=(?<mux_id>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*)"
             },
             "container-with-type": {

test/expected/test_cli.sh_0b3639753916f71254e8c9cce4ebb8bfd9978d3e.out

@@ -5239,13 +5239,19 @@
         },
         "demux": {
             "container": {
-                "pattern": "^(?:\\x1b\\[\\d*K)?(?<mux_id>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*)\\s+\\| (?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{9}Z )?(?<body>.*)"
+                "enabled": true,
+                "pattern": "^(?:\\x1b\\[\\d*K)?(?<mux_id>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*)\\s+\\| (?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.\\d{9}Z )?(?<body>.*)",
+                "control-pattern": ""
             },
             "container-with-type": {
-                "pattern": "^(?<mux_id>[a-zA-Z][\\w\\-]{3,}) (?<container_type>[a-zA-Z][\\w\\-]{3,}) (?<body>.*)"
+                "enabled": true,
+                "pattern": "^(?<mux_id>[a-zA-Z][\\w\\-]{3,}) (?<container_type>[a-zA-Z][\\w\\-]{3,}) (?<body>.*)",
+                "control-pattern": ""
             },
             "recv-with-pod": {
-                "pattern": "^(?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}(?:Z|[+\\-]\\d{2}:\\d{2})) source=[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]* (?<body>.*) kubernetes_host=(?<k8s_host>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*) kubernetes_pod_name=(?<mux_id>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*)"
+                "enabled": true,
+                "pattern": "^(?<timestamp>\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}\\.\\d{3}(?:Z|[+\\-]\\d{2}:\\d{2})) source=[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]* (?<body>.*) kubernetes_host=(?<k8s_host>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*) kubernetes_pod_name=(?<mux_id>[a-zA-Z0-9][a-zA-Z0-9_\\.\\-]*)",
+                "control-pattern": "^===== (?:START|END) =====$"
             }
         }
     },

test/expected/test_cli.sh_cc06341dd560f927512e92c7c0985ed8b25827ae.out

@@ -19,44 +19,45 @@
 /log/annotations/org.lnav.test/description -> {test_dir}/configs/installed/anno-test.json:6
 /log/annotations/org.lnav.test/handler -> {test_dir}/configs/installed/anno-test.json:8
 /log/date-time/convert-zoned-to-local -> root-config.json:18
-/log/demux/container-with-type/pattern -> root-config.json:35
+/log/demux/container-with-type/pattern -> root-config.json:36
 /log/demux/container/pattern -> root-config.json:29
-/log/demux/recv-with-pod/pattern -> root-config.json:32
-/tuning/archive-manager/cache-ttl -> root-config.json:42
-/tuning/archive-manager/min-free-space -> root-config.json:41
-/tuning/clipboard/impls/MacOS/find/read -> root-config.json:70
-/tuning/clipboard/impls/MacOS/find/write -> root-config.json:69
-/tuning/clipboard/impls/MacOS/general/read -> root-config.json:66
-/tuning/clipboard/impls/MacOS/general/write -> root-config.json:65
-/tuning/clipboard/impls/MacOS/test -> root-config.json:63
-/tuning/clipboard/impls/NeoVim/general/read -> root-config.json:98
-/tuning/clipboard/impls/NeoVim/general/write -> root-config.json:97
-/tuning/clipboard/impls/NeoVim/test -> root-config.json:95
-/tuning/clipboard/impls/Wayland/general/read -> root-config.json:77
-/tuning/clipboard/impls/Wayland/general/write -> root-config.json:76
-/tuning/clipboard/impls/Wayland/test -> root-config.json:74
-/tuning/clipboard/impls/Windows/general/write -> root-config.json:104
-/tuning/clipboard/impls/Windows/test -> root-config.json:102
-/tuning/clipboard/impls/X11-xclip/general/read -> root-config.json:84
-/tuning/clipboard/impls/X11-xclip/general/write -> root-config.json:83
-/tuning/clipboard/impls/X11-xclip/test -> root-config.json:81
-/tuning/clipboard/impls/tmux/general/read -> root-config.json:91
-/tuning/clipboard/impls/tmux/general/write -> root-config.json:90
-/tuning/clipboard/impls/tmux/test -> root-config.json:88
-/tuning/piper/max-size -> root-config.json:56
-/tuning/piper/rotations -> root-config.json:57
-/tuning/piper/ttl -> root-config.json:58
-/tuning/remote/ssh/command -> root-config.json:46
-/tuning/remote/ssh/config/BatchMode -> root-config.json:48
-/tuning/remote/ssh/config/ConnectTimeout -> root-config.json:49
-/tuning/remote/ssh/start-command -> root-config.json:51
-/tuning/remote/ssh/transfer-command -> root-config.json:52
-/tuning/url-scheme/docker-compose/handler -> root-config.json:114
-/tuning/url-scheme/docker/handler -> root-config.json:111
+/log/demux/recv-with-pod/control-pattern -> root-config.json:32
+/log/demux/recv-with-pod/pattern -> root-config.json:33
+/tuning/archive-manager/cache-ttl -> root-config.json:43
+/tuning/archive-manager/min-free-space -> root-config.json:42
+/tuning/clipboard/impls/MacOS/find/read -> root-config.json:71
+/tuning/clipboard/impls/MacOS/find/write -> root-config.json:70
+/tuning/clipboard/impls/MacOS/general/read -> root-config.json:67
+/tuning/clipboard/impls/MacOS/general/write -> root-config.json:66
+/tuning/clipboard/impls/MacOS/test -> root-config.json:64
+/tuning/clipboard/impls/NeoVim/general/read -> root-config.json:99
+/tuning/clipboard/impls/NeoVim/general/write -> root-config.json:98
+/tuning/clipboard/impls/NeoVim/test -> root-config.json:96
+/tuning/clipboard/impls/Wayland/general/read -> root-config.json:78
+/tuning/clipboard/impls/Wayland/general/write -> root-config.json:77
+/tuning/clipboard/impls/Wayland/test -> root-config.json:75
+/tuning/clipboard/impls/Windows/general/write -> root-config.json:105
+/tuning/clipboard/impls/Windows/test -> root-config.json:103
+/tuning/clipboard/impls/X11-xclip/general/read -> root-config.json:85
+/tuning/clipboard/impls/X11-xclip/general/write -> root-config.json:84
+/tuning/clipboard/impls/X11-xclip/test -> root-config.json:82
+/tuning/clipboard/impls/tmux/general/read -> root-config.json:92
+/tuning/clipboard/impls/tmux/general/write -> root-config.json:91
+/tuning/clipboard/impls/tmux/test -> root-config.json:89
+/tuning/piper/max-size -> root-config.json:57
+/tuning/piper/rotations -> root-config.json:58
+/tuning/piper/ttl -> root-config.json:59
+/tuning/remote/ssh/command -> root-config.json:47
+/tuning/remote/ssh/config/BatchMode -> root-config.json:49
+/tuning/remote/ssh/config/ConnectTimeout -> root-config.json:50
+/tuning/remote/ssh/start-command -> root-config.json:52
+/tuning/remote/ssh/transfer-command -> root-config.json:53
+/tuning/url-scheme/docker-compose/handler -> root-config.json:115
+/tuning/url-scheme/docker/handler -> root-config.json:112
 /tuning/url-scheme/hw/handler -> {test_dir}/configs/installed/hw-url-handler.json:6
-/tuning/url-scheme/journald/handler -> root-config.json:117
-/tuning/url-scheme/piper/handler -> root-config.json:120
-/tuning/url-scheme/podman/handler -> root-config.json:123
+/tuning/url-scheme/journald/handler -> root-config.json:118
+/tuning/url-scheme/piper/handler -> root-config.json:121
+/tuning/url-scheme/podman/handler -> root-config.json:124
 /ui/clock-format -> root-config.json:4
 /ui/default-colors -> root-config.json:6
 /ui/dim-text -> root-config.json:5