The fourth part of the changes is to transfer counters to shared memory.

Added to the cli: - Getting bus statistics. - Getting the values of the main counters (having fixed names).
yanet-platform · Jul 12, 2024 · 75da7ac · 75da7ac
1 parent 8e5b88c
commit 75da7ac
Show file tree

Hide file tree

Showing 9 changed files with 249 additions and 1 deletion.
diff --git a/autotest/units/001_one_port/075_counters_shared_memory/autotest.yaml b/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"
diff --git a/autotest/units/001_one_port/075_counters_shared_memory/controlplane.conf b/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"
+        }
+      }
+    }
+  }
+}
diff --git a/autotest/yanet-autotest-run.py b/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)
 

diff --git a/cli/bus.h b/cli/bus.h
@@ -0,0 +1,158 @@
+#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)
+{
+	uint64_t* requests = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_requests);
+	uint64_t* durations = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_durations);
+
+	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(common::sdp::DataPlaneInSharedMemory& sdp_data)
+{
+	uint64_t* buffer = common::sdp::ShiftBuffer<uint64_t*>(sdp_data.dataplane_data, sdp_data.metadata_bus.start_bus_errors);
+
+	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"), buffer[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
diff --git a/cli/helper.h b/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);
+	}
+}
diff --git a/cli/main.cpp b/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,11 +128,15 @@ 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); }},
+                    {"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); }}};

diff --git a/cli/telegraf.h b/cli/telegraf.h
@@ -694,4 +694,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);
+	}
+}
+
 }
diff --git a/common/idataplane.h b/common/idataplane.h
@@ -267,7 +267,12 @@ class dataPlane
 		int ret = connect(clientSocket, (struct sockaddr*)&address, sizeof(address));
 		if (ret == -1)
 		{
-			throw std::string("connect(): ") + strerror(errno);
+			int error = errno;
+			YANET_LOG_ERROR("Error connect to socket %s, error: %d - %s\n",
+			                common::idp::socketPath,
+			                error,
+			                strerror(error));
+			throw std::string("connect(): ") + strerror(error);
 		}
 	}
 

diff --git a/dataplane/bus.cpp b/dataplane/bus.cpp
@@ -404,6 +404,8 @@ void cBus::clientThread(int clientSocket)
 
 		std::chrono::duration<double> duration = std::chrono::system_clock::now() - startTime;
 
+		// The duration time is measured in milliseconds
+		stats.durations[(uint32_t)type] += static_cast<uint64_t>(1000 * duration.count());
 		YANET_LOG_DEBUG("request type %d processed - %.3f sec\n",
 		                (int)type,
 		                duration.count());