Skip to content

Commit 891c295

Browse files
gunlee01gunlee01
authored
Merge pull request #982 from scouter-project/develop
Develop
2 parents 3818904 + 474c456 commit 891c295

8 files changed

Lines changed: 120 additions & 10 deletions

File tree

‎scouter.agent.java/src/main/java/reactor/core/publisher/ScouterOptimizableOperatorProxy.java‎

Lines changed: 8 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,8 @@
2121
import scouter.agent.Logger;
2222
import scouter.agent.util.Tuple;
2323

24+
import java.lang.reflect.Method;
25+
2426
/**
2527
* @author Gun Lee (gunlee01@gmail.com) on 2020/08/08
2628
*/
@@ -31,7 +33,8 @@ public class ScouterOptimizableOperatorProxy {
3133
public static boolean accessible = false;
3234
public static boolean first = true;
3335

34-
public static Tuple.StringLongPair nameOnCheckpoint(Object candidate, int maxScanDepth) {
36+
public static Tuple.StringLongPair nameOnCheckpoint(Object candidate, int maxScanDepth, boolean isReactor34,
37+
Method isCheckpoint) {
3538
try {
3639
if (!accessible && first) {
3740
try {
@@ -54,12 +57,14 @@ public static Tuple.StringLongPair nameOnCheckpoint(Object candidate, int maxSca
5457
}
5558
if (closeAssembly instanceof MonoOnAssembly) {
5659
FluxOnAssembly.AssemblySnapshot snapshot = ((MonoOnAssembly) closeAssembly).stacktrace;
57-
if (snapshot != null && snapshot.checkpointed) {
60+
boolean cp = isReactor34 ? (Boolean) isCheckpoint.invoke(snapshot) : snapshot.checkpointed;
61+
if (snapshot != null && cp) {
5862
return new Tuple.StringLongPair(snapshot.cached, snapshot.hashCode());
5963
}
6064
} else if (closeAssembly instanceof FluxOnAssembly) {
6165
FluxOnAssembly.AssemblySnapshot snapshot = ((FluxOnAssembly) closeAssembly).snapshotStack;
62-
if (snapshot != null && snapshot.checkpointed) {
66+
boolean cp = isReactor34 ? (Boolean) isCheckpoint.invoke(snapshot) : snapshot.checkpointed;
67+
if (snapshot != null && cp) {
6368
return new Tuple.StringLongPair(snapshot.cached, snapshot.hashCode());
6469
}
6570
}

‎scouter.agent.java/src/main/java/scouter/agent/proxy/IReactiveSupport.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,4 +28,6 @@ public interface IReactiveSupport {
2828

2929
String dumpScannable(TraceContext traceContext, TraceContext.TimedScannable timedScannable, long now);
3030

31+
boolean isReactor34();
32+
3133
}

‎scouter.agent.java/src/main/java/scouter/agent/proxy/ReactiveSupportFactory.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,11 @@ public Object monoCoroutineContextHook(Object coroutineContext, TraceContext tra
3939
public String dumpScannable(TraceContext traceContext, TraceContext.TimedScannable timedScannable, long now) {
4040
return null;
4141
}
42+
43+
@Override
44+
public boolean isReactor34() {
45+
return false;
46+
}
4247
};
4348

4449
public static IReactiveSupport create(ClassLoader parent) {

‎scouter.agent.java/src/main/java/scouter/xtra/reactive/ReactiveSupport.java‎

Lines changed: 33 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,13 +42,33 @@
4242
import scouter.lang.step.ParameterizedMessageStep;
4343
import scouter.util.StringUtil;
4444

45+
import java.lang.reflect.Method;
4546
import java.util.function.BiFunction;
4647
import java.util.function.Consumer;
4748
import java.util.function.Function;
4849

4950
public class ReactiveSupport implements IReactiveSupport {
5051

5152
static Configure configure = Configure.getInstance();
53+
private Method subscriberContextMethod;
54+
private static Method isCheckpoint;
55+
private static boolean isReactor34;
56+
57+
public ReactiveSupport() {
58+
isReactor34 = ReactiveSupportUtils.isSupportReactor34();
59+
try {
60+
if (isReactor34) {
61+
subscriberContextMethod = Mono.class.getMethod("contextWrite", Function.class);
62+
Class<?> assemblySnapshotClass = Class.forName("reactor.core.publisher.FluxOnAssembly$AssemblySnapshot");
63+
isCheckpoint = assemblySnapshotClass.getDeclaredMethod("isCheckpoint");
64+
isCheckpoint.setAccessible(true);
65+
} else {
66+
subscriberContextMethod = Mono.class.getMethod("subscriberContext", Function.class);
67+
}
68+
} catch (Exception e) {
69+
throw new RuntimeException(e);
70+
}
71+
}
5272

5373
@Override
5474
public Object subscriptOnContext(Object mono0, final TraceContext traceContext) {
@@ -58,12 +78,17 @@ public Object subscriptOnContext(Object mono0, final TraceContext traceContext)
5878
}
5979
Mono<?> mono = (Mono<?>) mono0;
6080
traceContext.isReactiveTxidMarked = true;
61-
return mono.subscriberContext(new Function<Context, Context>() {
81+
82+
Mono<?> monoChain;
83+
Function<Context, Context> func = new Function<Context, Context>() {
6284
@Override
6385
public Context apply(Context context) {
6486
return context.put(TraceContext.class, traceContext);
6587
}
66-
}).doOnSuccess(new Consumer<Object>() {
88+
};
89+
90+
monoChain = (Mono<?>) subscriberContextMethod.invoke(mono, func);
91+
return monoChain.doOnSuccess(new Consumer<Object>() {
6792
@Override
6893
public void accept(Object o) {
6994
TraceMain.endHttpService(new TraceMain.Stat(traceContext), null);
@@ -164,7 +189,7 @@ public TxidLifter(CoreSubscriber<T> coreSubscriber, Scannable scannable, Publish
164189
this.traceContext = traceContext;
165190

166191
Tuple.StringLongPair checkpointPair = ScouterOptimizableOperatorProxy
167-
.nameOnCheckpoint(scannable, configure.profile_reactor_checkpoint_search_depth);
192+
.nameOnCheckpoint(scannable, configure.profile_reactor_checkpoint_search_depth, isReactor34, isCheckpoint);
168193
checkpointDesc = checkpointPair.aString;
169194

170195
Integer parentDepth = context.getOrDefault(SubscribeDepth.class, 0);
@@ -321,4 +346,9 @@ public String dumpScannable(TraceContext traceContext, TraceContext.TimedScannab
321346
ScouterOptimizableOperatorProxy.appendSources4Dump(scannable, builder, configure.profile_reactor_checkpoint_search_depth);
322347
return builder.toString();
323348
}
349+
350+
@Override
351+
public boolean isReactor34() {
352+
return isReactor34;
353+
}
324354
}

‎scouter.agent.java/src/main/java/scouter/xtra/reactive/ReactiveSupportUtils.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
package scouter.xtra.reactive;
2+
3+
import reactor.core.publisher.Mono;
4+
import scouter.agent.Logger;
5+
6+
import java.util.function.Function;
7+
8+
/**
9+
* @author Gun Lee (gunlee01@gmail.com) on 2/21/24
10+
*/
11+
public class ReactiveSupportUtils {
12+
13+
public static boolean isSupportReactor34() {
14+
try {
15+
Class<?> assemblySnapshotClass = Class.forName("reactor.core.publisher.FluxOnAssembly$AssemblySnapshot");
16+
assemblySnapshotClass.getDeclaredMethod("isCheckpoint");
17+
18+
Class<Mono> monoClass = Mono.class;
19+
Class<?>[] parameterTypes = new Class<?>[]{Function.class};
20+
monoClass.getMethod("contextWrite", parameterTypes);
21+
22+
return true;
23+
24+
} catch (ClassNotFoundException | NoSuchMethodException e) {
25+
e.printStackTrace();
26+
Logger.println("R301", e.getMessage());
27+
return false;
28+
} catch (Exception e) {
29+
e.printStackTrace();
30+
Logger.println("R302", e.getMessage(), e);
31+
return false;
32+
}
33+
}
34+
}

‎scouter.agent.java/src/main/java/scouter/xtra/reactive/ReactiveSupportWithCoroutine.java‎

Lines changed: 34 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -45,13 +45,33 @@
4545
import scouter.lang.step.ParameterizedMessageStep;
4646
import scouter.util.StringUtil;
4747

48+
import java.lang.reflect.Method;
4849
import java.util.function.BiFunction;
4950
import java.util.function.Consumer;
5051
import java.util.function.Function;
5152

5253
public class ReactiveSupportWithCoroutine implements IReactiveSupport {
5354

5455
static Configure configure = Configure.getInstance();
56+
private Method subscriberContextMethod;
57+
private static Method isCheckpoint;
58+
private static boolean isReactor34;
59+
60+
public ReactiveSupportWithCoroutine() {
61+
isReactor34 = ReactiveSupportUtils.isSupportReactor34();
62+
try {
63+
if (isReactor34) {
64+
subscriberContextMethod = Mono.class.getMethod("contextWrite", Function.class);
65+
Class<?> assemblySnapshotClass = Class.forName("reactor.core.publisher.FluxOnAssembly$AssemblySnapshot");
66+
isCheckpoint = assemblySnapshotClass.getDeclaredMethod("isCheckpoint");
67+
isCheckpoint.setAccessible(true);
68+
} else {
69+
subscriberContextMethod = Mono.class.getMethod("subscriberContext", Function.class);
70+
}
71+
} catch (Exception e) {
72+
throw new RuntimeException(e);
73+
}
74+
}
5575

5676
@Override
5777
public Object subscriptOnContext(Object mono0, final TraceContext traceContext) {
@@ -61,12 +81,16 @@ public Object subscriptOnContext(Object mono0, final TraceContext traceContext)
6181
}
6282
Mono<?> mono = (Mono<?>) mono0;
6383
traceContext.isReactiveTxidMarked = true;
64-
return mono.subscriberContext(new Function<Context, Context>() {
84+
Mono<?> monoChain;
85+
Function<Context, Context> func = new Function<Context, Context>() {
6586
@Override
6687
public Context apply(Context context) {
6788
return context.put(TraceContext.class, traceContext);
6889
}
69-
}).doOnSuccess(new Consumer<Object>() {
90+
};
91+
92+
monoChain = (Mono<?>) subscriberContextMethod.invoke(mono, func);
93+
return monoChain.doOnSuccess(new Consumer<Object>() {
7094
@Override
7195
public void accept(Object o) {
7296
TraceMain.endHttpService(new TraceMain.Stat(traceContext), null);
@@ -161,6 +185,7 @@ public static class TxidLifter<T> implements SpanSubscription<T>, Scannable {
161185
private final String checkpointDesc;
162186
private final Integer depth;
163187
private Subscription orgSubs;
188+
private boolean isReactor34;
164189

165190
private enum ReactorCheckPointType {
166191
ON_SUBSCRIBE,
@@ -176,9 +201,10 @@ public TxidLifter(CoreSubscriber<T> coreSubscriber, Scannable scannable, Publish
176201
this.scannable = scannable;
177202
this.publisher = publisher;
178203
this.traceContext = traceContext;
204+
this.isReactor34 = isReactor34;
179205

180206
Tuple.StringLongPair checkpointPair = ScouterOptimizableOperatorProxy
181-
.nameOnCheckpoint(scannable, configure.profile_reactor_checkpoint_search_depth);
207+
.nameOnCheckpoint(scannable, configure.profile_reactor_checkpoint_search_depth, isReactor34, isCheckpoint);
182208
checkpointDesc = checkpointPair.aString;
183209

184210
Integer parentDepth = context.getOrDefault(SubscribeDepth.class, 0);
@@ -335,4 +361,9 @@ public String dumpScannable(TraceContext traceContext, TraceContext.TimedScannab
335361
ScouterOptimizableOperatorProxy.appendSources4Dump(scannable, builder, configure.profile_reactor_checkpoint_search_depth);
336362
return builder.toString();
337363
}
364+
365+
@Override
366+
public boolean isReactor34() {
367+
return isReactor34;
368+
}
338369
}

‎scouter.server/src/main/java/scouter/server/CounterManager.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -538,6 +538,9 @@ public String normalizeCountersXml(String xmlString) throws JAXBException {
538538
if (counters.familys == null) {
539539
counters.familys = new Familys0(new ArrayList<>());
540540
}
541+
if (counters.familys.family == null) {
542+
counters.familys.family = new ArrayList<>();
543+
}
541544
List<Family0> families =
542545
counters.familys.family.stream()
543546
.filter(f -> !f.name.contains("zws-metric."))

‎scouter.webapp/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@
7373
<dependency>
7474
<groupId>com.fasterxml.jackson.core</groupId>
7575
<artifactId>jackson-core</artifactId>
76-
<version>2.8.11</version>
76+
<version>2.12.6</version>
7777
</dependency>
7878
<dependency>
7979
<groupId>com.fasterxml.jackson.core</groupId>

0 commit comments

Comments
 (0)