diff --git a/cf-agent/Makefile.am b/cf-agent/Makefile.am index 8e1a9815fa1..b97a814345b 100644 --- a/cf-agent/Makefile.am +++ b/cf-agent/Makefile.am @@ -77,6 +77,7 @@ libcf_agent_la_SOURCES = \ tokyo_check.c tokyo_check.h \ abstract_dir.c abstract_dir.h \ cf-agent.c \ + agent_operations.c agent_operations.h \ cf-agent-enterprise-stubs.c cf-agent-enterprise-stubs.h \ comparray.c comparray.h \ acl_posix.c acl_posix.h \ diff --git a/cf-agent/agent_operations.c b/cf-agent/agent_operations.c new file mode 100644 index 00000000000..cc44cdd03a3 --- /dev/null +++ b/cf-agent/agent_operations.c @@ -0,0 +1,764 @@ +/* + 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. +*/ + +/* ScheduleAgentOperations() and the promise actuator it uses: this is what + * cf-agent does with a single bundle (evaluate its promises in + * AGENT_TYPESEQUENCE order, keeping each one with KeepAgentPromise()). It is + * kept separate from cf-agent.c so that other components (e.g. cf-reactor, + * running the bundle named in an events promise's "then") can run a bundle + * the same way cf-agent does, without linking cf-agent's main(). */ + +#include +#include + +#include +#include /* ExpandPromise() */ +#include /* SpecialTypeBanner() */ +#include +#include +#include +#include /* IsRegexItemIn() */ +#include /* FullTextMatch() */ +#include +#include +#include +#include /* IsBuiltInPromiseType() */ +#include /* EvaluateCustomPromise() */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +extern int PR_KEPT; +extern int PR_REPAIRED; +extern int PR_NOTKEPT; + +int CFA_BACKGROUND = 0; /* GLOBAL_X */ +int CFA_BACKGROUND_LIMIT = 1; /* GLOBAL_P */ + +Item *PROCESSREFRESH = NULL; /* GLOBAL_P */ + +static const char *const AGENT_TYPESEQUENCE[] = +{ + "meta", + "vars", + "defaults", + "classes", /* Maelstrom order 2 */ + "users", + "files", + "packages", + "guest_environments", + "methods", + "processes", + "services", + "commands", + "storage", + "databases", + "reports", + NULL +}; + +static PromiseResult KeepAgentPromise(EvalContext *ctx, const Promise *pp, void *param); +static void NewTypeContext(TypeSequence type); +static void DeleteTypeContext(EvalContext *ctx, TypeSequence type); +static PromiseResult ParallelFindAndVerifyFilesPromises(EvalContext *ctx, const Promise *pp); +static void BannerStatusEnd(PromiseResult status, const char *type, char *name); +static void BannerStatusBegin(const char *type, char *name); +static PromiseResult DefaultVarPromise(EvalContext *ctx, const Promise *pp); +static int NoteBundleCompliance(const Bundle *bundle, int save_pr_kept, int save_pr_repaired, int save_pr_notkept, struct timespec start); + +/** + @brief + Wrapper around DefaultVarPromise to silence cast-function-type compiler warning in ScheduleAgentOperations + */ +static PromiseResult DefaultVarPromiseWrapper(EvalContext *ctx, const Promise *pp, void *param) { + UNUSED(param); + return DefaultVarPromise(ctx, pp); +} + +PromiseResult ScheduleAgentOperations(EvalContext *ctx, const Bundle *bp) +// NB - this function can be called recursively through "methods" +{ + if (EvalContextIsClassicOrder(ctx, bp)) + { + return ScheduleAgentOperationsNormalOrder(ctx, bp); + } + return ScheduleAgentOperationsTopDownOrder(ctx, bp); +} + +PromiseResult ScheduleAgentOperationsNormalOrder(EvalContext *ctx, const Bundle *bp) +{ + assert(bp != NULL); + + int save_pr_kept = PR_KEPT; + int save_pr_repaired = PR_REPAIRED; + int save_pr_notkept = PR_NOTKEPT; + struct timespec start = BeginMeasure(); + + if (PROCESSREFRESH == NULL || (PROCESSREFRESH && IsRegexItemIn(ctx, PROCESSREFRESH, bp->name))) + { + ClearProcessTable(); + } + + PromiseResult result = PROMISE_RESULT_SKIPPED; + + for (int pass = 1; pass < CF_DONEPASSES; pass++) + { + // Evaluate built-in (non-custom) promise types, according to type sequence (normal order): + for (TypeSequence type = 0; AGENT_TYPESEQUENCE[type] != NULL; type++) + { + const BundleSection *sp = BundleGetSection((Bundle *)bp, AGENT_TYPESEQUENCE[type]); + + if (!sp || SeqLength(sp->promises) == 0) + { + continue; + } + + NewTypeContext(type); + + SpecialTypeBanner(type, pass); + EvalContextStackPushBundleSectionFrame(ctx, sp); + + for (size_t ppi = 0; ppi < SeqLength(sp->promises); ppi++) + { + Promise *pp = SeqAt(sp->promises, ppi); + + EvalContextSetPass(ctx, pass); + + PromiseResult promise_result = ExpandPromise(ctx, pp, KeepAgentPromise, NULL); + result = PromiseResultUpdate(result, promise_result); + + if (EvalAborted(ctx) || BundleAbort(ctx)) + { + DeleteTypeContext(ctx, type); + EvalContextStackPopFrame(ctx); + NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); + return result; + } + } + + DeleteTypeContext(ctx, type); + EvalContextStackPopFrame(ctx); + + if (type == TYPE_SEQUENCE_CONTEXTS) + { + BundleResolve(ctx, bp); + BundleResolvePromiseType(ctx, bp, "defaults", DefaultVarPromiseWrapper); + } + } + + // Custom promises are evaluated at the end of an evaluation pass: + const size_t sections = SeqLength(bp->custom_sections); + for (size_t i = 0; i < sections; ++i) + { + BundleSection *section = SeqAt(bp->custom_sections, i); + + EvalContextStackPushBundleSectionFrame(ctx, section); + + const size_t promises = SeqLength(section->promises); + for (size_t ppi = 0; ppi < promises; ppi++) + { + Promise *pp = SeqAt(section->promises, ppi); + + EvalContextSetPass(ctx, pass); + + PromiseResult promise_result = ExpandPromise(ctx, pp, KeepAgentPromise, NULL); + result = PromiseResultUpdate(result, promise_result); + + if (EvalAborted(ctx) || BundleAbort(ctx)) + { + EvalContextStackPopFrame(ctx); + NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); + return result; + } + } + EvalContextStackPopFrame(ctx); + } + } + + NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); + return result; +} + +PromiseResult ScheduleAgentOperationsTopDownOrder(EvalContext *ctx, const Bundle *bp) +{ + assert(bp != NULL); + + int save_pr_kept = PR_KEPT; + int save_pr_repaired = PR_REPAIRED; + int save_pr_notkept = PR_NOTKEPT; + struct timespec start = BeginMeasure(); + + if (PROCESSREFRESH == NULL || (PROCESSREFRESH && IsRegexItemIn(ctx, PROCESSREFRESH, bp->name))) + { + ClearProcessTable(); + } + + PromiseResult result = PROMISE_RESULT_SKIPPED; + for (int pass = 1; pass < CF_DONEPASSES; pass++) + { + const char *last_promise_type = ""; + for (size_t ppi = 0; ppi < SeqLength(bp->all_promises); ppi++) + { + EvalContextSetPass(ctx, pass); + Promise *pp = SeqAt(bp->all_promises, ppi); + BundleSection *parent_section = pp->parent_section; + + if (!StringEqual(last_promise_type, parent_section->promise_type)) + { + SpecialTypeBannerFromString(parent_section->promise_type, pass); + } + last_promise_type = parent_section->promise_type; + + EvalContextStackPushBundleSectionFrame(ctx, parent_section); + + PromiseResult promise_result = ExpandPromise(ctx, pp, KeepAgentPromise, NULL); + result = PromiseResultUpdate(result, promise_result); + if (EvalAborted(ctx) || BundleAbort(ctx)) + { + EvalContextStackPopFrame(ctx); + NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); + return result; + } + EvalContextStackPopFrame(ctx); + } + } + + NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); + return result; +} + +/*********************************************************************/ + +static PromiseResult DefaultVarPromise(EvalContext *ctx, const Promise *pp) +{ + char *regex = PromiseGetConstraintAsRval(pp, "if_match_regex", RVAL_TYPE_SCALAR); + bool okay = true; + + + DataType value_type = CF_DATA_TYPE_NONE; + const void *value = NULL; + { + VarRef *ref = VarRefParseFromScope(pp->promiser, "this"); + value = EvalContextVariableGetPlaintext(ctx, ref, &value_type); + VarRefDestroy(ref); + } + + switch (value_type) + { + case CF_DATA_TYPE_STRING: + case CF_DATA_TYPE_INT: + case CF_DATA_TYPE_REAL: + if (regex && !FullTextMatch(ctx, regex, value)) + { + return PROMISE_RESULT_NOOP; + } + + if (regex == NULL) + { + return PROMISE_RESULT_NOOP; + } + break; + + case CF_DATA_TYPE_STRING_LIST: + case CF_DATA_TYPE_INT_LIST: + case CF_DATA_TYPE_REAL_LIST: + if (regex) + { + for (const Rlist *rp = value; rp != NULL; rp = rp->next) + { + if (FullTextMatch(ctx, regex, RlistScalarValue(rp))) + { + okay = false; + break; + } + } + + if (okay) + { + return PROMISE_RESULT_NOOP; + } + } + break; + + default: + break; + } + + { + VarRef *ref = VarRefParseFromBundle(pp->promiser, PromiseGetBundle(pp)); + EvalContextVariableRemove(ctx, ref); + VarRefDestroy(ref); + } + + return VerifyVarPromise(ctx, pp, NULL); +} + +static void LogVariableValue(const EvalContext *ctx, const Promise *pp) +{ + VarRef *ref = VarRefParseFromBundle(pp->promiser, PromiseGetBundle(pp)); + char *out = NULL; + + DataType type; + const void *var = EvalContextVariableGetPlaintext(ctx, ref, &type); + switch (type) + { + case CF_DATA_TYPE_INT: + case CF_DATA_TYPE_REAL: + case CF_DATA_TYPE_STRING: + out = xstrdup((char *) var); + break; + case CF_DATA_TYPE_INT_LIST: + case CF_DATA_TYPE_REAL_LIST: + case CF_DATA_TYPE_STRING_LIST: + { + size_t siz = CF_BUFSIZE; + size_t len = 0; + out = xcalloc(1, CF_BUFSIZE); + + for (Rlist *rp = (Rlist *) var; rp != NULL; rp = rp->next) + { + const char *s = (char *) rp->val.item; + + if (strlen(s) + len + 3 >= siz) // ", " + NULL + { + out = xrealloc(out, siz + CF_BUFSIZE); + siz += CF_BUFSIZE; + } + + if (len > 0) + { + len += strlcat(out, ", ", siz); + } + + len += strlcat(out, s, siz); + } + break; + } + case CF_DATA_TYPE_CONTAINER: + { + Writer *w = StringWriter(); + JsonWriteCompact(w, (JsonElement *) var); + out = StringWriterClose(w); + break; + } + default: + /* TODO is CF_DATA_TYPE_NONE acceptable? Today all meta variables + * are of this type. */ + /* UnexpectedError("Variable '%s' is of unknown type %d", */ + /* pp->promiser, type); */ + out = xstrdup("NONE"); + break; + } + + Log(LOG_LEVEL_DEBUG, "V: '%s' => '%s'", pp->promiser, out); + free(out); + VarRefDestroy(ref); +} + +static PromiseResult KeepAgentPromise(EvalContext *ctx, const Promise *pp, ARG_UNUSED void *param) +{ + assert(param == NULL); + assert(pp != NULL); + + BannerStatusBegin(PromiseGetPromiseType(pp), pp->promiser); + struct timespec start = BeginMeasure(); + PromiseResult result = PROMISE_RESULT_NOOP; + + if (strcmp("meta", PromiseGetPromiseType(pp)) == 0 || + strcmp("vars", PromiseGetPromiseType(pp)) == 0) + { + Log(LOG_LEVEL_VERBOSE, "V: Computing value of '%s'", pp->promiser); + + result = VerifyVarPromise(ctx, pp, NULL); + if (result != PROMISE_RESULT_FAIL) + { + if (LogGetGlobalLevel() >= LOG_LEVEL_DEBUG) + { + LogVariableValue(ctx, pp); + } + } + } + else if (strcmp("defaults", PromiseGetPromiseType(pp)) == 0) + { + result = DefaultVarPromise(ctx, pp); + } + else if (strcmp("classes", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyClassPromise(ctx, pp, NULL); + } + else if (strcmp("processes", PromiseGetPromiseType(pp)) == 0) + { + if (!LoadProcessTable()) + { + Log(LOG_LEVEL_ERR, "Unable to read the process table - cannot keep processes: type promises"); + return PROMISE_RESULT_FAIL; + } + result = VerifyProcessesPromise(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("storage", PromiseGetPromiseType(pp)) == 0) + { + result = FindAndVerifyStoragePromises(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("packages", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyPackagesPromise(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("users", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyUsersPromise(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + + else if (strcmp("files", PromiseGetPromiseType(pp)) == 0) + { + result = ParallelFindAndVerifyFilesPromises(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("commands", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyExecPromise(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("databases", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyDatabasePromises(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("methods", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyMethodsPromise(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("services", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyServicesPromise(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("guest_environments", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyEnvironmentsPromise(ctx, pp); + if (result != PROMISE_RESULT_SKIPPED) + { + EndMeasurePromise(start, pp); + } + } + else if (strcmp("reports", PromiseGetPromiseType(pp)) == 0) + { + result = VerifyReportPromise(ctx, pp); + } + else if (!IsBuiltInPromiseType(PromiseGetPromiseType(pp))) + { + result = EvaluateCustomPromise(ctx, pp); + } + else + { + result = PROMISE_RESULT_NOOP; + } + + BannerStatusEnd(result, PromiseGetPromiseType(pp), pp->promiser); + EvalContextLogPromiseIterationOutcome(ctx, pp, result); + return result; +} + +static void BannerStatusBegin(const char *type, char *name) +{ + if (StringEqual(type, "vars") || StringEqual(type, "classes")) + { + return; + } + Log(LOG_LEVEL_VERBOSE, "P: BEGIN %s promise (%.30s%s)", + type, name, + (strlen(name) > 30) ? "..." : ""); +} + +static void BannerStatusEnd(PromiseResult status, const char *type, char *name) +{ + if ((strcmp(type, "vars") == 0) || (strcmp(type, "classes") == 0)) + { + return; + } + + switch (status) + { + case PROMISE_RESULT_CHANGE: + Log(LOG_LEVEL_VERBOSE, "A: Promise REPAIRED"); + break; + + case PROMISE_RESULT_TIMEOUT: + Log(LOG_LEVEL_VERBOSE, "A: Promise TIMED-OUT"); + break; + + case PROMISE_RESULT_WARN: + case PROMISE_RESULT_FAIL: + case PROMISE_RESULT_INTERRUPTED: + Log(LOG_LEVEL_VERBOSE, "A: Promise NOT KEPT!"); + break; + + case PROMISE_RESULT_DENIED: + Log(LOG_LEVEL_VERBOSE, "A: Promise NOT KEPT - denied"); + break; + + case PROMISE_RESULT_NOOP: + Log(LOG_LEVEL_VERBOSE, "A: Promise was KEPT"); + break; + default: + return; + break; + } + + Log(LOG_LEVEL_VERBOSE, "P: END %s promise (%.30s%s)", + type, name, + (strlen(name) > 30) ? "..." : ""); +} + +/*********************************************************************/ +/* Type context */ +/*********************************************************************/ + +static void NewTypeContext(TypeSequence type) +{ +// get maxconnections + + switch (type) + { + case TYPE_SEQUENCE_ENVIRONMENTS: + NewEnvironmentsContext(); + break; + + case TYPE_SEQUENCE_FILES: + break; + + case TYPE_SEQUENCE_PROCESSES: + break; + + case TYPE_SEQUENCE_STORAGE: +#ifndef __MINGW32__ // TODO: Run if implemented on Windows + if (SeqLength(GetGlobalMountedFSList())) + { + DeleteMountInfo(GetGlobalMountedFSList()); + SeqClear(GetGlobalMountedFSList()); + } +#endif /* !__MINGW32__ */ + break; + + default: + break; + } + + return; +} + +/*********************************************************************/ + +static void DeleteTypeContext(EvalContext *ctx, TypeSequence type) +{ + switch (type) + { + case TYPE_SEQUENCE_ENVIRONMENTS: + DeleteEnvironmentsContext(); + break; + + case TYPE_SEQUENCE_FILES: + break; + + case TYPE_SEQUENCE_PROCESSES: + break; + + case TYPE_SEQUENCE_STORAGE: + DeleteStorageContext(); + break; + + case TYPE_SEQUENCE_PACKAGES: + ExecuteScheduledPackages(ctx); + CleanScheduledPackages(); + break; + + default: + break; + } +} + +/**************************************************************/ +/* Thread context */ +/**************************************************************/ + +#ifdef __MINGW32__ + +static PromiseResult ParallelFindAndVerifyFilesPromises(EvalContext *ctx, const Promise *pp) +{ + int background = PromiseGetConstraintAsBoolean(ctx, "background", pp); + + if (background) + { + Log(LOG_LEVEL_VERBOSE, "Background processing of files promises is not supported on Windows"); + } + + return FindAndVerifyFilesPromises(ctx, pp); +} + +#else /* !__MINGW32__ */ + +static PromiseResult ParallelFindAndVerifyFilesPromises(EvalContext *ctx, const Promise *pp) +{ + int background = PromiseGetConstraintAsBoolean(ctx, "background", pp); + pid_t child = 1; + PromiseResult result = PROMISE_RESULT_SKIPPED; + + if (background) + { + if (CFA_BACKGROUND < CFA_BACKGROUND_LIMIT) + { + CFA_BACKGROUND++; + Log(LOG_LEVEL_VERBOSE, "Spawning new process..."); + child = fork(); + + if (child == 0) + { + ALARM_PID = -1; + + result = PromiseResultUpdate(result, FindAndVerifyFilesPromises(ctx, pp)); + + Log(LOG_LEVEL_VERBOSE, "Exiting backgrounded promise"); + PromiseRef(LOG_LEVEL_VERBOSE, pp); + _exit(EXIT_SUCCESS); + // TODO: need to solve this + } + } + else + { + Log(LOG_LEVEL_VERBOSE, "Promised parallel execution promised but exceeded the max number of promised background tasks, so serializing"); + background = 0; + } + } + else + { + result = PromiseResultUpdate(result, FindAndVerifyFilesPromises(ctx, pp)); + } + + return result; +} + +#endif /* !__MINGW32__ */ + +/*********************************************************************/ +/* Compliance comp */ +/*********************************************************************/ + +static int NoteBundleCompliance(const Bundle *bundle, int save_pr_kept, int save_pr_repaired, int save_pr_notkept, struct timespec start) +{ + double delta_pr_kept, delta_pr_repaired, delta_pr_notkept; + double bundle_compliance = 0.0; + + delta_pr_kept = (double) (PR_KEPT - save_pr_kept); + delta_pr_notkept = (double) (PR_NOTKEPT - save_pr_notkept); + delta_pr_repaired = (double) (PR_REPAIRED - save_pr_repaired); + + Log(LOG_LEVEL_VERBOSE, "A: ..................................................."); + Log(LOG_LEVEL_VERBOSE, "A: Bundle Accounting Summary for '%s' in namespace %s", bundle->name, bundle->ns); + + if (delta_pr_kept + delta_pr_notkept + delta_pr_repaired <= 0) + { + Log(LOG_LEVEL_VERBOSE, "A: Zero promises executed for bundle '%s'", bundle->name); + Log(LOG_LEVEL_VERBOSE, "A: ..................................................."); + return PROMISE_RESULT_NOOP; + } + else + { + Log(LOG_LEVEL_VERBOSE, "A: Promises kept in '%s' = %.0lf", bundle->name, delta_pr_kept); + Log(LOG_LEVEL_VERBOSE, "A: Promises not kept in '%s' = %.0lf", bundle->name, delta_pr_notkept); + Log(LOG_LEVEL_VERBOSE, "A: Promises repaired in '%s' = %.0lf", bundle->name, delta_pr_repaired); + + bundle_compliance = (delta_pr_kept + delta_pr_repaired) / (delta_pr_kept + delta_pr_notkept + delta_pr_repaired); + + Log(LOG_LEVEL_VERBOSE, "A: Aggregate compliance (promises kept/repaired) for bundle '%s' = %.1lf%%", + bundle->name, bundle_compliance * 100.0); + + if (LogGetGlobalLevel() >= LOG_LEVEL_INFO) + { + char name[CF_MAXVARSIZE]; + snprintf(name, CF_MAXVARSIZE, "%s:%s", bundle->ns, bundle->name); + EndMeasure(name, start); + } + else + { + EndMeasure(NULL, start); + } + Log(LOG_LEVEL_VERBOSE, "A: ..................................................."); + } + + // return the worst case for the bundle status + + if (delta_pr_notkept > 0) + { + return PROMISE_RESULT_FAIL; + } + + if (delta_pr_repaired > 0) + { + return PROMISE_RESULT_CHANGE; + } + + return PROMISE_RESULT_NOOP; +} diff --git a/cf-agent/agent_operations.h b/cf-agent/agent_operations.h new file mode 100644 index 00000000000..e5dc6f1445a --- /dev/null +++ b/cf-agent/agent_operations.h @@ -0,0 +1,38 @@ +/* + 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_AGENT_OPERATIONS_H +#define CFENGINE_AGENT_OPERATIONS_H + +#include + +/* ScheduleAgentOperations() itself is declared in prototypes3.h. */ + +extern int CFA_BACKGROUND; /* GLOBAL_X */ +extern int CFA_BACKGROUND_LIMIT; /* GLOBAL_P, body agent control: max_children */ +extern Item *PROCESSREFRESH; /* GLOBAL_P, body agent control: refresh_processes */ + +PromiseResult ScheduleAgentOperations(EvalContext *ctx, const Bundle *bp); + +#endif diff --git a/cf-agent/cf-agent.c b/cf-agent/cf-agent.c index 4ffe508e72f..4a3a9612f44 100644 --- a/cf-agent/cf-agent.c +++ b/cf-agent/cf-agent.c @@ -29,6 +29,7 @@ #include #include +#include #include #include #include @@ -104,10 +105,6 @@ #include -extern int PR_KEPT; -extern int PR_REPAIRED; -extern int PR_NOTKEPT; - static bool ALLCLASSESREPORT = false; /* GLOBAL_P */ static bool ALWAYS_VALIDATE = false; /* GLOBAL_P */ static bool CFPARANOID = false; /* GLOBAL_P */ @@ -115,31 +112,6 @@ static bool PERFORM_DB_CHECK = false; static const Rlist *ACCESSLIST = NULL; /* GLOBAL_P */ -static int CFA_BACKGROUND = 0; /* GLOBAL_X */ -static int CFA_BACKGROUND_LIMIT = 1; /* GLOBAL_P */ - -static Item *PROCESSREFRESH = NULL; /* GLOBAL_P */ - -static const char *const AGENT_TYPESEQUENCE[] = -{ - "meta", - "vars", - "defaults", - "classes", /* Maelstrom order 2 */ - "users", - "files", - "packages", - "guest_environments", - "methods", - "processes", - "services", - "commands", - "storage", - "databases", - "reports", - NULL -}; - /*******************************************************************/ /* Agent specific variables */ /*******************************************************************/ @@ -151,20 +123,12 @@ static char **TranslateOldBootstrapOptionsConcatenated(int argc, char **argv); static void FreeFixedStringArray(int size, char **array); static void CheckAgentAccess(const Rlist *list, const Policy *policy); static void KeepControlPromises(EvalContext *ctx, const Policy *policy, GenericAgentConfig *config); -static PromiseResult KeepAgentPromise(EvalContext *ctx, const Promise *pp, void *param); -static void NewTypeContext(TypeSequence type); -static void DeleteTypeContext(EvalContext *ctx, TypeSequence type); -static PromiseResult ParallelFindAndVerifyFilesPromises(EvalContext *ctx, const Promise *pp); static bool VerifyBootstrap(bool skip_cf_execd_check); static void KeepPromiseBundles(EvalContext *ctx, const Policy *policy, GenericAgentConfig *config); static void KeepPromises(EvalContext *ctx, const Policy *policy, GenericAgentConfig *config); -static int NoteBundleCompliance(const Bundle *bundle, int save_pr_kept, int save_pr_repaired, int save_pr_notkept, struct timespec start); static void AllClassesReport(const EvalContext *ctx); static bool HasAvahiSupport(void); static int AutomaticBootstrap(GenericAgentConfig *config); -static void BannerStatusEnd(PromiseResult status, const char *type, char *name); -static void BannerStatusBegin(const char *type, char *name); -static PromiseResult DefaultVarPromise(EvalContext *ctx, const Promise *pp); static void WaitForBackgroundProcesses(); /*******************************************************************/ @@ -260,17 +224,6 @@ static const char *const HINTS[] = NULL }; -/** - @brief - Wrapper around DefaultVarPromise to silence cast-function-type compiler warning in ScheduleAgentOperations - */ -static PromiseResult DefaultVarPromiseWrapper(EvalContext *ctx, const Promise *pp, void *param) { - UNUSED(param); - return DefaultVarPromise(ctx, pp); -} - -/*******************************************************************/ - int main(int argc, char *argv[]) { SetupSignalsForAgent(); @@ -1599,160 +1552,6 @@ static void AllClassesReport(const EvalContext *ctx) } } -PromiseResult ScheduleAgentOperations(EvalContext *ctx, const Bundle *bp) -// NB - this function can be called recursively through "methods" -{ - if (EvalContextIsClassicOrder(ctx, bp)) - { - return ScheduleAgentOperationsNormalOrder(ctx, bp); - } - return ScheduleAgentOperationsTopDownOrder(ctx, bp); -} - -PromiseResult ScheduleAgentOperationsNormalOrder(EvalContext *ctx, const Bundle *bp) -{ - assert(bp != NULL); - - int save_pr_kept = PR_KEPT; - int save_pr_repaired = PR_REPAIRED; - int save_pr_notkept = PR_NOTKEPT; - struct timespec start = BeginMeasure(); - - if (PROCESSREFRESH == NULL || (PROCESSREFRESH && IsRegexItemIn(ctx, PROCESSREFRESH, bp->name))) - { - ClearProcessTable(); - } - - PromiseResult result = PROMISE_RESULT_SKIPPED; - - for (int pass = 1; pass < CF_DONEPASSES; pass++) - { - // Evaluate built-in (non-custom) promise types, according to type sequence (normal order): - for (TypeSequence type = 0; AGENT_TYPESEQUENCE[type] != NULL; type++) - { - const BundleSection *sp = BundleGetSection((Bundle *)bp, AGENT_TYPESEQUENCE[type]); - - if (!sp || SeqLength(sp->promises) == 0) - { - continue; - } - - NewTypeContext(type); - - SpecialTypeBanner(type, pass); - EvalContextStackPushBundleSectionFrame(ctx, sp); - - for (size_t ppi = 0; ppi < SeqLength(sp->promises); ppi++) - { - Promise *pp = SeqAt(sp->promises, ppi); - - EvalContextSetPass(ctx, pass); - - PromiseResult promise_result = ExpandPromise(ctx, pp, KeepAgentPromise, NULL); - result = PromiseResultUpdate(result, promise_result); - - if (EvalAborted(ctx) || BundleAbort(ctx)) - { - DeleteTypeContext(ctx, type); - EvalContextStackPopFrame(ctx); - NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); - return result; - } - } - - DeleteTypeContext(ctx, type); - EvalContextStackPopFrame(ctx); - - if (type == TYPE_SEQUENCE_CONTEXTS) - { - BundleResolve(ctx, bp); - BundleResolvePromiseType(ctx, bp, "defaults", DefaultVarPromiseWrapper); - } - } - - // Custom promises are evaluated at the end of an evaluation pass: - const size_t sections = SeqLength(bp->custom_sections); - for (size_t i = 0; i < sections; ++i) - { - BundleSection *section = SeqAt(bp->custom_sections, i); - - EvalContextStackPushBundleSectionFrame(ctx, section); - - const size_t promises = SeqLength(section->promises); - for (size_t ppi = 0; ppi < promises; ppi++) - { - Promise *pp = SeqAt(section->promises, ppi); - - EvalContextSetPass(ctx, pass); - - PromiseResult promise_result = ExpandPromise(ctx, pp, KeepAgentPromise, NULL); - result = PromiseResultUpdate(result, promise_result); - - if (EvalAborted(ctx) || BundleAbort(ctx)) - { - EvalContextStackPopFrame(ctx); - NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); - return result; - } - } - EvalContextStackPopFrame(ctx); - } - } - - NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); - return result; -} - -PromiseResult ScheduleAgentOperationsTopDownOrder(EvalContext *ctx, const Bundle *bp) -{ - assert(bp != NULL); - - int save_pr_kept = PR_KEPT; - int save_pr_repaired = PR_REPAIRED; - int save_pr_notkept = PR_NOTKEPT; - struct timespec start = BeginMeasure(); - - if (PROCESSREFRESH == NULL || (PROCESSREFRESH && IsRegexItemIn(ctx, PROCESSREFRESH, bp->name))) - { - ClearProcessTable(); - } - - PromiseResult result = PROMISE_RESULT_SKIPPED; - for (int pass = 1; pass < CF_DONEPASSES; pass++) - { - const char *last_promise_type = ""; - for (size_t ppi = 0; ppi < SeqLength(bp->all_promises); ppi++) - { - EvalContextSetPass(ctx, pass); - Promise *pp = SeqAt(bp->all_promises, ppi); - BundleSection *parent_section = pp->parent_section; - - if (!StringEqual(last_promise_type, parent_section->promise_type)) - { - SpecialTypeBannerFromString(parent_section->promise_type, pass); - } - last_promise_type = parent_section->promise_type; - - EvalContextStackPushBundleSectionFrame(ctx, parent_section); - - PromiseResult promise_result = ExpandPromise(ctx, pp, KeepAgentPromise, NULL); - result = PromiseResultUpdate(result, promise_result); - if (EvalAborted(ctx) || BundleAbort(ctx)) - { - EvalContextStackPopFrame(ctx); - NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); - return result; - } - EvalContextStackPopFrame(ctx); - } - } - - NoteBundleCompliance(bp, save_pr_kept, save_pr_repaired, save_pr_notkept, start); - return result; -} - -/*********************************************************************/ - #ifdef __MINGW32__ static void CheckAgentAccess(const Rlist *list, const Policy *policy) @@ -1821,449 +1620,6 @@ static void CheckAgentAccess(const Rlist *list, const Policy *policy) /*********************************************************************/ -static PromiseResult DefaultVarPromise(EvalContext *ctx, const Promise *pp) -{ - char *regex = PromiseGetConstraintAsRval(pp, "if_match_regex", RVAL_TYPE_SCALAR); - bool okay = true; - - - DataType value_type = CF_DATA_TYPE_NONE; - const void *value = NULL; - { - VarRef *ref = VarRefParseFromScope(pp->promiser, "this"); - value = EvalContextVariableGetPlaintext(ctx, ref, &value_type); - VarRefDestroy(ref); - } - - switch (value_type) - { - case CF_DATA_TYPE_STRING: - case CF_DATA_TYPE_INT: - case CF_DATA_TYPE_REAL: - if (regex && !FullTextMatch(ctx, regex, value)) - { - return PROMISE_RESULT_NOOP; - } - - if (regex == NULL) - { - return PROMISE_RESULT_NOOP; - } - break; - - case CF_DATA_TYPE_STRING_LIST: - case CF_DATA_TYPE_INT_LIST: - case CF_DATA_TYPE_REAL_LIST: - if (regex) - { - for (const Rlist *rp = value; rp != NULL; rp = rp->next) - { - if (FullTextMatch(ctx, regex, RlistScalarValue(rp))) - { - okay = false; - break; - } - } - - if (okay) - { - return PROMISE_RESULT_NOOP; - } - } - break; - - default: - break; - } - - { - VarRef *ref = VarRefParseFromBundle(pp->promiser, PromiseGetBundle(pp)); - EvalContextVariableRemove(ctx, ref); - VarRefDestroy(ref); - } - - return VerifyVarPromise(ctx, pp, NULL); -} - -static void LogVariableValue(const EvalContext *ctx, const Promise *pp) -{ - VarRef *ref = VarRefParseFromBundle(pp->promiser, PromiseGetBundle(pp)); - char *out = NULL; - - DataType type; - const void *var = EvalContextVariableGetPlaintext(ctx, ref, &type); - switch (type) - { - case CF_DATA_TYPE_INT: - case CF_DATA_TYPE_REAL: - case CF_DATA_TYPE_STRING: - out = xstrdup((char *) var); - break; - case CF_DATA_TYPE_INT_LIST: - case CF_DATA_TYPE_REAL_LIST: - case CF_DATA_TYPE_STRING_LIST: - { - size_t siz = CF_BUFSIZE; - size_t len = 0; - out = xcalloc(1, CF_BUFSIZE); - - for (Rlist *rp = (Rlist *) var; rp != NULL; rp = rp->next) - { - const char *s = (char *) rp->val.item; - - if (strlen(s) + len + 3 >= siz) // ", " + NULL - { - out = xrealloc(out, siz + CF_BUFSIZE); - siz += CF_BUFSIZE; - } - - if (len > 0) - { - len += strlcat(out, ", ", siz); - } - - len += strlcat(out, s, siz); - } - break; - } - case CF_DATA_TYPE_CONTAINER: - { - Writer *w = StringWriter(); - JsonWriteCompact(w, (JsonElement *) var); - out = StringWriterClose(w); - break; - } - default: - /* TODO is CF_DATA_TYPE_NONE acceptable? Today all meta variables - * are of this type. */ - /* UnexpectedError("Variable '%s' is of unknown type %d", */ - /* pp->promiser, type); */ - out = xstrdup("NONE"); - break; - } - - Log(LOG_LEVEL_DEBUG, "V: '%s' => '%s'", pp->promiser, out); - free(out); - VarRefDestroy(ref); -} - -static PromiseResult KeepAgentPromise(EvalContext *ctx, const Promise *pp, ARG_UNUSED void *param) -{ - assert(param == NULL); - assert(pp != NULL); - - BannerStatusBegin(PromiseGetPromiseType(pp), pp->promiser); - struct timespec start = BeginMeasure(); - PromiseResult result = PROMISE_RESULT_NOOP; - - if (strcmp("meta", PromiseGetPromiseType(pp)) == 0 || - strcmp("vars", PromiseGetPromiseType(pp)) == 0) - { - Log(LOG_LEVEL_VERBOSE, "V: Computing value of '%s'", pp->promiser); - - result = VerifyVarPromise(ctx, pp, NULL); - if (result != PROMISE_RESULT_FAIL) - { - if (LogGetGlobalLevel() >= LOG_LEVEL_DEBUG) - { - LogVariableValue(ctx, pp); - } - } - } - else if (strcmp("defaults", PromiseGetPromiseType(pp)) == 0) - { - result = DefaultVarPromise(ctx, pp); - } - else if (strcmp("classes", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyClassPromise(ctx, pp, NULL); - } - else if (strcmp("processes", PromiseGetPromiseType(pp)) == 0) - { - if (!LoadProcessTable()) - { - Log(LOG_LEVEL_ERR, "Unable to read the process table - cannot keep processes: type promises"); - return PROMISE_RESULT_FAIL; - } - result = VerifyProcessesPromise(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("storage", PromiseGetPromiseType(pp)) == 0) - { - result = FindAndVerifyStoragePromises(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("packages", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyPackagesPromise(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("users", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyUsersPromise(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - - else if (strcmp("files", PromiseGetPromiseType(pp)) == 0) - { - result = ParallelFindAndVerifyFilesPromises(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("commands", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyExecPromise(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("databases", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyDatabasePromises(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("methods", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyMethodsPromise(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("services", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyServicesPromise(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("guest_environments", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyEnvironmentsPromise(ctx, pp); - if (result != PROMISE_RESULT_SKIPPED) - { - EndMeasurePromise(start, pp); - } - } - else if (strcmp("reports", PromiseGetPromiseType(pp)) == 0) - { - result = VerifyReportPromise(ctx, pp); - } - else if (!IsBuiltInPromiseType(PromiseGetPromiseType(pp))) - { - result = EvaluateCustomPromise(ctx, pp); - } - else - { - result = PROMISE_RESULT_NOOP; - } - - BannerStatusEnd(result, PromiseGetPromiseType(pp), pp->promiser); - EvalContextLogPromiseIterationOutcome(ctx, pp, result); - return result; -} - -static void BannerStatusBegin(const char *type, char *name) -{ - if (StringEqual(type, "vars") || StringEqual(type, "classes")) - { - return; - } - Log(LOG_LEVEL_VERBOSE, "P: BEGIN %s promise (%.30s%s)", - type, name, - (strlen(name) > 30) ? "..." : ""); -} - -static void BannerStatusEnd(PromiseResult status, const char *type, char *name) -{ - if ((strcmp(type, "vars") == 0) || (strcmp(type, "classes") == 0)) - { - return; - } - - switch (status) - { - case PROMISE_RESULT_CHANGE: - Log(LOG_LEVEL_VERBOSE, "A: Promise REPAIRED"); - break; - - case PROMISE_RESULT_TIMEOUT: - Log(LOG_LEVEL_VERBOSE, "A: Promise TIMED-OUT"); - break; - - case PROMISE_RESULT_WARN: - case PROMISE_RESULT_FAIL: - case PROMISE_RESULT_INTERRUPTED: - Log(LOG_LEVEL_VERBOSE, "A: Promise NOT KEPT!"); - break; - - case PROMISE_RESULT_DENIED: - Log(LOG_LEVEL_VERBOSE, "A: Promise NOT KEPT - denied"); - break; - - case PROMISE_RESULT_NOOP: - Log(LOG_LEVEL_VERBOSE, "A: Promise was KEPT"); - break; - default: - return; - break; - } - - Log(LOG_LEVEL_VERBOSE, "P: END %s promise (%.30s%s)", - type, name, - (strlen(name) > 30) ? "..." : ""); -} - -/*********************************************************************/ -/* Type context */ -/*********************************************************************/ - -static void NewTypeContext(TypeSequence type) -{ -// get maxconnections - - switch (type) - { - case TYPE_SEQUENCE_ENVIRONMENTS: - NewEnvironmentsContext(); - break; - - case TYPE_SEQUENCE_FILES: - break; - - case TYPE_SEQUENCE_PROCESSES: - break; - - case TYPE_SEQUENCE_STORAGE: -#ifndef __MINGW32__ // TODO: Run if implemented on Windows - if (SeqLength(GetGlobalMountedFSList())) - { - DeleteMountInfo(GetGlobalMountedFSList()); - SeqClear(GetGlobalMountedFSList()); - } -#endif /* !__MINGW32__ */ - break; - - default: - break; - } - - return; -} - -/*********************************************************************/ - -static void DeleteTypeContext(EvalContext *ctx, TypeSequence type) -{ - switch (type) - { - case TYPE_SEQUENCE_ENVIRONMENTS: - DeleteEnvironmentsContext(); - break; - - case TYPE_SEQUENCE_FILES: - break; - - case TYPE_SEQUENCE_PROCESSES: - break; - - case TYPE_SEQUENCE_STORAGE: - DeleteStorageContext(); - break; - - case TYPE_SEQUENCE_PACKAGES: - ExecuteScheduledPackages(ctx); - CleanScheduledPackages(); - break; - - default: - break; - } -} - -/**************************************************************/ -/* Thread context */ -/**************************************************************/ - -#ifdef __MINGW32__ - -static PromiseResult ParallelFindAndVerifyFilesPromises(EvalContext *ctx, const Promise *pp) -{ - int background = PromiseGetConstraintAsBoolean(ctx, "background", pp); - - if (background) - { - Log(LOG_LEVEL_VERBOSE, "Background processing of files promises is not supported on Windows"); - } - - return FindAndVerifyFilesPromises(ctx, pp); -} - -#else /* !__MINGW32__ */ - -static PromiseResult ParallelFindAndVerifyFilesPromises(EvalContext *ctx, const Promise *pp) -{ - int background = PromiseGetConstraintAsBoolean(ctx, "background", pp); - pid_t child = 1; - PromiseResult result = PROMISE_RESULT_SKIPPED; - - if (background) - { - if (CFA_BACKGROUND < CFA_BACKGROUND_LIMIT) - { - CFA_BACKGROUND++; - Log(LOG_LEVEL_VERBOSE, "Spawning new process..."); - child = fork(); - - if (child == 0) - { - ALARM_PID = -1; - - result = PromiseResultUpdate(result, FindAndVerifyFilesPromises(ctx, pp)); - - Log(LOG_LEVEL_VERBOSE, "Exiting backgrounded promise"); - PromiseRef(LOG_LEVEL_VERBOSE, pp); - _exit(EXIT_SUCCESS); - // TODO: need to solve this - } - } - else - { - Log(LOG_LEVEL_VERBOSE, "Promised parallel execution promised but exceeded the max number of promised background tasks, so serializing"); - background = 0; - } - } - else - { - result = PromiseResultUpdate(result, FindAndVerifyFilesPromises(ctx, pp)); - } - - return result; -} - -#endif /* !__MINGW32__ */ - -/**************************************************************/ - static bool VerifyBootstrap(bool skip_cf_execd_check) { const char *policy_server = PolicyServerGet(); @@ -2302,67 +1658,6 @@ static bool VerifyBootstrap(bool skip_cf_execd_check) return true; } -/**************************************************************/ -/* Compliance comp */ -/**************************************************************/ - -static int NoteBundleCompliance(const Bundle *bundle, int save_pr_kept, int save_pr_repaired, int save_pr_notkept, struct timespec start) -{ - double delta_pr_kept, delta_pr_repaired, delta_pr_notkept; - double bundle_compliance = 0.0; - - delta_pr_kept = (double) (PR_KEPT - save_pr_kept); - delta_pr_notkept = (double) (PR_NOTKEPT - save_pr_notkept); - delta_pr_repaired = (double) (PR_REPAIRED - save_pr_repaired); - - Log(LOG_LEVEL_VERBOSE, "A: ..................................................."); - Log(LOG_LEVEL_VERBOSE, "A: Bundle Accounting Summary for '%s' in namespace %s", bundle->name, bundle->ns); - - if (delta_pr_kept + delta_pr_notkept + delta_pr_repaired <= 0) - { - Log(LOG_LEVEL_VERBOSE, "A: Zero promises executed for bundle '%s'", bundle->name); - Log(LOG_LEVEL_VERBOSE, "A: ..................................................."); - return PROMISE_RESULT_NOOP; - } - else - { - Log(LOG_LEVEL_VERBOSE, "A: Promises kept in '%s' = %.0lf", bundle->name, delta_pr_kept); - Log(LOG_LEVEL_VERBOSE, "A: Promises not kept in '%s' = %.0lf", bundle->name, delta_pr_notkept); - Log(LOG_LEVEL_VERBOSE, "A: Promises repaired in '%s' = %.0lf", bundle->name, delta_pr_repaired); - - bundle_compliance = (delta_pr_kept + delta_pr_repaired) / (delta_pr_kept + delta_pr_notkept + delta_pr_repaired); - - Log(LOG_LEVEL_VERBOSE, "A: Aggregate compliance (promises kept/repaired) for bundle '%s' = %.1lf%%", - bundle->name, bundle_compliance * 100.0); - - if (LogGetGlobalLevel() >= LOG_LEVEL_INFO) - { - char name[CF_MAXVARSIZE]; - snprintf(name, CF_MAXVARSIZE, "%s:%s", bundle->ns, bundle->name); - EndMeasure(name, start); - } - else - { - EndMeasure(NULL, start); - } - Log(LOG_LEVEL_VERBOSE, "A: ..................................................."); - } - - // return the worst case for the bundle status - - if (delta_pr_notkept > 0) - { - return PROMISE_RESULT_FAIL; - } - - if (delta_pr_repaired > 0) - { - return PROMISE_RESULT_CHANGE; - } - - return PROMISE_RESULT_NOOP; -} - #if defined(HAVE_AVAHI_CLIENT_CLIENT_H) && defined(HAVE_AVAHI_COMMON_ADDRESS_H) static bool HasAvahiSupport(void) diff --git a/cf-reactor/Makefile.am b/cf-reactor/Makefile.am index 443e919f112..659f842258b 100644 --- a/cf-reactor/Makefile.am +++ b/cf-reactor/Makefile.am @@ -39,11 +39,15 @@ 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 \ + file_watcher.c file_watcher.h if !BUILTIN_EXTENSIONS bin_PROGRAMS = cf-reactor - cf_reactor_LDADD = libcf-reactor.la + cf_reactor_LDADD = libcf-reactor.la ../cf-agent/libcf-agent.la cf_reactor_SOURCES = endif 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..04cda1e6abe 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(ctx, policy, &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/file_watcher.c b/cf-reactor/file_watcher.c new file mode 100644 index 00000000000..7417bcbf2a7 --- /dev/null +++ b/cf-reactor/file_watcher.c @@ -0,0 +1,88 @@ +/* + 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 + +// =========== code for EVENT_FILE_DELETED event type =========== + +typedef struct +{ + char *path; + bool existed_last_check; +} FileWatcherPayload; + +/* Only ENOENT/ENOTDIR mean the path genuinely doesn't exist. Any other + * stat() failure (e.g. EACCES, ESTALE) is a transient/permission error, not + * a deletion, so treat the file as still present rather than misreporting + * it as deleted. */ +static bool FileExists(const char *path) +{ + struct stat sb; + if (stat(path, &sb) == 0) + { + return true; + } + + if (errno == ENOENT || errno == ENOTDIR) + { + return false; + } + + Log(LOG_LEVEL_ERR, "Unable to stat '%s' while checking for file deletion: %s", path, GetErrorStr()); + return true; +} + +void *FileWatcherPayloadNew(const char *path) +{ + FileWatcherPayload *pl = (FileWatcherPayload *) xmalloc(sizeof(FileWatcherPayload)); + + pl->path = xstrdup(path); + pl->existed_last_check = FileExists(path); + + return (void *) pl; +} + +bool CheckFileExists(void *payload) +{ + FileWatcherPayload *fwp = payload; + + bool exists_now = FileExists(fwp->path); + bool deleted = fwp->existed_last_check && !exists_now; + + fwp->existed_last_check = exists_now; + return deleted; +} + +void DestroyFileWatcherPayload(void *payload) +{ + FileWatcherPayload *fwp = payload; + free(fwp->path); + free(fwp); +} + diff --git a/cf-reactor/file_watcher.h b/cf-reactor/file_watcher.h new file mode 100644 index 00000000000..5b3b5104d19 --- /dev/null +++ b/cf-reactor/file_watcher.h @@ -0,0 +1,35 @@ +/* + 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_FILE_WATCHER_H +#define CFENGINE_FILE_WATCHER_H + +#include + + +void *FileWatcherPayloadNew(const char *path); +bool CheckFileExists(void *payload); +void DestroyFileWatcherPayload(void *payload); + +#endif diff --git a/cf-reactor/reactor_context.c b/cf-reactor/reactor_context.c index 71f7b7bc3a1..71946dfdcf0 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(EvalContext *ctx, Policy *policy, 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(ctx, policy, 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..05a0ced5871 100644 --- a/cf-reactor/reactor_context.h +++ b/cf-reactor/reactor_context.h @@ -26,11 +26,13 @@ #define CFENGINE_REACTOR_CONTEXT_H #include +#include #include typedef enum { - REACTOR_FD_NOVA + REACTOR_FD_NOVA, + REACTOR_FD_WATCHER } ReactorFdType; /** @@ -52,9 +54,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(EvalContext *ctx, Policy *policy, 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..a49fe986139 --- /dev/null +++ b/cf-reactor/reactor_transform.c @@ -0,0 +1,145 @@ +/* + 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 +#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"); + + // 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); + WatcherRegister(key, EVENT_FILE_DELETED, FileWatcherPayloadNew(path), then_constraint->rval, 1); +} + +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) +{ + WatcherRegistryClear(); + + 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..b089a9b53cb --- /dev/null +++ b/cf-reactor/reactor_transform.h @@ -0,0 +1,42 @@ +/* + 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: + * keep its "meta", "vars" and "classes" promises, and register a watcher + * (see watcher.h) for each of its "events" promises. + * + * Safe to call again for every re-read of the policy: it first discards all + * previously registered watchers (see WatcherRegistryClear()), so the + * result always reflects only the given policy, not a stale prior one. + */ +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..e682870d095 --- /dev/null +++ b/cf-reactor/watcher.c @@ -0,0 +1,335 @@ +/* + 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 +#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() */ + +/* Guards `watchers` (and, for the same atomic-rebuild reason, its paired + * `event_to_bundle`) against the watcher thread (WatcherThreadMain()) + * iterating over `watchers` concurrently with the main thread re-registering + * watchers from a freshly (re-)read policy (WatcherRegister(), + * WatcherRegistryClear()). */ +static pthread_mutex_t watchers_mutex = PTHREAD_MUTEX_INITIALIZER; +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, free); +} + +void WatcherRegistryFinalize(void) +{ + SeqDestroy(watchers); + watchers = NULL; + MapDestroy(event_to_bundle); + event_to_bundle = NULL; +} + +void WatcherRegistryClear(void) +{ + assert(watchers != NULL && event_to_bundle != NULL); + + pthread_mutex_lock(&watchers_mutex); + + SeqDestroy(watchers); + watchers = SeqNew(4, WatcherDestroy); + + MapDestroy(event_to_bundle); + event_to_bundle = MapNew(StringHash_untyped, StringEqual_untyped, NULL, NULL); + + pthread_mutex_unlock(&watchers_mutex); +} + +static Rval *AllocateRval(Rval val) +{ + Rval *new = xmalloc(sizeof(Rval)); + + new->item = val.item; + new->type = val.type; + + return new; +} + +void WatcherRegister(const char *key, EventType type, void *state, Rval val, time_t interval) +{ + assert(key != NULL); + assert(watchers != NULL && event_to_bundle != NULL); + + WatcherCheckFn check_callback = NULL; + WatcherStateDestroyFn destroy_state = NULL; + switch (type) + { + case EVENT_FILE_DELETED: + check_callback = CheckFileExists; + destroy_state = DestroyFileWatcherPayload; + break; + // TODO: add more cases + + default: + ProgrammingError("Unknown reactor event type %d for watcher '%s'", (int) type, key); + } + + pthread_mutex_lock(&watchers_mutex); + + if (MapHasKey(event_to_bundle, key)) + { + Log(LOG_LEVEL_ERR, "Reactor watcher key '%s' is already registered, ignoring the duplicate", key); + pthread_mutex_unlock(&watchers_mutex); + 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, AllocateRval(val)); + + pthread_mutex_unlock(&watchers_mutex); +} + +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; + + pthread_mutex_lock(&watchers_mutex); + + 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) + { + ThreadedQueuePush(event_queue, SafeStringDuplicate(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); + } + + pthread_mutex_unlock(&watchers_mutex); + + 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, free); + + 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; +} + +static void RunBundleFromRval(EvalContext *ctx, Policy *policy, Rval val) +{ + const char *name = NULL; + const Rlist *args = NULL; + if (val.type == RVAL_TYPE_FNCALL) + { + name = RvalFnCallValue(val)->name; + args = RvalFnCallValue(val)->args; + } + else + { + name = RvalScalarValue(val); + args = NULL; + } + + EvalContextSetBundleArgs(ctx, args); + const Bundle *bp = EvalContextResolveBundleExpression(ctx, policy, name, "agent"); + if (bp == NULL) + { + bp = EvalContextResolveBundleExpression(ctx, policy, name, "common"); + } + if (bp == NULL) + { + Log(LOG_LEVEL_ERR, "Failed to resolve bundle from '%s'", name); + return; + } + BundleBanner(bp, args); + EvalContextStackPushBundleFrame(ctx, bp, args, false, NULL); + ScheduleAgentOperations(ctx, bp); + EvalContextStackPopFrame(ctx); + EndBundleBanner(bp); + if (EvalAborted(ctx)) + { + Log(LOG_LEVEL_ERR, "Eval aborted"); + } +} + +void EventWatcherHandleEvents(EvalContext *ctx, Policy *policy, 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; + + pthread_mutex_lock(&watchers_mutex); + Rval *bundle = MapGet(event_to_bundle, key); + pthread_mutex_unlock(&watchers_mutex); + + if (bundle == NULL) + { + Log(LOG_LEVEL_VERBOSE, "Reactor watcher '%s' fired but is no longer registered, ignoring", key); + } + else + { + Log(LOG_LEVEL_NOTICE, "Reactor watcher '%s' fired", key); + RunBundleFromRval(ctx, policy, *bundle); + } + + free(item); + } +} + +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..4afee23d4ca --- /dev/null +++ b/cf-reactor/watcher.h @@ -0,0 +1,61 @@ +/* + 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 Discard every currently registered watcher and start over. Call + * this before re-registering watchers from a freshly (re-)read policy. + */ +void WatcherRegistryClear(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, Rval val, time_t interval); +bool EventWatcherInitialize(int *fd); +void EventWatcherHandleEvents(EvalContext *ctx, Policy *policy, 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); +}