Skip to content

Commit d43403d

Browse files
albertoredalbertored
committed
Reader setup with httpd server
1 parent b8b3bf8 commit d43403d

File tree

3 files changed

+179
-45
lines changed

3 files changed

+179
-45
lines changed

‎apps/opentelemetry_experimental/src/otel_metric_reader.erl‎

Lines changed: 28 additions & 45 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,8 @@
2626

2727
-export([start_link/3,
2828
collect/1,
29-
shutdown/1]).
29+
shutdown/1,
30+
collect_/4]).
3031

3132
-export([init/1,
3233
handle_call/3,
@@ -64,7 +65,7 @@ start_link(ReaderId, ProviderSup, Config) ->
6465
gen_server:start_link(?MODULE, [ReaderId, ProviderSup, Config], []).
6566

6667
collect(ReaderPid) ->
67-
ReaderPid ! collect.
68+
gen_server:call(ReaderPid, collect).
6869

6970
shutdown(ReaderPid) ->
7071
gen_server:call(ReaderPid, shutdown).
@@ -111,62 +112,44 @@ handle_continue(register_with_server, State=#state{provider_sup=ProviderSup,
111112

112113
handle_call(shutdown, _From, State) ->
113114
{reply, ok, State};
114-
handle_call(_, _From, State) ->
115-
{noreply, State}.
116-
117-
handle_cast(_, State) ->
118-
{noreply, State}.
119-
120-
handle_info(collect, State=#state{exporter=undefined,
121-
export_interval_ms=ExporterIntervalMs,
122-
tref=TRef}) when TRef =/= undefined andalso
123-
ExporterIntervalMs =/= undefined ->
124-
erlang:cancel_timer(TRef, [{async, true}]),
125-
NewTRef = erlang:send_after(ExporterIntervalMs, self(), collect),
126-
{noreply, State#state{tref=NewTRef}};
127-
handle_info(collect, State=#state{id=ReaderId,
128-
exporter={ExporterModule, Config},
129-
export_interval_ms=undefined,
130-
tref=undefined,
115+
handle_call(collect, _From, State=#state{id=ReaderId,
116+
exporter=Exporter,
131117
callbacks_tab=CallbacksTab,
132118
view_aggregation_tab=ViewAggregationTab,
133119
metrics_tab=MetricsTab,
134120
resource=Resource
135121
}) ->
136-
%% collect from view aggregations table and then export
137-
Metrics = collect_(CallbacksTab, ViewAggregationTab, MetricsTab, ReaderId),
138-
139-
otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config),
140-
141-
{noreply, State};
142-
handle_info(collect, State=#state{id=ReaderId,
143-
exporter={ExporterModule, Config},
144-
export_interval_ms=ExporterIntervalMs,
145-
tref=TRef,
146-
callbacks_tab=CallbacksTab,
147-
view_aggregation_tab=ViewAggregationTab,
148-
metrics_tab=MetricsTab,
149-
resource=Resource
150-
}) when TRef =/= undefined andalso
151-
ExporterIntervalMs =/= undefined ->
152-
erlang:cancel_timer(TRef, [{async, true}]),
153-
NewTRef = erlang:send_after(ExporterIntervalMs, self(), collect),
154-
155-
%% collect from view aggregations table and then export
156-
Metrics = collect_(CallbacksTab, ViewAggregationTab, MetricsTab, ReaderId),
157-
158-
159-
otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config),
122+
TRef = update_timer(State#state.tref, State#state.export_interval_ms),
123+
Reply = collect_and_export(ReaderId, Exporter, CallbacksTab, ViewAggregationTab, MetricsTab, Resource),
124+
{reply, Reply, State#state{tref=TRef}};
125+
handle_call(_, _From, State) ->
126+
{noreply, State}.
160127

161-
{noreply, State#state{tref=NewTRef}};
162-
%% no tref or exporter, do nothing at all
128+
handle_info(collect, State) ->
129+
{reply, _, NewState} = handle_call(collect, undefined, State),
130+
{noreply, NewState};
163131
handle_info(_, State) ->
164132
{noreply, State}.
165133

134+
handle_cast(_, State) ->
135+
{noreply, State}.
136+
166137
code_change(State) ->
167138
{ok, State}.
168139

169140
%%
141+
collect_and_export(_ReaderId, undefined, _CallbacksTab, _ViewAggregationTab, _MetricsTab, _Resource) ->
142+
ok;
143+
collect_and_export(ReaderId, {ExporterModule, Config}, CallbacksTab, ViewAggregationTab, MetricsTab, Resource) ->
144+
%% collect from view aggregations table and then export
145+
Metrics = collect_(CallbacksTab, ViewAggregationTab, MetricsTab, ReaderId),
146+
otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config).
147+
148+
update_timer(undefined, undefined) ->
149+
undefined;
150+
update_timer(TRef, ExporterIntervalMs) ->
151+
erlang:cancel_timer(TRef, [{async, true}]),
152+
erlang:send_after(ExporterIntervalMs, self(), collect).
170153

171154
-spec collect_(any(), ets:table(), any(), reference()) -> [any()].
172155
collect_(CallbacksTab, ViewAggregationTab, MetricsTab, ReaderId) ->

‎apps/opentelemetry_experimental/src/otel_metric_reader_periodic.erl‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
%%%------------------------------------------------------------------------
2+
%% Copyright 2022, OpenTelemetry Authors
3+
%% Licensed under the Apache License, Version 2.0 (the "License");
4+
%% you may not use this file except in compliance with the License.
5+
%% You may obtain a copy of the License at
6+
%%
7+
%% http://www.apache.org/licenses/LICENSE-2.0
8+
%%
9+
%% Unless required by applicable law or agreed to in writing, software
10+
%% distributed under the License is distributed on an "AS IS" BASIS,
11+
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
%% See the License for the specific language governing permissions and
13+
%% limitations under the License.
14+
%%
15+
%% @doc TODO
16+
%% @end
17+
%%%-------------------------------------------------------------------------
18+
-module(otel_metric_reader_periodic).
19+
20+
-export([start_link/3,
21+
collect/1,
22+
shutdown/1]).
23+
24+
start_link(ReaderId, ProviderSup, Config) ->
25+
ConfigUpdated = maps:update_with(export_interval_ms, fun(ExporterIntervalMs) -> ExporterIntervalMs end, 60000, Config),
26+
gen_server:start_link(otel_metric_reader, [ReaderId, ProviderSup, ConfigUpdated], []).
27+
28+
collect(ReaderPid) ->
29+
otel_metric_reader:collect(ReaderPid).
30+
31+
shutdown(ReaderPid) ->
32+
otel_metric_reader:shutdown(ReaderPid).

‎apps/opentelemetry_experimental/src/otel_metric_reader_prometheus.erl‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
%%%------------------------------------------------------------------------
2+
%% Copyright 2022, OpenTelemetry Authors
3+
%% Licensed under the Apache License, Version 2.0 (the "License");
4+
%% you may not use this file except in compliance with the License.
5+
%% You may obtain a copy of the License at
6+
%%
7+
%% http://www.apache.org/licenses/LICENSE-2.0
8+
%%
9+
%% Unless required by applicable law or agreed to in writing, software
10+
%% distributed under the License is distributed on an "AS IS" BASIS,
11+
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
%% See the License for the specific language governing permissions and
13+
%% limitations under the License.
14+
%%
15+
%% @doc TODO
16+
%% @end
17+
%%%-------------------------------------------------------------------------
18+
-module(otel_metric_reader_prometheus).
19+
20+
-behaviour(supervisor).
21+
22+
-export([start_link/3,
23+
collect/1,
24+
shutdown/1]).
25+
26+
-export([init/1,
27+
do/1]).
28+
29+
-include_lib("kernel/include/logger.hrl").
30+
-include_lib("inets/include/httpd.hrl").
31+
-include_lib("opentelemetry_api_experimental/include/otel_metrics.hrl").
32+
33+
-define(TEMPORALITY_MAPPING, #{
34+
?KIND_COUNTER => ?TEMPORALITY_CUMULATIVE,
35+
?KIND_OBSERVABLE_COUNTER => ?TEMPORALITY_CUMULATIVE,
36+
?KIND_HISTOGRAM => ?TEMPORALITY_CUMULATIVE,
37+
?KIND_OBSERVABLE_GAUGE => ?TEMPORALITY_CUMULATIVE,
38+
?KIND_UPDOWN_COUNTER => ?TEMPORALITY_CUMULATIVE,
39+
?KIND_OBSERVABLE_UPDOWNCOUNTER => ?TEMPORALITY_CUMULATIVE
40+
}).
41+
42+
start_link(ReaderId, ProviderSup, Config) ->
43+
supervisor:start_link(?MODULE, [ReaderId, ProviderSup, Config]).
44+
45+
init([ReaderId, ProviderSup, Config]) ->
46+
% TODO warning if default_temporality_mapping, export_interval_ms, exporter
47+
% are present in the configuration
48+
Config1 = maps:put(default_temporality_mapping, ?TEMPORALITY_MAPPING, Config),
49+
Config2 = maps:remove(export_interval_ms, Config1),
50+
Config3 = maps:put(exporter, {otel_metric_exporter_prometheus, Config2}, Config2),
51+
52+
SupFlags = #{strategy => one_for_one,
53+
intensity => 5,
54+
period => 10},
55+
56+
ReaderChildSpec = #{
57+
id => ReaderId,
58+
start => {otel_metric_reader, start_link, [ReaderId, ProviderSup, Config3]},
59+
type => worker,
60+
restart => permanent,
61+
shutdown => 1000
62+
},
63+
64+
ChildSpecs = case maps:get(endpoint_port, Config, undefined) of
65+
undefined ->
66+
[ReaderChildSpec];
67+
HttpdPort when is_integer(HttpdPort) ->
68+
HttpdOpts = [
69+
{server_name, "OTel Prometheus exporter"},
70+
{server_tokens, {private, "TODO"}},
71+
{server_root, "/tmp"},
72+
{document_root, "/tmp"},
73+
{port, HttpdPort},
74+
{modules, [?MODULE]},
75+
{otel_metric_reader, {self(), ReaderId}},
76+
{pt_key, make_ref()}
77+
],
78+
HttpdChildSpec = #{
79+
id => make_ref(),
80+
start => {inets, start, [httpd, HttpdOpts, stand_alone]},
81+
type => worker,
82+
restart => permanent,
83+
shutdown => 1000
84+
},
85+
[ReaderChildSpec, HttpdChildSpec]
86+
end,
87+
88+
{ok, {SupFlags, ChildSpecs}}.
89+
90+
collect(ReaderPid) ->
91+
otel_metric_reader:collect(ReaderPid).
92+
93+
shutdown(ReaderPid) ->
94+
otel_metric_reader:shutdown(ReaderPid).
95+
96+
do(#mod{method="GET",request_uri="/metrics",config_db=ConfigDb}) ->
97+
ReaderPid = get_reader_pid(ConfigDb),
98+
Metrics = collect(ReaderPid),
99+
Headers = [
100+
{code, 200},
101+
{content_length, integer_to_list(iolist_size(Metrics))},
102+
{content_type, "text/plain; version=0.0.4"}
103+
],
104+
{proceed, [{response, {response, Headers, Metrics}}]};
105+
do(#mod{}) ->
106+
{proceed, [{response, {404, "Not found"}}]}.
107+
108+
get_reader_pid(ConfigDb) ->
109+
[PTKey] = ets:lookup_element(ConfigDb, pt_key, 2),
110+
case persistent_term:get(PTKey, undefined) of
111+
undefined ->
112+
[{ReaderSupPid, ReaderId}] = ets:lookup_element(ConfigDb, otel_metric_reader, 2),
113+
Children = supervisor:which_children(ReaderSupPid),
114+
{value, {_, ReaderPid, _, _}} = lists:search(fun({Id, _, _, _}) -> Id == ReaderId end, Children),
115+
persistent_term:put(PTKey, ReaderPid),
116+
ReaderPid;
117+
ReaderPid ->
118+
ReaderPid
119+
end.

0 commit comments

Comments
 (0)