storage_backlog: changed the release strategy

The output queue release strategy was changed from a transaction based
approach to a non transaction based approach in order to allow
rollover to work properly.

Additionally, the default behavior when chunks are mapped was changed
to set them down in order to honor max_chunks_up.

Signed-off-by: Leonardo Alminana <[email protected]>

Author: Leonardo Alminana

plugins/in_storage_backlog/sb.c

@@ -91,7 +91,7 @@ static int sb_append_chunk_to_segregated_backlogs(struct cio_chunk  *target_chun
 int sb_segregate_chunks(struct flb_config *config);
 
 int sb_release_output_queue_space(struct flb_output_instance *output_plugin,
-                                  size_t                      required_space);
+                                  ssize_t                    *required_space);
 
 ssize_t sb_get_releasable_output_queue_space(struct flb_output_instance *output_plugin,
                                              size_t                      required_space);
@@ -441,9 +441,10 @@ ssize_t sb_get_releasable_output_queue_space(struct flb_output_instance *output_
 }
 
 int sb_release_output_queue_space(struct flb_output_instance *output_plugin,
-                                  size_t                      required_space)
+                                  ssize_t                    *required_space)
 {
     struct mk_list      *chunk_iterator_tmp;
+    struct cio_chunk    *underlying_chunk;
     struct mk_list      *chunk_iterator;
     size_t               released_space;
     struct flb_sb       *context;
@@ -469,18 +470,17 @@ int sb_release_output_queue_space(struct flb_output_instance *output_plugin,
         chunk = mk_list_entry(chunk_iterator, struct sb_out_chunk, _head);
 
         released_space += chunk->size;
+        underlying_chunk = chunk->chunk;
 
-        cio_chunk_close(chunk->chunk, FLB_TRUE);
-        sb_remove_chunk_from_segregated_backlogs(chunk->chunk, context);
+        sb_remove_chunk_from_segregated_backlogs(underlying_chunk, context);
+        cio_chunk_close(underlying_chunk, FLB_TRUE);
 
-        if (released_space >= required_space) {
+        if (released_space >= *required_space) {
             break;
         }
     }
 
-    if (released_space < required_space) {
-        return -3;
-    }
+    *required_space -= released_space;
 
     return 0;
 }
@@ -618,6 +618,7 @@ static int cb_queue_chunks(struct flb_input_instance *in,
                  * queue but we need to leave it in the remainder queues.
                  */
                 sb_remove_chunk_from_segregated_backlogs(chunk_instance->chunk, ctx);
+                cio_chunk_down(ch);
 
                 /* check our limits */
                 total += size;