1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
#include <gtest/gtest.h>
#include <stdint.h>
#include <atomic>
#include <chrono>
#include <initializer_list> // IWYU pragma: keep
#include <random>
#include <thread>
#include <utility>
#include <vector>
#include "common.h"
#include "opentelemetry/context/context.h" // IWYU pragma: keep
#include "opentelemetry/metrics/meter.h"
#include "opentelemetry/metrics/sync_instruments.h"
#include "opentelemetry/nostd/function_ref.h"
#include "opentelemetry/nostd/string_view.h"
#include "opentelemetry/nostd/unique_ptr.h"
#include "opentelemetry/nostd/variant.h"
#include "opentelemetry/sdk/common/exporter_utils.h"
#include "opentelemetry/sdk/instrumentationscope/instrumentation_scope.h"
#include "opentelemetry/sdk/metrics/data/metric_data.h"
#include "opentelemetry/sdk/metrics/data/point_data.h"
#include "opentelemetry/sdk/metrics/export/metric_producer.h"
#include "opentelemetry/sdk/metrics/instruments.h"
#include "opentelemetry/sdk/metrics/meter_provider.h"
#include "opentelemetry/sdk/metrics/metric_reader.h"
#include "opentelemetry/sdk/metrics/push_metric_exporter.h"
using namespace opentelemetry;
using namespace opentelemetry::sdk::instrumentationscope;
using namespace opentelemetry::sdk::metrics;
class MockMetricExporterForStress : public opentelemetry::sdk::metrics::PushMetricExporter
{
public:
MockMetricExporterForStress() = default;
opentelemetry::sdk::metrics::AggregationTemporality GetAggregationTemporality(
opentelemetry::sdk::metrics::InstrumentType) const noexcept override
{
return AggregationTemporality::kDelta;
}
opentelemetry::sdk::common::ExportResult Export(
const opentelemetry::sdk::metrics::ResourceMetrics &) noexcept override
{
return opentelemetry::sdk::common::ExportResult::kSuccess;
}
bool ForceFlush(std::chrono::microseconds) noexcept override { return true; }
bool Shutdown(std::chrono::microseconds) noexcept override { return true; }
};
TEST(HistogramStress, UnsignedInt64)
{
MeterProvider mp;
auto m = mp.GetMeter("meter1", "version1", "schema1");
std::unique_ptr<MockMetricExporterForStress> exporter(new MockMetricExporterForStress());
std::shared_ptr<MetricReader> reader{new MockMetricReader(std::move(exporter))};
mp.AddMetricReader(reader);
auto h = m->CreateUInt64Histogram("histogram1", "histogram1_description", "histogram1_unit");
//
// Start a dedicated thread to collect the metrics
//
std::vector<HistogramPointData> actuals;
auto stop_collecting = std::make_shared<std::atomic<bool>>(false);
auto collect_thread = std::thread([&reader, &actuals, stop_collecting]() {
while (!*stop_collecting)
{
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
reader->Collect([&](ResourceMetrics &rm) {
for (const ScopeMetrics &smd : rm.scope_metric_data_)
{
for (const MetricData &md : smd.metric_data_)
{
for (const PointDataAttributes &dp : md.point_data_attr_)
{
actuals.push_back(opentelemetry::nostd::get<HistogramPointData>(dp.point_data));
}
}
}
return true;
});
}
});
//
// Start logging threads
//
int record_thread_count = std::thread::hardware_concurrency() - 1;
if (record_thread_count <= 0)
{
record_thread_count = 1;
}
std::vector<std::thread> threads(record_thread_count);
constexpr int iterations_per_thread = 2000000;
auto expected_sum = std::make_shared<std::atomic<uint64_t>>(0);
for (int i = 0; i < record_thread_count; ++i)
{
threads[i] = std::thread([&] {
std::random_device rd;
std::mt19937 random_engine(rd());
std::uniform_int_distribution<> gen_random(1, 20000);
for (int j = 0; j < iterations_per_thread; ++j)
{
int64_t val = gen_random(random_engine);
expected_sum->fetch_add(val, std::memory_order_relaxed);
h->Record(val, {});
}
});
}
for (int i = 0; i < record_thread_count; ++i)
{
threads[i].join();
}
//
// Stop the dedicated collection thread
//
*stop_collecting = true;
collect_thread.join();
//
// run the the final collection
//
reader->Collect([&](ResourceMetrics &rm) {
for (const ScopeMetrics &smd : rm.scope_metric_data_)
{
for (const MetricData &md : smd.metric_data_)
{
for (const PointDataAttributes &dp : md.point_data_attr_)
{
actuals.push_back(opentelemetry::nostd::get<HistogramPointData>(dp.point_data));
}
}
}
return true;
});
//
// Aggregate the results
//
int64_t expected_count = record_thread_count * iterations_per_thread;
int64_t collected_count = 0;
int64_t collected_sum = 0;
for (const auto &actual : actuals)
{
int64_t collected_bucket_sum = 0;
for (const auto &count : actual.counts_)
{
collected_bucket_sum += count;
}
ASSERT_EQ(collected_bucket_sum, actual.count_);
collected_sum += opentelemetry::nostd::get<int64_t>(actual.sum_);
collected_count += actual.count_;
}
ASSERT_EQ(expected_count, collected_count);
ASSERT_EQ(*expected_sum, collected_sum);
}
|
