[exporter/mezmoexporter] Add the Mezmo Log exporter

.github/CODEOWNERS

@@ -43,6 +43,7 @@ exporter/kafkaexporter/                              @open-telemetry/collector-c
 exporter/loadbalancingexporter/                      @open-telemetry/collector-contrib-approvers @jpkrohling
 exporter/logzioexporter/                             @open-telemetry/collector-contrib-approvers @jkowall @Doron-Bargo @yotamloe
 exporter/lokiexporter/                               @open-telemetry/collector-contrib-approvers @gramidt @jpkrohling
+exporter/mezmoexporter/                              @open-telemetry/collector-contrib-approvers @dashpole @billmeyer @gjanco
 exporter/newrelicexporter/                           @open-telemetry/collector-contrib-approvers @alanwest @jack-berg @nrcventura
 exporter/observiqexporter/                           @open-telemetry/collector-contrib-approvers @binaryfissiongames
 exporter/prometheusexporter/                         @open-telemetry/collector-contrib-approvers @Aneurysm9

exporter/mezmoexporter/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

exporter/mezmoexporter/README.md

@@ -0,0 +1,30 @@
+# Mezmo Exporter
+
+This exporter supports sending OpenTelemetry log data to [LogDNA (Mezmo)](https://logdna.com).
+
+# Configuration options:
+
+- `ingest_url` (required): Use the default URL of `https://logs.logdna.com/log/ingest` unless a custom URL has been provided for your account.
+- `ingest_key` (required): Ingestion key used to send log data to LogDNA.  See [Ingestion Keys](https://docs.logdna.com/docs/ingestion-key) for more details.
+
+# Example:
+## Simple Log Data
+
+```yaml
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: ":4317"
+
+exporters:
+  mezmo:
+    ingest_url = "https://logs.logdna.com/log/ingest"
+    ingest_key = "00000000000000000000000000000000"
+
+service:
+  pipelines:
+    logs:
+      receivers: [ otlp ]
+      exporters: [ mezmo ]
+```

exporter/mezmoexporter/config.go

@@ -0,0 +1,79 @@
+// 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 mezmoexporter
+
+import (
+	"fmt"
+	"net/url"
+	"time"
+
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/config/confighttp"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+)
+
+const (
+	// defaultTimeout
+	defaultTimeout time.Duration = 5 * time.Second
+
+	// defaultIngestURL
+	defaultIngestURL = "https://logs.logdna.com/log/ingest"
+
+	// See https://docs.logdna.com/docs/ingestion#service-limits for details
+
+	// Maximum payload in bytes that can be POST'd to the REST endpoint
+	maxBodySize     = 10 * 1024 * 1024
+	maxMessageSize  = 16 * 1024
+	maxMetaDataSize = 32 * 1024
+	maxAppnameLen   = 512
+	maxLogLevelLen  = 80
+)
+
+// Config defines configuration for Mezmo exporter.
+type Config struct {
+	config.ExporterSettings       `mapstructure:",squash"`
+	confighttp.HTTPClientSettings `mapstructure:",squash"` // squash ensures fields are correctly decoded in embedded struct.
+	exporterhelper.QueueSettings  `mapstructure:"sending_queue"`
+	exporterhelper.RetrySettings  `mapstructure:"retry_on_failure"`
+
+	// IngestURL is the URL to send telemetry to.
+	IngestURL string `mapstructure:"ingest_url"`
+
+	// Token is the authentication token provided by Mezmo.
+	IngestKey string `mapstructure:"ingest_key"`
+}
+
+// returns default http client settings
+func createDefaultHTTPClientSettings() confighttp.HTTPClientSettings {
+	return confighttp.HTTPClientSettings{
+		Timeout: defaultTimeout,
+	}
+}
+
+func (c *Config) Validate() error {
+	var err error
+	var parsed *url.URL
+
+	parsed, err = url.Parse(c.IngestURL)
+	if c.IngestURL == "" || err != nil {
+		return fmt.Errorf("\"ingest_url\" must be a valid URL")
+	}
+
+	if parsed.Host == "" {
+		return fmt.Errorf("\"ingest_url\" must contain a valid host")
+	}
+
+	return nil
+}

exporter/mezmoexporter/config_test.go

@@ -0,0 +1,47 @@
+// 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 mezmoexporter
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/config"
+	"go.opentelemetry.io/collector/service/servicetest"
+)
+
+func TestLoadConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.Nil(t, err)
+
+	factory := NewFactory()
+	factories.Exporters[typeStr] = factory
+	cfg, err := servicetest.LoadConfigAndValidate(filepath.Join("testdata", "config.yaml"), factories)
+
+	require.NoError(t, err)
+	require.NotNil(t, cfg)
+
+	loadedCfg := cfg.Exporters[config.NewComponentID(typeStr)]
+
+	// IngestURL doesn't have a default value so set it directly.
+	expectedCfg := factory.CreateDefaultConfig().(*Config)
+	expectedCfg.IngestURL = defaultIngestURL
+	expectedCfg.IngestKey = "00000000000000000000000000000000"
+
+	assert.Equal(t, expectedCfg, loadedCfg)
+}

exporter/mezmoexporter/doc.go

@@ -0,0 +1,16 @@
+// Copyright 2020, 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 mezmoexporter implements an exporter that sends data to Mezmo.
+package mezmoexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/mezmoexporter"

exporter/mezmoexporter/example/otel-collector-config.yaml

@@ -0,0 +1,17 @@
+receivers:
+  nop:
+
+processors:
+  nop:
+
+exporters:
+  mezmo:
+    ingest_url: "https://logs.logdna.com/log/ingest"
+    ingest_key: "00000000000000000000000000000000"
+
+service:
+  pipelines:
+    logs:
+      receivers: [ nop ]
+      processors: [ nop ]
+      exporters: [ mezmo ]

exporter/mezmoexporter/exporter.go

@@ -0,0 +1,166 @@
+// 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 mezmoexporter
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"sync"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/plog"
+)
+
+type mezmoExporter struct {
+	config   *Config
+	settings component.TelemetrySettings
+	client   *http.Client
+	wg       sync.WaitGroup
+}
+
+type MezmoLogLine struct {
+	Timestamp int64             `json:"timestamp"`
+	Line      string            `json:"line"`
+	App       string            `json:"app"`
+	Level     string            `json:"level"`
+	Meta      map[string]string `json:"meta"`
+}
+
+type MezmoLogBody struct {
+	Lines []MezmoLogLine `json:"lines"`
+}
+
+func newLogsExporter(config *Config, settings component.TelemetrySettings) *mezmoExporter {
+	var e = &mezmoExporter{
+		config:   config,
+		settings: settings,
+	}
+	return e
+}
+
+func (m *mezmoExporter) pushLogData(ctx context.Context, ld plog.Logs) error {
+	m.wg.Add(1)
+	defer m.wg.Done()
+
+	return m.logDataToMezmo(ld)
+}
+
+func (m *mezmoExporter) start(_ context.Context, host component.Host) (err error) {
+	m.client, err = m.config.HTTPClientSettings.ToClient(host.GetExtensions(), m.settings)
+	return err
+}
+
+func (m *mezmoExporter) stop(context.Context) (err error) {
+	m.wg.Wait()
+	return nil
+}
+
+func (m *mezmoExporter) logDataToMezmo(ld plog.Logs) error {
+	var errs error
+
+	var lines []MezmoLogLine
+
+	// Convert the log resources to mezmo lines...
+	resourceLogs := ld.ResourceLogs()
+	for i := 0; i < resourceLogs.Len(); i++ {
+		ills := resourceLogs.At(i).ScopeLogs()
+		for j := 0; j < ills.Len(); j++ {
+			logs := ills.At(j).LogRecords()
+
+			for k := 0; k < logs.Len(); k++ {
+				log := logs.At(k)
+
+				// Convert Attributes to meta fields being mindful of the maxMetaDataSize restriction
+				attrs := map[string]string{}
+				attrs["trace.id"] = log.TraceID().HexString()
+				attrs["span.id"] = log.SpanID().HexString()
+				log.Attributes().Range(func(k string, v pcommon.Value) bool {
+					attrs[k] = truncateString(v.StringVal(), maxMetaDataSize)
+					return true
+				})
+
+				s, _ := log.Attributes().Get("appname")
+				app := s.StringVal()
+
+				line := MezmoLogLine{
+					Timestamp: log.Timestamp().AsTime().UTC().UnixMilli(),
+					Line:      truncateString(log.Body().StringVal(), maxMessageSize),
+					App:       truncateString(app, maxAppnameLen),
+					Level:     truncateString(log.SeverityText(), maxLogLevelLen),
+					Meta:      attrs,
+				}
+				lines = append(lines, line)
+			}
+		}
+	}
+
+	// Send them to Mezmo in batches < 10MB in size
+	var b strings.Builder
+	b.WriteString("{\"lines\": [")
+
+	var lineBytes []byte
+	for i, line := range lines {
+		if lineBytes, errs = json.Marshal(line); errs != nil {
+			return fmt.Errorf("error Creating JSON payload: %s", errs)
+		}
+
+		var newBufSize = b.Len() + len(lineBytes)
+		if newBufSize >= maxBodySize-2 {
+			str := b.String()
+			str = str[:len(str)-1] + "]}"
+			if errs = m.sendLinesToMezmo(str); errs != nil {
+				return errs
+			}
+			b.Reset()
+			b.WriteString("{\"lines\": [")
+		}
+
+		b.Write(lineBytes)
+
+		if i < len(lines)-1 {
+			b.WriteRune(',')
+		}
+	}
+
+	str := b.String() + "]}"
+	if errs = m.sendLinesToMezmo(str); errs != nil {
+		return errs
+	}
+
+	return nil
+}
+
+func (m *mezmoExporter) sendLinesToMezmo(post string) (errs error) {
+	var hostname = "otel"
+
+	url := fmt.Sprintf("%s?hostname=%s", m.config.IngestURL, hostname)
+
+	req, _ := http.NewRequest("POST", url, bytes.NewBuffer([]byte(post)))
+	req.Header.Add("Accept", "application/json")
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("apikey", m.config.IngestKey)
+
+	var res *http.Response
+	if res, errs = http.DefaultClient.Do(req); errs != nil {
+		return fmt.Errorf("failed to POST log to Mezmo: %s", errs)
+	}
+
+	return res.Body.Close()
+}