Counters to shared memory

autotest/autotest.cpp

@@ -78,6 +78,11 @@ eResult tAutotest::init(const std::string& binaryPath,
 		return ret;
 	}
 
+	if (auto ret = common::sdp::SdpClient::ReadSharedMemoryData(sdp_data, true); ret != eResult::success)
+	{
+		return ret;
+	}
+
 	return eResult::success;
 }
 
@@ -1746,7 +1751,7 @@ bool tAutotest::step_memorize_counter_value(const YAML::Node& yamlStep)
 
 	uint32_t coreId = std::stoi(yamlStep.as<std::string>().substr(delim_pos + 1));
 
-	const auto response = dataPlane.get_counter_by_name({counter_name, coreId});
+	const auto response = common::sdp::SdpClient::GetCounterByName(sdp_data, counter_name, coreId);
 
 	if (response.empty())
 	{
@@ -1805,7 +1810,7 @@ bool tAutotest::step_diff_with_kept_counter_value(const YAML::Node& yamlStep)
 		return false;
 	}
 
-	const auto response = dataPlane.get_counter_by_name({counter_name, coreId});
+	const auto response = common::sdp::SdpClient::GetCounterByName(sdp_data, counter_name, coreId);
 
 	if (response.empty())
 	{

autotest/autotest.h

@@ -11,6 +11,7 @@
 #include "common/icontrolplane.h"
 #include "common/idataplane.h"
 #include "common/result.h"
+#include "common/sdpclient.h"
 
 namespace nAutotest
 {
@@ -96,6 +97,7 @@ class tAutotest
 
 	interface::dataPlane dataPlane;
 	interface::controlPlane controlPlane;
+	common::sdp::DataPlaneInSharedMemory sdp_data;
 
 	common::idp::getConfig::response dataPlaneConfig;
 	common::idp::get_shm_info::response dataPlaneSharedMemory;

autotest/units/001_one_port/075_counters_shared_memory/autotest.yaml

@@ -0,0 +1,5 @@
+steps:
+- cli: "telegraf counters"
+- cli: "telegraf bus"
+- cli: "bus errors"
+- cli: "bus requests"

autotest/units/001_one_port/075_counters_shared_memory/controlplane.conf

@@ -0,0 +1,32 @@
+{
+  "modules": {
+    "lp0.100": {
+      "type": "logicalPort",
+      "physicalPort": "kni0",
+      "vlanId": "100",
+      "macAddress": "00:11:22:33:44:55",
+      "nextModule": "vrf0"
+    },
+    "lp0.200": {
+      "type": "logicalPort",
+      "physicalPort": "kni0",
+      "vlanId": "200",
+      "macAddress": "00:11:22:33:44:55",
+      "nextModule": "vrf0"
+    },
+    "vrf0": {
+      "type": "route",
+      "interfaces": {
+        "kni0.100": {
+          "nextModule": "lp0.100"
+        },
+        "kni0.200": {
+          "ipv4Prefix": "200.0.0.2/24",
+          "neighborIPv4Address": "200.0.0.1",
+          "neighborMacAddress": "00:00:00:11:11:11",
+          "nextModule": "lp0.200"
+        }
+      }
+    }
+  }
+}

autotest/yanet-autotest-run.py

@@ -86,6 +86,7 @@ def start(self, dataplane_conf_path, units):
         os.makedirs("/run/yanet", exist_ok=True)
 
         self.run_dataplane(dataplane_conf_path)
+        time.sleep(5)
         self.run_controlplane()
         self.run_autotest(units)
 

cli/bus.h

@@ -0,0 +1,160 @@
+#pragma once
+
+#include "helper.h"
+#include "influxdb_format.h"
+
+namespace bus
+{
+
+using bus_request_info = std::tuple<std::string, uint64_t, uint64_t>;
+
+inline std::vector<bus_request_info> get_bus_requests(common::sdp::DataPlaneInSharedMemory& sdp_data)
+{
+	auto [requests, errors, durations] = sdp_data.BuffersBus();
+	(void)errors;
+
+	std::map<common::idp::requestType, std::string> names = {
+	        {common::idp::requestType::updateGlobalBase, "updateGlobalBase"},
+	        {common::idp::requestType::updateGlobalBaseBalancer, "updateGlobalBaseBalancer"},
+	        {common::idp::requestType::getGlobalBase, "getGlobalBase"},
+	        {common::idp::requestType::getWorkerStats, "getWorkerStats"},
+	        {common::idp::requestType::getSlowWorkerStats, "getSlowWorkerStats"},
+	        {common::idp::requestType::get_worker_gc_stats, "get_worker_gc_stats"},
+	        {common::idp::requestType::get_dregress_counters, "get_dregress_counters"},
+	        {common::idp::requestType::get_ports_stats, "get_ports_stats"},
+	        {common::idp::requestType::get_ports_stats_extended, "get_ports_stats_extended"},
+	        {common::idp::requestType::getControlPlanePortStats, "getControlPlanePortStats"},
+	        {common::idp::requestType::getPortStatsEx, "getPortStatsEx"},
+	        {common::idp::requestType::getFragmentationStats, "getFragmentationStats"},
+	        {common::idp::requestType::getFWState, "getFWState"},
+	        {common::idp::requestType::getFWStateStats, "getFWStateStats"},
+	        {common::idp::requestType::clearFWState, "clearFWState"},
+	        {common::idp::requestType::getConfig, "getConfig"},
+	        {common::idp::requestType::getErrors, "getErrors"},
+	        {common::idp::requestType::getReport, "getReport"},
+	        {common::idp::requestType::lpm4LookupAddress, "lpm4LookupAddress"},
+	        {common::idp::requestType::lpm6LookupAddress, "lpm6LookupAddress"},
+	        {common::idp::requestType::nat64stateful_state, "nat64stateful_state"},
+	        {common::idp::requestType::balancer_connection, "balancer_connection"},
+	        {common::idp::requestType::balancer_service_connections, "balancer_service_connections"},
+	        {common::idp::requestType::balancer_real_connections, "balancer_real_connections"},
+	        {common::idp::requestType::limits, "limits"},
+	        {common::idp::requestType::samples, "samples"},
+	        {common::idp::requestType::debug_latch_update, "debug_latch_update"},
+	        {common::idp::requestType::unrdup_vip_to_balancers, "unrdup_vip_to_balancers"},
+	        {common::idp::requestType::update_vip_vport_proto, "update_vip_vport_proto"},
+	        {common::idp::requestType::version, "version"},
+	        {common::idp::requestType::get_shm_info, "get_shm_info"},
+	        {common::idp::requestType::get_shm_tsc_info, "get_shm_tsc_info"},
+	        {common::idp::requestType::set_shm_tsc_state, "set_shm_tsc_state"},
+	        {common::idp::requestType::dump_physical_port, "dump_physical_port"},
+	        {common::idp::requestType::balancer_state_clear, "balancer_state_clear"},
+	        {common::idp::requestType::neighbor_show, "neighbor_show"},
+	        {common::idp::requestType::neighbor_insert, "neighbor_insert"},
+	        {common::idp::requestType::neighbor_remove, "neighbor_remove"},
+	        {common::idp::requestType::neighbor_clear, "neighbor_clear"},
+	        {common::idp::requestType::neighbor_flush, "neighbor_flush"},
+	        {common::idp::requestType::neighbor_update_interfaces, "neighbor_update_interfaces"},
+	        {common::idp::requestType::neighbor_stats, "neighbor_stats"},
+	        {common::idp::requestType::memory_manager_update, "memory_manager_update"},
+	        {common::idp::requestType::memory_manager_stats, "memory_manager_stats"}};
+
+	std::vector<bus_request_info> result;
+	for (uint32_t index = 0; index < (uint32_t)common::idp::requestType::size; ++index)
+	{
+		if ((requests[index] != 0) || (durations[index] != 0))
+		{
+			const auto& iter = names.find(static_cast<common::idp::requestType>(index));
+			result.emplace_back((iter != names.end() ? iter->second : "unknown"), requests[index], durations[index]);
+		}
+	}
+
+	return result;
+}
+
+inline void bus_requests()
+{
+	common::sdp::DataPlaneInSharedMemory sdp_data;
+	OpenSharedMemoryDataplaneBuffers(sdp_data, false);
+	auto requests = get_bus_requests(sdp_data);
+
+	table_t table;
+	table.insert("request", "count", "duration_ms");
+	for (const auto& [request, count, duration] : requests)
+	{
+		if ((count != 0) || (duration != 0))
+		{
+			table.insert(request, count, duration);
+		}
+	}
+
+	table.print();
+}
+
+inline std::vector<std::pair<std::string, uint64_t>> get_bus_errors(const common::sdp::DataPlaneInSharedMemory& sdp_data)
+{
+	auto [requests, errors, durations] = sdp_data.BuffersBus();
+	(void)requests;
+	(void)durations;
+
+	std::map<common::idp::errorType, std::string> names = {
+	        {common::idp::errorType::busRead, "busRead"},
+	        {common::idp::errorType::busWrite, "busWrite"},
+	        {common::idp::errorType::busParse, "busParse"},
+	};
+
+	std::vector<std::pair<std::string, uint64_t>> result;
+	for (uint32_t index = 0; index < (uint32_t)common::idp::errorType::size; ++index)
+	{
+		const auto& iter = names.find(static_cast<common::idp::errorType>(index));
+		result.emplace_back((iter != names.end() ? iter->second : "unknown"), errors[index]);
+	}
+
+	return result;
+}
+
+inline void bus_errors()
+{
+	common::sdp::DataPlaneInSharedMemory sdp_data;
+	OpenSharedMemoryDataplaneBuffers(sdp_data, false);
+	auto errors = get_bus_errors(sdp_data);
+
+	table_t table;
+	table.insert("error", "count");
+	for (const auto& [error, count] : errors)
+	{
+		table.insert(error, count);
+	}
+
+	table.print();
+}
+
+inline void bus_telegraf()
+{
+	common::sdp::DataPlaneInSharedMemory sdp_data;
+	OpenSharedMemoryDataplaneBuffers(sdp_data, false);
+
+	auto errors = get_bus_errors(sdp_data);
+	std::vector<influxdb_format::value_t> infl_errors;
+	for (const auto& [error, count] : errors)
+	{
+		infl_errors.emplace_back(error.data(), count);
+	}
+	influxdb_format::print("bus_errors", {}, infl_errors);
+
+	auto requests = get_bus_requests(sdp_data);
+	if (!requests.empty())
+	{
+		std::vector<influxdb_format::value_t> infl_counts;
+		std::vector<influxdb_format::value_t> infl_durations;
+		for (const auto& [request, count, duration] : requests)
+		{
+			infl_counts.emplace_back(request.data(), count);
+			infl_durations.emplace_back(request.data(), duration);
+		}
+		influxdb_format::print("bus_counts", {}, infl_counts);
+		influxdb_format::print("bus_durations", {}, infl_durations);
+	}
+}
+
+} // namespace bus

cli/develop.h

@@ -9,6 +9,7 @@
 
 #include "common/icontrolplane.h"
 #include "common/idataplane.h"
+#include "common/sdpclient.h"
 #include "common/tsc_deltas.h"
 
 #include "helper.h"
@@ -151,7 +152,7 @@ void counter(const uint32_t& counter_id,
 		}
 	}
 
-	const auto response = dataplane.getCounters(counter_ids);
+	const auto response = common::sdp::SdpClient::GetCounters(counter_ids);
 
 	table_t table;
 	table.insert("counter_id",

cli/helper.h

@@ -12,6 +12,7 @@
 
 #include <nlohmann/json.hpp>
 
+#include "common/sdpclient.h"
 #include "converter.h"
 
 template<typename type>
@@ -466,3 +467,12 @@ class table_t
 	std::vector<std::vector<std::string>> table;
 	std::vector<uint32_t> columnLengths;
 };
+
+void OpenSharedMemoryDataplaneBuffers(common::sdp::DataPlaneInSharedMemory& sdp_data, bool open_workers_data)
+{
+	if (common::sdp::SdpClient::ReadSharedMemoryData(sdp_data, open_workers_data) != eResult::success)
+	{
+		YANET_LOG_ERROR("Error openning shared memory dataplane buffers\n");
+		std::exit(1);
+	}
+}

cli/main.cpp

@@ -5,6 +5,7 @@
 
 #include "acl.h"
 #include "balancer.h"
+#include "bus.h"
 #include "config.h"
 #include "convert.h"
 #include "develop.h"
@@ -127,12 +128,18 @@ std::vector<std::tuple<std::string,
                     {"telegraf balancer service", "", [](const auto& args) { call(telegraf::balancer::service, args); }},
                     {"telegraf other", "", [](const auto& args) { call(telegraf::other, args); }},
                     {"telegraf tun64", "", [](const auto& args) { call(telegraf::mappings, args); }},
+                    {"telegraf counters", "", [](const auto& args) { call(telegraf::main_counters, args); }},
+                    {"telegraf bus", "", [](const auto& args) { call(bus::bus_telegraf, args); }},
                     {},
                     {"reload", "", [](const auto& args) { call(config::reload, args); }},
                     {"version", "", [](const auto& args) { call(show::version, args); }},
                     {"latch update dataplane", "<latch name> <state>", [](const auto& args) { call(latch::dataplane_update, args); }},
                     {"counter", "[counter_name] <core_id>", [](const auto& args) { call(show::counter_by_name, args); }},
                     {"counters stat", "", [](const auto& args) { call(show::counters_stat, args); }},
+
+                    {"bus requests", "", [](const auto& args) { call(bus::bus_requests, args); }},
+                    {"bus errors", "", [](const auto& args) { call(bus::bus_errors, args); }},
+
                     {"latch update dataplane", "<latch name> <state>", [](const auto& args) { call(latch::dataplane_update, args); }},
                     {},
                     {"convert logical_module", "", [](const auto& args) { call(convert::logical_module, args); }}};

cli/show.h

@@ -6,6 +6,7 @@
 
 #include "common/icontrolplane.h"
 #include "common/idataplane.h"
+#include "common/sdpclient.h"
 #include "common/tsc_deltas.h"
 #include "common/version.h"
 
@@ -800,9 +801,7 @@ inline void version()
 inline void counter_by_name(std::string counter_name,
                             const std::optional<tCoreId>& core_id)
 {
-	interface::dataPlane dataplane;
-
-	const auto response = dataplane.get_counter_by_name({counter_name, core_id});
+	const auto response = common::sdp::SdpClient::GetCounterByName(counter_name, core_id);
 
 	if (response.empty())
 	{

cli/telegraf.h

@@ -3,6 +3,7 @@
 #include "common/counters.h"
 #include "common/icontrolplane.h"
 #include "common/idataplane.h"
+#include "common/sdpclient.h"
 
 #include "helper.h"
 #include "influxdb_format.h"
@@ -100,7 +101,7 @@ void unsafe()
 	const auto [responseWorkers, responseWorkerGCs, responseSlowWorkerHashtableGC, responseFragmentation, responseFWState, responseTun64, response_nat64stateful, responseControlplane] = controlplane.telegraf_unsafe();
 	const auto& [responseSlowWorker, hashtable_gc] = responseSlowWorkerHashtableGC;
 
-	const auto static_counters = dataplane.getCounters(vector_range(0, (tCounterId)common::globalBase::static_counter_type::size));
+	const auto static_counters = common::sdp::SdpClient::GetCounters(vector_range(0, (tCounterId)common::globalBase::static_counter_type::size));
 	const auto neighbor_stats = dataplane.neighbor_stats();
 	const auto memory_stats = dataplane.memory_manager_stats();
 	const auto& [memory_groups, memory_objects] = memory_stats;
@@ -701,4 +702,34 @@ void service()
 
 }
 
+void main_counters()
+{
+	common::sdp::DataPlaneInSharedMemory sdp_data;
+	OpenSharedMemoryDataplaneBuffers(sdp_data, true);
+
+	for (const auto& [coreId, worker_info] : sdp_data.workers)
+	{
+		std::vector<influxdb_format::value_t> values;
+		uint64_t* buffer = common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer,
+		                                                       sdp_data.metadata_worker.start_counters);
+		for (const auto& [name, index] : sdp_data.metadata_worker.counter_positions)
+		{
+			values.emplace_back(name.data(), buffer[index]);
+		}
+		influxdb_format::print("worker", {{"coreId", coreId}}, values);
+	}
+
+	for (const auto& [coreId, worker_info] : sdp_data.workers_gc)
+	{
+		std::vector<influxdb_format::value_t> values;
+		uint64_t* buffer = common::sdp::ShiftBuffer<uint64_t*>(worker_info.buffer,
+		                                                       sdp_data.metadata_worker.start_counters);
+		for (const auto& [name, index] : sdp_data.metadata_worker_gc.counter_positions)
+		{
+			values.emplace_back(name.data(), buffer[index]);
+		}
+		influxdb_format::print("worker_gc", {{"coreId", coreId}}, values);
+	}
+}
+
 }