Merging upstream version 1.44.3.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

5 files changed, 1134 insertions, 0 deletions

diff --git a/fluent-bit/plugins/in_winlog/CMakeLists.txt b/fluent-bit/plugins/in_winlog/CMakeLists.txt
new file mode 100644
index 000000000..7b8b31563
--- /dev/null
+++ b/fluent-bit/plugins/in_winlog/CMakeLists.txt
@@ -0,0 +1,6 @@
+set(src
+    in_winlog.c
+    pack.c
+    winlog.c)
+
+FLB_PLUGIN(in_winlog "${src}" "advapi32")
diff --git a/fluent-bit/plugins/in_winlog/in_winlog.c b/fluent-bit/plugins/in_winlog/in_winlog.c
new file mode 100644
index 000000000..1d5398635
--- /dev/null
+++ b/fluent-bit/plugins/in_winlog/in_winlog.c
@@ -0,0 +1,267 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_compat.h>
+#include <fluent-bit/flb_input_plugin.h>
+#include <fluent-bit/flb_kernel.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_sqldb.h>
+#include "winlog.h"
+
+#define DEFAULT_INTERVAL_SEC  1
+#define DEFAULT_INTERVAL_NSEC 0
+#define DEFAULT_BUFFER_SIZE 0x7ffff /* Max size allowed by Win32 (512kb) */
+
+static int in_winlog_collect(struct flb_input_instance *ins,
+                             struct flb_config *config, void *in_context);
+
+static int in_winlog_init(struct flb_input_instance *in,
+                          struct flb_config *config, void *data)
+{
+    int ret;
+    const char *tmp;
+    struct mk_list *head;
+    struct winlog_channel *ch;
+    struct winlog_config *ctx;
+
+    /* Initialize context */
+    ctx = flb_calloc(1, sizeof(struct winlog_config));
+    if (!ctx) {
+        flb_errno();
+        return -1;
+    }
+    ctx->ins = in;
+
+    /* Load the config map */
+    ret = flb_input_config_map_set(in, (void *) ctx);
+    if (ret == -1) {
+        flb_free(ctx);
+        return -1;
+    }
+
+    /* Read Buffer */
+    ctx->bufsize = DEFAULT_BUFFER_SIZE;
+    ctx->buf = flb_malloc(ctx->bufsize);
+    if (!ctx->buf) {
+        flb_errno();
+        flb_free(ctx);
+    }
+
+    /* Open channels */
+    tmp = flb_input_get_property("channels", in);
+    if (!tmp) {
+        flb_plg_debug(ctx->ins, "no channel provided. listening to 'Application'");
+        tmp = "Application";
+    }
+
+    ctx->active_channel = winlog_open_all(tmp);
+    if (!ctx->active_channel) {
+        flb_plg_error(ctx->ins, "failed to open channels");
+        flb_free(ctx->buf);
+        flb_free(ctx);
+        return -1;
+    }
+
+    /* Initialize SQLite DB (optional) */
+    tmp = flb_input_get_property("db", in);
+    if (tmp) {
+        ctx->db = flb_sqldb_open(tmp, in->name, config);
+        if (!ctx->db) {
+            flb_plg_error(ctx->ins, "could not open/create database");
+            winlog_close_all(ctx->active_channel);
+            flb_free(ctx->buf);
+            flb_free(ctx);
+            return -1;
+        }
+
+        ret = flb_sqldb_query(ctx->db, SQL_CREATE_CHANNELS, NULL, NULL);
+        if (ret != FLB_OK) {
+            flb_plg_error(ctx->ins, "could not create 'channels' table");
+            flb_sqldb_close(ctx->db);
+            winlog_close_all(ctx->active_channel);
+            flb_free(ctx->buf);
+            flb_free(ctx);
+            return -1;
+        }
+
+        mk_list_foreach(head, ctx->active_channel) {
+            ch = mk_list_entry(head, struct winlog_channel, _head);
+            winlog_sqlite_load(ch, ctx->db);
+            flb_plg_debug(ctx->ins, "load channel<%s record=%u time=%u>",
+                          ch->name, ch->record_number, ch->time_written);
+        }
+    }
+
+    /* Set the context */
+    flb_input_set_context(in, ctx);
+
+    /* Set the collector */
+    ret = flb_input_set_collector_time(in,
+                                       in_winlog_collect,
+                                       ctx->interval_sec,
+                                       ctx->interval_nsec,
+                                       config);
+    if (ret == -1) {
+        flb_plg_error(ctx->ins, "could not set up a collector");
+    }
+    ctx->coll_fd = ret;
+
+    return 0;
+}
+
+static int in_winlog_read_channel(struct flb_input_instance *ins,
+                                  struct winlog_config *ctx,
+                                  struct winlog_channel *ch)
+{
+    unsigned int read;
+    char *ptr;
+    PEVENTLOGRECORD evt;
+    msgpack_packer mp_pck;
+    msgpack_sbuffer mp_sbuf;
+
+    if (winlog_read(ch, ctx->buf, ctx->bufsize, &read)) {
+        flb_plg_error(ctx->ins, "failed to read '%s'", ch->name);
+        return -1;
+    }
+    if (read == 0) {
+        return 0;
+    }
+    flb_plg_debug(ctx->ins, "read %u bytes from '%s'", read, ch->name);
+
+    msgpack_sbuffer_init(&mp_sbuf);
+    msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write);
+
+    ptr = ctx->buf;
+    while (ptr < ctx->buf + read) {
+        evt = (PEVENTLOGRECORD) ptr;
+
+        winlog_pack_event(&mp_pck, evt, ch, ctx);
+
+        ch->record_number = evt->RecordNumber;
+        ch->time_written = evt->TimeWritten;
+
+        ptr += evt->Length;
+    }
+
+    if (ctx->db) {
+        flb_plg_debug(ctx->ins, "save channel<%s record=%u time=%u>",
+                      ch->name, ch->record_number, ch->time_written);
+        winlog_sqlite_save(ch, ctx->db);
+    }
+
+    flb_input_log_append(ins, NULL, 0, mp_sbuf.data, mp_sbuf.size);
+
+    msgpack_sbuffer_destroy(&mp_sbuf);
+    return 0;
+}
+
+static int in_winlog_collect(struct flb_input_instance *ins,
+                             struct flb_config *config, void *in_context)
+{
+    struct winlog_config *ctx = in_context;
+    struct mk_list *head;
+    struct winlog_channel *ch;
+
+    mk_list_foreach(head, ctx->active_channel) {
+        ch = mk_list_entry(head, struct winlog_channel, _head);
+        in_winlog_read_channel(ins, ctx, ch);
+    }
+    return 0;
+}
+
+static void in_winlog_pause(void *data, struct flb_config *config)
+{
+    struct winlog_config *ctx = data;
+    flb_input_collector_pause(ctx->coll_fd, ctx->ins);
+}
+
+static void in_winlog_resume(void *data, struct flb_config *config)
+{
+    struct winlog_config *ctx = data;
+    flb_input_collector_resume(ctx->coll_fd, ctx->ins);
+}
+
+static int in_winlog_exit(void *data, struct flb_config *config)
+{
+    struct winlog_config *ctx = data;
+
+    if (!ctx) {
+        return 0;
+    }
+
+    winlog_close_all(ctx->active_channel);
+
+    if (ctx->db) {
+        flb_sqldb_close(ctx->db);
+    }
+    flb_free(ctx->buf);
+    flb_free(ctx);
+
+    return 0;
+}
+
+static struct flb_config_map config_map[] = {
+    {
+      FLB_CONFIG_MAP_STR, "channels", NULL,
+      0, FLB_FALSE, 0,
+      "Specify a comma-separated list of channels to read from"
+    },
+    {
+      FLB_CONFIG_MAP_STR, "db", NULL,
+      0, FLB_FALSE, 0,
+      "Specify DB file to save read offsets"
+    },
+    {
+      FLB_CONFIG_MAP_TIME, "interval_sec", "1s",
+      0, FLB_TRUE, offsetof(struct winlog_config, interval_sec),
+      "Set the polling interval for each channel"
+    },
+    {
+      FLB_CONFIG_MAP_INT, "interval_nsec", "0",
+      0, FLB_TRUE, offsetof(struct winlog_config, interval_nsec),
+      "Set the polling interval for each channel (sub seconds)"
+    },
+    {
+      FLB_CONFIG_MAP_BOOL, "string_inserts", "true",
+      0, FLB_TRUE, offsetof(struct winlog_config, string_inserts),
+      "Whether to include StringInserts in output records"
+    },
+    {
+      FLB_CONFIG_MAP_BOOL, "use_ansi", "false",
+      0, FLB_TRUE, offsetof(struct winlog_config, use_ansi),
+      "Use ANSI encoding on eventlog messages"
+    },
+
+    /* EOF */
+    {0}
+};
+
+struct flb_input_plugin in_winlog_plugin = {
+    .name         = "winlog",
+    .description  = "Windows Event Log",
+    .cb_init      = in_winlog_init,
+    .cb_pre_run   = NULL,
+    .cb_collect   = in_winlog_collect,
+    .cb_flush_buf = NULL,
+    .cb_pause     = in_winlog_pause,
+    .cb_resume    = in_winlog_resume,
+    .cb_exit      = in_winlog_exit,
+    .config_map   = config_map
+};
diff --git a/fluent-bit/plugins/in_winlog/pack.c b/fluent-bit/plugins/in_winlog/pack.c
new file mode 100644
index 000000000..4547d5514
--- /dev/null
+++ b/fluent-bit/plugins/in_winlog/pack.c
@@ -0,0 +1,451 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_compat.h>
+#include <fluent-bit/flb_mem.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_pack.h>
+#include <fluent-bit/flb_input_plugin.h>
+#include <msgpack.h>
+#include <sddl.h>
+#include <locale.h>
+#include "winlog.h"
+
+#define REGKEY_MAXLEN 256
+#define FMT_ISO8601 "%Y-%m-%d %H:%M:%S %z"
+#define FMT_EVTLOG L"SYSTEM\\CurrentControlSet\\Services\\EventLog\\%S\\%s"
+#define FMT_EVTALT L"SOFTWARE\\Microsoft\\Windows\\CurrentVersion\\WINEVT\\Publishers\\%s"
+
+/* 127 is the max number of function params */
+#define PARAM_MAXNUM 127
+
+#define SRCNAME(evt) ((wchar_t *) ((char *) (evt) + sizeof(EVENTLOGRECORD)))
+#define BINDATA(evt) ((unsigned char *) (evt) + (evt)->DataOffset)
+
+static void pack_nullstr(msgpack_packer *mp_pck)
+{
+    msgpack_pack_str(mp_pck, 0);
+    msgpack_pack_str_body(mp_pck, "", 0);
+}
+
+static int pack_wstr(msgpack_packer *mp_pck, wchar_t *wstr, int use_ansi)
+{
+    int size;
+    char *buf;
+    UINT codePage = CP_UTF8;
+    if (use_ansi) {
+        codePage = CP_ACP;
+    }
+
+    /* Compute the buffer size first */
+    size = WideCharToMultiByte(codePage, 0, wstr, -1, NULL, 0, NULL, NULL);
+    if (size == 0) {
+        return -1;
+    }
+
+    buf = flb_malloc(size);
+    if (buf == NULL) {
+        flb_errno();
+        return -1;
+    }
+
+    /* Convert UTF-16 into UTF-8/System code Page encoding */
+    size = WideCharToMultiByte(codePage, 0, wstr, -1, buf, size, NULL, NULL);
+    if (size == 0) {
+        flb_free(buf);
+        return -1;
+    }
+
+    /* Pack buf except the trailing '\0' */
+    msgpack_pack_str(mp_pck, size - 1);
+    msgpack_pack_str_body(mp_pck, buf, size - 1);
+    flb_free(buf);
+    return 0;
+}
+
+static int pack_time(msgpack_packer *mp_pck, int time)
+{
+    size_t len;
+    struct tm tm;
+    char buf[64];
+    _locale_t locale;
+
+    if (_localtime32_s(&tm, &time)) {
+        flb_errno();
+        return -1;
+    }
+
+    locale = _get_current_locale();
+    if (locale == NULL) {
+        return -1;
+    }
+
+    len = _strftime_l(buf, 64, FMT_ISO8601, &tm, locale);
+    if (len == 0) {
+        flb_errno();
+        _free_locale(locale);
+        return -1;
+    }
+    _free_locale(locale);
+    msgpack_pack_str(mp_pck, len);
+    msgpack_pack_str_body(mp_pck, buf, len);
+
+    return 0;
+}
+
+static int pack_event_type(msgpack_packer *mp_pck, int type)
+{
+    switch (type) {
+    case EVENTLOG_SUCCESS:
+        msgpack_pack_str(mp_pck, 7);
+        msgpack_pack_str_body(mp_pck, "Success", 7);
+        break;
+    case EVENTLOG_INFORMATION_TYPE:
+        msgpack_pack_str(mp_pck, 11);
+        msgpack_pack_str_body(mp_pck, "Information", 11);
+        break;
+    case EVENTLOG_WARNING_TYPE:
+        msgpack_pack_str(mp_pck, 7);
+        msgpack_pack_str_body(mp_pck, "Warning", 7);
+        break;
+    case EVENTLOG_ERROR_TYPE:
+        msgpack_pack_str(mp_pck, 5);
+        msgpack_pack_str_body(mp_pck, "Error", 5);
+        break;
+    case EVENTLOG_AUDIT_SUCCESS:
+        msgpack_pack_str(mp_pck, 12);
+        msgpack_pack_str_body(mp_pck, "SuccessAudit", 12);
+        break;
+    case EVENTLOG_AUDIT_FAILURE:
+        msgpack_pack_str(mp_pck, 12);
+        msgpack_pack_str_body(mp_pck, "FailureAudit", 12);
+        break;
+    default:
+        return -1;
+    }
+    return 0;
+}
+
+static int pack_binary(msgpack_packer *mp_pck, unsigned char *bin, int len)
+{
+    const char *hex = "0123456789abcdef";
+    char *buf;
+    int size = len * 2;
+    int i;
+
+    if (len == 0) {
+        pack_nullstr(mp_pck);
+        return 0;
+    }
+
+    buf = flb_malloc(size);
+    if (buf == NULL) {
+        flb_errno();
+        return -1;
+    }
+
+    for (i = 0; i < len; i++) {
+        buf[2*i]   = hex[bin[i] / 16];
+        buf[2*i+1] = hex[bin[i] % 16];
+    }
+    msgpack_pack_str(mp_pck, size);
+    msgpack_pack_str_body(mp_pck, buf, size);
+    flb_free(buf);
+    return 0;
+}
+
+static int pack_sid(msgpack_packer *mp_pck, PEVENTLOGRECORD evt,
+                    struct winlog_config *ctx)
+{
+    size_t size;
+    char *buf;
+    char *sid = (char *) evt + evt->UserSidOffset;
+
+    if (evt->UserSidLength == 0) {
+        pack_nullstr(mp_pck);
+        return 0;
+    }
+
+    if (!ConvertSidToStringSidA(sid, &buf)) {
+        flb_plg_error(ctx->ins, "fail to convert SID: %i", GetLastError());
+        return -1;
+    }
+
+    size = strlen(buf);
+    msgpack_pack_str(mp_pck, size);
+    msgpack_pack_str_body(mp_pck, buf, size);
+
+    LocalFree(buf);
+    return 0;
+}
+
+static wchar_t *read_registry(HKEY hkey, wchar_t *key, wchar_t *val)
+{
+    int ret;
+    int size;
+    wchar_t *buf;
+    unsigned int flags = RRF_RT_REG_EXPAND_SZ | RRF_RT_REG_SZ;
+
+    /* Get the buffer size first */
+    ret = RegGetValueW(hkey, key, val, flags, NULL, NULL, &size);
+    if (ret != ERROR_SUCCESS) {
+        return NULL;
+    }
+
+    buf = flb_malloc(size);
+    if (buf == NULL) {
+        flb_errno();
+        return NULL;
+    }
+
+    /* Read data into buffer */
+    ret = RegGetValueW(hkey, key, val, flags, NULL, buf, &size);
+    if (ret != ERROR_SUCCESS) {
+        flb_free(buf);
+        return NULL;
+    }
+    return buf;
+}
+
+static wchar_t *query_guid(wchar_t *guid)
+{
+    int ret;
+    wchar_t key[REGKEY_MAXLEN];
+
+    ret = swprintf_s(key, REGKEY_MAXLEN, FMT_EVTALT, guid);
+    if (ret == -1) {
+        flb_errno();
+        return NULL;
+    }
+
+    return read_registry(HKEY_LOCAL_MACHINE, key, L"MessageFileName");
+}
+
+static int pack_message(msgpack_packer *mp_pck, PEVENTLOGRECORD evt,
+                        struct winlog_channel *ch, struct winlog_config *ctx)
+{
+    int ret;
+    int i;
+    HMODULE hfile;
+    wchar_t key[REGKEY_MAXLEN];
+    wchar_t *msg;
+    wchar_t *paths;
+    wchar_t *path;
+    wchar_t *guid;
+    wchar_t *state;
+    wchar_t *tmp;
+    DWORD_PTR *args = NULL;
+
+    ret = swprintf_s(key, REGKEY_MAXLEN, FMT_EVTLOG, ch->name, SRCNAME(evt));
+    if (ret == -1) {
+        flb_errno();
+        return -1;
+    }
+
+    guid = read_registry(HKEY_LOCAL_MACHINE, key, L"ProviderGuid");
+    if (guid) {
+        paths = query_guid(guid);
+        flb_free(guid);
+    }
+    else {
+        paths = read_registry(HKEY_LOCAL_MACHINE, key, L"EventMessageFile");
+    }
+
+    if (paths == NULL) {
+        return -1;
+    }
+
+    if (evt->NumStrings) {
+        args = flb_calloc(PARAM_MAXNUM, sizeof(DWORD_PTR));
+        if (args == NULL) {
+            flb_errno();
+            flb_free(paths);
+            return -1;
+        }
+
+        tmp = (wchar_t *) ((char *) evt + evt->StringOffset);
+        for (i = 0; i < evt->NumStrings; i++) {
+            args[i] = (DWORD_PTR) tmp;
+            tmp += wcslen(tmp) + 1;
+        }
+    }
+
+    path = paths;
+    wcstok_s(path, L";", &state);
+    while (path) {
+        hfile = LoadLibraryExW(path, NULL, LOAD_LIBRARY_AS_DATAFILE);
+        if (hfile == NULL) {
+            path = wcstok_s(NULL , L";", &state);
+            continue;
+        }
+
+        ret = FormatMessageW(FORMAT_MESSAGE_FROM_HMODULE |
+                             FORMAT_MESSAGE_ALLOCATE_BUFFER |
+                             FORMAT_MESSAGE_ARGUMENT_ARRAY,
+                             hfile,        /* lpSource */
+                             evt->EventID, /* dwMessageId */
+                             0,            /* dwLanguageId */
+                             (LPWSTR) &msg,/* lpBuffer */
+                             0,            /* nSize */
+                             (va_list *) args);
+        if (ret > 0) {
+            ret = pack_wstr(mp_pck, msg, ctx->use_ansi);
+            LocalFree(msg);
+            FreeLibrary(hfile);
+            flb_free(paths);
+            flb_free(args);
+            return ret;
+        }
+        FreeLibrary(hfile);
+        path = wcstok_s(NULL , L";", &state);
+    }
+
+    flb_free(paths);
+    flb_free(args);
+    return -1;
+}
+
+static void pack_strings(msgpack_packer *mp_pck, PEVENTLOGRECORD evt, int use_ansi)
+{
+    int i;
+    int len;
+    wchar_t *wstr = (wchar_t *) ((char *) evt + evt->StringOffset);
+
+    msgpack_pack_array(mp_pck, evt->NumStrings);
+
+    for (i = 0; i < evt->NumStrings; i++) {
+        if (pack_wstr(mp_pck, wstr, use_ansi)) {
+            pack_nullstr(mp_pck);
+        }
+        wstr += wcslen(wstr) + 1;
+    }
+}
+
+void winlog_pack_event(msgpack_packer *mp_pck, PEVENTLOGRECORD evt,
+                       struct winlog_channel *ch, struct winlog_config *ctx)
+{
+    wchar_t *source_name = SRCNAME(evt);
+    wchar_t *computer_name = source_name + wcslen(source_name) + 1;
+    size_t len;
+    int count = 13;
+
+    if (ctx->string_inserts) {
+        count++;
+    }
+
+    msgpack_pack_array(mp_pck, 2);
+    flb_pack_time_now(mp_pck);
+
+    msgpack_pack_map(mp_pck, count);
+
+    /* RecordNumber */
+    msgpack_pack_str(mp_pck, 12);
+    msgpack_pack_str_body(mp_pck, "RecordNumber", 12);
+    msgpack_pack_uint32(mp_pck, evt->RecordNumber);
+
+    /* TimeGenerated */
+    msgpack_pack_str(mp_pck, 13);
+    msgpack_pack_str_body(mp_pck, "TimeGenerated", 13);
+    if (pack_time(mp_pck, evt->TimeGenerated)) {
+        flb_plg_error(ctx->ins, "invalid TimeGenerated %i", evt->TimeGenerated);
+        pack_nullstr(mp_pck);
+    }
+
+    /* TimeWritten */
+    msgpack_pack_str(mp_pck, 11);
+    msgpack_pack_str_body(mp_pck, "TimeWritten", 11);
+    if (pack_time(mp_pck, evt->TimeWritten)) {
+        flb_plg_error(ctx->ins, "invalid TimeWritten %i", evt->TimeWritten);
+        pack_nullstr(mp_pck);
+    }
+
+    /* EventId */
+    msgpack_pack_str(mp_pck, 7);
+    msgpack_pack_str_body(mp_pck, "EventID", 7);
+    msgpack_pack_uint16(mp_pck, evt->EventID & 0xffff);
+
+    /* Qualifiers */
+    msgpack_pack_str(mp_pck, 10);
+    msgpack_pack_str_body(mp_pck, "Qualifiers", 10);
+    msgpack_pack_uint16(mp_pck, evt->EventID >> 16);
+
+    /* EventType */
+    msgpack_pack_str(mp_pck, 9);
+    msgpack_pack_str_body(mp_pck, "EventType", 9);
+    if (pack_event_type(mp_pck, evt->EventType)) {
+        flb_plg_error(ctx->ins, "invalid EventType %i", evt->EventType);
+        pack_nullstr(mp_pck);
+    }
+
+    /* EventCategory */
+    msgpack_pack_str(mp_pck, 13);
+    msgpack_pack_str_body(mp_pck, "EventCategory", 13);
+    msgpack_pack_uint16(mp_pck, evt->EventCategory);
+
+    /* Channel */
+    len = strlen(ch->name);
+    msgpack_pack_str(mp_pck, 7);
+    msgpack_pack_str_body(mp_pck, "Channel", 7);
+    msgpack_pack_str(mp_pck, len);
+    msgpack_pack_str_body(mp_pck, ch->name, len);
+
+    /* Source Name */
+    msgpack_pack_str(mp_pck, 10);
+    msgpack_pack_str_body(mp_pck, "SourceName", 10);
+    if (pack_wstr(mp_pck, source_name, ctx->use_ansi)) {
+        flb_plg_error(ctx->ins, "invalid SourceName '%ls'", source_name);
+        pack_nullstr(mp_pck);
+    }
+
+    /* Computer Name */
+    msgpack_pack_str(mp_pck, 12);
+    msgpack_pack_str_body(mp_pck, "ComputerName", 12);
+    if (pack_wstr(mp_pck, computer_name, ctx->use_ansi)) {
+        flb_plg_error(ctx->ins, "invalid ComputerName '%ls'", computer_name);
+        pack_nullstr(mp_pck);
+    }
+
+    /* Event-specific Data */
+    msgpack_pack_str(mp_pck, 4);
+    msgpack_pack_str_body(mp_pck, "Data", 4);
+    if (pack_binary(mp_pck, BINDATA(evt), evt->DataLength)) {
+        pack_nullstr(mp_pck);
+    }
+
+    /* Sid */
+    msgpack_pack_str(mp_pck, 3);
+    msgpack_pack_str_body(mp_pck, "Sid", 3);
+    if (pack_sid(mp_pck, evt, ctx)) {
+        pack_nullstr(mp_pck);
+    }
+
+    /* Message */
+    msgpack_pack_str(mp_pck, 7);
+    msgpack_pack_str_body(mp_pck, "Message", 7);
+    if (pack_message(mp_pck, evt, ch, ctx)) {
+        pack_nullstr(mp_pck);
+    }
+
+    /* StringInserts (optional) */
+    if (ctx->string_inserts) {
+        msgpack_pack_str(mp_pck, 13);
+        msgpack_pack_str_body(mp_pck, "StringInserts", 13);
+        pack_strings(mp_pck, evt, ctx->use_ansi);
+    }
+}
diff --git a/fluent-bit/plugins/in_winlog/winlog.c b/fluent-bit/plugins/in_winlog/winlog.c
new file mode 100644
index 000000000..b6064b1f1
--- /dev/null
+++ b/fluent-bit/plugins/in_winlog/winlog.c
@@ -0,0 +1,300 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#include <fluent-bit/flb_compat.h>
+#include <fluent-bit/flb_info.h>
+#include <fluent-bit/flb_utils.h>
+#include <fluent-bit/flb_sqldb.h>
+#include <fluent-bit/flb_input.h>
+#include "winlog.h"
+
+struct winlog_channel *winlog_open(const char *channel)
+{
+    struct winlog_channel *ch;
+
+    ch = flb_calloc(1, sizeof(struct winlog_channel));
+    if (!ch) {
+        flb_errno();
+        return NULL;
+    }
+
+    ch->name = flb_strdup(channel);
+    if (!ch->name) {
+        flb_errno();
+        flb_free(ch);
+        return NULL;
+    }
+
+    ch->h = OpenEventLogA(NULL, channel);
+    if (!ch->h) {
+        flb_error("[in_winlog] cannot open '%s' (%i)", channel, GetLastError());
+        flb_free(ch->name);
+        flb_free(ch);
+        return NULL;
+    }
+
+    return ch;
+}
+
+void winlog_close(struct winlog_channel *ch)
+{
+    flb_free(ch->name);
+    CloseEventLog(ch->h);
+    flb_free(ch);
+}
+
+/*
+ * This routine is called when Windows Event Log was cleared
+ * while reading (e.g. running Clear-EventLog on PowerShell).
+ *
+ * In such a case, the only neat thing to do is to reopen the
+ * channel and start reading from the beginning.
+ */
+int winlog_on_cleared(struct winlog_channel *ch)
+{
+    HANDLE h;
+
+    h = OpenEventLogA(NULL, ch->name);
+    if (!h) {
+        flb_error("[in_winlog] cannot open '%s' (%i)", ch->name, GetLastError());
+        return -1;
+    }
+
+    if (ch->h) {
+        CloseEventLog(ch->h);
+    }
+
+    ch->h = h;
+    ch->seek = 0;
+    return 0;
+}
+
+
+/*
+ * ReadEventLog() has a known bug that SEEK_READ fails when the log file
+ * is too big.
+ *
+ * winlog_seek() is a workaround for the issue, which emulates seek
+ * by reading the stream until it reaches the target record.
+ *
+ * https://support.microsoft.com/en-hk/help/177199/
+ */
+static int winlog_seek(struct winlog_channel *ch, char *buf,
+                       unsigned int size, unsigned int *read)
+{
+    char *p;
+    char *end;
+    PEVENTLOGRECORD evt;
+
+    ch->seek = 0;
+    while (1) {
+        if (winlog_read(ch, buf, size, read)) {
+            return -1;
+        }
+        if (*read == 0) {
+            flb_trace("[in_winlog] seek '%s' to EOF", ch->name);
+            return 0;
+        }
+
+        p = buf;
+        end = buf + *read;
+        while (p < end) {
+            evt = (PEVENTLOGRECORD) p;
+
+            /* If the record is newer than the last record we've read,
+             * stop immediately.
+             */
+            if (evt->TimeWritten > ch->time_written) {
+                *read = (end - p);
+                memmove(buf, p, *read);
+                flb_trace("[in_winlog] seek '%s' to RecordNumber=%u (time)",
+                          ch->name, evt->RecordNumber);
+                return 0;
+            }
+            if (evt->TimeWritten == ch->time_written) {
+
+               /* If the record was written at the same time, compare
+                * the record number.
+                *
+                * Note! Since Windows would reset RecordNumber occasionally,
+                * this comparison is not completely reliable.
+                */
+                if (evt->RecordNumber > ch->record_number) {
+                    *read = (end - p);
+                    memmove(buf, p, *read);
+                    flb_trace("[in_winlog] seek '%s' to RecordNumber=%u",
+                              ch->name, evt->RecordNumber);
+                    return 0;
+                }
+            }
+            p += evt->Length;
+        }
+    }
+}
+
+/*
+ * Read from an open Windows Event Log channel.
+ */
+int winlog_read(struct winlog_channel *ch, char *buf, unsigned int size,
+                unsigned int *read)
+{
+    unsigned int flags;
+    unsigned int req;
+    unsigned int err;
+
+    if (ch->seek) {
+        flags = EVENTLOG_SEEK_READ;
+    } else {
+        flags = EVENTLOG_SEQUENTIAL_READ | EVENTLOG_FORWARDS_READ;
+    }
+
+    /*
+     * Note: ReadEventLogW() ignores `ch->record_number` (dwRecordOffset)
+     * if EVENTLOG_SEEK_READ is not set.
+     */
+    if (!ReadEventLogW(ch->h, flags, ch->record_number, buf, size, read, &req)) {
+        switch (err = GetLastError()) {
+            case ERROR_HANDLE_EOF:
+                break;
+            case ERROR_INVALID_PARAMETER:
+                return winlog_seek(ch, buf, size, read);
+            case ERROR_EVENTLOG_FILE_CHANGED:
+                flb_info("[in_winlog] channel '%s' is cleared. reopen it.", ch->name);
+                return winlog_on_cleared(ch);
+            default:
+                flb_error("[in_winlog] cannot read '%s' (%i)", ch->name, err);
+                return -1;
+        }
+    }
+    ch->seek = 0;
+    return 0;
+}
+
+/*
+ * Open multiple channels at once. The return value is a linked
+ * list of window_channel objects.
+ *
+ * "channels" are comma-separated names like "Setup,Security".
+ */
+struct mk_list *winlog_open_all(const char *channels)
+{
+    char *tmp;
+    char *channel;
+    char *state;
+    struct winlog_channel *ch;
+    struct mk_list *list;
+
+    tmp = flb_strdup(channels);
+    if (!tmp) {
+        flb_errno();
+        return NULL;
+    }
+
+    list = flb_malloc(sizeof(struct mk_list));
+    if (!list) {
+        flb_errno();
+        flb_free(tmp);
+        return NULL;
+    }
+    mk_list_init(list);
+
+    channel = strtok_s(tmp , ",", &state);
+    while (channel) {
+        ch = winlog_open(channel);
+        if (!ch) {
+            flb_free(tmp);
+            winlog_close_all(list);
+            return NULL;
+        }
+        mk_list_add(&ch->_head, list);
+        channel = strtok_s(NULL, ",", &state);
+    }
+    flb_free(tmp);
+    return list;
+}
+
+void winlog_close_all(struct mk_list *list)
+{
+    struct winlog_channel *ch;
+    struct mk_list *head;
+    struct mk_list *tmp;
+
+    mk_list_foreach_safe(head, tmp, list) {
+        ch = mk_list_entry(head, struct winlog_channel, _head);
+        mk_list_del(&ch->_head);
+        winlog_close(ch);
+    }
+    flb_free(list);
+}
+
+/*
+ * Callback function for flb_sqldb_query().
+ */
+static int winlog_sqlite_callback(void *data, int argc, char **argv, char **cols)
+{
+    struct winlog_sqlite_record *p = data;
+
+    p->name = argv[0];
+    p->record_number = (unsigned int) strtoul(argv[1], NULL, 10);
+    p->time_written = (unsigned int) strtoul(argv[2], NULL, 10);
+    p->created = (unsigned int) strtoul(argv[3], NULL, 10);
+    return 0;
+}
+
+/*
+ * Load the read offset from SQLite DB.
+ */
+int winlog_sqlite_load(struct winlog_channel *ch, struct flb_sqldb *db)
+{
+    int ret;
+    char query[1024];
+    struct winlog_sqlite_record record = {0};
+
+    snprintf(query, sizeof(query) - 1, SQL_GET_CHANNEL, ch->name);
+
+    ret = flb_sqldb_query(db, query, winlog_sqlite_callback, &record);
+    if (ret == FLB_ERROR) {
+        return -1;
+    }
+
+    if (record.name) {
+        ch->record_number = record.record_number;
+        ch->time_written = record.time_written;
+        ch->seek = 1;
+    }
+    return 0;
+}
+
+/*
+ * Save the read offset into SQLite DB.
+ */
+int winlog_sqlite_save(struct winlog_channel *ch, struct flb_sqldb *db)
+{
+    int ret;
+    char query[1024];
+
+    snprintf(query, sizeof(query) - 1, SQL_UPDATE_CHANNEL,
+             ch->name, ch->record_number, ch->time_written, time(NULL));
+
+    ret = flb_sqldb_query(db, query, NULL, NULL);
+    if (ret == FLB_ERROR) {
+        return -1;
+    }
+    return 0;
+}
diff --git a/fluent-bit/plugins/in_winlog/winlog.h b/fluent-bit/plugins/in_winlog/winlog.h
new file mode 100644
index 000000000..007a996ed
--- /dev/null
+++ b/fluent-bit/plugins/in_winlog/winlog.h
@@ -0,0 +1,110 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  Fluent Bit
+ *  ==========
+ *  Copyright (C) 2015-2022 The Fluent Bit Authors
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+#ifndef FLB_WINLOG_H
+#define FLB_WINLOG_H
+
+struct winlog_config {
+    unsigned int interval_sec;
+    unsigned int interval_nsec;
+    unsigned int bufsize;
+    int string_inserts;
+    int use_ansi;
+    char *buf;
+    struct mk_list *active_channel;
+    struct flb_sqldb *db;
+    flb_pipefd_t coll_fd;
+    struct flb_input_instance *ins;
+};
+
+struct winlog_channel {
+    HANDLE h;
+    char *name;
+    unsigned int record_number;
+    unsigned int time_written;
+    unsigned int seek;
+    struct mk_list _head;
+};
+
+struct winlog_sqlite_record {
+    char *name;
+    unsigned int record_number;
+    unsigned int time_written;
+    unsigned int created;
+};
+
+/*
+ * Open a Windows Event Log channel.
+ */
+struct winlog_channel *winlog_open(const char *channel);
+void winlog_close(struct winlog_channel *ch);
+
+/*
+ * Read records from a channel.
+ */
+int winlog_read(struct winlog_channel *ch, char *buf, unsigned int size, unsigned int *read);
+
+/*
+ * A bulk API to handle multiple channels at once using mk_list.
+ *
+ * "channels" are comma-separated names like "Setup,Security".
+ */
+struct mk_list *winlog_open_all(const char *channels);
+void winlog_close_all(struct mk_list *list);
+
+void winlog_pack_event(msgpack_packer *mp_pck, PEVENTLOGRECORD evt,
+                       struct winlog_channel *ch, struct winlog_config *ctx);
+
+/*
+ * Save the read offset to disk.
+ */
+int winlog_sqlite_load(struct winlog_channel *ch, struct flb_sqldb *db);
+int winlog_sqlite_save(struct winlog_channel *ch, struct flb_sqldb *db);
+
+/*
+ * SQL templates
+ */
+#define SQL_CREATE_CHANNELS                                         \
+    "CREATE TABLE IF NOT EXISTS in_winlog_channels ("               \
+    "  name    TEXT PRIMARY KEY,"                                   \
+    "  record_number INTEGER,"                                      \
+    "  time_written INTEGER,"                                       \
+    "  created INTEGER"                                             \
+    ");"
+
+#define SQL_GET_CHANNEL                                             \
+    "SELECT name, record_number, time_written, created"             \
+    " FROM in_winlog_channels WHERE name = '%s';"
+
+/*
+ * This uses UPCERT i.e. execute INSERT first and fall back to
+ * UPDATE if the entry already exists. It saves the trouble of
+ * doing an existence check manually.
+ *
+ * https://www.sqlite.org/lang_UPSERT.html
+ */
+#define SQL_UPDATE_CHANNEL                                          \
+    "INSERT INTO in_winlog_channels"                                \
+    "  (name, record_number, time_written, created)"                \
+    "  VALUES ('%s', %u, %u, %llu)"                                   \
+    "  ON CONFLICT(name) DO UPDATE"                                 \
+    "  SET record_number = excluded.record_number,"                 \
+    "      time_written = excluded.time_written"
+
+#endif