Add support for multiple pipelines in OpenTelemetryBuilder with Activity #735
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,105 @@ | ||
| // <copyright file="BroadcastActivityProcessor.cs" company="OpenTelemetry Authors"> | ||
| // Copyright The OpenTelemetry Authors | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
| // </copyright> | ||
|
|
||
| using System; | ||
| using System.Collections.Generic; | ||
| using System.Diagnostics; | ||
| using System.Linq; | ||
| using System.Threading; | ||
| using System.Threading.Tasks; | ||
| using OpenTelemetry.Internal; | ||
|
|
||
| namespace OpenTelemetry.Trace.Export.Internal | ||
| { | ||
| internal class BroadcastActivityProcessor : ActivityProcessor, IDisposable | ||
| { | ||
| private readonly IEnumerable<ActivityProcessor> processors; | ||
|
Member
There was a problem hiding this comment. Potential perf concern: |
||
|
|
||
| public BroadcastActivityProcessor(IEnumerable<ActivityProcessor> processors) | ||
| { | ||
| if (processors == null) | ||
| { | ||
| throw new ArgumentNullException(nameof(processors)); | ||
| } | ||
|
|
||
| if (!processors.Any()) | ||
| { | ||
| throw new ArgumentException($"{nameof(processors)} collection is empty"); | ||
| } | ||
|
|
||
| this.processors = processors; | ||
|
Member
There was a problem hiding this comment. Should this be a shallow copy? What if the passed in |
||
| } | ||
|
|
||
| public override void OnEnd(Activity activity) | ||
| { | ||
| foreach (var processor in this.processors) | ||
| { | ||
| try | ||
| { | ||
| processor.OnEnd(activity); | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| OpenTelemetrySdkEventSource.Log.SpanProcessorException("OnEnd", e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public override void OnStart(Activity activity) | ||
|
Member
There was a problem hiding this comment. nit: Probably more common to have Start before End. |
||
| { | ||
| foreach (var processor in this.processors) | ||
| { | ||
| try | ||
| { | ||
| processor.OnStart(activity); | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| OpenTelemetrySdkEventSource.Log.SpanProcessorException("OnStart", e); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public override Task ShutdownAsync(CancellationToken cancellationToken) | ||
| { | ||
| var tasks = new List<Task>(); | ||
| foreach (var processor in this.processors) | ||
| { | ||
| tasks.Add(processor.ShutdownAsync(cancellationToken)); | ||
| } | ||
|
|
||
| return Task.WhenAll(tasks); | ||
|
Contributor
There was a problem hiding this comment. The
Member
Author
There was a problem hiding this comment. The current code is vulnerable to one misbehaving processor. Not just in shutdown but even in start/stop as well. We need to address it properly. |
||
| } | ||
|
|
||
| public void Dispose() | ||
|
Member
There was a problem hiding this comment. This is different from what I've normally seen, worth double checking https://docs.microsoft.com/en-us/dotnet/standard/garbage-collection/implementing-dispose#implement-the-dispose-pattern.
Member
There was a problem hiding this comment. IMO because it's
Member
There was a problem hiding this comment. There are few things we need to address:
Member
Author
There was a problem hiding this comment. Agree with all of the above - But my goal with this PR is to just mimic the existing behavior with Spans (this pr was made with pure 'copy paste & minor adjustments' ). I'd prefer to get the parity 1st, then fix underlying issues. Let me know if this approach is okay?
Member
There was a problem hiding this comment. Yeah, these are not blockers for this PR.
Member
Author
There was a problem hiding this comment.
Yes this is needed. Modified OTSDK to call dispose on the Activityprocessor it has.
Member
Author
There was a problem hiding this comment. Will follow up on why Span was not calling Shutdown in SimpleProcessor! (Was doing it right for batching one). Anyway - both will be fixed in next PR. |
||
| { | ||
| foreach (var processor in this.processors) | ||
| { | ||
| try | ||
| { | ||
| if (processor is IDisposable disposable) | ||
| { | ||
| disposable.Dispose(); | ||
| } | ||
| } | ||
| catch (Exception e) | ||
| { | ||
| OpenTelemetrySdkEventSource.Log.SpanProcessorException("Dispose", e); | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The name is misleading since "broadcast" normally means there is no need to specify the target.
Probably borrow from Java/Python:
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
FanOutActivityProcessor? 🤷