Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,10 @@ Increment the:
* [CODE HEALTH] Fix remaining misc-override-with-different-visibility warnings
[#4280](https://github.com/open-telemetry/opentelemetry-cpp/pull/4280)

* [SDK] Complete exemplar filtering: the exemplar filter(`AlwaysOn`/
`AlwaysOff`/`TraceBased`)
[#4267](https://github.com/open-telemetry/opentelemetry-cpp/pull/4267)

Breaking changes:

* [METRICS SDK] Rename Base2 Exponential Histogram Aggregation config field
Expand Down
2 changes: 1 addition & 1 deletion ci/do_ci.sh
Original file line number Diff line number Diff line change
Expand Up @@ -632,7 +632,7 @@ elif [[ "$1" == "bazel.tsan" ]]; then
exit 0
elif [[ "$1" == "bazel.valgrind" ]]; then
bazel $BAZEL_STARTUP_OPTIONS build $BAZEL_OPTIONS_ASYNC //...
bazel $BAZEL_STARTUP_OPTIONS test --test_timeout=600 --run_under="/usr/bin/valgrind --leak-check=full --error-exitcode=1 --errors-for-leak-kinds=definite --suppressions=\"${SRC_DIR}/ci/valgrind-suppressions\"" $BAZEL_TEST_OPTIONS_ASYNC //...
bazel $BAZEL_STARTUP_OPTIONS test --test_timeout=900 --run_under="/usr/bin/valgrind --leak-check=full --error-exitcode=1 --errors-for-leak-kinds=definite --suppressions=\"${SRC_DIR}/ci/valgrind-suppressions\"" $BAZEL_TEST_OPTIONS_ASYNC //...
exit 0
elif [[ "$1" == "benchmark" ]]; then
[ -z "${BENCHMARK_DIR}" ] && export BENCHMARK_DIR=$HOME/benchmark
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,111 @@
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0

#pragma once

#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW

# include <stdint.h>
# include <memory>
# include <utility>
# include <vector>

# include "opentelemetry/common/timestamp.h"
# include "opentelemetry/context/context.h"
# include "opentelemetry/nostd/shared_ptr.h"
# include "opentelemetry/sdk/metrics/data/exemplar_data.h"
# include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
# include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
# include "opentelemetry/trace/context.h"
# include "opentelemetry/trace/span_context.h"
# include "opentelemetry/version.h"

OPENTELEMETRY_BEGIN_NAMESPACE
namespace sdk
{
namespace metrics
{

/**
* A reservoir that pre-filters measurements according to an ExemplarFilterType before
* delegating eligible ones to a wrapped reservoir.
*/
class FilteredExemplarReservoir final : public ExemplarReservoir
{
public:
FilteredExemplarReservoir(ExemplarFilterType filter_type,
nostd::shared_ptr<ExemplarReservoir> reservoir)
: should_sample_(SelectFilter(filter_type)), reservoir_(std::move(reservoir))
{}

void OfferMeasurement(int64_t value,
const MetricAttributes &attributes,
const opentelemetry::context::Context &context,
const opentelemetry::common::SystemTimestamp &timestamp) noexcept override
{
if (should_sample_(context))
{
reservoir_->OfferMeasurement(value, attributes, context, timestamp);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think AlwaysOn still does not fully work here. This forwards the measurement, but the real reservoir drops it later if there is no valid span context. Since trace/span ids are optional for exemplars, AlwaysOn + Context{} should still produce an exemplar without trace ids. Can we add a test that collects from the real reservoir and checks the returned exemplar is non-null?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same point for the above #4267 (comment)

}
}

void OfferMeasurement(double value,
const MetricAttributes &attributes,
const opentelemetry::context::Context &context,
const opentelemetry::common::SystemTimestamp &timestamp) noexcept override
{
if (should_sample_(context))
{
reservoir_->OfferMeasurement(value, attributes, context, timestamp);
}
}

std::vector<std::shared_ptr<ExemplarData>> CollectAndReset(
const MetricAttributes &pointAttributes) noexcept override
{
return reservoir_->CollectAndReset(pointAttributes);
}

private:
using ShouldSampleFn = bool (*)(const opentelemetry::context::Context &context);

static bool AlwaysOn(const opentelemetry::context::Context & /* context */) noexcept
{
return true;
}

static bool AlwaysOff(const opentelemetry::context::Context & /* context */) noexcept
{
return false;
}

static bool TraceBased(const opentelemetry::context::Context &context) noexcept
{
const opentelemetry::trace::SpanContext span_context =
opentelemetry::trace::GetSpanContext(context);
return span_context.IsValid() && span_context.IsSampled();
}

static ShouldSampleFn SelectFilter(ExemplarFilterType filter_type) noexcept
{
switch (filter_type)
{
case ExemplarFilterType::kAlwaysOn:
return &AlwaysOn;
case ExemplarFilterType::kAlwaysOff:
return &AlwaysOff;
case ExemplarFilterType::kTraceBased:
return &TraceBased;
}
return &TraceBased; // unreachable; all enumerators handled above
}

ShouldSampleFn should_sample_;
nostd::shared_ptr<ExemplarReservoir> reservoir_;
};

} // namespace metrics
} // namespace sdk
OPENTELEMETRY_END_NAMESPACE

#endif // ENABLE_METRICS_EXEMPLAR_PREVIEW
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ class ExemplarReservoir

static nostd::shared_ptr<ExemplarReservoir> GetSimpleFilteredExemplarReservoir(
ExemplarFilterType filter_type,
std::shared_ptr<ExemplarReservoir> reservoir);
nostd::shared_ptr<ExemplarReservoir> reservoir);

static nostd::shared_ptr<ExemplarReservoir> GetSimpleFixedSizeExemplarReservoir(
size_t size,
Expand Down
26 changes: 16 additions & 10 deletions sdk/include/opentelemetry/sdk/metrics/exemplar/reservoir_utils.h
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@
# include "opentelemetry/common/macros.h"
# include "opentelemetry/sdk/metrics/aggregation/aggregation_config.h"
# include "opentelemetry/sdk/metrics/exemplar/aligned_histogram_bucket_exemplar_reservoir.h"
# include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
# include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
# include "opentelemetry/sdk/metrics/exemplar/simple_fixed_size_exemplar_reservoir.h"
# include "opentelemetry/version.h"

Expand Down Expand Up @@ -48,7 +50,8 @@ static inline size_t GetSimpleFixedReservoirDefaultSize(const AggregationType ag
static inline nostd::shared_ptr<ExemplarReservoir> GetExemplarReservoir(
const AggregationType agg_type,
const AggregationConfig *agg_config,
const InstrumentDescriptor &instrument_descriptor)
const InstrumentDescriptor &instrument_descriptor,
ExemplarFilterType filter_type)
{
if (agg_type == AggregationType::kHistogram)
{
Expand All @@ -61,18 +64,21 @@ static inline nostd::shared_ptr<ExemplarReservoir> GetExemplarReservoir(
//
if (histogram_agg_config != nullptr && histogram_agg_config->boundaries_.size() > 1)
{
return nostd::shared_ptr<ExemplarReservoir>(new AlignedHistogramBucketExemplarReservoir(
histogram_agg_config->boundaries_.size(),
AlignedHistogramBucketExemplarReservoir::GetHistogramCellSelector(
histogram_agg_config->boundaries_),
GetMapAndResetCellMethod(instrument_descriptor)));
return ExemplarReservoir::GetSimpleFilteredExemplarReservoir(
filter_type,
nostd::shared_ptr<ExemplarReservoir>(new AlignedHistogramBucketExemplarReservoir(
histogram_agg_config->boundaries_.size(),
AlignedHistogramBucketExemplarReservoir::GetHistogramCellSelector(
histogram_agg_config->boundaries_),
GetMapAndResetCellMethod(instrument_descriptor))));
}
}

return nostd::shared_ptr<ExemplarReservoir>(new SimpleFixedSizeExemplarReservoir(
GetSimpleFixedReservoirDefaultSize(agg_type, agg_config),
SimpleFixedSizeExemplarReservoir::GetSimpleFixedSizeCellSelector(),
GetMapAndResetCellMethod(instrument_descriptor)));
return ExemplarReservoir::GetSimpleFilteredExemplarReservoir(

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For AlwaysOff, can we return GetNoExemplarReservoir() directly? Otherwise every metric recording still goes through the exemplar filter path just to drop the sample, so AlwaysOff is not really the same as disabling exemplar sampling

@proost proost Jul 22, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For the performance, you are right. But according to spec, looks like current implementation is correct.

Here is what i understand:

ExemplarFilter decide whether measurement is eligible for the examplar. ExamplarReservoir decide wheter examplar to be stored or not.

If i missunderstand the spec, please let me know where I got it wrong. (spec is bit confused to me.)

filter_type, nostd::shared_ptr<ExemplarReservoir>(new SimpleFixedSizeExemplarReservoir(
GetSimpleFixedReservoirDefaultSize(agg_type, agg_config),
SimpleFixedSizeExemplarReservoir::GetSimpleFixedSizeCellSelector(),
GetMapAndResetCellMethod(instrument_descriptor))));
}
} // namespace metrics
} // namespace sdk
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
#include "opentelemetry/sdk/metrics/aggregation/default_aggregation.h"

#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
# include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
# include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
#endif

Expand All @@ -37,7 +36,6 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
AsyncMetricStorage(const InstrumentDescriptor &instrument_descriptor,
const AggregationType aggregation_type,
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
ExemplarFilterType exempler_filter_type,
nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
#endif
const AggregationConfig *aggregation_config)
Expand All @@ -49,7 +47,6 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
delta_hash_map_(
std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
exemplar_filter_type_(exempler_filter_type),
exemplar_reservoir_(std::move(exemplar_reservoir)),
#endif
temporal_metric_storage_(instrument_descriptor, aggregation_type, aggregation_config)
Expand All @@ -66,11 +63,8 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
for (auto &measurement : measurements)
{
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
if (exemplar_filter_type_ == ExemplarFilterType::kAlwaysOn)
{
exemplar_reservoir_->OfferMeasurement(measurement.second, {}, {},
std::chrono::system_clock::now());
}
exemplar_reservoir_->OfferMeasurement(measurement.second, {}, {},
std::chrono::system_clock::now());
#endif

auto aggr = DefaultAggregation::CreateAggregation(aggregation_type_, instrument_descriptor_);
Expand Down Expand Up @@ -146,7 +140,6 @@ class AsyncMetricStorage : public MetricStorage, public AsyncWritableMetricStora
std::unique_ptr<AttributesHashMap> delta_hash_map_;
opentelemetry::common::SpinLockMutex hashmap_lock_;
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
ExemplarFilterType exemplar_filter_type_;
nostd::shared_ptr<ExemplarReservoir> exemplar_reservoir_;
#endif
TemporalMetricStorage temporal_metric_storage_;
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@
#endif

#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
# include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
# include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
#endif

Expand All @@ -48,25 +47,11 @@ namespace metrics
class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
{

#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW

static inline bool EnableExamplarFilter(ExemplarFilterType filter_type,
const opentelemetry::context::Context &context)
{
return filter_type == ExemplarFilterType::kAlwaysOn ||
(filter_type == ExemplarFilterType::kTraceBased &&
opentelemetry::trace::GetSpan(context)->GetContext().IsValid() &&
opentelemetry::trace::GetSpan(context)->GetContext().IsSampled());
}

#endif // ENABLE_METRICS_EXEMPLAR_PREVIEW

public:
SyncMetricStorage(const InstrumentDescriptor &instrument_descriptor,
const AggregationType aggregation_type,
std::shared_ptr<const AttributesProcessor> attributes_processor,
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
ExemplarFilterType exempler_filter_type,
nostd::shared_ptr<ExemplarReservoir> &&exemplar_reservoir,
#endif
const AggregationConfig *aggregation_config)
Expand All @@ -76,7 +61,6 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
std::make_unique<AttributesHashMap>(aggregation_config_->cardinality_limit_)),
attributes_processor_(std::move(attributes_processor)),
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
exemplar_filter_type_(exempler_filter_type),
exemplar_reservoir_(std::move(exemplar_reservoir)),
#endif
temporal_metric_storage_(instrument_descriptor, aggregation_type, aggregation_config)
Expand All @@ -97,10 +81,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
return;
}
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
if (EnableExamplarFilter(exemplar_filter_type_, context))
{
exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now());
}
exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now());
#endif
static MetricAttributes attr = MetricAttributes{};
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
Expand All @@ -123,11 +104,8 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
return;
}
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
if (EnableExamplarFilter(exemplar_filter_type_, context))
{
exemplar_reservoir_->OfferMeasurement(value, attributes, context,
std::chrono::system_clock::now());
}
exemplar_reservoir_->OfferMeasurement(value, attributes, context,
std::chrono::system_clock::now());
#endif

MetricAttributes attr{attributes, attributes_processor_.get()};
Expand Down Expand Up @@ -155,10 +133,7 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
return;
}
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
if (EnableExamplarFilter(exemplar_filter_type_, context))
{
exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now());
}
exemplar_reservoir_->OfferMeasurement(value, {}, context, std::chrono::system_clock::now());
#endif
static MetricAttributes attr = MetricAttributes{};
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
Expand All @@ -181,11 +156,8 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
return;
}
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
if (EnableExamplarFilter(exemplar_filter_type_, context))
{
exemplar_reservoir_->OfferMeasurement(value, attributes, context,
std::chrono::system_clock::now());
}
exemplar_reservoir_->OfferMeasurement(value, attributes, context,
std::chrono::system_clock::now());
#endif
MetricAttributes attr{attributes, attributes_processor_.get()};
std::lock_guard<opentelemetry::common::SpinLockMutex> guard(attribute_hashmap_lock_);
Expand Down Expand Up @@ -296,7 +268,6 @@ class SyncMetricStorage : public MetricStorage, public SyncWritableMetricStorage
std::function<std::unique_ptr<Aggregation>()> create_default_aggregation_;
std::shared_ptr<const AttributesProcessor> attributes_processor_;
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
ExemplarFilterType exemplar_filter_type_;
nostd::shared_ptr<ExemplarReservoir> exemplar_reservoir_;
#endif
TemporalMetricStorage temporal_metric_storage_;
Expand Down
11 changes: 11 additions & 0 deletions sdk/src/metrics/exemplar/reservoir.cc
Original file line number Diff line number Diff line change
Expand Up @@ -4,9 +4,12 @@
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW

# include <stddef.h>
# include <utility>

# include "opentelemetry/nostd/shared_ptr.h"
# include "opentelemetry/sdk/metrics/exemplar/aligned_histogram_bucket_exemplar_reservoir.h"
# include "opentelemetry/sdk/metrics/exemplar/filter_type.h"
# include "opentelemetry/sdk/metrics/exemplar/filtered_exemplar_reservoir.h"
# include "opentelemetry/sdk/metrics/exemplar/no_exemplar_reservoir.h"
# include "opentelemetry/sdk/metrics/exemplar/reservoir.h"
# include "opentelemetry/sdk/metrics/exemplar/reservoir_cell.h"
Expand All @@ -20,6 +23,14 @@ namespace sdk
namespace metrics
{

nostd::shared_ptr<ExemplarReservoir> ExemplarReservoir::GetSimpleFilteredExemplarReservoir(
ExemplarFilterType filter_type,
nostd::shared_ptr<ExemplarReservoir> reservoir)
{
return nostd::shared_ptr<ExemplarReservoir>{
new FilteredExemplarReservoir{filter_type, std::move(reservoir)}};
}

nostd::shared_ptr<ExemplarReservoir> ExemplarReservoir::GetSimpleFixedSizeExemplarReservoir(
size_t size,
const std::shared_ptr<ReservoirCellSelector> &reservoir_cell_selector,
Expand Down
6 changes: 2 additions & 4 deletions sdk/src/metrics/meter.cc
Original file line number Diff line number Diff line change
Expand Up @@ -545,9 +545,8 @@ std::unique_ptr<SyncWritableMetricStorage> Meter::RegisterSyncMetricStorage(
sync_storage = std::shared_ptr<SyncMetricStorage>(new SyncMetricStorage(
view_instr_desc, view.GetAggregationType(), view.GetAttributesProcessor(),
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
exemplar_filter_type,
GetExemplarReservoir(view.GetAggregationType(), view.GetAggregationConfig(),
view_instr_desc),
view_instr_desc, exemplar_filter_type),
#endif
view.GetAggregationConfig()));
storage_registry_.insert({view_instr_desc, sync_storage});
Expand Down Expand Up @@ -618,9 +617,8 @@ std::unique_ptr<AsyncWritableMetricStorage> Meter::RegisterAsyncMetricStorage(
async_storage = std::shared_ptr<AsyncMetricStorage>(new AsyncMetricStorage(
view_instr_desc, view.GetAggregationType(),
#ifdef ENABLE_METRICS_EXEMPLAR_PREVIEW
exemplar_filter_type,
GetExemplarReservoir(view.GetAggregationType(), view.GetAggregationConfig(),
view_instr_desc),
view_instr_desc, exemplar_filter_type),
#endif
view.GetAggregationConfig()));
storage_registry_.insert({view_instr_desc, async_storage});
Expand Down
Loading
Loading