core: fixes for issues reported by Coverity

plugins/in_emitter/emitter.c

@@ -107,8 +107,7 @@ int static do_in_emitter_add_record(struct em_chunk *ec,
                                ec->mp_sbuf.data,
                                ec->mp_sbuf.size);
     if (ret == -1) {
-        flb_plg_error(ctx->ins, "error registering chunk with tag: %s",
-                      ec->tag);
+        flb_plg_error(ctx->ins, "error registering chunk with tag: %s", ec->tag);
         /* Release the echunk */
         em_chunk_destroy(ec);
         return -1;
@@ -156,7 +155,7 @@ int in_emitter_add_record(const char *tag, int tag_len,
         i_ref->i_ins = i_ins;
         mk_list_add(&i_ref->_head, &ctx->i_ins_list);
         /* If in_emitter is paused, but new input plugin is not paused, pause it */
-        if (flb_input_buf_paused(ctx->ins) == FLB_TRUE && 
+        if (flb_input_buf_paused(ctx->ins) == FLB_TRUE &&
                 flb_input_buf_paused(i_ins) == FLB_FALSE) {
             flb_input_pause(i_ins);
         }
@@ -229,7 +228,7 @@ static int in_emitter_ingest_ring_buffer(struct flb_input_instance *in,
     (void) in;
 
 
-    while ((ret = flb_ring_buffer_read(ctx->msgs, (void *)&ec, 
+    while ((ret = flb_ring_buffer_read(ctx->msgs, (void *)&ec,
                                        sizeof(struct em_chunk))) == 0) {
         ret = flb_input_log_append(in,
                                    ec.tag, flb_sds_len(ec.tag),
@@ -260,8 +259,6 @@ static int cb_queue_chunks(struct flb_input_instance *in,
         /* Associate this backlog chunk to this instance into the engine */
         ret = do_in_emitter_add_record(echunk, in);
         if (ret == -1) {
-            flb_error("[in_emitter] error registering chunk with tag: %s",
-                      echunk->tag);
             continue;
         }
     }
@@ -400,7 +397,7 @@ static int cb_emitter_exit(void *data, struct flb_config *config)
     }
 
     if (ctx->msgs) {
-        while ((ret = flb_ring_buffer_read(ctx->msgs, (void *)&ec, 
+        while ((ret = flb_ring_buffer_read(ctx->msgs, (void *)&ec,
                                         sizeof(struct em_chunk))) == 0) {
             flb_sds_destroy(ec.tag);
             msgpack_sbuffer_destroy(&ec.mp_sbuf);

plugins/in_forward/fw_conn.c

@@ -147,11 +147,12 @@ struct fw_conn *fw_conn_add(struct flb_connection *connection, struct flb_in_fw_
         helo = flb_malloc(sizeof(struct flb_in_fw_helo));
         if (!helo) {
             flb_errno();
-
             return NULL;
         }
+
         ret = fw_prot_secure_forward_handshake_start(ctx->ins, connection, helo);
         if (ret != 0) {
+            flb_free(helo);
             return NULL;
         }
 

plugins/in_forward/fw_prot.c

@@ -630,6 +630,7 @@ static int check_ping(struct flb_input_instance *ins,
     if (o.type != MSGPACK_OBJECT_STR) {
         flb_plg_error(ins, "Invalid password_digest type message");
         flb_free(serverside);
+        flb_free(hostname);
         msgpack_unpacked_destroy(&result);
         return -1;
     }

plugins/in_tail/tail_file.c

@@ -876,7 +876,7 @@ static int set_file_position(struct flb_tail_config *ctx,
     }
     file->offset = ret;
 
-    if (file->decompression_context == NULL) {        
+    if (file->decompression_context == NULL) {
         file->stream_offset = ret;
     }
 
@@ -946,13 +946,13 @@ int flb_tail_file_append(char *path, struct stat *st, int mode,
     if (flb_tail_file_exists(st, ctx) == FLB_TRUE) {
         return -1;
     }
-   
+
     #ifdef __linux__
     if (ctx->file_cache_advise) {
         flb_plg_debug(ctx->ins, "file will be read in POSIX_FADV_DONTNEED mode %s", path);
     }
     #endif
-    
+
     fd = open(path, O_RDONLY);
     if (fd == -1) {
         flb_errno();
@@ -1044,8 +1044,9 @@ int flb_tail_file_append(char *path, struct stat *st, int mode,
      * for path_key to continue working after rotation. */
     file->orig_name = flb_strdup(file->name);
     if (!file->orig_name) {
-        flb_free(file->name);
         flb_errno();
+        flb_free(file->name);
+        file->name = NULL;
         goto error;
     }
     file->orig_name_len = file->name_len;
@@ -1449,7 +1450,7 @@ int flb_tail_file_chunk(struct flb_tail_file *file)
         }
     }
     #endif
-    
+
     read_size = file_buffer_capacity;
 
     if (file->decompression_context != NULL) {

plugins/out_kafka/kafka.c

@@ -105,7 +105,7 @@ int produce_message(struct flb_time *tm, msgpack_object *map,
     msgpack_packer mp_pck;
     msgpack_object key;
     msgpack_object val;
-    flb_sds_t s;
+    flb_sds_t s = NULL;
 
 #ifdef FLB_HAVE_AVRO_ENCODER
     // used to flag when a buffer needs to be freed for avro

plugins/out_stackdriver/stackdriver.c

@@ -1723,7 +1723,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
 
     /* Parameters for trace */
     int trace_extracted = FLB_FALSE;
-    flb_sds_t trace;
+    flb_sds_t trace = NULL;
     char stackdriver_trace[PATH_MAX];
     const char *new_trace;
 
@@ -2357,6 +2357,7 @@ static flb_sds_t stackdriver_format(struct flb_stackdriver *ctx,
             flb_plg_error(ctx->ins, "the type of payload labels should be map");
             flb_sds_destroy(operation_id);
             flb_sds_destroy(operation_producer);
+            flb_sds_destroy(trace);
             flb_log_event_decoder_destroy(&log_decoder);
             msgpack_sbuffer_destroy(&mp_sbuf);
             return NULL;