feat(clp-s): Support per-archive group-by count aggregation output to stdout and the results cache.

components/core/src/clp_s/AggregationSink.cpp

@@ -1,5 +1,6 @@
 #include "AggregationSink.hpp"
 
+#include <cstdint>
 #include <iostream>
 #include <string_view>
 #include <variant>
@@ -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

components/core/src/clp_s/AggregationSink.hpp

@@ -1,6 +1,7 @@
 #ifndef CLP_S_AGGREGATIONSINK_HPP
 #define CLP_S_AGGREGATIONSINK_HPP
 
+#include <cstdint>
 #include <string>
 #include <string_view>
 #include <vector>
@@ -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<bsoncxx::document::value> m_results;
+    ErrorCode m_flush_error{ErrorCode::ErrorCodeSuccess};
 };
 }  // namespace clp_s
 

components/core/src/clp_s/CommandLineArguments.cpp

@@ -794,6 +794,14 @@ CommandLineArguments::parse_arguments(int argc, char const** argv) {
                 "max",
                 po::value<std::string>(&aggregation_field)->value_name("FIELD"),
                 "Find the maximum value of the given field"
+            )(
+                "unique",
+                po::value<std::string>(&aggregation_field)->value_name("FIELD"),
+                "Find the distinct values of the given field"
+            )(
+                "count-by",
+                po::value<std::string>(&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<MinMaxAggregator>(m_aggregator.value()))
+        || (false == std::holds_alternative<CountAggregator>(m_aggregator.value())
+            && false == std::holds_alternative<CountByTimeAggregator>(m_aggregator.value())))
     {
         throw std::invalid_argument(
                 "The reducer output handler currently only supports count and count-by-time"

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,