From 245955f695b186469ef3186afe097b02cf484c95 Mon Sep 17 00:00:00 2001
From: Abhijat Malviya <abhijat@redpanda.com>
Date: Wed, 8 Nov 2023 12:37:58 +0530
Subject: [PATCH] cloud_storage: Log client in remote segment

(cherry picked from commit c720b5dbc762cc47434ddfb42de8b750a82a894e)
---
 src/v/cloud_storage/remote_segment.cc | 18 ++++++++++++++----
 1 file changed, 14 insertions(+), 4 deletions(-)

diff --git a/src/v/cloud_storage/remote_segment.cc b/src/v/cloud_storage/remote_segment.cc
index 8bcc34e54fad7..a352b4c518a58 100644
--- a/src/v/cloud_storage/remote_segment.cc
+++ b/src/v/cloud_storage/remote_segment.cc
@@ -1338,11 +1338,12 @@ remote_segment_batch_reader::read_some(
             const auto msg = fmt::format(
               "segment_reader is stuck, segment ntp: {}, _cur_rp_offset: "
               "{}, "
-              "_bytes_consumed: {}, parser error state: {}",
+              "_bytes_consumed: {}, parser error state: {}, client: {}",
               _seg->get_ntp(),
               _cur_rp_offset,
               _bytes_consumed,
-              _parser->error());
+              _parser->error(),
+              _config.client_address);
             if (_parser->error() == storage::parser_errc::end_of_stream) {
                 vlog(_ctxlog.info, "{}", msg);
             } else {
@@ -1362,8 +1363,10 @@ remote_segment_batch_reader::init_parser() {
     ss::gate::holder h(_gate);
     vlog(
       _ctxlog.debug,
-      "remote_segment_batch_reader::init_parser, start_offset: {}",
-      _config.start_offset);
+      "remote_segment_batch_reader::init_parser (creating stream), "
+      "start_offset: {}, client: {}",
+      _config.start_offset,
+      _config.client_address);
 
     auto stream_off = co_await _seg->offset_data_stream(
       model::offset_cast(_config.start_offset),
@@ -1371,6 +1374,13 @@ remote_segment_batch_reader::init_parser() {
       _config.first_timestamp,
       priority_manager::local().shadow_indexing_priority());
 
+    vlog(
+      _ctxlog.debug,
+      "remote_segment_batch_reader::init_parser (stream created), "
+      "start_offset: {}, client: {}",
+      _config.start_offset,
+      _config.client_address);
+
     auto parser = std::make_unique<storage::continuous_batch_parser>(
       std::make_unique<remote_segment_batch_consumer>(
         _config, *this, _seg->get_term(), _seg->get_ntp(), _rtc),