Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update ForceFlush behavior to meet with the latest spec requirements #2388

Merged
merged 4 commits into from
Sep 20, 2021
+63 −80
Merged

Update ForceFlush behavior to meet with the latest spec requirements #2388

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a91e22c
Update ForceFlush behavior to meet with the latest spec requirements
reyang Sep 20, 2021
f54fd31
update changelog
reyang Sep 20, 2021
0fadd33
update error message
reyang Sep 20, 2021
a5a19d7
update test case
reyang Sep 20, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 5 additions & 5 deletions src/OpenTelemetry/BaseExporter.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,8 +62,8 @@ public abstract class BaseExporter<T> : IDisposable
/// shutdown completed or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand All @@ -79,7 +79,7 @@ public bool Shutdown(int timeoutMilliseconds = Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

if (Interlocked.Increment(ref this.shutdownCount) > 1)
Expand Down Expand Up @@ -110,8 +110,8 @@ public void Dispose()
/// thread until shutdown completed or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand Down
20 changes: 10 additions & 10 deletions src/OpenTelemetry/BaseProcessor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,8 +68,8 @@ public virtual void OnEnd(T data)
/// completed, shutdown signaled or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when flush succeeded; otherwise, <c>false</c>.
Expand All @@ -84,7 +84,7 @@ public bool ForceFlush(int timeoutMilliseconds = Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

try
Expand All @@ -103,8 +103,8 @@ public bool ForceFlush(int timeoutMilliseconds = Timeout.Infinite)
/// shutdown completed or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand All @@ -120,7 +120,7 @@ public bool Shutdown(int timeoutMilliseconds = Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

if (Interlocked.Increment(ref this.shutdownCount) > 1)
Expand Down Expand Up @@ -156,8 +156,8 @@ internal virtual void SetParentProvider(BaseProvider parentProvider)
/// thread until flush completed, shutdown signaled or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when flush succeeded; otherwise, <c>false</c>.
Expand All @@ -177,8 +177,8 @@ protected virtual bool OnForceFlush(int timeoutMilliseconds)
/// thread until shutdown completed or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand Down
5 changes: 5 additions & 0 deletions src/OpenTelemetry/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,11 @@

## Unreleased

* Changed `CompositeProcessor<T>.OnForceFlush` to meet with the spec
requirement. Now the SDK will invoke `ForceFlush` on all registered
processors, even if there is a timeout.
([#2388](https://github.com/open-telemetry/opentelemetry-dotnet/pull/2388))

## 1.2.0-alpha3

Released 2021-Sep-13
Expand Down
19 changes: 5 additions & 14 deletions src/OpenTelemetry/CompositeProcessor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,37 +95,28 @@ public override void OnStart(T data)
/// <inheritdoc/>
protected override bool OnForceFlush(int timeoutMilliseconds)
{
var result = true;
var cur = this.head;

var sw = Stopwatch.StartNew();

while (cur != null)
{
if (timeoutMilliseconds == Timeout.Infinite)
{
_ = cur.Value.ForceFlush(Timeout.Infinite);
result = cur.Value.ForceFlush(Timeout.Infinite) && result;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can use &= operator

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think no, &= has different semantic (short-circuit ForceFlush when result is true) than what we need (ForceFlush would always be called).

}
else
{
var timeout = timeoutMilliseconds - sw.ElapsedMilliseconds;

if (timeout <= 0)
{
return false;
}

var succeeded = cur.Value.ForceFlush((int)timeout);

if (!succeeded)
{
return false;
}
// notify all the processors, even if we run overtime
result = cur.Value.ForceFlush((int)Math.Max(timeout, 0)) && result;
}

cur = cur.Next;
}

return true;
return result;
}

/// <inheritdoc/>
Expand Down
5 changes: 0 additions & 5 deletions src/OpenTelemetry/Metrics/CompositeMetricReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,11 +79,6 @@ protected override bool ProcessMetrics(Batch<Metric> metrics, int timeoutMillise
/// <inheritdoc/>
protected override bool OnCollect(int timeoutMilliseconds = Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
}

var result = true;
var cur = this.head;
var sw = Stopwatch.StartNew();
Expand Down
28 changes: 14 additions & 14 deletions src/OpenTelemetry/Metrics/MeterProviderExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,8 +29,8 @@ public static class MeterProviderExtensions
/// </summary>
/// <param name="provider">MeterProviderSdk instance on which ForceFlush will be called.</param>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when force flush succeeded; otherwise, <c>false</c>.
Expand All @@ -48,13 +48,13 @@ public static bool ForceFlush(this MeterProvider provider, int timeoutMillisecon
throw new ArgumentNullException(nameof(provider));
}

if (provider is MeterProviderSdk meterProviderSdk)
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
}
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

if (provider is MeterProviderSdk meterProviderSdk)
{
try
{
return meterProviderSdk.OnForceFlush(timeoutMilliseconds);
Expand All @@ -76,8 +76,8 @@ public static bool ForceFlush(this MeterProvider provider, int timeoutMillisecon
/// </summary>
/// <param name="provider">MeterProviderSdk instance on which Shutdown will be called.</param>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand All @@ -96,13 +96,13 @@ public static bool Shutdown(this MeterProvider provider, int timeoutMilliseconds
throw new ArgumentNullException(nameof(provider));
}

if (provider is MeterProviderSdk meterProviderSdk)
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
}
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

if (provider is MeterProviderSdk meterProviderSdk)
{
if (Interlocked.Increment(ref meterProviderSdk.ShutdownCount) > 1)
{
return false; // shutdown already called
Expand Down
8 changes: 4 additions & 4 deletions src/OpenTelemetry/Metrics/MeterProviderSdk.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,8 +187,8 @@ internal Batch<Metric> Collect()
/// thread until flush completed or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when flush succeeded; otherwise, <c>false</c>.
Expand All @@ -208,8 +208,8 @@ internal bool OnForceFlush(int timeoutMilliseconds)
/// thread until shutdown completed or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand Down
4 changes: 2 additions & 2 deletions src/OpenTelemetry/Metrics/MetricReader.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ public bool Collect(int timeoutMilliseconds = Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

try
Expand Down Expand Up @@ -106,7 +106,7 @@ public bool Shutdown(int timeoutMilliseconds = Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

if (Interlocked.Increment(ref this.shutdownCount) > 1)
Expand Down
28 changes: 14 additions & 14 deletions src/OpenTelemetry/Trace/TracerProviderExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,8 @@ public static TracerProvider AddProcessor(this TracerProvider provider, BaseProc
/// </summary>
/// <param name="provider">TracerProviderSdk instance on which ForceFlush will be called.</param>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when force flush succeeded; otherwise, <c>false</c>.
Expand All @@ -68,13 +68,13 @@ public static bool ForceFlush(this TracerProvider provider, int timeoutMilliseco
throw new ArgumentNullException(nameof(provider));
}

if (provider is TracerProviderSdk tracerProviderSdk)
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
}
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

if (provider is TracerProviderSdk tracerProviderSdk)
{
try
{
return tracerProviderSdk.OnForceFlush(timeoutMilliseconds);
Expand All @@ -95,8 +95,8 @@ public static bool ForceFlush(this TracerProvider provider, int timeoutMilliseco
/// </summary>
/// <param name="provider">TracerProviderSdk instance on which Shutdown will be called.</param>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand All @@ -115,13 +115,13 @@ public static bool Shutdown(this TracerProvider provider, int timeoutMillisecond
throw new ArgumentNullException(nameof(provider));
}

if (provider is TracerProviderSdk tracerProviderSdk)
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
if (timeoutMilliseconds < 0 && timeoutMilliseconds != Timeout.Infinite)
{
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative.");
}
throw new ArgumentOutOfRangeException(nameof(timeoutMilliseconds), timeoutMilliseconds, "timeoutMilliseconds should be non-negative or Timeout.Infinite.");
}

if (provider is TracerProviderSdk tracerProviderSdk)
{
if (Interlocked.Increment(ref tracerProviderSdk.ShutdownCount) > 1)
{
return false; // shutdown already called
Expand Down
4 changes: 2 additions & 2 deletions src/OpenTelemetry/Trace/TracerProviderSdk.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -265,8 +265,8 @@ internal bool OnForceFlush(int timeoutMilliseconds)
/// thread until shutdown completed or timed out.
/// </summary>
/// <param name="timeoutMilliseconds">
/// The number of milliseconds to wait, or <c>Timeout.Infinite</c> to
/// wait indefinitely.
/// The number (non-negative) of milliseconds to wait, or
/// <c>Timeout.Infinite</c> to wait indefinitely.
/// </param>
/// <returns>
/// Returns <c>true</c> when shutdown succeeded; otherwise, <c>false</c>.
Expand Down
12 changes: 2 additions & 10 deletions test/OpenTelemetry.Tests/Trace/CompositeActivityProcessorTests.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,16 +101,8 @@ public void CompositeActivityProcessor_ForceFlush(int timeout)
{
processor.ForceFlush(timeout);

if (timeout != 0)
{
Assert.True(p1.ForceFlushCalled);
Assert.True(p2.ForceFlushCalled);
}
else
{
Assert.False(p1.ForceFlushCalled);
Assert.False(p2.ForceFlushCalled);
}
Assert.True(p1.ForceFlushCalled);
Assert.True(p2.ForceFlushCalled);
}
}
}
Expand Down