[v24.1.x] Fix http client segfault in metrics_reporter

src/v/cluster/metrics_reporter.cc

@@ -396,6 +396,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.
@@ -451,22 +466,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();
 

src/v/http/client.cc

@@ -287,8 +287,6 @@ iobuf_to_constbufseq(const iobuf& iobuf) {
     return seq;
 }
 
-ss::future<> client::response_stream::shutdown() { return _client->stop(); }
-
 /// Return failed future if ec is set, otherwise return future in ready state
 static ss::future<iobuf>
 fail_on_error(prefix_logger& ctxlog, const boost::beast::error_code& ec) {

src/v/http/include/http/client.h

@@ -84,6 +84,7 @@ class client : protected net::base_transport {
       ss::shared_ptr<client_probe> probe,
       ss::lowres_clock::duration max_idle_time = {});
 
+    /// Stop must be called before destroying the client object.
     ss::future<> stop();
     using net::base_transport::shutdown;
     using net::base_transport::wait_input_shutdown;
@@ -108,9 +109,6 @@ class client : protected net::base_transport {
         response_stream& operator=(response_stream const&) = delete;
         response_stream operator=(response_stream&&) = delete;
 
-        /// \brief Shutdown connection gracefully
-        ss::future<> shutdown();
-
         /// Return true if the whole http payload is received and parsed
         bool is_done() const;