Arrow Exporter load prioritizer apparatus

.github/workflows/ci.yml

@@ -24,7 +24,7 @@ jobs:
         go-version: "1.21"
 
     - name: Build all modules
-      run: make build || true
+      run: make build
 
     - name: Test all modules
-      run: make test || true
+      run: make test

CHANGELOG.md

@@ -8,6 +8,8 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## Unreleased
 
+- Add load prioritization mechanism and "leastloaded" policy. [#178](https://github.com/open-telemetry/otel-arrow/pull/178)
+
 ## [0.21.0](https://github.com/open-telemetry/otel-arrow/releases/tag/v0.21.0) - 2024-04-10
 
 - Bump versions to latest 0.98.0 1.25.0 [#175](https://github.com/open-telemetry/otel-arrow/pull/175)

collector/cmd/otelarrowcol/go.sum

@@ -793,26 +793,26 @@ github.com/open-telemetry/opentelemetry-collector-contrib/extension/pprofextensi
 github.com/open-telemetry/opentelemetry-collector-contrib/extension/pprofextension v0.98.0/go.mod h1:b5y3WDWK6mqTL6U6eGqDqK2F3O9GncmvtsWmQ03UoVk=
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.98.0 h1:rbfZJ4YA82WSC9HW2jgmY1UZEsOYte8VTjrBFU6dv5Y=
 github.com/open-telemetry/opentelemetry-collector-contrib/internal/common v0.98.0/go.mod h1:KOTp7iN8sWTmqk/SxbMqvM2KXUdgXRWb6hI84E5arLY=
-github.com/open-telemetry/otel-arrow v0.21.0 h1:MWm//AMwchAXBtTpPM4qbp8c4qFYpZNNJnEPyl7EvoQ=
-github.com/open-telemetry/otel-arrow v0.21.0/go.mod h1:lCYR/TB4fYeKGVkwwDPIHRVW7KcoF3e+mTF164p6uoo=
-github.com/open-telemetry/otel-arrow/collector v0.21.0 h1:DWo3OK2kvmGXcgoTc7dRBrADuwu0PKX31KcG/q7fElk=
-github.com/open-telemetry/otel-arrow/collector v0.21.0/go.mod h1:MDDGj7VCOA6ya0d+xbPneynT5gC7TCHV3vVeXDEo41g=
-github.com/open-telemetry/otel-arrow/collector/connector/validationconnector v0.21.0 h1:nZ0zAYh9RUTRmIEigY+4OMt/mNVmkKhk6QAWLsTk5pM=
-github.com/open-telemetry/otel-arrow/collector/connector/validationconnector v0.21.0/go.mod h1:UdVJl2OoCOdvjvjcBcpNzw/oCs22DTZs8sqX/frUjw4=
-github.com/open-telemetry/otel-arrow/collector/exporter/fileexporter v0.21.0 h1:cwpBST2VxZ/fInOmj2x0IokmRfX+eeimpap2b2SPaRw=
-github.com/open-telemetry/otel-arrow/collector/exporter/fileexporter v0.21.0/go.mod h1:ruDh+tsLVYyaKhGuinIs96sb9uvXRl4lEy7m7VDE9+s=
-github.com/open-telemetry/otel-arrow/collector/exporter/otelarrowexporter v0.21.0 h1:N9R19/ascPkyzACWYpmnJD+fGtJyV+xddmYStjO/EvA=
-github.com/open-telemetry/otel-arrow/collector/exporter/otelarrowexporter v0.21.0/go.mod h1:ID/YmgkHY+fEUO1ZoraUHWROFecZnSWDo2BH2AWpwV8=
-github.com/open-telemetry/otel-arrow/collector/processor/concurrentbatchprocessor v0.21.0 h1:1OwSAt/Mgws9BjYeWJIYohizyIiuPTQb14H6vqk2joQ=
-github.com/open-telemetry/otel-arrow/collector/processor/concurrentbatchprocessor v0.21.0/go.mod h1:jhxhrxEviCFlBdso59LS7aeNDfE+U5EEgSPYvsjE+f0=
-github.com/open-telemetry/otel-arrow/collector/processor/experimentprocessor v0.21.0 h1:jB6d8VzWGEY1Vqmo2dOO6WzTjGxsym8g1pRTKIl6o5Y=
-github.com/open-telemetry/otel-arrow/collector/processor/experimentprocessor v0.21.0/go.mod h1:qtboHJOOcDLLo5m9BkakGnEha/7tZRRfUjTSnL+VLB8=
-github.com/open-telemetry/otel-arrow/collector/processor/obfuscationprocessor v0.21.0 h1:0kCHmlwCWh3spFpXhCt5WLWImRAmuAvHjld5kUJoCfM=
-github.com/open-telemetry/otel-arrow/collector/processor/obfuscationprocessor v0.21.0/go.mod h1:5aucYwnpl0uZDR5gU4kDPxQn5Cx5njNqG1ZsS9FNKo8=
-github.com/open-telemetry/otel-arrow/collector/receiver/filereceiver v0.21.0 h1:YSDbPz8HWS8WsYREEBgU6BQ+4mD3iP5zA3jVJwx14h8=
-github.com/open-telemetry/otel-arrow/collector/receiver/filereceiver v0.21.0/go.mod h1:+b7cyHH/e0v7DxXV8vipYDtnEflWsI/HBb2+35FjC+4=
-github.com/open-telemetry/otel-arrow/collector/receiver/otelarrowreceiver v0.21.0 h1:UVGTHve4UGW+ffAvu1Z5M6jcufHfJBeIUQEVJS/wI2U=
-github.com/open-telemetry/otel-arrow/collector/receiver/otelarrowreceiver v0.21.0/go.mod h1:h0zvjU/w/an91AQWELeoD9EzbrCAhljeSUuKSy6T1dI=
+github.com/open-telemetry/otel-arrow v0.21.0 h1:lmR2ajHIXiuick7l7XKfN9XnIbTWjcHv4/j5zDWgOWY=
+github.com/open-telemetry/otel-arrow v0.21.0/go.mod h1:alPrug7QOMk2WnNs+ORUOahekkElkQlUh54UButGELc=
+github.com/open-telemetry/otel-arrow/collector v0.21.0 h1:Zenyy3trGpwgpCrNG+SOOY4rFcDdAWntURMRLe30jz0=
+github.com/open-telemetry/otel-arrow/collector v0.21.0/go.mod h1:7C68tM7TET9pBAKEvIR5wT+rTMv8ns7+0YOgHw73qtQ=
+github.com/open-telemetry/otel-arrow/collector/connector/validationconnector v0.21.0 h1:xXxv8xkDz0vGprU0qNPNTWq7nHDXDYGgB0PtG3g2mIA=
+github.com/open-telemetry/otel-arrow/collector/connector/validationconnector v0.21.0/go.mod h1:nS0jwMCwcV12A+L1Di2EMlzVSDSoFxn6M+22gjupeRQ=
+github.com/open-telemetry/otel-arrow/collector/exporter/fileexporter v0.21.0 h1:U1Fg4UHb1s0bLUhsCHE8M4rn9IWALu4wYEkXRndipVk=
+github.com/open-telemetry/otel-arrow/collector/exporter/fileexporter v0.21.0/go.mod h1:wVHCf3F53+FMXDpeu4ESX7tT4N3lEqCSuKwoJLXbp2g=
+github.com/open-telemetry/otel-arrow/collector/exporter/otelarrowexporter v0.21.0 h1:NY7qbcUa8i8u8ze8GR84qrEQgFhotz+i+CNe9z7wiDg=
+github.com/open-telemetry/otel-arrow/collector/exporter/otelarrowexporter v0.21.0/go.mod h1:tQ9FscjuVOEJ2rmJSYoA/sl9CmZfbiEJjKFMmANOctM=
+github.com/open-telemetry/otel-arrow/collector/processor/concurrentbatchprocessor v0.21.0 h1:dBi3T8l6nvq4gM9ks56MGs7+3etz7NGE720hHryIwfg=
+github.com/open-telemetry/otel-arrow/collector/processor/concurrentbatchprocessor v0.21.0/go.mod h1:weO95udPGxPvroUyd9gLw4JgMyIeRFSGu8zIDRc2Y+g=
+github.com/open-telemetry/otel-arrow/collector/processor/experimentprocessor v0.21.0 h1:nb1jNqiM3ws2L8hAsB5YfB+vQT3SdyL057SntTIEA5s=
+github.com/open-telemetry/otel-arrow/collector/processor/experimentprocessor v0.21.0/go.mod h1:w3hv5AkhfakLq12nbz2skZ/e/Qqm8Om1qWE93S0SUm0=
+github.com/open-telemetry/otel-arrow/collector/processor/obfuscationprocessor v0.21.0 h1:0Y1tLc8+V1f4RnFCMx3EdaE3938p16z/Fkxnorwboq8=
+github.com/open-telemetry/otel-arrow/collector/processor/obfuscationprocessor v0.21.0/go.mod h1:TufpYpZ+RIzosn0xgBGg1FkgU0ZGoQZd93qVKNdUyYM=
+github.com/open-telemetry/otel-arrow/collector/receiver/filereceiver v0.21.0 h1:4at/i9ghKCifuTjXw0bfOfZsk7Iu4R0bgJnqt2YDvZU=
+github.com/open-telemetry/otel-arrow/collector/receiver/filereceiver v0.21.0/go.mod h1:+TM8UHPCpuvM4aefkts7J7S/IxIjpyzbZhPynG8s7O0=
+github.com/open-telemetry/otel-arrow/collector/receiver/otelarrowreceiver v0.21.0 h1:m6hne3ecPF5y48D9LVfiuYuIqRYN+tlg1dofyviIY3Y=
+github.com/open-telemetry/otel-arrow/collector/receiver/otelarrowreceiver v0.21.0/go.mod h1:pVrBChLbT8XbF6li/DDNuNhcBdckhnabLONK4Vsgr1o=
 github.com/opentracing/opentracing-go v1.0.2/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
 github.com/opentracing/opentracing-go v1.0.3-0.20180606204148-bd9c31933947/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
 github.com/opentracing/opentracing-go v1.1.0/go.mod h1:UkNAQd3GIcIGf0SeVgPpRdFStlNbqXla1AfSYxPUl2o=
@@ -1664,7 +1664,7 @@ google.golang.org/api v0.15.0/go.mod h1:iLdEw5Ide6rF15KTC1Kkl0iskquN2gFfn9o9XIsb
 google.golang.org/api v0.17.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
 google.golang.org/api v0.18.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
 google.golang.org/api v0.19.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
-google.golang.org/api v0.21.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
+google.golang.org/api v0.20.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
 google.golang.org/api v0.22.0/go.mod h1:BwFmGc8tA3vsd7r/7kR8DY7iEEGSU04BFxCo5jP/sfE=
 google.golang.org/api v0.24.0/go.mod h1:lIXQywCXRcnZPGlsd8NbLnOjtAoL6em04bJ9+z0MncE=
 google.golang.org/api v0.28.0/go.mod h1:lIXQywCXRcnZPGlsd8NbLnOjtAoL6em04bJ9+z0MncE=

collector/connector/validationconnector/go.sum

@@ -43,8 +43,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
-github.com/open-telemetry/otel-arrow v0.21.0 h1:MWm//AMwchAXBtTpPM4qbp8c4qFYpZNNJnEPyl7EvoQ=
-github.com/open-telemetry/otel-arrow v0.21.0/go.mod h1:lCYR/TB4fYeKGVkwwDPIHRVW7KcoF3e+mTF164p6uoo=
+github.com/open-telemetry/otel-arrow v0.21.0 h1:lmR2ajHIXiuick7l7XKfN9XnIbTWjcHv4/j5zDWgOWY=
+github.com/open-telemetry/otel-arrow v0.21.0/go.mod h1:alPrug7QOMk2WnNs+ORUOahekkElkQlUh54UButGELc=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v1.19.0 h1:ygXvpU1AoN1MhdzckN+PyD9QJOSD4x7kmXYlnfbA6JU=

collector/exporter/fileexporter/go.sum

@@ -50,8 +50,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w
 github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
 github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
-github.com/open-telemetry/otel-arrow/collector v0.21.0 h1:DWo3OK2kvmGXcgoTc7dRBrADuwu0PKX31KcG/q7fElk=
-github.com/open-telemetry/otel-arrow/collector v0.21.0/go.mod h1:MDDGj7VCOA6ya0d+xbPneynT5gC7TCHV3vVeXDEo41g=
+github.com/open-telemetry/otel-arrow/collector v0.21.0 h1:Zenyy3trGpwgpCrNG+SOOY4rFcDdAWntURMRLe30jz0=
+github.com/open-telemetry/otel-arrow/collector v0.21.0/go.mod h1:7C68tM7TET9pBAKEvIR5wT+rTMv8ns7+0YOgHw73qtQ=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_golang v1.19.0 h1:ygXvpU1AoN1MhdzckN+PyD9QJOSD4x7kmXYlnfbA6JU=

collector/exporter/otelarrowexporter/README.md

@@ -106,6 +106,20 @@ The following settings determine the resources that the exporter will use:
 - `num_streams` (default: number of CPUs): the number of concurrent Arrow streams
 - `max_stream_lifetime` (default: unlimited): duration after which streams are recycled.
 
+#### Load balancing
+
+The `arrow` configuration block includes a configurable prioritization
+policy.  The default policy, named "fifo", distributes work to the
+stream with the first-available writer.
+
+- `prioritizer` (default: fifo): may be set to "leastloadedN" where N <= `num_streams`.
+
+An alternative policy named "leastloadedN" is available, for values of
+N up to the number of configured streams (e.g., leastloaded2,
+leastloaded10).  This prioritizer makes a random selection of N
+streams and chooses the stream with the least number of outstanding
+work items.
+
 ### Network Configuration
 
 This component uses `round_robin` by default as the gRPC load

collector/exporter/otelarrowexporter/config.go

@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/open-telemetry/otel-arrow/collector/compression/zstd"
+	"github.com/open-telemetry/otel-arrow/collector/exporter/otelarrowexporter/internal/arrow"
 	"github.com/open-telemetry/otel-arrow/pkg/config"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config/configcompression"
@@ -73,6 +74,10 @@ type ArrowConfig struct {
 	// to standard OTLP.  If the Arrow service is unavailable, it
 	// will retry and/or fail.
 	DisableDowngrade bool `mapstructure:"disable_downgrade"`
+
+	// Prioritizer is a policy name for how load is distributed
+	// across streams.
+	Prioritizer arrow.PrioritizerName `mapstructure:"prioritizer"`
 }
 
 var _ component.Config = (*Config)(nil)
@@ -93,6 +98,10 @@ func (cfg *ArrowConfig) Validate() error {
 		return fmt.Errorf("zstd encoder: invalid configuration: %w", err)
 	}
 
+	if err := cfg.Prioritizer.Validate(); err != nil {
+		return fmt.Errorf("invalid prioritizer: %w", err)
+	}
+
 	// The cfg.PayloadCompression field is validated by the underlying library,
 	// but we only support Zstd or none.
 	switch cfg.PayloadCompression {

collector/exporter/otelarrowexporter/config_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/open-telemetry/otel-arrow/collector/compression/zstd"
+	"github.com/open-telemetry/otel-arrow/collector/exporter/otelarrowexporter/internal/arrow"
 	"github.com/open-telemetry/otel-arrow/pkg/config"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -32,6 +33,7 @@ func TestUnmarshalDefaultConfig(t *testing.T) {
 	assert.NoError(t, component.UnmarshalConfig(cm, cfg))
 	assert.Equal(t, factory.CreateDefaultConfig(), cfg)
 	assert.Equal(t, "round_robin", cfg.(*Config).ClientConfig.BalancerName)
+	assert.Equal(t, arrow.DefaultPrioritizer, cfg.(*Config).Arrow.Prioritizer)
 }
 
 func TestUnmarshalConfig(t *testing.T) {
@@ -86,6 +88,7 @@ func TestUnmarshalConfig(t *testing.T) {
 				MaxStreamLifetime:  2 * time.Hour,
 				PayloadCompression: configcompression.TypeZstd,
 				Zstd:               zstd.DefaultEncoderConfig(),
+				Prioritizer:        "leastloaded8",
 			},
 		}, cfg)
 }

collector/exporter/otelarrowexporter/factory.go

@@ -57,7 +57,8 @@ func createDefaultConfig() component.Config {
 			NumStreams:        runtime.NumCPU(),
 			MaxStreamLifetime: time.Hour,
 
-			Zstd: zstd.DefaultEncoderConfig(),
+			Zstd:        zstd.DefaultEncoderConfig(),
+			Prioritizer: arrow.DefaultPrioritizer,
 
 			// PayloadCompression is off by default because gRPC
 			// compression is on by default, above.

collector/exporter/otelarrowexporter/factory_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/open-telemetry/otel-arrow/collector/compression/zstd"
+	"github.com/open-telemetry/otel-arrow/collector/exporter/otelarrowexporter/internal/arrow"
 	"github.com/open-telemetry/otel-arrow/collector/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -41,6 +42,7 @@ func TestCreateDefaultConfig(t *testing.T) {
 		MaxStreamLifetime:  time.Hour,
 		PayloadCompression: "",
 		Zstd:               zstd.DefaultEncoderConfig(),
+		Prioritizer:        arrow.DefaultPrioritizer,
 	})
 }
 

collector/exporter/otelarrowexporter/go.sum

@@ -88,10 +88,10 @@ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjY
 github.com/mostynb/go-grpc-compression v1.2.2 h1:XaDbnRvt2+1vgr0b/l0qh4mJAfIxE0bKXtz2Znl3GGI=
 github.com/mostynb/go-grpc-compression v1.2.2/go.mod h1:GOCr2KBxXcblCuczg3YdLQlcin1/NfyDA348ckuCH6w=
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
-github.com/open-telemetry/otel-arrow v0.21.0 h1:MWm//AMwchAXBtTpPM4qbp8c4qFYpZNNJnEPyl7EvoQ=
-github.com/open-telemetry/otel-arrow v0.21.0/go.mod h1:lCYR/TB4fYeKGVkwwDPIHRVW7KcoF3e+mTF164p6uoo=
-github.com/open-telemetry/otel-arrow/collector v0.21.0 h1:DWo3OK2kvmGXcgoTc7dRBrADuwu0PKX31KcG/q7fElk=
-github.com/open-telemetry/otel-arrow/collector v0.21.0/go.mod h1:MDDGj7VCOA6ya0d+xbPneynT5gC7TCHV3vVeXDEo41g=
+github.com/open-telemetry/otel-arrow v0.21.0 h1:lmR2ajHIXiuick7l7XKfN9XnIbTWjcHv4/j5zDWgOWY=
+github.com/open-telemetry/otel-arrow v0.21.0/go.mod h1:alPrug7QOMk2WnNs+ORUOahekkElkQlUh54UButGELc=
+github.com/open-telemetry/otel-arrow/collector v0.21.0 h1:Zenyy3trGpwgpCrNG+SOOY4rFcDdAWntURMRLe30jz0=
+github.com/open-telemetry/otel-arrow/collector v0.21.0/go.mod h1:7C68tM7TET9pBAKEvIR5wT+rTMv8ns7+0YOgHw73qtQ=
 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=