Use correct metric type in OTLP exporter

reducer/metric_info.cc

@@ -20,47 +20,56 @@ namespace reducer {
 TcpMetricInfo TcpMetricInfo::bytes{
     TcpMetrics::bytes,
     "The total number of TCP bytes between the source and destination measured for the prior thirty seconds.",
-    UNIT_BYTES};
+    UNIT_BYTES,
+    MetricTypeSum};
 
 TcpMetricInfo TcpMetricInfo::rtt_num_measurements{
     TcpMetrics::rtt_num_measurements,
     "The number of measurements made in calculating the current RTT average value.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeGauge};
 
 TcpMetricInfo TcpMetricInfo::active{
     TcpMetrics::active,
     "The number of TCP connections considered to be open and alive between the source and destination at the point the measurement was taken.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeGauge};
 
 TcpMetricInfo TcpMetricInfo::rtt_average{
     TcpMetrics::rtt_average,
     "The computed average round trip time between the source and destination as measured in microseconds.",
-    UNIT_MICROSECONDS};
+    UNIT_MICROSECONDS,
+    MetricTypeGauge};
 
 TcpMetricInfo TcpMetricInfo::packets{
     TcpMetrics::packets,
     "The total number of TCP packets between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 TcpMetricInfo TcpMetricInfo::retrans{
     TcpMetrics::retrans,
     "The total number of TCP retransmission requests between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 TcpMetricInfo TcpMetricInfo::syn_timeouts{
     TcpMetrics::syn_timeouts,
     "The total number of TCP SYN timeouts between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 TcpMetricInfo TcpMetricInfo::new_sockets{
     TcpMetrics::new_sockets,
     "The total number of new TCP sockets opened between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 TcpMetricInfo TcpMetricInfo::resets{
     TcpMetrics::resets,
     "The total number of TCP resets sent between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 ////////////////////////////////////////////////////////////////////////////////
 // UDP
@@ -69,22 +78,26 @@ TcpMetricInfo TcpMetricInfo::resets{
 UdpMetricInfo UdpMetricInfo::bytes{
     UdpMetrics::bytes,
     "The total number of UDP bytes between the source and destination measured for the prior thirty seconds.",
-    UNIT_BYTES};
+    UNIT_BYTES,
+    MetricTypeSum};
 
 UdpMetricInfo UdpMetricInfo::packets{
     UdpMetrics::packets,
     "The total number of UDP packets between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 UdpMetricInfo UdpMetricInfo::active{
     UdpMetrics::active,
     "The number of UDP connections considered to be open and alive between the source and destination at the point the measurement was taken.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeGauge};
 
 UdpMetricInfo UdpMetricInfo::drops{
     UdpMetrics::drops,
     "The total number of UDP connections dropped between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 ////////////////////////////////////////////////////////////////////////////////
 // DNS
@@ -95,29 +108,34 @@ DnsMetricInfo DnsMetricInfo::client_duration_average{
     "This metric is the average duration in microseconds from when the client sends a DNS request, until the response is received back from the server."
     " As such, it includes the communication round-trip times, plus the server processing latency."
     " Computed by the summation of all times, divided by dns.responses.",
-    UNIT_MICROSECONDS};
+    UNIT_MICROSECONDS,
+    MetricTypeGauge};
 
 DnsMetricInfo DnsMetricInfo::server_duration_average{
     DnsMetrics::server_duration_average,
     "This metric is the average duration in microseconds for the server to respond to a request received locally. "
     " Thus, it does not include the network latency from or to the client."
     " Computed by the summation of all times, divided by dns.responses.",
-    UNIT_MICROSECONDS};
+    UNIT_MICROSECONDS,
+    MetricTypeGauge};
 
 DnsMetricInfo DnsMetricInfo::active_sockets{
     DnsMetrics::active_sockets,
     "The number of DNS connections for which measurements were taken in the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeGauge};
 
 DnsMetricInfo DnsMetricInfo::responses{
     DnsMetrics::responses,
     "The total number of DNS responses sent between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 DnsMetricInfo DnsMetricInfo::timeouts{
     DnsMetrics::timeouts,
     "The total number of DNS timeouts between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 
 ////////////////////////////////////////////////////////////////////////////////
 // HTTP
@@ -128,23 +146,27 @@ HttpMetricInfo HttpMetricInfo::client_duration_average{
     "This metric is the average duration in microseconds from when the client sends an HTTP request, until the response is received back from the server."
     " As such, it includes the communication round-trip times, plus the server processing latency."
     " Computed by summation of all times, divided by http.active_sockets.",
-    UNIT_MICROSECONDS};
+    UNIT_MICROSECONDS,
+    MetricTypeGauge};
 
 HttpMetricInfo HttpMetricInfo::server_duration_average{
     HttpMetrics::server_duration_average,
     "This metric is the average duration in microseconds for the server to respond to a request received locally."
     " Thus, it does not include the network latency from or to the client."
     " Computed by summation of all times, divided by http.active_sockets.",
-    UNIT_MICROSECONDS};
+    UNIT_MICROSECONDS,
+    MetricTypeGauge};
 
 HttpMetricInfo HttpMetricInfo::active_sockets{
     HttpMetrics::active_sockets,
     "The number of unencrypted HTTPv1 connections for which measurements were taken in the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeGauge};
 
 HttpMetricInfo HttpMetricInfo::status_code{
     HttpMetrics::status_code,
     "For a given class of response code (see 'response_code' dimension), the number of times an unencrypted server sent an"
     " HTTPv1 status code between the source and destination measured for the prior thirty seconds.",
-    UNIT_DIMENSIONLESS};
+    UNIT_DIMENSIONLESS,
+    MetricTypeSum};
 } // namespace reducer

reducer/metric_info.h

@@ -9,6 +9,14 @@
 #include <string_view>
 
 namespace reducer {
+// Metric Types
+//
+enum MetricType {
+  // OTLP Metric Data Sum
+  MetricTypeSum,
+  // OTLP Metric Data Gauge
+  MetricTypeGauge,
+};
 
 // Information associated with a metric.
 //
@@ -19,9 +27,12 @@ struct MetricInfo {
   const std::string description;
   // Unit in which the metric value is reported (format described in http://unitsofmeasure.org/ucum.html).
   const std::string unit;
+  // Metric Type
+  MetricType type;
 
-  explicit MetricInfo(std::string_view name_, std::string_view description_ = "", std::string_view unit_ = "")
-      : name(name_), description(description_), unit(unit_)
+  explicit MetricInfo(
+      std::string_view name_, std::string_view description_ = "", std::string_view unit_ = "", MetricType type_ = MetricTypeSum)
+      : name(name_), description(description_), unit(unit_), type(type_)
   {}
 };
 
@@ -32,8 +43,8 @@ template <typename T> struct OutboundMetricInfo : public MetricInfo {
   typedef T metric_group_t;
   const metric_group_t metric;
 
-  OutboundMetricInfo(T metric_, std::string_view description_, std::string_view unit_)
-      : MetricInfo(to_string(metric_), description_, unit_), metric(metric_)
+  OutboundMetricInfo(T metric_, std::string_view description_, std::string_view unit_, MetricType type_ = MetricTypeSum)
+      : MetricInfo(to_string(metric_), description_, unit_, type_), metric(metric_)
   {}
 };
 

reducer/otlp_grpc_formatter.cc

@@ -49,6 +49,8 @@ OtlpGrpcFormatter::OtlpGrpcFormatter(Publisher::WriterPtr const &writer) : write
   sum_.set_aggregation_temporality(opentelemetry::proto::metrics::v1::AggregationTemporality::AGGREGATION_TEMPORALITY_DELTA);
   sum_.set_is_monotonic(true);
   sum_.add_data_points();
+
+  gauge_.add_data_points();
 }
 
 OtlpGrpcFormatter::~OtlpGrpcFormatter()
@@ -128,30 +130,36 @@ void OtlpGrpcFormatter::format(
     metric.set_description(metric_info.description.data(), metric_info.description.size());
   }
 
-  auto data_point = sum_.mutable_data_points(0);
-
   if (labels_changed) {
     SCOPED_TIMING(OtlpGrpcFormatterFormatLabelsChanged);
-    data_point->clear_attributes();
+    data_point_.clear_attributes();
     for (auto const &[key, value] : labels) {
-      auto attribute = data_point->add_attributes();
+      auto attribute = data_point_.add_attributes();
       attribute->set_key(key.data(), key.size());
       attribute->mutable_value()->set_string_value(value.data(), value.size());
     }
   }
 
   if (timestamp_changed) {
-    data_point->set_time_unix_nano(integer_time<std::chrono::nanoseconds>(timestamp));
+    // set the start time to the timestamp minus 30 seconds.
+    data_point_.set_start_time_unix_nano(integer_time<std::chrono::nanoseconds>(timestamp) - int64_t(30000000000));
+    data_point_.set_time_unix_nano(integer_time<std::chrono::nanoseconds>(timestamp));
   }
 
   std::visit(
       overloaded_visitor{
-          [&](auto val) { data_point->set_as_int(val); },
-          [&](double val) { data_point->set_as_double(val); },
+          [&](auto val) { data_point_.set_as_int(val); },
+          [&](double val) { data_point_.set_as_double(val); },
       },
       metric_value);
 
-  *metric.mutable_sum() = sum_;
+  if (metric_info.type == MetricTypeSum) {
+    *sum_.mutable_data_points(0) = data_point_;
+    *metric.mutable_sum() = sum_;
+  } else {
+    *gauge_.mutable_data_points(0) = data_point_;
+    *metric.mutable_gauge() = gauge_;
+  }
 
   *scope_metrics_->add_metrics() = std::move(metric);
   STOP_TIMING(OtlpGrpcFormatterFormatMetric);

reducer/otlp_grpc_formatter.h

@@ -66,6 +66,8 @@ class OtlpGrpcFormatter : public TsdbFormatter {
   // avoid regenerating common portions.
   ExportMetricsServiceRequest metrics_request_;
   opentelemetry::proto::metrics::v1::Sum sum_;
+  opentelemetry::proto::metrics::v1::Gauge gauge_;
+  opentelemetry::proto::metrics::v1::NumberDataPoint data_point_;
   opentelemetry::proto::metrics::v1::ScopeMetrics *scope_metrics_;
 
   OtlpGrpcPublisher::WriterPtr const &writer_;