feat(instrumentation,telemetry): Incorporate thread pool telemetry

The thread pool telemetry is captured from the
dedicated thread pool provider. Presently, we
collect three different event types including
the call stacks. More specifically, SubmitThreadpoolWork, SubmitThreadpoolCallback, and
SetThreadpoolTimer events.
Additionally, the callback addresses are symbolized
to derive to function name, and when the callback is
the ZwContinue or RtlCaputreContext function call, we
also try to decode the CONTEXT structure and symbolize
the instruction pointer address.

Author: rabbitstack

configs/fibratus.yml

@@ -229,6 +229,9 @@ kstream:
   # Determines whether DNS client events are collected
   #enable-dns: true
 
+  # Determines whether thread pool events are collected
+  #enable-threadpool: true
+
   # Indicates if stack enrichment is enabled for eligible events
   #stack-enrichment: true
 

internal/etw/source.go

@@ -147,6 +147,7 @@ func (e *EventSource) Open(config *config.Config) error {
 		config.Kstream.EnableMemKevents = config.Kstream.EnableMemKevents && (e.r.HasMemEvents || (config.Yara.Enabled && !config.Yara.SkipAllocs))
 		config.Kstream.EnableDNSEvents = config.Kstream.EnableDNSEvents && e.r.HasDNSEvents
 		config.Kstream.EnableAuditAPIEvents = config.Kstream.EnableAuditAPIEvents && e.r.HasAuditAPIEvents
+		config.Kstream.EnableThreadpoolEvents = config.Kstream.EnableThreadpoolEvents && e.r.HasThreadpoolEvents
 		for _, ktype := range ktypes.All() {
 			if ktype == ktypes.CreateProcess || ktype == ktypes.TerminateProcess ||
 				ktype == ktypes.LoadImage || ktype == ktypes.UnloadImage {
@@ -189,6 +190,9 @@ func (e *EventSource) Open(config *config.Config) error {
 	if config.Kstream.EnableAuditAPIEvents {
 		e.addTrace(etw.KernelAuditAPICallsSession, etw.KernelAuditAPICallsGUID)
 	}
+	if config.Kstream.EnableThreadpoolEvents {
+		e.addTrace(etw.ThreadpoolSession, etw.ThreadpoolGUID)
+	}
 
 	for _, trace := range e.traces {
 		err := trace.Start()
@@ -226,6 +230,7 @@ func (e *EventSource) Open(config *config.Config) error {
 		// Init consumer and open the trace for processing
 		consumer := NewConsumer(e.psnap, e.hsnap, config, e.sequencer, e.evts)
 		consumer.SetFilter(e.filter)
+
 		// Attach event listeners
 		for _, lis := range e.listeners {
 			consumer.q.RegisterListener(lis)

internal/etw/stackext.go

@@ -100,3 +100,12 @@ func (s *StackExtensions) EnableMemoryCallstack() {
 		s.AddStackTracing(ktypes.VirtualAlloc)
 	}
 }
+
+// EnableThreadpoolCallstack enables stack tracing for thread pool events.
+func (s *StackExtensions) EnableThreadpoolCallstack() {
+	if s.config.EnableThreadpoolEvents {
+		s.AddStackTracing(ktypes.SubmitThreadpoolWork)
+		s.AddStackTracing(ktypes.SubmitThreadpoolCallback)
+		s.AddStackTracing(ktypes.SetThreadpoolTimer)
+	}
+}

internal/etw/trace.go

@@ -150,6 +150,10 @@ func (t *Trace) enableCallstacks() {
 	if t.IsSystemRegistryTrace() {
 		t.stackExtensions.EnableRegistryCallstack()
 	}
+
+	if t.IsThreadpoolTrace() {
+		t.stackExtensions.EnableThreadpoolCallstack()
+	}
 }
 
 // Start registers and starts an event tracing session.
@@ -202,7 +206,9 @@ func (t *Trace) Start() error {
 			log.Warnf("unable to set empty system flags: %v", err)
 			return nil
 		}
+
 		sysTraceFlags[0] = flags
+
 		// enable object manager tracking
 		if cfg.EnableHandleKevents {
 			sysTraceFlags[4] = etw.Handle
@@ -225,13 +231,14 @@ func (t *Trace) Start() error {
 	// enrichment is enabled, it is necessary to instruct the provider
 	// to emit stack addresses in the extended data item section when
 	// writing events to the session buffers
-	if cfg.StackEnrichment && !t.IsSystemProvider() {
+	if cfg.StackEnrichment && !t.IsSystemProvider() && !t.IsThreadpoolTrace() {
 		return etw.EnableTraceWithOpts(t.GUID, t.startHandle, t.Keywords, etw.EnableTraceOpts{WithStacktrace: true})
 	} else if cfg.StackEnrichment && len(t.stackExtensions.EventIds()) > 0 {
 		if err := etw.EnableStackTracing(t.startHandle, t.stackExtensions.EventIds()); err != nil {
 			return fmt.Errorf("fail to enable system events callstack tracing: %v", err)
 		}
 	}
+
 	if t.IsSystemRegistryTrace() {
 		if err := etw.EnableTrace(t.GUID, t.startHandle, t.Keywords); err != nil {
 			return err
@@ -249,6 +256,7 @@ func (t *Trace) Start() error {
 		sysTraceFlags[0] = etw.Registry
 		return etw.SetTraceSystemFlags(handle, sysTraceFlags)
 	}
+
 	return etw.EnableTrace(t.GUID, t.startHandle, t.Keywords)
 }
 
@@ -343,6 +351,9 @@ func (t *Trace) IsKernelTrace() bool { return t.GUID == etw.KernelTraceControlGU
 // IsSystemRegistryTrace determines if this is the system registry logger trace.
 func (t *Trace) IsSystemRegistryTrace() bool { return t.GUID == etw.SystemRegistryProviderID }
 
+// IsThreadpoolTrace determines if this is the thread pool logger trace.
+func (t *Trace) IsThreadpoolTrace() bool { return t.GUID == etw.ThreadpoolGUID }
+
 // IsSystemProvider determines if this is one of the granular system provider traces.
 func (t *Trace) IsSystemProvider() bool {
 	return t.GUID == etw.SystemIOProviderID || t.GUID == etw.SystemRegistryProviderID || t.GUID == etw.SystemProcessProviderID || t.GUID == etw.SystemMemoryProviderID

pkg/config/config_windows.go

@@ -411,6 +411,7 @@ func (c *Config) addFlags() {
 		c.flags.Bool(enableMemKevents, true, "Determines whether memory manager kernel events are collected by Kernel Logger provider")
 		c.flags.Bool(enableAuditAPIEvents, true, "Determines whether kernel audit API calls events are published")
 		c.flags.Bool(enableDNSEvents, true, "Determines whether DNS client events are enabled")
+		c.flags.Bool(enableThreadpoolEvents, true, "Determines whether thread pool events are published")
 		c.flags.Bool(stackEnrichment, true, "Indicates if stack enrichment is enabled for eligible events")
 		c.flags.Int(bufferSize, int(maxBufferSize), "Represents the amount of memory allocated for each event tracing session buffer, in kilobytes. The buffer size affects the rate at which buffers fill and must be flushed (small buffer size requires less memory but it increases the rate at which buffers must be flushed)")
 		c.flags.Int(minBuffers, int(defaultMinBuffers), "Determines the minimum number of buffers allocated for the event tracing session's buffer pool")

pkg/config/filters.go

@@ -170,19 +170,20 @@ func (ctx *ActionContext) UniquePids() []uint32 {
 // enabling/disabling event providers/types
 // dynamically.
 type RulesCompileResult struct {
-	HasProcEvents     bool
-	HasThreadEvents   bool
-	HasImageEvents    bool
-	HasFileEvents     bool
-	HasNetworkEvents  bool
-	HasRegistryEvents bool
-	HasHandleEvents   bool
-	HasMemEvents      bool
-	HasVAMapEvents    bool
-	HasDNSEvents      bool
-	HasAuditAPIEvents bool
-	UsedEvents        []ktypes.Ktype
-	NumberRules       int
+	HasProcEvents       bool
+	HasThreadEvents     bool
+	HasImageEvents      bool
+	HasFileEvents       bool
+	HasNetworkEvents    bool
+	HasRegistryEvents   bool
+	HasHandleEvents     bool
+	HasMemEvents        bool
+	HasVAMapEvents      bool
+	HasDNSEvents        bool
+	HasAuditAPIEvents   bool
+	HasThreadpoolEvents bool
+	UsedEvents          []ktypes.Ktype
+	NumberRules         int
 }
 
 func (r RulesCompileResult) ContainsEvent(ktype ktypes.Ktype) bool {
@@ -217,6 +218,7 @@ func (r RulesCompileResult) String() string {
 		HasVAMapEvents: %t
 		HasAuditAPIEvents: %t
 		HasDNSEvents: %t
+		HasThreadpoolEvents: %t
 		Events: %s`,
 		r.HasProcEvents,
 		r.HasThreadEvents,
@@ -229,6 +231,7 @@ func (r RulesCompileResult) String() string {
 		r.HasVAMapEvents,
 		r.HasAuditAPIEvents,
 		r.HasDNSEvents,
+		r.HasThreadpoolEvents,
 		strings.Join(events, ", "),
 	)
 }

pkg/config/kstream.go

@@ -32,21 +32,22 @@ import (
 )
 
 const (
-	enableThreadKevents   = "kstream.enable-thread"
-	enableRegistryKevents = "kstream.enable-registry"
-	enableNetKevents      = "kstream.enable-net"
-	enableFileIOKevents   = "kstream.enable-fileio"
-	enableVAMapKevents    = "kstream.enable-vamap"
-	enableImageKevents    = "kstream.enable-image"
-	enableHandleKevents   = "kstream.enable-handle"
-	enableMemKevents      = "kstream.enable-mem"
-	enableAuditAPIEvents  = "kstream.enable-audit-api"
-	enableDNSEvents       = "kstream.enable-dns"
-	stackEnrichment       = "kstream.stack-enrichment"
-	bufferSize            = "kstream.buffer-size"
-	minBuffers            = "kstream.min-buffers"
-	maxBuffers            = "kstream.max-buffers"
-	flushInterval         = "kstream.flush-interval"
+	enableThreadKevents    = "kstream.enable-thread"
+	enableRegistryKevents  = "kstream.enable-registry"
+	enableNetKevents       = "kstream.enable-net"
+	enableFileIOKevents    = "kstream.enable-fileio"
+	enableVAMapKevents     = "kstream.enable-vamap"
+	enableImageKevents     = "kstream.enable-image"
+	enableHandleKevents    = "kstream.enable-handle"
+	enableMemKevents       = "kstream.enable-mem"
+	enableAuditAPIEvents   = "kstream.enable-audit-api"
+	enableDNSEvents        = "kstream.enable-dns"
+	enableThreadpoolEvents = "kstream.enable-threadpool"
+	stackEnrichment        = "kstream.stack-enrichment"
+	bufferSize             = "kstream.buffer-size"
+	minBuffers             = "kstream.min-buffers"
+	maxBuffers             = "kstream.max-buffers"
+	flushInterval          = "kstream.flush-interval"
 
 	excludedEvents = "kstream.blacklist.events"
 	excludedImages = "kstream.blacklist.images"
@@ -82,6 +83,8 @@ type KstreamConfig struct {
 	EnableAuditAPIEvents bool `json:"enable-audit-api" yaml:"enable-audit-api"`
 	// EnableDNSEvents indicates if DNS client events are enabled
 	EnableDNSEvents bool `json:"enable-dns" yaml:"enable-dns"`
+	// EnableThreadpoolEvents indicates if thread pool events are enabled
+	EnableThreadpoolEvents bool `json:"enable-threadpool" yaml:"enable-threadpool"`
 	// StackEnrichment indicates if stack enrichment is enabled for eligible events.
 	StackEnrichment bool `json:"stack-enrichment" yaml:"stack-enrichment"`
 	// BufferSize represents the amount of memory allocated for each event tracing session buffer, in kilobytes.
@@ -115,6 +118,7 @@ func (c *KstreamConfig) initFromViper(v *viper.Viper) {
 	c.EnableMemKevents = v.GetBool(enableMemKevents)
 	c.EnableAuditAPIEvents = v.GetBool(enableAuditAPIEvents)
 	c.EnableDNSEvents = v.GetBool(enableDNSEvents)
+	c.EnableThreadpoolEvents = v.GetBool(enableThreadpoolEvents)
 	c.StackEnrichment = v.GetBool(stackEnrichment)
 	c.BufferSize = uint32(v.GetInt(bufferSize))
 	c.MinBuffers = uint32(v.GetInt(minBuffers))

pkg/config/schema_windows.go

@@ -179,25 +179,26 @@ var schema = `
 		"kstream": {
 			"type": "object",
 			"properties": {
-				"enable-thread": 	{"type": "boolean"},
-				"enable-image": 	{"type": "boolean"},
-				"enable-registry": 	{"type": "boolean"},
-				"enable-fileio": 	{"type": "boolean"},
-				"enable-vamap":		{"type": "boolean"},
-				"enable-handle": 	{"type": "boolean"},
-				"enable-net": 		{"type": "boolean"},
-				"enable-mem": 		{"type": "boolean"},
-				"enable-audit-api": {"type": "boolean"},
-				"enable-dns": 		{"type": "boolean"},
-				"stack-enrichment": {"type": "boolean"},
-				"min-buffers": 		{"type": "integer", "minimum": 1, "maximum": {{ .MinBuffers }}},
-				"max-buffers": 		{"type": "integer", "minimum": 2, "maximum": {{ .MaxBuffers }}},
-				"buffer-size":		{"type": "integer", "maximum": {{ .MaxBufferSize }}},
+				"enable-thread":			{"type": "boolean"},
+				"enable-image":				{"type": "boolean"},
+				"enable-registry":		{"type": "boolean"},
+				"enable-fileio":			{"type": "boolean"},
+				"enable-vamap":				{"type": "boolean"},
+				"enable-handle": 			{"type": "boolean"},
+				"enable-net": 				{"type": "boolean"},
+				"enable-mem": 				{"type": "boolean"},
+				"enable-audit-api": 	{"type": "boolean"},
+				"enable-dns": 				{"type": "boolean"},
+				"enable-threadpool":	{"type": "boolean"},
+				"stack-enrichment":		{"type": "boolean"},
+				"min-buffers":				{"type": "integer", "minimum": 1, "maximum": {{ .MinBuffers }}},
+				"max-buffers":				{"type": "integer", "minimum": 2, "maximum": {{ .MaxBuffers }}},
+				"buffer-size":				{"type": "integer", "maximum": {{ .MaxBufferSize }}},
                 "flush-interval":	{"type": "string", "minLength": 2, "pattern": "[0-9]+s"},
 				"blacklist":		{
 					"type": "object",
 					"properties":	{
-						"events":	{"type": "array", "items": {"type": "string", "enum": ["CreateThread", "TerminateThread", "OpenProcess", "OpenThread", "SetThreadContext", "LoadImage", "UnloadImage", "CreateFile", "CloseFile", "ReadFile", "WriteFile", "DeleteFile", "RenameFile", "SetFileInformation", "EnumDirectory", "MapViewFile", "UnmapViewFile", "RegCreateKey", "RegOpenKey", "RegSetValue", "RegQueryValue", "RegQueryKey", "RegDeleteKey", "RegDeleteValue", "RegCloseKey", "Accept", "Send", "Recv", "Connect", "Disconnect", "Reconnect", "Retransmit", "CreateHandle", "CloseHandle", "DuplicateHandle", "QueryDns", "ReplyDns", "VirtualAlloc", "VirtualFree", "CreateSymbolicLinkObject"]}},
+						"events":	{"type": "array", "items": {"type": "string", "enum": ["CreateThread", "TerminateThread", "OpenProcess", "OpenThread", "SetThreadContext", "LoadImage", "UnloadImage", "CreateFile", "CloseFile", "ReadFile", "WriteFile", "DeleteFile", "RenameFile", "SetFileInformation", "EnumDirectory", "MapViewFile", "UnmapViewFile", "RegCreateKey", "RegOpenKey", "RegSetValue", "RegQueryValue", "RegQueryKey", "RegDeleteKey", "RegDeleteValue", "RegCloseKey", "Accept", "Send", "Recv", "Connect", "Disconnect", "Reconnect", "Retransmit", "CreateHandle", "CloseHandle", "DuplicateHandle", "QueryDns", "ReplyDns", "VirtualAlloc", "VirtualFree", "CreateSymbolicLinkObject", "SubmitThreadpoolWork", "SubmitThreadpoolCallback", "SetThreadpoolTimer"]}},
 						"images":	{"type": "array", "items": {"type": "string", "minLength": 1}}
 					},
 					"additionalProperties": false

pkg/filter/rules.go

@@ -647,6 +647,8 @@ func (r *Rules) buildCompileResult() *config.RulesCompileResult {
 								rs.HasMemEvents = true
 							case ktypes.Handle:
 								rs.HasHandleEvents = true
+							case ktypes.Threadpool:
+								rs.HasThreadpoolEvents = true
 							}
 							if typ == ktypes.MapViewFile || typ == ktypes.UnmapViewFile {
 								rs.HasVAMapEvents = true

pkg/kevent/kevent_windows.go

@@ -556,6 +556,16 @@ func (e *Kevent) Summary() string {
 	case ktypes.ReplyDNS:
 		dnsName := e.GetParamAsString(kparams.DNSName)
 		return printSummary(e, fmt.Sprintf("received DNS response for <code>%s</code> query", dnsName))
+	case ktypes.CreateSymbolicLinkObject:
+		src := e.GetParamAsString(kparams.LinkSource)
+		target := e.GetParamAsString(kparams.LinkTarget)
+		return printSummary(e, fmt.Sprintf("created symbolic link from %s to %s", src, target))
+	case ktypes.SubmitThreadpoolWork:
+		return printSummary(e, "enqueued the work item to the thread pool")
+	case ktypes.SubmitThreadpoolCallback:
+		return printSummary(e, "Submitted the thread pool callback for execution within the work item")
+	case ktypes.SetThreadpoolTimer:
+		return printSummary(e, "set thread pool timer object")
 	}
 	return ""
 }