(otelarrowreceiver): asynchronous stream operations #181

moh-osman3 · 2024-04-23T12:05:16Z

This PR

resolves Arrow receiver load handling improvements #182
applies an request admission memory limiter merged in (boundedqueue): new semaphore implementation #174

…ompSz in exporter

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go

api/experimental/arrow/v1/arrow_service.pb.go

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

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go

collector/go.mod

jmacd · 2024-05-07T23:10:10Z

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go

+			if err != nil {
+				response.bytesToRelease = int64(0)
+			} else {
+				response.bytesToRelease = uncompSize


This looks better than before, thanks for working on it.

Thanks for the helpful review!

jmacd · 2024-05-07T23:21:57Z

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go

+	err = r.boundedQueue.Acquire(ctx, int64(prevAcquiredBytes))
+	if err != nil {
+		return fmt.Errorf("breaking stream: %v", err)
+	}


Breaking the stream when there are too many waiters -- LGTM.

jmacd · 2024-05-07T23:22:16Z

@lquerel please take a look, thanks!

lquerel

Overall, it seems good to me. However, I suggest improving the logs when the size specified in the header does not match the actual size, for the reasons mentioned in the comments.

lquerel · 2024-05-08T05:38:01Z

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go

+	// bounded queue to memory limit based on incoming uncompressed request size and waiters.
+	// Acquire will fail immediately if there are too many waiters,
+	// or will otherwise block until timeout or enough memory becomes available.
+	err = r.boundedQueue.Acquire(ctx, int64(prevAcquiredBytes))


What prevents a buggy (or misbehaving) client/exporter from setting the 'otlp-pdata-size' arbitrarily high? Without an upper limit or some other form of protection, could this create a type of DoS attack at minimal cost? Could we detect when the 'otlp-pdata-size' does not match the actual size of the message after decompression and then ban the sender for a period of time?

EDIT: Okay, there is detection in place with logging. Could we create some form of structured logs/events containing enough information to enable external (or internal) banning of the same clients that send us invalid data? The ban mechanism could be the topic of a future PR or be an external process triggered by these events.

Thanks for the review! Yes we log in the case that the header otlp-pdata-size, but I think you are correct that we could face a DoS attack if we accept requests with headers that acquire a large portion of the semaphore. I can file an issue for this and would love to here more about how to get the appropriate client information that would allow us to block these clients.

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go

lquerel · 2024-05-08T05:56:51Z

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go

+		if sizeHeaderFound {
+			// a mismatch between header set by exporter and the uncompSize just calculated.
+			r.telemetry.Logger.Debug("mismatch between uncompressed size in receiver and otlp-pdata-size header", zap.Int("uncompsize", int(uncompSize)), zap.Int("otlp-pdata-size", int(response.bytesToRelease)))
+		} else if diff < 0 {
+			// proto.Size() on compressed request was greater than pdata uncompressed size.
+			r.telemetry.Logger.Debug("uncompressed size is less than compressed size", zap.Int("uncompressed", int(uncompSize)), zap.Int("compressed", int(response.bytesToRelease)))
+		}


In my opinion, we must ensure that these structured logs contain enough information to enable us to ban buggy or misbehaving clients/exporters.

If any significant work is needed for this, I'd like to file an issue and come back to it. OTLP doesn't have an in-band way to identify the sender -- I believe we have minimal information in the gRPC peer struct, plus any headers we add ourselves. If we could standardize on a useful header for the exporter to emit, then we could log it here.

I created this issue #185 to work on in a followup

However after thinking about this more I'm wondering how reliable the address is in the client.Info that we get from the incoming stream context. Does this address show up consistently https://github.com/open-telemetry/opentelemetry-collector/blob/main/client/client.go#L93

Could this be enough information to identify harmful clients and block them from sending requests to the receiver?

@lquerel @jmacd added the client.Info.Addr to the log message if it is available. This issue #185 can be used to track a followup to ban clients that are misusing the header, but I would like to know your thoughts on the best way to go about this. Thanks

I don't know if client.Info is reliable. But we can probably find a way to get the client IP address at the gRPC connection level.

Included PRs: - #174 - #181 - #184 - #186 - #188 --------- Co-authored-by: Joshua MacDonald <[email protected]>

This PR adds information about the new config options added in #181 to the otelarrowreceiver readme. --------- Co-authored-by: Joshua MacDonald <[email protected]>

…ments & restructuring (#205) Restructure receiver code to improve readability. There are a number of metrics that are incremented when a batch starts being processed and are decremented when the batch is finished, but the control flow that maintained the balance of these updates was convoluted. The root-cause of #204 is that Arrow batches meant for a consumer to be processed in order were processed out-of-order. There was a large function body which served two purposes: consume Arrow data of the appropriate kind, enter data for the pipeline to consume next. This had to be split into two parts and should have been done as part of #181. (I, as reviewer, missed this and find, in hindsight, that the code is not easy to follow.) This improves the code structure by moving all stateful aspects of starting/finishing a request into a new `inFlightData` object which has a deferrable method to finish the request. Here, we keep: 1. The `inFlightWG` done count 2. The active requests metric 3. The active items metric 4. The active bytes metric 5. The bytes-acquired from the semaphore 6. A per-request span covering Arrow decode 7. Netstat-related instrumentation Authorization now happens before acquiring from the semaphore. A number of `fmt.Errorf()` calls are replaced with `status.Errorf(...)` and a specific error code. The tests are updated to be more specific. Several Arrow tests were accidentally canceling the test before an expected error condition was actually tested, they have been audited and improved. One new concurrent-receiver test was added. Fixes #204.

moh-osman3 added 10 commits April 25, 2024 08:36

add semaphore files

9a73c2a

add uncompSz to proto defintion and apply semaphore

e30c220

add AdmissionLimitMiB separate from MemoryLimitMiB, always record unc…

3ec41e4

…ompSz in exporter

closer to working

8e16255

return error in goroutine

2a3fa37

refactor solve race, broken tests still

45f2f9a

almost

3e68d50

painful but fixed?

212fca9

rm

0d9f180

rebase and fix tests again

366936e

moh-osman3 force-pushed the mohosman/apply-bounded-semaphore branch from 348bb99 to 366936e Compare April 25, 2024 13:59

moh-osman3 added 2 commits April 25, 2024 12:08

fix race

4c84297

remove unneded package

e1b920a

moh-osman3 marked this pull request as ready for review April 25, 2024 16:20

moh-osman3 requested review from lquerel, jmacd and codeboten as code owners April 25, 2024 16:20

jmacd reviewed Apr 25, 2024

View reviewed changes

collector/receiver/otelarrowreceiver/internal/arrow/arrow.go Outdated Show resolved Hide resolved

moh-osman3 added 3 commits April 25, 2024 13:47

remove replace

f93a37c

review feedback

77364fe

remove unneeded arg

a8448e4

jmacd reviewed Apr 30, 2024

View reviewed changes

api/experimental/arrow/v1/arrow_service.pb.go Outdated Show resolved Hide resolved

moh-osman3 added 2 commits May 1, 2024 08:31

rm uncompressed_size from proto

12f1b8c

add header, acquire diff

6f8ba19

jmacd reviewed May 2, 2024

View reviewed changes

moh-osman3 added 5 commits May 2, 2024 15:41

fix potential deadlock

af29366

unit tests

3b09b6c

use assertEqualUnsortedSpans

1258605

remove unneeded helper functions and use otelAssert.Equiv

25366e0

update changelog

40f7fa9

jmacd reviewed May 7, 2024

View reviewed changes

collector/go.mod Outdated Show resolved Hide resolved

moh-osman3 added 4 commits May 7, 2024 18:50

improve readability with response.bytesToRelease

7db9a40

remove deprecated

8d45b90

gofmt

8ea4949

rm newline

aacba4e

jmacd reviewed May 7, 2024

View reviewed changes

jmacd approved these changes May 7, 2024

View reviewed changes

go mod tidy

b97b100

lquerel reviewed May 8, 2024

View reviewed changes

moh-osman3 mentioned this pull request May 8, 2024

(otelarrowreceiver): blocking clients that misuse the otlp-pdata-size header #185

Open

moh-osman3 and others added 2 commits May 9, 2024 09:49

add client address if available

e9e2d31

Merge branch 'main' into mohosman/apply-bounded-semaphore

985ab4b

lquerel approved these changes May 9, 2024

View reviewed changes

jmacd merged commit 62b0393 into open-telemetry:main May 9, 2024
2 checks passed

moh-osman3 mentioned this pull request May 9, 2024

Release otel-arrow v0.23.0 #187

Merged

jmacd added a commit that referenced this pull request May 9, 2024

Release otel-arrow v0.23.0 (#187)

f72161c

Included PRs: - #174 - #181 - #184 - #186 - #188 --------- Co-authored-by: Joshua MacDonald <[email protected]>

moh-osman3 mentioned this pull request May 10, 2024

(otelarrowreceiver): document new config options #191

Merged

This was referenced May 28, 2024

Receiver crash report: concurrency bug #204

Closed

[otel-arrow/receiver] Receiver concurrency fixes; readability improvements & restructuring #205

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

(otelarrowreceiver): asynchronous stream operations #181

(otelarrowreceiver): asynchronous stream operations #181

moh-osman3 commented Apr 23, 2024 •

edited

Loading

jmacd May 7, 2024

moh-osman3 May 7, 2024

jmacd May 7, 2024

jmacd commented May 7, 2024

lquerel left a comment

lquerel May 8, 2024

moh-osman3 May 8, 2024

lquerel May 8, 2024

jmacd May 8, 2024

moh-osman3 May 8, 2024 •

edited

Loading

moh-osman3 May 8, 2024

moh-osman3 May 8, 2024

moh-osman3 May 9, 2024

lquerel May 9, 2024

(otelarrowreceiver): asynchronous stream operations #181

(otelarrowreceiver): asynchronous stream operations #181

Conversation

moh-osman3 commented Apr 23, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jmacd commented May 7, 2024

lquerel left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

moh-osman3 May 8, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

moh-osman3 commented Apr 23, 2024 •

edited

Loading

moh-osman3 May 8, 2024 •

edited

Loading