feat(interactive): Add readiness probe for graph_db_service (#4586)

For the queries service, add an HTTP API to detect whether it is OK.

Author: zhanglei1949

flex/engines/http_server/handler/graph_db_http_handler.cc

@@ -881,6 +881,35 @@ class adhoc_query_handler : public StoppableHandler {
 #endif
 };
 
+class service_status_handler : public seastar::httpd::handler_base {
+ public:
+  service_status_handler() {}
+  ~service_status_handler() override = default;
+
+  seastar::future<std::unique_ptr<seastar::httpd::reply>> handle(
+      const seastar::sstring& path,
+      std::unique_ptr<seastar::httpd::request> req,
+      std::unique_ptr<seastar::httpd::reply> rep) override {
+    if (req->_method == "GET") {
+      if (path.find("ready") != seastar::sstring::npos) {
+        auto& graph_db_service = GraphDBService::get();
+        rep->set_content_type("application/json");
+        if (graph_db_service.is_actors_running()) {
+          return new_reply(std::move(rep),
+                           seastar::httpd::reply::status_type::ok,
+                           "Service Is Ready");
+        } else {
+          return new_reply(
+              std::move(rep),
+              seastar::httpd::reply::status_type::service_unavailable,
+              "Service Is Not Ready");
+        }
+      }
+    }
+    return new_bad_request_reply(std::move(rep), "Unsupported action");
+  }
+};
+
 ///////////////////////////graph_db_http_handler/////////////////////////////
 
 graph_db_http_handler::graph_db_http_handler(uint16_t http_port,
@@ -1081,6 +1110,10 @@ seastar::future<> graph_db_http_handler::set_routes() {
       r.add(match_rule, OPERATIONS[i]);
     }
 
+    r.add(seastar::httpd::operation_type::GET,
+          seastar::httpd::url("/v1/service/ready"),
+          new service_status_handler());
+
     return seastar::make_ready_future<>();
   });
 }

flex/engines/http_server/handler/http_utils.cc

@@ -17,17 +17,23 @@
 
 namespace server {
 
-seastar::future<std::unique_ptr<seastar::httpd::reply>> new_bad_request_reply(
-    std::unique_ptr<seastar::httpd::reply> rep, const std::string& msg) {
-  rep->set_status(seastar::httpd::reply::status_type::bad_request);
+seastar::future<std::unique_ptr<seastar::httpd::reply>> new_reply(
+    std::unique_ptr<seastar::httpd::reply> rep,
+    seastar::httpd::reply::status_type status, const std::string& msg) {
+  rep->set_status(status);
   rep->set_content_type("application/json");
-  gs::Status status = gs::Status(gs::StatusCode::BAD_REQUEST, msg);
   rep->write_body("json", seastar::sstring(msg));
   rep->done();
   return seastar::make_ready_future<std::unique_ptr<seastar::httpd::reply>>(
       std::move(rep));
 }
 
+seastar::future<std::unique_ptr<seastar::httpd::reply>> new_bad_request_reply(
+    std::unique_ptr<seastar::httpd::reply> rep, const std::string& msg) {
+  return new_reply(std::move(rep),
+                   seastar::httpd::reply::status_type::bad_request, msg);
+}
+
 seastar::future<std::unique_ptr<seastar::httpd::reply>>
 new_internal_error_reply(std::unique_ptr<seastar::httpd::reply> rep,
                          const std::string& msg) {

flex/engines/http_server/handler/http_utils.h

@@ -12,6 +12,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+#include <seastar/http/httpd.hh>
 #include "flex/engines/http_server/types.h"
 #include "flex/utils/result.h"
 #include "seastar/http/common.hh"
@@ -22,6 +23,10 @@
 
 namespace server {
 
+seastar::future<std::unique_ptr<seastar::httpd::reply>> new_reply(
+    std::unique_ptr<seastar::httpd::reply> rep,
+    seastar::httpd::reply::status_type status, const std::string& msg);
+
 seastar::future<std::unique_ptr<seastar::httpd::reply>> new_bad_request_reply(
     std::unique_ptr<seastar::httpd::reply> rep, const std::string& msg);
 

flex/utils/remote/oss_storage.cc

@@ -50,6 +50,10 @@ void OSSConf::load_conf_from_env() {
       bucket_name_ = bucket_name;
     }
   }
+  if (std::getenv(kOSSConcurrency)) {
+    concurrency_ = std::stoi(std::getenv(kOSSConcurrency));
+  }
+  LOG(INFO) << "OSS concurrency: " << concurrency_;
 }
 
 class UserRetryStrategy : public AlibabaCloud::OSS::RetryStrategy {
@@ -155,9 +159,8 @@ gs::Status OSSRemoteStorageUploader::Put(const std::string& local_path,
                                                  remote_path, local_path);
   request.MetaData().addHeader("x-oss-forbid-overwrite", "true");
   request.setPartSize(conf_.partition_size_);
-  request.setThreadNum(
-      conf_.uploading_concurrency_);  // Increase the thread number to improve
-                                      // the upload speed
+  request.setThreadNum(conf_.concurrency_);  // Increase the thread number to
+                                             // improve the upload speed
   auto outcome = client_->ResumableUploadObject(request);
   if (!outcome.isSuccess()) {
     std::string error_string = "OSS ResumableUploadObject from local " +
@@ -231,9 +234,8 @@ gs::Status OSSRemoteStorageDownloader::Get(const std::string& remote_path,
   AlibabaCloud::OSS::DownloadObjectRequest request(conf_.bucket_name_,
                                                    remote_path, local_path);
   request.setPartSize(conf_.partition_size_);
-  request.setThreadNum(
-      conf_.uploading_concurrency_);  // Increase the thread number to improve
-                                      // the download speed
+  request.setThreadNum(conf_.concurrency_);  // Increase the thread number to
+                                             // improve the download speed
   auto outcome = client_->ResumableDownloadObject(request);
   if (!outcome.isSuccess()) {
     std::string error_string = oss_outcome_to_string(

flex/utils/remote/oss_storage.h

@@ -32,12 +32,13 @@ struct OSSConf {
   static constexpr const char* kOSSAccessKeySecret = "OSS_ACCESS_KEY_SECRET";
   static constexpr const char* kOSSEndpoint = "OSS_ENDPOINT";
   static constexpr const char* kOSSBucketName = "OSS_BUCKET_NAME";
+  static constexpr const char* kOSSConcurrency = "OSS_CONCURRENCY";
   // Avoid storing or printing the accesskey_id and accesskey_secret
   std::string accesskey_id_;
   std::string accesskey_secret_;
   std::string endpoint_;
   std::string bucket_name_;
-  int32_t uploading_concurrency_ = 4;
+  int32_t concurrency_ = 4;
   uint64_t partition_size_ = 1024 * 1024 * 128;
   AlibabaCloud::OSS::ClientConfiguration client_conf_;