rabbitstack
diff --git a/‎internal/bootstrap/bootstrap.go
Copy file name to clipboard
+37-61Lines changed: 37 additions & 61 deletions b/‎internal/bootstrap/bootstrap.go
Copy file name to clipboard
+37-61Lines changed: 37 additions & 61 deletions
diff --git a/‎internal/bootstrap/source.go
Copy file name to clipboard
+74Lines changed: 74 additions & 0 deletions b/‎internal/bootstrap/source.go
Copy file name to clipboard
+74Lines changed: 74 additions & 0 deletions
@@ -29,7 +29,6 @@ import (
 	"github.com/rabbitstack/fibratus/pkg/filter"
 	"github.com/rabbitstack/fibratus/pkg/handle"
 	"github.com/rabbitstack/fibratus/pkg/kcap"
-	"github.com/rabbitstack/fibratus/pkg/kstream"
 	"github.com/rabbitstack/fibratus/pkg/ps"
 	"github.com/rabbitstack/fibratus/pkg/symbolize"
 	"github.com/rabbitstack/fibratus/pkg/sys"
@@ -46,16 +45,16 @@ import (
 var ErrAlreadyRunning = errors.New("an instance of Fibratus process is already running in the system")
 
 // App centralizes the core building blocks responsible
-// for event acquisition, captures handling, filament
-// execution and event routing to the output sinks.
+// for event acquisition, rule engine initialization,
+// captures handling, filament execution and event routing
+// to the output sinks.
 type App struct {
 	config     *config.Config
-	controller *kstream.Controller
+	evs        *EventSourceControl
 	symbolizer *symbolize.Symbolizer
 	rules      *filter.Rules
 	hsnap      handle.Snapshotter
 	psnap      ps.Snapshotter
-	consumer   kstream.Consumer
 	filament   filament.Filament
 	agg        *aggregator.BufferedAggregator
 	writer     kcap.Writer
@@ -153,31 +152,26 @@ func NewApp(cfg *config.Config, options ...Option) (*App, error) {
 		log.Info("rule engine is disabled")
 	}
 
-	controller := kstream.NewController(cfg, res)
+	evs := NewEventSourceControl(psnap, hsnap, cfg, res)
 
 	app := &App{
-		config:     cfg,
-		controller: controller,
-		rules:      rules,
-		hsnap:      hsnap,
-		psnap:      psnap,
-		consumer:   kstream.NewConsumer(controller, psnap, hsnap, cfg),
-		signals:    sigs,
+		config:  cfg,
+		evs:     evs,
+		rules:   rules,
+		hsnap:   hsnap,
+		psnap:   psnap,
+		signals: sigs,
 	}
 	return app, nil
 }
 
-// Run starts configured tracing sessions and opens the event
-// stream to start consuming events. Depending on whether the
-// filament is provided, this method will either spin up a filament
-// or set up the aggregator to start forwarding events to the rule
-// engine and output sinks.
+// Run configure and opens the event source to start consuming events.
+// Depending on whether the filament is provided, this method will either
+// spin up a filament or set up the aggregator to start forwarding events
+// to the rule engine and output sinks.
 func (f *App) Run(args []string) error {
-	if f.controller == nil {
-		panic("controller is nil")
-	}
-	if f.consumer == nil {
-		panic("consumer is nil")
+	if f.evs == nil {
+		panic("event source is nil")
 	}
 	cfg := f.config
 
@@ -188,11 +182,6 @@ func (f *App) Run(args []string) error {
 	log.Infof("bootstrapping with pid %d. Version: %s", os.Getpid(), version.Get())
 	log.Infof("configuration dump %s", cfg.Print())
 
-	err := f.controller.Start()
-	if err != nil {
-		return err
-	}
-
 	// build the filter from the CLI argument. If we got
 	// a valid expression the filter is attached to the
 	// event consumer
@@ -201,14 +190,14 @@ func (f *App) Run(args []string) error {
 		return err
 	}
 	if kfilter != nil {
-		f.consumer.SetFilter(kfilter)
+		f.evs.SetFilter(kfilter)
 	}
 	// user can either instruct to bootstrap a filament or
 	// start a regular run. We'll set up the corresponding
 	// components accordingly to what we got from the CLI options.
 	// If a filament was given, we'll assign it the previous filter
 	// if it wasn't provided in the filament init function.
-	// Finally, we open the kernel stream flow and run the filament i.e.
+	// Finally, we open the event source and run the filament i.e.
 	// Python main thread in a new goroutine.
 	// In case of a regular run, we additionally set up the aggregator.
 	// The aggregator will grab the events from the queue, assemble them
@@ -220,19 +209,19 @@ func (f *App) Run(args []string) error {
 			return err
 		}
 		if f.filament.Filter() != nil {
-			f.consumer.SetFilter(f.filament.Filter())
+			f.evs.SetFilter(f.filament.Filter())
 		}
-		err = f.consumer.Open()
+		err = f.evs.Open(cfg)
 		if err != nil {
-			return multierror.Wrap(err, f.controller.Close())
+			return multierror.Wrap(err, f.evs.Close())
 		}
 		// load alert senders so emitting alerts is possible from filaments
 		err = alertsender.LoadAll(cfg.Alertsenders)
 		if err != nil {
 			log.Warnf("couldn't load alertsenders: %v", err)
 		}
 		go func() {
-			err = f.filament.Run(f.consumer.Events(), f.consumer.Errors())
+			err = f.filament.Run(f.evs.Events(), f.evs.Errors())
 			if err != nil {
 				log.Errorf("filament failed: %v", err)
 				f.stop()
@@ -242,29 +231,28 @@ func (f *App) Run(args []string) error {
 		// register stack symbolizer
 		if cfg.Kstream.StackEnrichment {
 			f.symbolizer = symbolize.NewSymbolizer(symbolize.NewDebugHelpResolver(cfg), f.psnap, cfg, false)
-			f.consumer.RegisterEventListener(f.symbolizer)
+			f.evs.RegisterEventListener(f.symbolizer)
 		}
 		// register rule engine
 		if f.rules != nil {
-			f.consumer.RegisterEventListener(f.rules)
+			f.evs.RegisterEventListener(f.rules)
 		}
 		// register YARA scanner
 		if cfg.Yara.Enabled {
 			scanner, err := yara.NewScanner(f.psnap, cfg.Yara)
 			if err != nil {
 				return err
 			}
-			f.consumer.RegisterEventListener(scanner)
+			f.evs.RegisterEventListener(scanner)
 		}
-		err = f.consumer.Open()
+		err = f.evs.Open(cfg)
 		if err != nil {
-			return multierror.Wrap(err, f.controller.Close())
+			return multierror.Wrap(err, f.evs.Close())
 		}
 		// set up the aggregator that forwards events to outputs
-		// and calls event listeners such as the rule engine
 		f.agg, err = aggregator.NewBuffered(
-			f.consumer.Events(),
-			f.consumer.Errors(),
+			f.evs.Events(),
+			f.evs.Errors(),
 			cfg.Aggregator,
 			cfg.Output,
 			cfg.Transformers,
@@ -280,37 +268,30 @@ func (f *App) Run(args []string) error {
 
 // WriteCapture writes the event stream to the capture file.
 func (f *App) WriteCapture(args []string) error {
-	if f.controller == nil {
-		panic("controller is nil")
-	}
-	if f.consumer == nil {
-		panic("consumer is nil")
+	if f.evs == nil {
+		panic("event source is nil")
 	}
 
 	if !f.isSingleInstance() {
 		return ErrAlreadyRunning
 	}
 
-	err := f.controller.Start()
-	if err != nil {
-		return err
-	}
 	kfilter, err := filter.NewFromCLI(args, f.config)
 	if err != nil {
 		return err
 	}
 	if kfilter != nil {
-		f.consumer.SetFilter(kfilter)
+		f.evs.SetFilter(kfilter)
 	}
-	err = f.consumer.Open()
+	err = f.evs.Open(f.config)
 	if err != nil {
 		return err
 	}
 	f.writer, err = kcap.NewWriter(f.config.KcapFile, f.psnap, f.hsnap)
 	if err != nil {
 		return err
 	}
-	errsChan := f.writer.Write(f.consumer.Events(), f.consumer.Errors())
+	errsChan := f.writer.Write(f.evs.Events(), f.evs.Errors())
 	go func() {
 		for err := range errsChan {
 			log.Warnf("fail to write event to capture: %v", err)
@@ -386,16 +367,11 @@ func (f *App) Wait() {
 // Shutdown is responsible for tearing down everything gracefully.
 func (f *App) Shutdown() error {
 	errs := make([]error, 0)
-	if f.controller != nil {
-		if err := f.controller.Close(); err != nil {
-			errs = append(errs, err)
-		}
-	}
 	if f.symbolizer != nil {
 		f.symbolizer.Close()
 	}
-	if f.consumer != nil {
-		if err := f.consumer.Close(); err != nil {
+	if f.evs != nil {
+		if err := f.evs.Close(); err != nil {
 			errs = append(errs, err)
 		}
 	}
 
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2021-2022 by Nedim Sabic Sabic
+ * https://www.fibratus.io
+ * All Rights Reserved.
+ *
+ * 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
+ *
+ *  https://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.
+ */
+
+package bootstrap
+
+import (
+	"github.com/rabbitstack/fibratus/internal/etw"
+	"github.com/rabbitstack/fibratus/pkg/config"
+	"github.com/rabbitstack/fibratus/pkg/filter"
+	"github.com/rabbitstack/fibratus/pkg/handle"
+	"github.com/rabbitstack/fibratus/pkg/kevent"
+	"github.com/rabbitstack/fibratus/pkg/ksource"
+	"github.com/rabbitstack/fibratus/pkg/ps"
+)
+
+// EventSourceControl abstracts away the management of event sources.
+// Presently, system events are captured by ETW infra, in the future
+// additional instrumentation engines can be introduced and the event
+// source will automatically decide which is the best engine to operate
+// with. As an example, eBPF instrumentation may gain traction in the
+// future, and the systems that support eBPF can provide a richer spectrum
+// of telemetry than the ETW subsystem. In this scenario, the event source
+// control will bootstrap the instrumentation engine based on eBPF.
+type EventSourceControl struct {
+	evs ksource.EventSource
+}
+
+func NewEventSourceControl(
+	psnap ps.Snapshotter,
+	hsnap handle.Snapshotter,
+	config *config.Config,
+	compiler *config.RulesCompileResult,
+) *EventSourceControl {
+	return &EventSourceControl{evs: etw.NewEventSource(psnap, hsnap, config, compiler)}
+}
+
+func (s *EventSourceControl) Open(config *config.Config) error {
+	return s.evs.Open(config)
+}
+
+func (s *EventSourceControl) Close() error {
+	return s.evs.Close()
+}
+
+func (s *EventSourceControl) Errors() <-chan error {
+	return s.evs.Errors()
+}
+
+func (s *EventSourceControl) Events() <-chan *kevent.Kevent {
+	return s.evs.Events()
+}
+
+func (s *EventSourceControl) SetFilter(f filter.Filter) {
+	s.evs.SetFilter(f)
+}
+
+func (s *EventSourceControl) RegisterEventListener(lis kevent.Listener) {
+	s.evs.RegisterEventListener(lis)
+}