[OpAMP] Refactor OpAMP communication pipe by RassK · Pull Request #4930 · open-telemetry/opentelemetry-dotnet-contrib

RassK · 2026-08-04T11:12:11Z

What

Design discussion issue open-telemetry/opamp-spec#366

This is about ensuring that communication is according to spec.

When the Agent wants to send a message to the Server and the Agent has previously sent a request to the Server that is not yet responded, the Client MUST wait until the response is received before a new request can be made.
src: https://opentelemetry.io/docs/specs/opamp/#plain-http-transport

Since spec is currently not clear that messages can be accepted in sync or async manner. This PR is a basis to support both options via configuration. If it's decided that the client must block the pipe until a full response is constructed, a follow up is needed.

Changes

Makes sure the client is waiting for server to respond and blocks the pipe.
Uses accumulator message to "queue" client's messages

Breaking changes ❗

No need for async send messages, since messages are always accumulated.

Notes

~~❗ This PR is a preview and a discussion object how to proceed to support corner cases in the spec.~~
Since there seems to be a consensus with this PR, we can move forward

Merge requirement checklist

CONTRIBUTING guidelines followed (license requirements, nullable enabled, static analysis, etc.)
Unit tests added/updated
Appropriate CHANGELOG.md files updated for non-trivial changes < TODO until the final form is decided
Changes in public API reviewed (if applicable)

opentelemetry-pr-dashboard · 2026-08-04T11:13:08Z

Pull request dashboard status

Waiting on the author · refreshed 2026-08-19 12:31 UTC

Respond to 1 review item (e.g. link a commit, explain why not, ask a follow-up):

Inline threads: 1

Status above doesn't look right?

Just replied or pushed? Anything around or after the refresh time above may not be picked up yet — give it a few minutes.
Should this be with reviewers? Comment /dashboard route:reviewers to route it to them.
Anything wrong — including the routing? Report it with what you expected; it helps us improve the dashboard.

codecov · 2026-08-04T11:14:23Z

Codecov Report

❌ Patch coverage is 96.69811% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 78.09%. Comparing base (abdb29d) to head (d288329).
⚠️ Report is 5 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
...c/OpenTelemetry.OpAmp.Client/Internal/OpAmpPipe.cs	94.87%	6 Missing ⚠️
...nt/Internal/Services/Heartbeat/HeartbeatService.cs	80.00%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #4930      +/-   ##
==========================================
+ Coverage   77.89%   78.09%   +0.20%     
==========================================
  Files         474      476       +2     
  Lines       20279    20306      +27     
==========================================
+ Hits        15796    15858      +62     
+ Misses       4483     4448      -35

Flag	Coverage Δ
unittests-OpAmp.Client	`90.69% <96.69%> (+3.37%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...penTelemetry.OpAmp.Client/Internal/FrameBuilder.cs	`82.10% <100.00%> (+0.12%)`	⬆️
...nTelemetry.OpAmp.Client/Internal/FrameProcessor.cs	`86.56% <100.00%> (+1.08%)`	⬆️
...etry.OpAmp.Client/Internal/MessageBuilderHelper.cs	`100.00% <100.00%> (ø)`
...nt/Internal/Messages/AgentIdentificationMessage.cs	`0.00% <ø> (ø)`
...y.OpAmp.Client/Internal/Messages/CommandMessage.cs	`0.00% <ø> (ø)`
...ent/Internal/Messages/ConnectionSettingsMessage.cs	`100.00% <ø> (ø)`
...p.Client/Internal/Messages/ErrorResponseMessage.cs	`0.00% <ø> (ø)`
...ient/Internal/Messages/PackagesAvailableMessage.cs	`0.00% <ø> (ø)`
...p.Client/Internal/Messages/ServerToAgentMessage.cs	`100.00% <100.00%> (ø)`
...ry.OpAmp.Client/Internal/OpAmpClientEventSource.cs	`80.59% <100.00%> (+31.59%)`	⬆️
... and 4 more

... and 1 file with indirect coverage changes

🚀 New features to boost your workflow:

📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

…metry-dotnet-contrib into opamp-com-reworks

stevejgordon

The overall direction looks reasonable and the pipe model is the right approach to satisfy the HTTP transport spec requirement.

A few initial code comments to consider. I will do another pass soon.

stevejgordon · 2026-08-07T08:11:51Z

+        this.TryFlush();
+    }
+
+    public Task FlushAsync()


Should this accept a CancellationToken passed through from StopAsync?

Correct, we discovered this as well in the internal overhaul to prevent hangs in the pipe. Using cancellation token here gives the control to user.

0b456cc

stevejgordon · 2026-08-07T08:16:08Z

+    {
+        lock (this.frameLock)
+        {
+            this.isBusy = false;


This is correct for the "happy path", but what if the the web socket connection is closed or dropped? I think we'd need a mechanism for WsReciever to notify through when that happens so isBusy can be reset?

Correct, retry / reconnect paths are missing currently intentionally. Seems a larger scope to focus on separately.

stevejgordon · 2026-08-07T08:16:57Z

+    public async Task StopAsync(CancellationToken token = default)
+    {
+        // Drain queued data.
+        await this.FlushAsync()


Pass the CancellationToken here?

#4930 (comment)

stevejgordon · 2026-08-07T08:17:08Z

+        this.AppendMessage(MessageBuilderHelper.AppendAgentDisconnect);
+
+        // Send disconnect.
+        await this.FlushAsync()


#4930 (comment)

stevejgordon · 2026-08-07T08:20:17Z

    IFrameBuilder AddCustomMessage(string capability, string type, ReadOnlyMemory<byte> data);

-    AgentToServer Build();
+    IFrameBuilder Clear();


Does this belong on this interface? It's beyond the concern of this abstraction. It seems to only be used from a test and could still existing on the FrameBuilder directly for that call site.

Seems it became a leftover, removed.

stevejgordon · 2026-08-07T08:30:00Z

+                .ConfigureAwait(false);
+        }
+
+        this.AppendMessage(MessageBuilderHelper.AppendIdentification);


Should we flush here to ensure identification is sent before heartbeats (and other services) are started? Otherwise, a lost identification message will cause heartbeats to be sent to a server it had never seen.

TryFlush should send it instantly since the pipe is initially free. I added Flush just in case, so nothing weird should not happen.

stevejgordon · 2026-08-07T08:31:34Z

 internal sealed class FrameProcessor
 {
    private readonly ConcurrentDictionary<Type, IReadOnlyList<object>> listeners = [];
+    private readonly ConcurrentBag<Action<ServerToAgent>> internalListeners = [];


Does this need to be ConcurrentBag? One one internal listener is accepted in the ctor, can we just store that?

The only issue today is that the pipe does not have a control over the processor (it does not construct it). Pipe itself is a user, like any internal services could be.

That doesn't explain the need for ConcurrentBag. Looking at the code, OpAmpPipe is the only caller of SubscribeToServerMessages, and it calls it exactly once from its constructor. There's also no Unsubscribe method, so the concurrent-collection semantics are never exercised.

A simple Action<ServerToAgent>? field would be clearer and more appropriate here (as it stands) and doesn't carry the misleading implication that subscribers can be dynamically added/removed concurrently.

Simple Action<ServerToAgent>? would meant that the owner should have set it. But because OpAmpPipe doesn't own it, then the control who sets it becomes vague. I do understand that there is a single use currently but such pattern leaves too many doors open.

I refactored this and removed that new functionality and reused the sub/unsub pattern already existing. This also helps in the future if you need to check multiple fields (sub-messages) without having any sync logic of waiting for multiple callbacks.

40877b3

…metry-dotnet-contrib into opamp-com-reworks

stevejgordon · 2026-08-19T10:02:17Z


+        foreach (var listener in this.internalListeners)
+        {
+            listener.Invoke(message);


Should we wrap this in try/catch to avoid exceptions bubbling out? We already do that for public listeners.

stevejgordon · 2026-08-19T10:12:56Z

+
+namespace OpenTelemetry.OpAmp.Client.Internal;
+
+internal sealed class OpAmpPipe : IDisposable


Genral comment for the OpAmpPipe implementation - For HTTP, gating on server-frame receipt looks spec-aligned: every POST gets a ServerToAgent response and PlainHttpTransport processes it before SendAsync returns. For WebSocket, the spec (in my re-reading) is full-duplex with no response requirement, so only clearing isBusy in OnServerFrameReceived can stall the pipe if the server doesn’t reply to every agent send (or unblock on an unrelated server message). I think we need to consider transport-specific pipe behavior: response-gated for HTTP, send-completion-gated for WebSocket.

stevejgordon · 2026-08-19T10:18:46Z

+        {
+            OpAmpClientEventSource.Log.SendingMessage();
+
+            await this.transport.SendAsync(message, this.tokenSource.Token)


This token is only cancelled on dispose. Should we accept the tokens through from FlushAsync/StopAsync or link the the caller cancellation tokens with tokenSource.Token to ensure correct in-flight cancellation?

seems a good catch, 3ae9596

stevejgordon · 2026-08-19T10:21:00Z


-            await this.dispatcher.DispatchHeartbeatAsync(report, this.cts.Token)
-                .ConfigureAwait(false);
+            this.pipe.AppendMessage(MessageBuilderHelper.AppendHeartbeat(report));


This no longer passes the cancellation token. Can this race with Stop? Would checking this.cts.IsCancellationRequested first before appending be reasonable here?

yep, seems IsCancellationRequested check is appropriate here. d288329

stevejgordon · 2026-08-19T10:27:01Z

@RassK Sorry, my feedback and your new commit overlapped, so some of my comments may no longer apply. I like the direction of the refactor.

stevejgordon · 2026-08-19T10:28:29Z

    {
        var message = ServerToAgent.Parser.ParseFrom(sequence);

+        this.Dispatch(new ServerToAgentMessage(message));


Worth checking listeners.ContainsKey(typeof(ServerToAgentMessage)) before allocating and dispatching a message no one is subscriber to?

that might be useful for other types as well.

RassK added 6 commits August 3, 2026 12:31

replace framedispatcher with pipe.

be5b46b

update concurrency tests

e508633

fix race conditions

f02c450

fix docs

0a3f174

fix logs

eb6502d

fixes

c301171

RassK requested a review from a team as a code owner August 4, 2026 11:12

github-actions Bot assigned RassK and stevejgordon Aug 4, 2026

github-actions Bot requested a review from stevejgordon August 4, 2026 11:12

github-actions Bot added the comp:opamp.client Things related to OpenTelemetry.OpAmp.Client label Aug 4, 2026

opentelemetry-pr-dashboard Bot mentioned this pull request Aug 4, 2026

Pull Request Dashboard #4846

Open

RassK added 12 commits August 4, 2026 15:17

fix double line

6ee9658

Merge branch 'main' into opamp-com-reworks

79eed9d

add fb clear tests

4bb0853

Merge branch 'opamp-com-reworks' of https://github.com/RassK/opentele…

1adf896

…metry-dotnet-contrib into opamp-com-reworks

fix public listener throw releasing pipe

625433a

Merge branch 'opamp-com-reworks' of https://github.com/RassK/opentele…

849e366

…metry-dotnet-contrib into opamp-com-reworks

remove force flush

5be8bb7

fix heartbeat tests

0df50b0

fix docs and names

de7e977

fix flush

9b1c80d

fix comment

2b50a48

Merge branch 'main' into opamp-com-reworks

c0bc6b4

stevejgordon reviewed Aug 7, 2026

View reviewed changes

RassK added 3 commits August 7, 2026 12:13

add cancellation token and enable net462 opamp pipe tests

0b456cc

Merge branch 'opamp-com-reworks' of https://github.com/RassK/opentele…

ece20da

…metry-dotnet-contrib into opamp-com-reworks

fixes

4234dbe

RassK added 2 commits August 7, 2026 12:37

add flush

f4ed7b0

Add changelog entry

b8a31e7

Kielek requested a review from stevejgordon August 13, 2026 10:45

RassK added 2 commits August 19, 2026 13:03

remove internal listeners

40877b3

Merge branch 'main' into opamp-com-reworks

af953be

stevejgordon reviewed Aug 19, 2026

View reviewed changes

RassK added 2 commits August 19, 2026 14:45

fix cancellationtoken

3ae9596

add cancel check

d288329


		namespace OpenTelemetry.OpAmp.Client.Internal;

		internal sealed class OpAmpPipe : IDisposable

Uh oh!

Conversation

RassK commented Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What

Changes

Breaking changes ❗

Notes

Merge requirement checklist

Uh oh!

opentelemetry-pr-dashboard Bot commented Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Pull request dashboard status

Uh oh!

codecov Bot commented Aug 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

stevejgordon left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

stevejgordon commented Aug 19, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

RassK commented Aug 4, 2026 •

edited

Loading

opentelemetry-pr-dashboard Bot commented Aug 4, 2026 •

edited

Loading

codecov Bot commented Aug 4, 2026 •

edited

Loading

stevejgordon commented Aug 19, 2026 •

edited

Loading