Merge branch 'main' into pr-query-log-correlation

Author: jackluo923

components/core/src/clp_s/AggregationSink.cpp

@@ -0,0 +1,64 @@
+#include "AggregationSink.hpp"
+
+#include <iostream>
+#include <string_view>
+#include <variant>
+
+#include <bsoncxx/builder/basic/document.hpp>
+#include <bsoncxx/builder/basic/kvp.hpp>
+#include <mongocxx/client.hpp>
+#include <mongocxx/collection.hpp>
+#include <mongocxx/exception/exception.hpp>
+#include <nlohmann/json.hpp>
+
+#include <clp_s/archive_constants.hpp>
+#include <clp_s/ErrorCode.hpp>
+#include <clp_s/ResultsCacheUtils.hpp>
+
+using std::string_view;
+
+namespace clp_s {
+auto StdoutSink::write(AggregationResult const& result) -> void {
+    nlohmann::json document;
+    document[constants::results_cache::search::cArchiveId] = m_archive_id;
+    for (auto const& [key, value] : result) {
+        std::visit([&](auto const& field_value) { document[key] = field_value; }, value);
+    }
+    std::cout << document.dump() << '\n';
+}
+
+ResultsCacheSink::ResultsCacheSink(string_view uri, string_view collection, string_view archive_id)
+        : m_archive_id{archive_id} {
+    m_collection = connect_to_results_cache(uri, collection, m_client);
+}
+
+auto ResultsCacheSink::write(AggregationResult const& result) -> void {
+    bsoncxx::builder::basic::document document;
+    document.append(
+            bsoncxx::builder::basic::kvp(constants::results_cache::search::cArchiveId, m_archive_id)
+    );
+    for (auto const& [key, value] : result) {
+        std::visit(
+                [&](auto const& field_value) {
+                    document.append(bsoncxx::builder::basic::kvp(key, field_value));
+                },
+                value
+        );
+    }
+    m_results.push_back(document.extract());
+}
+
+auto ResultsCacheSink::finish() -> ErrorCode {
+    if (m_results.empty()) {
+        return ErrorCode::ErrorCodeSuccess;
+    }
+
+    try {
+        m_collection.insert_many(m_results);
+        m_results.clear();
+    } catch (mongocxx::exception const& e) {
+        return ErrorCode::ErrorCodeFailureDbBulkWrite;
+    }
+    return ErrorCode::ErrorCodeSuccess;
+}
+}  // namespace clp_s

components/core/src/clp_s/AggregationSink.hpp

@@ -0,0 +1,98 @@
+#ifndef CLP_S_AGGREGATIONSINK_HPP
+#define CLP_S_AGGREGATIONSINK_HPP
+
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include <bsoncxx/document/value.hpp>
+#include <mongocxx/client.hpp>
+#include <mongocxx/collection.hpp>
+
+#include <clp_s/aggregators.hpp>
+#include <clp_s/ErrorCode.hpp>
+
+namespace clp_s {
+/**
+ * Consumes an aggregation's result documents and writes them to a destination.
+ */
+class AggregationSink {
+public:
+    // Constructors
+    AggregationSink() = default;
+
+    // Delete copy constructor and assignment operator
+    AggregationSink(AggregationSink const&) = delete;
+    auto operator=(AggregationSink const&) -> AggregationSink& = delete;
+
+    // Delete move constructor and assignment operator
+    AggregationSink(AggregationSink&&) = delete;
+    auto operator=(AggregationSink&&) -> AggregationSink& = delete;
+
+    // Destructor
+    virtual ~AggregationSink() = default;
+
+    // Methods
+    /**
+     * Writes one result document.
+     * @param result The result document to write.
+     */
+    virtual auto write(AggregationResult const& result) -> void = 0;
+
+    /**
+     * Flushes any buffered results.
+     * @return ErrorCodeSuccess on success or relevant error code on error
+     */
+    [[nodiscard]] virtual auto finish() -> ErrorCode = 0;
+};
+
+/**
+ * Sink that writes aggregation results to standard output as newline-delimited JSON.
+ */
+class StdoutSink : public AggregationSink {
+public:
+    // Constructors
+    explicit StdoutSink(std::string_view archive_id) : m_archive_id{archive_id} {}
+
+    // Methods implementing AggregationSink
+    auto write(AggregationResult const& result) -> void override;
+
+    [[nodiscard]] auto finish() -> ErrorCode override { return ErrorCode::ErrorCodeSuccess; }
+
+private:
+    // Data members
+    std::string m_archive_id;
+};
+
+/**
+ * Sink that writes aggregation results to a MongoDB results-cache collection.
+ */
+class ResultsCacheSink : public AggregationSink {
+public:
+    // Constructors
+    ResultsCacheSink(
+            std::string_view uri,
+            std::string_view collection,
+            std::string_view archive_id
+    );
+
+    // Methods implementing AggregationSink
+    auto write(AggregationResult const& result) -> void override;
+
+    /**
+     * Flushes the buffered result documents.
+     * @return ErrorCodeSuccess on success
+     * @return ErrorCodeFailureDbBulkWrite on database error
+     */
+    [[nodiscard]] auto finish() -> ErrorCode override;
+
+private:
+    // Data members
+    mongocxx::client m_client;
+    mongocxx::collection m_collection;
+    std::string m_archive_id;
+    std::vector<bsoncxx::document::value> m_results;
+};
+}  // namespace clp_s
+
+#endif  // CLP_S_AGGREGATIONSINK_HPP

components/core/src/clp_s/CMakeLists.txt

@@ -480,13 +480,19 @@ endif()
 
 set(
         CLP_S_EXE_SOURCES
+        AggregationSink.cpp
+        AggregationSink.hpp
+        aggregators.cpp
+        aggregators.hpp
         CommandLineArguments.cpp
         CommandLineArguments.hpp
         ErrorCode.hpp
         kv_ir_search.cpp
         kv_ir_search.hpp
         OutputHandlerImpl.cpp
         OutputHandlerImpl.hpp
+        ResultsCacheUtils.cpp
+        ResultsCacheUtils.hpp
         TraceableException.hpp
 )
 
@@ -544,6 +550,7 @@ if(CLP_BUILD_TESTING)
         target_sources(
                 clp_s_unit_test_sources
                 INTERFACE
+                filter/tests/test-clp_s-bitmap_view.cpp
                 filter/tests/test-clp_s-bloom_filter.cpp
                 filter/tests/test-clp_s-xxhash.cpp
                 tests/clp_s_test_utils.cpp

components/core/src/clp_s/CommandLineArguments.cpp

@@ -5,11 +5,14 @@
 #include <iostream>
 #include <optional>
 #include <string_view>
+#include <variant>
 
 #include <boost/program_options.hpp>
 #include <fmt/format.h>
 #include <spdlog/spdlog.h>
 
+#include <clp_s/search/ast/SearchUtils.hpp>
+
 #include "../clp/type_utils.hpp"
 #include "../reducer/types.hpp"
 #include "FileReader.hpp"
@@ -772,15 +775,25 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
             // clang-format on
             search_options.add(match_options);
 
+            int64_t count_by_time_bucket_size_millisecs{};
+            std::string aggregation_field;
             po::options_description aggregation_options("Aggregation Controls");
             // clang-format off
             aggregation_options.add_options()(
                 "count",
                 "Count the number of results"
             )(
                 "count-by-time",
-                po::value<int64_t>(&m_count_by_time_bucket_size_ms)->value_name("SIZE"),
+                po::value<int64_t>(&count_by_time_bucket_size_millisecs)->value_name("SIZE"),
                 "Count the number of results in each time span of the given size (ms)"
+            )(
+                "min",
+                po::value<std::string>(&aggregation_field)->value_name("FIELD"),
+                "Find the minimum value of the given field"
+            )(
+                "max",
+                po::value<std::string>(&aggregation_field)->value_name("FIELD"),
+                "Find the maximum value of the given field"
             );
             // clang-format on
             search_options.add(aggregation_options);
@@ -952,6 +965,14 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
                           << std::endl;
                 std::cerr << "  " << m_program_name << R"( s archives-dir "level: INFO")"
                           << " --count" << std::endl;
+                std::cerr << std::endl;
+
+                std::cerr << "  # Search archives in archives-dir for logs matching a KQL query"
+                             R"( "level: INFO" and output the maximum value of field "latency" to)"
+                             " stdout"
+                          << std::endl;
+                std::cerr << "  " << m_program_name << R"( s archives-dir "level: INFO")"
+                          << " --max latency" << std::endl;
 
                 po::options_description visible_options;
                 visible_options.add(general_options);
@@ -1026,9 +1047,10 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
                 throw std::invalid_argument("clp-s only supports one output handler at a time");
             }
 
-            m_aggregation_type = parse_aggregation_options(
+            m_aggregator = parse_aggregation_options(
                     parsed_command_line_options,
-                    m_count_by_time_bucket_size_ms
+                    count_by_time_bucket_size_millisecs,
+                    aggregation_field
             );
 
             for (auto const& [output_handler_name, output_handler_options] : output_options_map) {
@@ -1122,31 +1144,50 @@ void CommandLineArguments::parse_network_dest_output_handler_options(
 
 auto CommandLineArguments::parse_aggregation_options(
         po::variables_map const& parsed_options,
-        int64_t count_by_time_bucket_size_ms
-) -> std::optional<AggregationType> {
-    std::optional<AggregationType> aggregation_type;
-    if (parsed_options.count("count")) {
-        aggregation_type = AggregationType::Count;
-    }
-    if (parsed_options.count("count-by-time")) {
-        if (aggregation_type.has_value()) {
+        int64_t count_by_time_bucket_size_millisecs,
+        std::string_view aggregation_field
+) -> std::optional<Aggregator> {
+    std::optional<Aggregator> aggregator;
+    auto const set_aggregator = [&](Aggregator value) {
+        if (aggregator.has_value()) {
             throw std::invalid_argument(
-                    "The --count-by-time and --count options are mutually exclusive."
+                    "The --count, --count-by-time, --min, and --max options are mutually exclusive."
             );
         }
+        aggregator = std::move(value);
+    };
+    auto const validate_aggregation_field = [&]() {
+        if (aggregation_field.empty()) {
+            throw std::invalid_argument("The --min and --max options require a field.");
+        }
+        if (search::ast::has_unescaped_wildcards(aggregation_field)) {
+            throw std::invalid_argument("The --min and --max field must not contain wildcards.");
+        }
+    };
 
-        if (count_by_time_bucket_size_ms <= 0) {
+    if (parsed_options.count("count")) {
+        set_aggregator(CountAggregator{});
+    }
+    if (parsed_options.count("count-by-time")) {
+        if (count_by_time_bucket_size_millisecs <= 0) {
             throw std::invalid_argument("Value for count-by-time must be greater than zero.");
         }
-
-        aggregation_type = AggregationType::CountByTime;
+        set_aggregator(CountByTimeAggregator{count_by_time_bucket_size_millisecs});
+    }
+    if (parsed_options.count("min")) {
+        validate_aggregation_field();
+        set_aggregator(MinMaxAggregator{false, aggregation_field});
     }
-    return aggregation_type;
+    if (parsed_options.count("max")) {
+        validate_aggregation_field();
+        set_aggregator(MinMaxAggregator{true, aggregation_field});
+    }
+    return aggregator;
 }
 
 auto CommandLineArguments::reject_aggregation_for_handler(std::string_view handler_name) const
         -> void {
-    if (m_aggregation_type.has_value()) {
+    if (m_aggregator.has_value()) {
         throw std::invalid_argument(
                 fmt::format("The {} output handler does not support aggregations.", handler_name)
         );
@@ -1183,7 +1224,9 @@ void CommandLineArguments::parse_reducer_output_handler_options(
         throw std::invalid_argument("job-id cannot be negative.");
     }
 
-    if (false == m_aggregation_type.has_value()) {
+    if (false == m_aggregator.has_value()
+        || std::holds_alternative<MinMaxAggregator>(m_aggregator.value()))
+    {
         throw std::invalid_argument(
                 "The reducer output handler currently only supports count and count-by-time"
                 " aggregations."