mirror of
https://github.com/rsyslog/rsyslog.git
synced 2026-08-24 19:24:14 +02:00
imdocker: recover from stale since anchors
Why: imdocker can lose discovery of new containers when Docker rejects the internal since=<id> anchor after that container was removed. Impact: imdocker retries discovery without replaying containers it already attached, while keeping the remembered-container set bounded to the containers Docker still reports. Before/After: before a stale anchor could stall discovery and the recovery seen-set could grow without bound; after imdocker clears the bad anchor, rebuilds the latest anchor while scanning the full list, prunes stale seen IDs, and scans only unseen containers. Technical Overview: Track container IDs that have successfully received a log request in a per-instance seen table. Detect Docker's stale-anchor response by HTTP 404 plus an error body that references the exact since id. When that happens, clear the last-container anchor and retry the container list without since. The recovery scan records the current Docker list, updates the latest-container anchor before skipping already-seen IDs, and prunes seen IDs that Docker no longer reports, so completed containers are not reattached and high-churn environments do not accumulate stale IDs indefinitely. Add a Docker-backed regression that removes the anchor container and verifies a later container is still read. Closes: https://github.com/rsyslog/rsyslog/issues/4526 Validation: - devtools/format-code.sh --git-changed - devtools/format-code.sh --git-changed --check --check-if-available - git diff --check - bash -n tests/imdocker-since-deleted-anchor.sh - shellcheck -S warning tests/imdocker-since-deleted-anchor.sh - devtools/check-test-antipatterns.sh tests/imdocker-since-deleted-anchor.sh - make -C contrib/imdocker CC=gcc -j60 imdocker_la-imdocker.lo - reconfigured local worktree from stale CC=clang to CC=gcc after clang rejected GCC-only -Wjump-misses-init - make -j60 - tests/imdocker-since-deleted-anchor.sh With the help of AI-Agents: OpenAI Codex
This commit is contained in:
Rainer Gerhards
parent
cfb5fc4da7
commit
ec0bfbd1ec
@ -143,6 +143,7 @@ typedef struct docker_cont_logs_inst_s {
|
||||
|
||||
typedef struct docker_cont_log_instances_s {
|
||||
struct hashtable *ht_container_log_insts;
|
||||
struct hashtable *ht_seen_container_ids;
|
||||
pthread_mutex_t mut;
|
||||
CURLM *curlm;
|
||||
/* track the latest created container */
|
||||
@ -191,6 +192,8 @@ static rsRetVal dockerContLogReqsGet(docker_cont_log_instances_t *pThis,
|
||||
static rsRetVal dockerContLogReqsPrint(docker_cont_log_instances_t *pThis);
|
||||
static rsRetVal dockerContLogReqsAdd(docker_cont_log_instances_t *pThis, docker_cont_logs_inst_t *pContLogsReqInst);
|
||||
static rsRetVal dockerContLogReqsRemove(docker_cont_log_instances_t *pThis, const char *id);
|
||||
static rsRetVal dockerContLogReqsRememberSeen(docker_cont_log_instances_t *pThis, const char *id);
|
||||
static rsRetVal dockerContLogReqsClearLastContainer(docker_cont_log_instances_t *pThis);
|
||||
|
||||
/* docker_container_info_t */
|
||||
static rsRetVal dockerContainerInfoNew(docker_container_info_t **pThis);
|
||||
@ -719,6 +722,7 @@ static rsRetVal dockerContLogReqsNew(docker_cont_log_instances_t **ppThis) {
|
||||
CHKmalloc(pThis);
|
||||
CHKmalloc(pThis->ht_container_log_insts =
|
||||
create_hashtable(7, hash_from_string, key_equals_string, dockerContLogReqsDestructForHashtable));
|
||||
CHKmalloc(pThis->ht_seen_container_ids = create_hashtable(7, hash_from_string, key_equals_string, NULL));
|
||||
|
||||
CHKiConcCtrl(pthread_mutex_init(&pThis->mut, NULL));
|
||||
|
||||
@ -747,6 +751,11 @@ static rsRetVal dockerContLogReqsDestruct(docker_cont_log_instances_t *pThis) {
|
||||
hashtable_destroy(pThis->ht_container_log_insts, 1);
|
||||
pthread_mutex_unlock(&pThis->mut);
|
||||
}
|
||||
if (pThis->ht_seen_container_ids) {
|
||||
pthread_mutex_lock(&pThis->mut);
|
||||
hashtable_destroy(pThis->ht_seen_container_ids, 0);
|
||||
pthread_mutex_unlock(&pThis->mut);
|
||||
}
|
||||
if (pThis->last_container_id) {
|
||||
free(pThis->last_container_id);
|
||||
}
|
||||
@ -841,6 +850,87 @@ finalize_it:
|
||||
RETiRet;
|
||||
}
|
||||
|
||||
static rsRetVal dockerContLogReqsRememberId(struct hashtable *table, const char *id) {
|
||||
DEFiRet;
|
||||
uchar *keyName = NULL;
|
||||
|
||||
if (table && id && hashtable_search(table, (void *)id) == NULL) {
|
||||
CHKmalloc(keyName = (uchar *)strdup(id));
|
||||
if (!hashtable_insert(table, keyName, keyName)) {
|
||||
ABORT_FINALIZE(RS_RET_ERR);
|
||||
}
|
||||
keyName = NULL;
|
||||
}
|
||||
finalize_it:
|
||||
free(keyName);
|
||||
RETiRet;
|
||||
}
|
||||
|
||||
static rsRetVal dockerContLogReqsRememberSeen(docker_cont_log_instances_t *pThis, const char *id) {
|
||||
return dockerContLogReqsRememberId(pThis ? pThis->ht_seen_container_ids : NULL, id);
|
||||
}
|
||||
|
||||
static rsRetVal dockerContLogReqsPruneSeen(docker_cont_log_instances_t *pThis, struct hashtable *ht_current_ids) {
|
||||
DEFiRet;
|
||||
int ret = 0;
|
||||
struct hashtable_itr *itr = NULL;
|
||||
|
||||
if (!pThis || !pThis->ht_seen_container_ids || !ht_current_ids ||
|
||||
hashtable_count(pThis->ht_seen_container_ids) == 0) {
|
||||
FINALIZE;
|
||||
}
|
||||
|
||||
CHKmalloc(itr = hashtable_iterator(pThis->ht_seen_container_ids));
|
||||
do {
|
||||
char *id = hashtable_iterator_key(itr);
|
||||
if (hashtable_search(ht_current_ids, id) == NULL) {
|
||||
DBGPRINTF("prune stale seen container %s\n", id);
|
||||
ret = hashtable_iterator_remove(itr);
|
||||
} else {
|
||||
ret = hashtable_iterator_advance(itr);
|
||||
}
|
||||
} while (ret);
|
||||
|
||||
finalize_it:
|
||||
free(itr);
|
||||
RETiRet;
|
||||
}
|
||||
|
||||
static rsRetVal dockerContLogReqsUpdateLastContainer(docker_cont_log_instances_t *pThis,
|
||||
const char *containerId,
|
||||
uint64_t created) {
|
||||
DEFiRet;
|
||||
uchar *new_last_container_id = NULL;
|
||||
|
||||
if (pThis && containerId && pThis->last_container_created < created) {
|
||||
CHKmalloc(new_last_container_id = (uchar *)strdup(containerId));
|
||||
pThis->last_container_created = created;
|
||||
free(pThis->last_container_id);
|
||||
pThis->last_container_id = new_last_container_id;
|
||||
new_last_container_id = NULL;
|
||||
DBGPRINTF("last_container_id updated: ('%s', %u)\n", pThis->last_container_id,
|
||||
(unsigned)pThis->last_container_created);
|
||||
}
|
||||
|
||||
finalize_it:
|
||||
free(new_last_container_id);
|
||||
RETiRet;
|
||||
}
|
||||
|
||||
static rsRetVal dockerContLogReqsClearLastContainer(docker_cont_log_instances_t *pThis) {
|
||||
DEFiRet;
|
||||
|
||||
if (pThis) {
|
||||
CHKiConcCtrl(pthread_mutex_lock(&pThis->mut));
|
||||
free(pThis->last_container_id);
|
||||
pThis->last_container_id = NULL;
|
||||
pThis->last_container_created = 0;
|
||||
pthread_mutex_unlock(&pThis->mut);
|
||||
}
|
||||
finalize_it:
|
||||
RETiRet;
|
||||
}
|
||||
|
||||
static rsRetVal dockerContainerInfoNew(docker_container_info_t **ppThis) {
|
||||
DEFiRet;
|
||||
docker_container_info_t *pThis = calloc(1, sizeof(docker_container_info_t));
|
||||
@ -1497,6 +1587,7 @@ static char *dupDockerContainerName(const char *pname) {
|
||||
static rsRetVal process_json(sbool isInit, const char *json, docker_cont_log_instances_t *pInstances) {
|
||||
DEFiRet;
|
||||
struct fjson_object *json_obj = NULL;
|
||||
struct hashtable *ht_current_container_ids = NULL;
|
||||
int mut_locked = 0;
|
||||
DBGPRINTF("%s() - parsing json=%s\n", __FUNCTION__, json);
|
||||
|
||||
@ -1510,6 +1601,7 @@ static rsRetVal process_json(sbool isInit, const char *json, docker_cont_log_ins
|
||||
}
|
||||
|
||||
int length = fjson_object_array_length(json_obj);
|
||||
CHKmalloc(ht_current_container_ids = create_hashtable(7, hash_from_string, key_equals_string, NULL));
|
||||
/* LOCK the update process. */
|
||||
CHKiConcCtrl(pthread_mutex_lock(&pInstances->mut));
|
||||
mut_locked = 1;
|
||||
@ -1553,31 +1645,25 @@ static rsRetVal process_json(sbool isInit, const char *json, docker_cont_log_ins
|
||||
}
|
||||
|
||||
if (containerId) {
|
||||
CHKiRet(dockerContLogReqsRememberId(ht_current_container_ids, containerId));
|
||||
CHKiRet(dockerContLogReqsUpdateLastContainer(pInstances, containerId, containerInfo.created));
|
||||
if (hashtable_search(pInstances->ht_seen_container_ids, (void *)containerId) != NULL) {
|
||||
DBGPRINTF("skip already seen container %s\n", containerId);
|
||||
continue;
|
||||
}
|
||||
docker_cont_logs_inst_t *pInst = NULL;
|
||||
iRet = dockerContLogReqsGet(pInstances, &pInst, containerId);
|
||||
if (iRet == RS_RET_NOT_FOUND) {
|
||||
uchar *new_last_container_id = NULL;
|
||||
#ifdef USE_MULTI_LINE
|
||||
if (dockerContLogsInstNew(&pInst, containerId, &containerInfo, SubmitMsg2)
|
||||
#else
|
||||
if (dockerContLogsInstNew(&pInst, containerId, &containerInfo, SubmitMsg)
|
||||
#endif
|
||||
== RS_RET_OK) {
|
||||
if (pInstances->last_container_created < containerInfo.created) {
|
||||
CHKmalloc(new_last_container_id = (uchar *)strdup(containerId));
|
||||
pInstances->last_container_created = containerInfo.created;
|
||||
free(pInstances->last_container_id);
|
||||
pInstances->last_container_id = new_last_container_id;
|
||||
new_last_container_id = NULL;
|
||||
DBGPRINTF("last_container_id updated: ('%s', %u)\n", pInstances->last_container_id,
|
||||
(unsigned)pInstances->last_container_created);
|
||||
}
|
||||
iRet = dockerContLogsInstSetUrlById(isInit, pInst, pInstances->curlm, containerId);
|
||||
if (iRet != RS_RET_OK) {
|
||||
dockerContLogsInstDestruct(pInst);
|
||||
pInst = NULL;
|
||||
free(new_last_container_id);
|
||||
new_last_container_id = NULL;
|
||||
ABORT_FINALIZE(iRet);
|
||||
}
|
||||
|
||||
@ -1585,31 +1671,69 @@ static rsRetVal process_json(sbool isInit, const char *json, docker_cont_log_ins
|
||||
if (iRet != RS_RET_OK) {
|
||||
dockerContLogsInstDestruct(pInst);
|
||||
pInst = NULL;
|
||||
free(new_last_container_id);
|
||||
new_last_container_id = NULL;
|
||||
ABORT_FINALIZE(iRet);
|
||||
}
|
||||
CHKiRet(dockerContLogReqsRememberSeen(pInstances, containerId));
|
||||
pInst = NULL;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
CHKiRet(dockerContLogReqsPruneSeen(pInstances, ht_current_container_ids));
|
||||
|
||||
finalize_it:
|
||||
if (mut_locked) {
|
||||
pthread_mutex_unlock(&pInstances->mut);
|
||||
}
|
||||
if (ht_current_container_ids) {
|
||||
hashtable_destroy(ht_current_container_ids, 0);
|
||||
}
|
||||
if (json_obj) {
|
||||
json_object_put(json_obj);
|
||||
}
|
||||
RETiRet;
|
||||
}
|
||||
|
||||
static rsRetVal getContainerIds(sbool isInit, docker_cont_log_instances_t *pInstances, const char *url) {
|
||||
static sbool containerListErrorReferencesId(const char *json, const char *id) {
|
||||
sbool referencesId = 0;
|
||||
struct fjson_object *json_obj = NULL;
|
||||
struct fjson_object *message_obj = NULL;
|
||||
|
||||
if (json == NULL || id == NULL) {
|
||||
goto done;
|
||||
}
|
||||
|
||||
json_obj = fjson_tokener_parse(json);
|
||||
if (!json_obj || !fjson_object_is_type(json_obj, fjson_type_object)) {
|
||||
goto done;
|
||||
}
|
||||
if (!fjson_object_object_get_ex(json_obj, "message", &message_obj)) {
|
||||
goto done;
|
||||
}
|
||||
|
||||
const char *const message = fjson_object_get_string(message_obj);
|
||||
referencesId = message != NULL && strstr(message, id) != NULL;
|
||||
|
||||
done:
|
||||
if (json_obj) {
|
||||
json_object_put(json_obj);
|
||||
}
|
||||
return referencesId;
|
||||
}
|
||||
|
||||
static rsRetVal getContainerIds(sbool isInit,
|
||||
docker_cont_log_instances_t *pInstances,
|
||||
const char *url,
|
||||
const char *sinceId,
|
||||
sbool *const pSinceRejected) {
|
||||
DEFiRet;
|
||||
imdocker_req_t *req = NULL;
|
||||
|
||||
if (pSinceRejected != NULL) {
|
||||
*pSinceRejected = 0;
|
||||
}
|
||||
|
||||
CHKiRet(imdockerReqNew(&req));
|
||||
|
||||
CURLcode response = docker_get(req, url);
|
||||
@ -1618,6 +1742,15 @@ static rsRetVal getContainerIds(sbool isInit, docker_cont_log_instances_t *pInst
|
||||
ABORT_FINALIZE(RS_RET_ERR);
|
||||
}
|
||||
|
||||
if (pSinceRejected != NULL) {
|
||||
long http_status = 0;
|
||||
curl_easy_getinfo(req->curl, CURLINFO_RESPONSE_CODE, &http_status);
|
||||
if (http_status == 404 && containerListErrorReferencesId((const char *)req->buf->data, sinceId)) {
|
||||
*pSinceRejected = 1;
|
||||
ABORT_FINALIZE(RS_RET_OK);
|
||||
}
|
||||
}
|
||||
|
||||
CHKiRet(process_json(isInit, (const char *)req->buf->data, pInstances));
|
||||
|
||||
finalize_it:
|
||||
@ -1638,15 +1771,32 @@ static rsRetVal getContainerIdsAndAppend(sbool isInit, docker_cont_log_instances
|
||||
}
|
||||
|
||||
if (pInstances->last_container_id) {
|
||||
sbool sinceRejected = 0;
|
||||
CHKiRet(allocContainersListUrl(&url, pApiAddr, runModConf->apiVersionStr, runModConf->listContainersOptions,
|
||||
pInstances->last_container_id));
|
||||
DBGPRINTF("listcontainers url: %s\n", url);
|
||||
CHKiRet(getContainerIds(isInit, pInstances, (const char *)url, (const char *)pInstances->last_container_id,
|
||||
&sinceRejected));
|
||||
if (sinceRejected) {
|
||||
LogMsg(0, RS_RET_OK_WARN, LOG_WARNING,
|
||||
"imdocker: Docker rejected the last-seen container id; retrying container discovery without the "
|
||||
"since filter and ignoring containers already seen");
|
||||
CHKiRet(dockerContLogReqsClearLastContainer(pInstances));
|
||||
free(url);
|
||||
url = NULL;
|
||||
CHKiRet(allocContainersListUrl(&url, pApiAddr, runModConf->apiVersionStr, runModConf->listContainersOptions,
|
||||
NULL));
|
||||
} else {
|
||||
goto after_get_container_ids;
|
||||
}
|
||||
} else {
|
||||
CHKiRet(
|
||||
allocContainersListUrl(&url, pApiAddr, runModConf->apiVersionStr, runModConf->listContainersOptions, NULL));
|
||||
}
|
||||
DBGPRINTF("listcontainers url: %s\n", url);
|
||||
|
||||
CHKiRet(getContainerIds(isInit, pInstances, (const char *)url));
|
||||
CHKiRet(getContainerIds(isInit, pInstances, (const char *)url, NULL, NULL));
|
||||
after_get_container_ids:
|
||||
if (Debug) {
|
||||
dockerContLogReqsPrint(pInstances);
|
||||
}
|
||||
|
||||
@ -1120,6 +1120,7 @@ TESTS_OMAZUREDCE_YAML = \
|
||||
TESTS_IMDOCKER = \
|
||||
imdocker-basic.sh \
|
||||
imdocker-image-name.sh \
|
||||
imdocker-since-deleted-anchor.sh \
|
||||
imdocker-long-logline.sh \
|
||||
imdocker-long-options.sh \
|
||||
imdocker-new-logs-from-start.sh \
|
||||
|
||||
46
tests/imdocker-since-deleted-anchor.sh
Executable file
46
tests/imdocker-since-deleted-anchor.sh
Executable file
@ -0,0 +1,46 @@
|
||||
#!/bin/bash
|
||||
# Regression test for imdocker container discovery after Docker rejects the
|
||||
# internal "since=<last container id>" anchor. The oracle is that rsyslog first
|
||||
# reads a short-lived container, then still discovers and reads a second
|
||||
# container after the first one has been removed.
|
||||
# This is part of the rsyslog testbench, licensed under ASL 2.0
|
||||
. ${srcdir:=.}/diag.sh init
|
||||
|
||||
command -v docker >/dev/null 2>&1 || skip_test
|
||||
docker info >/dev/null 2>&1 || skip_test
|
||||
|
||||
COOKIE=$(LC_ALL=C tr -dc 'a-zA-Z0-9' < /dev/urandom | head -c 10)
|
||||
OLD_CONTAINER="rsyslog-${COOKIE}-old"
|
||||
NEW_CONTAINER="rsyslog-${COOKIE}-new"
|
||||
|
||||
cleanup_docker() {
|
||||
docker rm -f "$OLD_CONTAINER" "$NEW_CONTAINER" >/dev/null 2>&1 || true
|
||||
}
|
||||
trap cleanup_docker EXIT
|
||||
cleanup_docker
|
||||
|
||||
generate_conf
|
||||
add_conf '
|
||||
template(name="outfmt" type="string" string="%$!metadata!Names% %msg%\n")
|
||||
module(load="../contrib/imdocker/.libs/imdocker"
|
||||
PollingInterval="1"
|
||||
ListContainersOptions="all=true"
|
||||
GetContainerLogOptions="timestamps=0&follow=1&stdout=1&stderr=0&tail=all")
|
||||
if $!metadata!Names == "'$OLD_CONTAINER'" or $!metadata!Names == "'$NEW_CONTAINER'" then {
|
||||
action(type="omfile" template="outfmt" file="'$RSYSLOG_OUT_LOG'")
|
||||
}
|
||||
'
|
||||
|
||||
startup
|
||||
|
||||
docker run --name "$OLD_CONTAINER" alpine /bin/sh -c 'echo old-anchor'
|
||||
wait_content "old-anchor"
|
||||
docker rm "$OLD_CONTAINER"
|
||||
|
||||
docker run --name "$NEW_CONTAINER" alpine /bin/sh -c 'echo new-after-deleted-anchor'
|
||||
wait_content "new-after-deleted-anchor"
|
||||
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
exit_test
|
||||
Loading…
x
Reference in New Issue
Block a user