mirror of
https://github.com/apache/doris.git
synced 2026-11-04 18:02:19 +00:00
### 1. What does this PR do?
BE and Cloud Recycler previously maintained separate object-storage
abstractions and separate S3/Azure implementations. Although both sides
called the same cloud-provider SDKs, credential construction, error
conversion, metrics, pagination, batch deletion, and compatibility
behavior were duplicated and could evolve differently.
This PR consolidates those implementations under `common/cpp/client`
while keeping the original decorator-oriented client abstraction:
- `ObjStorageClient` is the common interface used by upper layers and
implemented by all concrete clients and decorators.
- `S3ObjStorageClient` and `AzureObjStorageClient` implement provider
operations through the AWS and Azure SDKs.
- `RateLimitedObjStorageClient` implements the same interface and
decorates another `ObjStorageClient`.
- BE's `S3ClientFactory` and Recycler's `S3Accessor` construct the
provider client and wrap it with their own rate-limit policy.
- Shared request/response types, page-based listing, lazy iteration,
recursive deletion orchestration, credentials, metrics, and error
conversion live in the common layer.
Before the refactor, BE and Recycler reached the SDKs through separate
stacks:
```text
BE call sites Recycler call sites
| |
v v
BE object client Recycler S3Accessor
+ BE limiter + Recycler limiter
| |
v v
BE S3 / Azure clients Recycler S3 / Azure clients
| |
v v
AWS / Azure SDK AWS / Azure SDK
```
After the refactor, BE and Recycler keep their own upper-layer
composition points and policies, while both use the same common
interface, decorator, and provider implementations:
```text
AFTER
+-------------------------- BE --------------------------+ +---------------------- Recycler ----------------------+
| BE call sites S3RateLimiterManager | | Recycler call sites AccessorRateLimiter |
| | | | | | | |
| v v | | v v |
| S3ClientFactory BE policy | | S3Accessor Recycler policy |
+-----------------------------+--------------------------+ +-----------------------------+-------------------------+
| |
+------------------------------+-------------------------------+
|
v
ObjStorageClient
^
|
RateLimitedObjStorageClient
(policy_: ObjStorageRateLimitPolicy)
|
+----------+----------+
| |
v v
S3ObjStorageClient AzureObjStorageClient
| |
v v
AWS SDK Azure SDK
```
This keeps cross-cutting behavior composable. Future decorators, such as
tracing, retries, or additional admission controls, can be added without
changing provider implementations or introducing a facade coupled to
every policy.
### 2. How are the different behaviors unified?
| Behavior | Unified behavior |
| --- | --- |
| Client API | BE and Recycler use the same `ObjStorageClient` interface
and request/response types. `doris::io` aliases keep BE call sites
source-compatible, while Recycler adapters preserve its integer-facing
API. |
| Provider clients | `S3ObjStorageClient` and `AzureObjStorageClient`
are shared by both callers and directly implement `ObjStorageClient`. |
| Rate limiting | `RateLimitedObjStorageClient` wraps any
`ObjStorageClient`. BE and Recycler inject their own
`ObjStorageRateLimitPolicy` when constructing the client. |
| Error model | `ObjectStorageResponse` carries a Doris status and
preserves the HTTP status and request ID when available. Recycler
adapters retain caller-facing contracts such as the 0/1/negative
`exists` result. |
| Listing | `list_objects` returns one page. `ObjectListIterator` owns
the continuation token and requests the next page only after consuming
the current page. |
| Missing S3 prefix | The shared S3 implementation preserves the
S3-compatible `NoSuchKey`-as-empty behavior for both callers. |
| Batch deletion | Provider implementations enforce their SDK limits: up
to 1000 objects for S3 and 256 subrequests for Azure. |
| Recursive deletion | `ObjStorageClient` provides the shared
provider-independent listing, expiration filtering, grouping, execution,
and error-propagation flow. Recycler can supply its `SyncExecutor`
adapter; BE uses the synchronous fallback. |
| AWS credentials | `AwsCredentialFactory` implements static
credentials, the default provider chain, role ARN, and external ID once.
|
| Azure credentials | `AzureAuthFactory` creates the shared-key
credential and container client for both callers. Shared error
conversion retains the BE TLS diagnostic context. |
| Metrics | Provider clients share request-failure metrics and latency
timers. The previous BE slow-S3-upload logging behavior is retained. |
### 3. Rate-limit boundary and follow-up
The decorator currently preserves the pre-refactor BE semantics:
- A direct `list_objects` or `delete_objects` call is admitted once
before delegation.
- `delete_objects_recursively` is admitted once as one logical PUT
operation and then delegates to the wrapped client.
- Therefore, list pages and provider delete batches performed inside
recursive deletion do not yet pass back through the decorator
individually.
Making recursive deletion invoke the decorated page-list and
delete-batch operations requires a separate change to avoid either
bypassing the decorator or double-counting requests. That request-level
admission behavior will be handled in a follow-up PR rather than mixed
into this structural refactor. This temporarily changes Recycler
recursive-delete admission to match the existing BE behavior.
127 lines
5.3 KiB
C++
127 lines
5.3 KiB
C++
// 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.
|
|
|
|
#pragma once
|
|
|
|
#include <aws/core/utils/Array.h>
|
|
#include <aws/core/utils/HashingUtils.h>
|
|
#include <cpp/sync_point.h>
|
|
#include <fmt/core.h>
|
|
#include <gen_cpp/Status_types.h>
|
|
#include <glog/logging.h>
|
|
|
|
#include <algorithm>
|
|
#include <azure/core/http/http.hpp>
|
|
#include <azure/core/http/http_status_code.hpp>
|
|
#include <azure/core/io/body_stream.hpp>
|
|
#include <azure/storage/blobs.hpp>
|
|
#include <azure/storage/blobs/blob_batch.hpp>
|
|
#include <azure/storage/blobs/blob_client.hpp>
|
|
#ifdef USE_AZURE
|
|
#include <azure/core/diagnostics/logger.hpp>
|
|
#include <azure/storage/blobs/blob_container_client.hpp>
|
|
#endif
|
|
#include <azure/storage/blobs/blob_sas_builder.hpp>
|
|
#include <azure/storage/blobs/rest_client.hpp>
|
|
#include <azure/storage/common/account_sas_builder.hpp>
|
|
#include <azure/storage/common/storage_credential.hpp>
|
|
#include <azure/storage/common/storage_exception.hpp>
|
|
#include <chrono>
|
|
#include <exception>
|
|
#include <iterator>
|
|
#include <ranges>
|
|
|
|
#include "client_bvar.h"
|
|
#include "cpp/obj_retry_strategy.h"
|
|
#include "cpp/util.h"
|
|
#include "obj_storage_client.h"
|
|
#include "s3_common.h"
|
|
|
|
namespace Azure::Storage::Blobs {
|
|
class BlobContainerClient;
|
|
} // namespace Azure::Storage::Blobs
|
|
|
|
namespace doris {
|
|
|
|
using namespace Azure::Storage::Blobs;
|
|
|
|
static const Azure::DateTime SystemClockEpoch {1970, 1, 1};
|
|
|
|
bool is_azure_tls_ca_error_message(std::string_view message);
|
|
std::string build_azure_tls_debug_suffix(std::string_view error_message,
|
|
std::string_view tls_debug_context);
|
|
std::string azure_multipart_block_id(std::string_view upload_id, int part_num);
|
|
std::string build_azure_batch_delete_failure_message(const ObjStoragePath& opts,
|
|
std::string_view key);
|
|
|
|
class ObjClientHolder;
|
|
|
|
class AzureObjStorageClient final : public ObjStorageClient {
|
|
public:
|
|
AzureObjStorageClient(
|
|
std::shared_ptr<Azure::Storage::Blobs::BlobContainerClient> client,
|
|
ObjStorageEndpointInfo config,
|
|
std::shared_ptr<Azure::Storage::StorageSharedKeyCredential> credential = nullptr)
|
|
: _config(std::move(config)),
|
|
_client(std::move(client)),
|
|
_credential(std::move(credential)) {}
|
|
~AzureObjStorageClient() override = default;
|
|
ObjStorageUploadResult create_multipart_upload(const ObjStoragePath& opts) override;
|
|
ObjStorageResponse put_object(const ObjStoragePath& opts, std::string_view stream) override;
|
|
ObjStorageUploadResult upload_part(const ObjStoragePath& opts, const std::string& upload_id,
|
|
std::string_view, int partNum) override;
|
|
ObjStorageResponse complete_multipart_upload(
|
|
const ObjStoragePath& opts, const std::string& upload_id,
|
|
const std::vector<ObjStorageCompletedPart>& completed_parts) override;
|
|
ObjStorageHeadResult head_object(const ObjStoragePath& opts) override;
|
|
ObjStorageResponse get_object(const ObjStoragePath& opts, void* buffer, size_t offset,
|
|
size_t bytes_read, size_t* size_return) override;
|
|
ObjStorageResponse delete_objects(const ObjStoragePath& opts,
|
|
std::vector<std::string> objs) override;
|
|
ObjStorageResponse delete_object(const ObjStoragePath& opts) override;
|
|
std::string generate_presigned_url(const ObjStoragePath& opts,
|
|
int64_t expiration_secs) override;
|
|
ObjStorageResponse get_lifecycle(const std::string& bucket, int64_t* expiration_days) override;
|
|
|
|
ObjStorageResponse check_versioning(const std::string& bucket) override;
|
|
|
|
ObjStorageResponse abort_multipart_upload(const ObjStoragePath& opts,
|
|
const std::string& upload_id) override;
|
|
ObjStorageCapabilities capabilities() const override {
|
|
return {.max_delete_batch = 256, .max_list_page = 5000};
|
|
}
|
|
|
|
protected:
|
|
ObjStorageListPageResult list_objects_page(const ObjStoragePath& path,
|
|
std::string_view continuation_token) override;
|
|
|
|
private:
|
|
ObjStorageEndpointInfo _config;
|
|
std::shared_ptr<Azure::Storage::Blobs::BlobContainerClient> _client;
|
|
std::shared_ptr<Azure::Storage::StorageSharedKeyCredential> _credential;
|
|
};
|
|
|
|
} // namespace doris
|
|
|
|
namespace doris::io {
|
|
using ::doris::AzureObjStorageClient;
|
|
using ::doris::azure_multipart_block_id;
|
|
using ::doris::build_azure_batch_delete_failure_message;
|
|
using ::doris::build_azure_tls_debug_suffix;
|
|
using ::doris::is_azure_tls_ca_error_message;
|
|
} // namespace doris::io
|