(otelarrowreceiver): asynchronous stream operations

CHANGELOG.md

@@ -8,7 +8,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## Unreleased
 
-- Added a semaphore package to limit bytes admitted and total number of waiters. [#174](https://github.com/open-telemetry/otel-arrow/pull/174)
+- Refactor otelarrowreceiver to do stream.Recv, request processing, and stream.Send in separate goroutines. [#181](https://github.com/open-telemetry/otel-arrow/pull/181)
+
+- Add a semaphore package to limit bytes admitted and total number of waiters. [#174](https://github.com/open-telemetry/otel-arrow/pull/174)
 
 ## [0.22.0](https://github.com/open-telemetry/otel-arrow/releases/tag/v0.22.0) - 2024-04-16
 

collector/examples/bridge/edge-collector.yaml

@@ -49,4 +49,4 @@ service:
       address: 127.0.0.1:8888
       level: detailed
     logs:
-      level: info
+      level: info

collector/examples/bridge/saas-collector.yaml

@@ -57,4 +57,4 @@ service:
   telemetry:
     metrics:
       address: 127.0.0.1:8889
-      level: normal
+      level: normal

collector/exporter/otelarrowexporter/go.mod

@@ -16,7 +16,6 @@ require (
 	go.opentelemetry.io/collector/config/configgrpc v0.98.0
 	go.opentelemetry.io/collector/config/configopaque v1.5.0
 	go.opentelemetry.io/collector/config/configretry v0.98.0
-	go.opentelemetry.io/collector/config/configtelemetry v0.98.0
 	go.opentelemetry.io/collector/config/configtls v0.98.0
 	go.opentelemetry.io/collector/confmap v0.98.0
 	go.opentelemetry.io/collector/consumer v0.98.0
@@ -75,6 +74,7 @@ require (
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.opentelemetry.io/collector/config/confignet v0.98.0 // indirect
+	go.opentelemetry.io/collector/config/configtelemetry v0.98.0 // indirect
 	go.opentelemetry.io/collector/config/internal v0.98.0 // indirect
 	go.opentelemetry.io/collector/featuregate v1.5.0 // indirect
 	go.opentelemetry.io/collector/receiver v0.98.0 // indirect

collector/exporter/otelarrowexporter/internal/arrow/exporter.go

@@ -7,14 +7,14 @@ import (
 	"context"
 	"errors"
 	"math/rand"
+	"strconv"
 	"sync"
 	"time"
 
 	arrowpb "github.com/open-telemetry/otel-arrow/api/experimental/arrow/v1"
 	"github.com/open-telemetry/otel-arrow/collector/netstats"
 	arrowRecord "github.com/open-telemetry/otel-arrow/pkg/otel/arrow_record"
 	"go.opentelemetry.io/collector/component"
-	"go.opentelemetry.io/collector/config/configtelemetry"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
@@ -285,19 +285,22 @@ func (e *Exporter) SendAndWait(ctx context.Context, data any) (bool, error) {
 	// exporter, because of the optimization phase performed in the
 	// conversion to Arrow.
 	var uncompSize int
-	if e.telemetry.MetricsLevel > configtelemetry.LevelNormal {
-		switch data := data.(type) {
-		case ptrace.Traces:
-			var sizer ptrace.ProtoMarshaler
-			uncompSize = sizer.TracesSize(data)
-		case plog.Logs:
-			var sizer plog.ProtoMarshaler
-			uncompSize = sizer.LogsSize(data)
-		case pmetric.Metrics:
-			var sizer pmetric.ProtoMarshaler
-			uncompSize = sizer.MetricsSize(data)
-		}
+	switch data := data.(type) {
+	case ptrace.Traces:
+		var sizer ptrace.ProtoMarshaler
+		uncompSize = sizer.TracesSize(data)
+	case plog.Logs:
+		var sizer plog.ProtoMarshaler
+		uncompSize = sizer.LogsSize(data)
+	case pmetric.Metrics:
+		var sizer pmetric.ProtoMarshaler
+		uncompSize = sizer.MetricsSize(data)
+	}
+
+	if md == nil {
+		md = make(map[string]string)
 	}
+	md["otlp-pdata-size"] = strconv.Itoa(uncompSize)
 
 	wri := writeItem{
 		records:     data,

collector/exporter/otelarrowexporter/internal/arrow/exporter_test.go

@@ -604,10 +604,13 @@ func TestArrowExporterHeaders(t *testing.T) {
 			md := metadata.MD{
 				"expected1": []string{"metadata1"},
 				"expected2": []string{fmt.Sprint(times)},
+				"otlp-pdata-size": []string{"329"},
 			}
 			expectOutput = append(expectOutput, md)
 		} else {
-			expectOutput = append(expectOutput, nil)
+			expectOutput = append(expectOutput, metadata.MD{
+				"otlp-pdata-size": []string{"329"},
+			})
 		}
 
 		sent, err := tc.exporter.SendAndWait(ctx, input)
@@ -678,10 +681,13 @@ func TestArrowExporterIsTraced(t *testing.T) {
 
 					md := metadata.MD{
 						"traceparent": []string{expectMap["traceparent"]},
+						"otlp-pdata-size": []string{"329"},
 					}
 					expectOutput = append(expectOutput, md)
 				} else {
-					expectOutput = append(expectOutput, nil)
+					expectOutput = append(expectOutput, metadata.MD{
+						"otlp-pdata-size": []string{"329"},
+					})
 				}
 
 				sent, err := tc.exporter.SendAndWait(ctx, input)

collector/go.mod

@@ -32,6 +32,7 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect

collector/go.sum

@@ -18,6 +18,8 @@ github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 h1:TQcrn6Wq+sKGkpyPvppOz99zsM
 github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1/go.mod h1:oJDH3BJKyqBA2TXFhDsKDGDTlndYOZ6rGS0BRZIxGhM=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
+github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
+github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=

collector/receiver/otelarrowreceiver/config.go

@@ -24,6 +24,13 @@ type ArrowConfig struct {
 	// passing through, they will see ResourceExhausted errors.
 	MemoryLimitMiB uint64 `mapstructure:"memory_limit_mib"`
 
+	AdmissionLimitMiB uint64 `mapstructure:"admission_limit_mib"`
+
+	// WaiterLimit is the limit on the number of waiters waiting to be processed and consumed.
+	// This is a dimension of memory limiting to ensure waiters are not consuming an
+	// unexpectedly large amount of memory in the arrow receiver.
+	WaiterLimit int64 `mapstructure:"waiter_limit"`
+
 	// Zstd settings apply to OTel-Arrow use of gRPC specifically.
 	Zstd zstd.DecoderConfig `mapstructure:"zstd"`
 }

collector/receiver/otelarrowreceiver/config_test.go

@@ -78,6 +78,7 @@ func TestUnmarshalConfig(t *testing.T) {
 				},
 				Arrow: ArrowConfig{
 					MemoryLimitMiB: 123,
+					AdmissionLimitMiB: 80,
 				},
 			},
 		}, cfg)
@@ -102,6 +103,7 @@ func TestUnmarshalConfigUnix(t *testing.T) {
 				},
 				Arrow: ArrowConfig{
 					MemoryLimitMiB: defaultMemoryLimitMiB,
+					AdmissionLimitMiB: defaultAdmissionLimitMiB,
 				},
 			},
 		}, cfg)

collector/receiver/otelarrowreceiver/factory.go

@@ -20,6 +20,7 @@ const (
 	defaultGRPCEndpoint = "0.0.0.0:4317"
 
 	defaultMemoryLimitMiB = 128
+	defaultAdmissionLimitMiB = defaultMemoryLimitMiB / 2
 )
 
 // NewFactory creates a new OTLP receiver factory.
@@ -46,6 +47,7 @@ func createDefaultConfig() component.Config {
 			},
 			Arrow: ArrowConfig{
 				MemoryLimitMiB: defaultMemoryLimitMiB,
+				AdmissionLimitMiB: defaultAdmissionLimitMiB,
 			},
 		},
 	}

collector/receiver/otelarrowreceiver/go.mod

@@ -34,7 +34,9 @@ require (
 	github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
 	github.com/apache/arrow/go/v14 v14.0.2 // indirect
 	github.com/axiomhq/hyperloglog v0.0.0-20230201085229-3ddf4bad03dc // indirect
+	github.com/bahlo/generic-list-go v0.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/buger/jsonparser v1.1.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/dgryski/go-metro v0.0.0-20180109044635-280f6062b5bc // indirect
@@ -55,6 +57,7 @@ require (
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
 	github.com/knadh/koanf/v2 v2.1.1 // indirect
+	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
@@ -66,6 +69,7 @@ require (
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.48.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
+	github.com/wk8/go-ordered-map/v2 v2.1.8 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	github.com/zeebo/xxh3 v1.0.2 // indirect
 	go.opentelemetry.io/collector/config/configcompression v1.5.0 // indirect

collector/receiver/otelarrowreceiver/go.sum

@@ -85,7 +85,6 @@ github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLA
 github.com/open-telemetry/otel-arrow v0.22.0 h1:G1jgtqAM2ho5pyKQ4tyrDzk9Y0VcJ+GZQRJgN26vRlI=
 github.com/open-telemetry/otel-arrow v0.22.0/go.mod h1:F50XFaiNfkfB0MYftZIUKFULm6pxfGqjbgQzevi+65M=
 github.com/open-telemetry/otel-arrow/collector v0.22.0 h1:lHFjzkh5PbsiW8B63SRntnP9W7bLCXV9lslO4zI0s/Y=
-github.com/open-telemetry/otel-arrow/collector v0.22.0/go.mod h1:R7hRwuGDxoGLB27dkJUFKDK7mGG7Yb02ODnLHx8Whis=
 github.com/pierrec/lz4/v4 v4.1.18 h1:xaKrnTkyoqfh1YItXl56+6KJNVYWlEEPuAQW9xsplYQ=
 github.com/pierrec/lz4/v4 v4.1.18/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=