Make health server optional (#966)

* Make health server optional

Related: #943

* fix test

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  

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)

cmd/flowlogs-pipeline/main_test.go

@@ -54,7 +54,7 @@ func TestPipelineConfigSetup(t *testing.T) {
     "PipeLine": "[{\"name\":\"grpc\"},{\"follows\":\"grpc\",\"name\":\"enrich\"},{\"follows\":\"enrich\",\"name\":\"loki\"},{\"follows\":\"enrich\",\"name\":\"prometheus\"}]",
     "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"kubernetes\":{\"ipField\":\"SrcAddr\",\"output\":\"SrcK8S\"},\"type\":\"add_kubernetes\"},{\"kubernetes\":{\"ipField\":\"DstAddr\",\"output\":\"DstK8S\"},\"type\":\"add_kubernetes\"},{\"add_service\":{\"input\":\"DstPort\",\"output\":\"Service\",\"protocol\":\"Proto\"},\"type\":\"add_service\"},{\"add_subnet\":{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"subnet_mask\":\"/16\"},\"type\":\"add_subnet\"}]},\"type\":\"network\"}},{\"name\":\"loki\",\"write\":{\"loki\":{\"batchSize\":102400,\"batchWait\":\"1s\",\"clientConfig\":{\"follow_redirects\":false,\"proxy_url\":null,\"tls_config\":{\"insecure_skip_verify\":false}},\"labels\":[\"SrcK8S_Namespace\",\"SrcK8S_OwnerName\",\"DstK8S_Namespace\",\"DstK8S_OwnerName\",\"FlowDirection\"],\"maxBackoff\":\"5m0s\",\"maxRetries\":10,\"minBackoff\":\"1s\",\"staticLabels\":{\"app\":\"netobserv-flowcollector\"},\"tenantID\":\"netobserv\",\"timeout\":\"10s\",\"timestampLabel\":\"TimeFlowEndMs\",\"timestampScale\":\"1ms\",\"url\":\"http://loki.netobserv.svc:3100/\"},\"type\":\"loki\"}},{\"encode\":{\"prom\":{\"metrics\":[{\"buckets\":null,\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
     "Health": {
-        "Port": "8080"
+        "Port": 8080
     },
     "Profile": {
         "Port": 0

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)