lokiexporter: Implementation ( PR 2/3 ) (#2262)

* added initial loki exporter implementation Signed-off-by: Granville Schmidt <[email protected]> * updated exporter to use consumererrors to enable exporthelper retry logic Signed-off-by: Granville Schmidt <[email protected]> * added memory limiter processor to example Signed-off-by: Granville Schmidt <[email protected]>
pmatyjasek-sumo · Apr 28, 2021 · d528dd9 · d528dd9
1 parent e4b0f98
commit d528dd9
Show file tree

Hide file tree

Showing 19 changed files with 1,785 additions and 426 deletions.
diff --git a/exporter/lokiexporter/README.md b/exporter/lokiexporter/README.md
@@ -8,18 +8,21 @@ Supported pipeline types: logs
 
 The following settings are required:
 
-- `endpoint` (no default): The target URL to send Loki log streams to (
-  e.g.: https://loki.example.com:3100/loki/api/v1/push).
-
-
-- `attributes_for_labels` (no default): List of allowed attributes to be added as labels to Loki log 
-  streams. This is a safety net to help prevent accidentally adding dynamic labels that may significantly increase 
-  cardinality, thus having a performance impact on your Loki instance. See the 
+- `endpoint` (no default): The target URL to send Loki log streams to (e.g.: http://loki:3100/loki/api/v1/push).
+
+- `labels.attributes` (no default): Map of attributes names to valid Loki label names (must match "^[a-zA-Z_][a-zA-Z0-9_]*$") 
+  allowed to be added as labels to Loki log streams. Logs that do not have at least one of these attributes will be dropped. 
+  This is a safety net to help prevent accidentally adding dynamic labels that may significantly increase cardinality, 
+  thus having a performance impact on your Loki instance. See the 
   [Loki label best practices](https://grafana.com/docs/loki/latest/best-practices/current-best-practices/) page for 
   additional details on the types of labels you may want to associate with log streams.
 
 The following settings can be optionally configured:
 
+- `tenant_id` (no default): The tenant ID used to identify the tenant the logs are associated to. This will set the 
+  "X-Scope-OrgID" header used by Loki. If left unset, this header will not be added.
+
+
 - `insecure` (default = false): When set to true disables verifying the server's certificate chain and host name. The
   connection is still encrypted but server identity is not verified.
 - `ca_file` (no default) Path to the CA cert to verify the server being connected to. Should only be used if `insecure` 
@@ -35,20 +38,24 @@ The following settings can be optionally configured:
 - `write_buffer_size` (default = 512 * 1024): WriteBufferSize for HTTP client.
 
 
-- `headers` (no default): Name/value pairs added to the HTTP request headers. Loki by default uses the "X-Scope-OrgID"
-  header to identify the tenant the log is associated to.
+- `headers` (no default): Name/value pairs added to the HTTP request headers.
 
 Example:
 
 ```yaml
 loki:
-  endpoint: https://loki.example.com:3100/loki/api/v1/push
-  attributes_for_labels:
-    - container.name
-    - k8s.cluster.name
-    - severity
+  endpoint: http://loki:3100/loki/api/v1/push
+  tenant_id: "example"
+  labels:
+    attributes:
+      # Allowing 'container.name' attribute and transform it to 'container_name', which is a valid Loki label name.
+      container.name: "container_name"
+      # Allowing 'k8s.cluster.name' attribute and transform it to 'k8s_cluster_name', which is a valid Loki label name.
+      k8s.cluster.name: "k8s_cluster_name"
+      # Allowing 'severity' attribute and not providing a mapping, since the attribute name is a valid Loki label name.
+      severity: ""
   headers:
-    "X-Scope-OrgID": "example"
+    "X-Custom-Header": "loki_rocks"
 ```
 
 The full list of settings exposed for this exporter are documented [here](./config.go) with detailed sample

diff --git a/exporter/lokiexporter/config.go b/exporter/lokiexporter/config.go
@@ -15,6 +15,10 @@
 package lokiexporter
 
 import (
+	"fmt"
+	"net/url"
+
+	"github.com/prometheus/common/model"
 	"go.opentelemetry.io/collector/config/confighttp"
 	"go.opentelemetry.io/collector/config/configmodels"
 	"go.opentelemetry.io/collector/exporter/exporterhelper"
@@ -27,6 +31,60 @@ type Config struct {
 	exporterhelper.QueueSettings  `mapstructure:"sending_queue"`
 	exporterhelper.RetrySettings  `mapstructure:"retry_on_failure"`
 
-	// The attributes that are allowed to be added as labels on the log stream sent to Loki.
-	AttributesForLabels []string `mapstructure:"attributes_for_labels"`
+	// TenantID defines the tenant ID to associate log streams with.
+	TenantID string `mapstructure:"tenant_id"`
+
+	// Labels defines how labels should be applied to log streams sent to Loki.
+	Labels LabelsConfig `mapstructure:"labels"`
+}
+
+func (c *Config) validate() error {
+	if _, err := url.Parse(c.Endpoint); c.Endpoint == "" || err != nil {
+		return fmt.Errorf("\"endpoint\" must be a valid URL")
+	}
+
+	if err := c.Labels.validate(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// LabelsConfig defines the labels-related configuration
+type LabelsConfig struct {
+	// Attributes are the attributes that are allowed to be added as labels on a log stream.
+	Attributes map[string]string `mapstructure:"attributes"`
+}
+
+func (c *LabelsConfig) validate() error {
+	if len(c.Attributes) == 0 {
+		return fmt.Errorf("\"labels.attributes\" must be configured with at least one attribute")
+	}
+
+	labelNameInvalidErr := "the label `%s` in \"labels.attributes\" is not a valid label name. Label names must match " + model.LabelNameRE.String()
+	for l, v := range c.Attributes {
+		if len(v) > 0 && !model.LabelName(v).IsValid() {
+			return fmt.Errorf(labelNameInvalidErr, v)
+		} else if len(v) == 0 && !model.LabelName(l).IsValid() {
+			return fmt.Errorf(labelNameInvalidErr, l)
+		}
+	}
+
+	return nil
+}
+
+// getAttributes creates a lookup of allowed attributes to valid Loki label names.
+func (c *LabelsConfig) getAttributes() map[string]model.LabelName {
+	attributes := map[string]model.LabelName{}
+
+	for attrName, lblName := range c.Attributes {
+		if len(lblName) > 0 {
+			attributes[attrName] = model.LabelName(lblName)
+			continue
+		}
+
+		attributes[attrName] = model.LabelName(attrName)
+	}
+
+	return attributes
 }
diff --git a/exporter/lokiexporter/config_test.go b/exporter/lokiexporter/config_test.go
@@ -19,6 +19,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component/componenttest"
@@ -48,7 +49,7 @@ func TestLoadConfig(t *testing.T) {
 		ExporterSettings: configmodels.ExporterSettings{TypeVal: typeStr, NameVal: "loki/allsettings"},
 		HTTPClientSettings: confighttp.HTTPClientSettings{
 			Headers: map[string]string{
-				"x-scope-orgid": "example",
+				"x-custom-header": "loki_rocks",
 			},
 			Endpoint: "https://loki:3100/loki/api/v1/push",
 			TLSSetting: configtls.TLSClientSetting{
@@ -74,7 +75,236 @@ func TestLoadConfig(t *testing.T) {
 			NumConsumers: 2,
 			QueueSize:    10,
 		},
-		AttributesForLabels: []string{conventions.AttributeContainerName, conventions.AttributeK8sCluster, "severity"},
+		TenantID: "example",
+		Labels: LabelsConfig{
+			Attributes: map[string]string{
+				conventions.AttributeContainerName: "container_name",
+				conventions.AttributeK8sCluster:    "k8s_cluster_name",
+				"severity":                         "severity",
+			},
+		},
+	}
+	require.Equal(t, &expectedCfg, actualCfg)
+}
+
+func TestConfig_validate(t *testing.T) {
+	const validEndpoint = "https://validendpoint.local"
+
+	validLabelsConfig := LabelsConfig{
+		Attributes: testValidAttributesWithMapping,
+	}
+
+	type fields struct {
+		ExporterSettings configmodels.ExporterSettings
+		Endpoint         string
+		Source           string
+		CredentialFile   string
+		Audience         string
+		Labels           LabelsConfig
+	}
+	tests := []struct {
+		name         string
+		fields       fields
+		errorMessage string
+		shouldError  bool
+	}{
+		{
+			name: "with valid endpoint",
+			fields: fields{
+				Endpoint: validEndpoint,
+				Labels:   validLabelsConfig,
+			},
+			shouldError: false,
+		},
+		{
+			name: "with missing endpoint",
+			fields: fields{
+				Endpoint: "",
+				Labels:   validLabelsConfig,
+			},
+			errorMessage: "\"endpoint\" must be a valid URL",
+			shouldError:  true,
+		},
+		{
+			name: "with invalid endpoint",
+			fields: fields{
+				Endpoint: "this://is:an:invalid:endpoint.com",
+				Labels:   validLabelsConfig,
+			},
+			errorMessage: "\"endpoint\" must be a valid URL",
+			shouldError:  true,
+		},
+		{
+			name: "with missing `labels.attributes`",
+			fields: fields{
+				Endpoint: validEndpoint,
+				Labels: LabelsConfig{
+					Attributes: nil,
+				},
+			},
+			errorMessage: "\"labels.attributes\" must be configured with at least one attribute",
+			shouldError:  true,
+		},
+		{
+			name: "with `labels.attributes` set",
+			fields: fields{
+				Endpoint: validEndpoint,
+				Labels: LabelsConfig{
+					Attributes: testValidAttributesWithMapping,
+				},
+			},
+			shouldError: false,
+		},
+		{
+			name: "with valid `labels` config",
+			fields: fields{
+				Endpoint: validEndpoint,
+				Labels:   validLabelsConfig,
+			},
+			shouldError: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			factory := NewFactory()
+			cfg := factory.CreateDefaultConfig().(*Config)
+			cfg.ExporterSettings = tt.fields.ExporterSettings
+			cfg.Endpoint = tt.fields.Endpoint
+			cfg.Labels = tt.fields.Labels
+
+			err := cfg.validate()
+			if (err != nil) != tt.shouldError {
+				t.Errorf("validate() error = %v, shouldError %v", err, tt.shouldError)
+				return
+			}
+
+			if tt.shouldError {
+				assert.Error(t, err)
+				if len(tt.errorMessage) != 0 {
+					assert.Equal(t, tt.errorMessage, err.Error())
+				}
+			}
+		})
+	}
+}
+
+func TestLabelsConfig_validate(t *testing.T) {
+	tests := []struct {
+		name         string
+		labels       LabelsConfig
+		errorMessage string
+		shouldError  bool
+	}{
+		{
+			name: "with no attributes",
+			labels: LabelsConfig{
+				Attributes: map[string]string{},
+			},
+			errorMessage: "\"labels.attributes\" must be configured with at least one attribute",
+			shouldError:  true,
+		},
+		{
+			name: "with valid attribute label map",
+			labels: LabelsConfig{
+				Attributes: map[string]string{
+					"some.attribute": "some_attribute",
+				},
+			},
+			shouldError: false,
+		},
+		{
+			name: "with invalid attribute label map",
+			labels: LabelsConfig{
+				Attributes: map[string]string{
+					"some.attribute": "invalid.label.name",
+				},
+			},
+			errorMessage: "the label `invalid.label.name` in \"labels.attributes\" is not a valid label name. Label names must match " + model.LabelNameRE.String(),
+			shouldError:  true,
+		},
+		{
+			name: "with attribute having an invalid label name and no map configured",
+			labels: LabelsConfig{
+				Attributes: map[string]string{
+					"invalid.attribute": "",
+				},
+			},
+			errorMessage: "the label `invalid.attribute` in \"labels.attributes\" is not a valid label name. Label names must match " + model.LabelNameRE.String(),
+			shouldError:  true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.labels.validate()
+			if (err != nil) != tt.shouldError {
+				t.Errorf("validate() error = %v, shouldError %v", err, tt.shouldError)
+				return
+			}
+
+			if tt.shouldError {
+				assert.Error(t, err)
+				if len(tt.errorMessage) != 0 {
+					assert.Equal(t, tt.errorMessage, err.Error())
+				}
+			}
+		})
+	}
+}
+
+func TestLabelsConfig_getAttributes(t *testing.T) {
+	tests := []struct {
+		name            string
+		labels          LabelsConfig
+		expectedMapping map[string]model.LabelName
+	}{
+		{
+			name: "with attributes without label mapping",
+			labels: LabelsConfig{
+				Attributes: map[string]string{
+					"attribute_1": "",
+					"attribute_2": "",
+				},
+			},
+			expectedMapping: map[string]model.LabelName{
+				"attribute_1": model.LabelName("attribute_1"),
+				"attribute_2": model.LabelName("attribute_2"),
+			},
+		},
+		{
+			name: "with attributes and label mapping",
+			labels: LabelsConfig{
+				Attributes: map[string]string{
+					"attribute.1": "attribute_1",
+					"attribute.2": "attribute_2",
+				},
+			},
+			expectedMapping: map[string]model.LabelName{
+				"attribute.1": model.LabelName("attribute_1"),
+				"attribute.2": model.LabelName("attribute_2"),
+			},
+		},
+		{
+			name: "with attributes and without label mapping",
+			labels: LabelsConfig{
+				Attributes: map[string]string{
+					"attribute.1": "attribute_1",
+					"attribute2":  "",
+				},
+			},
+			expectedMapping: map[string]model.LabelName{
+				"attribute.1": model.LabelName("attribute_1"),
+				"attribute2":  model.LabelName("attribute2"),
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mapping := tt.labels.getAttributes()
+
+			assert.Equal(t, tt.expectedMapping, mapping)
+		})
 	}
-	assert.Equal(t, &expectedCfg, actualCfg)
 }