Fix first bucket otlp issue if invalid end stamp (#665)

src/AbstractMetricsManager.h

@@ -218,7 +218,7 @@ class AbstractMetricsBucket
 
     virtual void to_json(json &j) const = 0;
     virtual void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const = 0;
-    virtual void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const = 0;
+    virtual void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const = 0;
     virtual void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) = 0;
 };
 
@@ -553,8 +553,13 @@ class AbstractMetricsManager
         if (!_tap_name.empty() && add_labels.find("tap") == add_labels.end()) {
             add_labels["tap"] = _tap_name;
         }
-
-        _metric_buckets.at(period)->to_opentelemetry(scope, add_labels);
+        auto bucket = _metric_buckets.at(period).get();
+        auto start_ts = bucket->start_tstamp();
+        auto end_ts = bucket->end_tstamp();
+        if (!end_ts.tv_sec) {
+            timespec_get(&end_ts, TIME_UTC);
+        }
+        bucket->to_opentelemetry(scope, start_ts, end_ts, add_labels);
     }
 
     void window_external_opentelemetry(metrics::v1::ScopeMetrics &scope, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) const
@@ -563,7 +568,13 @@ class AbstractMetricsManager
             return;
         }
         // static because caller guarantees only our own bucket type
-        static_cast<MetricsBucketClass *>(bucket)->to_opentelemetry(scope, add_labels);
+        auto sbucket = static_cast<MetricsBucketClass *>(bucket);
+        auto start_ts = sbucket->start_tstamp();
+        auto end_ts = sbucket->end_tstamp();
+        if (!end_ts.tv_sec) {
+            timespec_get(&end_ts, TIME_UTC);
+        }
+        sbucket->to_opentelemetry(scope, start_ts, end_ts, add_labels);
     }
 
     void window_external_prometheus(std::stringstream &out, AbstractMetricsBucket *bucket, Metric::LabelMap add_labels = {}) const

src/handlers/bgp/BgpStreamHandler.cpp

@@ -189,11 +189,8 @@ void BgpMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap ad
     _counters.filtered.to_prometheus(out, add_labels);
 }
 
-void BgpMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void BgpMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     _rate_total.to_opentelemetry(scope, start_ts, end_ts, add_labels);
 
     {

src/handlers/bgp/BgpStreamHandler.h

@@ -78,7 +78,7 @@ class BgpMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t, uint64_t) override
     {
     }

src/handlers/dhcp/DhcpStreamHandler.cpp

@@ -150,11 +150,8 @@ void DhcpMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a
     _dhcp_topServers.to_prometheus(out, add_labels);
 }
 
-void DhcpMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void DhcpMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     _rate_total.to_opentelemetry(scope, start_ts, end_ts, add_labels);
 
     {

src/handlers/dhcp/DhcpStreamHandler.h

@@ -100,7 +100,7 @@ class DhcpMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t, uint64_t) override
     {
     }

src/handlers/dns/v1/DnsStreamHandler.cpp

@@ -1238,11 +1238,8 @@ void DnsMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap ad
     });
 }
 
-void DnsMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void DnsMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     _rate_total.to_opentelemetry(scope, start_ts, end_ts, add_labels);
 
     {

src/handlers/dns/v1/DnsStreamHandler.h

@@ -199,7 +199,7 @@ class DnsMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override
     {
         _dns_topGeoLocECS.set_settings(topn_count, percentile_threshold);

src/handlers/dns/v2/DnsStreamHandler.cpp

@@ -840,11 +840,8 @@ void DnsMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap ad
     }
 }
 
-void DnsMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void DnsMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     for (auto &dns : _dns) {
         auto dir_labels = add_labels;
         dir_labels["direction"] = _dir_str.at(dns.first);

src/handlers/dns/v2/DnsStreamHandler.h

@@ -362,7 +362,7 @@ class DnsMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override
     {
         _topn_count = topn_count;

src/handlers/flow/FlowStreamHandler.cpp

@@ -885,11 +885,8 @@ void FlowMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a
     }
 }
 
-void FlowMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void FlowMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     std::shared_lock r_lock(_mutex);
 
     for (const auto &device : _devices_metrics) {

src/handlers/flow/FlowStreamHandler.h

@@ -275,7 +275,7 @@ class FlowMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override
     {
         _topn_count = topn_count;

src/handlers/input_resources/InputResourcesStreamHandler.cpp

@@ -165,11 +165,8 @@ void InputResourcesMetricsBucket::to_prometheus(std::stringstream &out, Metric::
     _handler_count.to_prometheus(out, add_labels);
 }
 
-void InputResourcesMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void InputResourcesMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     {
         auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe
 

src/handlers/input_resources/InputResourcesStreamHandler.h

@@ -53,7 +53,7 @@ class InputResourcesMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t, uint64_t) override
     {
     }

src/handlers/mock/MockStreamHandler.cpp

@@ -78,11 +78,8 @@ void MockMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a
     _counters.mock_counter.to_prometheus(out, add_labels);
 }
 
-void MockMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void MockMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     std::shared_lock r_lock(_mutex);
 
     _counters.mock_counter.to_opentelemetry(scope, start_ts, end_ts, add_labels);

src/handlers/mock/MockStreamHandler.h

@@ -7,9 +7,9 @@
 #include "AbstractMetricsManager.h"
 #include "MockInputStream.h"
 #include "StreamHandler.h"
-#include <spdlog/spdlog.h>
 #include <Corrade/Utility/Debug.h>
 #include <limits>
+#include <spdlog/spdlog.h>
 #include <string>
 
 namespace visor::handler::mock {
@@ -50,7 +50,7 @@ class MockMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t, uint64_t) override
     {
     }

src/handlers/net/v1/NetStreamHandler.cpp

@@ -387,11 +387,8 @@ void NetworkMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMa
     _payload_size.to_prometheus(out, add_labels);
 }
 
-void NetworkMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void NetworkMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     _rate_in.to_opentelemetry(scope, start_ts, end_ts, add_labels);
     _rate_out.to_opentelemetry(scope, start_ts, end_ts, add_labels);
     _rate_total.to_opentelemetry(scope, start_ts, end_ts, add_labels);

src/handlers/net/v1/NetStreamHandler.h

@@ -139,7 +139,7 @@ class NetworkMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override
     {
         _topGeoLoc.set_settings(topn_count, percentile_threshold);

src/handlers/net/v2/NetStreamHandler.cpp

@@ -382,11 +382,8 @@ void NetworkMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMa
     }
 }
 
-void NetworkMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void NetworkMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     if (group_enabled(group::NetMetrics::Quantiles)) {
         for (auto &net : _net) {
             auto dir_labels = add_labels;

src/handlers/net/v2/NetStreamHandler.h

@@ -192,7 +192,7 @@ class NetworkMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t topn_count, uint64_t percentile_threshold) override
     {
         _topn_count = topn_count;

src/handlers/netprobe/NetProbeStreamHandler.cpp

@@ -186,11 +186,8 @@ void NetProbeMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelM
     }
 }
 
-void NetProbeMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void NetProbeMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     std::shared_lock r_lock(_mutex);
 
     for (const auto &target : _targets_metrics) {

src/handlers/netprobe/NetProbeStreamHandler.h

@@ -15,8 +15,8 @@
 #pragma GCC diagnostic ignored "-Wold-style-cast"
 #pragma GCC diagnostic ignored "-Wzero-as-null-pointer-constant"
 #endif
-#include <VisorTcpLayer.h>
 #include <IcmpLayer.h>
+#include <VisorTcpLayer.h>
 #ifdef __GNUC__
 #pragma GCC diagnostic pop
 #endif
@@ -81,7 +81,7 @@ class NetProbeMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t, uint64_t) override
     {
     }

src/handlers/pcap/PcapStreamHandler.cpp

@@ -98,19 +98,15 @@ void PcapMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap a
     _counters.pcap_if_drop.to_prometheus(out, add_labels);
 }
 
-void PcapMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels) const
+void PcapMetricsBucket::to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels) const
 {
-    auto start_ts = start_tstamp();
-    auto end_ts = end_tstamp();
-
     std::shared_lock r_lock(_mutex);
 
     _counters.pcap_TCP_reassembly_errors.to_opentelemetry(scope, start_ts, end_ts, add_labels);
     _counters.pcap_os_drop.to_opentelemetry(scope, start_ts, end_ts, add_labels);
     _counters.pcap_if_drop.to_opentelemetry(scope, start_ts, end_ts, add_labels);
 }
 
-
 void PcapMetricsBucket::to_json(json &j) const
 {
     std::shared_lock r_lock(_mutex);

src/handlers/pcap/PcapStreamHandler.h

@@ -59,7 +59,7 @@ class PcapMetricsBucket final : public visor::AbstractMetricsBucket
     void specialized_merge(const AbstractMetricsBucket &other, Metric::Aggregate agg_operator) override;
     void to_json(json &j) const override;
     void to_prometheus(std::stringstream &out, Metric::LabelMap add_labels = {}) const override;
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap add_labels = {}) const override;
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &start_ts, timespec &end_ts, Metric::LabelMap add_labels = {}) const override;
     void update_topn_metrics(size_t, uint64_t) override
     {
     }

src/tests/test_metrics.cpp

@@ -17,7 +17,7 @@ class TestMetricsBucket final : public AbstractMetricsBucket
     {
         out << "test_performed" << std::endl;
     }
-    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, Metric::LabelMap) const
+    void to_opentelemetry(metrics::v1::ScopeMetrics &scope, timespec &, timespec &, Metric::LabelMap) const
     {
         scope.add_metrics()->set_name("test1");
         scope.add_metrics()->set_name("test2");