feat(output): New http output (#110)

* base HTTP output structure

* more work around HTTP output

* HTTP output tests, registration and polishing

* HTTP output config and validation schema

Author: rabbitstack

cmd/fibratus/app/control_service_windows.go

@@ -20,20 +20,22 @@ package app
 
 import (
 	"fmt"
+	"time"
+
 	"github.com/rabbitstack/fibratus/cmd/fibratus/common"
 	"github.com/rabbitstack/fibratus/pkg/aggregator"
 	"github.com/rabbitstack/fibratus/pkg/api"
 	"github.com/rabbitstack/fibratus/pkg/config"
 	"github.com/rabbitstack/fibratus/pkg/handle"
 	"github.com/rabbitstack/fibratus/pkg/kstream"
 	"github.com/rabbitstack/fibratus/pkg/ps"
+	ver "github.com/rabbitstack/fibratus/pkg/util/version"
 	"github.com/spf13/cobra"
 	"golang.org/x/sys/windows"
 	"golang.org/x/sys/windows/svc"
 	"golang.org/x/sys/windows/svc/debug"
 	"golang.org/x/sys/windows/svc/eventlog"
 	"golang.org/x/sys/windows/svc/mgr"
-	"time"
 )
 
 var startSvcCmd = &cobra.Command{
@@ -216,7 +218,7 @@ func (s *fsvc) run() error {
 	if err := common.Init(svcConfig, true); err != nil {
 		return err
 	}
-
+	ver.Set(version)
 	ctrl = kstream.NewKtraceController(svcConfig.Kstream)
 	err := ctrl.StartKtrace()
 	if err != nil {

cmd/fibratus/app/replay_windows.go

@@ -20,13 +20,15 @@ package app
 
 import (
 	"context"
+
 	"github.com/rabbitstack/fibratus/cmd/fibratus/common"
 	"github.com/rabbitstack/fibratus/pkg/aggregator"
 	"github.com/rabbitstack/fibratus/pkg/api"
 	"github.com/rabbitstack/fibratus/pkg/config"
 	"github.com/rabbitstack/fibratus/pkg/filament"
 	"github.com/rabbitstack/fibratus/pkg/filter"
 	"github.com/rabbitstack/fibratus/pkg/kcap"
+	ver "github.com/rabbitstack/fibratus/pkg/util/version"
 	"github.com/spf13/cobra"
 )
 
@@ -50,7 +52,7 @@ func replay(cmd *cobra.Command, args []string) error {
 	if err := common.Init(replayConfig, false); err != nil {
 		return err
 	}
-
+	ver.Set(version)
 	// set up the signals
 	stopCh := common.Signals()
 

cmd/fibratus/app/run_windows.go

@@ -21,6 +21,8 @@ package app
 import (
 	"os"
 
+	ver "github.com/rabbitstack/fibratus/pkg/util/version"
+
 	"github.com/rabbitstack/fibratus/cmd/fibratus/common"
 	"github.com/rabbitstack/fibratus/pkg/aggregator"
 	"github.com/rabbitstack/fibratus/pkg/alertsender"
@@ -71,7 +73,7 @@ func run(cmd *cobra.Command, args []string) error {
 	if err := common.Init(cfg, true); err != nil {
 		return err
 	}
-
+	ver.Set(version)
 	// set up the signals
 	stopCh := common.Signals()
 

configs/fibratus.yml

@@ -359,6 +359,55 @@ output:
     # Indicates if the chain and host verification stage is skipped
     #tls-insecure-skip-verify: false
 
+  # HTTP output sends event batches to HTTP servers.
+  http:
+    # Indicates if the HTTP output is enabled
+    enabled: false
+
+    # List of endpoints to which the events are sent
+    #endpoints:
+    #  - http://localhost:8081
+
+    # Represents the timeout for the HTTP requests
+    #timeout: 5s
+
+    # Specifies the HTTP proxy URL. It overrides the HTTP proxy URL as indicated by the environment variables
+    #proxy-url: ""
+
+    # The username for HTTP proxy authentication
+    #proxy-username: ""
+
+    # The password for HTTP proxy authentication
+    #proxy-password: ""
+
+    # Determines the HTTP verb to use in requests
+    #method: POST
+
+    # Username for the basic HTTP authentication
+    #username: ""
+
+    # Password for the basic HTTP authentication
+    #password: ""
+
+    # If enabled, the HTTP body is compressed with gzip compression
+    #enable-gzip: false
+
+    # List of arbitrary headers to include in HTTP requests
+    #headers:
+    #  api-key: ""
+
+    # Path to the public/private key file
+    #tls-key:
+
+    # Path to certificate file
+    #tls-cert:
+
+    # Represents the path of the certificate file that is associated with the Certification Authority (CA)
+    #tls-ca:
+
+    # Indicates if the chain and host verification stage is skipped
+    #tls-insecure-skip-verify: false
+
 # =============================== Portable Executable (PE) =============================
 
 # Tweaks for controlling the fetching of the PE (Portable Executable) metadata from the process' binary image.

pkg/aggregator/aggregator.go

@@ -21,20 +21,25 @@ package aggregator
 import (
 	"errors"
 	"expvar"
+	"time"
+
 	"github.com/rabbitstack/fibratus/pkg/aggregator/transformers"
 	"github.com/rabbitstack/fibratus/pkg/alertsender"
 	"github.com/rabbitstack/fibratus/pkg/kevent"
 	"github.com/rabbitstack/fibratus/pkg/outputs"
 	log "github.com/sirupsen/logrus"
-	"time"
+
 	// initialize outputs
 	_ "github.com/rabbitstack/fibratus/pkg/outputs/amqp"
 	_ "github.com/rabbitstack/fibratus/pkg/outputs/console"
 	_ "github.com/rabbitstack/fibratus/pkg/outputs/elasticsearch"
+	_ "github.com/rabbitstack/fibratus/pkg/outputs/http"
 	_ "github.com/rabbitstack/fibratus/pkg/outputs/null"
+
 	// initialize alert senders
 	_ "github.com/rabbitstack/fibratus/pkg/alertsender/mail"
 	_ "github.com/rabbitstack/fibratus/pkg/alertsender/slack"
+
 	// initialize transformers
 	_ "github.com/rabbitstack/fibratus/pkg/aggregator/transformers/remove"
 	_ "github.com/rabbitstack/fibratus/pkg/aggregator/transformers/rename"

pkg/config/_fixtures/fibratus.yml

@@ -162,6 +162,12 @@ output:
     exchange-type: topic
     routing-key: fibratus
     vhost: /
+  http:
+    enabled: false
+    endpoints:
+      - http://localhost:8081
+    timeout: 5s
+    method: POST
 
 # =============================== Portable Executable (PE) =============================
 

pkg/config/_fixtures/http-output.yml

@@ -0,0 +1,30 @@
+kstream:
+  max-buffers: 10
+  min-buffers: 8
+  flush-interval: 1s
+  blacklist:
+    events:
+      - CreateThread
+
+filament: top_hives_io
+
+output:
+  console:
+    enabled: false
+    format: pretty
+  elasticsearch:
+  http:
+    enabled: true
+    endpoints:
+      - http://localhost:8081
+      - http://localhost:8082
+    timeout: 2s
+    proxy-url: http://192.168.1.8:3123
+    proxy-username: bunny
+    proxy-password: bunny
+    username: basic
+    password: basic
+    enable-gzip: true
+    headers:
+      api-Key: kkvvkk
+      header2: value2

pkg/config/config_windows.go

@@ -24,6 +24,8 @@ import (
 	"io/ioutil"
 	"time"
 
+	"github.com/rabbitstack/fibratus/pkg/outputs/http"
+
 	"github.com/rabbitstack/fibratus/pkg/aggregator"
 	"github.com/rabbitstack/fibratus/pkg/aggregator/transformers"
 	removet "github.com/rabbitstack/fibratus/pkg/aggregator/transformers/remove"
@@ -185,6 +187,7 @@ func NewWithOpts(options ...Option) *Config {
 		console.AddFlags(flagSet)
 		amqp.AddFlags(flagSet)
 		elasticsearch.AddFlags(flagSet)
+		http.AddFlags(flagSet)
 		removet.AddFlags(flagSet)
 		replacet.AddFlags(flagSet)
 		renamet.AddFlags(flagSet)

pkg/config/output.go

@@ -21,15 +21,17 @@ package config
 import (
 	"errors"
 	"fmt"
+	"reflect"
+	"strconv"
+
 	"github.com/rabbitstack/fibratus/pkg/outputs"
 	"github.com/rabbitstack/fibratus/pkg/outputs/amqp"
 	"github.com/rabbitstack/fibratus/pkg/outputs/console"
 	"github.com/rabbitstack/fibratus/pkg/outputs/elasticsearch"
+	"github.com/rabbitstack/fibratus/pkg/outputs/http"
 	"github.com/rabbitstack/fibratus/pkg/outputs/null"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/sys/windows/svc"
-	"reflect"
-	"strconv"
 )
 
 var errNoOutputSection = errors.New("no output section in config")
@@ -71,8 +73,8 @@ func (c *Config) tryLoadOutput() error {
 	}
 
 	for typ, config := range mapping {
-		switch typ {
-		case "console":
+		switch outputs.TypeFromString(typ) {
+		case outputs.Console:
 			var consoleConfig console.Config
 			if err := decode(config, &consoleConfig); err != nil {
 				return errOutputConfig(typ, err)
@@ -82,7 +84,7 @@ func (c *Config) tryLoadOutput() error {
 			}
 			c.Output.Type, c.Output.Output = outputs.Console, consoleConfig
 
-		case "amqp":
+		case outputs.AMQP:
 			var amqpConfig amqp.Config
 			if err := decode(config, &amqpConfig); err != nil {
 				return errOutputConfig(typ, err)
@@ -92,7 +94,7 @@ func (c *Config) tryLoadOutput() error {
 			}
 			c.Output.Type, c.Output.Output = outputs.AMQP, amqpConfig
 
-		case "elasticsearch":
+		case outputs.Elasticsearch:
 			var esConfig elasticsearch.Config
 			if err := decode(config, &esConfig); err != nil {
 				return errOutputConfig(typ, err)
@@ -101,13 +103,22 @@ func (c *Config) tryLoadOutput() error {
 				continue
 			}
 			c.Output.Type, c.Output.Output = outputs.Elasticsearch, esConfig
+
+		case outputs.HTTP:
+			var httpConfig http.Config
+			if err := decode(config, &httpConfig); err != nil {
+				return errOutputConfig(typ, err)
+			}
+			if !httpConfig.Enabled {
+				continue
+			}
+			c.Output.Type, c.Output.Output = outputs.HTTP, httpConfig
 		}
 	}
 
 	// if it is not an interactive session but the console output is enabled
 	// we default to null output and warn about that
-	in, err := svc.IsAnInteractiveSession()
-	if err == nil && !in && c.Output.Output != nil {
+	if isWindowsService() && c.Output.Output != nil {
 		if c.Output.Type == outputs.Console {
 			log.Warn("running in non-interactive session with console output. " +
 				"Please configure a different output type. Defaulting to null output")
@@ -135,3 +146,12 @@ func findActiveOutputs(outputs map[string]interface{}) []string {
 	}
 	return outputTypes
 }
+
+// isWindowsService returns true if the process is running inside Windows Service.
+func isWindowsService() bool {
+	interactive, err := svc.IsAnInteractiveSession()
+	if err != nil {
+		return false
+	}
+	return !interactive
+}

pkg/config/output_test.go

@@ -19,14 +19,16 @@
 package config
 
 import (
+	"testing"
+	"time"
+
 	"github.com/rabbitstack/fibratus/pkg/outputs/amqp"
+	"github.com/rabbitstack/fibratus/pkg/outputs/http"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"testing"
-	"time"
 )
 
-func TestOutput(t *testing.T) {
+func TestAMQPOutput(t *testing.T) {
 	c := NewWithOpts(WithRun())
 
 	err := c.flags.Parse([]string{"--config-file=_fixtures/output.yml"})
@@ -47,3 +49,31 @@ func TestOutput(t *testing.T) {
 	assert.Equal(t, "fibratus", amqpConfig.RoutingKey)
 	assert.Equal(t, "/", amqpConfig.Vhost)
 }
+
+func TestHTTPOutput(t *testing.T) {
+	c := NewWithOpts(WithRun())
+
+	err := c.flags.Parse([]string{"--config-file=_fixtures/http-output.yml"})
+	require.NoError(t, c.viper.BindPFlags(c.flags))
+	require.NoError(t, err)
+	require.NoError(t, c.TryLoadFile(c.GetConfigFile()))
+
+	require.NoError(t, c.Init())
+
+	require.NotNil(t, c.Output)
+	require.IsType(t, http.Config{}, c.Output.Output)
+
+	httpConfig := c.Output.Output.(http.Config)
+	assert.True(t, httpConfig.Enabled)
+	assert.Len(t, httpConfig.Endpoints, 2)
+	assert.Contains(t, httpConfig.Endpoints, "http://localhost:8081")
+	assert.Equal(t, time.Second*2, httpConfig.Timeout)
+	assert.Equal(t, "http://192.168.1.8:3123", httpConfig.ProxyURL)
+	assert.Equal(t, "bunny", httpConfig.ProxyUsername)
+	assert.Equal(t, "bunny", httpConfig.ProxyPassword)
+	assert.True(t, httpConfig.EnableGzip)
+	assert.Equal(t, "basic", httpConfig.Username)
+	assert.Equal(t, "basic", httpConfig.Password)
+	assert.Len(t, httpConfig.Headers, 2)
+	assert.Equal(t, "kkvvkk", httpConfig.Headers["api-key"])
+}