diff --git a/.github/workflows/pr-closed-docker.yaml b/.github/workflows/pr-closed-docker.yaml index 24c0af2a67f..b51c5523f09 100644 --- a/.github/workflows/pr-closed-docker.yaml +++ b/.github/workflows/pr-closed-docker.yaml @@ -12,7 +12,27 @@ jobs: # We may need a specific token here with `packages:admin` privileges which is not available to GITHUB_TOKEN packages: write steps: + - name: Check whether PR images exist + id: package + uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0 + with: + script: | + try { + await github.rest.packages.getPackageForOrganization({ + org: 'fluent', + package_type: 'container', + package_name: `fluent-bit/pr-${context.issue.number}` + }); + return true; + } catch (error) { + if (error.status !== 404) { + throw error; + } + core.info('No container images were published for this PR.'); + return false; + } - uses: vlaurin/action-ghcr-prune@0cf7d39f88546edd31965acba78cdcb0be14d641 # v0.6.0 + if: steps.package.outputs.result == 'true' with: organization: fluent container: fluent-bit/pr-${{ github.event.number }} diff --git a/plugins/out_s3/s3.c b/plugins/out_s3/s3.c index e2c730d20b5..e0aae318c88 100644 --- a/plugins/out_s3/s3.c +++ b/plugins/out_s3/s3.c @@ -57,16 +57,96 @@ struct worker_info { FLB_TLS_DEFINE(struct worker_info, s3_worker_info); -#ifdef FLB_SYSTEM_WINDOWS -static int setenv(const char *name, const char *value, int overwrite) +struct s3_test_env_entry { + char *name; + char *value; + struct mk_list _head; +}; + +static pthread_mutex_t s3_test_env_mutex = PTHREAD_MUTEX_INITIALIZER; +static struct mk_list s3_test_env = {&s3_test_env, &s3_test_env}; + +/* Values remain immutable until reset, after all test workers have stopped. */ +char *flb_s3_test_env_get(const char *name) { - if (overwrite == 0 && getenv(name) != NULL) { - return 0; + struct mk_list *head; + struct s3_test_env_entry *entry; + char *value; + + pthread_mutex_lock(&s3_test_env_mutex); + mk_list_foreach_r(head, &s3_test_env) { + entry = mk_list_entry(head, struct s3_test_env_entry, _head); + if (strcmp(entry->name, name) == 0) { + value = entry->value; + pthread_mutex_unlock(&s3_test_env_mutex); + return value; + } + } + pthread_mutex_unlock(&s3_test_env_mutex); + return getenv(name); +} + +int flb_s3_test_env_set(const char *name, const char *value, int overwrite) +{ + struct s3_test_env_entry *entry; + struct s3_test_env_entry *previous; + struct mk_list *head; + + entry = flb_calloc(1, sizeof(*entry)); + if (entry == NULL) { + return -1; + } + entry->name = flb_strdup(name); + entry->value = value == NULL ? NULL : flb_strdup(value); + if (entry->name == NULL || (value != NULL && entry->value == NULL)) { + flb_free(entry->name); + flb_free(entry->value); + flb_free(entry); + return -1; } - return _putenv_s(name, value); + pthread_mutex_lock(&s3_test_env_mutex); + if (!overwrite) { + mk_list_foreach_r(head, &s3_test_env) { + previous = mk_list_entry(head, struct s3_test_env_entry, _head); + if (strcmp(previous->name, name) == 0) { + if (previous->value != NULL) { + pthread_mutex_unlock(&s3_test_env_mutex); + flb_free(entry->name); + flb_free(entry->value); + flb_free(entry); + return 0; + } + break; + } + } + } + mk_list_add(&entry->_head, &s3_test_env); + pthread_mutex_unlock(&s3_test_env_mutex); + return 0; +} + +int flb_s3_test_env_unset(const char *name) +{ + return flb_s3_test_env_set(name, NULL, 1); +} + +void flb_s3_test_env_reset(void) +{ + struct mk_list *head; + struct mk_list *tmp; + struct s3_test_env_entry *entry; + + pthread_mutex_lock(&s3_test_env_mutex); + mk_list_foreach_safe(head, tmp, &s3_test_env) { + entry = mk_list_entry(head, struct s3_test_env_entry, _head); + mk_list_del(&entry->_head); + flb_free(entry->name); + flb_free(entry->value); + flb_free(entry); + } + pthread_mutex_unlock(&s3_test_env_mutex); } -#endif static int s3_timer_create(struct flb_s3 *ctx); @@ -413,7 +493,7 @@ static char *mock_error_response(char *error_env_var) char *error = NULL; int len = 0; - err_val = getenv(error_env_var); + err_val = flb_s3_test_env_get(error_env_var); if (err_val != NULL && strlen(err_val) > 0) { error = flb_calloc(strlen(err_val) + 1, sizeof(char)); if (error == NULL) { @@ -432,7 +512,7 @@ static char *mock_error_response(char *error_env_var) int s3_plugin_under_test() { - if (getenv("FLB_S3_PLUGIN_UNDER_TEST") != NULL) { + if (flb_s3_test_env_get("FLB_S3_PLUGIN_UNDER_TEST") != NULL) { return FLB_TRUE; } @@ -522,7 +602,7 @@ int create_headers(struct flb_s3 *ctx, char *body_md5, }; /* - * Track mock API call counts via env vars (e.g. TEST_PutObject_CALL_COUNT) + * Track mock API call counts in synchronized test state * so tests can assert the exact number of S3 API attempts. */ static void mock_s3_call_increment_counter(char *api) @@ -533,11 +613,11 @@ static void mock_s3_call_increment_counter(char *api) char buf[16]; snprintf(env_var, sizeof(env_var), "TEST_%s_CALL_COUNT", api); - val = getenv(env_var); + val = flb_s3_test_env_get(env_var); count = val ? atoi(val) : 0; count++; snprintf(buf, sizeof(buf), "%d", count); - setenv(env_var, buf, 1); + flb_s3_test_env_set(env_var, buf, 1); } static void mock_s3_call_record_uri(char *api, char *uri) @@ -546,18 +626,18 @@ static void mock_s3_call_record_uri(char *api, char *uri) char uri_env_var[64]; char *count; - if (getenv("TEST_RECORD_S3_URIS") == NULL) { + if (flb_s3_test_env_get("TEST_RECORD_S3_URIS") == NULL) { return; } snprintf(count_env_var, sizeof(count_env_var), "TEST_%s_CALL_COUNT", api); - count = getenv(count_env_var); + count = flb_s3_test_env_get(count_env_var); if (count == NULL) { return; } snprintf(uri_env_var, sizeof(uri_env_var), "TEST_%s_URI_%s", api, count); - setenv(uri_env_var, uri, 1); + flb_s3_test_env_set(uri_env_var, uri, 1); } struct flb_http_client *mock_s3_call(char *error_env_var, char *api) @@ -655,7 +735,7 @@ static struct flb_http_client *mock_s3_call_for_tag(char *error_env_var, { char *error_tag; - error_tag = getenv(error_tag_env_var); + error_tag = flb_s3_test_env_get(error_tag_env_var); if (error_tag != NULL && (strlen(error_tag) != tag_len || memcmp(error_tag, tag, tag_len) != 0)) { return mock_s3_call("TEST_S3_OTHER_TAG_ERROR", api); @@ -2100,7 +2180,7 @@ static int construct_request_buffer(struct flb_s3 *ctx, flb_sds_t new_data, int ret; if (s3_plugin_under_test() == FLB_TRUE && - getenv("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR") != NULL) { + flb_s3_test_env_get("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR") != NULL) { return -1; } diff --git a/plugins/out_s3/s3.h b/plugins/out_s3/s3.h index 526f56de708..25c6bb402c1 100644 --- a/plugins/out_s3/s3.h +++ b/plugins/out_s3/s3.h @@ -257,6 +257,10 @@ void multipart_upload_destroy(struct multipart_upload *m_upload); struct flb_http_client *mock_s3_call(char *error_env_var, char *api); int s3_plugin_under_test(); +char *flb_s3_test_env_get(const char *name); +int flb_s3_test_env_set(const char *name, const char *value, int overwrite); +int flb_s3_test_env_unset(const char *name); +void flb_s3_test_env_reset(void); int get_md5_base64(char *buf, size_t buf_size, char *md5_str, size_t md5_str_size); diff --git a/sdk/browser/fluent-bit-pthread.js b/sdk/browser/fluent-bit-pthread.js index ae64808a66d..5461d8fee90 100644 --- a/sdk/browser/fluent-bit-pthread.js +++ b/sdk/browser/fluent-bit-pthread.js @@ -18,8 +18,9 @@ self.addEventListener('error', event => report('error', event.message || 'Pthrea self.addEventListener('message', event => { if (event.data?.flbShutdown === 1) { event.stopImmediatePropagation(); - self.postMessage({flbPthread: 1, stage: 'closed'}); + // Set the closing flag before allowing the owner to finish teardown. self.close(); + self.postMessage({flbPthread: 1, stage: 'closed'}); return; } // CMD_LOAD in the pinned Emscripten 6.0.9 libpthread.js. diff --git a/sdk/browser/fluent-bit-worker.js b/sdk/browser/fluent-bit-worker.js index df5ef1d9493..62ddf41554b 100644 --- a/sdk/browser/fluent-bit-worker.js +++ b/sdk/browser/fluent-bit-worker.js @@ -53,6 +53,7 @@ self.Worker = class extends BrowserWorker { #resolveReady; #rejectReady; #resolveClosed; + #shutdownPromise; #ready = new Promise((resolve, reject) => { this.#resolveReady = resolve; this.#rejectReady = reject; }); constructor(url, options) { const prepared = preparedWorkers.get(options?.name); @@ -104,19 +105,33 @@ self.Worker = class extends BrowserWorker { }); } get ready() { return this.#ready; } - async shutdown() { + shutdown() { + if (this.#shutdownPromise) { return this.#shutdownPromise; } // Let initialized children close before their owner exits. Terminating // both levels at once can retain a Chromium worker after startup fails. - let timer; - try { - await new Promise(resolve => { - this.#resolveClosed = resolve; - timer = setTimeout(resolve, 250); - super.postMessage({flbShutdown: 1}); - }); - } - catch (_) { /* An unavailable child will be terminated below. */ } - finally { clearTimeout(timer); this.terminate(); } + this.#rejectReady(new Error('Pthread initialization cancelled')); + this.#load = null; + this.#shutdownPromise = (async () => { + let timer; + try { + await new Promise(resolve => { + this.#resolveClosed = resolve; + timer = setTimeout(resolve, 250); + super.postMessage({flbShutdown: 1}); + }); + } + catch (_) { /* An unavailable child will be terminated below. */ } + finally { + clearTimeout(timer); + // Dispatch child termination while the owner is still alive. + // A self-close acknowledgement can precede Chromium releasing + // the nested WASM worker; let that termination finish first. + super.terminate(); + await new Promise(resolve => setTimeout(resolve, 250)); + pthreadWorkers.delete(this); + } + })(); + return this.#shutdownPromise; } postMessage(...args) { if (args[0]?.cmd === PTHREAD_LOAD && !this.#runtimeReady) { @@ -127,6 +142,12 @@ self.Worker = class extends BrowserWorker { super.postMessage(...args); } terminate() { + // Emscripten also terminates the pool during exit. Keep those children + // tracked until shutdown finishes so the SDK can await their cleanup. + if (this.#shutdownPromise || active?.operation === 4 || state === 'failed') { + void this.shutdown(); + return; + } this.#rejectReady(new Error('Pthread initialization cancelled')); this.#load = null; pthreadWorkers.delete(this); @@ -224,9 +245,9 @@ async function fatal(message, code = 'E_RUNTIME') { if (wasmInitialization) { await wasmInitialization.catch(() => {}); } releaseStorage?.(); flushLogs(); - post({type: 'fatal', code, message: String(message)}); // Nothing can recover this runtime; release the worker's own event loop. self.close(); + post({type: 'fatal', code, message: String(message), closed: true}); } async function lockStorage(namespace) { @@ -397,8 +418,8 @@ async function execute(request) { } if (method === 'destroy') { await command(id, 4); - await drainLogs(); await Promise.all([...pthreadWorkers].map(worker => worker.shutdown())); + await drainLogs(); releaseStorage?.(); state = 'destroyed'; return null; @@ -432,11 +453,11 @@ self.onmessage = async ({data}) => { executing = true; try { const value = await execute(data); - post({type: 'response', id: data.id, ok: true, value}); if (state === 'destroyed') { clearTimeout(logTimer); self.close(); } + post({type: 'response', id: data.id, ok: true, value, closed: state === 'destroyed'}); } catch (error) { await drainLogs(); diff --git a/sdk/browser/fluent-bit.js b/sdk/browser/fluent-bit.js index f83fbe1d84c..04a4f8edb27 100644 --- a/sdk/browser/fluent-bit.js +++ b/sdk/browser/fluent-bit.js @@ -58,6 +58,8 @@ export function getBrowserSupport() { class FluentBit { #worker; + #workerClosed = false; + #terminationTimer; #options; #state = 'loading'; #info; @@ -121,11 +123,21 @@ class FluentBit { this.#state = state; this.#callback('onStateChange', state); } + #terminateWorker() { + if (this.#workerClosed) { + // Give self-close time to finish before enforcing the fallback. + // Immediate termination can interrupt nested worker cleanup. + if (this.#terminationTimer === undefined) { + this.#terminationTimer = setTimeout(() => this.#worker?.terminate(), 250); + } + } + else { this.#worker?.terminate(); } + } #fatal(failure) { if (['failed', 'destroyed'].includes(this.#state)) { return; } clearTimeout(this.#loadTimer); this.#detachAbort(); - this.#worker?.terminate(); + this.#terminateWorker(); this.#setState('failed'); this.#rejectReady(failure); if (this.#pending) { @@ -180,6 +192,7 @@ class FluentBit { this.#resolveReady(); } else if (message.type === 'fatal') { + this.#workerClosed = message.closed === true; this.#fatal(error(message.code || 'E_RUNTIME', message.message)); } else if (message.type === 'response' && this.#pending?.id === message.id) { @@ -191,7 +204,14 @@ class FluentBit { const pending = this.#pending; this.#pending = null; clearTimeout(pending.timer); - if (message.ok) { pending.resolve(message.value); } + if (message.ok) { + // A terminal reply confirms that the owner called self.close(). + // Interrupting it with terminate() can retain nested Chromium workers. + if (pending.method === 'destroy' && message.closed === true) { + this.#workerClosed = true; + } + pending.resolve(message.value); + } else { const failure = error(message.code, message.message); pending.reject(failure); @@ -209,7 +229,7 @@ class FluentBit { return new Promise((resolve, reject) => { const timer = setTimeout(() => this.#fatal(error('E_TIMEOUT', `${method} timed out; runtime terminated`)), this.#options.operationTimeoutMs); - this.#pending = {id, resolve, reject, timer}; + this.#pending = {id, method, resolve, reject, timer}; try { this.#worker.postMessage({abi: ABI, type: 'request', id, method, payload}); } catch (_) { this.#fatal(error('E_PROTOCOL', 'Unable to send SDK command')); } }); @@ -347,7 +367,7 @@ class FluentBit { this.#pending.reject(error('E_DESTROYED', 'SDK instance was destroyed')); this.#pending = null; } - this.#worker?.terminate(); + this.#terminateWorker(); this.#setState('destroyed'); } })(); diff --git a/src/flb_lib.c b/src/flb_lib.c index f4dab73948c..0631660b863 100644 --- a/src/flb_lib.c +++ b/src/flb_lib.c @@ -1054,9 +1054,6 @@ int flb_stop(flb_ctx_t *ctx) * the service exited for some reason (plugin action). Always * wait and double check that the child thread is not running. */ -#if defined(FLB_SYSTEM_MACOS) - pthread_cancel(tid); -#endif pthread_join(tid, NULL); return 0; } @@ -1073,9 +1070,7 @@ int flb_stop(flb_ctx_t *ctx) flb_debug("[lib] sending STOP signal to the engine"); flb_engine_exit(ctx->config); -#if defined(FLB_SYSTEM_MACOS) - pthread_cancel(tid); -#endif + /* Let engine cleanup join output workers before releasing their state. */ ret = pthread_join(tid, NULL); if (ret != 0) { flb_errno(); diff --git a/tests/internal/CMakeLists.txt b/tests/internal/CMakeLists.txt index 4e60f79673c..e3545df6435 100644 --- a/tests/internal/CMakeLists.txt +++ b/tests/internal/CMakeLists.txt @@ -301,7 +301,12 @@ function(prepare_unit_tests TEST_PREFIX SOURCEFILES) target_link_libraries(${source_file_we} flb-sp) endif() - target_link_libraries(${source_file_we} ${FLB_TEST_LINK_LIBRARY} cfl-static) + if("${source_file_we}" STREQUAL "flb-it-network_dns") + # Keep upstream calls and DNS thread-local state in the test executable. + target_link_libraries(${source_file_we} fluent-bit-static cfl-static) + else() + target_link_libraries(${source_file_we} ${FLB_TEST_LINK_LIBRARY} cfl-static) + endif() if(FLB_AVRO_ENCODER) target_link_libraries(${source_file_we} avro-static jansson) @@ -317,6 +322,13 @@ endfunction(prepare_unit_tests) prepare_unit_tests(flb-it- "${UNIT_TESTS_FILES}") +if(TARGET flb-it-network_dns) + # Use the real resolver implementation with a test-local c-ares initializer. + target_sources(flb-it-network_dns PRIVATE ${PROJECT_SOURCE_DIR}/src/flb_network.c) + target_compile_definitions(flb-it-network_dns PRIVATE + ares_init_options=flb_test_ares_init_options) +endif() + if(TARGET flb-it-engine_dispatch) target_include_directories(flb-it-engine_dispatch PRIVATE ${PROJECT_SOURCE_DIR}/lib/chunkio/deps) diff --git a/tests/internal/input_chunk.c b/tests/internal/input_chunk.c index 0aa8b407453..76c8d63875b 100644 --- a/tests/internal/input_chunk.c +++ b/tests/internal/input_chunk.c @@ -744,85 +744,64 @@ void flb_test_input_chunk_correct_total_records(void) void flb_test_input_chunk_grouped_auto_records(void) { int ret; - int in_ffd; - int out_ffd; char *payload; size_t payload_size; - flb_ctx_t *ctx; + struct flb_config *cfg; struct flb_input_instance *i_ins; + struct flb_output_instance *o_ins; + struct cio_ctx *cio; + struct cio_options opts; struct mk_list *head; + struct mk_list *tmp; struct flb_input_chunk *ic; payload = NULL; payload_size = 0; - ret = build_grouped_log_payload(&payload, &payload_size); - TEST_CHECK(ret == 0); - if (ret != 0) { - return; - } + TEST_ASSERT(ret == 0); - ctx = flb_create(); - TEST_CHECK(ctx != NULL); - if (!ctx) { - flb_free(payload); - return; - } - - ret = flb_service_set(ctx, "flush", "1", "grace", "1", NULL); - TEST_CHECK(ret == 0); - - in_ffd = flb_input(ctx, (char *) "lib", NULL); - TEST_CHECK(in_ffd >= 0); - flb_input_set(ctx, in_ffd, "tag", "test", NULL); - - out_ffd = flb_output(ctx, (char *) "null", NULL); - TEST_CHECK(out_ffd >= 0); - flb_output_set(ctx, out_ffd, "match", "test", NULL); - - ret = flb_start(ctx); - TEST_CHECK(ret == 0); - if (ret != 0) { - flb_destroy(ctx); - flb_free(payload); - return; - } + /* Inspect the chunk synchronously so the engine cannot flush it first. */ + flb_init_env(); + cfg = flb_config_init(); + TEST_ASSERT(cfg != NULL); + cfg->evl = mk_event_loop_create(256); + TEST_ASSERT(cfg->evl != NULL); + flb_log_create(cfg, FLB_LOG_STDERR, FLB_LOG_DEBUG, NULL); - i_ins = mk_list_entry_first(&ctx->config->inputs, - struct flb_input_instance, - _head); - TEST_CHECK(i_ins != NULL); - if (!i_ins) { - flb_stop(ctx); - flb_destroy(ctx); - flb_free(payload); - return; - } + i_ins = flb_input_new(cfg, "lib", NULL, FLB_TRUE); + TEST_ASSERT(i_ins != NULL); + flb_input_set_property(i_ins, "tag", "test"); + i_ins->storage_type = CIO_STORE_MEM; - ret = flb_input_chunk_append_raw(i_ins, - FLB_INPUT_LOGS, - 0, - "test", - 4, - payload, - payload_size); + cio_options_init(&opts); + cio = cio_create(&opts); + TEST_ASSERT(cio != NULL); + ret = flb_storage_input_create(cio, i_ins); + TEST_ASSERT(ret == 0); + ret = flb_input_init_all(cfg); + TEST_ASSERT(ret == 0); + + o_ins = flb_output_new(cfg, "null", NULL, FLB_TRUE); + TEST_ASSERT(o_ins != NULL); + flb_output_set_property(o_ins, "match", "test"); + ret = flb_router_io_set(cfg); + TEST_ASSERT(ret == 0); + + ret = flb_input_chunk_append_raw(i_ins, FLB_INPUT_LOGS, 0, + "test", 4, payload, payload_size); TEST_CHECK(ret == 0); - - flb_time_msleep(200); - - ic = NULL; - mk_list_foreach(head, &i_ins->chunks) { + TEST_CHECK(mk_list_size(&i_ins->chunks) == 1); + mk_list_foreach_safe(head, tmp, &i_ins->chunks) { ic = mk_list_entry(head, struct flb_input_chunk, _head); - break; - } - - TEST_CHECK(ic != NULL); - if (ic != NULL) { TEST_CHECK(ic->total_records == 1); + flb_input_chunk_destroy(ic, FLB_TRUE); } - flb_stop(ctx); - flb_destroy(ctx); + cio_destroy(cio); + flb_router_exit(cfg); + flb_input_exit_all(cfg); + flb_output_exit(cfg); + flb_config_exit(cfg); flb_free(payload); } diff --git a/tests/internal/network_dns.c b/tests/internal/network_dns.c index 659cf0a4482..efed014abbe 100644 --- a/tests/internal/network_dns.c +++ b/tests/internal/network_dns.c @@ -8,10 +8,105 @@ #include #include #include +#include +#ifdef ares_init_options +#undef ares_init_options +#endif #include #include "flb_tests_internal.h" +static unsigned short dns_server_port; + +/* Only this test target redirects c-ares to a loopback DNS server. */ +int flb_test_ares_init_options(ares_channel *channel, struct ares_options *options, + int optmask) +{ + int ret; + char server[64]; + + ret = ares_init_options(channel, options, optmask); + if (ret != ARES_SUCCESS) { + return ret; + } + + snprintf(server, sizeof(server), "127.0.0.1:%u", dns_server_port); + ret = ares_set_servers_ports_csv(*channel, server); + if (ret != ARES_SUCCESS) { + ares_destroy(*channel); + *channel = NULL; + } + return ret; +} + +struct dns_server { + flb_sockfd_t listener; + flb_sockfd_t client; + unsigned char packet[1024]; + size_t received; + size_t sent; +}; + +static void dns_server_start(struct dns_server *server) +{ + struct sockaddr_in address; + socklen_t address_size; + + memset(server, 0, sizeof(*server)); + server->client = FLB_INVALID_SOCKET; + server->listener = socket(AF_INET, SOCK_STREAM, 0); + TEST_ASSERT(server->listener != FLB_INVALID_SOCKET); + memset(&address, 0, sizeof(address)); + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + TEST_ASSERT(bind(server->listener, (struct sockaddr *) &address, sizeof(address)) == 0); + TEST_ASSERT(listen(server->listener, 1) == 0); + address_size = sizeof(address); + TEST_ASSERT(getsockname(server->listener, (struct sockaddr *) &address, &address_size) == 0); + dns_server_port = ntohs(address.sin_port); + TEST_ASSERT(flb_net_socket_nonblocking(server->listener) == 0); +} + +static void dns_server_respond(struct dns_server *server) +{ + int ret; + size_t packet_size; + + if (server->client == FLB_INVALID_SOCKET) { + server->client = accept(server->listener, NULL, NULL); + if (server->client == FLB_INVALID_SOCKET) { + return; + } + TEST_ASSERT(flb_net_socket_nonblocking(server->client) == 0); + } + + if (server->sent == 0) { + ret = recv(server->client, (char *) server->packet + server->received, + sizeof(server->packet) - server->received, 0); + if (ret <= 0) { + return; + } + server->received += ret; + } + if (server->received < 2) { + return; + } + packet_size = 2 + ((size_t) server->packet[0] << 8) + server->packet[1]; + TEST_ASSERT(packet_size >= 14 && packet_size <= sizeof(server->packet)); + if (server->received < packet_size) { + return; + } + + /* Echo the framed question and EDNS data with an NXDOMAIN response. */ + server->packet[4] = 0x81; + server->packet[5] = 0x83; + ret = send(server->client, (char *) server->packet + server->sent, + packet_size - server->sent, 0); + if (ret > 0) { + server->sent += ret; + } +} + struct dns_timeout_context { char *mode; int result; @@ -67,12 +162,14 @@ static void test_dns_timeout(char *mode, int upstream_first, int timeout) struct flb_sched_timer *deadline = NULL; struct mk_list *head; struct mk_event_loop *evl; + struct dns_server server; #ifdef FLB_SYSTEM_WINDOWS WSADATA wsa_data; TEST_ASSERT(WSAStartup(MAKEWORD(2, 2), &wsa_data) == 0); #endif + dns_server_start(&server); flb_coro_init(); flb_coro_thread_init(); flb_engine_evl_init(); @@ -145,11 +242,13 @@ static void test_dns_timeout(char *mode, int upstream_first, int timeout) } /* With a deadline, withhold DNS events to simulate an unresponsive server. - * Without one, let the configured resolver complete the .invalid lookup so - * the coroutine and lookup context can be cleaned up normally. + * Without one, answer locally so completion never depends on runner DNS. */ limit = time(NULL) + 5; while (!context.done && time(NULL) < limit) { + if (timeout <= 0) { + dns_server_respond(&server); + } mk_event_wait_2(evl, 100); mk_event_foreach(event, evl) { if (timeout <= 0 && event->type == FLB_ENGINE_EV_CUSTOM) { @@ -172,7 +271,7 @@ static void test_dns_timeout(char *mode, int upstream_first, int timeout) TEST_CHECK(context.result == ARES_ETIMEOUT); } else { - TEST_CHECK(context.result != ARES_ETIMEOUT); + TEST_CHECK(context.result == ARES_ENOTFOUND); } } else { @@ -193,6 +292,10 @@ static void test_dns_timeout(char *mode, int upstream_first, int timeout) flb_net_dns_ctx_set(NULL); mk_event_loop_destroy(evl); ares_library_cleanup(); + if (server.client != FLB_INVALID_SOCKET) { + flb_socket_close(server.client); + } + flb_socket_close(server.listener); #ifdef FLB_SYSTEM_WINDOWS WSACleanup(); #endif diff --git a/tests/internal/oauth2.c b/tests/internal/oauth2.c index 05abfb3f5b7..fd2ba6e1098 100644 --- a/tests/internal/oauth2.c +++ b/tests/internal/oauth2.c @@ -1042,7 +1042,7 @@ void test_caching_and_refresh(void) config = flb_config_init(); TEST_CHECK(config != NULL); - ret = oauth2_mock_server_start(&server, 65, 0); + ret = oauth2_mock_server_start(&server, 3600, 0); TEST_CHECK(ret == 0); ctx = create_oauth_ctx(config, &server, 58); @@ -1067,7 +1067,8 @@ void test_caching_and_refresh(void) TEST_CHECK(strcmp(token, "mock-token-1") == 0); TEST_CHECK(server.token_requests == 1); - sleep(2); + /* Exercise the refresh boundary without a one-second cache lifetime. */ + ctx->expires_at = time(NULL) + ctx->refresh_skew; ret = flb_oauth2_get_access_token(ctx, &token, FLB_FALSE); TEST_CHECK(ret == 0); diff --git a/tests/runtime/core_shutdown_spin.c b/tests/runtime/core_shutdown_spin.c index 961fb941366..f1d5b4c51f4 100644 --- a/tests/runtime/core_shutdown_spin.c +++ b/tests/runtime/core_shutdown_spin.c @@ -19,6 +19,8 @@ #include #include +#include +#include #include #include #include @@ -103,8 +105,76 @@ void flb_test_duplicate_stop_no_spin(void) } } +static pthread_mutex_t worker_exit_mutex = PTHREAD_MUTEX_INITIALIZER; +static int worker_exit_started; +static int worker_exit_finished; + +static int slow_worker_exit(void *context, struct flb_config *config) +{ + pthread_mutex_lock(&worker_exit_mutex); + worker_exit_started++; + pthread_mutex_unlock(&worker_exit_mutex); + + /* Keep the engine in pthread_join() when the caller invokes flb_stop(). */ + flb_time_msleep(300); + + pthread_mutex_lock(&worker_exit_mutex); + worker_exit_finished++; + pthread_mutex_unlock(&worker_exit_mutex); + return 0; +} + +static void flb_test_stop_waits_for_output_workers(void) +{ + flb_ctx_t *ctx; + struct flb_output_instance *output; + struct flb_output_plugin plugin; + int out_ffd; + int started; + int finished; + int i; + + worker_exit_started = 0; + worker_exit_finished = 0; + ctx = flb_create(); + TEST_ASSERT(ctx != NULL); + TEST_ASSERT(flb_service_set(ctx, "Grace", "1", "Log_Level", "error", NULL) == 0); + TEST_ASSERT(flb_input(ctx, "lib", NULL) >= 0); + out_ffd = flb_output(ctx, "null", NULL); + TEST_ASSERT(out_ffd >= 0); + TEST_ASSERT(flb_output_set(ctx, out_ffd, "match", "*", "workers", "2", NULL) == 0); + + output = mk_list_entry_first(&ctx->config->outputs, struct flb_output_instance, _head); + plugin = *output->p; + plugin.cb_worker_exit = slow_worker_exit; + output->p = &plugin; + TEST_ASSERT(flb_start(ctx) == 0); + TEST_ASSERT(flb_engine_exit(ctx->config) >= 0); + + started = 0; + for (i = 0; i < 500 && started == 0; i++) { + flb_time_msleep(10); + pthread_mutex_lock(&worker_exit_mutex); + started = worker_exit_started; + pthread_mutex_unlock(&worker_exit_mutex); + } + TEST_ASSERT(started > 0); + TEST_CHECK(flb_stop(ctx) == 0); + + pthread_mutex_lock(&worker_exit_mutex); + finished = worker_exit_finished; + pthread_mutex_unlock(&worker_exit_mutex); + TEST_CHECK_(finished == 2, "only %d of 2 workers exited before flb_stop returned", finished); + if (finished != 2) { + /* A cancelled engine may leave workers using the context until process exit. */ + return; + } + flb_destroy(ctx); +} + /* Test list */ TEST_LIST = { {"duplicate_stop_no_spin", flb_test_duplicate_stop_no_spin}, + {"stop_waits_for_output_workers", flb_test_stop_waits_for_output_workers}, {NULL, NULL} }; diff --git a/tests/runtime/filter_rewrite_tag.c b/tests/runtime/filter_rewrite_tag.c index 14c33435b08..c511fdb9fad 100644 --- a/tests/runtime/filter_rewrite_tag.c +++ b/tests/runtime/filter_rewrite_tag.c @@ -802,6 +802,21 @@ static void append_group_test_record(struct flb_log_event_encoder *encoder, int TEST_CHECK(ret == FLB_EVENT_ENCODER_SUCCESS); } +static int cb_append_group(struct flb_input_instance *input, + struct flb_config *config, void *context) +{ + int ret; + struct flb_log_event_encoder *encoder = input->data; + + (void) config; + (void) context; + + ret = flb_input_chunk_append_raw(input, FLB_INPUT_LOGS, 3, "rewrite", 7, + encoder->output_buffer, encoder->output_length); + TEST_CHECK(ret == 0); + return ret; +} + static void check_group_rewrite(int reverse, int keep, int unmatched, int same_tag) { int ret; @@ -816,6 +831,7 @@ static void check_group_rewrite(int reverse, int keep, int unmatched, int same_t struct flb_lib_out_cb callbacks[3]; struct filter_test *ctx; struct flb_input_instance *input; + struct flb_input_plugin input_plugin; struct flb_log_event_encoder encoder; for (i = 0; i < 3; i++) { @@ -840,15 +856,10 @@ static void check_group_rewrite(int reverse, int keep, int unmatched, int same_t TEST_CHECK(ret == 0); } clear_output_num(); - ret = flb_start(ctx->flb); - if (!TEST_CHECK(ret == 0)) { - flb_destroy(ctx->flb); - flb_free(ctx); - return; - } ret = flb_log_event_encoder_init(&encoder, FLB_LOG_EVENT_FORMAT_DEFAULT); if (!TEST_CHECK(ret == FLB_EVENT_ENCODER_SUCCESS)) { - filter_test_destroy(ctx); + flb_destroy(ctx->flb); + flb_free(ctx); return; } ret = flb_log_event_encoder_group_init(&encoder); @@ -874,14 +885,22 @@ static void check_group_rewrite(int reverse, int keep, int unmatched, int same_t append_group_test_record(&encoder, 2); input = flb_input_get_instance(ctx->flb->config, ctx->i_ffd); - if (TEST_CHECK(input != NULL)) { - ret = flb_input_chunk_append_raw(input, FLB_INPUT_LOGS, 3, "rewrite", 7, - encoder.output_buffer, encoder.output_length); - TEST_CHECK(ret == 0); + TEST_ASSERT(input != NULL); + /* Inject on the engine thread before its emitter collector can run. */ + input_plugin = *input->p; + input_plugin.cb_pre_run = cb_append_group; + input->p = &input_plugin; + input->data = &encoder; + ret = flb_start(ctx->flb); + if (!TEST_CHECK(ret == 0)) { + flb_destroy(ctx->flb); + flb_free(ctx); + flb_log_event_encoder_destroy(&encoder); + return; } - flb_log_event_encoder_destroy(&encoder); wait_for_output_num(5000, keep ? 6 : 3, &got); filter_test_destroy(ctx); + flb_log_event_encoder_destroy(&encoder); TEST_CHECK(get_output_num() == (keep ? 6 : 3)); for (id = 0; id < 3; id++) { diff --git a/tests/runtime/in_simple_systems.c b/tests/runtime/in_simple_systems.c index f20dd477eb9..e8723b8e1de 100644 --- a/tests/runtime/in_simple_systems.c +++ b/tests/runtime/in_simple_systems.c @@ -95,6 +95,7 @@ int callback_add_record(void* data, size_t size, void* cb_data) ctx->records[ctx->num_records].size = size; ctx->records[ctx->num_records].data = data; ctx->num_records++; + set_result(ctx->num_records); } return 0; } @@ -306,7 +307,8 @@ void do_test_records_single(char *system, void (*records_cb)(struct callback_rec flb_destroy(ctx); } -void do_test_records_wait_time(char *system, int wait_time, void (*records_cb)(struct callback_records *), ...) +void do_test_records_wait_time(char *system, int wait_time, int minimum_records, + void (*records_cb)(struct callback_records *), ...) { flb_ctx_t *ctx = NULL; int in_ffd; @@ -355,6 +357,13 @@ void do_test_records_wait_time(char *system, int wait_time, void (*records_cb)(s /* Set wait_time plus 2 sec passed. It must have flushed */ flb_time_msleep((wait_time + 2) * 1000); + /* Allow a delayed flush to finish while the engine can still deliver data. + * Read the count published atomically by the output callback. + */ + for (i = 0; i < 50 && get_result() < minimum_records; i++) { + flb_time_msleep(100); + } + flb_stop(ctx); records_cb(records); @@ -557,11 +566,6 @@ void flb_test_dummy_records_message_copies_100(struct callback_records *records) void flb_test_dummy_records_message_rate(struct callback_records *records) { - int trys; - - for (trys = 0; trys < 20 && records->num_records < 20; trys++) { - flb_time_msleep(1000); - } TEST_CHECK(records->num_records >= 20); } @@ -577,7 +581,7 @@ void flb_test_dummy_records_message_interval_nsec(struct callback_records *recor void flb_test_dummy_records_message_flush_on_startup(struct callback_records *records) { - TEST_CHECK(records->num_records >= 2); + TEST_CHECK(records->num_records == 1); } void flb_test_in_dummy_flush(void) @@ -610,24 +614,28 @@ void flb_test_in_dummy_flush(void) do_test_records_single("dummy", flb_test_dummy_records_message_copies_100, "copies", "100", NULL); - do_test_records_wait_time("dummy", 1, flb_test_dummy_records_message_rate, - "rate", "20", - NULL); - do_test_records_wait_time("dummy", 2, flb_test_dummy_records_message_interval_sec, + do_test_records_wait_time("dummy", 2, 1, flb_test_dummy_records_message_interval_sec, "interval_sec", "2", "interval_nsec", "0", NULL); - do_test_records_wait_time("dummy", 1, flb_test_dummy_records_message_interval_nsec, + do_test_records_wait_time("dummy", 1, 1, flb_test_dummy_records_message_interval_nsec, "interval_sec", "0", "interval_nsec", "700000000", NULL); - do_test_records_wait_time("dummy", 5, flb_test_dummy_records_message_flush_on_startup, - "interval_sec", "5", + /* The periodic collector must not satisfy the startup assertion. */ + do_test_records_wait_time("dummy", 1, 1, flb_test_dummy_records_message_flush_on_startup, + "interval_sec", "60", "interval_nsec", "0", "flush_on_startup", "true", NULL); } +void flb_test_in_dummy_rate(void) +{ + do_test_records_wait_time("dummy", 1, 20, flb_test_dummy_records_message_rate, + "rate", "20", NULL); +} + void flb_test_in_dummy_thread_flush(void) { do_test("dummy_thread", NULL); @@ -692,6 +700,7 @@ TEST_LIST = { #endif #ifdef in_dummy {"dummy_flush", flb_test_in_dummy_flush}, + {"dummy_rate", flb_test_in_dummy_rate}, #endif #ifdef in_mem {"mem_flush", flb_test_in_mem_flush}, diff --git a/tests/runtime/out_s3.c b/tests/runtime/out_s3.c index be279b29ef3..1a14e5e4ba9 100644 --- a/tests/runtime/out_s3.c +++ b/tests/runtime/out_s3.c @@ -4,6 +4,7 @@ #include #include "../../plugins/out_s3/s3.h" #include "../../plugins/out_s3/s3_store.h" +#define TEST_FINI flb_s3_test_env_reset() #include "flb_tests_runtime.h" #include "../include/flb_tests_tmpdir.h" #include @@ -152,7 +153,7 @@ static int get_s3_call_count(const char *api) char *value; snprintf(name, sizeof(name), "TEST_%s_CALL_COUNT", api); - value = getenv(name); + value = flb_s3_test_env_get(name); return value ? atoi(value) : 0; } @@ -445,7 +446,7 @@ void flb_test_s3_multipart_success(void) } /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -469,18 +470,18 @@ void flb_test_s3_multipart_success(void) wait_for_s3_call_count("CompleteMultipartUpload", 1); - call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_CompleteMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 CompleteMultipartUpload call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -494,7 +495,7 @@ void flb_test_s3_putobject_success(void) int call_count; /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -520,15 +521,15 @@ void flb_test_s3_putobject_success(void) wait_for_s3_call_count("PutObject", 1); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 PutObject call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); } void flb_test_s3_putobject_error(void) @@ -548,8 +549,8 @@ void flb_test_s3_putobject_error(void) } /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); @@ -576,16 +577,16 @@ void flb_test_s3_putobject_error(void) wait_for_s3_call_count("PutObject", 1); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count >= 1, "Expected >= 1 PutObject calls, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -608,8 +609,8 @@ void flb_test_s3_create_upload_error(void) } /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_CREATE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_CREATE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); @@ -639,7 +640,7 @@ void flb_test_s3_create_upload_error(void) wait_for_s3_call_count("CreateMultipartUpload", 2); wait_for_file_count_at_most(s3_ctx->stream_active->path, 0); - call_count_str = getenv("TEST_CreateMultipartUpload_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_CreateMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 2, "Expected CreateMultipartUpload to stop after two attempts, got %d", @@ -647,19 +648,19 @@ void flb_test_s3_create_upload_error(void) TEST_CHECK_(count_files_recursive(s3_ctx->stream_active->path) == 0, "Expected retry-exhausted chunk to be deleted"); - call_count_str = getenv("TEST_UploadPart_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_UploadPart_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 0, "Expected 0 UploadPart calls, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_CREATE_MULTIPART_UPLOAD_ERROR"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_CREATE_MULTIPART_UPLOAD_ERROR"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -680,8 +681,8 @@ void flb_test_s3_upload_part_error(void) } /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_UPLOAD_PART_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_UPLOAD_PART_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); @@ -705,24 +706,24 @@ void flb_test_s3_upload_part_error(void) wait_for_s3_call_count("UploadPart", 1); - call_count_str = getenv("TEST_UploadPart_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_UploadPart_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count >= 1, "Expected >= 1 UploadPart calls, got %d", call_count); - call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_CompleteMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 0, "Expected 0 CompleteMultipartUpload calls, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_UPLOAD_PART_ERROR"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_UPLOAD_PART_ERROR"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -744,8 +745,8 @@ void flb_test_s3_complete_upload_error(void) } /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); @@ -772,7 +773,7 @@ void flb_test_s3_complete_upload_error(void) wait_for_s3_call_count("CompleteMultipartUpload", 2); - call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_CompleteMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count >= 2, "Expected >= 2 CompleteMultipartUpload calls (retried), got %d", @@ -784,12 +785,12 @@ void flb_test_s3_complete_upload_error(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -808,8 +809,8 @@ void flb_test_s3_ordered_retry_uses_backoff_deadline(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_UPLOAD_PART_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_UPLOAD_PART_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -844,10 +845,10 @@ void flb_test_s3_ordered_retry_uses_backoff_deadline(void) if (ret != 0) { flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_UPLOAD_PART_ERROR"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_UPLOAD_PART_ERROR"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); flb_free(store_dir); return; } @@ -875,10 +876,10 @@ void flb_test_s3_ordered_retry_uses_backoff_deadline(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_UPLOAD_PART_ERROR"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_UPLOAD_PART_ERROR"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); flb_free(store_dir); } @@ -899,9 +900,9 @@ void flb_test_s3_ordered_timer_isolates_tag_backoff(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_RECORD_S3_URIS", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR_TAG", "oldest", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_RECORD_S3_URIS", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR_TAG", "oldest", 1); ctx = flb_create(); for (index = 0; index < 2; index++) { @@ -935,7 +936,7 @@ void flb_test_s3_ordered_timer_isolates_tag_backoff(void) s3_ctx = get_s3_context(ctx); TEST_CHECK(s3_ctx != NULL); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); ret = prepare_s3_ordered_files(s3_ctx); TEST_CHECK_(ret == 0, "Expected both ordered chunks to be created"); if (ret != 0) { @@ -949,18 +950,18 @@ void flb_test_s3_ordered_timer_isolates_tag_backoff(void) get_s3_call_count("PutObject")); TEST_CHECK_(s3_ctx->retry_time == 0, "Expected the healthy upload to preserve the global retry reset behavior"); - uri = getenv("TEST_PutObject_URI_1"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_1"); TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest") == 0, "Expected oldest timed-out chunk first, got %s", uri ? uri : "(null)"); - uri = getenv("TEST_PutObject_URI_2"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_2"); TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/later") == 0, "Expected healthy tag during oldest tag's backoff, got %s", uri ? uri : "(null)"); - unsetenv("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); wait_for_s3_call_count("PutObject", 3); - uri = getenv("TEST_PutObject_URI_3"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_3"); TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest") == 0, "Expected failed tag to retry at its deadline, got %s", uri ? uri : "(null)"); @@ -969,14 +970,14 @@ void flb_test_s3_ordered_timer_isolates_tag_backoff(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PUT_OBJECT_ERROR_TAG"); - unsetenv("TEST_RECORD_S3_URIS"); - unsetenv("TEST_PutObject_CALL_COUNT"); - unsetenv("TEST_PutObject_URI_1"); - unsetenv("TEST_PutObject_URI_2"); - unsetenv("TEST_PutObject_URI_3"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR_TAG"); + flb_s3_test_env_unset("TEST_RECORD_S3_URIS"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_URI_1"); + flb_s3_test_env_unset("TEST_PutObject_URI_2"); + flb_s3_test_env_unset("TEST_PutObject_URI_3"); flb_free(store_dir); } @@ -995,7 +996,7 @@ void flb_test_s3_ordered_construct_error_exhausts_chunk(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -1022,7 +1023,7 @@ void flb_test_s3_ordered_construct_error_exhausts_chunk(void) TEST_CHECK(ret >= 0); s3_ctx = get_s3_context(ctx); TEST_CHECK(s3_ctx != NULL); - setenv("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR", "true", 1); + flb_s3_test_env_set("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR", "true", 1); ret = wait_for_s3_file_create_time(s3_ctx, "construct-error", 15, time(NULL) - 61, NULL, 0); TEST_CHECK_(ret == 0, "Expected construct-error chunk to be created"); @@ -1044,12 +1045,12 @@ void flb_test_s3_ordered_construct_error_exhausts_chunk(void) "Expected retry delay to reset after terminal cleanup"); cleanup: - unsetenv("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR"); + flb_s3_test_env_unset("TEST_CONSTRUCT_REQUEST_BUFFER_ERROR"); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1070,8 +1071,8 @@ void flb_test_s3_ordered_backoff_does_not_starve_completion(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); for (index = 0; index < 2; index++) { @@ -1123,7 +1124,7 @@ void flb_test_s3_ordered_backoff_does_not_starve_completion(void) goto cleanup; } s3_file->create_time = time(NULL) - 61; - setenv("TEST_CREATE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("TEST_CREATE_MULTIPART_UPLOAD_ERROR", ERROR_ACCESS_DENIED, 1); ret = flb_lib_push(ctx, input_fds[1], (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); TEST_CHECK(ret >= 0); @@ -1134,16 +1135,16 @@ void flb_test_s3_ordered_backoff_does_not_starve_completion(void) "Expected pending completion to run while queue head was backing off"); cleanup: - unsetenv("TEST_CREATE_MULTIPART_UPLOAD_ERROR"); - unsetenv("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR"); + flb_s3_test_env_unset("TEST_CREATE_MULTIPART_UPLOAD_ERROR"); + flb_s3_test_env_unset("TEST_COMPLETE_MULTIPART_UPLOAD_ERROR"); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1164,8 +1165,8 @@ void flb_test_s3_ordered_shared_upload_retries_safely(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_UPLOAD_PART_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_UPLOAD_PART_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -1247,12 +1248,12 @@ void flb_test_s3_ordered_shared_upload_retries_safely(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_UPLOAD_PART_ERROR"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_UPLOAD_PART_ERROR"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1266,7 +1267,7 @@ void flb_test_s3_compression_gzip(void) int call_count; /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -1290,18 +1291,18 @@ void flb_test_s3_compression_gzip(void) wait_for_s3_call_count("CompleteMultipartUpload", 1); - call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_CompleteMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 CompleteMultipartUpload call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); } void flb_test_s3_compression_gzip_putobject(void) @@ -1314,7 +1315,7 @@ void flb_test_s3_compression_gzip_putobject(void) int call_count; /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -1340,15 +1341,15 @@ void flb_test_s3_compression_gzip_putobject(void) wait_for_s3_call_count("PutObject", 1); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 PutObject call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); } void flb_test_s3_compression_zstd(void) @@ -1361,7 +1362,7 @@ void flb_test_s3_compression_zstd(void) int call_count; /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -1385,18 +1386,18 @@ void flb_test_s3_compression_zstd(void) wait_for_s3_call_count("CompleteMultipartUpload", 1); - call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_CompleteMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 CompleteMultipartUpload call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); } void flb_test_s3_compression_zstd_putobject(void) @@ -1409,7 +1410,7 @@ void flb_test_s3_compression_zstd_putobject(void) int call_count; /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -1435,15 +1436,15 @@ void flb_test_s3_compression_zstd_putobject(void) wait_for_s3_call_count("PutObject", 1); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 PutObject call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); } void flb_test_s3_compression_snappy(void) @@ -1456,7 +1457,7 @@ void flb_test_s3_compression_snappy(void) int call_count; /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -1480,18 +1481,18 @@ void flb_test_s3_compression_snappy(void) wait_for_s3_call_count("CompleteMultipartUpload", 1); - call_count_str = getenv("TEST_CompleteMultipartUpload_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_CompleteMultipartUpload_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 CompleteMultipartUpload call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_CreateMultipartUpload_CALL_COUNT"); - unsetenv("TEST_UploadPart_CALL_COUNT"); - unsetenv("TEST_CompleteMultipartUpload_CALL_COUNT"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_CreateMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_UploadPart_CALL_COUNT"); + flb_s3_test_env_unset("TEST_CompleteMultipartUpload_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); } void flb_test_s3_compression_snappy_putobject(void) @@ -1504,7 +1505,7 @@ void flb_test_s3_compression_snappy_putobject(void) int call_count; /* mocks calls- signals that we are in test mode */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -1530,15 +1531,15 @@ void flb_test_s3_compression_snappy_putobject(void) wait_for_s3_call_count("PutObject", 1); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 PutObject call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); } void flb_test_s3_preserve_data_ordering(void) @@ -1557,7 +1558,7 @@ void flb_test_s3_preserve_data_ordering(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -1584,15 +1585,15 @@ void flb_test_s3_preserve_data_ordering(void) wait_for_s3_call_count("PutObject", 1); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 PutObject call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1617,8 +1618,8 @@ void flb_test_s3_putobject_retry_limit_semantics(void) } /* Use mocks without flush bypass so the plugin's internal retry runs */ - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); @@ -1641,7 +1642,7 @@ void flb_test_s3_putobject_retry_limit_semantics(void) TEST_CHECK(ret == 0); /* Reset counter after startup so we only count test-driven attempts */ - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); /* Wait until the initial attempt and one retry have run. */ flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); @@ -1650,7 +1651,7 @@ void flb_test_s3_putobject_retry_limit_semantics(void) flb_stop(ctx); flb_destroy(ctx); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; /* retry_limit=1: 1 initial attempt + 1 retry = 2 PutObject calls */ @@ -1658,9 +1659,9 @@ void flb_test_s3_putobject_retry_limit_semantics(void) "Expected 2 PutObject calls (1 attempt + 1 retry), got %d", call_count); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1683,8 +1684,8 @@ void flb_test_s3_default_retry_limit(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); @@ -1706,7 +1707,7 @@ void flb_test_s3_default_retry_limit(void) ret = flb_start(ctx); TEST_CHECK(ret == 0); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); /* Wait for the initial attempt and all five default retries. */ flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); @@ -1715,16 +1716,16 @@ void flb_test_s3_default_retry_limit(void) flb_stop(ctx); flb_destroy(ctx); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 6, "Expected 6 PutObject calls (default retry_limit=5), got %d", call_count); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1747,8 +1748,8 @@ void flb_test_s3_default_retry_exhausted_action_quarantine(void) snprintf(quarantine_dir, sizeof(quarantine_dir), "%s/fluent/quarantine", store_dir); - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); @@ -1771,7 +1772,7 @@ void flb_test_s3_default_retry_exhausted_action_quarantine(void) ret = flb_start(ctx); TEST_CHECK(ret == 0); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); wait_for_s3_call_count("PutObject", 2); wait_for_file_count(quarantine_dir, 1); @@ -1784,9 +1785,9 @@ void flb_test_s3_default_retry_exhausted_action_quarantine(void) "Expected quarantined file(s), got %d", file_count); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1808,9 +1809,9 @@ void flb_test_s3_ordered_index_keeps_global_queue_order(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_RECORD_S3_URIS", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR_TAG", "oldest", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_RECORD_S3_URIS", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR_TAG", "oldest", 1); ctx = flb_create(); for (index = 0; index < 2; index++) { @@ -1844,7 +1845,7 @@ void flb_test_s3_ordered_index_keeps_global_queue_order(void) s3_ctx = get_s3_context(ctx); TEST_CHECK(s3_ctx != NULL); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); ret = prepare_s3_ordered_files(s3_ctx); TEST_CHECK_(ret == 0, "Expected both ordered chunks to be created"); if (ret != 0) { @@ -1856,18 +1857,18 @@ void flb_test_s3_ordered_index_keeps_global_queue_order(void) TEST_CHECK_(get_s3_call_count("PutObject") == 1, "Expected $INDEX to preserve global stop-at-head ordering, got %d calls", get_s3_call_count("PutObject")); - uri = getenv("TEST_PutObject_URI_1"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_1"); TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest/0") == 0, "Expected oldest indexed chunk first, got %s", uri ? uri : "(null)"); - unsetenv("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); wait_for_s3_call_count("PutObject", 3); - uri = getenv("TEST_PutObject_URI_2"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_2"); TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/oldest/0") == 0, "Expected failed indexed chunk to retry before later tag, got %s", uri ? uri : "(null)"); - uri = getenv("TEST_PutObject_URI_3"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_3"); TEST_CHECK_(uri != NULL && strcmp(uri, "/fluent/queue/later/1") == 0, "Expected later indexed chunk after retry, got %s", uri ? uri : "(null)"); @@ -1876,14 +1877,14 @@ void flb_test_s3_ordered_index_keeps_global_queue_order(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PUT_OBJECT_ERROR_TAG"); - unsetenv("TEST_RECORD_S3_URIS"); - unsetenv("TEST_PutObject_CALL_COUNT"); - unsetenv("TEST_PutObject_URI_1"); - unsetenv("TEST_PutObject_URI_2"); - unsetenv("TEST_PutObject_URI_3"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR_TAG"); + flb_s3_test_env_unset("TEST_RECORD_S3_URIS"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_URI_1"); + flb_s3_test_env_unset("TEST_PutObject_URI_2"); + flb_s3_test_env_unset("TEST_PutObject_URI_3"); flb_free(store_dir); } @@ -1906,7 +1907,7 @@ void flb_test_s3_empty_upload_queue_file_deleted(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -1959,8 +1960,8 @@ void flb_test_s3_empty_upload_queue_file_deleted(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -1984,7 +1985,7 @@ void flb_test_s3_near_full_buffer_append_succeeds(void) memset(payload, 'a', sizeof(payload)); memset(excess_payload, 'b', sizeof(excess_payload)); - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -2029,8 +2030,8 @@ void flb_test_s3_near_full_buffer_append_succeeds(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -2054,9 +2055,9 @@ void flb_test_s3_startup_index_order_after_failure(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); - setenv("TEST_RECORD_S3_URIS", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("TEST_RECORD_S3_URIS", "true", 1); ctx = flb_create(); for (index = 0; index < 2; index++) { @@ -2096,8 +2097,8 @@ void flb_test_s3_startup_index_order_after_failure(void) flb_destroy(ctx); flb_time_msleep(1100); - unsetenv("TEST_PutObject_CALL_COUNT"); - unsetenv("TEST_PutObject_URI_1"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_URI_1"); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -2125,7 +2126,7 @@ void flb_test_s3_startup_index_order_after_failure(void) TEST_CHECK_(call_count == 1, "Expected $INDEX recovery to stop after one failure, got %d attempts", call_count); - uri = getenv("TEST_PutObject_URI_1"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_1"); TEST_CHECK_(uri != NULL && (strcmp(uri, "/fluent/recovery/index-test-0/0") == 0 || strcmp(uri, "/fluent/recovery/index-test-1/0") == 0), @@ -2138,18 +2139,18 @@ void flb_test_s3_startup_index_order_after_failure(void) "Expected failed request to roll index back to 0, got %" PRIu64, s3_ctx->seq_index); - unsetenv("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); ret = flb_lib_push(ctx, in_ffd, (char *) JSON_TD, (int) sizeof(JSON_TD) - 1); TEST_CHECK(ret >= 0); wait_for_s3_call_count("PutObject", 3); - uri = getenv("TEST_PutObject_URI_2"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_2"); TEST_CHECK_(uri != NULL && - getenv("TEST_PutObject_URI_1") != NULL && - strcmp(uri, getenv("TEST_PutObject_URI_1")) == 0, + flb_s3_test_env_get("TEST_PutObject_URI_1") != NULL && + strcmp(uri, flb_s3_test_env_get("TEST_PutObject_URI_1")) == 0, "Expected failed chunk retry to keep its tag and index, got %s", uri ? uri : "(null)"); - uri = getenv("TEST_PutObject_URI_3"); + uri = flb_s3_test_env_get("TEST_PutObject_URI_3"); TEST_CHECK_(uri != NULL && (strcmp(uri, "/fluent/recovery/index-test-0/1") == 0 || strcmp(uri, "/fluent/recovery/index-test-1/1") == 0), @@ -2162,14 +2163,14 @@ void flb_test_s3_startup_index_order_after_failure(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_RECORD_S3_URIS"); - unsetenv("TEST_PutObject_CALL_COUNT"); - unsetenv("TEST_PutObject_URI_1"); - unsetenv("TEST_PutObject_URI_2"); - unsetenv("TEST_PutObject_URI_3"); - unsetenv("TEST_PutObject_URI_4"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_RECORD_S3_URIS"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_URI_1"); + flb_s3_test_env_unset("TEST_PutObject_URI_2"); + flb_s3_test_env_unset("TEST_PutObject_URI_3"); + flb_s3_test_env_unset("TEST_PutObject_URI_4"); flb_free(store_dir); } @@ -2197,8 +2198,8 @@ void flb_test_s3_startup_buffer_size_accounting(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); - setenv("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("TEST_PUT_OBJECT_ERROR", ERROR_ACCESS_DENIED, 1); ctx = flb_create(); for (index = 0; index < S3_TEST_STARTUP_FILE_COUNT; index++) { @@ -2250,7 +2251,7 @@ void flb_test_s3_startup_buffer_size_accounting(void) /* Ensure the next process gets a distinct active stream directory. */ flb_time_msleep(1100); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -2292,8 +2293,8 @@ void flb_test_s3_startup_buffer_size_accounting(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "lib", NULL); @@ -2333,9 +2334,9 @@ void flb_test_s3_startup_buffer_size_accounting(void) flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PUT_OBJECT_ERROR"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PUT_OBJECT_ERROR"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } diff --git a/tests/runtime/out_s3_otlp_json.c b/tests/runtime/out_s3_otlp_json.c index 2ed480b428a..70ac6f7185d 100644 --- a/tests/runtime/out_s3_otlp_json.c +++ b/tests/runtime/out_s3_otlp_json.c @@ -1,5 +1,7 @@ /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ #include +#include "../../plugins/out_s3/s3.h" +#define TEST_FINI flb_s3_test_env_reset() #include "flb_tests_runtime.h" #include "../include/flb_tests_tmpdir.h" #include "../../plugins/in_opentelemetry/opentelemetry.h" @@ -81,7 +83,7 @@ void flb_test_s3_format_otlp_json(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); in_ffd = flb_input(ctx, (char *) "opentelemetry", NULL); @@ -110,15 +112,15 @@ void flb_test_s3_format_otlp_json(void) TEST_CHECK(ret == 0); sleep(10); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 PutObject call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } @@ -138,7 +140,7 @@ void flb_test_s3_format_otlp_json_with_compression(void) return; } - setenv("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); + flb_s3_test_env_set("FLB_S3_PLUGIN_UNDER_TEST", "true", 1); ctx = flb_create(); @@ -169,15 +171,15 @@ void flb_test_s3_format_otlp_json_with_compression(void) TEST_CHECK(ret == 0); sleep(10); - call_count_str = getenv("TEST_PutObject_CALL_COUNT"); + call_count_str = flb_s3_test_env_get("TEST_PutObject_CALL_COUNT"); call_count = call_count_str ? atoi(call_count_str) : 0; TEST_CHECK_(call_count == 1, "Expected 1 PutObject call, got %d", call_count); flb_stop(ctx); flb_destroy(ctx); - unsetenv("FLB_S3_PLUGIN_UNDER_TEST"); - unsetenv("TEST_PutObject_CALL_COUNT"); + flb_s3_test_env_unset("FLB_S3_PLUGIN_UNDER_TEST"); + flb_s3_test_env_unset("TEST_PutObject_CALL_COUNT"); flb_free(store_dir); } diff --git a/tests/runtime_shell/in_http_tls_expect.ps1 b/tests/runtime_shell/in_http_tls_expect.ps1 index f0d0c634119..f349a773dfb 100644 --- a/tests/runtime_shell/in_http_tls_expect.ps1 +++ b/tests/runtime_shell/in_http_tls_expect.ps1 @@ -31,7 +31,7 @@ try { $env:SIGNAL_FILE_PATH = $signalFile $env:LISTENER_VHOST = "leo.vcap.me" $env:LISTENER_HOST = "127.0.0.1" - $env:LISTENER_PORT = "50000" + $env:LISTENER_PORT = "20500" $config = Get-RuntimeConfigPath "in_http_tls_expect.conf" $process = Start-FluentBit $config diff --git a/tests/runtime_shell/in_http_tls_expect.sh b/tests/runtime_shell/in_http_tls_expect.sh index fb47e2ae243..838c3e0378d 100755 --- a/tests/runtime_shell/in_http_tls_expect.sh +++ b/tests/runtime_shell/in_http_tls_expect.sh @@ -18,7 +18,7 @@ test_in_http_tls_filter_expect() { export SIGNAL_FILE_PATH="/tmp/fb_signal_$$" export LISTENER_VHOST=leo.vcap.me export LISTENER_HOST=127.0.0.1 - export LISTENER_PORT=50000 + export LISTENER_PORT=20500 input_generator & diff --git a/tests/runtime_shell/in_syslog_tcp_plaintext_expect.ps1 b/tests/runtime_shell/in_syslog_tcp_plaintext_expect.ps1 index aec1243157b..82dec6682b6 100644 --- a/tests/runtime_shell/in_syslog_tcp_plaintext_expect.ps1 +++ b/tests/runtime_shell/in_syslog_tcp_plaintext_expect.ps1 @@ -10,7 +10,7 @@ try { $signalFile = Join-Path $tempDirectory "signal.log" $env:SIGNAL_FILE_PATH = $signalFile $env:LISTENER_HOST = "127.0.0.1" - $env:LISTENER_PORT = "50001" + $env:LISTENER_PORT = "20501" $config = Get-RuntimeConfigPath "in_syslog_tcp_plaintext_expect.conf" $process = Start-FluentBit $config diff --git a/tests/runtime_shell/in_syslog_tcp_plaintext_expect.sh b/tests/runtime_shell/in_syslog_tcp_plaintext_expect.sh index 414d4069880..32dc0e7b263 100755 --- a/tests/runtime_shell/in_syslog_tcp_plaintext_expect.sh +++ b/tests/runtime_shell/in_syslog_tcp_plaintext_expect.sh @@ -15,7 +15,7 @@ input_generator() { test_in_syslog_tcp_plaintext_filter_expect() { export SIGNAL_FILE_PATH="/tmp/fb_signal_$$" export LISTENER_HOST=127.0.0.1 - export LISTENER_PORT=50001 + export LISTENER_PORT=20501 input_generator & diff --git a/tests/runtime_shell/in_syslog_tcp_tls_expect.ps1 b/tests/runtime_shell/in_syslog_tcp_tls_expect.ps1 index faa4df5388f..0449b52f482 100644 --- a/tests/runtime_shell/in_syslog_tcp_tls_expect.ps1 +++ b/tests/runtime_shell/in_syslog_tcp_tls_expect.ps1 @@ -12,7 +12,7 @@ try { $env:SIGNAL_FILE_PATH = $signalFile $env:LISTENER_VHOST = "leo.vcap.me" $env:LISTENER_HOST = "127.0.0.1" - $env:LISTENER_PORT = "50002" + $env:LISTENER_PORT = "20502" $config = Get-RuntimeConfigPath "in_syslog_tcp_tls_expect.conf" $process = Start-FluentBit $config diff --git a/tests/runtime_shell/in_syslog_tcp_tls_expect.sh b/tests/runtime_shell/in_syslog_tcp_tls_expect.sh index d76c0e3c0ce..fbe92ba8c02 100755 --- a/tests/runtime_shell/in_syslog_tcp_tls_expect.sh +++ b/tests/runtime_shell/in_syslog_tcp_tls_expect.sh @@ -16,7 +16,7 @@ test_in_syslog_tcp_plaintext_filter_expect() { export SIGNAL_FILE_PATH="/tmp/fb_signal_$$" export LISTENER_VHOST=leo.vcap.me export LISTENER_HOST=127.0.0.1 - export LISTENER_PORT=50002 + export LISTENER_PORT=20502 input_generator & diff --git a/tests/wasm/sdk_test.py b/tests/wasm/sdk_test.py index 20861baf61f..263a07b9634 100644 --- a/tests/wasm/sdk_test.py +++ b/tests/wasm/sdk_test.py @@ -63,6 +63,15 @@ def send_head(self): return super().send_head() +class WasmFaultHandler(BrowserHandler): + def send_head(self): + if self.path == '/sdk/fluent-bit-runtime.wasm': + self.server.wasm_fault_requests += 1 + self.send_error(503, 'Injected WASM download failure') + return None + return super().send_head() + + def check_storage_failure(page): result = page.evaluate('''async () => { const {createFluentBit} = await import('/sdk/fluent-bit.js'); @@ -180,8 +189,11 @@ def main(): page.on('console', lambda event: print(event.text, flush=True)) errors = [] workers = set() + worker_phases = {} + phase = 'lifecycle' def worker_started(worker): workers.add(worker) + worker_phases[worker] = phase worker.on('close', lambda: workers.discard(worker)) page.on('worker', worker_started) page.on('pageerror', lambda event: errors.append(str(event))) @@ -197,20 +209,26 @@ def worker_started(worker): assert result['passed'] and not errors, errors if not args.fts_only: server.RequestHandlerClass = functools.partial(StorageFaultHandler, build_dir=args.build_dir) + phase = 'storage recovery' check_storage_failure(page) - server.RequestHandlerClass = functools.partial(BrowserHandler, build_dir=args.build_dir) - page.route('**/sdk/fluent-bit-runtime.wasm', lambda route: route.abort()) + phase = 'download failure' + server.wasm_fault_requests = 0 + server.RequestHandlerClass = functools.partial( + WasmFaultHandler, build_dir=args.build_dir) failure = page.evaluate("""async () => { const {createFluentBit} = await import('/sdk/fluent-bit.js'); - try { await createFluentBit(); return null; } + try { await createFluentBit({initTimeoutMs: 10000}); return null; } catch (error) { return error.code; } }""") assert failure in ('E_LOAD', 'E_RUNTIME', 'E_WORKER'), failure + assert server.wasm_fault_requests == 1 for _ in range(100): if not workers: break page.wait_for_timeout(50) - assert not workers, f'{len(workers)} SDK workers survived destroy' + assert not workers, ( + 'SDK workers survived destroy: ' + f'{[(worker.url, worker_phases[worker]) for worker in workers]}') print(result) finally: server.shutdown() diff --git a/tests/wasm/sdk_unit_test.mjs b/tests/wasm/sdk_unit_test.mjs index 906ee82e614..4bf7641e692 100644 --- a/tests/wasm/sdk_unit_test.mjs +++ b/tests/wasm/sdk_unit_test.mjs @@ -6,13 +6,16 @@ import {createFluentBit, getBrowserSupport} from '../../sdk/browser/fluent-bit.j const instances = []; let reply; class FakeWorker { - constructor() { this.terminated = false; instances.push(this); } + constructor() { this.terminated = false; this.closed = false; instances.push(this); } postMessage(message) { queueMicrotask(() => { if (!this.terminated) { reply(this, message); } }); } - emit(message) { this.onmessage?.({data: {abi: 1, ...message}}); } + emit(message) { + if (message.closed === true) { this.closed = true; } + this.onmessage?.({data: {abi: 1, ...message}}); + } terminate() { this.terminated = true; } } const defaults = {workerUrl: 'https://sdk.test/sdk/fluent-bit-worker.js'}; @@ -29,7 +32,7 @@ beforeEach(() => { } }; }); -afterEach(() => { for (const worker of instances) { assert.equal(worker.terminated, true); } }); +afterEach(() => { for (const worker of instances) { assert.equal(worker.terminated || worker.closed, true); } }); test('preflight rejects missing isolation without creating a worker', async () => { globalThis.crossOriginIsolated = false; @@ -224,3 +227,30 @@ test('forced destruction after a startup response cannot revive the instance', a await assert.rejects(sdk.start({yaml: 'pipeline: {}'}), {code: 'E_STATE'}); assert.equal(sdk.state, 'destroyed'); }); + +test('destroy lets an acknowledged owner close before enforcing termination', async () => { + const sdk = await createFluentBit(defaults); + const priorReply = reply; + reply = (worker, message) => { + if (message.method === 'destroy') { + worker.emit({type: 'response', id: message.id, ok: true, closed: true}); + } + else { priorReply(worker, message); } + }; + await sdk.destroy(); + assert.equal(instances[0].closed, true); + assert.equal(instances[0].terminated, false); + await new Promise(resolve => setTimeout(resolve, 300)); + assert.equal(instances[0].terminated, true); + assert.equal(sdk.state, 'destroyed'); +}); + +test('fatal closure acknowledgement delays the termination fallback', async () => { + reply = worker => worker.emit({type: 'fatal', code: 'E_LOAD', + message: 'Injected download failure', closed: true}); + await assert.rejects(createFluentBit(defaults), {code: 'E_LOAD'}); + assert.equal(instances[0].closed, true); + assert.equal(instances[0].terminated, false); + await new Promise(resolve => setTimeout(resolve, 300)); + assert.equal(instances[0].terminated, true); +}); diff --git a/tests/wasm/sdk_worker_unit_test.mjs b/tests/wasm/sdk_worker_unit_test.mjs index b2de2810e04..e35d3a91283 100644 --- a/tests/wasm/sdk_worker_unit_test.mjs +++ b/tests/wasm/sdk_worker_unit_test.mjs @@ -4,10 +4,10 @@ import assert from 'node:assert/strict'; import {readFileSync} from 'node:fs'; import vm from 'node:vm'; -function worker() { +function worker(BrowserWorker = class {}) { const source = readFileSync(new URL('../../sdk/browser/fluent-bit-worker.js', import.meta.url), 'utf8'); const context = vm.createContext({TextEncoder, setTimeout, clearTimeout, - self: {Worker: class {}, addEventListener() {}}}); + self: {Worker: BrowserWorker, addEventListener() {}}}); // Startup is deliberately omitted: exercise the real command state machine // with a controllable storage backend, including rejected IndexedDB commits. vm.runInContext(source.replaceAll('import.meta.url', '"https://sdk.test/worker.js"'), context); @@ -59,6 +59,48 @@ test('dynamic pthread creation cannot exceed the qualified pool', () => { assert.throws(() => sdk.run('new self.Worker("unexpected.js")'), {code: 'E_LIMIT'}); }); +class FakeBrowserWorker { + constructor() { this.listeners = new Map(); this.messages = []; this.terminated = false; } + addEventListener(type, listener) { this.listeners.set(type, listener); } + postMessage(message) { this.messages.push(message); } + terminate() { this.terminated = true; } +} + +test('runtime termination retains children until acknowledged cleanup finishes', async () => { + const sdk = worker(FakeBrowserWorker); + sdk.run('active = {operation: 4}'); + const child = sdk.run('const child = new self.Worker("pthread.js"); child.terminate(); child'); + assert.equal(sdk.run('pthreadWorkers.size'), 1); + assert.equal(child.terminated, false); + child.listeners.get('message')({data: {flbPthread: 1, stage: 'closed'}, + stopImmediatePropagation() {}}); + await new Promise(resolve => setTimeout(resolve, 0)); + assert.equal(child.terminated, true); + assert.equal(sdk.run('pthreadWorkers.size'), 1); + await sdk.run('child.shutdown()'); + assert.equal(sdk.run('pthreadWorkers.size'), 0); + assert.equal(child.terminated, true); + sdk.run('active = null; child.terminate()'); + assert.equal(child.messages.length, 1); +}); + +test('unresponsive pthread shutdown remains bounded and forces termination', async () => { + const sdk = worker(FakeBrowserWorker); + sdk.run('active = {operation: 4}'); + const child = sdk.run('const child = new self.Worker("pthread.js"); child.terminate(); child'); + assert.equal(sdk.run('pthreadWorkers.size'), 1); + await sdk.run('child.shutdown()'); + assert.equal(sdk.run('pthreadWorkers.size'), 0); + assert.equal(child.terminated, true); +}); + +test('ordinary pthread termination still stops the worker immediately', () => { + const sdk = worker(FakeBrowserWorker); + const child = sdk.run('const child = new self.Worker("pthread.js"); child.terminate(); child'); + assert.equal(sdk.run('pthreadWorkers.size'), 0); + assert.equal(child.terminated, true); +}); + test('failed persistent startup requires checkpoint recovery too', async () => { const sdk = worker(); sdk.run("state = 'stopped'; checkpointRequired = false; command = async () => ({result: -3})"); @@ -70,23 +112,49 @@ test('failed persistent startup requires checkpoint recovery too', async () => { assert.equal((await sdk.request('getStats')).checkpointRequired, false); }); -test('graceful destruction drains logs and children before replying and closing', async () => { +test('graceful destruction drains child shutdown logs before replying and closing', async () => { const sdk = worker(); sdk.run(` const teardown = []; + let closingReply; state = 'stopped'; command = async () => { teardown.push('engine'); return {result: 0}; }; drainLogs = async () => { teardown.push('logs'); }; pthreadWorkers.add({shutdown: async () => { teardown.push('child'); }}); releaseStorage = () => teardown.push('storage'); self.postMessage = message => { - if (message.type === 'response' && message.ok) teardown.push('response'); + if (message.type === 'response' && message.ok) { + closingReply = message.closed; + teardown.push('response'); + } }; self.close = () => teardown.push('close'); `); await sdk.run(`self.onmessage({data: {abi: 1, type: 'request', id: 1, method: 'destroy', payload: {}}})`); assert.deepEqual(Array.from(sdk.run('teardown')), - ['engine', 'logs', 'child', 'storage', 'response', 'close']); + ['engine', 'child', 'logs', 'storage', 'close', 'response']); + assert.equal(sdk.run('closingReply'), true); assert.equal(sdk.run('state'), 'destroyed'); }); + +test('fatal startup cleanup marks the owner closing before notifying its parent', async () => { + const sdk = worker(); + sdk.run(` + const teardown = []; + let closingReply; + pthreadWorkers.add({shutdown: async () => teardown.push('child')}); + releaseStorage = () => teardown.push('storage'); + self.close = () => teardown.push('close'); + self.postMessage = message => { + if (message.type === 'fatal') { + closingReply = message.closed; + teardown.push('fatal'); + } + }; + `); + await sdk.run('fatal("injected startup failure")'); + assert.deepEqual(Array.from(sdk.run('teardown')), ['child', 'storage', 'close', 'fatal']); + assert.equal(sdk.run('closingReply'), true); + assert.equal(sdk.run('state'), 'failed'); +});