fluent-bit: tenant ID configuration (#1273)

* fluent-bit: tenant ID configuration

Both promtail and the fluentd plugin already have support for sending a
tenant ID, and all necessary for doing so in fluent-bit is actually
passing through the configuration item. Configuration is aligned with
the fluentd plugin for consistency.

We're lucky, since no configuration will result in an empty string that
will be treated as no tenant ID by the Loki shipping client.

Signed-off-by: Jens Erat <email@jenserat.de>

* promtail: minor typo in documentation

Signed-off-by: Jens Erat <email@jenserat.de>

cmd/fluent-bit/README.md

@@ -11,6 +11,7 @@ This plugin is implemented with [Fluent Bit's Go plugin](https://github.com/flue
 | Key           | Description                                   | Default                             |
 | --------------|-----------------------------------------------|-------------------------------------|
 | Url           | Url of loki server API endpoint.               | http://localhost:3100/loki/api/v1/push |
+| TenantID      | The tenant ID used by default to push logs to Loki. If omitted or empty it assumes Loki is running in single-tenant mode and no `X-Scope-OrgID` header is sent.               | "" |
 | BatchWait     | Time to wait before send a log batch to Loki, full or not. (unit: sec) | 1 second   |
 | BatchSize     | Log batch size to send a log batch to Loki (unit: Bytes).    | 10 KiB (10 * 1024 Bytes) |
 | Labels        | labels for API requests.                       | {job="fluent-bit"}                    |

cmd/fluent-bit/config.go

@@ -61,6 +61,9 @@ func parseConfig(cfg ConfigGetter) (*config, error) {
 	}
 	res.clientConfig.URL = clientURL
 
+	// cfg.Get will return empty string if not set, which is handled by the client library as no tenant
+	res.clientConfig.TenantID = cfg.Get("TenantID")
+
 	batchWait := cfg.Get("BatchWait")
 	if batchWait != "" {
 		batchWaitValue, err := strconv.Atoi(batchWait)

cmd/fluent-bit/config_test.go

@@ -49,6 +49,7 @@ func Test_parseConfig(t *testing.T) {
 		{"setting values",
 			map[string]string{
 				"URL":           "http://somewhere.com:3100/loki/api/v1/push",
+				"TenantID":      "my-tenant-id",
 				"LineFormat":    "key_value",
 				"LogLevel":      "warn",
 				"Labels":        `{app="foo"}`,
@@ -62,6 +63,7 @@ func Test_parseConfig(t *testing.T) {
 				lineFormat: kvPairFormat,
 				clientConfig: client.Config{
 					URL:            mustParseURL("http://somewhere.com:3100/loki/api/v1/push"),
+					TenantID:       "my-tenant-id",
 					BatchSize:      100,
 					BatchWait:      30 * time.Second,
 					ExternalLabels: lokiflag.LabelSet{LabelSet: model.LabelSet{"app": "foo"}},
@@ -89,6 +91,7 @@ func Test_parseConfig(t *testing.T) {
 				lineFormat: kvPairFormat,
 				clientConfig: client.Config{
 					URL:            mustParseURL("http://somewhere.com:3100/loki/api/v1/push"),
+					TenantID:       "", // empty as not set in fluent-bit plugin config map
 					BatchSize:      100,
 					BatchWait:      30 * time.Second,
 					ExternalLabels: lokiflag.LabelSet{LabelSet: model.LabelSet{"app": "foo"}},
@@ -148,6 +151,9 @@ func assertConfig(t *testing.T, expected, actual *config) {
 	if !reflect.DeepEqual(expected.clientConfig.URL, actual.clientConfig.URL) {
 		t.Errorf("incorrect URL want:%v got:%v", expected.clientConfig.URL, actual.clientConfig.URL)
 	}
+	if !reflect.DeepEqual(expected.clientConfig.TenantID, actual.clientConfig.TenantID) {
+		t.Errorf("incorrect TenantID want:%v got:%v", expected.clientConfig.TenantID, actual.clientConfig.TenantID)
+	}
 	if !reflect.DeepEqual(expected.lineFormat, actual.lineFormat) {
 		t.Errorf("incorrect lineFormat want:%v got:%v", expected.lineFormat, actual.lineFormat)
 	}

cmd/fluent-bit/out_loki.go

@@ -48,6 +48,7 @@ func FLBPluginInit(ctx unsafe.Pointer) int {
 	logger = newLogger(conf.logLevel)
 	level.Info(logger).Log("[flb-go]", "Starting fluent-bit-go-loki", "version", version.Info())
 	level.Info(logger).Log("[flb-go]", "provided parameter", "URL", conf.clientConfig.URL)
+	level.Info(logger).Log("[flb-go]", "provided parameter", "TenantID", conf.clientConfig.TenantID)
 	level.Info(logger).Log("[flb-go]", "provided parameter", "BatchWait", conf.clientConfig.BatchWait)
 	level.Info(logger).Log("[flb-go]", "provided parameter", "BatchSize", conf.clientConfig.BatchSize)
 	level.Info(logger).Log("[flb-go]", "provided parameter", "Labels", conf.clientConfig.ExternalLabels)

docs/clients/promtail/configuration.md

@@ -138,7 +138,7 @@ url: <string>
 
 # The tenant ID used by default to push logs to Loki. If omitted or empty
 # it assumes Loki is running in single-tenant mode and no X-Scope-OrgID header
-# it sent.
+# is sent.
 [tenant_id: <string>]
 
 # Maximum amount of time to wait before sending a batch, even if that