From 99bad45892e3efb01fafa06ce8301b1922cda293 Mon Sep 17 00:00:00 2001 From: ryanohnemus Date: Tue, 30 Apr 2024 16:35:28 -0500 Subject: [PATCH] in_kubernetes_events: add tests and move connection stream up to context Signed-off-by: ryanohnemus --- .../in_kubernetes_events/kubernetes_events.c | 131 ++++--- .../in_kubernetes_events/kubernetes_events.h | 3 + .../kubernetes_events_conf.c | 10 + tests/runtime/CMakeLists.txt | 1 + .../eventlist_v1_with_creationTimestamp.json | 55 +++ .../eventlist_v1_with_creationTimestamp.out | 1 + .../eventlist_v1_with_lastTimestamp.json | 55 +++ .../eventlist_v1_with_lastTimestamp.out | 1 + tests/runtime/data/in_kubernetes_events/token | 1 + tests/runtime/in_kubernetes_events.c | 356 ++++++++++++++++++ 10 files changed, 564 insertions(+), 50 deletions(-) create mode 100644 tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.json create mode 100644 tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.out create mode 100644 tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.json create mode 100644 tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.out create mode 100644 tests/runtime/data/in_kubernetes_events/token create mode 100644 tests/runtime/in_kubernetes_events.c diff --git a/plugins/in_kubernetes_events/kubernetes_events.c b/plugins/in_kubernetes_events/kubernetes_events.c index 0a439ae4fab..b2a012c6539 100644 --- a/plugins/in_kubernetes_events/kubernetes_events.c +++ b/plugins/in_kubernetes_events/kubernetes_events.c @@ -586,7 +586,6 @@ static int process_event_list(struct k8s_events *ctx, char *in_data, size_t in_s } static struct flb_http_client *make_event_watch_api_request(struct k8s_events *ctx, - struct flb_connection *u_conn, uint64_t max_resource_version) { flb_sds_t url; @@ -603,21 +602,20 @@ static struct flb_http_client *make_event_watch_api_request(struct k8s_events *c flb_sds_printf(&url, "?watch=1&resourceVersion=%llu", max_resource_version); flb_plg_info(ctx->ins, "Requesting %s", url); - c = flb_http_client(u_conn, FLB_HTTP_GET, url, + c = flb_http_client(ctx->current_connection, FLB_HTTP_GET, url, NULL, 0, ctx->api_host, ctx->api_port, NULL, 0); flb_sds_destroy(url); return c; } static struct flb_http_client *make_event_list_api_request(struct k8s_events *ctx, - struct flb_connection *u_conn, flb_sds_t continue_token) { flb_sds_t url; struct flb_http_client *c; if (continue_token == NULL && ctx->limit_request == 0 && ctx->namespace == NULL) { - return flb_http_client(u_conn, FLB_HTTP_GET, K8S_EVENTS_KUBE_API_URI, + return flb_http_client(ctx->current_connection, FLB_HTTP_GET, K8S_EVENTS_KUBE_API_URI, NULL, 0, ctx->api_host, ctx->api_port, NULL, 0); } @@ -637,7 +635,7 @@ static struct flb_http_client *make_event_list_api_request(struct k8s_events *ct } flb_sds_printf(&url, "limit=%d", ctx->limit_request); } - c = flb_http_client(u_conn, FLB_HTTP_GET, url, + c = flb_http_client(ctx->current_connection, FLB_HTTP_GET, url, NULL, 0, ctx->api_host, ctx->api_port, NULL, 0); flb_sds_destroy(url); return c; @@ -788,50 +786,50 @@ static void initialize_http_client(struct flb_http_client* c, struct k8s_events* } } -static int k8s_events_collect(struct flb_input_instance *ins, - struct flb_config *config, void *in_context) +static int check_and_init_stream(struct k8s_events *ctx) { - int ret; - size_t b_sent; - struct flb_connection *u_conn = NULL; - struct flb_http_client *c = NULL; - struct k8s_events *ctx = in_context; + /* Returns FLB_TRUE if stream has been initialized */ flb_sds_t continue_token = NULL; uint64_t max_resource_version = 0; - size_t bytes_consumed; - int chunk_proc_ret; + size_t b_sent; + int ret; + struct flb_http_client *c = NULL; - if (pthread_mutex_trylock(&ctx->lock) != 0) { - FLB_INPUT_RETURN(0); + /* if the streaming client is already active, just return it */ + if(ctx->streaming_client) { + return FLB_TRUE; } - u_conn = flb_upstream_conn_get(ctx->upstream); - if (!u_conn) { - flb_plg_error(ins, "upstream connection initialization error"); - goto exit; - } + /* setup connection if one does not exist */ + if(!ctx->current_connection) { + ctx->current_connection = flb_upstream_conn_get(ctx->upstream); + if (!ctx->current_connection) { + flb_plg_error(ctx->ins, "upstream connection initialization error"); + goto failure; + } - ret = refresh_token_if_needed(ctx); - if (ret == -1) { - flb_plg_error(ctx->ins, "failed to refresh token"); - goto exit; + ret = refresh_token_if_needed(ctx); + if (ret == -1) { + flb_plg_error(ctx->ins, "failed to refresh token"); + goto failure; + } } do { - c = make_event_list_api_request(ctx, u_conn, continue_token); + c = make_event_list_api_request(ctx, continue_token); if (continue_token != NULL) { flb_sds_destroy(continue_token); continue_token = NULL; } if (!c) { - flb_plg_error(ins, "unable to create http client"); - goto exit; + flb_plg_error(ctx->ins, "unable to create http client"); + goto failure; } initialize_http_client(c, ctx); ret = flb_http_do(c, &b_sent); if (ret != 0) { - flb_plg_error(ins, "http do error"); - goto exit; + flb_plg_error(ctx->ins, "http do error"); + goto failure; } if (c->resp.status == 200 && c->resp.payload_size > 0) { @@ -846,7 +844,7 @@ static int k8s_events_collect(struct flb_input_instance *ins, else { flb_plg_error(ctx->ins, "http_status=%i", c->resp.status); } - goto exit; + goto failure; } flb_http_client_destroy(c); c = NULL; @@ -860,48 +858,81 @@ static int k8s_events_collect(struct flb_input_instance *ins, /* Now that we've done a full list, we can use the resource version and do a watch * to stream updates efficiently */ - c = make_event_watch_api_request(ctx, u_conn, max_resource_version); - if (!c) { - flb_plg_error(ins, "unable to create http client"); - goto exit; + ctx->streaming_client = make_event_watch_api_request(ctx, max_resource_version); + if (!ctx->streaming_client) { + flb_plg_error(ctx->ins, "unable to create http client"); + goto failure; } - initialize_http_client(c, ctx); + initialize_http_client(ctx->streaming_client, ctx); /* Watch will stream chunked json data, so we only send * the http request, then use flb_http_get_response_data * to attempt processing on available streamed data */ b_sent = 0; - ret = flb_http_do_request(c, &b_sent); + ret = flb_http_do_request(ctx->streaming_client, &b_sent); if (ret != 0) { - flb_plg_error(ins, "http do request error"); - goto exit; + flb_plg_error(ctx->ins, "http do request error"); + goto failure; + } + + return FLB_TRUE; + +failure: + if (c) { + flb_http_client_destroy(c); + } + if (ctx->streaming_client) { + flb_http_client_destroy(ctx->streaming_client); + ctx->streaming_client = NULL; + } + if (ctx->current_connection) { + flb_upstream_conn_release(ctx->current_connection); + ctx->current_connection = NULL; + } + return FLB_FALSE; +} + +static int k8s_events_collect(struct flb_input_instance *ins, + struct flb_config *config, void *in_context) +{ + int ret; + struct k8s_events *ctx = in_context; + size_t bytes_consumed; + int chunk_proc_ret; + + if (pthread_mutex_trylock(&ctx->lock) != 0) { + FLB_INPUT_RETURN(0); + } + + if (check_and_init_stream(ctx) == FLB_FALSE) { + FLB_INPUT_RETURN(0); } ret = FLB_HTTP_MORE; bytes_consumed = 0; chunk_proc_ret = 0; while ((ret == FLB_HTTP_MORE || ret == FLB_HTTP_CHUNK_AVAILABLE) && chunk_proc_ret == 0) { - ret = flb_http_get_response_data(c, bytes_consumed); + ret = flb_http_get_response_data(ctx->streaming_client, bytes_consumed); bytes_consumed = 0; - if( c->resp.status == 200 && ret == FLB_HTTP_CHUNK_AVAILABLE ) { - chunk_proc_ret = process_http_chunk(ctx, c, &bytes_consumed); + if(ctx->streaming_client->resp.status == 200 && ret == FLB_HTTP_CHUNK_AVAILABLE ) { + chunk_proc_ret = process_http_chunk(ctx, ctx->streaming_client, &bytes_consumed); } } /* NOTE: skipping any processing after streaming socket closes */ - if (c->resp.status != 200) { - flb_plg_warn(ins, "events watch failure, http_status=%d payload=%s", c->resp.status, c->resp.payload); + if (ctx->streaming_client->resp.status != 200) { + flb_plg_warn(ins, "events watch failure, http_status=%d payload=%s", + ctx->streaming_client->resp.status, ctx->streaming_client->resp.payload); + + flb_http_client_destroy(ctx->streaming_client); + flb_upstream_conn_release(ctx->current_connection); + ctx->streaming_client = NULL; + ctx->current_connection = NULL; } exit: pthread_mutex_unlock(&ctx->lock); - if (c) { - flb_http_client_destroy(c); - } - if (u_conn) { - flb_upstream_conn_release(u_conn); - } FLB_INPUT_RETURN(0); } diff --git a/plugins/in_kubernetes_events/kubernetes_events.h b/plugins/in_kubernetes_events/kubernetes_events.h index b5ec7db1c9c..734b63a2afb 100644 --- a/plugins/in_kubernetes_events/kubernetes_events.h +++ b/plugins/in_kubernetes_events/kubernetes_events.h @@ -82,6 +82,9 @@ struct k8s_events { struct flb_upstream *upstream; struct flb_input_instance *ins; + struct flb_connection *current_connection; + struct flb_http_client *streaming_client; + /* limit for event queries */ int limit_request; /* last highest seen resource_version */ diff --git a/plugins/in_kubernetes_events/kubernetes_events_conf.c b/plugins/in_kubernetes_events/kubernetes_events_conf.c index bc14b0ad09f..e40d67b415e 100644 --- a/plugins/in_kubernetes_events/kubernetes_events_conf.c +++ b/plugins/in_kubernetes_events/kubernetes_events_conf.c @@ -94,6 +94,8 @@ static int network_init(struct k8s_events *ctx, struct flb_config *config) int io_type = FLB_IO_TCP; ctx->upstream = NULL; + ctx->current_connection = NULL; + ctx->streaming_client = NULL; if (ctx->api_https == FLB_TRUE) { if (!ctx->tls_ca_path && !ctx->tls_ca_file) { @@ -280,6 +282,14 @@ void k8s_events_conf_destroy(struct k8s_events *ctx) flb_ra_destroy(ctx->ra_resource_version); } + if(ctx->streaming_client) { + flb_http_client_destroy(ctx->streaming_client); + } + + if(ctx->current_connection) { + flb_upstream_conn_release(ctx->current_connection); + } + if (ctx->upstream) { flb_upstream_destroy(ctx->upstream); } diff --git a/tests/runtime/CMakeLists.txt b/tests/runtime/CMakeLists.txt index f3e901b6b7f..9e5cc4670e4 100644 --- a/tests/runtime/CMakeLists.txt +++ b/tests/runtime/CMakeLists.txt @@ -57,6 +57,7 @@ if(FLB_OUT_LIB) FLB_RT_TEST(FLB_IN_TCP "in_tcp.c") FLB_RT_TEST(FLB_IN_FORWARD "in_forward.c") FLB_RT_TEST(FLB_IN_FLUENTBIT_METRICS "in_fluentbit_metrics.c") + FLB_RT_TEST(FLB_IN_KUBERNETES_EVENTS "in_kubernetes_events.c") endif() # Filter Plugins diff --git a/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.json b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.json new file mode 100644 index 00000000000..8ac6fb1ad9e --- /dev/null +++ b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.json @@ -0,0 +1,55 @@ +{ + "kind": "EventList", + "apiVersion": "v1", + "metadata": { + "resourceVersion": "177157" + }, + "items": [ + { + "metadata": { + "name": "fluent-bit-78945dccd8-2g7qg.17a3c80ba0453aee", + "namespace": "default", + "uid": "6e3013d5-a79b-4dc4-b6c0-6b652302672e", + "resourceVersion": "176761", + "creationTimestamp": "2023-12-24T13:37:16Z", + "managedFields": [ + { + "manager": "kube-scheduler", + "operation": "Update", + "apiVersion": "events.k8s.io/v1", + "time": "2023-12-24T13:37:16Z", + "fieldsType": "FieldsV1", + "fieldsV1": { + "f:action": {}, + "f:eventTime": {}, + "f:note": {}, + "f:reason": {}, + "f:regarding": {}, + "f:reportingController": {}, + "f:reportingInstance": {}, + "f:type": {} + } + } + ] + }, + "involvedObject": { + "kind": "Pod", + "namespace": "default", + "name": "fluent-bit-78945dccd8-2g7qg", + "uid": "ed7de8ff-61fb-40bb-9ecb-55a801a4cd89", + "apiVersion": "v1", + "resourceVersion": "176749" + }, + "reason": "FailedScheduling", + "message": "0/1 nodes are available: 1 Insufficient cpu. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod..", + "source": {}, + "firstTimestamp": null, + "lastTimestamp": null, + "type": "Warning", + "eventTime": "2023-12-24T13:37:16.335172Z", + "action": "Scheduling", + "reportingComponent": "default-scheduler", + "reportingInstance": "default-scheduler-minikube" + } + ] +} diff --git a/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.out b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.out new file mode 100644 index 00000000000..0ee4783201a --- /dev/null +++ b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_creationTimestamp.out @@ -0,0 +1 @@ +[1703425036.000000,{"metadata":{"name":"fluent-bit-78945dccd8-2g7qg.17a3c80ba0453aee","namespace":"default","uid":"6e3013d5-a79b-4dc4-b6c0-6b652302672e","resourceVersion":"176761","creationTimestamp":"2023-12-24T13:37:16Z","managedFields":[{"manager":"kube-scheduler","operation":"Update","apiVersion":"events.k8s.io/v1","time":"2023-12-24T13:37:16Z","fieldsType":"FieldsV1","fieldsV1":{"f:action":{},"f:eventTime":{},"f:note":{},"f:reason":{},"f:regarding":{},"f:reportingController":{},"f:reportingInstance":{},"f:type":{}}}]},"involvedObject":{"kind":"Pod","namespace":"default","name":"fluent-bit-78945dccd8-2g7qg","uid":"ed7de8ff-61fb-40bb-9ecb-55a801a4cd89","apiVersion":"v1","resourceVersion":"176749"},"reason":"FailedScheduling","message":"0/1 nodes are available: 1 Insufficient cpu. preemption: 0/1 nodes are available: 1 No preemption victims found for incoming pod..","source":{},"firstTimestamp":null,"lastTimestamp":null,"type":"Warning","eventTime":"2023-12-24T13:37:16.335172Z","action":"Scheduling","reportingComponent":"default-scheduler","reportingInstance":"default-scheduler-minikube"}] \ No newline at end of file diff --git a/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.json b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.json new file mode 100644 index 00000000000..8c0f9b6c5c1 --- /dev/null +++ b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.json @@ -0,0 +1,55 @@ +{ + "kind": "EventList", + "apiVersion": "v1", + "metadata": { + "resourceVersion": "177157" + }, + "items": [ + { + "metadata": { + "name": ".17a3ba8b4aa36c81", + "namespace": "default", + "uid": "ec5546b7-f1b9-4e61-a90c-a1f3b611edbc", + "resourceVersion": "174688", + "creationTimestamp": "2023-12-24T09:30:07Z", + "managedFields": [ + { + "manager": "storage-provisioner", + "operation": "Update", + "apiVersion": "v1", + "time": "2023-12-24T09:30:07Z", + "fieldsType": "FieldsV1", + "fieldsV1": { + "f:count": {}, + "f:firstTimestamp": {}, + "f:involvedObject": {}, + "f:lastTimestamp": {}, + "f:message": {}, + "f:reason": {}, + "f:source": { + "f:component": {} + }, + "f:type": {} + } + } + ] + }, + "involvedObject": { + "kind": "Endpoints", + "apiVersion": "v1" + }, + "reason": "LeaderElection", + "message": "minikube_31f5cdfb-29b0-4f84-9f9c-585088e9235f stopped leading", + "source": { + "component": "k8s.io/minikube-hostpath_minikube_31f5cdfb-29b0-4f84-9f9c-585088e9235f" + }, + "firstTimestamp": "2023-12-24T09:29:51Z", + "lastTimestamp": "2023-12-24T09:29:51Z", + "count": 1, + "type": "Normal", + "eventTime": null, + "reportingComponent": "", + "reportingInstance": "" + } + ] +} diff --git a/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.out b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.out new file mode 100644 index 00000000000..35ff686ed31 --- /dev/null +++ b/tests/runtime/data/in_kubernetes_events/eventlist_v1_with_lastTimestamp.out @@ -0,0 +1 @@ +[1703410191.000000,{"metadata":{"name":".17a3ba8b4aa36c81","namespace":"default","uid":"ec5546b7-f1b9-4e61-a90c-a1f3b611edbc","resourceVersion":"174688","creationTimestamp":"2023-12-24T09:30:07Z","managedFields":[{"manager":"storage-provisioner","operation":"Update","apiVersion":"v1","time":"2023-12-24T09:30:07Z","fieldsType":"FieldsV1","fieldsV1":{"f:count":{},"f:firstTimestamp":{},"f:involvedObject":{},"f:lastTimestamp":{},"f:message":{},"f:reason":{},"f:source":{"f:component":{}},"f:type":{}}}]},"involvedObject":{"kind":"Endpoints","apiVersion":"v1"},"reason":"LeaderElection","message":"minikube_31f5cdfb-29b0-4f84-9f9c-585088e9235f stopped leading","source":{"component":"k8s.io/minikube-hostpath_minikube_31f5cdfb-29b0-4f84-9f9c-585088e9235f"},"firstTimestamp":"2023-12-24T09:29:51Z","lastTimestamp":"2023-12-24T09:29:51Z","count":1,"type":"Normal","eventTime":null,"reportingComponent":"","reportingInstance":""}] \ No newline at end of file diff --git a/tests/runtime/data/in_kubernetes_events/token b/tests/runtime/data/in_kubernetes_events/token new file mode 100644 index 00000000000..e6eca93f66e --- /dev/null +++ b/tests/runtime/data/in_kubernetes_events/token @@ -0,0 +1 @@ +fakeTokenFile diff --git a/tests/runtime/in_kubernetes_events.c b/tests/runtime/in_kubernetes_events.c new file mode 100644 index 00000000000..de054fac2c7 --- /dev/null +++ b/tests/runtime/in_kubernetes_events.c @@ -0,0 +1,356 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2019-2022 The Fluent Bit Authors + * Copyright (C) 2015-2018 Treasure Data Inc. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "flb_tests_runtime.h" + +#define JSON_CONTENT_TYPE "application/json" + +#define KUBE_API_HOST "127.0.0.1" +#define KUBE_API_PORT 8449 + +#define V1_EVENTS "/v1/api/events" +#define IN_KUBERNETES_EVENTS_DATA_PATH FLB_TESTS_DATA_PATH "/data/in_kubernetes_events" +#define KUBE_TOKEN_FILE FLB_TESTS_DATA_PATH "/data/in_kubernetes_events/token" + +struct test_ctx { + flb_ctx_t *flb; /* Fluent Bit library context */ + int i_ffd; /* Input fd */ + int f_ffd; /* Filter fd (unused) */ + int o_ffd; /* Output fd */ + +}; + +struct test_k8s_server_ctx { + mk_ctx_t *ctx; /* Monkey HTTP Context */ + int vid; /* Virtual Host ID */ + int mq_id; /* Message Queue ID */ + struct mk_event_loop *evl; + char json_input_file[1024]; +}; + + +pthread_mutex_t result_mutex = PTHREAD_MUTEX_INITIALIZER; +int num_output = 0; +static int get_output_num() +{ + int ret; + pthread_mutex_lock(&result_mutex); + ret = num_output; + pthread_mutex_unlock(&result_mutex); + + return ret; +} + +static void set_output_num(int num) +{ + pthread_mutex_lock(&result_mutex); + num_output = num; + pthread_mutex_unlock(&result_mutex); +} + +static void clear_output_num() +{ + set_output_num(0); +} + +static flb_sds_t read_file(const char *filename) +{ + int fd = -1; + struct stat sb; + int ret; + flb_sds_t payload = NULL; + + fd = open(filename, O_RDONLY, 0); + if (fd != -1) { + if (fstat(fd, &sb) == 0) { + payload = flb_sds_create_size(sb.st_size+1); + if (!payload) { + flb_errno(); + } + else { + ret = read(fd, payload, sb.st_size); + if (ret != sb.st_size) { + flb_error("Problem reading file: %s", filename); + } + payload[sb.st_size] = '\0'; + } + } + close(fd); + } else { + flb_error("Unable to open test file: %s", filename); + } + + return payload; +} + +/* Callback to check expected results */ +static int cb_check_result_json(void *record, size_t size, void *data) +{ + char *p; + flb_sds_t expected; + char *result; + int num = get_output_num(); + const char *filename; + char full_filename[1024]; + + set_output_num(num+1); + + filename = (const char *) data; + result = (char *) record; + + sprintf(full_filename, "%s/%s.out", IN_KUBERNETES_EVENTS_DATA_PATH, filename); + expected = read_file(full_filename); + + p = strstr(result, expected); + TEST_CHECK(p != NULL); + + if (p == NULL) { + flb_error("Expected to find: '%s' in result '%s'", + expected, result); + } + + flb_free(record); + if (expected) { + flb_sds_destroy(expected); + } + return 0; +} + +static void cb_root(mk_request_t *request, void *data) +{ + flb_sds_t payload; + struct test_k8s_server_ctx *server = data; + payload = read_file(server->json_input_file); + + if (request->query_string.data && strstr(request->query_string.data, "watch=1") != NULL) { + // NOTE/TODO: stream via watch not currently supported, this should become 200 status + // and chunked response when we do support it + mk_http_status(request, 500); + mk_http_done(request); + } + else { + mk_http_status(request, 200); + mk_http_header(request, "Content-Type", 12, JSON_CONTENT_TYPE, 16); + mk_http_send(request, payload, strlen(payload), NULL); + mk_http_done(request); + } + + if (payload) { + flb_sds_destroy(payload); + } +} + +struct test_k8s_server_ctx *initialize_mock_k8s_api(const char* filename) +{ + int vid; + char tmp[32]; + struct test_k8s_server_ctx *server; + + server = flb_calloc(1, sizeof(struct test_k8s_server_ctx)); + if (!server) { + flb_errno(); + return NULL; + } + + sprintf(server->json_input_file, "%s/%s.json", + IN_KUBERNETES_EVENTS_DATA_PATH, filename); + + /* Create HTTP server context */ + server->ctx = mk_create(); + if (!server->ctx) { + flb_error("[http_server] could not create context"); + flb_free(server); + return NULL; + } + + /* Compose listen address */ + snprintf(tmp, sizeof(tmp) -1, "%s:%d", KUBE_API_HOST, KUBE_API_PORT); + mk_config_set(server->ctx, "Listen", tmp, NULL); + vid = mk_vhost_create(server->ctx, NULL); + server->vid = vid; + + /* Setup virtual host */ + mk_vhost_set(server->ctx, vid, "Name", "kubernetes-api", NULL); + + /* Root */ + mk_vhost_handler(server->ctx, vid, "/", cb_root, server); + + mk_start(server->ctx); + + return server; +} + +static struct test_ctx *test_ctx_create(struct flb_lib_out_cb *data) +{ + int i_ffd; + int o_ffd; + struct test_ctx *ctx = NULL; + char kube_url[512] = {0}; + + + ctx = flb_calloc(1, sizeof(struct test_ctx)); + if (!TEST_CHECK(ctx != NULL)) { + TEST_MSG("flb_calloc failed"); + flb_errno(); + return NULL; + } + + /* Service config */ + ctx->flb = flb_create(); + flb_service_set(ctx->flb, + "Flush", "0.200000000", + "Grace", "3", + "Log_Level", "debug", + NULL); + + /* Input */ + i_ffd = flb_input(ctx->flb, (char *) "kubernetes_events", NULL); + TEST_CHECK(i_ffd >= 0); + ctx->i_ffd = i_ffd; + + sprintf(kube_url, "http://%s:%d", KUBE_API_HOST, KUBE_API_PORT); + TEST_CHECK(flb_input_set(ctx->flb, i_ffd, + "kube_url", kube_url, + "kube_token_file", KUBE_TOKEN_FILE, + "kube_retention_time", "365000d", + "tls", "off", + "interval_sec", "1", + "interval_nsec", "0", + NULL) == 0); + + /* Output */ + o_ffd = flb_output(ctx->flb, (char *) "lib", (void *) data); + ctx->o_ffd = o_ffd; + + flb_output_set(ctx->flb, ctx->o_ffd, + "match", "*", + "format", "json", + NULL); + + return ctx; +} + +static void test_ctx_destroy(struct test_ctx *ctx) +{ + TEST_CHECK(ctx != NULL); + + flb_stop(ctx->flb); + flb_destroy(ctx->flb); + flb_free(ctx); + +} + +static void mock_k8s_api_destroy(struct test_k8s_server_ctx* server) +{ + TEST_CHECK(server != NULL); + mk_stop(server->ctx); + mk_destroy(server->ctx); + flb_free(server); +} + +void flb_test_events_v1_with_lastTimestamp() +{ + struct flb_lib_out_cb cb_data; + struct test_ctx *ctx; + int ret; + int num; + const char *filename = "eventlist_v1_with_lastTimestamp"; + + clear_output_num(); + + cb_data.cb = cb_check_result_json; + cb_data.data = (void *)filename; + + ctx = test_ctx_create(&cb_data); + if (!TEST_CHECK(ctx != NULL)) { + TEST_MSG("test_ctx_create failed"); + exit(EXIT_FAILURE); + } + + struct test_k8s_server_ctx* k8s_server = initialize_mock_k8s_api( + filename + ); + + ret = flb_start(ctx->flb); + TEST_CHECK(ret == 0); + + // waiting to flush + flb_time_msleep(1500); + + num = get_output_num(); + if (!TEST_CHECK(num > 0)) { + TEST_MSG("no outputs"); + } + mock_k8s_api_destroy(k8s_server); + test_ctx_destroy(ctx); +} + +void flb_test_events_v1_with_creationTimestamp() +{ + struct flb_lib_out_cb cb_data; + struct test_ctx *ctx; + int ret; + int num; + const char *filename = "eventlist_v1_with_creationTimestamp"; + + clear_output_num(); + + cb_data.cb = cb_check_result_json; + cb_data.data = (void *)filename; + + ctx = test_ctx_create(&cb_data); + if (!TEST_CHECK(ctx != NULL)) { + TEST_MSG("test_ctx_create failed"); + exit(EXIT_FAILURE); + } + + struct test_k8s_server_ctx* k8s_server = initialize_mock_k8s_api( + filename + ); + + ret = flb_start(ctx->flb); + TEST_CHECK(ret == 0); + + // waiting to flush + flb_time_msleep(1500); + + num = get_output_num(); + if (!TEST_CHECK(num > 0)) { + TEST_MSG("no outputs"); + } + mock_k8s_api_destroy(k8s_server); + test_ctx_destroy(ctx); +} + +TEST_LIST = { + {"events_v1_with_lastTimestamp", flb_test_events_v1_with_lastTimestamp}, + {"events_v1_with_creationTimestamp", flb_test_events_v1_with_creationTimestamp}, + {NULL, NULL} +}; +