GH-1324 Restored support for the old format of metric Message

- brought back the Metric class from the old boot
- modified Metric class slightly to adjust for the type of info available thru micrometer
- added 'meterFilter' property for metric filtering

Resolves #1324

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetrics.java

@@ -0,0 +1,79 @@
+/*
+ * 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.Collection;
+import java.util.Date;
+import java.util.Map;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+
+
+/**
+ * @author Vinicius Carvalho
+ * @author Oleg Zhurakousky
+ */
+@JsonPropertyOrder({ "name", "createdTime", "properties", "metrics" })
+class ApplicationMetrics {
+
+	@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timezone = "UTC")
+	private final Date createdTime;
+
+	private String name;
+
+	private Collection<Metric<Number>> metrics;
+
+	private Map<String, Object> properties;
+
+	@JsonCreator
+	ApplicationMetrics(@JsonProperty("name") String name, @JsonProperty("metrics") Collection<Metric<Number>> metrics) {
+		this.name = name;
+		this.metrics = metrics;
+		this.createdTime = new Date();
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public Collection<Metric<Number>> getMetrics() {
+		return metrics;
+	}
+
+	public void setMetrics(Collection<Metric<Number>> metrics) {
+		this.metrics = metrics;
+	}
+
+	public Date getCreatedTime() {
+		return createdTime;
+	}
+
+	public Map<String, Object> getProperties() {
+		return properties;
+	}
+
+	public void setProperties(Map<String, Object> properties) {
+		this.properties = properties;
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/ApplicationMetricsProperties.java

@@ -16,6 +16,7 @@
 
 package org.springframework.cloud.stream.micrometer;
 
+import java.time.Duration;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -46,8 +47,17 @@ public class ApplicationMetricsProperties implements EnvironmentAware, Applicati
 
 	public static final String PREFIX = "spring.cloud.stream.metrics";
 
+	public static final String EXPORT_FILTER = PREFIX + ".filter";
+
 	private static final Bindable<Map<String, String>> STRING_STRING_MAP = Bindable.mapOf(String.class, String.class);
 
+
+	/**
+	 * Pattern to control the 'meters' one wants to capture. By default all 'meters' will be captured.
+	 * For example, 'spring.integration.*' will only capture metric information for meters whose name starts with 'spring.integration'.
+	 */
+	private String meterFilter;
+
 	/**
 	 * 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}}}
@@ -65,7 +75,7 @@ public class ApplicationMetricsProperties implements EnvironmentAware, Applicati
 	 * Interval expressed as Duration for scheduling metrics snapshots publishing.
 	 * Defaults to PT60S (60 sec)
 	 */
-	private String scheduleInterval;
+	private Duration scheduleInterval;
 
 	/**
 	 * List of properties that are going to be appended to each message. This gets
@@ -111,14 +121,22 @@ public class ApplicationMetricsProperties implements EnvironmentAware, Applicati
 		return this.exportProperties;
 	}
 
-	public String getScheduleInterval() {
+	public Duration getScheduleInterval() {
 		return scheduleInterval;
 	}
 
-	public void setScheduleInterval(String scheduleInterval) {
+	public void setScheduleInterval(Duration scheduleInterval) {
 		this.scheduleInterval = scheduleInterval;
 	}
 
+	public String getMeterFilter() {
+		return this.meterFilter;
+	}
+
+	public void setMeterFilter(String meterFilter) {
+		this.meterFilter = meterFilter;
+	}
+
 	private boolean isMatch(String name, String[] includes, String[] excludes) {
 		if (ObjectUtils.isEmpty(includes)
 				|| PatternMatchUtils.simpleMatch(includes, name)) {

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DefaultDestinationPublishingMeterRegistry.java

@@ -16,11 +16,8 @@
 
 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;
@@ -51,8 +48,6 @@ 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;
@@ -69,7 +64,6 @@ import org.springframework.messaging.support.GenericMessage;
  * @author Oleg Zhurakousky
  *
  * @since 2.0
- *
  */
 class DefaultDestinationPublishingMeterRegistry extends MeterRegistry implements SmartLifecycle {
 
@@ -79,8 +73,6 @@ class DefaultDestinationPublishingMeterRegistry extends MeterRegistry implements
 
 	private final Consumer<String> metricsConsumer;
 
-	private final DecimalFormat format = new DecimalFormat("#.####");
-
 	private final ApplicationMetricsProperties applicationProperties;
 
 	private final ObjectMapper objectMapper = new ObjectMapper();
@@ -150,23 +142,25 @@ class DefaultDestinationPublishingMeterRegistry extends MeterRegistry implements
 	}
 
 	protected void publish() {
-		List<Map<String, Object>> aggregatedMeters = new ArrayList<>();
+		List<Metric<Number>> 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));
+				aggregatedMeters.add(toTimerMetric((Timer) meter));
+			}
+			else if (meter instanceof DistributionSummary) {
+				aggregatedMeters.add(toSummaryMetric((DistributionSummary) meter));
 			}
 		}
-		Map<String, Object> 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);
+		if (!aggregatedMeters.isEmpty()) {
+			ApplicationMetrics metrics = new ApplicationMetrics(this.applicationProperties.getKey(), aggregatedMeters);
+			metrics.setProperties(this.applicationProperties.getExportProperties());
+			try {
+				String jsonString = this.objectMapper.writeValueAsString(metrics);
+				this.metricsConsumer.accept(jsonString);
+			}
+			catch (JsonProcessingException e) {
+				logger.warn("Error producing JSON String representation metric data", e);
+			}
 		}
 	}
 
@@ -194,8 +188,7 @@ class DefaultDestinationPublishingMeterRegistry extends MeterRegistry implements
 	}
 
 	@Override
-	protected DistributionSummary newDistributionSummary(Id id, DistributionStatisticConfig distributionStatisticConfig,
-			double scale) {
+	protected DistributionSummary newDistributionSummary(Id id, DistributionStatisticConfig distributionStatisticConfig, double scale) {
 		return new CumulativeDistributionSummary(id, clock, distributionStatisticConfig, scale);
 	}
 
@@ -214,63 +207,12 @@ class DefaultDestinationPublishingMeterRegistry extends MeterRegistry implements
 				TimeUnit.MILLISECONDS);
 	}
 
-	private Map<String, Object> summaryMap(DistributionSummary summary) {
-		List<Field> fields = this.gatherSnapshotFields(summary.takeSnapshot(false));
-		return this.toMeterMap(fields, summary.getId());
+	private Metric<Number> toSummaryMetric(DistributionSummary summary) {
+		return new Metric<Number>(summary.getId(), summary.takeSnapshot(false));
 	}
 
-	private Map<String, Object> timerMap(Timer timer) {
-		List<Field> fields = this.gatherSnapshotFields(timer.takeSnapshot(false));
-		return this.toMeterMap(fields, timer.getId());
-	}
-
-	private List<Field> gatherSnapshotFields(HistogramSnapshot snapshot) {
-		List<Field> 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<String, Object> toMeterMap(List<Field> fields, Meter.Id id) {
-		Map<String, Object> 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 Metric<Number> toTimerMetric(Timer timer) {
+		return new Metric<Number>(timer.getId(), timer.takeSnapshot(false));
 	}
 
 	/**

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/DestinationPublishingMetricsAutoConfiguration.java

@@ -17,6 +17,7 @@
 package org.springframework.cloud.stream.micrometer;
 
 import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.config.MeterFilter;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
@@ -35,13 +36,14 @@ 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;
+import org.springframework.util.PatternMatchUtils;
+import org.springframework.util.StringUtils;
 
 /**
  *
  * @author Oleg Zhurakousky
  *
  * @since 2.0
- *
  */
 @Configuration
 @AutoConfigureBefore(SimpleMetricsExportAutoConfiguration.class)
@@ -63,7 +65,12 @@ public class DestinationPublishingMetricsAutoConfiguration {
 			ApplicationMetricsProperties applicationMetricsProperties,
 			MetersPublisherBinding publisherBinding,
 			MetricsPublisherConfig metricsPublisherConfig, Clock clock) {
-		return new DefaultDestinationPublishingMeterRegistry(applicationMetricsProperties, publisherBinding, metricsPublisherConfig, clock);
+		DefaultDestinationPublishingMeterRegistry registry = new DefaultDestinationPublishingMeterRegistry(applicationMetricsProperties, publisherBinding, metricsPublisherConfig, clock);
+
+		if (StringUtils.hasText(applicationMetricsProperties.getMeterFilter())) {
+			registry.config().meterFilter(MeterFilter.denyUnless(id -> PatternMatchUtils.simpleMatch(applicationMetricsProperties.getMeterFilter(), id.getName())));
+		}
+		return registry;
 	}
 
 	@Bean

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/Metric.java

@@ -0,0 +1,107 @@
+/*
+ * 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.util.Date;
+import java.util.concurrent.TimeUnit;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import com.fasterxml.jackson.annotation.JsonPropertyOrder;
+
+import io.micrometer.core.instrument.Meter;
+import io.micrometer.core.instrument.distribution.HistogramSnapshot;
+
+/**
+ * Immutable class that wraps the micrometer's {@link HistogramSnapshot}.
+ *
+ * @param <T> the value of type {@link Number}
+ *
+ * @author Oleg Zhurakousky
+ */
+@JsonPropertyOrder({ "id", "timestamp", "sum", "count", "mean", "upper", "total"})
+class Metric<T extends Number> {
+
+	private final Date timestamp;
+
+	private final Meter.Id id;
+
+	private final Number sum;
+
+	private final Number count;
+
+	private final Number mean;
+
+	private final Number upper;
+
+	private final Number total;
+
+
+	/**
+	 * Create a new {@link Metric} instance.
+	 * @param id Meter id
+	 * @param snapshot instance of HistogramSnapshot
+	 */
+	Metric(Meter.Id id, HistogramSnapshot snapshot) {
+		this.timestamp = new Date();
+		this.id = id;
+		this.sum = snapshot.total(TimeUnit.MILLISECONDS);
+		this.count = snapshot.count();
+		this.mean = snapshot.mean(TimeUnit.MILLISECONDS);
+		this.upper = snapshot.max(TimeUnit.MILLISECONDS);
+		this.total = snapshot.total(TimeUnit.MILLISECONDS);
+	}
+
+	public Meter.Id getId() {
+		return this.id;
+	}
+
+	@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", timezone = "UTC")
+	public Date getTimestamp() {
+		return this.timestamp;
+	}
+
+	public Number getSum() {
+		return sum;
+	}
+
+	public Number getCount() {
+		return count;
+	}
+
+	public Number getMean() {
+		return mean;
+	}
+
+	public Number getUpper() {
+		return upper;
+	}
+
+	public Number getTotal() {
+		return total;
+	}
+
+	@Override
+	public String toString() {
+		return "Metric [id=" + this.id +
+				", sum=" + this.sum +
+				", count=" + this.count +
+				", mean=" + this.mean +
+				", upper=" + this.upper +
+				", total=" + this.total +
+				", timestamp=" + this.timestamp + "]";
+	}
+}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/micrometer/MetricsPublisherConfig.java

@@ -42,7 +42,7 @@ class MetricsPublisherConfig implements StepRegistryConfig {
 	public String get(String key) {
 		String value = null;
 		if (key.equals(this.prefix() + ".step")) {
-			value = this.applicationMetricsProperties.getScheduleInterval();
+			value = this.applicationMetricsProperties.getScheduleInterval().toString();
 		}
 		return value;
 	}