arrow

Documentation

Index

• Constants
• Variables
• type AnyStreamClient
• type Exporter
  • func NewExporter(maxStreamLifetime time.Duration, numStreams int, ...) *Exporter
  • func (e *Exporter) SendAndWait(ctx context.Context, data any) (bool, error)
  • func (e *Exporter) Shutdown(_ context.Context) error
  • func (e *Exporter) Start(ctx context.Context) error
• type PrioritizerName
  • func (p PrioritizerName) Validate() error
• type Stream
• type StreamClientFunc
  • func MakeAnyStreamClient[T AnyStreamClient](method string, ...) StreamClientFunc

Constants

const (
	// DefaultMaxStreamLifetime is 30 seconds, because the
	// marginal compression benefit of a longer OTel-Arrow stream
	// is limited after 100s of batches.
	DefaultMaxStreamLifetime = 30 * time.Second

	// DefaultPayloadCompression is "zstd" so that Arrow IPC
	// payloads use Arrow-configured Zstd over the payload
	// independently of whatever compression gRPC may have
	// configured.  This is on by default, achieving "double
	// compression" because:
	// (a) relatively cheap in CPU terms
	// (b) minor compression benefit
	// (c) helps stay under gRPC request size limits
	DefaultPayloadCompression configcompression.Type = "zstd"
)

Variables

var (
	// DefaultNumStreams is half the number of CPUs.  This is
	// selected as an estimate of relatively how much work is
	// being performed by the exporter compared with other
	// components in the system.
	DefaultNumStreams = max(1, runtime.NumCPU()/2)
)

Defaults settings should use relatively few resources, so that users are required to explicitly configure large instances.

var ErrStreamRestarting = status.Error(codes.Aborted, "stream is restarting")

Types

type AnyStreamClient

type AnyStreamClient interface {
	Send(*arrowpb.BatchArrowRecords) error
	Recv() (*arrowpb.BatchStatus, error)
	grpc.ClientStream
}

AnyStreamClient is the interface supported by all Arrow streams.

type Exporter

type Exporter struct {
	// contains filtered or unexported fields
}

Exporter is 1:1 with exporter, isolates arrow-specific functionality.

func NewExporter

func NewExporter(
	maxStreamLifetime time.Duration,
	numStreams int,
	prioritizerName PrioritizerName,
	disableDowngrade bool,
	telemetry component.TelemetrySettings,
	grpcOptions []grpc.CallOption,
	newProducer func() arrowRecord.ProducerAPI,
	streamClient StreamClientFunc,
	perRPCCredentials credentials.PerRPCCredentials,
	netReporter netstats.Interface,
) *Exporter

NewExporter configures a new Exporter.

func (*Exporter) SendAndWait

func (e *Exporter) SendAndWait(ctx context.Context, data any) (bool, error)

SendAndWait tries to send using an Arrow stream. The results are:

(true, nil): Arrow send: success at consumer (false, nil): Arrow is not supported by the server, caller expected to fallback. (true, non-nil): Arrow send: server response may be permanent or allow retry. (false, non-nil): Context timeout prevents retry.

consumer should fall back to standard OTLP, (true, nil)

func (*Exporter) Shutdown

func (e *Exporter) Shutdown(_ context.Context) error

Shutdown returns when all Arrow-associated goroutines have returned.

func (*Exporter) Start

func (e *Exporter) Start(ctx context.Context) error

Start creates the background context used by all streams and starts a stream controller, which initializes the initial set of streams.

type PrioritizerName

type PrioritizerName string

const (
	DefaultPrioritizer         PrioritizerName = LeastLoadedPrioritizer
	LeastLoadedPrioritizer     PrioritizerName = llPrefix
	LeastLoadedTwoPrioritizer  PrioritizerName = llPrefix + "2"
	LeastLoadedFourPrioritizer PrioritizerName = llPrefix + "4"
)

func (PrioritizerName) Validate

func (p PrioritizerName) Validate() error

Validate implements component.ConfigValidator

type Stream

type Stream struct {
	// contains filtered or unexported fields
}

Stream is 1:1 with gRPC stream.

type StreamClientFunc

type StreamClientFunc func(context.Context, ...grpc.CallOption) (AnyStreamClient, string, error)

streamClientFunc is a constructor for AnyStreamClients. These return the method name to assist with instrumentation, since the gRPC stats handler isn't able to see the correct uncompressed size.

func MakeAnyStreamClient

func MakeAnyStreamClient[T AnyStreamClient](method string, clientFunc func(ctx context.Context, opts ...grpc.CallOption) (T, error)) StreamClientFunc

MakeAnyStreamClient accepts any Arrow-like stream and turns it into an AnyStreamClient. The method name is carried through because once constructed, gRPC clients will not reveal their service and method names.