out_azure_kusto: added buffering fixes

plugins/out_azure_kusto/CMakeLists.txt

@@ -1,7 +1,8 @@
 set(src
-  azure_kusto.c
-  azure_kusto_conf.c
-  azure_kusto_ingest.c
-  )
+        azure_kusto.c
+        azure_kusto_conf.c
+        azure_kusto_ingest.c
+        azure_kusto_store.c
+)
 
-FLB_PLUGIN(out_azure_kusto "${src}" "")
+FLB_PLUGIN(out_azure_kusto "${src}" "")

plugins/out_azure_kusto/azure_kusto.h

@@ -26,6 +26,13 @@
 #include <fluent-bit/flb_sds.h>
 #include <fluent-bit/flb_upstream_ha.h>
 
+#include <fluent-bit/flb_scheduler.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_time.h>
+#include <dirent.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
 /* refresh token every 50 minutes */
 #define FLB_AZURE_KUSTO_TOKEN_REFRESH 3000
 
@@ -53,14 +60,23 @@
 
 #define FLB_AZURE_KUSTO_INGEST_ENDPOINT_CONNECTION_TIMEOUT "60"
 
+#define FLB_AZURE_KUSTO_BUFFER_DIR_MAX_SIZE "8G"  // 8GB buffer directory size
+#define UPLOAD_TIMER_MAX_WAIT 180000
+#define UPLOAD_TIMER_MIN_WAIT 18000
+#define MAX_FILE_SIZE         4000000000 // 4GB
+
+#define FLB_AZURE_IMDS_ENDPOINT "/metadata/identity/oauth2/token"
+#define FLB_AZURE_IMDS_API_VERSION "2018-02-01"
+#define FLB_AZURE_IMDS_RESOURCE "https://api.kusto.windows.net/"
+
 
 struct flb_azure_kusto_resources {
     struct flb_upstream_ha *blob_ha;
     struct flb_upstream_ha *queue_ha;
     flb_sds_t identity_token;
 
     /* used to reload resouces after some time */
-    time_t load_time;
+    uint64_t load_time;
 };
 
 struct flb_azure_kusto {
@@ -74,6 +90,7 @@ struct flb_azure_kusto {
     flb_sds_t ingestion_mapping_reference;
 
     int ingestion_endpoint_connect_timeout;
+    int io_timeout;
 
     /* compress payload */
     int compression_enabled;
@@ -87,14 +104,19 @@ struct flb_azure_kusto {
     int include_time_key;
     flb_sds_t time_key;
 
-    /* --- internal data --- */
+    flb_sds_t azure_kusto_buffer_key;
 
-    flb_sds_t ingestion_mgmt_endpoint;
+    /* --- internal data --- */
 
     /* oauth2 context */
     flb_sds_t oauth_url;
+    //flb_sds_t imds_url;
+    int use_imds;
     struct flb_oauth2 *o;
 
+    int timer_created;
+    int timer_ms;
+
     /* mutex for acquiring oauth tokens */
     pthread_mutex_t token_mutex;
 
@@ -106,9 +128,36 @@ struct flb_azure_kusto {
 
     pthread_mutex_t blob_mutex;
 
+    pthread_mutex_t buffer_mutex;
+
+    int buffering_enabled;
+
+    size_t file_size;
+    time_t upload_timeout;
+    time_t retry_time;
+
+    int buffer_file_delete_early;
+    int unify_tag;
+    int blob_uri_length;
+    int scheduler_max_retries;
+    int delete_on_max_upload_error;
+
+    int has_old_buffers;
+    size_t store_dir_limit_size;
+    /* track the total amount of buffered data */
+    size_t current_buffer_size;
+    flb_sds_t buffer_dir;
+    char *store_dir;
+    struct flb_fstore *fs;
+    struct flb_fstore_stream *stream_active;  /* default active stream */
+    struct flb_fstore_stream *stream_upload;
+
+
     /* Upstream connection to the backend server */
     struct flb_upstream *u;
 
+    struct flb_upstream *imds_upstream;
+
     /* Fluent Bit context */
     struct flb_config *config;
 
@@ -119,4 +168,4 @@ struct flb_azure_kusto {
 flb_sds_t get_azure_kusto_token(struct flb_azure_kusto *ctx);
 flb_sds_t execute_ingest_csl_command(struct flb_azure_kusto *ctx, const char *csl);
 
-#endif
+#endif