Fluent Forward Receiver

exporter/exportertest/sink_exporter.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 
 	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumerdata"
 	"go.opentelemetry.io/collector/consumer/pdata"
 	"go.opentelemetry.io/collector/consumer/pdatautil"
@@ -234,14 +235,16 @@ func (sme *SinkMetricsExporter) Shutdown(context.Context) error {
 	return nil
 }
 
-// SinkLogExporter acts as a metrics receiver for use in tests.
+// SinkLogExporter acts as a logs receiver for use in tests.
 type SinkLogExporter struct {
 	consumeLogError error // to be returned by ConsumeLog, if set
 	mu              sync.Mutex
 	logs            []data.Logs
 	logRecordsCount int
 }
 
+var _ consumer.LogConsumer = new(SinkLogExporter)
+
 // SetConsumeLogError sets an error that will be returned by ConsumeLog
 func (sle *SinkLogExporter) SetConsumeLogError(err error) {
 	sle.mu.Lock()
@@ -287,6 +290,14 @@ func (sle *SinkLogExporter) LogRecordsCount() int {
 	return sle.logRecordsCount
 }
 
+// Reset deletes any existing logs.
+func (sle *SinkLogExporter) Reset() {
+	sle.mu.Lock()
+	defer sle.mu.Unlock()
+
+	sle.logs = nil
+}
+
 // Shutdown stops the exporter and is invoked during shutdown.
 func (sle *SinkLogExporter) Shutdown(context.Context) error {
 	return nil

exporter/fileexporter/file_exporter_test.go

@@ -14,10 +14,8 @@
 package fileexporter
 
 import (
-	"bytes"
 	"context"
 	"encoding/json"
-	"errors"
 	"strconv"
 	"testing"
 	"time"
@@ -33,26 +31,11 @@ import (
 	otlpcommon "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/common/v1"
 	logspb "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/logs/v1"
 	otresourcepb "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/resource/v1"
+	"go.opentelemetry.io/collector/testutil"
 )
 
-type mockFile struct {
-	buf    bytes.Buffer
-	maxLen int
-}
-
-func (mf *mockFile) Write(p []byte) (n int, err error) {
-	if mf.maxLen != 0 && len(p)+mf.buf.Len() > mf.maxLen {
-		return 0, errors.New("buffer would be filled by write")
-	}
-	return mf.buf.Write(p)
-}
-
-func (mf *mockFile) Close() error {
-	return nil
-}
-
 func TestFileTraceExporterNoErrors(t *testing.T) {
-	mf := &mockFile{}
+	mf := &testutil.LimitedWriter{}
 	lte := &Exporter{file: mf}
 	require.NotNil(t, lte)
 
@@ -78,7 +61,7 @@ func TestFileTraceExporterNoErrors(t *testing.T) {
 	assert.NoError(t, lte.Shutdown(context.Background()))
 
 	var j map[string]interface{}
-	assert.NoError(t, json.Unmarshal(mf.buf.Bytes(), &j))
+	assert.NoError(t, json.Unmarshal(mf.Bytes(), &j))
 
 	assert.EqualValues(t, j,
 		map[string]interface{}{
@@ -102,7 +85,7 @@ func TestFileTraceExporterNoErrors(t *testing.T) {
 }
 
 func TestFileMetricsExporterNoErrors(t *testing.T) {
-	mf := &mockFile{}
+	mf := &testutil.LimitedWriter{}
 	lme := &Exporter{file: mf}
 	require.NotNil(t, lme)
 
@@ -132,7 +115,7 @@ func TestFileMetricsExporterNoErrors(t *testing.T) {
 	assert.NoError(t, lme.Shutdown(context.Background()))
 
 	var j map[string]interface{}
-	assert.NoError(t, json.Unmarshal(mf.buf.Bytes(), &j))
+	assert.NoError(t, json.Unmarshal(mf.Bytes(), &j))
 
 	assert.EqualValues(t, j,
 		map[string]interface{}{
@@ -162,7 +145,7 @@ func TestFileMetricsExporterNoErrors(t *testing.T) {
 }
 
 func TestFileLogsExporterNoErrors(t *testing.T) {
-	mf := &mockFile{}
+	mf := &testutil.LimitedWriter{}
 	exporter := &Exporter{file: mf}
 	require.NotNil(t, exporter)
 
@@ -216,7 +199,7 @@ func TestFileLogsExporterNoErrors(t *testing.T) {
 	assert.NoError(t, exporter.ConsumeLogs(context.Background(), data.LogsFromProto(ld)))
 	assert.NoError(t, exporter.Shutdown(context.Background()))
 
-	decoder := json.NewDecoder(&mf.buf)
+	decoder := json.NewDecoder(mf)
 	var j map[string]interface{}
 	assert.NoError(t, decoder.Decode(&j))
 
@@ -342,8 +325,8 @@ func TestFileLogsExporterErrors(t *testing.T) {
 	for i := range cases {
 		maxLen := cases[i].MaxLen
 		t.Run(cases[i].Name, func(t *testing.T) {
-			mf := &mockFile{
-				maxLen: maxLen,
+			mf := &testutil.LimitedWriter{
+				MaxLen: maxLen,
 			}
 			exporter := &Exporter{file: mf}
 			require.NotNil(t, exporter)

go.mod

@@ -31,6 +31,7 @@ require (
 	github.com/ory/go-acc v0.2.1
 	github.com/ory/x v0.0.109 // indirect
 	github.com/pavius/impi v0.0.3
+	github.com/philhofer/fwd v1.0.0 // indirect
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.5.1
 	github.com/prometheus/common v0.10.0
@@ -45,6 +46,7 @@ require (
 	github.com/spf13/viper v1.7.0
 	github.com/stretchr/testify v1.6.1
 	github.com/tcnksm/ghr v0.13.0
+	github.com/tinylib/msgp v1.1.2
 	github.com/uber/jaeger-lib v2.2.0+incompatible
 	go.opencensus.io v0.22.3
 	go.uber.org/atomic v1.6.0

go.sum

@@ -1053,6 +1053,8 @@ github.com/performancecopilot/speed v3.0.0+incompatible/go.mod h1:/CLtqpZ5gBg1M9
 github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU=
 github.com/phayes/checkstyle v0.0.0-20170904204023-bfd46e6a821d h1:CdDQnGF8Nq9ocOS/xlSptM1N3BbrA6/kmaep5ggwaIA=
 github.com/phayes/checkstyle v0.0.0-20170904204023-bfd46e6a821d/go.mod h1:3OzsM7FXDQlpCiw2j81fOmAwQLnZnLGXVKUzeKQXIAw=
+github.com/philhofer/fwd v1.0.0 h1:UbZqGr5Y38ApvM/V/jEljVxwocdweyH+vmYvRPBnbqQ=
+github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU=
 github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v1.0.2-0.20190131084431-473cd7ce01a1/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
 github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY=
@@ -1167,6 +1169,7 @@ github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAm
 github.com/shirou/gopsutil v0.0.0-20190901111213-e4ec7b275ada/go.mod h1:WWnYX4lzhCH5h/3YBfyVA3VbLYjlMZZAQcW9ojMexNc=
 github.com/shirou/gopsutil v0.0.0-20200517204708-c89193f22d93 h1:+ZhxoIovCjs+mkd0pCBqczqvx/vl+emW8x04WM15Y7M=
 github.com/shirou/gopsutil v0.0.0-20200517204708-c89193f22d93/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
+github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 h1:udFKJ0aHUL60LboW/A+DfgoHVedieIzIXE8uylPue0U=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc=
 github.com/shopspring/decimal v0.0.0-20180709203117-cd690d0c9e24/go.mod h1:M+9NzErvs504Cn4c5DxATwIqPbtswREoFCre64PpcG4=
 github.com/shurcooL/go v0.0.0-20180423040247-9e1955d9fb6e h1:MZM7FHLqUHYI0Y/mQAt3d2aYa0SiNms/hFqC9qJYolM=
@@ -1275,6 +1278,8 @@ github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhV
 github.com/tidwall/sjson v1.0.4/go.mod h1:bURseu1nuBkFpIES5cz6zBtjmYeOQmEESshn7VpF15Y=
 github.com/timakin/bodyclose v0.0.0-20190930140734-f7f2e9bca95e h1:RumXZ56IrCj4CL+g1b9OL/oH0QnsF976bC8xQFYUD5Q=
 github.com/timakin/bodyclose v0.0.0-20190930140734-f7f2e9bca95e/go.mod h1:Qimiffbc6q9tBWlVV6x0P9sat/ao1xEkREYPPj9hphk=
+github.com/tinylib/msgp v1.1.2 h1:gWmO7n0Ys2RBEb7GPYB9Ujq8Mk5p2U08lRnmMcGy6BQ=
+github.com/tinylib/msgp v1.1.2/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20170815181823-89b8d40f7ca8/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tmc/grpc-websocket-proxy v0.0.0-20190109142713-0ad062ec5ee5/go.mod h1:ncp9v5uamzpCO7NfCPTXjqaC+bZgJeR0sMTm6dMHP7U=
 github.com/tommy-muehle/go-mnd v1.3.1-0.20200224220436-e6f9a994e8fa h1:RC4maTWLKKwb7p1cnoygsbKIgNlJqSYBeAFON3Ar8As=

internal/data/common.go

@@ -0,0 +1,33 @@
+// Copyright 2020 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 data
+
+import commonproto "go.opentelemetry.io/collector/internal/data/opentelemetry-proto-gen/common/v1"
+
+func NewStringValue(s string) *commonproto.AnyValue {
+	return &commonproto.AnyValue{Value: &commonproto.AnyValue_StringValue{StringValue: s}}
+}
+
+func NewIntValue(i int64) *commonproto.AnyValue {
+	return &commonproto.AnyValue{Value: &commonproto.AnyValue_IntValue{IntValue: i}}
+}
+
+func NewDoubleValue(d float64) *commonproto.AnyValue {
+	return &commonproto.AnyValue{Value: &commonproto.AnyValue_DoubleValue{DoubleValue: d}}
+}
+
+func NewBoolValue(b bool) *commonproto.AnyValue {
+	return &commonproto.AnyValue{Value: &commonproto.AnyValue_BoolValue{BoolValue: b}}
+}

internal/data/common_test.go

@@ -0,0 +1,28 @@
+// Copyright 2020 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 data
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAnyValueHelpers(t *testing.T) {
+	require.NotNil(t, NewStringValue("test"))
+	require.NotNil(t, NewIntValue(1))
+	require.NotNil(t, NewDoubleValue(1.0))
+	require.NotNil(t, NewBoolValue(true))
+}

receiver/fluentforwardreceiver/README.md

@@ -0,0 +1,33 @@
+# Fluent Forward Receiver
+
+This receiver runs a TCP server that accepts events via the [Fluent Forward
+protocol](https://github.com/fluent/fluentd/wiki/Forward-Protocol-Specification-v1).
+
+This receiver:
+
+ - Does **not** support TLS or the handshake portion of the Forward protocol.
+ - Does support acknowledgments of events that have the `chunk` option, as per the spec.
+ - Supports all three event types (message, forward, packed forward, including
+   compressed packed forward)
+ - Supports listening on a Unix domain socket by making the `listenAddress`
+   option of the form `unix://<path to socket>`.
+ - If using TCP, it will start a UDP server on the same port to deliver
+   heartbeat echos, as per the spec.
+
+Here is a basic example config that makes the receiver listen on all interfaces
+on port 8006:
+
+```yaml
+receivers:
+  fluentforward:
+    listenAddress: 0.0.0.0:8006
+```
+
+
+## Development
+
+If you are working on this receiver and need to regenerate any of the message
+pack autogenerated code, just run `go generate` on this package and its
+subpackages.  You can get the `msgp` binary by just running `go get -u -t
+github.com/tinylib/msgp`, and make sure the Go binary path is on your shell's
+PATH.

receiver/fluentforwardreceiver/ack.go

@@ -0,0 +1,36 @@
+// Copyright 2020 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 fluentforwardreceiver
+
+import "github.com/tinylib/msgp/msgp"
+
+type AckResponse struct {
+	Ack string `msg:"ack"`
+}
+
+func (z AckResponse) EncodeMsg(en *msgp.Writer) error {
+	// map header, size 1
+	// write "ack"
+	err := en.Append(0x81, 0xa3, 0x61, 0x63, 0x6b)
+	if err != nil {
+		return err
+	}
+
+	err = en.WriteString(z.Ack)
+	if err != nil {
+		return msgp.WrapError(err, "Ack")
+	}
+	return nil
+}

receiver/fluentforwardreceiver/ack_test.go

@@ -0,0 +1,51 @@
+// Copyright 2020 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 fluentforwardreceiver
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/tinylib/msgp/msgp"
+
+	"go.opentelemetry.io/collector/testutil"
+)
+
+func msgpWriterWithLimit(l int) *msgp.Writer {
+	// NewWriterSize forces size to be at least 18 bytes so just use that as
+	// the floor and write nulls to those first 18 bytes to make the limit
+	// truly l.
+	w := msgp.NewWriterSize(&testutil.LimitedWriter{
+		MaxLen: l,
+	}, 18+l)
+	w.Write(bytes.Repeat([]byte{0x00}, 18))
+	return w
+}
+
+func TestAckEncoding(t *testing.T) {
+	a := &AckResponse{
+		Ack: "test",
+	}
+
+	err := a.EncodeMsg(msgpWriterWithLimit(1000))
+	require.Nil(t, err)
+
+	err = a.EncodeMsg(msgpWriterWithLimit(4))
+	require.NotNil(t, err)
+
+	err = a.EncodeMsg(msgpWriterWithLimit(7))
+	require.NotNil(t, err)
+}