ceph/src/jaegertracing/opentelemetry-cpp/sdk/test/_metrics/sketch_aggregator_test.cc

   1 // Copyright The OpenTelemetry Authors
   2 // SPDX-License-Identifier: Apache-2.0
   3
   4 #ifdef ENABLE_METRICS_PREVIEW
   5 #  include "opentelemetry/sdk/_metrics/aggregator/sketch_aggregator.h"
   6
   7 #  include <gtest/gtest.h>
   8 #  include <iostream>
   9 #  include <numeric>
  10 #  include <thread>
  11
  12 namespace metrics_api = opentelemetry::metrics;
  13
  14 OPENTELEMETRY_BEGIN_NAMESPACE
  15 namespace sdk
  16 {
  17 namespace metrics
  18 {
  19
  20 // Test updating with a uniform set of updates
  21 TEST(Sketch, UniformValues)
  22 {
  23   SketchAggregator<int> alpha(metrics_api::InstrumentKind::ValueRecorder, .000005);
  24
  25   EXPECT_EQ(alpha.get_aggregator_kind(), AggregatorKind::Sketch);
  26
  27   alpha.checkpoint();
  28   EXPECT_EQ(alpha.get_checkpoint().size(), 2);
  29   EXPECT_EQ(alpha.get_boundaries().size(), 0);
  30   EXPECT_EQ(alpha.get_counts().size(), 0);
  31
  32   for (int i = 0; i < 60; i++)
  33   {
  34     alpha.update(i);
  35   }
  36
  37   alpha.checkpoint();
  38
  39   EXPECT_EQ(alpha.get_boundaries().size(), 60);
  40   EXPECT_EQ(alpha.get_counts().size(), 60);
  41
  42   EXPECT_EQ(alpha.get_checkpoint()[0], 1770);
  43   EXPECT_EQ(alpha.get_checkpoint()[1], 60);
  44 }
  45
  46 // Test updating with a normal distribution
  47 TEST(Sketch, NormalValues)
  48 {
  49   SketchAggregator<int> alpha(metrics_api::InstrumentKind::ValueRecorder, .0005);
  50
  51   std::vector<int> vals{1, 3, 3, 5, 5, 5, 7, 7, 7, 7, 9, 9, 9, 11, 11, 13};
  52   for (int i : vals)
  53   {
  54     alpha.update(i);
  55   }
  56   alpha.checkpoint();
  57
  58   EXPECT_EQ(alpha.get_checkpoint()[0], std::accumulate(vals.begin(), vals.end(), 0));
  59   EXPECT_EQ(alpha.get_checkpoint()[1], vals.size());
  60
  61   std::vector<int> correct = {1, 2, 3, 4, 3, 2, 1};
  62   EXPECT_EQ(alpha.get_counts(), correct);
  63
  64   std::vector<double> captured_bounds = alpha.get_boundaries();
  65   for (size_t i = 0; i < captured_bounds.size(); i++)
  66   {
  67     captured_bounds[i] = round(captured_bounds[i]);
  68   }
  69
  70   // It is not guaranteed that bounds are correct once the bucket sizes pass 1000
  71   std::vector<double> correct_bounds = {1, 3, 5, 7, 9, 11, 13};
  72   EXPECT_EQ(captured_bounds, correct_bounds);
  73 }
  74
  75 int randVal()
  76 {
  77   return rand() % 100000;
  78 }
  79
  80 /** Note that in this case, "Large" refers to a number of distinct values which exceed the maximum
  81  * number of allowed buckets.
  82  */
  83 TEST(Sketch, QuantileSmall)
  84 {
  85   SketchAggregator<int> alpha(metrics_api::InstrumentKind::ValueRecorder, .00005);
  86
  87   std::vector<int> vals1(2048);
  88   std::generate(vals1.begin(), vals1.end(), randVal);
  89
  90   std::vector<int> vals2(2048);
  91   std::generate(vals1.begin(), vals1.end(), randVal);
  92
  93   for (int i : vals1)
  94   {
  95     alpha.update(i);
  96   }
  97   alpha.checkpoint();
  98   std::sort(vals1.begin(), vals1.end());
  99
 100   EXPECT_TRUE(abs(alpha.get_quantiles(.25) - vals1[2048 * .25 - 1]) <= 10);
 101   EXPECT_TRUE(abs(alpha.get_quantiles(.50) - vals1[2048 * .50 - 1]) <= 10);
 102   EXPECT_TRUE(abs(alpha.get_quantiles(.75) - vals1[2048 * .75 - 1]) <= 10);
 103 }
 104
 105 TEST(Sketch, UpdateQuantileLarge)
 106 {
 107   SketchAggregator<int> alpha(metrics_api::InstrumentKind::ValueRecorder, .0005, 7);
 108   std::vector<int> vals{1, 3, 3, 5, 5, 5, 7, 7, 7, 7, 9, 9, 9, 11, 11, 13};
 109   for (int i : vals)
 110   {
 111     alpha.update(i);
 112   }
 113
 114   // This addition should trigger the "1" and "3" buckets to merge
 115   alpha.update(15);
 116   alpha.checkpoint();
 117
 118   std::vector<int> correct = {3, 3, 4, 3, 2, 1, 1};
 119   EXPECT_EQ(alpha.get_counts(), correct);
 120
 121   for (int i : vals)
 122   {
 123     alpha.update(i);
 124   }
 125   alpha.update(15);
 126   alpha.update(17);
 127   alpha.checkpoint();
 128
 129   correct = {6, 4, 3, 2, 1, 1, 1};
 130   EXPECT_EQ(alpha.get_counts(), correct);
 131 }
 132
 133 TEST(Sketch, MergeSmall)
 134 {
 135   SketchAggregator<int> alpha(metrics_api::InstrumentKind::ValueRecorder, .0005);
 136   SketchAggregator<int> beta(metrics_api::InstrumentKind::ValueRecorder, .0005);
 137
 138   std::vector<int> vals{1, 3, 3, 5, 5, 5, 7, 7, 7, 7, 9, 9, 9, 11, 11, 13};
 139   for (int i : vals)
 140   {
 141     alpha.update(i);
 142   }
 143
 144   std::vector<int> otherVals{1, 1, 1, 1, 11, 11, 13, 13, 13, 15};
 145   for (int i : otherVals)
 146   {
 147     beta.update(i);
 148   }
 149
 150   alpha.merge(beta);
 151   alpha.checkpoint();
 152
 153   EXPECT_EQ(alpha.get_checkpoint()[0], std::accumulate(vals.begin(), vals.end(), 0) +
 154                                            std::accumulate(otherVals.begin(), otherVals.end(), 0));
 155   EXPECT_EQ(alpha.get_checkpoint()[1], vals.size() + otherVals.size());
 156
 157   std::vector<int> correct = {5, 2, 3, 4, 3, 4, 4, 1};
 158   EXPECT_EQ(alpha.get_counts(), correct);
 159 }
 160
 161 TEST(Sketch, MergeLarge)
 162 {
 163   SketchAggregator<int> alpha(metrics_api::InstrumentKind::ValueRecorder, .0005, 7);
 164   SketchAggregator<int> beta(metrics_api::InstrumentKind::ValueRecorder, .0005, 7);
 165
 166   std::vector<int> vals{1, 3, 3, 5, 5, 5, 7, 7, 7, 7, 9, 9, 9, 11, 11, 13};
 167   for (int i : vals)
 168   {
 169     alpha.update(i);
 170   }
 171
 172   std::vector<int> otherVals{1, 1, 1, 1, 11, 11, 13, 13, 13, 15};
 173   for (int i : otherVals)
 174   {
 175     beta.update(i);
 176   }
 177
 178   alpha.merge(beta);
 179   alpha.checkpoint();
 180
 181   EXPECT_EQ(alpha.get_checkpoint()[0], std::accumulate(vals.begin(), vals.end(), 0) +
 182                                            std::accumulate(otherVals.begin(), otherVals.end(), 0));
 183   EXPECT_EQ(alpha.get_checkpoint()[1], vals.size() + otherVals.size());
 184
 185   std::vector<int> correct = {7, 3, 4, 3, 4, 4, 1};
 186   EXPECT_EQ(alpha.get_counts(), correct);
 187 }
 188
 189 // Update callback used to validate multi-threaded performance
 190 void sketchUpdateCallback(Aggregator<int> &agg, std::vector<int> vals)
 191 {
 192   for (int i : vals)
 193   {
 194     agg.update(i);
 195   }
 196 }
 197
 198 TEST(Sketch, Concurrency)
 199 {
 200   SketchAggregator<int> alpha(metrics_api::InstrumentKind::ValueRecorder, .0005, 20);
 201
 202   std::vector<int> vals1(1000);
 203   std::generate(vals1.begin(), vals1.end(), randVal);
 204
 205   std::vector<int> vals2(1000);
 206   std::generate(vals2.begin(), vals2.end(), randVal);
 207
 208   std::thread first(sketchUpdateCallback, std::ref(alpha), vals1);
 209   std::thread second(sketchUpdateCallback, std::ref(alpha), vals2);
 210
 211   first.join();
 212   second.join();
 213
 214   SketchAggregator<int> beta(metrics_api::InstrumentKind::ValueRecorder, .0005, 20);
 215
 216   for (int i : vals1)
 217   {
 218     beta.update(i);
 219   }
 220   for (int i : vals2)
 221   {
 222     beta.update(i);
 223   }
 224
 225   alpha.checkpoint();
 226   beta.checkpoint();
 227
 228   EXPECT_EQ(alpha.get_checkpoint(), beta.get_checkpoint());
 229   EXPECT_EQ(alpha.get_counts(), beta.get_counts());
 230   EXPECT_EQ(alpha.get_boundaries(), beta.get_boundaries());
 231 }
 232
 233 #  if __EXCEPTIONS
 234
 235 TEST(Sketch, Errors)
 236 {
 237
 238   SketchAggregator<int> tol1(metrics_api::InstrumentKind::ValueRecorder, .000005);
 239   SketchAggregator<int> tol2(metrics_api::InstrumentKind::ValueRecorder, .005);
 240   SketchAggregator<int> sz1(metrics_api::InstrumentKind::ValueRecorder, .000005, 2938);
 241   SketchAggregator<int> sz2(metrics_api::InstrumentKind::ValueRecorder, .000005);
 242
 243   EXPECT_ANY_THROW(tol1.merge(tol2));
 244   EXPECT_ANY_THROW(sz1.merge(sz2));
 245   EXPECT_ANY_THROW(tol1.get_quantiles(-.000001));
 246   EXPECT_ANY_THROW(tol1.get_quantiles(1.000001));
 247 }
 248
 249 #  endif
 250
 251 }  // namespace metrics
 252 }  // namespace sdk
 253 OPENTELEMETRY_END_NAMESPACE
 254 #endif