feat(interactive): Implement oss uploader and downloader for interactive (#4577)

- Introducing `aliyun-oss-cpp-sdk` as third-party
- Implement oss-object-reader/write, allowing dumping graph data to oss
and load graph data from oss.

Author: zhanglei1949

.gitmodules

@@ -13,3 +13,6 @@
 [submodule "flex/third_party/parallel-hashmap"]
 	path = flex/third_party/parallel-hashmap
 	url = https://github.com/greg7mdp/parallel-hashmap.git
+[submodule "flex/third_party/aliyun-oss-cpp-sdk"]
+	path = flex/third_party/aliyun-oss-cpp-sdk
+	url = https://github.com/aliyun/aliyun-oss-cpp-sdk.git

flex/CMakeLists.txt

@@ -17,6 +17,7 @@ option(USE_PTHASH "Whether to use pthash" OFF)
 option(OPTIMIZE_FOR_HOST "Whether to optimize on host" ON) # Whether to build optimized code on host
 option(USE_STATIC_ARROW "Whether to use static arrow" OFF) # Whether to link arrow statically, default is OFF
 option(BUILD_WITH_OTEL "Whether to build with opentelemetry-cpp" OFF) # Whether to build with opentelemetry-cpp, default is OFF
+option(BUILD_WITH_OSS "Whether to build with oss support" OFF) # Whether to build with oss support, default is OFF
 
 #print options
 message(STATUS "Build test: ${BUILD_TEST}")
@@ -51,6 +52,13 @@ if(LINUX_KERNEL_VERSION VERSION_GREATER_EQUAL "4.5")
     add_definitions(-DUSE_COPY_FILE_RANGE)
 endif()
 
+if (BUILD_WITH_OSS)
+    add_definitions(-DBUILD_WITH_OSS)
+    include_directories(${CMAKE_CURRENT_SOURCE_DIR}/third_party/aliyun-oss-cpp-sdk/sdk/include)
+    set(TARGET_OUTPUT_NAME_PREFIX "alibabacloud-oss-" CACHE STRING "The target's output name prefix")
+    add_subdirectory(third_party/aliyun-oss-cpp-sdk)
+endif()
+
 include_directories(${CMAKE_CURRENT_SOURCE_DIR}/../)
 
 if(USE_PTHASH)

flex/bin/CMakeLists.txt

@@ -44,6 +44,9 @@ install(PROGRAMS load_plan_and_gen.sh DESTINATION bin)
 include_directories(${Boost_INCLUDE_DIRS})
 add_executable(bulk_loader bulk_loader.cc)
 target_link_libraries(bulk_loader flex_rt_mutable_graph ${GFLAGS_LIBRARIES})
+if (BUILD_WITH_OSS)
+        target_link_libraries(bulk_loader cpp-sdk)
+endif()
 install_without_export_flex_target(bulk_loader)
 
 add_executable(stored_procedure_runner stored_procedure_runner.cc)

flex/bin/bulk_loader.cc

@@ -24,6 +24,11 @@
 #include "flex/engines/graph_db/database/graph_db.h"
 #include "flex/engines/http_server/options.h"
 
+#ifdef BUILD_WITH_OSS
+#include <boost/process.hpp>
+#include "flex/utils/remote/oss_storage.h"
+#endif
+
 namespace bpo = boost::program_options;
 
 static std::string work_dir;
@@ -44,6 +49,73 @@ void signal_handler(int signal) {
   }
 }
 
+#ifdef BUILD_WITH_OSS
+
+void check_oss_object_not_exist(std::string& data_path,
+                                std::string& object_path,
+                                gs::OSSConf& oss_conf) {
+  auto pos = data_path.find("/", 6);
+  if (pos == std::string::npos) {
+    LOG(FATAL) << "Invalid data path: " << data_path;
+  }
+  oss_conf.bucket_name_ = data_path.substr(6, pos - 6);
+  object_path = data_path.substr(pos + 1);
+  oss_conf.load_conf_from_env();
+  // check whether the object exists
+  auto oss_reader = std::make_shared<gs::OSSRemoteStorageDownloader>(oss_conf);
+  if (!oss_reader || !oss_reader->Open().ok()) {
+    LOG(FATAL) << "Failed to open oss reader";
+  }
+  std::vector<std::string> path_list;
+  auto status = oss_reader->List(object_path, path_list);
+  if (status.ok() && path_list.size() > 0) {
+    LOG(FATAL) << "Object already exists: " << object_path
+               << ", list size: " << path_list.size()
+               << ", please remove the object and try again.";
+  }
+  // use a random directory
+  data_path = "/tmp/" + std::to_string(time(nullptr));
+}
+
+int32_t upload_data_dir_to_oss(const std::filesystem::path& data_dir_path,
+                               const std::string& object_path,
+                               const gs::OSSConf& oss_conf) {
+  // zip the data directory
+  std::string zip_file = data_dir_path.string() + ".zip";
+  std::string zip_cmd = "zip -r " + zip_file + " " + data_dir_path.string();
+  boost::process::child zip_process(zip_cmd);
+  zip_process.wait();
+
+  int res = zip_process.exit_code();
+  if (res != 0) {
+    LOG(ERROR) << "Failed to zip data directory: " << zip_cmd
+               << ", code: " << res;
+    return -1;
+  }
+
+  auto oss_writer = std::make_shared<gs::OSSRemoteStorageUploader>(oss_conf);
+  if (!oss_writer || !oss_writer->Open().ok()) {
+    LOG(ERROR) << "Failed to open oss writer";
+    return -1;
+  }
+  auto status = oss_writer->Put(zip_file, object_path, false);
+  if (!status.ok()) {
+    LOG(ERROR) << "Failed to upload data to oss: " << status.ToString();
+    return -1;
+  }
+  status = oss_writer->Close();
+  if (!status.ok()) {
+    LOG(ERROR) << "Failed to close oss writer: " << status.ToString();
+    return -1;
+  }
+  LOG(INFO) << "Successfully uploaded data to oss: " << object_path
+            << ", it is in zip format";
+  std::filesystem::remove(zip_file);
+  std::filesystem::remove_all(data_dir_path);
+  return 0;
+}
+#endif
+
 int main(int argc, char** argv) {
   bpo::options_description desc("Usage:");
   /**
@@ -90,6 +162,17 @@ int main(int argc, char** argv) {
   }
 
   std::string data_path = "";
+  /**
+   * If the data path is an oss path, the data will be uploaded to oss after
+   * loading to a temporary directory. To improve the performance of the
+   * performance, bulk_loader will zip the data directory before uploading.
+   * The data path should be in the format of oss://bucket_name/object_path
+   */
+#ifdef BUILD_WITH_OSS
+  bool upload_to_oss = false;
+  std::string object_path = "";
+  auto oss_conf = gs::OSSConf();
+#endif
   std::string bulk_load_config_path = "";
   std::string graph_schema_path = "";
 
@@ -141,6 +224,16 @@ int main(int argc, char** argv) {
         vm["use-mmap-vector"].as<bool>());
   }
 
+  if (data_path.find("oss://") == 0) {
+#ifdef BUILD_WITH_OSS
+    upload_to_oss = true;
+    check_oss_object_not_exist(data_path, object_path, oss_conf);
+#else
+    LOG(ERROR) << "OSS is not supported in this build";
+    return -1;
+#endif
+  }
+
   std::filesystem::path data_dir_path(data_path);
   if (!std::filesystem::exists(data_dir_path)) {
     std::filesystem::create_directory(data_dir_path);
@@ -185,5 +278,11 @@ int main(int argc, char** argv) {
   t += grape::GetCurrentTime();
   LOG(INFO) << "Finished bulk loading in " << t << " seconds.";
 
+#ifdef BUILD_WITH_OSS
+  if (upload_to_oss) {
+    return upload_data_dir_to_oss(data_dir_path, object_path, oss_conf);
+  }
+#endif
+
   return 0;
 }

flex/tests/hqps/CMakeLists.txt

@@ -1,11 +1,17 @@
 
 # file(GLOB_RECURSE GS_TEST_FILES "${CMAKE_CURRENT_SOURCE_DIR}/*.cc")
 file(GLOB GS_TEST_FILES RELATIVE "${CMAKE_CURRENT_SOURCE_DIR}" "${CMAKE_CURRENT_SOURCE_DIR}/*.cc")
+if (NOT BUILD_WITH_OSS)
+        list(REMOVE_ITEM GS_TEST_FILES "oss_test.cc")
+endif()
 
 foreach(f ${GS_TEST_FILES})
         string(REGEX MATCH "^(.*)\\.[^.]*$" dummy ${f})
         set(T_NAME ${CMAKE_MATCH_1})
         message(STATUS "Found graphscope test - " ${T_NAME})
         add_executable(${T_NAME} ${CMAKE_CURRENT_SOURCE_DIR}/${T_NAME}.cc)
-        target_link_libraries(${T_NAME} flex_plan_proto  flex_graph_db)
+        target_link_libraries(${T_NAME} flex_plan_proto  flex_graph_db flex_utils)
+        if (BUILD_WITH_OSS)
+                target_link_libraries(${T_NAME} cpp-sdk)
+        endif()
 endforeach()

flex/tests/hqps/oss_test.cc

@@ -0,0 +1,63 @@
+
+/** Copyright 2020 Alibaba Group Holding Limited.
+ *
+ * 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.
+ */
+
+#include "flex/utils/remote/oss_storage.h"
+
+#include <glog/logging.h>
+
+int main(int argc, char** argv) {
+  if (argc != 6) {
+    std::cerr << "usage: oss_object_writer <access-key> <access-secret> "
+                 "<endpoint> <bucket> <input-file> "
+              << std::endl;
+  }
+
+  gs::OSSConf conf;
+  conf.accesskey_id_ = argv[1];
+  conf.accesskey_secret_ = argv[2];
+  conf.endpoint_ = argv[3];
+  conf.bucket_name_ = argv[4];
+
+  gs::OSSRemoteStorageUploader writer(conf);
+  gs::OSSRemoteStorageDownloader reader(conf);
+
+  CHECK(writer.Open().ok()) << "Open OSS writer failed";
+  CHECK(reader.Open().ok()) << "Open OSS reader failed";
+
+  std::string input_file = argv[5];
+  std::string object_name = "test_object";
+  if (writer.Put(input_file, object_name).ok()) {
+    LOG(INFO) << "Put object " << object_name << " success";
+  } else {
+    LOG(ERROR) << "Put object " << object_name << " failed";
+  }
+
+  std::string output_file = "output_file";
+  if (reader.Get(object_name, output_file).ok()) {
+    LOG(INFO) << "Get object " << object_name << " success";
+  } else {
+    LOG(ERROR) << "Get object " << object_name << " failed";
+  }
+
+  // delete object
+  if (writer.Delete(object_name).ok()) {
+    LOG(INFO) << "Delete object " << object_name << " success";
+  } else {
+    LOG(ERROR) << "Delete object " << object_name << " failed";
+  }
+
+  return 0;
+}

flex/third_party/aliyun-oss-cpp-sdk

@@ -73,12 +73,21 @@ install(DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/flex
 
 ############################################ Build Utils Library ############################################
 file(GLOB_RECURSE UTILS_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/*.cc")
+
+if (NOT BUILD_WITH_OSS)
+        list(REMOVE_ITEM UTILS_SRC_FILES "${CMAKE_CURRENT_SOURCE_DIR}/remote/oss_storage.cc")
+endif()
+
+
 add_library(flex_utils SHARED ${UTILS_SRC_FILES} ${ERROR_PROTO_SRCS})
 target_include_directories(flex_utils PUBLIC 
         $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}> 
         $<BUILD_INTERFACE:${CMAKE_CURRENT_BINARY_DIR}>)
 # Link the static library of arrow, to save the trouble of linking the shared library of arrow
 target_link_libraries(flex_utils PUBLIC ${Protobuf_LIBRARIES} ${ARROW_LIB} ${YAML_CPP_LIBRARIES} ${Boost_LIBRARIES} ${GLOG_LIBRARIES})
+if (BUILD_WITH_OSS)
+        target_link_libraries(flex_utils PRIVATE cpp-sdk)
+endif()
 
 
 find_package(mimalloc 1.8 REQUIRED)

flex/utils/CMakeLists.txt

@@ -0,0 +1,45 @@
+/** Copyright 2020 Alibaba Group Holding Limited.
+ *
+ * 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.
+ */
+
+#include "flex/utils/file_utils.h"
+
+namespace gs {
+
+bool read_string_from_file(const std::string& file_path, std::string& content) {
+  std::ifstream inputFile(file_path);
+
+  if (!inputFile.is_open()) {
+    LOG(ERROR) << "Error: Could not open the file " << file_path;
+    return false;
+  }
+  std::ostringstream buffer;
+  buffer << inputFile.rdbuf();
+  content = buffer.str();
+  return true;
+}
+
+bool write_string_to_file(const std::string& content,
+                          const std::string& file_path) {
+  std::ofstream outputFile(file_path, std::ios::out | std::ios::trunc);
+
+  if (!outputFile.is_open()) {
+    LOG(ERROR) << "Error: Could not open the file " << file_path;
+    return false;
+  }
+  outputFile << content;
+  return true;
+}
+
+}  // namespace gs

flex/utils/file_utils.cc

@@ -0,0 +1,35 @@
+/** Copyright 2020 Alibaba Group Holding Limited.
+ *
+ * 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.
+ */
+
+#ifndef UTILS_FILE_UTILS_H_
+#define UTILS_FILE_UTILS_H_
+
+#include <fstream>
+#include <iostream>
+#include <sstream>
+#include <string>
+
+#include <glog/logging.h>
+
+namespace gs {
+
+bool read_string_from_file(const std::string& file_path, std::string& content);
+
+bool write_string_to_file(const std::string& content,
+                          const std::string& file_path);
+
+}  // namespace gs
+
+#endif  // UTILS_FILE_UTILS_H_