-
Notifications
You must be signed in to change notification settings - Fork 879
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix netty 4.1 instrumentation not removing future listeners (#2851)
* Fix netty 4.1 instrumentation not removing future listeners * Code review follow-up * Use InstrumentationContext
- Loading branch information
Mateusz Rzeszutek
authored
Apr 27, 2021
1 parent
2df0bb4
commit 77f8be8
Showing
5 changed files
with
232 additions
and
32 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
100 changes: 100 additions & 0 deletions
100
...in/java/io/opentelemetry/javaagent/instrumentation/netty/v4_1/FutureListenerWrappers.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| package io.opentelemetry.javaagent.instrumentation.netty.v4_1; | ||
|
|
||
| import io.netty.util.concurrent.Future; | ||
| import io.netty.util.concurrent.GenericFutureListener; | ||
| import io.netty.util.concurrent.GenericProgressiveFutureListener; | ||
| import io.netty.util.concurrent.ProgressiveFuture; | ||
| import io.opentelemetry.context.Context; | ||
| import io.opentelemetry.context.Scope; | ||
| import io.opentelemetry.javaagent.instrumentation.api.ContextStore; | ||
|
|
||
| public final class FutureListenerWrappers { | ||
| @SuppressWarnings("unchecked") | ||
| public static GenericFutureListener<? extends Future<? super Void>> wrap( | ||
| ContextStore<GenericFutureListener, GenericFutureListener> contextStore, | ||
| Context context, | ||
| GenericFutureListener<? extends Future<? super Void>> delegate) { | ||
| if (delegate instanceof WrappedFutureListener | ||
| || delegate instanceof WrappedProgressiveFutureListener) { | ||
| return delegate; | ||
| } | ||
| return (GenericFutureListener<? extends Future<? super Void>>) | ||
| contextStore.putIfAbsent( | ||
| delegate, | ||
| () -> { | ||
| if (delegate instanceof GenericProgressiveFutureListener) { | ||
| return new WrappedProgressiveFutureListener( | ||
| context, | ||
| (GenericProgressiveFutureListener<ProgressiveFuture<? super Void>>) delegate); | ||
| } else { | ||
| return new WrappedFutureListener( | ||
| context, (GenericFutureListener<Future<? super Void>>) delegate); | ||
| } | ||
| }); | ||
| } | ||
|
|
||
| public static GenericFutureListener<? extends Future<? super Void>> getWrapper( | ||
| ContextStore<GenericFutureListener, GenericFutureListener> contextStore, | ||
| GenericFutureListener<? extends Future<? super Void>> delegate) { | ||
| GenericFutureListener<? extends Future<? super Void>> wrapper = | ||
| (GenericFutureListener<? extends Future<? super Void>>) contextStore.get(delegate); | ||
| return wrapper == null ? delegate : wrapper; | ||
| } | ||
|
|
||
| private static final class WrappedFutureListener | ||
| implements GenericFutureListener<Future<? super Void>> { | ||
|
|
||
| private final Context context; | ||
| private final GenericFutureListener<Future<? super Void>> delegate; | ||
|
|
||
| private WrappedFutureListener( | ||
| Context context, GenericFutureListener<Future<? super Void>> delegate) { | ||
| this.context = context; | ||
| this.delegate = delegate; | ||
| } | ||
|
|
||
| @Override | ||
| public void operationComplete(Future<? super Void> future) throws Exception { | ||
| try (Scope ignored = context.makeCurrent()) { | ||
| delegate.operationComplete(future); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static final class WrappedProgressiveFutureListener | ||
| implements GenericProgressiveFutureListener<ProgressiveFuture<? super Void>> { | ||
|
|
||
| private final Context context; | ||
| private final GenericProgressiveFutureListener<ProgressiveFuture<? super Void>> delegate; | ||
|
|
||
| private WrappedProgressiveFutureListener( | ||
| Context context, | ||
| GenericProgressiveFutureListener<ProgressiveFuture<? super Void>> delegate) { | ||
| this.context = context; | ||
| this.delegate = delegate; | ||
| } | ||
|
|
||
| @Override | ||
| public void operationProgressed( | ||
| ProgressiveFuture<? super Void> progressiveFuture, long l, long l1) throws Exception { | ||
| try (Scope ignored = context.makeCurrent()) { | ||
| delegate.operationProgressed(progressiveFuture, l, l1); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void operationComplete(ProgressiveFuture<? super Void> progressiveFuture) | ||
| throws Exception { | ||
| try (Scope ignored = context.makeCurrent()) { | ||
| delegate.operationComplete(progressiveFuture); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private FutureListenerWrappers() {} | ||
| } |
28 changes: 0 additions & 28 deletions
28
...ain/java/io/opentelemetry/javaagent/instrumentation/netty/v4_1/WrappedFutureListener.java
This file was deleted.
Oops, something went wrong.
57 changes: 57 additions & 0 deletions
57
instrumentation/netty/netty-4.1/javaagent/src/test/groovy/ChannelFutureTest.groovy
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| /* | ||
| * Copyright The OpenTelemetry Authors | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
|
|
||
| import io.netty.channel.embedded.EmbeddedChannel | ||
| import io.netty.util.concurrent.Future | ||
| import io.netty.util.concurrent.GenericFutureListener | ||
| import io.netty.util.concurrent.GenericProgressiveFutureListener | ||
| import io.netty.util.concurrent.ProgressiveFuture | ||
| import io.opentelemetry.instrumentation.test.AgentInstrumentationSpecification | ||
| import java.util.concurrent.TimeUnit | ||
| import java.util.concurrent.atomic.AtomicInteger | ||
|
|
||
| class ChannelFutureTest extends AgentInstrumentationSpecification { | ||
| // regression test for https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/2705 | ||
| def "should clean up wrapped listeners"() { | ||
| given: | ||
| def channel = new EmbeddedChannel() | ||
| def counter = new AtomicInteger() | ||
|
|
||
| def listener1 = newListener(counter) | ||
| channel.closeFuture().addListener(listener1) | ||
| channel.closeFuture().removeListener(listener1) | ||
|
|
||
| def listener2 = newListener(counter) | ||
| def listener3 = newProgressiveListener(counter) | ||
| channel.closeFuture().addListeners(listener2, listener3) | ||
| channel.closeFuture().removeListeners(listener2, listener3) | ||
|
|
||
| when: | ||
| channel.close().await(5, TimeUnit.SECONDS) | ||
|
|
||
| then: | ||
| counter.get() == 0 | ||
| } | ||
|
|
||
| private static GenericFutureListener newListener(AtomicInteger counter) { | ||
| new GenericFutureListener() { | ||
| void operationComplete(Future future) throws Exception { | ||
| counter.incrementAndGet() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static GenericFutureListener newProgressiveListener(AtomicInteger counter) { | ||
| new GenericProgressiveFutureListener() { | ||
| void operationProgressed(ProgressiveFuture future, long progress, long total) throws Exception { | ||
| counter.incrementAndGet() | ||
| } | ||
|
|
||
| void operationComplete(Future future) throws Exception { | ||
| counter.incrementAndGet() | ||
| } | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters