risingwave_object_store/object/opendal_engine/
s3.rs

1// Copyright 2026 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::sync::Arc;
16use std::time::Duration;
17
18use opendal::layers::LoggingLayer;
19use opendal::services::S3;
20use opendal::{HttpTransporter, OperationContext, Operator};
21use opendal_http_transport_reqwest::ReqwestTransport;
22use risingwave_common::config::ObjectStoreConfig;
23use thiserror_ext::AsReport as _;
24
25use super::rotating_transport::RotatingTransport;
26use super::{MediaType, OpendalObjectStore, new_operator};
27use crate::object::object_metrics::ObjectStoreMetrics;
28use crate::object::{ObjectError, ObjectResult};
29
30impl OpendalObjectStore {
31    /// create opendal s3 engine.
32    pub fn new_s3_engine(
33        bucket: String,
34        config: Arc<ObjectStoreConfig>,
35        metrics: Arc<ObjectStoreMetrics>,
36    ) -> ObjectResult<Self> {
37        // Create s3 builder.
38        let mut builder = S3::default().bucket(&bucket);
39        // For AWS S3, there is no need to set an endpoint; for other S3 compatible object stores, it is necessary to set this field.
40        if let Ok(endpoint_url) = std::env::var("RW_S3_ENDPOINT") {
41            builder = builder.endpoint(&endpoint_url);
42        }
43
44        if std::env::var("RW_IS_FORCE_PATH_STYLE").is_err() {
45            builder = builder.enable_virtual_host_style();
46        }
47
48        let transport = Self::new_http_transport(config.clone())?;
49
50        let op = new_operator(
51            &config,
52            Operator::new(builder)?
53                .with_context(OperationContext::new().with_http_transport(transport))
54                .layer(LoggingLayer::default()),
55        );
56
57        Ok(Self {
58            op,
59            media_type: MediaType::S3,
60            config,
61            metrics,
62        })
63    }
64
65    /// Creates a minio client. The server should be like `minio://key:secret@address:port/bucket`.
66    pub fn new_minio_engine(
67        server: &str,
68        config: Arc<ObjectStoreConfig>,
69        metrics: Arc<ObjectStoreMetrics>,
70    ) -> ObjectResult<Self> {
71        let server = server.strip_prefix("minio://").unwrap();
72        let (access_key_id, rest) = server.split_once(':').unwrap();
73        let (secret_access_key, mut rest) = rest.split_once('@').unwrap();
74
75        let endpoint_prefix = if let Some(rest_stripped) = rest.strip_prefix("https://") {
76            rest = rest_stripped;
77            "https://"
78        } else if let Some(rest_stripped) = rest.strip_prefix("http://") {
79            rest = rest_stripped;
80            "http://"
81        } else {
82            "http://"
83        };
84        let (address, bucket) = rest.split_once('/').unwrap();
85        let builder = S3::default()
86            .bucket(bucket)
87            .region("custom")
88            .access_key_id(access_key_id)
89            .secret_access_key(secret_access_key)
90            .endpoint(&format!("{}{}", endpoint_prefix, address))
91            .disable_config_load();
92
93        let transport = Self::new_http_transport(config.clone())?;
94
95        let op = new_operator(
96            &config,
97            Operator::new(builder)?
98                .with_context(OperationContext::new().with_http_transport(transport))
99                .layer(LoggingLayer::default()),
100        );
101
102        Ok(Self {
103            op,
104            media_type: MediaType::Minio,
105            config,
106            metrics,
107        })
108    }
109
110    fn new_http_transport(config: Arc<ObjectStoreConfig>) -> ObjectResult<HttpTransporter> {
111        match config.s3.http_client_rotation_interval_ms {
112            None => Ok(HttpTransporter::new(ReqwestTransport::new(
113                Self::new_http_client(&config)?,
114            ))),
115            Some(interval_ms) => Ok(HttpTransporter::new(RotatingTransport::new(
116                move || {
117                    Self::new_http_client(&config)
118                        .map(ReqwestTransport::new)
119                        .map_err(|err| {
120                            opendal::Error::new(
121                                opendal::ErrorKind::Unexpected,
122                                "failed to build rotating HTTP client",
123                            )
124                            .set_source(err)
125                        })
126                },
127                Duration::from_millis(interval_ms),
128            )?)),
129        }
130    }
131
132    pub fn new_http_client(config: &ObjectStoreConfig) -> ObjectResult<reqwest::Client> {
133        let mut client_builder = reqwest::ClientBuilder::new();
134
135        if let Some(keepalive_ms) = config.s3.keepalive_ms.as_ref() {
136            client_builder = client_builder.tcp_keepalive(Duration::from_millis(*keepalive_ms));
137        }
138
139        if let Some(nodelay) = config.s3.nodelay.as_ref() {
140            client_builder = client_builder.tcp_nodelay(*nodelay);
141        }
142        client_builder.build().map_err(|e| {
143            ObjectError::internal(format!("failed to build HTTP client: {}", e.as_report()))
144        })
145    }
146}
risingwave_object_store/object/opendal_engine/s3.rs

risingwave_object_store/object/opendal_engine/
s3.rs
