Prepare a VAST v2.4.2 release (#3048)

This PR backports #3047 and prepares the repository for a new patch
release on the v2.4.x branch.

changelog/v2.4.2/changes/3047.md

@@ -0,0 +1 @@
+VAST's rebuilding and compaction features now interfere less with queries.

libvast/include/vast/query_context.hpp

@@ -127,13 +127,12 @@ struct query_context {
   vast::ids ids = {};
 
   struct priority {
-    constexpr static uint8_t high = 100;
-    constexpr static uint8_t normal = 25;
-    constexpr static uint8_t low = 1;
+    constexpr static uint64_t normal = 1'000;
+    constexpr static uint64_t low = 1;
   };
 
   /// The query priority.
-  uint8_t priority = priority::normal;
+  uint64_t priority = priority::normal;
 
   /// The issuer of the query.
   std::string issuer = {};

libvast/include/vast/system/index.hpp

@@ -225,7 +225,9 @@ struct index_state {
 
   // -- query handling ---------------------------------------------------------
 
-  void schedule_lookups();
+  /// Schedules partitions for lookups. Returns the number of newly scheduled
+  /// partitions.
+  [[nodiscard]] auto schedule_lookups() -> size_t;
 
   // -- introspection ----------------------------------------------------------
 

libvast/src/query_queue.cpp

@@ -23,7 +23,10 @@ std::size_t memusage(const std::vector<query_queue::entry>& entries) {
 
 bool operator<(const query_queue::entry& lhs,
                const query_queue::entry& rhs) noexcept {
-  return lhs.priority < rhs.priority;
+  const auto lhs_num_queries = lhs.queries.size();
+  const auto rhs_num_queries = rhs.queries.size();
+  return std::tie(lhs.priority, lhs_num_queries)
+         < std::tie(rhs.priority, rhs_num_queries);
 }
 
 bool operator==(const query_queue::entry& lhs, const uuid& rhs) noexcept {

libvast/src/system/index.cpp

@@ -888,20 +888,21 @@ void index_state::add_partition_creation_listener(
 
 // -- query handling ---------------------------------------------------------
 
-void index_state::schedule_lookups() {
+auto index_state::schedule_lookups() -> size_t {
   if (!pending_queries.has_work())
-    return;
+    return 0u;
   auto t = timer::start(scheduler_measurement);
   auto num_scheduled = size_t{0};
   auto on_return = caf::detail::make_scope_guard([&] {
     t.stop(num_scheduled);
   });
+  const size_t previous_partition_lookups = running_partition_lookups;
   while (running_partition_lookups < max_concurrent_partition_lookups) {
     // 1. Get the partition with the highest accumulated priority.
     auto next = pending_queries.next();
     if (!next) {
       VAST_DEBUG("{} did not find a partition to query", *self);
-      return;
+      break;
     }
     auto immediate_completion = [&](const query_queue::entry& x) {
       for (auto qid : x.queries)
@@ -970,7 +971,10 @@ void index_state::schedule_lookups() {
         *cnt -= 1;
         if (*cnt == 0) {
           --running_partition_lookups;
-          schedule_lookups();
+          const auto num_scheduled = schedule_lookups();
+          VAST_DEBUG("{} scheduled {} partitions after completion of a "
+                     "previously scheduled lookup",
+                     *self, num_scheduled);
         }
       };
       self
@@ -998,6 +1002,8 @@ void index_state::schedule_lookups() {
     running_partition_lookups++;
     num_scheduled++;
   }
+  VAST_ASSERT_CHEAP(running_partition_lookups >= previous_partition_lookups);
+  return running_partition_lookups - previous_partition_lookups;
 }
 
 // -- introspection ----------------------------------------------------------
@@ -1007,7 +1013,6 @@ namespace {
 struct query_counters {
   size_t num_low_prio = 0;
   size_t num_normal_prio = 0;
-  size_t num_high_prio = 0;
   size_t num_custom_prio = 0;
 };
 
@@ -1018,8 +1023,6 @@ auto get_query_counters(const query_queue& pending_queries) {
       result.num_low_prio++;
     else if (q.query_context.priority == query_context::priority::normal)
       result.num_normal_prio++;
-    else if (q.query_context.priority == query_context::priority::high)
-      result.num_high_prio++;
     else
       result.num_custom_prio++;
   }
@@ -1040,7 +1043,6 @@ void index_state::send_report() {
       {"scheduler.backlog.custom", query_counters.num_custom_prio},
       {"scheduler.backlog.low", query_counters.num_low_prio},
       {"scheduler.backlog.normal", query_counters.num_normal_prio},
-      {"scheduler.backlog.high", query_counters.num_high_prio},
       {"scheduler.partition.pending", pending_queries.num_partitions()},
       {"scheduler.partition.materializations", materializations},
       {"scheduler.partition.lookups", counters.partition_lookups},
@@ -1549,7 +1551,10 @@ index(index_actor::stateful_pointer<index_state> self,
                   std::move(candidates)))
               rp.deliver(err);
             rp.deliver(query_cursor{query_id, num_candidates, scheduled});
-            self->state.schedule_lookups();
+            const auto num_scheduled = self->state.schedule_lookups();
+            VAST_DEBUG("{} scheduled {} partitions for lookup after a new "
+                       "query came in",
+                       *self, num_scheduled);
           },
           [rp](const caf::error& e) mutable {
             rp.deliver(caf::make_error(
@@ -1568,7 +1573,10 @@ index(index_actor::stateful_pointer<index_state> self,
       if (auto err
           = self->state.pending_queries.activate(query_id, num_partitions))
         VAST_WARN("{} can't activate unknown query: {}", *self, err);
-      self->state.schedule_lookups();
+      const auto num_scheduled = self->state.schedule_lookups();
+      VAST_DEBUG("{} scheduled {} partitions following the request to "
+                 "activate {} partitions for query {}",
+                 *self, num_scheduled, num_partitions, query_id);
     },
     [self](atom::erase, uuid partition_id) -> caf::result<atom::done> {
       VAST_VERBOSE("{} erases partition {}", *self, partition_id);
@@ -1821,7 +1829,9 @@ index(index_actor::stateful_pointer<index_state> self,
         pipeline->name(), partition_transfomer, match_everything);
       auto transform_id = self->state.pending_queries.create_query_id();
       query_context.id = transform_id;
-      query_context.priority = query_context::priority::high;
+      // We set the query priority for partition transforms to zero so they
+      // always get less priority than queries.
+      query_context.priority = 0;
       VAST_DEBUG("{} emplaces {} for pipeline {}", *self, query_context,
                  pipeline->name());
       auto input_size = detail::narrow_cast<uint32_t>(old_partition_ids.size());
@@ -1833,7 +1843,10 @@ index(index_actor::stateful_pointer<index_state> self,
                     .requested_partitions = input_size},
         std::vector{old_partition_ids});
       VAST_ASSERT(err == caf::none);
-      self->state.schedule_lookups();
+      const auto num_scheduled = self->state.schedule_lookups();
+      VAST_DEBUG("{} scheduled {} partitions following a request to "
+                 "transform partitions",
+                 *self, num_scheduled);
       auto marker_path = self->state.marker_path(transform_id);
       auto rp = self->make_response_promise<std::vector<partition_info>>();
       auto deliver

libvast/test/query_queue.cpp

@@ -75,7 +75,7 @@ uuid make_insert(query_queue& q, std::vector<uuid>&& candidates) {
 
 uuid make_insert(query_queue& q, std::vector<uuid>&& candidates,
                  uint32_t taste_size,
-                 uint8_t priority = query_context::priority::normal) {
+                 uint64_t priority = query_context::priority::normal) {
   uint32_t cands_size = candidates.size();
   auto query_context = make_random_query_context();
   query_context.priority = priority;

python/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "vast"
-version = "2.4.1"
+version = "2.4.2"
 description = "A security telemetry engine for detection and response"
 authors = ["Tenzir <engineering@tenzir.com>"]
 maintainers = ["Tenzir <engineering@tenzir.com>"]

version.json

@@ -4,7 +4,7 @@
     "annotated git tag without the leading 'v'.",
     "This value gets updated automatically by `scripts/prepare-release`."
   ],
-  "vast-version-fallback": "2.4.1",
+  "vast-version-fallback": "2.4.2",
   "vast-partition-version_COMMENT": [
     "The partition version. This number must be bumped alongside the release",
     "version for releases that contain major format changes to the on-disk",