Skip to content

Commit e01062f

Browse files
cosmo0920cosmo0920
committed
tests: internal: Add a regression test case for stale tasks
Signed-off-by: Hiroshi Hatake <hiroshi@chronosphere.io>
1 parent 5dadedd commit e01062f

2 files changed

Lines changed: 360 additions & 0 deletions

File tree

‎tests/internal/CMakeLists.txt‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,7 @@ set(UNIT_TESTS_FILES
6464
opentelemetry.c
6565
storage_dlq.c
6666
engine_adaptive_flush.c
67+
engine_dispatch.c
6768
)
6869

6970
if(FLB_OUT_AZURE_BLOB)
@@ -279,6 +280,11 @@ endfunction(prepare_unit_tests)
279280

280281
prepare_unit_tests(flb-it- "${UNIT_TESTS_FILES}")
281282

283+
if(TARGET flb-it-engine_dispatch)
284+
target_include_directories(flb-it-engine_dispatch PRIVATE
285+
${PROJECT_SOURCE_DIR}/lib/chunkio/deps)
286+
endif()
287+
282288
option(FLB_TESTS_FIPS_ENABLED "Run tests that require an installed OpenSSL FIPS provider" OFF)
283289
if(TARGET flb-it-fips)
284290
if(FLB_OUT_S3)

‎tests/internal/engine_dispatch.c‎

Lines changed: 354 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,354 @@
1+
/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
2+
3+
#include <stdlib.h>
4+
#include <string.h>
5+
6+
#include <fluent-bit/flb_config.h>
7+
#include <fluent-bit/flb_engine_dispatch.h>
8+
#include <fluent-bit/flb_input.h>
9+
#include <fluent-bit/flb_input_chunk.h>
10+
#include <fluent-bit/flb_mem.h>
11+
#include <fluent-bit/flb_output.h>
12+
#include <fluent-bit/flb_scheduler.h>
13+
#include <fluent-bit/flb_socket.h>
14+
#include <fluent-bit/flb_storage.h>
15+
#include <fluent-bit/flb_task.h>
16+
17+
#include <chunkio/chunkio.h>
18+
#include <chunkio/cio_memfs.h>
19+
20+
#include "flb_tests_internal.h"
21+
22+
struct test_ctx {
23+
struct flb_config *config;
24+
struct cio_ctx *cio;
25+
struct flb_input_instance *input;
26+
};
27+
28+
static void test_ctx_destroy(struct test_ctx *ctx)
29+
{
30+
if (ctx == NULL) {
31+
return;
32+
}
33+
34+
if (ctx->config != NULL) {
35+
flb_input_exit_all(ctx->config);
36+
}
37+
38+
if (ctx->cio != NULL) {
39+
cio_destroy(ctx->cio);
40+
ctx->config->cio = NULL;
41+
}
42+
43+
if (ctx->config != NULL) {
44+
flb_config_exit(ctx->config);
45+
}
46+
47+
flb_free(ctx);
48+
}
49+
50+
static struct test_ctx *test_ctx_create(void)
51+
{
52+
int ret;
53+
struct test_ctx *ctx;
54+
struct cio_options options;
55+
#ifdef _WIN32
56+
WSADATA wsa_data;
57+
#endif
58+
59+
ctx = flb_calloc(1, sizeof(struct test_ctx));
60+
if (ctx == NULL) {
61+
flb_errno();
62+
return NULL;
63+
}
64+
65+
ctx->config = flb_config_init();
66+
if (ctx->config == NULL) {
67+
test_ctx_destroy(ctx);
68+
return NULL;
69+
}
70+
71+
#ifdef _WIN32
72+
WSAStartup(0x0201, &wsa_data);
73+
#endif
74+
75+
ctx->config->evl = mk_event_loop_create(8);
76+
if (ctx->config->evl == NULL) {
77+
test_ctx_destroy(ctx);
78+
return NULL;
79+
}
80+
81+
ctx->config->sched = flb_sched_create(ctx->config, ctx->config->evl);
82+
if (ctx->config->sched == NULL) {
83+
test_ctx_destroy(ctx);
84+
return NULL;
85+
}
86+
87+
cio_options_init(&options);
88+
options.flags = CIO_OPEN;
89+
ctx->cio = cio_create(&options);
90+
if (ctx->cio == NULL) {
91+
test_ctx_destroy(ctx);
92+
return NULL;
93+
}
94+
ctx->config->cio = ctx->cio;
95+
96+
ctx->input = flb_input_new(ctx->config, "dummy", NULL, FLB_FALSE);
97+
if (ctx->input == NULL) {
98+
test_ctx_destroy(ctx);
99+
return NULL;
100+
}
101+
102+
ret = flb_storage_input_create(ctx->cio, ctx->input);
103+
if (ret != 0) {
104+
test_ctx_destroy(ctx);
105+
return NULL;
106+
}
107+
108+
return ctx;
109+
}
110+
111+
static struct flb_task_retry *create_retry_dispatch_task(
112+
struct test_ctx *ctx,
113+
struct flb_output_instance *output,
114+
int *task_id,
115+
char **chunk_buffer)
116+
{
117+
struct cio_memfs *memfs;
118+
struct flb_input_chunk *chunk;
119+
struct flb_task *task;
120+
struct flb_task_route *route;
121+
struct flb_task_retry *retry;
122+
123+
chunk = flb_input_chunk_create(ctx->input, FLB_INPUT_LOGS, "test", 4);
124+
if (chunk == NULL) {
125+
return NULL;
126+
}
127+
128+
task = task_alloc(ctx->config);
129+
if (task == NULL) {
130+
flb_input_chunk_destroy(chunk, FLB_TRUE);
131+
return NULL;
132+
}
133+
134+
task->i_ins = ctx->input;
135+
task->ic = chunk;
136+
chunk->task = task;
137+
chunk->busy = FLB_TRUE;
138+
mk_list_add(&task->_head, &ctx->input->tasks);
139+
140+
route = flb_calloc(1, sizeof(struct flb_task_route));
141+
if (route == NULL) {
142+
flb_task_destroy(task, FLB_TRUE);
143+
return NULL;
144+
}
145+
route->out = output;
146+
route->status = FLB_TASK_ROUTE_INACTIVE;
147+
mk_list_add(&route->_head, &task->routes);
148+
149+
retry = flb_calloc(1, sizeof(struct flb_task_retry));
150+
if (retry == NULL) {
151+
flb_task_destroy(task, FLB_TRUE);
152+
return NULL;
153+
}
154+
retry->attempts = 1;
155+
retry->o_ins = output;
156+
retry->parent = task;
157+
mk_list_add(&retry->_head, &task->retries);
158+
159+
/* Force flb_input_chunk_flush() to return NULL without altering production code. */
160+
memfs = ((struct cio_chunk *) chunk->chunk)->backend;
161+
*chunk_buffer = memfs->buf_data;
162+
memfs->buf_data = NULL;
163+
*task_id = task->id;
164+
165+
return retry;
166+
}
167+
168+
static void test_retry_flush_failure_releases_last_task_owner(void)
169+
{
170+
int ret;
171+
int task_id;
172+
char *chunk_buffer;
173+
struct test_ctx *ctx;
174+
struct flb_task *replacement;
175+
struct flb_task_retry *retry;
176+
struct flb_output_instance output;
177+
178+
ctx = test_ctx_create();
179+
TEST_CHECK(ctx != NULL);
180+
if (ctx == NULL) {
181+
return;
182+
}
183+
184+
memset(&output, 0, sizeof(output));
185+
retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer);
186+
TEST_CHECK(retry != NULL);
187+
if (retry == NULL) {
188+
test_ctx_destroy(ctx);
189+
return;
190+
}
191+
192+
ret = flb_engine_dispatch_retry(retry, ctx->config);
193+
TEST_CHECK(ret == -1);
194+
TEST_CHECK(ctx->config->task_map[task_id].task == NULL);
195+
TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0);
196+
TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0);
197+
198+
free(chunk_buffer);
199+
200+
replacement = task_alloc(ctx->config);
201+
TEST_CHECK(replacement != NULL);
202+
if (replacement != NULL) {
203+
TEST_CHECK(replacement->id == task_id);
204+
flb_task_destroy(replacement, FLB_TRUE);
205+
}
206+
207+
test_ctx_destroy(ctx);
208+
}
209+
210+
static void test_retry_flush_failure_preserves_active_task_owner(void)
211+
{
212+
int ret;
213+
int task_id;
214+
char *chunk_buffer;
215+
struct cio_memfs *memfs;
216+
struct test_ctx *ctx;
217+
struct flb_input_chunk *chunk;
218+
struct flb_task *task;
219+
struct flb_task_retry *retry;
220+
struct flb_output_instance output;
221+
222+
ctx = test_ctx_create();
223+
TEST_CHECK(ctx != NULL);
224+
if (ctx == NULL) {
225+
return;
226+
}
227+
228+
memset(&output, 0, sizeof(output));
229+
retry = create_retry_dispatch_task(ctx, &output, &task_id, &chunk_buffer);
230+
TEST_CHECK(retry != NULL);
231+
if (retry == NULL) {
232+
test_ctx_destroy(ctx);
233+
return;
234+
}
235+
236+
task = retry->parent;
237+
task->users = 1;
238+
239+
ret = flb_engine_dispatch_retry(retry, ctx->config);
240+
TEST_CHECK(ret == -1);
241+
TEST_CHECK(ctx->config->task_map[task_id].task == task);
242+
TEST_CHECK(task->users == 1);
243+
TEST_CHECK(mk_list_size(&task->retries) == 0);
244+
TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1);
245+
TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1);
246+
247+
if (ctx->config->task_map[task_id].task == task) {
248+
chunk = task->ic;
249+
memfs = ((struct cio_chunk *) chunk->chunk)->backend;
250+
memfs->buf_data = chunk_buffer;
251+
task->users = 0;
252+
flb_task_users_release(task);
253+
}
254+
else {
255+
free(chunk_buffer);
256+
}
257+
258+
TEST_CHECK(ctx->config->task_map[task_id].task == NULL);
259+
TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0);
260+
TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0);
261+
262+
test_ctx_destroy(ctx);
263+
}
264+
265+
static void test_retry_flush_failure_preserves_pending_retry(void)
266+
{
267+
int ret;
268+
int task_id;
269+
char *chunk_buffer;
270+
struct cio_memfs *memfs;
271+
struct test_ctx *ctx;
272+
struct flb_input_chunk *chunk;
273+
struct flb_task *task;
274+
struct flb_task_route *route;
275+
struct flb_task_retry *remaining_retry;
276+
struct flb_task_retry *retry;
277+
struct flb_output_instance output_a;
278+
struct flb_output_instance output_b;
279+
280+
ctx = test_ctx_create();
281+
TEST_CHECK(ctx != NULL);
282+
if (ctx == NULL) {
283+
return;
284+
}
285+
286+
memset(&output_a, 0, sizeof(output_a));
287+
memset(&output_b, 0, sizeof(output_b));
288+
retry = create_retry_dispatch_task(ctx, &output_a, &task_id, &chunk_buffer);
289+
TEST_CHECK(retry != NULL);
290+
if (retry == NULL) {
291+
test_ctx_destroy(ctx);
292+
return;
293+
}
294+
task = retry->parent;
295+
296+
route = flb_calloc(1, sizeof(struct flb_task_route));
297+
remaining_retry = flb_calloc(1, sizeof(struct flb_task_retry));
298+
TEST_CHECK(route != NULL);
299+
TEST_CHECK(remaining_retry != NULL);
300+
if (route == NULL || remaining_retry == NULL) {
301+
flb_free(route);
302+
flb_free(remaining_retry);
303+
chunk = task->ic;
304+
memfs = ((struct cio_chunk *) chunk->chunk)->backend;
305+
memfs->buf_data = chunk_buffer;
306+
flb_task_destroy(task, FLB_TRUE);
307+
test_ctx_destroy(ctx);
308+
return;
309+
}
310+
311+
route->out = &output_b;
312+
route->status = FLB_TASK_ROUTE_INACTIVE;
313+
mk_list_add(&route->_head, &task->routes);
314+
315+
remaining_retry->attempts = 1;
316+
remaining_retry->o_ins = &output_b;
317+
remaining_retry->parent = task;
318+
mk_list_add(&remaining_retry->_head, &task->retries);
319+
320+
ret = flb_engine_dispatch_retry(retry, ctx->config);
321+
TEST_CHECK(ret == -1);
322+
TEST_CHECK(ctx->config->task_map[task_id].task == task);
323+
TEST_CHECK(task->users == 0);
324+
TEST_CHECK(mk_list_size(&task->retries) == 1);
325+
TEST_CHECK(mk_list_size(&ctx->input->tasks) == 1);
326+
TEST_CHECK(mk_list_size(&ctx->input->chunks) == 1);
327+
328+
if (ctx->config->task_map[task_id].task == task) {
329+
chunk = task->ic;
330+
memfs = ((struct cio_chunk *) chunk->chunk)->backend;
331+
memfs->buf_data = chunk_buffer;
332+
flb_task_retry_destroy(remaining_retry);
333+
flb_task_users_release(task);
334+
}
335+
else {
336+
free(chunk_buffer);
337+
}
338+
339+
TEST_CHECK(ctx->config->task_map[task_id].task == NULL);
340+
TEST_CHECK(mk_list_size(&ctx->input->tasks) == 0);
341+
TEST_CHECK(mk_list_size(&ctx->input->chunks) == 0);
342+
343+
test_ctx_destroy(ctx);
344+
}
345+
346+
TEST_LIST = {
347+
{ "retry_flush_failure_releases_last_task_owner",
348+
test_retry_flush_failure_releases_last_task_owner },
349+
{ "retry_flush_failure_preserves_active_task_owner",
350+
test_retry_flush_failure_preserves_active_task_owner },
351+
{ "retry_flush_failure_preserves_pending_retry",
352+
test_retry_flush_failure_preserves_pending_retry },
353+
{ 0 }
354+
};

0 commit comments

Comments
 (0)