-module(oc_stat_view). -export([register/5, deregister/1, subscribe/1, subscribe/5, batch_subscribe/1, subscribed/1, unsubscribe/1, registered/1]). -export([measure_views/1, subscribed/0, add_sample/3, export/1]). -export_types([name/0, description/0, view_data/0]). -export(['__init_backend__'/0]). -include("opencensus.hrl"). -define(NAME_POS, 2). -define(SUBSCRIBED_POS, 3). -type name() :: atom() | binary() | string(). -type description() :: binary() | string(). -type view_data() :: #{name := name(), description := description(), ctags := oc_tags:tags(), data := oc_stat_aggregation:data()}. -spec register(name(), description(), oc_tags:tags(), measure_name(), aggregation()) -> ok. register(Name, Description, Tags, Measure, Aggregation) -> %% TODO: check Measure exists? register(Name, Description, Tags, Measure, Aggregation, false). -spec deregister(name()) -> ok. deregister(Name) -> ets:delete(?MODULE, Name), ok. -spec subscribe(map() | name()) -> ok. subscribe(#{name := Name, description := Description, tags := Tags, measure := Measure, aggregation := Aggregation}) -> subscribe(Name, Description, Tags, Measure, Aggregation); subscribe(Name) -> ets:update_element(?MODULE, Name, {?SUBSCRIBED_POS, true}), ok. -spec subscribe(name(), description(), oc_tags:tags(), measure_name(), aggregation()) -> ok. subscribe(Name, Description, Tags, Measure, Aggregation) -> %% TODO: check Measure exists? register(Name, Description, Tags, Measure, Aggregation, true). %% @doc %% Subscribe many `Views' at once. %% @end -spec batch_subscribe(list(name() | map())) -> ok. batch_subscribe(Views) when is_list(Views) -> [ok = subscribe(View) || View <- Views], ok. -spec register(name(), description(), oc_tags:tags(), measure_name(), aggregation(), boolean()) -> ok | no_return(). register(Name, Description, Tags, Measure, Aggregation, Subscribed) -> NAggregation = normalize_aggregation(Aggregation), NTags = normalize_tags(Tags), case ets:match_object(?MODULE, {Measure, Name, '_', '_', '_', '_'}) of [{Measure, Name, '_', Description, NTags, NAggregation}] -> ets:update_element(?MODULE, Name, {?SUBSCRIBED_POS, Subscribed}); [_] -> erlang:error({already_exists, Name}); _ -> {AggregationModule, AggregationOptions} = NAggregation, {_, Keys} = NTags, %% TODO: transaction? %% HACK: Keys reversed because tag_value reverses NAggregationOptions = AggregationModule:init(Name, lists:reverse(Keys), AggregationOptions), ets:insert(?MODULE, {Measure, Name, Subscribed, Description, NTags, {AggregationModule, NAggregationOptions}}) end, ok. -spec unsubscribe(name()) -> ok. unsubscribe(Name) -> ets:update_element(?MODULE, Name, {?SUBSCRIBED_POS, false}), ok. %% @doc %% Checks whether a view `Name' is registered. %% @end -spec registered(name()) -> boolean(). registered(Name) -> ets:lookup_element(?MODULE, Name, ?NAME_POS) =/= []. measure_views(Measure) -> ets:lookup(?MODULE, Measure). subscribed({_Measure, _Name, Subscribed, _Description, _Tags, _Aggregation}) -> Subscribed. subscribed() -> ets:match_object(?MODULE, {'_', '_', true, '_', '_', '_'}). add_sample({_Measure, Name, _Subscribed, _Description, ViewTags, Aggregation}, ContextTags, Value) -> {_, Keys} = ViewTags, TagValues = tag_values(ContextTags, Keys), {AggregationModule, AggregationOptions} = Aggregation, AggregationModule:add_sample(Name, TagValues, Value, AggregationOptions). -spec export(tuple()) -> view_data(). export({_Measure, Name, _, Description, ViewTags, Aggregation}) -> {AggregationModule, AggregationOptions} = Aggregation, {CTags, _Keys} = ViewTags, #{name => Name, description => Description, ctags => CTags, data => AggregationModule:export(Name, AggregationOptions)}. normalize_aggregation({Module, Options}) -> {Module, Options}; normalize_aggregation(Module) when is_atom(Module) -> {Module, []}. normalize_tags([]) -> {#{}, []}; normalize_tags(Tags) -> normalize_tags(Tags, {#{}, []}). normalize_tags([], {Map, List}) -> {Map, lists:reverse(List)}; normalize_tags([First|Rest], {Map, List}) when is_map(First) -> normalize_tags(Rest, {maps:merge(Map, First), List}); normalize_tags([First|Rest], {Map, List}) when is_atom(First) -> normalize_tags(Rest, {Map, [First | List]}). tag_values(Tags, Keys) -> lists:foldl(fun(Key, Acc) -> [maps:get(Key, Tags, undefined) | Acc] end, [], Keys). '__init_backend__'() -> ?MODULE = ets:new(?MODULE, [bag, named_table, public, {read_concurrency, true}]), ok.