Merge remote-tracking branch 'origin/topic/awelzel/prom-callbacks-2'

* origin/topic/awelzel/prom-callbacks-2:
  Update broker submodule
  telemetry: Move callbacks to Zeek
  auxil/prometheus-cpp: Pin to 1.2.4

(cherry picked from commit f24bc1ee88)

auxil/broker

@@ -1 +1 @@
-Subproject commit 5b3ed87a93b2ded1f3c95ff1a3b99e2c6ab84ef4
+Subproject commit a80bf420aa6f55b4eb959ae89c184522a096a119

auxil/prometheus-cpp

@@ -1 +1 @@
-Subproject commit 4649065e2a1dd21c81e41cd6007dce5486b77fc0
+Subproject commit ad99e21f4706193670c42b36c9824dc997f4c475

src/Timer.cc

@@ -2,14 +2,11 @@
 
 #include "zeek/Timer.h"
 
-#include "zeek/zeek-config.h"
-
 #include "zeek/Desc.h"
 #include "zeek/NetVar.h"
 #include "zeek/RunState.h"
 #include "zeek/broker/Manager.h"
 #include "zeek/iosource/Manager.h"
-#include "zeek/iosource/PktSrc.h"
 #include "zeek/util.h"
 
 namespace zeek::detail {

src/telemetry/Counter.cc

@@ -2,27 +2,17 @@
 
 using namespace zeek::telemetry;
 
-Counter::Counter(FamilyType* family, const prometheus::Labels& labels, prometheus::CollectCallbackPtr callback) noexcept
+Counter::Counter(FamilyType* family, const prometheus::Labels& labels, detail::CollectCallbackPtr callback) noexcept
-    : handle(family->Add(labels)), labels(labels) {
+    : family(family), handle(family->Add(labels)), labels(labels), callback(std::move(callback)) {}
-    if ( callback ) {
-        handle.AddCollectCallback(std::move(callback));
-        has_callback = true;
-    }
-}
 
 double Counter::Value() const noexcept {
-    if ( has_callback ) {
+    if ( callback )
-        // Use Collect() here instead of Value() to correctly handle metrics with
+        return callback();
-        // callbacks.
-        auto metric = handle.Collect();
-        return metric.counter.value;
-    }
 
     return handle.Value();
 }
 
-std::shared_ptr<Counter> CounterFamily::GetOrAdd(Span<const LabelView> labels,
+std::shared_ptr<Counter> CounterFamily::GetOrAdd(Span<const LabelView> labels, detail::CollectCallbackPtr callback) {
-                                                 prometheus::CollectCallbackPtr callback) {
     prometheus::Labels p_labels = detail::BuildPrometheusLabels(labels);
 
     auto check = [&](const std::shared_ptr<Counter>& counter) { return counter->CompareLabels(p_labels); };
@@ -36,6 +26,15 @@ std::shared_ptr<Counter> CounterFamily::GetOrAdd(Span<const LabelView> labels,
 }
 
 std::shared_ptr<Counter> CounterFamily::GetOrAdd(std::initializer_list<LabelView> labels,
-                                                 prometheus::CollectCallbackPtr callback) {
+                                                 detail::CollectCallbackPtr callback) {
     return GetOrAdd(Span{labels.begin(), labels.size()}, std::move(callback));
 }
+
+void CounterFamily::RunCallbacks() {
+    for ( auto& c : counters ) {
+        if ( c->HasCallback() ) {
+            double val = c->RunCallback();
+            c->Set(val);
+        }
+    }
+}

src/telemetry/Counter.h

@@ -4,7 +4,6 @@
 
 #include <prometheus/counter.h>
 #include <prometheus/family.h>
-#include <cstdint>
 #include <initializer_list>
 #include <memory>
 
@@ -15,6 +14,12 @@
 
 namespace zeek::telemetry {
 
+namespace detail {
+using CollectCallbackPtr = std::function<double()>;
+}
+
+class CounterFamily;
+
 /**
  * A handle to a metric that can only go up.
  */
@@ -26,7 +31,7 @@ public:
     using FamilyType = prometheus::Family<Handle>;
 
     explicit Counter(FamilyType* family, const prometheus::Labels& labels,
-                     prometheus::CollectCallbackPtr callback = nullptr) noexcept;
+                     detail::CollectCallbackPtr callback = nullptr) noexcept;
 
     /**
      * Increments the value by 1.
@@ -55,10 +60,21 @@ public:
 
     bool CompareLabels(const prometheus::Labels& lbls) const { return labels == lbls; }
 
+    bool HasCallback() const noexcept { return callback != nullptr; }
+    double RunCallback() const { return callback(); }
+
 private:
+    friend class CounterFamily;
+    void Set(double val) {
+        // Counter has no Set(), but we can fake it.
+        handle.Reset();
+        handle.Increment(val);
+    }
+
+    FamilyType* family = nullptr;
     Handle& handle;
     prometheus::Labels labels;
-    bool has_callback = false;
+    detail::CollectCallbackPtr callback;
 };
 
 using CounterPtr = std::shared_ptr<Counter>;
@@ -74,15 +90,17 @@ public:
      * Returns the metrics handle for given labels, creating a new instance
      * lazily if necessary.
      */
-    CounterPtr GetOrAdd(Span<const LabelView> labels, prometheus::CollectCallbackPtr callback = nullptr);
+    CounterPtr GetOrAdd(Span<const LabelView> labels, detail::CollectCallbackPtr callback = nullptr);
 
     /**
      * @copydoc GetOrAdd
      */
-    CounterPtr GetOrAdd(std::initializer_list<LabelView> labels, prometheus::CollectCallbackPtr callback = nullptr);
+    CounterPtr GetOrAdd(std::initializer_list<LabelView> labels, detail::CollectCallbackPtr callback = nullptr);
 
     zeek_int_t MetricType() const noexcept override { return BifEnum::Telemetry::MetricType::COUNTER; }
 
+    void RunCallbacks() override;
+
 private:
     prometheus::Family<prometheus::Counter>* family;
     std::vector<CounterPtr> counters;

src/telemetry/Gauge.cc

@@ -3,26 +3,16 @@
 using namespace zeek::telemetry;
 
 double Gauge::Value() const noexcept {
-    if ( has_callback ) {
+    if ( callback )
-        // Use Collect() here instead of Value() to correctly handle metrics
+        return callback();
-        // with callbacks.
-        auto metric = handle.Collect();
-        return metric.gauge.value;
-    }
 
     return handle.Value();
 }
 
+Gauge::Gauge(FamilyType* family, const prometheus::Labels& labels, detail::CollectCallbackPtr callback) noexcept
+    : family(family), handle(family->Add(labels)), labels(labels), callback(std::move(callback)) {}
 
-Gauge::Gauge(FamilyType* family, const prometheus::Labels& labels, prometheus::CollectCallbackPtr callback) noexcept
+std::shared_ptr<Gauge> GaugeFamily::GetOrAdd(Span<const LabelView> labels, detail::CollectCallbackPtr callback) {
-    : handle(family->Add(labels)), labels(labels) {
-    if ( callback ) {
-        handle.AddCollectCallback(std::move(callback));
-        has_callback = true;
-    }
-}
-
-std::shared_ptr<Gauge> GaugeFamily::GetOrAdd(Span<const LabelView> labels, prometheus::CollectCallbackPtr callback) {
     prometheus::Labels p_labels = detail::BuildPrometheusLabels(labels);
 
     auto check = [&](const std::shared_ptr<Gauge>& gauge) { return gauge->CompareLabels(p_labels); };
@@ -36,6 +26,13 @@ std::shared_ptr<Gauge> GaugeFamily::GetOrAdd(Span<const LabelView> labels, prome
 }
 
 std::shared_ptr<Gauge> GaugeFamily::GetOrAdd(std::initializer_list<LabelView> labels,
-                                             prometheus::CollectCallbackPtr callback) {
+                                             detail::CollectCallbackPtr callback) {
     return GetOrAdd(Span{labels.begin(), labels.size()}, std::move(callback));
 }
+
+void GaugeFamily::RunCallbacks() {
+    for ( const auto& g : gauges ) {
+        if ( g->HasCallback() )
+            g->Set(g->RunCallback());
+    }
+}

src/telemetry/Gauge.h

@@ -4,7 +4,7 @@
 
 #include <prometheus/family.h>
 #include <prometheus/gauge.h>
-#include <cstdint>
+#include <unistd.h>
 #include <initializer_list>
 #include <memory>
 
@@ -15,6 +15,10 @@
 
 namespace zeek::telemetry {
 
+namespace detail {
+using CollectCallbackPtr = std::function<double()>;
+}
+
 /**
  * A handle to a metric that can count up and down.
  */
@@ -26,7 +30,7 @@ public:
     using FamilyType = prometheus::Family<Handle>;
 
     explicit Gauge(FamilyType* family, const prometheus::Labels& labels,
-                   prometheus::CollectCallbackPtr callback = nullptr) noexcept;
+                   detail::CollectCallbackPtr callback = nullptr) noexcept;
 
     /**
      * Increments the value by 1.
@@ -57,6 +61,11 @@ public:
      */
     void Dec(double amount) noexcept { handle.Decrement(amount); }
 
+    /**
+     * Set the value by @p val.
+     */
+    void Set(double val) noexcept { handle.Set(val); }
+
     /**
      * Decrements the value by 1.
      * @return The new value.
@@ -73,10 +82,14 @@ public:
 
     bool CompareLabels(const prometheus::Labels& lbls) const { return labels == lbls; }
 
+    bool HasCallback() const noexcept { return callback != nullptr; }
+    double RunCallback() const { return callback(); }
+
 private:
+    FamilyType* family = nullptr;
     Handle& handle;
     prometheus::Labels labels;
-    bool has_callback = false;
+    detail::CollectCallbackPtr callback;
 };
 
 using GaugePtr = std::shared_ptr<Gauge>;
@@ -89,18 +102,20 @@ public:
      * Returns the metrics handle for given labels, creating a new instance
      * lazily if necessary.
      */
-    GaugePtr GetOrAdd(Span<const LabelView> labels, prometheus::CollectCallbackPtr callback = nullptr);
+    GaugePtr GetOrAdd(Span<const LabelView> labels, detail::CollectCallbackPtr callback = nullptr);
 
     /**
      * @copydoc GetOrAdd
      */
-    GaugePtr GetOrAdd(std::initializer_list<LabelView> labels, prometheus::CollectCallbackPtr callback = nullptr);
+    GaugePtr GetOrAdd(std::initializer_list<LabelView> labels, detail::CollectCallbackPtr callback = nullptr);
 
     zeek_int_t MetricType() const noexcept override { return BifEnum::Telemetry::MetricType::GAUGE; }
 
     GaugeFamily(prometheus::Family<prometheus::Gauge>* family, Span<const std::string_view> labels)
         : MetricFamily(labels), family(family) {}
 
+    void RunCallbacks() override;
+
 private:
     prometheus::Family<prometheus::Gauge>* family;
     std::vector<GaugePtr> gauges;

src/telemetry/Histogram.h

@@ -66,6 +66,8 @@ public:
 
     zeek_int_t MetricType() const noexcept override { return BifEnum::Telemetry::MetricType::HISTOGRAM; }
 
+    void RunCallbacks() override {}
+
 private:
     prometheus::Family<prometheus::Histogram>* family;
     prometheus::Histogram::BucketBoundaries boundaries;

src/telemetry/Manager.cc

@@ -122,36 +122,16 @@ void Manager::InitPostScript() {
         return &this->current_process_stats;
     };
     rss_gauge = GaugeInstance("process", "resident_memory", {}, "Resident memory size", "bytes",
-                              []() -> prometheus::ClientMetric {
+                              []() { return static_cast<double>(get_stats()->rss); });
-                                  auto* s = get_stats();
-                                  prometheus::ClientMetric metric;
-                                  metric.gauge.value = static_cast<double>(s->rss);
-                                  return metric;
-                              });
 
     vms_gauge = GaugeInstance("process", "virtual_memory", {}, "Virtual memory size", "bytes",
-                              []() -> prometheus::ClientMetric {
+                              []() { return static_cast<double>(get_stats()->vms); });
-                                  auto* s = get_stats();
-                                  prometheus::ClientMetric metric;
-                                  metric.gauge.value = static_cast<double>(s->vms);
-                                  return metric;
-                              });
 
     cpu_gauge = GaugeInstance("process", "cpu", {}, "Total user and system CPU time spent", "seconds",
-                              []() -> prometheus::ClientMetric {
+                              []() { return get_stats()->cpu; });
-                                  auto* s = get_stats();
-                                  prometheus::ClientMetric metric;
-                                  metric.gauge.value = s->cpu;
-                                  return metric;
-                              });
 
     fds_gauge = GaugeInstance("process", "open_fds", {}, "Number of open file descriptors", "",
-                              []() -> prometheus::ClientMetric {
+                              []() { return static_cast<double>(get_stats()->fds); });
-                                  auto* s = get_stats();
-                                  prometheus::ClientMetric metric;
-                                  metric.gauge.value = static_cast<double>(s->fds);
-                                  return metric;
-                              });
 #endif
 
     iosource_mgr->RegisterFd(collector_flare.FD(), this);
@@ -491,7 +471,7 @@ CounterFamilyPtr Manager::CounterFamily(std::string_view prefix, std::string_vie
 
 CounterPtr Manager::CounterInstance(std::string_view prefix, std::string_view name, Span<const LabelView> labels,
                                     std::string_view helptext, std::string_view unit,
-                                    prometheus::CollectCallbackPtr callback) {
+                                    detail::CollectCallbackPtr callback) {
     return WithLabelNames(labels, [&, this](auto labelNames) {
         auto family = CounterFamily(prefix, name, labelNames, helptext, unit);
         return family->GetOrAdd(labels, callback);
@@ -500,7 +480,7 @@ CounterPtr Manager::CounterInstance(std::string_view prefix, std::string_view na
 
 CounterPtr Manager::CounterInstance(std::string_view prefix, std::string_view name,
                                     std::initializer_list<LabelView> labels, std::string_view helptext,
-                                    std::string_view unit, prometheus::CollectCallbackPtr callback) {
+                                    std::string_view unit, detail::CollectCallbackPtr callback) {
     auto lbl_span = Span{labels.begin(), labels.size()};
     return CounterInstance(prefix, name, lbl_span, helptext, unit, std::move(callback));
 }
@@ -529,8 +509,7 @@ GaugeFamilyPtr Manager::GaugeFamily(std::string_view prefix, std::string_view na
 }
 
 GaugePtr Manager::GaugeInstance(std::string_view prefix, std::string_view name, Span<const LabelView> labels,
-                                std::string_view helptext, std::string_view unit,
+                                std::string_view helptext, std::string_view unit, detail::CollectCallbackPtr callback) {
-                                prometheus::CollectCallbackPtr callback) {
     return WithLabelNames(labels, [&, this](auto labelNames) {
         auto family = GaugeFamily(prefix, name, labelNames, helptext, unit);
         return family->GetOrAdd(labels, callback);
@@ -538,8 +517,7 @@ GaugePtr Manager::GaugeInstance(std::string_view prefix, std::string_view name,
 }
 
 GaugePtr Manager::GaugeInstance(std::string_view prefix, std::string_view name, std::initializer_list<LabelView> labels,
-                                std::string_view helptext, std::string_view unit,
+                                std::string_view helptext, std::string_view unit, detail::CollectCallbackPtr callback) {
-                                prometheus::CollectCallbackPtr callback) {
     auto lbl_span = Span{labels.begin(), labels.size()};
     return GaugeInstance(prefix, name, lbl_span, helptext, unit, std::move(callback));
 }
@@ -588,7 +566,9 @@ void Manager::ProcessFd(int fd, int flags) {
 
     collector_flare.Extinguish();
 
-    prometheus_registry->UpdateViaCallbacks();
+    for ( const auto& [name, f] : families )
+        f->RunCallbacks();
+
     collector_response_idx = collector_request_idx;
 
     lk.unlock();

src/telemetry/Manager.h

@@ -31,6 +31,10 @@ class Registry;
 
 namespace zeek::telemetry {
 
+namespace detail {
+using CollectCallbackPtr = std::function<double()>;
+}
+
 class ZeekCollectable;
 
 /**
@@ -98,12 +102,12 @@ public:
      */
     CounterPtr CounterInstance(std::string_view prefix, std::string_view name, Span<const LabelView> labels,
                                std::string_view helptext, std::string_view unit = "",
-                               prometheus::CollectCallbackPtr callback = nullptr);
+                               detail::CollectCallbackPtr callback = nullptr);
 
     /// @copydoc counterInstance
     CounterPtr CounterInstance(std::string_view prefix, std::string_view name, std::initializer_list<LabelView> labels,
                                std::string_view helptext, std::string_view unit = "",
-                               prometheus::CollectCallbackPtr callback = nullptr);
+                               detail::CollectCallbackPtr callback = nullptr);
 
     /**
      * @return A gauge metric family. Creates the family lazily if necessary.
@@ -134,12 +138,12 @@ public:
      */
     GaugePtr GaugeInstance(std::string_view prefix, std::string_view name, Span<const LabelView> labels,
                            std::string_view helptext, std::string_view unit = "",
-                           prometheus::CollectCallbackPtr callback = nullptr);
+                           detail::CollectCallbackPtr callback = nullptr);
 
     /// @copydoc GaugeInstance
     GaugePtr GaugeInstance(std::string_view prefix, std::string_view name, std::initializer_list<LabelView> labels,
                            std::string_view helptext, std::string_view unit = "",
-                           prometheus::CollectCallbackPtr callback = nullptr);
+                           detail::CollectCallbackPtr callback = nullptr);
 
     // Forces the compiler to use the type `Span<const T>` instead of trying to
     // match parameters to a `span`.

src/telemetry/MetricFamily.h

@@ -22,6 +22,8 @@ public:
 
     std::vector<std::string> LabelNames() const { return label_names; }
 
+    virtual void RunCallbacks() = 0;
+
 protected:
     MetricFamily(Span<const std::string_view> labels) {
         for ( const auto& lbl : labels )

src/telemetry/Utils.h

@@ -5,7 +5,6 @@
 #include <string_view>
 
 #include "zeek/Span.h"
-#include "zeek/Val.h"
 
 namespace zeek::telemetry {