Skip to content

Commit 6de9672

Browse files
albertoredalbertored
committed
Reader setup with httpd server
1 parent aa4089e commit 6de9672

File tree

3 files changed

+191
-40
lines changed

3 files changed

+191
-40
lines changed

‎apps/opentelemetry_experimental/src/otel_metric_reader.erl‎

Lines changed: 40 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,8 @@
2727
-export([start_link/3,
2828
collect/1,
2929
checkpoint_generation/1,
30-
shutdown/1]).
30+
shutdown/1,
31+
collect_/4]).
3132

3233
-export([init/1,
3334
handle_call/3,
@@ -67,7 +68,7 @@ start_link(ReaderId, ProviderSup, Config) ->
6768
gen_server:start_link(?MODULE, [ReaderId, ProviderSup, Config], []).
6869

6970
collect(ReaderPid) ->
70-
ReaderPid ! collect.
71+
gen_server:call(ReaderPid, collect).
7172

7273
shutdown(ReaderPid) ->
7374
gen_server:call(ReaderPid, shutdown).
@@ -140,64 +141,63 @@ handle_continue(register_with_server, State=#state{provider_sup=ProviderSup,
140141

141142
handle_call(shutdown, _From, State) ->
142143
{reply, ok, State};
143-
handle_call(_, _From, State) ->
144-
{noreply, State}.
145-
146-
handle_cast(_, State) ->
147-
{noreply, State}.
148144

149-
%% eqwalizer:fixme get an unbound record error until the fixme for state record is resolved
150-
handle_info(collect, State=#state{exporter=undefined,
151-
export_interval_ms=ExporterIntervalMs,
152-
tref=TRef}) when TRef =/= undefined andalso
153-
ExporterIntervalMs =/= undefined ->
154-
erlang:cancel_timer(TRef, [{async, true}]),
155-
NewTRef = erlang:send_after(ExporterIntervalMs, self(), collect),
156-
{noreply, State#state{tref=NewTRef}};
157-
handle_info(collect, State=#state{id=ReaderId,
158-
exporter={ExporterModule, Config},
159-
export_interval_ms=undefined,
160-
tref=undefined,
145+
handle_call(collect, _From, State=#state{id=ReaderId,
146+
exporter=Exporter,
161147
callbacks_tab=CallbacksTab,
162148
streams_tab=StreamsTab,
163149
metrics_tab=MetricsTab,
164150
exemplars_tab=ExemplarsTab,
165151
resource=Resource,
166152
producers=Producers
167153
}) ->
154+
TRef = update_timer(State#state.tref, State#state.export_interval_ms),
155+
Reply = collect_and_export(ReaderId, Exporter, CallbacksTab, ViewAggregationTab, MetricsTab, Resource),
156+
{reply, Reply, State#state{tref=TRef}};
157+
158+
handle_call(collect, _From, State=#state{id=ReaderId,
159+
exporter={ExporterModule, Config},
160+
export_interval_ms=ExporterIntervalMs,
161+
tref=TRef,
162+
callbacks_tab=CallbacksTab,
163+
streams_tab=StreamsTab,
164+
metrics_tab=MetricsTab,
165+
exemplars_tab=ExemplarsTab,
166+
resource=Resource,
167+
producers=Producers
168+
}) ->
169+
TRef = update_timer(State#state.tref, State#state.export_interval_ms),
170+
168171
Metrics = run_collection(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId, Producers),
169172

170-
otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config),
173+
Reply = otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config),
171174

172-
{noreply, State};
173-
handle_info(collect, State=#state{id=ReaderId,
174-
exporter={ExporterModule, Config},
175-
export_interval_ms=ExporterIntervalMs,
176-
tref=TRef,
177-
callbacks_tab=CallbacksTab,
178-
streams_tab=StreamsTab,
179-
metrics_tab=MetricsTab,
180-
exemplars_tab=ExemplarsTab,
181-
resource=Resource,
182-
producers=Producers
183-
}) when TRef =/= undefined andalso
184-
ExporterIntervalMs =/= undefined ->
185-
erlang:cancel_timer(TRef, [{async, true}]),
186-
NewTRef = erlang:send_after(ExporterIntervalMs, self(), collect),
175+
{reply, Reply, State#state{tref=TRef}};
187176

188-
Metrics = run_collection(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId, Producers),
177+
%% no exporter, do nothing at all
178+
handle_call(collect, _From, State) ->
179+
{reply, ok, State};
189180

190-
otel_exporter:export_metrics(ExporterModule, Metrics, Resource, Config),
181+
handle_call(_, _From, State) ->
182+
{noreply, State}.
191183

192-
{noreply, State#state{tref=NewTRef}};
193-
%% no tref or exporter, do nothing at all
184+
handle_info(collect, State) ->
185+
{reply, _, NewState} = handle_call(collect, undefined, State),
186+
{noreply, NewState};
194187
handle_info(_, State) ->
195188
{noreply, State}.
196189

190+
handle_cast(_, State) ->
191+
{noreply, State}.
192+
197193
code_change(State) ->
198194
{ok, State}.
199195

200-
%%
196+
update_timer(undefined, undefined) ->
197+
undefined;
198+
update_timer(TRef, ExporterIntervalMs) ->
199+
erlang:cancel_timer(TRef, [{async, true}]),
200+
erlang:send_after(ExporterIntervalMs, self(), collect).
201201

202202
run_collection(CallbacksTab, StreamsTab, MetricsTab, ExemplarsTab, ReaderId, Producers) ->
203203
%% collect from view aggregations table and then export

‎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)