feat: Add support for log-level filtering of structured IR streams.

src/clp_ffi_js/constants.hpp

@@ -19,6 +19,7 @@ enum class LogLevel : std::uint8_t {
     FATAL,
 };
 constexpr LogLevel cValidLogLevelsBeginIdx{LogLevel::TRACE};
+constexpr LogLevel cValidLogLevelsEndIdx{LogLevel::FATAL};
 
 /**
  * Strings corresponding to `LogLevel`.

src/clp_ffi_js/ir/LogEventWithFilterData.hpp

@@ -0,0 +1,70 @@
+#ifndef CLP_FFI_JS_IR_LOGEVENTWITHFILTERDATA_HPP
+#define CLP_FFI_JS_IR_LOGEVENTWITHFILTERDATA_HPP
+
+#include <concepts>
+#include <utility>
+
+#include <clp/ffi/KeyValuePairLogEvent.hpp>
+#include <clp/ir/LogEvent.hpp>
+#include <clp/ir/types.hpp>
+
+#include <clp_ffi_js/constants.hpp>
+
+namespace clp_ffi_js::ir {
+using clp::ir::four_byte_encoded_variable_t;
+using UnstructuredLogEvent = clp::ir::LogEvent<four_byte_encoded_variable_t>;
+using StructuredLogEvent = clp::ffi::KeyValuePairLogEvent;
+
+// Concept defining valid log event types.
+// TODO: Extend valid log event types when filtering support is added for structured logs.
+template <typename T>
+concept ValidLogEventTypes
+        = std::same_as<T, UnstructuredLogEvent> || std::same_as<T, StructuredLogEvent>;
+
+/**
+ * A templated class that extends a log event type with processed versions of some of its fields,
+ * specifically the fields that are used for filtering in the `StreamReader` classes and their
+ * callers.
+ *
+ * @tparam LogEvent The type of the log event.
+ */
+template <typename LogEvent>
+requires ValidLogEventTypes<LogEvent>
+class LogEventWithFilterData {
+public:
+    // Constructor
+    LogEventWithFilterData(
+            LogEvent log_event,
+            LogLevel log_level,
+            clp::ir::epoch_time_ms_t timestamp
+    )
+            : m_log_event{std::move(log_event)},
+              m_log_level{log_level},
+              m_timestamp{timestamp} {}
+
+    // Disable copy constructor and assignment operator
+    LogEventWithFilterData(LogEventWithFilterData const&) = delete;
+    auto operator=(LogEventWithFilterData const&) -> LogEventWithFilterData& = delete;
+
+    // Default move constructor and assignment operator
+    LogEventWithFilterData(LogEventWithFilterData&&) = default;
+    auto operator=(LogEventWithFilterData&&) -> LogEventWithFilterData& = default;
+
+    // Destructor
+    ~LogEventWithFilterData() = default;
+
+    [[nodiscard]] auto get_log_event() const -> LogEvent const& { return m_log_event; }
+
+    [[nodiscard]] auto get_log_level() const -> LogLevel { return m_log_level; }
+
+    [[nodiscard]] auto get_timestamp() const -> clp::ir::epoch_time_ms_t { return m_timestamp; }
+
+private:
+    LogEvent m_log_event;
+    LogLevel m_log_level;
+    clp::ir::epoch_time_ms_t m_timestamp;
+};
+
+}  // namespace clp_ffi_js::ir
+
+#endif  // CLP_FFI_JS_IR_LOGEVENTWITHFILTERDATA_HPP

src/clp_ffi_js/ir/StreamReader.hpp

@@ -23,6 +23,12 @@ enum class StreamType : uint8_t {
     Unstructured,
 };
 
+/**
+ * Mapping between an index in the filtered log events collection to an index in the unfiltered
+ * log events collection.
+ */
+using FilteredLogEventsMap = std::optional<std::vector<size_t>>;
+
 /**
  * Class to deserialize and decode Zstandard-compressed CLP IR streams as well as format decoded
  * log events.

src/clp_ffi_js/ir/StructuredIrStreamReader.cpp

@@ -13,39 +13,42 @@
 #include <clp/ErrorCode.hpp>
 #include <clp/ffi/ir_stream/Deserializer.hpp>
 #include <clp/ffi/KeyValuePairLogEvent.hpp>
-#include <clp/ffi/Value.hpp>
-#include <clp/ir/types.hpp>
 #include <clp/TraceableException.hpp>
 #include <emscripten/em_asm.h>
 #include <emscripten/val.h>
 #include <spdlog/spdlog.h>
 
 #include <clp_ffi_js/ClpFfiJsException.hpp>
 #include <clp_ffi_js/constants.hpp>
+#include <clp_ffi_js/ir/LogEventWithFilterData.hpp>
 #include <clp_ffi_js/ir/StreamReader.hpp>
 #include <clp_ffi_js/ir/StreamReaderDataContext.hpp>
+#include <clp_ffi_js/ir/StructuredIrUnitHandler.hpp>
+#include <clp_ffi_js/ir/utils.hpp>
 
 namespace clp_ffi_js::ir {
 namespace {
 constexpr std::string_view cEmptyJsonStr{"{}"};
 constexpr std::string_view cLogLevelFilteringNotSupportedErrorMsg{
         "Log level filtering is not yet supported in this reader."
 };
+constexpr std::string_view cReaderOptionsLogLevelKey{"logLevelKey"};
 constexpr std::string_view cReaderOptionsTimestampKey{"timestampKey"};
 }  // namespace
 
-using clp::ir::four_byte_encoded_variable_t;
-
 auto StructuredIrStreamReader::create(
         std::unique_ptr<ZstdDecompressor>&& zstd_decompressor,
         clp::Array<char> data_array,
         ReaderOptions const& reader_options
 ) -> StructuredIrStreamReader {
-    auto deserialized_log_events{std::make_shared<std::vector<clp::ffi::KeyValuePairLogEvent>>()};
+    auto deserialized_log_events{
+            std::make_shared<std::vector<LogEventWithFilterData<StructuredLogEvent>>>()
+    };
     auto result{StructuredIrDeserializer::create(
             *zstd_decompressor,
-            IrUnitHandler{
+            StructuredIrUnitHandler{
                     deserialized_log_events,
+                    reader_options[cReaderOptionsLogLevelKey.data()].as<std::string>(),
                     reader_options[cReaderOptionsTimestampKey.data()].as<std::string>()
             }
     )};
@@ -75,15 +78,19 @@ auto StructuredIrStreamReader::get_num_events_buffered() const -> size_t {
 }
 
 auto StructuredIrStreamReader::get_filtered_log_event_map() const -> FilteredLogEventMapTsType {
-    SPDLOG_ERROR(cLogLevelFilteringNotSupportedErrorMsg);
-    return FilteredLogEventMapTsType{emscripten::val::null()};
+    if (false == m_filtered_log_event_map.has_value()) {
+        return FilteredLogEventMapTsType{emscripten::val::null()};
+    }
+
+    return FilteredLogEventMapTsType{emscripten::val::array(m_filtered_log_event_map.value())};
 }
 
 void StructuredIrStreamReader::filter_log_events(LogLevelFilterTsType const& log_level_filter) {
-    if (log_level_filter.isNull()) {
-        return;
-    }
-    SPDLOG_ERROR(cLogLevelFilteringNotSupportedErrorMsg);
+    filter_deserialized_events(
+            log_level_filter,
+            m_filtered_log_event_map,
+            *m_deserialized_log_events
+    );
 }
 
 auto StructuredIrStreamReader::deserialize_stream() -> size_t {
@@ -119,9 +126,6 @@ auto StructuredIrStreamReader::deserialize_stream() -> size_t {
                 )
         };
     }
-    m_timestamp_node_id = m_stream_reader_data_context->get_deserializer()
-                                  .get_ir_unit_handler()
-                                  .get_timestamp_node_id();
     m_stream_reader_data_context.reset(nullptr);
     return m_deserialized_log_events->size();
 }
@@ -140,9 +144,10 @@ auto StructuredIrStreamReader::decode_range(size_t begin_idx, size_t end_idx, bo
     auto const results{emscripten::val::array()};
 
     for (size_t log_event_idx = begin_idx; log_event_idx < end_idx; ++log_event_idx) {
-        auto const& log_event{m_deserialized_log_events->at(log_event_idx)};
+        auto const& log_event_with_filter_data{m_deserialized_log_events->at(log_event_idx)};
+        auto const& structured_log_event = log_event_with_filter_data.get_log_event();
 
-        auto const json_result{log_event.serialize_to_json()};
+        auto const json_result{structured_log_event.serialize_to_json()};
         std::string json_str{cEmptyJsonStr};
         if (false == json_result.has_value()) {
             auto error_code{json_result.error()};
@@ -155,26 +160,12 @@ auto StructuredIrStreamReader::decode_range(size_t begin_idx, size_t end_idx, bo
             json_str = json_result.value().dump();
         }
 
-        auto const& id_value_pairs{log_event.get_node_id_value_pairs()};
-        clp::ffi::value_int_t timestamp{0};
-        if (m_timestamp_node_id.has_value()) {
-            auto const& timestamp_pair{id_value_pairs.at(m_timestamp_node_id.value())};
-            if (timestamp_pair.has_value()) {
-                if (timestamp_pair->is<clp::ffi::value_int_t>()) {
-                    timestamp = timestamp_pair.value().get_immutable_view<clp::ffi::value_int_t>();
-                } else {
-                    // TODO: Add support for parsing timestamp values of string type.
-                    SPDLOG_ERROR("Unable to parse timestamp for log_event_idx={}", log_event_idx);
-                }
-            }
-        }
-
         EM_ASM(
                 { Emval.toValue($0).push([UTF8ToString($1), $2, $3, $4]); },
                 results.as_handle(),
                 json_str.c_str(),
-                timestamp,
-                LogLevel::NONE,
+                log_event_with_filter_data.get_timestamp(),
+                log_event_with_filter_data.get_log_level(),
                 log_event_idx + 1
         );
     }
@@ -184,7 +175,8 @@ auto StructuredIrStreamReader::decode_range(size_t begin_idx, size_t end_idx, bo
 
 StructuredIrStreamReader::StructuredIrStreamReader(
         StreamReaderDataContext<StructuredIrDeserializer>&& stream_reader_data_context,
-        std::shared_ptr<std::vector<clp::ffi::KeyValuePairLogEvent>> deserialized_log_events
+        std::shared_ptr<std::vector<LogEventWithFilterData<StructuredLogEvent>>>
+                deserialized_log_events
 )
         : m_deserialized_log_events{std::move(deserialized_log_events)},
           m_stream_reader_data_context{