diff options
Diffstat (limited to 'src/fluent-bit/plugins/in_kubernetes_events')
6 files changed, 1465 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/in_kubernetes_events/CMakeLists.txt b/src/fluent-bit/plugins/in_kubernetes_events/CMakeLists.txt new file mode 100644 index 000000000..b860a55e3 --- /dev/null +++ b/src/fluent-bit/plugins/in_kubernetes_events/CMakeLists.txt @@ -0,0 +1,5 @@ +set(src + kubernetes_events_conf.c + kubernetes_events.c) + +FLB_PLUGIN(in_kubernetes_events "${src}" "") diff --git a/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events.c b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events.c new file mode 100644 index 000000000..97719fba6 --- /dev/null +++ b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events.c @@ -0,0 +1,921 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2023 The Fluent Bit Authors + * + * 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 <sys/types.h> +#include <sys/stat.h> + +#include <fluent-bit/flb_input_plugin.h> +#include <fluent-bit/flb_network.h> +#include <fluent-bit/flb_pack.h> +#include <fluent-bit/flb_utils.h> +#include <fluent-bit/flb_error.h> +#include <fluent-bit/flb_compat.h> +#include <fluent-bit/flb_ra_key.h> +#include <fluent-bit/flb_time.h> +#include <fluent-bit/flb_strptime.h> +#include <fluent-bit/flb_parser.h> +#include <fluent-bit/flb_log_event_encoder.h> +#include <fluent-bit/flb_compat.h> + +#include "kubernetes_events.h" +#include "kubernetes_events_conf.h" + +#ifdef FLB_HAVE_SQLDB +#include "kubernetes_events_sql.h" +static int k8s_events_sql_insert_event(struct k8s_events *ctx, msgpack_object *item); +#endif + +static int file_to_buffer(const char *path, + char **out_buf, size_t *out_size) +{ + int ret; + int len; + char *buf; + ssize_t bytes; + FILE *fp; + struct stat st; + + if (!(fp = fopen(path, "r"))) { + return -1; + } + + ret = stat(path, &st); + if (ret == -1) { + flb_errno(); + fclose(fp); + return -1; + } + + buf = flb_calloc(1, (st.st_size + 1)); + if (!buf) { + flb_errno(); + fclose(fp); + return -1; + } + + bytes = fread(buf, st.st_size, 1, fp); + if (bytes < 1) { + flb_free(buf); + fclose(fp); + return -1; + } + + fclose(fp); + + // trim new lines + for (len = st.st_size; len > 0; len--) { + if (buf[len-1] != '\n' && buf[len-1] != '\r') { + break; + } + } + buf[len] = '\0'; + + *out_buf = buf; + *out_size = len; + + return 0; +} + +/* Set K8s Authorization Token and get HTTP Auth Header */ +static int get_http_auth_header(struct k8s_events *ctx) +{ + int ret; + char *temp; + char *tk = NULL; + size_t tk_size = 0; + + if (!ctx->token_file || strlen(ctx->token_file) == 0) { + return 0; + } + + ret = file_to_buffer(ctx->token_file, &tk, &tk_size); + if (ret == -1) { + flb_plg_warn(ctx->ins, "cannot open %s", ctx->token_file); + return -1; + } + ctx->token_created = time(NULL); + + /* Token */ + if (ctx->token != NULL) { + flb_free(ctx->token); + } + ctx->token = tk; + ctx->token_len = tk_size; + + /* HTTP Auth Header */ + if (ctx->auth == NULL) { + ctx->auth = flb_malloc(tk_size + 32); + } + else if (ctx->auth_len < tk_size + 32) { + temp = flb_realloc(ctx->auth, tk_size + 32); + if (temp == NULL) { + flb_errno(); + flb_free(ctx->auth); + ctx->auth = NULL; + return -1; + } + ctx->auth = temp; + } + + if (!ctx->auth) { + return -1; + } + + ctx->auth_len = snprintf(ctx->auth, tk_size + 32, "Bearer %s", tk); + return 0; +} + +/* Refresh HTTP Auth Header if K8s Authorization Token is expired */ +static int refresh_token_if_needed(struct k8s_events *ctx) +{ + int expired = FLB_FALSE; + int ret; + + if (!ctx->token_file || strlen(ctx->token_file) == 0) { + return 0; + } + + if (ctx->token_created > 0) { + if (time(NULL) > ctx->token_created + ctx->token_ttl) { + expired = FLB_TRUE; + } + } + + if (expired || ctx->token_created == 0) { + ret = get_http_auth_header(ctx); + if (ret == -1) { + return -1; + } + } + + return 0; +} +static int timestamp_lookup(struct k8s_events *ctx, char *ts, struct flb_time *time) +{ + struct flb_tm tm = { 0 }; + + if (flb_strptime(ts, "%Y-%m-%dT%H:%M:%SZ", &tm) == NULL) { + return -1; + } + + time->tm.tv_sec = flb_parser_tm2time(&tm); + time->tm.tv_nsec = 0; + + return 0; +} + +static msgpack_object *record_get_field_ptr(msgpack_object *obj, const char *fieldname) +{ + int i; + msgpack_object *k; + msgpack_object *v; + + if (obj->type != MSGPACK_OBJECT_MAP) { + return NULL; + } + + for (i = 0; i < obj->via.map.size; i++) { + k = &obj->via.map.ptr[i].key; + if (k->type != MSGPACK_OBJECT_STR) { + continue; + } + + if (strncmp(k->via.str.ptr, fieldname, strlen(fieldname)) == 0) { + v = &obj->via.map.ptr[i].val; + return v; + } + } + return NULL; +} + +static int record_get_field_sds(msgpack_object *obj, const char *fieldname, flb_sds_t *val) +{ + msgpack_object *v; + + v = record_get_field_ptr(obj, fieldname); + if (v == NULL) { + return 0; + } + if (v->type != MSGPACK_OBJECT_STR) { + return -1; + } + + *val = flb_sds_create_len(v->via.str.ptr, v->via.str.size); + return 0; +} + +static int record_get_field_time(msgpack_object *obj, const char *fieldname, time_t *val) +{ + msgpack_object *v; + struct flb_tm tm = { 0 }; + + v = record_get_field_ptr(obj, fieldname); + if (v == NULL) { + return -1; + } + if (v->type != MSGPACK_OBJECT_STR) { + return -1; + } + + if (flb_strptime(v->via.str.ptr, "%Y-%m-%dT%H:%M:%SZ", &tm) == NULL) { + return -2; + } + + *val = mktime(&tm.tm); + return 0; +} + +static int record_get_field_uint64(msgpack_object *obj, const char *fieldname, uint64_t *val) +{ + msgpack_object *v; + char *end; + + v = record_get_field_ptr(obj, fieldname); + if (v == NULL) { + return -1; + } + + // 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)) { + return -1; + } + return 0; + } + if (v->type == MSGPACK_OBJECT_POSITIVE_INTEGER) { + *val = v->via.u64; + return 0; + } + if (v->type == MSGPACK_OBJECT_NEGATIVE_INTEGER) { + *val = (uint64_t)v->via.i64; + return 0; + } + return -1; +} + +static int item_get_timestamp(msgpack_object *obj, time_t *event_time) +{ + int ret; + msgpack_object *metadata; + + // 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; + } + + ret = record_get_field_time(obj, "firstTimestamp", event_time); + if (ret != -1) { + return FLB_TRUE; + } + + metadata = record_get_field_ptr(obj, "metadata"); + if (metadata == NULL) { + return FLB_FALSE; + } + + ret = record_get_field_time(metadata, "creationTimestamp", event_time); + if (ret != -1) { + return FLB_TRUE; + } + + return FLB_FALSE; +} + +static bool check_event_is_filtered(struct k8s_events *ctx, msgpack_object *obj) +{ + int ret; + time_t event_time; + time_t now; + msgpack_object *metadata; + flb_sds_t uid; + uint64_t resource_version; + + ret = item_get_timestamp(obj, &event_time); + if (ret == -FLB_FALSE) { + flb_plg_error(ctx->ins, "Cannot get timestamp for item in response"); + return FLB_FALSE; + } + + now = (time_t)(cfl_time_now() / 1000000000); + if (event_time < (now - ctx->retention_time)) { + flb_plg_debug(ctx->ins, "Item is older than retention_time: %ld < %ld", + event_time, (now - ctx->retention_time)); + return FLB_TRUE; + } + + metadata = record_get_field_ptr(obj, "metadata"); + if (metadata == NULL) { + flb_plg_error(ctx->ins, "Cannot unpack item metadata in response"); + return FLB_FALSE; + } + + ret = record_get_field_uint64(metadata, "resourceVersion", &resource_version); + if (ret == -1) { + flb_plg_error(ctx->ins, "Cannot get resourceVersion for item in response"); + return FLB_FALSE; + } + + ret = record_get_field_sds(metadata, "uid", &uid); + if (ret == -1) { + flb_plg_error(ctx->ins, "Cannot get resourceVersion for item in response"); + return FLB_FALSE; + } + + +#ifdef FLB_HAVE_SQLDB + bool exists; + + + if (ctx->db) { + sqlite3_bind_text(ctx->stmt_get_kubernetes_event_exists_by_uid, + 1, uid, -1, NULL); + ret = sqlite3_step(ctx->stmt_get_kubernetes_event_exists_by_uid); + if (ret != SQLITE_ROW) { + if (ret != SQLITE_DONE) { + flb_plg_error(ctx->ins, "cannot execute kubernetes event exists"); + } + sqlite3_clear_bindings(ctx->stmt_get_kubernetes_event_exists_by_uid); + sqlite3_reset(ctx->stmt_get_kubernetes_event_exists_by_uid); + flb_sds_destroy(uid); + return FLB_FALSE; + } + + exists = sqlite3_column_int64(ctx->stmt_get_kubernetes_event_exists_by_uid, 0); + + flb_plg_debug(ctx->ins, "is_filtered: uid=%s exists=%d", uid, exists); + sqlite3_clear_bindings(ctx->stmt_get_kubernetes_event_exists_by_uid); + sqlite3_reset(ctx->stmt_get_kubernetes_event_exists_by_uid); + flb_sds_destroy(uid); + + return exists > 0 ? FLB_TRUE : FLB_FALSE; + } +#endif + + // 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: %lu (< %lu)", resource_version, + ctx->last_resource_version); + flb_sds_destroy(uid); + return FLB_TRUE; + } + + flb_sds_destroy(uid); + 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) +{ + int i; + int ret = -1; + int root_type; + size_t consumed = 0; + char *buf_data; + size_t buf_size; + size_t off = 0; + struct flb_time ts; + struct flb_ra_value *rval; + 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; + + + ret = flb_pack_json(in_data, in_size, &buf_data, &buf_size, &root_type, &consumed); + if (ret == -1) { + flb_plg_error(ctx->ins, "could not process payload, incomplete or bad formed JSON"); + goto json_error; + } + + /* 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"); + goto unpack_error; + } + + /* lookup the items array */ + root = result.data; + if (root.type != MSGPACK_OBJECT_MAP) { + 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 + for (i = 0; i < root.via.map.size; i++) { + k = root.via.map.ptr[i].key; + if (k.type != MSGPACK_OBJECT_STR) { + continue; + } + + if (strncmp(k.via.str.ptr, "items", 5) == 0) { + items = &root.via.map.ptr[i].val; + if (items->type != MSGPACK_OBJECT_ARRAY) { + flb_plg_error(ctx->ins, "Cannot unpack items"); + goto msg_error; + } + } + + if (strncmp(k.via.str.ptr, "metadata", 8) == 0) { + metadata = &root.via.map.ptr[i].val; + if (metadata->type != MSGPACK_OBJECT_MAP) { + flb_plg_error(ctx->ins, "Cannot unpack metadata"); + goto msg_error; + } + } + } + + if (items == NULL) { + flb_plg_error(ctx->ins, "Cannot find items in response"); + goto msg_error; + } + + if (metadata == NULL) { + flb_plg_error(ctx->ins, "Cannot find metatada in response"); + goto msg_error; + } + + ret = record_get_field_sds(metadata, "continue", continue_token); + if (ret == -1) { + if (ret == -1) { + flb_plg_error(ctx->ins, "Cannot process continue token"); + 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; + } + } + + /* 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; + } + + if (check_event_is_filtered(ctx, item) == FLB_TRUE) { + continue; + } + +#ifdef FLB_HAVE_SQLDB + if (ctx->db) { + k8s_events_sql_insert_event(ctx, item); + } +#endif + + /* get event timestamp */ + rval = flb_ra_get_value_object(ctx->ra_timestamp, *item); + if (!rval || rval->type != FLB_RA_STRING) { + flb_plg_error(ctx->ins, "cannot retrieve event timestamp"); + goto msg_error; + } + + /* convert timestamp */ + ret = timestamp_lookup(ctx, rval->val.string, &ts); + if (ret == -1) { + flb_plg_error(ctx->ins, "cannot lookup event timestamp"); + flb_ra_key_value_destroy(rval); + goto msg_error; + } + + /* 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: %lu", resource_version); + } + flb_ra_key_value_destroy(rval); + } + + if (ctx->encoder->output_length > 0) { + flb_input_log_append(ctx->ins, NULL, 0, + ctx->encoder->output_buffer, + ctx->encoder->output_length); + } + +msg_error: + msgpack_unpacked_destroy(&result); +unpack_error: + flb_free(buf_data); +json_error: + 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) +{ + 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, + 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) + + strlen(ctx->namespace)); + flb_sds_printf(&url, K8S_EVENTS_KUBE_NAMESPACE_API_URI, ctx->namespace); + } + + flb_sds_cat_safe(&url, "?", 1); + if (ctx->limit_request) { + if (continue_token != NULL) { + flb_sds_printf(&url, "continue=%s&", continue_token); + } + flb_sds_printf(&url, "limit=%d", ctx->limit_request); + } + c = flb_http_client(u_conn, FLB_HTTP_GET, url, + NULL, 0, ctx->api_host, ctx->api_port, NULL, 0); + flb_sds_destroy(url); + return c; +} + +#ifdef FLB_HAVE_SQLDB + +static int k8s_events_cleanup_db(struct flb_input_instance *ins, + struct flb_config *config, void *in_context) +{ + int ret; + struct k8s_events *ctx = (struct k8s_events *)in_context; + time_t retention_time_ago; + time_t now = (cfl_time_now() / 1000000000); + + if (ctx->db == NULL) { + FLB_INPUT_RETURN(0); + } + + retention_time_ago = now - (ctx->retention_time); + sqlite3_bind_int64(ctx->stmt_delete_old_kubernetes_events, + 1, (int64_t)retention_time_ago); + ret = sqlite3_step(ctx->stmt_delete_old_kubernetes_events); + if (ret != SQLITE_ROW && ret != SQLITE_DONE) { + flb_plg_error(ctx->ins, "cannot execute delete old kubernetes events"); + } + + sqlite3_clear_bindings(ctx->stmt_delete_old_kubernetes_events); + sqlite3_reset(ctx->stmt_delete_old_kubernetes_events); + + FLB_INPUT_RETURN(0); +} + +static int k8s_events_sql_insert_event(struct k8s_events *ctx, msgpack_object *item) +{ + int ret; + uint64_t resource_version; + time_t last; + msgpack_object *meta; + flb_sds_t uid; + + + meta = record_get_field_ptr(item, "meta"); + if (meta == NULL) { + flb_plg_error(ctx->ins, "unable to find metadata to save event"); + return -1; + } + + ret = record_get_field_uint64(meta, "resourceVersion", &resource_version); + if (ret == -1) { + flb_plg_error(ctx->ins, "unable to find resourceVersion in metadata to save event"); + return -1; + } + + ret = record_get_field_sds(meta, "uid", &uid); + if (ret == -1) { + flb_plg_error(ctx->ins, "unable to find uid in metadata to save event"); + return -1; + } + + ret = item_get_timestamp(item, &last); + if (ret == -FLB_FALSE) { + flb_plg_error(ctx->ins, "Cannot get timestamp for item to save it"); + return -1; + } + + if (ret == -2) { + flb_plg_error(ctx->ins, "unable to parse lastTimestamp in item to save event"); + flb_sds_destroy(uid); + return -1; + } + + /* Bind parameters */ + sqlite3_bind_text(ctx->stmt_insert_kubernetes_event, 1, uid, -1, 0); + sqlite3_bind_int64(ctx->stmt_insert_kubernetes_event, 2, resource_version); + sqlite3_bind_int64(ctx->stmt_insert_kubernetes_event, 3, (int64_t)last); + + /* Run the insert */ + ret = sqlite3_step(ctx->stmt_insert_kubernetes_event); + if (ret != SQLITE_DONE) { + sqlite3_clear_bindings(ctx->stmt_insert_kubernetes_event); + sqlite3_reset(ctx->stmt_insert_kubernetes_event); + flb_plg_error(ctx->ins, "cannot execute insert kubernetes event %s inode=%lu", + uid, resource_version); + flb_sds_destroy(uid); + return -1; + } + + flb_plg_debug(ctx->ins, + "inserted k8s event: uid=%s, resource_version=%lu, last=%ld", + uid, resource_version, last); + sqlite3_clear_bindings(ctx->stmt_insert_kubernetes_event); + sqlite3_reset(ctx->stmt_insert_kubernetes_event); + + flb_sds_destroy(uid); + return flb_sqldb_last_id(ctx->db); +} + +#endif + +static int k8s_events_collect(struct flb_input_instance *ins, + struct flb_config *config, void *in_context) +{ + int ret; + size_t b_sent; + struct flb_connection *u_conn = NULL; + struct flb_http_client *c = NULL; + struct k8s_events *ctx = in_context; + flb_sds_t continue_token = NULL; + uint64_t max_resource_version = 0; + + if (pthread_mutex_trylock(&ctx->lock) != 0) { + FLB_INPUT_RETURN(0); + } + + u_conn = flb_upstream_conn_get(ctx->upstream); + if (!u_conn) { + flb_plg_error(ins, "upstream connection initialization error"); + goto exit; + } + + ret = refresh_token_if_needed(ctx); + if (ret == -1) { + flb_plg_error(ctx->ins, "failed to refresh token"); + goto exit; + } + + do { + c = make_event_api_request(ctx, u_conn, 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_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); + } + + ret = flb_http_do(c, &b_sent); + if (ret != 0) { + flb_plg_error(ins, "http do error"); + goto exit; + } + + if (c->resp.status == 200) { + ret = process_events(ctx, c->resp.payload, c->resp.payload_size, &max_resource_version, &continue_token); + } + 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); + } + } + flb_http_client_destroy(c); + c = NULL; + } while(continue_token != NULL); + + if (max_resource_version > ctx->last_resource_version) { + flb_plg_debug(ctx->ins, "set last resourceVersion=%lu", max_resource_version); + ctx->last_resource_version = max_resource_version; + } + +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); +} + +static int k8s_events_init(struct flb_input_instance *ins, + struct flb_config *config, void *data) +{ + struct k8s_events *ctx = NULL; + + ctx = k8s_events_conf_create(ins); + if (!ctx) { + return -1; + } + + ctx->coll_id = flb_input_set_collector_time(ins, + k8s_events_collect, + ctx->interval_sec, + ctx->interval_nsec, + config); + +#ifdef FLB_HAVE_SQLDB + if (ctx->db) { + ctx->coll_cleanup_id = flb_input_set_collector_time(ins, + k8s_events_cleanup_db, + ctx->interval_sec, + ctx->interval_nsec, + config); + } +#endif + + return 0; +} + +static int k8s_events_exit(void *data, struct flb_config *config) +{ + struct k8s_events *ctx = data; + + if (!ctx) { + return 0; + } + + k8s_events_conf_destroy(ctx); + return 0; +} + +/* Configuration properties map */ +static struct flb_config_map config_map[] = { + /* Full Kubernetes API server URL */ + { + FLB_CONFIG_MAP_STR, "kube_url", "https://kubernetes.default.svc", + 0, FLB_FALSE, 0, + "Kubernetes API server URL" + }, + + /* Refresh interval */ + { + FLB_CONFIG_MAP_INT, "interval_sec", DEFAULT_INTERVAL_SEC, + 0, FLB_TRUE, offsetof(struct k8s_events, interval_sec), + "Set the polling interval for each channel" + }, + { + FLB_CONFIG_MAP_INT, "interval_nsec", DEFAULT_INTERVAL_NSEC, + 0, FLB_TRUE, offsetof(struct k8s_events, interval_nsec), + "Set the polling interval for each channel (sub seconds)" + }, + + /* TLS: set debug 'level' */ + { + FLB_CONFIG_MAP_INT, "tls.debug", "0", + 0, FLB_TRUE, offsetof(struct k8s_events, tls_debug), + "set TLS debug level: 0 (no debug), 1 (error), " + "2 (state change), 3 (info) and 4 (verbose)" + }, + + /* TLS: enable verification */ + { + FLB_CONFIG_MAP_BOOL, "tls.verify", "true", + 0, FLB_TRUE, offsetof(struct k8s_events, tls_verify), + "enable or disable verification of TLS peer certificate" + }, + + /* TLS: set tls.vhost feature */ + { + FLB_CONFIG_MAP_STR, "tls.vhost", NULL, + 0, FLB_TRUE, offsetof(struct k8s_events, tls_vhost), + "set optional TLS virtual host" + }, + + /* Kubernetes TLS: CA file */ + { + FLB_CONFIG_MAP_STR, "kube_ca_file", K8S_EVENTS_KUBE_CA, + 0, FLB_TRUE, offsetof(struct k8s_events, tls_ca_file), + "Kubernetes TLS CA file" + }, + + /* Kubernetes TLS: CA certs path */ + { + FLB_CONFIG_MAP_STR, "kube_ca_path", NULL, + 0, FLB_TRUE, offsetof(struct k8s_events, tls_ca_path), + "Kubernetes TLS ca path" + }, + + /* Kubernetes Token file */ + { + FLB_CONFIG_MAP_STR, "kube_token_file", K8S_EVENTS_KUBE_TOKEN, + 0, FLB_TRUE, offsetof(struct k8s_events, token_file), + "Kubernetes authorization token file" + }, + + /* Kubernetes Token file TTL */ + { + FLB_CONFIG_MAP_TIME, "kube_token_ttl", "10m", + 0, FLB_TRUE, offsetof(struct k8s_events, token_ttl), + "kubernetes token ttl, until it is reread from the token file. Default: 10m" + }, + + { + FLB_CONFIG_MAP_INT, "kube_request_limit", "0", + 0, FLB_TRUE, offsetof(struct k8s_events, limit_request), + "kubernetes limit parameter for events query, no limit applied when set to 0" + }, + + { + FLB_CONFIG_MAP_TIME, "kube_retention_time", "1h", + 0, FLB_TRUE, offsetof(struct k8s_events, retention_time), + "kubernetes retention time for events. Default: 1h" + }, + + { + FLB_CONFIG_MAP_STR, "kube_namespace", NULL, + 0, FLB_TRUE, offsetof(struct k8s_events, namespace), + "kubernetes namespace to get events from, gets event from all namespaces by default." + }, + +#ifdef FLB_HAVE_SQLDB + { + FLB_CONFIG_MAP_STR, "db", NULL, + 0, FLB_FALSE, 0, + "set a database file to keep track of recorded kubernetes events." + }, + { + FLB_CONFIG_MAP_STR, "db.sync", "normal", + 0, FLB_FALSE, 0, + "set a database sync method. values: extra, full, normal and off." + }, +#endif + + /* EOF */ + {0} +}; + +/* Plugin reference */ +struct flb_input_plugin in_kubernetes_events_plugin = { + .name = "kubernetes_events", + .description = "Kubernetes Events", + .cb_init = k8s_events_init, + .cb_pre_run = NULL, + .cb_collect = k8s_events_collect, + .cb_flush_buf = NULL, + .cb_exit = k8s_events_exit, + .config_map = config_map, + .flags = FLB_INPUT_NET | FLB_INPUT_CORO | FLB_INPUT_THREADED +}; diff --git a/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events.h b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events.h new file mode 100644 index 000000000..3afd48570 --- /dev/null +++ b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events.h @@ -0,0 +1,106 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2023 The Fluent Bit Authors + * + * 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. + */ + +#ifndef FLB_IN_KUBERNETES_EVENTS_H +#define FLB_IN_KUBERNETES_EVENTS_H + +#include <fluent-bit/flb_input_plugin.h> +#include <fluent-bit/flb_info.h> +#include <fluent-bit/flb_upstream.h> +#include <fluent-bit/flb_record_accessor.h> +#include <fluent-bit/flb_sqldb.h> + +#define DEFAULT_INTERVAL_SEC "0" +#define DEFAULT_INTERVAL_NSEC "500000000" + +/* Filter context */ +struct k8s_events { + int coll_id; + int coll_cleanup_id; + int interval_sec; /* interval collection time (Second) */ + int interval_nsec; /* interval collection time (Nanosecond) */ + int retention_time; /* retention time limit, default 1 hour */ + + /* Configuration parameters */ + char *api_host; + int api_port; + int api_https; + int tls_debug; + int tls_verify; + int kube_token_ttl; + flb_sds_t namespace; + + /* API Server end point */ + char kube_url[1024]; + + /* TLS CA certificate file */ + char *tls_ca_path; + char *tls_ca_file; + + /* TLS virtual host (optional), set by configmap */ + flb_sds_t tls_vhost; + + /* Kubernetes Token from FLB_KUBE_TOKEN file */ + char *token_file; + char *token; + int token_ttl; + size_t token_len; + int token_created; + + /* Pre-formatted HTTP Authorization header value */ + char *auth; + size_t auth_len; + + int dns_retries; + int dns_wait_time; + + struct flb_tls *tls; + + struct flb_log_event_encoder *encoder; + + /* record accessor */ + struct flb_record_accessor *ra_timestamp; + struct flb_record_accessor *ra_resource_version; + + /* others */ + struct flb_config *config; + struct flb_upstream *upstream; + struct flb_input_instance *ins; + + /* limit for event queries */ + int limit_request; + /* last highest seen resource_version */ + uint64_t last_resource_version; + +#ifdef FLB_HAVE_SQLDB + /* State database */ + struct flb_sqldb *db; + int db_sync; + int db_locking; + flb_sds_t db_journal_mode; + sqlite3_stmt *stmt_get_kubernetes_event_exists_by_uid; + sqlite3_stmt *stmt_insert_kubernetes_event; + sqlite3_stmt *stmt_delete_old_kubernetes_events; +#endif + + /* concurrency lock */ + pthread_mutex_t lock; +}; + +#endif
\ No newline at end of file diff --git a/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_conf.c b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_conf.c new file mode 100644 index 000000000..4f67d8cfc --- /dev/null +++ b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_conf.c @@ -0,0 +1,326 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2023 The Fluent Bit Authors + * + * 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 "kubernetes_events_conf.h" + +#ifdef FLB_HAVE_SQLDB +#include "kubernetes_events_sql.h" + + +/* Open or create database required by tail plugin */ +static struct flb_sqldb *flb_kubernetes_event_db_open(const char *path, + struct flb_input_instance *in, + struct k8s_events *ctx, + struct flb_config *config) +{ + int ret; + char tmp[64]; + struct flb_sqldb *db; + + /* Open/create the database */ + db = flb_sqldb_open(path, in->name, config); + if (!db) { + return NULL; + } + + /* Create table schema if it don't exists */ + ret = flb_sqldb_query(db, SQL_CREATE_KUBERNETES_EVENTS, NULL, NULL); + if (ret != FLB_OK) { + flb_plg_error(ctx->ins, "db: could not create 'in_kubernetes_events' table"); + flb_sqldb_close(db); + return NULL; + } + + if (ctx->db_sync >= 0) { + snprintf(tmp, sizeof(tmp) - 1, SQL_PRAGMA_SYNC, + ctx->db_sync); + ret = flb_sqldb_query(db, tmp, NULL, NULL); + if (ret != FLB_OK) { + flb_plg_error(ctx->ins, "db could not set pragma 'sync'"); + flb_sqldb_close(db); + return NULL; + } + } + + if (ctx->db_locking == FLB_TRUE) { + ret = flb_sqldb_query(db, SQL_PRAGMA_LOCKING_MODE, NULL, NULL); + if (ret != FLB_OK) { + flb_plg_error(ctx->ins, "db: could not set pragma 'locking_mode'"); + flb_sqldb_close(db); + return NULL; + } + } + + if (ctx->db_journal_mode) { + snprintf(tmp, sizeof(tmp) - 1, SQL_PRAGMA_JOURNAL_MODE, + ctx->db_journal_mode); + ret = flb_sqldb_query(db, tmp, NULL, NULL); + if (ret != FLB_OK) { + flb_plg_error(ctx->ins, "db could not set pragma 'journal_mode'"); + flb_sqldb_close(db); + return NULL; + } + } + + return db; +} + +static int flb_kubernetes_event_db_close(struct flb_sqldb *db) +{ + flb_sqldb_close(db); + return 0; +} + +#endif + +static int network_init(struct k8s_events *ctx, struct flb_config *config) +{ + int io_type = FLB_IO_TCP; + + ctx->upstream = NULL; + + if (ctx->api_https == FLB_TRUE) { + if (!ctx->tls_ca_path && !ctx->tls_ca_file) { + ctx->tls_ca_file = flb_strdup(K8S_EVENTS_KUBE_CA); + } + + /* create a custom TLS context since we use user-defined certs */ + ctx->tls = flb_tls_create(FLB_TLS_CLIENT_MODE, + ctx->tls_verify, + ctx->tls_debug, + ctx->tls_vhost, + ctx->tls_ca_path, + ctx->tls_ca_file, + NULL, NULL, NULL); + if (!ctx->tls) { + return -1; + } + + io_type = FLB_IO_TLS; + } + + /* Create an Upstream context */ + ctx->upstream = flb_upstream_create(config, + ctx->api_host, + ctx->api_port, + io_type, + ctx->tls); + if (!ctx->upstream) { + flb_plg_error(ctx->ins, "network initialization failed"); + return -1; + } + + return 0; +} + +struct k8s_events *k8s_events_conf_create(struct flb_input_instance *ins) +{ + int off; + int ret; + const char *p; + const char *url; + const char *tmp; + struct k8s_events *ctx = NULL; + pthread_mutexattr_t attr; + + ctx = flb_calloc(1, sizeof(struct k8s_events)); + if (!ctx) { + flb_errno(); + return NULL; + } + ctx->ins = ins; + + pthread_mutexattr_init(&attr); + pthread_mutex_init(&ctx->lock, &attr); + + /* Load the config map */ + ret = flb_input_config_map_set(ins, (void *) ctx); + if (ret == -1) { + flb_free(ctx); + return NULL; + } + flb_input_set_context(ins, ctx); + + ctx->encoder = flb_log_event_encoder_create(FLB_LOG_EVENT_FORMAT_DEFAULT); + if (!ctx->encoder) { + flb_plg_error(ins, "could not initialize event encoder"); + k8s_events_conf_destroy(ctx); + return NULL; + } + + /* Record accessor pattern */ + ctx->ra_timestamp = flb_ra_create(K8S_EVENTS_RA_TIMESTAMP, FLB_TRUE); + if (!ctx->ra_timestamp) { + flb_plg_error(ctx->ins, + "could not create record accessor for metadata items"); + k8s_events_conf_destroy(ctx); + return NULL; + } + + ctx->ra_resource_version = flb_ra_create(K8S_EVENTS_RA_RESOURCE_VERSION, FLB_TRUE); + if (!ctx->ra_resource_version) { + flb_plg_error(ctx->ins, "could not create record accessor for resource version"); + k8s_events_conf_destroy(ctx); + return NULL; + } + + /* Get Kubernetes API server */ + url = flb_input_get_property("kube_url", ins); + if (!url) { + ctx->api_host = flb_strdup(K8S_EVENTS_KUBE_API_HOST); + ctx->api_port = K8S_EVENTS_KUBE_API_PORT; + ctx->api_https = FLB_TRUE; + } + else { + tmp = url; + + /* Check the protocol */ + if (strncmp(tmp, "http://", 7) == 0) { + off = 7; + ctx->api_https = FLB_FALSE; + } + else if (strncmp(tmp, "https://", 8) == 0) { + off = 8; + ctx->api_https = FLB_TRUE; + } + else { + k8s_events_conf_destroy(ctx); + return NULL; + } + + /* Get hostname and TCP port */ + p = url + off; + tmp = strchr(p, ':'); + if (tmp) { + ctx->api_host = flb_strndup(p, tmp - p); + tmp++; + ctx->api_port = atoi(tmp); + } + else { + ctx->api_host = flb_strdup(p); + ctx->api_port = K8S_EVENTS_KUBE_API_PORT; + } + } + snprintf(ctx->kube_url, sizeof(ctx->kube_url) - 1, + "%s://%s:%i", + ctx->api_https ? "https" : "http", + ctx->api_host, ctx->api_port); + + flb_plg_info(ctx->ins, "API server: %s", ctx->kube_url); + + /* network setup */ + ret = network_init(ctx, ins->config); + if (ret == -1) { + k8s_events_conf_destroy(ctx); + return NULL; + } + +#ifdef FLB_HAVE_SQLDB + /* Initialize database */ + tmp = flb_input_get_property("db", ins); + if (tmp) { + ctx->db = flb_kubernetes_event_db_open(tmp, ins, ctx, ins->config); + if (!ctx->db) { + flb_plg_error(ctx->ins, "could not open/create database"); + k8s_events_conf_destroy(ctx); + return NULL; + } + } + + if (ctx->db) { + ret = sqlite3_prepare_v2(ctx->db->handler, + SQL_KUBERNETES_EVENT_EXISTS_BY_UID, + -1, + &ctx->stmt_get_kubernetes_event_exists_by_uid, + 0); + if (ret != SQLITE_OK) { + flb_plg_error(ctx->ins, "error preparing database SQL statement: stmt_get_kubernetes_event_exists_by_uid"); + k8s_events_conf_destroy(ctx); + return NULL; + } + + ret = sqlite3_prepare_v2(ctx->db->handler, + SQL_INSERT_KUBERNETES_EVENTS, + -1, + &ctx->stmt_insert_kubernetes_event, + 0); + if (ret != SQLITE_OK) { + flb_plg_error(ctx->ins, "error preparing database SQL statement: stmt_insert_kubernetes_event"); + k8s_events_conf_destroy(ctx); + return NULL; + } + + ret = sqlite3_prepare_v2(ctx->db->handler, + SQL_DELETE_OLD_KUBERNETES_EVENTS, + -1, + &ctx->stmt_delete_old_kubernetes_events, + 0); + if (ret != SQLITE_OK) { + flb_plg_error(ctx->ins, "error preparing database SQL statement: stmt_delete_old_kubernetes_events"); + k8s_events_conf_destroy(ctx); + return NULL; + } + } +#endif + + return ctx; +} + +void k8s_events_conf_destroy(struct k8s_events *ctx) +{ + if (ctx->ra_timestamp) { + flb_ra_destroy(ctx->ra_timestamp); + } + + if (ctx->ra_resource_version) { + flb_ra_destroy(ctx->ra_resource_version); + } + + if (ctx->upstream) { + flb_upstream_destroy(ctx->upstream); + } + + if (ctx->encoder) { + flb_log_event_encoder_destroy(ctx->encoder); + } + + if (ctx->api_host) { + flb_free(ctx->api_host); + } + if (ctx->token) { + flb_free(ctx->token); + } + if (ctx->auth) { + flb_free(ctx->auth); + } + +#ifdef FLB_HAVE_TLS + if (ctx->tls) { + flb_tls_destroy(ctx->tls); + } +#endif + +#ifdef FLB_HAVE_SQLDB + if (ctx->db) { + flb_kubernetes_event_db_close(ctx->db); + } +#endif + + flb_free(ctx); +} diff --git a/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_conf.h b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_conf.h new file mode 100644 index 000000000..9d6b54197 --- /dev/null +++ b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_conf.h @@ -0,0 +1,47 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2023 The Fluent Bit Authors + * + * 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. + */ + +#ifndef FLB_IN_KUBERNETES_EVENTS_CONF_H +#define FLB_IN_KUBERNETES_EVENTS_CONF_H + +#include <fluent-bit/flb_input_plugin.h> +#include <fluent-bit/flb_record_accessor.h> +#include <fluent-bit/flb_log_event_encoder.h> + +#include "kubernetes_events.h" + +/* Kubernetes API server info */ +#define K8S_EVENTS_KUBE_API_HOST "kubernetes.default.svc" +#define K8S_EVENTS_KUBE_API_PORT 443 +// /apis/events.k8s.io/v1/events +// /apis/events.k8s.io/v1/namespaces/{namespace}/events +#define K8S_EVENTS_KUBE_API_URI "/api/v1/events" +#define K8S_EVENTS_KUBE_NAMESPACE_API_URI "/api/v1/namespaces/%s/events" + +/* secrets */ +#define K8S_EVENTS_KUBE_TOKEN "/var/run/secrets/kubernetes.io/serviceaccount/token" +#define K8S_EVENTS_KUBE_CA "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt" + +#define K8S_EVENTS_RA_TIMESTAMP "$metadata['creationTimestamp']" +#define K8S_EVENTS_RA_RESOURCE_VERSION "$metadata['resourceVersion']" + +struct k8s_events *k8s_events_conf_create(struct flb_input_instance *ins); +void k8s_events_conf_destroy(struct k8s_events *ctx); + +#endif
\ No newline at end of file diff --git a/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_sql.h b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_sql.h new file mode 100644 index 000000000..3076791cc --- /dev/null +++ b/src/fluent-bit/plugins/in_kubernetes_events/kubernetes_events_sql.h @@ -0,0 +1,60 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2023 The Fluent Bit Authors + * + * 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. + */ + +#ifndef FLB_KUBERNETES_EVENTS_SQL_H +#define FLB_KUBERNETES_EVENTS_SQL_H + +/* + * In Fluent Bit we try to have a common convention for table names, + * if the table belongs to an input/output plugin, use the plugins name + * with the name of the object or type. + * + * in_kubernetes_events plugin table to track kubernetes events: + * in_kubernetes_events + */ +#define SQL_CREATE_KUBERNETES_EVENTS \ + "CREATE TABLE IF NOT EXISTS in_kubernetes_events (" \ + " id INTEGER PRIMARY KEY," \ + " uid TEXT NOT NULL," \ + " resourceVersion INTEGER NOT NULL," \ + " created INTEGER NOT NULL" \ + ");" + +#define SQL_KUBERNETES_EVENT_EXISTS_BY_UID \ + "SELECT COUNT(id) " \ + " FROM in_kubernetes_events " \ + " WHERE uid=@uid;" + +#define SQL_INSERT_KUBERNETES_EVENTS \ + "INSERT INTO in_kubernetes_events (uid, resourceVersion, created)" \ + " VALUES (@uid, @resourceVersion, @created);" + +#define SQL_DELETE_OLD_KUBERNETES_EVENTS \ + "DELETE FROM in_kubernetes_events WHERE created <= @createdBefore;" + +#define SQL_PRAGMA_SYNC \ + "PRAGMA synchronous=%i;" + +#define SQL_PRAGMA_JOURNAL_MODE \ + "PRAGMA journal_mode=%s;" + +#define SQL_PRAGMA_LOCKING_MODE \ + "PRAGMA locking_mode=EXCLUSIVE;" + +#endif |