Fix attribute filtering for synchronous instruments. (#2472)

sdk/include/opentelemetry/sdk/metrics/data/exemplar_data.h

@@ -6,8 +6,8 @@
 #include <memory>
 
 #include "opentelemetry/common/timestamp.h"
-#include "opentelemetry/sdk/common/attribute_utils.h"
 #include "opentelemetry/sdk/metrics/data/metric_data.h"
+#include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
 #include "opentelemetry/trace/span_context.h"
 #include "opentelemetry/version.h"
 
@@ -16,7 +16,7 @@ namespace sdk
 {
 namespace metrics
 {
-using MetricAttributes = opentelemetry::sdk::common::OrderedAttributeMap;
+using MetricAttributes = opentelemetry::sdk::metrics::FilteredOrderedAttributeMap;
 /**
  * A sample input measurement.
  *

sdk/include/opentelemetry/sdk/metrics/exemplar/filter.h

@@ -5,7 +5,7 @@
 
 #include <memory>
 
-#include "opentelemetry/sdk/common/attribute_utils.h"
+#include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -23,7 +23,7 @@ class OrderedAttributeMap;
 
 namespace metrics
 {
-using MetricAttributes = opentelemetry::sdk::common::OrderedAttributeMap;
+using MetricAttributes = opentelemetry::sdk::metrics::FilteredOrderedAttributeMap;
 
 /**
  * Exemplar filters are used to pre-filter measurements before attempting to store them in a

sdk/include/opentelemetry/sdk/metrics/instruments.h

@@ -6,7 +6,7 @@
 #include <functional>
 #include <string>
 
-#include "opentelemetry/sdk/common/attribute_utils.h"
+#include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
@@ -63,7 +63,7 @@ struct InstrumentDescriptor
   InstrumentValueType value_type_;
 };
 
-using MetricAttributes               = opentelemetry::sdk::common::OrderedAttributeMap;
+using MetricAttributes               = opentelemetry::sdk::metrics::FilteredOrderedAttributeMap;
 using AggregationTemporalitySelector = std::function<AggregationTemporality(InstrumentType)>;
 
 /*class InstrumentSelector {

sdk/include/opentelemetry/sdk/metrics/observer_result.h

@@ -30,15 +30,7 @@ class ObserverResultT final : public opentelemetry::metrics::ObserverResultT<T>
 
   void Observe(T value, const opentelemetry::common::KeyValueIterable &attributes) noexcept override
   {
-    if (attributes_processor_)
-    {
-      auto attr = attributes_processor_->process(attributes);
-      data_.insert({attr, value});
-    }
-    else
-    {
-      data_.insert({MetricAttributes{attributes}, value});
-    }
+    data_.insert({MetricAttributes{attributes, attributes_processor_}, value});
   }
 
   const std::unordered_map<MetricAttributes, T, AttributeHashGenerator> &GetMeasurements()

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

@@ -69,6 +69,7 @@ class AttributesHashMap
    * value and store in the hash
    */
   Aggregation *GetOrSetDefault(const opentelemetry::common::KeyValueIterable &attributes,
+                               const AttributesProcessor *attributes_processor,
                                std::function<std::unique_ptr<Aggregation>()> aggregation_callback,
                                size_t hash)
   {
@@ -83,7 +84,7 @@ class AttributesHashMap
       return GetOrSetOveflowAttributes(aggregation_callback);
     }
 
-    MetricAttributes attr{attributes};
+    MetricAttributes attr{attributes, attributes_processor};
 
     hash_map_[hash] = {attr, aggregation_callback()};
     return hash_map_[hash].second.get();
@@ -133,6 +134,7 @@ class AttributesHashMap
    * Set the value for given key, overwriting the value if already present
    */
   void Set(const opentelemetry::common::KeyValueIterable &attributes,
+           const AttributesProcessor *attributes_processor,
            std::unique_ptr<Aggregation> aggr,
            size_t hash)
   {
@@ -149,7 +151,7 @@ class AttributesHashMap
     }
     else
     {
-      MetricAttributes attr{attributes};
+      MetricAttributes attr{attributes, attributes_processor};
       hash_map_[hash] = {attr, std::move(aggr)};
     }
   }
@@ -169,8 +171,7 @@ class AttributesHashMap
     }
     else
     {
-      MetricAttributes attr{attributes};
-      hash_map_[hash] = {attr, std::move(aggr)};
+      hash_map_[hash] = {attributes, std::move(aggr)};
     }
   }
 

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

@@ -0,0 +1,38 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#pragma once
+
+#include <map>
+#include <string>
+#include "opentelemetry/sdk/common/attributemap_hash.h"
+#include "opentelemetry/version.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+class AttributesProcessor;
+class FilteredOrderedAttributeMap : public opentelemetry::sdk::common::OrderedAttributeMap
+{
+public:
+  FilteredOrderedAttributeMap() = default;
+  FilteredOrderedAttributeMap(
+      std::initializer_list<std::pair<nostd::string_view, opentelemetry::common::AttributeValue>>
+          attributes)
+      : OrderedAttributeMap(attributes)
+  {}
+  FilteredOrderedAttributeMap(const opentelemetry::common::KeyValueIterable &attributes)
+      : FilteredOrderedAttributeMap(attributes, nullptr)
+  {}
+  FilteredOrderedAttributeMap(const opentelemetry::common::KeyValueIterable &attributes,
+                              const opentelemetry::sdk::metrics::AttributesProcessor *processor);
+  FilteredOrderedAttributeMap(
+      std::initializer_list<std::pair<nostd::string_view, opentelemetry::common::AttributeValue>>
+          attributes,
+      const opentelemetry::sdk::metrics::AttributesProcessor *processor);
+};
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE

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

@@ -99,7 +99,8 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
         });
 
     std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
-    attributes_hashmap_->GetOrSetDefault(attributes, create_default_aggregation_, hash)
+    attributes_hashmap_
+        ->GetOrSetDefault(attributes, attributes_processor_, create_default_aggregation_, hash)
         ->Aggregate(value);
   }
 
@@ -148,7 +149,8 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
           }
         });
     std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
-    attributes_hashmap_->GetOrSetDefault(attributes, create_default_aggregation_, hash)
+    attributes_hashmap_
+        ->GetOrSetDefault(attributes, attributes_processor_, create_default_aggregation_, hash)
         ->Aggregate(value);
   }
 

sdk/include/opentelemetry/sdk/metrics/view/attributes_processor.h

@@ -5,19 +5,18 @@
 
 #include <string>
 #include <unordered_map>
-
 #include "opentelemetry/common/key_value_iterable.h"
 #include "opentelemetry/nostd/string_view.h"
-#include "opentelemetry/sdk/common/attribute_utils.h"
 #include "opentelemetry/sdk/metrics/instruments.h"
+#include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
 #include "opentelemetry/version.h"
 
 OPENTELEMETRY_BEGIN_NAMESPACE
 namespace sdk
 {
 namespace metrics
 {
-using MetricAttributes = opentelemetry::sdk::common::OrderedAttributeMap;
+using MetricAttributes = opentelemetry::sdk::metrics::FilteredOrderedAttributeMap;
 
 /**
  * The AttributesProcessor is responsible for customizing which

sdk/src/metrics/CMakeLists.txt

@@ -13,6 +13,7 @@ add_library(
   instrument_metadata_validator.cc
   export/periodic_exporting_metric_reader.cc
   export/periodic_exporting_metric_reader_factory.cc
+  state/filtered_ordered_attribute_map.cc
   state/metric_collector.cc
   state/observable_registry.cc
   state/sync_metric_storage.cc

sdk/src/metrics/state/filtered_ordered_attribute_map.cc

@@ -0,0 +1,43 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+#include "opentelemetry/sdk/metrics/state/filtered_ordered_attribute_map.h"
+#include "opentelemetry/sdk/metrics/view/attributes_processor.h"
+
+OPENTELEMETRY_BEGIN_NAMESPACE
+namespace sdk
+{
+namespace metrics
+{
+FilteredOrderedAttributeMap::FilteredOrderedAttributeMap(
+    const opentelemetry::common::KeyValueIterable &attributes,
+    const AttributesProcessor *processor)
+    : OrderedAttributeMap()
+{
+  attributes.ForEachKeyValue(
+      [&](nostd::string_view key, opentelemetry::common::AttributeValue value) noexcept {
+        if (processor && processor->isPresent(key))
+        {
+          SetAttribute(key, value);
+        }
+        return true;
+      });
+}
+
+FilteredOrderedAttributeMap::FilteredOrderedAttributeMap(
+    std::initializer_list<std::pair<nostd::string_view, opentelemetry::common::AttributeValue>>
+        attributes,
+    const AttributesProcessor *processor)
+    : OrderedAttributeMap()
+{
+  for (auto &kv : attributes)
+  {
+    if (processor && processor->isPresent(kv.first))
+    {
+      SetAttribute(kv.first, kv.second);
+    }
+  }
+}
+}  // namespace metrics
+}  // namespace sdk
+OPENTELEMETRY_END_NAMESPACE

sdk/test/metrics/cardinality_limit_test.cc

@@ -26,8 +26,8 @@ TEST(CardinalityLimit, AttributesHashMapBasicTests)
   long record_value = 100;
   for (auto i = 0; i < 10; i++)
   {
-    OrderedAttributeMap attributes = {{"key", std::to_string(i)}};
-    auto hash                      = opentelemetry::sdk::common::GetHashForAttributeMap(attributes);
+    FilteredOrderedAttributeMap attributes = {{"key", std::to_string(i)}};
+    auto hash = opentelemetry::sdk::common::GetHashForAttributeMap(attributes);
     static_cast<LongSumAggregation *>(
         hash_map.GetOrSetDefault(attributes, aggregation_callback, hash))
         ->Aggregate(record_value);
@@ -37,16 +37,16 @@ TEST(CardinalityLimit, AttributesHashMapBasicTests)
   // overflowmetric point.
   for (auto i = 10; i < 15; i++)
   {
-    OrderedAttributeMap attributes = {{"key", std::to_string(i)}};
-    auto hash                      = opentelemetry::sdk::common::GetHashForAttributeMap(attributes);
+    FilteredOrderedAttributeMap attributes = {{"key", std::to_string(i)}};
+    auto hash = opentelemetry::sdk::common::GetHashForAttributeMap(attributes);
     static_cast<LongSumAggregation *>(
         hash_map.GetOrSetDefault(attributes, aggregation_callback, hash))
         ->Aggregate(record_value);
   }
   EXPECT_EQ(hash_map.Size(), 10);  // only one more metric point should be added as overflow.
   // get the overflow metric point
   auto agg = hash_map.GetOrSetDefault(
-      OrderedAttributeMap({{kAttributesLimitOverflowKey, kAttributesLimitOverflowValue}}),
+      FilteredOrderedAttributeMap({{kAttributesLimitOverflowKey, kAttributesLimitOverflowValue}}),
       aggregation_callback, kOverflowAttributesHash);
   EXPECT_NE(agg, nullptr);
   auto sum_agg = static_cast<LongSumAggregation *>(agg);