feat(object store): support obs (#13844) (#13869)

Co-authored-by: congyi wang <58715567+wcy-fdu@users.noreply.github.com>

risedev.yml

@@ -191,6 +191,19 @@ profile:
       - use: compactor
       # - use: prometheus
       # - use: grafana
+  obs:
+    steps:
+      # - use: etcd
+      - use: meta-node
+      - use: compute-node
+      - use: frontend
+      # If you want to use obs as storage backend, configure bucket name:
+      - use: opendal
+        engine: obs
+        bucket: bucket-name
+      - use: compactor
+      # - use: prometheus
+      # - use: grafana
 
   oss:
     steps:

src/object_store/src/object/mod.rs

@@ -810,6 +810,15 @@ pub async fn build_remote_object_store(
                     .monitored(metrics),
             )
         }
+        obs if obs.starts_with("obs://") => {
+            let obs = obs.strip_prefix("obs://").unwrap();
+            let (bucket, root) = obs.split_once('@').unwrap_or((obs, ""));
+            ObjectStoreImpl::Opendal(
+                OpendalObjectStore::new_obs_engine(bucket.to_string(), root.to_string())
+                    .unwrap()
+                    .monitored(metrics),
+            )
+        }
 
         oss if oss.starts_with("oss://") => {
             let oss = oss.strip_prefix("oss://").unwrap();

src/object_store/src/object/opendal_engine/mod.rs

@@ -24,6 +24,8 @@ pub mod webhdfs;
 
 pub mod gcs;
 
+pub mod obs;
+
 pub mod oss;
 
 pub mod azblob;

src/object_store/src/object/opendal_engine/obs.rs

@@ -0,0 +1,52 @@
+// Copyright 2023 RisingWave Labs
+//
+// Licensed 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.
+
+use opendal::layers::{LoggingLayer, RetryLayer};
+use opendal::services::Obs;
+use opendal::Operator;
+
+use super::{EngineType, OpendalObjectStore};
+use crate::object::ObjectResult;
+
+impl OpendalObjectStore {
+    /// create opendal obs engine.
+    pub fn new_obs_engine(bucket: String, root: String) -> ObjectResult<Self> {
+        // Create obs backend builder.
+        let mut builder = Obs::default();
+
+        builder.bucket(&bucket);
+
+        builder.root(&root);
+
+        let endpoint = std::env::var("OBS_ENDPOINT")
+            .unwrap_or_else(|_| panic!("OBS_ENDPOINT not found from environment variables"));
+        let access_key_id = std::env::var("OBS_ACCESS_KEY_ID")
+            .unwrap_or_else(|_| panic!("OBS_ACCESS_KEY_ID not found from environment variables"));
+        let secret_access_key = std::env::var("OBS_SECRET_ACCESS_KEY").unwrap_or_else(|_| {
+            panic!("OBS_SECRET_ACCESS_KEY not found from environment variables")
+        });
+
+        builder.endpoint(&endpoint);
+        builder.access_key_id(&access_key_id);
+        builder.secret_access_key(&secret_access_key);
+        let op: Operator = Operator::new(builder)?
+            .layer(LoggingLayer::default())
+            .layer(RetryLayer::default())
+            .finish();
+        Ok(Self {
+            op,
+            engine_type: EngineType::Obs,
+        })
+    }
+}

src/object_store/src/object/opendal_engine/opendal_object_store.rs

@@ -37,6 +37,7 @@ pub enum EngineType {
     Memory,
     Hdfs,
     Gcs,
+    Obs,
     Oss,
     Webhdfs,
     Azblob,
@@ -189,6 +190,7 @@ impl ObjectStore for OpendalObjectStore {
             EngineType::Memory => "Memory",
             EngineType::Hdfs => "Hdfs",
             EngineType::Gcs => "Gcs",
+            EngineType::Obs => "Obs",
             EngineType::Oss => "Oss",
             EngineType::Webhdfs => "Webhdfs",
             EngineType::Azblob => "Azblob",

src/risedevtool/src/task/utils.rs

@@ -122,6 +122,10 @@ pub fn add_hummock_backend(
                 cmd.arg("--state-store")
                 .arg(format!("hummock+gcs://{}", opendal.bucket));
             }
+            else if opendal.engine == "obs"{
+                cmd.arg("--state-store")
+                .arg(format!("hummock+obs://{}", opendal.bucket));
+            }
             else if opendal.engine == "oss"{
                 cmd.arg("--state-store")
                 .arg(format!("hummock+oss://{}", opendal.bucket));