[enhance](S3) Print the error detail for every s3 operation

be/src/io/fs/buffered_reader.cpp

@@ -474,6 +474,10 @@ void PrefetchBuffer::prefetch_buffer() {
         return;
     }
     if (!s.ok() && _offset < _reader->size()) {
+        // We should print the error msg since this buffer might not be accessed by the consumer
+        // which would result in the status being missed
+        LOG_WARNING("prefetch path {} failed, offset {}, error {}", _reader->path().native(),
+                    _offset, s.to_string());
         _prefetch_status = std::move(s);
     }
     _buffer_status = BufferStatus::PREFETCHED;

be/src/io/fs/s3_file_reader.cpp

@@ -97,8 +97,10 @@ Status S3FileReader::read_at_impl(size_t offset, Slice result, size_t* bytes_rea
     auto outcome = client->GetObject(request);
     s3_bvar::s3_get_total << 1;
     if (!outcome.IsSuccess()) {
-        return Status::IOError("failed to read from {}: {}", _path.native(),
-                               outcome.GetError().GetMessage());
+        return Status::IOError("failed to read from {}: {}, exception {}, error code {}",
+                               _path.native(), outcome.GetError().GetMessage(),
+                               outcome.GetError().GetExceptionName(),
+                               outcome.GetError().GetResponseCode());
     }
     *bytes_read = outcome.GetResult().GetContentLength();
     if (*bytes_read != bytes_req) {

be/src/io/fs/s3_file_system.cpp

@@ -537,9 +537,10 @@ Status S3FileSystem::get_key(const Path& path, std::string* key) const {
 
 template <typename AwsOutcome>
 std::string S3FileSystem::error_msg(const std::string& key, const AwsOutcome& outcome) const {
-    return fmt::format("(endpoint: {}, bucket: {}, key:{}, {}), {}", _s3_conf.endpoint,
-                       _s3_conf.bucket, key, outcome.GetError().GetExceptionName(),
-                       outcome.GetError().GetMessage());
+    return fmt::format("(endpoint: {}, bucket: {}, key:{}, {}), {}, error code {}",
+                       _s3_conf.endpoint, _s3_conf.bucket, key,
+                       outcome.GetError().GetExceptionName(), outcome.GetError().GetMessage(),
+                       outcome.GetError().GetResponseCode());
 }
 
 std::string S3FileSystem::error_msg(const std::string& key, const std::string& err) const {

be/src/io/fs/s3_file_writer.cpp

@@ -132,8 +132,11 @@ Status S3FileWriter::_create_multi_upload_request() {
         _upload_id = outcome.GetResult().GetUploadId();
         return Status::OK();
     }
-    return Status::IOError("failed to create multipart upload(bucket={}, key={}, upload_id={}): {}",
-                           _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage());
+    return Status::IOError(
+            "failed to create multipart upload(bucket={}, key={}, upload_id={}): {}, exception {}, "
+            "error code {}",
+            _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage(),
+            outcome.GetError().GetExceptionName(), outcome.GetError().GetResponseCode());
 }
 
 void S3FileWriter::_wait_until_finish(std::string_view task_name) {
@@ -185,8 +188,11 @@ Status S3FileWriter::abort() {
         _aborted = true;
         return Status::OK();
     }
-    return Status::IOError("failed to abort multipart upload(bucket={}, key={}, upload_id={}): {}",
-                           _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage());
+    return Status::IOError(
+            "failed to abort multipart upload(bucket={}, key={}, upload_id={}): {}, exception {}, "
+            "error code {}",
+            _bucket, _path.native(), _upload_id, outcome.GetError().GetMessage(),
+            outcome.GetError().GetExceptionName(), outcome.GetError().GetResponseCode());
 }
 
 Status S3FileWriter::close() {
@@ -340,9 +346,12 @@ void S3FileWriter::_upload_one_part(int64_t part_num, UploadFileBuffer& buf) {
     });
     if (!upload_part_outcome.IsSuccess()) {
         auto s = Status::IOError(
-                "failed to upload part (bucket={}, key={}, part_num={}, up_load_id={}): {}",
+                "failed to upload part (bucket={}, key={}, part_num={}, up_load_id={}): {}, "
+                "exception {}, error code {}",
                 _bucket, _path.native(), part_num, _upload_id,
-                upload_part_outcome.GetError().GetMessage());
+                upload_part_outcome.GetError().GetMessage(),
+                upload_part_outcome.GetError().GetExceptionName(),
+                upload_part_outcome.GetError().GetResponseCode());
         LOG_WARNING(s.to_string());
         buf.set_val(s);
         return;
@@ -414,8 +423,11 @@ Status S3FileWriter::_complete() {
 
     if (!compute_outcome.IsSuccess()) {
         auto s = Status::IOError(
-                "failed to create complete multi part upload (bucket={}, key={}): {}", _bucket,
-                _path.native(), compute_outcome.GetError().GetMessage());
+                "failed to create complete multi part upload (bucket={}, key={}): {}, exception "
+                "{}, error code {}",
+                _bucket, _path.native(), compute_outcome.GetError().GetMessage(),
+                compute_outcome.GetError().GetExceptionName(),
+                compute_outcome.GetError().GetResponseCode());
         LOG_WARNING(s.to_string());
         return s;
     }
@@ -463,10 +475,10 @@ void S3FileWriter::_put_object(UploadFileBuffer& buf) {
     auto response = _client->PutObject(request);
     s3_bvar::s3_put_total << 1;
     if (!response.IsSuccess()) {
-        _st = Status::InternalError("Error: [{}:{}, responseCode:{}]",
-                                    response.GetError().GetExceptionName(),
-                                    response.GetError().GetMessage(),
-                                    static_cast<int>(response.GetError().GetResponseCode()));
+        _st = Status::InternalError(
+                "failed to put object (bucket={}, key={}), Error: [{}:{}, responseCode:{}]",
+                _bucket, _path.native(), response.GetError().GetExceptionName(),
+                response.GetError().GetMessage(), response.GetError().GetResponseCode());
         LOG(WARNING) << _st;
         buf.set_val(_st);
         return;