in_calyptia_fleet: windows support.

cmake/windows-setup.cmake

@@ -60,8 +60,6 @@ if(FLB_WINDOWS_DEFAULTS)
   set(FLB_IN_EMITTER            Yes)
   set(FLB_IN_PODMAN_METRICS      No)
   set(FLB_IN_ELASTICSEARCH      Yes)
-  # disable calyptia fleet management for now
-  set(FLB_IN_CALYPTIA_FLEET     No)
   set(FLB_IN_SPLUNK             Yes)
 
   # OUTPUT plugins

plugins/in_calyptia_fleet/in_calyptia_fleet.c

@@ -348,8 +348,11 @@ static void *do_reload(void *data)
     reload->flb->config->conf_path_file = reload->cfg_path;
 
     sleep(5);
+#ifndef FLB_SYSTEM_WINDOWS
     kill(getpid(), SIGHUP);
-
+#else
+	GenerateConsoleCtrlEvent(1 /* CTRL_BREAK_EVENT_1 */, 0);
+#endif
     return NULL;
 }
 
@@ -690,6 +693,33 @@ static int get_calyptia_fleet_id_by_name(struct flb_in_calyptia_fleet_config *ct
     return 0;
 }
 
+#ifdef FLB_SYSTEM_WINDOWS
+#define link(a, b) CreateHardLinkA(b, a, 0)
+
+ssize_t readlink(const char *path, char *realpath, size_t srealpath) {
+    HANDLE hFile;
+    DWORD ret;
+
+    hFile = CreateFile(path, GENERIC_READ, FILE_SHARE_READ, NULL, OPEN_EXISTING, 
+                       FILE_ATTRIBUTE_NORMAL, NULL);
+
+    if (hFile == INVALID_HANDLE_VALUE) {
+        return -1;
+    }
+
+    ret = GetFinalPathNameByHandleA(hFile, realpath, srealpath, VOLUME_NAME_NT);
+
+    if (ret < srealpath) {
+        CloseHandle(hFile);
+        return -1;
+    }
+
+    CloseHandle(hFile);
+    return ret;
+}
+
+#endif
+
 /* cb_collect callback */
 static int in_calyptia_fleet_collect(struct flb_input_instance *ins,
                                      struct flb_config *config, 
@@ -712,6 +742,10 @@ static int in_calyptia_fleet_collect(struct flb_input_instance *ins,
     char *data;
     size_t b_sent;
     int ret = -1;
+#ifdef FLB_SYSTEM_WINDOWS
+    DWORD err;
+    LPSTR lpMsg;
+#endif
 
     u_conn = flb_upstream_conn_get(ctx->u);
 
@@ -865,7 +899,16 @@ static int in_calyptia_fleet_collect(struct flb_input_instance *ins,
             flb_sds_destroy(cfgoldname);
         }
 
-        symlink(cfgname, cfgnewname);
+        if (!link(cfgname, cfgnewname)) {
+#ifdef FLB_SYSTEM_WINDOWS
+            err = GetLastError();
+            FormatMessageA(FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_ALLOCATE_BUFFER,
+                          NULL, err, 0, &lpMsg, 0, NULL);
+            flb_plg_error(ctx->ins, "unable to create hard link: %s", lpMsg);
+#else
+            flb_errno();
+#endif
+        }
     }
 
     if (ctx->config_timestamp < time_last_modified) {
@@ -898,12 +941,18 @@ static int in_calyptia_fleet_collect(struct flb_input_instance *ins,
     FLB_INPUT_RETURN(ret);
 }
 
+#ifdef FLB_SYSTEM_WINDOWS
+#define _mkdir(a, b) mkdir(a)
+#else
+#define _mkdir(a, b) mkdir(a, b)
+#endif
+
 /* recursively create directories, based on:
  *   https://stackoverflow.com/a/2336245
  * who found it at:
  *   http://nion.modprobe.de/blog/archives/357-Recursive-directory-creation.html
  */
-static int _mkdir(const char *dir, int perms) {
+static int __mkdir(const char *dir, int perms) {
     char tmp[255];
     char *ptr = NULL;
     size_t len;
@@ -923,23 +972,23 @@ static int _mkdir(const char *dir, int perms) {
         if (*ptr == '/') {
             *ptr = 0;
             if (access(tmp, F_OK) != 0) {
-                ret = mkdir(tmp, perms);
+                ret = _mkdir(tmp, perms);
                 if (ret != 0) {
                     return ret;
                 }
             }
             *ptr = '/';
         }
     }
-    return mkdir(tmp, perms);
+    return _mkdir(tmp, perms);
 }
 
 static int create_fleet_directory(struct flb_in_calyptia_fleet_config *ctx)
 {
     flb_sds_t myfleetdir;
 
     if (access(ctx->config_dir, F_OK) != 0) {
-        if (_mkdir(ctx->config_dir, 0700) != 0) {
+        if (__mkdir(ctx->config_dir, 0700) != 0) {
             return -1;
         }
     }
@@ -956,7 +1005,7 @@ static int create_fleet_directory(struct flb_in_calyptia_fleet_config *ctx)
     }
 
     if (access(myfleetdir, F_OK) != 0) {
-        if (_mkdir(myfleetdir, 0700) !=0) {
+        if (__mkdir(myfleetdir, 0700) !=0) {
             return -1;
         }
     }

src/fluent-bit.c

@@ -616,13 +616,50 @@ static void flb_signal_handler(int signal)
     }
 }
 
+#ifdef FLB_SYSTEM_WINDOWS
+#include <ConsoleApi.h>
+
+static flb_ctx_t *handler_ctx = NULL;
+static struct flb_cf *handler_opts = NULL;
+static int handler_signal = 0;
+
+void flb_console_handler_set_ctx(flb_ctx_t *ctx, struct flb_cf *cf_opts)
+{
+    handler_ctx = ctx;
+    handler_opts = cf_opts;
+}
+
+static BOOL WINAPI flb_console_handler(DWORD evType)
+{
+    switch(evType) {
+    case 1 /* CTRL_BREAK_EVENT_1 */:
+        if (flb_bin_restarting == FLB_RELOAD_IDLE) {
+            flb_bin_restarting = FLB_RELOAD_IN_PROGRESS;
+            /* signal the main loop to execute reload. this is necessary since
+             * all signal handlers in win32 are executed on their own thread.
+             */
+            handler_signal = 1;
+            flb_bin_restarting = FLB_RELOAD_IDLE;
+        }
+        else {
+            flb_utils_error(FLB_ERR_RELOADING_IN_PROGRESS);
+        }
+        break;
+    }
+    return 1;
+}
+#endif
+
 static void flb_signal_init()
 {
     signal(SIGINT,  &flb_signal_handler_break_loop);
 #ifndef FLB_SYSTEM_WINDOWS
     signal(SIGQUIT, &flb_signal_handler_break_loop);
     signal(SIGHUP,  &flb_signal_handler);
     signal(SIGCONT, &flb_signal_handler);
+#else
+    /* Use SetConsoleCtrlHandler on windows to simulate SIGHUP */
+    SetConsoleCtrlHandler(flb_console_handler, 1);
 #endif
     signal(SIGTERM, &flb_signal_handler_break_loop);
     signal(SIGSEGV, &flb_signal_handler);
@@ -836,12 +873,15 @@ static int parse_trace_pipeline(flb_ctx_t *ctx, const char *pipeline, char **tra
     }
 
     mk_list_foreach(cur, parts) {
-	key = NULL;
-	value = NULL;
+        key = NULL;
+        value = NULL;
+
         part = mk_list_entry(cur, struct flb_split_entry, _head);
+
         if (parse_trace_pipeline_prop(ctx, part->value, &key, &value) == FLB_ERROR) {
             return FLB_ERROR;
         }
+
         if (strcmp(key, "input") == 0) {
             if (*trace_input != NULL) {
                 flb_free(*trace_input);
@@ -874,12 +914,14 @@ static int parse_trace_pipeline(flb_ctx_t *ctx, const char *pipeline, char **tra
                                    (char *)propname, strlen(propname),
                                    (char *)propval, strlen(propval));
         }
-	if (key != NULL) {
-		mk_mem_free(key);
-	}
-	if (value != NULL) {
-		flb_free(value);
-	}
+
+        if (key != NULL) {
+            mk_mem_free(key);
+        }
+
+        if (value != NULL) {
+            flb_free(value);
+        }
     }
 
     flb_utils_split_free(parts);
@@ -995,6 +1037,10 @@ int flb_main(int argc, char **argv)
     cf = config->cf_main;
     service = cf_opts->service;
 
+#ifdef FLB_SYSTEM_WINDOWS
+    flb_console_handler_set_ctx(ctx, cf_opts);
+#endif
+
     /* Add reference for cf_opts */
     config->cf_opts = cf_opts;
 
@@ -1315,8 +1361,19 @@ int flb_main(int argc, char **argv)
     while (ctx->status == FLB_LIB_OK && exit_signal == 0) {
         sleep(1);
 
+#ifdef FLB_SYSTEM_WINDOWS
+        if (handler_signal == 1) {
+            handler_signal = 0;
+            flb_reload(ctx, cf_opts);
+        }
+#endif
+
         /* set the context again before checking the status again */
         ctx = flb_context_get();
+
+#ifdef FLB_SYSTEM_WINDOWS
+        flb_console_handler_set_ctx(ctx, cf_opts);
+#endif
     }
 
     if (exit_signal) {