otel-metricfilter-fn

CHANGELOG.md

@@ -21,6 +21,9 @@ Increment the:
 * [SDK] Fix instrumentation scope attributes evaluated in equal method
   [#3214](https://github.com/open-telemetry/opentelemetry-cpp/pull/3214)
 
+* [SDK] Implement spec: MetricFilter
+  [#3235](https://github.com/open-telemetry/opentelemetry-cpp/pull/3235)
+
 * [EXPORTER] Fix scope attributes missing from otlp traces metrics
   [#3185](https://github.com/open-telemetry/opentelemetry-cpp/pull/3185)
 

sdk/include/opentelemetry/sdk/metrics/export/metric_filter.h

@@ -0,0 +1,112 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <functional>
+#include <memory>
+
+#include "opentelemetry/nostd/string_view.h"
+#include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h"
+#include "opentelemetry/sdk/metrics/data/metric_data.h"
+#include "opentelemetry/sdk/metrics/instruments.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+
+/**
+ * MetricFilter defines the interface which enables the MetricReader’s
+ * registered MetricProducers or the SDK’s MetricProducer to filter aggregated
+ * data points (Metric Points) inside its Produce operation. The filtering is
+ * done at the MetricProducer for performance reasons.
+ *
+ * The MetricFilter allows filtering an entire metric stream - dropping or
+ * allowing all its attribute sets - by its TestMetric operation, which accepts
+ * the metric stream information (scope, name, kind and unit) and returns an
+ * enumeration: kAccept, kDrop or kAcceptPartial. If the latter returned, the
+ * TestAttributes operation is to be called per attribute set of that metric
+ * stream, returning an enumeration determining if the data point for that
+ * (metric stream, attributes) pair is to be allowed in the result of the
+ * MetricProducer Produce operation.
+ */
+class MetricFilter
+{
+public:
+  enum class MetricFilterResult
+  {
+    kAccept,
+    kDrop,
+    kAcceptPartial,
+  };
+
+  enum class AttributesFilterResult
+  {
+    kAccept,
+    kDrop,
+  };
+
+  using TestMetricFn = std::function<MetricFilterResult(
+      const opentelemetry::sdk::instrumentationscope::InstrumentationScope &scope,
+      opentelemetry::nostd::string_view name,
+      const InstrumentType &type,
+      opentelemetry::nostd::string_view unit)>;
+
+  using TestAttributesFn = std::function<AttributesFilterResult(
+      const opentelemetry::sdk::instrumentationscope::InstrumentationScope &scope,
+      opentelemetry::nostd::string_view name,
+      const InstrumentType &type,
+      opentelemetry::nostd::string_view unit,
+      const PointAttributes &attributes)>;
+
+  // static
+  static std::unique_ptr<MetricFilter> Create(TestMetricFn test_metric_fn,
+                                              TestAttributesFn test_attributes_fn)
+  {
+    return std::make_unique<MetricFilter>(test_metric_fn, test_attributes_fn);
+  }
+
+  MetricFilter(TestMetricFn test_metric_fn, TestAttributesFn test_attributes_fn)
+      : test_metric_fn_(test_metric_fn), test_attributes_fn_(test_attributes_fn)
+  {}
+
+  /**
+   * TestMetric is called once for every metric stream, in each MetricProducer
+   * Produce operation.
+   */
+  MetricFilterResult TestMetric(
+      const opentelemetry::sdk::instrumentationscope::InstrumentationScope &scope,
+      opentelemetry::nostd::string_view name,
+      const InstrumentType &type,
+      opentelemetry::nostd::string_view unit)
+  {
+    return test_metric_fn_(scope, name, type, unit);
+  }
+
+  /**
+   * TestAttributes determines for a given metric stream and attribute set if
+   * it should be allowed or filtered out.
+   *
+   * This operation should only be called if TestMetric operation returned
+   * kAcceptPartial for the given metric stream arguments.
+   */
+  AttributesFilterResult TestAttributes(
+      const opentelemetry::sdk::instrumentationscope::InstrumentationScope &scope,
+      opentelemetry::nostd::string_view name,
+      const InstrumentType &type,
+      opentelemetry::nostd::string_view unit,
+      const PointAttributes &attributes)
+  {
+    return test_attributes_fn_(scope, name, type, unit, attributes);
+  }
+
+private:
+  TestMetricFn test_metric_fn_;
+  TestAttributesFn test_attributes_fn_;
+};
+
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE

sdk/include/opentelemetry/sdk/metrics/export/metric_producer.h

@@ -3,12 +3,14 @@
 
 #pragma once
 
+#include <memory>
 #include <utility>
 #include <vector>
 
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/nostd/variant.h"
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
+#include "opentelemetry/sdk/metrics/export/metric_filter.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -80,7 +82,9 @@ struct ResourceMetrics
 class MetricProducer
 {
 public:
-  MetricProducer()          = default;
+  MetricProducer(std::unique_ptr<MetricFilter> metric_filter = nullptr)
+      : metric_filter_(std::move(metric_filter))
+  {}
   virtual ~MetricProducer() = default;
 
   MetricProducer(const MetricProducer &)  = delete;
@@ -107,6 +111,8 @@ class MetricProducer
    * partial failure.
    */
   virtual Result Produce() noexcept = 0;
+
+  std::unique_ptr<MetricFilter> metric_filter_;
 };
 
 }  // namespace metrics

sdk/include/opentelemetry/sdk/metrics/meter_context.h

@@ -13,6 +13,7 @@
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/nostd/span.h"
 #include "opentelemetry/nostd/string_view.h"
+#include "opentelemetry/sdk/metrics/export/metric_filter.h"
 #include "opentelemetry/sdk/metrics/metric_reader.h"
 #include "opentelemetry/sdk/metrics/state/metric_collector.h"
 #include "opentelemetry/sdk/metrics/view/instrument_selector.h"
@@ -96,14 +97,20 @@ class MeterContext : public std::enable_shared_from_this<MeterContext>
   opentelemetry::common::SystemTimestamp GetSDKStartTime() noexcept;
 
   /**
-   * Attaches a metric reader to list of configured readers for this Meter context.
-   * @param reader The metric reader for this meter context. This
-   * must not be a nullptr.
+   * Create a MetricCollector from a MetricReader using this MeterContext and add it to the list of
+   * configured collectors.
+   * @param reader The MetricReader for which a MetricCollector is to be created. This must not be a
+   * nullptr.
+   * @param metric_filter The optional MetricFilter used when creating the MetricCollector.
+   * @return The MetricCollector created.
    *
    * Note: This reader may not receive any in-flight meter data, but will get newly created meter
-   * data. Note: This method is not thread safe, and should ideally be called from main thread.
+   * data.
+   * Note: This method is not thread safe, and should ideally be called from main thread.
    */
-  void AddMetricReader(std::shared_ptr<MetricReader> reader) noexcept;
+  std::shared_ptr<MetricCollector> AddMetricReader(
+      std::shared_ptr<MetricReader> reader,
+      std::unique_ptr<MetricFilter> metric_filter = nullptr) noexcept;
 
   /**
    * Attaches a View to list of configured Views for this Meter context.

sdk/include/opentelemetry/sdk/metrics/meter_provider.h

@@ -11,6 +11,7 @@
 #include "opentelemetry/metrics/meter_provider.h"
 #include "opentelemetry/nostd/shared_ptr.h"
 #include "opentelemetry/nostd/string_view.h"
+#include "opentelemetry/sdk/metrics/export/metric_filter.h"
 #include "opentelemetry/sdk/metrics/meter_context.h"
 #include "opentelemetry/sdk/metrics/metric_reader.h"
 #include "opentelemetry/sdk/metrics/view/instrument_selector.h"
@@ -79,14 +80,20 @@ class OPENTELEMETRY_EXPORT MeterProvider final : public opentelemetry::metrics::
   const sdk::resource::Resource &GetResource() const noexcept;
 
   /**
-   * Attaches a metric reader to list of configured readers for this Meter providers.
-   * @param reader The metric reader for this meter provider. This
-   * must not be a nullptr.
+   * Create a MetricCollector from a MetricReader using the MeterContext of this MeterProvider and
+   * add it to the list of configured collectors.
+   * @param reader The MetricReader for which a MetricCollector is to be created. This must not be a
+   * nullptr.
+   * @param metric_filter The optional MetricFilter used when creating the MetricCollector.
+   * @return The MetricCollector created.
    *
    * Note: This reader may not receive any in-flight meter data, but will get newly created meter
-   * data. Note: This method is not thread safe, and should ideally be called from main thread.
+   * data.
+   * Note: This method is not thread safe, and should ideally be called from main thread.
    */
-  void AddMetricReader(std::shared_ptr<MetricReader> reader) noexcept;
+  std::shared_ptr<MetricCollector> AddMetricReader(
+      std::shared_ptr<MetricReader> reader,
+      std::unique_ptr<MetricFilter> metric_filter = nullptr) noexcept;
 
   /**
    * Attaches a View to list of configured Views for this Meter provider.

sdk/include/opentelemetry/sdk/metrics/state/metric_collector.h

@@ -7,6 +7,7 @@
 #include <memory>
 
 #include "opentelemetry/nostd/function_ref.h"
+#include "opentelemetry/sdk/metrics/export/metric_filter.h"
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
 #include "opentelemetry/sdk/metrics/metric_reader.h"
@@ -40,7 +41,9 @@ class CollectorHandle
 class MetricCollector : public MetricProducer, public CollectorHandle
 {
 public:
-  MetricCollector(MeterContext *context, std::shared_ptr<MetricReader> metric_reader);
+  MetricCollector(MeterContext *context,
+                  std::shared_ptr<MetricReader> metric_reader,
+                  std::unique_ptr<MetricFilter> metric_filter = nullptr);
 
   ~MetricCollector() override = default;
 
@@ -62,6 +65,7 @@ class MetricCollector : public MetricProducer, public CollectorHandle
 private:
   MeterContext *meter_context_;
   std::shared_ptr<MetricReader> metric_reader_;
+  std::unique_ptr<MetricFilter> metric_filter_;
 };
 }  // namespace metrics
 }  // namespace sdk

sdk/src/metrics/meter_context.cc

@@ -79,10 +79,14 @@ opentelemetry::common::SystemTimestamp MeterContext::GetSDKStartTime() noexcept
   return sdk_start_ts_;
 }
 
-void MeterContext::AddMetricReader(std::shared_ptr<MetricReader> reader) noexcept
+std::shared_ptr<MetricCollector> MeterContext::AddMetricReader(
+    std::shared_ptr<MetricReader> reader,
+    std::unique_ptr<MetricFilter> metric_filter) noexcept
 {
-  auto collector = std::shared_ptr<MetricCollector>{new MetricCollector(this, std::move(reader))};
+  auto collector = std::shared_ptr<MetricCollector>{
+      new MetricCollector(this, std::move(reader), std::move(metric_filter))};
   collectors_.push_back(collector);
+  return collector;
 }
 
 void MeterContext::AddView(std::unique_ptr<InstrumentSelector> instrument_selector,

sdk/src/metrics/meter_provider.cc

@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0
 
 #include <chrono>
+#include <memory>
 #include <mutex>
 #include <utility>
 
@@ -12,6 +13,8 @@
 #include "opentelemetry/nostd/string_view.h"
 #include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h"
+#include "opentelemetry/sdk/metrics/export/metric_filter.h"
+#include "opentelemetry/sdk/metrics/export/metric_producer.h"
 #include "opentelemetry/sdk/metrics/meter.h"
 #include "opentelemetry/sdk/metrics/meter_context.h"
 #include "opentelemetry/sdk/metrics/meter_provider.h"
@@ -107,9 +110,11 @@ const resource::Resource &MeterProvider::GetResource() const noexcept
   return context_->GetResource();
 }
 
-void MeterProvider::AddMetricReader(std::shared_ptr<MetricReader> reader) noexcept
+std::shared_ptr<MetricCollector> MeterProvider::AddMetricReader(
+    std::shared_ptr<MetricReader> reader,
+    std::unique_ptr<MetricFilter> metric_filter) noexcept
 {
-  context_->AddMetricReader(std::move(reader));
+  return context_->AddMetricReader(std::move(reader), std::move(metric_filter));
 }
 
 void MeterProvider::AddView(std::unique_ptr<InstrumentSelector> instrument_selector,

sdk/src/metrics/state/metric_collector.cc

@@ -11,6 +11,7 @@
 #include "opentelemetry/nostd/function_ref.h"
 #include "opentelemetry/sdk/common/global_log_handler.h"
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
+#include "opentelemetry/sdk/metrics/export/metric_filter.h"
 #include "opentelemetry/sdk/metrics/export/metric_producer.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
 #include "opentelemetry/sdk/metrics/meter.h"
@@ -28,8 +29,11 @@ namespace metrics
 using opentelemetry::sdk::resource::Resource;
 
 MetricCollector::MetricCollector(opentelemetry::sdk::metrics::MeterContext *context,
-                                 std::shared_ptr<MetricReader> metric_reader)
-    : meter_context_{context}, metric_reader_{std::move(metric_reader)}
+                                 std::shared_ptr<MetricReader> metric_reader,
+                                 std::unique_ptr<MetricFilter> metric_filter)
+    : meter_context_{context},
+      metric_reader_{std::move(metric_reader)},
+      metric_filter_(std::move(metric_filter))
 {
   metric_reader_->SetMetricProducer(this);
 }
@@ -63,12 +67,61 @@ MetricProducer::Result MetricCollector::Produce() noexcept
   meter_context_->ForEachMeter([&](const std::shared_ptr<Meter> &meter) noexcept {
     auto collection_ts = std::chrono::system_clock::now();
     auto metric_data   = meter->Collect(this, collection_ts);
-    if (!metric_data.empty())
+    if (metric_data.empty())
+    {
+      return true;
+    }
+    ScopeMetrics scope_metrics;
+    scope_metrics.metric_data_ = std::move(metric_data);
+    scope_metrics.scope_       = meter->GetInstrumentationScope();
+    if (!this->metric_filter_)
     {
-      ScopeMetrics scope_metrics;
-      scope_metrics.metric_data_ = std::move(metric_data);
-      scope_metrics.scope_       = meter->GetInstrumentationScope();
       resource_metrics.scope_metric_data_.emplace_back(std::move(scope_metrics));
+      return true;
+    }
+
+    ScopeMetrics filtered_scope_metrics;
+    filtered_scope_metrics.scope_ = meter->GetInstrumentationScope();
+    for (MetricData &metric : scope_metrics.metric_data_)
+    {
+      const opentelemetry::sdk::instrumentationscope::InstrumentationScope &scope =
+          *scope_metrics.scope_;
+      opentelemetry::nostd::string_view name = metric.instrument_descriptor.name_;
+      const InstrumentType &type             = metric.instrument_descriptor.type_;
+      opentelemetry::nostd::string_view unit = metric.instrument_descriptor.unit_;
+
+      MetricFilter::MetricFilterResult metric_filter_result =
+          this->metric_filter_->TestMetric(scope, name, type, unit);
+      if (metric_filter_result == MetricFilter::MetricFilterResult::kAccept)
+      {
+        filtered_scope_metrics.metric_data_.emplace_back(std::move(metric));
+        continue;
+      }
+      else if (metric_filter_result == MetricFilter::MetricFilterResult::kDrop)
+      {
+        continue;
+      }
+
+      std::vector<PointDataAttributes> filtered_point_data_attrs;
+      for (PointDataAttributes &point_data_attr : metric.point_data_attr_)
+      {
+        const PointAttributes &attributes = point_data_attr.attributes;
+        MetricFilter::AttributesFilterResult attributes_filter_result =
+            this->metric_filter_->TestAttributes(scope, name, type, unit, attributes);
+        if (attributes_filter_result == MetricFilter::AttributesFilterResult::kAccept)
+        {
+          filtered_point_data_attrs.emplace_back(std::move(point_data_attr));
+        }
+      }
+      if (!filtered_point_data_attrs.empty())
+      {
+        metric.point_data_attr_ = std::move(filtered_point_data_attrs);
+        filtered_scope_metrics.metric_data_.emplace_back(std::move(metric));
+      }
+    }
+    if (!filtered_scope_metrics.metric_data_.empty())
+    {
+      resource_metrics.scope_metric_data_.emplace_back(std::move(filtered_scope_metrics));
     }
     return true;
   });

sdk/test/metrics/CMakeLists.txt

@@ -27,6 +27,7 @@ foreach(
   observer_result_test
   sync_instruments_test
   async_instruments_test
+  metric_collector_test
   metric_reader_test
   observable_registry_test
   periodic_exporting_metric_reader_test