This is an automated email from the ASF dual-hosted git repository.
gavinchou pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 6b5c53fdb4e [fix](s3) Keep the response stream usable when an error
body overflows the read buffer (#66557)
6b5c53fdb4e is described below
commit 6b5c53fdb4e55024fc761347648eb3950194d2e6
Author: Xin Liao <[email protected]>
AuthorDate: Mon Aug 31 10:36:54 2026 +0800
[fix](s3) Keep the response stream usable when an error body overflows the
read buffer (#66557)
Reading from object storage fails from time to time with
```
[INTERNAL_ERROR]failed to read from <key>: Failed to flush response stream
(eof: 0, bad: 1) code=-1 type=1, request_id=failed to read
```
and succeeds when the same statement is run again. It has been hit by
queries
reading a rowset, by compaction, by an outfile export and by the
download of an
inverted index, always on an object storage that was answering `429` or
`503` at
that moment.
`S3ObjStorageClient::get_object()` hands the buffer of the caller to the
SDK as
the response stream of the request, sized exactly like the requested
range. The
SDK writes the body of every response into that stream, the body of an
error
response included. The XML document of a `429 SlowDown` is a few hundred
bytes,
so a small ranged read cannot hold it - the read of the footer of a
packed file
asks for 12 bytes. `PreallocatedStreamBuf` does not implement
`overflow()`, so
the stream turns bad, the write callback of curl reports a short write
and curl
aborts the transfer with `CURLE_WRITE_ERROR`.
The status code of the response is lost from there on:
`CurlHttpClient::MakeRequest()` reads `CURLINFO_RESPONSE_CODE` only when
curl
succeeded, so the code stays at `REQUEST_NOT_MADE` (-1), and the flush
check at
the end of the same function replaces the retryable `NETWORK_CONNECTION`
classification with `INTERNAL_FAILURE` (1).
`S3CustomRetryStrategy::ShouldRetry()`
declines to retry an error classified that way, and so does
`S3FileReader::read_at_impl()`, which retries on `429` alone. A
throttling error
the server asked us to retry cancels the statement of the user instead,
which is
why running it again works.
This also means the error carries no evidence of what really happened:
the code
of the response, the exception name and the request id of the object
storage are
all gone by the time the message is built.
The fix lets the response stream grow: the body is written into the
buffer of the
caller as long as it fits, which is the case for every successful ranged
read and
keeps that path free of copies, and the remainder spills into a buffer
of the
stream itself, truncated at 1MB because only error documents are
expected to
overflow. The stream never turns bad, so curl completes the transfer,
the SDK
records the real status code and parses the error out of the body, and
both the
retry of the SDK and the retry of `S3FileReader` on `429` work again.
A server or a proxy answering a ranged read with the whole object
overflows the
buffer as well. Such a read is still rejected, by the length check that
follows
the request, and now with a message that says so.
Two misleading messages are fixed along the way:
- `request_id=failed to read` is not a request id of the object storage.
It is
the string `S3FileReader` appended behind the empty request id of a
failure
raised by the client itself. The append is dropped and an empty request
id is
printed as `<empty>`.
- The message of a failed read named neither the bucket nor the offset,
leaving
`failed to read from :` in the log whenever the key was empty.
---
be/src/io/fs/s3_file_reader.cpp | 10 +-
be/test/io/fs/s3_response_stream_test.cpp | 189 ++++++++++++++++++++++++
common/cpp/obj-client/s3_common.h | 145 +++++++++++++++++-
common/cpp/obj-client/s3_obj_storage_client.cpp | 20 ++-
4 files changed, 352 insertions(+), 12 deletions(-)
diff --git a/be/src/io/fs/s3_file_reader.cpp b/be/src/io/fs/s3_file_reader.cpp
index 0786a2555cf..5151d53b939 100644
--- a/be/src/io/fs/s3_file_reader.cpp
+++ b/be/src/io/fs/s3_file_reader.cpp
@@ -162,6 +162,7 @@ Status S3FileReader::read_at_impl(size_t offset, Slice
result, size_t* bytes_rea
SCOPED_RAW_TIMER(&_s3_stats.total_get_request_time_ns);
int total_sleep_time = 0;
+ Status last_error;
while (retry_count <= max_retries) {
*bytes_read = 0;
s3_file_reader_read_counter << 1;
@@ -174,6 +175,7 @@ Status S3FileReader::read_at_impl(size_t offset, Slice
result, size_t* bytes_rea
if (resp.http_code ==
static_cast<int>(Aws::Http::HttpResponseCode::TOO_MANY_REQUESTS)) {
s3_file_reader_too_many_request_counter << 1;
+ last_error = Status(resp.status.code,
std::move(resp.status.msg));
retry_count++;
int wait_time = std::min(base_wait_time * (1 << retry_count),
max_wait_time); // Exponential backoff
@@ -184,8 +186,7 @@ Status S3FileReader::read_at_impl(size_t offset, Slice
result, size_t* bytes_rea
continue;
} else {
// Handle other errors
- return std::move(Status(resp.status.code,
std::move(resp.status.msg))
- .append("failed to read"));
+ return {resp.status.code, std::move(resp.status.msg)};
}
}
if (*bytes_read != bytes_req) {
@@ -208,8 +209,9 @@ Status S3FileReader::read_at_impl(size_t offset, Slice
result, size_t* bytes_rea
}
std::string msg = fmt::format(
"failed to get object, path={} offset={} bytes_req={}
bytes_read={} file_size={} "
- "tries={}",
- _path.native(), offset, bytes_req, *bytes_read, _file_size,
(max_retries + 1));
+ "tries={}, last error: [{}]",
+ _path.native(), offset, bytes_req, *bytes_read, _file_size,
(max_retries + 1),
+ last_error.msg());
LOG(WARNING) << msg;
return Status::InternalError(msg);
}
diff --git a/be/test/io/fs/s3_response_stream_test.cpp
b/be/test/io/fs/s3_response_stream_test.cpp
new file mode 100644
index 00000000000..a874cdc1c09
--- /dev/null
+++ b/be/test/io/fs/s3_response_stream_test.cpp
@@ -0,0 +1,189 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+
+#include <sstream>
+#include <string>
+#include <vector>
+
+#include "cpp/obj-client/s3_common.h"
+
+namespace doris {
+
+namespace {
+
+// What the SDK does with the body of a response it has to build an error from.
+std::string drain(std::iostream& stream) {
+ std::stringstream out;
+ out << stream.rdbuf();
+ return out.str();
+}
+
+// The XML body a MinIO answers a throttled ranged read with, shortened.
+constexpr char SLOW_DOWN_BODY[] =
+ R"(<?xml version="1.0"
encoding="UTF-8"?><Error><Code>SlowDown</Code><Message>Please )"
+ R"(reduce your request
rate.</Message><Key>data/packed_file/2666/x.bin</Key></Error>)";
+
+} // namespace
+
+// A body of the requested size lands in the buffer of the caller, without a
copy.
+TEST(S3ResponseStreamTest, BodyFits) {
+ std::string body(64, 'a');
+ std::vector<char> buffer(body.size());
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ stream.write(body.data(), body.size());
+ stream.flush();
+
+ EXPECT_FALSE(stream.fail());
+ EXPECT_EQ(body, std::string(buffer.data(), buffer.size()));
+ EXPECT_EQ(static_cast<std::streampos>(body.size()), stream.tellp());
+ EXPECT_EQ(body, drain(stream));
+}
+
+// An error body larger than the range of the read leaves the stream usable,
which is what
+// keeps curl from aborting the transfer and the SDK from losing the status
code.
+TEST(S3ResponseStreamTest, ErrorBodyOverflowsInOneWrite) {
+ std::string body(SLOW_DOWN_BODY);
+ // A read of the footer of a packed file is far smaller than the error
document.
+ std::vector<char> buffer(12);
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ stream.write(body.data(), body.size());
+ stream.flush();
+
+ EXPECT_FALSE(stream.fail());
+ EXPECT_EQ(static_cast<std::streampos>(body.size()), stream.tellp());
+ EXPECT_EQ(body, drain(stream));
+}
+
+// curl hands the body over in chunks, so the overflow can happen in the
middle of one.
+TEST(S3ResponseStreamTest, ErrorBodyOverflowsAcrossWrites) {
+ std::string body(SLOW_DOWN_BODY);
+ std::vector<char> buffer(16);
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ size_t chunk = 7;
+ for (size_t pos = 0; pos < body.size(); pos += chunk) {
+ stream.write(body.data() + pos, std::min(chunk, body.size() - pos));
+ }
+ stream.flush();
+
+ EXPECT_FALSE(stream.fail());
+ EXPECT_EQ(static_cast<std::streampos>(body.size()), stream.tellp());
+ // The bytes written before the overflow are kept, so the body stays
contiguous.
+ EXPECT_EQ(body, drain(stream));
+}
+
+// A body written one character at a time goes through overflow() instead of
xsputn().
+TEST(S3ResponseStreamTest, ErrorBodyOverflowsCharByChar) {
+ std::string body(SLOW_DOWN_BODY);
+ std::vector<char> buffer(4);
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ for (char c : body) {
+ stream.put(c);
+ }
+ stream.flush();
+
+ EXPECT_FALSE(stream.fail());
+ EXPECT_EQ(body, drain(stream));
+}
+
+// A server answering a ranged read with the whole object must not blow up the
memory of the
+// backend. The body is truncated, the stream stays good and the read is
rejected later on by
+// the length check of the caller.
+TEST(S3ResponseStreamTest, OversizedBodyIsTruncated) {
+ std::string body(S3ResponseStreamBuf::MAX_SPILL_SIZE + 4096, 'x');
+ std::vector<char> buffer(8);
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ stream.write(body.data(), body.size());
+ stream.flush();
+
+ EXPECT_FALSE(stream.fail());
+
EXPECT_EQ(static_cast<std::streampos>(S3ResponseStreamBuf::MAX_SPILL_SIZE),
stream.tellp());
+ EXPECT_EQ(S3ResponseStreamBuf::MAX_SPILL_SIZE, drain(stream).size());
+}
+
+// The same, with a buffer of the caller larger than the bound of the spill: a
prefetched
+// read asks for `remote_storage_read_buffer_mb` at a time and a download for
the whole file,
+// so the bytes moved out of that buffer on the overflow have to be truncated
as well.
+TEST(S3ResponseStreamTest, OversizedBodyIsTruncatedWithLargeBuffer) {
+ std::vector<char> buffer(2 * S3ResponseStreamBuf::MAX_SPILL_SIZE);
+ std::string body(4 * S3ResponseStreamBuf::MAX_SPILL_SIZE, 'x');
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ // curl hands the body over in chunks of `CURL_MAX_WRITE_SIZE`, so the
buffer of the caller
+ // is filled before a write overflows it.
+ size_t chunk = 16384;
+ for (size_t pos = 0; pos < body.size(); pos += chunk) {
+ stream.write(body.data() + pos, std::min(chunk, body.size() - pos));
+ }
+ stream.flush();
+
+ EXPECT_FALSE(stream.fail());
+
EXPECT_EQ(static_cast<std::streampos>(S3ResponseStreamBuf::MAX_SPILL_SIZE),
stream.tellp());
+ EXPECT_EQ(S3ResponseStreamBuf::MAX_SPILL_SIZE, drain(stream).size());
+}
+
+// A truncated body is still a body the SDK rewinds and reads to its end.
+TEST(S3ResponseStreamTest, SeekTruncatedBody) {
+ std::vector<char> buffer(2 * S3ResponseStreamBuf::MAX_SPILL_SIZE);
+ std::string body(4 * S3ResponseStreamBuf::MAX_SPILL_SIZE, 'x');
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ stream.write(body.data(), body.size());
+
+ EXPECT_EQ(S3ResponseStreamBuf::MAX_SPILL_SIZE, drain(stream).size());
+ stream.clear();
+ EXPECT_EQ(std::streampos(0), stream.seekg(0).tellg());
+ EXPECT_EQ(S3ResponseStreamBuf::MAX_SPILL_SIZE, drain(stream).size());
+ // Past the end of what has been kept.
+ stream.clear();
+ EXPECT_TRUE(stream.seekg(S3ResponseStreamBuf::MAX_SPILL_SIZE + 1).fail());
+}
+
+// The SDK rewinds the body before parsing an error out of it.
+TEST(S3ResponseStreamTest, SeekBackAndForth) {
+ std::string body(SLOW_DOWN_BODY);
+ std::vector<char> buffer(12);
+
+ S3ResponseStream stream(buffer.data(), buffer.size());
+ stream.write(body.data(), body.size());
+
+ EXPECT_EQ(body, drain(stream));
+ stream.clear();
+ stream.seekg(0);
+ EXPECT_EQ(body, drain(stream));
+
+ stream.clear();
+ stream.seekg(2);
+ EXPECT_EQ(body.substr(2), drain(stream));
+}
+
+// An empty body is what tells the SDK to build the error out of the status
code alone.
+TEST(S3ResponseStreamTest, EmptyBody) {
+ std::vector<char> buffer(16);
+ S3ResponseStream stream(buffer.data(), buffer.size());
+
+ EXPECT_EQ(std::streampos(0), stream.tellp());
+ EXPECT_TRUE(drain(stream).empty());
+}
+
+} // namespace doris
diff --git a/common/cpp/obj-client/s3_common.h
b/common/cpp/obj-client/s3_common.h
index 2f420455227..c5eaae2162a 100644
--- a/common/cpp/obj-client/s3_common.h
+++ b/common/cpp/obj-client/s3_common.h
@@ -20,6 +20,11 @@
#include <aws/core/utils/memory/stl/AWSStreamFwd.h>
#include <aws/core/utils/stream/PreallocatedStreamBuf.h>
+#include <algorithm>
+#include <cstring>
+#include <streambuf>
+#include <vector>
+
namespace doris {
// A non-copying iostream.
@@ -34,12 +39,148 @@ public:
std::iostream(this) {}
};
+// The AWS SDK writes the body of every response into the stream built by the
response
+// stream factory of the request, whatever the status of that response is.
Reading an
+// object range straight into the buffer of the caller therefore breaks as
soon as the
+// server answers with an error: the XML body of a `429 SlowDown` is a few
hundred bytes
+// and does not fit into the buffer of a small range read.
`PreallocatedStreamBuf` does not
+// implement `overflow()`, so the stream turns bad, curl aborts the transfer
with
+// `CURLE_WRITE_ERROR`, and the SDK reports an `INTERNAL_FAILURE` named
"Failed to flush
+// response stream" while never recording the status code of the response.
Both the retry
+// strategy of the SDK and the retry of `S3FileReader` key on that status
code, so an error
+// the server asked us to retry ends up cancelling the query instead.
+//
+// This stream buffer writes into the buffer of the caller as long as the body
fits, which
+// is the case for every successful ranged read, and spills the rest into a
buffer of its
+// own. The stream never turns bad, so the SDK reports the real status code
and can parse
+// the error out of the body.
+class S3ResponseStreamBuf final : public std::streambuf {
+public:
+ // Bodies beyond this size are truncated. Only error documents are
expected to overflow
+ // and one is a few hundred bytes, so this leaves them two orders of
magnitude of room
+ // while bounding what a single failing request can hold. Kept small on
purpose: this
+ // buffer is allocated on the transport thread of the SDK, out of the
reach of the memory
+ // tracker of the query, and every concurrent read that fails holds one of
its own.
+ static constexpr size_t MAX_SPILL_SIZE = 64 * 1024;
+
+ S3ResponseStreamBuf(void* buf, size_t nbytes) :
_buf(static_cast<char*>(buf)) {
+ setp(_buf, _buf + nbytes);
+ setg(_buf, _buf, _buf);
+ }
+
+protected:
+ std::streamsize xsputn(const char* s, std::streamsize n) override {
+ if (!_spilled) {
+ if (n <= epptr() - pptr()) {
+ std::memcpy(pptr(), s, n);
+ pbump(static_cast<int>(n));
+ return n;
+ }
+ _spill_over();
+ }
+ // Saturating on its own: the spill is clamped when it is filled from
the buffer of
+ // the caller, and this must not underflow into an unbounded write if
it ever is not.
+ auto room = _spill.size() < MAX_SPILL_SIZE ? MAX_SPILL_SIZE -
_spill.size() : 0;
+ auto writable = std::min(static_cast<size_t>(n), room);
+ _spill.insert(_spill.end(), s, s + writable);
+ // Always report the whole write as consumed. A short write is what
makes curl
+ // abort the transfer and lose the status code of the response.
+ return n;
+ }
+
+ int_type overflow(int_type ch) override {
+ if (traits_type::eq_int_type(ch, traits_type::eof())) {
+ return traits_type::not_eof(ch);
+ }
+ auto c = traits_type::to_char_type(ch);
+ xsputn(&c, 1);
+ return ch;
+ }
+
+ int_type underflow() override {
+ _reset_get_area(_read_pos());
+ if (gptr() == egptr()) {
+ return traits_type::eof();
+ }
+ return traits_type::to_int_type(*gptr());
+ }
+
+ pos_type seekoff(off_type off, std::ios_base::seekdir dir,
+ std::ios_base::openmode which) override {
+ auto size = static_cast<off_type>(_written());
+ if ((which & std::ios_base::out) && !(which & std::ios_base::in)) {
+ // The SDK only asks for the write position, to tell an empty body
apart from a
+ // body it has to parse. Moving the write pointer is not supported.
+ return dir == std::ios_base::cur && off == 0 ? pos_type(size) :
pos_type(off_type(-1));
+ }
+ // A seek asking for both areas at once, which is what the default
argument of
+ // `pubseekoff()` and `pubseekpos()` does, is served as a seek of the
read area. The
+ // write area is append only, so there is nothing to move there.
+ off_type pos = off;
+ if (dir == std::ios_base::cur) {
+ pos += static_cast<off_type>(_read_pos());
+ } else if (dir == std::ios_base::end) {
+ pos += size;
+ }
+ if (pos < 0 || pos > size) {
+ return pos_type(off_type(-1));
+ }
+ _reset_get_area(static_cast<size_t>(pos));
+ return pos_type(pos);
+ }
+
+ pos_type seekpos(pos_type pos, std::ios_base::openmode which) override {
+ return seekoff(pos, std::ios_base::beg, which);
+ }
+
+private:
+ // Moves what has been written so far into the spill buffer, so that the
body stays
+ // contiguous and the SDK can parse the error out of it. Truncated right
here: the buffer
+ // of the caller is the size of the range that was asked for,
`remote_storage_read_buffer_mb`
+ // of it for a prefetched read and the whole file for a download, so it
can be far larger
+ // than the bound of the spill. Starting the spill beyond its own bound
would leave no room
+ // for the truncation to ever apply and let a server answering a ranged
read with the whole
+ // object be buffered in full.
+ void _spill_over() {
+ auto kept = std::min(static_cast<size_t>(pptr() - _buf),
MAX_SPILL_SIZE);
+ _spill.assign(_buf, _buf + kept);
+ setp(nullptr, nullptr);
+ _spilled = true;
+ }
+
+ // Bytes of the body held by this buffer, truncation excluded.
+ size_t _written() const { return _spilled ? _spill.size() : pptr() - _buf;
}
+
+ // Both areas start at the same logical offset, so the read position
survives a spill.
+ size_t _read_pos() const { return gptr() - eback(); }
+
+ void _reset_get_area(size_t pos) {
+ char* begin = _spilled ? _spill.data() : _buf;
+ auto size = _written();
+ pos = std::min(pos, size);
+ setg(begin, begin + pos, begin + size);
+ }
+
+ char* _buf;
+ std::vector<char> _spill;
+ bool _spilled = false;
+};
+
+class S3ResponseStream final : public std::iostream {
+public:
+ S3ResponseStream(void* buf, size_t nbytes) : std::iostream(&_buf),
_buf(buf, nbytes) {}
+
+private:
+ S3ResponseStreamBuf _buf;
+};
+
// By default, the AWS SDK reads object data into an auto-growing StringStream.
-// To avoid copies, read directly into our preallocated buffer instead.
+// To avoid copies, read the body directly into our preallocated buffer
instead, and keep
+// only what does not fit, which is an error document, in a buffer of the
stream itself.
// See https://github.com/aws/aws-sdk-cpp/issues/64 for an alternative but
// functionally similar recipe.
inline Aws::IOStreamFactory AwsWriteableStreamFactory(void* buf, int64_t
nbytes) {
- return [=]() { return Aws::New<StringViewStream>("", buf, nbytes); };
+ return [=]() { return Aws::New<S3ResponseStream>("", buf,
static_cast<size_t>(nbytes)); };
}
} // namespace doris
diff --git a/common/cpp/obj-client/s3_obj_storage_client.cpp
b/common/cpp/obj-client/s3_obj_storage_client.cpp
index c1e84f9a5ae..df9210f612e 100644
--- a/common/cpp/obj-client/s3_obj_storage_client.cpp
+++ b/common/cpp/obj-client/s3_obj_storage_client.cpp
@@ -58,10 +58,14 @@ std::string object_identity(const ObjStoragePath& opts) {
}
std::string s3_error_message(const Aws::S3::S3Error& error, std::string_view
message) {
+ // A failure raised by the client itself carries no request id, and a
dangling
+ // `request_id=` has been read as a request id of the object storage.
+ std::string request_id =
+ error.GetRequestId().empty() ? "<empty>" :
error.GetRequestId().c_str();
return fmt::format("{}: {} {} code={}, type={}, request_id={}", message,
error.GetExceptionName(), error.GetMessage(),
static_cast<int>(error.GetResponseCode()),
- static_cast<int>(error.GetErrorType()),
error.GetRequestId());
+ static_cast<int>(error.GetErrorType()), request_id);
}
} // namespace
@@ -300,21 +304,25 @@ ObjStorageResponse S3ObjStorageClient::get_object(const
ObjStoragePath& opts, vo
if (!outcome.IsSuccess()) {
record_s3_request_failed(outcome.GetError());
return ObjStorageResponse {
- .status = s3fs_error(outcome.GetError(), fmt::format("failed
to get object: {}",
-
object_identity(opts))),
+ .status = s3fs_error(
+ outcome.GetError(),
+ fmt::format("failed to get object: bucket={} object={}
offset={} size={}",
+ opts.bucket, object_identity(opts),
offset, bytes_read)),
.http_code =
static_cast<int>(outcome.GetError().GetResponseCode()),
.request_id = outcome.GetError().GetRequestId(),
};
}
*size_return = outcome.GetResult().GetContentLength();
+ // Short read, or a server or a proxy answering a ranged read with the
whole object.
SYNC_POINT_CALLBACK("s3_obj_storage_client::get_object", size_return);
if (*size_return != bytes_read) {
const auto& request_id = outcome.GetResult().GetRequestId();
return ObjStorageResponse {
.status = {ObjStorageStatus::INTERNAL_ERROR,
- fmt::format("incomplete read from {}, expect {},
got {}, request_id={}",
- object_identity(opts), bytes_read,
*size_return,
- request_id)},
+ fmt::format("incomplete read from bucket={}
object={} offset={}, expect "
+ "{}, got {}, request_id={}",
+ opts.bucket, object_identity(opts),
offset, bytes_read,
+ *size_return, request_id)},
.request_id = request_id};
}
return ObjStorageResponse::OK();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
