in_docker_events: re-initialize docker socket when dockerd is restarted(#3439)

Signed-off-by: Takahiro Yamashita <[email protected]>

Author: nokute78

plugins/in_docker_events/docker_events.c

@@ -102,7 +102,9 @@ static int in_de_collect(struct flb_input_instance *ins,
     size_t out_size = 0;
     struct flb_time out_time;
 
-    if ((ret = read(ctx->fd, ctx->buf, ctx->buf_size - 1)) > 0) {
+    ret = read(ctx->fd, ctx->buf, ctx->buf_size - 1);
+
+    if (ret > 0) {
         str_len = ret;
         ctx->buf[str_len] = '\0';
 
@@ -153,6 +155,39 @@ static int in_de_collect(struct flb_input_instance *ins,
             }
         }
     }
+    else if (ret == 0) {
+        /* EOF */
+        /* docker service may be restarted */
+        flb_plg_info(ctx->ins, "EOF detected. Re-initialize");
+        /* remove old socket collector */
+        ret = flb_input_collector_delete(ctx->coll_id, ins);
+        close(ctx->fd);
+        if (ret < 0) {
+            flb_plg_error(ctx->ins, "failed to pause event");
+            return -1;
+        }
+        ctx->fd = -1;
+        /* create socket again */
+        if (de_unix_create(ctx) < 0) {
+            flb_plg_error(ctx->ins, "failed to re-initialize socket");
+            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");
+            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");
+            return -1;
+        }
+    }
     else {
         error = errno;
         flb_plg_error(ctx->ins, "read returned error: %d, %s", error,
@@ -194,8 +229,9 @@ static int in_de_init(struct flb_input_instance *ins,
         return -1;
     }
 
-    if (flb_input_set_collector_event(ins, in_de_collect,
-                                      ctx->fd, config) == -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);

plugins/in_docker_events/docker_events.h

@@ -33,6 +33,7 @@
 struct flb_in_de_config
 {
     int fd;                         /* File descriptor */
+    int coll_id;                    /* collector id */
     flb_sds_t unix_path;            /* Unix path for socket */
     char *buf;
     size_t buf_size;