/src/fluent-bit/plugins/in_docker_events/docker_events.c

Source
/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2026 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_input_plugin.h>
#include <fluent-bit/flb_network.h>
#include <fluent-bit/flb_pack.h>
#include <sys/socket.h>
#include <sys/un.h>

#include "docker_events.h"
#include "docker_events_config.h"


/**
 * Creates the connection to docker's unix socket and sends the
 * HTTP GET /events
 *
 * @param ctx  Pointer to flb_in_de_config
 *
 * @return int 0 on success, -1 on failure
 */
static int de_unix_create(struct flb_in_de_config *ctx)
{
    ssize_t bytes;
    unsigned long len;
    size_t address_length;
    struct sockaddr_un address;
    char request[512];

    ctx->fd = flb_net_socket_create(AF_UNIX, FLB_FALSE);
    if (ctx->fd == -1) {
        return -1;
    }

    /* Prepare the unix socket path */
    len = strlen(ctx->unix_path);
    address.sun_family = AF_UNIX;
    sprintf(address.sun_path, "%s", ctx->unix_path);
    address_length = sizeof(address.sun_family) + len + 1;
    if (connect(ctx->fd, (struct sockaddr *)&address, address_length) == -1) {
        flb_errno();
        close(ctx->fd);
        return -1;
    }

    strcpy(request, "GET /events HTTP/1.0\r\n\r\n");
    flb_plg_trace(ctx->ins, "writing to socket %s", request);
    write(ctx->fd, request, strlen(request));

    /* Read the initial http response */
    bytes = read(ctx->fd, ctx->buf, ctx->buf_size - 1);
    if (bytes == -1) {
        flb_errno();
    }
    flb_plg_debug(ctx->ins, "read %zu bytes from socket", bytes);

    return 0;
}

static int in_de_collect(struct flb_input_instance *ins,
                         struct flb_config *config, void *in_context);

static int reconnect_docker_sock(struct flb_input_instance *ins,
                                 struct flb_config *config,
                                 struct flb_in_de_config *ctx)
{
    int ret;

    /* remove old socket collector */
    if (ctx->coll_id >= 0) {
        ret = flb_input_collector_delete(ctx->coll_id, ins);
        if (ret < 0) {
            flb_plg_error(ctx->ins, "failed to pause event");
            return -1;
        }
        ctx->coll_id = -1;
    }
    if (ctx->fd > 0) {
        flb_plg_debug(ctx->ins, "close socket fd=%d", ctx->fd);
        close(ctx->fd);
        ctx->fd = -1;
    }

    /* create socket again */
    if (de_unix_create(ctx) < 0) {
        flb_plg_error(ctx->ins, "failed to re-initialize socket");
        if (ctx->fd > 0) {
            flb_plg_debug(ctx->ins, "close socket fd=%d", ctx->fd);
            close(ctx->fd);
            ctx->fd = -1;
        }
        return -1;
    }
    /* set event */
    ctx->coll_id = flb_input_set_collector_event(ins,
                                                 in_de_collect,
                                                 ctx->fd, config);
    if (ctx->coll_id < 0) {
        flb_plg_error(ctx->ins,
                      "could not set collector for IN_DOCKER_EVENTS plugin");
        close(ctx->fd);
        ctx->fd = -1;
        return -1;
    }
    ret = flb_input_collector_start(ctx->coll_id, ins);
    if (ret < 0) {
        flb_plg_error(ctx->ins,
                      "could not start collector for IN_DOCKER_EVENTS plugin");
        flb_input_collector_delete(ctx->coll_id, ins);
        close(ctx->fd);
        ctx->coll_id = -1;
        ctx->fd = -1;
        return -1;
    }

    flb_plg_info(ctx->ins, "Reconnect successful");
    return 0;
}

static int cb_reconnect(struct flb_input_instance *ins,
                       struct flb_config *config,
                       void *in_context)
{
    struct flb_in_de_config *ctx = in_context;
    int ret;

    flb_plg_info(ctx->ins, "Retry(%d/%d)",
                 ctx->current_retries, ctx->reconnect_retry_limits);
    ret = reconnect_docker_sock(ins, config, ctx);
    if (ret < 0) {
        /* Failed to reconnect */
        ctx->current_retries++;
        if (ctx->current_retries > ctx->reconnect_retry_limits) {
            /* give up */
            flb_plg_error(ctx->ins, "Failed to retry. Giving up...");
            goto cb_reconnect_end;
        }
        flb_plg_info(ctx->ins, "Failed. Waiting for next retry..");
        return 0;
    }

 cb_reconnect_end:
    if(flb_input_collector_delete(ctx->retry_coll_id, ins) < 0) {
        flb_plg_error(ctx->ins, "failed to delete timer event");
    }
    ctx->current_retries = 0;
    ctx->retry_coll_id = -1;
    return ret;
}

static int create_reconnect_event(struct flb_input_instance *ins,
                                  struct flb_config *config,
                                  struct flb_in_de_config *ctx)
{
    int ret;

    if (ctx->retry_coll_id >= 0) {
        flb_plg_debug(ctx->ins, "already retring ?");
        return 0;
    }

    /* try before creating event to stop incoming event */
    ret = reconnect_docker_sock(ins, config, ctx);
    if (ret == 0) {        
        return 0;
    }

    ctx->current_retries = 1;
    ctx->retry_coll_id = flb_input_set_collector_time(ins,
                                                      cb_reconnect,
                                                      ctx->reconnect_retry_interval,
                                                      0,
                                                      config);
    if (ctx->retry_coll_id < 0) {
        flb_plg_error(ctx->ins, "failed to create timer event");
        return -1;
    }
    ret = flb_input_collector_start(ctx->retry_coll_id, ins);
    if (ret < 0) {
        flb_plg_error(ctx->ins, "failed to start timer event");
        flb_input_collector_delete(ctx->retry_coll_id, ins);
        ctx->retry_coll_id = -1;
        return -1;
    }
    flb_plg_info(ctx->ins, "create reconnect event. interval=%d second",
                 ctx->reconnect_retry_interval);

    return 0;
}

static int is_recoverable_error(int error)
{
    /* ENOTTY: 
          It reports on Docker in Docker mode.
          https://github.com/fluent/fluent-bit/issues/3439#issuecomment-831424674
     */
    if (error == ENOTTY || error == EBADF) {
        return FLB_TRUE;
    }
    return FLB_FALSE;
}


/**
 * Callback function to process events recieved on the unix
 * socket.
 *
 * @param ins           Pointer to flb_input_instance
 * @param config        Pointer to flb_config
 * @param in_context    void Pointer used to cast to
 *                      flb_in_de_config
 *
 * @return int Always returns success
 */
static int in_de_collect(struct flb_input_instance *ins,
                         struct flb_config *config, void *in_context)
{
    int ret = 0;
    int error;
    size_t str_len = 0;
    struct flb_in_de_config *ctx = in_context;

    /* variables for parser */
    int parser_ret = -1;
    void *out_buf = NULL;
    size_t out_size = 0;
    struct flb_time out_time;

    ret = read(ctx->fd, ctx->buf, ctx->buf_size - 1);

    if (ret > 0) {
        str_len = ret;
        ctx->buf[str_len] = '\0';

        ret = flb_log_event_encoder_begin_record(&ctx->log_encoder);

        if (!ctx->parser) {
            /* Initialize local msgpack buffer */
            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                ret = flb_log_event_encoder_set_current_timestamp(
                        &ctx->log_encoder);
            }

            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                ret = flb_log_event_encoder_append_body_values(
                        &ctx->log_encoder,
                        FLB_LOG_EVENT_CSTRING_VALUE(ctx->key),
                        FLB_LOG_EVENT_STRING_VALUE(ctx->buf, str_len));
            }

            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                ret = flb_log_event_encoder_commit_record(&ctx->log_encoder);
            }

            if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                flb_input_log_append(ins, NULL, 0,
                                     ctx->log_encoder.output_buffer,
                                     ctx->log_encoder.output_length);

            }
            else {
                flb_plg_error(ctx->ins, "Error encoding record : %d", ret);
            }
        }
        else {
            flb_time_get(&out_time);

            parser_ret = flb_parser_do(ctx->parser, ctx->buf, str_len - 1,
                                       &out_buf, &out_size, &out_time);
            if (parser_ret >= 0) {
                if (flb_time_to_nanosec(&out_time) == 0L) {
                    flb_time_get(&out_time);
                }

                if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                    ret = flb_log_event_encoder_set_timestamp(
                            &ctx->log_encoder,
                            &out_time);
                }

                if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                    ret = flb_log_event_encoder_set_body_from_raw_msgpack(
                            &ctx->log_encoder,
                            out_buf,
                            out_size);
                }

                if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                    ret = flb_log_event_encoder_commit_record(&ctx->log_encoder);
                }

                if (ret == FLB_EVENT_ENCODER_SUCCESS) {
                    flb_input_log_append(ins, NULL, 0,
                                         ctx->log_encoder.output_buffer,
                                         ctx->log_encoder.output_length);

                }
                else {
                    flb_plg_error(ctx->ins, "Error encoding record : %d", ret);
                }


                flb_free(out_buf);
            }
            else {
                flb_plg_trace(ctx->ins, "tried to parse: %s", ctx->buf);
                flb_plg_trace(ctx->ins, "buf_size %zu", ctx->buf_size);
                flb_plg_error(ctx->ins, "parser returned an error: %d",
                              parser_ret);
            }
        }

        flb_log_event_encoder_reset(&ctx->log_encoder);
    }
    else if (ret == 0) {
        /* EOF */

        /* docker service may be restarted */
        flb_plg_info(ctx->ins, "EOF detected. Re-initialize");
        if (ctx->reconnect_retry_limits > 0) {
            ret = create_reconnect_event(ins, config, ctx);
            if (ret < 0) {
                return ret;
            }
        }
    }
    else {
        error = errno;
        flb_plg_error(ctx->ins, "read returned error: %d, %s", error,
                      strerror(error));
        if (is_recoverable_error(error)) {
            if (ctx->reconnect_retry_limits > 0) {
                ret = create_reconnect_event(ins, config, ctx);
                if (ret < 0) {
                    return ret;
                }
            }
        }
    }

    return 0;
}

/**
 * Callback function to initialize docker events plugin
 *
 * @param ins     Pointer to flb_input_instance
 * @param config  Pointer to flb_config
 * @param data    Unused
 *
 * @return int 0 on success, -1 on failure
 */
static int in_de_init(struct flb_input_instance *ins,
                      struct flb_config *config, void *data)
{
    struct flb_in_de_config *ctx = NULL;
    (void) data;

    /* Allocate space for the configuration */
    ctx = de_config_init(ins, config);
    if (!ctx) {
        return -1;
    }
    ctx->ins = ins;
    ctx->retry_coll_id = -1;
    ctx->current_retries = 0;

    /* Set the context */
    flb_input_set_context(ins, ctx);

    if (de_unix_create(ctx) != 0) {
        flb_plg_error(ctx->ins, "could not listen on unix://%s",
                      ctx->unix_path);
        de_config_destroy(ctx);
        return -1;
    }

    ctx->coll_id = flb_input_set_collector_event(ins, in_de_collect,
                                                 ctx->fd, config);
    if(ctx->coll_id < 0){
        flb_plg_error(ctx->ins,
                      "could not set collector for IN_DOCKER_EVENTS plugin");
        de_config_destroy(ctx);
        return -1;
    }

    flb_plg_info(ctx->ins, "listening for events on %s", ctx->unix_path);
    return 0;
}

/**
 * Callback exit function to cleanup plugin
 *
 * @param data    Pointer cast to flb_in_de_config
 * @param config  Unused
 *
 * @return int    Always returns 0
 */
static int in_de_exit(void *data, struct flb_config *config)
{
    (void) config;
    struct flb_in_de_config *ctx = data;

    if (!ctx) {
        return 0;
    }

    de_config_destroy(ctx);

    return 0;
}

/* Configuration properties map */
static struct flb_config_map config_map[] = {
    {
     FLB_CONFIG_MAP_STR, "unix_path", DEFAULT_UNIX_SOCKET_PATH,
     0, FLB_TRUE, offsetof(struct flb_in_de_config, unix_path),
     "Define Docker unix socket path to read events"
    },
    {
     FLB_CONFIG_MAP_SIZE, "buffer_size", "8k",
     0, FLB_TRUE, offsetof(struct flb_in_de_config, buf_size),
     "Set buffer size to read events"
    },
    {
     FLB_CONFIG_MAP_STR, "parser", NULL,
      0, FLB_FALSE, 0,
     "Optional parser for records, if not set, records are packages under 'key'"
    },
    {
     FLB_CONFIG_MAP_STR, "key", DEFAULT_FIELD_NAME,
     0, FLB_TRUE, offsetof(struct flb_in_de_config, key),
     "Set the key name to store unparsed Docker events"
    },
    {
     FLB_CONFIG_MAP_INT, "reconnect.retry_limits", "5",
     0, FLB_TRUE, offsetof(struct flb_in_de_config, reconnect_retry_limits),
     "Maximum number to retry to connect docker socket"
    },
    {
     FLB_CONFIG_MAP_INT, "reconnect.retry_interval", "1",
     0, FLB_TRUE, offsetof(struct flb_in_de_config, reconnect_retry_interval),
     "Retry interval to connect docker socket"
    },
    /* EOF */
    {0}
};

/* Plugin reference */
struct flb_input_plugin in_docker_events_plugin = {
    .name         = "docker_events",
    .description  = "Docker events",
    .cb_init      = in_de_init,
    .cb_pre_run   = NULL,
    .cb_collect   = in_de_collect,
    .cb_flush_buf = NULL,
    .cb_exit      = in_de_exit,
    .config_map   = config_map,
    .flags        = FLB_INPUT_NET
};

Coverage Report

Created: 2026-08-09 07:14

Line	Count	Source
1		/* -- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -- */
2
3		/* Fluent Bit
4		* ==========
5		* Copyright (C) 2015-2026 The Fluent Bit Authors
6		*
7		* Licensed under the Apache License, Version 2.0 (the "License");
8		* you may not use this file except in compliance with the License.
9		* You may obtain a copy of the License at
10		*
11		* http://www.apache.org/licenses/LICENSE-2.0
12		*
13		* Unless required by applicable law or agreed to in writing, software
14		* distributed under the License is distributed on an "AS IS" BASIS,
15		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16		* See the License for the specific language governing permissions and
17		* limitations under the License.
18		*/
19
20		#include <fluent-bit/flb_input_plugin.h>
21		#include <fluent-bit/flb_network.h>
22		#include <fluent-bit/flb_pack.h>
23		#include <sys/socket.h>
24		#include <sys/un.h>
25
26		#include "docker_events.h"
27		#include "docker_events_config.h"
28
29
30		/**
31		* Creates the connection to docker's unix socket and sends the
32		* HTTP GET /events
33		*
34		* @param ctx Pointer to flb_in_de_config
35		*
36		* @return int 0 on success, -1 on failure
37		*/
38		static int de_unix_create(struct flb_in_de_config *ctx)
39	0	{
40	0	ssize_t bytes;
41	0	unsigned long len;
42	0	size_t address_length;
43	0	struct sockaddr_un address;
44	0	char request[512];
45
46	0	ctx->fd = flb_net_socket_create(AF_UNIX, FLB_FALSE);
47	0	if (ctx->fd == -1) {
48	0	return -1;
49	0	}
50
51		/* Prepare the unix socket path */
52	0	len = strlen(ctx->unix_path);
53	0	address.sun_family = AF_UNIX;
54	0	sprintf(address.sun_path, "%s", ctx->unix_path);
55	0	address_length = sizeof(address.sun_family) + len + 1;
56	0	if (connect(ctx->fd, (struct sockaddr *)&address, address_length) == -1) {
57	0	flb_errno();
58	0	close(ctx->fd);
59	0	return -1;
60	0	}
61
62	0	strcpy(request, "GET /events HTTP/1.0\r\n\r\n");
63	0	flb_plg_trace(ctx->ins, "writing to socket %s", request);
64	0	write(ctx->fd, request, strlen(request));
65
66		/* Read the initial http response */
67	0	bytes = read(ctx->fd, ctx->buf, ctx->buf_size - 1);
68	0	if (bytes == -1) {
69	0	flb_errno();
70	0	}
71	0	flb_plg_debug(ctx->ins, "read %zu bytes from socket", bytes);
72
73	0	return 0;
74	0	}
75
76		static int in_de_collect(struct flb_input_instance *ins,
77		struct flb_config config, void in_context);
78
79		static int reconnect_docker_sock(struct flb_input_instance *ins,
80		struct flb_config *config,
81		struct flb_in_de_config *ctx)
82	0	{
83	0	int ret;
84
85		/* remove old socket collector */
86	0	if (ctx->coll_id >= 0) {
87	0	ret = flb_input_collector_delete(ctx->coll_id, ins);
88	0	if (ret < 0) {
89	0	flb_plg_error(ctx->ins, "failed to pause event");
90	0	return -1;
91	0	}
92	0	ctx->coll_id = -1;
93	0	}
94	0	if (ctx->fd > 0) {
95	0	flb_plg_debug(ctx->ins, "close socket fd=%d", ctx->fd);
96	0	close(ctx->fd);
97	0	ctx->fd = -1;
98	0	}
99
100		/* create socket again */
101	0	if (de_unix_create(ctx) < 0) {
102	0	flb_plg_error(ctx->ins, "failed to re-initialize socket");
103	0	if (ctx->fd > 0) {
104	0	flb_plg_debug(ctx->ins, "close socket fd=%d", ctx->fd);
105	0	close(ctx->fd);
106	0	ctx->fd = -1;
107	0	}
108	0	return -1;
109	0	}
110		/* set event */
111	0	ctx->coll_id = flb_input_set_collector_event(ins,
112	0	in_de_collect,
113	0	ctx->fd, config);
114	0	if (ctx->coll_id < 0) {
115	0	flb_plg_error(ctx->ins,
116	0	"could not set collector for IN_DOCKER_EVENTS plugin");
117	0	close(ctx->fd);
118	0	ctx->fd = -1;
119	0	return -1;
120	0	}
121	0	ret = flb_input_collector_start(ctx->coll_id, ins);
122	0	if (ret < 0) {
123	0	flb_plg_error(ctx->ins,
124	0	"could not start collector for IN_DOCKER_EVENTS plugin");
125	0	flb_input_collector_delete(ctx->coll_id, ins);
126	0	close(ctx->fd);
127	0	ctx->coll_id = -1;
128	0	ctx->fd = -1;
129	0	return -1;
130	0	}
131
132	0	flb_plg_info(ctx->ins, "Reconnect successful");
133	0	return 0;
134	0	}
135
136		static int cb_reconnect(struct flb_input_instance *ins,
137		struct flb_config *config,
138		void *in_context)
139	0	{
140	0	struct flb_in_de_config *ctx = in_context;
141	0	int ret;
142
143	0	flb_plg_info(ctx->ins, "Retry(%d/%d)",
144	0	ctx->current_retries, ctx->reconnect_retry_limits);
145	0	ret = reconnect_docker_sock(ins, config, ctx);
146	0	if (ret < 0) {
147		/* Failed to reconnect */
148	0	ctx->current_retries++;
149	0	if (ctx->current_retries > ctx->reconnect_retry_limits) {
150		/* give up */
151	0	flb_plg_error(ctx->ins, "Failed to retry. Giving up...");
152	0	goto cb_reconnect_end;
153	0	}
154	0	flb_plg_info(ctx->ins, "Failed. Waiting for next retry..");
155	0	return 0;
156	0	}
157
158	0	cb_reconnect_end:
159	0	if(flb_input_collector_delete(ctx->retry_coll_id, ins) < 0) {
160	0	flb_plg_error(ctx->ins, "failed to delete timer event");
161	0	}
162	0	ctx->current_retries = 0;
163	0	ctx->retry_coll_id = -1;
164	0	return ret;
165	0	}
166
167		static int create_reconnect_event(struct flb_input_instance *ins,
168		struct flb_config *config,
169		struct flb_in_de_config *ctx)
170	0	{
171	0	int ret;
172
173	0	if (ctx->retry_coll_id >= 0) {
174	0	flb_plg_debug(ctx->ins, "already retring ?");
175	0	return 0;
176	0	}
177
178		/* try before creating event to stop incoming event */
179	0	ret = reconnect_docker_sock(ins, config, ctx);
180	0	if (ret == 0) {
181	0	return 0;
182	0	}
183
184	0	ctx->current_retries = 1;
185	0	ctx->retry_coll_id = flb_input_set_collector_time(ins,
186	0	cb_reconnect,
187	0	ctx->reconnect_retry_interval,
188	0	0,
189	0	config);
190	0	if (ctx->retry_coll_id < 0) {
191	0	flb_plg_error(ctx->ins, "failed to create timer event");
192	0	return -1;
193	0	}
194	0	ret = flb_input_collector_start(ctx->retry_coll_id, ins);
195	0	if (ret < 0) {
196	0	flb_plg_error(ctx->ins, "failed to start timer event");
197	0	flb_input_collector_delete(ctx->retry_coll_id, ins);
198	0	ctx->retry_coll_id = -1;
199	0	return -1;
200	0	}
201	0	flb_plg_info(ctx->ins, "create reconnect event. interval=%d second",
202	0	ctx->reconnect_retry_interval);
203
204	0	return 0;
205	0	}
206
207		static int is_recoverable_error(int error)
208	0	{
209		/* ENOTTY:
210		It reports on Docker in Docker mode.
211		https://github.com/fluent/fluent-bit/issues/3439#issuecomment-831424674
212		*/
213	0	if (error == ENOTTY \|\| error == EBADF) {
214	0	return FLB_TRUE;
215	0	}
216	0	return FLB_FALSE;
217	0	}
218
219
220		/**
221		* Callback function to process events recieved on the unix
222		* socket.
223		*
224		* @param ins Pointer to flb_input_instance
225		* @param config Pointer to flb_config
226		* @param in_context void Pointer used to cast to
227		* flb_in_de_config
228		*
229		* @return int Always returns success
230		*/
231		static int in_de_collect(struct flb_input_instance *ins,
232		struct flb_config config, void in_context)
233	0	{
234	0	int ret = 0;
235	0	int error;
236	0	size_t str_len = 0;
237	0	struct flb_in_de_config *ctx = in_context;
238
239		/* variables for parser */
240	0	int parser_ret = -1;
241	0	void *out_buf = NULL;
242	0	size_t out_size = 0;
243	0	struct flb_time out_time;
244
245	0	ret = read(ctx->fd, ctx->buf, ctx->buf_size - 1);
246
247	0	if (ret > 0) {
248	0	str_len = ret;
249	0	ctx->buf[str_len] = '\0';
250
251	0	ret = flb_log_event_encoder_begin_record(&ctx->log_encoder);
252
253	0	if (!ctx->parser) {
254		/* Initialize local msgpack buffer */
255	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
256	0	ret = flb_log_event_encoder_set_current_timestamp(
257	0	&ctx->log_encoder);
258	0	}
259
260	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
261	0	ret = flb_log_event_encoder_append_body_values(
262	0	&ctx->log_encoder,
263	0	FLB_LOG_EVENT_CSTRING_VALUE(ctx->key),
264	0	FLB_LOG_EVENT_STRING_VALUE(ctx->buf, str_len));
265	0	}
266
267	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
268	0	ret = flb_log_event_encoder_commit_record(&ctx->log_encoder);
269	0	}
270
271	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
272	0	flb_input_log_append(ins, NULL, 0,
273	0	ctx->log_encoder.output_buffer,
274	0	ctx->log_encoder.output_length);
275
276	0	}
277	0	else {
278	0	flb_plg_error(ctx->ins, "Error encoding record : %d", ret);
279	0	}
280	0	}
281	0	else {
282	0	flb_time_get(&out_time);
283
284	0	parser_ret = flb_parser_do(ctx->parser, ctx->buf, str_len - 1,
285	0	&out_buf, &out_size, &out_time);
286	0	if (parser_ret >= 0) {
287	0	if (flb_time_to_nanosec(&out_time) == 0L) {
288	0	flb_time_get(&out_time);
289	0	}
290
291	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
292	0	ret = flb_log_event_encoder_set_timestamp(
293	0	&ctx->log_encoder,
294	0	&out_time);
295	0	}
296
297	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
298	0	ret = flb_log_event_encoder_set_body_from_raw_msgpack(
299	0	&ctx->log_encoder,
300	0	out_buf,
301	0	out_size);
302	0	}
303
304	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
305	0	ret = flb_log_event_encoder_commit_record(&ctx->log_encoder);
306	0	}
307
308	0	if (ret == FLB_EVENT_ENCODER_SUCCESS) {
309	0	flb_input_log_append(ins, NULL, 0,
310	0	ctx->log_encoder.output_buffer,
311	0	ctx->log_encoder.output_length);
312
313	0	}
314	0	else {
315	0	flb_plg_error(ctx->ins, "Error encoding record : %d", ret);
316	0	}
317
318
319	0	flb_free(out_buf);
320	0	}
321	0	else {
322	0	flb_plg_trace(ctx->ins, "tried to parse: %s", ctx->buf);
323	0	flb_plg_trace(ctx->ins, "buf_size %zu", ctx->buf_size);
324	0	flb_plg_error(ctx->ins, "parser returned an error: %d",
325	0	parser_ret);
326	0	}
327	0	}
328
329	0	flb_log_event_encoder_reset(&ctx->log_encoder);
330	0	}
331	0	else if (ret == 0) {
332		/* EOF */
333
334		/* docker service may be restarted */
335	0	flb_plg_info(ctx->ins, "EOF detected. Re-initialize");
336	0	if (ctx->reconnect_retry_limits > 0) {
337	0	ret = create_reconnect_event(ins, config, ctx);
338	0	if (ret < 0) {
339	0	return ret;
340	0	}
341	0	}
342	0	}
343	0	else {
344	0	error = errno;
345	0	flb_plg_error(ctx->ins, "read returned error: %d, %s", error,
346	0	strerror(error));
347	0	if (is_recoverable_error(error)) {
348	0	if (ctx->reconnect_retry_limits > 0) {
349	0	ret = create_reconnect_event(ins, config, ctx);
350	0	if (ret < 0) {
351	0	return ret;
352	0	}
353	0	}
354	0	}
355	0	}
356
357	0	return 0;
358	0	}
359
360		/**
361		* Callback function to initialize docker events plugin
362		*
363		* @param ins Pointer to flb_input_instance
364		* @param config Pointer to flb_config
365		* @param data Unused
366		*
367		* @return int 0 on success, -1 on failure
368		*/
369		static int in_de_init(struct flb_input_instance *ins,
370		struct flb_config config, void data)
371	0	{
372	0	struct flb_in_de_config *ctx = NULL;
373	0	(void) data;
374
375		/* Allocate space for the configuration */
376	0	ctx = de_config_init(ins, config);
377	0	if (!ctx) {
378	0	return -1;
379	0	}
380	0	ctx->ins = ins;
381	0	ctx->retry_coll_id = -1;
382	0	ctx->current_retries = 0;
383
384		/* Set the context */
385	0	flb_input_set_context(ins, ctx);
386
387	0	if (de_unix_create(ctx) != 0) {
388	0	flb_plg_error(ctx->ins, "could not listen on unix://%s",
389	0	ctx->unix_path);
390	0	de_config_destroy(ctx);
391	0	return -1;
392	0	}
393
394	0	ctx->coll_id = flb_input_set_collector_event(ins, in_de_collect,
395	0	ctx->fd, config);
396	0	if(ctx->coll_id < 0){
397	0	flb_plg_error(ctx->ins,
398	0	"could not set collector for IN_DOCKER_EVENTS plugin");
399	0	de_config_destroy(ctx);
400	0	return -1;
401	0	}
402
403	0	flb_plg_info(ctx->ins, "listening for events on %s", ctx->unix_path);
404	0	return 0;
405	0	}
406
407		/**
408		* Callback exit function to cleanup plugin
409		*
410		* @param data Pointer cast to flb_in_de_config
411		* @param config Unused
412		*
413		* @return int Always returns 0
414		*/
415		static int in_de_exit(void data, struct flb_config config)
416	0	{
417	0	(void) config;
418	0	struct flb_in_de_config *ctx = data;
419
420	0	if (!ctx) {
421	0	return 0;
422	0	}
423
424	0	de_config_destroy(ctx);
425
426	0	return 0;
427	0	}
428
429		/* Configuration properties map */
430		static struct flb_config_map config_map[] = {
431		{
432		FLB_CONFIG_MAP_STR, "unix_path", DEFAULT_UNIX_SOCKET_PATH,
433		0, FLB_TRUE, offsetof(struct flb_in_de_config, unix_path),
434		"Define Docker unix socket path to read events"
435		},
436		{
437		FLB_CONFIG_MAP_SIZE, "buffer_size", "8k",
438		0, FLB_TRUE, offsetof(struct flb_in_de_config, buf_size),
439		"Set buffer size to read events"
440		},
441		{
442		FLB_CONFIG_MAP_STR, "parser", NULL,
443		0, FLB_FALSE, 0,
444		"Optional parser for records, if not set, records are packages under 'key'"
445		},
446		{
447		FLB_CONFIG_MAP_STR, "key", DEFAULT_FIELD_NAME,
448		0, FLB_TRUE, offsetof(struct flb_in_de_config, key),
449		"Set the key name to store unparsed Docker events"
450		},
451		{
452		FLB_CONFIG_MAP_INT, "reconnect.retry_limits", "5",
453		0, FLB_TRUE, offsetof(struct flb_in_de_config, reconnect_retry_limits),
454		"Maximum number to retry to connect docker socket"
455		},
456		{
457		FLB_CONFIG_MAP_INT, "reconnect.retry_interval", "1",
458		0, FLB_TRUE, offsetof(struct flb_in_de_config, reconnect_retry_interval),
459		"Retry interval to connect docker socket"
460		},
461		/* EOF */
462		{0}
463		};
464
465		/* Plugin reference */
466		struct flb_input_plugin in_docker_events_plugin = {
467		.name = "docker_events",
468		.description = "Docker events",
469		.cb_init = in_de_init,
470		.cb_pre_run = NULL,
471		.cb_collect = in_de_collect,
472		.cb_flush_buf = NULL,
473		.cb_exit = in_de_exit,
474		.config_map = config_map,
475		.flags = FLB_INPUT_NET
476		};