in_storage_backlog: extra checks and use content size for validation
@@ -77,6 +77,8 @@ static int cb_queue_chunks(struct flb_input_instance *in,
* All chunks on this backlog are 'file' based, always try to set
* them up. We validate the status.
*/
+ ret = cio_chunk_is_up(sbc->chunk);
+ if (ret == CIO_FALSE) {
ret = cio_chunk_up(sbc->chunk);
if (ret == CIO_CORRUPTED) {
flb_plg_error(ctx->ins, "removing corrupted chunk from the "
@@ -91,15 +93,21 @@ static int cb_queue_chunks(struct flb_input_instance *in,
continue;
}
+ }
+
/* get the number of bytes being used by the chunk */
- size = cio_chunk_get_real_size(sbc->chunk);
+ size = cio_chunk_get_content_size(sbc->chunk);
if (size <= 0) {
+ flb_plg_error(ctx->ins, "removing empty chunk from the "
+ "queue %s:%s",
+ sbc->stream->name, sbc->chunk->name);
+ cio_chunk_close(sbc->chunk, FLB_TRUE);
+ mk_list_del(&sbc->_head);
+ flb_free(sbc);
continue;
}
ch = sbc->chunk;
- flb_plg_info(ctx->ins, "queueing %s:%s",
- sbc->stream->name, sbc->chunk->name);
/* Associate this backlog chunk to this instance into the engine */
ic = flb_input_chunk_map(in, ch);
@@ -113,6 +121,9 @@ static int cb_queue_chunks(struct flb_input_instance *in,
continue;
}
+ flb_plg_info(ctx->ins, "queueing %s:%s",
+ sbc->stream->name, sbc->chunk->name);
+
/* remove the reference, it's on the engine hands now */
mk_list_del(&sbc->_head);
flb_free(sbc);
