[Donation Proposal]: OTEL Arrow Adapter

[lquerel]

Description

This project extends, in a compatible way, the existing OTLP protocol with a generic columnar representation for metrics, logs, and traces based on the cross-language Apache Arrow transport system for columnar data. This extension significantly improves the efficiency of the OTLP protocol for scenarios involving the transmission of large batches of OTLP entities. Results show a 2-3 times better compression rate for typical data, while Apache Arrow’s zero-copy serialization and deserialization techniques help lower overhead.

The first phase of this project will deliver the following primary components, contained in the donated repository:

1. Golang-based adapter reference implementation that implements the translation to and from OTLP-Arrow using the OTel Collector’s “pdata” interface.
2. Protobuf definition for OTLP-Arrow that would be migrated into the opentelemetry-proto repo as it stabilizes. This includes a representation for multi-variate metrics which allows metrics with shared attributes to be compactly represented and processed.
3. Arrow-enabled OTel Collector Exporter and Receiver components that are drop-in compatible with the OTLP Exporter and Receiver core components.
4. Arrow-enabled OTel-Go SDK “shim” to the Arrow-enabled Exporter component, which is mainly useful for validation purposes.

In a second phase of the project, we propose developing new OpenTelemetry components and mechanisms for exchanging and processing data with the Apache Arrow ecosystem, including:

1. Arrow-enabled Collector: an OTel Collector pipeline for processing Arrow batches directly
2. Support for reading/writing Apache Parquet files, and other methods to leverage non-Golang Apache Arrow ecosystem components (e.g., the Arrow Query engine)
3. Arrow-enabled OpenTelemetry SDKs (i.e., without a “shim” and adapter).

The repository for donation: https://github.com/f5/otel-arrow-adapter

Collector repo fork containing the drop-in compatible exporter and receiver components under development: https://github.com/open-telemetry/experimental-arrow-collector

More details on the associated OTEP text: https://github.com/lquerel/oteps/blob/main/text/0156-columnar-encoding.md. The OTEP is still pending, unmerged.

Benefits to the OpenTelemetry community

Compared to the existing OpenTelemetry protocol this compatible extension has the following improvements:

• Reduce the bandwidth requirements of the protocol. The two main levers are: 1) a better representation of the telemetry data based on a columnar representation, 2) a stream-oriented gRPC endpoint that is more efficient to transmit batches of OTLP entities.
• Provide a more optimal representation for multivariate time-series data. With the current version of the OpenTelemetry protocol, users have to transform multivariate time-series (i.e multiple related metrics sharing the same attributes and timestamp) into a collection of univariate time-series resulting in a large amount of duplication and additional overhead covering the entire chain from exporters to backends.
• Provide more advanced and efficient telemetry data processing capabilities. Increasing data volume, cost efficiency, and data minimization require additional data processing capabilities such as data projection, aggregation, and filtering.

Reasons for donation

The proposed protocol is tightly integrated with the existing OTLP protocol. A fallback mechanism between the two protocols has been implemented via a new pair of OTLP Arrow Exporter/Receiver that are drop-in compatible with the OTLP Exporter/Receiver core component, justifying the integration of this extension directly into the upstream project.

Repository

https://github.com/f5/otel-arrow-adapter

Existing usage

The project is developed and maintained jointly by F5 and Lightstep. F5 seeks to align the open-source community built around NGINX with the open-source community built around OpenTelemetry; this effort will deliver industrial-grade analytics capability to the telemetry data produced by its software components for its customers. Improving the compression performance and representation of multivariate time series are also among F5's goals. Lightstep, wanting to recommend OpenTelemetry collectors for its customers to use, seeks to improve the compression performance and efficiency achieved by OTLP when used for bulk data transport.

We are actively testing the components for donation on production data using an OpenTelemetry collector built from our drop-in compatible OTLP exporter and receiver. We are extending test coverage and eliminating gaps. We are planning to start beta tests with the community by providing documentation and tooling for benchmarking and troubleshooting.

Maintenance

F5 and Lightstep will continue to develop and maintain the project. We will encourage and help all new contributors to participate in this project. We are open to suggestions and ideas.

Our current roadmap is as follows:

1. Continue to work on performance, reliability and robustness, extending test coverage, testing with more production data.
2. Phase 2 of this project
   a) New client SDK natively supporting OTLP Arrow and multivariate metrics.
   b) Continuing the migration to Apache Arrow to achieve end-to-end performance gains.
   c) Integrating processing, aggregation, and filtering capabilities that leverages the Apache Arrow eco-system.
   d) Parquet integration.

Licenses

This project is licensed under the terms of the Apache 2.0 open source license.

Trademarks

The Apache Arrow project is used in this project.

Other notes

No response

[gramidt]

I am beyond excited to see this proposal! Excellent work, and I can't wait to see phase 2 of this project. 🚀

[jsuereth]

I think it might be good to call out some of the large technical differences in OTLP-arrow vs. OTLP and when to use it vs. other options.

I was looking over details.

1. This new protocol is streaming gRPC. That's a change from batch OTLP. How would this interact with load balancers and the overall design goals of OTLP?
2. Is the new protocol stateful (or able to be stateful?) I saw it has optional dictionaries that can be passed and such. I'm curious if this is the case.

I'm definitely supportive of OTLP-arrow as an viable alternative to OTLP where it makes sense, but I'd love if you could outline where it shines and where it doesn't. Specifically when should you pick one vs. the other.

[lquerel]

Thanks @jsuereth for your support.

Regarding your first comment, I added the following note in the OTEP to explain the motivations behind streaming gRPC for OTLP Arrow.

Unary RPC vs Stream RPC: We use a stream-oriented protocol to get rid of the overhead of specifying the schema and dictionaries for each batch. A state will be maintained receiver side to keep track of the schemas and dictionaries. The Arrow IPC format has been designed to follow this pattern and also allows the dictionaries to be sent incrementally. Similarly, ZSTD dictionaries can also be transferred to the RPC stream to optimize the transfer of small batches (for more details see the description of the compression field in the following paragraphs). To mitigate the usual pitfalls of a stream-oriented protocol please see this paragraph in the implementation recommendations section.

For the second comment, I confirm that the protocol is stateful. We use this property to maintain schemas and dictionaries on the receiver side to amortize their cost. This small extra cost in the first messages is actually paid back quite quickly and allows us to obtain gains on the size of the messages as well as improvements on the compression rate.

For your last question, the following two scenarios are particularly suitable for OTLP Arrow (non exhaustive list):

• Production environments, consisting of a large number of OTEL data producers, which send their telemetry data to an external system (via the Internet). The use of an Arrow OTLP collector acting as a batching point at the output of the environment will result in substantial savings in network costs.
• Individual producers of telemetry data with massive trace streams (e.g. API gateway, Load balancer, ...) will be able to use this new protocol with a native OTLP Arrow client (currently in design phase). This is also valid for systems that emit massive amounts of metrics and logs.

A future enhancement (based on ZSTD dictionaries) should also make this project interesting for scenarios where the batch size is small.

[tigrannajaryan]

A couple questions:

• There is no support for this new protocol if you use plain HTTP, it requires gRPC?
• Since it uses gRPC streams how does it handle the load balancing? Streams are long-lived and can cause imbalance over time.

[lquerel]

@tigrannajaryan

The design currently calls for the use of gRPC streams to benefit from OTLP-Arrow transport. We believe that some of this benefit can be had even for unary gRPC and HTTP requests with large request batches to amortize sending of dictionary and schema information. This remains an area for study.

Regarding the "imbalance over time" question, we rely on the MaxAge and MaxAgeGrace of the gRPC connection to mitigate with this kind of issue. This is discussed in the "Implement recommendations" section of the OTEP.

[lquerel]

@jsuereth @tigrannajaryan

I updated my benchmark tool to compare the three following experiments:

• OTLP
• OTLP Arrow stream mode
• OTLP Arrow unary RPC mode (or plain HTTP mode)

The following results are based on traces coming from a production environment (~400K traces).

The streaming mode is definitely the best among these 3 experiments. This protocol is stateful and allows to :

• amortize the initial cost of the schema
• transfer delta dictionaries to remove entirely redundant information between batches in the same stream

It is interesting to note that for batches with more than 100 traces, the unary RPC mode is slightly better than the standard OTLP protocol. This is due to the organization of the data being better suited to compression.

[tigrannajaryan]

I updated my benchmark tool to compare the three following experiments

@lquerel thanks for the update! What does the batch size reflect? Is it literally the number of traces or number of spans? If number of traces, how many spans in each trace?

[tigrannajaryan]

@lquerel I am surprised streaming saves so much for large batch sizes. I would expect that for sufficiently large batches the dictionary is a small part of it so re-sending it with every batch shouldn't result in much bigger payloads.
Do you have this particular benchmark's code that I can run myself and understand a bit better?

[tigrannajaryan]

Also, if I understand correctly the columnar format shows little advantage in compressed form (and sometimes is even worse, for very small batches). So, essentially most of the gains are from streaming not from using columnar format.

[lquerel]

@tigrannajaryan The following pull request contains the modifications to benchmark the "unary RPC" mode for traces and logs. I'm still working on finalizing a memory optimisation for metrics. These benchmarks reproduces all the steps from the conversion between OTLP and OTLP Arrow, the serialization, the compression, ... These steps are cleary represented in the output of the benchmark tool.

f5/otel-arrow-adapter#108

To run the benchmark you need to get some data in protobuf format. Basically a Trace Request serialized in binary format and saved as a file. To run the benchmark for traces I'm using the following command.

go run tools/trace_benchmark/main.go -unaryrpc traces.bin.

traces.bin being the file containing the traces. The file must contain enough traces to run all the different tested scenarios (batch sizes 10, 100, 1000, 2000, 5000, 10000). I personally used a dataset of ~400k traces.

I will soon update the benchmark tool to produce more detailed stats in order to present the distribution of spans and traces in the input file.

UPDATE: stats on the data set.

[tigrannajaryan]

traces.bin being the file containing the traces.

Can you share the traces.bin that you used? I want to rerun the same benchmarks.

[lquerel]

@tigrannajaryan

Regarding the impact of streaming on the compression ratio. Not having to send dictionaries over and over again for many fields is definitely an advantage. By nature, a dictionary doesn't compress well because there aren't many repetitions, but when you can get rid of dictionaries entirely, it's directly a net gain. This is what happens in the scenario using streaming. However, the gain may vary a bit depending on the nature of the data set.

In the case of the unary rpc scenario, we pay both the overhead of the schema and the dictionaries that are rebuilt for each batch and that in comparison with a good compression algorithm do not do much better in unary rpc mode. However in terms of allocation, serialization and processing speed (e.g. for filtering, aggregation), the columnar format will be incomparably better (visible in phase 2). It is theoretically possible to do better I think by sorting some columns to optimize the compression rate. This is a well known optimization in the world of columnar data. However it is quite difficult to implement in a general way in the OTel context because the OTel entities are strongly hierarchical and the attributes are not uniform between the different entities which compose the batches. This will be the subject of a future optimization that will not impact the protocol and the data format.

Finally, the fact that the compression ratio is a bit worse with small batches was expected. The goal of this effort is to optimize batch sizes of at least 100-200 elements.

[jmacd]

For the record:

This new protocol is streaming gRPC. That's a change from batch OTLP. How would this interact with load balancers and the overall design goals of OTLP?

We did a study of the impact of unary vs. streaming gRPC. There is nothing preventing us from using unary RPC, but you will need large batches to get good compression and you will still be leaving compression "on the table".

Is the new protocol stateful (or able to be stateful?) I saw it has optional dictionaries that can be passed and such. I'm curious if this is the case.

It is point-to-point stateful. You can reset a connection at any time and it will rebuild the state needed in a future connection.