From: Michael Tremer Date: Thu, 9 Jul 2026 17:51:14 +0000 (+0000) Subject: client: Add some scaffolding for a HTTP client X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=e945f6e879120cc5395f0d149e969b1fbd5dc517;p=telemetry.git client: Add some scaffolding for a HTTP client Signed-off-by: Michael Tremer --- diff --git a/Makefile.am b/Makefile.am index f84e653..ddb96d1 100644 --- a/Makefile.am +++ b/Makefile.am @@ -96,6 +96,8 @@ dist_telemetryd_SOURCES = \ src/daemon/buffer.h \ src/daemon/bus.c \ src/daemon/bus.h \ + src/daemon/client.c \ + src/daemon/client.h \ src/daemon/colors.h \ src/daemon/command.c \ src/daemon/command.h \ diff --git a/src/daemon/client.c b/src/daemon/client.c new file mode 100644 index 0000000..e2f3eb3 --- /dev/null +++ b/src/daemon/client.c @@ -0,0 +1,403 @@ +/*############################################################################# +# # +# telemetryd - The IPFire Telemetry Collection Service # +# Copyright (C) 2026 IPFire Development Team # +# # +# This program is free software: you can redistribute it and/or modify # +# it under the terms of the GNU General Public License as published by # +# the Free Software Foundation, either version 3 of the License, or # +# (at your option) any later version. # +# # +# This program is distributed in the hope that it will be useful, # +# but WITHOUT ANY WARRANTY; without even the implied warranty of # +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # +# GNU General Public License for more details. # +# # +# You should have received a copy of the GNU General Public License # +# along with this program. If not, see . # +# # +#############################################################################*/ + +#include + +#include + +#include + +#include "ctx.h" +#include "client.h" + +struct td_client { + td_ctx* ctx; + int nrefs; + + // Event Loop + sd_event* loop; + + // cURL Multi Handle + CURLM* curl; + + // Timer + sd_event_source* timer; + + // How many are still running? + int still_running; +}; + +static int td_client_action(td_client* self) { + void* p = NULL; + + CURLMsg* msg = NULL; + int msgs_left = 0; + + for (;;) { + // Read the next message + msg = curl_multi_info_read(self->curl, &msgs_left); + if (!msg) + break; + + switch (msg->msg) { + case CURLMSG_DONE: + // Update reference to transfer + curl_easy_getinfo(msg->easy_handle, CURLINFO_PRIVATE, &p); + +#if 0 + // Call the done callback + r = td_request_done(p, self->loop, msg->data.result); + if (r < 0) { + switch (-r) { + case EAGAIN: + // Launch it again + // XXX TODO + + // Move to the next message + continue; + + // Handle signal that the handle should not be dequeued, yet + case EINPROGRESS: + continue; + + // Raise any errors + default: + return r; + } + } +#endif + break; + + default: + ERROR(self->ctx, "Received unhandled cURL message %u\n", msg->msg); + break; + } + } + + return 0; +} + +static int td_client_timer_callback(CURLM* multi, long timeout_ms, void* data) { + td_client* self = data; + int r; + + // A negative value indicates that we should remove the timer + if (timeout_ms < 0) { + if (self->timer) { + r = sd_event_source_set_enabled(self->timer, SD_EVENT_OFF); + if (r < 0) + ERROR(self->ctx, "Could not disarm the timer: %s\n", strerror(-r)); + } + + return 0; + } + + // Set the timer + r = sd_event_source_set_time_relative(self->timer, timeout_ms * 1000); + if (r < 0) { + ERROR(self->ctx, "Could not set timer: %s\n", strerror(-r)); + + return r; + } + + // Have the timer fire once + r = sd_event_source_set_enabled(self->timer, SD_EVENT_ONESHOT); + if (r < 0) { + ERROR(self->ctx, "Could not enable the timer: %s\n", strerror(-r)); + + return r; + } + + DEBUG(self->ctx, "cURL set a timer for %ldms\n", timeout_ms); + + return 0; +} + +static int td_client_timer_fired(sd_event_source* s, uint64_t usec, void* data) { + td_client* self = data; + int r; + + DEBUG(self->ctx, "cURL timer fired\n"); + + r = curl_multi_socket_action(self->curl, CURL_SOCKET_TIMEOUT, 0, &self->still_running); + if (r) + return r; + + // Check for any messages + return td_client_action(self); +} + +static int td_client_setup_timer(td_client* self) { + int r; + + // Create a new timer + r = sd_event_add_time_relative(self->loop, &self->timer, CLOCK_MONOTONIC, 0, 0, + td_client_timer_fired, self); + if (r < 0) { + ERROR(self->ctx, "Could not set timer: %s\n", strerror(-r)); + + return r; + } + + return 0; +} + +static int td_client_socket(sd_event_source* s, int fd, uint32_t events, void* data) { + td_client* self = data; + int r; + + int action = 0; + + if (events & EPOLLIN) + action |= CURL_CSELECT_IN; + + if (events & EPOLLOUT) + action |= CURL_CSELECT_OUT; + + //DEBUG(self->ctx, "cURL has activity on socket %d\n", fd); + + // Inform cURL about some socket activity + r = curl_multi_socket_action(self->curl, fd, action, &self->still_running); + if (r) + return r; + + // Check for any messages + r = td_client_action(self); + if (r) + return r; + + // Disarm the timer + if (self->still_running <= 0) { + if (self->timer) { + r = sd_event_source_set_enabled(self->timer, SD_EVENT_OFF); + if (r < 0) { + ERROR(self->ctx, "Could not disarm the timer: %s\n", strerror(-r)); + + return r; + } + } + } + + return 0; +} + +static int td_client_socket_callback(CURL* e, curl_socket_t fd, int what, void* data, void* p) { + td_client* self = data; + sd_event_source* s = p; + uint32_t events = 0; + int r; + + // Remove the socket? + if (what == CURL_POLL_REMOVE) { + // Disable the event + r = sd_event_source_set_enabled(s, SD_EVENT_OFF); + if (r < 0) + ERROR(self->ctx, "Could not disable fd %d: %s\n", fd, strerror(-r)); + + DEBUG(self->ctx, "cURL deregistered socket %d\n", fd); + + return r; + } + + // Do we need to read from this socket? + if (what & CURL_POLL_IN) + events |= EPOLLIN; + + // Do we need to write to this socket? + if (what & CURL_POLL_OUT) + events |= EPOLLOUT; + + // Change events? + if (s) { + r = sd_event_source_set_io_events(s, events); + if (r < 0) { + ERROR(self->ctx, "Could not change events for socket %d: %s\n", + fd, strerror(-r)); + + return r; + } + + DEBUG(self->ctx, "cURL changed socket %d\n", fd); + + return 0; + } + + // Add the socket to the event loop + r = sd_event_add_io(self->loop, &s, fd, events, td_client_socket, self); + if (r < 0) { + ERROR(self->ctx, "Could not register socket %d: %s\n", fd, strerror(-r)); + + goto ERROR; + } + + // Store the event source + curl_multi_assign(self->curl, fd, sd_event_source_ref(s)); + + DEBUG(self->ctx, "cURL registered socket %d\n", fd); + +ERROR: + if (s) + sd_event_source_unref(s); + + return r; +} + +static int td_client_setup_curl(td_client* self) { + int r; + + // Initialize cURL + r = curl_global_init(CURL_GLOBAL_DEFAULT); + if (r) { + ERROR(self->ctx, "Could not initialize cURL: %d\n", r); + return r; + } + + // Create a new multi handle + self->curl = curl_multi_init(); + if (!self->curl) { + ERROR(self->ctx, "Could not create cURL multi handle\n"); + return 1; + } + + // Register with the event loop + r = curl_multi_setopt(self->curl, CURLMOPT_TIMERFUNCTION, td_client_timer_callback); + if (r) { + ERROR(self->ctx, "Could not register the timer function: %s\n", + curl_multi_strerror(r)); + + return r; + } + + r = curl_multi_setopt(self->curl, CURLMOPT_TIMERDATA, self); + if (r) { + ERROR(self->ctx, "Could not register the timer data: %s\n", + curl_multi_strerror(r)); + + return r; + } + + r = curl_multi_setopt(self->curl, CURLMOPT_SOCKETFUNCTION, td_client_socket_callback); + if (r) { + ERROR(self->ctx, "Could not register the socket function: %s\n", + curl_multi_strerror(r)); + + return r; + } + + r = curl_multi_setopt(self->curl, CURLMOPT_SOCKETDATA, self); + if (r) { + ERROR(self->ctx, "Could not register the socket data: %s\n", + curl_multi_strerror(r)); + + return r; + } + + // Set the number of open idle connections + r = curl_multi_setopt(self->curl, CURLMOPT_MAXCONNECTS, 16L); + if (r) { + ERROR(self->ctx, "Could not set max parallel transfers: %s\n", + curl_multi_strerror(r)); + + return r; + } + + // Do not open more than eight connections to the same host + r = curl_multi_setopt(self->curl, CURLMOPT_MAX_HOST_CONNECTIONS, 8L); + if (r) { + ERROR(self->ctx, "Could not set max host connections: %s\n", + curl_multi_strerror(r)); + + return r; + } + + // Limit total number of open connections + r = curl_multi_setopt(self->curl, CURLMOPT_MAX_TOTAL_CONNECTIONS, 64L); + if (r) { + ERROR(self->ctx, "Could not set max total connections: %s\n", + curl_multi_strerror(r)); + + return r; + } + + return 0; +} + +static void td_client_free(td_client* self) { + if (self->timer) + sd_event_source_unref(self->timer); + if (self->curl) + curl_multi_cleanup(self->curl); + if (self->loop) + sd_event_unref(self->loop); + if (self->ctx) + td_ctx_unref(self->ctx); + free(self); +} + +int td_client_create(td_client** client, td_ctx* ctx, td_daemon* daemon) { + td_client* self = NULL; + int r; + + // Allocate some memory + self = calloc(1, sizeof(*self)); + if (!self) + return -errno; + + // Store a reference to the context + self->ctx = td_ctx_ref(ctx); + + // Initialize the reference counter + self->nrefs = 1; + + // Fetch the event loop + self->loop = td_daemon_loop(daemon); + + // Setup timer + r = td_client_setup_timer(self); + if (r < 0) + goto ERROR; + + // Setup cURL + r = td_client_setup_curl(self); + if (r < 0) + goto ERROR; + + // Return the pointer + *client = self; + + return 0; + +ERROR: + td_client_free(self); + return r; +} + +td_client* td_client_ref(td_client* self) { + ++self->nrefs; + return self; +} + +td_client* td_client_unref(td_client* self) { + if (--self->nrefs > 0) + return self; + + td_client_free(self); + return NULL; +} diff --git a/src/daemon/client.h b/src/daemon/client.h new file mode 100644 index 0000000..a3870ab --- /dev/null +++ b/src/daemon/client.h @@ -0,0 +1,35 @@ +/*############################################################################# +# # +# telemetryd - The IPFire Telemetry Collection Service # +# Copyright (C) 2026 IPFire Development Team # +# # +# This program is free software: you can redistribute it and/or modify # +# it under the terms of the GNU General Public License as published by # +# the Free Software Foundation, either version 3 of the License, or # +# (at your option) any later version. # +# # +# This program is distributed in the hope that it will be useful, # +# but WITHOUT ANY WARRANTY; without even the implied warranty of # +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # +# GNU General Public License for more details. # +# # +# You should have received a copy of the GNU General Public License # +# along with this program. If not, see . # +# # +#############################################################################*/ + +#ifndef TELEMETRY_CLIENT_H +#define TELEMETRY_CLIENT_H + +typedef struct td_client td_client; + +#include "ctx.h" +#include "daemon.h" + +int td_client_create(td_client** client, + td_ctx* ctx, td_daemon* daemon); + +td_client* td_client_ref(td_client* self); +td_client* td_client_unref(td_client* self); + +#endif /* TELEMETRY_CLIENT_H */ diff --git a/src/daemon/daemon.c b/src/daemon/daemon.c index b45ae9e..e39c7a3 100644 --- a/src/daemon/daemon.c +++ b/src/daemon/daemon.c @@ -28,6 +28,7 @@ #include #include "bus.h" +#include "client.h" #include "ctx.h" #include "daemon.h" #include "graphs.h" @@ -80,6 +81,9 @@ struct td_daemon { // Netlink td_netlink* netlink; + // HTTP Client + td_client* client; + // Path char path[PATH_MAX]; }; @@ -100,6 +104,11 @@ static int td_daemon_init(sd_event_source* source, void* data) { if (r < 0) goto ERROR; + // Initialize the client + r = td_client_create(&self->client, self->ctx, self); + if (r < 0) + goto ERROR; + #ifdef HAVE_SENSORS // Initialize sensors r = td_sensors_init(self->ctx); @@ -126,6 +135,12 @@ static int td_daemon_on_exit(sd_event_source* source, void* data) { td_sensors_cleanup(self->ctx); #endif /* HAVE_SENSORS */ + // Cleanup client + if (self->client) { + td_client_unref(self->client); + self->client = NULL; + } + // Free all graphs if (self->graphs) { td_graphs_unref(self->graphs);