diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-05 12:08:03 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2024-05-05 12:08:18 +0000 |
commit | 5da14042f70711ea5cf66e034699730335462f66 (patch) | |
tree | 0f6354ccac934ed87a2d555f45be4c831cf92f4a /src/fluent-bit/plugins/in_docker/docker.c | |
parent | Releasing debian version 1.44.3-2. (diff) | |
download | netdata-5da14042f70711ea5cf66e034699730335462f66.tar.xz netdata-5da14042f70711ea5cf66e034699730335462f66.zip |
Merging upstream version 1.45.3+dfsg.
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'src/fluent-bit/plugins/in_docker/docker.c')
-rw-r--r-- | src/fluent-bit/plugins/in_docker/docker.c | 560 |
1 files changed, 560 insertions, 0 deletions
diff --git a/src/fluent-bit/plugins/in_docker/docker.c b/src/fluent-bit/plugins/in_docker/docker.c new file mode 100644 index 000000000..135c9f6b4 --- /dev/null +++ b/src/fluent-bit/plugins/in_docker/docker.c @@ -0,0 +1,560 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2022 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 <fluent-bit/flb_info.h> +#include <fluent-bit/flb_input.h> +#include <fluent-bit/flb_input_plugin.h> +#include <fluent-bit/flb_config.h> +#include <fluent-bit/flb_pack.h> +#include <fluent-bit/flb_kv.h> + +#include <stdio.h> +#include <dirent.h> +#include <string.h> +#include <stdlib.h> +#include <stdint.h> + +#include "docker.h" + +static int cb_docker_collect(struct flb_input_instance *i_ins, + struct flb_config *config, void *in_context); + +docker_info *in_docker_init_docker_info(char *id) +{ + int len; + docker_info *docker; + + docker = flb_malloc(sizeof(docker_info)); + if (!docker) { + flb_errno(); + return NULL; + } + + len = strlen(id); + docker->id = flb_malloc(sizeof(char)*(len + 1)); + if (!docker->id) { + flb_errno(); + flb_free(docker); + return NULL; + } + strcpy(docker->id, id); + docker->id[len] = '\0'; + + return docker; +} + +static docker_snapshot *init_snapshot(char *id) +{ + int id_len; + docker_snapshot *snapshot; + + snapshot = (docker_snapshot *) flb_malloc(sizeof(docker_snapshot)); + if (!snapshot) { + flb_errno(); + return NULL; + } + + id_len = strlen(id) + 1; + snapshot->id = (char *) flb_malloc((id_len)*sizeof(char)); + if (!snapshot->id) { + flb_errno(); + flb_free(snapshot); + return NULL; + } + strcpy(snapshot->id, id); + + return snapshot; +} + +static bool is_exists(struct mk_list *list, char *id) +{ + int id_len; + char *cmp; + docker_info *item; + bool result = false; + struct mk_list *head; + + if (!list || !id) { + return result; + } + + mk_list_foreach(head, list) { + item = mk_list_entry(head, docker_info, _head); + + /* id could be of length 12 or 64 */ + id_len = strlen(item->id); + cmp = flb_calloc(id_len + 1, sizeof(char)); + if (!cmp) { + flb_errno(); + return NULL; + } + memcpy(cmp, id, id_len); + if (strcmp(item->id, cmp) == 0) { + result = true; + } + flb_free(cmp); + } + + return result; +} + +static void free_snapshots(struct mk_list *snaps); +/* Returns dockers CPU/Memory metrics. */ +static struct mk_list *get_docker_stats(struct flb_docker *ctx, struct mk_list *dockers) +{ + docker_snapshot *snapshot; + struct docker_info *docker; + struct mk_list *head; + struct mk_list *snapshots; + + if (!dockers) { + return NULL; + } + + snapshots = flb_malloc(sizeof(struct mk_list)); + if (!snapshots) { + flb_errno(); + return NULL; + } + + mk_list_init(snapshots); + mk_list_foreach(head, dockers) { + docker = mk_list_entry(head, docker_info, _head); + snapshot = init_snapshot(docker->id); + if (snapshot == NULL) { + free_snapshots(snapshots); + return NULL; + } + snapshot->name = ctx->cgroup_api.get_container_name(ctx, docker->id); + if (snapshot->name == NULL) { + free_snapshots(snapshots); + flb_free(snapshot->id); + flb_free(snapshot); + return NULL; + } + snapshot->cpu = ctx->cgroup_api.get_cpu_snapshot(ctx, docker->id); + if (snapshot->cpu == NULL) { + free_snapshots(snapshots); + flb_free(snapshot->name); + flb_free(snapshot->id); + flb_free(snapshot); + return NULL; + } + snapshot->mem = ctx->cgroup_api.get_mem_snapshot(ctx, docker->id); + if (snapshot->mem == NULL) { + free_snapshots(snapshots); + flb_free(snapshot->cpu); + flb_free(snapshot->name); + flb_free(snapshot->id); + flb_free(snapshot); + return NULL; + } + + mk_list_add(&snapshot->_head, snapshots); + } + + return snapshots; +} + +/* Returns a list of docker ids from space delimited string. */ +static struct mk_list *get_ids_from_str(char *space_delimited_str) +{ + struct mk_list *str_parts; + struct mk_list *parts_head; + struct mk_list *tmp; + struct flb_split_entry *part; + struct mk_list *dockers; + docker_info *docker; + + dockers = flb_malloc(sizeof(struct mk_list)); + if (!dockers) { + flb_errno(); + return NULL; + } + + mk_list_init(dockers); + str_parts = flb_utils_split(space_delimited_str, ' ', 256); + mk_list_foreach_safe(parts_head, tmp, str_parts) { + part = mk_list_entry(parts_head, struct flb_split_entry, _head); + if (part->len == DOCKER_LONG_ID_LEN + || part->len == DOCKER_SHORT_ID_LEN) { + docker = in_docker_init_docker_info(part->value); + mk_list_add(&docker->_head, dockers); + } + } + + flb_utils_split_free(str_parts); + return dockers; +} + +/* Initializes blacklist/whitelist. */ +static void init_filter_lists(struct flb_input_instance *f_ins, + struct flb_docker *ctx) +{ + struct mk_list *head; + struct flb_kv *kv; + + ctx->whitelist = NULL; + ctx->blacklist = NULL; + + /* Iterate all filter properties */ + mk_list_foreach(head, &f_ins->properties) { + kv = mk_list_entry(head, struct flb_kv, _head); + + if (strcasecmp(kv->key, "include") == 0) { + ctx->whitelist = get_ids_from_str(kv->val); + } + else if (strcasecmp(kv->key, "exclude") == 0) { + ctx->blacklist = get_ids_from_str(kv->val); + } + } +} + +/* Filters list of active dockers as per config. This returns a new list */ +static struct mk_list *apply_filters(struct flb_docker *ctx, + struct mk_list *dockers) +{ + struct mk_list *head; + struct mk_list *tmp; + docker_info *new; + docker_info *docker; + struct mk_list *filtered; + + if (ctx->whitelist == NULL && ctx->blacklist == NULL) { + return dockers; + } + + filtered = flb_malloc(sizeof(struct mk_list)); + if (!filtered) { + flb_errno(); + return NULL; + } + + mk_list_init(filtered); + + /* whitelist */ + mk_list_foreach_safe(head, tmp, dockers) { + docker = mk_list_entry(head, docker_info, _head); + if (ctx->whitelist == NULL) { + new = in_docker_init_docker_info(docker->id); + mk_list_add(&new->_head, filtered); + } + else { + if (is_exists(ctx->whitelist, docker->id)) { + new = in_docker_init_docker_info(docker->id); + mk_list_add(&new->_head, filtered); + } + } + } + + /* blacklist */ + if (ctx->blacklist != NULL) { + mk_list_foreach_safe(head, tmp, filtered) { + docker = mk_list_entry(head, docker_info, _head); + if (is_exists(ctx->blacklist, docker->id)) { + mk_list_del(&docker->_head); + flb_free(docker->id); + flb_free(docker); + } + } + } + + return filtered; +} + +/* Init Docker input */ +static int cb_docker_init(struct flb_input_instance *in, + struct flb_config *config, void *data) +{ + int ret; + struct flb_docker *ctx; + + /* Allocate space for the configuration */ + ctx = flb_calloc(1, sizeof(struct flb_docker)); + if (!ctx) { + flb_errno(); + return -1; + } + ctx->ins = in; + in_docker_set_cgroup_api_v1(&ctx->cgroup_api); /* TODO: support cgroup v2*/ + + init_filter_lists(in, ctx); + + /* Set the context */ + flb_input_set_context(in, ctx); + + /* Load the config map */ + ret = flb_input_config_map_set(in, (void *)ctx); + if (ret == -1) { + flb_free(ctx); + flb_plg_error(in, "unable to load configuration."); + return -1; + } + + if (ctx->interval_sec <= 0 && ctx->interval_nsec <= 0) { + ctx->interval_sec = atoi(DEFAULT_INTERVAL_SEC); + ctx->interval_nsec = atoi(DEFAULT_INTERVAL_NSEC); + } + + /* Set our collector based on time, CPU usage every 1 second */ + ret = flb_input_set_collector_time(in, + cb_docker_collect, ctx->interval_sec, + ctx->interval_nsec, config); + if (ret == -1) { + flb_plg_error(ctx->ins, "Could not set collector for Docker input plugin"); + flb_free(ctx); + return -1; + } + ctx->coll_fd = ret; + + ret = flb_log_event_encoder_init(&ctx->log_encoder, + FLB_LOG_EVENT_FORMAT_DEFAULT); + + if (ret != FLB_EVENT_ENCODER_SUCCESS) { + flb_plg_error(ctx->ins, "error initializing event encoder : %d", ret); + flb_free(ctx); + return -1; + } + + return ret; +} + +/* Flush snapshot as a message for output. */ +static void flush_snapshot(struct flb_docker *ctx, + struct flb_input_instance *i_ins, + docker_snapshot *snapshot) +{ + int result; + + if (!snapshot) { + return; + } + + result = flb_log_event_encoder_begin_record(&ctx->log_encoder); + + if (result == FLB_EVENT_ENCODER_SUCCESS) { + result = flb_log_event_encoder_set_current_timestamp( + &ctx->log_encoder); + } + + if (result == FLB_EVENT_ENCODER_SUCCESS) { + result = flb_log_event_encoder_append_body_values( + &ctx->log_encoder, + /* Docker ID [12 chars] */ + FLB_LOG_EVENT_CSTRING_VALUE("id"), + FLB_LOG_EVENT_STRING_VALUE(snapshot->id, DOCKER_SHORT_ID_LEN), + + /* Docker Name */ + FLB_LOG_EVENT_CSTRING_VALUE("name"), + FLB_LOG_EVENT_CSTRING_VALUE(snapshot->name), + + /* CPU used [nanoseconds] */ + FLB_LOG_EVENT_CSTRING_VALUE("cpu_used"), + FLB_LOG_EVENT_UINT32_VALUE(snapshot->cpu->used), + + /* Memory used [bytes] */ + FLB_LOG_EVENT_CSTRING_VALUE("mem_used"), + FLB_LOG_EVENT_UINT32_VALUE(snapshot->mem->used), + + /* Memory limit [bytes] */ + FLB_LOG_EVENT_CSTRING_VALUE("mem_limit"), + FLB_LOG_EVENT_UINT64_VALUE(snapshot->mem->limit)); + } + + if (result == FLB_EVENT_ENCODER_SUCCESS) { + result = flb_log_event_encoder_commit_record(&ctx->log_encoder); + } + + flb_trace("[in_docker] ID %s CPU %lu MEMORY %ld", snapshot->id, + snapshot->cpu->used, snapshot->mem->used); + + if (result == FLB_EVENT_ENCODER_SUCCESS) { + flb_input_log_append(i_ins, NULL, 0, + ctx->log_encoder.output_buffer, + ctx->log_encoder.output_length); + + } + else { + flb_plg_error(i_ins, "Error encoding record : %d", result); + } + + flb_log_event_encoder_reset(&ctx->log_encoder); +} + +static void flush_snapshots(struct flb_docker *ctx, + struct flb_input_instance *i_ins, + struct mk_list *snapshots) +{ + struct mk_list *head; + docker_snapshot *snapshot; + + mk_list_foreach(head, snapshots) { + snapshot = mk_list_entry(head, docker_snapshot, _head); + flush_snapshot(ctx, i_ins, snapshot); + } +} + +static void free_snapshots(struct mk_list *snaps) +{ + struct docker_snapshot *snap; + struct mk_list *tmp; + struct mk_list *head; + + if (snaps == NULL) { + return; + } + + mk_list_foreach_safe(head, tmp, snaps) { + snap = mk_list_entry(head, docker_snapshot, _head); + flb_free(snap->id); + flb_free(snap->name); + flb_free(snap->cpu); + flb_free(snap->mem); + flb_free(snap); + } + flb_free(snaps); +} + +static void free_docker_list(struct mk_list *dockers) +{ + struct mk_list *head; + struct mk_list *tmp; + struct docker_info *docker; + + if (dockers == NULL) { + return; + } + + mk_list_foreach_safe(head, tmp, dockers) { + docker = mk_list_entry(head, docker_info, _head); + flb_free(docker->id); + flb_free(docker); + } + flb_free(dockers); +} + +/* Callback to gather Docker CPU/Memory usage. */ +static int cb_docker_collect(struct flb_input_instance *ins, + struct flb_config *config, void *in_context) +{ + struct mk_list *active; + struct mk_list *filtered; + struct mk_list *snaps; + struct flb_docker *ctx = in_context; + (void) config; + + /* Get current active dockers. */ + active = ctx->cgroup_api.get_active_docker_ids(); + + filtered = apply_filters(ctx, active); + if (!filtered) { + free_docker_list(active); + return 0; + } + + /* Get Mem/CPU stats of dockers. */ + snaps = get_docker_stats(ctx, filtered); + if (!snaps) { + free_docker_list(active); + if (active != filtered) { + /* apply_filters can return the address of acive. + * In that case, filtered is already freed. + */ + free_docker_list(filtered); + } + return 0; + } + + flush_snapshots(ctx, ins, snaps); + + free_snapshots(snaps); + free_docker_list(active); + + if (ctx->whitelist != NULL || ctx->blacklist != NULL) { + free_docker_list(filtered); + } + + return 0; +} + +static void cb_docker_pause(void *data, struct flb_config *config) +{ + struct flb_docker *ctx = data; + flb_input_collector_pause(ctx->coll_fd, ctx->ins); +} + +static void cb_docker_resume(void *data, struct flb_config *config) +{ + struct flb_docker *ctx = data; + flb_input_collector_resume(ctx->coll_fd, ctx->ins); +} + +static int cb_docker_exit(void *data, struct flb_config *config) +{ + (void) *config; + struct flb_docker *ctx = data; + + /* done */ + flb_log_event_encoder_destroy(&ctx->log_encoder); + + free_docker_list(ctx->whitelist); + free_docker_list(ctx->blacklist); + flb_free(ctx); + + return 0; +} + +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_INT, "interval_sec", DEFAULT_INTERVAL_SEC, + 0, FLB_TRUE, offsetof(struct flb_docker, interval_sec), + "Set the collector interval" + }, + { + FLB_CONFIG_MAP_INT, "interval_nsec", DEFAULT_INTERVAL_NSEC, + 0, FLB_TRUE, offsetof(struct flb_docker, interval_nsec), + "Set the collector interval (nanoseconds)" + }, + { + FLB_CONFIG_MAP_STR, "include", NULL, + 0, FLB_FALSE, 0, + "A space-separated list of containers to include" + }, + { + FLB_CONFIG_MAP_STR, "exclude", NULL, + 0, FLB_FALSE, 0, + "A space-separated list of containers to exclude" + }, + /* EOF */ + {0} +}; + +/* Plugin reference */ +struct flb_input_plugin in_docker_plugin = { + .name = "docker", + .description = "Docker containers metrics", + .cb_init = cb_docker_init, + .cb_pre_run = NULL, + .cb_collect = cb_docker_collect, + .cb_flush_buf = NULL, + .cb_pause = cb_docker_pause, + .cb_resume = cb_docker_resume, + .cb_exit = cb_docker_exit, + .config_map = config_map +}; |