Add metrics

Makefile

@@ -24,7 +24,7 @@ plugin:
 
 .PHONY: docker-images
 docker-images:
-	@docker build -t $(FLUENT_BIT_TO_LOKI_IMAGE_REPOSITORY):$(FLUENT_BIT_TO_LOKI_IMAGE_TAG) -t $(FLUENT_BIT_TO_LOKI_IMAGE_REPOSITORY):latest -f Dockerfile --target fluent-bit .
+	@docker build -t $(FLUENT_BIT_TO_LOKI_IMAGE_REPOSITORY):$(FLUENT_BIT_TO_LOKI_IMAGE_TAG) -t $(FLUENT_BIT_TO_LOKI_IMAGE_REPOSITORY):latest -f Dockerfile --target carrier .
 
 .PHONY: docker-push
 docker-push:

README.md

@@ -38,7 +38,9 @@ It also adds additional configurations that aim to improve plugin's performance
 | TagKey | The key of the record which holds the tag. The tag should not be nested | "tag"
 | TagPrefix | The prefix of the tag. In the prefix no metadata will be searched. The prefix must not contain group expression(`()`). | none
 | TagExpression | The regex expression which will be used for matching the metadata retrieved from the tag. It contains 3 group expressions (`()`): `pod name`, `namespace` and the `container name` | "\\.(.*)_(.*)_(.*)-.*\\.log"
-| DropLogEntryWithoutK8sMetadata | When metadata is missing for the log entry, it will be dropped | `false` 
+| DropLogEntryWithoutK8sMetadata | When metadata is missing for the log entry, it will be dropped | `false`
+| MetricsTickWindow | The time in which the metrics will be kept in seconds | `3600`
+| MetricsTickInterval | Update time for the metrics in seconds  | `30`
 
 
 ### Labels

cmd/out_loki.go

@@ -26,9 +26,15 @@ import (
 
 	"github.com/gardener/logging/fluent-bit-to-loki/pkg/config"
 	"github.com/gardener/logging/fluent-bit-to-loki/pkg/lokiplugin"
+	"github.com/gardener/logging/fluent-bit-to-loki/pkg/metrics"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 )
+import (
+	"net/http"
+
+	"github.com/prometheus/client_golang/prometheus/promhttp"
+)
 
 var (
 	// registered loki plugin instances, required for disposal during shutdown
@@ -39,6 +45,12 @@ var (
 )
 
 func init() {
+	// metrics
+	go func() {
+		http.Handle("/metrics", promhttp.Handler())
+		http.ListenAndServe(":2021", nil)
+	}()
+
 	var logLevel logging.Level
 	_ = logLevel.Set("info")
 	logger = log.With(newLogger(logLevel), "ts", log.DefaultTimestampUTC, "caller", "main")
@@ -70,6 +82,7 @@ func FLBPluginRegister(ctx unsafe.Pointer) int {
 func FLBPluginInit(ctx unsafe.Pointer) int {
 	conf, err := config.ParseConfig(&pluginConfig{ctx: ctx})
 	if err != nil {
+		metrics.ErrorsMetric.Add(1, metrics.ErrorFLBPluginInit)
 		level.Error(logger).Log("[flb-go]", "failed to launch", "error", err)
 		return output.FLB_ERROR
 	}
@@ -112,9 +125,12 @@ func FLBPluginInit(ctx unsafe.Pointer) int {
 	level.Info(paramLogger).Log("TagPrefix", fmt.Sprintf("%+v", conf.KubernetesMetadata.TagPrefix))
 	level.Info(paramLogger).Log("TagExpression", fmt.Sprintf("%+v", conf.KubernetesMetadata.TagExpression))
 	level.Info(paramLogger).Log("DropLogEntryWithoutK8sMetadata", fmt.Sprintf("%+v", conf.KubernetesMetadata.DropLogEntryWithoutK8sMetadata))
+	level.Info(paramLogger).Log("MetricsTickWindow", fmt.Sprintf("%+v", conf.Metrics.MetricsTickWindow))
+	level.Info(paramLogger).Log("MetricsTickInterval", fmt.Sprintf("%+v", conf.Metrics.MetricsTickInterval))
 
 	plugin, err := lokiplugin.NewPlugin(informer, conf, logger)
 	if err != nil {
+		metrics.ErrorsMetric.Add(1, metrics.ErrorNewPlugin)
 		level.Error(logger).Log("newPlugin", err)
 		return output.FLB_ERROR
 	}
@@ -131,6 +147,7 @@ func FLBPluginInit(ctx unsafe.Pointer) int {
 func FLBPluginFlushCtx(ctx, data unsafe.Pointer, length C.int, _ *C.char) int {
 	plugin := output.FLBPluginGetContext(ctx).(lokiplugin.Loki)
 	if plugin == nil {
+		metrics.ErrorsMetric.Add(1, metrics.ErrorFLBPluginFlushCtx)
 		level.Error(logger).Log("[flb-go]", "plugin not initialized")
 		return output.FLB_ERROR
 	}
@@ -180,6 +197,7 @@ func FLBPluginExit() int {
 		plugin.Close()
 	}
 	close(informerStopChan)
+	close(lokiplugin.StopChn)
 	return output.FLB_OK
 }
 

go.mod

@@ -16,7 +16,8 @@ require (
 	github.com/json-iterator/go v1.1.10
 	github.com/onsi/ginkgo v1.11.0
 	github.com/onsi/gomega v1.8.1
-	github.com/prometheus/common v0.10.0
+	github.com/prometheus/client_golang v1.8.0
+	github.com/prometheus/common v0.14.0
 	github.com/weaveworks/common v0.0.0-20200625145055-4b1847531bc9
 	k8s.io/apimachinery v0.18.6
 	k8s.io/client-go v12.0.0+incompatible
@@ -30,6 +31,7 @@ replace (
 
 	github.com/Azure/go-autorest => github.com/Azure/go-autorest v13.3.0+incompatible
 	github.com/hpcloud/tail => github.com/grafana/tail v0.0.0-20191024143944-0b54ddf21fe7
+	github.com/prometheus/common => github.com/prometheus/common v0.10.0
 
 	golang.org/x/net v0.0.0-20190813000000-74dc4d7220e7 => golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7
 	k8s.io/api => k8s.io/api v0.17.9

go.sum

@@ -549,6 +549,8 @@ github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvq
 github.com/golang/protobuf v1.4.1/go.mod h1:U8fpvMrcmy5pZrNK1lt4xCsGvpyWQ/VVv6QDs8UjoX8=
 github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
 github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/golang/protobuf v1.4.3 h1:JjCZWpVbqXDqFVmTfYWEVTMIYrL/NPdPSCHPJ0T/raM=
+github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
 github.com/golang/snappy v0.0.0-20170215233205-553a64147049/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.0-20180518054509-2e65f85255db/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
@@ -1010,6 +1012,8 @@ github.com/prometheus/client_golang v1.4.1/go.mod h1:e9GMxYsXl05ICDXkRhurwBS4Q3O
 github.com/prometheus/client_golang v1.6.0/go.mod h1:ZLOG9ck3JLRdB5MgO8f+lLTe83AXG6ro35rLTxvnIl4=
 github.com/prometheus/client_golang v1.7.1 h1:NTGy1Ja9pByO+xAeH/qiWnLrKtr3hJPNjaVUwnjpdpA=
 github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M=
+github.com/prometheus/client_golang v1.8.0 h1:zvJNkoCFAnYFNC24FV8nW4JdRJ3GIFcLbg65lL/JDcw=
+github.com/prometheus/client_golang v1.8.0/go.mod h1:O9VU6huf47PktckDQfMTX0Y8tY0/7TSWwj+ITvv0TnM=
 github.com/prometheus/client_model v0.0.0-20170216185247-6f3806018612/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
 github.com/prometheus/client_model v0.0.0-20190115171406-56726106282f/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo=
@@ -1019,19 +1023,6 @@ github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:
 github.com/prometheus/client_model v0.1.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/prometheus/client_model v0.2.0 h1:uq5h0d+GuxiXLJLNABMgp2qUWDPiLvgCzz2dUR+/W/M=
 github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
-github.com/prometheus/common v0.0.0-20180518154759-7600349dcfe1/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
-github.com/prometheus/common v0.0.0-20180801064454-c7de2306084e/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
-github.com/prometheus/common v0.0.0-20181113130724-41aa239b4cce/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
-github.com/prometheus/common v0.0.0-20181126121408-4724e9255275/go.mod h1:daVV7qP5qjZbuso7PdcryaAu0sAZbrN9i7WWcTMWvro=
-github.com/prometheus/common v0.2.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
-github.com/prometheus/common v0.4.0/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
-github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4=
-github.com/prometheus/common v0.6.0 h1:kRhiuYSXR3+uv2IbVbZhUxK5zVD/2pp3Gd2PpvPkpEo=
-github.com/prometheus/common v0.6.0/go.mod h1:eBmuwkDJBwy6iBfxCBob6t6dR6ENT/y+J+Zk0j9GMYc=
-github.com/prometheus/common v0.7.0/go.mod h1:DjGbpBbp5NYNiECxcL/VnbXCCaQpKd3tt26CguLLsqA=
-github.com/prometheus/common v0.8.0/go.mod h1:PC/OgXc+UN7B4ALwvn1yzVZmVwvhXp5JsbBv6wSv6i0=
-github.com/prometheus/common v0.9.1 h1:KOMtN28tlbam3/7ZKEYKHhKoJZYYj3gMH4uc62x7X7U=
-github.com/prometheus/common v0.9.1/go.mod h1:yhUN8i9wzaXS3w1O07YhxHEBxD+W35wd8bs7vj7HSQ4=
 github.com/prometheus/common v0.10.0 h1:RyRA7RzGXQZiW+tGMr7sxa85G1z0yOpM1qq5c8lNawc=
 github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo=
 github.com/prometheus/node_exporter v1.0.0-rc.0.0.20200428091818-01054558c289 h1:dTUS1vaLWq+Y6XKOTnrFpoVsQKLCbCp1OLj24TDi7oM=
@@ -1053,6 +1044,8 @@ github.com/prometheus/procfs v0.0.11 h1:DhHlBtkHWPYi8O2y31JkK0TF+DGM+51OopZjH/Ia
 github.com/prometheus/procfs v0.0.11/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/procfs v0.1.3 h1:F0+tqvhOksq22sc6iCHF5WGlWjdwj92p0udFh1VFBS8=
 github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
+github.com/prometheus/procfs v0.2.0 h1:wH4vA7pcjKuZzjF7lM8awk4fnuJO6idemZXoKnULUx4=
+github.com/prometheus/procfs v0.2.0/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU=
 github.com/prometheus/prometheus v0.0.0-20180315085919-58e2a31db8de/go.mod h1:oAIUtOny2rjMX0OWN5vPR5/q/twIROJvdqnQKDdil/s=
 github.com/prometheus/prometheus v0.0.0-20190818123050-43acd0e2e93f/go.mod h1:rMTlmxGCvukf2KMu3fClMDKLLoJ5hl61MhcJ7xKakf0=
 github.com/prometheus/prometheus v1.8.2-0.20200107122003-4708915ac6ef/go.mod h1:7U90zPoLkWjEIQcy/rweQla82OCTUzxVHE51G3OhJbI=
@@ -1472,6 +1465,8 @@ golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200724161237-0e2f3a69832c h1:UIcGWL6/wpCfyGuJnRFJRurA+yj8RrW7Q6x2YMCXt6c=
 golang.org/x/sys v0.0.0-20200724161237-0e2f3a69832c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201015000850-e3ed0017c211 h1:9UQO31fZ+0aKQOFldThf7BKPMJTiBfWycGh/u3UoO88=
+golang.org/x/sys v0.0.0-20201015000850-e3ed0017c211/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.0.0-20160726164857-2910a502d2bf/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=

pkg/buffer/dque.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/gardener/logging/fluent-bit-to-loki/pkg/config"
+	"github.com/gardener/logging/fluent-bit-to-loki/pkg/metrics"
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
@@ -81,6 +82,7 @@ func (c *dqueClient) dequeuer() {
 			case dque.ErrQueueClosed:
 				return
 			default:
+				metrics.ErrorsMetric.Add(1, metrics.ErrorDequeuer)
 				level.Error(c.logger).Log("msg", "error dequeuing record", "error", err, "queue", c.queue.Name)
 				continue
 			}
@@ -89,12 +91,14 @@ func (c *dqueClient) dequeuer() {
 		// Assert type of the response to an Item pointer so we can work with it
 		record, ok := entry.(*dqueEntry)
 		if !ok {
+			metrics.ErrorsMetric.Add(1, metrics.ErrorDequeuerNotValidType)
 			level.Error(c.logger).Log("msg", "error dequeued record is not an valid type", "queue", c.queue.Name)
 			continue
 		}
 
 		level.Debug(c.logger).Log("msg", "sending record to Loki", "url", c.url, "record", record.String())
 		if err := c.loki.Handle(record.LabelSet, record.TimeStamp, record.Line); err != nil {
+			metrics.ErrorsMetric.Add(1, metrics.ErrorDequeuerSendRecourd)
 			level.Error(c.logger).Log("msg", fmt.Sprintf("error sending record to Loki %s", c.url), "error", err)
 		}
 		level.Debug(c.logger).Log("msg", "successful sent record to Loki", "host", c.url, "record", record.String())

pkg/client/client.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/gardener/logging/fluent-bit-to-loki/pkg/buffer"
 	"github.com/gardener/logging/fluent-bit-to-loki/pkg/config"
+	"github.com/gardener/logging/fluent-bit-to-loki/pkg/metrics"
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
@@ -74,10 +75,11 @@ func (c *clientWrapper) Handle(ls model.LabelSet, t time.Time, s string) error {
 	lastTimeStamp, ok := c.lastTS[key]
 	if ok && t.Before(lastTimeStamp) {
 		diff := lastTimeStamp.Sub(t)
-		c.logOutOfOrderDifference(diff, "msg", "log entry out of order. Its timestamp is going to be overwritten", "difference", diff.String(), "LastTimestamp", lastTimeStamp.String(), "IncommingTimestamp", t.String(), "URL", c.url, "Stream", ls.String())
+		c.logOutOfOrderDifference(ls, diff, "msg", "log entry out of order. Its timestamp is going to be overwritten", "difference", diff.String(), "LastTimestamp", lastTimeStamp.String(), "IncommingTimestamp", t.String(), "URL", c.url, "Stream", ls.String())
 		t = lastTimeStamp
 	}
 	c.lastTS[key] = t
+
 	return c.lokiclient.Handle(ls, t, s)
 }
 
@@ -87,12 +89,25 @@ func (c *clientWrapper) Stop() {
 	c.lokiclient.Stop()
 }
 
-func (c *clientWrapper) logOutOfOrderDifference(timeDiff time.Duration, keyvals ...interface{}) {
+func (c *clientWrapper) logOutOfOrderDifference(ls model.LabelSet, timeDiff time.Duration, keyvals ...interface{}) {
+	var severity string
+
 	if timeDiff.Seconds() < 1 {
+		severity = "debug"
 		level.Debug(c.logger).Log(keyvals...)
 	} else if timeDiff.Seconds() <= 5 {
+		severity = "info"
 		level.Info(c.logger).Log(keyvals...)
 	} else {
+		severity = "warn"
 		level.Warn(c.logger).Log(keyvals...)
 	}
+
+	if podName, ok := ls["pod_name"]; ok {
+		metrics.OutOfOrderMetric.Add(1, severity, string(podName))
+	} else if systemdComponent, ok := ls["systemd_component"]; ok {
+		metrics.OutOfOrderMetric.Add(1, severity, string(systemdComponent))
+	} else {
+		metrics.OutOfOrderMetric.Add(1, severity, "")
+	}
 }

pkg/client/client_test.go

@@ -74,6 +74,10 @@ var _ = Describe("Client", func() {
 		DynamicHostPrefix: "http://loki.",
 		DynamicHostSuffix: ".svc:3100/loki/api/v1/push",
 		DynamicHostRegex:  "shoot--",
+		Metrics: config.MetricsConfig{
+			MetricsTickWindow:   60,
+			MetricsTickInterval: 30,
+		},
 	}
 
 	logger := log.NewLogfmtLogger(log.NewSyncWriter(os.Stderr))

pkg/config/config.go

@@ -53,6 +53,12 @@ const (
 
 	// DefaultKubernetesMetadataTagPrefix represents the prefix of the entry's tag
 	DefaultKubernetesMetadataTagPrefix = "kubernetes\\.var\\.log\\.containers"
+
+	// DefaultMetricsTickWindow is the default metrics tick window
+	DefaultMetricsTickWindow = 3600
+
+	// DefaultMetricsTickInterval is the default metrics tick interval
+	DefaultMetricsTickInterval = 30
 )
 
 //Config holds all of the needet properties of the loki output plugin
@@ -72,6 +78,7 @@ type Config struct {
 	DynamicHostSuffix    string
 	DynamicHostRegex     string
 	KubernetesMetadata   KubernetesMetadataExtraction
+	Metrics              MetricsConfig
 }
 
 // BufferConfig contains the buffer settings
@@ -98,6 +105,12 @@ type KubernetesMetadataExtraction struct {
 	TagExpression                      string
 }
 
+// MetricsConfig holds the configurations for the metrics
+type MetricsConfig struct {
+	MetricsTickWindow   int
+	MetricsTickInterval int
+}
+
 // DefaultBufferConfig holds the configurations for using output buffer
 var DefaultBufferConfig = BufferConfig{
 	Buffer:     false,
@@ -368,5 +381,25 @@ func ParseConfig(cfg Getter) (*Config, error) {
 		}
 	}
 
+	metricsTickWindow := cfg.Get("MetricsTickWindow")
+	if metricsTickWindow != "" {
+		res.Metrics.MetricsTickWindow, err = strconv.Atoi(metricsTickWindow)
+		if err != nil {
+			return nil, fmt.Errorf("cannot convert MetricsTickWindow %v to integer, error: %v", metricsTickWindow, err)
+		}
+	} else {
+		res.Metrics.MetricsTickWindow = DefaultMetricsTickWindow
+	}
+
+	metricsTickInterval := cfg.Get("MetricsTickInterval")
+	if metricsTickInterval != "" {
+		res.Metrics.MetricsTickInterval, err = strconv.Atoi(metricsTickInterval)
+		if err != nil {
+			return nil, fmt.Errorf("cannot convert MetricsTickInterval %v to integer, error: %v", metricsTickInterval, err)
+		}
+	} else {
+		res.Metrics.MetricsTickInterval = DefaultMetricsTickInterval
+	}
+
 	return res, nil
 }