Skip to content

Commit 87840bc

Browse files
garyrussellartembilan
garyrussell
authored and
artembilan
committed
GH-3056: Add Micrometer Metrics to Runtime Graph
Resolves #3056 * Retrieve meters each time the graph is rendered, not just when it is built.
1 parent 528203e commit 87840bc

File tree

15 files changed

+652
-90
lines changed

15 files changed

+652
-90
lines changed

‎spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationGraphServer.java‎

Lines changed: 71 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
import org.springframework.context.ApplicationContextAware;
3636
import org.springframework.context.ApplicationListener;
3737
import org.springframework.context.event.ContextRefreshedEvent;
38+
import org.springframework.integration.core.MessageSource;
3839
import org.springframework.integration.endpoint.IntegrationConsumer;
3940
import org.springframework.integration.endpoint.MessageProducerSupport;
4041
import org.springframework.integration.endpoint.PollingConsumer;
@@ -50,6 +51,8 @@
5051
import org.springframework.lang.Nullable;
5152
import org.springframework.messaging.MessageChannel;
5253
import org.springframework.messaging.MessageHandler;
54+
import org.springframework.messaging.PollableChannel;
55+
import org.springframework.util.ClassUtils;
5356

5457
/**
5558
* Builds the runtime object model graph.
@@ -62,7 +65,9 @@
6265
*/
6366
public class IntegrationGraphServer implements ApplicationContextAware, ApplicationListener<ContextRefreshedEvent> {
6467

65-
private static final float GRAPH_VERSION = 1.0f;
68+
private static final float GRAPH_VERSION = 1.1f;
69+
70+
private static MicrometerNodeEnhancer micrometerEnhancer;
6671

6772
private final NodeFactory nodeFactory = new NodeFactory();
6873

@@ -152,6 +157,10 @@ protected <T> Map<String, T> getBeansOfType(Class<T> type) {
152157
}
153158

154159
private synchronized Graph buildGraph() {
160+
if (micrometerEnhancer == null && ClassUtils.isPresent("io.micrometer.core.instrument.MeterRegistry",
161+
this.applicationContext.getClassLoader())) {
162+
micrometerEnhancer = new MicrometerNodeEnhancer(this.applicationContext);
163+
}
155164
String implementationVersion = IntegrationGraphServer.class.getPackage().getImplementationVersion();
156165
if (implementationVersion == null) {
157166
implementationVersion = "unknown - is Spring Integration running from the distribution jar?";
@@ -347,91 +356,128 @@ private static final class NodeFactory {
347356
super();
348357
}
349358

350-
private MessageChannelNode channelNode(String name, MessageChannel channel) {
351-
return new MessageChannelNode(this.nodeId.incrementAndGet(), name, channel);
359+
MessageChannelNode channelNode(String name, MessageChannel channel) {
360+
MessageChannelNode node;
361+
if (channel instanceof PollableChannel) {
362+
node = new PollableChannelNode(this.nodeId.incrementAndGet(), name, channel);
363+
}
364+
else {
365+
node = new MessageChannelNode(this.nodeId.incrementAndGet(), name, channel);
366+
}
367+
if (IntegrationGraphServer.micrometerEnhancer != null) {
368+
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
369+
}
370+
return node;
352371
}
353372

354-
private MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) {
373+
MessageGatewayNode gatewayNode(String name, MessagingGatewaySupport gateway) {
355374
String errorChannel = channelToBeanName(gateway.getErrorChannel());
356375
String requestChannel = channelToBeanName(gateway.getRequestChannel());
357376
return new MessageGatewayNode(this.nodeId.incrementAndGet(), name, gateway, requestChannel, errorChannel);
358377
}
359378

360379
@Nullable
361-
private String channelToBeanName(MessageChannel messageChannel) {
380+
String channelToBeanName(MessageChannel messageChannel) {
362381
return messageChannel instanceof NamedComponent
363382
? ((NamedComponent) messageChannel).getBeanName()
364383
: Objects.toString(messageChannel, null);
365384
}
366385

367-
private MessageProducerNode producerNode(String name, MessageProducerSupport producer) {
386+
MessageProducerNode producerNode(String name, MessageProducerSupport producer) {
368387
String errorChannel = channelToBeanName(producer.getErrorChannel());
369388
String outputChannel = channelToBeanName(producer.getOutputChannel());
370389
return new MessageProducerNode(this.nodeId.incrementAndGet(), name, producer,
371390
outputChannel, errorChannel);
372391
}
373392

374-
private MessageSourceNode sourceNode(String name, SourcePollingChannelAdapter adapter) {
393+
MessageSourceNode sourceNode(String name, SourcePollingChannelAdapter adapter) {
375394
String errorChannel = channelToBeanName(adapter.getDefaultErrorChannel());
376395
String outputChannel = channelToBeanName(adapter.getOutputChannel());
377-
return new MessageSourceNode(this.nodeId.incrementAndGet(), name, adapter.getMessageSource(),
396+
String nameToUse = name;
397+
MessageSource<?> source = adapter.getMessageSource();
398+
if (source instanceof NamedComponent) {
399+
nameToUse = ((NamedComponent) source).getComponentName();
400+
}
401+
MessageSourceNode node = new MessageSourceNode(this.nodeId.incrementAndGet(), nameToUse, source,
378402
outputChannel, errorChannel);
403+
if (IntegrationGraphServer.micrometerEnhancer != null) {
404+
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
405+
}
406+
return node;
379407
}
380408

381-
private MessageHandlerNode handlerNode(String name, IntegrationConsumer consumer) {
409+
MessageHandlerNode handlerNode(String nameArg, IntegrationConsumer consumer) {
382410
String outputChannelName = channelToBeanName(consumer.getOutputChannel());
383411
MessageHandler handler = consumer.getHandler();
412+
MessageHandlerNode node;
413+
String name = nameArg;
414+
if (handler instanceof NamedComponent) {
415+
name = ((NamedComponent) handler).getComponentName();
416+
}
384417
if (handler instanceof CompositeMessageHandler) {
385-
return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null,
418+
node = compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName, null,
386419
false);
387420
}
388421
else if (handler instanceof DiscardingMessageHandler) {
389-
return discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, null,
422+
node = discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName, null,
390423
false);
391424
}
392425
else if (handler instanceof MappingMessageRouterManagement) {
393-
return routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler,
426+
node = routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler,
394427
outputChannelName, null, false);
395428
}
396429
else if (handler instanceof RecipientListRouterManagement) {
397-
return recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler,
430+
node = recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler,
398431
outputChannelName, null, false);
399432
}
400433
else {
401434
String inputChannel = channelToBeanName(consumer.getInputChannel());
402-
return new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
435+
node = new MessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
403436
inputChannel, outputChannelName);
404437
}
438+
if (IntegrationGraphServer.micrometerEnhancer != null) {
439+
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
440+
}
441+
return node;
405442
}
406443

407-
private MessageHandlerNode polledHandlerNode(String name, PollingConsumer consumer) {
444+
MessageHandlerNode polledHandlerNode(String nameArg, PollingConsumer consumer) {
408445
String outputChannelName = channelToBeanName(consumer.getOutputChannel());
409446
String errorChannel = channelToBeanName(consumer.getDefaultErrorChannel());
410447
MessageHandler handler = consumer.getHandler();
448+
MessageHandlerNode node;
449+
String name = nameArg;
450+
if (handler instanceof NamedComponent) {
451+
name = ((NamedComponent) handler).getComponentName();
452+
}
411453
if (handler instanceof CompositeMessageHandler) {
412-
return compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName,
454+
node = compositeHandler(name, consumer, (CompositeMessageHandler) handler, outputChannelName,
413455
errorChannel, true);
414456
}
415457
else if (handler instanceof DiscardingMessageHandler) {
416-
return discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName,
458+
node = discardingHandler(name, consumer, (DiscardingMessageHandler) handler, outputChannelName,
417459
errorChannel, true);
418460
}
419461
else if (handler instanceof MappingMessageRouterManagement) {
420-
return routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler,
462+
node = routingHandler(name, consumer, handler, (MappingMessageRouterManagement) handler,
421463
outputChannelName, errorChannel, true);
422464
}
423465
else if (handler instanceof RecipientListRouterManagement) {
424-
return recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler,
466+
node = recipientListRoutingHandler(name, consumer, handler, (RecipientListRouterManagement) handler,
425467
outputChannelName, errorChannel, true);
426468
}
427469
else {
428470
String inputChannel = channelToBeanName(consumer.getInputChannel());
429-
return new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
471+
node = new ErrorCapableMessageHandlerNode(this.nodeId.incrementAndGet(), name, handler,
430472
inputChannel, outputChannelName, errorChannel);
431473
}
474+
if (IntegrationGraphServer.micrometerEnhancer != null) {
475+
node = IntegrationGraphServer.micrometerEnhancer.enhance(node);
476+
}
477+
return node;
432478
}
433479

434-
private MessageHandlerNode compositeHandler(String name, IntegrationConsumer consumer,
480+
MessageHandlerNode compositeHandler(String name, IntegrationConsumer consumer,
435481
CompositeMessageHandler handler, String output, String errors, boolean polled) {
436482

437483
List<CompositeMessageHandlerNode.InnerHandler> innerHandlers =
@@ -453,7 +499,7 @@ private MessageHandlerNode compositeHandler(String name, IntegrationConsumer con
453499
inputChannel, output, innerHandlers);
454500
}
455501

456-
private MessageHandlerNode discardingHandler(String name, IntegrationConsumer consumer,
502+
MessageHandlerNode discardingHandler(String name, IntegrationConsumer consumer,
457503
DiscardingMessageHandler handler, String output, String errors, boolean polled) {
458504

459505
String discards = channelToBeanName(handler.getDiscardChannel());
@@ -465,7 +511,7 @@ private MessageHandlerNode discardingHandler(String name, IntegrationConsumer co
465511
inputChannel, output, discards);
466512
}
467513

468-
private MessageHandlerNode routingHandler(String name, IntegrationConsumer consumer, MessageHandler handler,
514+
MessageHandlerNode routingHandler(String name, IntegrationConsumer consumer, MessageHandler handler,
469515
MappingMessageRouterManagement router, String output, String errors, boolean polled) {
470516

471517
Collection<String> routes =
@@ -481,7 +527,7 @@ private MessageHandlerNode routingHandler(String name, IntegrationConsumer consu
481527
inputChannel, output, routes);
482528
}
483529

484-
private MessageHandlerNode recipientListRoutingHandler(String name, IntegrationConsumer consumer,
530+
MessageHandlerNode recipientListRoutingHandler(String name, IntegrationConsumer consumer,
485531
MessageHandler handler, RecipientListRouterManagement router, String output, String errors,
486532
boolean polled) {
487533

@@ -499,7 +545,7 @@ private MessageHandlerNode recipientListRoutingHandler(String name, IntegrationC
499545
inputChannel, output, routes);
500546
}
501547

502-
private void reset() {
548+
void reset() {
503549
this.nodeId.set(0);
504550
}
505551

‎spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,10 +109,16 @@ public void addProperties(@Nullable Map<String, Object> props) {
109109

110110
public static class Stats {
111111

112+
private final String deprecated = "stats are deprecated in favor of sendTimers and receiveCounters";
113+
112114
protected boolean isAvailable() {
113115
return false;
114116
}
115117

118+
public String getDeprecated() {
119+
return this.deprecated;
120+
}
121+
116122
}
117123

118124
}

‎spring-integration-core/src/main/java/org/springframework/integration/graph/MessageChannelNode.java‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
package org.springframework.integration.graph;
1818

19+
import java.util.function.Supplier;
20+
1921
import org.springframework.messaging.MessageChannel;
2022

2123
/**
@@ -27,7 +29,9 @@
2729
*
2830
*/
2931
@SuppressWarnings("deprecation")
30-
public class MessageChannelNode extends IntegrationNode {
32+
public class MessageChannelNode extends IntegrationNode implements SendTimersAware {
33+
34+
private Supplier<SendTimers> sendTimers;
3135

3236
public MessageChannelNode(int nodeId, String name, MessageChannel channel) {
3337
super(nodeId, name, channel,
@@ -36,6 +40,14 @@ public MessageChannelNode(int nodeId, String name, MessageChannel channel) {
3640
: new IntegrationNode.Stats());
3741
}
3842

43+
public SendTimers getSendTimers() {
44+
return this.sendTimers.get();
45+
}
46+
47+
@Override
48+
public void sendTimers(Supplier<SendTimers> timers) {
49+
this.sendTimers = timers;
50+
}
3951

4052
public static final class Stats extends IntegrationNode.Stats {
4153

‎spring-integration-core/src/main/java/org/springframework/integration/graph/MessageHandlerNode.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
package org.springframework.integration.graph;
1818

19+
import java.util.function.Supplier;
20+
1921
import org.springframework.messaging.MessageHandler;
2022

2123
/**
@@ -27,10 +29,12 @@
2729
*
2830
*/
2931
@SuppressWarnings("deprecation")
30-
public class MessageHandlerNode extends EndpointNode {
32+
public class MessageHandlerNode extends EndpointNode implements SendTimersAware {
3133

3234
private final String input;
3335

36+
private Supplier<SendTimers> sendTimers;
37+
3438
public MessageHandlerNode(int nodeId, String name, MessageHandler handler, String input, String output) {
3539
super(nodeId, name, handler, output,
3640
handler instanceof org.springframework.integration.support.management.MessageHandlerMetrics
@@ -43,6 +47,15 @@ public String getInput() {
4347
return this.input;
4448
}
4549

50+
public SendTimers getSendTimers() {
51+
return this.sendTimers.get();
52+
}
53+
54+
@Override
55+
public void sendTimers(Supplier<SendTimers> timers) {
56+
this.sendTimers = timers;
57+
}
58+
4659
public static final class Stats extends IntegrationNode.Stats {
4760

4861
private final org.springframework.integration.support.management.MessageHandlerMetrics handler;

‎spring-integration-core/src/main/java/org/springframework/integration/graph/MessageSourceNode.java‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
package org.springframework.integration.graph;
1818

19+
import java.util.function.Supplier;
20+
1921
import org.springframework.integration.core.MessageSource;
2022

2123
/**
@@ -27,7 +29,9 @@
2729
*
2830
*/
2931
@SuppressWarnings("deprecation")
30-
public class MessageSourceNode extends ErrorCapableEndpointNode {
32+
public class MessageSourceNode extends ErrorCapableEndpointNode implements ReceiveCountersAware {
33+
34+
private Supplier<ReceiveCounters> receiveCounters;
3135

3236
public MessageSourceNode(int nodeId, String name, MessageSource<?> messageSource, String output, String errors) {
3337
super(nodeId, name, messageSource, output, errors,
@@ -37,6 +41,14 @@ public MessageSourceNode(int nodeId, String name, MessageSource<?> messageSource
3741
: new IntegrationNode.Stats());
3842
}
3943

44+
public ReceiveCounters getReceiveCounters() {
45+
return this.receiveCounters.get();
46+
}
47+
48+
@Override
49+
public void receiveCounters(Supplier<ReceiveCounters> counters) {
50+
this.receiveCounters = counters;
51+
}
4052

4153
public static final class Stats extends IntegrationNode.Stats {
4254

0 commit comments

Comments
 (0)