SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Expose channel metrics for aggregate applications

Browse Source 
Fixes #660 #661

* Fix an issue where aggregated applications were not enabling web endpoints
* Add spring-integration-jmx as mandatory dependency, so that SCSt applications
  expose metrics by default;
* Remove `AggregatorParentConfiguration` as redundant
* Register a `PublicMetrics` instance for each aggregate, using a customized version
  of SpringIntegrationMetricReader
This commit is contained in:
Marius Bogoevici
2016-09-21 02:39:42 -04:00
committed by Ilayaperumal Gopinathan
parent d11b3cf08d
commit 810803859a
11 changed files with 202 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-cloud-stream/pom.xml
View File

@@ -34,6 +34,10 @@
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jmx</artifactId>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-tuple</artifactId>

37
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplication.java
View File

@@ -27,6 +27,7 @@ import org.springframework.messaging.SubscribableChannel;
/**
* Class that is responsible for embedding apps using shared channel registry.
*
* @author Marius Bogoevici
* @author Ilayaperumal Gopinathan
* @author Venil Noronha
@@ -35,45 +36,35 @@ abstract class AggregateApplication {
private static final String SPRING_CLOUD_STREAM_INTERNAL_PREFIX = "spring.cloud.stream.internal";
private static final String CHANNEL_NAMESPACE_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
private static final String CHANNEL_NAMESPACE_PROPERTY_NAME =
SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".channelNamespace";
private static final String SELF_CONTAINED_APP_PROPERTY_NAME = SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".selfContained";
private static final String SELF_CONTAINED_APP_PROPERTY_NAME =
SPRING_CLOUD_STREAM_INTERNAL_PREFIX + ".selfContained";
public static final String INPUT_CHANNEL_NAME = "input";
public static final String OUTPUT_CHANNEL_NAME = "output";
static ConfigurableApplicationContext createParentContext(Object[] sources, String[] args, boolean selfContained) {
static ConfigurableApplicationContext createParentContext(Object[] sources, String[] args, final
boolean selfContained, boolean webEnvironment,
boolean headless) {
SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder();
aggregatorParentConfiguration
.sources(AggregatorParentConfiguration.class)
.sources(sources)
.web(false)
.headless(true)
.web(webEnvironment)
.headless(headless)
.properties("spring.jmx.default-domain="
+ AggregatorParentConfiguration.class.getName(),
+ AggregateApplicationBuilder.ParentConfiguration.class.getName(),
SELF_CONTAINED_APP_PROPERTY_NAME + "=" + selfContained);
return aggregatorParentConfiguration.run(args);
}
static ConfigurableApplicationContext createParentContext(ConfigurableApplicationContext parentContext, String[] args,
boolean selfContained) {
SpringApplicationBuilder aggregatorParentConfiguration = new SpringApplicationBuilder();
aggregatorParentConfiguration
.sources(AggregatorParentConfiguration.class)
.web(false)
.headless(true)
.properties("spring.jmx.default-domain="
+ AggregatorParentConfiguration.class.getName(),
SELF_CONTAINED_APP_PROPERTY_NAME + "=" + selfContained)
.parent(parentContext);
return aggregatorParentConfiguration.run(args);
}
static String getNamespace(String appClassName, int index) {
static String getDefaultNamespace(String appClassName, int index) {
return appClassName + "_" + index;
}
protected static SpringApplicationBuilder embedApp(
ConfigurableApplicationContext parentContext, String namespace,
Class<?> app) {
@@ -81,7 +72,7 @@ abstract class AggregateApplication {
.web(false)
.main(app)
.bannerMode(Mode.OFF)
.properties("spring.jmx.default-domain=" + app)
.properties("spring.jmx.default-domain=" + namespace)
.properties(CHANNEL_NAMESPACE_PROPERTY_NAME + "=" + namespace)
.registerShutdownHook(false)
.parent(parentContext);

70
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregateApplicationBuilder.java
View File

@@ -21,9 +21,15 @@ import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import org.springframework.boot.actuate.endpoint.MetricReaderPublicMetrics;
import org.springframework.boot.actuate.endpoint.MetricsEndpoint;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.util.StringUtils;
/**
@@ -50,12 +56,16 @@ public class AggregateApplicationBuilder {
private List<String> parentArgs = new ArrayList<>();
private boolean headless = true;
private boolean webEnvironment = true;
public AggregateApplicationBuilder(String... args) {
this(new Object[]{ParentConfiguration.class}, args);
this(new Object[]{ ParentConfiguration.class }, args);
}
public AggregateApplicationBuilder(Object source, String... args) {
this(new Object[]{source}, args);
this(new Object[]{ source }, args);
}
public AggregateApplicationBuilder(Object[] sources, String[] args) {
@@ -75,7 +85,7 @@ public class AggregateApplicationBuilder {
}
public AggregateApplicationBuilder parent(Object source, String... args) {
return parent(new Object[]{source}, args);
return parent(new Object[]{ source }, args);
}
public AggregateApplicationBuilder parent(Object[] sources, String[] args) {
@@ -84,6 +94,30 @@ public class AggregateApplicationBuilder {
return this;
}
/**
* Flag to explicitly request a web or non-web environment.
*
* @param webEnvironment true if the application has a web environment
* @return the AggregateApplicationBuilder being constructed
* @see SpringApplicationBuilder#web(boolean)
*/
public AggregateApplicationBuilder web(boolean webEnvironment) {
this.webEnvironment = webEnvironment;
return this;
}
/**
* Configures the headless attribute of the build application.
*
* @param headless true if the application is headless
* @return the AggregateApplicationBuilder being constructed
* @see SpringApplicationBuilder#headless(boolean)
*/
public AggregateApplicationBuilder headless(boolean headless) {
this.headless = headless;
return this;
}
public SourceConfigurer from(Class<?> app) {
SourceConfigurer sourceConfigurer = new SourceConfigurer(app);
this.sourceConfigurer = sourceConfigurer;
@@ -110,12 +144,13 @@ public class AggregateApplicationBuilder {
Class<?> appToEmbed = appConfigurer.getApp();
// Always update namespace before preparing SharedChannelRegistry
if (appConfigurer.namespace == null) {
appConfigurer.namespace = AggregateApplication.getNamespace(appConfigurer.getApp().getName(), i);
appConfigurer.namespace = AggregateApplication.getDefaultNamespace(appConfigurer.getApp().getName(),
i);
}
appsToEmbed.put(appToEmbed, appConfigurer.namespace);
}
this.parentContext = AggregateApplication.createParentContext(this.parentSources.toArray(new Object[0]),
this.parentArgs.toArray(new String[0]), areAppsSelfContained());
this.parentArgs.toArray(new String[0]), selfContained(), this.webEnvironment, this.headless);
SharedChannelRegistry sharedChannelRegistry = this.parentContext.getBean(SharedChannelRegistry.class);
AggregateApplication.prepareSharedChannelRegistry(sharedChannelRegistry, appsToEmbed);
for (int i = apps.size() - 1; i >= 0; i--) {
@@ -125,7 +160,7 @@ public class AggregateApplicationBuilder {
return this.parentContext;
}
private boolean areAppsSelfContained() {
private boolean selfContained() {
return (this.sourceConfigurer != null) && (this.sinkConfigurer != null);
}
@@ -225,9 +260,15 @@ public class AggregateApplicationBuilder {
}
void embed() {
childContext(this.app, AggregateApplicationBuilder.this.parentContext,
this.namespace).args(this.args).config(this.names)
.profiles(this.profiles).run();
final ConfigurableApplicationContext childContext =
childContext(this.app, AggregateApplicationBuilder.this.parentContext,
this.namespace).args(this.args).config(this.names)
.profiles(this.profiles).run();
AggregateApplicationBuilder.this.parentContext.getBeanFactory().getBean(
MetricsEndpoint.class).registerPublicMetrics(
new MetricReaderPublicMetrics(
new NamespaceAwareSpringIntegrationMetricReader(this.namespace, childContext.getBean(
IntegrationMBeanExporter.class))));
}
}
@@ -262,7 +303,7 @@ public class AggregateApplicationBuilder {
return this;
}
public void run() {
public ConfigurableApplicationContext run() {
List<String> args = new ArrayList<String>();
if (this.args != null) {
args.addAll(Arrays.asList(this.args));
@@ -270,13 +311,18 @@ public class AggregateApplicationBuilder {
if (this.configName != null) {
args.add("--spring.config.name=" + this.configName);
}
this.builder.run(args.toArray(new String[0]));
return this.builder.run(args.toArray(new String[0]));
}
}
@EnableAutoConfiguration
@EnableBinding
public static class ParentConfiguration {
@Bean
@ConditionalOnMissingBean(SharedChannelRegistry.class)
public SharedChannelRegistry sharedChannelRegistry() {
return new SharedChannelRegistry();
}
}
}

38
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/AggregatorParentConfiguration.java
View File

@@ -1,38 +0,0 @@
/*
* Copyright 2015-2016 the original author or 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.
*/
package org.springframework.cloud.stream.aggregate;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.context.annotation.Bean;
/**
* Basic configuration for an aggregator application parent.
*
* @author Marius Bogoevici
*/
@EnableAutoConfiguration
@EnableBinding
public class AggregatorParentConfiguration {
@Bean
@ConditionalOnMissingBean(SharedChannelRegistry.class)
public SharedChannelRegistry sharedChannelRegistry() {
return new SharedChannelRegistry();
}
}

104
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/NamespaceAwareSpringIntegrationMetricReader.java Normal file
View File

@@ -0,0 +1,104 @@
/*
* Copyright 2016 the original author or 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.
*/
package org.springframework.cloud.stream.aggregate;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import org.springframework.boot.actuate.metrics.Metric;
import org.springframework.boot.actuate.metrics.reader.MetricReader;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.integration.support.management.Statistics;
import org.springframework.util.Assert;
/**
* A customized version of {@link org.springframework.boot.actuate.metrics.integration.SpringIntegrationMetricReader} that
* provides support for customizing channels with a namespace prefix.
*
* @author Marius Bogoevici
* @see org.springframework.boot.actuate.metrics.integration.SpringIntegrationMetricReader for original implementation
*/
public class NamespaceAwareSpringIntegrationMetricReader implements MetricReader {
private final String namespace;
private final IntegrationMBeanExporter exporter;
public NamespaceAwareSpringIntegrationMetricReader(String namespace, IntegrationMBeanExporter exporter) {
Assert.hasText(namespace, "cannot be null or empty String");
Assert.notNull(exporter, "cannot be null");
this.namespace = namespace;
this.exporter = exporter;
}
@Override
public Metric<?> findOne(String metricName) {
return null;
}
@Override
public Iterable<Metric<?>> findAll() {
IntegrationMBeanExporter exporter = this.exporter;
List<Metric<?>> metrics = new ArrayList<Metric<?>>();
for (String name : exporter.getChannelNames()) {
String prefix = "integration.channel." + namespace + "." + name;
metrics.addAll(getStatistics(prefix + ".errorRate",
exporter.getChannelErrorRate(name)));
metrics.add(new Metric<Long>(prefix + ".sendCount",
exporter.getChannelSendCountLong(name)));
metrics.addAll(getStatistics(prefix + ".sendRate",
exporter.getChannelSendRate(name)));
metrics.add(new Metric<Long>(prefix + ".receiveCount",
exporter.getChannelReceiveCountLong(name)));
}
for (String name : exporter.getHandlerNames()) {
metrics.addAll(getStatistics("integration." + namespace + ".handler." + name + ".duration",
exporter.getHandlerDuration(name)));
}
metrics.add(new Metric<Integer>("integration." + namespace + ".activeHandlerCount",
exporter.getActiveHandlerCount()));
metrics.add(new Metric<Integer>("integration." + namespace + ".handlerCount",
exporter.getHandlerCount()));
metrics.add(new Metric<Integer>("integration." + namespace + ".channelCount",
exporter.getChannelCount()));
metrics.add(new Metric<Integer>("integration." + namespace + ".queuedMessageCount",
exporter.getQueuedMessageCount()));
return metrics;
}
private Collection<? extends Metric<?>> getStatistics(String name,
Statistics statistic) {
List<Metric<?>> metrics = new ArrayList<Metric<?>>();
metrics.add(new Metric<Double>(name + ".mean", statistic.getMean()));
metrics.add(new Metric<Double>(name + ".max", statistic.getMax()));
metrics.add(new Metric<Double>(name + ".min", statistic.getMin()));
metrics.add(
new Metric<Double>(name + ".stdev", statistic.getStandardDeviation()));
metrics.add(new Metric<Long>(name + ".count", statistic.getCountLong()));
return metrics;
}
@Override
public long count() {
int totalChannelCount = this.exporter.getChannelCount() * 11;
int totalHandlerCount = this.exporter.getHandlerCount() * 5;
return totalChannelCount + totalHandlerCount + 4;
}
}

4
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/aggregate/SharedChannelRegistry.java
View File

@@ -38,10 +38,10 @@ public class SharedChannelRegistry {
}
public void register(String id, MessageChannel messageChannel) {
sharedChannels.put(id, messageChannel);
this.sharedChannels.put(id, messageChannel);
}
public Map<String, MessageChannel> getAll() {
return Collections.unmodifiableMap(sharedChannels);
return Collections.unmodifiableMap(this.sharedChannels);
}
}

32
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/aggregation/ModuleAggregationTest.java
View File

@@ -16,13 +16,12 @@
package org.springframework.cloud.stream.aggregation;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.cloud.stream.aggregate.AggregateApplicationBuilder;
import org.springframework.cloud.stream.aggregate.SharedChannelRegistry;
@@ -34,7 +33,6 @@ import org.springframework.cloud.stream.utils.MockBinderRegistryConfiguration;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.ReflectionUtils;
import static org.assertj.core.api.Assertions.assertThat;
@@ -65,24 +63,22 @@ public class ModuleAggregationTest {
argsToVerify.add("--foo1=bar1");
argsToVerify.add("--foo2=bar2");
argsToVerify.add("--foo3=bar3");
argsToVerify.add("--server.port=0");
AggregateApplicationBuilder aggregateApplicationBuilder =
new AggregateApplicationBuilder(MockBinderRegistryConfiguration.class,
"--foo1=bar1");
aggregateApplicationBuilder.parent(DummyConfig.class, "--foo2=bar2")
.from(TestSource.class)
.namespace("foo").to(TestProcessor.class).namespace("bar")
.run("--foo3=bar3");
Field parentArgsField = ReflectionUtils.findField(AggregateApplicationBuilder.class,"parentArgs", List.class);
ReflectionUtils.makeAccessible(parentArgsField);
Field parentSourcesField = ReflectionUtils.findField(AggregateApplicationBuilder.class,"parentSources", List.class);
ReflectionUtils.makeAccessible(parentSourcesField);
String args = ReflectionUtils.getField(parentArgsField, aggregateApplicationBuilder).toString();
Assert.assertEquals(args, argsToVerify.toString());
List<Object> sources = ((List<Object>)ReflectionUtils.getField(parentSourcesField, aggregateApplicationBuilder));
Assert.assertTrue(sources.size() == 3);
Assert.assertTrue(sources.contains(AggregateApplicationBuilder.ParentConfiguration.class));
Assert.assertTrue(sources.contains(MockBinderRegistryConfiguration.class));
Assert.assertTrue(sources.contains(DummyConfig.class));
final ConfigurableApplicationContext context =
aggregateApplicationBuilder.parent(DummyConfig.class, "--foo2=bar2")
.from(TestSource.class)
.namespace("foo").to(TestProcessor.class).namespace("bar")
.run("--foo3=bar3", "--server.port=0");
DirectFieldAccessor aggregateApplicationBuilderAccessor = new DirectFieldAccessor(aggregateApplicationBuilder);
assertThat((List<String>) aggregateApplicationBuilderAccessor.getPropertyValue(
"parentArgs")).containsExactlyInAnyOrder(argsToVerify.toArray(new String[argsToVerify.size()]));
List<Object> sources = (List<Object>) aggregateApplicationBuilderAccessor.getPropertyValue("parentSources");
assertThat(sources).containsExactlyInAnyOrder(AggregateApplicationBuilder.ParentConfiguration.class,
MockBinderRegistryConfiguration.class, DummyConfig.class);
context.close();
}
@Test

5
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/BinderFactoryConfigurationTests.java
View File

@@ -25,7 +25,6 @@ import org.junit.Test;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.aggregate.AggregatorParentConfiguration;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.stub1.StubBinder1;
import org.springframework.cloud.stream.binder.stub1.StubBinder1Configuration;
@@ -64,7 +63,7 @@ public class BinderFactoryConfigurationTests {
public void loadBinderTypeRegistryWithNonSelfContainedAggregatorApp() throws Exception {
try {
createBinderTestContextWithSources(
new Class[]{SimpleApplication.class, AggregatorParentConfiguration.class}, new String[]{},
new Class[]{SimpleApplication.class}, new String[]{},
"spring.cloud.stream.internal.selfContained=false");
fail();
}
@@ -78,7 +77,7 @@ public class BinderFactoryConfigurationTests {
@Test
public void loadBinderTypeRegistryWithSelfContainedAggregatorApp() throws Exception {
createBinderTestContextWithSources(
new Class[] { SimpleApplication.class, AggregatorParentConfiguration.class}, new String[] {},
new Class[] { SimpleApplication.class}, new String[] {},
"spring.cloud.stream.internal.selfContained=true");
}

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/HealthIndicatorsConfigurationTests.java
View File

@@ -71,6 +71,7 @@ public class HealthIndicatorsConfigurationTests {
assertThat(healthIndicators.get("binder1").health().getStatus()).isEqualTo(Status.UP);
assertThat(healthIndicators).containsKey("binder2");
assertThat(healthIndicators.get("binder2").health().getStatus()).isEqualTo(Status.UNKNOWN);
context.close();
}
@Test
@@ -92,6 +93,7 @@ public class HealthIndicatorsConfigurationTests {
}
assertThat(context.getBean("testHealthIndicator1", CompositeHealthIndicator.class)).isNotNull();
assertThat(context.getBean("testHealthIndicator2", CompositeHealthIndicator.class)).isNotNull();
context.close();
}
public static ConfigurableApplicationContext createBinderTestContext(

1
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/InputOutputBindingOrderTest.java
View File

@@ -56,6 +56,7 @@ public class InputOutputBindingOrderTest {
assertThat(someLifecycle.isRunning());
applicationContext.close();
assertThat(someLifecycle.isRunning()).isFalse();
applicationContext.close();
}
@EnableBinding(Processor.class)

2
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/LifecycleBinderTests.java
View File

@@ -37,7 +37,7 @@ import static org.assertj.core.api.Assertions.assertThat;
public class LifecycleBinderTests {
@Test
public void testNonSmartLifecyclesStarted() {
public void testOnlySmartLifecyclesStarted() {
ConfigurableApplicationContext applicationContext = SpringApplication.run(TestSource.class, "--server.port=-1");
SimpleLifecycle simpleLifecycle = applicationContext.getBean(SimpleLifecycle.class);
assertThat(simpleLifecycle.isRunning()).isFalse();