diff --git a/src/fne/network/MetadataNetwork.cpp b/src/fne/network/MetadataNetwork.cpp index 1124e8c4..a647e323 100644 --- a/src/fne/network/MetadataNetwork.cpp +++ b/src/fne/network/MetadataNetwork.cpp @@ -20,6 +20,7 @@ using namespace network; using namespace network::callhandler; #include +#include // --------------------------------------------------------------------------- // Public Class Members @@ -36,7 +37,15 @@ MetadataNetwork::MetadataNetwork(HostFNE* host, TrafficNetwork* trafficNetwork, m_status(NET_STAT_INVALID), m_peerReplicaActPkt(), m_peerTreeListPkt(), - m_threadPool(workerCnt, "meta") + m_threadPool(workerCnt, "meta"), + m_healthDataPlaneEpoch(0U), + m_healthLastDataPlaneTick(0U), + m_healthPacketsReceived(0U), + m_healthLastPacketReceived(0U), + m_healthTasksSubmitted(0U), + m_healthTasksCompleted(0U), + m_healthLastTaskProgress(0U), + m_healthTaskEnqueueFailures(0U) { assert(trafficNetwork != nullptr); assert(host != nullptr); @@ -59,6 +68,10 @@ void MetadataNetwork::setPresharedKey(const uint8_t* presharedKey) void MetadataNetwork::processNetwork() { + const uint64_t healthNow = std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); + m_healthDataPlaneEpoch.fetch_add(1U, std::memory_order_relaxed); + m_healthLastDataPlaneTick.store(healthNow, std::memory_order_relaxed); + if (m_status != NET_STAT_MST_RUNNING) { return; } @@ -72,6 +85,9 @@ void MetadataNetwork::processNetwork() // read message UInt8Array buffer = m_frameQueue->read(length, address, addrLen, &rtpHeader, &fneHeader); if (length > 0) { + m_healthPacketsReceived.fetch_add(1U, std::memory_order_relaxed); + m_healthLastPacketReceived.store(healthNow, std::memory_order_relaxed); + if (m_debug) Utils::dump(1U, "MetadataNetwork::processNetwork(), Network Message", buffer.get(), length); @@ -86,14 +102,25 @@ void MetadataNetwork::processNetwork() req->addrLen = addrLen; req->rtpHeader = rtpHeader; req->fneHeader = fneHeader; + req->healthMetadataTracked = true; req->length = length; req->buffer = new uint8_t[length]; ::memcpy(req->buffer, buffer.get(), length); + const uint64_t pendingBefore = m_healthTasksSubmitted.load(std::memory_order_relaxed) - m_healthTasksCompleted.load(std::memory_order_relaxed); + if (pendingBefore == 0U) + m_healthLastTaskProgress.store(healthNow, std::memory_order_relaxed); + m_healthTasksSubmitted.fetch_add(1U, std::memory_order_relaxed); + if (!m_threadPool.enqueue(new_pooltask(taskNetworkRx, req))) { + m_healthTaskEnqueueFailures.fetch_add(1U, std::memory_order_relaxed); + m_healthTasksCompleted.fetch_add(1U, std::memory_order_relaxed); + m_healthLastTaskProgress.store(healthNow, std::memory_order_relaxed); + LogError(LOG_NET, "Failed to task enqueue network packet request, peerId = %u, %s:%u", peerId, udp::Socket::address(address).c_str(), udp::Socket::port(address)); + if (req != nullptr) { if (req->buffer != nullptr) delete[] req->buffer; @@ -269,6 +296,18 @@ void MetadataNetwork::taskNetworkRx(NetPacketRequest* req) return; } + const bool healthMetadataTracked = req->healthMetadataTracked; + + // defines a lambda function to mark the completion of a health task + const auto healthTaskComplete = [mdNetwork, healthMetadataTracked]() { + if (!healthMetadataTracked) + return; + + mdNetwork->m_healthTasksCompleted.fetch_add(1U, std::memory_order_relaxed); + const uint64_t now = std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); + mdNetwork->m_healthLastTaskProgress.store(now, std::memory_order_relaxed); + }; + if (req == nullptr) return; @@ -298,9 +337,12 @@ void MetadataNetwork::taskNetworkRx(NetPacketRequest* req) if (func == NET_FUNC::TRANSFER && (subFunc == NET_SUBFUNC::TRANSFER_SUBFUNC_ACTIVITY || subFunc == NET_SUBFUNC::TRANSFER_SUBFUNC_DIAG) && !(req->length > TRANSFER_PCKT_HDR_LEN && req->length <= static_cast(DATA_PACKET_LENGTH))) { LogWarning(LOG_MASTER, "PEER %u malformed FNE transfer packet, subfunc = $%02X, length = %d", peerId, subFunc, req->length); + if (req->buffer != nullptr) delete[] req->buffer; delete req; + + healthTaskComplete(); return; } @@ -313,5 +355,7 @@ void MetadataNetwork::taskNetworkRx(NetPacketRequest* req) if (req->buffer != nullptr) delete[] req->buffer; delete req; + + healthTaskComplete(); } } diff --git a/src/fne/network/MetadataNetwork.h b/src/fne/network/MetadataNetwork.h index 980d0060..6ecd11bf 100644 --- a/src/fne/network/MetadataNetwork.h +++ b/src/fne/network/MetadataNetwork.h @@ -25,6 +25,7 @@ #include #include #include +#include // --------------------------------------------------------------------------- // Class Prototypes @@ -35,6 +36,7 @@ class FNETestHooks; #endif class HOST_SW_API HostFNE; +namespace fne_restapi { class HOST_SW_API RESTAPI; } namespace network { @@ -111,6 +113,8 @@ namespace network TrafficNetwork* m_trafficNetwork; HostFNE* m_host; + friend class ::fne_restapi::RESTAPI; + std::string m_address; uint16_t m_port; @@ -153,6 +157,18 @@ namespace network ThreadPool m_threadPool; + // passive health telemetry -- these counters are intentionally lock-free so + // the REST thread can diagnose a stalled worker without waiting + // for a lock held by that worker + std::atomic m_healthDataPlaneEpoch; + std::atomic m_healthLastDataPlaneTick; + std::atomic m_healthPacketsReceived; + std::atomic m_healthLastPacketReceived; + std::atomic m_healthTasksSubmitted; + std::atomic m_healthTasksCompleted; + std::atomic m_healthLastTaskProgress; + std::atomic m_healthTaskEnqueueFailures; + /** * @brief Finds a packet buffer entry in the map. * @param pktMap Instance of the PacketBufferMap class. diff --git a/src/fne/network/TrafficNetwork.cpp b/src/fne/network/TrafficNetwork.cpp index 6d206504..a4c9798f 100644 --- a/src/fne/network/TrafficNetwork.cpp +++ b/src/fne/network/TrafficNetwork.cpp @@ -157,6 +157,16 @@ TrafficNetwork::TrafficNetwork(HostFNE* host, const std::string& address, uint16 m_jitterMaxWait(40000U), m_threadPool(workerCnt, "fne"), m_metadataUpdateThreadPool(workerCnt / 2U, "mupdt"), + m_healthDataPlaneEpoch(0U), + m_healthLastDataPlaneTick(0U), + m_healthPacketsReceived(0U), + m_healthProtocolPacketsReceived(0U), + m_healthLastPacketReceived(0U), + m_healthLastProtocolPacketReceived(0U), + m_healthTasksSubmitted(0U), + m_healthTasksCompleted(0U), + m_healthLastTaskProgress(0U), + m_healthTaskEnqueueFailures(0U), m_metadataUpdateMutex(), m_metadataUpdateState(), m_disablePacketData(false), @@ -572,6 +582,10 @@ void TrafficNetwork::setPresharedKey(const uint8_t* presharedKey) void TrafficNetwork::processNetwork() { + const uint64_t healthNow = std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); + m_healthDataPlaneEpoch.fetch_add(1U, std::memory_order_relaxed); + m_healthLastDataPlaneTick.store(healthNow, std::memory_order_relaxed); + if (m_status != NET_STAT_MST_RUNNING) { return; } @@ -585,6 +599,13 @@ void TrafficNetwork::processNetwork() // read message UInt8Array buffer = m_frameQueue->read(length, address, addrLen, &rtpHeader, &fneHeader); if (length > 0) { + m_healthPacketsReceived.fetch_add(1U, std::memory_order_relaxed); + m_healthLastPacketReceived.store(healthNow, std::memory_order_relaxed); + if (fneHeader.getFunction() == NET_FUNC::PROTOCOL) { + m_healthProtocolPacketsReceived.fetch_add(1U, std::memory_order_relaxed); + m_healthLastProtocolPacketReceived.store(healthNow, std::memory_order_relaxed); + } + if (m_debug) Utils::dump(1U, "TrafficNetwork::processNetwork(), Network Message", buffer.get(), length); @@ -607,9 +628,19 @@ void TrafficNetwork::processNetwork() ::memcpy(req->buffer, buffer.get(), length); // enqueue the task + const uint64_t pendingBefore = m_healthTasksSubmitted.load(std::memory_order_relaxed) - m_healthTasksCompleted.load(std::memory_order_relaxed); + if (pendingBefore == 0U) + m_healthLastTaskProgress.store(healthNow, std::memory_order_relaxed); + + m_healthTasksSubmitted.fetch_add(1U, std::memory_order_relaxed); if (!m_threadPool.enqueue(new_pooltask(taskNetworkRx, req))) { + m_healthTaskEnqueueFailures.fetch_add(1U, std::memory_order_relaxed); + m_healthTasksCompleted.fetch_add(1U, std::memory_order_relaxed); + m_healthLastTaskProgress.store(healthNow, std::memory_order_relaxed); + LogError(LOG_NET, "Failed to task enqueue network packet request, peerId = %u, %s:%u", peerId, udp::Socket::address(address).c_str(), udp::Socket::port(address)); + if (req != nullptr) { if (req->buffer != nullptr) delete[] req->buffer; @@ -1158,6 +1189,13 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) return; } + // define a lambda function to mark the completion of a health task and update the last task progress timestamp + const auto healthTaskComplete = [network]() { + network->m_healthTasksCompleted.fetch_add(1U, std::memory_order_relaxed); + const uint64_t now = std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); + network->m_healthLastTaskProgress.store(now, std::memory_order_relaxed); + }; + if (req == nullptr) return; @@ -1203,6 +1241,7 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) delete[] req->buffer; delete req; + healthTaskComplete(); return; } @@ -1230,16 +1269,22 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) if ((func == NET_FUNC::RPTK && !validRepeaterAuthLength(req->length)) || (func == NET_FUNC::RPTC && !validRepeaterConfigLength(req->length))) { LogWarning(LOG_MASTER, "PEER %u malformed FNE login packet, func = $%02X, length = %d", peerId, func, req->length); + if (req->buffer != nullptr) delete[] req->buffer; delete req; + + healthTaskComplete(); return; } // bryanb: temporary support to allow announce packets on the traffic port but ultimately // this should be removed and handled like TRANSFER is handled here if (func == NET_FUNC::ANNOUNCE) { + healthTaskComplete(); + network->m_host->m_mdNetwork->taskNetworkRx(req); + return; // don't break, return because taskNetworkRx will cleanup req } @@ -1254,6 +1299,8 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) if (req->buffer != nullptr) delete[] req->buffer; delete req; + + healthTaskComplete(); } } diff --git a/src/fne/network/TrafficNetwork.h b/src/fne/network/TrafficNetwork.h index 6f824c83..ce561938 100644 --- a/src/fne/network/TrafficNetwork.h +++ b/src/fne/network/TrafficNetwork.h @@ -156,6 +156,7 @@ namespace network uint8_t* buffer = nullptr; //!< Raw data buffer uint64_t pktRxTime; //!< Packet receive time + bool healthMetadataTracked = false; //!< Health task originated on metadata port. }; // --------------------------------------------------------------------------- @@ -469,6 +470,20 @@ namespace network ThreadPool m_threadPool; ThreadPool m_metadataUpdateThreadPool; + // passive health telemetry -- these counters are intentionally lock-free so + // the REST thread can diagnose a stalled worker without waiting + // for a lock held by that worker + std::atomic m_healthDataPlaneEpoch; + std::atomic m_healthLastDataPlaneTick; + std::atomic m_healthPacketsReceived; + std::atomic m_healthProtocolPacketsReceived; + std::atomic m_healthLastPacketReceived; + std::atomic m_healthLastProtocolPacketReceived; + std::atomic m_healthTasksSubmitted; + std::atomic m_healthTasksCompleted; + std::atomic m_healthLastTaskProgress; + std::atomic m_healthTaskEnqueueFailures; + /** * @brief Represents the state of a metadata update for a given peer ID. * @ingroup fne_network diff --git a/src/fne/restapi/RESTAPI.cpp b/src/fne/restapi/RESTAPI.cpp index bd559a09..7d22affe 100644 --- a/src/fne/restapi/RESTAPI.cpp +++ b/src/fne/restapi/RESTAPI.cpp @@ -691,6 +691,7 @@ void RESTAPI::initializeEndpoints() m_dispatcher.match(FNE_GET_RELOAD_CRYPTO).get(REST_API_BIND(RESTAPI::restAPI_GetReloadCrypto, this)); m_dispatcher.match(FNE_GET_STATS).get(REST_API_BIND(RESTAPI::restAPI_GetStats, this)); + m_dispatcher.match(FNE_GET_HEALTH).get(REST_API_BIND(RESTAPI::restAPI_GetHealth, this)); m_dispatcher.match(FNE_GET_RESET_TOTAL_CALLS).get(REST_API_BIND(RESTAPI::restAPI_GetResetTotalCalls, this)); m_dispatcher.match(FNE_GET_RESET_ACTIVE_CALLS).get(REST_API_BIND(RESTAPI::restAPI_GetResetActiveCalls, this)); m_dispatcher.match(FNE_GET_RESET_CALL_COLLISIONS).get(REST_API_BIND(RESTAPI::restAPI_GetResetCallCollisions, this)); @@ -2199,6 +2200,300 @@ void RESTAPI::restAPI_GetStats(const HTTPPayload& request, HTTPPayload& reply, c reply.payload(response); } +/* REST API endpoint; reports passive FNE health information. */ + +void RESTAPI::restAPI_GetHealth(const HTTPPayload& request, HTTPPayload& reply, const RequestMatch& match) +{ + if (!validateAuth(request, reply)) { + return; + } + + json::object response = json::object(); + setResponseDefaultStatus(response); + + json::object health = json::object(); + json::array reasonCodes = json::array(); + json::array reasons = json::array(); + + const uint64_t now = std::chrono::duration_cast(std::chrono::system_clock::now().time_since_epoch()).count(); + const auto ageMs = [now](uint64_t timestamp) -> uint64_t { + return timestamp == 0U || timestamp > now ? 0U : now - timestamp; + }; + + const uint32_t localFneId = m_network != nullptr ? m_network->m_peerId : 0U; + health["fneId"].set(localFneId); + health["observedAtMs"].set(now); + + json::object process = json::object(); + process["status"].set("healthy"); + const bool restApiResponsive = true; + process["restApiResponsive"].set(restApiResponsive); + health["process"].set(process); + + // check if the traffic network is available and healthy + if (m_network == nullptr) { + health["overall"].set("failed"); + reasonCodes.push_back(json::value("TRAFFIC_NETWORK_UNAVAILABLE")); + reasons.push_back(json::value("traffic network is not initialized")); + health["reasonCodes"].set(reasonCodes); + health["reasons"].set(reasons); + response["health"].set(health); + reply.payload(response); + return; + } + + const uint64_t loopStallThreshold = 5000U; + const uint64_t trafficEpoch = m_network->m_healthDataPlaneEpoch.load(std::memory_order_relaxed); + const uint64_t trafficLastTick = m_network->m_healthLastDataPlaneTick.load(std::memory_order_relaxed); + const uint64_t trafficTickAge = ageMs(trafficLastTick); + const bool trafficPlaneInitializing = trafficLastTick == 0U; + const bool trafficPlaneStalled = !trafficPlaneInitializing && trafficTickAge > loopStallThreshold; + + network::MetadataNetwork* mdNetwork = m_host->m_mdNetwork; + const bool metadataUnavailable = mdNetwork == nullptr; + const uint64_t metadataEpoch = metadataUnavailable ? 0U : mdNetwork->m_healthDataPlaneEpoch.load(std::memory_order_relaxed); + const uint64_t metadataLastTick = metadataUnavailable ? 0U : mdNetwork->m_healthLastDataPlaneTick.load(std::memory_order_relaxed); + const uint64_t metadataTickAge = ageMs(metadataLastTick); + const bool metadataPlaneInitializing = !metadataUnavailable && metadataLastTick == 0U; + const bool metadataPlaneStalled = metadataUnavailable || + (!metadataPlaneInitializing && metadataTickAge > loopStallThreshold); + const bool dataPlaneInitializing = trafficPlaneInitializing || metadataPlaneInitializing; + const bool dataPlaneStalled = trafficPlaneStalled || metadataPlaneStalled; + + // construct JSON objects representing the status of the traffic, metadata, and overall data planes + json::object trafficPlane = json::object(); + trafficPlane["status"].set(trafficPlaneInitializing ? "initializing" : (trafficPlaneStalled ? "failed" : "healthy")); + trafficPlane["epoch"].set(trafficEpoch); + trafficPlane["lastTickAgeMs"].set(trafficTickAge); + + json::object metadataPlane = json::object(); + metadataPlane["status"].set(metadataUnavailable ? "unavailable" : (metadataPlaneInitializing ? "initializing" : (metadataPlaneStalled ? "failed" : "healthy"))); + metadataPlane["epoch"].set(metadataEpoch); + metadataPlane["lastTickAgeMs"].set(metadataTickAge); + + json::object dataPlane = json::object(); + dataPlane["status"].set(dataPlaneInitializing ? "initializing" : (dataPlaneStalled ? "failed" : "healthy")); + dataPlane["loopStallThresholdMs"].set(loopStallThreshold); + dataPlane["traffic"].set(trafficPlane); + dataPlane["metadata"].set(metadataPlane); + health["dataPlane"].set(dataPlane); + + const uint64_t submitted = m_network->m_healthTasksSubmitted.load(std::memory_order_relaxed); + const uint64_t completed = m_network->m_healthTasksCompleted.load(std::memory_order_relaxed); + const uint64_t inFlight = submitted >= completed ? submitted - completed : 0U; + const uint64_t lastTaskProgress = m_network->m_healthLastTaskProgress.load(std::memory_order_relaxed); + const uint64_t taskProgressAge = inFlight > 0U ? ageMs(lastTaskProgress) : 0U; + const uint64_t queueStallThreshold = 5000U; + const bool trafficQueueStalled = inFlight > 0U && taskProgressAge > queueStallThreshold; + + json::object inboundQueue = json::object(); + inboundQueue["submittedTotal"].set(submitted); + inboundQueue["completedTotal"].set(completed); + inboundQueue["inFlight"].set(inFlight); + inboundQueue["lastProgressAgeMs"].set(taskProgressAge); + const uint64_t trafficEnqueueFailures = m_network->m_healthTaskEnqueueFailures.load(std::memory_order_relaxed); + const bool trafficQueueProgressing = !trafficQueueStalled; + inboundQueue["enqueueFailures"].set(trafficEnqueueFailures); + inboundQueue["progressing"].set(trafficQueueProgressing); + + const uint64_t metadataSubmitted = metadataUnavailable ? 0U : mdNetwork->m_healthTasksSubmitted.load(std::memory_order_relaxed); + const uint64_t metadataCompleted = metadataUnavailable ? 0U : mdNetwork->m_healthTasksCompleted.load(std::memory_order_relaxed); + const uint64_t metadataInFlight = metadataSubmitted >= metadataCompleted ? metadataSubmitted - metadataCompleted : 0U; + const uint64_t metadataLastTaskProgress = metadataUnavailable ? 0U : mdNetwork->m_healthLastTaskProgress.load(std::memory_order_relaxed); + const uint64_t metadataTaskProgressAge = metadataInFlight > 0U ? ageMs(metadataLastTaskProgress) : 0U; + const bool metadataQueueStalled = !metadataUnavailable && metadataInFlight > 0U && + metadataTaskProgressAge > queueStallThreshold; + const bool queueStalled = trafficQueueStalled || metadataQueueStalled; + + json::object metadataQueue = json::object(); + metadataQueue["submittedTotal"].set(metadataSubmitted); + metadataQueue["completedTotal"].set(metadataCompleted); + metadataQueue["inFlight"].set(metadataInFlight); + metadataQueue["lastProgressAgeMs"].set(metadataTaskProgressAge); + const uint64_t metadataEnqueueFailures = metadataUnavailable ? 0U : mdNetwork->m_healthTaskEnqueueFailures.load(std::memory_order_relaxed); + const bool metadataQueueProgressing = !metadataQueueStalled && !metadataUnavailable; + metadataQueue["enqueueFailures"].set(metadataEnqueueFailures); + metadataQueue["progressing"].set(metadataQueueProgressing); + + json::object queues = json::object(); + queues["status"].set(queueStalled ? "failed" : "healthy"); + queues["stallThresholdMs"].set(queueStallThreshold); + queues["trafficInboundWorkerQueue"].set(inboundQueue); + queues["metadataInboundWorkerQueue"].set(metadataQueue); + health["queues"].set(queues); + + // construct JSON array representing the status of upstream peer networks + json::array upstreams = json::array(); + uint32_t configuredUpstreams = 0U; + uint32_t connectedUpstreams = 0U; + for (const auto& entry : m_host->m_peerNetworks) { + network::PeerNetwork* upstream = entry.second; + if (upstream == nullptr) + continue; + + json::object obj = json::object(); + const bool enabled = upstream->isEnabled(); + const bool connected = upstream->getStatus() == NET_STAT_RUNNING; + if (enabled) + configuredUpstreams++; + if (enabled && connected) + connectedUpstreams++; + + obj["peerId"].set(entry.first); + const uint32_t remotePeerId = upstream->getRemotePeerId(); + obj["remotePeerId"].set(remotePeerId); + obj["enabled"].set(enabled); + obj["connected"].set(connected); + upstreams.push_back(json::value(obj)); + } + + // construct JSON array representing the status of child peer connections + json::array children = json::array(); + uint32_t connectedPeers = 0U; + uint32_t connectedChildren = 0U; + bool metadataDegraded = false; + m_network->m_peers.shared_lock(); + for (const auto& entry : m_network->m_peers) { + network::FNEPeerConnection* peer = entry.second; + if (peer == nullptr) + continue; + if (peer->connected()) + connectedPeers++; + if (peer->peerClass() != PEER_CONN_CLASS_NEIGHBOR) + continue; + + if (peer->connected()) + connectedChildren++; + if (peer->missedMetadataUpdates() > 0U) + metadataDegraded = true; + + json::object obj = json::object(); + obj["peerId"].set(entry.first); + obj["identity"].set(peer->identity()); + const bool peerConnected = peer->connected(); + const uint64_t lastPingAge = ageMs(peer->lastPing()); + const uint32_t pingsReceived = peer->pingsReceived(); + const uint32_t missedMetadataUpdates = peer->missedMetadataUpdates(); + obj["connected"].set(peerConnected); + obj["lastPingAgeMs"].set(lastPingAge); + obj["pingsReceived"].set(pingsReceived); + obj["missedMetadataUpdates"].set(missedMetadataUpdates); + children.push_back(json::value(obj)); + } + m_network->m_peers.shared_unlock(); + + // construct JSON object representing the connectivity status of the network + const bool partitioned = configuredUpstreams > 0U && connectedUpstreams == 0U; + const bool upstreamDegraded = configuredUpstreams > connectedUpstreams && connectedUpstreams > 0U; + json::object connectivity = json::object(); + connectivity["status"].set(partitioned ? "partitioned" : ((upstreamDegraded || metadataDegraded) ? "degraded" : "healthy")); + connectivity["configuredUpstreams"].set(configuredUpstreams); + connectivity["connectedUpstreams"].set(connectedUpstreams); + connectivity["connectedPeers"].set(connectedPeers); + connectivity["connectedChildren"].set(connectedChildren); + connectivity["upstreams"].set(upstreams); + connectivity["children"].set(children); + health["connectivity"].set(connectivity); + + // construct JSON object representing the status of network traffic + const uint64_t lastPacket = m_network->m_healthLastPacketReceived.load(std::memory_order_relaxed); + const uint64_t lastProtocolPacket = m_network->m_healthLastProtocolPacketReceived.load(std::memory_order_relaxed); + const uint64_t trafficIdleThreshold = 60000U; + const bool trafficSeen = lastProtocolPacket != 0U; + const bool trafficIdle = !trafficSeen || ageMs(lastProtocolPacket) > trafficIdleThreshold; + json::object traffic = json::object(); + traffic["status"].set(trafficIdle ? "idle" : "healthy"); + const uint64_t packetsReceived = m_network->m_healthPacketsReceived.load(std::memory_order_relaxed); + const uint64_t protocolPacketsReceived = m_network->m_healthProtocolPacketsReceived.load(std::memory_order_relaxed); + const uint64_t lastPacketAge = ageMs(lastPacket); + const uint64_t lastProtocolPacketAge = ageMs(lastProtocolPacket); + const uint64_t metadataPacketsReceived = metadataUnavailable ? 0U : mdNetwork->m_healthPacketsReceived.load(std::memory_order_relaxed); + const uint64_t lastMetadataPacketAge = metadataUnavailable ? 0U : ageMs(mdNetwork->m_healthLastPacketReceived.load(std::memory_order_relaxed)); + const int32_t activeCalls = network::TrafficNetwork::MetricsLogging::getTotalActiveCalls(); + traffic["packetsReceived"].set(packetsReceived); + traffic["protocolPacketsReceived"].set(protocolPacketsReceived); + traffic["lastPacketAgeMs"].set(lastPacketAge); + traffic["lastProtocolPacketAgeMs"].set(lastProtocolPacketAge); + traffic["metadataPacketsReceived"].set(metadataPacketsReceived); + traffic["lastMetadataPacketAgeMs"].set(lastMetadataPacketAge); + traffic["idleThresholdMs"].set(trafficIdleThreshold); + traffic["activeCalls"].set(activeCalls); + health["traffic"].set(traffic); + + // construct JSON object representing the status of the spanning tree network + json::object spanningTree = json::object(); + const bool treeAvailable = m_network->m_treeRoot != nullptr; + const uint32_t knownNodes = treeAvailable ? network::SpanningTree::countChildren(m_network->m_treeRoot) + 1U : 0U; + spanningTree["status"].set(treeAvailable ? "healthy" : "unavailable"); + const bool spanningTreeEnabled = m_network->m_enableSpanningTree; + const uint32_t rootId = treeAvailable ? m_network->m_treeRoot->id() : 0U; + spanningTree["enabled"].set(spanningTreeEnabled); + spanningTree["rootId"].set(rootId); + spanningTree["localRole"].set(treeAvailable && m_network->m_treeRoot->id() == m_network->m_peerId ? "root" : "node"); + spanningTree["knownNodes"].set(knownNodes); + health["spanningTree"].set(spanningTree); + + // determine the overall health status of the network based on individual components + std::string overall = "healthy"; + if (dataPlaneInitializing) { + overall = "initializing"; + } + + if (trafficIdle && overall == "healthy") { + overall = "idle"; + } + + if (upstreamDegraded || metadataDegraded || !treeAvailable) { + overall = "degraded"; + } + + if (partitioned) { + overall = "partitioned"; + reasonCodes.push_back(json::value("ALL_UPSTREAMS_DISCONNECTED")); + reasons.push_back(json::value("all configured upstream FNE connections are disconnected")); + } + + if (dataPlaneStalled || queueStalled) { + overall = "failed"; + } + + if (trafficPlaneStalled) { + reasonCodes.push_back(json::value("TRAFFIC_DATA_PLANE_STALLED")); + reasons.push_back(json::value("traffic processing loop is not progressing")); + } + + if (metadataPlaneStalled) { + reasonCodes.push_back(json::value(metadataUnavailable ? "METADATA_NETWORK_UNAVAILABLE" : "METADATA_DATA_PLANE_STALLED")); + reasons.push_back(json::value(metadataUnavailable ? "metadata network is not initialized" : "metadata processing loop is not progressing")); + } + + if (trafficQueueStalled) { + reasonCodes.push_back(json::value("TRAFFIC_INBOUND_QUEUE_STALLED")); + reasons.push_back(json::value("traffic inbound worker queue contains stale work")); + } + + if (metadataQueueStalled) { + reasonCodes.push_back(json::value("METADATA_INBOUND_QUEUE_STALLED")); + reasons.push_back(json::value("metadata inbound worker queue contains stale work")); + } + + if (upstreamDegraded) { + reasonCodes.push_back(json::value("UPSTREAM_CONNECTIVITY_DEGRADED")); + reasons.push_back(json::value("one or more configured upstream FNE connections are disconnected")); + } + + if (metadataDegraded) { + reasonCodes.push_back(json::value("METADATA_UPDATES_MISSED")); + reasons.push_back(json::value("one or more child FNEs have missed metadata updates")); + } + + health["overall"].set(overall); + health["reasonCodes"].set(reasonCodes); + health["reasons"].set(reasons); + response["health"].set(health); + reply.payload(response); +} + /* REST API endpoint; implements get reset total calls request. */ void RESTAPI::restAPI_GetResetTotalCalls(const HTTPPayload& request, HTTPPayload& reply, const RequestMatch& match) diff --git a/src/fne/restapi/RESTAPI.h b/src/fne/restapi/RESTAPI.h index 9db2b755..69b96237 100644 --- a/src/fne/restapi/RESTAPI.h +++ b/src/fne/restapi/RESTAPI.h @@ -434,6 +434,11 @@ namespace fne_restapi */ void restAPI_GetStats(const HTTPPayload& request, HTTPPayload& reply, const restapi::RequestMatch& match); + /** + * @brief REST API endpoint; reports passive FNE health information. + */ + void restAPI_GetHealth(const HTTPPayload& request, HTTPPayload& reply, const restapi::RequestMatch& match); + /** * @brief REST API endpoint; implements put reset total calls request. * @param request HTTP request. diff --git a/src/fne/restapi/RESTDefines.h b/src/fne/restapi/RESTDefines.h index ea077001..8ca79fd7 100644 --- a/src/fne/restapi/RESTDefines.h +++ b/src/fne/restapi/RESTDefines.h @@ -64,6 +64,7 @@ #define FNE_GET_RELOAD_CRYPTO "/reload-crypto" #define FNE_GET_STATS "/stats" +#define FNE_GET_HEALTH "/health" #define FNE_GET_RESET_TOTAL_CALLS "/stat-reset-total-calls" #define FNE_GET_RESET_ACTIVE_CALLS "/stat-reset-active-calls" #define FNE_GET_RESET_CALL_COLLISIONS "/stat-reset-call-collisions" diff --git a/src/remote/RESTClientMain.cpp b/src/remote/RESTClientMain.cpp index 6072341c..8c7fd10a 100644 --- a/src/remote/RESTClientMain.cpp +++ b/src/remote/RESTClientMain.cpp @@ -61,6 +61,7 @@ #define RCD_FNE_SAVE_PEER_ACL "fne-peer-commit" #define RCD_FNE_GET_STATS "fne-stats" +#define RCD_FNE_GET_HEALTH "fne-health" #define RCD_FNE_PUT_RESET_TOTAL_CALLS "fne-reset-total-calls" #define RCD_FNE_PUT_RESET_ACTIVE_CALLS "fne-reset-active-calls" #define RCD_FNE_PUT_RESET_CALL_COLLISIONS "fne-reset-call-collisions" @@ -992,6 +993,9 @@ int main(int argc, char** argv) else if (rcom == RCD_FNE_GET_STATS) { retCode = client->send(HTTP_GET, FNE_GET_STATS, json::object(), response); } + else if (rcom == RCD_FNE_GET_HEALTH) { + retCode = client->send(HTTP_GET, FNE_GET_HEALTH, json::object(), response); + } else if (rcom == RCD_FNE_PUT_RESET_TOTAL_CALLS) { retCode = client->send(HTTP_GET, FNE_GET_RESET_TOTAL_CALLS, json::object(), response); }