From c341cb6042402e1d2c2fde20e74d01055899f4d9 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Tue, 7 Jul 2026 01:25:08 +0900 Subject: [PATCH 1/6] in_etw: implement Event Tracing for Windows input Signed-off-by: Hiroshi Hatake --- cmake/plugins_options.cmake | 1 + cmake/windows-setup.cmake | 1 + plugins/CMakeLists.txt | 1 + plugins/in_etw/CMakeLists.txt | 4 + plugins/in_etw/etw.c | 1269 +++++++++++++++++++++++++++++++++ plugins/in_etw/etw.h | 73 ++ 6 files changed, 1349 insertions(+) create mode 100644 plugins/in_etw/CMakeLists.txt create mode 100644 plugins/in_etw/etw.c create mode 100644 plugins/in_etw/etw.h diff --git a/cmake/plugins_options.cmake b/cmake/plugins_options.cmake index 7fbd0a96bd5..aa8bfd05bda 100644 --- a/cmake/plugins_options.cmake +++ b/cmake/plugins_options.cmake @@ -61,6 +61,7 @@ DEFINE_OPTION(FLB_IN_TCP "Enable TCP input plugin" DEFINE_OPTION(FLB_IN_THERMAL "Enable Thermal plugin" ON) DEFINE_OPTION(FLB_IN_UDP "Enable UDP input plugin" ON) DEFINE_OPTION(FLB_IN_UNIX_SOCKET "Enable Unix socket input plugin" OFF) +DEFINE_OPTION(FLB_IN_ETW "Enable Event Tracing for Windows input plugin" OFF) DEFINE_OPTION(FLB_IN_WINLOG "Enable Windows Log input plugin" OFF) DEFINE_OPTION(FLB_IN_WINDOWS_EXPORTER_METRICS "Enable windows exporter metrics input plugin" ON) DEFINE_OPTION(FLB_IN_WINEVTLOG "Enable Windows EvtLog input plugin" OFF) diff --git a/cmake/windows-setup.cmake b/cmake/windows-setup.cmake index f253382a0cc..48a749b4d29 100644 --- a/cmake/windows-setup.cmake +++ b/cmake/windows-setup.cmake @@ -71,6 +71,7 @@ if(FLB_WINDOWS_DEFAULTS) set(FLB_IN_SYSTEMD No) set(FLB_IN_DUMMY Yes) set(FLB_IN_NETIF No) + set(FLB_IN_ETW Yes) set(FLB_IN_WINLOG Yes) set(FLB_IN_WINSTAT Yes) set(FLB_IN_WINEVTLOG Yes) diff --git a/plugins/CMakeLists.txt b/plugins/CMakeLists.txt index 36fcc94ad12..c885f51fda3 100644 --- a/plugins/CMakeLists.txt +++ b/plugins/CMakeLists.txt @@ -311,6 +311,7 @@ if (FLB_STREAM_PROCESSOR) endif() if (FLB_SYSTEM_WINDOWS) + REGISTER_IN_PLUGIN("in_etw") REGISTER_IN_PLUGIN("in_winlog") REGISTER_IN_PLUGIN("in_winstat") REGISTER_IN_PLUGIN("in_winevtlog") diff --git a/plugins/in_etw/CMakeLists.txt b/plugins/in_etw/CMakeLists.txt new file mode 100644 index 00000000000..52fb0e7bcf8 --- /dev/null +++ b/plugins/in_etw/CMakeLists.txt @@ -0,0 +1,4 @@ +set(src + etw.c) + +FLB_PLUGIN(in_etw "${src}" "advapi32;tdh;ole32") diff --git a/plugins/in_etw/etw.c b/plugins/in_etw/etw.c new file mode 100644 index 00000000000..d4b1aff5fa1 --- /dev/null +++ b/plugins/in_etw/etw.c @@ -0,0 +1,1269 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2026 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 +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include + +#include "etw.h" + +#define FLB_IN_ETW_UNIX_EPOCH_IN_100NS 116444736000000000ULL +#define FLB_IN_ETW_PROVIDER_MAP_SIZE 14 + +static void pack_cstr(msgpack_packer *mp_pck, const char *str) +{ + size_t len; + + len = strlen(str); + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, str, len); +} + +static int guid_to_string(const GUID *guid, char *buf, size_t size) +{ + int ret; + + ret = snprintf(buf, size, + "{%08" PRIx32 "-%04x-%04x-%02x%02x-" + "%02x%02x%02x%02x%02x%02x}", + (uint32_t) guid->Data1, + guid->Data2, + guid->Data3, + guid->Data4[0], + guid->Data4[1], + guid->Data4[2], + guid->Data4[3], + guid->Data4[4], + guid->Data4[5], + guid->Data4[6], + guid->Data4[7]); + + if (ret < 0 || (size_t) ret >= size) { + return -1; + } + + return 0; +} + +static int parse_guid(const char *str, GUID *guid) +{ + unsigned int d1; + unsigned int d2; + unsigned int d3; + unsigned int d4[8]; + int ret; + + ret = sscanf(str, + "{%8x-%4x-%4x-%2x%2x-%2x%2x%2x%2x%2x%2x}", + &d1, &d2, &d3, + &d4[0], &d4[1], &d4[2], &d4[3], + &d4[4], &d4[5], &d4[6], &d4[7]); + + if (ret != 11) { + ret = sscanf(str, + "%8x-%4x-%4x-%2x%2x-%2x%2x%2x%2x%2x%2x", + &d1, &d2, &d3, + &d4[0], &d4[1], &d4[2], &d4[3], + &d4[4], &d4[5], &d4[6], &d4[7]); + } + + if (ret != 11) { + return -1; + } + + guid->Data1 = d1; + guid->Data2 = (unsigned short) d2; + guid->Data3 = (unsigned short) d3; + guid->Data4[0] = (unsigned char) d4[0]; + guid->Data4[1] = (unsigned char) d4[1]; + guid->Data4[2] = (unsigned char) d4[2]; + guid->Data4[3] = (unsigned char) d4[3]; + guid->Data4[4] = (unsigned char) d4[4]; + guid->Data4[5] = (unsigned char) d4[5]; + guid->Data4[6] = (unsigned char) d4[6]; + guid->Data4[7] = (unsigned char) d4[7]; + + return 0; +} + +static int parse_uint64(const char *str, ULONGLONG *out) +{ + char *end; + unsigned long long value; + + if (str == NULL || str[0] == '\0') { + return -1; + } + + value = strtoull(str, &end, 0); + if (end == str || *end != '\0') { + return -1; + } + + *out = value; + + return 0; +} + +static WCHAR *utf8_to_wide(const char *str) +{ + int size; + WCHAR *buf; + + size = MultiByteToWideChar(CP_UTF8, 0, str, -1, NULL, 0); + if (size == 0) { + return NULL; + } + + buf = flb_calloc(size, sizeof(WCHAR)); + if (buf == NULL) { + flb_errno(); + return NULL; + } + + if (MultiByteToWideChar(CP_UTF8, 0, str, -1, buf, size) == 0) { + flb_free(buf); + return NULL; + } + + return buf; +} + +static char *wide_to_utf8(const WCHAR *str, int bytes, int *out_len) +{ + int wide_len; + int size; + char *buf; + + if (str == NULL) { + return NULL; + } + + if (bytes > 0) { + wide_len = bytes / sizeof(WCHAR); + while (wide_len > 0 && str[wide_len - 1] == L'\0') { + wide_len--; + } + } + else { + wide_len = -1; + } + + size = WideCharToMultiByte(CP_UTF8, 0, str, wide_len, NULL, 0, NULL, NULL); + if (size == 0) { + return NULL; + } + + buf = flb_malloc(size + 1); + if (buf == NULL) { + flb_errno(); + return NULL; + } + + if (WideCharToMultiByte(CP_UTF8, 0, str, wide_len, buf, size, NULL, NULL) == 0) { + flb_free(buf); + return NULL; + } + + buf[size] = '\0'; + *out_len = size; + + return buf; +} + +static void pack_wide_string(msgpack_packer *mp_pck, const WCHAR *str, int bytes) +{ + int len; + char *utf8; + + utf8 = wide_to_utf8(str, bytes, &len); + if (utf8 == NULL) { + msgpack_pack_nil(mp_pck); + return; + } + + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, utf8, len); + flb_free(utf8); +} + +static int resolve_provider_name(struct flb_etw *ctx, const WCHAR *name, GUID *guid) +{ + ULONG size; + TDHSTATUS status; + PROVIDER_ENUMERATION_INFO *providers; + PROVIDER_ENUMERATION_INFO *new_providers; + TRACE_PROVIDER_INFO *info; + WCHAR *provider_name; + ULONG i; + + size = 0; + providers = NULL; + + while (1) { + status = TdhEnumerateProviders(providers, &size); + if (status == ERROR_SUCCESS) { + break; + } + + if (status != ERROR_INSUFFICIENT_BUFFER) { + flb_plg_error(ctx->ins, + "TdhEnumerateProviders failed while resolving '%S' (status=%lu)", + name, status); + flb_free(providers); + return -1; + } + + new_providers = flb_realloc(providers, size); + if (new_providers == NULL) { + flb_errno(); + flb_free(providers); + return -1; + } + providers = new_providers; + } + + for (i = 0; i < providers->NumberOfProviders; i++) { + info = &providers->TraceProviderInfoArray[i]; + provider_name = (WCHAR *) ((char *) providers + info->ProviderNameOffset); + + if (_wcsicmp(provider_name, name) == 0) { + *guid = info->ProviderGuid; + flb_free(providers); + return 0; + } + } + + flb_plg_error(ctx->ins, + "ETW provider_name '%S' was not found in registered providers", + name); + flb_free(providers); + + return -1; +} + +static EVENT_TRACE_PROPERTIES *create_trace_properties(struct flb_etw *ctx) +{ + size_t name_size; + size_t props_size; + EVENT_TRACE_PROPERTIES *props; + + name_size = (wcslen(ctx->session_name_wide) + 1) * sizeof(WCHAR); + props_size = sizeof(EVENT_TRACE_PROPERTIES) + name_size; + + props = flb_calloc(1, props_size); + if (props == NULL) { + flb_errno(); + return NULL; + } + + props->Wnode.BufferSize = (ULONG) props_size; + props->Wnode.Guid = ctx->session_guid; + props->Wnode.Flags = WNODE_FLAG_TRACED_GUID; + props->Wnode.ClientContext = 2; + props->LogFileMode = EVENT_TRACE_REAL_TIME_MODE; + props->LoggerNameOffset = sizeof(EVENT_TRACE_PROPERTIES); + if (ctx->buffer_size > 0) { + props->BufferSize = (ULONG) ctx->buffer_size; + } + if (ctx->minimum_buffers > 0) { + props->MinimumBuffers = (ULONG) ctx->minimum_buffers; + } + if (ctx->maximum_buffers > 0) { + props->MaximumBuffers = (ULONG) ctx->maximum_buffers; + } + if (ctx->flush_timer > 0) { + props->FlushTimer = (ULONG) ctx->flush_timer; + } + memcpy((char *) props + props->LoggerNameOffset, ctx->session_name_wide, name_size); + + return props; +} + +static void etw_timestamp_to_flb_time(const EVENT_HEADER *header, struct flb_time *tm) +{ + ULONGLONG ts; + ULONGLONG unix_100ns; + + ts = (ULONGLONG) header->TimeStamp.QuadPart; + if (ts <= FLB_IN_ETW_UNIX_EPOCH_IN_100NS) { + flb_time_get(tm); + return; + } + + unix_100ns = ts - FLB_IN_ETW_UNIX_EPOCH_IN_100NS; + flb_time_set(tm, + (time_t) (unix_100ns / 10000000ULL), + (long) ((unix_100ns % 10000000ULL) * 100ULL)); +} + +static void pack_guid_value(msgpack_packer *mp_pck, const GUID *guid) +{ + char buf[64]; + + if (guid_to_string(guid, buf, sizeof(buf)) == 0) { + pack_cstr(mp_pck, buf); + } + else { + msgpack_pack_nil(mp_pck); + } +} + +static void pack_filetime_value(msgpack_packer *mp_pck, const FILETIME *filetime) +{ + ULONGLONG ts; + ULONGLONG unix_100ns; + + ts = (((ULONGLONG) filetime->dwHighDateTime) << 32) | filetime->dwLowDateTime; + if (ts <= FLB_IN_ETW_UNIX_EPOCH_IN_100NS) { + msgpack_pack_nil(mp_pck); + return; + } + + unix_100ns = ts - FLB_IN_ETW_UNIX_EPOCH_IN_100NS; + msgpack_pack_uint64(mp_pck, unix_100ns / 10000000ULL); +} + +static int pack_nil_if_property_too_small(msgpack_packer *mp_pck, ULONG size, + size_t expected) +{ + if (size < expected) { + msgpack_pack_nil(mp_pck); + return FLB_TRUE; + } + + return FLB_FALSE; +} + +static void pack_property_value(msgpack_packer *mp_pck, USHORT in_type, + BYTE *buf, ULONG size) +{ + char *str; + size_t len; + + switch (in_type) { + case TDH_INTYPE_UNICODESTRING: + pack_wide_string(mp_pck, (WCHAR *) buf, size); + break; + case TDH_INTYPE_ANSISTRING: + len = strnlen((char *) buf, size); + msgpack_pack_str(mp_pck, len); + msgpack_pack_str_body(mp_pck, (char *) buf, len); + break; + case TDH_INTYPE_INT8: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT8))) { + break; + } + msgpack_pack_int8(mp_pck, *((INT8 *) buf)); + break; + case TDH_INTYPE_UINT8: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT8))) { + break; + } + msgpack_pack_uint8(mp_pck, *((UINT8 *) buf)); + break; + case TDH_INTYPE_INT16: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT16))) { + break; + } + msgpack_pack_int16(mp_pck, *((INT16 *) buf)); + break; + case TDH_INTYPE_UINT16: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT16))) { + break; + } + msgpack_pack_uint16(mp_pck, *((UINT16 *) buf)); + break; + case TDH_INTYPE_INT32: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT32))) { + break; + } + msgpack_pack_int32(mp_pck, *((INT32 *) buf)); + break; + case TDH_INTYPE_UINT32: + case TDH_INTYPE_HEXINT32: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT32))) { + break; + } + msgpack_pack_uint32(mp_pck, *((UINT32 *) buf)); + break; + case TDH_INTYPE_INT64: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT64))) { + break; + } + msgpack_pack_int64(mp_pck, *((INT64 *) buf)); + break; + case TDH_INTYPE_UINT64: + case TDH_INTYPE_HEXINT64: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT64))) { + break; + } + msgpack_pack_uint64(mp_pck, *((UINT64 *) buf)); + break; + case TDH_INTYPE_BOOLEAN: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(BOOL))) { + break; + } + if (*((BOOL *) buf)) { + msgpack_pack_true(mp_pck); + } + else { + msgpack_pack_false(mp_pck); + } + break; + case TDH_INTYPE_FLOAT: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(float))) { + break; + } + msgpack_pack_float(mp_pck, *((float *) buf)); + break; + case TDH_INTYPE_DOUBLE: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(double))) { + break; + } + msgpack_pack_double(mp_pck, *((double *) buf)); + break; + case TDH_INTYPE_GUID: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(GUID))) { + break; + } + pack_guid_value(mp_pck, (GUID *) buf); + break; + case TDH_INTYPE_FILETIME: + if (pack_nil_if_property_too_small(mp_pck, size, sizeof(FILETIME))) { + break; + } + pack_filetime_value(mp_pck, (FILETIME *) buf); + break; + case TDH_INTYPE_POINTER: + if (size == sizeof(UINT64)) { + msgpack_pack_uint64(mp_pck, *((UINT64 *) buf)); + } + else if (size == sizeof(UINT32)) { + msgpack_pack_uint32(mp_pck, *((UINT32 *) buf)); + } + else { + msgpack_pack_nil(mp_pck); + } + break; + default: + if (size == 0) { + msgpack_pack_nil(mp_pck); + } + else { + str = (char *) buf; + msgpack_pack_bin(mp_pck, size); + msgpack_pack_bin_body(mp_pck, str, size); + } + break; + } +} + +static void pack_payload_property(msgpack_packer *mp_pck, PEVENT_RECORD record, + TRACE_EVENT_INFO *info, ULONG index) +{ + ULONG size; + TDHSTATUS status; + BYTE *buf; + PROPERTY_DATA_DESCRIPTOR desc; + EVENT_PROPERTY_INFO *prop; + + prop = &info->EventPropertyInfoArray[index]; + if (prop->Flags & PropertyStruct) { + msgpack_pack_nil(mp_pck); + return; + } + + memset(&desc, 0, sizeof(desc)); + desc.PropertyName = (ULONGLONG) ((char *) info + prop->NameOffset); + desc.ArrayIndex = ULONG_MAX; + + size = 0; + status = TdhGetPropertySize(record, 0, NULL, 1, &desc, &size); + if (status != ERROR_SUCCESS || size == 0) { + msgpack_pack_nil(mp_pck); + return; + } + + buf = flb_malloc(size); + if (buf == NULL) { + flb_errno(); + msgpack_pack_nil(mp_pck); + return; + } + + status = TdhGetProperty(record, 0, NULL, 1, &desc, size, buf); + if (status == ERROR_SUCCESS) { + pack_property_value(mp_pck, prop->nonStructType.InType, buf, size); + } + else { + msgpack_pack_nil(mp_pck); + } + + flb_free(buf); +} + +static void pack_payload(msgpack_packer *mp_pck, PEVENT_RECORD record) +{ + ULONG size; + ULONG i; + TDHSTATUS status; + TRACE_EVENT_INFO *info; + EVENT_PROPERTY_INFO *prop; + WCHAR *name; + char fallback[32]; + + size = 0; + status = TdhGetEventInformation(record, 0, NULL, NULL, &size); + if (status != ERROR_INSUFFICIENT_BUFFER) { + msgpack_pack_map(mp_pck, 0); + return; + } + + info = flb_malloc(size); + if (info == NULL) { + flb_errno(); + msgpack_pack_map(mp_pck, 0); + return; + } + + status = TdhGetEventInformation(record, 0, NULL, info, &size); + if (status != ERROR_SUCCESS) { + flb_free(info); + msgpack_pack_map(mp_pck, 0); + return; + } + + msgpack_pack_map(mp_pck, info->TopLevelPropertyCount); + for (i = 0; i < info->TopLevelPropertyCount; i++) { + prop = &info->EventPropertyInfoArray[i]; + if (prop->NameOffset > 0) { + name = (WCHAR *) ((char *) info + prop->NameOffset); + pack_wide_string(mp_pck, name, 0); + } + else { + snprintf(fallback, sizeof(fallback), "property_%lu", i); + pack_cstr(mp_pck, fallback); + } + + pack_payload_property(mp_pck, record, info, i); + } + + flb_free(info); +} + +static void pack_related_activity_id(msgpack_packer *mp_pck, PEVENT_RECORD record) +{ + USHORT i; + PEVENT_HEADER_EXTENDED_DATA_ITEM item; + PEVENT_EXTENDED_ITEM_RELATED_ACTIVITYID related; + + for (i = 0; i < record->ExtendedDataCount; i++) { + item = &record->ExtendedData[i]; + if (item->ExtType != EVENT_HEADER_EXT_TYPE_RELATED_ACTIVITYID) { + continue; + } + + if (item->DataPtr == 0 || + item->DataSize < sizeof(EVENT_EXTENDED_ITEM_RELATED_ACTIVITYID)) { + msgpack_pack_nil(mp_pck); + return; + } + + related = (PEVENT_EXTENDED_ITEM_RELATED_ACTIVITYID) (uintptr_t) item->DataPtr; + pack_guid_value(mp_pck, &related->RelatedActivityId); + return; + } + + msgpack_pack_nil(mp_pck); +} + +static TRACEHANDLE etw_exchange_trace(struct flb_etw *ctx, TRACEHANDLE value) +{ + return (TRACEHANDLE) InterlockedExchange64((volatile LONG64 *) &ctx->trace, + (LONG64) value); +} + +static TRACEHANDLE etw_exchange_session(struct flb_etw *ctx, TRACEHANDLE value) +{ + return (TRACEHANDLE) InterlockedExchange64((volatile LONG64 *) &ctx->session, + (LONG64) value); +} + +static TRACEHANDLE etw_get_session(struct flb_etw *ctx) +{ + return (TRACEHANDLE) InterlockedCompareExchange64((volatile LONG64 *) &ctx->session, + 0, 0); +} + +static int etw_update_loss_metrics(struct flb_etw *ctx) +{ + ULONG status; + TRACEHANDLE session; + uint64_t timestamp; + char *name; + LONG query_errors; + + if (ctx->cmt_events_lost == NULL || + ctx->cmt_realtime_buffers_lost == NULL || + ctx->properties == NULL) { + return 0; + } + + session = etw_get_session(ctx); + if (session == 0) { + return 0; + } + + status = ControlTraceW(session, + ctx->session_name_wide, + ctx->properties, + EVENT_TRACE_CONTROL_QUERY); + if (status != ERROR_SUCCESS) { + query_errors = InterlockedIncrement(&ctx->query_errors); + if (query_errors == 1 || query_errors % 1000 == 0) { + flb_plg_warn(ctx->ins, + "ControlTrace query failed for session '%S' (status=%lu)", + ctx->session_name_wide, status); + } + return -1; + } + + name = (char *) flb_input_name(ctx->ins); + timestamp = cfl_time_now(); + + cmt_gauge_set(ctx->cmt_events_lost, + timestamp, + (double) ctx->properties->EventsLost, + 1, (char *[]) {name}); + cmt_gauge_set(ctx->cmt_realtime_buffers_lost, + timestamp, + (double) ctx->properties->RealTimeBuffersLost, + 1, (char *[]) {name}); + + return 0; +} + +static int etw_loss_metrics_collect(struct flb_input_instance *in, + struct flb_config *config, void *data) +{ + struct flb_etw *ctx; + + ctx = data; + etw_update_loss_metrics(ctx); + + return 0; +} + +static VOID WINAPI etw_event_callback(PEVENT_RECORD record) +{ + struct flb_etw *ctx; + struct flb_time timestamp; + msgpack_sbuffer mp_sbuf; + msgpack_packer mp_pck; + LONG append_errors; + int ret; + + ctx = (struct flb_etw *) record->UserContext; + if (ctx == NULL) { + return; + } + + if (InterlockedCompareExchange(&ctx->paused, 0, 0) || + InterlockedCompareExchange(&ctx->exiting, 0, 0)) { + return; + } + + etw_timestamp_to_flb_time(&record->EventHeader, ×tamp); + + msgpack_sbuffer_init(&mp_sbuf); + msgpack_packer_init(&mp_pck, &mp_sbuf, msgpack_sbuffer_write); + + msgpack_pack_array(&mp_pck, 2); + flb_time_append_to_msgpack(×tamp, &mp_pck, FLB_TIME_ETFMT_INT); + msgpack_pack_map(&mp_pck, FLB_IN_ETW_PROVIDER_MAP_SIZE); + + pack_cstr(&mp_pck, "provider_guid"); + pack_guid_value(&mp_pck, &record->EventHeader.ProviderId); + + pack_cstr(&mp_pck, "provider_name"); + if (ctx->provider_name != NULL) { + pack_cstr(&mp_pck, ctx->provider_name); + } + else { + msgpack_pack_nil(&mp_pck); + } + + pack_cstr(&mp_pck, "event_id"); + msgpack_pack_uint16(&mp_pck, record->EventHeader.EventDescriptor.Id); + + pack_cstr(&mp_pck, "version"); + msgpack_pack_uint8(&mp_pck, record->EventHeader.EventDescriptor.Version); + + pack_cstr(&mp_pck, "channel"); + msgpack_pack_uint8(&mp_pck, record->EventHeader.EventDescriptor.Channel); + + pack_cstr(&mp_pck, "level"); + msgpack_pack_uint8(&mp_pck, record->EventHeader.EventDescriptor.Level); + + pack_cstr(&mp_pck, "task"); + msgpack_pack_uint16(&mp_pck, record->EventHeader.EventDescriptor.Task); + + pack_cstr(&mp_pck, "opcode"); + msgpack_pack_uint8(&mp_pck, record->EventHeader.EventDescriptor.Opcode); + + pack_cstr(&mp_pck, "keywords"); + msgpack_pack_uint64(&mp_pck, record->EventHeader.EventDescriptor.Keyword); + + pack_cstr(&mp_pck, "process_id"); + msgpack_pack_uint32(&mp_pck, record->EventHeader.ProcessId); + + pack_cstr(&mp_pck, "thread_id"); + msgpack_pack_uint32(&mp_pck, record->EventHeader.ThreadId); + + pack_cstr(&mp_pck, "activity_id"); + pack_guid_value(&mp_pck, &record->EventHeader.ActivityId); + + pack_cstr(&mp_pck, "related_activity_id"); + pack_related_activity_id(&mp_pck, record); + + pack_cstr(&mp_pck, "payload"); + pack_payload(&mp_pck, record); + + ret = flb_input_log_append(ctx->ins, NULL, 0, mp_sbuf.data, mp_sbuf.size); + if (ret == -1) { + append_errors = InterlockedIncrement(&ctx->append_errors); + if (append_errors == 1 || append_errors % 1000 == 0) { + flb_plg_warn(ctx->ins, + "failed to append ETW event record (%ld failures)", + append_errors); + } + } + + msgpack_sbuffer_destroy(&mp_sbuf); +} + +static void etw_close_trace(struct flb_etw *ctx) +{ + TRACEHANDLE trace; + + trace = etw_exchange_trace(ctx, INVALID_PROCESSTRACE_HANDLE); + if (trace != INVALID_PROCESSTRACE_HANDLE) { + CloseTrace(trace); + } +} + +static ULONG etw_stop_session_by_name(struct flb_etw *ctx) +{ + return ControlTraceW(0, + ctx->session_name_wide, + ctx->properties, + EVENT_TRACE_CONTROL_STOP); +} + +static void etw_disable_provider(struct flb_etw *ctx, TRACEHANDLE session) +{ + ULONG status; + + if (session == 0) { + return; + } + + status = EnableTraceEx2(session, + &ctx->provider_guid, + EVENT_CONTROL_CODE_DISABLE_PROVIDER, + (UCHAR) ctx->level, + ctx->match_any_keyword, + ctx->match_all_keyword, + 0, + NULL); + if (status != ERROR_SUCCESS && status != ERROR_WMI_INSTANCE_NOT_FOUND) { + flb_plg_warn(ctx->ins, "EnableTraceEx2 disable failed (status=%lu)", status); + } +} + +static void etw_stop_session(struct flb_etw *ctx) +{ + TRACEHANDLE session; + ULONG status; + + session = etw_exchange_session(ctx, 0); + if (session != 0 && ctx->properties != NULL) { + etw_disable_provider(ctx, session); + status = ControlTraceW(session, + ctx->session_name_wide, + ctx->properties, + EVENT_TRACE_CONTROL_STOP); + if (status != ERROR_SUCCESS && status != ERROR_WMI_INSTANCE_NOT_FOUND) { + flb_plg_warn(ctx->ins, "ControlTrace stop failed (status=%lu)", status); + } + } +} + +static ULONG etw_start_session(struct flb_etw *ctx) +{ + ULONG status; + + status = StartTraceW(&ctx->session, ctx->session_name_wide, ctx->properties); + if (status != ERROR_ALREADY_EXISTS) { + return status; + } + + flb_plg_warn(ctx->ins, + "ETW session '%S' already exists; stopping stale session and retrying", + ctx->session_name_wide); + + status = etw_stop_session_by_name(ctx); + if (status != ERROR_SUCCESS && status != ERROR_WMI_INSTANCE_NOT_FOUND) { + flb_plg_error(ctx->ins, + "could not stop existing ETW session '%S' (status=%lu)", + ctx->session_name_wide, status); + return ERROR_ALREADY_EXISTS; + } + + return StartTraceW(&ctx->session, ctx->session_name_wide, ctx->properties); +} + +static void *etw_worker(void *data) +{ + ULONG status; + EVENT_TRACE_LOGFILEW logfile; + struct flb_etw *ctx; + + ctx = data; + + status = etw_start_session(ctx); + if (status != ERROR_SUCCESS) { + if (status == ERROR_ALREADY_EXISTS) { + flb_plg_error(ctx->ins, + "ETW session '%S' already exists and could not be reused", + ctx->session_name_wide); + } + else { + flb_plg_error(ctx->ins, "StartTrace failed for session '%S' (status=%lu)", + ctx->session_name_wide, status); + } + return NULL; + } + + status = EnableTraceEx2(ctx->session, + &ctx->provider_guid, + EVENT_CONTROL_CODE_ENABLE_PROVIDER, + (UCHAR) ctx->level, + ctx->match_any_keyword, + ctx->match_all_keyword, + 0, + NULL); + if (status != ERROR_SUCCESS) { + flb_plg_error(ctx->ins, "EnableTraceEx2 failed (status=%lu)", status); + etw_stop_session(ctx); + return NULL; + } + + memset(&logfile, 0, sizeof(logfile)); + logfile.LoggerName = ctx->session_name_wide; + logfile.ProcessTraceMode = PROCESS_TRACE_MODE_REAL_TIME | + PROCESS_TRACE_MODE_EVENT_RECORD; + logfile.EventRecordCallback = etw_event_callback; + logfile.Context = ctx; + + etw_exchange_trace(ctx, OpenTraceW(&logfile)); + if (ctx->trace == INVALID_PROCESSTRACE_HANDLE) { + flb_plg_error(ctx->ins, "OpenTrace failed (error=%lu)", GetLastError()); + etw_stop_session(ctx); + return NULL; + } + + status = ProcessTrace(&ctx->trace, 1, NULL, NULL); + if (status != ERROR_SUCCESS && !InterlockedCompareExchange(&ctx->exiting, 0, 0)) { + flb_plg_error(ctx->ins, "ProcessTrace failed (status=%lu)", status); + } + + etw_close_trace(ctx); + + if (!InterlockedCompareExchange(&ctx->exiting, 0, 0)) { + etw_stop_session(ctx); + } + + return NULL; +} + +static void etw_config_destroy(struct flb_etw *ctx) +{ + if (ctx == NULL) { + return; + } + + if (ctx->thread_created) { + InterlockedExchange(&ctx->exiting, 1); + etw_close_trace(ctx); + etw_stop_session(ctx); + pthread_join(ctx->thread, NULL); + } + + if (ctx->session_name_wide != NULL) { + flb_free(ctx->session_name_wide); + } + + if (ctx->properties != NULL) { + flb_free(ctx->properties); + } + + flb_free(ctx); +} + +static int configure_provider(struct flb_etw *ctx) +{ + WCHAR *provider_name_wide; + GUID name_guid; + char guid_buf[64]; + int have_guid; + int have_name; + + have_guid = ctx->provider_guid_str != NULL && ctx->provider_guid_str[0] != '\0'; + have_name = ctx->provider_name != NULL && ctx->provider_name[0] != '\0'; + + if (!have_guid && !have_name) { + flb_plg_error(ctx->ins, "either provider_guid or provider_name must be set"); + return -1; + } + + if (have_guid && parse_guid(ctx->provider_guid_str, &ctx->provider_guid) != 0) { + flb_plg_error(ctx->ins, "invalid provider_guid '%s'", ctx->provider_guid_str); + return -1; + } + + if (!have_name) { + return 0; + } + + provider_name_wide = utf8_to_wide(ctx->provider_name); + if (provider_name_wide == NULL) { + flb_plg_error(ctx->ins, "could not convert provider_name '%s' to UTF-16", + ctx->provider_name); + return -1; + } + + if (resolve_provider_name(ctx, provider_name_wide, &name_guid) != 0) { + flb_free(provider_name_wide); + return -1; + } + flb_free(provider_name_wide); + + if (have_guid && memcmp(&ctx->provider_guid, &name_guid, sizeof(GUID)) != 0) { + guid_to_string(&name_guid, guid_buf, sizeof(guid_buf)); + flb_plg_error(ctx->ins, + "provider_name '%s' resolves to %s, which does not match provider_guid '%s'", + ctx->provider_name, guid_buf, ctx->provider_guid_str); + return -1; + } + + ctx->provider_guid = name_guid; + + return 0; +} + +static int configure_session_properties(struct flb_etw *ctx) +{ + if (ctx->buffer_size < 0) { + flb_plg_error(ctx->ins, "buffer_size must be zero or greater"); + return -1; + } + + if (ctx->minimum_buffers < 0) { + flb_plg_error(ctx->ins, "minimum_buffers must be zero or greater"); + return -1; + } + + if (ctx->maximum_buffers < 0) { + flb_plg_error(ctx->ins, "maximum_buffers must be zero or greater"); + return -1; + } + + if (ctx->flush_timer < 0) { + flb_plg_error(ctx->ins, "flush_timer must be zero or greater"); + return -1; + } + + if (ctx->minimum_buffers > 0 && + ctx->maximum_buffers > 0 && + ctx->minimum_buffers > ctx->maximum_buffers) { + flb_plg_error(ctx->ins, + "minimum_buffers must be less than or equal to maximum_buffers"); + return -1; + } + + return 0; +} + +static int configure_loss_metrics(struct flb_etw *ctx, struct flb_config *config) +{ + int ret; + time_t interval; + + ctx->loss_metrics_collector_id = -1; + + if (ctx->ins->cmt == NULL) { + return 0; + } + + ctx->cmt_events_lost = cmt_gauge_create(ctx->ins->cmt, + "fluentbit", "input", + "etw_events_lost", + "Number of ETW events lost by the session.", + 1, (char *[]) {"name"}); + if (ctx->cmt_events_lost == NULL) { + flb_plg_error(ctx->ins, "could not create ETW events lost metric"); + return -1; + } + + ctx->cmt_realtime_buffers_lost = + cmt_gauge_create(ctx->ins->cmt, + "fluentbit", "input", + "etw_realtime_buffers_lost", + "Number of ETW real-time buffers lost by the session.", + 1, (char *[]) {"name"}); + if (ctx->cmt_realtime_buffers_lost == NULL) { + flb_plg_error(ctx->ins, "could not create ETW real-time buffers lost metric"); + return -1; + } + + interval = ctx->flush_timer > 0 ? ctx->flush_timer : 1; + ret = flb_input_set_collector_time(ctx->ins, + etw_loss_metrics_collect, + interval, 0, + config); + if (ret == -1) { + flb_plg_error(ctx->ins, "could not set ETW loss metrics collector"); + return -1; + } + + ctx->loss_metrics_collector_id = ret; + + return 0; +} + +static int in_etw_init(struct flb_input_instance *in, + struct flb_config *config, void *data) +{ + HRESULT hres; + int ret; + struct flb_etw *ctx; + + ctx = flb_calloc(1, sizeof(struct flb_etw)); + if (ctx == NULL) { + flb_errno(); + return -1; + } + + ctx->ins = in; + ctx->trace = INVALID_PROCESSTRACE_HANDLE; + ctx->loss_metrics_collector_id = -1; + + ret = flb_input_config_map_set(in, ctx); + if (ret == -1) { + etw_config_destroy(ctx); + return -1; + } + + if (ctx->level < 0 || ctx->level > 255) { + flb_plg_error(ctx->ins, "level must be between 0 and 255"); + etw_config_destroy(ctx); + return -1; + } + + if (parse_uint64(ctx->match_any_keyword_str, &ctx->match_any_keyword) != 0) { + flb_plg_error(ctx->ins, "invalid match_any_keyword '%s'", + ctx->match_any_keyword_str); + etw_config_destroy(ctx); + return -1; + } + + if (parse_uint64(ctx->match_all_keyword_str, &ctx->match_all_keyword) != 0) { + flb_plg_error(ctx->ins, "invalid match_all_keyword '%s'", + ctx->match_all_keyword_str); + etw_config_destroy(ctx); + return -1; + } + + if (configure_session_properties(ctx) != 0) { + etw_config_destroy(ctx); + return -1; + } + + if (configure_provider(ctx) != 0) { + etw_config_destroy(ctx); + return -1; + } + + ctx->session_name_wide = utf8_to_wide(ctx->session_name); + if (ctx->session_name_wide == NULL) { + flb_plg_error(ctx->ins, "could not convert session_name '%s' to UTF-16", + ctx->session_name); + etw_config_destroy(ctx); + return -1; + } + + hres = CoCreateGuid(&ctx->session_guid); + if (FAILED(hres)) { + flb_plg_error(ctx->ins, "could not generate ETW session GUID (hr=0x%08lx)", + (unsigned long) hres); + etw_config_destroy(ctx); + return -1; + } + + ctx->properties = create_trace_properties(ctx); + if (ctx->properties == NULL) { + etw_config_destroy(ctx); + return -1; + } + + flb_input_set_context(in, ctx); + + if (configure_loss_metrics(ctx, config) != 0) { + etw_config_destroy(ctx); + return -1; + } + + ret = pthread_create(&ctx->thread, NULL, etw_worker, ctx); + if (ret != 0) { + flb_plg_error(ctx->ins, "could not create ETW worker thread"); + etw_config_destroy(ctx); + return -1; + } + ctx->thread_created = FLB_TRUE; + + return 0; +} + +static void in_etw_pause(void *data, struct flb_config *config) +{ + struct flb_etw *ctx; + + ctx = data; + InterlockedExchange(&ctx->paused, 1); + if (ctx->loss_metrics_collector_id >= 0) { + flb_input_collector_pause(ctx->loss_metrics_collector_id, ctx->ins); + } +} + +static void in_etw_resume(void *data, struct flb_config *config) +{ + struct flb_etw *ctx; + + ctx = data; + InterlockedExchange(&ctx->paused, 0); + if (ctx->loss_metrics_collector_id >= 0) { + flb_input_collector_resume(ctx->loss_metrics_collector_id, ctx->ins); + } +} + +static int in_etw_exit(void *data, struct flb_config *config) +{ + struct flb_etw *ctx; + + ctx = data; + etw_config_destroy(ctx); + + return 0; +} + +static struct flb_config_map config_map[] = { + { + FLB_CONFIG_MAP_STR, "provider_guid", NULL, + 0, FLB_TRUE, offsetof(struct flb_etw, provider_guid_str), + "ETW provider GUID to enable" + }, + { + FLB_CONFIG_MAP_STR, "provider_name", NULL, + 0, FLB_TRUE, offsetof(struct flb_etw, provider_name), + "ETW provider name to resolve and enable" + }, + { + FLB_CONFIG_MAP_STR, "session_name", FLB_IN_ETW_DEFAULT_SESSION_NAME, + 0, FLB_TRUE, offsetof(struct flb_etw, session_name), + "ETW real-time session name" + }, + { + FLB_CONFIG_MAP_INT, "level", FLB_IN_ETW_DEFAULT_LEVEL, + 0, FLB_TRUE, offsetof(struct flb_etw, level), + "ETW provider level" + }, + { + FLB_CONFIG_MAP_STR, "match_any_keyword", FLB_IN_ETW_DEFAULT_MATCH_ANY, + 0, FLB_TRUE, offsetof(struct flb_etw, match_any_keyword_str), + "ETW MatchAnyKeyword mask" + }, + { + FLB_CONFIG_MAP_STR, "match_all_keyword", FLB_IN_ETW_DEFAULT_MATCH_ALL, + 0, FLB_TRUE, offsetof(struct flb_etw, match_all_keyword_str), + "ETW MatchAllKeyword mask" + }, + { + FLB_CONFIG_MAP_INT, "buffer_size", FLB_IN_ETW_DEFAULT_BUFFER_SIZE, + 0, FLB_TRUE, offsetof(struct flb_etw, buffer_size), + "ETW session buffer size in kilobytes. The session buffer memory upper " + "bound is buffer_size KB * maximum_buffers; the default is 64 KB * 32 " + "= 2048 KB. Zero uses the Windows default." + }, + { + FLB_CONFIG_MAP_INT, "minimum_buffers", FLB_IN_ETW_DEFAULT_MIN_BUFFERS, + 0, FLB_TRUE, offsetof(struct flb_etw, minimum_buffers), + "Minimum number of ETW session buffers. Zero uses the Windows default." + }, + { + FLB_CONFIG_MAP_INT, "maximum_buffers", FLB_IN_ETW_DEFAULT_MAX_BUFFERS, + 0, FLB_TRUE, offsetof(struct flb_etw, maximum_buffers), + "Maximum number of ETW session buffers. The session buffer memory upper " + "bound is buffer_size KB * maximum_buffers; the default is 64 KB * 32 " + "= 2048 KB. Zero uses the Windows default." + }, + { + FLB_CONFIG_MAP_INT, "flush_timer", FLB_IN_ETW_DEFAULT_FLUSH_TIMER, + 0, FLB_TRUE, offsetof(struct flb_etw, flush_timer), + "ETW session flush timer in seconds. Zero uses the Windows default." + }, + {0} +}; + +struct flb_input_plugin in_etw_plugin = { + .name = "event_tracing_windows", + .description = "Event Tracing for Windows", + .cb_init = in_etw_init, + .cb_pre_run = NULL, + .cb_collect = NULL, + .cb_flush_buf = NULL, + .cb_pause = in_etw_pause, + .cb_resume = in_etw_resume, + .cb_exit = in_etw_exit, + .config_map = config_map +}; diff --git a/plugins/in_etw/etw.h b/plugins/in_etw/etw.h new file mode 100644 index 00000000000..f735a6dec7c --- /dev/null +++ b/plugins/in_etw/etw.h @@ -0,0 +1,73 @@ +/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ + +/* Fluent Bit + * ========== + * Copyright (C) 2015-2026 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_IN_ETW_H +#define FLB_IN_ETW_H + +#include +#include + +#include +#include +#include + +#define FLB_IN_ETW_DEFAULT_SESSION_NAME "fluent-bit-event-tracing-windows" +#define FLB_IN_ETW_DEFAULT_LEVEL "5" +#define FLB_IN_ETW_DEFAULT_MATCH_ANY "0xffffffffffffffff" +#define FLB_IN_ETW_DEFAULT_MATCH_ALL "0" +#define FLB_IN_ETW_DEFAULT_BUFFER_SIZE "64" +#define FLB_IN_ETW_DEFAULT_MIN_BUFFERS "4" +#define FLB_IN_ETW_DEFAULT_MAX_BUFFERS "32" +#define FLB_IN_ETW_DEFAULT_FLUSH_TIMER "1" + +struct flb_etw { + flb_sds_t provider_guid_str; + flb_sds_t provider_name; + flb_sds_t session_name; + flb_sds_t match_any_keyword_str; + flb_sds_t match_all_keyword_str; + int level; + int buffer_size; + int minimum_buffers; + int maximum_buffers; + int flush_timer; + + GUID provider_guid; + ULONGLONG match_any_keyword; + ULONGLONG match_all_keyword; + GUID session_guid; + WCHAR *session_name_wide; + EVENT_TRACE_PROPERTIES *properties; + TRACEHANDLE session; + TRACEHANDLE trace; + + LONG exiting; + LONG paused; + LONG append_errors; + LONG query_errors; + int thread_created; + int loss_metrics_collector_id; + pthread_t thread; + + struct cmt_gauge *cmt_events_lost; + struct cmt_gauge *cmt_realtime_buffers_lost; + struct flb_input_instance *ins; +}; + +#endif From efc4a0d85a73dc9129c71d071478f8f366d8028f Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Wed, 8 Jul 2026 16:01:00 +0900 Subject: [PATCH 2/6] in_etw: Handle mutex operations instead of atomic Signed-off-by: Hiroshi Hatake --- plugins/in_etw/etw.c | 60 +++++++++++++++++++++++++++++++++++--------- plugins/in_etw/etw.h | 1 + 2 files changed, 49 insertions(+), 12 deletions(-) diff --git a/plugins/in_etw/etw.c b/plugins/in_etw/etw.c index d4b1aff5fa1..0c05ba1c9b3 100644 --- a/plugins/in_etw/etw.c +++ b/plugins/in_etw/etw.c @@ -609,20 +609,37 @@ static void pack_related_activity_id(msgpack_packer *mp_pck, PEVENT_RECORD recor static TRACEHANDLE etw_exchange_trace(struct flb_etw *ctx, TRACEHANDLE value) { - return (TRACEHANDLE) InterlockedExchange64((volatile LONG64 *) &ctx->trace, - (LONG64) value); + TRACEHANDLE previous; + + EnterCriticalSection(&ctx->handle_lock); + previous = ctx->trace; + ctx->trace = value; + LeaveCriticalSection(&ctx->handle_lock); + + return previous; } static TRACEHANDLE etw_exchange_session(struct flb_etw *ctx, TRACEHANDLE value) { - return (TRACEHANDLE) InterlockedExchange64((volatile LONG64 *) &ctx->session, - (LONG64) value); + TRACEHANDLE previous; + + EnterCriticalSection(&ctx->handle_lock); + previous = ctx->session; + ctx->session = value; + LeaveCriticalSection(&ctx->handle_lock); + + return previous; } static TRACEHANDLE etw_get_session(struct flb_etw *ctx) { - return (TRACEHANDLE) InterlockedCompareExchange64((volatile LONG64 *) &ctx->session, - 0, 0); + TRACEHANDLE session; + + EnterCriticalSection(&ctx->handle_lock); + session = ctx->session; + LeaveCriticalSection(&ctx->handle_lock); + + return session; } static int etw_update_loss_metrics(struct flb_etw *ctx) @@ -832,9 +849,14 @@ static void etw_stop_session(struct flb_etw *ctx) static ULONG etw_start_session(struct flb_etw *ctx) { ULONG status; + TRACEHANDLE session; - status = StartTraceW(&ctx->session, ctx->session_name_wide, ctx->properties); + session = 0; + status = StartTraceW(&session, ctx->session_name_wide, ctx->properties); if (status != ERROR_ALREADY_EXISTS) { + if (status == ERROR_SUCCESS) { + etw_exchange_session(ctx, session); + } return status; } @@ -850,12 +872,20 @@ static ULONG etw_start_session(struct flb_etw *ctx) return ERROR_ALREADY_EXISTS; } - return StartTraceW(&ctx->session, ctx->session_name_wide, ctx->properties); + session = 0; + status = StartTraceW(&session, ctx->session_name_wide, ctx->properties); + if (status == ERROR_SUCCESS) { + etw_exchange_session(ctx, session); + } + + return status; } static void *etw_worker(void *data) { ULONG status; + TRACEHANDLE session; + TRACEHANDLE trace; EVENT_TRACE_LOGFILEW logfile; struct flb_etw *ctx; @@ -875,7 +905,9 @@ static void *etw_worker(void *data) return NULL; } - status = EnableTraceEx2(ctx->session, + session = etw_get_session(ctx); + + status = EnableTraceEx2(session, &ctx->provider_guid, EVENT_CONTROL_CODE_ENABLE_PROVIDER, (UCHAR) ctx->level, @@ -896,14 +928,15 @@ static void *etw_worker(void *data) logfile.EventRecordCallback = etw_event_callback; logfile.Context = ctx; - etw_exchange_trace(ctx, OpenTraceW(&logfile)); - if (ctx->trace == INVALID_PROCESSTRACE_HANDLE) { + trace = OpenTraceW(&logfile); + etw_exchange_trace(ctx, trace); + if (trace == INVALID_PROCESSTRACE_HANDLE) { flb_plg_error(ctx->ins, "OpenTrace failed (error=%lu)", GetLastError()); etw_stop_session(ctx); return NULL; } - status = ProcessTrace(&ctx->trace, 1, NULL, NULL); + status = ProcessTrace(&trace, 1, NULL, NULL); if (status != ERROR_SUCCESS && !InterlockedCompareExchange(&ctx->exiting, 0, 0)) { flb_plg_error(ctx->ins, "ProcessTrace failed (status=%lu)", status); } @@ -938,6 +971,8 @@ static void etw_config_destroy(struct flb_etw *ctx) flb_free(ctx->properties); } + DeleteCriticalSection(&ctx->handle_lock); + flb_free(ctx); } @@ -1088,6 +1123,7 @@ static int in_etw_init(struct flb_input_instance *in, ctx->ins = in; ctx->trace = INVALID_PROCESSTRACE_HANDLE; ctx->loss_metrics_collector_id = -1; + InitializeCriticalSection(&ctx->handle_lock); ret = flb_input_config_map_set(in, ctx); if (ret == -1) { diff --git a/plugins/in_etw/etw.h b/plugins/in_etw/etw.h index f735a6dec7c..d71a5f66f7b 100644 --- a/plugins/in_etw/etw.h +++ b/plugins/in_etw/etw.h @@ -56,6 +56,7 @@ struct flb_etw { EVENT_TRACE_PROPERTIES *properties; TRACEHANDLE session; TRACEHANDLE trace; + CRITICAL_SECTION handle_lock; LONG exiting; LONG paused; From 846e9dd5109ce92d33ec4fc32e0c07b54576f009 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Wed, 8 Jul 2026 18:14:26 +0900 Subject: [PATCH 3/6] in_etw: Add kernel log mode and session validations Signed-off-by: Hiroshi Hatake --- plugins/in_etw/etw.c | 332 ++++++++++++++++++++++++++++++++++++++++--- plugins/in_etw/etw.h | 14 ++ 2 files changed, 327 insertions(+), 19 deletions(-) diff --git a/plugins/in_etw/etw.c b/plugins/in_etw/etw.c index 0c05ba1c9b3..a0f014025b8 100644 --- a/plugins/in_etw/etw.c +++ b/plugins/in_etw/etw.c @@ -28,6 +28,7 @@ #include #include +#include #include #include #include @@ -38,6 +39,14 @@ #define FLB_IN_ETW_UNIX_EPOCH_IN_100NS 116444736000000000ULL #define FLB_IN_ETW_PROVIDER_MAP_SIZE 14 +#define FLB_IN_ETW_DEFAULT_KERNEL_FLAGS "process,thread,image_load" + +static const GUID flb_etw_system_trace_control_guid = { + 0x9e814aad, + 0x3204, + 0x11d2, + {0x9a, 0x82, 0x00, 0x60, 0x08, 0xa8, 0x69, 0x39} +}; static void pack_cstr(msgpack_packer *mp_pck, const char *str) { @@ -134,6 +143,194 @@ static int parse_uint64(const char *str, ULONGLONG *out) return 0; } +static char *trim_token(char *str) +{ + char *end; + + while (isspace((unsigned char) *str)) { + str++; + } + + if (*str == '\0') { + return str; + } + + end = str + strlen(str) - 1; + while (end > str && isspace((unsigned char) *end)) { + *end = '\0'; + end--; + } + + return str; +} + +static int has_empty_csv_token(const char *str) +{ + const char *p; + int has_token; + + has_token = FLB_FALSE; + for (p = str; *p != '\0'; p++) { + if (*p == ',') { + if (has_token == FLB_FALSE) { + return FLB_TRUE; + } + has_token = FLB_FALSE; + continue; + } + + if (!isspace((unsigned char) *p)) { + has_token = FLB_TRUE; + } + } + + return has_token == FLB_FALSE; +} + +static int parse_session_type(struct flb_etw *ctx) +{ + if (ctx->session_type_str == NULL || + strcasecmp(ctx->session_type_str, "provider") == 0) { + ctx->session_type = FLB_IN_ETW_SESSION_PROVIDER; + return 0; + } + + if (strcasecmp(ctx->session_type_str, "system") == 0) { + ctx->session_type = FLB_IN_ETW_SESSION_SYSTEM; + return 0; + } + + flb_plg_error(ctx->ins, + "invalid session_type '%s' (expected 'provider' or 'system')", + ctx->session_type_str); + + return -1; +} + +static int parse_stale_session_action(struct flb_etw *ctx) +{ + if (ctx->stale_session_action_str == NULL || + strcasecmp(ctx->stale_session_action_str, "stop") == 0) { + ctx->stale_session_action = FLB_IN_ETW_STALE_ACTION_STOP; + return 0; + } + + if (strcasecmp(ctx->stale_session_action_str, "fail") == 0) { + ctx->stale_session_action = FLB_IN_ETW_STALE_ACTION_FAIL; + return 0; + } + + flb_plg_error(ctx->ins, + "invalid stale_session_action '%s' (expected 'stop' or 'fail')", + ctx->stale_session_action_str); + + return -1; +} + +static int kernel_flag_from_name(const char *name, ULONG *flag) +{ + if (strcasecmp(name, "process") == 0) { + *flag = EVENT_TRACE_FLAG_PROCESS; + return 0; + } + + if (strcasecmp(name, "thread") == 0) { + *flag = EVENT_TRACE_FLAG_THREAD; + return 0; + } + + if (strcasecmp(name, "image_load") == 0 || + strcasecmp(name, "image") == 0) { + *flag = EVENT_TRACE_FLAG_IMAGE_LOAD; + return 0; + } + + if (strcasecmp(name, "cswitch") == 0 || + strcasecmp(name, "context_switch") == 0) { + *flag = EVENT_TRACE_FLAG_CSWITCH; + return 0; + } + + if (strcasecmp(name, "disk_io") == 0 || + strcasecmp(name, "disk") == 0) { + *flag = EVENT_TRACE_FLAG_DISK_IO; + return 0; + } + + if (strcasecmp(name, "tcpip") == 0 || + strcasecmp(name, "network_tcpip") == 0) { + *flag = EVENT_TRACE_FLAG_NETWORK_TCPIP; + return 0; + } + + return -1; +} + +static int parse_kernel_flags(struct flb_etw *ctx) +{ + ULONG flag; + ULONGLONG numeric_flags; + char *tmp; + char *token; + char *context; + + if (ctx->session_type != FLB_IN_ETW_SESSION_SYSTEM) { + return 0; + } + + if (parse_uint64(ctx->kernel_flags_str, &numeric_flags) == 0) { + if (numeric_flags > ULONG_MAX) { + flb_plg_error(ctx->ins, "kernel_flags numeric value is too large"); + return -1; + } + if (numeric_flags == 0) { + flb_plg_error(ctx->ins, "kernel_flags must enable at least one flag"); + return -1; + } + ctx->kernel_flags = (ULONG) numeric_flags; + return 0; + } + + if (has_empty_csv_token(ctx->kernel_flags_str)) { + flb_plg_error(ctx->ins, "kernel_flags contains an empty token"); + return -1; + } + + tmp = flb_strdup(ctx->kernel_flags_str); + if (tmp == NULL) { + flb_errno(); + return -1; + } + + token = strtok_s(tmp, ",", &context); + while (token != NULL) { + token = trim_token(token); + if (token[0] == '\0') { + flb_plg_error(ctx->ins, "kernel_flags contains an empty token"); + flb_free(tmp); + return -1; + } + + if (kernel_flag_from_name(token, &flag) != 0) { + flb_plg_error(ctx->ins, "unknown kernel_flags token '%s'", token); + flb_free(tmp); + return -1; + } + + ctx->kernel_flags |= flag; + token = strtok_s(NULL, ",", &context); + } + + flb_free(tmp); + + if (ctx->kernel_flags == 0) { + flb_plg_error(ctx->ins, "kernel_flags must enable at least one flag"); + return -1; + } + + return 0; +} + static WCHAR *utf8_to_wide(const char *str) { int size; @@ -287,10 +484,19 @@ static EVENT_TRACE_PROPERTIES *create_trace_properties(struct flb_etw *ctx) } props->Wnode.BufferSize = (ULONG) props_size; - props->Wnode.Guid = ctx->session_guid; + if (ctx->session_type == FLB_IN_ETW_SESSION_SYSTEM) { + props->Wnode.Guid = flb_etw_system_trace_control_guid; + } + else { + props->Wnode.Guid = ctx->session_guid; + } props->Wnode.Flags = WNODE_FLAG_TRACED_GUID; props->Wnode.ClientContext = 2; props->LogFileMode = EVENT_TRACE_REAL_TIME_MODE; + if (ctx->session_type == FLB_IN_ETW_SESSION_SYSTEM) { + props->LogFileMode |= EVENT_TRACE_SYSTEM_LOGGER_MODE; + props->EnableFlags = ctx->kernel_flags; + } props->LoggerNameOffset = sizeof(EVENT_TRACE_PROPERTIES); if (ctx->buffer_size > 0) { props->BufferSize = (ULONG) ctx->buffer_size; @@ -811,7 +1017,7 @@ static void etw_disable_provider(struct flb_etw *ctx, TRACEHANDLE session) { ULONG status; - if (session == 0) { + if (session == 0 || ctx->session_type == FLB_IN_ETW_SESSION_SYSTEM) { return; } @@ -860,6 +1066,10 @@ static ULONG etw_start_session(struct flb_etw *ctx) return status; } + if (ctx->stale_session_action == FLB_IN_ETW_STALE_ACTION_FAIL) { + return ERROR_ALREADY_EXISTS; + } + flb_plg_warn(ctx->ins, "ETW session '%S' already exists; stopping stale session and retrying", ctx->session_name_wide); @@ -905,20 +1115,22 @@ static void *etw_worker(void *data) return NULL; } - session = etw_get_session(ctx); - - status = EnableTraceEx2(session, - &ctx->provider_guid, - EVENT_CONTROL_CODE_ENABLE_PROVIDER, - (UCHAR) ctx->level, - ctx->match_any_keyword, - ctx->match_all_keyword, - 0, - NULL); - if (status != ERROR_SUCCESS) { - flb_plg_error(ctx->ins, "EnableTraceEx2 failed (status=%lu)", status); - etw_stop_session(ctx); - return NULL; + if (ctx->session_type == FLB_IN_ETW_SESSION_PROVIDER) { + session = etw_get_session(ctx); + + status = EnableTraceEx2(session, + &ctx->provider_guid, + EVENT_CONTROL_CODE_ENABLE_PROVIDER, + (UCHAR) ctx->level, + ctx->match_any_keyword, + ctx->match_all_keyword, + 0, + NULL); + if (status != ERROR_SUCCESS) { + flb_plg_error(ctx->ins, "EnableTraceEx2 failed (status=%lu)", status); + etw_stop_session(ctx); + return NULL; + } } memset(&logfile, 0, sizeof(logfile)); @@ -987,6 +1199,15 @@ static int configure_provider(struct flb_etw *ctx) have_guid = ctx->provider_guid_str != NULL && ctx->provider_guid_str[0] != '\0'; have_name = ctx->provider_name != NULL && ctx->provider_name[0] != '\0'; + if (ctx->session_type == FLB_IN_ETW_SESSION_SYSTEM) { + if (have_guid || have_name) { + flb_plg_error(ctx->ins, + "provider_guid/provider_name cannot be used with session_type system"); + return -1; + } + return 0; + } + if (!have_guid && !have_name) { flb_plg_error(ctx->ins, "either provider_guid or provider_name must be set"); return -1; @@ -1060,6 +1281,36 @@ static int configure_session_properties(struct flb_etw *ctx) return 0; } +static int configure_session_name(struct flb_etw *ctx) +{ + if (ctx->session_type != FLB_IN_ETW_SESSION_SYSTEM) { + return 0; + } + + if (ctx->session_name == NULL || + strcmp(ctx->session_name, FLB_IN_ETW_DEFAULT_SESSION_NAME) == 0 || + strcmp(ctx->session_name, KERNEL_LOGGER_NAMEA) == 0) { + return 0; + } + + flb_plg_error(ctx->ins, + "session_type system requires session_name '%s'", + KERNEL_LOGGER_NAMEA); + + return -1; +} + +static const char *effective_session_name(struct flb_etw *ctx) +{ + if (ctx->session_type == FLB_IN_ETW_SESSION_SYSTEM && + ctx->session_name != NULL && + strcmp(ctx->session_name, FLB_IN_ETW_DEFAULT_SESSION_NAME) == 0) { + return KERNEL_LOGGER_NAMEA; + } + + return ctx->session_name; +} + static int configure_loss_metrics(struct flb_etw *ctx, struct flb_config *config) { int ret; @@ -1112,6 +1363,7 @@ static int in_etw_init(struct flb_input_instance *in, { HRESULT hres; int ret; + const char *session_name; struct flb_etw *ctx; ctx = flb_calloc(1, sizeof(struct flb_etw)); @@ -1131,6 +1383,16 @@ static int in_etw_init(struct flb_input_instance *in, return -1; } + if (parse_session_type(ctx) != 0) { + etw_config_destroy(ctx); + return -1; + } + + if (parse_stale_session_action(ctx) != 0) { + etw_config_destroy(ctx); + return -1; + } + if (ctx->level < 0 || ctx->level > 255) { flb_plg_error(ctx->ins, "level must be between 0 and 255"); etw_config_destroy(ctx); @@ -1151,20 +1413,31 @@ static int in_etw_init(struct flb_input_instance *in, return -1; } + if (parse_kernel_flags(ctx) != 0) { + etw_config_destroy(ctx); + return -1; + } + if (configure_session_properties(ctx) != 0) { etw_config_destroy(ctx); return -1; } + if (configure_session_name(ctx) != 0) { + etw_config_destroy(ctx); + return -1; + } + if (configure_provider(ctx) != 0) { etw_config_destroy(ctx); return -1; } - ctx->session_name_wide = utf8_to_wide(ctx->session_name); + session_name = effective_session_name(ctx); + ctx->session_name_wide = utf8_to_wide(session_name); if (ctx->session_name_wide == NULL) { flb_plg_error(ctx->ins, "could not convert session_name '%s' to UTF-16", - ctx->session_name); + session_name); etw_config_destroy(ctx); return -1; } @@ -1247,7 +1520,21 @@ static struct flb_config_map config_map[] = { { FLB_CONFIG_MAP_STR, "session_name", FLB_IN_ETW_DEFAULT_SESSION_NAME, 0, FLB_TRUE, offsetof(struct flb_etw, session_name), - "ETW real-time session name" + "ETW real-time session name. When session_type is system and this value " + "is left as the default, the plugin uses the Windows kernel logger " + "session name 'NT Kernel Logger'." + }, + { + FLB_CONFIG_MAP_STR, "session_type", FLB_IN_ETW_DEFAULT_SESSION_TYPE, + 0, FLB_TRUE, offsetof(struct flb_etw, session_type_str), + "ETW session type: provider for a real-time provider consumer, or system " + "for a system logger session using kernel_flags." + }, + { + FLB_CONFIG_MAP_STR, "stale_session_action", FLB_IN_ETW_DEFAULT_STALE_ACTION, + 0, FLB_TRUE, offsetof(struct flb_etw, stale_session_action_str), + "Action when the ETW session already exists: stop stops the existing " + "session and retries; fail returns an error without stopping it." }, { FLB_CONFIG_MAP_INT, "level", FLB_IN_ETW_DEFAULT_LEVEL, @@ -1264,6 +1551,13 @@ static struct flb_config_map config_map[] = { 0, FLB_TRUE, offsetof(struct flb_etw, match_all_keyword_str), "ETW MatchAllKeyword mask" }, + { + FLB_CONFIG_MAP_STR, "kernel_flags", FLB_IN_ETW_DEFAULT_KERNEL_FLAGS, + 0, FLB_TRUE, offsetof(struct flb_etw, kernel_flags_str), + "Comma-separated kernel flags used with session_type system. Supported " + "names: process, thread, image_load, cswitch, tcpip, disk_io. A numeric " + "EVENT_TRACE_FLAG_* mask is also accepted." + }, { FLB_CONFIG_MAP_INT, "buffer_size", FLB_IN_ETW_DEFAULT_BUFFER_SIZE, 0, FLB_TRUE, offsetof(struct flb_etw, buffer_size), diff --git a/plugins/in_etw/etw.h b/plugins/in_etw/etw.h index d71a5f66f7b..64707e9a6d5 100644 --- a/plugins/in_etw/etw.h +++ b/plugins/in_etw/etw.h @@ -35,6 +35,14 @@ #define FLB_IN_ETW_DEFAULT_MIN_BUFFERS "4" #define FLB_IN_ETW_DEFAULT_MAX_BUFFERS "32" #define FLB_IN_ETW_DEFAULT_FLUSH_TIMER "1" +#define FLB_IN_ETW_DEFAULT_STALE_ACTION "stop" +#define FLB_IN_ETW_DEFAULT_SESSION_TYPE "provider" + +#define FLB_IN_ETW_SESSION_PROVIDER 0 +#define FLB_IN_ETW_SESSION_SYSTEM 1 + +#define FLB_IN_ETW_STALE_ACTION_STOP 0 +#define FLB_IN_ETW_STALE_ACTION_FAIL 1 struct flb_etw { flb_sds_t provider_guid_str; @@ -42,15 +50,21 @@ struct flb_etw { flb_sds_t session_name; flb_sds_t match_any_keyword_str; flb_sds_t match_all_keyword_str; + flb_sds_t stale_session_action_str; + flb_sds_t session_type_str; + flb_sds_t kernel_flags_str; int level; int buffer_size; int minimum_buffers; int maximum_buffers; int flush_timer; + int stale_session_action; + int session_type; GUID provider_guid; ULONGLONG match_any_keyword; ULONGLONG match_all_keyword; + ULONG kernel_flags; GUID session_guid; WCHAR *session_name_wide; EVENT_TRACE_PROPERTIES *properties; From 18a871cabce5181a84e61e81e1d5106e7772be0d Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Wed, 8 Jul 2026 19:08:28 +0900 Subject: [PATCH 4/6] in_etw: Handle lifecycles and length more rigidly Signed-off-by: Hiroshi Hatake --- plugins/in_etw/etw.c | 92 +++++++++++++++++++++++++++++++++++++++----- plugins/in_etw/etw.h | 5 +++ 2 files changed, 87 insertions(+), 10 deletions(-) diff --git a/plugins/in_etw/etw.c b/plugins/in_etw/etw.c index a0f014025b8..98b9535bca0 100644 --- a/plugins/in_etw/etw.c +++ b/plugins/in_etw/etw.c @@ -392,7 +392,12 @@ static char *wide_to_utf8(const WCHAR *str, int bytes, int *out_len) } buf[size] = '\0'; - *out_len = size; + if (wide_len == -1) { + *out_len = size - 1; + } + else { + *out_len = size; + } return buf; } @@ -1007,10 +1012,46 @@ static void etw_close_trace(struct flb_etw *ctx) static ULONG etw_stop_session_by_name(struct flb_etw *ctx) { - return ControlTraceW(0, - ctx->session_name_wide, - ctx->properties, - EVENT_TRACE_CONTROL_STOP); + ULONG status; + EVENT_TRACE_PROPERTIES *properties; + + properties = create_trace_properties(ctx); + if (properties == NULL) { + return ERROR_NOT_ENOUGH_MEMORY; + } + + status = ControlTraceW(0, + ctx->session_name_wide, + properties, + EVENT_TRACE_CONTROL_STOP); + flb_free(properties); + + return status; +} + +static void etw_signal_startup(struct flb_etw *ctx, int status) +{ + pthread_mutex_lock(&ctx->startup_lock); + if (!ctx->startup_done) { + ctx->startup_status = status; + ctx->startup_done = FLB_TRUE; + pthread_cond_signal(&ctx->startup_cond); + } + pthread_mutex_unlock(&ctx->startup_lock); +} + +static int etw_wait_startup(struct flb_etw *ctx) +{ + int status; + + pthread_mutex_lock(&ctx->startup_lock); + while (!ctx->startup_done) { + pthread_cond_wait(&ctx->startup_cond, &ctx->startup_lock); + } + status = ctx->startup_status; + pthread_mutex_unlock(&ctx->startup_lock); + + return status; } static void etw_disable_provider(struct flb_etw *ctx, TRACEHANDLE session) @@ -1112,6 +1153,7 @@ static void *etw_worker(void *data) flb_plg_error(ctx->ins, "StartTrace failed for session '%S' (status=%lu)", ctx->session_name_wide, status); } + etw_signal_startup(ctx, -1); return NULL; } @@ -1129,6 +1171,7 @@ static void *etw_worker(void *data) if (status != ERROR_SUCCESS) { flb_plg_error(ctx->ins, "EnableTraceEx2 failed (status=%lu)", status); etw_stop_session(ctx); + etw_signal_startup(ctx, -1); return NULL; } } @@ -1145,9 +1188,12 @@ static void *etw_worker(void *data) if (trace == INVALID_PROCESSTRACE_HANDLE) { flb_plg_error(ctx->ins, "OpenTrace failed (error=%lu)", GetLastError()); etw_stop_session(ctx); + etw_signal_startup(ctx, -1); return NULL; } + etw_signal_startup(ctx, 0); + status = ProcessTrace(&trace, 1, NULL, NULL); if (status != ERROR_SUCCESS && !InterlockedCompareExchange(&ctx->exiting, 0, 0)) { flb_plg_error(ctx->ins, "ProcessTrace failed (status=%lu)", status); @@ -1183,6 +1229,11 @@ static void etw_config_destroy(struct flb_etw *ctx) flb_free(ctx->properties); } + if (ctx->startup_sync_initialized) { + pthread_cond_destroy(&ctx->startup_cond); + pthread_mutex_destroy(&ctx->startup_lock); + } + DeleteCriticalSection(&ctx->handle_lock); flb_free(ctx); @@ -1377,6 +1428,22 @@ static int in_etw_init(struct flb_input_instance *in, ctx->loss_metrics_collector_id = -1; InitializeCriticalSection(&ctx->handle_lock); + ret = pthread_mutex_init(&ctx->startup_lock, NULL); + if (ret != 0) { + DeleteCriticalSection(&ctx->handle_lock); + flb_free(ctx); + return -1; + } + + ret = pthread_cond_init(&ctx->startup_cond, NULL); + if (ret != 0) { + pthread_mutex_destroy(&ctx->startup_lock); + DeleteCriticalSection(&ctx->handle_lock); + flb_free(ctx); + return -1; + } + ctx->startup_sync_initialized = FLB_TRUE; + ret = flb_input_config_map_set(in, ctx); if (ret == -1) { etw_config_destroy(ctx); @@ -1458,11 +1525,6 @@ static int in_etw_init(struct flb_input_instance *in, flb_input_set_context(in, ctx); - if (configure_loss_metrics(ctx, config) != 0) { - etw_config_destroy(ctx); - return -1; - } - ret = pthread_create(&ctx->thread, NULL, etw_worker, ctx); if (ret != 0) { flb_plg_error(ctx->ins, "could not create ETW worker thread"); @@ -1471,6 +1533,16 @@ static int in_etw_init(struct flb_input_instance *in, } ctx->thread_created = FLB_TRUE; + if (etw_wait_startup(ctx) != 0) { + etw_config_destroy(ctx); + return -1; + } + + if (configure_loss_metrics(ctx, config) != 0) { + etw_config_destroy(ctx); + return -1; + } + return 0; } diff --git a/plugins/in_etw/etw.h b/plugins/in_etw/etw.h index 64707e9a6d5..169c5eeee51 100644 --- a/plugins/in_etw/etw.h +++ b/plugins/in_etw/etw.h @@ -71,11 +71,16 @@ struct flb_etw { TRACEHANDLE session; TRACEHANDLE trace; CRITICAL_SECTION handle_lock; + pthread_mutex_t startup_lock; + pthread_cond_t startup_cond; LONG exiting; LONG paused; LONG append_errors; LONG query_errors; + int startup_sync_initialized; + int startup_done; + int startup_status; int thread_created; int loss_metrics_collector_id; pthread_t thread; From ad8ce10f841188d2ea5d3ded2c840d669542eb7f Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Wed, 8 Jul 2026 23:08:07 +0900 Subject: [PATCH 5/6] in_etw: Adjust cunsuming workers lifecycles on pause/resume Signed-off-by: Hiroshi Hatake --- plugins/in_etw/etw.c | 93 ++++++++++++++++++++++++++++++++++++-------- plugins/in_etw/etw.h | 1 + 2 files changed, 78 insertions(+), 16 deletions(-) diff --git a/plugins/in_etw/etw.c b/plugins/in_etw/etw.c index 98b9535bca0..087b15db1db 100644 --- a/plugins/in_etw/etw.c +++ b/plugins/in_etw/etw.c @@ -907,6 +907,11 @@ static int etw_loss_metrics_collect(struct flb_input_instance *in, struct flb_etw *ctx; ctx = data; + if (InterlockedCompareExchange(&ctx->paused, 0, 0) || + InterlockedCompareExchange(&ctx->exiting, 0, 0)) { + return 0; + } + etw_update_loss_metrics(ctx); return 0; @@ -926,8 +931,11 @@ static VOID WINAPI etw_event_callback(PEVENT_RECORD record) return; } + EnterCriticalSection(&ctx->callback_lock); + if (InterlockedCompareExchange(&ctx->paused, 0, 0) || InterlockedCompareExchange(&ctx->exiting, 0, 0)) { + LeaveCriticalSection(&ctx->callback_lock); return; } @@ -998,6 +1006,7 @@ static VOID WINAPI etw_event_callback(PEVENT_RECORD record) } msgpack_sbuffer_destroy(&mp_sbuf); + LeaveCriticalSection(&ctx->callback_lock); } static void etw_close_trace(struct flb_etw *ctx) @@ -1195,7 +1204,9 @@ static void *etw_worker(void *data) etw_signal_startup(ctx, 0); status = ProcessTrace(&trace, 1, NULL, NULL); - if (status != ERROR_SUCCESS && !InterlockedCompareExchange(&ctx->exiting, 0, 0)) { + if (status != ERROR_SUCCESS && + !InterlockedCompareExchange(&ctx->paused, 0, 0) && + !InterlockedCompareExchange(&ctx->exiting, 0, 0)) { flb_plg_error(ctx->ins, "ProcessTrace failed (status=%lu)", status); } @@ -1208,19 +1219,67 @@ static void *etw_worker(void *data) return NULL; } -static void etw_config_destroy(struct flb_etw *ctx) +static void etw_reset_startup(struct flb_etw *ctx) { - if (ctx == NULL) { + pthread_mutex_lock(&ctx->startup_lock); + ctx->startup_done = FLB_FALSE; + ctx->startup_status = 0; + pthread_mutex_unlock(&ctx->startup_lock); +} + +static int etw_start_worker(struct flb_etw *ctx) +{ + int ret; + + if (ctx->thread_created) { + return 0; + } + + etw_reset_startup(ctx); + + ret = pthread_create(&ctx->thread, NULL, etw_worker, ctx); + if (ret != 0) { + flb_plg_error(ctx->ins, "could not create ETW worker thread"); + return -1; + } + ctx->thread_created = FLB_TRUE; + + if (etw_wait_startup(ctx) != 0) { + pthread_join(ctx->thread, NULL); + ctx->thread_created = FLB_FALSE; + return -1; + } + + return 0; +} + +static void etw_stop_worker(struct flb_etw *ctx, int exiting) +{ + if (!ctx->thread_created) { return; } - if (ctx->thread_created) { + if (exiting) { InterlockedExchange(&ctx->exiting, 1); - etw_close_trace(ctx); - etw_stop_session(ctx); - pthread_join(ctx->thread, NULL); } + EnterCriticalSection(&ctx->callback_lock); + LeaveCriticalSection(&ctx->callback_lock); + + etw_close_trace(ctx); + etw_stop_session(ctx); + pthread_join(ctx->thread, NULL); + ctx->thread_created = FLB_FALSE; +} + +static void etw_config_destroy(struct flb_etw *ctx) +{ + if (ctx == NULL) { + return; + } + + etw_stop_worker(ctx, FLB_TRUE); + if (ctx->session_name_wide != NULL) { flb_free(ctx->session_name_wide); } @@ -1235,6 +1294,7 @@ static void etw_config_destroy(struct flb_etw *ctx) } DeleteCriticalSection(&ctx->handle_lock); + DeleteCriticalSection(&ctx->callback_lock); flb_free(ctx); } @@ -1427,9 +1487,11 @@ static int in_etw_init(struct flb_input_instance *in, ctx->trace = INVALID_PROCESSTRACE_HANDLE; ctx->loss_metrics_collector_id = -1; InitializeCriticalSection(&ctx->handle_lock); + InitializeCriticalSection(&ctx->callback_lock); ret = pthread_mutex_init(&ctx->startup_lock, NULL); if (ret != 0) { + DeleteCriticalSection(&ctx->callback_lock); DeleteCriticalSection(&ctx->handle_lock); flb_free(ctx); return -1; @@ -1438,6 +1500,7 @@ static int in_etw_init(struct flb_input_instance *in, ret = pthread_cond_init(&ctx->startup_cond, NULL); if (ret != 0) { pthread_mutex_destroy(&ctx->startup_lock); + DeleteCriticalSection(&ctx->callback_lock); DeleteCriticalSection(&ctx->handle_lock); flb_free(ctx); return -1; @@ -1525,15 +1588,7 @@ static int in_etw_init(struct flb_input_instance *in, flb_input_set_context(in, ctx); - ret = pthread_create(&ctx->thread, NULL, etw_worker, ctx); - if (ret != 0) { - flb_plg_error(ctx->ins, "could not create ETW worker thread"); - etw_config_destroy(ctx); - return -1; - } - ctx->thread_created = FLB_TRUE; - - if (etw_wait_startup(ctx) != 0) { + if (etw_start_worker(ctx) != 0) { etw_config_destroy(ctx); return -1; } @@ -1555,6 +1610,7 @@ static void in_etw_pause(void *data, struct flb_config *config) if (ctx->loss_metrics_collector_id >= 0) { flb_input_collector_pause(ctx->loss_metrics_collector_id, ctx->ins); } + etw_stop_worker(ctx, FLB_FALSE); } static void in_etw_resume(void *data, struct flb_config *config) @@ -1563,6 +1619,11 @@ static void in_etw_resume(void *data, struct flb_config *config) ctx = data; InterlockedExchange(&ctx->paused, 0); + if (etw_start_worker(ctx) != 0) { + InterlockedExchange(&ctx->paused, 1); + return; + } + if (ctx->loss_metrics_collector_id >= 0) { flb_input_collector_resume(ctx->loss_metrics_collector_id, ctx->ins); } diff --git a/plugins/in_etw/etw.h b/plugins/in_etw/etw.h index 169c5eeee51..dda68ee5c22 100644 --- a/plugins/in_etw/etw.h +++ b/plugins/in_etw/etw.h @@ -71,6 +71,7 @@ struct flb_etw { TRACEHANDLE session; TRACEHANDLE trace; CRITICAL_SECTION handle_lock; + CRITICAL_SECTION callback_lock; pthread_mutex_t startup_lock; pthread_cond_t startup_cond; From 66449ac0dc65a62886fcf03c2a589673d2a88433 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Fri, 10 Jul 2026 13:31:38 +0900 Subject: [PATCH 6/6] in_etw: harden ETW ingestion and session handling Signed-off-by: Hiroshi Hatake --- plugins/in_etw/etw.c | 234 ++++++++++++++++++++++++++++++++----------- plugins/in_etw/etw.h | 2 +- 2 files changed, 178 insertions(+), 58 deletions(-) diff --git a/plugins/in_etw/etw.c b/plugins/in_etw/etw.c index 087b15db1db..b490df192c4 100644 --- a/plugins/in_etw/etw.c +++ b/plugins/in_etw/etw.c @@ -29,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -89,20 +90,21 @@ static int parse_guid(const char *str, GUID *guid) unsigned int d2; unsigned int d3; unsigned int d4[8]; + char trailing; int ret; ret = sscanf(str, - "{%8x-%4x-%4x-%2x%2x-%2x%2x%2x%2x%2x%2x}", + "{%8x-%4x-%4x-%2x%2x-%2x%2x%2x%2x%2x%2x}%c", &d1, &d2, &d3, &d4[0], &d4[1], &d4[2], &d4[3], - &d4[4], &d4[5], &d4[6], &d4[7]); + &d4[4], &d4[5], &d4[6], &d4[7], &trailing); if (ret != 11) { ret = sscanf(str, - "%8x-%4x-%4x-%2x%2x-%2x%2x%2x%2x%2x%2x", + "%8x-%4x-%4x-%2x%2x-%2x%2x%2x%2x%2x%2x%c", &d1, &d2, &d3, &d4[0], &d4[1], &d4[2], &d4[3], - &d4[4], &d4[5], &d4[6], &d4[7]); + &d4[4], &d4[5], &d4[6], &d4[7], &trailing); } if (ret != 11) { @@ -129,12 +131,13 @@ static int parse_uint64(const char *str, ULONGLONG *out) char *end; unsigned long long value; - if (str == NULL || str[0] == '\0') { + if (str == NULL || str[0] == '\0' || str[0] == '-') { return -1; } + errno = 0; value = strtoull(str, &end, 0); - if (end == str || *end != '\0') { + if (end == str || *end != '\0' || errno == ERANGE) { return -1; } @@ -418,6 +421,46 @@ static void pack_wide_string(msgpack_packer *mp_pck, const WCHAR *str, int bytes flb_free(utf8); } +static int pack_bounded_wide_string(msgpack_packer *mp_pck, const WCHAR *str, + size_t bytes) +{ + size_t chars; + size_t len; + int utf8_len; + char *utf8; + + if (bytes > INT_MAX || bytes < sizeof(WCHAR)) { + return -1; + } + + chars = bytes / sizeof(WCHAR); + for (len = 0; len < chars; len++) { + if (str[len] == L'\0') { + break; + } + } + + if (len == chars) { + return -1; + } + + if (len == 0) { + msgpack_pack_str(mp_pck, 0); + return 0; + } + + utf8 = wide_to_utf8(str, (int) (len * sizeof(WCHAR)), &utf8_len); + if (utf8 == NULL) { + return -1; + } + + msgpack_pack_str(mp_pck, utf8_len); + msgpack_pack_str_body(mp_pck, utf8, utf8_len); + flb_free(utf8); + + return 0; +} + static int resolve_provider_name(struct flb_etw *ctx, const WCHAR *name, GUID *guid) { ULONG size; @@ -564,14 +607,21 @@ static void pack_filetime_value(msgpack_packer *mp_pck, const FILETIME *filetime msgpack_pack_uint64(mp_pck, unix_100ns / 10000000ULL); } -static int pack_nil_if_property_too_small(msgpack_packer *mp_pck, ULONG size, - size_t expected) +static int pack_non_scalar_property_value(msgpack_packer *mp_pck, + const BYTE *buf, + ULONG size, size_t expected) { if (size < expected) { msgpack_pack_nil(mp_pck); return FLB_TRUE; } + if (size > expected) { + msgpack_pack_bin(mp_pck, size); + msgpack_pack_bin_body(mp_pck, buf, size); + return FLB_TRUE; + } + return FLB_FALSE; } @@ -591,57 +641,57 @@ static void pack_property_value(msgpack_packer *mp_pck, USHORT in_type, msgpack_pack_str_body(mp_pck, (char *) buf, len); break; case TDH_INTYPE_INT8: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT8))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(INT8))) { break; } msgpack_pack_int8(mp_pck, *((INT8 *) buf)); break; case TDH_INTYPE_UINT8: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT8))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(UINT8))) { break; } msgpack_pack_uint8(mp_pck, *((UINT8 *) buf)); break; case TDH_INTYPE_INT16: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT16))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(INT16))) { break; } msgpack_pack_int16(mp_pck, *((INT16 *) buf)); break; case TDH_INTYPE_UINT16: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT16))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(UINT16))) { break; } msgpack_pack_uint16(mp_pck, *((UINT16 *) buf)); break; case TDH_INTYPE_INT32: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT32))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(INT32))) { break; } msgpack_pack_int32(mp_pck, *((INT32 *) buf)); break; case TDH_INTYPE_UINT32: case TDH_INTYPE_HEXINT32: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT32))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(UINT32))) { break; } msgpack_pack_uint32(mp_pck, *((UINT32 *) buf)); break; case TDH_INTYPE_INT64: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(INT64))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(INT64))) { break; } msgpack_pack_int64(mp_pck, *((INT64 *) buf)); break; case TDH_INTYPE_UINT64: case TDH_INTYPE_HEXINT64: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(UINT64))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(UINT64))) { break; } msgpack_pack_uint64(mp_pck, *((UINT64 *) buf)); break; case TDH_INTYPE_BOOLEAN: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(BOOL))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(BOOL))) { break; } if (*((BOOL *) buf)) { @@ -652,25 +702,25 @@ static void pack_property_value(msgpack_packer *mp_pck, USHORT in_type, } break; case TDH_INTYPE_FLOAT: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(float))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(float))) { break; } msgpack_pack_float(mp_pck, *((float *) buf)); break; case TDH_INTYPE_DOUBLE: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(double))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(double))) { break; } msgpack_pack_double(mp_pck, *((double *) buf)); break; case TDH_INTYPE_GUID: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(GUID))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(GUID))) { break; } pack_guid_value(mp_pck, (GUID *) buf); break; case TDH_INTYPE_FILETIME: - if (pack_nil_if_property_too_small(mp_pck, size, sizeof(FILETIME))) { + if (pack_non_scalar_property_value(mp_pck, buf, size, sizeof(FILETIME))) { break; } pack_filetime_value(mp_pck, (FILETIME *) buf); @@ -700,7 +750,8 @@ static void pack_property_value(msgpack_packer *mp_pck, USHORT in_type, } static void pack_payload_property(msgpack_packer *mp_pck, PEVENT_RECORD record, - TRACE_EVENT_INFO *info, ULONG index) + TRACE_EVENT_INFO *info, ULONG info_size, + ULONG index) { ULONG size; TDHSTATUS status; @@ -714,6 +765,12 @@ static void pack_payload_property(msgpack_packer *mp_pck, PEVENT_RECORD record, return; } + if (prop->NameOffset == 0 || + prop->NameOffset > info_size - sizeof(WCHAR)) { + msgpack_pack_nil(mp_pck); + return; + } + memset(&desc, 0, sizeof(desc)); desc.PropertyName = (ULONGLONG) ((char *) info + prop->NameOffset); desc.ArrayIndex = ULONG_MAX; @@ -752,6 +809,7 @@ static void pack_payload(msgpack_packer *mp_pck, PEVENT_RECORD record) EVENT_PROPERTY_INFO *prop; WCHAR *name; char fallback[32]; + size_t property_info_size; size = 0; status = TdhGetEventInformation(record, 0, NULL, NULL, &size); @@ -774,19 +832,36 @@ static void pack_payload(msgpack_packer *mp_pck, PEVENT_RECORD record) return; } + if (size < offsetof(TRACE_EVENT_INFO, EventPropertyInfoArray)) { + flb_free(info); + msgpack_pack_map(mp_pck, 0); + return; + } + + property_info_size = (size - offsetof(TRACE_EVENT_INFO, EventPropertyInfoArray)) / + sizeof(EVENT_PROPERTY_INFO); + if (info->TopLevelPropertyCount > property_info_size) { + flb_free(info); + msgpack_pack_map(mp_pck, 0); + return; + } + msgpack_pack_map(mp_pck, info->TopLevelPropertyCount); for (i = 0; i < info->TopLevelPropertyCount; i++) { prop = &info->EventPropertyInfoArray[i]; - if (prop->NameOffset > 0) { + if (prop->NameOffset > 0 && + prop->NameOffset <= size - sizeof(WCHAR)) { name = (WCHAR *) ((char *) info + prop->NameOffset); - pack_wide_string(mp_pck, name, 0); - } - else { - snprintf(fallback, sizeof(fallback), "property_%lu", i); - pack_cstr(mp_pck, fallback); + if (pack_bounded_wide_string(mp_pck, name, + size - prop->NameOffset) == 0) { + pack_payload_property(mp_pck, record, info, size, i); + continue; + } } - pack_payload_property(mp_pck, record, info, i); + snprintf(fallback, sizeof(fallback), "property_%lu", i); + pack_cstr(mp_pck, fallback); + pack_payload_property(mp_pck, record, info, size, i); } flb_free(info); @@ -857,13 +932,15 @@ static int etw_update_loss_metrics(struct flb_etw *ctx) { ULONG status; TRACEHANDLE session; + EVENT_TRACE_PROPERTIES *properties; uint64_t timestamp; + ULONG events_lost; + ULONG realtime_buffers_lost; char *name; LONG query_errors; if (ctx->cmt_events_lost == NULL || - ctx->cmt_realtime_buffers_lost == NULL || - ctx->properties == NULL) { + ctx->cmt_realtime_buffers_lost == NULL) { return 0; } @@ -872,11 +949,17 @@ static int etw_update_loss_metrics(struct flb_etw *ctx) return 0; } + properties = create_trace_properties(ctx); + if (properties == NULL) { + return -1; + } + status = ControlTraceW(session, ctx->session_name_wide, - ctx->properties, + properties, EVENT_TRACE_CONTROL_QUERY); if (status != ERROR_SUCCESS) { + flb_free(properties); query_errors = InterlockedIncrement(&ctx->query_errors); if (query_errors == 1 || query_errors % 1000 == 0) { flb_plg_warn(ctx->ins, @@ -886,16 +969,20 @@ static int etw_update_loss_metrics(struct flb_etw *ctx) return -1; } + events_lost = properties->EventsLost; + realtime_buffers_lost = properties->RealTimeBuffersLost; + flb_free(properties); + name = (char *) flb_input_name(ctx->ins); timestamp = cfl_time_now(); cmt_gauge_set(ctx->cmt_events_lost, timestamp, - (double) ctx->properties->EventsLost, + (double) events_lost, 1, (char *[]) {name}); cmt_gauge_set(ctx->cmt_realtime_buffers_lost, timestamp, - (double) ctx->properties->RealTimeBuffersLost, + (double) realtime_buffers_lost, 1, (char *[]) {name}); return 0; @@ -995,12 +1082,13 @@ static VOID WINAPI etw_event_callback(PEVENT_RECORD record) pack_cstr(&mp_pck, "payload"); pack_payload(&mp_pck, record); - ret = flb_input_log_append(ctx->ins, NULL, 0, mp_sbuf.data, mp_sbuf.size); - if (ret == -1) { + ret = flb_input_ingress_queue_log(ctx->ins, NULL, 0, + mp_sbuf.data, mp_sbuf.size); + if (ret != 0) { append_errors = InterlockedIncrement(&ctx->append_errors); if (append_errors == 1 || append_errors % 1000 == 0) { flb_plg_warn(ctx->ins, - "failed to append ETW event record (%ld failures)", + "failed to enqueue ETW event record (%ld failures)", append_errors); } } @@ -1088,17 +1176,28 @@ static void etw_stop_session(struct flb_etw *ctx) { TRACEHANDLE session; ULONG status; + EVENT_TRACE_PROPERTIES *properties; + + properties = create_trace_properties(ctx); + if (properties == NULL) { + flb_plg_warn(ctx->ins, "could not allocate ETW stop properties"); + return; + } session = etw_exchange_session(ctx, 0); - if (session != 0 && ctx->properties != NULL) { - etw_disable_provider(ctx, session); - status = ControlTraceW(session, - ctx->session_name_wide, - ctx->properties, - EVENT_TRACE_CONTROL_STOP); - if (status != ERROR_SUCCESS && status != ERROR_WMI_INSTANCE_NOT_FOUND) { - flb_plg_warn(ctx->ins, "ControlTrace stop failed (status=%lu)", status); - } + if (session == 0) { + flb_free(properties); + return; + } + + etw_disable_provider(ctx, session); + status = ControlTraceW(session, + ctx->session_name_wide, + properties, + EVENT_TRACE_CONTROL_STOP); + flb_free(properties); + if (status != ERROR_SUCCESS && status != ERROR_WMI_INSTANCE_NOT_FOUND) { + flb_plg_warn(ctx->ins, "ControlTrace stop failed (status=%lu)", status); } } @@ -1106,9 +1205,16 @@ static ULONG etw_start_session(struct flb_etw *ctx) { ULONG status; TRACEHANDLE session; + EVENT_TRACE_PROPERTIES *properties; session = 0; - status = StartTraceW(&session, ctx->session_name_wide, ctx->properties); + properties = create_trace_properties(ctx); + if (properties == NULL) { + return ERROR_NOT_ENOUGH_MEMORY; + } + + status = StartTraceW(&session, ctx->session_name_wide, properties); + flb_free(properties); if (status != ERROR_ALREADY_EXISTS) { if (status == ERROR_SUCCESS) { etw_exchange_session(ctx, session); @@ -1133,7 +1239,13 @@ static ULONG etw_start_session(struct flb_etw *ctx) } session = 0; - status = StartTraceW(&session, ctx->session_name_wide, ctx->properties); + properties = create_trace_properties(ctx); + if (properties == NULL) { + return ERROR_NOT_ENOUGH_MEMORY; + } + + status = StartTraceW(&session, ctx->session_name_wide, properties); + flb_free(properties); if (status == ERROR_SUCCESS) { etw_exchange_session(ctx, session); } @@ -1163,6 +1275,7 @@ static void *etw_worker(void *data) ctx->session_name_wide, status); } etw_signal_startup(ctx, -1); + InterlockedExchange(&ctx->worker_finished, 1); return NULL; } @@ -1181,6 +1294,7 @@ static void *etw_worker(void *data) flb_plg_error(ctx->ins, "EnableTraceEx2 failed (status=%lu)", status); etw_stop_session(ctx); etw_signal_startup(ctx, -1); + InterlockedExchange(&ctx->worker_finished, 1); return NULL; } } @@ -1198,6 +1312,7 @@ static void *etw_worker(void *data) flb_plg_error(ctx->ins, "OpenTrace failed (error=%lu)", GetLastError()); etw_stop_session(ctx); etw_signal_startup(ctx, -1); + InterlockedExchange(&ctx->worker_finished, 1); return NULL; } @@ -1216,6 +1331,8 @@ static void *etw_worker(void *data) etw_stop_session(ctx); } + InterlockedExchange(&ctx->worker_finished, 1); + return NULL; } @@ -1232,10 +1349,16 @@ static int etw_start_worker(struct flb_etw *ctx) int ret; if (ctx->thread_created) { - return 0; + if (!InterlockedCompareExchange(&ctx->worker_finished, 0, 0)) { + return 0; + } + + pthread_join(ctx->thread, NULL); + ctx->thread_created = FLB_FALSE; } etw_reset_startup(ctx); + InterlockedExchange(&ctx->worker_finished, 0); ret = pthread_create(&ctx->thread, NULL, etw_worker, ctx); if (ret != 0) { @@ -1284,10 +1407,6 @@ static void etw_config_destroy(struct flb_etw *ctx) flb_free(ctx->session_name_wide); } - if (ctx->properties != NULL) { - flb_free(ctx->properties); - } - if (ctx->startup_sync_initialized) { pthread_cond_destroy(&ctx->startup_cond); pthread_mutex_destroy(&ctx->startup_lock); @@ -1580,14 +1699,15 @@ static int in_etw_init(struct flb_input_instance *in, return -1; } - ctx->properties = create_trace_properties(ctx); - if (ctx->properties == NULL) { + flb_input_set_context(in, ctx); + + ret = flb_input_ingress_enable(in); + if (ret != 0) { + flb_plg_error(ctx->ins, "could not enable ETW ingress queue"); etw_config_destroy(ctx); return -1; } - flb_input_set_context(in, ctx); - if (etw_start_worker(ctx) != 0) { etw_config_destroy(ctx); return -1; diff --git a/plugins/in_etw/etw.h b/plugins/in_etw/etw.h index dda68ee5c22..2ac2fea12dd 100644 --- a/plugins/in_etw/etw.h +++ b/plugins/in_etw/etw.h @@ -67,7 +67,6 @@ struct flb_etw { ULONG kernel_flags; GUID session_guid; WCHAR *session_name_wide; - EVENT_TRACE_PROPERTIES *properties; TRACEHANDLE session; TRACEHANDLE trace; CRITICAL_SECTION handle_lock; @@ -77,6 +76,7 @@ struct flb_etw { LONG exiting; LONG paused; + LONG worker_finished; LONG append_errors; LONG query_errors; int startup_sync_initialized;