Move prometheus server to main (#399)

KalmanMeth · web-flow · commit 04f8a14d4ccf · 2023-03-08T09:41:44.000+02:00
* move prom server

* moved prometheus server creation to main

* removed prometheus port from tests

* explicitly ignored return value of Shutdown

* allow multiple instances of encode prom by ignoring duplicated metric name

* fix handling of port in confgenerator

* addressed review comments
diff --git a/README.md b/README.md
@@ -49,15 +49,15 @@ Usage:
   flowlogs-pipeline [flags]  
   
 Flags:  
-      --config string             config file (default is $HOME/.flowlogs-pipeline)  
-      --health.address string     Health server address (default "0.0.0.0")  
-      --health.port string        Health server port (default "8080")  
-  -h, --help                      help for flowlogs-pipeline  
-      --log-level string          Log level: debug, info, warning, error (default "error")  
-      --metrics-settings string   json for global metrics settings  
-      --parameters string         json of config file parameters field  
-      --pipeline string           json of config file pipeline field  
-      --profile.port int          Go pprof tool port (default: disabled)
+      --config string            config file (default is $HOME/.flowlogs-pipeline)  
+      --health.address string    Health server address (default "0.0.0.0")  
+      --health.port string       Health server port (default "8080")  
+  -h, --help                     help for flowlogs-pipeline  
+      --log-level string         Log level: debug, info, warning, error (default "error")  
+      --metricsSettings string   json for global metrics settings  
+      --parameters string        json of config file parameters field  
+      --pipeline string          json of config file pipeline field  
+      --profile.port int         Go pprof tool port (default: disabled)
 ```
 <!---END-AUTO-flowlogs-pipeline_help--->
 
diff --git a/cmd/flowlogs-pipeline/main.go b/cmd/flowlogs-pipeline/main.go
@@ -18,6 +18,8 @@
 package main
 
 import (
+	"context"
+	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -142,7 +144,7 @@ func initFlags() {
 	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")
-	rootCmd.PersistentFlags().StringVar(&opts.MetricsSettings, "metrics-settings", "", "json for global metrics settings")
+	rootCmd.PersistentFlags().StringVar(&opts.MetricsSettings, "metricsSettings", "", "json for global metrics settings")
 }
 
 func main() {
@@ -177,6 +179,20 @@ func run() {
 	// Setup (threads) exit manager
 	utils.SetupElegantExit()
 
+	// create prometheus server for operational metrics
+	// if value of address is empty, then by default it will take 0.0.0.0
+	addr := fmt.Sprintf("%s:%v", cfg.MetricsSettings.Address, cfg.MetricsSettings.Port)
+	log.Infof("startServer: addr = %s", addr)
+	promServer := &http.Server{
+		Addr: addr,
+		// TLS clients must use TLS 1.2 or higher
+		TLSConfig: &tls.Config{
+			MinVersion: tls.VersionTLS12,
+		},
+	}
+	tlsConfig := cfg.MetricsSettings.TLS
+	go utils.StartPromServer(tlsConfig, promServer, !cfg.MetricsSettings.NoPanic)
+
 	// Create new flows pipeline
 	mainPipeline, err = pipeline.NewPipeline(&cfg)
 	if err != nil {
@@ -198,6 +214,8 @@ func run() {
 	// Starts the flows pipeline
 	mainPipeline.Run()
 
+	_ = promServer.Shutdown(context.Background())
+
 	// Give all threads a chance to exit and then exit the process
 	time.Sleep(time.Second)
 	log.Debugf("exiting main run")
diff --git a/cmd/flowlogs-pipeline/main_test.go b/cmd/flowlogs-pipeline/main_test.go
@@ -52,7 +52,7 @@ func TestPipelineConfigSetup(t *testing.T) {
 
 	js := `{
     "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\":[{\"input\":\"SrcAddr\",\"output\":\"SrcK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstAddr\",\"output\":\"DstK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstPort\",\"output\":\"Service\",\"parameters\":\"Proto\",\"type\":\"add_service\"},{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"parameters\":\"/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,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"port\":9102,\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
+    "Parameters": "[{\"ingest\":{\"grpc\":{\"port\":2055},\"type\":\"grpc\"},\"name\":\"grpc\"},{\"name\":\"enrich\",\"transform\":{\"network\":{\"rules\":[{\"input\":\"SrcAddr\",\"output\":\"SrcK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstAddr\",\"output\":\"DstK8S\",\"type\":\"add_kubernetes\"},{\"input\":\"DstPort\",\"output\":\"Service\",\"parameters\":\"Proto\",\"type\":\"add_service\"},{\"input\":\"SrcAddr\",\"output\":\"SrcSubnet\",\"parameters\":\"/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,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\",\"SrcK8S_Namespace\"],\"name\":\"bandwidth_per_network_service_per_namespace\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"SrcSubnet\"],\"name\":\"bandwidth_per_source_subnet\",\"type\":\"counter\",\"valueKey\":\"Bytes\"},{\"buckets\":null,\"filter\":{\"key\":\"\",\"value\":\"\"},\"labels\":[\"Service\"],\"name\":\"network_service_total\",\"type\":\"counter\",\"valueKey\":\"\"}],\"prefix\":\"netobserv_\"},\"type\":\"prom\"},\"name\":\"prometheus\"}]",
     "Health": {
         "Port": "8080"
     },
diff --git a/docs/api.md b/docs/api.md
@@ -17,13 +17,8 @@ Following is the supported API format for prometheus encode:
                  valueKey: entry key from which to resolve metric value
                  labels: labels to be associated with the metric
                  buckets: histogram buckets
-         address: address to expose "/metrics" endpoint
-         port: port number to expose "/metrics" endpoint
          prefix: prefix added to each metric name
          expiryTime: seconds of no-flow to wait before deleting prometheus data item
-         tls: TLS configuration for the prometheus endpoint
-             certPath: path to the certificate file
-             keyPath: path to the key file
          maxMetrics: maximum number of metrics to report (default: unlimited)
 </pre>
 ## Kafka encode API
diff --git a/network_definitions/config.yaml b/network_definitions/config.yaml
@@ -88,7 +88,6 @@ extract:
 encode:
   prom:
     prefix: flp_
-    port: 9102
 write:
   type: loki
   loki:
diff --git a/pkg/api/encode_prom.go b/pkg/api/encode_prom.go
@@ -24,11 +24,8 @@ type PromTLSConf struct {
 
 type PromEncode struct {
 	Metrics    PromMetricsItems `yaml:"metrics,omitempty" json:"metrics,omitempty" doc:"list of prometheus metric definitions, each includes:"`
-	Address    string           `yaml:"address,omitempty" json:"address,omitempty" doc:"address to expose \"/metrics\" endpoint"`
-	Port       int              `yaml:"port,omitempty" json:"port,omitempty" doc:"port number to expose \"/metrics\" endpoint"`
 	Prefix     string           `yaml:"prefix,omitempty" json:"prefix,omitempty" doc:"prefix added to each metric name"`
 	ExpiryTime int              `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"seconds of no-flow to wait before deleting prometheus data item"`
-	TLS        *PromTLSConf     `yaml:"tls,omitempty" json:"tls,omitempty" doc:"TLS configuration for the prometheus endpoint"`
 	MaxMetrics int              `yaml:"maxMetrics,omitempty" json:"maxMetrics,omitempty" doc:"maximum number of metrics to report (default: unlimited)"`
 }
 
diff --git a/pkg/confgen/confgen_test.go b/pkg/confgen/confgen_test.go
@@ -142,7 +142,6 @@ func Test_RunShortConfGen(t *testing.T) {
 	// Expects prom encode
 	require.Len(t, out.Parameters[3].Encode.Prom.Metrics, 1)
 	require.Equal(t, &api.PromEncode{
-		Port:   9102,
 		Prefix: "flp_",
 		Metrics: api.PromMetricsItems{{
 			Name:     "test_metric",
@@ -222,7 +221,6 @@ func Test_RunConfGenNoAgg(t *testing.T) {
 	// Expects prom encode
 	require.Len(t, out.Parameters[2].Encode.Prom.Metrics, 1)
 	require.Equal(t, &api.PromEncode{
-		Port:   9102,
 		Prefix: "flp_",
 		Metrics: api.PromMetricsItems{{
 			Name:     "test_metric",
@@ -325,7 +323,6 @@ func Test_RunLongConfGen(t *testing.T) {
 	// Expects prom encode; make sure type "histogram" is changed to "agg_histogram"
 	require.Len(t, out.Parameters[4].Encode.Prom.Metrics, 2)
 	require.Equal(t, &api.PromEncode{
-		Port:   9102,
 		Prefix: "flp_",
 		Metrics: api.PromMetricsItems{{
 			Name:     "test_metric",
diff --git a/pkg/confgen/config_test.go b/pkg/confgen/config_test.go
@@ -31,7 +31,6 @@ func expectedConfig() *Config {
 		Description: "test description",
 		Encode: config.Encode{
 			Prom: &api.PromEncode{
-				Port:   7777,
 				Prefix: "prefix",
 			},
 		},
@@ -54,7 +53,6 @@ ingest:
     port: 8888
 encode:
   prom:
-    port: 7777
     prefix: prefix
 `
 
diff --git a/pkg/confgen/flowlogs2metrics_config.go b/pkg/confgen/flowlogs2metrics_config.go
@@ -56,8 +56,6 @@ func (cg *ConfGen) GenerateFlowlogs2PipelineConfig() *config.ConfigFileStruct {
 	}
 	if len(cg.promMetrics) > 0 {
 		metricsNode.EncodePrometheus("encode_prom", api.PromEncode{
-			Address: cg.config.Encode.Prom.Address,
-			Port:    cg.config.Encode.Prom.Port,
 			Prefix:  cg.config.Encode.Prom.Prefix,
 			Metrics: cg.promMetrics,
 		})
@@ -69,6 +67,10 @@ func (cg *ConfGen) GenerateFlowlogs2PipelineConfig() *config.ConfigFileStruct {
 		LogLevel:   "error",
 		Pipeline:   pipeline.GetStages(),
 		Parameters: pipeline.GetStageParams(),
+		MetricsSettings: config.MetricsSettings{
+			Port:   9102,
+			Prefix: "flp_op",
+		},
 	}
 }
 
diff --git a/pkg/config/config.go b/pkg/config/config.go
@@ -36,9 +36,9 @@ type Options struct {
 
 type ConfigFileStruct struct {
 	LogLevel        string          `yaml:"log-level,omitempty" json:"log-level,omitempty"`
+	MetricsSettings MetricsSettings `yaml:"metricsSettings,omitempty" json:"metricsSettings,omitempty"`
 	Pipeline        []Stage         `yaml:"pipeline,omitempty" json:"pipeline,omitempty"`
 	Parameters      []StageParam    `yaml:"parameters,omitempty" json:"parameters,omitempty"`
-	MetricsSettings MetricsSettings `yaml:"metricsSettings,omitempty" json:"metricsSettings,omitempty"`
 	PerfSettings    PerfSettings    `yaml:"perfSettings,omitempty" json:"perfSettings,omitempty"`
 }
 
@@ -55,12 +55,11 @@ type Profile struct {
 // Also, currently FLP doesn't support defining more than one PromEncode stage. If this feature is added later, these global settings
 // will help configuring common setting for all PromEncode stages - PromEncode settings would then act as overrides.
 type MetricsSettings struct {
-	// TODO: manage global metrics server, ie. not coupled to PromEncode, cf https://github.com/netobserv/flowlogs-pipeline/issues/302
-	// Port       int              `yaml:"port,omitempty" json:"port,omitempty" doc:"port number to expose \"/metrics\" endpoint"`
-	// TLS        *PromTLSConf     `yaml:"tls,omitempty" json:"tls,omitempty" doc:"TLS configuration for the prometheus endpoint"`
-	// ExpiryTime int              `yaml:"expiryTime,omitempty" json:"expiryTime,omitempty" doc:"seconds of no-flow to wait before deleting prometheus data item"`
-	Prefix  string `yaml:"prefix,omitempty" json:"prefix,omitempty"`
-	NoPanic bool   `yaml:"noPanic,omitempty" json:"noPanic,omitempty"`
+	Address string           `yaml:"address,omitempty" json:"address,omitempty" doc:"address to expose \"/metrics\" endpoint"`
+	Port    int              `yaml:"port,omitempty" json:"port,omitempty" doc:"port number to expose \"/metrics\" endpoint"`
+	TLS     *api.PromTLSConf `yaml:"tls,omitempty" json:"tls,omitempty" doc:"TLS configuration for the prometheus endpoint"`
+	Prefix  string           `yaml:"prefix,omitempty" json:"prefix,omitempty"`
+	NoPanic bool             `yaml:"noPanic,omitempty" json:"noPanic,omitempty"`
 }
 
 // PerfSettings allows setting some internal configuration parameters
diff --git a/pkg/config/pipeline_builder_test.go b/pkg/config/pipeline_builder_test.go
@@ -129,7 +129,6 @@ func TestKafkaPromPipeline(t *testing.T) {
 			Labels:   []string{"by", "aggregate"},
 			Buckets:  []float64{},
 		}},
-		Port:   9090,
 		Prefix: "flp_",
 	})
 	stages := pl.GetStages()
@@ -160,7 +159,7 @@ func TestKafkaPromPipeline(t *testing.T) {
 
 	b, err = json.Marshal(params[4])
 	require.NoError(t, err)
-	require.JSONEq(t, `{"name":"prom","encode":{"type":"prom","prom":{"metrics":[{"name":"connections_per_source_as","type":"counter","filter":{"key":"name","value":"src_as_connection_count"},"valueKey":"recent_count","labels":["by","aggregate"],"buckets":[]}],"port":9090,"prefix":"flp_"}}}`, string(b))
+	require.JSONEq(t, `{"name":"prom","encode":{"type":"prom","prom":{"metrics":[{"name":"connections_per_source_as","type":"counter","filter":{"key":"name","value":"src_as_connection_count"},"valueKey":"recent_count","labels":["by","aggregate"],"buckets":[]}],"prefix":"flp_"}}}`, string(b))
 }
 
 func TestForkPipeline(t *testing.T) {
diff --git a/pkg/operational/metrics.go b/pkg/operational/metrics.go
@@ -111,7 +111,9 @@ func NewMetrics(settings *config.MetricsSettings) *Metrics {
 func (o *Metrics) register(c prometheus.Collector, name string) {
 	err := prometheus.DefaultRegisterer.Register(c)
 	if err != nil {
-		if o.settings.NoPanic {
+		if _, ok := err.(prometheus.AlreadyRegisteredError); ok {
+			logrus.Errorf("metrics registration error [%s]: %v", name, err)
+		} else if o.settings.NoPanic {
 			logrus.Errorf("metrics registration error [%s]: %v", name, err)
 		} else {
 			logrus.Panicf("metrics registration error [%s]: %v", name, err)
diff --git a/pkg/pipeline/aggregate_prom_test.go b/pkg/pipeline/aggregate_prom_test.go
@@ -64,7 +64,6 @@ parameters:
    encode:
      type: prom
      prom:
-       port: 9103
        prefix: test_
        expiryTime: 1
        metrics:
diff --git a/pkg/pipeline/encode/encode_prom.go b/pkg/pipeline/encode/encode_prom.go
@@ -18,11 +18,7 @@
 package encode
 
 import (
-	"context"
-	"crypto/tls"
 	"fmt"
-	"net/http"
-	"os"
 	"strings"
 	"time"
 
@@ -33,7 +29,6 @@ import (
 	"github.com/netobserv/flowlogs-pipeline/pkg/utils"
 
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/client_golang/prometheus/promhttp"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -62,8 +57,6 @@ type EncodeProm struct {
 	expiryTime       time.Duration
 	mCache           *putils.TimedCache
 	exitChan         <-chan struct{}
-	server           *http.Server
-	tlsConfig        *api.PromTLSConf
 	metricsProcessed prometheus.Counter
 	metricsDropped   prometheus.Counter
 	errorsCounter    *prometheus.CounterVec
@@ -261,30 +254,6 @@ func (e *EncodeProm) cleanupExpiredEntriesLoop() {
 	}
 }
 
-// startServer listens for prometheus resource usage requests
-func (e *EncodeProm) startServer() {
-	log.Debugf("entering startServer")
-
-	// The Handler function provides a default handler to expose metrics
-	// via an HTTP server. "/metrics" is the usual endpoint for that.
-	http.Handle("/metrics", promhttp.Handler())
-
-	var err error
-	if e.tlsConfig != nil {
-		err = e.server.ListenAndServeTLS(e.tlsConfig.CertPath, e.tlsConfig.KeyPath)
-	} else {
-		err = e.server.ListenAndServe()
-	}
-	if err != nil && err != http.ErrServerClosed {
-		log.Errorf("error in http.ListenAndServe: %v", err)
-		os.Exit(1)
-	}
-}
-
-func (e *EncodeProm) closeServer(ctx context.Context) error {
-	return e.server.Shutdown(ctx)
-}
-
 func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (Encoder, error) {
 	cfg := api.PromEncode{}
 	if params.Encode != nil && params.Encode.Prom != nil {
@@ -365,19 +334,7 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 	log.Debugf("histos = %v", histos)
 	log.Debugf("aggHistos = %v", aggHistos)
 
-	// if value of address is empty, then by default it will take 0.0.0.0
-	addr := fmt.Sprintf("%s:%v", cfg.Address, cfg.Port)
-	log.Infof("startServer: addr = %s", addr)
-
 	w := &EncodeProm{
-		server: &http.Server{
-			Addr: addr,
-			// TLS clients must use TLS 1.2 or higher
-			TLSConfig: &tls.Config{
-				MinVersion: tls.VersionTLS12,
-			},
-		},
-		tlsConfig:        cfg.TLS,
 		counters:         counters,
 		gauges:           gauges,
 		histos:           histos,
@@ -389,7 +346,6 @@ func NewEncodeProm(opMetrics *operational.Metrics, params config.StageParam) (En
 		metricsDropped:   opMetrics.NewCounter(&metricsDropped, params.Name),
 		errorsCounter:    opMetrics.NewCounterVec(&encodePromErrors),
 	}
-	go w.startServer()
 	go w.cleanupExpiredEntriesLoop()
 	return w, nil
 }
diff --git a/pkg/pipeline/encode/encode_prom_test.go b/pkg/pipeline/encode/encode_prom_test.go
diff --git a/pkg/pipeline/encode/prom_cache_test.go b/pkg/pipeline/encode/prom_cache_test.go
diff --git a/pkg/pipeline/utils/prom_server.go b/pkg/pipeline/utils/prom_server.go
diff --git a/pkg/test/network_defs.go b/pkg/test/network_defs.go

Original file line number	Diff line number	Diff line change
`@@ -56,8 +56,6 @@ func (cg ConfGen) GenerateFlowlogs2PipelineConfig() config.ConfigFileStruct {`
`56`	`56`	`}`
`57`	`57`	`if len(cg.promMetrics) > 0 {`
`58`	`58`	`metricsNode.EncodePrometheus("encode_prom", api.PromEncode{`
`59`		`- Address: cg.config.Encode.Prom.Address,`
`60`		`- Port: cg.config.Encode.Prom.Port,`
`61`	`59`	`Prefix: cg.config.Encode.Prom.Prefix,`
`62`	`60`	`Metrics: cg.promMetrics,`
`63`	`61`	`})`
`@@ -69,6 +67,10 @@ func (cg ConfGen) GenerateFlowlogs2PipelineConfig() config.ConfigFileStruct {`
`69`	`67`	`LogLevel: "error",`
`70`	`68`	`Pipeline: pipeline.GetStages(),`
`71`	`69`	`Parameters: pipeline.GetStageParams(),`
	`70`	`+ MetricsSettings: config.MetricsSettings{`
	`71`	`+ Port: 9102,`
	`72`	`+ Prefix: "flp_op",`
	`73`	`+ },`
`72`	`74`	`}`
`73`	`75`	`}`
`74`	`76`