Merge pull request #9722 from HJianBo/more-confs-for-prometheus

feat: more confs for prometheus pushing

apps/emqx_dashboard/src/emqx_dashboard_swagger.erl

@@ -708,6 +708,8 @@ typename_to_spec("qos()", _Mod) ->
     #{type => integer, minimum => 0, maximum => 2, example => 0};
 typename_to_spec("{binary(), binary()}", _Mod) ->
     #{type => object, example => #{}};
+typename_to_spec("{string(), string()}", _Mod) ->
+    #{type => object, example => #{}};
 typename_to_spec("comma_separated_list()", _Mod) ->
     #{type => string, example => <<"item1,item2">>};
 typename_to_spec("comma_separated_binary()", _Mod) ->

apps/emqx_prometheus/i18n/emqx_prometheus_schema_i18n.conf

@@ -24,6 +24,35 @@ emqx_prometheus_schema {
       zh: """数据推送间隔"""
     }
   }
+
+  headers {
+    desc {
+      en: """A list of HTTP Headers when pushing to Push Gateway.<br/>
+For example, <code> { Authorization = "some-authz-tokens"}</code>"""
+      zh: """推送到 Push Gateway 的 HTTP Headers 列表。<br/>
+例如，<code> { Authorization = "some-authz-tokens"}</code>"""
+    }
+  }
+
+  job_name {
+    desc {
+      en: """Job Name that is pushed to the Push Gateway. Available variables:<br/>
+- ${name}: Name of EMQX node.<br/>
+- ${host}: Host name of EMQX node.<br/>
+For example, when the EMQX node name is <code>emqx@127.0.0.1</code> then the <code>name</code> variable takes value <code>emqx</code> and the <code>host</code> variable takes value <code>127.0.0.1</code>.<br/>
+
+Default value is: <code>${name}/instance/${name}~${host}</code>
+"""
+      zh: """推送到 Push Gateway 的 Job 名称。可用变量为：<br/>
+- ${name}: EMQX 节点的名称。
+- ${host}: EMQX 节点主机名。
+
+例如，当 EMQX 节点名为 <code>emqx@127.0.0.1</code> 则 name 变量的值为 <code>emqx</code>，host 变量的值为 <code>127.0.0.1</code>。<br/>
+
+默认值为: <code>${name}/instance/${name}~${host}</code>"""
+    }
+  }
+
   enable {
     desc {
       en: """Turn Prometheus data pushing on or off"""

apps/emqx_prometheus/src/emqx_prometheus.app.src

@@ -2,7 +2,7 @@
 {application, emqx_prometheus, [
     {description, "Prometheus for EMQX"},
     % strict semver, bump manually!
-    {vsn, "5.0.3"},
+    {vsn, "5.0.4"},
     {modules, []},
     {registered, [emqx_prometheus_sup]},
     {applications, [kernel, stdlib, prometheus, emqx]},

apps/emqx_prometheus/src/emqx_prometheus.erl

@@ -98,8 +98,13 @@ handle_cast(_Msg, State) ->
     {noreply, State}.
 
 handle_info({timeout, Timer, ?TIMER_MSG}, State = #{timer := Timer}) ->
-    #{interval := Interval, push_gateway_server := Server} = opts(),
+    #{
-    PushRes = push_to_push_gateway(Server),
+        interval := Interval,
+        headers := Headers,
+        job_name := JobName,
+        push_gateway_server := Server
+    } = opts(),
+    PushRes = push_to_push_gateway(Server, Headers, JobName),
     NewTimer = ensure_timer(Interval),
     NewState = maps:update_with(PushRes, fun(C) -> C + 1 end, 1, State#{timer => NewTimer}),
     %% Data is too big, hibernate for saving memory and stop system monitor warning.
@@ -107,18 +112,27 @@ handle_info({timeout, Timer, ?TIMER_MSG}, State = #{timer := Timer}) ->
 handle_info(_Msg, State) ->
     {noreply, State}.
 
-push_to_push_gateway(Uri) ->
+push_to_push_gateway(Uri, Headers, JobName) when is_list(Headers) ->
     [Name, Ip] = string:tokens(atom_to_list(node()), "@"),
-    Url = lists:concat([Uri, "/metrics/job/", Name, "/instance/", Name, "~", Ip]),
+    JobName1 = emqx_placeholder:preproc_tmpl(JobName),
+    JobName2 = binary_to_list(
+        emqx_placeholder:proc_tmpl(
+            JobName1,
+            #{<<"name">> => Name, <<"host">> => Ip}
+        )
+    ),
+
+    Url = lists:concat([Uri, "/metrics/job/", JobName2]),
     Data = prometheus_text_format:format(),
-    case httpc:request(post, {Url, [], "text/plain", Data}, ?HTTP_OPTIONS, []) of
+    case httpc:request(post, {Url, Headers, "text/plain", Data}, ?HTTP_OPTIONS, []) of
-        {ok, {{"HTTP/1.1", 200, "OK"}, _Headers, _Body}} ->
+        {ok, {{"HTTP/1.1", 200, _}, _RespHeaders, _RespBody}} ->
             ok;
         Error ->
             ?SLOG(error, #{
                 msg => "post_to_push_gateway_failed",
                 error => Error,
-                url => Url
+                url => Url,
+                headers => Headers
             }),
             failed
     end.

apps/emqx_prometheus/src/emqx_prometheus_api.erl

@@ -121,6 +121,8 @@ prometheus_config_example() ->
         enable => true,
         interval => "15s",
         push_gateway_server => <<"http://127.0.0.1:9091">>,
+        headers => #{'header-name' => 'header-value'},
+        job_name => <<"${name}/instance/${name}~${host}">>,
         vm_dist_collector => enabled,
         mnesia_collector => enabled,
         vm_statistics_collector => enabled,

apps/emqx_prometheus/src/emqx_prometheus_schema.erl

@@ -25,7 +25,8 @@
     roots/0,
     fields/1,
     desc/1,
-    translation/1
+    translation/1,
+    convert_headers/1
 ]).
 
 namespace() -> "prometheus".
@@ -52,6 +53,26 @@ fields("prometheus") ->
                     desc => ?DESC(interval)
                 }
             )},
+        {headers,
+            ?HOCON(
+                list({string(), string()}),
+                #{
+                    default => #{},
+                    required => false,
+                    converter => fun ?MODULE:convert_headers/1,
+                    desc => ?DESC(headers)
+                }
+            )},
+        {job_name,
+            ?HOCON(
+                binary(),
+                #{
+                    default => <<"${name}/instance/${name}~${host}">>,
+                    required => true,
+                    desc => ?DESC(job_name)
+                }
+            )},
+
         {enable,
             ?HOCON(
                 boolean(),
@@ -126,6 +147,17 @@ fields("prometheus") ->
 desc("prometheus") -> ?DESC(prometheus);
 desc(_) -> undefined.
 
+convert_headers(Headers) when is_map(Headers) ->
+    maps:fold(
+        fun(K, V, Acc) ->
+            [{binary_to_list(K), binary_to_list(V)} | Acc]
+        end,
+        [],
+        Headers
+    );
+convert_headers(Headers) when is_list(Headers) ->
+    Headers.
+
 %% for CI test, CI don't load the whole emqx_conf_schema.
 translation(Name) ->
     emqx_conf_schema:translation(Name).

apps/emqx_prometheus/test/emqx_prometheus_SUITE.erl

@@ -27,6 +27,8 @@
     "prometheus {\n"
     "  push_gateway_server = \"http://127.0.0.1:9091\"\n"
     "  interval = \"1s\"\n"
+    "  headers = { Authorization = \"some-authz-tokens\"}\n"
+    "  job_name = \"${name}~${host}\"\n"
     "  enable = true\n"
     "  vm_dist_collector = enabled\n"
     "  mnesia_collector = enabled\n"
@@ -85,6 +87,25 @@ t_collector_no_crash_test(_) ->
     prometheus_text_format:format(),
     ok.
 
+t_assert_push(_) ->
+    meck:new(httpc, [passthrough]),
+    Self = self(),
+    AssertPush = fun(Method, Req = {Url, Headers, ContentType, _Data}, HttpOpts, Opts) ->
+        ?assertEqual(post, Method),
+        ?assertMatch("http://127.0.0.1:9091/metrics/job/test~127.0.0.1", Url),
+        ?assertEqual([{"Authorization", "some-authz-tokens"}], Headers),
+        ?assertEqual("text/plain", ContentType),
+        Self ! pass,
+        meck:passthrough([Method, Req, HttpOpts, Opts])
+    end,
+    meck:expect(httpc, request, AssertPush),
+    ?assertMatch(ok, emqx_prometheus_sup:start_child(emqx_prometheus)),
+    receive
+        pass -> ok
+    after 2000 ->
+        ct:fail(assert_push_request_failed)
+    end.
+
 t_only_for_coverage(_) ->
     ?assertEqual("5.0.0", emqx_prometheus_proto_v1:introduced_in()),
     ok.

changes/v5.0.15/feat-9722.en.md

@@ -0,0 +1,3 @@
+Add the following configuration options for Pushing metrics to Prometheus Push Gateway:
+- `headers`: Allows custom HTTP request headers.
+- `job_name`: allows to customize the name of the Job pushed to Push Gateway.

changes/v5.0.15/feat-9722.zh.md

@@ -0,0 +1,3 @@
+为 Prometheus 推送到 Push Gateway 新增以下配置项：
+- `headers`：允许自定义 HTTP 请求头。
+- `job_name`：允许自定义推送到 Push Gateway 的 Job 名称。