e2ap,metrics: extending metrics hub so sources and subscrbiers can be mapped to each other in a custom fashion.

Author: yagoda

apps/gnb/gnb.cpp

@@ -410,9 +410,8 @@ int main(int argc, char** argv)
   gnb_console_helper console(*epoll_broker);
   console.on_app_starting();
 
-  std::unique_ptr<metrics_hub>           hub = std::make_unique<metrics_hub>(*workers.metrics_hub_exec.get());
-  std::unique_ptr<e2_du_metrics_manager> e2_du_metric_manager = std::make_unique<e2_du_metrics_manager>();
-  hub->add_subscriber(*e2_du_metric_manager.get());
+  std::unique_ptr<metrics_hub> hub = std::make_unique<metrics_hub>(*workers.metrics_hub_exec.get());
+  std::vector<std::unique_ptr<e2_du_metrics_manager>> e2_du_metric_managers;
 
   // Create NGAP adapter.
   std::unique_ptr<srsran::srs_cu_cp::ngap_network_adapter> ngap_adapter =
@@ -508,7 +507,7 @@ int main(int argc, char** argv)
                                                           *mac_p,
                                                           console,
                                                           e2_gw,
-                                                          *e2_du_metric_manager.get(),
+                                                          e2_du_metric_managers,
                                                           *hub);
 
   for (unsigned sector_id = 0, sector_end = du_inst.size(); sector_id != sector_end; ++sector_id) {

apps/gnb/gnb_du_factory.cpp

@@ -52,24 +52,44 @@ static du_low_configuration create_du_low_config(const gnb_appconfig&
   return du_lo_cfg;
 }
 
-std::vector<std::unique_ptr<du>> srsran::make_gnb_dus(const gnb_appconfig&                  gnb_cfg,
-                                                      worker_manager&                       workers,
-                                                      upper_phy_rg_gateway&                 rg_gateway,
-                                                      upper_phy_rx_symbol_request_notifier& rx_symbol_request_notifier,
-                                                      srs_du::f1c_connection_client&        f1c_client_handler,
-                                                      srs_du::f1u_du_gateway&               f1u_gw,
-                                                      timer_manager&                        timer_mng,
-                                                      mac_pcap&                             mac_p,
-                                                      gnb_console_helper&                   console_helper,
-                                                      e2_connection_client&                 e2_client_handler,
-                                                      e2_du_metrics_manager&                e2_du_metric_manager,
-                                                      metrics_hub&                          metrics_hub)
+std::vector<std::unique_ptr<du>>
+srsran::make_gnb_dus(const gnb_appconfig&                                 gnb_cfg,
+                     worker_manager&                                      workers,
+                     upper_phy_rg_gateway&                                rg_gateway,
+                     upper_phy_rx_symbol_request_notifier&                rx_symbol_request_notifier,
+                     srs_du::f1c_connection_client&                       f1c_client_handler,
+                     srs_du::f1u_du_gateway&                              f1u_gw,
+                     timer_manager&                                       timer_mng,
+                     mac_pcap&                                            mac_p,
+                     gnb_console_helper&                                  console_helper,
+                     e2_connection_client&                                e2_client_handler,
+                     std::vector<std::unique_ptr<e2_du_metrics_manager>>& e2_du_metric_managers,
+                     metrics_hub&                                         metrics_hub)
 {
   // DU cell config
   std::vector<du_cell_config> du_cells = generate_du_cell_config(gnb_cfg);
   console_helper.set_cells(du_cells);
-  metrics_hub.add_subscriber(console_helper.get_metrics_notifier());
 
+  // Set up metrics hub with DU sources and e2 subscribers if enabled.
+  for (unsigned i = 0; i < gnb_cfg.cells_cfg.size(); i++) {
+    std::string source_name = "DU " + std::to_string(i);
+    unsigned    source_idx  = metrics_hub.add_source(source_name);
+    if (gnb_cfg.e2_cfg.enable_du_e2) {
+      e2_du_metric_managers.push_back(std::make_unique<e2_du_metrics_manager>());
+      auto sub = metrics_hub.add_subscriber(*e2_du_metric_managers.back().get());
+      metrics_hub.connect_subscriber_to_source(source_idx, sub);
+    }
+  }
+  // This source will aggregate the metrics from all DU sources.
+  unsigned console_source_agg_idx = metrics_hub.add_source("console aggregator");
+  auto console_agg_subscriber = metrics_hub.add_subscriber(*metrics_hub.get_source_notifier(console_source_agg_idx));
+  // Connecting all DU metric sources to the console subscriber via the aggregator source.
+  for (unsigned i = 0; i < du_cells.size(); i++) {
+    metrics_hub.connect_subscriber_to_source(i, console_agg_subscriber);
+  }
+  // Adding console as a subscriber to metrics_hub with the console aggregator as a source.
+  auto console_subscriber = metrics_hub.add_subscriber(console_helper.get_metrics_notifier());
+  metrics_hub.connect_subscriber_to_source(console_source_agg_idx, console_subscriber);
   std::vector<std::unique_ptr<du>> du_insts;
   for (unsigned i = 0, e = du_cells.size(); i != e; ++i) {
     // Create a gNB config with one cell.
@@ -107,12 +127,12 @@ std::vector<std::unique_ptr<du>> srsran::make_gnb_dus(const gnb_appconfig&
     du_hi_cfg.gnb_du_name                    = fmt::format("srsdu{}", du_hi_cfg.gnb_du_id);
     du_hi_cfg.du_bind_addr                   = {fmt::format("127.0.0.{}", du_hi_cfg.gnb_du_id)};
     du_hi_cfg.mac_cfg                        = generate_mac_expert_config(gnb_cfg);
-    du_hi_cfg.metrics_notifier               = &metrics_hub;
+    du_hi_cfg.metrics_notifier               = metrics_hub.get_source_notifier(i);
     du_hi_cfg.sched_cfg                      = generate_scheduler_expert_config(gnb_cfg);
     if (gnb_cfg.e2_cfg.enable_du_e2) {
       du_hi_cfg.e2_client            = &e2_client_handler;
       du_hi_cfg.e2ap_config          = generate_e2_config(gnb_cfg);
-      du_hi_cfg.e2_du_metric_manager = &e2_du_metric_manager;
+      du_hi_cfg.e2_du_metric_manager = &(*(e2_du_metric_managers[i].get()));
     }
     if (gnb_cfg.test_mode_cfg.test_ue.rnti != INVALID_RNTI) {
       du_hi_cfg.test_cfg.test_ue = srs_du::du_test_config::test_ue_config{gnb_cfg.test_mode_cfg.test_ue.rnti,

apps/gnb/gnb_du_factory.h

@@ -16,6 +16,7 @@
 #include "srsran/du/du.h"
 #include "srsran/du/du_cell_config.h"
 #include "srsran/e2/e2_du_metrics_manager.h"
+
 namespace srsran {
 
 class upper_phy_rg_gateway;
@@ -32,17 +33,18 @@ class f1u_du_gateway;
 } // namespace srs_du
 
 /// \brief Instanties a list of Distributed Unit (DU) given a gNB application configuration.
-std::vector<std::unique_ptr<du>> make_gnb_dus(const gnb_appconfig&                  gnb_cfg,
-                                              worker_manager&                       workers,
-                                              upper_phy_rg_gateway&                 rg_gateway,
-                                              upper_phy_rx_symbol_request_notifier& rx_symbol_request_notifier,
-                                              srs_du::f1c_connection_client&        f1c_client_handler,
-                                              srs_du::f1u_du_gateway&               f1u_gw,
-                                              timer_manager&                        timer_mng,
-                                              mac_pcap&                             mac_p,
-                                              gnb_console_helper&                   console_helper,
-                                              e2_connection_client&                 e2_client_handler,
-                                              e2_du_metrics_manager&                e2_du_metric_manager,
-                                              metrics_hub&                          metrics_hub);
+std::vector<std::unique_ptr<du>>
+make_gnb_dus(const gnb_appconfig&                                 gnb_cfg,
+             worker_manager&                                      workers,
+             upper_phy_rg_gateway&                                rg_gateway,
+             upper_phy_rx_symbol_request_notifier&                rx_symbol_request_notifier,
+             srs_du::f1c_connection_client&                       f1c_client_handler,
+             srs_du::f1u_du_gateway&                              f1u_gw,
+             timer_manager&                                       timer_mng,
+             mac_pcap&                                            mac_p,
+             gnb_console_helper&                                  console_helper,
+             e2_connection_client&                                e2_client_handler,
+             std::vector<std::unique_ptr<e2_du_metrics_manager>>& e2_du_metric_managers,
+             metrics_hub&                                         metrics_hub);
 
 } // namespace srsran

apps/gnb/helpers/metrics_hub.cpp

@@ -12,7 +12,10 @@
 
 using namespace srsran;
 
-metrics_hub::metrics_hub(task_executor& du_executor_) : du_executor(du_executor_) {}
+metrics_hub::metrics_hub(task_executor& du_executor_) :
+  logger(srslog::fetch_basic_logger("METRICS HUB")), du_executor(du_executor_)
+{
+}
 void metrics_hub::report_metrics(span<const scheduler_ue_metrics> ue_metrics)
 {
   std::vector<scheduler_ue_metrics> ue_metrics_copy(ue_metrics.begin(), ue_metrics.end());
@@ -23,7 +26,56 @@ void metrics_hub::report_metrics(span<const scheduler_ue_metrics> ue_metrics)
     }
   });
 }
-void metrics_hub::add_subscriber(scheduler_ue_metrics_notifier& subscriber)
+unsigned metrics_hub::add_source(std::string source_name)
+{
+  sources.push_back(std::make_unique<metrics_hub_source>(du_executor, source_name));
+  return sources.size() - 1;
+}
+
+void metrics_hub::connect_subscriber_to_source(unsigned source_idx, scheduler_ue_metrics_notifier* subscriber)
+{
+  if (source_idx >= sources.size() || subscriber == nullptr) {
+    logger.warning("Cannot connect subscriber to source");
+    return;
+  }
+  sources[source_idx]->add_subscriber(*subscriber);
+}
+
+scheduler_ue_metrics_notifier* metrics_hub::add_subscriber(scheduler_ue_metrics_notifier& subscriber)
+{
+  subscribers.push_back(&subscriber);
+  return (subscribers.at(subscribers.size() - 1));
+}
+
+scheduler_ue_metrics_notifier* metrics_hub::get_source_notifier(unsigned idx)
+{
+  if (idx >= sources.size()) {
+    logger.warning("Source index out of range");
+    return nullptr;
+  }
+  return &(*sources[idx]);
+}
+
+scheduler_ue_metrics_notifier* metrics_hub::get_subscriber_notifier(unsigned idx)
+{
+  if (idx >= subscribers.size()) {
+    logger.warning("Subscriber index out of range");
+  }
+  return subscribers[idx];
+}
+
+void metrics_hub::metrics_hub_source::report_metrics(span<const scheduler_ue_metrics> ue_metrics)
+{
+  std::vector<scheduler_ue_metrics> ue_metrics_copy(ue_metrics.begin(), ue_metrics.end());
+  du_executor.execute([this, ue_metrics_copy]() {
+    for (auto& subscriber : subscribers) {
+      span<const scheduler_ue_metrics> ue_metrics_span(ue_metrics_copy);
+      subscriber->report_metrics(ue_metrics_span);
+    }
+  });
+}
+
+void metrics_hub::metrics_hub_source::add_subscriber(scheduler_ue_metrics_notifier& subscriber)
 {
   subscribers.push_back(&subscriber);
 }

apps/gnb/helpers/metrics_hub.h

@@ -21,11 +21,54 @@ class metrics_hub : public scheduler_ue_metrics_notifier
 public:
   metrics_hub(task_executor& du_executor_);
   void report_metrics(span<const scheduler_ue_metrics> ue_metrics) override;
-  void add_subscriber(scheduler_ue_metrics_notifier& subscriber);
+
+  /// \brief  adds a subscriber to the metrics hub.
+  /// \param[in] subscriber reference to the new subscriber.
+  /// \return returns a pointer to the new subscriber.
+  scheduler_ue_metrics_notifier* add_subscriber(scheduler_ue_metrics_notifier& subscriber);
+
+  /// \brief creates a new souce object in the metrics hub.
+  /// \param[in]  gives the source a name.
+  /// \return returns the index of the new source.
+  unsigned add_source(std::string source_name);
+
+  /// \brief connects a subscriber to a source.
+  /// \param[in] index of the source to be connected.
+  /// \param[in] pointer to the subscriber to be connected.
+  void connect_subscriber_to_source(unsigned source_idx, scheduler_ue_metrics_notifier* subscriber);
+
+  /// \brief retrives a pointer to the source notifier at the given index.
+  /// \param[in] index of desired source notifier.
+  /// \return returns a pointer to the source notifier.
+  scheduler_ue_metrics_notifier* get_source_notifier(unsigned idx);
+
+  /// \brief retrives a pointer to the subscriber notifier at the given index.
+  /// \param index of the desired subscriber.
+  /// \return returns a pointer to the subscriber notifier.
+  scheduler_ue_metrics_notifier* get_subscriber_notifier(unsigned idx);
+
+  /// This class is used to create a source object in the metrics hub to which one or more subscribers can be connected.
+  class metrics_hub_source : public scheduler_ue_metrics_notifier
+  {
+  public:
+    metrics_hub_source(task_executor& du_executor_, std::string _source_name) :
+      du_executor(du_executor_), source_name(_source_name)
+    {
+    }
+    void           report_metrics(span<const scheduler_ue_metrics> ue_metrics) override;
+    void           add_subscriber(scheduler_ue_metrics_notifier& subscriber);
+    task_executor& du_executor;
+    std::string    source_name;
+
+  private:
+    std::vector<scheduler_ue_metrics_notifier*> subscribers;
+  };
 
 private:
-  std::vector<scheduler_ue_metrics_notifier*> subscribers;
-  task_executor&                              du_executor;
+  std::vector<scheduler_ue_metrics_notifier*>      subscribers;
+  std::vector<std::unique_ptr<metrics_hub_source>> sources;
+  srslog::basic_logger&                            logger;
+  task_executor&                                   du_executor;
 };
 
 } // namespace srsran

include/srsran/du_high/du_high_configuration.h

@@ -35,7 +35,7 @@ struct du_high_configuration {
   timer_manager*                     timers               = nullptr;
   scheduler_ue_metrics_notifier*     metrics_notifier     = nullptr;
   e2_connection_client*              e2_client            = nullptr;
-  e2_du_metrics_manager*             e2_du_metric_manager = nullptr;
+  e2_du_metrics_interface*           e2_du_metric_manager = nullptr;
   std::string                        gnb_du_name;
   uint64_t                           gnb_du_id;
   transport_layer_address            du_bind_addr;

lib/e2/common/e2_du_metrics_manager.cpp

@@ -29,5 +29,8 @@ void e2_du_metrics_manager::report_metrics(span<const scheduler_ue_metrics> ue_m
 
 void e2_du_metrics_manager::get_metrics(scheduler_ue_metrics& ue_metrics)
 {
+  if (ue_metrics_queue.empty()) {
+    return;
+  }
   ue_metrics = ue_metrics_queue.front();
 }