c/metrics_reporter: ensure http client is stopped before destroying

In case of exceptions, destroying the client may not be safe because the
underlying output stream may not be fully flushed.

src/v/cluster/metrics_reporter.cc

@@ -398,6 +398,21 @@ ss::future<http::client> metrics_reporter::make_http_client() {
     co_return http::client(client_configuration, _as.local());
 }
 
+ss::future<>
+metrics_reporter::do_send_metrics(http::client& client, iobuf body) {
+    auto timeout = config::shard_local_cfg().metrics_reporter_tick_interval();
+    auto res = co_await client.get_connected(timeout, _logger);
+    // skip sending metrics, unable to connect
+    if (res != http::reconnect_result_t::connected) {
+        vlog(
+          _logger.trace, "unable to send metrics report, connection timeout");
+        co_return;
+    }
+    auto resp_stream = co_await client.post(
+      _address.path, std::move(body), http::content_type::json, timeout);
+    co_await resp_stream->prefetch_headers();
+}
+
 ss::future<> metrics_reporter::do_report_metrics() {
     // try initializing cluster info, if it is already present this operation
     // does nothing.
@@ -453,22 +468,10 @@ ss::future<> metrics_reporter::do_report_metrics() {
     }
     auto out = serialize_metrics_snapshot(snapshot.value());
     try {
-        // prepare http client
-        auto client = co_await make_http_client();
-        auto timeout
-          = config::shard_local_cfg().metrics_reporter_tick_interval();
-        auto res = co_await client.get_connected(timeout, _logger);
-        // skip sending metrics, unable to connect
-        if (res != http::reconnect_result_t::connected) {
-            vlog(
-              _logger.trace,
-              "unable to send metrics report, connection timeout");
-            co_return;
-        }
-        auto resp_stream = co_await client.post(
-          _address.path, std::move(out), http::content_type::json, timeout);
-        co_await resp_stream->prefetch_headers();
-        co_await resp_stream->shutdown();
+        co_await http::with_client(
+          co_await make_http_client(), [this, &out](http::client& client) {
+              return do_send_metrics(client, std::move(out));
+          });
         _last_success = ss::lowres_clock::now();
     } catch (...) {
         vlog(

src/v/cluster/metrics_reporter.h

@@ -103,6 +103,7 @@ class metrics_reporter {
     ss::future<result<metrics_snapshot>> build_metrics_snapshot();
 
     ss::future<http::client> make_http_client();
+    ss::future<> do_send_metrics(http::client&, iobuf body);
     ss::future<> try_initialize_cluster_info();
     ss::future<> propagate_cluster_id();