support deprecated option + changelog + unit tests

.chloggen/admission-control-for-otlp.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: otelarrowreceiver
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add admission control in the otelarrow receiver's standard otlp data path. 
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [35021]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

receiver/otelarrowreceiver/config.go

@@ -14,9 +14,9 @@ import (
 
 // Protocols is the configuration for the supported protocols.
 type Protocols struct {
-	GRPC      configgrpc.ServerConfig `mapstructure:"grpc"`
-	Arrow     ArrowConfig             `mapstructure:"arrow"`
-	Admission AdmissionConfig         `mapstructure:"admission"`
+	GRPC  configgrpc.ServerConfig `mapstructure:"grpc"`
+	Arrow ArrowConfig             `mapstructure:"arrow"`
+	Admission AdmissionConfig     `mapstructure:"admission"`
 }
 
 type AdmissionConfig struct {
@@ -38,6 +38,12 @@ type ArrowConfig struct {
 	// passing through, they will see ResourceExhausted errors.
 	MemoryLimitMiB uint64 `mapstructure:"memory_limit_mib"`
 
+	// Deprecated: This field is no longer supported, use cfg.Admission.AdmissionLimitMiB instead.
+	AdmissionLimitMiB uint64 `mapstructure:"admission_limit_mib"`
+
+	// Deprecated: This field is no longer supported, use cfg.Admission.WaiterLimit instead.
+	WaiterLimit int64 `mapstructure:"waiter_limit"`
+
 	// Zstd settings apply to OTel-Arrow use of gRPC specifically.
 	Zstd zstd.DecoderConfig `mapstructure:"zstd"`
 }
@@ -57,3 +63,19 @@ func (cfg *ArrowConfig) Validate() error {
 	}
 	return nil
 }
+
+func (cfg *Config) Validate() error {
+	if err := cfg.GRPC.Validate(); err != nil {
+		return err
+	}
+	if err := cfg.Arrow.Validate(); err != nil {
+		return err
+	}
+	if cfg.Arrow.AdmissionLimitMiB != 0 {
+		cfg.Admission.AdmissionLimitMiB = cfg.Arrow.AdmissionLimitMiB
+	}
+	if cfg.Arrow.WaiterLimit != 0 {
+		cfg.Admission.WaiterLimit = cfg.Arrow.WaiterLimit
+	}
+	return nil
+}

receiver/otelarrowreceiver/internal/common/blocking_consumer.go

@@ -0,0 +1,46 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package common
+
+import(
+	"context"
+
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+)
+
+type BlockingConsumer struct{
+	block chan struct{}
+}
+
+func NewBlockingConsumer() *BlockingConsumer {
+	return &BlockingConsumer{
+		block: make(chan struct{}),
+	}
+}
+
+func (bc *BlockingConsumer) ConsumeTraces(ctx context.Context, _ ptrace.Traces) error {
+	<-bc.block
+	return nil
+}
+
+func (bc *BlockingConsumer) ConsumeMetrics(ctx context.Context, _ pmetric.Metrics) error {
+	<-bc.block
+	return nil
+}
+
+func (bc *BlockingConsumer) ConsumeLogs(ctx context.Context, _ plog.Logs) error {
+	<-bc.block
+	return nil
+}
+
+func (bc *BlockingConsumer) Unblock() {
+	close(bc.block)
+}
+
+func (bc *BlockingConsumer) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}

receiver/otelarrowreceiver/internal/logs/otlp_test.go

@@ -7,7 +7,9 @@ import (
 	"context"
 	"errors"
 	"net"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -19,8 +21,16 @@ import (
 	"go.opentelemetry.io/collector/receiver/receivertest"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"go.uber.org/multierr"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testdata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelarrowreceiver/internal/common"
+)
+
+const (
+	maxWaiters = 10
+	maxBytes = int64(250)
 )
 
 func TestExport(t *testing.T) {
@@ -57,6 +67,61 @@ func TestExport_ErrorConsumer(t *testing.T) {
 	assert.Equal(t, plogotlp.ExportResponse{}, resp)
 }
 
+func TestExport_AdmissionLimitBytesExceeded(t *testing.T) {
+	ld := testdata.GenerateLogs(10)
+	logSink := new(consumertest.LogsSink)
+	req := plogotlp.NewExportRequestFromLogs(ld)
+
+	logClient := makeLogsServiceClient(t, logSink)
+	resp, err := logClient.Export(context.Background(), req)
+	assert.EqualError(t, err, "rpc error: code = Unknown desc = rejecting request, request size larger than configured limit")
+	assert.Equal(t, plogotlp.ExportResponse{}, resp)
+}
+
+func TestExport_TooManyWaiters(t *testing.T) {
+	bc := common.NewBlockingConsumer()
+
+	logsClient := makeLogsServiceClient(t, bc)
+	bg := context.Background()
+	var errs, err error
+	ld := testdata.GenerateLogs(1)
+	req := plogotlp.NewExportRequestFromLogs(ld)
+	var mtx sync.Mutex
+	numResponses := 0
+	// Send request that will acquire all of the semaphores bytes and block.
+	go func() {
+		_, err = logsClient.Export(bg, req)
+		mtx.Lock()
+		errs = multierr.Append(errs, err)
+		numResponses+=1
+		mtx.Unlock()
+	}()
+
+	for i := 0; i < maxWaiters+1; i++ {
+		go func() {
+			_, err := logsClient.Export(bg, req)
+			mtx.Lock()
+			errs = multierr.Append(errs, err)
+			numResponses+=1
+			mtx.Unlock()
+		}()
+	}
+
+	// sleep so all async requests are blocked on semaphore Acquire.
+	time.Sleep(1 * time.Second)
+
+	// unblock and wait for errors to be returned and written.
+	bc.Unblock()
+	assert.Eventually(t, func() bool {
+		mtx.Lock()
+		defer mtx.Unlock()
+		errSlice := multierr.Errors(errs)
+		return numResponses == maxWaiters+2 && len(errSlice) == 1
+	}, 3 * time.Second, 10 * time.Millisecond)
+
+	assert.ErrorContains(t, errs, "too many waiters")
+}
+
 func makeLogsServiceClient(t *testing.T, lc consumer.Logs) plogotlp.GRPCClient {
 	addr := otlpReceiverOnGRPCServer(t, lc)
 	cc, err := grpc.NewClient(addr.String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
@@ -84,7 +149,9 @@ func otlpReceiverOnGRPCServer(t *testing.T, lc consumer.Logs) net.Addr {
 		ReceiverCreateSettings: set,
 	})
 	require.NoError(t, err)
-	r := New(lc, obsrecv)
+
+	bq := admission.NewBoundedQueue(maxBytes, maxWaiters)
+	r := New(lc, obsrecv, bq)
 	// Now run it as a gRPC server
 	srv := grpc.NewServer()
 	plogotlp.RegisterGRPCServer(srv, r)

receiver/otelarrowreceiver/internal/metrics/otlp_test.go

@@ -7,7 +7,9 @@ import (
 	"context"
 	"errors"
 	"net"
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -19,8 +21,16 @@ import (
 	"go.opentelemetry.io/collector/receiver/receivertest"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
+	"go.uber.org/multierr"
 
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/admission"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/otelarrow/testdata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/receiver/otelarrowreceiver/internal/common"
+)
+
+const (
+	maxWaiters = 10
+	maxBytes = int64(250)
 )
 
 func TestExport(t *testing.T) {
@@ -57,6 +67,61 @@ func TestExport_ErrorConsumer(t *testing.T) {
 	assert.Equal(t, pmetricotlp.ExportResponse{}, resp)
 }
 
+func TestExport_AdmissionLimitBytesExceeded(t *testing.T) {
+	md := testdata.GenerateMetrics(10)
+	metricSink := new(consumertest.MetricsSink)
+	req := pmetricotlp.NewExportRequestFromMetrics(md)
+
+	metricsClient := makeMetricsServiceClient(t, metricSink)
+	resp, err := metricsClient.Export(context.Background(), req)
+	assert.EqualError(t, err, "rpc error: code = Unknown desc = rejecting request, request size larger than configured limit")
+	assert.Equal(t, pmetricotlp.ExportResponse{}, resp)
+}
+
+func TestExport_TooManyWaiters(t *testing.T) {
+	bc := common.NewBlockingConsumer()
+
+	metricsClient := makeMetricsServiceClient(t, bc)
+	bg := context.Background()
+	var errs, err error
+	md := testdata.GenerateMetrics(1)
+	req := pmetricotlp.NewExportRequestFromMetrics(md)
+	var mtx sync.Mutex
+	numResponses := 0
+	// Send request that will acquire all of the semaphores bytes and block.
+	go func() {
+		_, err = metricsClient.Export(bg, req)
+		mtx.Lock()
+		errs = multierr.Append(errs, err)
+		numResponses+=1
+		mtx.Unlock()
+	}()
+
+	for i := 0; i < maxWaiters+1; i++ {
+		go func() {
+			_, err := metricsClient.Export(bg, req)
+			mtx.Lock()
+			errs = multierr.Append(errs, err)
+			numResponses+=1
+			mtx.Unlock()
+		}()
+	}
+
+	// sleep so all async requests are blocked on semaphore Acquire.
+	time.Sleep(1 * time.Second)
+
+	// unblock and wait for errors to be returned and written.
+	bc.Unblock()
+	assert.Eventually(t, func() bool {
+		mtx.Lock()
+		defer mtx.Unlock()
+		errSlice := multierr.Errors(errs)
+		return numResponses == maxWaiters+2 && len(errSlice) == 1
+	}, 3 * time.Second, 10 * time.Millisecond)
+
+	assert.ErrorContains(t, errs, "too many waiters")
+}
+
 func makeMetricsServiceClient(t *testing.T, mc consumer.Metrics) pmetricotlp.GRPCClient {
 	addr := otlpReceiverOnGRPCServer(t, mc)
 
@@ -85,7 +150,9 @@ func otlpReceiverOnGRPCServer(t *testing.T, mc consumer.Metrics) net.Addr {
 		ReceiverCreateSettings: set,
 	})
 	require.NoError(t, err)
-	r := New(mc, obsrecv)
+
+	bq := admission.NewBoundedQueue(maxBytes, maxWaiters)
+	r := New(mc, obsrecv, bq)
 	// Now run it as a gRPC server
 	srv := grpc.NewServer()
 	pmetricotlp.RegisterGRPCServer(srv, r)

receiver/otelarrowreceiver/internal/trace/otlp.go

@@ -5,7 +5,6 @@ package trace // import "github.com/open-telemetry/opentelemetry-collector-contr
 
 import (
 	"context"
-	"fmt"
 
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/ptrace"
@@ -47,7 +46,6 @@ func (r *Receiver) Export(ctx context.Context, req ptraceotlp.ExportRequest) (pt
 	ctx = r.obsrecv.StartTracesOp(ctx)
 
 	sizeBytes := int64(r.sizer.TracesSize(req.Traces()))
-	fmt.Println(sizeBytes)
 	err := r.boundedQueue.Acquire(ctx, sizeBytes)
 	if err != nil {
 		return ptraceotlp.NewExportResponse(), err