feat: fabric-bridge auto-fetch target_info from remote target node
Initiator can now auto-discover target_info by polling the target's
HTTP control server. Just set target_host and target_port in config:
--config '{"mode":"initiator",...,"target_host":"10.10.100.1","target_port":9100}'
No more manual copy-paste of target_info JSON between hosts.
This commit is contained in:
@@ -4,6 +4,14 @@
|
|||||||
|
|
||||||
#include <spdlog/spdlog.h>
|
#include <spdlog/spdlog.h>
|
||||||
|
|
||||||
|
#ifndef _WIN32
|
||||||
|
#include <sys/socket.h>
|
||||||
|
#include <netinet/in.h>
|
||||||
|
#include <arpa/inet.h>
|
||||||
|
#include <unistd.h>
|
||||||
|
#include <netdb.h>
|
||||||
|
#endif
|
||||||
|
|
||||||
namespace dmf_node {
|
namespace dmf_node {
|
||||||
|
|
||||||
FabricBridgeNode::~FabricBridgeNode() {
|
FabricBridgeNode::~FabricBridgeNode() {
|
||||||
@@ -49,6 +57,12 @@ void FabricBridgeNode::configure(const nlohmann::json& params) {
|
|||||||
spdlog::info("Fabric-bridge: target_info set, signaling startup");
|
spdlog::info("Fabric-bridge: target_info set, signaling startup");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (params.contains("target_host")) {
|
||||||
|
target_host_ = params["target_host"].get<std::string>();
|
||||||
|
target_port_ = params.value("target_port", 9100);
|
||||||
|
spdlog::info("Fabric-bridge: will fetch target_info from {}:{}", target_host_, target_port_);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
|
void FabricBridgeNode::on_add_writer(const std::string& port_id, mxlFlowWriter writer) {
|
||||||
@@ -85,8 +99,99 @@ void FabricBridgeNode::on_remove_reader(const std::string& port_id) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void FabricBridgeNode::fetch_target_info() {
|
||||||
|
if (target_host_.empty()) return;
|
||||||
|
|
||||||
|
spdlog::info("Fabric-bridge: fetching target_info from {}:{}", target_host_, target_port_);
|
||||||
|
|
||||||
|
std::string request = "POST /cmd HTTP/1.1\r\n"
|
||||||
|
"Host: " + target_host_ + "\r\n"
|
||||||
|
"Content-Type: application/json\r\n"
|
||||||
|
"Connection: close\r\n"
|
||||||
|
"Content-Length: 20\r\n"
|
||||||
|
"\r\n"
|
||||||
|
"{\"cmd\":\"status\"}";
|
||||||
|
|
||||||
|
for (int attempt = 0; attempt < 30 && running_ && target_info_str_.empty(); ++attempt) {
|
||||||
|
int sock = socket(AF_INET, SOCK_STREAM, 0);
|
||||||
|
if (sock < 0) {
|
||||||
|
spdlog::warn("Fabric-bridge: socket() failed, retry {}...", attempt + 1);
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
struct hostent* he = gethostbyname(target_host_.c_str());
|
||||||
|
if (!he) {
|
||||||
|
spdlog::warn("Fabric-bridge: cannot resolve {}, retry {}...", target_host_, attempt + 1);
|
||||||
|
close(sock);
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
struct sockaddr_in addr{};
|
||||||
|
addr.sin_family = AF_INET;
|
||||||
|
addr.sin_port = htons(target_port_);
|
||||||
|
std::memcpy(&addr.sin_addr, he->h_addr_list[0], he->h_length);
|
||||||
|
|
||||||
|
if (connect(sock, reinterpret_cast<struct sockaddr*>(&addr), sizeof(addr)) < 0) {
|
||||||
|
spdlog::warn("Fabric-bridge: connect to {}:{} failed, retry {}...", target_host_, target_port_, attempt + 1);
|
||||||
|
close(sock);
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (send(sock, request.c_str(), request.size(), 0) < 0) {
|
||||||
|
spdlog::warn("Fabric-bridge: send failed, retry {}...", attempt + 1);
|
||||||
|
close(sock);
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
std::string response;
|
||||||
|
char buf[4096];
|
||||||
|
ssize_t n;
|
||||||
|
while ((n = recv(sock, buf, sizeof(buf), 0)) > 0) {
|
||||||
|
response.append(buf, n);
|
||||||
|
}
|
||||||
|
close(sock);
|
||||||
|
|
||||||
|
auto body_pos = response.find("\r\n\r\n");
|
||||||
|
if (body_pos == std::string::npos) {
|
||||||
|
spdlog::warn("Fabric-bridge: no HTTP body in response, retry {}...", attempt + 1);
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
auto json = nlohmann::json::parse(response.substr(body_pos + 4));
|
||||||
|
if (json.contains("data") && json["data"].contains("target_info")) {
|
||||||
|
auto ti = json["data"]["target_info"].get<std::string>();
|
||||||
|
if (!ti.empty()) {
|
||||||
|
target_info_str_ = ti;
|
||||||
|
start_requested_.store(true);
|
||||||
|
spdlog::info("Fabric-bridge: fetched target_info ({} bytes)", target_info_str_.size());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} catch (const nlohmann::json::exception& e) {
|
||||||
|
spdlog::warn("Fabric-bridge: JSON parse error: {}", e.what());
|
||||||
|
}
|
||||||
|
|
||||||
|
spdlog::info("Fabric-bridge: target_info not ready yet, retry {}...", attempt + 1);
|
||||||
|
std::this_thread::sleep_for(std::chrono::milliseconds(500));
|
||||||
|
}
|
||||||
|
|
||||||
|
if (target_info_str_.empty()) {
|
||||||
|
spdlog::error("Fabric-bridge: failed to fetch target_info after 30 attempts");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
void FabricBridgeNode::process() {
|
void FabricBridgeNode::process() {
|
||||||
if (!worker_thread_.has_value() && start_requested_.load()) {
|
if (!worker_thread_.has_value() && start_requested_.load()) {
|
||||||
|
if (mode_ == FabricMode::Initiator && !target_host_.empty() && target_info_str_.empty()) {
|
||||||
|
fetch_target_info();
|
||||||
|
}
|
||||||
|
if (start_requested_.load()) {
|
||||||
running_ = true;
|
running_ = true;
|
||||||
spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target");
|
spdlog::info("Fabric-bridge: starting worker thread (mode={})", mode_ == FabricMode::Initiator ? "initiator" : "target");
|
||||||
if (mode_ == FabricMode::Initiator) {
|
if (mode_ == FabricMode::Initiator) {
|
||||||
@@ -95,6 +200,7 @@ void FabricBridgeNode::process() {
|
|||||||
worker_thread_ = std::thread(&FabricBridgeNode::run_target, this);
|
worker_thread_ = std::thread(&FabricBridgeNode::run_target, this);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
std::this_thread::sleep_for(std::chrono::milliseconds(100));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -51,6 +51,11 @@ private:
|
|||||||
|
|
||||||
std::string target_info_str_;
|
std::string target_info_str_;
|
||||||
|
|
||||||
|
std::string target_host_;
|
||||||
|
uint16_t target_port_ = 9100;
|
||||||
|
|
||||||
|
void fetch_target_info();
|
||||||
|
|
||||||
mxlFabricsInstance fabrics_instance_ = nullptr;
|
mxlFabricsInstance fabrics_instance_ = nullptr;
|
||||||
mxlFabricsInitiator initiator_ = nullptr;
|
mxlFabricsInitiator initiator_ = nullptr;
|
||||||
mxlFabricsTarget target_ = nullptr;
|
mxlFabricsTarget target_ = nullptr;
|
||||||
|
|||||||
Reference in New Issue
Block a user