Make health server optional

Related: #943

Author: jotak

README.md

@@ -52,7 +52,7 @@ Flags:
       --config string              config file (default is $HOME/.flowlogs-pipeline)  
       --dynamicParameters string   json of configmap location for dynamic parameters  
       --health.address string      Health server address (default "0.0.0.0")  
-      --health.port string         Health server port (default "8080")  
+      --health.port int            Health server port (default: disable health server)   
   -h, --help                       help for flowlogs-pipeline  
       --log-level string           Log level: debug, info, warning, error (default "error")  
       --metricsSettings string     json for global metrics settings  
@@ -955,6 +955,7 @@ Images
   image-push            Push MULTIARCH_TARGETS images  
   manifest-build        Build MULTIARCH_TARGETS manifest  
   manifest-push         Push MULTIARCH_TARGETS manifest  
+  extract-binaries      Extract all MULTIARCH_TARGETS binaries  
   goyacc                Regenerate filters query langage  
   
 kubernetes  

cmd/flowlogs-pipeline/main.go

@@ -143,7 +143,7 @@ func initFlags() {
 	rootCmd.PersistentFlags().StringVar(&cfgFile, "config", "", fmt.Sprintf("config file (default is $HOME/%s)", defaultLogFileName))
 	rootCmd.PersistentFlags().StringVar(&logLevel, "log-level", "error", "Log level: debug, info, warning, error")
 	rootCmd.PersistentFlags().StringVar(&opts.Health.Address, "health.address", "0.0.0.0", "Health server address")
-	rootCmd.PersistentFlags().StringVar(&opts.Health.Port, "health.port", "8080", "Health server port")
+	rootCmd.PersistentFlags().IntVar(&opts.Health.Port, "health.port", 0, "Health server port (default: disable health server) ")
 	rootCmd.PersistentFlags().IntVar(&opts.Profile.Port, "profile.port", 0, "Go pprof tool port (default: disabled)")
 	rootCmd.PersistentFlags().StringVar(&opts.PipeLine, "pipeline", "", "json of config file pipeline field")
 	rootCmd.PersistentFlags().StringVar(&opts.Parameters, "parameters", "", "json of config file parameters field")
@@ -199,15 +199,20 @@ func run() {
 	}
 
 	// Start health report server
-	healthServer := operational.NewHealthServer(&opts, mainPipeline.IsAlive, mainPipeline.IsReady)
+	var healthServer *http.Server
+	if opts.Health.Port > 0 {
+		healthServer = operational.NewHealthServer(&opts, mainPipeline.IsAlive, mainPipeline.IsReady)
+	}
 
 	// Starts the flows pipeline
 	mainPipeline.Run()
 
 	if promServer != nil {
 		_ = promServer.Shutdown(context.Background())
 	}
-	_ = healthServer.Shutdown(context.Background())
+	if healthServer != nil {
+		_ = healthServer.Shutdown(context.Background())
+	}
 
 	// Give all threads a chance to exit and then exit the process
 	time.Sleep(time.Second)

docs/api.md

@@ -229,6 +229,7 @@ Following is the supported API format for filter transformations:
                                      input: entry input field
                                      value: specified value of input field:
                                      castInt: set true to cast the value field as an int (numeric values are float64 otherwise)
+         samplingField: sampling field name to be set when sampling is used; if the field already exists in flows, its value is multiplied with the new sampling
 </pre>
 ## Transform Network API
 Following is the supported API format for network transformations:

hack/examples/docker-ipfix-config.yaml

@@ -1,4 +1,6 @@
 log-level: info
+health:
+  port: 8080
 pipeline:
   - name: ingest
   - name: conntrack

pkg/config/config.go

@@ -60,7 +60,7 @@ type HotReloadStruct struct {
 
 type Health struct {
 	Address string
-	Port    string
+	Port    int
 }
 
 type Profile struct {

pkg/operational/health.go

@@ -18,7 +18,7 @@
 package operational
 
 import (
-	"net"
+	"fmt"
 	"net/http"
 	"time"
 
@@ -30,7 +30,7 @@ import (
 
 func NewHealthServer(opts *config.Options, isAlive healthcheck.Check, isReady healthcheck.Check) *http.Server {
 	handler := healthcheck.NewHandler()
-	address := net.JoinHostPort(opts.Health.Address, opts.Health.Port)
+	address := fmt.Sprintf("%s:%d", opts.Health.Address, opts.Health.Port)
 	handler.AddLivenessCheck("PipelineCheck", isAlive)
 	handler.AddReadinessCheck("PipelineCheck", isReady)
 

pkg/pipeline/health_test.go

@@ -35,7 +35,7 @@ func TestNewHealthServer(t *testing.T) {
 
 	type args struct {
 		pipeline Pipeline
-		port     string
+		port     int
 		address  string
 	}
 	type want struct {
@@ -47,15 +47,15 @@ func TestNewHealthServer(t *testing.T) {
 		args args
 		want want
 	}{
-		{name: "pipeline running", args: args{pipeline: Pipeline{IsRunning: true}, port: "7000", address: "0.0.0.0"}, want: want{statusCode: 200}},
-		{name: "pipeline not running", args: args{pipeline: Pipeline{IsRunning: false}, port: "7001", address: "0.0.0.0"}, want: want{statusCode: 503}},
+		{name: "pipeline running", args: args{pipeline: Pipeline{IsRunning: true}, port: 7000, address: "0.0.0.0"}, want: want{statusCode: 200}},
+		{name: "pipeline not running", args: args{pipeline: Pipeline{IsRunning: false}, port: 7001, address: "0.0.0.0"}, want: want{statusCode: 503}},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 
 			opts := config.Options{Health: config.Health{Port: tt.args.port, Address: tt.args.address}}
-			expectedAddr := fmt.Sprintf("%s:%s", opts.Health.Address, opts.Health.Port)
+			expectedAddr := fmt.Sprintf("%s:%d", opts.Health.Address, opts.Health.Port)
 			server := operational.NewHealthServer(&opts, tt.args.pipeline.IsAlive, tt.args.pipeline.IsReady)
 			require.NotNil(t, server)
 			require.Equal(t, expectedAddr, server.Addr)