Skip to content

in_calyptia_fleet: Use symlinks to manage fleet config files #10755

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

in_calyptia_fleet: Use symlinks to manage fleet config files #10755

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2fbb73c
in_calyptia_fleet: Use symlinks to manage fleet config files
alecholmes Aug 19, 2025
d772de3
Keep windows
alecholmes Aug 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
234 changes: 173 additions & 61 deletions plugins/in_calyptia_fleet/in_calyptia_fleet.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,7 @@ static flb_sds_t time_fleet_config_filename(struct flb_in_calyptia_fleet_config
{
char s_last_modified[32];

snprintf(s_last_modified, sizeof(s_last_modified)-1, "%d", (int)t);
snprintf(s_last_modified, sizeof(s_last_modified), "%lld", (long long)t);
return fleet_config_filename(ctx, s_last_modified);
}

Expand Down Expand Up @@ -485,40 +485,6 @@ static void *do_reload(void *data)
return NULL;
}

static int test_config_is_valid(struct flb_in_calyptia_fleet_config *ctx,
flb_sds_t cfgpath)
{
return FLB_TRUE;
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

test_config_is_valid is a no-op in practice and according to @pwhelan we'll never be ably to fully validate up front. Removing this function.

struct flb_cf *conf;
int ret = FLB_FALSE;

if (cfgpath == NULL) {
return FLB_FALSE;
}

conf = flb_cf_create();
if (conf == NULL) {
flb_plg_debug(ctx->ins, "unable to create config during validation test: %s",
cfgpath);
goto config_init_error;
}

conf = flb_cf_create_from_file(conf, cfgpath);
if (conf == NULL) {
flb_plg_debug(ctx->ins,
"unable to create config from file during validation test: %s",
cfgpath);
goto cf_create_from_file_error;
}

ret = FLB_TRUE;

cf_create_from_file_error:
flb_cf_destroy(conf);
config_init_error:
return ret;
}

static int parse_config_name_timestamp(struct flb_in_calyptia_fleet_config *ctx,
const char *cfgpath,
long *config_timestamp)
Expand All @@ -537,12 +503,11 @@ static int parse_config_name_timestamp(struct flb_in_calyptia_fleet_config *ctx,
/* Prevent undefined references due to use of readlink */
#ifndef FLB_SYSTEM_WINDOWS
case FLB_TRUE:

len = readlink(cfgpath, realname, sizeof(realname));

if (len > sizeof(realname)) {
len = readlink(cfgpath, realname, sizeof(realname) - 1);
if (len < 0 || len >= (sizeof(realname) - 1)) {
return FLB_FALSE;
}
realname[len] = '\0';
break;
#endif /* FLB_SYSTEM_WINDOWS */
case FLB_FALSE:
Expand Down Expand Up @@ -600,6 +565,10 @@ static int execute_reload(struct flb_in_calyptia_fleet_config *ctx, flb_sds_t cf
}

reload = flb_calloc(1, sizeof(struct reload_ctx));
if (reload == NULL) {
flb_sds_destroy(cfgpath);
return FLB_FALSE;
}
reload->flb = flb;
reload->cfg_path = cfgpath;

Expand All @@ -614,6 +583,7 @@ static int execute_reload(struct flb_in_calyptia_fleet_config *ctx, flb_sds_t cf
flb_input_collector_resume(ctx->collect_fd, ctx->ins);
}

flb_free(reload);
flb_sds_destroy(cfgpath);
return FLB_FALSE;
}
Expand All @@ -624,17 +594,6 @@ static int execute_reload(struct flb_in_calyptia_fleet_config *ctx, flb_sds_t cf
*/
flb_plg_info(ctx->ins, "loading configuration from %s.", reload->cfg_path);

if (test_config_is_valid(ctx, reload->cfg_path) == FLB_FALSE) {
flb_plg_error(ctx->ins, "unable to load configuration.");

if (ctx->collect_fd > 0) {
flb_input_collector_resume(ctx->collect_fd, ctx->ins);
}

flb_sds_destroy(cfgpath);
return FLB_FALSE;
}

if (fleet_cur_chdir(ctx) == -1) {
flb_errno();
flb_plg_error(ctx->ins, "unable to change to configuration directory");
Expand Down Expand Up @@ -1025,7 +984,7 @@ static int check_timestamp_is_newer(struct flb_in_calyptia_fleet_config *ctx, ti
/* Check if existing file timestamp is greater than or equal to new timestamp */
if (file_timestamp >= new_timestamp) {
flb_plg_debug(ctx->ins,
"existing file with timestamp %ld >= new timestamp %ld",
"existing file with timestamp %lld >= new timestamp %lld",
(long long)file_timestamp, (long long)new_timestamp);
ret = FLB_FALSE;
break;
Expand Down Expand Up @@ -1311,8 +1270,6 @@ static struct cfl_array *read_glob(const char *path)
return read_glob_win(path, NULL);
}

#endif

static int cfl_array_qsort_conf_files(const void *arg_a, const void *arg_b)
{
struct cfl_variant *var_a = (struct cfl_variant *)*(void **)arg_a;
Expand Down Expand Up @@ -1341,7 +1298,7 @@ static int cfl_array_qsort_conf_files(const void *arg_a, const void *arg_b)
return strcmp(var_a->data.as_string, var_b->data.as_string);
}

static int calyptia_config_delete_old_dir(const char *cfgpath)
static int calyptia_config_delete_old_dir_win(const char *cfgpath)
{
flb_sds_t cfg_glob;
char *ext;
Expand Down Expand Up @@ -1395,7 +1352,7 @@ static int calyptia_config_delete_old_dir(const char *cfgpath)
return FLB_TRUE;
}

static int calyptia_config_delete_old(struct flb_in_calyptia_fleet_config *ctx)
static int calyptia_config_delete_old_win(struct flb_in_calyptia_fleet_config *ctx)
{
struct cfl_array *confs;
struct cfl_array *tconfs;
Expand Down Expand Up @@ -1452,7 +1409,7 @@ static int calyptia_config_delete_old(struct flb_in_calyptia_fleet_config *ctx)

for (idx = 0; idx < (((ssize_t)tconfs->entry_count) -3); idx++) {
unlink(tconfs->entries[idx]->data.as_string);
calyptia_config_delete_old_dir(tconfs->entries[idx]->data.as_string);
calyptia_config_delete_old_dir_win(tconfs->entries[idx]->data.as_string);
}

cfl_array_destroy(confs);
Expand All @@ -1462,7 +1419,7 @@ static int calyptia_config_delete_old(struct flb_in_calyptia_fleet_config *ctx)
return 0;
}

static flb_sds_t calyptia_config_get_newest(struct flb_in_calyptia_fleet_config *ctx)
static flb_sds_t calyptia_config_get_newest_win(struct flb_in_calyptia_fleet_config *ctx)
{
struct cfl_array *inis;
flb_sds_t glob_conf_files = NULL;
Expand Down Expand Up @@ -1519,8 +1476,162 @@ static flb_sds_t calyptia_config_get_newest(struct flb_in_calyptia_fleet_config
return cfgnewname;
}

#endif

#ifndef FLB_SYSTEM_WINDOWS

/**
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I deleted a couple of funcs and replaced them new ones. The diff doesn't display well since the old and new code is vaguely similar.

* Deletes the directory at path and all the files in it.
* FLB_TRUE is returned if successful, otherwise FLB_FALSE.
*/
static int delete_dir(const char *path)
{
flb_sds_t glob_pattern;
struct cfl_array *files;
struct stat st;
int idx;
int ret = FLB_TRUE;

if (path == NULL) {
return FLB_FALSE;
}
glob_pattern = flb_sds_create(path);
if (glob_pattern == NULL) {
return FLB_FALSE;
}
if (flb_sds_cat_safe(&glob_pattern, PATH_SEPARATOR "*", strlen(PATH_SEPARATOR "*")) != 0) {
flb_sds_destroy(glob_pattern);
return FLB_FALSE;
}
files = read_glob(glob_pattern);
flb_sds_destroy(glob_pattern);
if (files != NULL) {
for (idx = 0; idx < ((ssize_t)files->entry_count); idx++) {
if (stat(files->entries[idx]->data.as_string, &st) == 0) {
if (S_ISDIR(st.st_mode)) {
if (delete_dir(files->entries[idx]->data.as_string) != FLB_TRUE) {
ret = FLB_FALSE;
}
} else {
if (unlink(files->entries[idx]->data.as_string) != 0) {
ret = FLB_FALSE;
}
}
} else {
ret = FLB_FALSE;
}
}
cfl_array_destroy(files);
}
if (rmdir(path) != 0) {
ret = FLB_FALSE;
}
return ret;
}
/**
* Deletes config files and directories that are referenced (directly or
* indirectly) by the symlink at the value returned by the function pointed to
* by config_filename_func.
*
* Returns FLB_TRUE if successful, otherwise FLB_FALSE.
*/
static int calyptia_config_delete_by_symlink(struct flb_in_calyptia_fleet_config *ctx,
flb_sds_t (*config_filename_func)(struct flb_in_calyptia_fleet_config *))
{
struct cfl_array *confs;
flb_sds_t config_path;
char realname[CALYPTIA_MAX_DIR_SIZE] = {0};
ssize_t len;
char *ext;
int idx;
struct stat entry_stat;
const char *entry_path;

if (ctx == NULL) {
return FLB_FALSE;
}
/* Get the symlink path and extract prefix from its target */
config_path = config_filename_func(ctx);
if (config_path == NULL) {
flb_plg_error(ctx->ins, "error getting config path");
return FLB_FALSE;
}
/* Follow the symlink to get the target filename */
len = readlink(config_path, realname, sizeof(realname) - 1);
if (len < 0 || len >= (sizeof(realname) - 1)) {
flb_plg_error(ctx->ins, "error resolving symlink: %s", config_path);
flb_sds_destroy(config_path);
return FLB_FALSE;
}
realname[len] = '\0';

/* Replace the extension with a glob (e.g. "/a/b.yaml" -> "/a/b*") */
ext = strrchr(realname, '.');
if (ext == NULL) {
flb_plg_error(ctx->ins, "symlink target has no extension: %s", realname);
flb_sds_destroy(config_path);
return FLB_FALSE;
}
strcpy(ext, "*");

/* Delete all files and directories that match the prefix pattern */
confs = read_glob(realname);
if (confs == NULL) {
flb_plg_warn(ctx->ins, "config glob did not return any files: %s", realname);
flb_sds_destroy(config_path);
return FLB_FALSE;
}
for (idx = 0; idx < confs->entry_count; idx++) {
entry_path = confs->entries[idx]->data.as_string;
if (stat(entry_path, &entry_stat) == 0) {
if (S_ISDIR(entry_stat.st_mode)) {
flb_plg_info(ctx->ins, "deleting config directory: %s", entry_path);
if (delete_dir(entry_path) == FLB_FALSE) {
flb_plg_warn(ctx->ins, "unable to delete config directory: %s", entry_path);
}
} else {
flb_plg_info(ctx->ins, "deleting config file: %s", entry_path);
if (unlink(entry_path) != 0) {
flb_plg_warn(ctx->ins, "unable to delete config file: %s", entry_path);
}
}
}
}

flb_plg_info(ctx->ins, "deleting config symlink: %s", config_path);
if (unlink(config_path) != 0) {
flb_plg_warn(ctx->ins, "unable to delete config symlink: %s", config_path);
}

cfl_array_destroy(confs);
flb_sds_destroy(config_path);
return FLB_TRUE;
}

/* Wraps new_fleet_config_filename macro for use by calyptia_config_delete_by_symlink. */
static flb_sds_t new_fleet_config_filename_func(struct flb_in_calyptia_fleet_config *ctx)
{
return new_fleet_config_filename(ctx);
}

/* Wraps old_fleet_config_filename macro for use by calyptia_config_delete_by_symlink */
static flb_sds_t old_fleet_config_filename_func(struct flb_in_calyptia_fleet_config *ctx)
{
return old_fleet_config_filename(ctx);
}

/** Deletes config files and directories referenced by the new config symlink. */
static int calyptia_config_delete_new(struct flb_in_calyptia_fleet_config *ctx)
{
return calyptia_config_delete_by_symlink(ctx, new_fleet_config_filename_func);
}

/** Deletes config files and directories referenced by the old config symlink. */
static int calyptia_config_delete_old(struct flb_in_calyptia_fleet_config *ctx)
{
return calyptia_config_delete_by_symlink(ctx, old_fleet_config_filename_func);
}

static int calyptia_config_add(struct flb_in_calyptia_fleet_config *ctx,
const char *cfgname)
{
Expand Down Expand Up @@ -1592,6 +1703,7 @@ static int calyptia_config_commit(struct flb_in_calyptia_fleet_config *ctx)
}

if (exists_old_fleet_config(ctx) == FLB_TRUE) {
calyptia_config_delete_old(ctx);
unlink(cfgoldname);
}

Expand All @@ -1601,7 +1713,6 @@ static int calyptia_config_commit(struct flb_in_calyptia_fleet_config *ctx)
}
}

calyptia_config_delete_old(ctx);
rc = FLB_TRUE;

error:
Expand Down Expand Up @@ -1670,7 +1781,7 @@ static int calyptia_config_add(struct flb_in_calyptia_fleet_config *ctx,

static int calyptia_config_commit(struct flb_in_calyptia_fleet_config *ctx)
{
calyptia_config_delete_old(ctx);
calyptia_config_delete_old_win(ctx);
return FLB_TRUE;
}

Expand Down Expand Up @@ -2088,9 +2199,9 @@ static int load_fleet_config(struct flb_in_calyptia_fleet_config *ctx)
else if (exists_new_fleet_config(ctx) == FLB_TRUE) {
return execute_reload(ctx, new_fleet_config_filename(ctx));
}
#ifdef FLB_SYSTEM_WINDOWS
else {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The branches above will load either current or new config. No need to pick otherwise random config and a subsequent PR will refactor this function a bit anyway.

cfgnewname = calyptia_config_get_newest(ctx);

cfgnewname = calyptia_config_get_newest_win(ctx);
if (cfgnewname != NULL) {
flb_plg_debug(ctx->ins, "loading newest configuration: %s", cfgnewname);
return execute_reload(ctx, cfgnewname);
Expand All @@ -2099,6 +2210,7 @@ static int load_fleet_config(struct flb_in_calyptia_fleet_config *ctx)
flb_plg_warn(ctx->ins, "unable to find latest configuration file");
}
}
#endif
}
else {
flb_plg_debug(ctx->ins, "we are already using a configuration file: %s",
Expand Down
Loading