feat: add prometheus push mode support

docs/en/prometheus_metrics.md

@@ -19,6 +19,14 @@ By default, tRPC-Cpp framework does not compile related code with prometheus. To
 
 ## Bazel
 
+Before compile, please add prometheus deps in your WORKSPACE file.
+
+```python
+load("@com_github_jupp0r_prometheus_cpp//bazel:repositories.bzl", "prometheus_cpp_repositories")
+
+prometheus_cpp_repositories()
+```
+
 Add the `"trpc_include_prometheus"` compilation option during Bazel compilation.
 
 For example, add it in `.bazelrc` file.
@@ -66,6 +74,20 @@ The description of the configuration item are as follow.
 | histogram_module_cfg | Sequences | No, the default value is [1, 10, 100, 1000] | Statistical interval for latency distribution in ModuleReport, measured in milliseconds. |
 | const_labels | Mappings | No, the default value is empty. | Default labels attached to each RPC statistical data. |
 
+By default, we use pull mode to report. If you want to use push mode, please add below configuration to your yaml file.
+
+```yaml
+plugins:
+  metrics:
+    prometheus:
+      push_mode:
+        enable: true
+        gateway_host: 127.0.0.1
+        gateway_port: 9091
+        job_name: trpc_prometheus_push_metrics
+        interval_ms: 10000 # 默认上报间隔是10s
+```
+
 ## ModuleReport
 
 **ModuleReport refers to reporting metrics data of RPC inter-module calls, including caller reporting (for tracking client-side invocation information) and callee reporting (for tracking server-side invocation information).**

docs/zh/prometheus_metrics.md

@@ -19,6 +19,14 @@ tRPC-Cpp 框架默认不会编译 Prometheus 相关的代码。若要开启，
 
 ## Bazel 启用方式
 
+编译前，需要增加prometheus的依赖在WORKSPACE文件里。
+
+```python
+load("@com_github_jupp0r_prometheus_cpp//bazel:repositories.bzl", "prometheus_cpp_repositories")
+
+prometheus_cpp_repositories()
+```
+
 bazel 编译时加上`“trpc_include_prometheus”`编译选项。
 
 例如在 `.bazelrc` 中加上：
@@ -59,6 +67,20 @@ plugins:
         key2: value2
 ```
 
+默认采用pull模式，如果需要启用push模式，则需要增加类似如下的配置：
+
+```yaml
+plugins:
+  metrics:
+    prometheus:
+      push_mode:
+        enable: true
+        gateway_host: 127.0.0.1
+        gateway_port: 9091
+        job_name: trpc_prometheus_push_metrics
+        interval_ms: 10000 # 默认上报间隔是10s
+```
+
 配置项说明：
 
 | 参数 | 类型 | 是否必须配置 | 说明 |

examples/features/prometheus/README.md

@@ -32,6 +32,14 @@ examples/features/prometheus/
 
 * Compilation
 
+Before compile, please add prometheus deps in your WORKSPACE.
+
+```python
+load("@com_github_jupp0r_prometheus_cpp//bazel:repositories.bzl", "prometheus_cpp_repositories")
+
+prometheus_cpp_repositories()
+```
+
 We can run the following command to compile the demo.
 
 ```shell

examples/features/prometheus/proxy/trpc_cpp_fiber.yaml

@@ -44,6 +44,12 @@ plugins:
       const_labels:
         const_key1: const_value1
         const_key2: const_value2
+      # push_mode:
+      #   enable: true
+      #   gateway_host: 127.0.0.1
+      #   gateway_port: 9091
+      #   job_name: trpc_prometheus_push_metrics
+      #   interval_ms: 1000
   log:
     default:
       - name: default

examples/features/prometheus/run.sh

@@ -1,12 +1,25 @@
 bazel build //examples/helloworld/...
+
+cat > WORKSPACE <<EOF
+workspace(name = "trpc_cpp")
+
+load("//trpc:workspace.bzl", "trpc_workspace")
+
+trpc_workspace()
+
+load("@com_github_jupp0r_prometheus_cpp//bazel:repositories.bzl", "prometheus_cpp_repositories")
+
+prometheus_cpp_repositories()
+EOF
+
 bazel build //examples/features/prometheus/... --define trpc_include_prometheus=true
 
 echo "begin"
 ./bazel-bin/examples/helloworld/helloworld_svr --config=examples/helloworld/conf/trpc_cpp_fiber.yaml &
 sleep 1
 ./bazel-bin/examples/features/prometheus/proxy/forward_server --config=examples/features/prometheus/proxy/trpc_cpp_fiber.yaml &
 sleep 1
-./bazel-bin/examples/features/prometheus/client/client_config --config=examples/features/prometheus/client/trpc_cpp_fiber.yaml
+./bazel-bin/examples/features/prometheus/client/client --client_config=examples/features/prometheus/client/trpc_cpp_fiber.yaml
 
 killall helloworld_svr
 if [ $? -ne 0 ]; then

third_party/com_github_thalhammer_jwt_cpp/jwt_cpp.BUILD

@@ -4,7 +4,8 @@ package(
 
 cc_library(
     name = "jwt-cpp",
-    hdrs = glob(["**/*.h"]),
+    hdrs = glob(["include/jwt-cpp/**/*.h", "include/picojson/*.h"]),
+    includes = ["include"],
     deps = [
       "@com_github_openssl_openssl//:libcrypto",
       "@com_github_openssl_openssl//:libssl",

trpc/admin/BUILD

@@ -364,11 +364,13 @@ cc_library(
             "@com_github_jupp0r_prometheus_cpp//pull",
             "@com_github_thalhammer_jwt_cpp//:jwt-cpp",
             "//trpc/metrics/prometheus:prometheus_metrics",
+            "//trpc/metrics/prometheus:prometheus_conf_parser",
         ],
         "//trpc:include_metrics_prometheus": [
             "@com_github_jupp0r_prometheus_cpp//pull",
             "@com_github_thalhammer_jwt_cpp//:jwt-cpp",
             "//trpc/metrics/prometheus:prometheus_metrics",
+            "//trpc/metrics/prometheus:prometheus_conf_parser",
         ],
     }),
 )

trpc/admin/prometheus_handler.cc

@@ -14,8 +14,11 @@
 #ifdef TRPC_BUILD_INCLUDE_PROMETHEUS
 #include "trpc/admin/prometheus_handler.h"
 
+#include "jwt-cpp/jwt.h"
 #include "prometheus/text_serializer.h"
 
+#include "trpc/metrics/prometheus/prometheus_conf_parser.h"
+
 namespace {
 bool JwtValidate(const std::string& token, std::map<std::string, std::string>& auth_cfg) {
   try {
@@ -88,19 +91,6 @@ bool PrometheusHandler::CheckBasicAuth(std::string token) {
     TRPC_FMT_ERROR("error token: {}", token);
     return false;
   }
-
-  std::string username_pwd = http::Base64Decode(std::begin(splited[1]), std::end(splited[1]));
-  auto sp = Split(username_pwd, ':');
-  if (sp.size() != 2) {
-    TRPC_FMT_ERROR("error token: {}", token);
-    return false;
-  }
-
-  auto username = sp[0], pwd = sp[1];
-  if (username != auth_cfg_["username"] || pwd != auth_cfg_["password"]) {
-    TRPC_FMT_ERROR("error username or password: username: {}, password: {}", username, pwd);
-    return false;
-  }
   return true;
 }
 
@@ -113,20 +103,10 @@ void PrometheusHandler::CommandHandle(http::HttpRequestPtr req, rapidjson::Value
   std::string token = req->GetHeader("authorization");
 
   if (!auth_cfg_.empty()) {
-    if (auth_cfg_.count("username") && auth_cfg_.count("password")) {
-      // push mode
-      // use the basic auth if already config the username and password.
-      if (!CheckBasicAuth(token)) {
-        result.AddMember("message", "wrong request without right username or password", alloc);
-        return;
-      }
-    } else {
-      // pull mode
-      // use the json web token auth.
-      if (!CheckTokenAuth(token)) {
-        result.AddMember("message", "wrong request without right token", alloc);
-        return;
-      }
+    // use the json web token auth.
+    if (!CheckTokenAuth(token)) {
+      result.AddMember("message", "wrong request without right token", alloc);
+      return;
     }
   }
 

trpc/admin/prometheus_handler.h

@@ -14,8 +14,6 @@
 #ifdef TRPC_BUILD_INCLUDE_PROMETHEUS
 #pragma once
 
-#include <jwt-cpp/jwt.h>
-
 #include "trpc/admin/admin_handler.h"
 #include "trpc/common/config/trpc_config.h"
 #include "trpc/log/trpc_log.h"

trpc/metrics/prometheus/BUILD

@@ -75,13 +75,16 @@ cc_library(
         "//trpc/util:prometheus",
         "//trpc/common/config:trpc_config",
         "//trpc/metrics",
+        "//trpc/runtime/common:periphery_task_scheduler",
     ] + select({
         "//conditions:default": [],
         "//trpc:trpc_include_prometheus": [
             "@com_github_jupp0r_prometheus_cpp//pull",
+            "@com_github_jupp0r_prometheus_cpp//push",
         ],
         "//trpc:include_metrics_prometheus": [
             "@com_github_jupp0r_prometheus_cpp//pull",
+            "@com_github_jupp0r_prometheus_cpp//push",
         ],
     }),
 )

trpc/metrics/prometheus/prometheus_conf.cc

@@ -30,32 +30,18 @@ void PrometheusConfig::Display() const {
     TRPC_LOG_DEBUG(label.first << ":" << label.second);
   }
 
-  TRPC_LOG_DEBUG("--------------------------------");
-}
-
-}  // namespace trpc
-
-namespace YAML {
+  TRPC_LOG_DEBUG("auth_cfg:");
+  for (auto auth : auth_cfg) {
+    TRPC_LOG_DEBUG(auth.first << ":" << auth.second);
+  }
 
-YAML::Node convert<trpc::PrometheusConfig>::encode(const trpc::PrometheusConfig& config) {
-  YAML::Node node;
-  node["histogram_module_cfg"] = config.histogram_module_cfg;
-  node["const_labels"] = config.const_labels;
-  node["auth_cfg"] = config.auth_cfg;
-  return node;
-}
+  TRPC_LOG_DEBUG("push_mode enable:" << push_mode.enable);
+  TRPC_LOG_DEBUG("push_mode gateway_host:" << push_mode.gateway_host);
+  TRPC_LOG_DEBUG("push_mode gateway_port:" << push_mode.gateway_port);
+  TRPC_LOG_DEBUG("push_mode job_name:" << push_mode.job_name);
+  TRPC_LOG_DEBUG("push_mode interval_ms:" << push_mode.interval_ms);
 
-bool convert<trpc::PrometheusConfig>::decode(const YAML::Node& node, trpc::PrometheusConfig& config) {
-  if (node["histogram_module_cfg"]) {
-    config.histogram_module_cfg = node["histogram_module_cfg"].as<std::vector<double>>();
-  }
-  if (node["const_labels"]) {
-    config.const_labels = node["const_labels"].as<std::map<std::string, std::string>>();
-  }
-  if (node["auth_cfg"]) {
-    config.auth_cfg = node["auth_cfg"].as<std::map<std::string, std::string>>();
-  }
-  return true;
+  TRPC_LOG_DEBUG("--------------------------------");
 }
 
-}  // namespace YAML
+}  // namespace trpc

trpc/metrics/prometheus/prometheus_conf.h

@@ -31,17 +31,15 @@ struct PrometheusConfig {
 
   std::map<std::string, std::string> auth_cfg;
 
+  struct PushMode {
+    bool enable = false;
+    std::string gateway_host;
+    std::string gateway_port;
+    std::string job_name;
+    int interval_ms = 10000;
+  } push_mode;
+
   void Display() const;
 };
 
 }  // namespace trpc
-
-namespace YAML {
-
-template <>
-struct convert<trpc::PrometheusConfig> {
-  static YAML::Node encode(const trpc::PrometheusConfig& config);
-  static bool decode(const YAML::Node& node, trpc::PrometheusConfig& config);
-};
-
-}  // namespace YAML

trpc/metrics/prometheus/prometheus_conf_parser.h

@@ -26,7 +26,13 @@ struct convert<trpc::PrometheusConfig> {
     node["histogram_module_cfg"] = conf.histogram_module_cfg;
     node["const_labels"] = conf.const_labels;
     node["auth_cfg"] = conf.auth_cfg;
-
+    node["push_mode"]["enable"] = conf.push_mode.enable;
+    if (conf.push_mode.enable) {
+      node["push_mode"]["gateway_host"] = conf.push_mode.gateway_host;
+      node["push_mode"]["gateway_port"] = conf.push_mode.gateway_port;
+      node["push_mode"]["job_name"] = conf.push_mode.job_name;
+      node["push_mode"]["interval_ms"] = conf.push_mode.interval_ms;
+    }
     return node;
   }
 
@@ -42,6 +48,16 @@ struct convert<trpc::PrometheusConfig> {
     if (node["auth_cfg"]) {
       conf.auth_cfg = node["auth_cfg"].as<std::map<std::string, std::string>>();
     }
+    if (node["push_mode"]) {
+      const auto& push_mode = node["push_mode"];
+      conf.push_mode.enable = push_mode["enable"].as<bool>(false);
+      if (conf.push_mode.enable) {
+        conf.push_mode.gateway_host = push_mode["gateway_host"].as<std::string>();
+        conf.push_mode.gateway_port = push_mode["gateway_port"].as<std::string>();
+        conf.push_mode.job_name = push_mode["job_name"].as<std::string>();
+        conf.push_mode.interval_ms = push_mode["interval_ms"].as<int>(10000);
+      }
+    }
     return true;
   }
 };

trpc/metrics/prometheus/prometheus_metrics.cc

@@ -15,9 +15,15 @@
 #include "trpc/metrics/prometheus/prometheus_metrics.h"
 
 #include "trpc/common/config/trpc_config.h"
+#include "trpc/metrics/prometheus/prometheus_conf_parser.h"
+#include "trpc/runtime/common/periphery_task_scheduler.h"
+
+#include "prometheus/gateway.h"
 
 namespace trpc {
 
+constexpr const int kPushToGatewaySucc = 200;
+
 int PrometheusMetrics::Init() noexcept {
   bool ret = TrpcConfig::GetInstance()->GetPluginConfig<PrometheusConfig>(
       "metrics", trpc::prometheus::kPrometheusMetricsName, prometheus_conf_);
@@ -47,6 +53,47 @@ int PrometheusMetrics::Init() noexcept {
   return 0;
 }
 
+void PrometheusMetrics::Start() noexcept {
+  // initialize prometheus pusher task
+  if (prometheus_conf_.push_mode.enable) {
+    if (push_gateway_task_id_ == 0) {
+      const auto& push_conf = prometheus_conf_.push_mode;
+      const auto& auth_conf = prometheus_conf_.auth_cfg;
+      std::string username, password;
+      if (!auth_conf.empty()) {
+        auto it = auth_conf.find("username");
+        if (it != auth_conf.end()) {
+          username = it->second;
+        }
+        it = auth_conf.find("password");
+        if (it != auth_conf.end()) {
+          password = it->second;
+        }
+      }
+      ::prometheus::Labels labels;
+      std::unique_ptr<::prometheus::Gateway> gateway = std::make_unique<::prometheus::Gateway>(
+          push_conf.gateway_host, push_conf.gateway_port, push_conf.job_name, labels, username, password);
+      gateway->RegisterCollectable(trpc::prometheus::GetRegistry());
+      push_gateway_task_id_ = PeripheryTaskScheduler::GetInstance()->SubmitInnerPeriodicalTask(
+          [gateway = std::move(gateway)]() {
+            int ret = gateway->Push();
+            if (ret != kPushToGatewaySucc) {
+              TRPC_FMT_ERROR("Failed to push metrics to the gateway");
+            }
+          },
+          prometheus_conf_.push_mode.interval_ms, "PrometheusPushGatewayTask");
+    }
+  }
+}
+
+void PrometheusMetrics::Stop() noexcept {
+  if (push_gateway_task_id_ != 0) {
+    PeripheryTaskScheduler::GetInstance()->StopInnerTask(push_gateway_task_id_);
+    PeripheryTaskScheduler::GetInstance()->JoinInnerTask(push_gateway_task_id_);
+    push_gateway_task_id_ = 0;
+  }
+}
+
 int PrometheusMetrics::ModuleReport(const ModuleMetricsInfo& info) {
   // report the number of calls and the time it takes to execute.
   if (info.source == kMetricsCallerSource) {  // caller report