Loki Exporter - Adding a feature for loki exporter to encode JSON for log entry

exporter/lokiexporter/README.md

@@ -44,6 +44,8 @@ The following settings can be optionally configured:
 
 - `headers` (no default): Name/value pairs added to the HTTP request headers.
 
+- `format` (default = body): Set the log entry line format. This can be set to 'json' (the entire json encoded log record) or 'body' (the log record body field as a string).
+
 Example:
 
 ```yaml

exporter/lokiexporter/config.go

@@ -36,6 +36,8 @@ type Config struct {
 
 	// Labels defines how labels should be applied to log streams sent to Loki.
 	Labels LabelsConfig `mapstructure:"labels"`
+	// Allows you to choose the entry format in the exporter
+	Format string `mapstructure:"format"`
 }
 
 func (c *Config) validate() error {

exporter/lokiexporter/config_test.go

@@ -42,7 +42,7 @@ func TestLoadConfig(t *testing.T) {
 	require.NoError(t, err)
 	require.NotNil(t, cfg)
 
-	assert.Equal(t, 2, len(cfg.Exporters))
+	assert.Equal(t, 3, len(cfg.Exporters))
 
 	actualCfg := cfg.Exporters[config.NewComponentIDWithName(typeStr, "allsettings")].(*Config)
 	expectedCfg := Config{
@@ -87,6 +87,59 @@ func TestLoadConfig(t *testing.T) {
 				"severity":      "severity",
 			},
 		},
+		Format: "body",
+	}
+	require.Equal(t, &expectedCfg, actualCfg)
+}
+
+func TestJsonLoadConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.Nil(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[config.Type(typeStr)] = factory
+	cfg, err := configtest.LoadConfig(path.Join(".", "testdata", "config.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	assert.Equal(t, 3, len(cfg.Exporters))
+
+	actualCfg := cfg.Exporters[config.NewComponentIDWithName(typeStr, "json")].(*Config)
+	expectedCfg := Config{
+		ExporterSettings: config.NewExporterSettings(config.NewComponentIDWithName(typeStr, "json")),
+		HTTPClientSettings: confighttp.HTTPClientSettings{
+			Headers:  map[string]string{},
+			Endpoint: "https://loki:3100/loki/api/v1/push",
+			TLSSetting: configtls.TLSClientSetting{
+				TLSSetting: configtls.TLSSetting{
+					CAFile:   "",
+					CertFile: "",
+					KeyFile:  "",
+				},
+				Insecure: false,
+			},
+			ReadBufferSize:  0,
+			WriteBufferSize: 524288,
+			Timeout:         30000000000,
+		},
+		RetrySettings: exporterhelper.RetrySettings{
+			Enabled:         true,
+			InitialInterval: 5000000000,
+			MaxInterval:     30000000000,
+			MaxElapsedTime:  300000000000,
+		},
+		QueueSettings: exporterhelper.QueueSettings{
+			Enabled:      true,
+			NumConsumers: 10,
+			QueueSize:    5000,
+		},
+		TenantID: "example",
+		Labels: LabelsConfig{
+			Attributes:         map[string]string{},
+			ResourceAttributes: map[string]string{},
+		},
+		Format: "json",
 	}
 	require.Equal(t, &expectedCfg, actualCfg)
 }

exporter/lokiexporter/encode_json.go

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lokiexporter
+
+import (
+	"encoding/json"
+
+	"go.opentelemetry.io/collector/model/pdata"
+)
+
+// JSON representation of the LogRecord as described by https://developers.google.com/protocol-buffers/docs/proto3#json
+
+type lokiEntry struct {
+	Name       string                 `json:"name,omitempty"`
+	Body       string                 `json:"body,omitempty"`
+	TraceID    string                 `json:"traceid,omitempty"`
+	SpanID     string                 `json:"spanid,omitempty"`
+	Severity   string                 `json:"severity,omitempty"`
+	Attributes map[string]interface{} `json:"attributes,omitempty"`
+	Resources  map[string]interface{} `json:"resources,omitempty"`
+}
+
+func encodeJSON(lr pdata.LogRecord, res pdata.Resource) (string, error) {
+	var logRecord lokiEntry
+	var jsonRecord []byte
+
+	logRecord = lokiEntry{
+		Name:       lr.Name(),
+		Body:       lr.Body().StringVal(),
+		TraceID:    lr.TraceID().HexString(),
+		SpanID:     lr.SpanID().HexString(),
+		Severity:   lr.SeverityText(),
+		Attributes: lr.Attributes().AsRaw(),
+		Resources:  res.Attributes().AsRaw(),
+	}
+
+	jsonRecord, err := json.Marshal(logRecord)
+	if err != nil {
+		return "", err
+	}
+	return string(jsonRecord), nil
+}

exporter/lokiexporter/encode_json_test.go

@@ -0,0 +1,52 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lokiexporter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/model/pdata"
+)
+
+func exampleLog() (pdata.LogRecord, pdata.Resource) {
+
+	buffer := pdata.NewLogRecord()
+	buffer.Body().SetStringVal("Example log")
+	buffer.SetName("name")
+	buffer.SetSeverityText("error")
+	buffer.Attributes().Insert("attr1", pdata.NewAttributeValueString("1"))
+	buffer.Attributes().Insert("attr2", pdata.NewAttributeValueString("2"))
+	buffer.SetTraceID(pdata.NewTraceID([16]byte{1, 2, 3, 4}))
+	buffer.SetSpanID(pdata.NewSpanID([8]byte{5, 6, 7, 8}))
+
+	resource := pdata.NewResource()
+	resource.Attributes().Insert("host.name", pdata.NewAttributeValueString("something"))
+
+	return buffer, resource
+}
+
+func exampleJSON() string {
+	jsonExample := `{"name":"name","body":"Example log","traceid":"01020304000000000000000000000000","spanid":"0506070800000000","severity":"error","attributes":{"attr1":"1","attr2":"2"},"resources":{"host.name":"something"}}`
+	return jsonExample
+}
+
+func TestConvert(t *testing.T) {
+	in := exampleJSON()
+	out, err := encodeJSON(exampleLog())
+	t.Log(in)
+	t.Log(out, err)
+	assert.Equal(t, in, out)
+}

exporter/lokiexporter/example/docker-compose.yml

@@ -13,8 +13,6 @@ services:
     container_name: otel
     command:
       - "--config=/etc/otel-collector-config.yml"
-      # Memory Ballast size should be max 1/3 to 1/2 of memory.
-      - "--mem-ballast-size-mib=683"
     volumes:
       - ./otel-collector-config.yml:/etc/otel-collector-config.yml
     ports:

exporter/lokiexporter/example/otel-collector-config.yml

@@ -24,15 +24,23 @@ processors:
 exporters:
   loki:
     endpoint: "http://loki:3100/loki/api/v1/push"
+    # Will encode the whole OTEL message in json
+    # This format happen after extracting labels
+    format: json
     labels:
       attributes:
         container_name: ""
         source: ""
+      resources:
+        host.name: "hostname"
 
 extensions:
   health_check:
   pprof:
   zpages:
+  memory_ballast:
+    # Memory Ballast size should be max 1/3 to 1/2 of memory.
+    size_mib: 64
 
 service:
   extensions: [pprof, zpages, health_check]

exporter/lokiexporter/exporter.go

@@ -36,17 +36,24 @@ import (
 )
 
 type lokiExporter struct {
-	config *Config
-	logger *zap.Logger
-	client *http.Client
-	wg     sync.WaitGroup
+	config  *Config
+	logger  *zap.Logger
+	client  *http.Client
+	wg      sync.WaitGroup
+	convert func(pdata.LogRecord, pdata.Resource) (*logproto.Entry, error)
 }
 
 func newExporter(config *Config, logger *zap.Logger) *lokiExporter {
-	return &lokiExporter{
+	lokiexporter := &lokiExporter{
 		config: config,
 		logger: logger,
 	}
+	if config.Format == "json" {
+		lokiexporter.convert = convertLogToJSONEntry
+	} else {
+		lokiexporter.convert = convertLogToLokiEntry
+	}
+	return lokiexporter
 }
 
 func (l *lokiExporter) pushLogData(ctx context.Context, ld pdata.Logs) error {
@@ -135,7 +142,15 @@ func (l *lokiExporter) logDataToLoki(ld pdata.Logs) (pr *logproto.PushRequest, n
 					continue
 				}
 				labels := mergedLabels.String()
-				entry := convertLogToLokiEntry(log)
+				var entry *logproto.Entry
+				var err error
+				entry, err = l.convert(log, resource)
+				if err != nil {
+					// Couldn't convert to JSON so dropping log.
+					numDroppedLogs++
+					l.logger.Error("Failed to convert to JSON - Dropping Log", zap.Error(err))
+					continue
+				}
 
 				if stream, ok := streams[labels]; ok {
 					stream.Entries = append(stream.Entries, *entry)
@@ -195,9 +210,20 @@ func (l *lokiExporter) convertAttributesToLabels(attributes pdata.AttributeMap,
 	return ls
 }
 
-func convertLogToLokiEntry(lr pdata.LogRecord) *logproto.Entry {
+func convertLogToLokiEntry(lr pdata.LogRecord, res pdata.Resource) (*logproto.Entry, error) {
 	return &logproto.Entry{
 		Timestamp: time.Unix(0, int64(lr.Timestamp())),
 		Line:      lr.Body().StringVal(),
+	}, nil
+}
+
+func convertLogToJSONEntry(lr pdata.LogRecord, res pdata.Resource) (*logproto.Entry, error) {
+	line, err := encodeJSON(lr, res)
+	if err != nil {
+		return nil, err
 	}
+	return &logproto.Entry{
+		Timestamp: time.Unix(0, int64(lr.Timestamp())),
+		Line:      line,
+	}, nil
 }

exporter/lokiexporter/exporter_test.go

@@ -477,8 +477,10 @@ func TestExporter_convertLogToLokiEntry(t *testing.T) {
 	lr := pdata.NewLogRecord()
 	lr.Body().SetStringVal("log message")
 	lr.SetTimestamp(ts)
+	res := pdata.NewResource()
+	res.Attributes().Insert("host.name", pdata.NewAttributeValueString("something"))
 
-	entry := convertLogToLokiEntry(lr)
+	entry, _ := convertLogToLokiEntry(lr, res)
 
 	expEntry := &logproto.Entry{
 		Timestamp: time.Unix(0, int64(lr.Timestamp())),
@@ -576,3 +578,21 @@ func TestExporter_stopAlwaysReturnsNil(t *testing.T) {
 	require.NotNil(t, exp)
 	require.NoError(t, exp.stop(context.Background()))
 }
+
+func TestExporter_convertLogtoJSONEntry(t *testing.T) {
+	ts := pdata.Timestamp(int64(1) * time.Millisecond.Nanoseconds())
+	lr := pdata.NewLogRecord()
+	lr.Body().SetStringVal("log message")
+	lr.SetTimestamp(ts)
+	res := pdata.NewResource()
+	res.Attributes().Insert("host.name", pdata.NewAttributeValueString("something"))
+
+	entry, err := convertLogToJSONEntry(lr, res)
+	expEntry := &logproto.Entry{
+		Timestamp: time.Unix(0, int64(lr.Timestamp())),
+		Line:      `{"body":"log message","resources":{"host.name":"something"}}`,
+	}
+	require.Nil(t, err)
+	require.NotNil(t, entry)
+	require.Equal(t, expEntry, entry)
+}

exporter/lokiexporter/factory.go

@@ -48,6 +48,7 @@ func createDefaultConfig() config.Exporter {
 		RetrySettings: exporterhelper.DefaultRetrySettings(),
 		QueueSettings: exporterhelper.DefaultQueueSettings(),
 		TenantID:      "",
+		Format:        "body",
 		Labels: LabelsConfig{
 			Attributes:         map[string]string{},
 			ResourceAttributes: map[string]string{},

exporter/lokiexporter/testdata/config.yaml

@@ -13,6 +13,10 @@ exporters:
         container.name: "container_name"
         k8s.cluster.name: "k8s_cluster_name"
         severity: "severity"
+  loki/json:
+    endpoint: "https://loki:3100/loki/api/v1/push"
+    tenant_id: "example"
+    format: "json"
   loki/allsettings:
     endpoint: "https://loki:3100/loki/api/v1/push"
     tenant_id: "example"
@@ -48,4 +52,4 @@ service:
     logs:
       receivers: [ nop ]
       processors: [ nop ]
-      exporters: [ loki, loki/allsettings ]
+      exporters: [ loki, loki/allsettings, loki/json ]