feat(interactive): Introduce max_content_size for admin_service (#4491)

zhanglei1949 · web-flow · commit 508c4200885f · 2025-02-17T10:45:24.000+08:00
As titled.
diff --git a/.github/workflows/flex-interactive.yml b/.github/workflows/flex-interactive.yml
@@ -95,10 +95,8 @@ jobs:
         GS_TEST_DIR: /tmp/gstest
       run: |
         # launch service: ${COORDINATOR_PORT} for coordinator http port; ${CYPHER_PORT} for cypher port;
-        cat >>  ${GITHUB_WORKSPACE}/flex/tests/hqps/interactive_config_test.yaml << EOF
-        coordinator:
-          max_content_length: 1048576 # 1MB
-        EOF
+        # replace max_content_length to 1MB
+        sed -i 's/max_content_length: .*/max_content_length: 1MB/g' ${GITHUB_WORKSPACE}/flex/tests/hqps/interactive_config_test.yaml
         gsctl instance deploy --type interactive --image-registry graphscope --image-tag latest --cypher-port ${CYPHER_PORT} \
               --coordinator-port ${COORDINATOR_PORT} --config ${GITHUB_WORKSPACE}/flex/tests/hqps/interactive_config_test.yaml \
               --admin-port 7778 --storedproc-port 10001
diff --git a/coordinator/gscoordinator/coordinator.py b/coordinator/gscoordinator/coordinator.py
@@ -37,6 +37,7 @@
 from gscoordinator.flex.core.config import SOLUTION
 from gscoordinator.flex.encoder import JSONEncoder
 from gscoordinator.monitor import Monitor
+from gscoordinator.utils import human_readable_to_bytes
 
 logger = logging.getLogger("graphscope")
 
@@ -134,7 +135,9 @@ def start_http_service(config):
     initialize_client_wrapper(config)
     app = connexion.App(__name__, specification_dir="./flex/openapi/")
     app.app.json_encoder = JSONEncoder
-    app.app.config["MAX_CONTENT_LENGTH"] = config.coordinator.max_content_length
+    app.app.config["MAX_CONTENT_LENGTH"] = human_readable_to_bytes(
+        config.coordinator.max_content_length
+    )
     app.add_api(
         "openapi.yaml",
         arguments={"title": "GraphScope FLEX HTTP SERVICE API"},
diff --git a/coordinator/gscoordinator/utils.py b/coordinator/gscoordinator/utils.py
@@ -2170,3 +2170,36 @@ def replace_string_in_dict(dict_obj, old, new):
     elif isinstance(dict_obj, str):
         return dict_obj.replace(old, new)
     return dict_obj
+
+
+# Reference: https://gist.github.com/beugley/ccd69945346759eb6142272a6d69b4e0
+def human_readable_to_bytes(size):
+    """Given a human-readable byte string (e.g. 2G, 30M, 20K),
+    return the number of bytes.  Will raise an exception if the argument has
+    unexpected form.
+    """
+    # Try to parse the size as if the unit was coded on 1 char.
+    try:
+        numeric_size = float(size[:-1])
+        unit = size[-1]
+    except ValueError:
+        try:
+            # Try to parse the size as if the unit was coded on 2 chars.
+            numeric_size = float(size[:-2])
+            unit = size[-2:-1]
+        except ValueError:
+            raise ValueError("Can't convert %r to bytes" % size)
+
+    unit = unit.upper()
+
+    # Now we have a numeric value and a unit. Check the unit and convert to bytes.
+    if unit == "G":
+        bytes = numeric_size * 1073741824
+    elif unit == "M":
+        bytes = numeric_size * 1048576
+    elif unit == "K":
+        bytes = numeric_size * 1024
+    else:
+        bytes = numeric_size
+
+    return int(bytes)
diff --git a/docs/flex/interactive/configuration.md b/docs/flex/interactive/configuration.md
@@ -61,20 +61,6 @@ gsctl instance deploy --type interactive --config ./interactive_config.yaml
 Please be aware that you're not required to configure every option. Simply adjust the settings that are relevant to your needs. Any options left unconfigured will automatically adopt their default values, as detailed in the following sections.
 ```
 
-#### Customize Coordinator Service Configuration
-
-When deploying Interactive, a supplementary service called the Coordinator is also initiated. This service functions similarly to the `ApiServer` in Kubernetes, providing users with a consistent and standardized set of APIs to interact with the GraphScope platform.
-You could also customize the configuration for coordinator service, just put the configuration in `coordinator` section.
-
-An example of a coordinator related configuration is as follows:
-
-```yaml
-coordinator:
-  max_content_length: 5242880 # maximum bytes of content the coordinator can handle
-```
-
-For more details, consult [config.py](https://github.com/alibaba/GraphScope/blob/main/python/graphscope/config.py).
-
 
 ##### Sample Configuration
 Here's a glimpse of what a typical YAML configuration file might look like:
@@ -134,7 +120,7 @@ In this following table, we use the `.` notation to represent the hierarchy with
 | compiler.planner.rules.NotMatchToAntiJoinRule | N/A | An optimization rule that transforms a "not exist" pattern into an anti-join operation  | 0.0.1 |
 | compiler.query_timeout  | 3000000   ｜ The maximum time for compiler to wait engine's reply, in `ms`  | 0.0.3 | 
 | http_service.sharding_mode | exclusive | The sharding mode for http service, In exclusive mode, one shard is reserved exclusively for service admin request. In cooperative, both query request and admin request could be served by any shard. | 0.5 |
-| coordinator.max_content_length | 1073741824 | The maximum length of a http request that coordinator could handle | 0.5 |
+| http_service.max_content_length | 1GB | The maximum length of a http request that admin http service could handle | 0.5 |
 
 #### TODOs
 
diff --git a/flex/engines/http_server/actor/admin_actor.act.cc b/flex/engines/http_server/actor/admin_actor.act.cc
@@ -1345,9 +1345,9 @@ seastar::future<admin_query_result> admin_actor::run_get_graph_statistic(
 }
 
 seastar::future<admin_query_result> admin_actor::upload_file(
-    query_param&& query_param) {
-  auto& content = query_param.content;
-  auto upload_res = WorkDirManipulator::CreateFile(content);
+    graph_management_param&& query_param) {
+  auto upload_res = WorkDirManipulator::CreateFile(query_param.content.first,
+                                                   query_param.content.second);
   if (upload_res.ok()) {
     auto value = upload_res.value();
     return seastar::make_ready_future<admin_query_result>(
diff --git a/flex/engines/http_server/actor/admin_actor.act.h b/flex/engines/http_server/actor/admin_actor.act.h
@@ -72,7 +72,7 @@ class ANNOTATION(actor:impl) admin_actor : public hiactor::actor {
 
   seastar::future<admin_query_result> ANNOTATION(actor:method) run_get_graph_statistic(query_param&& param);
 
-  seastar::future<admin_query_result> ANNOTATION(actor:method) upload_file(query_param&& param);
+  seastar::future<admin_query_result> ANNOTATION(actor:method) upload_file(graph_management_param&& file_name_content);
 
   // DECLARE_RUN_QUERIES;
   /// Declare `do_work` func here, no need to implement.
diff --git a/flex/engines/http_server/graph_db_service.cc b/flex/engines/http_server/graph_db_service.cc
@@ -48,7 +48,8 @@ ServiceConfig::ServiceConfig()
       metadata_store_type_(gs::MetadataStoreType::kLocalFile),
       log_level(DEFAULT_LOG_LEVEL),
       verbose_level(DEFAULT_VERBOSE_LEVEL),
-      sharding_mode(DEFAULT_SHARDING_MODE) {}
+      sharding_mode(DEFAULT_SHARDING_MODE),
+      admin_svc_max_content_length(DEFAULT_MAX_CONTENT_LENGTH) {}
 
 const std::string GraphDBService::DEFAULT_GRAPH_NAME = "modern_graph";
 const std::string GraphDBService::DEFAULT_INTERACTIVE_HOME = "/opt/flex/";
@@ -121,7 +122,8 @@ void GraphDBService::init(const ServiceConfig& config) {
       config.enable_adhoc_handler);
   if (config.start_admin_service) {
     admin_hdl_ = std::make_unique<admin_http_handler>(
-        config.admin_port, config.get_exclusive_shard_id());
+        config.admin_port, config.get_exclusive_shard_id(),
+        config.admin_svc_max_content_length);
   }
 
   initialized_.store(true);
diff --git a/flex/engines/http_server/graph_db_service.h b/flex/engines/http_server/graph_db_service.h
@@ -47,6 +47,8 @@ struct ServiceConfig {
       0;  // 0 = INFO, 1 = WARNING, 2 = ERROR, 3 = FATAL
   static constexpr const ShardingMode DEFAULT_SHARDING_MODE =
       ShardingMode::EXCLUSIVE;
+  static constexpr const uint32_t DEFAULT_MAX_CONTENT_LENGTH =
+      1024 * 1024 * 1024;  // 1GB
 
   // Those has default value
   uint32_t bolt_port;
@@ -80,7 +82,8 @@ struct ServiceConfig {
 
   // Those has not default value
   std::string default_graph;
-  std::string engine_config_path;  // used for codegen.
+  std::string engine_config_path;       // used for codegen.
+  size_t admin_svc_max_content_length;  // max content length for admin service.
 
   ServiceConfig();
 
@@ -95,6 +98,20 @@ struct ServiceConfig {
     }
   }
 
+  void set_admin_svc_max_content_length(size_t max_content_length) {
+    this->admin_svc_max_content_length = max_content_length;
+  }
+
+  void set_admin_svc_max_content_length(const std::string& max_content_length) {
+    auto val = gs::human_readable_to_bytes(max_content_length);
+    if (val == 0) {
+      LOG(ERROR) << "Invalid max_content_length: " << max_content_length << ", "
+                 << "use default value: " << DEFAULT_MAX_CONTENT_LENGTH;
+      val = DEFAULT_MAX_CONTENT_LENGTH;
+    }
+    set_admin_svc_max_content_length(val);
+  }
+
   int32_t get_exclusive_shard_id() const {
     return sharding_mode == ShardingMode::EXCLUSIVE ? shard_num - 1 : -1;
   }
@@ -286,6 +303,12 @@ struct convert<server::ServiceConfig> {
         service_config.set_sharding_mode(sharding_mode);
         VLOG(1) << "sharding_mode: " << sharding_mode;
       }
+      if (http_service_node["max_content_length"]) {
+        service_config.set_admin_svc_max_content_length(
+            http_service_node["max_content_length"].as<std::string>());
+        LOG(INFO) << "max_content_length: "
+                  << service_config.admin_svc_max_content_length;
+      }
     } else {
       LOG(ERROR) << "Fail to find http_service configuration";
       return false;
diff --git a/flex/engines/http_server/handler/admin_http_handler.cc b/flex/engines/http_server/handler/admin_http_handler.cc
@@ -98,8 +98,8 @@ class admin_file_upload_handler_impl : public seastar::httpd::handler_base {
           std::move(results));
     } else {
       return admin_actor_refs_[dst_executor]
-          .upload_file(
-              query_param{std::move(file_name_and_contents[cur_ind].second)})
+          .upload_file(graph_management_param{
+              std::move(file_name_and_contents[cur_ind])})
           .then_wrapped([this, dst_executor, cur_ind,
                          file_name_and_contents =
                              std::move(file_name_and_contents),
@@ -165,8 +165,8 @@ class admin_file_upload_handler_impl : public seastar::httpd::handler_base {
     VLOG(10) << "filenames: " << gs::to_string(filenames);
     VLOG(10) << "content types" << gs::to_string(content_types);
     std::vector<std::pair<seastar::sstring, seastar::sstring>> res;
-    for (size_t i = 0; i < names.size(); ++i) {
-      res.emplace_back(names[i], contents[i]);
+    for (size_t i = 0; i < filenames.size(); ++i) {
+      res.emplace_back(filenames[i], contents[i]);
     }
     return res;
   }
@@ -642,15 +642,22 @@ class admin_http_job_handler_impl : public seastar::httpd::handler_base {
 };
 
 admin_http_handler::admin_http_handler(uint16_t http_port,
-                                       int32_t exclusive_shard_id)
-    : http_port_(http_port), exclusive_shard_id_(exclusive_shard_id) {}
+                                       int32_t exclusive_shard_id,
+                                       size_t max_content_length)
+    : http_port_(http_port),
+      exclusive_shard_id_(exclusive_shard_id),
+      max_content_length_(max_content_length) {}
 
 void admin_http_handler::start() {
   auto fut = seastar::alien::submit_to(
       *seastar::alien::internal::default_instance, 0, [this] {
         return server_.start()
             .then([this] { return set_routes(); })
-            .then([this] { return server_.listen(http_port_); })
+            .then([this] {
+              server_.server().local().set_content_length_limit(
+                  max_content_length_);
+              return server_.listen(http_port_);
+            })
             .then([this] {
               fmt::print(
                   "HQPS admin http handler is listening on port {} ...\n",
diff --git a/flex/engines/http_server/handler/admin_http_handler.h b/flex/engines/http_server/handler/admin_http_handler.h
@@ -28,7 +28,8 @@ namespace server {
 class InteractiveAdminService;
 class admin_http_handler {
  public:
-  admin_http_handler(uint16_t http_port, int32_t exclusive_shard_id);
+  admin_http_handler(uint16_t http_port, int32_t exclusive_shard_id,
+                     size_t max_content_length);
 
   void start();
   void stop();
@@ -39,6 +40,7 @@ class admin_http_handler {
  private:
   const uint16_t http_port_;
   int32_t exclusive_shard_id_;  // -1 means not exclusive
+  size_t max_content_length_;
   seastar::httpd::http_server_control server_;
 };
 
diff --git a/flex/engines/http_server/workdir_manipulator.cc b/flex/engines/http_server/workdir_manipulator.cc
@@ -639,7 +639,7 @@ std::string WorkDirManipulator::GetLogDir() {
 }
 
 std::string WorkDirManipulator::GetUploadDir() {
-  auto upload_dir = workspace + UPLOAD_DIR;
+  auto upload_dir = workspace + "/" + UPLOAD_DIR;
   if (!std::filesystem::exists(upload_dir)) {
     std::filesystem::create_directory(upload_dir);
   }
@@ -678,27 +678,21 @@ gs::Result<std::string> WorkDirManipulator::CommitTempIndices(
 }
 
 gs::Result<std::string> WorkDirManipulator::CreateFile(
-    const seastar::sstring& content) {
+    const seastar::sstring& raw_file_name, const seastar::sstring& content) {
   if (content.size() == 0) {
     return {gs::Status(gs::StatusCode::INVALID_ARGUMENT, "Content is empty")};
   }
-  if (content.size() > MAX_CONTENT_SIZE) {
-    return {
-        gs::Status(gs::StatusCode::INVALID_ARGUMENT,
-                   "Content is too large" + std::to_string(content.size()))};
-  }
 
+  std::string file_name = GetUploadDir() + "/" + raw_file_name.c_str();
   // get the timestamp as the file name
-  boost::uuids::uuid uuid = boost::uuids::random_generator()();
-  auto file_name = GetUploadDir() + "/" + boost::uuids::to_string(uuid);
   std::ofstream fout(file_name);
   if (!fout.is_open()) {
     return {gs::Status(gs::StatusCode::PERMISSION_DENIED, "Fail to open file")};
   }
   fout << content;
   fout.close();
   LOG(INFO) << "Successfully create file: " << file_name;
-  return file_name;
+  return gs::Result<std::string>(file_name);
 }
 
 // graph_name can be a path, first try as it is absolute path, or
diff --git a/flex/engines/http_server/workdir_manipulator.h b/flex/engines/http_server/workdir_manipulator.h
@@ -61,7 +61,6 @@ class WorkDirManipulator {
   static const std::string TMP_DIR;
   static const std::string GRAPH_LOADER_BIN;
   static const std::string UPLOAD_DIR;
-  static constexpr int32_t MAX_CONTENT_SIZE = 100 * 1024 * 1024;  // 100MB
 
   static void SetWorkspace(const std::string& workspace_path);
 
@@ -182,7 +181,8 @@ class WorkDirManipulator {
   //       them after the loading process.
   //       TODO(zhanglei): Consider the bulk loading may fail, so we will
   //       automatically clear the uploaded files after a period of time.
-  static gs::Result<std::string> CreateFile(const seastar::sstring& content);
+  static gs::Result<std::string> CreateFile(const seastar::sstring& file_name,
+                                            const seastar::sstring& content);
 
  private:
   static std::string get_tmp_bulk_loading_job_log_path(
diff --git a/flex/tests/hqps/interactive_config_test.yaml b/flex/tests/hqps/interactive_config_test.yaml
@@ -41,5 +41,4 @@ http_service:
   default_listen_address: localhost
   admin_port: 7777
   query_port: 10000
-coordinator:
-  max_content_length: 1048576 # 1MB
+  max_content_length: 1GB
diff --git a/flex/utils/service_utils.cc b/flex/utils/service_utils.cc
@@ -139,4 +139,60 @@ std::string memory_to_mb_str(uint64_t mem_bytes) {
   return std::to_string(mem_mb) + "MB";
 }
 
+// Possible input: 1KB, 1B, 1K, 2Gi, 4GB
+size_t human_readable_to_bytes(const std::string& human_readable_bytes) {
+  // Check if the input is empty
+  if (human_readable_bytes.empty()) {
+    return 0;
+  }
+
+  // Define the multipliers for various size units
+  static std::unordered_map<std::string, size_t> multipliers = {
+      {"B", 1},
+      {"KB", 1024ul},
+      {"MB", 1024ul * 1024},
+      {"GB", 1024ul * 1024 * 1024},
+      {"KiB", 1024ul},
+      {"MiB", 1024ul * 1024},
+      {"GiB", 1024ul * 1024 * 1024}};
+
+  size_t pos = 0;
+
+  // Read and validate the numeric part
+  while (pos < human_readable_bytes.size() &&
+         (isdigit(human_readable_bytes[pos]) ||
+          human_readable_bytes[pos] == '.' ||
+          human_readable_bytes[pos] == ' ')) {
+    pos++;
+  }
+
+  // If no numeric part, return 0
+  if (pos == 0) {
+    return 0;
+  }
+
+  // Extract the numeric portion as a string
+  std::string number_str = human_readable_bytes.substr(0, pos);
+  double number = std::stod(number_str);  // Convert to double for calculation
+
+  // Read the unit part
+  std::string unit;
+  if (pos < human_readable_bytes.size()) {
+    unit = human_readable_bytes.substr(pos);
+  }
+
+  // Normalize the unit to uppercase
+  if (!unit.empty()) {
+    std::transform(unit.begin(), unit.end(), unit.begin(), ::toupper);
+  }
+
+  // If the unit is not in the multipliers map, return 0
+  if (multipliers.count(unit) == 0) {
+    return 0;
+  }
+
+  // Calculate bytes
+  return static_cast<size_t>(number * multipliers[unit]);
+}
+
 }  // namespace gs
diff --git a/flex/utils/service_utils.h b/flex/utils/service_utils.h
@@ -196,6 +196,8 @@ std::pair<double, double> get_current_cpu_usage();
 
 std::string memory_to_mb_str(uint64_t mem_bytes);
 
+size_t human_readable_to_bytes(const std::string& human_readable);
+
 }  // namespace gs
 
 #endif  // SERVICE_UTILS_H
diff --git a/k8s/dockerfiles/flex-interactive.Dockerfile b/k8s/dockerfiles/flex-interactive.Dockerfile
diff --git a/k8s/dockerfiles/interactive-entrypoint.sh b/k8s/dockerfiles/interactive-entrypoint.sh
diff --git a/python/graphscope/config.py b/python/graphscope/config.py
diff --git a/python/graphscope/gsctl/commands/dev.py b/python/graphscope/gsctl/commands/dev.py
