Merge pull request #553 from tsloughter/view-attr-filter

metric views: support filtering attributes

apps/opentelemetry_experimental/src/otel_aggregation.erl

@@ -49,19 +49,26 @@
       ViewAggregation :: #view_aggregation{},
       CollectionStartTime :: integer().
 
-maybe_init_aggregate(MetricsTab, ViewAggregation=#view_aggregation{aggregation_module=AggregationModule},
+maybe_init_aggregate(MetricsTab, ViewAggregation=#view_aggregation{aggregation_module=AggregationModule,
+                                                                   attribute_keys=AttributeKeys},
                      Value, Attributes) ->
-    case AggregationModule:aggregate(MetricsTab, ViewAggregation, Value, Attributes) of
+    FilteredAttributes = filter_attributes(AttributeKeys, Attributes),
+    case AggregationModule:aggregate(MetricsTab, ViewAggregation, Value, FilteredAttributes) of
         true ->
             ok;
         false ->
             %% entry doesn't exist, create it and rerun the aggregate function
-            Metric = AggregationModule:init(ViewAggregation, Attributes),
+            Metric = AggregationModule:init(ViewAggregation, FilteredAttributes),
             %% don't overwrite a possible concurrent measurement doing the same
             _ = ets:insert_new(MetricsTab, Metric),
-            AggregationModule:aggregate(MetricsTab, ViewAggregation, Value, Attributes)
+            AggregationModule:aggregate(MetricsTab, ViewAggregation, Value, FilteredAttributes)
     end.
 
+filter_attributes(undefined, Attributes) ->
+    Attributes;
+filter_attributes(Keys, Attributes) ->
+    maps:with(Keys, Attributes).
+
 -spec default_mapping() -> #{otel_instrument:kind() => module()}.
 default_mapping() ->
     #{?KIND_COUNTER => otel_aggregation_sum,

apps/opentelemetry_experimental/src/otel_meter_server.erl

@@ -390,7 +390,10 @@ handle_measurement(Meter, Name, Number, Attributes, ViewAggregationsTab, Metrics
 
 update_aggregations(Value, Attributes, ViewAggregations, MetricsTab) ->
     lists:foreach(fun(ViewAggregation=#view_aggregation{}) ->
-                          otel_aggregation:maybe_init_aggregate(MetricsTab, ViewAggregation, Value, Attributes);
+                          otel_aggregation:maybe_init_aggregate(MetricsTab,
+                                                                ViewAggregation,
+                                                                Value,
+                                                                Attributes);
                      (_) ->
                           ok
                   end, ViewAggregations).
@@ -400,6 +403,18 @@ per_reader_aggregations(Reader, Instrument, ViewAggregations) ->
     [view_aggregation_for_reader(Instrument, ViewAggregation, View, Reader)
      || {View, ViewAggregation} <- ViewAggregations].
 
+view_aggregation_for_reader(Instrument=#instrument{kind=Kind}, ViewAggregation, View=#view{attribute_keys=AttributeKeys},
+                            Reader=#reader{id=Id,
+                                           default_temporality_mapping=ReaderTemporalityMapping}) ->
+    AggregationModule = aggregation_module(Instrument, View, Reader),
+    Temporality = maps:get(Kind, ReaderTemporalityMapping, ?AGGREGATION_TEMPORALITY_UNSPECIFIED),
+
+    ViewAggregation#view_aggregation{
+      reader=Id,
+      attribute_keys=AttributeKeys,
+      aggregation_module=AggregationModule,
+      aggregation_options=#{},
+      temporality=Temporality};
 view_aggregation_for_reader(Instrument=#instrument{kind=Kind}, ViewAggregation, View,
                             Reader=#reader{id=Id,
                                            default_temporality_mapping=ReaderTemporalityMapping}) ->
@@ -408,10 +423,12 @@ view_aggregation_for_reader(Instrument=#instrument{kind=Kind}, ViewAggregation,
 
     ViewAggregation#view_aggregation{
       reader=Id,
+      attribute_keys=undefined,
       aggregation_module=AggregationModule,
       aggregation_options=#{},
       temporality=Temporality}.
 
+
 %% no aggregation defined for the View, so get the aggregation from the Reader
 %% the Reader's mapping of Instrument Kind to Aggregation was merged with the
 %% global default, so any missing Kind entries are filled in from the global

apps/opentelemetry_experimental/src/otel_view.erl

@@ -61,7 +61,7 @@ new(Criteria, Config) ->
     #view{name=CriteriaInstrumentName,
           instrument_matchspec=Matchspec,
           description=maps:get(description, Config, undefined),
-          attribute_keys=maps:get(attribute_keys, Config, []),
+          attribute_keys=maps:get(attribute_keys, Config, undefined),
           aggregation_module=maps:get(aggregation_module, Config, undefined),
           aggregation_options=maps:get(aggregation_options, Config, #{})}.
 
@@ -84,6 +84,7 @@ match_instrument_to_views(Instrument=#instrument{name=InstrumentName,
     Scope = otel_meter:scope(Meter),
     case lists:filtermap(fun(View=#view{name=ViewName,
                                         description=ViewDescription,
+                                        attribute_keys=AttributeKeys,
                                         aggregation_options=AggregationOptions,
                                         instrument_matchspec=Matchspec}) ->
                                  case ets:match_spec_run([Instrument], Matchspec) of
@@ -96,6 +97,7 @@ match_instrument_to_views(Instrument=#instrument{name=InstrumentName,
                                                                          instrument=Instrument,
                                                                          temporality=Temporality,
                                                                          is_monotonic=IsMonotonic,
+                                                                         attribute_keys=AttributeKeys,
                                                                          aggregation_options=AggregationOptions,
                                                                          description=value_or(ViewDescription,
                                                                                               Description)
@@ -108,6 +110,7 @@ match_instrument_to_views(Instrument=#instrument{name=InstrumentName,
                                            instrument=Instrument,
                                            temporality=Temporality,
                                            is_monotonic=IsMonotonic,
+                                           attribute_keys=undefined,
                                            aggregation_options=#{},
                                            description=Description}}];
         Aggs ->

apps/opentelemetry_experimental/src/otel_view.hrl

@@ -2,7 +2,7 @@
         {name                    :: otel_instrument:name(),
          instrument_matchspec    :: ets:compiled_match_spec(),
          description             :: unicode:unicode_binary() | undefined,
-         attribute_keys          :: [opentelemetry:attribute_key()],
+         attribute_keys          :: [opentelemetry:attribute_key()] | undefined,
          aggregation_module      :: module(),
          aggregation_options=#{} :: map()}).
 
@@ -13,6 +13,8 @@
          instrument :: otel_instrument:t(),
          reader :: reference() | undefined,
 
+         attribute_keys :: [opentelemetry:attribute_key()] | undefined,
+
          aggregation_module :: module(),
          aggregation_options :: map(),
 

apps/opentelemetry_experimental/test/otel_metrics_SUITE.erl

@@ -81,7 +81,8 @@ all() ->
     [default_view, provider_test, view_creation_test, counter_add, multiple_readers,
      explicit_histograms, delta_explicit_histograms, delta_counter, cumulative_counter,
      kill_reader, kill_server, observable_counter, observable_updown_counter, observable_gauge,
-     multi_instrument_callback, using_macros, float_counter, float_updown_counter, float_histogram].
+     multi_instrument_callback, using_macros, float_counter, float_updown_counter, float_histogram,
+     sync_filtered_attributes, async_filtered_attributes].
 
 init_per_suite(Config) ->
     application:load(opentelemetry_experimental),
@@ -922,3 +923,76 @@ multi_instrument_callback(_Config) ->
     ?assertLastValueReceive(GaugeName, GaugeDesc, Unit, [{5, #{<<"a">> => <<"b">>}}]),
 
     ok.
+
+sync_filtered_attributes(_Config) ->
+    Meter = opentelemetry_experimental:get_meter(?MODULE),
+
+    CounterName = a_counter,
+    CounterDesc = <<"counter description">>,
+    CounterUnit = kb,
+
+    Counter = otel_counter:create(Meter, CounterName,
+                                  #{description => CounterDesc,
+                                    unit => CounterUnit}),
+    ?assertMatch(#instrument{meter = {otel_meter_default, _},
+                             module = otel_meter_default,
+                             name = CounterName,
+                             description = CounterDesc,
+                             kind = counter,
+                             unit = CounterUnit}, Counter),
+
+    ?assert(otel_meter_server:add_view(view_a, #{instrument_name => CounterName},
+                                       #{aggregation_module => otel_aggregation_sum,
+                                         attribute_keys => [a, b]})),
+
+    ?assertEqual(ok, otel_counter:add(Counter, 2, #{a => 1, b => 2, c => 3})),
+    ?assertEqual(ok, otel_counter:add(Counter, 5, #{a => 1, b => 2})),
+    ?assertEqual(ok, ?counter_add(CounterName, 5, #{a => 1, b => 2, c => 3})),
+
+    otel_meter_server:force_flush(),
+
+    ?assertSumReceive(CounterName, CounterDesc, kb,
+                      [{7, #{a => 1, b => 2, c => 3}}, {5, #{a => 1, b => 2}}]),
+    ?assertSumReceive(view_a, CounterDesc, kb,
+                      [{12, #{a => 1, b => 2}}]),
+
+
+    ok.
+
+async_filtered_attributes(_Config) ->
+    DefaultMeter = otel_meter_default,
+
+    Meter = opentelemetry_experimental:get_meter(),
+    ?assertMatch({DefaultMeter, _}, Meter),
+
+    CounterName = a_observable_counter,
+    CounterDesc = <<"observable counter description">>,
+    CounterUnit = kb,
+
+    ?assert(otel_meter_server:add_view(#{instrument_name => CounterName},
+                                       #{aggregation_module => otel_aggregation_sum,
+                                         attribute_keys => [a]})),
+
+    Counter = otel_meter:create_observable_counter(Meter, CounterName,
+                                                   fun(_Args) ->
+                                                           MeasurementAttributes = #{a => b,
+                                                                                     c => d},
+                                                           {4, MeasurementAttributes}
+                                                   end,
+                                                   [],
+                                                   #{description => CounterDesc,
+                                                     unit => CounterUnit}),
+
+    ?assertMatch(#instrument{meter = {DefaultMeter,_},
+                             module = DefaultMeter,
+                             name = CounterName,
+                             description = CounterDesc,
+                             kind = observable_counter,
+                             unit = CounterUnit,
+                             callback=_}, Counter),
+
+    otel_meter_server:force_flush(),
+
+    ?assertSumReceive(CounterName, <<"observable counter description">>, kb, [{4, #{a => b}}]),
+
+    ok.