Merge pull request #11 from tlm-solutions/feature/prometheus-exporter

Prometheus exporter for Layer 2 Lower MAC

.gitignore

@@ -3,3 +3,4 @@
 /cmake-build-release
 .idea
 /result
+/.cache

CMakeLists.txt

@@ -12,11 +12,15 @@ add_executable(tetra-decoder
         src/streaming_ordered_output_thread_pool_executor.cpp
         src/bit_stream_decoder.cpp
         src/iq_stream_decoder.cpp
+        src/prometheus.cpp
+        src/l2/access_assignment_channel.cpp
+        src/l2/broadcast_synchronization_channel.cpp
+        src/l2/logical_link_control.cpp
         src/l2/lower_mac.cpp
         src/l2/lower_mac_coding.cpp
+        src/l2/timebase_counter.cpp
         src/l2/upper_mac.cpp
         src/l2/upper_mac_fragmentation.cpp
-        src/l2/logical_link_control.cpp
         src/l3/mobile_link_entity.cpp
         src/l3/mobile_management.cpp
         src/l3/circuit_mode_control_entity.cpp
@@ -44,14 +48,15 @@ find_package(cxxopts CONFIG REQUIRED)
 find_package(ZLIB REQUIRED)
 find_package(fmt REQUIRED)
 find_package(nlohmann_json REQUIRED)
+find_package(prometheus-cpp CONFIG REQUIRED)
 
 include_directories(${CMAKE_SOURCE_DIR}/include)
 
 if (NOT NIX_BUILD)
 	target_link_libraries(tetra-decoder cxxopts::cxxopts)
 endif()
 
-target_link_libraries(tetra-decoder ZLIB::ZLIB fmt::fmt nlohmann_json::nlohmann_json viterbi)
+target_link_libraries(tetra-decoder ZLIB::ZLIB fmt::fmt nlohmann_json::nlohmann_json viterbi prometheus-cpp::pull)
 target_link_libraries(tetra-viterbi viterbi)
 
 install(TARGETS tetra-decoder DESTINATION bin)

derivation.nix

@@ -5,6 +5,8 @@
 , zlib
 , fmt
 , nlohmann_json
+, prometheus-cpp
+, curlFull
 }:
 clangStdenv.mkDerivation {
   name = "tetra-decoder";
@@ -13,7 +15,13 @@ clangStdenv.mkDerivation {
   src = ./.;
 
   nativeBuildInputs = [ cmake pkg-config fmt ];
-  buildInputs = [ cxxopts zlib nlohmann_json ];
+  buildInputs = [
+    cxxopts
+    zlib
+    nlohmann_json
+    curlFull
+    prometheus-cpp
+  ];
 
   cmakeFlags = [ "-DNIX_BUILD=ON" ];
 

include/decoder.hpp

@@ -40,7 +40,8 @@ class Decoder {
   public:
     Decoder(unsigned int receive_port, unsigned int send_port, bool packed, std::optional<std::string> input_file,
             std::optional<std::string> output_file, bool iq_or_bit_stream,
-            std::optional<unsigned int> uplink_scrambling_code);
+            std::optional<unsigned int> uplink_scrambling_code,
+            std::shared_ptr<PrometheusExporter>& prometheus_exporter);
     ~Decoder();
 
     void main_loop();

include/l2/access_assignment_channel.hpp

@@ -0,0 +1,30 @@
+/*
+ * Copyright (C) 2024 Transit Live Mapping Solutionss
+ * All rights reserved.
+ *
+ * Authors:
+ *   Marenz Schmidl
+ */
+
+#pragma once
+
+#include "burst_type.hpp"
+#include "l2/timebase_counter.hpp"
+#include <cstdint>
+#include <optional>
+#include <ostream>
+#include <vector>
+
+enum DownlinkUsage { CommonControl, Unallocated, AssignedControl, CommonAndAssignedControl, Traffic };
+
+struct AccessAssignmentChannel {
+    DownlinkUsage downlink_usage;
+    std::optional<int> downlink_traffic_usage_marker;
+
+    AccessAssignmentChannel() = delete;
+    AccessAssignmentChannel(BurstType burst_type, const TimebaseCounter& time, const std::vector<uint8_t>& data);
+
+    friend auto operator<<(std::ostream& stream, const AccessAssignmentChannel& aac) -> std::ostream&;
+};
+
+auto operator<<(std::ostream& stream, const AccessAssignmentChannel& aac) -> std::ostream&;

include/l2/broadcast_synchronization_channel.hpp

@@ -0,0 +1,41 @@
+/*
+ * Copyright (C) 2024 Transit Live Mapping Solutions
+ * All rights reserved.
+ *
+ * Authors:
+ *   Marenz Schmidl
+ */
+
+#pragma once
+
+#include "burst_type.hpp"
+#include "l2/timebase_counter.hpp"
+#include <cstdint>
+#include <vector>
+
+struct BroadcastSynchronizationChannel {
+  public:
+    uint8_t system_code = 0;
+    uint32_t color_code = 0;
+    TimebaseCounter time{};
+    uint8_t sharing_mode = 0;
+    uint8_t time_slot_reserved_frames = 0;
+    uint8_t up_lane_dtx = 0;
+    uint8_t frame_18_extension = 0;
+
+    uint32_t scrambling_code = 0;
+
+    uint32_t mobile_country_code = 0;
+    uint32_t mobile_network_code = 0;
+    uint8_t dNwrk_broadcast_broadcast_supported = 0;
+    uint8_t dNwrk_broadcast_enquiry_supported = 0;
+    uint8_t cell_load_ca = 0;
+    uint8_t late_entry_supported = 0;
+
+    BroadcastSynchronizationChannel() = default;
+    BroadcastSynchronizationChannel(const BurstType burst_type, const std::vector<uint8_t>& data);
+
+    friend auto operator<<(std::ostream& stream, const BroadcastSynchronizationChannel& bsc) -> std::ostream&;
+};
+
+auto operator<<(std::ostream& stream, const BroadcastSynchronizationChannel& bsc) -> std::ostream&;

include/l2/lower_mac.hpp

@@ -9,33 +9,187 @@
 
 #pragma once
 
+#include "l2/broadcast_synchronization_channel.hpp"
+#include "l2/timebase_counter.hpp"
 #include <cstdint>
 #include <memory>
+#include <optional>
 #include <vector>
 
 #include <burst_type.hpp>
 #include <l2/upper_mac.hpp>
+#include <prometheus.h>
 #include <reporter.hpp>
 #include <utils/viter_bi_codec.hpp>
 
+/// The class to provide prometheus metrics to the lower mac
+class LowerMacPrometheusCounters {
+  private:
+    /// The family of counters for received bursts
+    prometheus::Family<prometheus::Counter>& burst_received_count_family_;
+    /// The counter for the received ControlUplinkBurst
+    prometheus::Counter& control_uplink_burst_received_count_;
+    /// The counter for the received NormalUplinkBurst
+    prometheus::Counter& normal_uplink_burst_received_count_;
+    /// The counter for the received NormalUplinkBurstSplit
+    prometheus::Counter& normal_uplink_burst_split_received_count_;
+    /// The counter for the received NormalDownlinkBurst
+    prometheus::Counter& normal_downlink_burst_received_count_;
+    /// The counter for the received NormalDownlinkBurstSplit
+    prometheus::Counter& normal_downlink_burst_split_received_count_;
+    /// The counter for the received SynchronizationBurst
+    prometheus::Counter& synchronization_burst_received_count_;
+
+    /// The family of counters for decoding errors on received bursts in the lower MAC
+    prometheus::Family<prometheus::Counter>& burst_lower_mac_decode_error_count_family_;
+    /// The counter for the received ControlUplinkBurst with decode errors
+    prometheus::Counter& control_uplink_burst_lower_mac_decode_error_count_;
+    /// The counter for the received NormalUplinkBurst with decode errors
+    prometheus::Counter& normal_uplink_burst_lower_mac_decode_error_count_;
+    /// The counter for the received NormalUplinkBurstSplit with decode errors
+    prometheus::Counter& normal_uplink_burst_split_lower_mac_decode_error_count_;
+    /// The counter for the received NormalDownlinkBurst with decode errors
+    prometheus::Counter& normal_downlink_burst_lower_mac_decode_error_count_;
+    /// The counter for the received NormalDownlinkBurstSplit with decode errors
+    prometheus::Counter& normal_downlink_burst_split_lower_mac_decode_error_count_;
+    /// The counter for the received SynchronizationBurst with decode errors
+    prometheus::Counter& synchronization_burst_lower_mac_decode_error_count_;
+
+    /// The family of counters for mismatched number of bursts in the downlink lower MAC
+    prometheus::Family<prometheus::Counter>& burst_lower_mac_mismatch_count_family_;
+    /// The counter for the skipped bursts (lost bursts) in the downlink lower MAC
+    prometheus::Counter& lower_mac_burst_skipped_count_;
+    /// The counter for the too many bursts in the downlink lower MAC
+    prometheus::Counter& lower_mac_burst_too_many_count_;
+
+  public:
+    LowerMacPrometheusCounters(std::shared_ptr<PrometheusExporter>& prometheus_exporter)
+        : burst_received_count_family_(prometheus_exporter->burst_received_count())
+        , control_uplink_burst_received_count_(burst_received_count_family_.Add({{"burst_type", "ControlUplinkBurst"}}))
+        , normal_uplink_burst_received_count_(burst_received_count_family_.Add({{"burst_type", "NormalUplinkBurst"}}))
+        , normal_uplink_burst_split_received_count_(
+              burst_received_count_family_.Add({{"burst_type", "NormalUplinkBurstSplit"}}))
+        , normal_downlink_burst_received_count_(
+              burst_received_count_family_.Add({{"burst_type", "NormalDownlinkBurst"}}))
+        , normal_downlink_burst_split_received_count_(
+              burst_received_count_family_.Add({{"burst_type", "NormalDownlinkBurstSplit"}}))
+        , synchronization_burst_received_count_(
+              burst_received_count_family_.Add({{"burst_type", "SynchronizationBurst"}}))
+        , burst_lower_mac_decode_error_count_family_(prometheus_exporter->burst_lower_mac_decode_error_count())
+        , control_uplink_burst_lower_mac_decode_error_count_(
+              burst_lower_mac_decode_error_count_family_.Add({{"burst_type", "ControlUplinkBurst"}}))
+        , normal_uplink_burst_lower_mac_decode_error_count_(
+              burst_lower_mac_decode_error_count_family_.Add({{"burst_type", "NormalUplinkBurst"}}))
+        , normal_uplink_burst_split_lower_mac_decode_error_count_(
+              burst_lower_mac_decode_error_count_family_.Add({{"burst_type", "NormalUplinkBurstSplit"}}))
+        , normal_downlink_burst_lower_mac_decode_error_count_(
+              burst_lower_mac_decode_error_count_family_.Add({{"burst_type", "NormalDownlinkBurst"}}))
+        , normal_downlink_burst_split_lower_mac_decode_error_count_(
+              burst_lower_mac_decode_error_count_family_.Add({{"burst_type", "NormalDownlinkBurstSplit"}}))
+        , synchronization_burst_lower_mac_decode_error_count_(
+              burst_lower_mac_decode_error_count_family_.Add({{"burst_type", "SynchronizationBurst"}}))
+        , burst_lower_mac_mismatch_count_family_(prometheus_exporter->burst_lower_mac_mismatch_count())
+        , lower_mac_burst_skipped_count_(burst_lower_mac_mismatch_count_family_.Add({{"mismatch_type", "Skipped"}}))
+        , lower_mac_burst_too_many_count_(
+              burst_lower_mac_mismatch_count_family_.Add({{"mismatch_type", "Too many"}})){};
+
+    /// This function is called for every burst. It increments the counter associated to the burst type.
+    /// \param burst_type the type of the burst for which to increment the counter
+    /// \param decode_error true if there was an error decoding the packets on the lower mac (crc16)
+    auto increment(BurstType burst_type, bool decode_error) -> void {
+        switch (burst_type) {
+        case BurstType::ControlUplinkBurst:
+            control_uplink_burst_received_count_.Increment();
+            break;
+        case BurstType::NormalUplinkBurst:
+            normal_uplink_burst_received_count_.Increment();
+            break;
+        case BurstType::NormalUplinkBurstSplit:
+            normal_uplink_burst_split_received_count_.Increment();
+            break;
+        case BurstType::NormalDownlinkBurst:
+            normal_downlink_burst_received_count_.Increment();
+            break;
+        case BurstType::NormalDownlinkBurstSplit:
+            normal_downlink_burst_split_received_count_.Increment();
+            break;
+        case BurstType::SynchronizationBurst:
+            synchronization_burst_received_count_.Increment();
+            break;
+        }
+
+        if (decode_error) {
+            switch (burst_type) {
+            case BurstType::ControlUplinkBurst:
+                control_uplink_burst_lower_mac_decode_error_count_.Increment();
+                break;
+            case BurstType::NormalUplinkBurst:
+                normal_uplink_burst_lower_mac_decode_error_count_.Increment();
+                break;
+            case BurstType::NormalUplinkBurstSplit:
+                normal_uplink_burst_split_lower_mac_decode_error_count_.Increment();
+                break;
+            case BurstType::NormalDownlinkBurst:
+                normal_downlink_burst_lower_mac_decode_error_count_.Increment();
+                break;
+            case BurstType::NormalDownlinkBurstSplit:
+                normal_downlink_burst_split_lower_mac_decode_error_count_.Increment();
+                break;
+            case BurstType::SynchronizationBurst:
+                synchronization_burst_lower_mac_decode_error_count_.Increment();
+                break;
+            }
+        }
+    }
+
+    /// This function is called for Synchronization Bursts. It increments the counters for the missmatched received
+    /// burst counts.
+    /// \param current_timestamp the current timestamp of the synchronization burst
+    /// \param expected_timestamp the predicted timestamp of the synchronization burst
+    auto increment(const TimebaseCounter current_timestamp, const TimebaseCounter expected_timestamp) {
+        auto current_timestamp_count = current_timestamp.count();
+        auto expected_timestamp_count = expected_timestamp.count();
+
+        if (current_timestamp_count > expected_timestamp_count) {
+            auto missed_bursts = current_timestamp_count - expected_timestamp_count;
+            lower_mac_burst_skipped_count_.Increment(missed_bursts);
+        } else if (expected_timestamp_count > current_timestamp_count) {
+            auto too_many_bursts = expected_timestamp_count - current_timestamp_count;
+            lower_mac_burst_too_many_count_.Increment(too_many_bursts);
+        }
+    }
+};
+
 class LowerMac {
   public:
-    explicit LowerMac(std::shared_ptr<Reporter> reporter);
+    LowerMac() = delete;
+    LowerMac(std::shared_ptr<Reporter> reporter, std::shared_ptr<PrometheusExporter>& prometheus_exporter,
+             std::optional<uint32_t> scrambling_code = std::nullopt);
     ~LowerMac() = default;
 
-    // does the signal processing and then returns a list of function that need to be executed for data to be passed to
-    // upper mac sequentially.
-    // TODO: this is currently only done for uplink bursts
-    // Downlink burst get processed and passed to upper mac directly
+    // does the signal processing and then returns a list of function that need to be executed for data to be passed
+    // to upper mac sequentially.
+    [[nodiscard]] auto processChannels(const std::vector<uint8_t>& frame, BurstType burst_type,
+                                       const BroadcastSynchronizationChannel& bsc)
+        -> std::vector<std::function<void()>>;
+
+    /// handles the decoding of the synchronization bursts and once synchronized passes the data to the decoding of the
+    /// channels. keeps track of the current network time
     [[nodiscard]] auto process(const std::vector<uint8_t>& frame, BurstType burst_type)
         -> std::vector<std::function<void()>>;
-    void set_scrambling_code(unsigned int scrambling_code) { upper_mac_->set_scrambling_code(scrambling_code); };
 
   private:
     std::shared_ptr<Reporter> reporter_{};
     std::shared_ptr<ViterbiCodec> viter_bi_codec_1614_{};
     std::shared_ptr<UpperMac> upper_mac_{};
 
+    std::unique_ptr<LowerMacPrometheusCounters> metrics_;
+
+    /// The last received synchronization burst.
+    /// This include the current scrambling code. Set by Synchronization Burst on downlink or injected from the side for
+    /// uplink processing, as we decouple it from the downlink for data/control packets.
+    std::optional<BroadcastSynchronizationChannel> sync_;
+
     static auto descramble(const uint8_t* data, uint8_t* res, std::size_t len, uint32_t scramblingCode) noexcept
         -> void;
     static auto deinterleave(const uint8_t* data, uint8_t* res, std::size_t K, std::size_t a) noexcept -> void;

include/l2/timebase_counter.hpp

@@ -0,0 +1,48 @@
+/*
+ * Copyright (C) 2024 Transit Live Mapping Solutions
+ * All rights reserved.
+ *
+ * Authors:
+ *   Marenz Schmidl
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <ostream>
+#include <tuple>
+
+class TimebaseCounter {
+  private:
+    uint16_t time_slot_ = 1;
+    uint16_t frame_number_ = 1;
+    uint16_t multi_frame_number_ = 1;
+
+  public:
+    TimebaseCounter() = default;
+    TimebaseCounter(const uint16_t time_slot, const uint16_t frame_number, const uint16_t multi_frame_number)
+        : time_slot_(time_slot)
+        , frame_number_(frame_number)
+        , multi_frame_number_(multi_frame_number){};
+
+    [[nodiscard]] auto time_slot() const noexcept -> uint16_t { return time_slot_; };
+    [[nodiscard]] auto frame_number() const noexcept -> uint16_t { return frame_number_; };
+    [[nodiscard]] auto multi_frame_number() const noexcept -> uint16_t { return multi_frame_number_; };
+    /// convert the slot and frame numbers into a single value
+    [[nodiscard]] auto count() const noexcept -> unsigned {
+        return (time_slot_ - 1) + 4 * (frame_number_ - 1) + 4 * 18 * (multi_frame_number_ - 1);
+    }
+
+    [[nodiscard]] auto operator==(const TimebaseCounter& other) const noexcept -> bool {
+        return std::tie(time_slot_, frame_number_, multi_frame_number_) ==
+               std::tie(other.time_slot_, other.frame_number_, other.multi_frame_number_);
+    };
+
+    [[nodiscard]] auto operator!=(const TimebaseCounter& other) const noexcept -> bool { return !(*this == other); }
+
+    auto increment() noexcept -> void;
+
+    friend auto operator<<(std::ostream& stream, const TimebaseCounter& tc) -> std::ostream&;
+};
+
+auto operator<<(std::ostream& stream, const TimebaseCounter& tc) -> std::ostream&;