diff --git a/components/core/src/clp_s/AggregationSink.cpp b/components/core/src/clp_s/AggregationSink.cpp index cb06467b5..435a80c48 100644 --- a/components/core/src/clp_s/AggregationSink.cpp +++ b/components/core/src/clp_s/AggregationSink.cpp @@ -1,5 +1,6 @@ #include "AggregationSink.hpp" +#include #include #include #include @@ -27,12 +28,36 @@ auto StdoutSink::write(AggregationResult const& result) -> void { std::cout << document.dump() << '\n'; } -ResultsCacheSink::ResultsCacheSink(string_view uri, string_view collection, string_view archive_id) - : m_archive_id{archive_id} { +ResultsCacheSink::ResultsCacheSink( + string_view uri, + string_view collection, + uint64_t batch_size, + string_view archive_id +) + : m_batch_size{batch_size}, + m_archive_id{archive_id} { m_collection = connect_to_results_cache(uri, collection, m_client); } +auto ResultsCacheSink::flush_buffer() -> ErrorCode { + if (m_results.empty()) { + return ErrorCode::ErrorCodeSuccess; + } + + try { + m_collection.insert_many(m_results); + } catch (mongocxx::exception const&) { + return ErrorCode::ErrorCodeFailureDbBulkWrite; + } + m_results.clear(); + return ErrorCode::ErrorCodeSuccess; +} + auto ResultsCacheSink::write(AggregationResult const& result) -> void { + if (ErrorCode::ErrorCodeSuccess != m_flush_error) { + return; + } + bsoncxx::builder::basic::document document; document.append( bsoncxx::builder::basic::kvp(constants::results_cache::search::cArchiveId, m_archive_id) @@ -46,19 +71,16 @@ auto ResultsCacheSink::write(AggregationResult const& result) -> void { ); } m_results.push_back(document.extract()); -} -auto ResultsCacheSink::finish() -> ErrorCode { - if (m_results.empty()) { - return ErrorCode::ErrorCodeSuccess; + if (m_results.size() >= m_batch_size) { + m_flush_error = flush_buffer(); } +} - try { - m_collection.insert_many(m_results); - m_results.clear(); - } catch (mongocxx::exception const& e) { - return ErrorCode::ErrorCodeFailureDbBulkWrite; +auto ResultsCacheSink::finish() -> ErrorCode { + if (ErrorCode::ErrorCodeSuccess != m_flush_error) { + return m_flush_error; } - return ErrorCode::ErrorCodeSuccess; + return flush_buffer(); } } // namespace clp_s diff --git a/components/core/src/clp_s/AggregationSink.hpp b/components/core/src/clp_s/AggregationSink.hpp index 794c08458..004b5ef4b 100644 --- a/components/core/src/clp_s/AggregationSink.hpp +++ b/components/core/src/clp_s/AggregationSink.hpp @@ -1,6 +1,7 @@ #ifndef CLP_S_AGGREGATIONSINK_HPP #define CLP_S_AGGREGATIONSINK_HPP +#include #include #include #include @@ -73,25 +74,41 @@ class ResultsCacheSink : public AggregationSink { ResultsCacheSink( std::string_view uri, std::string_view collection, + uint64_t batch_size, std::string_view archive_id ); // Methods implementing AggregationSink + /** + * Buffers a result document, flushing the buffer to the database once it reaches the batch + * size. Documents are dropped after an earlier flush failure; the error surfaces in `finish()`. + * @param result The result document to write. + */ auto write(AggregationResult const& result) -> void override; /** - * Flushes the buffered result documents. + * Flushes any remaining buffered result documents. * @return ErrorCodeSuccess on success - * @return ErrorCodeFailureDbBulkWrite on database error + * @return ErrorCodeFailureDbBulkWrite if this flush or an earlier batched flush failed */ [[nodiscard]] auto finish() -> ErrorCode override; private: + // Methods + /** + * Inserts the buffered result documents into the collection. + * @return ErrorCodeSuccess on success + * @return ErrorCodeFailureDbBulkWrite on database error + */ + [[nodiscard]] auto flush_buffer() -> ErrorCode; + // Data members mongocxx::client m_client; mongocxx::collection m_collection; + uint64_t m_batch_size; std::string m_archive_id; std::vector m_results; + ErrorCode m_flush_error{ErrorCode::ErrorCodeSuccess}; }; } // namespace clp_s diff --git a/components/core/src/clp_s/CommandLineArguments.cpp b/components/core/src/clp_s/CommandLineArguments.cpp index 3d1afe185..f607efcff 100644 --- a/components/core/src/clp_s/CommandLineArguments.cpp +++ b/components/core/src/clp_s/CommandLineArguments.cpp @@ -794,6 +794,14 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) { "max", po::value(&aggregation_field)->value_name("FIELD"), "Find the maximum value of the given field" + )( + "unique", + po::value(&aggregation_field)->value_name("FIELD"), + "Find the distinct values of the given field" + )( + "count-by", + po::value(&aggregation_field)->value_name("FIELD"), + "Count the number of results grouped by the value of the given field" ); // clang-format on search_options.add(aggregation_options); @@ -973,6 +981,14 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) { << std::endl; std::cerr << " " << m_program_name << R"( s archives-dir "level: INFO")" << " --max latency" << 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 distinct values of field "latency")" + " to stdout" + << std::endl; + std::cerr << " " << m_program_name << R"( s archives-dir "level: INFO")" + << " --unique latency" << std::endl; po::options_description visible_options; visible_options.add(general_options); @@ -1151,17 +1167,22 @@ auto CommandLineArguments::parse_aggregation_options( auto const set_aggregator = [&](Aggregator value) { if (aggregator.has_value()) { throw std::invalid_argument( - "The --count, --count-by-time, --min, and --max options are mutually exclusive." + "The --count, --count-by-time, --min, --max, --unique, and --count-by 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."); + throw std::invalid_argument( + "The --min, --max, --unique, and --count-by 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."); + throw std::invalid_argument( + "The --min, --max, --unique, and --count-by field must not contain wildcards." + ); } }; @@ -1182,6 +1203,14 @@ auto CommandLineArguments::parse_aggregation_options( validate_aggregation_field(); set_aggregator(MinMaxAggregator{true, aggregation_field}); } + if (parsed_options.count("unique")) { + validate_aggregation_field(); + set_aggregator(UniqueAggregator{aggregation_field}); + } + if (parsed_options.count("count-by")) { + validate_aggregation_field(); + set_aggregator(GroupByCountAggregator{aggregation_field}); + } return aggregator; } @@ -1225,7 +1254,8 @@ void CommandLineArguments::parse_reducer_output_handler_options( } if (false == m_aggregator.has_value() - || std::holds_alternative(m_aggregator.value())) + || (false == std::holds_alternative(m_aggregator.value()) + && false == std::holds_alternative(m_aggregator.value()))) { throw std::invalid_argument( "The reducer output handler currently only supports count and count-by-time" diff --git a/components/core/src/clp_s/CommandLineArguments.hpp b/components/core/src/clp_s/CommandLineArguments.hpp index 30d12b998..4ecfcb6ea 100644 --- a/components/core/src/clp_s/CommandLineArguments.hpp +++ b/components/core/src/clp_s/CommandLineArguments.hpp @@ -162,10 +162,10 @@ class CommandLineArguments { * @param parsed_options * @param count_by_time_bucket_size_millisecs Bucket size for count-by-time. Only used by that * option. - * @param aggregation_field Field for min/max. Only used by those options. + * @param aggregation_field Field for min/max/unique. Only used by those options. * @return The requested aggregation, or std::nullopt if none was requested. * @throws std::invalid_argument if multiple aggregations are specified, the bucket size is - * non-positive, or a min/max field is empty or contains wildcards. + * non-positive, or a min/max/unique field is empty or contains wildcards. */ [[nodiscard]] static auto parse_aggregation_options( boost::program_options::variables_map const& parsed_options, diff --git a/components/core/src/clp_s/aggregators.cpp b/components/core/src/clp_s/aggregators.cpp index 9976ddb9e..fa8c123eb 100644 --- a/components/core/src/clp_s/aggregators.cpp +++ b/components/core/src/clp_s/aggregators.cpp @@ -1,6 +1,7 @@ #include "aggregators.hpp" #include +#include #include #include #include @@ -18,6 +19,82 @@ using std::string; using std::string_view; namespace clp_s { +namespace { +/** + * Tokenizes an aggregation's target field into its key path. + * @param field + * @param field_path Returns the path's keys. + * @throws std::invalid_argument if `field` is malformed, or if it names a non-default namespace. + */ +auto tokenize_aggregation_field(string_view field, std::vector& field_path) -> void { + string descriptor_namespace; + if (false + == search::ast::tokenize_column_descriptor(string{field}, field_path, descriptor_namespace)) + { + throw std::invalid_argument("Invalid aggregation field: " + string{field}); + } + if (false == descriptor_namespace.empty()) { + throw std::invalid_argument( + "The aggregation field must be in the default namespace; namespaced fields (e.g. " + "the auto-generated \"@\" namespace) are not supported." + ); + } +} + +/** + * Parses `message` as JSON and locates the value at `field_path`. + * @param message A matched record, marshalled to a JSON string. + * @param field_path + * @param doc Returns the parsed document. + * @return The value at `field_path`. + * @return nullptr if `message` isn't valid JSON, or if `field_path` doesn't resolve to a value. + */ +auto +find_field_value(string_view message, std::vector const& field_path, nlohmann::json& doc) + -> nlohmann::json const* { + try { + doc = nlohmann::json::parse(message); + } catch (nlohmann::json::exception const&) { + return nullptr; + } + + nlohmann::json const* current{&doc}; + for (auto const& key : field_path) { + if (false == current->is_object()) { + return nullptr; + } + auto const it{current->find(key)}; + if (current->end() == it) { + return nullptr; + } + current = &it.value(); + } + return current; +} + +/** + * Converts a scalar JSON value to an `AggregationValue`. + * @param value + * @return `value` as an `AggregationValue` if it's an integer, float, string, or boolean. + * @return std::nullopt otherwise. + */ +auto to_aggregation_value(nlohmann::json const& value) -> std::optional { + if (value.is_number_integer()) { + return value.get(); + } + if (value.is_number_float()) { + return value.get(); + } + if (value.is_string()) { + return value.get(); + } + if (value.is_boolean()) { + return value.get(); + } + return std::nullopt; +} +} // namespace + auto CountAggregator::get_results() const -> std::vector { if (0 == m_count) { return {}; @@ -39,25 +116,39 @@ auto CountByTimeAggregator::get_results() const -> std::vector void { + nlohmann::json doc; + auto const* const value{find_field_value(message, m_field_path, doc)}; + if (nullptr == value) { + return; + } + auto aggregation_value{to_aggregation_value(*value)}; + if (aggregation_value.has_value()) { + m_counts[std::move(aggregation_value.value())] += 1; + } +} + +auto GroupByCountAggregator::get_results() const -> std::vector { + std::vector results; + results.reserve(m_counts.size()); + for (auto const& [value, count] : m_counts) { + AggregationResult result; + result.emplace_back(constants::results_cache::search::cField, m_field); + result.emplace_back(constants::results_cache::search::cValue, value); + result.emplace_back(constants::results_cache::search::cCount, count); + results.push_back(std::move(result)); + } + return results; +} + MinMaxAggregator::MinMaxAggregator(bool find_max, string_view field) : m_find_max{find_max}, m_field{field} { - string descriptor_namespace; - if (false - == search::ast::tokenize_column_descriptor( - string{field}, - m_field_path, - descriptor_namespace - )) - { - throw std::invalid_argument("Invalid --min/--max field: " + string{field}); - } - if (false == descriptor_namespace.empty()) { - throw std::invalid_argument( - "The --min/--max field must be in the default namespace; namespaced fields (e.g. " - "the auto-generated \"@\" namespace) are not supported." - ); - } + tokenize_aggregation_field(field, m_field_path); } auto MinMaxAggregator::beats_extreme(Extreme candidate) const -> bool { @@ -74,29 +165,13 @@ auto MinMaxAggregator::beats_extreme(Extreme candidate) const -> bool { auto MinMaxAggregator::add_record(string_view message, epochtime_t) -> void { nlohmann::json doc; - try { - doc = nlohmann::json::parse(message); - } catch (nlohmann::json::exception const&) { - return; - } - - nlohmann::json const* node{&doc}; - for (auto const& key : m_field_path) { - if (false == node->is_object()) { - return; - } - auto const it{node->find(key)}; - if (node->end() == it) { - return; - } - node = &it.value(); - } - - if (false == node->is_number()) { + auto const* const value{find_field_value(message, m_field_path, doc)}; + if (nullptr == value || false == value->is_number()) { return; } Extreme const candidate{ - node->is_number_integer() ? Extreme{node->get()} : Extreme{node->get()} + value->is_number_integer() ? Extreme{value->get()} + : Extreme{value->get()} }; if (false == m_extreme.has_value() || beats_extreme(candidate)) { m_extreme = candidate; @@ -119,4 +194,32 @@ auto MinMaxAggregator::get_results() const -> std::vector { result.emplace_back(key, value); return {std::move(result)}; } + +UniqueAggregator::UniqueAggregator(string_view field) : m_field{field} { + tokenize_aggregation_field(field, m_field_path); +} + +auto UniqueAggregator::add_record(string_view message, epochtime_t) -> void { + nlohmann::json doc; + auto const* const value{find_field_value(message, m_field_path, doc)}; + if (nullptr == value) { + return; + } + auto aggregation_value{to_aggregation_value(*value)}; + if (aggregation_value.has_value()) { + m_values.emplace(std::move(aggregation_value.value())); + } +} + +auto UniqueAggregator::get_results() const -> std::vector { + std::vector results; + results.reserve(m_values.size()); + for (auto const& value : m_values) { + AggregationResult result; + result.emplace_back(constants::results_cache::search::cField, m_field); + result.emplace_back(constants::results_cache::search::cValue, value); + results.push_back(std::move(result)); + } + return results; +} } // namespace clp_s diff --git a/components/core/src/clp_s/aggregators.hpp b/components/core/src/clp_s/aggregators.hpp index f98858f10..740617afd 100644 --- a/components/core/src/clp_s/aggregators.hpp +++ b/components/core/src/clp_s/aggregators.hpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -18,7 +19,7 @@ namespace clp_s { /** * A single typed value in an aggregation's result document. */ -using AggregationValue = std::variant; +using AggregationValue = std::variant; /** * One aggregation's result document: an ordered list of typed key-value pairs. @@ -27,7 +28,7 @@ using AggregationResult = std::vector>; /** * Requirements a type must satisfy to be used as an aggregator. - * @tparam AggregatorType The type of the aggregator. + * @tparam AggregatorType */ template concept AggregatorReq = requires( @@ -120,6 +121,31 @@ class CountByTimeAggregator { std::map m_bucket_counts; }; +/** + * Counts the number of matched records grouped by the value of a target field. + */ +class GroupByCountAggregator { +public: + // Static constants + static constexpr bool cNeedsMetadata{false}; + static constexpr bool cNeedsMarshalledRecord{true}; + + // Constructors + explicit GroupByCountAggregator(std::string_view field); + + // Methods + auto add_record(std::string_view message, [[maybe_unused]] epochtime_t timestamp_millisecs) + -> void; + + [[nodiscard]] auto get_results() const -> std::vector; + +private: + // Data members + std::string m_field; + std::vector m_field_path; + std::map m_counts; +}; + /** * Tracks the minimum or maximum value of a target field across matched records. */ @@ -156,10 +182,40 @@ class MinMaxAggregator { std::optional m_extreme; }; +/** + * Collects the distinct values of a target field across matched records. + */ +class UniqueAggregator { +public: + // Static constants + static constexpr bool cNeedsMetadata{false}; + static constexpr bool cNeedsMarshalledRecord{true}; + + // Constructors + explicit UniqueAggregator(std::string_view field); + + // Methods + auto add_record(std::string_view message, [[maybe_unused]] epochtime_t timestamp_millisecs) + -> void; + + [[nodiscard]] auto get_results() const -> std::vector; + +private: + // Data members + std::string m_field; + std::vector m_field_path; + std::set m_values; +}; + /** * One of the supported aggregators that a search can apply to its matched records. */ -using Aggregator = std::variant; +using Aggregator = std::variant< + CountAggregator, + CountByTimeAggregator, + GroupByCountAggregator, + MinMaxAggregator, + UniqueAggregator>; } // namespace clp_s #endif // CLP_S_AGGREGATORS_HPP diff --git a/components/core/src/clp_s/archive_constants.hpp b/components/core/src/clp_s/archive_constants.hpp index 0ab8cf824..0dbf3d4c6 100644 --- a/components/core/src/clp_s/archive_constants.hpp +++ b/components/core/src/clp_s/archive_constants.hpp @@ -63,6 +63,7 @@ constexpr char cCount[]{"count"}; constexpr char cMin[]{"min"}; constexpr char cMax[]{"max"}; constexpr char cField[]{"field"}; +constexpr char cValue[]{"value"}; } // namespace results_cache::search } // namespace clp_s::constants #endif // CLP_S_ARCHIVE_CONSTANTS_HPP diff --git a/components/core/src/clp_s/clp-s.cpp b/components/core/src/clp_s/clp-s.cpp index d73b7d4d2..b538495a9 100644 --- a/components/core/src/clp_s/clp-s.cpp +++ b/components/core/src/clp_s/clp-s.cpp @@ -346,6 +346,7 @@ bool search_archive( std::make_unique( options.uri, options.collection, + options.batch_size, archive_reader->get_archive_id() ) );