update r09-receiver to allow for scraping the prometheus data

Author: marenz2569

CMakeLists.txt

@@ -1,8 +1,10 @@
 cmake_minimum_required(VERSION 3.22)
 
 add_executable(r09-receiver
-               src/main.cpp
                src/correlate_access_code_bb_ts_fl_impl.cpp
+               src/prometheus.cpp
+               src/prometheus_gauge_populator.cpp
+               src/r09-receiver.cpp
                src/rational_resampler_impl.cc)
 
 target_compile_options(r09-receiver PUBLIC -std=c++17 -Wall)
@@ -13,11 +15,12 @@ find_package(Gnuradio "3.8" REQUIRED)
 find_package(Boost REQUIRED)
 find_package(Volk REQUIRED)
 find_package(libenvpp REQUIRED)
+find_package(prometheus-cpp CONFIG REQUIRED)
 
 include(${OSMOSDR_DIR}/lib/cmake/osmosdr/gnuradio-osmosdrConfig.cmake)
 
 include_directories(${GNURADIO_ALL_INCLUDE_DIRS})
 
-target_link_libraries(r09-receiver log4cpp gnuradio-digital gnuradio-analog gnuradio-filter gnuradio-blocks gnuradio-fft gnuradio-runtime gnuradio-pmt volk gnuradio-osmosdr libenvpp::libenvpp)
+target_link_libraries(r09-receiver log4cpp gnuradio-digital gnuradio-analog gnuradio-filter gnuradio-blocks gnuradio-fft gnuradio-runtime gnuradio-pmt volk gnuradio-osmosdr libenvpp::libenvpp prometheus-cpp::pull)
 
 install(TARGETS r09-receiver DESTINATION bin)

README.md

@@ -16,6 +16,7 @@ DECODER\_RF | int of RF value for SDR
 DECODER\_IF | int of IF value for SDR 
 DECODER\_BB | int of BB value for SDR 
 DECODER\_DEVICE\_STRING | device string for osmosdr
+DECODER\_PROMETHEUS\_ADDRESS | the address where the appliaction is listening on for prometheus metric extraction (Format is host:port)
 
 ## building with nix flakes
 

nixos-module/default.nix

@@ -40,6 +40,16 @@ in
       default = 42;
       description = "";
     };
+    PrometheusHost = mkOption {
+      type = types.str;
+      default = "localhost";
+      description = ''The host of the prometheus exporter for the R09-receiver'';
+    };
+    PrometheusPort = mkOption {
+      type = types.port;
+      default = 9020;
+      description = ''The port of the prometheus exporter for the R09-receiver'';
+    };
     user = mkOption {
       type = types.str;
       default = "r09-receiver";
@@ -52,6 +62,16 @@ in
     };
   };
 
+  options.services.prometheus.exporters.r09-receiver = with lib; {
+    port = mkOption {
+      type = types.port;
+      default = 9020;
+      description = lib.mdDoc ''
+        Port to listen on.
+      '';
+    };
+  };
+
   config = lib.mkIf cfg.enable {
 
     hardware = {
@@ -74,6 +94,7 @@ in
         "DECODER_IF" = toString IF;
         "DECODER_BB" = toString BB;
         "DECODER_DEVICE_STRING" = device;
+        "DECODER_PROMETHEUS_ADDRESS" = "${PrometheusHost}:${toString PrometheusPort}";
       };
 
       serviceConfig = {
@@ -85,6 +106,9 @@ in
       };
     };
 
+    # provide the prometheus exporter information for scraping
+    services.prometheus.exporters.r09-receiver.port = cfg.PrometheusPort;
+
     users.groups."${cfg.group}" = { };
     users.users."${cfg.user}" = {
       name = cfg.user;

pkgs/r09-receiver.nix

@@ -13,6 +13,9 @@
 , fftwFloat
 , patchelf
 , libenvpp
+, prometheus-cpp
+, zlib
+, curlFull
 }:
 let
   decoder-dependencies = stdenv.mkDerivation {
@@ -82,7 +85,7 @@ let
     src = ./..;
 
     nativeBuildInputs = [ cmake pkg-config gnuradio_unwrapped osmosdr ];
-    buildInputs = [ log4cpp mpir gnuradio_unwrapped.boost.dev gmpxx.dev gnuradio_unwrapped.volk libenvpp ];
+    buildInputs = [ log4cpp mpir gnuradio_unwrapped.boost.dev gmpxx.dev gnuradio_unwrapped.volk libenvpp prometheus-cpp zlib curlFull ];
 
     cmakeFlags = [ "-DOSMOSDR_DIR=${osmosdr}" ];
   };

src/prometheus.cpp

@@ -0,0 +1,17 @@
+#include "prometheus.h"
+
+PrometheusExporter::PrometheusExporter(
+    const std::string &prometheus_host) noexcept {
+  exposer_ = std::make_unique<prometheus::Exposer>(prometheus_host);
+  registry_ = std::make_shared<prometheus::Registry>();
+
+  exposer_->RegisterCollectable(registry_);
+}
+
+auto PrometheusExporter::signal_strength() noexcept
+    -> prometheus::Family<prometheus::Gauge> & {
+  return prometheus::BuildGauge()
+      .Name("signal_strength")
+      .Help("Current Signal Strength")
+      .Register(*registry_);
+}

src/prometheus.h

@@ -0,0 +1,22 @@
+#ifndef PROMETHEUS_H
+#define PROMETHEUS_H
+
+#include <memory>
+
+#include <prometheus/counter.h>
+#include <prometheus/exposer.h>
+#include <prometheus/registry.h>
+
+class PrometheusExporter {
+private:
+  std::shared_ptr<prometheus::Registry> registry_;
+  std::unique_ptr<prometheus::Exposer> exposer_;
+
+public:
+  PrometheusExporter(const std::string &host) noexcept;
+  ~PrometheusExporter() noexcept = default;
+
+  auto signal_strength() noexcept -> prometheus::Family<prometheus::Gauge> &;
+};
+
+#endif // PROMETHEUS_H

src/prometheus_gauge_populator.cpp

@@ -0,0 +1,38 @@
+#include <gnuradio/io_signature.h>
+
+#include "prometheus_gauge_populator.h"
+
+namespace gr::prometheus {
+
+PrometheusGaugePopulator::sptr
+PrometheusGaugePopulator::make(::prometheus::Gauge &gauge) {
+  return gnuradio::get_initial_sptr(new PrometheusGaugePopulator(gauge));
+}
+
+PrometheusGaugePopulator::PrometheusGaugePopulator(::prometheus::Gauge &gauge)
+    : block(
+          /*name=*/"PrometheusGaugePopulator",
+          /*input_signature=*/
+          io_signature::make(/*min_streams=*/1, /*max_streams=*/1,
+                             /*sizeof_stream_items=*/sizeof(float)),
+          /*output_signature=*/
+          io_signature::make(/*min_streams=*/0, /*max_streams=*/0,
+                             /*sizeof_stream_items=*/0)),
+      gauge_(gauge) {}
+
+auto PrometheusGaugePopulator::general_work(
+    const int, gr_vector_int &, gr_vector_const_void_star &input_items,
+    gr_vector_void_star &) -> int {
+  const auto *in = (const float *)input_items[0];
+
+  // Consume one and write it to the gauge.
+  const auto item = *in;
+  consume_each(1);
+
+  gauge_.Set(item);
+
+  // We do not produce any items.
+  return 0;
+}
+
+} // namespace gr::prometheus

src/prometheus_gauge_populator.h

@@ -0,0 +1,33 @@
+#ifndef PROMETHEUS_GAUGE_POPULATOR_H
+#define PROMETHEUS_GAUGE_POPULATOR_H
+
+#include <gnuradio/block.h>
+#include <gnuradio/blocks/api.h>
+
+#include <prometheus/gauge.h>
+
+namespace gr::prometheus {
+
+/// This block takes a float as an input and writes in into a prometheus gauge
+class PrometheusGaugePopulator : virtual public block {
+private:
+  /// the prometheus gauge we are populating with this block
+  ::prometheus::Gauge &gauge_;
+
+public:
+  using sptr = boost::shared_ptr<PrometheusGaugePopulator>;
+
+  PrometheusGaugePopulator() = delete;
+
+  explicit PrometheusGaugePopulator(::prometheus::Gauge &gauge);
+
+  static auto make(::prometheus::Gauge &gauge) -> sptr;
+
+  auto general_work(int noutput_items, gr_vector_int &ninput_items,
+                    gr_vector_const_void_star &input_items,
+                    gr_vector_void_star &output_items) -> int override;
+};
+
+} // namespace gr::prometheus
+
+#endif // PROMETHEUS_GAUGE_POPULATOR_H

src/main.cpp renamed to src/r09-receiver.cpp

@@ -8,6 +8,7 @@
 #include <gnuradio/analog/pwr_squelch_cc.h>
 #include <gnuradio/analog/quadrature_demod_cf.h>
 #include <gnuradio/blocks/add_const_ff.h>
+#include <gnuradio/blocks/complex_to_mag_squared.h>
 #include <gnuradio/blocks/multiply_const.h>
 #include <gnuradio/blocks/socket_pdu.h>
 #include <gnuradio/blocks/tagged_stream_to_pdu.h>
@@ -27,6 +28,8 @@
 #include <osmosdr/source.h>
 
 #include "correlate_access_code_bb_ts_fl.h"
+#include "prometheus.h"
+#include "prometheus_gauge_populator.h"
 #include "rational_resampler.h"
 
 static auto print_gnuradio_diagnostics() -> void {
@@ -42,7 +45,8 @@ static auto print_gnuradio_diagnostics() -> void {
 }
 
 auto receiver_main(const int frequency, const int offset, const int RF,
-                   const int IF, const int BB, const std::string &device_string)
+                   const int IF, const int BB, const std::string &device_string,
+                   std::shared_ptr<PrometheusExporter> prometheus_exporter)
     -> void {
   float samp_rate = 2000000;
   float bandwidth_sdr = 1000000;
@@ -117,6 +121,32 @@ auto receiver_main(const int frequency, const int offset, const int RF,
   tb->connect(correlate, 0, taggedStreamToPdu, 0);
   tb->msg_connect(taggedStreamToPdu, "pdus", udp_client, "pdus");
 
+  // create blocks to save the power of the current channel if prometheus
+  // exporter is available
+  if (prometheus_exporter) {
+    auto &signal_strength = prometheus_exporter->signal_strength();
+    auto &stream_signal_strength =
+        signal_strength.Add({{"frequency", std::to_string(frequency + offset)},
+                             {"name", "R09 Receiver Average Signal Strength"}});
+
+    auto mag_squared = gr::blocks::complex_to_mag_squared::make();
+    // averaging filter over one second
+    unsigned tap_size = samp_rate / decimation;
+    std::vector<float> averaging_filter(/*count=*/tap_size,
+                                        /*alloc=*/1.0 / tap_size);
+    // do not decimate directly to the final frequency, since there will be some
+    // jitter
+    unsigned decimation = tap_size / 10;
+    auto fir = gr::filter::fir_filter_fff::make(/*decimation=*/decimation,
+                                                averaging_filter);
+    auto populator = gr::prometheus::PrometheusGaugePopulator::make(
+        /*gauge=*/stream_signal_strength);
+
+    tb->connect(xlat, 0, mag_squared, 0);
+    tb->connect(mag_squared, 0, fir, 0);
+    tb->connect(fir, 0, populator, 0);
+  }
+
   print_gnuradio_diagnostics();
 
   tb->start();
@@ -130,7 +160,7 @@ auto main(int argc, char **argv) -> int {
   float frequency;
   float offset;
   int RF, IF, BB;
-  std::string device_string;
+  std::string device_string, prometheus_addr;
 
   const auto frequency_id = pre.register_required_variable<float>("FREQUENCY");
   const auto offset_id = pre.register_required_variable<float>("OFFSET");
@@ -139,6 +169,8 @@ auto main(int argc, char **argv) -> int {
   const auto BB_id = pre.register_variable<int>("BB");
   const auto device_string_id =
       pre.register_variable<std::string>("DEVICE_STRING");
+  const auto prometheus_addr_id =
+      pre.register_variable<std::string>("PROMETHEUS_ADDRESS");
 
   const auto parsed_and_validated_pre = pre.parse_and_validate();
 
@@ -149,15 +181,23 @@ auto main(int argc, char **argv) -> int {
     IF = parsed_and_validated_pre.get_or(IF_id, 0);
     BB = parsed_and_validated_pre.get_or(BB_id, 0);
     device_string = parsed_and_validated_pre.get_or(device_string_id, "");
+    prometheus_addr = parsed_and_validated_pre.get_or(prometheus_addr_id, "");
   } else {
     std::cout << parsed_and_validated_pre.warning_message();
     std::cout << parsed_and_validated_pre.error_message();
 
     return EXIT_FAILURE;
   }
 
+  std::shared_ptr<PrometheusExporter> prometheus_exporter;
+
+  if (!prometheus_addr.empty()) {
+    prometheus_exporter = std::make_shared<PrometheusExporter>(prometheus_addr);
+  }
+
   try {
-    receiver_main(frequency, offset, RF, IF, BB, device_string);
+    receiver_main(frequency, offset, RF, IF, BB, device_string,
+                  prometheus_exporter);
   } catch (std::exception &e) {
     std::cerr << e.what() << std::endl;
     return EXIT_FAILURE;