feat(storage): support azblob (risingwavelabs#8257)

Author: wcy-fdu

risedev.yml

@@ -133,7 +133,7 @@ profile:
       - use: meta-node
       - use: compute-node
       - use: frontend
-      # If you want to use hdfs as storage backend, configure hdfs namenode and root path:
+      # If you want to use webhdfs as storage backend, configure hdfs namenode and root path:
       - use: opendal
         engine: webhdfs
         namenode: "127.0.0.1:9870"
@@ -148,7 +148,7 @@ profile:
       - use: meta-node
       - use: compute-node
       - use: frontend
-      # If you want to use google cloud stoage as storage backend, configure hdfs namenode and root path:
+      # If you want to use google cloud stoage as storage backend, configure bucket name and root path:
       - use: opendal
         engine: gcs
         bucket: bucket-name
@@ -163,10 +163,25 @@ profile:
       - use: meta-node
       - use: compute-node
       - use: frontend
-      # If you want to use google cloud stoage as storage backend, configure hdfs namenode and root path:
+      # If you want to use oss as storage backend, configure bucket name and root path:
       - use: opendal
         engine: oss
-        bucket: "risingwave-oss-wcy"
+        bucket: test-bucket
+        root: risingwave
+      - use: compactor
+      # - use: prometheus
+      # - use: grafana
+
+  azblob:
+    steps:
+      # - use: etcd
+      - use: meta-node
+      - use: compute-node
+      - use: frontend
+      # If you want to use azblob as storage backend, configure bucket(container) name and root path:
+      - use: opendal
+        engine: azblob
+        bucket: test-bucket
         root: risingwave
       - use: compactor
       # - use: prometheus
@@ -898,7 +913,7 @@ template:
     # Minio instances used by this compute node
     provide-minio: "minio*"
 
-    # AWS s3 bucket used by this compute node
+    # OpenDAL storage backend used by this compute node
     provide-opendal: "opendal*"
     # AWS s3 bucket used by this compute node
     provide-aws-s3: "aws-s3*"
@@ -957,7 +972,7 @@ template:
 
     engine: hdfs
 
-    namenode: 127.0.0.1:9000"
+    namenode: 127.0.0.1:9000
 
     bucket: risingwave-test
 

src/object_store/src/object/mod.rs

@@ -857,6 +857,15 @@ pub async fn parse_remote_object_store(
                     .monitored(metrics),
             )
         }
+        azblob if azblob.starts_with("azblob://") => {
+            let azblob = azblob.strip_prefix("azblob://").unwrap();
+            let (container_name, root) = azblob.split_once('@').unwrap();
+            ObjectStoreImpl::Opendal(
+                OpendalObjectStore::new_azblob_engine(container_name.to_string(), root.to_string())
+                    .unwrap()
+                    .monitored(metrics),
+            )
+        }
         fs if fs.starts_with("fs://") => {
             let fs = fs.strip_prefix("fs://").unwrap();
             let (_, root) = fs.split_once('@').unwrap();

src/object_store/src/object/opendal_engine/azblob.rs

@@ -0,0 +1,44 @@
+// 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::services::Azblob;
+use opendal::Operator;
+
+use super::{EngineType, OpendalObjectStore};
+use crate::object::ObjectResult;
+impl OpendalObjectStore {
+    /// create opendal azblob engine.
+    pub fn new_azblob_engine(container_name: String, root: String) -> ObjectResult<Self> {
+        // Create azblob backend builder.
+        let mut builder = Azblob::default();
+        builder.root(&root);
+        builder.container(&container_name);
+
+        let endpoint = std::env::var("AZBLOB_ENDPOINT")
+            .unwrap_or_else(|_| panic!("AZBLOB_ENDPOINT not found from environment variables"));
+        let account_name = std::env::var("AZBLOB_ACCOUNT_NAME")
+            .unwrap_or_else(|_| panic!("AZBLOB_ACCOUNT_NAME not found from environment variables"));
+        let account_key = std::env::var("AZBLOB_ACCOUNT_KEY")
+            .unwrap_or_else(|_| panic!("AZBLOB_ACCOUNT_KEY not found from environment variables"));
+
+        builder.endpoint(&endpoint);
+        builder.account_name(&account_name);
+        builder.account_key(&account_key);
+        let op: Operator = Operator::new(builder)?.finish();
+        Ok(Self {
+            op,
+            engine_type: EngineType::Azblob,
+        })
+    }
+}

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

@@ -26,5 +26,7 @@ pub mod gcs;
 pub use gcs::*;
 pub mod oss;
 pub use oss::*;
+pub mod azblob;
+pub use azblob::*;
 pub mod fs;
 pub use fs::*;

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

@@ -39,6 +39,7 @@ pub enum EngineType {
     Gcs,
     Oss,
     Webhdfs,
+    Azblob,
     Fs,
 }
 
@@ -186,6 +187,7 @@ impl ObjectStore for OpendalObjectStore {
             EngineType::Gcs => "Gcs",
             EngineType::Oss => "Oss",
             EngineType::Webhdfs => "Webhdfs",
+            EngineType::Azblob => "Azblob",
             EngineType::Fs => "Fs",
         }
     }

src/risedevtool/src/task/utils.rs

@@ -117,6 +117,11 @@ pub fn add_storage_backend(
                 .arg(format!("hummock+webhdfs://{}@{}", opendal.namenode, opendal.root));
                 true
             }
+            else if opendal.engine == "azblob"{
+                cmd.arg("--state-store")
+                .arg(format!("hummock+azblob://{}@{}", opendal.bucket, opendal.root));
+                true
+            }
             else if opendal.engine == "fs"{
                 cmd.arg("--state-store")
                 .arg(format!("hummock+fs://{}@{}", opendal.namenode, opendal.root));