blob: assorted improvements

plugins/in_blob/blob.c

@@ -28,10 +28,13 @@
 
 #include <sys/types.h>
 #include <sys/stat.h>
+#include <string.h>
 
 #ifndef FLB_SYSTEM_WINDOWS
 #include <glob.h>
 #include <fnmatch.h>
+#else
+#define strtok_r strtok_s
 #endif
 
 #include <stdio.h>
@@ -190,6 +193,128 @@ static inline int do_glob(const char *pattern,
     return ret;
 }
 
+#define MATCHING_METHOD_PRESENCE 0
+#define MATCHING_METHOD_EXACT    1
+
+static int apply_glob_pattern(flb_sds_t pattern, char *path) {
+    int        matching_method;
+    char      *strtok_context;
+    cfl_sds_t  local_pattern;
+    char      *filename_part;
+    char      *pattern_part;
+    char      *filename;
+    int        result;
+    size_t     index;
+
+    if (pattern == NULL || pattern[0] == '\0') {
+        return FLB_FALSE;
+    }
+
+    if (path == NULL || path[0] == '\0') {
+        return FLB_FALSE;
+    }
+
+    filename = NULL;
+
+#ifdef FLB_SYSTEM_WINDOWS
+    filename = strrchr(path, '\\');
+#endif
+
+    if (filename == NULL) {
+        filename = strrchr(path, '/');
+    }
+
+    if (filename != NULL) {
+        filename = &filename[1];
+    }
+    else {
+        filename = path;
+    }
+
+    if (filename[0] == '\0') {
+        return FLB_FALSE;
+    }
+
+    local_pattern = cfl_sds_create(pattern);
+
+    if (local_pattern == NULL) {
+        return FLB_FALSE;
+    }
+
+    result = FLB_FALSE;
+
+    if (strrchr(local_pattern, '*') != NULL) {
+        index = 0;
+        pattern_part = strtok_r(local_pattern, "*", &strtok_context);
+
+        if (local_pattern[0] == '*') {
+            matching_method = MATCHING_METHOD_PRESENCE;
+        }
+        else {
+            matching_method = MATCHING_METHOD_EXACT;
+        }
+
+        filename_part = filename;
+
+
+        while (pattern_part != NULL && filename_part != NULL) {
+            if (matching_method == MATCHING_METHOD_PRESENCE) {
+                filename_part = strstr(filename_part, pattern_part);
+            }
+            else {
+                result = strncmp(filename_part, pattern_part, strlen(pattern_part));
+
+                if (result != 0) {
+                    filename_part = NULL;
+                }
+
+                matching_method = MATCHING_METHOD_PRESENCE;
+            }
+
+            if (filename_part != NULL) {
+                filename_part = &filename_part[strlen(pattern_part)];
+            }
+
+            index++;
+            pattern_part = strtok_r(NULL, "*", &strtok_context);
+        }
+
+        result = FLB_FALSE;
+
+        if (pattern_part == NULL) {
+            if (pattern[strlen(pattern) - 1] != '*') {
+                if (filename_part != NULL) {
+                    if (filename_part[0] == '\0') {
+                        result = FLB_TRUE;
+                    }
+                }
+            }
+            else {
+                if (filename_part != NULL) {
+                    result = FLB_TRUE;
+                }
+            }
+        }
+        else {
+            result = FLB_FALSE;
+        }
+    }
+    else {
+        result = strcmp(filename, local_pattern);
+
+        if (result == 0) {
+            result = FLB_TRUE;
+        }
+        else {
+            result = FLB_FALSE;
+        }
+    }
+
+    cfl_sds_destroy(local_pattern);
+
+    return result;
+}
+
 /* This function recursively searches a directory tree using
  * a glob compatible pattern that implements the fluentd style
  * recursion wildcard **.
@@ -290,11 +415,20 @@ static ssize_t recursive_file_search(struct blob_ctx *ctx,
     else if (cfl_sds_len(local_pattern) == 0) {
         recursive_search_flag = FLB_FALSE;
     }
+    else if (cfl_sds_len(local_path) == 0) {
+        recursive_search_flag = FLB_FALSE;
+    }
 
     memset(&glob_context, 0, sizeof(glob_t));
 
     /* Scan the given path */
-    result = do_glob(local_path, GLOB_TILDE | GLOB_ERR, NULL, &glob_context);
+    if (local_path[0] != '\0') {
+        result = do_glob(local_path, GLOB_TILDE | GLOB_ERR, NULL, &glob_context);
+    }
+    else {
+        result = do_glob(local_pattern, GLOB_TILDE | GLOB_ERR, NULL, &glob_context);
+    }
+
     if (result != 0) {
         switch (result) {
         case GLOB_NOSPACE:
@@ -411,23 +545,29 @@ static ssize_t recursive_file_search(struct blob_ctx *ctx,
         else if (recursive_search_flag == FLB_FALSE &&
                  (S_ISREG(fs_entry_metadata.st_mode) ||
                   S_ISLNK(fs_entry_metadata.st_mode))) {
-            result = blob_file_append(ctx,
-                                      glob_context.gl_pathv[index],
-                                      &fs_entry_metadata);
 
-            if (result == 0) {
-                flb_plg_debug(ctx->ins,
-                              "blob scan add: %s, inode %" PRIu64,
-                              glob_context.gl_pathv[index],
-                              (uint64_t) fs_entry_metadata.st_ino);
-            }
-            else {
-                flb_plg_debug(ctx->ins,
-                              "blob scan skip: %s",
-                              glob_context.gl_pathv[index]);
-            }
+            result = apply_glob_pattern(ctx->exclude_pattern,
+                                        (char *) glob_context.gl_pathv[index]);
 
-            match_count++;
+            if (result == FLB_FALSE) {
+                result = blob_file_append(ctx,
+                                        glob_context.gl_pathv[index],
+                                        &fs_entry_metadata);
+
+                if (result == 0) {
+                    flb_plg_debug(ctx->ins,
+                                "blob scan add: %s, inode %" PRIu64,
+                                glob_context.gl_pathv[index],
+                                (uint64_t) fs_entry_metadata.st_ino);
+                }
+                else {
+                    flb_plg_debug(ctx->ins,
+                                "blob scan skip: %s",
+                                glob_context.gl_pathv[index]);
+                }
+
+                match_count++;
+            }
         }
     }
 
@@ -843,6 +983,11 @@ static struct flb_config_map config_map[] = {
      "Path to scan for blob/binary files"
     },
 
+    {
+     FLB_CONFIG_MAP_STR, "exclude_pattern", NULL,
+     0, FLB_TRUE, offsetof(struct blob_ctx, exclude_pattern),
+    },
+
 #ifdef FLB_HAVE_SQLDB
     {
      FLB_CONFIG_MAP_STR, "database_file", NULL,

plugins/in_blob/blob.h

@@ -73,6 +73,7 @@ struct blob_ctx {
 
     /* config map options */
     flb_sds_t path;
+    flb_sds_t exclude_pattern;
     flb_sds_t database_file;
     time_t scan_refresh_interval;
 

plugins/out_azure_blob/azure_blob.c

@@ -668,7 +668,7 @@ static int process_blob_chunk(struct flb_azure_blob *ctx, struct flb_event_chunk
             continue;
         }
 
-        ret = azb_db_file_insert(ctx, source, file_path, file_size);
+        ret = azb_db_file_insert(ctx, source, ctx->endpoint, file_path, file_size);
         if (ret == -1) {
             flb_plg_error(ctx->ins, "cannot insert blob file into database: %s (size=%lu)",
                           file_path, file_size);
@@ -708,6 +708,7 @@ static void cb_azb_blob_file_upload(struct flb_config *config, void *out_context
     uint64_t file_delivery_attempts;
     off_t offset_start;
     off_t offset_end;
+    cfl_sds_t file_destination = NULL;
     cfl_sds_t file_path = NULL;
     cfl_sds_t part_ids = NULL;
     cfl_sds_t source = NULL;
@@ -876,7 +877,8 @@ static void cb_azb_blob_file_upload(struct flb_config *config, void *out_context
                                     &offset_start, &offset_end,
                                     &part_delivery_attempts,
                                     &file_delivery_attempts,
-                                    &file_path);
+                                    &file_path,
+                                    &file_destination);
     if (ret == -1) {
         flb_plg_error(ctx->ins, "cannot get next blob file part");
         info->active_upload = FLB_FALSE;
@@ -891,6 +893,25 @@ static void cb_azb_blob_file_upload(struct flb_config *config, void *out_context
         /* just continue, the row info was retrieved */
     }
 
+    if (strcmp(file_destination, ctx->endpoint) != 0) {
+        flb_plg_info(ctx->ins,
+                     "endpoint change detected, restarting file : %s",
+                     file_path);
+
+        info->active_upload = FLB_FALSE;
+
+        /* we need to set the aborted state flag to wait for existing uploads
+         * to finish and then wipe the slate and start again but we don't want
+         * to increment the failure count in this case.
+         */
+        azb_db_file_set_aborted_state(ctx, file_id, file_path, 1);
+
+        cfl_sds_destroy(file_path);
+        cfl_sds_destroy(file_destination);
+
+        flb_sched_timer_cb_coro_return();
+    }
+
     /* since this is the first part we want to increment the files
      * delivery attempt counter.
      */
@@ -902,20 +923,31 @@ static void cb_azb_blob_file_upload(struct flb_config *config, void *out_context
     ret = flb_utils_read_file_offset(file_path, offset_start, offset_end, &out_buf, &out_size);
     if (ret == -1) {
         flb_plg_error(ctx->ins, "cannot read file part %s", file_path);
-        cfl_sds_destroy(file_path);
+
         info->active_upload = FLB_FALSE;
+
+        cfl_sds_destroy(file_path);
+        cfl_sds_destroy(file_destination);
+
         flb_sched_timer_cb_coro_return();
     }
 
     azb_db_file_part_delivery_attempts(ctx, file_id, part_id, ++part_delivery_attempts);
 
     flb_plg_debug(ctx->ins, "sending part file %s (id=%" PRIu64 " part_id=%" PRIu64 ")", file_path, id, part_id);
+
     ret = send_blob(config, NULL, ctx, FLB_EVENT_TYPE_BLOBS,
                     AZURE_BLOB_BLOCKBLOB, file_path, part_id, NULL, 0, out_buf, out_size);
+
     if (ret == FLB_OK) {
         ret = azb_db_file_part_uploaded(ctx, id);
+
         if (ret == -1) {
             info->active_upload = FLB_FALSE;
+
+            cfl_sds_destroy(file_path);
+            cfl_sds_destroy(file_destination);
+
             flb_sched_timer_cb_coro_return();
         }
     }
@@ -926,14 +958,16 @@ static void cb_azb_blob_file_upload(struct flb_config *config, void *out_context
             part_delivery_attempts >= ctx->part_delivery_attempt_limit) {
             azb_db_file_set_aborted_state(ctx, file_id, file_path, 1);
         }
-        /* FIXME */
     }
+
     info->active_upload = FLB_FALSE;
 
     if (out_buf) {
         flb_free(out_buf);
     }
+
     cfl_sds_destroy(file_path);
+    cfl_sds_destroy(file_destination);
 
     flb_sched_timer_cb_coro_return();
 }

plugins/out_azure_blob/azure_blob.h

@@ -108,6 +108,7 @@ struct flb_azure_blob {
     sqlite3_stmt *stmt_delete_file;
     sqlite3_stmt *stmt_abort_file;
     sqlite3_stmt *stmt_get_file;
+    sqlite3_stmt *stmt_update_file_destination;
     sqlite3_stmt *stmt_update_file_delivery_attempt_count;
     sqlite3_stmt *stmt_set_file_aborted_state;
     sqlite3_stmt *stmt_get_next_aborted_file;