diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index a31a35ed2..405c16e44 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -46,6 +46,7 @@ import org.springframework.cloud.stream.binding.SingleBindingTargetBindable; import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor; import org.springframework.cloud.stream.binding.SubscribableChannelBindingTargetFactory; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; +import org.springframework.cloud.stream.micrometer.DestinationPublishingMetricsAutoConfiguration; import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -69,6 +70,7 @@ import org.springframework.messaging.handler.annotation.support.DefaultMessageHa import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory; import org.springframework.scheduling.TaskScheduler; + /** * Configuration class that provides necessary beans for {@link MessageChannel} binding. * @@ -83,7 +85,7 @@ import org.springframework.scheduling.TaskScheduler; */ @Configuration @EnableConfigurationProperties({ BindingServiceProperties.class, SpringIntegrationProperties.class }) -@Import(ContentTypeConfiguration.class) +@Import({ContentTypeConfiguration.class, DestinationPublishingMetricsAutoConfiguration.class}) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) public class BindingServiceConfiguration { diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java new file mode 100644 index 000000000..746d34491 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java @@ -0,0 +1,169 @@ +/* + * Copyright 2017-2018 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.micrometer; + +import java.util.HashMap; +import java.util.Map; +import java.util.Map.Entry; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.beans.factory.config.BeanExpressionContext; +import org.springframework.beans.factory.config.BeanExpressionResolver; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.bind.BindResult; +import org.springframework.boot.context.properties.bind.Bindable; +import org.springframework.boot.context.properties.bind.Binder; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.EnvironmentAware; +import org.springframework.core.env.Environment; +import org.springframework.util.ObjectUtils; +import org.springframework.util.PatternMatchUtils; + +/** + * @author Vinicius Carvalho + * @author Janne Valkealahti + * @author Oleg Zhurakousky + */ +@ConfigurationProperties(prefix = ApplicationMetricsProperties.PREFIX) +public class ApplicationMetricsProperties implements EnvironmentAware, ApplicationContextAware { + + public static final String PREFIX = "spring.cloud.stream.metrics"; + + private static final Bindable> STRING_STRING_MAP = Bindable.mapOf(String.class, String.class); + + /** + * The name of the metric being emitted. Should be an unique value per application. + * Defaults to: ${spring.application.name:${vcap.application.name:${spring.config.name:application}}} + */ + @Value("${spring.application.name:${vcap.application.name:${spring.config.name:application}}}") + private String key; + + /** + * Application properties that should be added to the metrics payload + * For example: `spring.application**` + */ + private String[] properties; + + /** + * Interval expressed as Duration for scheduling metrics snapshots publishing. + * Defaults to PT60S (60 sec) + */ + private String scheduleInterval; + + /** + * List of properties that are going to be appended to each message. This gets + * populate by onApplicationEvent, once the context refreshes to avoid overhead of + * doing per message basis. + */ + private Map exportProperties; + + private Environment environment; + + private ApplicationContext applicationContext; + + @Override + public void setEnvironment(Environment environment) { + this.environment = environment; + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + public String getKey() { + return key; + } + + public void setKey(String key) { + this.key = key; + } + + public String[] getProperties() { + return properties; + } + + public void setProperties(String[] properties) { + this.properties = properties; + } + + public Map getExportProperties() { + if (this.exportProperties == null) { + this.exportProperties = buildExportProperties(); + } + return this.exportProperties; + } + + public String getScheduleInterval() { + return scheduleInterval; + } + + public void setScheduleInterval(String scheduleInterval) { + this.scheduleInterval = scheduleInterval; + } + + private boolean isMatch(String name, String[] includes, String[] excludes) { + if (ObjectUtils.isEmpty(includes) + || PatternMatchUtils.simpleMatch(includes, name)) { + return !PatternMatchUtils.simpleMatch(excludes, name); + } + return false; + } + + private Map buildExportProperties() { + Map props = new HashMap<>(); + if (!ObjectUtils.isEmpty(this.properties)) { + Map target = bindProperties(); + + BeanExpressionResolver beanExpressionResolver = ((ConfigurableApplicationContext) applicationContext) + .getBeanFactory().getBeanExpressionResolver(); + BeanExpressionContext expressionContext = new BeanExpressionContext( + ((ConfigurableApplicationContext) applicationContext).getBeanFactory(), null); + for (Entry entry : target.entrySet()) { + if (isMatch(entry.getKey(), this.properties, null)) { + String stringValue = ObjectUtils.nullSafeToString(entry.getValue()); + Object exportedValue = null; + if (stringValue != null) { + exportedValue = stringValue.startsWith("#{") + ? beanExpressionResolver.evaluate( + environment.resolvePlaceholders(stringValue), expressionContext) + : environment.resolvePlaceholders(stringValue); + } + + props.put(entry.getKey(), exportedValue); + } + } + } + return props; + } + + private Map bindProperties() { + Map target; + BindResult> bindResult = Binder.get(environment).bind("", STRING_STRING_MAP); + if (bindResult.isBound()) { + target = bindResult.get(); + } + else { + target = new HashMap<>(); + } + return target; + } + +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DefaultDestinationPublishingMeterRegistry.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DefaultDestinationPublishingMeterRegistry.java new file mode 100644 index 000000000..4d411553f --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DefaultDestinationPublishingMeterRegistry.java @@ -0,0 +1,292 @@ +/* + * Copyright 2018 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.micrometer; + +import java.text.DecimalFormat; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.function.ToDoubleFunction; +import java.util.function.ToLongFunction; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; + +import io.micrometer.core.instrument.Clock; +import io.micrometer.core.instrument.Counter; +import io.micrometer.core.instrument.DistributionSummary; +import io.micrometer.core.instrument.FunctionCounter; +import io.micrometer.core.instrument.FunctionTimer; +import io.micrometer.core.instrument.Gauge; +import io.micrometer.core.instrument.LongTaskTimer; +import io.micrometer.core.instrument.Measurement; +import io.micrometer.core.instrument.Meter; +import io.micrometer.core.instrument.Meter.Id; +import io.micrometer.core.instrument.Meter.Type; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.Timer; +import io.micrometer.core.instrument.cumulative.CumulativeCounter; +import io.micrometer.core.instrument.cumulative.CumulativeDistributionSummary; +import io.micrometer.core.instrument.cumulative.CumulativeFunctionCounter; +import io.micrometer.core.instrument.cumulative.CumulativeFunctionTimer; +import io.micrometer.core.instrument.cumulative.CumulativeTimer; +import io.micrometer.core.instrument.distribution.DistributionStatisticConfig; +import io.micrometer.core.instrument.distribution.HistogramSnapshot; +import io.micrometer.core.instrument.distribution.ValueAtPercentile; +import io.micrometer.core.instrument.distribution.pause.PauseDetector; +import io.micrometer.core.instrument.internal.DefaultGauge; +import io.micrometer.core.instrument.internal.DefaultLongTaskTimer; +import io.micrometer.core.instrument.internal.DefaultMeter; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.context.SmartLifecycle; +import org.springframework.messaging.support.GenericMessage; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.0 + * + */ +class DefaultDestinationPublishingMeterRegistry extends MeterRegistry implements SmartLifecycle { + + private static final Log logger = LogFactory.getLog(DefaultDestinationPublishingMeterRegistry.class); + + private final MetricsPublisherConfig metricsPublisherConfig; + + private final Consumer metricsConsumer; + + private final DecimalFormat format = new DecimalFormat("#.####"); + + private final ApplicationMetricsProperties applicationProperties; + + private final ObjectMapper objectMapper = new ObjectMapper(); + + private ScheduledFuture publisher; + + DefaultDestinationPublishingMeterRegistry(ApplicationMetricsProperties applicationProperties, + MetersPublisherBinding publisherBinding, MetricsPublisherConfig metricsPublisherConfig, Clock clock) { + super(clock); + this.metricsPublisherConfig = metricsPublisherConfig; + this.metricsConsumer = new MessageChannelPublisher(publisherBinding); + this.applicationProperties = applicationProperties; + } + + @Override + public void start() { + start(Executors.defaultThreadFactory()); + } + + @Override + public void stop() { + if (publisher != null) { + publisher.cancel(false); + publisher = null; + } + } + + @Override + public boolean isRunning() { + return this.publisher != null; + } + + @Override + public int getPhase() { + return 0; + } + + @Override + public boolean isAutoStartup() { + return true; + } + + @Override + public void stop(Runnable callback) { + this.stop(); + callback.run(); + } + + @Override + protected Gauge newGauge(Meter.Id id, T obj, ToDoubleFunction f) { + return new DefaultGauge<>(id, obj, f); + } + + @Override + protected Counter newCounter(Meter.Id id) { + return new CumulativeCounter(id); + } + + @Override + protected LongTaskTimer newLongTaskTimer(Meter.Id id) { + return new DefaultLongTaskTimer(id, clock); + } + + @Override + protected TimeUnit getBaseTimeUnit() { + return TimeUnit.MILLISECONDS; + } + + protected void publish() { + List> aggregatedMeters = new ArrayList<>(); + for (Meter meter : this.getMeters()) { + if (meter instanceof Timer) { + aggregatedMeters.add(timerMap((Timer) meter)); + } else if (meter instanceof DistributionSummary) { + aggregatedMeters.add(summaryMap((DistributionSummary) meter)); + } + } + Map messageMap = new LinkedHashMap<>(); + messageMap.put("name", this.applicationProperties.getKey()); + messageMap.put("properties", this.applicationProperties.getExportProperties()); + messageMap.put("meter-snapshots", aggregatedMeters); + try { + String jsonString = this.objectMapper.writeValueAsString(messageMap); + this.metricsConsumer.accept(jsonString); + } catch (JsonProcessingException e) { + logger.warn("Error producing JSON String representation metric data", e); + } + } + + @Override + protected Timer newTimer(Id id, DistributionStatisticConfig distributionStatisticConfig, + PauseDetector pauseDetector) { + return new CumulativeTimer(id, clock, distributionStatisticConfig, pauseDetector, getBaseTimeUnit()); + } + + @Override + protected FunctionTimer newFunctionTimer(Id id, T obj, ToLongFunction countFunction, + ToDoubleFunction totalTimeFunction, TimeUnit totalTimeFunctionUnits) { + return new CumulativeFunctionTimer(id, obj, countFunction, totalTimeFunction, totalTimeFunctionUnits, + getBaseTimeUnit()); + } + + @Override + protected FunctionCounter newFunctionCounter(Id id, T obj, ToDoubleFunction valueFunction) { + return new CumulativeFunctionCounter(id, obj, valueFunction); + } + + @Override + protected Meter newMeter(Id id, Type type, Iterable measurements) { + return new DefaultMeter(id, type, measurements); + } + + @Override + protected DistributionSummary newDistributionSummary(Id id, DistributionStatisticConfig distributionStatisticConfig, + double scale) { + return new CumulativeDistributionSummary(id, clock, distributionStatisticConfig, scale); + } + + @Override + protected DistributionStatisticConfig defaultHistogramConfig() { + return DistributionStatisticConfig.builder().expiry(metricsPublisherConfig.step()).build() + .merge(DistributionStatisticConfig.DEFAULT); + } + + private void start(ThreadFactory threadFactory) { + if (publisher != null) { + stop(); + } + publisher = Executors.newSingleThreadScheduledExecutor(threadFactory).scheduleAtFixedRate(this::publish, + metricsPublisherConfig.step().toMillis(), metricsPublisherConfig.step().toMillis(), + TimeUnit.MILLISECONDS); + } + + private Map summaryMap(DistributionSummary summary) { + List fields = this.gatherSnapshotFields(summary.takeSnapshot(false)); + return this.toMeterMap(fields, summary.getId()); + } + + private Map timerMap(Timer timer) { + List fields = this.gatherSnapshotFields(timer.takeSnapshot(false)); + return this.toMeterMap(fields, timer.getId()); + } + + private List gatherSnapshotFields(HistogramSnapshot snapshot) { + List fields = new ArrayList<>(); + fields.add(new Field("sum", snapshot.total(getBaseTimeUnit()))); + fields.add(new Field("count", snapshot.count())); + fields.add(new Field("mean", snapshot.mean(getBaseTimeUnit()))); + fields.add(new Field("upper", snapshot.max(getBaseTimeUnit()))); + fields.add(new Field("total", snapshot.total(getBaseTimeUnit()))); + + for (ValueAtPercentile v : snapshot.percentileValues()) { + fields.add(new Field(format.format(v.percentile()) + "_percentile", v.value(getBaseTimeUnit()))); + } + return fields; + } + + private Map toMeterMap(List fields, Meter.Id id) { + Map meterMap = new LinkedHashMap<>(); + meterMap.put("id", id); + meterMap.put("metrics", fields); + return meterMap; + } + + /** + * + */ + private class Field { + final String name; + final double value; + + private Field(String name, double value) { + this.name = name; + this.value = value; + } + + @SuppressWarnings("unused") // used by ObjectMapper + public String getName() { + return name; + } + + @SuppressWarnings("unused") // used by ObjectMapper + public double getValue() { + return value; + } + + @Override + public String toString() { + return name + "=" + format.format(value); + } + } + + /** + * + */ + private static final class MessageChannelPublisher implements Consumer { + private final MetersPublisherBinding metersPublisherBinding; + + MessageChannelPublisher(MetersPublisherBinding metersPublisherBinding) { + this.metersPublisherBinding = metersPublisherBinding; + } + + @Override + public void accept(String metricData) { + logger.trace(metricData); + this.metersPublisherBinding.applicationMetrics().send(new GenericMessage(metricData)); + } + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DestinationPublishingMetricsAutoConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DestinationPublishingMetricsAutoConfiguration.java new file mode 100644 index 000000000..70b202a5f --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DestinationPublishingMetricsAutoConfiguration.java @@ -0,0 +1,78 @@ +/* + * Copyright 2018 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.micrometer; + +import io.micrometer.core.instrument.Clock; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.config.BeanFactoryPostProcessor; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.beans.factory.support.RootBeanDefinition; +import org.springframework.boot.actuate.autoconfigure.metrics.MetricsAutoConfiguration; +import org.springframework.boot.actuate.autoconfigure.metrics.export.simple.SimpleMetricsExportAutoConfiguration; +import org.springframework.boot.autoconfigure.AutoConfigureAfter; +import org.springframework.boot.autoconfigure.AutoConfigureBefore; +import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.stream.binder.Binder; +import org.springframework.cloud.stream.binding.BindableProxyFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.0 + * + */ +@Configuration +@AutoConfigureBefore(SimpleMetricsExportAutoConfiguration.class) +@AutoConfigureAfter(MetricsAutoConfiguration.class) +@ConditionalOnClass(Binder.class) +@ConditionalOnProperty("spring.cloud.stream.bindings." + MetersPublisherBinding.APPLICATION_METRICS + ".destination") +@EnableConfigurationProperties(ApplicationMetricsProperties.class) +public class DestinationPublishingMetricsAutoConfiguration { + + @Bean + @ConditionalOnMissingBean + public MetricsPublisherConfig metricsPublisherConfig(ApplicationMetricsProperties metersPublisherProperties) { + return new MetricsPublisherConfig(metersPublisherProperties); + } + + @Bean + @ConditionalOnMissingBean + public DefaultDestinationPublishingMeterRegistry defaultDestinationPublishingMeterRegistry(ApplicationMetricsProperties applicationMetricsProperties, + MetersPublisherBinding publisherBinding, MetricsPublisherConfig metricsPublisherConfig, Clock clock) { + return new DefaultDestinationPublishingMeterRegistry(applicationMetricsProperties, publisherBinding, metricsPublisherConfig, clock); + } + + @Bean + public BeanFactoryPostProcessor metersPublisherBindingRegistrant() { + return new BeanFactoryPostProcessor() { + @Override + public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { + RootBeanDefinition emitterBindingDefinition = new RootBeanDefinition(BindableProxyFactory.class); + emitterBindingDefinition.getConstructorArgumentValues().addGenericArgumentValue(MetersPublisherBinding.class); + ((DefaultListableBeanFactory)beanFactory).registerBeanDefinition(MetersPublisherBinding.class.getName(), emitterBindingDefinition); + } + }; + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetersPublisherBinding.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetersPublisherBinding.java new file mode 100644 index 000000000..47decfbd1 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetersPublisherBinding.java @@ -0,0 +1,34 @@ +/* + * Copyright 2018 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.micrometer; + +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.messaging.MessageChannel; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.0 + */ +public interface MetersPublisherBinding { + + String APPLICATION_METRICS = "applicationMetrics"; + + @Output(APPLICATION_METRICS) + MessageChannel applicationMetrics(); +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java new file mode 100644 index 000000000..871c8279b --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java @@ -0,0 +1,49 @@ +/* + * Copyright 2018 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.micrometer; + +import io.micrometer.core.instrument.step.StepRegistryConfig; + +/** + * + * @author Oleg Zhurakousky + * + * @since 2.0 + * + */ +class MetricsPublisherConfig implements StepRegistryConfig { + + private final ApplicationMetricsProperties applicationMetricsProperties; + + MetricsPublisherConfig(ApplicationMetricsProperties applicationMetricsProperties) { + this.applicationMetricsProperties = applicationMetricsProperties; + } + + @Override + public String prefix() { + return ApplicationMetricsProperties.PREFIX; + } + + @Override + public String get(String key) { + String value = null; + if (key.equals(this.prefix() + ".step")) { + value = this.applicationMetricsProperties.getScheduleInterval(); + } + return value; + } +}