diff --git a/docs/api/rest.rst b/docs/api/rest.rst index 11c19cdf1..200795504 100644 --- a/docs/api/rest.rst +++ b/docs/api/rest.rst @@ -45,21 +45,34 @@ operations that read them. ``X-Medkit-No-Fan-Out`` Answer from this gateway alone: do not query aggregated peers and do not merge their items. Read by the **per-entity** resource-collection list - endpoints (data, operations, configurations, faults, logs) and by ``GET - /api/v1/version-info`` - exactly the operations that declare it in the - OpenAPI document. + endpoints (data, operations, configurations, faults, logs), by the global + ``GET /api/v1/faults`` and by ``GET /api/v1/version-info`` - exactly the + operations that declare it in the OpenAPI document. The gateway sets it on its own outbound peer requests, which stops bidirectional aggregation from recursing **on the routes that check it**. - The global ``GET /api/v1/faults`` does not check it, so it neither declares - the header nor honours it (see :ref:`the fan-out design note - `). **Presence-only.** The value is never read, so ``X-Medkit-No-Fan-Out: false`` suppresses fan-out exactly like any other value. The OpenAPI schema is a string rather than a boolean for that reason. Omit the header to get the aggregated answer. +``X-Medkit-Fan-Out-Visited`` + Read by the global ``GET /api/v1/faults`` only, and set by the gateway on + its own fan-out of that route: the instance ids of the gateways the request + has already passed through, comma-separated. Each aggregating gateway + publishes its id as ``x-medkit-instance-id`` on ``GET /api/v1/health``, + where its peers read it. A gateway does not ask a peer whose id is on the + path. A gateway asked anyway answers an empty list, because its faults are + already in the answer. Any other gateway answers with its own faults and + those of its peers, adding its id. This is what lets the global list reach + the end of a chain (A -> B -> C) while two gateways that aggregate each + other do not ask each other twice. It takes precedence over + ``X-Medkit-No-Fan-Out``, which the same request also carries. A path of 16 + ids is not extended: the gateway answers for itself and reports its peers + as ``too-deep``. Clients do not send it (see :ref:`the fan-out design note + `). + Server Capabilities ------------------- @@ -3892,6 +3905,13 @@ why, one entry per name, with ``reason`` one of: * - ``invalid-response`` - The peer answered with something that is not the JSON the route promises. + * - ``too-deep`` + - Global ``GET /faults`` only. The request had already passed through 16 + gateways, so this gateway did not ask the peer. + +On the global ``GET /faults`` a peer that could not reach one of its own +peers is reported partial in turn. The entry names both hops, +``/``, with the reason the peer gave. Without ``peer_failures`` a client cannot tell a busy subsystem from a dead one, which is the whole of what a partial answer is asked. diff --git a/docs/config/aggregation.rst b/docs/config/aggregation.rst index 34f4ea942..caacb6c6e 100644 --- a/docs/config/aggregation.rst +++ b/docs/config/aggregation.rst @@ -491,6 +491,14 @@ into the local response. This applies to: - ``GET /api/v1/{entity_type}/{id}/logs`` - ``GET /api/v1/faults`` (global fault list) +The global fault list reaches the whole chain: in A -> B -> C, ``GET +/api/v1/faults`` on A lists the faults of A, B and C. Each gateway adds its own +id to an ``X-Medkit-Fan-Out-Visited`` header on the peer requests and does not +ask a peer whose id is already there. Two gateways that list each other as +peers therefore do not pass the request back and forth. If B cannot reach C, +A's answer is partial and names ``subsystem_b/subsystem_c`` in +``failed_peers``. + If a peer request fails during fan-out (peer unreachable or non-2xx response), the response body includes: diff --git a/src/ros2_medkit_gateway/design/aggregation.rst b/src/ros2_medkit_gateway/design/aggregation.rst index 9d0361ec4..a308cf33b 100644 --- a/src/ros2_medkit_gateway/design/aggregation.rst +++ b/src/ros2_medkit_gateway/design/aggregation.rst @@ -621,26 +621,54 @@ healthy peers, collects the responses, and merges the ``items`` arrays. If some peers fail, the response body includes ``x-medkit.partial: true`` and ``x-medkit.failed_peers``. Fan-out requests carry an ``X-Medkit-No-Fan-Out`` header, and both helpers return early when the incoming request has it, so a -peer that aggregates back never fans out a second time. These are the routes -that declare the header in the OpenAPI document -(``RouteEntry::fan_out_aware()``). - -.. warning:: - - The global ``GET /api/v1/faults`` is **not** one of them. - ``FaultHandlers::list_all_faults`` calls ``fan_out_get()`` directly rather - than through either helper, so it never inspects ``X-Medkit-No-Fan-Out`` - - and ``fan_out_get()`` is the code that *sets* the header outbound. Nothing - else in ``aggregation/`` guards the loop. Two gateways that peer with each - other therefore recurse unbounded on this one route: A queries B, B queries - A, and each hop holds a ``std::async`` thread until its timeout, so the - thread cost grows with recursion depth. - - The route consequently does not declare the header either - advertising an - opt-out it ignores would be worse than silence. Fixing this means routing - ``list_all_faults`` through the helper (which changes the per-item wire - shape it deliberately preserves) or adding a loop guard inside - ``fan_out_get()``; both are aggregation changes, not documentation ones. +peer that aggregates back never fans out a second time. These routes and the +global fault list below are the ones that declare the header in the OpenAPI +document (``RouteEntry::fan_out_aware()``). + +**The global fault list** (``GET /api/v1/faults``) fans out as well, but it +cannot stop at the first peer. There is no routing table behind it: in a chain +A -> B -> C the only way A lists C's faults is for B to ask C. So +``FaultHandlers::list_all_faults`` (and ``PluginContext::list_all_faults``, +which calls the same route on the peers) sends ``X-Medkit-Fan-Out-Visited`` +next to ``X-Medkit-No-Fan-Out``. The header carries the instance ids of the +gateways the request has passed through. Each ``AggregationManager`` generates +its id at startup, because a peer name is what the aggregator calls a peer and +a gateway cannot recognise itself by it. It publishes the id as +``x-medkit-instance-id`` on ``/health``, and ``PeerClient::check_health`` +records each peer's. ``plan_global_fault_fan_out()`` in ``fan_out_helpers.hpp`` +reads the request: + +- No visited path and no ``X-Medkit-No-Fan-Out``: a client request. Answer + locally and fan out with the path set to this gateway's id. +- Visited path that does not name this gateway: answer locally and fan out + with this gateway's id added. ``X-Medkit-No-Fan-Out`` is ignored here, + because ``fan_out_get()`` puts it on every peer request. +- Visited path that names this gateway: the request came back round a loop. + Answer ``200`` with an empty list, since this gateway's faults are already + in the answer the first pass is building. +- A path of 16 ids, or one holding an id longer than 64 characters: answer + locally, do not extend it, and report every healthy peer as failed with + reason ``too-deep``, so a truncated list is not reported complete. +- ``X-Medkit-No-Fan-Out`` alone: answer locally only. That is a client opting + out, or a gateway that predates the visited path. + +``fan_out_get()`` does not send the request to a peer whose recorded id is on +the path. Two gateways that aggregate each other therefore do not ask each +other twice: A asks B with ``[A]``, and B asks only its other peers. This +matters for more than latency. A is holding a worker while it waits for B, so +a request sent back to A could only be answered once another of A's workers +is free. Under as many concurrent requests as A has workers, none would be, +and every one would wait for the peer timeout. The empty answer for a gateway +that finds itself on the path covers a peer whose id is not known yet (before +its first health check) or one that predates the id. + +A peer that answers partial makes the result partial too. Its failed peers are +named ``/`` with the reason it gave, so a dead gateway at the +end of a chain shows up at the head. ``x-medkit.count`` is the length of the +merged ``items``. + +The path does not deduplicate a gateway reached over two different routes in +the same request (A -> B -> C plus A -> C): C answers both, as it did before. **Target-filtered fan-out.** For per-entity paths, ``merge_peer_items()`` asks ``AggregationManager::get_peer_contributors(id)`` for the list of diff --git a/src/ros2_medkit_gateway/design/dto_contract.rst b/src/ros2_medkit_gateway/design/dto_contract.rst index bf3651976..a27d13b7c 100644 --- a/src/ros2_medkit_gateway/design/dto_contract.rst +++ b/src/ros2_medkit_gateway/design/dto_contract.rst @@ -714,7 +714,8 @@ Further ``RouteEntry`` knobs shape the published operation: - ``fan_out_aware()`` - this route reads the ``X-Medkit-No-Fan-Out`` request header, i.e. a client can ask it to answer from this gateway alone instead of merging aggregated peers. Carried by the routes whose handlers go through - ``fan_out_collection`` or ``merge_peer_items``. The declared schema is a bare + ``fan_out_collection`` or ``merge_peer_items``, and by the global ``GET + /faults``, which reads it through ``plan_global_fault_fan_out``. The declared schema is a bare string, not a boolean: the gateway tests ``has_header`` and never reads the value, so ``X-Medkit-No-Fan-Out: false`` still suppresses fan-out and a boolean schema would promise a generated client the opposite. Also diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/aggregation/aggregation_manager.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/aggregation/aggregation_manager.hpp index 8db3ed2be..16377e4ce 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/aggregation/aggregation_manager.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/aggregation/aggregation_manager.hpp @@ -32,6 +32,31 @@ namespace ros2_medkit_gateway { +/// Instance ids of the gateways a global `GET /faults` fan-out has already +/// passed through, comma-separated. A gateway that finds its own id here has +/// its faults in the answer already; one that does not answers for itself and +/// for its own peers. See `plan_global_fault_fan_out`. +inline constexpr const char * kFanOutVisitedHeader = "X-Medkit-Fan-Out-Visited"; + +/// The ids on an `X-Medkit-Fan-Out-Visited` value, trimmed, empty entries dropped. +inline std::vector split_fan_out_visited(const std::string & value) { + std::vector ids; + for (size_t start = 0; start <= value.size();) { + auto end = value.find(',', start); + if (end == std::string::npos) { + end = value.size(); + } + std::string id = value.substr(start, end - start); + id.erase(0, id.find_first_not_of(" \t")); + id.erase(id.find_last_not_of(" \t") + 1); + if (!id.empty()) { + ids.push_back(std::move(id)); + } + start = end + 1; + } + return ids; +} + /// A peer's name and where to reach it. struct PeerEndpoint { std::string name; @@ -193,6 +218,14 @@ class AggregationManager { /// Get the number of known peers (static + discovered) size_t peer_count() const; + /// Random id of this gateway process, put on `X-Medkit-Fan-Out-Visited`. + /// + /// Generated, not configured: peer names are what the aggregator calls a + /// peer, so a gateway cannot recognise itself by one. + const std::string & instance_id() const { + return instance_id_; + } + /** * @brief Add a dynamically discovered peer * @@ -392,10 +425,18 @@ class AggregationManager { * list means "no matching peers" and produces an empty result with * no fan-out. When null (default), all healthy peers are queried, * preserving the pre-filtering behavior for global endpoints. + * @param extra_headers Sent on every peer request next to + * `X-Medkit-No-Fan-Out`. When they carry `X-Medkit-Fan-Out-Visited`, + * a peer whose instance id is on that path is not asked at all: its + * answer is already counted, and asking it would hold a worker on + * both sides until it answered empty. A peer that reports itself + * partial makes this result partial too, with its failed peers named + * `/`. * @return FanOutResult with merged items and failure info. */ FanOutResult fan_out_get(const std::string & path, const std::string & auth_header, - const std::vector * target_peers = nullptr); + const std::vector * target_peers = nullptr, + const httplib::Headers & extra_headers = {}); /** * @brief Get peer status for /health endpoint @@ -406,6 +447,7 @@ class AggregationManager { private: AggregationConfig config_; log::Logger logger_; + std::string instance_id_; size_t static_peer_count_{0}; ///< Number of statically configured peers (not subject to max_discovered_peers) /// Whether a peer whose picture could not be refreshed can still be reached. /// diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/aggregation/peer_client.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/aggregation/peer_client.hpp index 75bb3b83b..859564edb 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/aggregation/peer_client.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/aggregation/peer_client.hpp @@ -178,11 +178,16 @@ class PeerClient { /// Check if the peer was healthy at last health check bool is_healthy() const; + /// The peer's `x-medkit-instance-id` from its last health check, empty if it + /// sent none. Lets a fan-out skip a peer already on the visited path. + std::string instance_id() const; + /** * @brief Perform a health check against the peer * * GETs /api/v1/health on the peer. Sets the internal healthy_ flag - * based on whether a 200 response was received. + * based on whether a 200 response was received, and records the peer's + * instance id. */ void check_health(); @@ -308,6 +313,9 @@ class PeerClient { std::atomic healthy_{false}; std::atomic shutdown_requested_{false}; + mutable std::mutex instance_id_mutex_; + std::string instance_id_; + std::mutex client_mutex_; std::unique_ptr client_; diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/http/fan_out_helpers.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/http/fan_out_helpers.hpp index 73c44baa0..94c1cb722 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/http/fan_out_helpers.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/http/fan_out_helpers.hpp @@ -14,7 +14,9 @@ #pragma once +#include #include +#include #include #include #include @@ -111,6 +113,69 @@ inline std::string build_fan_out_path(const httplib::Request & req) { return path; } +/// How a gateway answers the global `GET /faults`. +/// +/// That list has no routing table behind it: the only way to list the faults +/// of a gateway two hops away is for the hop in between to fan out again. So +/// the route does not stop at `X-Medkit-No-Fan-Out` the way the per-entity +/// collections do. Each gateway adds its instance id to +/// `X-Medkit-Fan-Out-Visited` instead. A peer whose id is on the path is not +/// asked, and one asked anyway (its id not yet learned from its health check) +/// answers an empty list. A chain A -> B -> C still reaches C, and two gateways +/// that aggregate each other do not ask each other twice. +struct GlobalFaultFanOut { + enum class Mode { + kFanOut, ///< Answer locally and ask the peers, sending `visited`. + kLocalOnly, ///< Answer from this gateway's fault manager only. + kAlreadyCounted, ///< This gateway is on the path: answer an empty list. + kTooDeep, ///< The path cannot be extended: answer locally, report the peers not asked. + }; + Mode mode{Mode::kLocalOnly}; + std::string visited; +}; + +/// Longest `X-Medkit-Fan-Out-Visited` path that is still extended. +inline constexpr size_t kMaxFanOutVisited = 16; +/// Longest instance id accepted on that path. Ours are 32 hex digits. +inline constexpr size_t kMaxFanOutVisitedIdLength = 64; + +/// `self_id` is `AggregationManager::instance_id()`, empty when aggregation is off. +/// +/// `X-Medkit-No-Fan-Out` alone keeps its meaning: answer from this gateway +/// alone. That is what a client opting out sends, and what a gateway that +/// predates the visited path sends. Next to the visited path it is ignored, +/// because `fan_out_get` puts it on every peer request. +inline GlobalFaultFanOut plan_global_fault_fan_out(const httplib::Request & req, const std::string & self_id) { + using Mode = GlobalFaultFanOut::Mode; + if (self_id.empty()) { + return {Mode::kLocalOnly, {}}; + } + if (!req.has_header(kFanOutVisitedHeader)) { + if (req.has_header("X-Medkit-No-Fan-Out")) { + return {Mode::kLocalOnly, {}}; + } + return {Mode::kFanOut, self_id}; + } + const auto ids = split_fan_out_visited(req.get_header_value(kFanOutVisitedHeader)); + if (std::find(ids.begin(), ids.end(), self_id) != ids.end()) { + return {Mode::kAlreadyCounted, {}}; + } + // A path this gateway cannot extend still gets its own faults: answering + // for itself is right, asking further is not. + bool readable = ids.size() < kMaxFanOutVisited; + std::string visited; + for (const auto & id : ids) { + readable = readable && id.size() <= kMaxFanOutVisitedIdLength; + visited += id; + visited += ','; + } + if (!readable) { + return {Mode::kTooDeep, {}}; + } + visited += self_id; + return {Mode::kFanOut, std::move(visited)}; +} + /// Fan-out GET to peer gateways and merge their items into `result["items"]`. /// Aggregation metadata (partial, failed_peers) is written into `ext_json` /// (a plain JSON object that callers fold into the response x-medkit block). diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/aggregation.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/aggregation.hpp index ae403ebc0..91baf6b9b 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/aggregation.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/aggregation.hpp @@ -64,9 +64,10 @@ inline constexpr std::string_view dto_name = "DroppedItem"; // tell a slow subsystem from a dead one. // // Wire keys: -// peer - peer name, matching the entry in failed_peers +// peer - peer name, matching the entry in failed_peers. On the global +// fault list a peer's own failed peer is named /. // reason - one of: timeout, unreachable, canceled, error-status, too-large, -// invalid-response +// invalid-response, too-deep (global fault list only) // --------------------------------------------------------------------------- struct PeerFailure { std::string peer; diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/health.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/health.hpp index 7cbcb7cd5..de01520d8 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/health.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/dto/health.hpp @@ -172,6 +172,10 @@ struct Health { /// why a peer refused their stream with 503. std::optional x_medkit_sse; std::optional peers; // free-form array of peer status objects + /// Random id of this gateway process, present when aggregation is on. Wire + /// key: "x-medkit-instance-id". An aggregating peer reads it in its health + /// check, so a global fault-list fan-out can skip a gateway it already passed. + std::optional x_medkit_instance_id; // NOT optional: these two are emitted in every mode, aggregation or not, so // the generated schema must list them in `required` and let a typed client // read them without a presence check. An empty `warnings` array is the @@ -190,8 +194,12 @@ inline constexpr auto dto_fields = std::make_tuple( field("discovery", &Health::discovery), field("x-medkit-data-provider", &Health::x_medkit_data_provider), field("x-medkit-subscription-executor", &Health::x_medkit_subscription_executor), field("x-medkit-entity-cache", &Health::x_medkit_entity_cache), field("x-medkit-sse", &Health::x_medkit_sse), - field("peers", &Health::peers), field("warning_schema_version", &Health::warning_schema_version), - field("warnings", &Health::warnings)); + field("peers", &Health::peers), + field("x-medkit-instance-id", &Health::x_medkit_instance_id, + "Random id of this gateway process, present when aggregation is enabled. Generated at startup and " + "carried on X-Medkit-Fan-Out-Visited by the global fault list, so a gateway already asked is not asked " + "again."), + field("warning_schema_version", &Health::warning_schema_version), field("warnings", &Health::warnings)); template <> inline constexpr std::string_view dto_name = "HealthStatus"; diff --git a/src/ros2_medkit_gateway/src/aggregation/aggregation_manager.cpp b/src/ros2_medkit_gateway/src/aggregation/aggregation_manager.cpp index c7d198085..f543ecc12 100644 --- a/src/ros2_medkit_gateway/src/aggregation/aggregation_manager.cpp +++ b/src/ros2_medkit_gateway/src/aggregation/aggregation_manager.cpp @@ -17,6 +17,7 @@ #include #include +#include #include #include #include @@ -32,6 +33,17 @@ namespace ros2_medkit_gateway { namespace { +std::string make_instance_id() { + static constexpr char kHex[] = "0123456789abcdef"; + std::random_device rd; + std::uniform_int_distribution digit(0, 15); + std::string id(32, '0'); + for (auto & c : id) { + c = kHex[digit(rd)]; + } + return id; +} + /** * @brief Extract the host portion from a URL * @@ -189,7 +201,7 @@ bool is_valid_peer_url(const std::string & url) { } // namespace AggregationManager::AggregationManager(const AggregationConfig & config, const log::Logger * logger) - : config_(config), logger_(logger ? *logger : log::Logger("aggregation_manager")) { + : config_(config), logger_(logger ? *logger : log::Logger("aggregation_manager")), instance_id_(make_instance_id()) { for (const auto & peer_cfg : config_.peers) { // Validate scheme for static peers (http/https only). // Unlike mDNS peers, loopback addresses are valid for static config @@ -827,7 +839,8 @@ void AggregationManager::forward_request(const std::string & peer_name, const ht AggregationManager::FanOutResult AggregationManager::fan_out_get(const std::string & path, const std::string & auth_header, - const std::vector * target_peers) { + const std::vector * target_peers, + const httplib::Headers & extra_headers) { // Per-peer result collected by each async task struct PeerResult { std::string peer_name; @@ -837,6 +850,8 @@ AggregationManager::FanOutResult AggregationManager::fan_out_get(const std::stri /// fan-out failure indistinguishable on the wire. std::string failure_reason; nlohmann::json items = nlohmann::json::array(); + /// What the peer itself could not reach, when it fanned out further. + std::vector nested_failures; }; // A non-null, empty target list means "no peers match" - skip fan-out @@ -849,6 +864,11 @@ AggregationManager::FanOutResult AggregationManager::fan_out_get(const std::stri return empty_result; } + std::vector visited; + if (auto it = extra_headers.find(kFanOutVisitedHeader); it != extra_headers.end()) { + visited = split_fan_out_visited(it->second); + } + // Snapshot healthy peers under lock, release before network I/O. // When target_peers is non-null, restrict the snapshot to matching peers. std::vector> snapshot; @@ -862,20 +882,31 @@ AggregationManager::FanOutResult AggregationManager::fan_out_get(const std::stri std::find(target_peers->begin(), target_peers->end(), peer->name()) == target_peers->end()) { continue; } + // Already answered for this request. Asked anyway it would answer empty, + // but only once one of its workers is free, and every worker it has may + // be waiting on this gateway. + if (!visited.empty()) { + const auto peer_id = peer->instance_id(); + if (!peer_id.empty() && std::find(visited.begin(), visited.end(), peer_id) != visited.end()) { + continue; + } + } snapshot.push_back(peer); } } // Fan out GET requests in parallel via std::async to reduce worst-case // latency from N * timeout_ms (sequential) to just timeout_ms (parallel). + httplib::Headers headers{{"X-Medkit-No-Fan-Out", "1"}}; + headers.insert(extra_headers.begin(), extra_headers.end()); std::vector> futures; futures.reserve(snapshot.size()); for (auto & peer : snapshot) { - futures.push_back(std::async(std::launch::async, [peer, path, auth_header]() { + futures.push_back(std::async(std::launch::async, [peer, path, auth_header, headers]() { PeerResult pr; pr.peer_name = peer->name(); - auto result = peer->forward_and_get_json("GET", path, auth_header, {{"X-Medkit-No-Fan-Out", "1"}}); + auto result = peer->forward_and_get_json("GET", path, auth_header, headers); if (!result.has_value()) { pr.success = false; pr.failure_reason = peer_failure_reason(result.error().kind); @@ -887,6 +918,47 @@ AggregationManager::FanOutResult AggregationManager::fan_out_get(const std::stri if (response_json.contains("items") && response_json["items"].is_array()) { pr.items = response_json["items"]; } + // A peer that fanned out further and could not reach everything says so. + // Dropping that would report this answer complete without those items. + const auto x_medkit = response_json.find("x-medkit"); + const bool peer_partial = x_medkit != response_json.end() && x_medkit->is_object() && + x_medkit->contains("partial") && (*x_medkit)["partial"].is_boolean() && + (*x_medkit)["partial"].get(); + if (peer_partial) { + std::vector nested_names; + if (auto names = x_medkit->find("failed_peers"); names != x_medkit->end() && names->is_array()) { + for (const auto & name : *names) { + if (name.is_string()) { + nested_names.push_back(name.get()); + } + } + } + for (const auto & name : nested_names) { + // A peer that predates peer_failures names no reason. Its partial + // flag meant a peer call that failed, which is what unreachable says. + std::string reason = "unreachable"; + if (auto failures = x_medkit->find("peer_failures"); failures != x_medkit->end() && failures->is_array()) { + for (const auto & failure : *failures) { + if (!failure.is_object()) { + continue; + } + const auto named = failure.find("peer"); + const auto why = failure.find("reason"); + if (named != failure.end() && named->is_string() && named->get() == name) { + if (why != failure.end() && why->is_string()) { + reason = why->get(); + } + break; + } + } + } + pr.nested_failures.push_back(dto::PeerFailure{pr.peer_name + "/" + name, std::move(reason)}); + } + // Partial with nobody named breaks the contract this route promises. + if (pr.nested_failures.empty()) { + pr.nested_failures.push_back(dto::PeerFailure{pr.peer_name, "invalid-response"}); + } + } return pr; })); } @@ -906,6 +978,11 @@ AggregationManager::FanOutResult AggregationManager::fan_out_get(const std::stri fan_out_result.merged_items.push_back(std::move(item)); fan_out_result.item_peers.push_back(pr.peer_name); } + for (auto & failure : pr.nested_failures) { + fan_out_result.is_partial = true; + fan_out_result.failed_peers.push_back(failure.peer); + fan_out_result.peer_failures.push_back(std::move(failure)); + } } return fan_out_result; diff --git a/src/ros2_medkit_gateway/src/core/aggregation/peer_client.cpp b/src/ros2_medkit_gateway/src/core/aggregation/peer_client.cpp index 1b37af5a6..481d6323e 100644 --- a/src/ros2_medkit_gateway/src/core/aggregation/peer_client.cpp +++ b/src/ros2_medkit_gateway/src/core/aggregation/peer_client.cpp @@ -705,7 +705,24 @@ void PeerClient::check_health() { headers.emplace("Authorization", peer_auth_header_); } auto result = client_->Get(std::string(API_PREFIX) + "/health", headers); - healthy_.store(result && result->status == 200); + const bool healthy = result && result->status == 200; + std::string instance_id; + if (healthy) { + auto body = nlohmann::json::parse(result->body, nullptr, false); + if (body.is_object() && body.contains("x-medkit-instance-id") && body["x-medkit-instance-id"].is_string()) { + instance_id = body["x-medkit-instance-id"].get(); + } + } + { + std::lock_guard id_lock(instance_id_mutex_); + instance_id_ = std::move(instance_id); + } + healthy_.store(healthy); +} + +std::string PeerClient::instance_id() const { + std::lock_guard lock(instance_id_mutex_); + return instance_id_; } tl::expected PeerClient::fetch_entities() { diff --git a/src/ros2_medkit_gateway/src/core/openapi/route_registry.cpp b/src/ros2_medkit_gateway/src/core/openapi/route_registry.cpp index f51871887..b47af89bb 100644 --- a/src/ros2_medkit_gateway/src/core/openapi/route_registry.cpp +++ b/src/ros2_medkit_gateway/src/core/openapi/route_registry.cpp @@ -393,12 +393,13 @@ RouteEntry & RouteEntry::fan_out_aware() { // The prose deliberately does not promise loop-freedom in general: the // gateway sets this header on its own outbound peer requests, but that only // terminates a recursion on routes that read it - which is exactly the set - // carrying this declaration. The global `GET /faults` fans out without - // checking it and so is not declared here. + // carrying this declaration. The global `GET /faults` reads it too, but its + // own fan-out also sends `X-Medkit-Fan-Out-Visited`, which takes precedence + // so a chain of gateways is still listed to the end. return header_param("X-Medkit-No-Fan-Out", "Present at any value: answer from this gateway alone and do not query " "aggregated peers. The gateway sets it on its own outbound peer requests, so " - "bidirectional aggregation terminates at the first peer for this operation.", + "bidirectional aggregation does not recurse on this operation.", false, nlohmann::json{{"type", "string"}}); } diff --git a/src/ros2_medkit_gateway/src/http/handlers/fault_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/fault_handlers.cpp index 0c8d80939..76f78412a 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/fault_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/fault_handlers.cpp @@ -459,6 +459,20 @@ http::Result FaultHandlers::list_all_faults(const http::Ty } const auto filter = *filter_result; +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-declarations" + const auto & raw_req = req.raw_for_framework(); +#pragma GCC diagnostic pop + auto * agg = ctx_.aggregation_manager(); + const auto fan_out = plan_global_fault_fan_out(raw_req, agg ? agg->instance_id() : std::string{}); + if (fan_out.mode == GlobalFaultFanOut::Mode::kAlreadyCounted) { + // The request came back round a loop of peers. The first time it passed + // here this gateway's faults went into the answer being built. + json response = {{"items", json::array()}}; + response["x-medkit"] = dto::JsonWriter::write(dto::FaultListXMedkit{}); + return wrap_list_result(std::move(response)); + } + auto fault_mgr = ctx_.services().fault_manager(); // Empty source_id = no filtering, return all faults auto result = fault_mgr->list_faults("", filter.include_pending, filter.include_confirmed, filter.include_cleared, @@ -473,7 +487,6 @@ http::Result FaultHandlers::list_all_faults(const http::Ty // x-medkit extension for ros2_medkit-specific fields (typed DTO) dto::FaultListXMedkit xm; - xm.count = result.data.value("count", static_cast(0)); xm.muted_count = result.data.value("muted_count", static_cast(0)); xm.cluster_count = result.data.value("cluster_count", static_cast(0)); @@ -490,12 +503,9 @@ http::Result FaultHandlers::list_all_faults(const http::Ty // path used the raw `fan_out_get` here (not the typed `fan_out_collection`) // because the per-item shape on the global list includes vendor extensions // that JsonReader would drop; preserve that behaviour. - if (auto * agg = ctx_.aggregation_manager()) { -#pragma GCC diagnostic push -#pragma GCC diagnostic ignored "-Wdeprecated-declarations" - const auto & raw_req = req.raw_for_framework(); -#pragma GCC diagnostic pop - auto fan_result = agg->fan_out_get(raw_req.path, raw_req.get_header_value("Authorization")); + if (agg && fan_out.mode == GlobalFaultFanOut::Mode::kFanOut) { + auto fan_result = agg->fan_out_get(build_fan_out_path(raw_req), raw_req.get_header_value("Authorization"), + nullptr, {{kFanOutVisitedHeader, fan_out.visited}}); if (fan_result.merged_items.is_array()) { for (const auto & item : fan_result.merged_items) { response["items"].push_back(item); @@ -507,6 +517,20 @@ http::Result FaultHandlers::list_all_faults(const http::Ty xm.peer_failures = fan_result.peer_failures; } } + if (agg && fan_out.mode == GlobalFaultFanOut::Mode::kTooDeep) { + // The path is at its limit, so the peers behind this gateway were not + // asked. Saying nothing would report a truncated list as complete. + for (const auto & peer : agg->healthy_peer_endpoints()) { + xm.partial = true; + if (!xm.failed_peers) { + xm.failed_peers.emplace(); + xm.peer_failures.emplace(); + } + xm.failed_peers->push_back(peer.name); + xm.peer_failures->push_back(dto::PeerFailure{peer.name, "too-deep"}); + } + } + xm.count = static_cast(response["items"].size()); response["x-medkit"] = dto::JsonWriter::write(xm); return wrap_list_result(std::move(response)); diff --git a/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp index fcdebc62f..ee60acf18 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp @@ -166,6 +166,7 @@ http::Result HealthHandlers::get_health(const http::TypedRequest & // Add peer status when aggregation is active if (auto * agg = ctx_.aggregation_manager()) { response.peers = agg->get_peer_status(); + response.x_medkit_instance_id = agg->instance_id(); // Aggregation warnings are only collectable when there is an aggregation // manager; the array itself is not conditional on one. diff --git a/src/ros2_medkit_gateway/src/http/rest_server.cpp b/src/ros2_medkit_gateway/src/http/rest_server.cpp index 2e13aba68..8e22d8669 100644 --- a/src/ros2_medkit_gateway/src/http/rest_server.cpp +++ b/src/ros2_medkit_gateway/src/http/rest_server.cpp @@ -23,6 +23,7 @@ #include #include +#include "ros2_medkit_gateway/aggregation/aggregation_manager.hpp" #include "ros2_medkit_gateway/core/auth/auth_middleware.hpp" #include "ros2_medkit_gateway/core/fault_trigger_engine.hpp" #include "ros2_medkit_gateway/core/http/error_codes.hpp" @@ -2200,6 +2201,13 @@ void RESTServer::setup_routes() { // 503 when the fault store cannot be read - same branch as the // per-entity list, and equally out of the recorder's reach. .errors({503}) + .fan_out_aware() + .header_param(kFanOutVisitedHeader, + "Set by an aggregating gateway on its own fan-out of this operation: the instance ids of " + "the gateways the request has passed through, comma-separated. A gateway on that path " + "answers an empty list, any other answers for itself and asks those of its peers not on " + "it. Takes precedence over X-Medkit-No-Fan-Out. Clients do not send it.", + /*required=*/false) .operation_id("listAllFaults") .query(); diff --git a/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp b/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp index 47bc64937..16925b17a 100644 --- a/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp +++ b/src/ros2_medkit_gateway/src/plugins/plugin_context.cpp @@ -270,9 +270,11 @@ class GatewayPluginContext : public RosPluginContext { // Fan-out to peer gateways: fetch all faults from /api/v1/faults on each peer. // Auth: see list_entity_faults - the ROS 2 service path forwards no inbound // Authorization header, so peers are queried with an empty credential. + // The visited path is what lets a peer ask its own peers, so the list + // reaches past the first hop without coming back here. auto * agg = node_->get_aggregation_manager(); if (agg) { - auto fan_result = agg->fan_out_get("/api/v1/faults", ""); + auto fan_result = agg->fan_out_get("/api/v1/faults", "", nullptr, {{kFanOutVisitedHeader, agg->instance_id()}}); if (fan_result.merged_items.is_array()) { // Ensure local response has a "faults" array to append to. if (!response.contains("faults") || !response["faults"].is_array()) { diff --git a/src/ros2_medkit_gateway/test/test_aggregation_manager.cpp b/src/ros2_medkit_gateway/test/test_aggregation_manager.cpp index 673bffda5..cd017fe73 100644 --- a/src/ros2_medkit_gateway/test/test_aggregation_manager.cpp +++ b/src/ros2_medkit_gateway/test/test_aggregation_manager.cpp @@ -1796,6 +1796,153 @@ TEST(AggregationManager, fan_out_get_partial_results_one_peer_fails) { EXPECT_EQ(result.failed_peers[0], "peer_bad"); } +TEST(AggregationManager, instance_id_is_random_per_manager) { + AggregationConfig config; + config.enabled = true; + AggregationManager first(config); + AggregationManager second(config); + EXPECT_EQ(first.instance_id().size(), 32u); + EXPECT_EQ(first.instance_id().find_first_not_of("0123456789abcdef"), std::string::npos); + EXPECT_NE(first.instance_id(), second.instance_id()); +} + +TEST(AggregationManager, fan_out_get_sends_extra_headers_next_to_no_fan_out) { + MockPeerServer mock; + std::string captured_visited; + bool captured_no_fan_out = false; + mock.server().Get("/api/v1/health", [](const httplib::Request &, httplib::Response & res) { + res.set_content(R"({"status":"healthy"})", "application/json"); + }); + mock.server().Get("/api/v1/faults", [&](const httplib::Request & req, httplib::Response & res) { + captured_visited = req.get_header_value(kFanOutVisitedHeader); + captured_no_fan_out = req.has_header("X-Medkit-No-Fan-Out"); + res.set_content(R"({"items":[]})", "application/json"); + }); + int port = mock.start(); + + AggregationConfig config; + config.enabled = true; + config.timeout_ms = 5000; + AggregationConfig::PeerConfig peer; + peer.url = "http://127.0.0.1:" + std::to_string(port); + peer.name = "peer"; + config.peers.push_back(peer); + + AggregationManager manager(config); + manager.check_all_health(); + ASSERT_EQ(manager.healthy_peer_count(), 1u); + + auto result = manager.fan_out_get("/api/v1/faults", "", nullptr, {{kFanOutVisitedHeader, "aaaa,bbbb"}}); + + EXPECT_FALSE(result.is_partial); + EXPECT_EQ(captured_visited, "aaaa,bbbb"); + EXPECT_TRUE(captured_no_fan_out); +} + +namespace { + +/// A peer answering /health with `instance_id` and /faults with `faults_body`. +void serve_fault_peer(MockPeerServer & mock, const std::string & instance_id, const std::string & faults_body, + int * fault_calls) { + mock.server().Get("/api/v1/health", [instance_id](const httplib::Request &, httplib::Response & res) { + nlohmann::json body = {{"status", "healthy"}}; + if (!instance_id.empty()) { + body["x-medkit-instance-id"] = instance_id; + } + res.set_content(body.dump(), "application/json"); + }); + mock.server().Get("/api/v1/faults", [faults_body, fault_calls](const httplib::Request &, httplib::Response & res) { + ++*fault_calls; + res.set_content(faults_body, "application/json"); + }); +} + +AggregationConfig one_peer_config(int port) { + AggregationConfig config; + config.enabled = true; + config.timeout_ms = 5000; + AggregationConfig::PeerConfig peer; + peer.url = "http://127.0.0.1:" + std::to_string(port); + peer.name = "peer_b"; + config.peers.push_back(peer); + return config; +} + +} // namespace + +TEST(AggregationManager, fan_out_get_does_not_ask_a_peer_already_on_the_visited_path) { + MockPeerServer mock; + int fault_calls = 0; + serve_fault_peer(mock, "bbbb", R"({"items":[{"fault_code":"X"}]})", &fault_calls); + AggregationManager manager(one_peer_config(mock.start())); + manager.check_all_health(); + ASSERT_EQ(manager.healthy_peer_count(), 1u); + + auto skipped = manager.fan_out_get("/api/v1/faults", "", nullptr, {{kFanOutVisitedHeader, "aaaa, bbbb"}}); + EXPECT_EQ(fault_calls, 0); + EXPECT_TRUE(skipped.merged_items.empty()); + EXPECT_FALSE(skipped.is_partial); + + auto asked = manager.fan_out_get("/api/v1/faults", "", nullptr, {{kFanOutVisitedHeader, "aaaa"}}); + EXPECT_EQ(fault_calls, 1); + EXPECT_EQ(asked.merged_items.size(), 1u); +} + +TEST(AggregationManager, fan_out_get_asks_a_peer_whose_id_is_unknown) { + // A peer that predates the instance id cannot be recognised; it is asked and + // its own visited check answers. + MockPeerServer mock; + int fault_calls = 0; + serve_fault_peer(mock, "", R"({"items":[]})", &fault_calls); + AggregationManager manager(one_peer_config(mock.start())); + manager.check_all_health(); + ASSERT_EQ(manager.healthy_peer_count(), 1u); + + manager.fan_out_get("/api/v1/faults", "", nullptr, {{kFanOutVisitedHeader, "aaaa"}}); + EXPECT_EQ(fault_calls, 1); +} + +TEST(AggregationManager, fan_out_get_reports_a_peer_that_answered_partial) { + MockPeerServer mock; + int fault_calls = 0; + serve_fault_peer(mock, "bbbb", + R"({"items":[{"fault_code":"B"}],"x-medkit":{"partial":true,"failed_peers":["peer_c"],)" + R"("peer_failures":[{"peer":"peer_c","reason":"timeout"}]}})", + &fault_calls); + AggregationManager manager(one_peer_config(mock.start())); + manager.check_all_health(); + ASSERT_EQ(manager.healthy_peer_count(), 1u); + + auto result = manager.fan_out_get("/api/v1/faults", ""); + + EXPECT_EQ(result.merged_items.size(), 1u); + EXPECT_TRUE(result.is_partial); + ASSERT_EQ(result.failed_peers.size(), 1u); + EXPECT_EQ(result.failed_peers[0], "peer_b/peer_c"); + ASSERT_EQ(result.peer_failures.size(), 1u); + EXPECT_EQ(result.peer_failures[0].peer, "peer_b/peer_c"); + EXPECT_EQ(result.peer_failures[0].reason, "timeout"); +} + +TEST(AggregationManager, fan_out_get_reads_a_malformed_partial_peer_without_throwing) { + MockPeerServer mock; + int fault_calls = 0; + serve_fault_peer(mock, "bbbb", + R"({"items":[],"x-medkit":{"partial":true,"failed_peers":["peer_c",7],)" + R"("peer_failures":[{"peer":3},{"peer":"peer_c","reason":false}]}})", + &fault_calls); + AggregationManager manager(one_peer_config(mock.start())); + manager.check_all_health(); + ASSERT_EQ(manager.healthy_peer_count(), 1u); + + auto result = manager.fan_out_get("/api/v1/faults", ""); + + EXPECT_TRUE(result.is_partial); + ASSERT_EQ(result.peer_failures.size(), 1u); + EXPECT_EQ(result.peer_failures[0].peer, "peer_b/peer_c"); + EXPECT_EQ(result.peer_failures[0].reason, "unreachable"); +} + TEST(AggregationManager, fan_out_get_forwards_auth_header_when_enabled) { MockPeerServer mock; std::string captured_auth; diff --git a/src/ros2_medkit_gateway/test/test_fan_out_helpers.cpp b/src/ros2_medkit_gateway/test/test_fan_out_helpers.cpp index bb835e8f4..c41a8a21e 100644 --- a/src/ros2_medkit_gateway/test/test_fan_out_helpers.cpp +++ b/src/ros2_medkit_gateway/test/test_fan_out_helpers.cpp @@ -127,6 +127,100 @@ TEST(FanOutHelpers, build_fan_out_path_encodes_keys) { EXPECT_TRUE(result.find("a%3Db=val") != std::string::npos); } +// ============================================================================= +// plan_global_fault_fan_out +// ============================================================================= + +namespace { + +const std::string kSelf = "0123456789abcdef0123456789abcdef"; + +httplib::Request global_faults_request() { + httplib::Request req; + req.path = "/api/v1/faults"; + return req; +} + +} // namespace + +TEST(FanOutHelpers, global_fault_plan_client_request_fans_out_with_own_id) { + auto plan = plan_global_fault_fan_out(global_faults_request(), kSelf); + EXPECT_EQ(plan.mode, GlobalFaultFanOut::Mode::kFanOut); + EXPECT_EQ(plan.visited, kSelf); +} + +TEST(FanOutHelpers, global_fault_plan_without_aggregation_is_local) { + auto req = global_faults_request(); + req.headers.emplace(kFanOutVisitedHeader, "peer_a"); + EXPECT_EQ(plan_global_fault_fan_out(req, "").mode, GlobalFaultFanOut::Mode::kLocalOnly); +} + +TEST(FanOutHelpers, global_fault_plan_no_fan_out_alone_is_local) { + // A client opting out, or a gateway that predates the visited path. + auto req = global_faults_request(); + req.headers.emplace("X-Medkit-No-Fan-Out", "1"); + EXPECT_EQ(plan_global_fault_fan_out(req, kSelf).mode, GlobalFaultFanOut::Mode::kLocalOnly); +} + +TEST(FanOutHelpers, global_fault_plan_peer_request_extends_the_path) { + // The middle of a chain: B must still ask C, so No-Fan-Out does not stop it. + auto req = global_faults_request(); + req.headers.emplace("X-Medkit-No-Fan-Out", "1"); + req.headers.emplace(kFanOutVisitedHeader, "aaaa, bbbb"); + auto plan = plan_global_fault_fan_out(req, kSelf); + EXPECT_EQ(plan.mode, GlobalFaultFanOut::Mode::kFanOut); + EXPECT_EQ(plan.visited, "aaaa,bbbb," + kSelf); +} + +TEST(FanOutHelpers, global_fault_plan_own_id_on_the_path_is_already_counted) { + auto req = global_faults_request(); + req.headers.emplace(kFanOutVisitedHeader, "aaaa," + kSelf + ",bbbb"); + EXPECT_EQ(plan_global_fault_fan_out(req, kSelf).mode, GlobalFaultFanOut::Mode::kAlreadyCounted); +} + +TEST(FanOutHelpers, global_fault_plan_own_id_wins_over_an_unreadable_path) { + // Answering locally here would list this gateway's faults twice. + auto req = global_faults_request(); + req.headers.emplace(kFanOutVisitedHeader, std::string(kMaxFanOutVisitedIdLength + 1, 'x') + "," + kSelf); + EXPECT_EQ(plan_global_fault_fan_out(req, kSelf).mode, GlobalFaultFanOut::Mode::kAlreadyCounted); +} + +TEST(FanOutHelpers, global_fault_plan_long_path_is_not_extended) { + std::string path; + for (size_t i = 0; i < kMaxFanOutVisited; ++i) { + path += (i ? "," : "") + std::to_string(i); + } + auto req = global_faults_request(); + req.headers.emplace(kFanOutVisitedHeader, path); + EXPECT_EQ(plan_global_fault_fan_out(req, kSelf).mode, GlobalFaultFanOut::Mode::kTooDeep); +} + +TEST(FanOutHelpers, global_fault_plan_oversized_id_is_not_extended) { + auto req = global_faults_request(); + req.headers.emplace(kFanOutVisitedHeader, std::string(kMaxFanOutVisitedIdLength + 1, 'x')); + EXPECT_EQ(plan_global_fault_fan_out(req, kSelf).mode, GlobalFaultFanOut::Mode::kTooDeep); +} + +TEST(FanOutHelpers, global_fault_plan_one_short_of_the_limit_is_extended) { + std::string path; + for (size_t i = 0; i + 1 < kMaxFanOutVisited; ++i) { + path += (i ? "," : "") + std::to_string(i); + } + auto req = global_faults_request(); + req.headers.emplace(kFanOutVisitedHeader, path); + auto plan = plan_global_fault_fan_out(req, kSelf); + EXPECT_EQ(plan.mode, GlobalFaultFanOut::Mode::kFanOut); + EXPECT_EQ(plan.visited, path + "," + kSelf); +} + +TEST(FanOutHelpers, global_fault_plan_skips_empty_entries) { + auto req = global_faults_request(); + req.headers.emplace(kFanOutVisitedHeader, " , aaaa,,"); + auto plan = plan_global_fault_fan_out(req, kSelf); + EXPECT_EQ(plan.mode, GlobalFaultFanOut::Mode::kFanOut); + EXPECT_EQ(plan.visited, "aaaa," + kSelf); +} + // ============================================================================= // merge_peer_items // ============================================================================= diff --git a/src/ros2_medkit_integration_tests/CMakeLists.txt b/src/ros2_medkit_integration_tests/CMakeLists.txt index 501072fcd..b1b21bb2f 100644 --- a/src/ros2_medkit_integration_tests/CMakeLists.txt +++ b/src/ros2_medkit_integration_tests/CMakeLists.txt @@ -279,7 +279,8 @@ if(BUILD_TESTING) test_rest_logger_names test_aggregation_time_budgets test_peer_failure_reasons - test_relay_peer_credential) + test_relay_peer_credential + test_mutual_peer_faults) set(_MULTI_GATEWAY_DOMAINS 4) # Two-gateway tests that need the second gateway isolated from the first and diff --git a/src/ros2_medkit_integration_tests/test/features/test_mutual_peer_faults.test.py b/src/ros2_medkit_integration_tests/test/features/test_mutual_peer_faults.test.py new file mode 100644 index 000000000..a40fac689 --- /dev/null +++ b/src/ros2_medkit_integration_tests/test/features/test_mutual_peer_faults.test.py @@ -0,0 +1,270 @@ +#!/usr/bin/env python3 +# Copyright 2026 mfaferek93 +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""GET /faults across a loop of peers and down a chain. + +Gateway A lists B as a peer and B lists A, so the two aggregate each other. B +also lists C, a leaf: A -> B -> C is the chain topology from +docs/config/aggregation.rst. Each gateway runs its own faulty lidar and fault +manager on its own domain, so each holds one LIDAR_RANGE_INVALID. + +GET /faults on A must list all three, each once: + +- before the visited path, A asked B, B asked A again, and the request bounced + until the peer timeouts; +- answering a peer's request from the local fault manager alone stops the + bounce but drops C, because B never asks it; +- a gateway that answered a request coming back round the loop with its own + faults would list them twice; +- a gateway that sent the request back round the loop at all would hold a + worker on each side, so enough concurrent requests end at the peer timeout. +""" + +from concurrent.futures import ThreadPoolExecutor +import time +import unittest + +from launch import LaunchDescription +from launch.actions import SetEnvironmentVariable, TimerAction +import launch_testing +import launch_testing.actions +import requests + +from ros2_medkit_test_utils.constants import ( + ALLOWED_EXIT_CODES, + API_BASE_PATH, + DISCOVERY_INTERVAL, + DISCOVERY_TIMEOUT, + FAULT_TIMEOUT, + GATEWAY_STARTUP_INTERVAL, + GATEWAY_STARTUP_TIMEOUT, + get_test_domain_id, + get_test_port, + get_time_scale, +) +from ros2_medkit_test_utils.gateway_test_case import GatewayTestCase +from ros2_medkit_test_utils.launch_helpers import ( + create_demo_nodes, + create_fault_manager_node, + create_gateway_node, +) + +A_PORT = get_test_port(0) +B_PORT = get_test_port(1) +C_PORT = get_test_port(2) +A_URL = f'http://localhost:{A_PORT}{API_BASE_PATH}' +B_URL = f'http://localhost:{B_PORT}{API_BASE_PATH}' +C_URL = f'http://localhost:{C_PORT}{API_BASE_PATH}' +A_DOMAIN_ID = get_test_domain_id(0) +B_DOMAIN_ID = get_test_domain_id(1) +C_DOMAIN_ID = get_test_domain_id(2) + +PEER_TIMEOUT_MS = int(2000 * get_time_scale()) +FAULT_CODE = 'LIDAR_RANGE_INVALID' +NO_FAN_OUT = {'X-Medkit-No-Fan-Out': '1'} +# Pinned to the gateway's default worker count, and as many concurrent +# requests as that: enough to tie up every worker A has. +HTTP_WORKERS = 6 +CONCURRENT_REQUESTS = HTTP_WORKERS + + +def _aggregation_params(peers): + return { + 'server.http_thread_pool_size': HTTP_WORKERS, + 'aggregation.enabled': True, + 'aggregation.timeout_ms': PEER_TIMEOUT_MS, + 'aggregation.announce': False, + 'aggregation.discover': False, + 'aggregation.peer_urls': [f'http://localhost:{port}' for port, _ in peers], + 'aggregation.peer_names': [name for _, name in peers], + } + + +def generate_test_description(): + b_env = {'ROS_DOMAIN_ID': str(B_DOMAIN_ID)} + c_env = {'ROS_DOMAIN_ID': str(C_DOMAIN_ID)} + + gateway_a = create_gateway_node( + port=A_PORT, + extra_params=_aggregation_params([(B_PORT, 'gateway_b')]), + ) + gateway_b = create_gateway_node( + name='gateway_b_node', + port=B_PORT, + extra_params=_aggregation_params([(A_PORT, 'gateway_a'), (C_PORT, 'gateway_c')]), + extra_env=b_env, + ) + gateway_c = create_gateway_node( + name='gateway_c_node', + port=C_PORT, + extra_env=c_env, + ) + + delayed = TimerAction( + period=2.0, + actions=( + create_demo_nodes(['lidar_sensor'], lidar_faulty=True) + + create_demo_nodes(['lidar_sensor'], lidar_faulty=True, extra_env=b_env) + + create_demo_nodes(['lidar_sensor'], lidar_faulty=True, extra_env=c_env) + + [ + create_fault_manager_node(rosbag_enabled=False), + create_fault_manager_node(rosbag_enabled=False, extra_env=b_env), + create_fault_manager_node(rosbag_enabled=False, extra_env=c_env), + ] + ), + ) + + return ( + LaunchDescription([ + SetEnvironmentVariable('ROS_DOMAIN_ID', str(A_DOMAIN_ID)), + gateway_a, + gateway_b, + gateway_c, + delayed, + launch_testing.actions.ReadyToTest(), + ]), + {'gateway_node': gateway_a, 'gateway_b': gateway_b, 'gateway_c': gateway_c}, + ) + + +def _first_seen(response): + return [f.get('first_occurred') for f in response.json().get('items', []) + if f.get('fault_code') == FAULT_CODE] + + +class TestMutualPeerFaults(GatewayTestCase): + + BASE_URL = A_URL + REQUIRED_APPS = {'lidar_sensor'} + + @classmethod + def setUpClass(cls): + super().setUpClass() + for url in (B_URL, C_URL): + cls._wait_for_health(url) + cls._wait_for_peers_online(A_URL, {'gateway_b'}) + cls._wait_for_peers_online(B_URL, {'gateway_a', 'gateway_c'}) + # Every gateway reports the same fault code, so the faults are told + # apart by when each fault manager first saw its own. + cls.first_seen = { + url: cls._wait_for_local_fault(url) for url in (A_URL, B_URL, C_URL) + } + assert len(set(cls.first_seen.values())) == 3, cls.first_seen + + @staticmethod + def _wait_for_health(url): + deadline = time.monotonic() + GATEWAY_STARTUP_TIMEOUT + while time.monotonic() < deadline: + try: + if requests.get(f'{url}/health', timeout=2).status_code == 200: + return + except requests.exceptions.RequestException: + pass + time.sleep(GATEWAY_STARTUP_INTERVAL) + raise AssertionError(f'{url} not healthy after {GATEWAY_STARTUP_TIMEOUT}s') + + @staticmethod + def _wait_for_peers_online(url, names): + # A skips a peer it holds as unhealthy without marking the answer + # partial, so a test that runs before its health check lands would + # fail on correct code. + deadline = time.monotonic() + DISCOVERY_TIMEOUT + last = None + while time.monotonic() < deadline: + try: + r = requests.get(f'{url}/health', timeout=5) + if r.status_code == 200: + last = r.json().get('peers', []) + online = {p.get('name') for p in last if p.get('status') == 'online'} + if names <= online: + return + except requests.exceptions.RequestException as exc: + last = str(exc) + time.sleep(DISCOVERY_INTERVAL) + raise AssertionError(f'{url} never saw {names} online; last saw {last}') + + @staticmethod + def _wait_for_local_fault(url): + deadline = time.monotonic() + FAULT_TIMEOUT + while time.monotonic() < deadline: + try: + r = requests.get(f'{url}/faults', headers=NO_FAN_OUT, timeout=5) + if r.status_code == 200: + seen = _first_seen(r) + if seen: + return seen[0] + except requests.exceptions.RequestException: + pass + time.sleep(1.0) + raise AssertionError(f'{FAULT_CODE} never reached the fault manager behind {url}') + + def _assert_complete(self, response, gateways): + self.assertEqual(response.status_code, 200, response.text) + body = response.json() + x_medkit = body.get('x-medkit', {}) + self.assertNotIn('partial', x_medkit, x_medkit) + self.assertNotIn('failed_peers', x_medkit, x_medkit) + self.assertEqual(x_medkit.get('count'), len(body.get('items', [])), x_medkit) + self.assertEqual( + sorted(_first_seen(response)), + sorted(self.first_seen[url] for url in gateways), + f'expected the faults of {gateways}, one each') + + def test_no_fan_out_request_answers_local_faults_only(self): + """X-Medkit-No-Fan-Out alone is answered from the local fault manager.""" + r = requests.get(f'{A_URL}/faults', headers=NO_FAN_OUT, timeout=30) + self._assert_complete(r, [A_URL]) + + def test_head_of_chain_lists_every_gateway_once(self): + """A lists A, B and C, without bouncing between A and B.""" + start = time.monotonic() + r = requests.get(f'{A_URL}/faults', timeout=30) + elapsed = time.monotonic() - start + self._assert_complete(r, [A_URL, B_URL, C_URL]) + self.assertLess(elapsed, PEER_TIMEOUT_MS / 1000.0) + + def test_middle_of_chain_lists_every_gateway_once(self): + """B asks A and C, and A does not ask B back.""" + r = requests.get(f'{B_URL}/faults', timeout=30) + self._assert_complete(r, [A_URL, B_URL, C_URL]) + + def test_status_filter_reaches_every_peer(self): + """The query string goes to the peers, so they filter too.""" + r = requests.get(f'{A_URL}/faults', params={'status': 'cleared'}, timeout=30) + self._assert_complete(r, []) + r = requests.get(f'{A_URL}/faults', params={'status': 'all'}, timeout=30) + self._assert_complete(r, [A_URL, B_URL, C_URL]) + + def test_concurrent_requests_do_not_exhaust_workers(self): + """As many concurrent requests as A has workers all answer complete. + + A request sent back to A would wait for a free worker on A while every + worker waits on B, so each would end at the peer timeout. + """ + with ThreadPoolExecutor(max_workers=CONCURRENT_REQUESTS) as pool: + responses = list(pool.map( + lambda _: requests.get(f'{A_URL}/faults', timeout=30), + range(CONCURRENT_REQUESTS))) + for r in responses: + self._assert_complete(r, [A_URL, B_URL, C_URL]) + + +@launch_testing.post_shutdown_test() +class TestShutdown(unittest.TestCase): + + def test_exit_codes(self, proc_info): + for info in proc_info: + self.assertIn(info.returncode, ALLOWED_EXIT_CODES, + f'{info.process_name} exited with code {info.returncode}')