debug_bundle: Reloading metadata on restart

Added functionality that will reload metadata from the kvstore after the
service restarts.

Signed-off-by: Michael Boquard <michael@redpanda.com>

Author: michael-redpanda

src/v/debug_bundle/debug_bundle_service.cc

@@ -131,6 +131,12 @@ ss::future<bytes> calculate_sha256_sum(std::string_view path) {
 
     co_return std::move(ctx).final();
 }
+
+ss::future<bool>
+validate_sha256_checksum(std::string_view path, bytes_view checksum) {
+    auto sum = co_await calculate_sha256_sum(path);
+    co_return sum == checksum;
+}
 } // namespace
 
 struct service::output_handler {
@@ -159,11 +165,19 @@ class service::debug_bundle_process {
       , _output_file_path(std::move(output_file_path))
       , _created_time(clock::now()) {
         _rpk_process->set_stdout_consumer(
-          output_handler{.output_buffer = _cout});
+          output_handler{.output_buffer = std::ref(_cout)});
         _rpk_process->set_stderr_consumer(
-          output_handler{.output_buffer = _cerr});
+          output_handler{.output_buffer = std::ref(_cerr)});
     }
 
+    explicit debug_bundle_process(metadata md)
+      : _job_id(md.job_id)
+      , _wait_result(md.get_wait_status())
+      , _output_file_path(md.debug_bundle_file_path)
+      , _cout(std::move(md.cout))
+      , _cerr(std::move(md.cerr))
+      , _created_time(md.get_created_at()) {}
+
     debug_bundle_process() = delete;
     debug_bundle_process(debug_bundle_process&&) = default;
     debug_bundle_process& operator=(debug_bundle_process&&) = default;
@@ -272,12 +286,15 @@ ss::future<> service::start() {
           _rpk_path_binding().native());
     }
 
+    co_await maybe_reload_previous_run();
+
     lg.debug("Service started");
 }
 
 ss::future<> service::stop() {
     lg.debug("Service stopping");
     if (ss::this_shard_id() == service_shard) {
+        auto units = co_await _process_control_mutex.get_units();
         if (is_running()) {
             try {
                 co_await _rpk_process->terminate(1s);
@@ -290,6 +307,7 @@ ss::future<> service::stop() {
         }
     }
     co_await _gate.close();
+    _rpk_process.reset(nullptr);
 }
 
 ss::future<result<void>> service::initiate_rpk_debug_bundle_collection(
@@ -615,6 +633,10 @@ ss::future<> service::cleanup_previous_run() const {
         co_await ss::remove_file(debug_bundle_file.native());
     }
 
+    co_await remove_kvstore_entry();
+}
+
+ss::future<> service::remove_kvstore_entry() const {
     co_await _kvstore->remove(
       storage::kvstore::key_space::debug_bundle,
       bytes::from_string(debug_bundle_metadata_key));
@@ -705,5 +727,56 @@ ss::future<> service::handle_wait_result(
         co_return;
     }
 }
+ss::future<> service::maybe_reload_previous_run() {
+    auto md_buf = _kvstore->get(
+      storage::kvstore::key_space::debug_bundle,
+      bytes::from_string(debug_bundle_metadata_key));
+
+    if (!md_buf) {
+        vlog(lg.debug, "No previous run detected");
+        co_return;
+    }
+
+    iobuf_parser p(std::move(*md_buf));
+    auto md = serde::read<metadata>(p);
+
+    auto run_was_successful = [&md]() {
+        auto wait_status = md.get_wait_status();
+        return std::holds_alternative<ss::experimental::process::wait_exited>(
+                 wait_status)
+               && std::get<ss::experimental::process::wait_exited>(wait_status)
+                      .exit_code
+                    == 0;
+    }();
+
+    if (
+      run_was_successful
+      && !co_await ss::file_exists(md.debug_bundle_file_path)) {
+        vlog(
+          lg.debug,
+          "Debug bundle file {} does not exist, cannot reload metadata",
+          md.debug_bundle_file_path);
+        co_return co_await remove_kvstore_entry();
+    }
+
+    if (
+      run_was_successful
+      && !co_await validate_sha256_checksum(
+        md.debug_bundle_file_path, md.sha256_checksum)) {
+        vlog(
+          lg.debug,
+          "Debug bundle file {} checksum mismatch",
+          md.debug_bundle_file_path);
+        co_await ss::remove_file(md.debug_bundle_file_path);
+        co_return co_await remove_kvstore_entry();
+    }
+
+    vlog(
+      lg.info,
+      "Detected a valid previous run that was {}successful",
+      run_was_successful ? "" : "un");
+
+    _rpk_process = std::make_unique<debug_bundle_process>(std::move(md));
+}
 
 } // namespace debug_bundle

src/v/debug_bundle/debug_bundle_service.h

@@ -143,6 +143,11 @@ class service final : public ss::peering_sharded_service<service> {
      */
     ss::future<> cleanup_previous_run() const;
 
+    /**
+     * @brief Removes debug bundle entry from kvstore
+     */
+    ss::future<> remove_kvstore_entry() const;
+
     /**
      * @brief Set the metadata object within the kvstore
      *
@@ -172,6 +177,15 @@ class service final : public ss::peering_sharded_service<service> {
     ss::future<> handle_wait_result(
       job_id_t job_id, ss::experimental::process::wait_status result);
 
+    /**
+     * @brief Attempts to reload a previous run
+     *
+     * If a previous run exists in the kvstore upon service start, will attempt
+     * to reload the metadata and make that debug bundle available to the user
+     *
+     */
+    ss::future<> maybe_reload_previous_run();
+
 private:
     /// Handler used to emplace stdout/stderr into a buffer
     struct output_handler;

src/v/debug_bundle/tests/debug_bundle_service_test.cc

@@ -83,7 +83,8 @@ struct debug_bundle_service_fixture : public seastar_test {
     }
 
     ss::future<debug_bundle::result<debug_bundle::debug_bundle_status_data>>
-    wait_for_process_to_finish(const std::chrono::seconds timeout) {
+    wait_for_process_to_finish(
+      const std::chrono::seconds timeout = std::chrono::seconds{10}) {
         const auto start_time = debug_bundle::clock::now();
         while (debug_bundle::clock::now() - start_time <= timeout) {
             auto status = co_await _service.local().rpk_debug_bundle_status();
@@ -190,7 +191,7 @@ TEST_F_CORO(debug_bundle_service_started_fixture, run_process) {
     EXPECT_EQ(status.assume_value().job_id, job_id);
     EXPECT_EQ(status.assume_value().file_name, fmt::format("{}.zip", job_id));
 
-    ASSERT_NO_THROW_CORO(status = co_await wait_for_process_to_finish(10s));
+    ASSERT_NO_THROW_CORO(status = co_await wait_for_process_to_finish());
 
     ASSERT_TRUE_CORO(status.has_value()) << res.assume_error().message();
 
@@ -512,7 +513,7 @@ TEST_F_CORO(
     auto term_res = co_await _service.local().cancel_rpk_debug_bundle(job_id);
     ASSERT_TRUE_CORO(term_res.has_value()) << term_res.assume_error().message();
 
-    ASSERT_NO_THROW_CORO(status = co_await wait_for_process_to_finish(10s));
+    ASSERT_NO_THROW_CORO(status = co_await wait_for_process_to_finish());
 
     ASSERT_TRUE_CORO(status.has_value()) << res.assume_error().message();
 
@@ -725,3 +726,128 @@ TEST_F_CORO(debug_bundle_service_started_fixture, validate_metadata) {
     EXPECT_EQ(
       metadata.sha256_checksum, co_await calculate_sha256_sum(job_file));
 }
+
+TEST_F_CORO(debug_bundle_service_started_fixture, validate_restart) {
+    debug_bundle::job_id_t job_id(uuid_t::create());
+
+    co_await run_bundle(job_id);
+
+    auto status = co_await _service.local().rpk_debug_bundle_status();
+    ASSERT_TRUE_CORO(status.has_value()) << status.assume_error().message();
+
+    co_await restart_service();
+
+    auto status_restart = co_await _service.local().rpk_debug_bundle_status();
+    ASSERT_TRUE_CORO(status_restart.has_value())
+      << status_restart.assume_error().message();
+
+    EXPECT_EQ(status.assume_value(), status_restart.assume_value());
+}
+
+TEST_F_CORO(
+  debug_bundle_service_started_fixture, validate_restart_remove_file) {
+    debug_bundle::job_id_t job_id(uuid_t::create());
+
+    co_await run_bundle(job_id);
+
+    auto file_path = co_await _service.local().rpk_debug_bundle_path(job_id);
+    ASSERT_TRUE_CORO(file_path.has_value())
+      << file_path.assume_error().message();
+
+    // Remove the file and restart the service - the metadata should be removed
+    co_await ss::remove_file(file_path.assume_value().native());
+
+    co_await restart_service();
+
+    auto status_restart = co_await _service.local().rpk_debug_bundle_status();
+    ASSERT_FALSE_CORO(status_restart.has_value());
+    EXPECT_EQ(
+      status_restart.assume_error().code(),
+      debug_bundle::error_code::debug_bundle_process_never_started);
+
+    EXPECT_FALSE(
+      _kvstore
+        ->get(
+          storage::kvstore::key_space::debug_bundle,
+          bytes::from_string(debug_bundle::service::debug_bundle_metadata_key))
+        .has_value());
+}
+
+TEST_F_CORO(debug_bundle_service_started_fixture, validate_invalid_sha) {
+    debug_bundle::job_id_t job_id(uuid_t::create());
+
+    co_await run_bundle(job_id);
+
+    ASSERT_NO_THROW_CORO(co_await wait_for_kvstore_to_populate(_kvstore.get()));
+
+    {
+        auto metadata_buf = _kvstore->get(
+          storage::kvstore::key_space::debug_bundle,
+          bytes::from_string(debug_bundle::service::debug_bundle_metadata_key));
+        ASSERT_TRUE_CORO(metadata_buf.has_value());
+        iobuf_parser parser(std::move(metadata_buf.value()));
+        auto metadata = serde::read<debug_bundle::metadata>(parser);
+        metadata.sha256_checksum = bytes::from_string("invalid");
+        iobuf buf;
+        serde::write(buf, std::move(metadata));
+        co_await _kvstore->put(
+          storage::kvstore::key_space::debug_bundle,
+          bytes::from_string(debug_bundle::service::debug_bundle_metadata_key),
+          std::move(buf));
+    }
+
+    co_await restart_service();
+
+    auto status_restart = co_await _service.local().rpk_debug_bundle_status();
+    ASSERT_FALSE_CORO(status_restart.has_value());
+    EXPECT_EQ(
+      status_restart.assume_error().code(),
+      debug_bundle::error_code::debug_bundle_process_never_started);
+
+    EXPECT_FALSE(
+      _kvstore
+        ->get(
+          storage::kvstore::key_space::debug_bundle,
+          bytes::from_string(debug_bundle::service::debug_bundle_metadata_key))
+        .has_value());
+}
+
+TEST_F_CORO(debug_bundle_service_started_fixture, restart_unsuccessful_run) {
+    debug_bundle::job_id_t job_id(uuid_t::create());
+
+    auto res = co_await _service.local().initiate_rpk_debug_bundle_collection(
+      job_id, {});
+    ASSERT_TRUE_CORO(res.has_value()) << res.assume_error().message();
+
+    auto status = co_await _service.local().rpk_debug_bundle_status();
+    ASSERT_TRUE_CORO(status.has_value()) << res.assume_error().message();
+
+    EXPECT_EQ(
+      status.assume_value().status, debug_bundle::debug_bundle_status::running);
+
+    auto term_res = co_await _service.local().cancel_rpk_debug_bundle(job_id);
+    ASSERT_TRUE_CORO(term_res.has_value()) << term_res.assume_error().message();
+
+    ASSERT_NO_THROW_CORO(status = co_await wait_for_process_to_finish());
+
+    ASSERT_TRUE_CORO(status.has_value()) << status.assume_error().message();
+
+    EXPECT_EQ(
+      status.assume_value().status, debug_bundle::debug_bundle_status::error);
+
+    co_await restart_service();
+
+    auto status_restart = co_await _service.local().rpk_debug_bundle_status();
+    ASSERT_TRUE_CORO(status_restart.has_value())
+      << status_restart.assume_error().message();
+    EXPECT_EQ(
+      status_restart.assume_value().status,
+      debug_bundle::debug_bundle_status::error);
+
+    EXPECT_TRUE(
+      _kvstore
+        ->get(
+          storage::kvstore::key_space::debug_bundle,
+          bytes::from_string(debug_bundle::service::debug_bundle_metadata_key))
+        .has_value());
+}

src/v/debug_bundle/types.h

@@ -129,6 +129,10 @@ struct debug_bundle_status_data {
     ss::sstring file_name;
     chunked_vector<ss::sstring> cout;
     chunked_vector<ss::sstring> cerr;
+
+    friend bool
+    operator==(const debug_bundle_status_data&, const debug_bundle_status_data&)
+      = default;
 };
 } // namespace debug_bundle