diff --git a/cf-reactor/Makefile.am b/cf-reactor/Makefile.am index 443e919f112..3fb1f2b1b09 100644 --- a/cf-reactor/Makefile.am +++ b/cf-reactor/Makefile.am @@ -39,7 +39,10 @@ libcf_reactor_la_LIBADD = ../libpromises/libpromises.la libcf_reactor_la_SOURCES = \ cf-reactor.c \ - reactor_context.c reactor_context.h + reactor_context.c reactor_context.h \ + reactor_transform.c reactor_transform.h \ + watcher.c watcher.h \ + wakeup_channel.c wakeup_channel.h if !BUILTIN_EXTENSIONS bin_PROGRAMS = cf-reactor diff --git a/cf-reactor/README.md b/cf-reactor/README.md index 261123b4340..0e9c41ad481 100644 --- a/cf-reactor/README.md +++ b/cf-reactor/README.md @@ -6,7 +6,7 @@ cf-reactor periodically reads the policy to set up the in-memory datastructures needed to react to events. It evaluates reactor bundles and parses events promises to create the corresponding `watchers`. These watchers stay active until the next policy read, when they are rebuilt. -Events promises are parsed into watcher instances, each holding: the event `key` from the events promise, an event `type` (e.g. file deletion), a `payload` with whatever state the check needs (e.g. the name of the file being watched), and a `check function` that detects the event by comparing the current state to the state recorded at the last check (e.g. whether the file still exists). It returns `true` only on the specific transition being watched for, so it fires once per change and does not stay `true` afterwards. +Events promises are parsed into watcher instances, each holding: the event `key` from the events promise, an event `type` (e.g. file deletion), a `state` with whatever state the check needs (e.g. the name of the file being watched), and a `check function` that detects the event by comparing the current state to the state recorded at the last check (e.g. whether the file still exists). It returns `true` only on the specific transition being watched for, so it fires once per change and does not stay `true` afterwards. ### Events @@ -26,28 +26,11 @@ On dispatch, the bundle corresponding to an event is run, however it doesn't do ## Implementation details -### Moving stuff around - -Rather than exposing the raw file-descriptor bookkeeping required for `select(2)`, we introduce a unified interface that serves both the reactor-plugin and event-driven code paths. This is achieved by encapsulating all relevant state in a context struct, `ReactorContext`: - -```C -typedef struct ReactorContext -{ - Seq *fds // array of ReactorFd, which holds the fd and some metadata - fd_set readfds; -} ReactorContext; -``` - -- `ReactorContextInitialize()`: initializes the reactor-plugin and event-driven code. Wraps `ReactorNovaInitialize()` -- `ReactorContextSetupFileDescriptors()`: populates readfds with the file descriptors to monitor, prior to the select() call. -- `ReactorContextHandleEvents()`: iterates over the file descriptors and dispatches the appropriate action based on which ones were signaled as ready. Wraps `ReactorNovaHandleTimeout` and `ReactorNovaHandleEvents()`. -- `ReactorContextFinalize()`: releases the daemon's associated resources. Wraps `ReactorNovaFinalize()`. - ### Tracking spec & Events In order to track all the events promises, we use two datastructures: a global list of `"Watcher"`, which is a struct associated with an event type and the promise name (also called `key`) and a global hashmap mapping this `key` to a `bundle` which is parsed from the policy. -cf-reactor reads the policy periodically, and when it does, rebuilds the list of watchers and the hashmap using the single function `WatcherRegister(key, event_type, payload, bundle, interval)`. Each events promise is associated with an event type, which is defined in `when` bodies: +cf-reactor reads the policy periodically, and when it does, rebuilds the list of watchers and the hashmap using the single function `WatcherRegister(key, event_type, state, bundle, interval)`. Each events promise is associated with an event type, which is defined in `when` bodies: ```cf3 body when file_deleted(filename) @@ -57,11 +40,11 @@ body when file_deleted(filename) ``` Every event type must have defined: -- A check function (called `check_fn` in `Watcher`): This is a function defined specifically for the event that checks if the conditions holds. For example, in case of file deletion, we check if the file doesn't exist anymore compare to the last time we checked. If yes, then it returns `true`. -- A `payload`: This is a struct whose interpretation depends on the event type (thus being declared as `void *`). We typically need some state that we compare between each event-check. In the case of file deletion, we need to know the name of the file we are watching, and whether the file existed last time we checked. -- A payload destroying function (called `destroy_payload`): This is simply a function to free the payload associated with the event type. +- A check function (called `check_callback` in `Watcher`): This is a function defined specifically for the event that checks if the conditions holds. For example, in case of file deletion, we check if the file doesn't exist anymore compare to the last time we checked. If yes, then it returns `true`. +- A `state`: This is a struct whose interpretation depends on the event type (thus being declared as `void *`). We typically need some state that we compare between each event-check. In the case of file deletion, we need to know the name of the file we are watching, and whether the file existed last time we checked. +- A state destroying function (called `destroy_state`): This is simply a function to free the state associated with the event type. -Also, we need a function that will create the payload. That's what `FileWatcherPayloadNew()` does. +Also, we need a function that will create the state. That's what `FileWatcherPayloadNew()` does. So each event type we add in the future just need to have these four things defined, and we need to create the `Watcher` object with the right functions inside `WatcherRegister()` and also call the right `"...PayloadNew()"` function. @@ -71,7 +54,7 @@ So each event type we add in the future just need to have these four things defi `ReactorContextInitialize()` sets up all the necessary data structures for polling, and then starts `WatcherThreadMain`, which polls for events as follows: - It iterates through each watcher in the global list of watchers. -- If the elapsed time exceeds the watcher's `interval`, it runs `check_fn` to determine whether an event has been triggered. +- If the elapsed time exceeds the watcher's `interval`, it runs `check_callback` to determine whether an event has been triggered. - If an event was triggered, it pushes the watcher's `key` (the promise name) onto a thread-safe queue, then signals the file descriptor via `WakeupChannelNotify`, which `select(2)` will pick up on its next iteration. It then goes back to sleep. In parallel, `EventWatcherHandleEvents`, called from within `ReactorContextHandleEvents`, reads from the file descriptor with `WakeupChannelReadFd()` once notified that an event has occurred, and pops the thread-safe queue until it's empty. Each key popped from the queue is looked up in the global hashmap to retrieve the corresponding bundle, which `cf-reactor` then runs (in another thread or subprocess) diff --git a/cf-reactor/cf-reactor.c b/cf-reactor/cf-reactor.c index 80169153a91..975d140031d 100644 --- a/cf-reactor/cf-reactor.c +++ b/cf-reactor/cf-reactor.c @@ -30,13 +30,17 @@ #include #include #include +#include /* LoadPolicy */ +#include /* UpdateLastPolicyUpdateTime */ +#include /* GetInputDir */ #include #include #include #include /* signal, kill */ -#include /* GetSignalPipe, MakeSignalPipe, IsPendingTermination, HandleSignalsForDaemon */ +#include /* GetSignalPipe, MakeSignalPipe, IsPendingTermination, HandleSignalsForDaemon, ReloadConfigRequested, ClearRequestReloadConfig */ #include #include +#include /*****************************************************************************/ /* Globals */ @@ -49,6 +53,7 @@ int NO_FORK = false; /*****************************************************************************/ #define DEFAULT_POLL_INTERVAL_SECS 30 +#define DEFAULT_POLICY_CHECK_INTERVAL_SECS (5 * 60) /*******************************************************************/ /* Command line options */ @@ -70,6 +75,7 @@ static const char *const CF_REACTOR_MANPAGE_LONG_DESCRIPTION = static const struct option OPTIONS[] = { {"debug", no_argument, 0, 'd'}, + {"file", required_argument, 0, 'f'}, {"no-fork", no_argument, 0, 'F'}, {"log-level", required_argument, 0, 'g'}, {"help", no_argument, 0, 'h'}, @@ -84,6 +90,7 @@ static const struct option OPTIONS[] = static const char *const HINTS[] = { "Enable debugging output and run in foreground", + "Specify an alternative input file than the default. This option is overridden by FILE if supplied as argument.", "Run as a foreground process (do not fork)", "Specify how detailed logs should be. Possible values: 'error', 'warning', 'notice', 'info', 'verbose', 'debug'", "Print the help message", @@ -102,11 +109,15 @@ static GenericAgentConfig *CheckOpts(int argc, char **argv) GenericAgentConfig *config = GenericAgentConfigNewDefault(AGENT_TYPE_REACTOR, GetTTYInteractive()); int longopt_idx; - while ((c = getopt_long(argc, argv, "dFg:hIlMvV", + while ((c = getopt_long(argc, argv, "df:Fg:hIlMvV", OPTIONS, &longopt_idx)) != -1) { switch (c) { + case 'f': + GenericAgentConfigSetInputFile(config, GetInputDir(), optarg); + MINUSF = true; + break; case 'd': LogSetGlobalLevel(LOG_LEVEL_DEBUG); NO_FORK = true; @@ -190,6 +201,51 @@ static GenericAgentConfig *CheckOpts(int argc, char **argv) /*****************************************************************************/ +static void CheckPolicyUpdates(EvalContext *ctx, Policy **policy, GenericAgentConfig *config) +{ + time_t validated_at = ReadTimestampFromPolicyValidatedFile(config, NULL); + + bool reload_policy = false; + if (config->agent_specific.daemon.last_validated_at < validated_at) + { + Log(LOG_LEVEL_VERBOSE, "New promises detected, reloading policy"); + reload_policy = true; + } + if (ReloadConfigRequested()) + { + Log(LOG_LEVEL_VERBOSE, "Policy reload requested, reloading policy"); + reload_policy = true; + } + + if (!reload_policy) + { + Log(LOG_LEVEL_DEBUG, "No new promises found"); + return; + } + + ClearRequestReloadConfig(); + config->agent_specific.daemon.last_validated_at = validated_at; + + if (!GenericAgentArePromisesValid(config)) + { + Log(LOG_LEVEL_INFO, "Reactor policy has errors, keeping the previous policy"); + return; + } + + Log(LOG_LEVEL_NOTICE, "Rereading policy file '%s'", config->input_file); + + EvalContextClear(ctx); + PolicyDestroy(*policy); + *policy = NULL; + + UpdateLastPolicyUpdateTime(ctx); + GenericAgentDiscoverContext(ctx, config, NULL); + + *policy = LoadPolicy(ctx, config); +} + +/*****************************************************************************/ + int main(int argc, char *argv[]) { @@ -197,6 +253,20 @@ int main(int argc, char *argv[]) EvalContext *ctx = EvalContextNew(); GenericAgentConfigApply(ctx, config); + const char *program_invocation_name = argv[0]; + const char *last_dir_sep = strrchr(program_invocation_name, FILE_SEPARATOR); + const char *program_name = (last_dir_sep != NULL ? last_dir_sep + 1 : program_invocation_name); + GenericAgentDiscoverContext(ctx, config, program_name); + + Policy *policy = SelectAndLoadPolicy(config, ctx, false, false); + if (!policy) + { + Log(LOG_LEVEL_ERR, "Error reading CFEngine policy. Exiting..."); + DoCleanupAndExit(EXIT_FAILURE); + } + + GenericAgentPostLoadInit(ctx); + #ifdef __MINGW32__ if (!NO_FORK) @@ -253,15 +323,21 @@ int main(int argc, char *argv[]) /* We need an initial value here for the first iteration of the cycle * below. */ time_t next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS; + time_t next_policy_check = time(NULL) + DEFAULT_POLICY_CHECK_INTERVAL_SECS; + + // Setup event watchers from policy + KeepReactorPromises(ctx, policy); + while (!IsPendingTermination()) { int max_fd = ReactorContextSetupFileDescriptors(&reactor_ctx); /* Determine how much time is remaining until the next tick. */ time_t last_tick = time(NULL); - time_t remaining = next_tick > last_tick ? next_tick - last_tick : 0; + time_t remaining_tick = next_tick > last_tick ? next_tick - last_tick : 0; + time_t remaining_policy_check = next_policy_check > last_tick ? next_policy_check - last_tick : 0; - struct timeval timeout = { .tv_sec = remaining }; + struct timeval timeout = { .tv_sec = MIN(remaining_tick, remaining_policy_check) }; int ret = select(max_fd, &reactor_ctx.readfds, NULL, NULL, &timeout); /* Reschedule the backstop tick against the current time (not @@ -290,14 +366,25 @@ int main(int argc, char *argv[]) { /*** timeout ***/ ReactorNovaHandleTimeout(&next_tick); - continue; } - /* else */ + else + { + ReactorContextHandleEvents(&reactor_ctx, &next_tick); + } + + + time_t now = time(NULL); + if (now >= next_policy_check || ReloadConfigRequested()) + { + CheckPolicyUpdates(ctx, &policy, config); + next_policy_check = now + DEFAULT_POLICY_CHECK_INTERVAL_SECS; - ReactorContextHandleEvents(&reactor_ctx, &next_tick); + KeepReactorPromises(ctx, policy); + } } ReactorContextFinalize(&reactor_ctx); + PolicyDestroy(policy); GenericAgentFinalize(ctx, config); CallCleanupFunctions(); diff --git a/cf-reactor/reactor_context.c b/cf-reactor/reactor_context.c index 71f7b7bc3a1..db78bb85e68 100644 --- a/cf-reactor/reactor_context.c +++ b/cf-reactor/reactor_context.c @@ -25,6 +25,7 @@ #include #include /* ReactorNova*() */ #include /* GetSignalPipe() */ +#include #include #define INIT_FD_COUNT 8 @@ -37,69 +38,88 @@ static ReactorFd *ReactorFdNew(int fd, ReactorFdType type) return rfd; } -bool ReactorContextInitialize(ReactorContext *ctx) +bool ReactorContextInitialize(ReactorContext *reactor_context) { - assert(ctx != NULL); + assert(reactor_context != NULL); - ctx->fds = SeqNew(INIT_FD_COUNT, free); + reactor_context->fds = SeqNew(INIT_FD_COUNT, free); + + WatcherRegistryInitialize(); // Initialize Nova fds { - ctx->max_nova_fds = ReactorNovaMaxFds(); - int *nova_fds = (int *) xmalloc(ctx->max_nova_fds * sizeof(int)); + reactor_context->max_nova_fds = ReactorNovaMaxFds(); + int *nova_fds = (int *) xmalloc(reactor_context->max_nova_fds * sizeof(int)); size_t num_nova_fds = 0; - if (!ReactorNovaInitialize(nova_fds, ctx->max_nova_fds, &num_nova_fds)) + if (!ReactorNovaInitialize(nova_fds, reactor_context->max_nova_fds, &num_nova_fds)) { free(nova_fds); - SeqDestroy(ctx->fds); - ctx->fds = NULL; + SeqDestroy(reactor_context->fds); + reactor_context->fds = NULL; return false; } + // num_nova_fds can never end up less than max_nova_fds here: Nova + // asserts internally that it always reports back the same fd count + // it advertises as its max (see the assert on poll_fd_idx in + // SetupEventProcessing(), nova/reactor-plugin/cf-reactor.c), + // so this loop always appends exactly max_nova_fds entries. for (size_t i = 0; i < num_nova_fds; i++) { - SeqAppend(ctx->fds, ReactorFdNew(nova_fds[i], REACTOR_FD_NOVA)); + SeqAppend(reactor_context->fds, ReactorFdNew(nova_fds[i], REACTOR_FD_NOVA)); } free(nova_fds); } - // TODO: initialize other event sources here. - + // Initialize event watcher fd + { + int watcher_fd; + if (!EventWatcherInitialize(&watcher_fd)) + { + ReactorNovaFinalize(); + WatcherRegistryFinalize(); + SeqDestroy(reactor_context->fds); + reactor_context->fds = NULL; + return false; + } + SeqAppend(reactor_context->fds, ReactorFdNew(watcher_fd, REACTOR_FD_WATCHER)); + } + return true; } -int ReactorContextSetupFileDescriptors(ReactorContext *ctx) +int ReactorContextSetupFileDescriptors(ReactorContext *reactor_context) { - assert(ctx != NULL); + assert(reactor_context != NULL); - FD_ZERO(&ctx->readfds); + FD_ZERO(&reactor_context->readfds); int signal_pipe = GetSignalPipe(); - FD_SET(signal_pipe, &ctx->readfds); + FD_SET(signal_pipe, &reactor_context->readfds); int max_fd = signal_pipe; - for (size_t i = 0; i < SeqLength(ctx->fds); i++) + for (size_t i = 0; i < SeqLength(reactor_context->fds); i++) { - const ReactorFd *rfd = SeqAt(ctx->fds, i); - FD_SET(rfd->fd, &ctx->readfds); + const ReactorFd *rfd = SeqAt(reactor_context->fds, i); + FD_SET(rfd->fd, &reactor_context->readfds); max_fd = MAX(rfd->fd, max_fd); } return max_fd + 1; } -static bool NovaHasTimedOut(const ReactorContext *ctx) +static bool NovaHasTimedOut(const ReactorContext *reactor_context) { - assert(ctx != NULL); - for (size_t i = 0; i < SeqLength(ctx->fds); i++) + assert(reactor_context != NULL); + for (size_t i = 0; i < SeqLength(reactor_context->fds); i++) { - const ReactorFd *rfd = SeqAt(ctx->fds, i); + const ReactorFd *rfd = SeqAt(reactor_context->fds, i); if (rfd->type != REACTOR_FD_NOVA) { continue; } - if (FD_ISSET(rfd->fd, &ctx->readfds)) + if (FD_ISSET(rfd->fd, &reactor_context->readfds)) { return false; } @@ -107,36 +127,40 @@ static bool NovaHasTimedOut(const ReactorContext *ctx) return true; } -static int *GetNovaFds(const ReactorContext *ctx) +static int *GetNovaFds(const ReactorContext *reactor_context) { - assert(ctx != NULL); - int *nova_fds = (int *) xmalloc(ctx->max_nova_fds * sizeof(int)); + assert(reactor_context != NULL); + int *nova_fds = (int *) xmalloc(reactor_context->max_nova_fds * sizeof(int)); size_t num_nova_fds = 0; - for (size_t i = 0; i < SeqLength(ctx->fds); i++) + for (size_t i = 0; i < SeqLength(reactor_context->fds); i++) { - const ReactorFd *rfd = SeqAt(ctx->fds, i); + const ReactorFd *rfd = SeqAt(reactor_context->fds, i); if (rfd->type != REACTOR_FD_NOVA) { continue; } - // Since we came so far, this should be always true - assert(num_nova_fds < ctx->max_nova_fds); + // This can never go out of bounds: reactor_context->fds always + // holds exactly max_nova_fds REACTOR_FD_NOVA entries (see the + // comment in ReactorContextInitialize()), so num_nova_fds cannot + // exceed max_nova_fds and nova_fds is always fully populated by + // the time this function returns. + assert(num_nova_fds < reactor_context->max_nova_fds); nova_fds[num_nova_fds++] = rfd->fd; } return nova_fds; } -static void SetNovaFds(ReactorContext *ctx, const int *nova_fds) +static void SetNovaFds(ReactorContext *reactor_context, const int *nova_fds) { - assert(ctx != NULL); + assert(reactor_context != NULL); size_t num_nova_fds = 0; - for (size_t i = 0; i < SeqLength(ctx->fds); i++) + for (size_t i = 0; i < SeqLength(reactor_context->fds); i++) { - ReactorFd *rfd = SeqAt(ctx->fds, i); + ReactorFd *rfd = SeqAt(reactor_context->fds, i); if (rfd->type != REACTOR_FD_NOVA) { @@ -145,21 +169,36 @@ static void SetNovaFds(ReactorContext *ctx, const int *nova_fds) rfd->fd = nova_fds[num_nova_fds++]; } } +static int GetWatcherFd(const ReactorContext *reactor_context) +{ + assert(reactor_context != NULL); + for (size_t i = 0; i < SeqLength(reactor_context->fds); i++) + { + const ReactorFd *rfd = SeqAt(reactor_context->fds, i); + + if (rfd->type == REACTOR_FD_WATCHER) + { + return rfd->fd; + } + } + ProgrammingError("Reactor was not initialized with event watcher fd"); + return 0; +} -void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick) +void ReactorContextHandleEvents(ReactorContext *reactor_context, time_t *next_tick) { - assert(ctx != NULL); + assert(reactor_context != NULL); - if (NovaHasTimedOut(ctx)) + if (NovaHasTimedOut(reactor_context)) { ReactorNovaHandleTimeout(next_tick); } else { - int *nova_fds = GetNovaFds(ctx); - ReactorNovaHandleEvents(&ctx->readfds, nova_fds, next_tick); + int *nova_fds = GetNovaFds(reactor_context); + ReactorNovaHandleEvents(&reactor_context->readfds, nova_fds, next_tick); // ReactorNova replaces the fd of broken connection - SetNovaFds(ctx, nova_fds); + SetNovaFds(reactor_context, nova_fds); free(nova_fds); } @@ -168,22 +207,23 @@ void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick) * promptly on a pending signal, but (per its own contract in * signals.c) it must be drained or it stays "ready" forever, which * would stop select() from ever blocking again. */ - if (FD_ISSET(GetSignalPipe(), &ctx->readfds)) + if (FD_ISSET(GetSignalPipe(), &reactor_context->readfds)) { unsigned char buf; while (recv(GetSignalPipe(), &buf, 1, 0) > 0) { /* drain */ } } - // TODO: handle events for other event sources here. + EventWatcherHandleEvents(GetWatcherFd(reactor_context), &reactor_context->readfds); } -void ReactorContextFinalize(ReactorContext *ctx) +void ReactorContextFinalize(ReactorContext *reactor_context) { - assert(ctx != NULL); + assert(reactor_context != NULL); + EventWatcherFinalize(); ReactorNovaFinalize(); - ctx->max_nova_fds = 0; + reactor_context->max_nova_fds = 0; - SeqDestroy(ctx->fds); - ctx->fds = NULL; + SeqDestroy(reactor_context->fds); + reactor_context->fds = NULL; } diff --git a/cf-reactor/reactor_context.h b/cf-reactor/reactor_context.h index 632664a5f74..3950b94696a 100644 --- a/cf-reactor/reactor_context.h +++ b/cf-reactor/reactor_context.h @@ -30,7 +30,8 @@ typedef enum { - REACTOR_FD_NOVA + REACTOR_FD_NOVA, + REACTOR_FD_WATCHER } ReactorFdType; /** @@ -52,9 +53,9 @@ typedef struct size_t max_nova_fds; } ReactorContext; -bool ReactorContextInitialize(ReactorContext *ctx); -int ReactorContextSetupFileDescriptors(ReactorContext *ctx); -void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick); -void ReactorContextFinalize(ReactorContext *ctx); +bool ReactorContextInitialize(ReactorContext *reactor_context); +int ReactorContextSetupFileDescriptors(ReactorContext *reactor_context); +void ReactorContextHandleEvents(ReactorContext *reactor_context, time_t *next_tick); +void ReactorContextFinalize(ReactorContext *reactor_context); #endif diff --git a/cf-reactor/reactor_transform.c b/cf-reactor/reactor_transform.c new file mode 100644 index 00000000000..481cdba1b9d --- /dev/null +++ b/cf-reactor/reactor_transform.c @@ -0,0 +1,173 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + 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; version 3. + + 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, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* Promise types evaluated within `bundle reactor NAME { ... }`. */ +static const char *const REACTOR_TYPESEQUENCE[] = +{ + "meta", + "vars", + "classes", + "events", + NULL +}; + +// Temporary implementation of KeepEventsPromise that doesn't reflect how we have defined when bodies: +// 1. We only expect one constraint, "file_deleted" +// 2. The original idea was that we could "or" constraints that are inside the same when body +// TODO: adapt this function to the original design +static void KeepEventsPromise(EvalContext *ctx, const Promise *pp) +{ + const char *key = pp->promiser; + + const char *path = PromiseGetConstraintAsRval(pp, "file_deleted", RVAL_TYPE_SCALAR); + if (path == NULL) + { + Log(LOG_LEVEL_WARNING, + "Reactor events promise '%s' must specify exactly one file to watch in its 'when' body, ignoring", + key); + return; + } + + Constraint *then_constraint = PromiseGetConstraint(pp, "then"); + const char *bundle_name = NULL; + if (then_constraint != NULL) + { + switch (then_constraint->rval.type) + { + case RVAL_TYPE_SCALAR: + bundle_name = RvalScalarValue(then_constraint->rval); + break; + case RVAL_TYPE_FNCALL: + bundle_name = RvalFnCallValue(then_constraint->rval)->name; + break; + default: + break; + } + } + + if (bundle_name == NULL) + { + Log(LOG_LEVEL_ERR, "Reactor events promise '%s' does not specify a 'then' bundle, ignoring", key); + return; + } + + const Bundle *bundle = EvalContextResolveBundleExpression(ctx, PromiseGetPolicy(pp), bundle_name, "agent"); + if (bundle == NULL) + { + bundle = EvalContextResolveBundleExpression(ctx, PromiseGetPolicy(pp), bundle_name, "common"); + } + if (bundle == NULL) + { + Log(LOG_LEVEL_ERR, "Reactor events promise '%s' refers to unknown bundle '%s', ignoring", key, bundle_name); + return; + } + + // TODO: Call WatcherRegister. We have here all the information we need (event type, state, events promiser) + Log(LOG_LEVEL_INFO, "Registering a file_deleted watcher with key '%s', on file '%s'", key, path); +} + +static PromiseResult KeepReactorPromise(EvalContext *ctx, const Promise *pp, ARG_UNUSED void *param) +{ + assert(param == NULL); + PromiseBanner(ctx, pp); + + if (strcmp(PromiseGetPromiseType(pp), "vars") == 0 || + strcmp(PromiseGetPromiseType(pp), "meta") == 0) + { + return VerifyVarPromise(ctx, pp, NULL); + } + + if (strcmp(PromiseGetPromiseType(pp), "classes") == 0) + { + return VerifyClassPromise(ctx, pp, NULL); + } + + if (strcmp(PromiseGetPromiseType(pp), "events") == 0) + { + KeepEventsPromise(ctx, pp); + return PROMISE_RESULT_NOOP; + } + + return PROMISE_RESULT_NOOP; +} + +static void EvaluateReactorBundle(EvalContext *ctx, const Bundle *bp) +{ + EvalContextStackPushBundleFrame(ctx, bp, NULL, false, NULL); + + for (int i = 0; REACTOR_TYPESEQUENCE[i] != NULL; i++) + { + const BundleSection *sp = BundleGetSection(bp, REACTOR_TYPESEQUENCE[i]); + if (!sp || SeqLength(sp->promises) == 0) + { + Log(LOG_LEVEL_DEBUG, "No promise type %s in bundle %s", + REACTOR_TYPESEQUENCE[i], bp->name); + continue; + } + + EvalContextStackPushBundleSectionFrame(ctx, sp); + for (size_t j = 0; j < SeqLength(sp->promises); j++) + { + Promise *pp = SeqAt(sp->promises, j); + ExpandPromise(ctx, pp, KeepReactorPromise, NULL); + } + EvalContextStackPopFrame(ctx); + } + + EvalContextStackPopFrame(ctx); +} + +void KeepReactorPromises(EvalContext *ctx, const Policy *policy) +{ + for (size_t i = 0; i < SeqLength(policy->bundles); i++) + { + Bundle *bp = SeqAt(policy->bundles, i); + if (strcmp(bp->type, CF_AGENTTYPES[AGENT_TYPE_REACTOR]) != 0) + { + continue; + } + + if (RlistLen(bp->args) > 0) + { + Log(LOG_LEVEL_WARNING, + "Cannot implicitly evaluate bundle '%s %s', as this bundle takes arguments.", + bp->type, bp->name); + continue; + } + + EvaluateReactorBundle(ctx, bp); + } +} diff --git a/cf-reactor/reactor_transform.h b/cf-reactor/reactor_transform.h new file mode 100644 index 00000000000..5fb31d20474 --- /dev/null +++ b/cf-reactor/reactor_transform.h @@ -0,0 +1,39 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + 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; version 3. + + 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, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#ifndef CFENGINE_REACTOR_TRANSFORM_H +#define CFENGINE_REACTOR_TRANSFORM_H + +#include +#include + +/** + * @brief Evaluate every `bundle reactor NAME { ... }` in the given policy. + * + * Only the "meta", "vars" and "classes" promise types are evaluated; + * "events" promises are not evaluated yet (see watcher.c). + */ +void KeepReactorPromises(EvalContext *ctx, const Policy *policy); + +#endif diff --git a/cf-reactor/wakeup_channel.c b/cf-reactor/wakeup_channel.c new file mode 100644 index 00000000000..e2aa7e740dd --- /dev/null +++ b/cf-reactor/wakeup_channel.c @@ -0,0 +1,92 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + 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; version 3. + + 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, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#include +#include +#include /* cf_closesocket() */ +#include /* MakeNonBlockingSocketPair() */ + +bool WakeupChannelOpen(WakeupChannel *channel) +{ + assert(channel != NULL); + + return MakeNonBlockingSocketPair(channel->fds); +} + +int WakeupChannelReadFd(const WakeupChannel *channel) +{ + assert(channel != NULL); + return channel->fds[0]; +} + +void WakeupChannelNotify(const WakeupChannel *channel) +{ + assert(channel != NULL); + + /* One byte is enough to wake the reader up; if the channel happens to + * already be full, the reader is already guaranteed to wake up because + * of what's queued, so a transient EAGAIN/EWOULDBLOCK here is fine. */ + unsigned char byte = 1; + if (send(channel->fds[1], (const char *) &byte, sizeof(byte), 0) < 0) + { + // These signal contention. Everything else is an error. + if (errno != EAGAIN +#ifndef __MINGW32__ + && errno != EWOULDBLOCK +#endif + ) + { + /* Not fatal: unlike SignalNotify(), this doesn't run in a + * signal handler, so there's no async-signal-safety reason to + * _exit() here. But it's still worth logging loudly: on a real + * failure the reactor's select() loop has no other way to + * notice a fired watcher event, since a select() timeout only + * re-arms ReactorNova, it doesn't drain event_queue (see + * EventWatcherHandleEvents()). */ + Log(LOG_LEVEL_ERR, "Could not notify wakeup channel (send: '%s')", GetErrorStr()); + } + } +} + +void WakeupChannelDrain(const WakeupChannel *channel) +{ + assert(channel != NULL); + + unsigned char buf; + while (recv(channel->fds[0], (char *) &buf, sizeof(buf), 0) > 0) { /* drain */ } +} + +void WakeupChannelClose(WakeupChannel *channel) +{ + assert(channel != NULL); + + for (int i = 0; i < 2; i++) + { + if (channel->fds[i] != -1) + { + cf_closesocket(channel->fds[i]); + channel->fds[i] = -1; + } + } +} diff --git a/cf-reactor/wakeup_channel.h b/cf-reactor/wakeup_channel.h new file mode 100644 index 00000000000..8fd2ab197fa --- /dev/null +++ b/cf-reactor/wakeup_channel.h @@ -0,0 +1,49 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + 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; version 3. + + 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, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#ifndef CFENGINE_WAKEUP_CHANNEL_H +#define CFENGINE_WAKEUP_CHANNEL_H + +#include + +/** + * @brief A cross-platform self-pipe: lets a background thread wake up the + * daemon's select(2) loop on demand. + * + * Any current or future background event source (the watcher subsystem + * today, potentially others later) that needs to interrupt select() should + * own one of these rather than inventing its own pipe/socketpair handling. + */ +typedef struct +{ + int fds[2]; /* [0] = read end, add to the select() fd_set; [1] = write end */ +} WakeupChannel; + +bool WakeupChannelOpen(WakeupChannel *channel); +int WakeupChannelReadFd(const WakeupChannel *channel); +void WakeupChannelNotify(const WakeupChannel *channel); +void WakeupChannelDrain(const WakeupChannel *channel); +void WakeupChannelClose(WakeupChannel *channel); + +#endif /* CFENGINE_WAKEUP_CHANNEL_H */ diff --git a/cf-reactor/watcher.c b/cf-reactor/watcher.c new file mode 100644 index 00000000000..3edf269e222 --- /dev/null +++ b/cf-reactor/watcher.c @@ -0,0 +1,247 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + 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; version 3. + + 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, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#include +#include +#include // MaskTerminationSignalsInThread() +#include // IsPendingTermination() +#include +#include +#include // StringHash_untyped(), StringEqual_untyped() +#include +#include +#include +#include + +/* Upper bound on how long the watcher thread ever sleeps in one go, so that + * IsPendingTermination() is re-checked at least this often during shutdown, + * regardless of what poll intervals watchers asked for. */ +#define MAX_WATCHER_THREAD_SLEEP_SECS 1 + +/* Not currently caller-configurable -- WatcherRegister() doesn't take an + * interval parameter -- so every watcher polls at this rate for now. */ +#define DEFAULT_WATCHER_POLL_INTERVAL_SECS 10 + +typedef struct +{ + char *key; + WatcherCheckFn check_callback; // resolved from `type` at WatcherRegister() time + WatcherStateDestroyFn destroy_state; + void *state; + time_t poll_interval_secs; + time_t next_due; +} Watcher; + +static void WatcherDestroy(void *item); /* defined below, next to WatcherRegister() */ + +// TODO: potential race condition. If the policy is reparsed and watchers are re-registered +// while the watcher thread is iterating over this Seq, +// it may read a Watcher that is being freed or reallocated concurrently +static Seq *watchers = NULL; +static Map *event_to_bundle = NULL; + +static WakeupChannel wakeup_channel; +static ThreadedQueue *event_queue = NULL; +static pthread_t watcher_thread; + +/*****************************************************************************/ + +void WatcherRegistryInitialize(void) +{ + assert(watchers == NULL); + assert(event_to_bundle == NULL); + + watchers = SeqNew(4, WatcherDestroy); + event_to_bundle = MapNew(StringHash_untyped, StringEqual_untyped, NULL, NULL); +} + +void WatcherRegistryFinalize(void) +{ + SeqDestroy(watchers); + watchers = NULL; + MapDestroy(event_to_bundle); + event_to_bundle = NULL; +} + +void WatcherRegister(const char *key, EventType type, void *state, Bundle *bundle, time_t interval) +{ + assert(key != NULL); + assert(bundle != NULL); + assert(watchers != NULL && event_to_bundle != NULL); + + WatcherCheckFn check_callback = NULL; + WatcherStateDestroyFn destroy_state = NULL; + switch (type) + { + + case EVENT_FILE_DELETED: + // TODO: initialize check_callback and destroy_state + break; + + default: + ProgrammingError("Unknown reactor event type %d for watcher '%s'", (int) type, key); + } + + if (MapHasKey(event_to_bundle, key)) + { + Log(LOG_LEVEL_ERR, "Reactor watcher key '%s' is already registered, ignoring the duplicate", key); + if (destroy_state != NULL) + { + destroy_state(state); + } + return; + } + + Watcher *w = xmalloc(sizeof(Watcher)); + w->key = xstrdup(key); + w->state = state; + w->poll_interval_secs = interval; + w->next_due = 0; /* due immediately on the watcher thread's first pass */ + w->check_callback = check_callback; + w->destroy_state = destroy_state; + + SeqAppend(watchers, w); + MapInsert(event_to_bundle, w->key, bundle); +} + +static void WatcherDestroy(void *item) +{ + Watcher *w = item; + if (w->destroy_state != NULL) + { + w->destroy_state(w->state); + } + free(w->key); + free(w); +} + +static void *WatcherThreadMain(ARG_UNUSED void *unused) +{ + /* Keep termination signals landing on the main thread (which owns the + * daemon's HandleSignalsForDaemon()-based shutdown), never on this one. + * No-op on Windows -- see signal_lib.h. */ +#ifndef __MINGW32__ + MaskTerminationSignalsInThread(); +#endif + + while (!IsPendingTermination()) + { + time_t now = time(NULL); + time_t sleep_for = MAX_WATCHER_THREAD_SLEEP_SECS; + bool any_event = false; + + for (size_t i = 0; i < SeqLength(watchers); i++) + { + Watcher *w = SeqAt(watchers, i); + + if (now >= w->next_due) + { + bool fired = w->check_callback(w->state); + w->next_due = now + w->poll_interval_secs; + if (fired) + { + /** TODO: potential use after free. If a policy reparse destroys this Watcher (and frees w->key) + * before the queued key is consumed, the consumer will read freed memory */ + ThreadedQueuePush(event_queue, w->key); + any_event = true; + } + } + + time_t until_due = (w->next_due > now) ? (w->next_due - now) : 0; + sleep_for = MIN(sleep_for, until_due); + } + + if (any_event) + { + WakeupChannelNotify(&wakeup_channel); + } + + if (sleep_for > 0) + { + sleep((unsigned int) sleep_for); + } + } + + return NULL; +} + +bool EventWatcherInitialize(int *fd) +{ + assert(fd != NULL); + assert(watchers != NULL && event_to_bundle != NULL); /* WatcherRegistryInitialize() first */ + + if (!WakeupChannelOpen(&wakeup_channel)) + { + Log(LOG_LEVEL_ERR, "Failed to open wakeup channel for cf-reactor event watcher"); + return false; + } + + event_queue = ThreadedQueueNew(16, NULL); + + int ret = pthread_create(&watcher_thread, NULL, WatcherThreadMain, NULL); + if (ret != 0) + { + Log(LOG_LEVEL_ERR, "Unable to start cf-reactor watcher thread: %s", GetErrorStrFromCode(ret)); + ThreadedQueueDestroy(event_queue); + event_queue = NULL; + WakeupChannelClose(&wakeup_channel); + return false; + } + + *fd = WakeupChannelReadFd(&wakeup_channel); + + Log(LOG_LEVEL_VERBOSE, "Started reactor watcher subsystem with %zu watcher(s)", SeqLength(watchers)); + return true; +} + +void EventWatcherHandleEvents(int fd, fd_set *readfds) +{ + assert(readfds != NULL); + + if (!FD_ISSET(fd, readfds)) + { + return; + } + + WakeupChannelDrain(&wakeup_channel); + + void *item; + while (ThreadedQueuePop(event_queue, &item, 0)) + { + const char *key = item; + ARG_UNUSED const Bundle *bundle = MapGet(event_to_bundle, key); + Log(LOG_LEVEL_NOTICE, "Reactor watcher '%s' fired", key); + // TODO: run bundle + } +} + +void EventWatcherFinalize(void) +{ + pthread_join(watcher_thread, NULL); + ThreadedQueueDestroy(event_queue); + event_queue = NULL; + WakeupChannelClose(&wakeup_channel); + + WatcherRegistryFinalize(); +} diff --git a/cf-reactor/watcher.h b/cf-reactor/watcher.h new file mode 100644 index 00000000000..4f458386e84 --- /dev/null +++ b/cf-reactor/watcher.h @@ -0,0 +1,55 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + 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; version 3. + + 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, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#ifndef CFENGINE_WATCHER_H +#define CFENGINE_WATCHER_H + +#include /* Bundle */ + +typedef enum +{ + EVENT_FILE_DELETED, +} EventType; + +typedef bool (*WatcherCheckFn)(void *state); +typedef void (*WatcherStateDestroyFn)(void *state); + +void WatcherRegistryInitialize(void); +void WatcherRegistryFinalize(void); + +/** + * @brief Register a specific watcher instance. + * + * @param key the events promise identifier + * @param type the type of watcher, defined in when bodies + * @param state the data used for by the watcher, depending on the type + * @param bundle the bundle to run on event + * @param interval interval between runs + */ +void WatcherRegister(const char *key, EventType type, void *state, Bundle *bundle, time_t interval); +bool EventWatcherInitialize(int *fd); +void EventWatcherHandleEvents(int fd, fd_set *readfds); +void EventWatcherFinalize(void); + +#endif diff --git a/libpromises/cf3parse_logic.h b/libpromises/cf3parse_logic.h index 2f98c586019..8ec5b36c369 100644 --- a/libpromises/cf3parse_logic.h +++ b/libpromises/cf3parse_logic.h @@ -236,6 +236,13 @@ static bool RelevantBundle(const char *agent, const char *blocktype) } } + // cf-reactor should keep agent bundles + if (StringEqual(agent, CF_AGENTTYPES[AGENT_TYPE_REACTOR]) + && StringEqual(blocktype, CF_AGENTTYPES[AGENT_TYPE_AGENT])) + { + return true; + } + DeleteItemList(ip); return false; } diff --git a/libpromises/generic_agent.c b/libpromises/generic_agent.c index 3e282c946ec..a31ec97230f 100644 --- a/libpromises/generic_agent.c +++ b/libpromises/generic_agent.c @@ -1199,7 +1199,8 @@ bool GenericAgentCheckPolicy(GenericAgentConfig *config, bool force_validation, { if (config->agent_type == AGENT_TYPE_SERVER || config->agent_type == AGENT_TYPE_MONITOR || - config->agent_type == AGENT_TYPE_EXECUTOR) + config->agent_type == AGENT_TYPE_EXECUTOR || + config->agent_type == AGENT_TYPE_REACTOR) { time_t validated_at = ReadTimestampFromPolicyValidatedFile(config, NULL); config->agent_specific.daemon.last_validated_at = validated_at; diff --git a/libpromises/mod_reactor.c b/libpromises/mod_reactor.c index c5a1d738ce7..0bc16cd1fc8 100644 --- a/libpromises/mod_reactor.c +++ b/libpromises/mod_reactor.c @@ -31,7 +31,7 @@ static const ConstraintSyntax when_constraints[] = CONSTRAINT_SYNTAX_GLOBAL, /* Row models */ - ConstraintSyntaxNewStringList("files_deleted", CF_ANYSTRING, "List of files to react for on deletion", SYNTAX_STATUS_NORMAL), + ConstraintSyntaxNewString("file_deleted", CF_ANYSTRING, "File to react for on deletion", SYNTAX_STATUS_NORMAL), ConstraintSyntaxNewNull() }; diff --git a/libpromises/signals.c b/libpromises/signals.c index 21d150cd65f..75370656ffe 100644 --- a/libpromises/signals.c +++ b/libpromises/signals.c @@ -27,6 +27,7 @@ #include /* GetStateDir() */ #include /* FILE_SEPARATOR */ #include /* TerminateCustomPromises */ +#include /* cf_closesocket() */ static bool PENDING_TERMINATION = false; /* GLOBAL_X */ @@ -72,45 +73,63 @@ static void CloseSignalPipe(void) } /** - * Make a pipe that can be used to flag that a signal has arrived. - * Using a pipe avoids race conditions, since it saves its values until emptied. - * Use GetSignalPipe() to get the pipe. + * Creates an AF_UNIX SOCK_STREAM socket pair and sets both ends + * non-blocking. * Note that we use a real socket as the pipe, because Windows only supports * using select() with real sockets. This means also using send() and recv() * instead of write() and read(). */ -void MakeSignalPipe(void) +bool MakeNonBlockingSocketPair(int fds[2]) { - if (socketpair(AF_UNIX, SOCK_STREAM, 0, SIGNAL_PIPE) != 0) + fds[0] = -1; + fds[1] = -1; + + if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) != 0) { - Log(LOG_LEVEL_CRIT, "Could not create internal communication pipe. Cannot continue. (socketpair: '%s')", - GetErrorStr()); - DoCleanupAndExit(EXIT_FAILURE); + Log(LOG_LEVEL_ERR, "Could not create socket pair (socketpair: '%s')", GetErrorStr()); + return false; } - RegisterCleanupFunction(&CloseSignalPipe); - for (int c = 0; c < 2; c++) { #ifdef __MINGW32__ u_long enable = 1; - int ret = ioctlsocket(SIGNAL_PIPE[c], FIONBIO, &enable); + int ret = ioctlsocket(fds[c], FIONBIO, &enable); #define CNTLNAME "ioctlsocket" #else /* Unix: */ - int ret = fcntl(SIGNAL_PIPE[c], F_SETFL, O_NONBLOCK); + int ret = fcntl(fds[c], F_SETFL, O_NONBLOCK); #define CNTLNAME "fcntl" #endif /* __MINGW32__ */ if (ret != 0) { - Log(LOG_LEVEL_CRIT, - "Could not unblock internal communication pipe. " - "Cannot continue. (" CNTLNAME ": '%s')", - GetErrorStr()); - DoCleanupAndExit(EXIT_FAILURE); + Log(LOG_LEVEL_ERR, "Could not set socket pair to non-blocking (" CNTLNAME ": '%s')", GetErrorStr()); + cf_closesocket(fds[0]); + cf_closesocket(fds[1]); + fds[0] = -1; + fds[1] = -1; + return false; } #undef CNTLNAME } + + return true; +} + +/** + * Make a pipe that can be used to flag that a signal has arrived. + * Using a pipe avoids race conditions, since it saves its values until emptied. + * Use GetSignalPipe() to get the pipe. + */ +void MakeSignalPipe(void) +{ + if (!MakeNonBlockingSocketPair(SIGNAL_PIPE)) + { + Log(LOG_LEVEL_CRIT, "Could not create internal communication pipe. Cannot continue."); + DoCleanupAndExit(EXIT_FAILURE); + } + + RegisterCleanupFunction(&CloseSignalPipe); } /** diff --git a/libpromises/signals.h b/libpromises/signals.h index 4ddbaf60266..03b73f74b78 100644 --- a/libpromises/signals.h +++ b/libpromises/signals.h @@ -36,6 +36,14 @@ void RequestReloadConfig(void); void MakeSignalPipe(void); int GetSignalPipe(void); + +/** + * Creates an AF_UNIX SOCK_STREAM socket pair with both ends set + * non-blocking, the way MakeSignalPipe() and cf-reactor's WakeupChannel + * both need. On failure, fds[0] and fds[1] are left at -1 and the cause is + * logged at LOG_LEVEL_ERR. + */ +bool MakeNonBlockingSocketPair(int fds[2]); void HandleSignalsForDaemon(int signum); void HandleSignalsForAgent(int signum); diff --git a/tests/unit/Makefile.am b/tests/unit/Makefile.am index 696c34e29c2..6acafb2e100 100644 --- a/tests/unit/Makefile.am +++ b/tests/unit/Makefile.am @@ -37,6 +37,7 @@ AM_CPPFLAGS = $(CORE_CPPFLAGS) \ -I$(srcdir)/../../cf-execd \ -I$(srcdir)/../../cf-key \ -I$(srcdir)/../../cf-check \ + -I$(srcdir)/../../cf-reactor \ -DTESTDATADIR='"$(srcdir)/data"' LDADD = ../../libpromises/libpromises.la libtest.la @@ -154,7 +155,8 @@ check_PROGRAMS = \ split_process_line_test \ new_packages_promise_test \ iteration_test \ - protocol_recv_overflow_test + protocol_recv_overflow_test \ + watcher_test if HAVE_AVAHI_CLIENT if HAVE_AVAHI_COMMON @@ -225,6 +227,10 @@ set_domainname_test_LDADD = libtest.la ../../libpromises/libpromises.la regex_test_SOURCES = regex_test.c ../../libpromises/match_scope.c +watcher_test_SOURCES = watcher_test.c \ + ../../cf-reactor/watcher.c \ + ../../cf-reactor/wakeup_channel.c + protocol_test_SOURCES = protocol_test.c \ ../../cf-serverd/server_common.c \ ../../cf-serverd/server_tls.c \ diff --git a/tests/unit/watcher_test.c b/tests/unit/watcher_test.c new file mode 100644 index 00000000000..9ef9d097cbb --- /dev/null +++ b/tests/unit/watcher_test.c @@ -0,0 +1,115 @@ +#include + +#include +#include +#include /* HandleSignalsForDaemon() */ +#include /* Bundle */ +#include /* LoggingPrivContext, LoggingPrivSetContext() */ + +#include +#include /* strstr() */ + +static int captured_err_count = 0; +static char captured_err_message[256]; + +static char *CaptureErrorLogHook(ARG_UNUSED LoggingPrivContext *pctx, LogLevel level, const char *message) +{ + if (level == LOG_LEVEL_ERR) + { + captured_err_count++; + strlcpy(captured_err_message, message, sizeof(captured_err_message)); + } + return (char *) message; +} + +static void test_registry_initialize_finalize(void) +{ + WatcherRegistryInitialize(); + WatcherRegistryFinalize(); +} + +static void test_watcher_register_single(void) +{ + WatcherRegistryInitialize(); + + /* WatcherRegister()/WatcherDestroy() never dereference the bundle + * pointer, they only store it in a map, so a fake non-NULL pointer is + * fine here. */ + Bundle *fake_bundle = (Bundle *) 0x1; + + WatcherRegister("test-event", EVENT_FILE_DELETED, NULL, fake_bundle, 5); + + WatcherRegistryFinalize(); +} + +static void test_watcher_register_duplicate_key_ignored(void) +{ + WatcherRegistryInitialize(); + + Bundle *fake_bundle_a = (Bundle *) 0x1; + Bundle *fake_bundle_b = (Bundle *) 0x2; + + WatcherRegister("dup-event", EVENT_FILE_DELETED, NULL, fake_bundle_a, 5); + + captured_err_count = 0; + captured_err_message[0] = '\0'; + LoggingPrivContext log_ctx = { .log_hook = CaptureErrorLogHook }; + LoggingPrivSetContext(&log_ctx); + + /* Registering the same key again must be rejected: an error is logged + * (not silently swallowed) and the first registration is kept, not + * replaced. */ + WatcherRegister("dup-event", EVENT_FILE_DELETED, NULL, fake_bundle_b, 5); + + LoggingPrivSetContext(NULL); + + assert_int_equal(captured_err_count, 1); + assert_true(strstr(captured_err_message, "dup-event") != NULL); + assert_true(strstr(captured_err_message, "already registered") != NULL); + + WatcherRegistryFinalize(); +} + +static void test_event_watcher_lifecycle(void) +{ + WatcherRegistryInitialize(); + + /* No watchers are registered here, and WatcherRegister() doesn't wire up + * a check_callback for EVENT_FILE_DELETED yet (see the TODO in + * WatcherRegister()), so there is nothing safe for the watcher thread to + * poll yet. Force IsPendingTermination() to true up front so + * WatcherThreadMain() exits its loop immediately instead of polling. */ + HandleSignalsForDaemon(SIGTERM); + + int fd = -1; + assert_true(EventWatcherInitialize(&fd)); + assert_true(fd >= 0); + + fd_set readfds; + FD_ZERO(&readfds); + /* fd not set: must return early without touching the wakeup channel or + * the event queue. */ + EventWatcherHandleEvents(fd, &readfds); + + FD_SET(fd, &readfds); + /* fd set but nothing queued: must drain the channel (no-op) and find + * nothing to pop from the event queue. */ + EventWatcherHandleEvents(fd, &readfds); + + /* Also joins the watcher thread and finalizes the watcher registry. */ + EventWatcherFinalize(); +} + +int main() +{ + PRINT_TEST_BANNER(); + const UnitTest tests[] = + { + unit_test(test_registry_initialize_finalize), + unit_test(test_watcher_register_single), + unit_test(test_watcher_register_duplicate_key_ignored), + unit_test(test_event_watcher_lifecycle), + }; + + return run_tests(tests); +}