feat: add config_repl Phase 1 — HEAD-file storage, SHA-256, trim, stale check, sanitise_cn
- config_repl.h/.c: atomic versioned file save, pointer file, trim, stale check - test_config_repl.c: 17 Unity tests covering all Phase 1 storage functions - typemap.h: add repl/heartbeat types + union members for xce_request_t/xce_reply_t - clusterconfig.h: add config_generation field to cluster_config_t (capnp schema ref) - meson.build: add config_repl.c and test_config_repl.c to build targets - controller_test.c: register test_config_repl dispatch function
This commit is contained in:
parent
1c4a103a66
commit
491c2550c2
7 changed files with 926 additions and 7 deletions
368
src/controller/config_repl.c
Normal file
368
src/controller/config_repl.c
Normal file
|
|
@ -0,0 +1,368 @@
|
|||
#define _GNU_SOURCE
|
||||
#include "config_repl.h"
|
||||
|
||||
#include <dirent.h>
|
||||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <limits.h>
|
||||
#include <openssl/sha.h>
|
||||
#include <pthread.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
#include <sys/stat.h>
|
||||
#include <sys/types.h>
|
||||
#include <time.h>
|
||||
#include <utime.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include "byteorder.h"
|
||||
#include "clusterconfig.h"
|
||||
#include "utils.h"
|
||||
|
||||
/* ── internal helpers ──────────────────────────── */
|
||||
|
||||
static void make_conf_dir(const char *workdir, char *out, size_t sz)
|
||||
{
|
||||
snprintf(out, sz, "%s/conf", workdir);
|
||||
}
|
||||
|
||||
static void bytes_to_hex(const uint8_t *b, size_t n, char *out)
|
||||
{
|
||||
static const char hex[] = "0123456789abcdef";
|
||||
for (size_t i = 0; i < n; i++) {
|
||||
out[2 * i] = hex[(b[i] >> 4) & 0xf];
|
||||
out[2 * i + 1] = hex[b[i] & 0xf];
|
||||
}
|
||||
out[2 * n] = '\0';
|
||||
}
|
||||
|
||||
/* ── conf_read_current_filename ─────────────────── */
|
||||
|
||||
int conf_read_current_filename(const char *workdir, char *out, size_t outsz)
|
||||
{
|
||||
char head_path[4096];
|
||||
snprintf(head_path, sizeof(head_path), "%s/conf/current", workdir);
|
||||
|
||||
FILE *f = fopen(head_path, "r");
|
||||
if (!f)
|
||||
return -ENOENT;
|
||||
|
||||
struct stat st;
|
||||
if (fstat(fileno(f), &st) == 0 && st.st_size == 0) {
|
||||
fclose(f);
|
||||
ERROR("conf/current is empty — treating as missing config (pointer file"
|
||||
" may be from an interrupted write)");
|
||||
return -ENODATA;
|
||||
}
|
||||
|
||||
if (!fgets(out, (int)outsz, f)) {
|
||||
fclose(f);
|
||||
return -EIO;
|
||||
}
|
||||
fclose(f);
|
||||
out[strcspn(out, "\n")] = '\0';
|
||||
|
||||
if (strchr(out, '/')) {
|
||||
ERROR("conf/current contains path separator — corrupted pointer file");
|
||||
return -EINVAL;
|
||||
}
|
||||
if (strncmp(out, "cluster.bin.", 12) != 0) {
|
||||
ERROR("conf/current '%s' does not start with 'cluster.bin.'", out);
|
||||
return -EINVAL;
|
||||
}
|
||||
const char *hex = out + 12;
|
||||
if (strlen(hex) != 12 || strspn(hex, "0123456789abcdef") != 12) {
|
||||
ERROR("conf/current suffix '%s' is not 12 lowercase hex chars", hex);
|
||||
return -EINVAL;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ── conf_save_config ───────────────────────────── */
|
||||
|
||||
int conf_save_config(myconf_t *myconf, const uint8_t *bytes, size_t len,
|
||||
const uint8_t sha256[32])
|
||||
{
|
||||
char conf_dir[4096];
|
||||
make_conf_dir(myconf->workdir, conf_dir, sizeof(conf_dir));
|
||||
|
||||
if (mkdir(conf_dir, 0700) != 0 && errno != EEXIST)
|
||||
return -errno;
|
||||
|
||||
/* Derive 12-char hex tag from first 6 bytes of SHA-256 */
|
||||
char hex_tag[13];
|
||||
bytes_to_hex(sha256, 6, hex_tag);
|
||||
|
||||
/* Step 1: write versioned file via thread-unique .tmp */
|
||||
char tmp_path[4096], ver_path[4096];
|
||||
snprintf(tmp_path, sizeof(tmp_path), "%s/cluster.bin.%s.%lu.tmp",
|
||||
conf_dir, hex_tag, (unsigned long)pthread_self());
|
||||
snprintf(ver_path, sizeof(ver_path), "%s/cluster.bin.%s",
|
||||
conf_dir, hex_tag);
|
||||
|
||||
int fd = open(tmp_path, O_WRONLY | O_CREAT | O_TRUNC, 0600);
|
||||
if (fd < 0)
|
||||
return -errno;
|
||||
if (write(fd, bytes, len) != (ssize_t)len) {
|
||||
close(fd);
|
||||
unlink(tmp_path);
|
||||
return -EIO;
|
||||
}
|
||||
if (fsync(fd) != 0) {
|
||||
close(fd);
|
||||
unlink(tmp_path);
|
||||
return -EIO;
|
||||
}
|
||||
close(fd);
|
||||
if (rename(tmp_path, ver_path) != 0) {
|
||||
unlink(tmp_path);
|
||||
return -errno;
|
||||
}
|
||||
|
||||
/* Step 2: update HEAD pointer via thread-unique .tmp */
|
||||
char ptr_tmp[4096], ptr_path[4096];
|
||||
snprintf(ptr_tmp, sizeof(ptr_tmp), "%s/current.%lu.tmp",
|
||||
conf_dir, (unsigned long)pthread_self());
|
||||
snprintf(ptr_path, sizeof(ptr_path), "%s/current", conf_dir);
|
||||
|
||||
char target[NAME_MAX];
|
||||
snprintf(target, sizeof(target), "cluster.bin.%s\n", hex_tag);
|
||||
|
||||
fd = open(ptr_tmp, O_WRONLY | O_CREAT | O_TRUNC, 0600);
|
||||
if (fd < 0)
|
||||
return -errno;
|
||||
if (write(fd, target, strlen(target)) != (ssize_t)strlen(target)) {
|
||||
close(fd);
|
||||
unlink(ptr_tmp);
|
||||
return -EIO;
|
||||
}
|
||||
fsync(fd);
|
||||
close(fd);
|
||||
if (rename(ptr_tmp, ptr_path) != 0) {
|
||||
unlink(ptr_tmp);
|
||||
return -errno;
|
||||
}
|
||||
|
||||
/* Step 2.5: fsync conf/ directory to persist the rename(2)s */
|
||||
fd = open(conf_dir, O_RDONLY);
|
||||
if (fd >= 0) {
|
||||
fsync(fd);
|
||||
close(fd);
|
||||
}
|
||||
|
||||
/* Step 3: trim old versions */
|
||||
conf_trim_old_versions(myconf->workdir, myconf->conf_keep_n);
|
||||
|
||||
/* Update conffile to the new versioned path (heap-allocated) */
|
||||
free(myconf->conffile);
|
||||
if (asprintf(&myconf->conffile, "%s/cluster.bin.%s", conf_dir, hex_tag) < 0)
|
||||
return -ENOMEM;
|
||||
|
||||
/* Update tag (first 8 bytes of sha256, host order) and full hash */
|
||||
uint64_t new_tag;
|
||||
memcpy(&new_tag, sha256, 8);
|
||||
myconf->config_version_tag = be64toh(new_tag);
|
||||
memcpy(myconf->config_sha256, sha256, 32);
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ── conf_read_raw ──────────────────────────────── */
|
||||
|
||||
int conf_read_raw(const char *conffile, uint8_t **bytes, size_t *len)
|
||||
{
|
||||
FILE *f = fopen(conffile, "rb");
|
||||
if (!f)
|
||||
return -errno;
|
||||
|
||||
if (fseek(f, 0, SEEK_END) != 0) {
|
||||
fclose(f);
|
||||
return -EIO;
|
||||
}
|
||||
long sz = ftell(f);
|
||||
if (sz < 0) {
|
||||
fclose(f);
|
||||
return -EIO;
|
||||
}
|
||||
fseek(f, 0, SEEK_SET);
|
||||
|
||||
if (sz == 0) {
|
||||
fclose(f);
|
||||
ERROR("Config file '%s' is empty — not a valid ClusterConfig blob",
|
||||
conffile);
|
||||
return -EIO;
|
||||
}
|
||||
if ((size_t)sz > CONF_MAX_SIZE) {
|
||||
fclose(f);
|
||||
ERROR("Config file '%s' exceeds CONF_MAX_SIZE (%zu bytes)",
|
||||
conffile, (size_t)sz);
|
||||
return -EFBIG;
|
||||
}
|
||||
|
||||
*bytes = (uint8_t *)malloc((size_t)sz);
|
||||
if (!*bytes) {
|
||||
fclose(f);
|
||||
return -ENOMEM;
|
||||
}
|
||||
|
||||
if (fread(*bytes, 1, (size_t)sz, f) != (size_t)sz) {
|
||||
free(*bytes);
|
||||
*bytes = NULL;
|
||||
fclose(f);
|
||||
return -EIO;
|
||||
}
|
||||
fclose(f);
|
||||
*len = (size_t)sz;
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ── conf_get_saved_tag ─────────────────────────── */
|
||||
|
||||
uint64_t conf_get_saved_tag(const char *workdir)
|
||||
{
|
||||
char fname[256] = {0};
|
||||
if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
|
||||
return 0;
|
||||
|
||||
char ver_path[4096];
|
||||
snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", workdir, fname);
|
||||
|
||||
uint8_t *bytes = NULL;
|
||||
size_t blen = 0;
|
||||
if (conf_read_raw(ver_path, &bytes, &blen) != 0)
|
||||
return 0;
|
||||
|
||||
uint8_t sha256[32];
|
||||
SHA256(bytes, blen, sha256);
|
||||
free(bytes);
|
||||
|
||||
uint64_t tag;
|
||||
memcpy(&tag, sha256, 8);
|
||||
return be64toh(tag);
|
||||
}
|
||||
|
||||
/* ── conf_check_stale ───────────────────────────── */
|
||||
|
||||
int conf_check_stale(const char *workdir, time_t stale_timeout_s)
|
||||
{
|
||||
if (stale_timeout_s == 0)
|
||||
return 0;
|
||||
|
||||
char fname[256] = {0};
|
||||
if (conf_read_current_filename(workdir, fname, sizeof(fname)) != 0)
|
||||
return 0;
|
||||
|
||||
char ver_path[4096];
|
||||
snprintf(ver_path, sizeof(ver_path), "%s/conf/%s", workdir, fname);
|
||||
|
||||
struct stat st;
|
||||
if (stat(ver_path, &st) != 0)
|
||||
return 0;
|
||||
|
||||
time_t age = time(NULL) - st.st_mtime;
|
||||
return (age > stale_timeout_s) ? 1 : 0;
|
||||
}
|
||||
|
||||
/* ── conf_trim_old_versions ─────────────────────── */
|
||||
|
||||
typedef struct {
|
||||
char name[NAME_MAX];
|
||||
time_t mtime;
|
||||
} ver_entry_t;
|
||||
|
||||
static int ver_entry_cmp_desc(const void *a, const void *b)
|
||||
{
|
||||
const ver_entry_t *ea = (const ver_entry_t *)a;
|
||||
const ver_entry_t *eb = (const ver_entry_t *)b;
|
||||
if (eb->mtime > ea->mtime) return 1;
|
||||
if (eb->mtime < ea->mtime) return -1;
|
||||
return 0;
|
||||
}
|
||||
|
||||
void conf_trim_old_versions(const char *workdir, int keep_n)
|
||||
{
|
||||
char conf_dir[4096];
|
||||
make_conf_dir(workdir, conf_dir, sizeof(conf_dir));
|
||||
|
||||
/* Read current pointer to protect it */
|
||||
char current_name[256] = {0};
|
||||
conf_read_current_filename(workdir, current_name, sizeof(current_name));
|
||||
|
||||
ver_entry_t entries[256];
|
||||
int n = 0;
|
||||
|
||||
DIR *d = opendir(conf_dir);
|
||||
if (!d)
|
||||
return;
|
||||
|
||||
struct dirent *e;
|
||||
while ((e = readdir(d)) != NULL && n < 256) {
|
||||
if (strncmp(e->d_name, "cluster.bin.", 12) != 0)
|
||||
continue;
|
||||
if (strstr(e->d_name, ".tmp") != NULL) {
|
||||
char full[4096];
|
||||
snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
|
||||
unlink(full);
|
||||
continue;
|
||||
}
|
||||
char full[4096];
|
||||
snprintf(full, sizeof(full), "%s/%s", conf_dir, e->d_name);
|
||||
struct stat st;
|
||||
if (stat(full, &st) == 0) {
|
||||
strncpy(entries[n].name, e->d_name, NAME_MAX - 1);
|
||||
entries[n].name[NAME_MAX - 1] = '\0';
|
||||
entries[n].mtime = st.st_mtime;
|
||||
n++;
|
||||
}
|
||||
}
|
||||
closedir(d);
|
||||
|
||||
/* Sort descending by mtime (newest first) */
|
||||
qsort(entries, (size_t)n, sizeof(entries[0]), ver_entry_cmp_desc);
|
||||
|
||||
/* Delete everything after keep_n, protecting the current pointer */
|
||||
for (int i = keep_n; i < n; i++) {
|
||||
if (strcmp(entries[i].name, current_name) == 0)
|
||||
continue;
|
||||
char full[4096];
|
||||
snprintf(full, sizeof(full), "%s/%s", conf_dir, entries[i].name);
|
||||
unlink(full);
|
||||
}
|
||||
}
|
||||
|
||||
/* ── config_contains_reserved_name ─────────────── */
|
||||
|
||||
int config_contains_reserved_name(cluster_config_t *cfg)
|
||||
{
|
||||
if (!cfg)
|
||||
return 0;
|
||||
/* Check queue names — xce_queue_t.queue is the queue name field */
|
||||
for (int i = 0; i < cfg->nqueues; i++) {
|
||||
if (cfg->queues[i] &&
|
||||
cfg->queues[i]->queue &&
|
||||
strcmp(cfg->queues[i]->queue, "lost_and_found") == 0)
|
||||
return 1;
|
||||
}
|
||||
/* Check host config names — xce_hostconfig_t.name */
|
||||
for (int i = 0; i < cfg->nhconfigs; i++) {
|
||||
if (cfg->hconfigs[i] &&
|
||||
cfg->hconfigs[i]->name &&
|
||||
strcmp(cfg->hconfigs[i]->name, "lost_and_found") == 0)
|
||||
return 1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
/* ── sanitise_cn ────────────────────────────────── */
|
||||
|
||||
char *sanitise_cn(const char *src, size_t maxlen)
|
||||
{
|
||||
static char buf[128];
|
||||
size_t n = maxlen < sizeof(buf) - 1 ? maxlen : sizeof(buf) - 1;
|
||||
size_t i;
|
||||
for (i = 0; i < n && src[i] != '\0'; i++)
|
||||
buf[i] = (src[i] >= 0x20 && src[i] < 0x7f) ? src[i] : '?';
|
||||
buf[i] = '\0';
|
||||
return buf;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue