diff --git a/plugins/in_kubernetes_events/kubernetes_events.c b/plugins/in_kubernetes_events/kubernetes_events.c index 9b98a88c0e9..276ec42c2f1 100644 --- a/plugins/in_kubernetes_events/kubernetes_events.c +++ b/plugins/in_kubernetes_events/kubernetes_events.c @@ -42,6 +42,8 @@ static int k8s_events_sql_insert_event(struct k8s_events *ctx, msgpack_object *item); #endif +#define JSON_ARRAY_DELIM "\r\n" + static int file_to_buffer(const char *path, char **out_buf, size_t *out_size) { @@ -79,7 +81,7 @@ static int file_to_buffer(const char *path, fclose(fp); - // trim new lines + /* trim new lines */ for (len = st.st_size; len > 0; len--) { if (buf[len-1] != '\n' && buf[len-1] != '\r') { break; @@ -241,7 +243,7 @@ static int record_get_field_uint64(msgpack_object *obj, const char *fieldname, u return -1; } - // attempt to parse string as number... + /* attempt to parse string as number... */ if (v->type == MSGPACK_OBJECT_STR) { *val = strtoul(v->via.str.ptr, &end, 10); if (end == NULL || (end < v->via.str.ptr + v->via.str.size)) { @@ -265,8 +267,9 @@ static int item_get_timestamp(msgpack_object *obj, struct flb_time *event_time) int ret; msgpack_object *metadata; - // some events can have lastTimestamp and firstTimestamp set to - // NULL while having metadata.creationTimestamp set. + /* some events can have lastTimestamp and firstTimestamp set to + * NULL while having metadata.creationTimestamp set. + */ ret = record_get_field_time(obj, "lastTimestamp", event_time); if (ret != -1) { return FLB_TRUE; @@ -294,15 +297,15 @@ static bool check_event_is_filtered(struct k8s_events *ctx, msgpack_object *obj, struct flb_time *event_time) { int ret; - time_t now; + uint64_t outdated; msgpack_object *metadata; flb_sds_t uid; uint64_t resource_version; - now = (time_t)(cfl_time_now() / 1000000000); - if (event_time->tm.tv_sec < (now - ctx->retention_time)) { + outdated = cfl_time_now() - (ctx->retention_time * 1000000000L); + if (flb_time_to_nanosec(event_time) < outdated) { flb_plg_debug(ctx->ins, "Item is older than retention_time: %ld < %ld", - *event_time, (now - ctx->retention_time)); + flb_time_to_nanosec(event_time), outdated); return FLB_TRUE; } @@ -354,7 +357,7 @@ static bool check_event_is_filtered(struct k8s_events *ctx, msgpack_object *obj, } #endif - // check if this is an old event. + /* check if this is an old event. */ if (ctx->last_resource_version && resource_version <= ctx->last_resource_version) { flb_plg_debug(ctx->ins, "skipping old object: %llu (< %llu)", resource_version, ctx->last_resource_version); @@ -366,7 +369,121 @@ static bool check_event_is_filtered(struct k8s_events *ctx, msgpack_object *obj, return FLB_FALSE; } -static int process_events(struct k8s_events *ctx, char *in_data, size_t in_size, uint64_t *max_resource_version, flb_sds_t *continue_token) + +static int process_event_object(struct k8s_events *ctx, flb_sds_t action, + msgpack_object *item) +{ + int ret = -1; + struct flb_time ts; + uint64_t resource_version; + msgpack_object* item_metadata; + + if(strncmp(action, "ADDED", 5) != 0 && strncmp(action, "MODIFIED", 8) != 0 ) { + /* We don't process DELETED nor BOOKMARK */ + return 0; + } + + item_metadata = record_get_field_ptr(item, "metadata"); + if (item_metadata == NULL) { + flb_plg_warn(ctx->ins, "Event without metadata"); + return -1; + } + ret = record_get_field_uint64(item_metadata, "resourceVersion", &resource_version); + if (ret == -1) { + return ret; + } + + /* reset the log encoder */ + flb_log_event_encoder_reset(ctx->encoder); + + /* print every item from the items array */ + if (item->type != MSGPACK_OBJECT_MAP) { + flb_plg_error(ctx->ins, "Cannot unpack item in response"); + return -1; + } + + /* get event timestamp */ + ret = item_get_timestamp(item, &ts); + if (ret == FLB_FALSE) { + flb_plg_error(ctx->ins, "cannot retrieve event timestamp"); + return -1; + } + + if (check_event_is_filtered(ctx, item, &ts) == FLB_TRUE) { + return 0; + } + +#ifdef FLB_HAVE_SQLDB + if (ctx->db) { + k8s_events_sql_insert_event(ctx, item); + } +#endif + + /* encode content as a log event */ + flb_log_event_encoder_begin_record(ctx->encoder); + flb_log_event_encoder_set_timestamp(ctx->encoder, &ts); + + ret = flb_log_event_encoder_set_body_from_msgpack_object(ctx->encoder, item); + if (ret == FLB_EVENT_ENCODER_SUCCESS) { + ret = flb_log_event_encoder_commit_record(ctx->encoder); + } + else { + flb_plg_warn(ctx->ins, "unable to encode: %llu", resource_version); + } + + if (ctx->encoder->output_length > 0) { + flb_input_log_append(ctx->ins, NULL, 0, + ctx->encoder->output_buffer, + ctx->encoder->output_length); + } + + return 0; +} + +static int process_watched_event(struct k8s_events *ctx, char *buf_data, size_t buf_size) { + int ret = -1; + size_t off = 0; + msgpack_unpacked result; + msgpack_object root; + msgpack_object *item = NULL; + flb_sds_t event_type = NULL; + + /* unpack */ + msgpack_unpacked_init(&result); + ret = msgpack_unpack_next(&result, buf_data, buf_size, &off); + if (ret != MSGPACK_UNPACK_SUCCESS) { + flb_plg_error(ctx->ins, "Cannot unpack response"); + return -1; + } + + root = result.data; + if (root.type != MSGPACK_OBJECT_MAP) { + return -1; + } + + ret = record_get_field_sds(&root, "type", &event_type); + if (ret == -1) { + flb_plg_warn(ctx->ins, "Streamed Event 'type' not found"); + goto msg_error; + } + + item = record_get_field_ptr(&root, "object"); + if (item == NULL || item->type != MSGPACK_OBJECT_MAP) { + flb_plg_warn(ctx->ins, "Streamed Event 'object' not found"); + ret = -1; + goto msg_error; + } + + ret = process_event_object(ctx, event_type, item); + +msg_error: + flb_sds_destroy(event_type); + msgpack_unpacked_destroy(&result); + return ret; +} + +static int process_event_list(struct k8s_events *ctx, char *in_data, size_t in_size, + uint64_t *max_resource_version, flb_sds_t *continue_token) { int i; int ret = -1; @@ -375,16 +492,13 @@ static int process_events(struct k8s_events *ctx, char *in_data, size_t in_size, char *buf_data; size_t buf_size; size_t off = 0; - struct flb_time ts; - uint64_t resource_version; msgpack_unpacked result; msgpack_object root; msgpack_object k; msgpack_object *items = NULL; msgpack_object *item = NULL; - msgpack_object *item_metadata = NULL; msgpack_object *metadata = NULL; - + const flb_sds_t action = "ADDED"; /* All items from a k8s list we consider as 'ADDED' */ ret = flb_pack_json(in_data, in_size, &buf_data, &buf_size, &root_type, &consumed); if (ret == -1) { @@ -406,8 +520,9 @@ static int process_events(struct k8s_events *ctx, char *in_data, size_t in_size, return -1; } - // Traverse the EventList for the metadata (for the continue token) and the items. - // https://kubernetes.io/docs/reference/kubernetes-api/cluster-resources/event-v1/#EventList + /* Traverse the EventList for the metadata (for the continue token) and the items. + * https://kubernetes.io/docs/reference/kubernetes-api/cluster-resources/event-v1/#EventList + */ for (i = 0; i < root.via.map.size; i++) { k = root.via.map.ptr[i].key; if (k.type != MSGPACK_OBJECT_STR) { @@ -437,82 +552,29 @@ static int process_events(struct k8s_events *ctx, char *in_data, size_t in_size, } if (metadata == NULL) { - flb_plg_error(ctx->ins, "Cannot find metatada in response"); + flb_plg_error(ctx->ins, "Cannot find metadata in response"); goto msg_error; } - ret = record_get_field_sds(metadata, "continue", continue_token); + ret = record_get_field_uint64(metadata, "resourceVersion", max_resource_version); if (ret == -1) { - if (ret == -1) { - flb_plg_error(ctx->ins, "Cannot process continue token"); + flb_plg_error(ctx->ins, "Cannot find EventList resourceVersion"); goto msg_error; - } } - for (i = 0; i < items->via.array.size; i++) { - if (items->via.array.ptr[i].type != MSGPACK_OBJECT_MAP) { - flb_plg_warn(ctx->ins, "Event that is not a map"); - continue; - } - item_metadata = record_get_field_ptr(&items->via.array.ptr[i], "metadata"); - if (item_metadata == NULL) { - flb_plg_warn(ctx->ins, "Event without metadata"); - continue; - } - ret = record_get_field_uint64(item_metadata, - "resourceVersion", &resource_version); - if (ret == -1) { - continue; - } - if (resource_version > *max_resource_version) { - *max_resource_version = resource_version; - } + ret = record_get_field_sds(metadata, "continue", continue_token); + if (ret == -1) { + flb_plg_error(ctx->ins, "Cannot process continue token"); + goto msg_error; } - /* reset the log encoder */ - flb_log_event_encoder_reset(ctx->encoder); - - /* print every item from the items array */ for (i = 0; i < items->via.array.size; i++) { item = &items->via.array.ptr[i]; if (item->type != MSGPACK_OBJECT_MAP) { flb_plg_error(ctx->ins, "Cannot unpack item in response"); goto msg_error; } - - /* get event timestamp */ - ret = item_get_timestamp(item, &ts); - if (ret == FLB_FALSE) { - flb_plg_error(ctx->ins, "cannot retrieve event timestamp"); - goto msg_error; - } - - if (check_event_is_filtered(ctx, item, (time_t *) &ts) == FLB_TRUE) { - continue; - } - -#ifdef FLB_HAVE_SQLDB - if (ctx->db) { - k8s_events_sql_insert_event(ctx, item); - } -#endif - - /* encode content as a log event */ - flb_log_event_encoder_begin_record(ctx->encoder); - flb_log_event_encoder_set_timestamp(ctx->encoder, &ts); - - ret = flb_log_event_encoder_set_body_from_msgpack_object(ctx->encoder, item); - if (ret == FLB_EVENT_ENCODER_SUCCESS) { - ret = flb_log_event_encoder_commit_record(ctx->encoder); - } else { - flb_plg_warn(ctx->ins, "unable to encode: %llu", resource_version); - } - } - - if (ctx->encoder->output_length > 0) { - flb_input_log_append(ctx->ins, NULL, 0, - ctx->encoder->output_buffer, - ctx->encoder->output_length); + process_event_object(ctx, action, item); } msg_error: @@ -523,23 +585,45 @@ static int process_events(struct k8s_events *ctx, char *in_data, size_t in_size, return ret; } -static struct flb_http_client *make_event_api_request(struct k8s_events *ctx, - struct flb_connection *u_conn, - flb_sds_t continue_token) +static struct flb_http_client *make_event_watch_api_request(struct k8s_events *ctx, + uint64_t max_resource_version) { flb_sds_t url; struct flb_http_client *c; + if (ctx->namespace == NULL) { + url = flb_sds_create(K8S_EVENTS_KUBE_API_URI); + } + else { + url = flb_sds_create_size(strlen(K8S_EVENTS_KUBE_NAMESPACE_API_URI) + + strlen(ctx->namespace)); + flb_sds_printf(&url, K8S_EVENTS_KUBE_NAMESPACE_API_URI, ctx->namespace); + } + + flb_sds_printf(&url, "?watch=1&resourceVersion=%llu", max_resource_version); + flb_plg_info(ctx->ins, "Requesting %s", 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, + 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); } if (ctx->namespace == NULL) { url = flb_sds_create(K8S_EVENTS_KUBE_API_URI); - } else { - url = flb_sds_create_size(strlen(K8S_EVENTS_KUBE_NAMESPACE_API_URI) + + } + else { + url = flb_sds_create_size(strlen(K8S_EVENTS_KUBE_NAMESPACE_API_URI) + strlen(ctx->namespace)); flb_sds_printf(&url, K8S_EVENTS_KUBE_NAMESPACE_API_URI, ctx->namespace); } @@ -551,7 +635,7 @@ static struct flb_http_client *make_event_api_request(struct k8s_events *ctx, } 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; @@ -641,8 +725,8 @@ static int k8s_events_sql_insert_event(struct k8s_events *ctx, msgpack_object *i } flb_plg_debug(ctx->ins, - "inserted k8s event: uid=%s, resource_version=%llu, last=%ld", - uid, resource_version, last); + "inserted k8s event: uid=%s, resource_version=%llu, last=%llu", + uid, resource_version, flb_time_to_nanosec(&last)); sqlite3_clear_bindings(ctx->stmt_insert_kubernetes_event); sqlite3_reset(ctx->stmt_insert_kubernetes_event); @@ -652,66 +736,115 @@ static int k8s_events_sql_insert_event(struct k8s_events *ctx, msgpack_object *i #endif -static int k8s_events_collect(struct flb_input_instance *ins, - struct flb_config *config, void *in_context) +static int process_http_chunk(struct k8s_events* ctx, struct flb_http_client *c, + size_t *bytes_consumed) { - int ret; - size_t b_sent; - struct flb_connection *u_conn = NULL; - struct flb_http_client *c = NULL; - struct k8s_events *ctx = in_context; + int ret = 0; + int root_type; + size_t consumed = 0; + char *buf_data = NULL; + size_t buf_size; + size_t token_size = 0; + char *token_start = 0; + char *token_end = NULL; + + token_start = c->resp.payload; + token_end = strpbrk(token_start, JSON_ARRAY_DELIM); + while ( token_end != NULL && ret == 0 ) { + token_size = token_end - token_start; + ret = flb_pack_json(token_start, token_size, &buf_data, &buf_size, &root_type, &consumed); + if (ret == -1) { + flb_plg_debug(ctx->ins, "could not process payload, incomplete or bad formed JSON: %s", + c->resp.payload); + } + else { + *bytes_consumed += token_size + 1; + ret = process_watched_event(ctx, buf_data, buf_size); + } + + flb_free(buf_data); + if (buf_data) { + buf_data = NULL; + } + token_start = token_end+1; + token_end = strpbrk(token_start, JSON_ARRAY_DELIM); + } + + if (buf_data) { + flb_free(buf_data); + } + return ret; +} + +static void initialize_http_client(struct flb_http_client* c, struct k8s_events* ctx) +{ + flb_http_buffer_size(c, 0); + + flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10); + if (ctx->auth_len > 0) { + flb_http_add_header(c, "Authorization", 13, ctx->auth, ctx->auth_len); + } +} + +static int check_and_init_stream(struct k8s_events *ctx) +{ + /* Returns FLB_TRUE if stream has been initialized */ flb_sds_t continue_token = NULL; uint64_t max_resource_version = 0; + 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_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; } - flb_http_buffer_size(c, 0); - - flb_http_add_header(c, "User-Agent", 10, "Fluent-Bit", 10); - if (ctx->auth_len > 0) { - flb_http_add_header(c, "Authorization", 13, ctx->auth, ctx->auth_len); - } - + 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) { - ret = process_events(ctx, c->resp.payload, c->resp.payload_size, &max_resource_version, &continue_token); + if (c->resp.status == 200 && c->resp.payload_size > 0) { + ret = process_event_list(ctx, c->resp.payload, c->resp.payload_size, + &max_resource_version, &continue_token); } - else { + else + { if (c->resp.payload_size > 0) { flb_plg_error(ctx->ins, "http_status=%i:\n%s", c->resp.status, c->resp.payload); } else { flb_plg_error(ctx->ins, "http_status=%i", c->resp.status); } + goto failure; } flb_http_client_destroy(c); c = NULL; @@ -722,14 +855,88 @@ static int k8s_events_collect(struct flb_input_instance *ins, ctx->last_resource_version = max_resource_version; } -exit: - pthread_mutex_unlock(&ctx->lock); + /* Now that we've done a full list, we can use the resource version and do a watch + * to stream updates efficiently + */ + 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(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(ctx->streaming_client, &b_sent); + if (ret != 0) { + flb_plg_error(ctx->ins, "http do request error"); + goto failure; + } + + return FLB_TRUE; + +failure: if (c) { flb_http_client_destroy(c); } - if (u_conn) { - flb_upstream_conn_release(u_conn); + 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(ctx->streaming_client, bytes_consumed); + bytes_consumed = 0; + 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 (ctx->streaming_client->resp.status != 200 || ret == FLB_HTTP_ERROR) { + if (ret == FLB_HTTP_ERROR) { + flb_plg_warn(ins, "kubernetes chunked stream error."); + } + else { + 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; + } + + pthread_mutex_unlock(&ctx->lock); FLB_INPUT_RETURN(0); } @@ -863,12 +1070,6 @@ static struct flb_config_map config_map[] = { "kubernetes namespace to get events from, gets event from all namespaces by default." }, - { - FLB_CONFIG_MAP_STR, "timestamp_key", NULL, - 0, FLB_TRUE, offsetof(struct k8s_events, timestamp_key), - "Deprecated. To be removed in v3.0" - }, - #ifdef FLB_HAVE_SQLDB { FLB_CONFIG_MAP_STR, "db", NULL, diff --git a/plugins/in_kubernetes_events/kubernetes_events.h b/plugins/in_kubernetes_events/kubernetes_events.h index a4bcbf19d11..734b63a2afb 100644 --- a/plugins/in_kubernetes_events/kubernetes_events.h +++ b/plugins/in_kubernetes_events/kubernetes_events.h @@ -74,9 +74,6 @@ struct k8s_events { struct flb_log_event_encoder *encoder; - /* timestamp key - deprecated, to be removed in v3.0 */ - flb_sds_t timestamp_key; - /* record accessor */ struct flb_record_accessor *ra_resource_version; @@ -85,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 */ @@ -105,4 +105,4 @@ struct k8s_events { pthread_mutex_t lock; }; -#endif \ No newline at end of file +#endif 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} +}; +