diff --git a/docs/api/rest.rst b/docs/api/rest.rst index 200795504..1c8e0336f 100644 --- a/docs/api/rest.rst +++ b/docs/api/rest.rst @@ -2158,7 +2158,11 @@ collection. - ``resource`` (string, required): Full SOVD resource URI to observe (e.g. ``/api/v1/apps/{id}/data/{topic}``, ``/api/v1/apps/{id}/faults``, ``/api/v1/functions/{id}/x-medkit-graph``). The URI ends at the collection - unless that collection's sampler narrows its payload to a named resource + unless that collection's sampler narrows its payload to a named resource. + The item path is percent-decoded as a GET path is, so an encoded topic or + plugin item name addresses the item GET of the same URI reads. A + member-qualified id (``member:/topic``) is not resolved by a subscription; + use the bare item id. - ``protocol`` (string, optional): Transport protocol. Only ``"sse"`` supported. Default: ``"sse"`` - ``interval`` (string, required): One of ``fast``, ``normal``, ``slow`` - ``duration`` (integer, required): Subscription lifetime in seconds. diff --git a/docs/config/server.rst b/docs/config/server.rst index 848dc55ae..1b1fc8c3b 100644 --- a/docs/config/server.rst +++ b/docs/config/server.rst @@ -696,7 +696,7 @@ Bulk Data Storage ----------------- Configure file-based bulk data storage for uploads (calibration files, firmware, etc.). -The ``rosbags`` category is always available via the Fault Manager and does not need +The ``rosbags`` category is served by the Fault Manager and does not need configuration. .. list-table:: @@ -733,8 +733,9 @@ Example: .. note:: - The ``rosbags`` category is always available through the Fault Manager and cannot be - used for uploads or deletes. It is automatically included in the category list. + The ``rosbags`` category is available through the Fault Manager and cannot be + used for uploads or deletes. It is included in the category list whenever the Fault + Manager serves rosbags, which is always the case on the ROS 2 node. .. note:: diff --git a/scripts/check_cores_without_ros/README.md b/scripts/check_cores_without_ros/README.md index 2469f83e3..360ca0097 100644 --- a/scripts/check_cores_without_ros/README.md +++ b/scripts/check_cores_without_ros/README.md @@ -17,6 +17,13 @@ not installed, so the PostgreSQL backend is not built. ## What `main.cpp` checks +- Cyclic subscriptions: with `register_builtin_samplers()` and the SSE transport, a subscription on a data resource (served by + a controlled `TopicDataProvider`), on a resource a plugin's `DataProvider` serves, and on faults each stream two events whose + values differ. +- Bulk data: a store built by `make_bulk_data_store()` takes an upload over HTTP, lists it, returns it byte-equal and deletes + it; an upload past `max_upload_size` is refused; the categories do not contain `rosbags` when the fault manager serves none. +- `GET /` capabilities: `bulk_data` and `cyclic_subscriptions` are false for services without a store, rosbags, transport or + sampler, and true as soon as a store or rosbags (bulk data), or a transport and a sampler (subscriptions), exist. - Gateway: with `server_name`, `vendor_name` and `server_version` set in `RestConfig`, `GET /api/v1/` returns the name and version, and `GET /api/v1/version-info` returns the vendor name and version. - Reporter core: a report below the local filter threshold is filtered, the next one reaches the sink, a PASSED report is diff --git a/scripts/check_cores_without_ros/main.cpp b/scripts/check_cores_without_ros/main.cpp index 6a8cf8d90..ff817f24d 100644 --- a/scripts/check_cores_without_ros/main.cpp +++ b/scripts/check_cores_without_ros/main.cpp @@ -17,12 +17,18 @@ #include #include +#include +#include #include #include #include #include #include +#include +#include #include +#include +#include #include #include #include @@ -40,18 +46,25 @@ #include "ros2_medkit_fault_manager/sqlite_fault_storage.hpp" #include "ros2_medkit_fault_reporter/core/reporter_core.hpp" #include "ros2_medkit_gateway/core/auth/auth_config.hpp" +#include "ros2_medkit_gateway/core/builtin_samplers.hpp" #include "ros2_medkit_gateway/core/config.hpp" +#include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" #include "ros2_medkit_gateway/core/discovery/entity_discovery.hpp" #include "ros2_medkit_gateway/core/gateway_services.hpp" +#include "ros2_medkit_gateway/core/http/handlers/sse_transport_provider.hpp" #include "ros2_medkit_gateway/core/http/rate_limiter.hpp" #include "ros2_medkit_gateway/core/http/rest_server.hpp" #include "ros2_medkit_gateway/core/http/sse_client_tracker.hpp" +#include "ros2_medkit_gateway/core/managers/bulk_data_store.hpp" #include "ros2_medkit_gateway/core/managers/configuration_manager.hpp" #include "ros2_medkit_gateway/core/managers/data_access_manager.hpp" #include "ros2_medkit_gateway/core/managers/fault_manager.hpp" #include "ros2_medkit_gateway/core/managers/operation_manager.hpp" #include "ros2_medkit_gateway/core/managers/subscription_manager.hpp" #include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp" +#include "ros2_medkit_gateway/core/plugins/gateway_plugin.hpp" +#include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/core/providers/data_provider.hpp" #include "ros2_medkit_gateway/core/resource_sampler.hpp" #include "ros2_medkit_gateway/core/subscription_transport.hpp" #include "ros2_medkit_gateway/core/transports/action_transport.hpp" @@ -118,10 +131,17 @@ class NoFaultTransport : public gw::FaultServiceTransport { const std::string & /*description*/, const std::string & /*source_id*/) override { return none(); } + // Once set, every list answers one fault whose occurrence_count grows with each call. gw::FaultResult list_faults(const std::string & /*source_id*/, bool /*include_prefailed*/, bool /*include_confirmed*/, bool /*include_cleared*/, bool /*include_healed*/, bool /*include_muted*/, bool /*include_clusters*/) override { - return none(); + if (!reports_faults.load()) { + return none(); + } + gw::json fault = {{"fault_code", "CHECK_FAULT"}, + {"occurrence_count", ++list_calls}, + {"reporting_sources", gw::json::array({"hosted_app"})}}; + return {true, {{"faults", gw::json::array({fault})}}, {}, gw::FaultFailure::Declined}; } gw::FaultWithEnvJsonResult get_fault_with_env(const std::string & /*fault_code*/, const std::string & /*source_id*/) override { @@ -148,6 +168,12 @@ class NoFaultTransport : public gw::FaultServiceTransport { bool is_available() const override { return false; } + bool serves_rosbags() const override { + return rosbags.load(); + } + std::atomic reports_faults{false}; + std::atomic rosbags{false}; + std::atomic list_calls{0}; private: static gw::FaultResult none() { @@ -155,6 +181,90 @@ class NoFaultTransport : public gw::FaultServiceTransport { } }; +// Answers every sample with a value that grows by one per call. +class CountingTopicProvider : public gw::TopicDataProvider { + public: + tl::expected sample(const std::string & topic, + std::chrono::milliseconds /*timeout*/) override { + gw::TopicSampleResult out; + out.topic_name = topic; + out.message_type = "check/msg/Counter"; + out.data = gw::json{{"value", ++samples}}; + out.has_data = true; + return out; + } + tl::expected, gw::ErrorInfo> + sample_parallel(const std::vector & topics, std::chrono::milliseconds timeout) override { + std::vector out; + out.reserve(topics.size()); + for (const auto & topic : topics) { + out.push_back(*sample(topic, timeout)); + } + return out; + } + std::optional get_topic_info(const std::string & /*topic*/) override { + return std::nullopt; + } + bool has_publishers(const std::string & /*topic*/) override { + return true; + } + std::vector discover(const std::string & /*namespace_prefix*/) override { + return {}; + } + std::vector discover_all() override { + return {}; + } + std::map build_component_topic_map() override { + return {}; + } + gw::ComponentTopics get_component_topics(const std::string & /*component_fqn*/) override { + return {}; + } + gw::TopicDiscoveryResult discover_topics_by_namespace() override { + return {}; + } + std::set discover_topic_namespaces() override { + return {}; + } + gw::ComponentTopics get_topics_for_namespace(const std::string & /*ns_prefix*/) override { + return {}; + } + std::vector get_topic_publishers(const std::string & /*topic*/) override { + return {}; + } + std::vector get_topic_subscribers(const std::string & /*topic*/) override { + return {}; + } + gw::TopicConnection get_topic_connection(const std::string & /*topic*/) override { + return {}; + } + std::atomic samples{0}; +}; + +// A plugin that serves the data of the entities it owns; every read counts up. +class CheckDataPlugin : public gw::GatewayPlugin, public gw::DataProvider { + public: + std::string name() const override { + return "check_plugin"; + } + void configure(const gw::json & /*config*/) override { + } + tl::expected + list_data(const std::string & /*entity_id*/) override { + return gw::dto::DataListResult{{{"items", gw::json::array()}}}; + } + tl::expected read_data(const std::string & /*entity_id*/, + const std::string & resource_name) override { + return gw::dto::DataValue{{{"source", "plugin"}, {"name", resource_name}, {"value", ++reads}}}; + } + tl::expected write_data(const std::string & /*entity_id*/, + const std::string & /*resource_name*/, + const gw::json & /*value*/) override { + return tl::make_unexpected(gw::DataProviderErrorInfo{gw::DataProviderError::ReadOnly, "read only", 405}); + } + std::atomic reads{0}; +}; + class NoParameterTransport : public gw::ParameterTransport { public: bool is_self_node(const std::string & /*node_name*/) const override { @@ -302,7 +412,8 @@ class NoTypeSchemas : public gw::TypeSchemaProvider { class MinimalServices : public gw::GatewayServices { public: MinimalServices() - : fault_(std::make_shared()) + : fault_transport_(std::make_shared()) + , fault_(fault_transport_) , config_(std::make_shared()) , data_(std::make_shared()) , operations_(std::make_shared(), std::make_shared(), nullptr) @@ -345,7 +456,7 @@ class MinimalServices : public gw::GatewayServices { return nullptr; } gw::PluginManager * plugin_manager() override { - return nullptr; + return plugins_.get(); } gw::ScriptManager * script_manager() override { return nullptr; @@ -360,7 +471,7 @@ class MinimalServices : public gw::GatewayServices { return nullptr; } gw::BulkDataStore * bulk_data_store() override { - return nullptr; + return bulk_store_.get(); } gw::FaultTriggerEngine * fault_trigger_engine() override { return nullptr; @@ -378,10 +489,32 @@ class MinimalServices : public gw::GatewayServices { return nullptr; } + gw::ThreadSafeEntityCache & mutable_cache() { + return cache_; + } + NoFaultTransport & fault_transport() { + return *fault_transport_; + } + // Plugins exist only after this call; the plugin owns the entity ids it is given. + CheckDataPlugin & add_data_plugin(const std::vector & owned_entities) { + plugins_ = std::make_unique(); + auto plugin = std::make_unique(); + auto & ref = *plugin; + plugins_->add_plugin(std::move(plugin)); + plugins_->register_entity_ownership("check_plugin", owned_entities); + return ref; + } + void set_bulk_store(std::unique_ptr store) { + bulk_store_ = std::move(store); + } + private: gw::ThreadSafeEntityCache cache_; EmptyDiscovery discovery_; NoTypeSchemas schemas_; + std::shared_ptr fault_transport_; + std::unique_ptr bulk_store_; + std::unique_ptr plugins_; gw::FaultManager fault_; gw::ConfigurationManager config_; gw::DataAccessManager data_; @@ -411,6 +544,122 @@ bool check(bool ok, const char * what) { std::printf("%s: %s\n", ok ? "ok " : "FAIL", what); return ok; } + +// A REST server on an ephemeral port, stopped and joined on scope exit. +class RunningServer { + public: + explicit RunningServer(MinimalServices & services) : port_(free_port()) { + gw::RestConfig rest; + rest.http_thread_pool_size = 6; + server_ = std::make_unique(services, nullptr, "127.0.0.1", port_, gw::CorsConfig{}, + gw::AuthConfig{}, gw::RateLimitConfig{}, rest); + thread_ = std::thread([this] { + server_->start(); + }); + for (int i = 0; i < 100 && !server_->is_running(); ++i) { + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + } + } + ~RunningServer() { + server_->stop(); + thread_.join(); + } + RunningServer(const RunningServer &) = delete; + RunningServer & operator=(const RunningServer &) = delete; + RunningServer(RunningServer &&) = delete; + RunningServer & operator=(RunningServer &&) = delete; + int port() const { + return port_; + } + + private: + int port_; + std::unique_ptr server_; + std::thread thread_; +}; + +// Reads SSE frames from `path` until `want` of them arrived, the stream ended or the read timed out. +std::vector read_events(httplib::Client & client, const std::string & path, size_t want) { + std::vector events; + std::string buffer; + client.Get( + path, httplib::Headers{}, + [](const httplib::Response &) { + return true; + }, + [&](const char * data, size_t len) { + buffer.append(data, len); + for (size_t end = buffer.find("\n\n"); end != std::string::npos; end = buffer.find("\n\n")) { + const std::string frame = buffer.substr(0, end); + buffer.erase(0, end + 2); + if (frame.rfind("data: ", 0) == 0) { + events.push_back(nlohmann::json::parse(frame.substr(6), nullptr, false)); + } + } + return events.size() < want; + }); + return events; +} + +httplib::Client make_client(int port) { + httplib::Client client("127.0.0.1", port); + client.set_connection_timeout(5); + client.set_read_timeout(5); + return client; +} + +// Creates a fast cyclic subscription on `resource`; returns {subscription id, events path}. +std::pair subscribe(httplib::Client & client, const std::string & resource, + const std::string & app = "hosted_app") { + const nlohmann::json body = {{"resource", resource}, {"interval", "fast"}, {"duration", 60}}; + auto res = client.Post("/api/v1/apps/" + app + "/cyclic-subscriptions", body.dump(), "application/json"); + if (!res || res->status != 201) { + std::printf("subscribe %s -> %d %s\n", resource.c_str(), res ? res->status : 0, res ? res->body.c_str() : ""); + return {}; + } + const auto json = nlohmann::json::parse(res->body, nullptr, false); + return {json.value("id", ""), json.value("event_source", "")}; +} + +// True when the first two sampled payloads exist and `field` differs between them. +bool values_change(const std::vector & events, + const std::function & pick) { + if (events.size() < 2 || !events[0].contains("payload") || !events[1].contains("payload")) { + return false; + } + const auto first = pick(events[0]["payload"]); + const auto second = pick(events[1]["payload"]); + return !first.is_null() && !second.is_null() && first != second; +} + +// `GET /` capabilities of a server over `services`. +nlohmann::json root_capabilities(MinimalServices & services) { + RunningServer running(services); + auto client = make_client(running.port()); + auto res = client.Get("/api/v1/"); + if (!res || res->status != 200) { + return nlohmann::json::object(); + } + return nlohmann::json::parse(res->body, nullptr, false).value("capabilities", nlohmann::json::object()); +} + +gw::App hosted_app() { + gw::App app; + app.id = "hosted_app"; + app.name = "Hosted app"; + app.external = true; + return app; +} + +void add_sse_transport(MinimalServices & services) { + services.transport_registry()->register_transport( + std::make_unique(*services.subscription_manager(), services.sse_client_tracker())); + services.subscription_manager()->set_on_removed([&services](const gw::CyclicSubscriptionInfo & info) { + if (auto * transport = services.transport_registry()->get_transport(info.protocol)) { + transport->stop(info.id); + } + }); +} } // namespace static int run() { @@ -709,6 +958,194 @@ static int run() { server_thread.join(); } + // A host that registers the samplers, a transport and a bulk-data store gets subscriptions and bulk data. + { + MinimalServices services; + CountingTopicProvider topics; + services.data_access_manager()->set_topic_data_provider(&topics); + services.fault_transport().reports_faults = true; + auto plugin_app = hosted_app(); + plugin_app.id = "plugin_app"; + services.mutable_cache().update_apps({hosted_app(), plugin_app}); + services.add_data_plugin({"plugin_app"}); + const auto store_dir = std::filesystem::temp_directory_path() / "check_cores_without_ros_bulk"; + std::filesystem::remove_all(store_dir); + gw::BulkDataConfig bulk; + bulk.storage_dir = store_dir.string(); + bulk.max_upload_size = 1024; + bulk.categories = {"calibration", ""}; + services.set_bulk_store(gw::make_bulk_data_store(bulk)); + gw::register_builtin_samplers(services); + add_sse_transport(services); + { + RunningServer running(services); + auto client = make_client(running.port()); + + // Cyclic subscription on a data resource. + const auto data_sub = subscribe(client, "/api/v1/apps/hosted_app/data/temperature"); + ok &= check(!data_sub.second.empty(), "a cyclic subscription on a data resource is created"); + const auto data_events = + data_sub.second.empty() ? std::vector{} : read_events(client, data_sub.second, 2); + ok &= check(data_events.size() == 2, "the data subscription streams two events"); + ok &= check(values_change(data_events, + [](const nlohmann::json & p) { + return p.contains("data") ? p["data"].value("value", nlohmann::json()) + : nlohmann::json(); + }), + "the data value differs between the two events"); + ok &= check(!data_events.empty() && data_events[0].contains("payload") && + data_events[0]["payload"].value("id", "") == "/temperature", + "the data payload names the resource"); + if (!data_sub.first.empty()) { + client.Delete("/api/v1/apps/hosted_app/cyclic-subscriptions/" + data_sub.first); + } + + // Cyclic subscription on a resource a plugin serves. + const auto plugin_sub = subscribe(client, "/api/v1/apps/plugin_app/data/speed", "plugin_app"); + ok &= check(!plugin_sub.second.empty(), "a cyclic subscription on a plugin-served data resource is created"); + const auto plugin_events = + plugin_sub.second.empty() ? std::vector{} : read_events(client, plugin_sub.second, 2); + ok &= check(plugin_events.size() == 2, "the plugin-data subscription streams two events"); + ok &= check(!plugin_events.empty() && plugin_events[0].contains("payload") && + plugin_events[0]["payload"].value("source", "") == "plugin" && + plugin_events[0]["payload"].value("name", "") == "speed", + "the plugin-data payload comes from the plugin's provider"); + ok &= check(values_change(plugin_events, + [](const nlohmann::json & p) { + return p.value("value", nlohmann::json()); + }), + "the plugin value differs between the two events"); + if (!plugin_sub.first.empty()) { + client.Delete("/api/v1/apps/plugin_app/cyclic-subscriptions/" + plugin_sub.first); + } + + // Cyclic subscription on faults. + const auto fault_sub = subscribe(client, "/api/v1/apps/hosted_app/faults"); + ok &= check(!fault_sub.second.empty(), "a cyclic subscription on faults is created"); + const auto fault_events = + fault_sub.second.empty() ? std::vector{} : read_events(client, fault_sub.second, 2); + ok &= check(fault_events.size() == 2, "the faults subscription streams two events"); + ok &= check(values_change(fault_events, + [](const nlohmann::json & p) { + return p.contains("faults") && p["faults"].is_array() && !p["faults"].empty() + ? p["faults"][0].value("occurrence_count", nlohmann::json()) + : nlohmann::json(); + }), + "the fault occurrence_count differs between the two events"); + if (!fault_sub.first.empty()) { + client.Delete("/api/v1/apps/hosted_app/cyclic-subscriptions/" + fault_sub.first); + } + + // Bulk data: upload, list, download, delete. + const std::string payload("bulk\0payload\xff\x01", 15); + const std::string base = "/api/v1/apps/hosted_app/bulk-data"; + httplib::MultipartFormDataItems items = {{"file", payload, "cal.bin", "application/octet-stream"}}; + auto uploaded = client.Post(base + "/calibration", items); + ok &= check(uploaded && uploaded->status == 201, "an upload to a configured category answers 201"); + const std::string item_id = uploaded && uploaded->status == 201 + ? nlohmann::json::parse(uploaded->body, nullptr, false).value("id", "") + : ""; + auto listed = client.Get(base + "/calibration"); + const auto listed_json = listed ? nlohmann::json::parse(listed->body, nullptr, false) : nlohmann::json(); + ok &= + check(listed && listed->status == 200 && listed_json.contains("items") && listed_json["items"].size() == 1 && + listed_json["items"][0].value("id", "") == item_id && !item_id.empty(), + "the uploaded item is listed"); + auto downloaded = client.Get(base + "/calibration/" + item_id); + ok &= check(downloaded && downloaded->status == 200 && downloaded->body == payload, "the download is byte-equal"); + auto removed = client.Delete(base + "/calibration/" + item_id); + ok &= check(removed && removed->status == 204, "the delete answers 204"); + auto relisted = client.Get(base + "/calibration"); + const auto relisted_json = relisted ? nlohmann::json::parse(relisted->body, nullptr, false) : nlohmann::json(); + ok &= check(relisted && relisted_json.contains("items") && relisted_json["items"].empty(), + "the deleted item is gone"); + auto too_big = client.Post( + base + "/calibration", + httplib::MultipartFormDataItems{{"file", std::string(2048, 'x'), "big.bin", "application/octet-stream"}}); + ok &= check(too_big && too_big->status >= 400, "an upload past max_upload_size is refused"); + + // Categories: the store's own, and no rosbags when the fault manager serves none. + auto categories = client.Get(base); + const auto categories_json = + categories ? nlohmann::json::parse(categories->body, nullptr, false) : nlohmann::json(); + const auto cat_items = categories_json.value("items", nlohmann::json::array()); + ok &= check(categories && categories->status == 200 && + std::find(cat_items.begin(), cat_items.end(), nlohmann::json("calibration")) != cat_items.end(), + "the categories list the configured category"); + ok &= check(std::find(cat_items.begin(), cat_items.end(), nlohmann::json("rosbags")) == cat_items.end(), + "the categories do not list rosbags when the fault manager serves none"); + + // A host without rosbags answers rosbags like any category that does not exist. + const auto error_code_of = [](const httplib::Result & r) { + return r ? nlohmann::json::parse(r->body, nullptr, false).value("error_code", "") : std::string{}; + }; + auto no_rosbag_list = client.Get(base + "/rosbags"); + auto no_such_list = client.Get(base + "/no_such_category"); + ok &= check(no_such_list && no_such_list->status == 404, "an unknown category list answers 404"); + ok &= check(no_rosbag_list && no_such_list && no_rosbag_list->status == no_such_list->status && + error_code_of(no_rosbag_list) == error_code_of(no_such_list), + "listing rosbags without rosbags answers like an unknown category"); + auto no_rosbag_item = client.Get(base + "/rosbags/some-recording"); + auto no_such_item = client.Get(base + "/no_such_category/some-recording"); + ok &= check(no_rosbag_item && no_rosbag_item->body.find("Unknown category: rosbags") != std::string::npos, + "the rosbag download error names the unknown category"); + ok &= check(no_rosbag_item && no_such_item && no_rosbag_item->status == no_such_item->status && + error_code_of(no_rosbag_item) == error_code_of(no_such_item), + "downloading a rosbag without rosbags answers like an unknown category"); + + auto root = client.Get("/api/v1/"); + const auto caps = + root ? nlohmann::json::parse(root->body, nullptr, false).value("capabilities", nlohmann::json::object()) + : nlohmann::json::object(); + ok &= check(caps.value("bulk_data", false), "GET / reports bulk_data true with a store"); + ok &= check(caps.value("cyclic_subscriptions", false), + "GET / reports cyclic_subscriptions true with samplers and a transport"); + } + services.transport_registry()->shutdown_all(*services.subscription_manager()); + std::filesystem::remove_all(store_dir); + } + + // Capabilities follow what the host provides. + { + MinimalServices bare; + const auto caps = root_capabilities(bare); + ok &= + check(caps.contains("bulk_data") && !caps.value("bulk_data", true), "no store and no rosbags: bulk_data false"); + ok &= check(caps.contains("cyclic_subscriptions") && !caps.value("cyclic_subscriptions", true), + "no transport and no sampler: cyclic_subscriptions false"); + + MinimalServices transport_only; + add_sse_transport(transport_only); + const auto transport_caps = root_capabilities(transport_only); + ok &= check(transport_caps.contains("cyclic_subscriptions") && !transport_caps.value("cyclic_subscriptions", true), + "a transport without a sampler: cyclic_subscriptions false"); + transport_only.transport_registry()->shutdown_all(*transport_only.subscription_manager()); + + MinimalServices sampler_only; + sampler_only.sampler_registry()->register_sampler( + "x-check", [](const std::string &, const std::string &) -> tl::expected { + return nlohmann::json::object(); + }); + const auto sampler_caps = root_capabilities(sampler_only); + ok &= check(sampler_caps.contains("cyclic_subscriptions") && !sampler_caps.value("cyclic_subscriptions", true), + "a sampler without a transport: cyclic_subscriptions false"); + + MinimalServices store_only; + gw::BulkDataConfig store_only_config; + store_only_config.storage_dir = + (std::filesystem::temp_directory_path() / "check_cores_without_ros_store_only").string(); + store_only.set_bulk_store(gw::make_bulk_data_store(store_only_config)); + store_only.fault_transport().rosbags = false; + const auto store_caps = root_capabilities(store_only); + ok &= check(store_caps.value("bulk_data", false), "a store without rosbags: bulk_data true"); + std::filesystem::remove_all(std::filesystem::temp_directory_path() / "check_cores_without_ros_store_only"); + + MinimalServices rosbags_only; + rosbags_only.fault_transport().rosbags = true; + const auto rosbag_caps = root_capabilities(rosbags_only); + ok &= check(rosbag_caps.value("bulk_data", false), "rosbags without a store: bulk_data true"); + } + return ok ? 0 : 1; } diff --git a/src/ros2_medkit_gateway/README.md b/src/ros2_medkit_gateway/README.md index 04125c74c..b24178ed9 100644 --- a/src/ros2_medkit_gateway/README.md +++ b/src/ros2_medkit_gateway/README.md @@ -66,7 +66,7 @@ All endpoints are prefixed with `/api/v1` for API versioning. ### Bulk Data Endpoints -- `GET /api/v1/{entity}/{id}/bulk-data` - List bulk-data categories (rosbags + configured) +- `GET /api/v1/{entity}/{id}/bulk-data` - List bulk-data categories (configured, plus `rosbags` when the fault manager serves them) - `GET /api/v1/{entity}/{id}/bulk-data/{category}` - List bulk-data items in a category - `GET /api/v1/{entity}/{id}/bulk-data/{category}/{item_id}` - Download a bulk-data file - `POST /api/v1/{entity}/{id}/bulk-data/{category}` - Upload bulk data (components/apps only) @@ -1761,6 +1761,11 @@ target_link_libraries(check_cores_without_ros PRIVATE gateway_core) The file takes a target the including build already defines, else `find_package`, else the copy vendored in `src/vendored/` (cpp-httplib, jwt-cpp, tl::expected). An existing `httplib::httplib` target is reused when it is older than 0.20 (or has no version), because 0.20 removed the multipart API the handlers use. `gateway_core` requires C++17 from its consumers (`cxx_std_17` is a public compile feature) and works with CMake 3.16. `-DENABLE_SWAGGER_UI=ON` works in both builds and needs network access and `xxd` at configure time. The check implements `GatewayServices` with minimal stand-ins and starts `RESTServer`. A program that hosts the core names its server through `RestConfig::server_name`, `vendor_name` and `server_version` (`name` and `version` on `GET /`, `vendor_info` on `GET /version-info`, and the OpenAPI title and version). The defaults are the ROS 2 node's values. `scripts/check_cores_without_ros/` builds and runs it on Ubuntu 20.04 with gcc 9, the oldest compiler the cores are checked with. +A program that hosts the core gets the cyclic-subscription samplers and the bulk-data store from the core, and passes the following: + +- **Samplers.** `register_builtin_samplers(services)` (`core/builtin_samplers.hpp`) registers `data`, `faults`, `configurations` and `logs` on `services.sampler_registry()`, plus `updates` when `services.update_manager()` is set. Each sampler reads through the same managers and helpers as its GET route. A `data` subscription on an entity a plugin owns reads through that plugin's `DataProvider` (or its `x-plc-data` route), and on any other entity through `services.data_access_manager()` and its `TopicDataProvider`. A collection with no manager behind it answers with the error text of its GET route, not a crash. The host passes `GatewayServices` (it must outlive the samplers) and registers a subscription transport itself, for example `SseTransportProvider`. `GET /` reports `cyclic_subscriptions` true only when a transport and at least one sampler are registered. +- **Bulk data.** `BulkDataConfig` (`core/config.hpp`: `storage_dir`, `max_upload_size`, `categories`, with the defaults of the `bulk_data.*` parameters) and `make_bulk_data_store(config)` (`core/managers/bulk_data_store.hpp`) build the store. The host returns it from `GatewayServices::bulk_data_store()` before it starts `RESTServer`, which sizes the upload limit from it. The `rosbags` category is listed only when `FaultServiceTransport::serves_rosbags()` is true; otherwise a request for it answers 404 like any unknown category. The default is true; a host whose fault manager records no rosbags overrides it. `GET /` reports `bulk_data` true only when a store exists or rosbags are served. + The `gateway_node` executable and existing test targets link `gateway_ros2`, so they transitively get both layers from a single dependency. The neutral contract is enforced by two CTest checks: - `gateway_core_purity` (linter label) - greps `core/` for any ROS-package include and fails on any match. diff --git a/src/ros2_medkit_gateway/design/index.rst b/src/ros2_medkit_gateway/design/index.rst index 09c3e0799..855db0262 100644 --- a/src/ros2_medkit_gateway/design/index.rst +++ b/src/ros2_medkit_gateway/design/index.rst @@ -35,6 +35,28 @@ The check implements ``GatewayServices`` with minimal stand-ins and starts defaults are the ROS 2 node's values. ``scripts/check_cores_without_ros/`` builds and runs it on Ubuntu 20.04 with gcc 9, the oldest compiler the cores are checked with. +A program that hosts the core also gets the cyclic-subscription samplers and the bulk-data +store from it: + +* ``register_builtin_samplers(services)`` (``core/builtin_samplers.hpp``) + registers ``data``, ``faults``, ``configurations`` and ``logs`` on + ``services.sampler_registry()``, plus ``updates`` when + ``services.update_manager()`` is set. Each sampler reads through the + managers and helpers of its GET route. ``data`` on a plugin-owned entity + goes through the plugin's ``DataProvider`` (``core/data/data_item_reader.hpp``, + shared with ``GET /{entity}/data/{item}``); on any other entity it goes + through ``DataAccessManager`` and its ``TopicDataProvider``. The host passes + ``GatewayServices``, which must outlive the samplers, and registers a + subscription transport itself. +* ``BulkDataConfig`` (``core/config.hpp``) and ``make_bulk_data_store()`` + (``core/managers/bulk_data_store.hpp``) build the store. The host returns it + from ``GatewayServices::bulk_data_store()`` before it starts ``RESTServer``. + The ``rosbags`` category is listed only when + ``FaultServiceTransport::serves_rosbags()`` is true (the default). +* ``GET /`` reports ``bulk_data`` true only when a store exists or rosbags are + served, and ``cyclic_subscriptions`` true only when a transport and at least + one sampler are registered. + The ``gateway_node`` executable and existing test targets link ``gateway_ros2``, so they transitively get both layers from a single dependency. The neutral contract is enforced by two CTest checks: diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/builtin_samplers.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/builtin_samplers.hpp new file mode 100644 index 000000000..087d51d99 --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/builtin_samplers.hpp @@ -0,0 +1,31 @@ +// Copyright 2026 bburda +// +// 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. + +#pragma once + +namespace ros2_medkit_gateway { + +class GatewayServices; + +/// Registers the built-in cyclic-subscription samplers on `services.sampler_registry()`: +/// `data`, `faults`, `configurations` and `logs`, plus `updates` when `services.update_manager()` +/// is set. Each sampler reads through the managers and helpers of the matching GET route, so a +/// resource the route serves can be subscribed to, whoever provides it. A collection the host has +/// no manager for answers with the error text of its GET route. +/// +/// `services` must outlive every sampler: the samplers keep a reference and read the manager +/// getters on each call. Call it once, before the first subscription is created. +void register_builtin_samplers(GatewayServices & services); + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp index 9fbc5bdaa..2472b9aa3 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/config.hpp @@ -46,6 +46,16 @@ struct RestConfig : ServerIdentity { std::string logger_name{"rest_server"}; }; +/// Where uploaded bulk data goes and what a host accepts. The defaults are the node's `bulk_data.*` defaults. +struct BulkDataConfig { + /// Base directory of the stored files. + std::string storage_dir{"/tmp/ros2_medkit_bulk_data"}; + /// Largest accepted upload in bytes. 0 means unlimited. A negative value is refused and the default is used. + int64_t max_upload_size{104857600}; + /// Accepted upload categories. Empty names are dropped. `rosbags` is never listed here. + std::vector categories; +}; + /** * @brief TLS/HTTPS configuration settings * diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/data_item_reader.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/data_item_reader.hpp new file mode 100644 index 000000000..6d4236327 --- /dev/null +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/data/data_item_reader.hpp @@ -0,0 +1,51 @@ +// Copyright 2026 bburda +// +// 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. + +#pragma once + +#include + +#include +#include + +#include "ros2_medkit_gateway/core/data/data_types.hpp" +#include "ros2_medkit_gateway/core/models/error_info.hpp" +#include "ros2_medkit_gateway/core/providers/data_provider.hpp" +#include "ros2_medkit_gateway/dto/data.hpp" + +namespace ros2_medkit_gateway { + +class DataAccessManager; +class PluginManager; + +/// An `x-medkit-plugin-error` with the status clamped to [400, 599] and the message cut to 512 characters. +ErrorInfo make_plugin_error(int http_status, const std::string & message, nlohmann::json extra_params = {}); + +/// The plugin error for a failed `DataProvider` call. +ErrorInfo make_provider_error(const DataProviderErrorInfo & info, const std::string & entity_id); + +/// A data item is addressed by its full topic path. The leading slash is optional in a route segment. +std::string to_full_topic_path(const std::string & topic_name); + +/// Reads one data item of a plugin-owned entity: through its `DataProvider`, else through its +/// `x-plc-data` route. `GET /{entity}/data/{item}` and the cyclic-subscription sampler both use it. +tl::expected read_plugin_data_item(PluginManager * plugins, const std::string & entity_id, + const std::string & topic_name); + +/// Samples one topic through the `TopicDataProvider` of `data`. Answers 503 when none is configured. +/// `GET /{entity}/data/{item}` and the cyclic-subscription sampler both use it. +tl::expected sample_data_topic(DataAccessManager & data, + const std::string & full_topic_path); + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp index 0c3982560..aeef86084 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/bulk_data_store.hpp @@ -15,6 +15,7 @@ #pragma once #include +#include #include #include #include @@ -23,6 +24,8 @@ #include #include +#include "ros2_medkit_gateway/core/config.hpp" + namespace ros2_medkit_gateway { /** @@ -148,4 +151,8 @@ class BulkDataStore { static tl::expected validate_path_component(const std::string & value, const std::string & name); }; +/// Builds the store a host serves bulk data from. Drops empty category names. A negative +/// `max_upload_size` is logged and replaced by the default of `BulkDataConfig`. +std::unique_ptr make_bulk_data_store(const BulkDataConfig & config); + } // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp index 2b6627b7f..1386d1daf 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/managers/fault_manager.hpp @@ -96,6 +96,9 @@ class FaultManager { /// Check if fault manager services are available. bool is_available() const; + /// Whether the fault manager can serve rosbag files (forwards to transport). + bool serves_rosbags() const; + /// Wait for fault manager services to become available (forwards to transport). bool wait_for_services(std::chrono::duration timeout); diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp index 9e2fc0c82..2d9084b23 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/subscription_transport.hpp @@ -90,6 +90,8 @@ class TransportRegistry { void register_transport(std::unique_ptr provider); SubscriptionTransportProvider * get_transport(const std::string & protocol) const; bool has_transport(const std::string & protocol) const; + /// True when no transport is registered. + bool empty() const; /// Shutdown all subscriptions by delegating to SubscriptionManager::shutdown(). /// Transport cleanup happens via the manager's on_removed callback, which diff --git a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp index 8a4c1288c..97f908424 100644 --- a/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp +++ b/src/ros2_medkit_gateway/include/ros2_medkit_gateway/core/transports/fault_service_transport.hpp @@ -63,6 +63,13 @@ class FaultServiceTransport { virtual bool wait_for_services(std::chrono::duration timeout) = 0; virtual bool is_available() const = 0; + + /// Whether this transport serves rosbag files (`get_rosbag`, `list_rosbags`). The bulk-data + /// `rosbags` category is listed only when it does. A host whose fault manager records no + /// rosbags overrides it to return false. + virtual bool serves_rosbags() const { + return true; + } }; } // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp b/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp new file mode 100644 index 000000000..5812c999c --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/builtin_samplers.cpp @@ -0,0 +1,198 @@ +// Copyright 2026 bburda +// +// 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. + +#include "ros2_medkit_gateway/core/builtin_samplers.hpp" + +#include +#include + +#include +#include + +#include "ros2_medkit_gateway/core/data/data_item_reader.hpp" +#include "ros2_medkit_gateway/core/faults/fault_scope.hpp" +#include "ros2_medkit_gateway/core/gateway_services.hpp" +#include "ros2_medkit_gateway/core/logs/log_scope.hpp" +#include "ros2_medkit_gateway/core/managers/configuration_manager.hpp" +#include "ros2_medkit_gateway/core/managers/data_access_manager.hpp" +#include "ros2_medkit_gateway/core/managers/fault_manager.hpp" +#include "ros2_medkit_gateway/core/managers/log_manager.hpp" +#include "ros2_medkit_gateway/core/managers/update_manager.hpp" +#include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp" +#include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/core/providers/update_types.hpp" +#include "ros2_medkit_gateway/core/resource_sampler.hpp" + +namespace ros2_medkit_gateway { + +namespace { + +using Sampled = tl::expected; + +Sampled sample_data(GatewayServices & services, const std::string & entity_id, const std::string & resource_path) { + // The item is what follows the separator slash, as in GET /{entity}/data/{item}. + const std::string item_name = + (!resource_path.empty() && resource_path.front() == '/') ? resource_path.substr(1) : resource_path; + // A plugin-owned entity is read through its provider, as GET /{entity}/data/{item} does. + auto * plugins = services.plugin_manager(); + if (plugins != nullptr && plugins->get_entity_owner(entity_id)) { + auto item = read_plugin_data_item(plugins, entity_id, item_name); + if (!item) { + return tl::make_unexpected(item.error().message); + } + return std::move(item->content); + } + const std::string topic_path = to_full_topic_path(item_name); + auto sample = sample_data_topic(*services.data_access_manager(), topic_path); + if (!sample) { + return tl::make_unexpected(std::string{"Topic sample failed: "} + sample.error().message); + } + if (sample->has_data && sample->data.has_value()) { + nlohmann::json payload; + payload["id"] = topic_path; + payload["data"] = *sample->data; + return payload; + } + return tl::make_unexpected("Topic data not available: " + topic_path); +} + +Sampled sample_faults(GatewayServices & services, const std::string & entity_id) { + auto * fault_mgr = services.fault_manager(); + if (!fault_mgr) { + return tl::make_unexpected(std::string("FaultManager not available")); + } + const auto & cache = services.entity_cache(); + auto entity_ref = cache.find_entity(entity_id); + if (!entity_ref) { + return tl::make_unexpected("Entity not found: " + entity_id); + } + // Same scoping as GET /{entity}/faults: shared source resolution (external app -> bare id, + // external component owns its own id) and the shared boundary-aware filter. + auto source_fqns = faults::resolve_entity_source_fqns(cache, entity_ref->type, entity_id); + auto result = fault_mgr->list_faults(""); + if (!result.success) { + return tl::make_unexpected(result.error_message); + } + auto filtered = faults::filter_faults_by_sources(result.data["faults"], source_fqns); + result.data["count"] = filtered.size(); + result.data["faults"] = std::move(filtered); + return result.data; +} + +Sampled sample_configurations(GatewayServices & services, const std::string & entity_id) { + auto * config_mgr = services.configuration_manager(); + if (!config_mgr) { + return tl::make_unexpected(std::string("ConfigurationManager not available")); + } + auto entity_configs = services.entity_cache().get_entity_configurations(entity_id); + nlohmann::json items = nlohmann::json::array(); + for (const auto & node : entity_configs.nodes) { + auto result = config_mgr->list_parameters(node.node_fqn); + if (result.success && result.data.is_array()) { + for (auto & param : result.data) { + items.push_back(std::move(param)); + } + } + } + nlohmann::json payload; + payload["items"] = std::move(items); + return payload; +} + +Sampled sample_logs(GatewayServices & services, const std::string & entity_id) { + auto * log_mgr = services.log_manager(); + if (!log_mgr) { + return tl::make_unexpected(std::string("LogManager not available")); + } + const auto & cache = services.entity_cache(); + auto entity_ref = cache.find_entity(entity_id); + if (!entity_ref) { + return tl::make_unexpected("Entity not found: " + entity_id); + } + // Same scoping as GET /{entity}/logs via the shared helper (external app -> bare id, external + // component owns its own id, plus the namespace-prefix merge for COMPONENT / AREA). + std::string entity_fqn; + if (entity_ref->type == SovdEntityType::COMPONENT) { + if (auto comp = cache.get_component(entity_id)) { + entity_fqn = comp->fqn; + } + } else if (entity_ref->type == SovdEntityType::AREA) { + if (auto area = cache.get_area(entity_id)) { + entity_fqn = area->namespace_path; + } + } else if (entity_ref->type == SovdEntityType::APP) { + if (auto app = cache.get_app(entity_id)) { + entity_fqn = app->effective_fqn(); + } + } + auto result = logs::query_entity_logs(*log_mgr, cache, entity_ref->type, entity_id, entity_fqn, "", ""); + if (!result.has_value()) { + return tl::make_unexpected(result.error()); + } + nlohmann::json payload; + payload["items"] = std::move(*result); + return payload; +} + +Sampled sample_updates(GatewayServices & services, const std::string & resource_path) { + auto * mgr = services.update_manager(); + if (!mgr || !mgr->has_backend()) { + return tl::make_unexpected(std::string("Update backend not available")); + } + auto result = mgr->get_status(resource_path); + if (!result) { + return tl::make_unexpected(result.error().message); + } + return update_status_to_json(*result); +} + +} // namespace + +void register_builtin_samplers(GatewayServices & services) { + auto & registry = *services.sampler_registry(); + registry.register_sampler( + "data", + [&services](const std::string & entity_id, const std::string & resource_path) { + return sample_data(services, entity_id, resource_path); + }, + /*is_builtin=*/true, /*honours_resource_path=*/true); + registry.register_sampler( + "faults", + [&services](const std::string & entity_id, const std::string & /*resource_path*/) { + return sample_faults(services, entity_id); + }, + /*is_builtin=*/true); + registry.register_sampler( + "configurations", + [&services](const std::string & entity_id, const std::string & /*resource_path*/) { + return sample_configurations(services, entity_id); + }, + /*is_builtin=*/true); + registry.register_sampler( + "logs", + [&services](const std::string & entity_id, const std::string & /*resource_path*/) { + return sample_logs(services, entity_id); + }, + /*is_builtin=*/true); + if (services.update_manager() != nullptr) { + registry.register_sampler( + "updates", + [&services](const std::string & /*entity_id*/, const std::string & resource_path) { + return sample_updates(services, resource_path); + }, + /*is_builtin=*/true, /*honours_resource_path=*/true); + } +} + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/data/data_item_reader.cpp b/src/ros2_medkit_gateway/src/core/data/data_item_reader.cpp new file mode 100644 index 000000000..4a974620b --- /dev/null +++ b/src/ros2_medkit_gateway/src/core/data/data_item_reader.cpp @@ -0,0 +1,93 @@ +// Copyright 2026 bburda +// +// 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. + +#include "ros2_medkit_gateway/core/data/data_item_reader.hpp" + +#include +#include +#include +#include +#include + +#include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" +#include "ros2_medkit_gateway/core/http/error_codes.hpp" +#include "ros2_medkit_gateway/core/logging.hpp" +#include "ros2_medkit_gateway/core/managers/data_access_manager.hpp" +#include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/http/handlers/handler_support.hpp" + +namespace ros2_medkit_gateway { + +using json = nlohmann::json; +using handlers::make_error; + +ErrorInfo make_plugin_error(int http_status, const std::string & message, json extra_params) { + static constexpr size_t kMaxMessageLength = 512; + int status = http_status < 400 ? 400 : (http_status > 599 ? 599 : http_status); + std::string msg = message.size() > kMaxMessageLength ? message.substr(0, kMaxMessageLength) + "..." : message; + return make_error(status, ERR_PLUGIN_ERROR, std::move(msg), std::move(extra_params)); +} + +ErrorInfo make_provider_error(const DataProviderErrorInfo & info, const std::string & entity_id) { + return make_plugin_error(info.http_status, info.message, json{{"entity_id", entity_id}}); +} + +std::string to_full_topic_path(const std::string & topic_name) { + if (topic_name.empty() || topic_name.front() == '/') { + return topic_name; + } + return "/" + topic_name; +} + +tl::expected read_plugin_data_item(PluginManager * plugins, const std::string & entity_id, + const std::string & topic_name) { + const log::Logger logger("rest_server"); + auto * data_prov = plugins ? plugins->get_data_provider_for_entity(entity_id) : nullptr; + if (data_prov == nullptr) { + // Same x-plc-data route fallback as the data list, single-item form. + if (auto via_route = plugins ? plugins->fetch_entity_data_via_route(entity_id, topic_name) : std::nullopt) { + return dto::DataValue{std::move(*via_route)}; + } + return tl::make_unexpected( + make_error(404, ERR_RESOURCE_NOT_FOUND, "No data provider for plugin entity '" + entity_id + "'")); + } + try { + auto result = data_prov->read_data(entity_id, topic_name); + if (!result) { + return tl::make_unexpected(make_provider_error(result.error(), entity_id)); + } + return std::move(*result); + } catch (const std::exception & e) { + MEDKIT_LOG_ERROR(logger, "Plugin DataProvider threw for entity '%s': %s", entity_id.c_str(), e.what()); + return tl::make_unexpected(make_plugin_error(500, "Plugin threw exception", json{{"entity_id", entity_id}})); + } catch (...) { + MEDKIT_LOG_ERROR(logger, "Plugin DataProvider threw unknown exception for entity '%s'", entity_id.c_str()); + return tl::make_unexpected( + make_plugin_error(500, "Plugin threw unknown exception", json{{"entity_id", entity_id}})); + } +} + +tl::expected sample_data_topic(DataAccessManager & data, + const std::string & full_topic_path) { + // Sampling goes through the pool-backed TopicDataProvider. + auto * provider = data.get_topic_data_provider(); + if (provider == nullptr) { + return tl::make_unexpected(make_error(503, ERR_SERVICE_UNAVAILABLE, "Topic sampling is not configured")); + } + const auto timeout_ms = + std::chrono::milliseconds{static_cast(std::max(data.get_topic_sample_timeout(), 0.0) * 1000.0)}; + return provider->sample(full_topic_path, timeout_ms); +} + +} // namespace ros2_medkit_gateway diff --git a/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp b/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp index 27e8384b1..9763426ab 100644 --- a/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp +++ b/src/ros2_medkit_gateway/src/core/managers/bulk_data_store.cpp @@ -14,13 +14,16 @@ #include "ros2_medkit_gateway/core/managers/bulk_data_store.hpp" +#include #include +#include #include #include #include #include #include "ros2_medkit_gateway/core/http/http_utils.hpp" +#include "ros2_medkit_gateway/core/logging.hpp" namespace ros2_medkit_gateway { @@ -35,6 +38,30 @@ BulkDataStore::BulkDataStore(const std::string & storage_dir, size_t max_upload_ std::filesystem::create_directories(storage_dir_, ec); } +std::unique_ptr make_bulk_data_store(const BulkDataConfig & config) { + // A negative value would wrap to a huge size_t and be handed to set_payload_max_length as an + // effectively unlimited cap. 0 already means unlimited by design, so a negative is refused rather + // than folded into it. + size_t max_upload = 0; + if (config.max_upload_size < 0) { + const BulkDataConfig defaults; + MEDKIT_LOG_WARN(log::Logger("bulk_data"), + "bulk_data.max_upload_size %" PRId64 + " is negative. Must be >= 0 (0 = unlimited). Using default %" PRId64 ".", + config.max_upload_size, defaults.max_upload_size); + max_upload = static_cast(defaults.max_upload_size); + } else { + max_upload = static_cast(config.max_upload_size); + } + auto categories = config.categories; + categories.erase(std::remove_if(categories.begin(), categories.end(), + [](const std::string & item) { + return item.empty(); + }), + categories.end()); + return std::make_unique(config.storage_dir, max_upload, std::move(categories)); +} + // --- Validation --- tl::expected BulkDataStore::validate_path_component(const std::string & value, diff --git a/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp b/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp index cb02ef227..c406cbac6 100644 --- a/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp +++ b/src/ros2_medkit_gateway/src/core/managers/fault_manager.cpp @@ -61,6 +61,10 @@ bool FaultManager::is_available() const { return transport_->is_available(); } +bool FaultManager::serves_rosbags() const { + return transport_->serves_rosbags(); +} + bool FaultManager::wait_for_services(std::chrono::duration timeout) { return transport_->wait_for_services(timeout); } diff --git a/src/ros2_medkit_gateway/src/core/subscription_transport.cpp b/src/ros2_medkit_gateway/src/core/subscription_transport.cpp index 2144def53..71b2ee71a 100644 --- a/src/ros2_medkit_gateway/src/core/subscription_transport.cpp +++ b/src/ros2_medkit_gateway/src/core/subscription_transport.cpp @@ -41,6 +41,11 @@ bool TransportRegistry::has_transport(const std::string & protocol) const { return transports_.count(protocol) > 0; } +bool TransportRegistry::empty() const { + std::shared_lock lock(mutex_); + return transports_.empty(); +} + void TransportRegistry::shutdown_all(SubscriptionManager & sub_mgr) { // Delegate to sub_mgr.shutdown() which triggers on_removed callbacks // for each active subscription, causing transport->stop() for each. diff --git a/src/ros2_medkit_gateway/src/gateway_node.cpp b/src/ros2_medkit_gateway/src/gateway_node.cpp index cb546c81e..0fb4343fe 100644 --- a/src/ros2_medkit_gateway/src/gateway_node.cpp +++ b/src/ros2_medkit_gateway/src/gateway_node.cpp @@ -30,12 +30,11 @@ #include #include "ros2_medkit_gateway/core/aggregation/network_utils.hpp" +#include "ros2_medkit_gateway/core/builtin_samplers.hpp" #include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" #include "ros2_medkit_gateway/core/discovery/merge_types.hpp" #include "ros2_medkit_gateway/core/discovery/refresh_debounce.hpp" #include "ros2_medkit_gateway/core/entity_validation.hpp" -#include "ros2_medkit_gateway/core/faults/fault_scope.hpp" -#include "ros2_medkit_gateway/core/logs/log_scope.hpp" #include "ros2_medkit_gateway/core/thread_pool_config.hpp" #include "ros2_medkit_gateway/fault_manager_paths.hpp" #include "ros2_medkit_gateway/param_utils.hpp" @@ -270,8 +269,9 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki declare_parameter("aggregation.peer_names", std::vector{""}); // Bulk data storage parameters - declare_parameter("bulk_data.storage_dir", "/tmp/ros2_medkit_bulk_data"); - declare_parameter("bulk_data.max_upload_size", 104857600); // 100MB + const BulkDataConfig bulk_data_defaults; + declare_parameter("bulk_data.storage_dir", bulk_data_defaults.storage_dir); + declare_parameter("bulk_data.max_upload_size", bulk_data_defaults.max_upload_size); declare_parameter("bulk_data.categories", std::vector{}); // Runtime (heuristic) discovery options @@ -692,34 +692,14 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki fault_mgr_ = std::make_unique(fault_service_transport_); // Initialize bulk data store - auto bd_storage_dir = get_parameter("bulk_data.storage_dir").as_string(); - // A negative value would wrap to a huge size_t and be handed to - // set_payload_max_length as an effectively unlimited cap. 0 already means - // unlimited by documented design, so a negative cannot be folded into it - // without turning a typo into that same silent "no limit". Reject it and say - // which value was refused. - const int64_t bd_max_upload_raw = get_parameter("bulk_data.max_upload_size").as_int(); - size_t bd_max_upload = 0; - if (bd_max_upload_raw < 0) { - RCLCPP_WARN(get_logger(), - "bulk_data.max_upload_size %" PRId64 - " is negative. Must be >= 0 (0 = unlimited). " - "Using default 104857600.", - bd_max_upload_raw); - bd_max_upload = 104857600; - } else { - bd_max_upload = static_cast(bd_max_upload_raw); - } - auto bd_categories = get_parameter("bulk_data.categories").as_string_array(); - bd_categories.erase(std::remove_if(bd_categories.begin(), bd_categories.end(), - [](const auto & item) { - return item.empty(); - }), - bd_categories.end()); - - bulk_data_store_ = std::make_unique(bd_storage_dir, bd_max_upload, bd_categories); - RCLCPP_INFO(get_logger(), "Bulk data store: dir=%s, max_upload=%zuB, categories=%zu", bd_storage_dir.c_str(), - bd_max_upload, bd_categories.size()); + BulkDataConfig bulk_data_config; + bulk_data_config.storage_dir = get_parameter("bulk_data.storage_dir").as_string(); + bulk_data_config.max_upload_size = get_parameter("bulk_data.max_upload_size").as_int(); + bulk_data_config.categories = get_parameter("bulk_data.categories").as_string_array(); + bulk_data_store_ = make_bulk_data_store(bulk_data_config); + RCLCPP_INFO(get_logger(), "Bulk data store: dir=%s, max_upload=%zuB, categories=%zu", + bulk_data_config.storage_dir.c_str(), bulk_data_store_->max_upload_bytes(), + bulk_data_store_->list_categories().size()); // Initialize subscription manager for cyclic subscriptions // Same wrap hazard as bulk_data.max_upload_size: a negative here becomes a @@ -1332,148 +1312,7 @@ GatewayNode::GatewayNode(const rclcpp::NodeOptions & options) : Node("ros2_medki } // Register built-in resource samplers - sampler_registry_->register_sampler( - "data", - [this](const std::string & /*entity_id*/, - const std::string & resource_path) -> tl::expected { - auto * dam = get_data_access_manager(); - if (!dam) { - return tl::make_unexpected(std::string("DataAccessManager not available")); - } - auto * provider = dam->get_topic_data_provider(); - if (!provider) { - return tl::make_unexpected(std::string("TopicDataProvider not configured")); - } - const auto timeout_ms = std::chrono::milliseconds{ - static_cast(std::max(dam->get_topic_sample_timeout(), 0.0) * 1000.0)}; - auto r = provider->sample(resource_path, timeout_ms); - if (!r) { - return tl::make_unexpected(std::string{"Topic sample failed: "} + r.error().message); - } - TopicSampleResult sample = *r; - if (sample.has_data && sample.data.has_value()) { - nlohmann::json payload; - payload["id"] = resource_path; - payload["data"] = *sample.data; - return payload; - } - return tl::make_unexpected("Topic data not available: " + resource_path); - }, - /*is_builtin=*/true, /*honours_resource_path=*/true); - - sampler_registry_->register_sampler( - "faults", - [this](const std::string & entity_id, - const std::string & /*resource_path*/) -> tl::expected { - auto * fault_mgr = get_fault_manager(); - if (!fault_mgr) { - return tl::make_unexpected(std::string("FaultManager not available")); - } - const auto & cache = get_thread_safe_cache(); - auto entity_ref = cache.find_entity(entity_id); - if (!entity_ref) { - return tl::make_unexpected("Entity not found: " + entity_id); - } - // Same scoping as GET /{entity}/faults: shared source resolution - // (external app -> bare id, external component owns its own id) and - // the shared boundary-aware filter. A private variant here answered - // "Entity no longer available" for external apps whose effective_fqn - // is empty while REST returned the fault. - auto source_fqns = faults::resolve_entity_source_fqns(cache, entity_ref->type, entity_id); - auto result = fault_mgr->list_faults(""); - if (!result.success) { - return tl::make_unexpected(result.error_message); - } - auto filtered = faults::filter_faults_by_sources(result.data["faults"], source_fqns); - result.data["count"] = filtered.size(); - result.data["faults"] = std::move(filtered); - return result.data; - }, - /*is_builtin=*/true); - - sampler_registry_->register_sampler( - "configurations", - [this](const std::string & entity_id, - const std::string & /*resource_path*/) -> tl::expected { - auto * config_mgr = get_configuration_manager(); - if (!config_mgr) { - return tl::make_unexpected(std::string("ConfigurationManager not available")); - } - const auto & cache = get_thread_safe_cache(); - auto entity_configs = cache.get_entity_configurations(entity_id); - nlohmann::json items = nlohmann::json::array(); - for (const auto & node : entity_configs.nodes) { - auto result = config_mgr->list_parameters(node.node_fqn); - if (result.success && result.data.is_array()) { - for (auto & param : result.data) { - items.push_back(std::move(param)); - } - } - } - nlohmann::json payload; - payload["items"] = std::move(items); - return payload; - }, - /*is_builtin=*/true); - - sampler_registry_->register_sampler( - "logs", - [this](const std::string & entity_id, - const std::string & /*resource_path*/) -> tl::expected { - auto * log_mgr = get_log_manager(); - if (!log_mgr) { - return tl::make_unexpected(std::string("LogManager not available")); - } - const auto & cache = get_thread_safe_cache(); - auto entity_ref = cache.find_entity(entity_id); - if (!entity_ref) { - return tl::make_unexpected("Entity not found: " + entity_id); - } - // Same scoping as GET /{entity}/logs via the shared helper (external - // app -> bare id, external component owns its own id, plus the - // namespace-prefix merge for COMPONENT / AREA). - std::string entity_fqn; - if (entity_ref->type == SovdEntityType::COMPONENT) { - if (auto comp = cache.get_component(entity_id)) { - entity_fqn = comp->fqn; - } - } else if (entity_ref->type == SovdEntityType::AREA) { - if (auto area = cache.get_area(entity_id)) { - entity_fqn = area->namespace_path; - } - } else if (entity_ref->type == SovdEntityType::APP) { - if (auto app = cache.get_app(entity_id)) { - entity_fqn = app->effective_fqn(); - } - } - auto result = logs::query_entity_logs(*log_mgr, cache, entity_ref->type, entity_id, entity_fqn, "", ""); - if (!result.has_value()) { - return tl::make_unexpected(result.error()); - } - nlohmann::json payload; - payload["items"] = std::move(*result); - return payload; - }, - /*is_builtin=*/true); - - // Register update status sampler (server-level, uses UpdateManager) - if (update_mgr_) { - sampler_registry_->register_sampler( - "updates", - [this](const std::string & /*entity_id*/, - const std::string & resource_path) -> tl::expected { - auto * mgr = get_update_manager(); - if (!mgr || !mgr->has_backend()) { - return tl::make_unexpected(std::string("Update backend not available")); - } - auto result = mgr->get_status(resource_path); - if (!result) { - return tl::make_unexpected(result.error().message); - } - return update_status_to_json(*result); - }, - /*is_builtin=*/true, /*honours_resource_path=*/true); - } + register_builtin_samplers(*this); RCLCPP_INFO(get_logger(), "Registered built-in resource samplers: data, faults, configurations, logs%s", update_mgr_ ? ", updates" : ""); diff --git a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp index 1c8063f89..35f99d713 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/bulkdata_handlers.cpp @@ -269,6 +269,14 @@ std::vector compute_bulkdata_source_filters(const ThreadSafeEntityC } // namespace detail +namespace { +/// Whether the fault manager serves the `rosbags` category. When it does not, the category is unknown. +bool rosbags_served(const HandlerContext & ctx) { + auto * fault_mgr = ctx.services().fault_manager(); + return fault_mgr != nullptr && fault_mgr->serves_rosbags(); +} +} // namespace + // --------------------------------------------------------------------------- // GET /{entity}/bulk-data - list categories // --------------------------------------------------------------------------- @@ -289,9 +297,11 @@ http::Result BulkDataHandlers::list_categories(const return tl::unexpected(make_error(400, ERR_COLLECTION_NOT_SUPPORTED, access.error().message)); } - // Build categories list: "rosbags" always available + BulkDataStore categories. + // Build categories list: "rosbags" when the fault manager serves them + BulkDataStore categories. dto::BulkDataCategoryList response; - response.items.push_back("rosbags"); // Always available via FaultManager + if (rosbags_served(ctx_)) { + response.items.push_back("rosbags"); + } auto * store = ctx_.bulk_data_store(); if (store) { @@ -328,7 +338,7 @@ BulkDataHandlers::list_descriptors(const http::TypedRequest & req) { return tl::unexpected(make_error(400, ERR_INVALID_REQUEST, "Missing category")); } - if (category == "rosbags") { + if (category == "rosbags" && rosbags_served(ctx_)) { // === Rosbags: served via FaultManager === auto fault_mgr = ctx_.services().fault_manager(); @@ -431,7 +441,7 @@ http::Result BulkDataHandlers::download(const http::TypedR std::string mimetype; std::string filename; - if (category == "rosbags") { + if (category == "rosbags" && rosbags_served(ctx_)) { // === Rosbags: served via FaultManager === // // The id is a recording id. A pre-#620 URL carrying a fault code still resolves: diff --git a/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp index ef44efffb..98905de6d 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/cyclic_subscription_handlers.cpp @@ -177,6 +177,14 @@ CyclicSubscriptionHandlers::post_subscription(const http::TypedRequest & req, } } + // GET refuses an update id that fails the entity-id rules; so does a subscription. + if (parsed->collection == "updates") { + if (auto id_ok = ctx_.validate_entity_id(parsed->resource_path); !id_ok) { + return tl::unexpected(make_error(400, ERR_X_MEDKIT_INVALID_RESOURCE_URI, id_ok.error(), + json{{"parameter", "resource"}, {"value", resource}})); + } + } + // Check sampler is registered if (!sampler_registry_.has_sampler(parsed->collection)) { return tl::unexpected(make_error(400, ERR_X_MEDKIT_COLLECTION_NOT_AVAILABLE, @@ -456,7 +464,8 @@ CyclicSubscriptionHandlers::parse_resource_uri(const std::string & resource) { parsed.entity_type = match[1].str(); parsed.entity_id = match[2].str(); parsed.collection = match[3].str(); - parsed.resource_path = match[4].matched ? match[4].str() : ""; + // Decoded as the GET path is, so a subscription reads the item GET reads. Before the '..' check. + parsed.resource_path = match[4].matched ? httplib::detail::decode_url(match[4].str(), false) : ""; // Security: reject '..' as a path segment (not as substring of e.g. '/..foo') if (!parsed.resource_path.empty()) { @@ -482,7 +491,7 @@ CyclicSubscriptionHandlers::parse_resource_uri(const std::string & resource) { parsed.entity_type = ""; // server-level, no entity parsed.entity_id = ""; parsed.collection = "updates"; - parsed.resource_path = match[1].str(); // update package ID + parsed.resource_path = httplib::detail::decode_url(match[1].str(), false); // update package ID return parsed; } diff --git a/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp b/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp index eb5533d89..43c7b7722 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/data_handlers.cpp @@ -29,6 +29,7 @@ #include #include +#include "ros2_medkit_gateway/core/data/data_item_reader.hpp" #include "ros2_medkit_gateway/core/data/topic_data_provider.hpp" #include "ros2_medkit_gateway/core/exceptions.hpp" #include "ros2_medkit_gateway/core/http/error_codes.hpp" @@ -55,21 +56,6 @@ using json = nlohmann::json; // Helper free functions // ============================================================================= -/// Sanitize a plugin-supplied error into the standard `x-medkit-plugin-error` -/// shape: clamp HTTP status to [400, 599] and truncate message at 512 chars. -ErrorInfo make_plugin_error(int http_status, const std::string & message, json extra_params = {}) { - static constexpr size_t kMaxMessageLength = 512; - int status = http_status < 400 ? 400 : (http_status > 599 ? 599 : http_status); - std::string msg = message.size() > kMaxMessageLength ? message.substr(0, kMaxMessageLength) + "..." : message; - return make_error(status, ERR_PLUGIN_ERROR, std::move(msg), std::move(extra_params)); -} - -/// Map a `DataProviderErrorInfo` (from the typed plugin ABI) into the SOVD -/// `x-medkit-plugin-error` wire shape via `make_plugin_error`. -ErrorInfo make_provider_error(const DataProviderErrorInfo & info, const std::string & entity_id) { - return make_plugin_error(info.http_status, info.message, json{{"entity_id", entity_id}}); -} - /// Read the first positional capture group (entity_id) with the same "missing /// capture is treated as 400 invalid-request" semantics as the other migrated /// handlers (matches the legacy `req.matches.size() < N` guard). @@ -91,15 +77,6 @@ tl::expected read_topic_id(const http::TypedRequest & re return tl::make_unexpected(make_error(400, ERR_INVALID_REQUEST, "Invalid request")); } -/// A data item is addressed by its full ROS topic path; the leading slash is -/// optional on the wire because the route captures a percent-decoded segment. -std::string to_full_topic_path(const std::string & topic_name) { - if (topic_name.empty() || topic_name.front() == '/') { - return topic_name; - } - return "/" + topic_name; -} - /// What one addressed data item resolves to: the ROS topic to act on, the id to /// echo back to the caller, the member the id named, whether this gateway's own /// walk records that member as a provider of the topic, and the member the tree @@ -637,32 +614,7 @@ http::Result DataHandlers::get_data_item(const http::TypedReques // Delegate to plugin DataProvider for plugin-owned entities. if (entity_info.is_plugin) { - auto * pmgr = ctx_.services().plugin_manager(); - auto * data_prov = pmgr ? pmgr->get_data_provider_for_entity(entity_id) : nullptr; - if (data_prov == nullptr) { - // Same x-plc-data route fallback as list_data above, single-item form. - if (auto via_route = pmgr ? pmgr->fetch_entity_data_via_route(entity_id, topic_name) : std::nullopt) { - return dto::DataValue{std::move(*via_route)}; - } - return tl::make_unexpected( - make_error(404, ERR_RESOURCE_NOT_FOUND, "No data provider for plugin entity '" + entity_id + "'")); - } - try { - auto result = data_prov->read_data(entity_id, topic_name); - if (!result) { - return tl::make_unexpected(make_provider_error(result.error(), entity_id)); - } - return std::move(*result); - } catch (const std::exception & e) { - MEDKIT_LOG_ERROR(HandlerContext::logger(), "Plugin DataProvider threw for entity '%s': %s", entity_id.c_str(), - e.what()); - return tl::make_unexpected(make_plugin_error(500, "Plugin threw exception", json{{"entity_id", entity_id}})); - } catch (...) { - MEDKIT_LOG_ERROR(HandlerContext::logger(), "Plugin DataProvider threw unknown exception for entity '%s'", - entity_id.c_str()); - return tl::make_unexpected( - make_plugin_error(500, "Plugin threw unknown exception", json{{"entity_id", entity_id}})); - } + return read_plugin_data_item(ctx_.services().plugin_manager(), entity_id, topic_name); } try { @@ -677,16 +629,7 @@ http::Result DataHandlers::get_data_item(const http::TypedReques } const std::string & full_topic_path = addressed->full_topic_path; - // Sampling goes through the pool-backed TopicDataProvider (issue #375 race - // fix). The provider is configured in main() before serving traffic. - auto data_access_mgr = ctx_.services().data_access_manager(); - const auto timeout_sec = data_access_mgr->get_topic_sample_timeout(); - auto * provider = data_access_mgr->get_topic_data_provider(); - if (provider == nullptr) { - return tl::make_unexpected(make_error(503, ERR_SERVICE_UNAVAILABLE, "Topic sampling is not configured")); - } - const auto timeout_ms = std::chrono::milliseconds{static_cast(std::max(timeout_sec, 0.0) * 1000.0)}; - auto r = provider->sample(full_topic_path, timeout_ms); + auto r = sample_data_topic(*ctx_.services().data_access_manager(), full_topic_path); if (!r) { // Propagate the provider's ErrorInfo verbatim: http_status, code (SOVD // constant or x-medkit-*), message, and structured params. Collapsing 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 5708b5ce4..591cac254 100644 --- a/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp +++ b/src/ros2_medkit_gateway/src/http/handlers/health_handlers.cpp @@ -27,12 +27,16 @@ #include "ros2_medkit_gateway/core/http/fan_out_helpers.hpp" #include "ros2_medkit_gateway/core/http/http_utils.hpp" #include "ros2_medkit_gateway/core/http/warning_codes.hpp" +#include "ros2_medkit_gateway/core/managers/bulk_data_store.hpp" +#include "ros2_medkit_gateway/core/managers/fault_manager.hpp" #include "ros2_medkit_gateway/core/managers/lock_manager.hpp" #include "ros2_medkit_gateway/core/managers/script_manager.hpp" #include "ros2_medkit_gateway/core/managers/trigger_manager.hpp" #include "ros2_medkit_gateway/core/managers/update_manager.hpp" #include "ros2_medkit_gateway/core/models/thread_safe_entity_cache.hpp" #include "ros2_medkit_gateway/core/plugins/plugin_manager.hpp" +#include "ros2_medkit_gateway/core/resource_sampler.hpp" +#include "ros2_medkit_gateway/core/subscription_transport.hpp" #include "ros2_medkit_gateway/core/version.hpp" #include "ros2_medkit_gateway/dto/health.hpp" @@ -285,8 +289,14 @@ http::Result HealthHandlers::get_root(const http::TypedReques capabilities.configurations = true; capabilities.faults = true; capabilities.logs = true; - capabilities.bulk_data = true; - capabilities.cyclic_subscriptions = true; + const auto * fault_mgr = ctx_.services().fault_manager(); + const auto * transports = ctx_.services().transport_registry(); + const auto * samplers = ctx_.services().sampler_registry(); + capabilities.bulk_data = + ctx_.services().bulk_data_store() != nullptr || (fault_mgr != nullptr && fault_mgr->serves_rosbags()); + capabilities.cyclic_subscriptions = ctx_.services().subscription_manager() != nullptr && transports != nullptr && + !transports->empty() && samplers != nullptr && + !samplers->collection_names().empty(); capabilities.locking = ctx_.services().lock_manager() != nullptr; capabilities.triggers = ctx_.services().trigger_manager() != nullptr; capabilities.updates = ctx_.services().update_manager() != nullptr; diff --git a/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp b/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp index ef8c841b2..c1a92d8ce 100644 --- a/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp +++ b/src/ros2_medkit_gateway/test/demo_nodes/test_data_provider_plugin.cpp @@ -97,6 +97,9 @@ class TestDataProviderPlugin : public GatewayPlugin, {"unit", "degC"}}); items.push_back( {{"id", "pressure"}, {"name", "Pressure"}, {"category", "currentData"}, {"value", 3.2}, {"unit", "bar"}}); + // Names that need percent-encoding in a URI, like PLC symbols. + items.push_back({{"id", "\"DB1\".temp"}, {"name", "DB1 temp"}, {"category", "currentData"}, {"value", 7.5}}); + items.push_back({{"id", "Motor Speed"}, {"name", "Motor Speed"}, {"category", "currentData"}, {"value", 1450}}); return dto::DataListResult{nlohmann::json{{"items", std::move(items)}}}; } diff --git a/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py b/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py index f5e09f946..befb3ffce 100644 --- a/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py +++ b/src/ros2_medkit_integration_tests/test/features/test_multi_collection_subscriptions.test.py @@ -27,6 +27,7 @@ import threading import time import unittest +from urllib.parse import quote import launch_testing import launch_testing.actions @@ -159,6 +160,48 @@ def test_data_subscription_create(self): self.assertIn('event_source', data) self.assertTrue(data['event_source'].endswith('/events')) + def _assert_encoded_topic_matches_get(self, encoded): + """Check that a subscription on an encoded topic URI delivers what GET returns.""" + resource = f'/api/v1/apps/{self.app_id}/data/{encoded}' + r = self._create_subscription(resource, interval='fast') + self.assertEqual(r.status_code, 201, f'Create failed: {r.text}') + body = r.json() + self.addCleanup(self._delete_subscription, body['id']) + self.assertEqual(body['observed_resource'], resource) + + event = None + with requests.get( + self.BASE_URL.removesuffix(API_BASE_PATH) + body['event_source'], + stream=True, timeout=15) as stream: + for line in stream.iter_lines(decode_unicode=True): + if line and line.startswith('data:'): + event = json.loads(line[len('data:'):].strip()) + break + self.assertIsNotNone(event, 'SSE stream ended without an event') + + get = requests.get(f'{self.BASE_URL}/apps/{self.app_id}/data/{encoded}', + timeout=5) + self.assertEqual(get.status_code, 200, get.text) + self.assertNotIn('error', event, f'Sample failed: {event}') + sample = event['payload'] + self.assertEqual(sample['id'], get.json()['id']) + self.assertEqual(set(sample['data']), set(get.json()['data'])) + + def test_encoded_topic_subscription_matches_get(self): + """Topic path encoded without its leading slash. + + @verifies REQ_INTEROP_089 + """ + self._assert_encoded_topic_matches_get( + quote(self.topic_id.lstrip('/'), safe='')) + + def test_encoded_topic_with_leading_slash_matches_get(self): + """Topic id encoded as listed, leading slash included. + + @verifies REQ_INTEROP_089 + """ + self._assert_encoded_topic_matches_get(quote(self.topic_id, safe='')) + def test_faults_subscription_create(self): """Create a subscription on faults collection. diff --git a/src/ros2_medkit_integration_tests/test/features/test_plugin_item_subscription.test.py b/src/ros2_medkit_integration_tests/test/features/test_plugin_item_subscription.test.py new file mode 100644 index 000000000..0d246d0fb --- /dev/null +++ b/src/ros2_medkit_integration_tests/test/features/test_plugin_item_subscription.test.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +# Copyright 2026 bburda +# +# 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. + +"""A data subscription on a plugin item reads the item GET of the same URI reads. + +The item names carry characters that a client percent-encodes in a URI path +(a space, a quote), like PLC symbols such as ``"DB1".temp``. +""" + +import json +import os +import unittest +from urllib.parse import quote + +from ament_index_python.packages import get_package_prefix +import launch_testing +import requests + +from ros2_medkit_test_utils.constants import ALLOWED_EXIT_CODES, API_BASE_PATH +from ros2_medkit_test_utils.gateway_test_case import GatewayTestCase +from ros2_medkit_test_utils.launch_helpers import create_test_launch + +PLUGIN_APP = 'test_plc_app' +PLAIN_ITEM = 'temperature' +QUOTED_ITEM = '"DB1".temp' +SPACED_ITEM = 'Motor Speed' + + +def _get_plugin_path(so_name): + pkg_prefix = get_package_prefix('ros2_medkit_gateway') + return os.path.join(pkg_prefix, 'lib', 'ros2_medkit_gateway', so_name) + + +def generate_test_description(): + return create_test_launch( + demo_nodes=[], + fault_manager=False, + gateway_params={ + 'plugins': ['test_data_provider'], + 'plugins.test_data_provider.path': + _get_plugin_path('libtest_data_provider_plugin.so'), + 'sse.max_clients': 10, + 'sse.max_subscriptions': 50, + }, + ) + + +class TestPluginItemSubscription(GatewayTestCase): + + MIN_EXPECTED_APPS = 1 + REQUIRED_APPS = {PLUGIN_APP} + + def _item_uri(self, item): + return f'{API_BASE_PATH}/apps/{PLUGIN_APP}/data/{quote(item, safe="")}' + + def _first_event(self, item, expect_status=201): + resource = self._item_uri(item) + r = requests.post( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/cyclic-subscriptions', + json={'resource': resource, 'interval': 'fast', 'duration': 30}, + timeout=5, + ) + self.assertEqual(r.status_code, expect_status, r.text) + if expect_status != 201: + return r + body = r.json() + # The resource is echoed as the client sent it. + self.assertEqual(body['observed_resource'], resource) + sub_id = body['id'] + try: + with requests.get(self.BASE_URL.removesuffix(API_BASE_PATH) + + body['event_source'], stream=True, + timeout=15) as stream: + for line in stream.iter_lines(decode_unicode=True): + if line and line.startswith('data:'): + return json.loads(line[len('data:'):].strip()) + finally: + requests.delete( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/cyclic-subscriptions/' + f'{sub_id}', timeout=5) + self.fail('SSE stream ended without an event') + + def _assert_event_matches_get(self, item): + get = requests.get( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/data/{quote(item, safe="")}', + timeout=5) + self.assertEqual(get.status_code, 200, get.text) + event = self._first_event(item) + self.assertEqual( + event.get('payload', event), get.json(), + f'event {event} differs from GET {get.json()} for {item!r}') + + def test_plain_item_matches_get(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + self._assert_event_matches_get(PLAIN_ITEM) + + def test_quoted_item_matches_get(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + self._assert_event_matches_get(QUOTED_ITEM) + + def test_spaced_item_matches_get(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + self._assert_event_matches_get(SPACED_ITEM) + + def test_encoded_dot_dot_segment_is_refused(self): + """Check one case. + + @verifies REQ_INTEROP_089 + """ + r = requests.post( + f'{self.BASE_URL}/apps/{PLUGIN_APP}/cyclic-subscriptions', + json={'resource': + f'{API_BASE_PATH}/apps/{PLUGIN_APP}/data/%2E%2E', + 'interval': 'fast', 'duration': 30}, + timeout=5, + ) + self.assertEqual(r.status_code, 400, r.text) + + +@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) diff --git a/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py b/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py index 3251bcd6f..2bb918890 100644 --- a/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py +++ b/src/ros2_medkit_integration_tests/test/features/test_update_status_subscription.test.py @@ -195,6 +195,51 @@ def test_02_subscription_listed_for_entity(self): sub_ids = [s['id'] for s in items] self.assertIn(sub_id, sub_ids) + def _create_update_subscription(self, update_id_segment): + return requests.post( + f'{self.BASE_URL}/apps/{self.app_id}/cyclic-subscriptions', + json={ + 'resource': f'/api/v1/updates/{update_id_segment}/status', + 'interval': 'fast', + 'duration': 60, + }, + timeout=5, + ) + + def test_02b_encoded_update_id_matches_get(self): + """A percent-encoded update id delivers the status GET returns.""" + self._register_package('sub-enc-pkg') + encoded = '%73ub-enc-pkg' + r = self._create_update_subscription(encoded) + self.assertEqual(r.status_code, 201, f'Create failed: {r.text}') + body = r.json() + self.addCleanup(self._cleanup_subscription, body['id']) + self.assertEqual( + body['observed_resource'], f'/api/v1/updates/{encoded}/status') + + event = None + with requests.get(f'http://127.0.0.1:{PORT}{body["event_source"]}', + stream=True, timeout=15) as stream: + for line in stream.iter_lines(decode_unicode=True): + if line and line.startswith('data:'): + event = json.loads(line[len('data:'):].strip()) + break + self.assertIsNotNone(event, 'SSE stream ended without an event') + self.assertNotIn('error', event, f'Sample failed: {event}') + + get = requests.get( + f'{self.BASE_URL}/updates/{encoded}/status', timeout=5) + self.assertEqual(get.status_code, 200, get.text) + self.assertEqual(event['payload'], get.json()) + + def test_02c_update_id_refused_by_get_is_refused(self): + """An id GET refuses (here '..' hidden as %2E%2E) is refused at create.""" + get = requests.get( + f'{self.BASE_URL}/updates/%2E%2E/status', timeout=5) + self.assertEqual(get.status_code, 400, get.text) + r = self._create_update_subscription('%2E%2E') + self.assertEqual(r.status_code, 400, r.text) + def test_03_sse_stream_delivers_update_progress(self): """SSE stream delivers status snapshots during update prepare.""" self._register_package('sub-sse-pkg')