feat(test): add e2e test for OpenDAL fs backend (risingwavelabs#8528)

Author: wcy-fdu

ci/scripts/e2e-test-parallel-for-opendal.sh

@@ -0,0 +1,63 @@
+#!/usr/bin/env bash
+
+# Exits as soon as any line fails.
+set -euo pipefail
+
+source ci/scripts/common.env.sh
+
+while getopts 'p:' opt; do
+    case ${opt} in
+        p )
+            profile=$OPTARG
+            ;;
+        \? )
+            echo "Invalid Option: -$OPTARG" 1>&2
+            exit 1
+            ;;
+        : )
+            echo "Invalid option: $OPTARG requires an argument" 1>&2
+            ;;
+    esac
+done
+shift $((OPTIND -1))
+
+echo "--- Download artifacts"
+mkdir -p target/debug
+buildkite-agent artifact download risingwave-"$profile" target/debug/
+buildkite-agent artifact download risedev-dev-"$profile" target/debug/
+buildkite-agent artifact download "e2e_test/generated/*" ./
+mv target/debug/risingwave-"$profile" target/debug/risingwave
+mv target/debug/risedev-dev-"$profile" target/debug/risedev-dev
+
+echo "--- Adjust permission"
+chmod +x ./target/debug/risingwave
+chmod +x ./target/debug/risedev-dev
+
+echo "--- Generate RiseDev CI config"
+cp ci/risedev-components.ci.env risedev-components.user.env
+
+echo "--- Prepare RiseDev dev cluster"
+cargo make pre-start-dev
+cargo make link-all-in-one-binaries
+
+host_args="-h localhost -p 4565 -h localhost -p 4566 -h localhost -p 4567"
+
+echo "--- e2e, ci-3cn-3fe-opendal-fs-backend, streaming"
+RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
+cargo make ci-start ci-3cn-3fe-opendal-fs-backend
+sqllogictest ${host_args} -d dev  './e2e_test/streaming/**/*.slt' -j 16 --junit "parallel-opendal-fs-backend-${profile}"
+
+echo "--- Kill cluster"
+rm -rf /tmp/rw_ci
+cargo make ci-kill
+
+
+echo "--- e2e, ci-3cn-3fe-opendal-fs-backend, batch"
+RUST_LOG="info,risingwave_stream=info,risingwave_batch=info,risingwave_storage=info" \
+cargo make ci-start ci-3cn-3fe-opendal-fs-backend
+sqllogictest ${host_args} -d dev  './e2e_test/ddl/**/*.slt' --junit "parallel-opendal-fs-backend-ddl-${profile}"
+sqllogictest ${host_args} -d dev  './e2e_test/batch/**/*.slt' -j 16 --junit "parallel-opendal-fs-backend-batch-${profile}"
+
+echo "--- Kill cluster"
+rm -rf /tmp/rw_ci
+cargo make ci-kill

ci/scripts/e2e-test-parallel-in-memory.sh

@@ -54,4 +54,4 @@ sqllogictest ${host_args} -d dev  './e2e_test/ddl/**/*.slt' --junit "parallel-in
 sqllogictest ${host_args} -d dev  './e2e_test/batch/**/*.slt' -j 16 --junit "parallel-in-memory-batch-${profile}"
 
 echo "--- Kill cluster"
-cargo make ci-kill
+cargo make ci-kill

ci/workflows/pull-request.yml

@@ -113,6 +113,21 @@ steps:
     timeout_in_minutes: 12
     retry: *auto-retry
 
+  - label: "end-to-end test for opendal (parallel)"
+    command: "ci/scripts/e2e-test-parallel-for-opendal.sh -p ci-dev"
+    depends_on:
+      - "build"
+      - "docslt"
+    plugins:
+      - gencer/cache#v2.4.10: *cargo-cache
+      - docker-compose#v4.9.0:
+          run: rw-build-env
+          config: ci/docker-compose.yml
+          mount-buildkite-agent: true
+      - ./ci/plugins/upload-failure-logs
+    timeout_in_minutes: 12
+    retry: *auto-retry
+
   - label: "end-to-end test (parallel, in-memory)"
     command: "ci/scripts/e2e-test-parallel-in-memory.sh -p ci-dev"
     depends_on: "build"

risedev.yml

@@ -499,6 +499,39 @@ profile:
         exporter-port: 2224
         health-check-port: 6788
 
+  ci-3cn-3fe-opendal-fs-backend:
+    config-path: src/config/ci.toml
+    steps:
+      # - use: etcd
+      #   unsafe-no-fsync: true
+      - use: meta-node
+      - use: opendal
+        engine: fs
+        bucket: ""
+        root: /tmp/rw_ci
+      - use: compute-node
+        port: 5687
+        exporter-port: 1222
+      - use: compute-node
+        port: 5688
+        exporter-port: 1223
+      - use: compute-node
+        port: 5689
+        exporter-port: 1224
+      - use: frontend
+        port: 4565
+        exporter-port: 2222
+        health-check-port: 6786
+      - use: frontend
+        port: 4566
+        exporter-port: 2223
+        health-check-port: 6787
+      - use: frontend
+        port: 4567
+        exporter-port: 2224
+        health-check-port: 6788
+      - use: compactor
+
   ci-pubsub:
     config-path: src/config/ci.toml
     steps:

src/object_store/src/object/mod.rs

@@ -857,6 +857,15 @@ pub async fn parse_remote_object_store(
                     .monitored(metrics),
             )
         }
+        fs if fs.starts_with("fs://") => {
+            let fs = fs.strip_prefix("fs://").unwrap();
+            let (_, root) = fs.split_once('@').unwrap();
+            ObjectStoreImpl::Opendal(
+                OpendalObjectStore::new_fs_engine(root.to_string())
+                    .unwrap()
+                    .monitored(metrics),
+            )
+        }
         s3_compatible if s3_compatible.starts_with("s3-compatible://") => {
             ObjectStoreImpl::S3Compatible(
                 S3ObjectStore::new_s3_compatible(

src/object_store/src/object/opendal_engine/fs.rs

@@ -0,0 +1,34 @@
+// 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::Fs;
+use opendal::Operator;
+
+use super::{EngineType, OpendalObjectStore};
+use crate::object::ObjectResult;
+impl OpendalObjectStore {
+    /// create opendal fs engine.
+    pub fn new_fs_engine(root: String) -> ObjectResult<Self> {
+        // Create fs backend builder.
+        let mut builder = Fs::default();
+
+        builder.root(&root);
+
+        let op: Operator = Operator::new(builder)?.finish();
+        Ok(Self {
+            op,
+            engine_type: EngineType::Fs,
+        })
+    }
+}

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

@@ -26,3 +26,5 @@ pub mod gcs;
 pub use gcs::*;
 pub mod oss;
 pub use oss::*;
+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,
+    Fs,
 }
 
 impl OpendalObjectStore {
@@ -185,6 +186,7 @@ impl ObjectStore for OpendalObjectStore {
             EngineType::Gcs => "Gcs",
             EngineType::Oss => "Oss",
             EngineType::Webhdfs => "Webhdfs",
+            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 == "fs"{
+                cmd.arg("--state-store")
+                .arg(format!("hummock+fs://{}@{}", opendal.namenode, opendal.root));
+                true
+            }
             else{
                 unimplemented!()
             }