Add gauge support for the Counter Consumer
- Add Micrometer GAUGE Meter metrics. Add new counter.meter-type property to select between counter and gauge meter types. - Add tests and stock-exchange demo application. - Update counter-consumer README. Resolves: https://github.com/spring-cloud/stream-applications/issues/69
This commit is contained in:
Christian Tzolov
@@ -1,23 +1,170 @@
|
||||
# Counter Consumer
|
||||
|
||||
A consumer that allows to compute multiple consumers from an incoming message.
|
||||
It uses micrometer internally and can use various popular TSDB technologies to persist the counter values.
|
||||
The `counter-consumer` is a Java https://docs.oracle.com/javase/8/docs/api/java/util/function/Consumer.html[Consumer<Message<?>>] that computes analytics from the input data messages and publishes them as metrics to various monitoring systems.
|
||||
It leverages the https://micrometer.io[micrometer library] for providing a uniform programming experience across the most popular https://micrometer.io/docs[monitoring systems] and uses https://docs.spring.io/spring-integration/reference/html/spel.html#spel[Spring Expression Language (SpEL)] for defining how the metric names, values and tags are computed from the input data.
|
||||
|
||||
It is worth to look at the https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/counter-sink/README.adoc[counter-sink] where this consumer is used to create a Spring Cloud Stream based counter sink.
|
||||
The counter-consumer can produce two metrics types:
|
||||
|
||||
- https://micrometer.io/docs/concepts#_counters[Counter] - reports a single metric, a count, that increments by a fixed, positive amount. Counters can be used for computing the rates of how the data changes in time.
|
||||
- https://micrometer.io/docs/concepts#_gauges[Gauge] - reports the current value. Typical examples for gauges would be the size of a collection or map or number of threads in a running state.
|
||||
|
||||
A https://micrometer.io/docs/concepts#_meters[Meter] (e.g Counter or Gauge) is uniquely identified by its `name` and `dimensions` (the term dimensions and tags is used interchangeably). Dimensions allow a particular named metric to be sliced to drill down and reason about the data.
|
||||
|
||||
NOTE: As a metrics is uniquely identified by its `name` and `dimensions`, you can assign multiple tags (e.g. key/value pairs) to every metric, but you cannot randomly change those tags afterwards! Monitoring systems such as Prometheus will complain if a metric with the same name has different sets of tags.
|
||||
|
||||
## Beans for injection
|
||||
|
||||
You can import `CounterConsumerConfiguration` in the application and then inject the following bean.
|
||||
Add the counter-consumer dependency to your POM:
|
||||
|
||||
`Consumer<Message<?>> counterConsumer`
|
||||
[source,xml]
|
||||
----
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud.fn</groupId>
|
||||
<artifactId>counter-consumer</artifactId>
|
||||
<version>1.0.0-SNAPSHOT</version>
|
||||
</dependency>
|
||||
----
|
||||
|
||||
You can use `counterConsumer` as a qualifier when injecting.
|
||||
Import the https://github.com/spring-cloud/stream-applications/blob/master/functions/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerConfiguration.java[CounterConsumerConfiguration] in the application and inject the following consumer bean:
|
||||
|
||||
[source,java]
|
||||
----
|
||||
Consumer<Message<?>> counterConsumer
|
||||
----
|
||||
|
||||
For every input https://docs.spring.io/spring-integration/reference/html/message.html[Message] the `counterConsumer` computes the defined metrics and eventually, with the help of the micrometer library, publishes them to the backend monitoring systems. The https://docs.spring.io/spring-integration/reference/html/message.html[Message] is a generic container for data. Each Message instance includes a payload and headers containing user-extensible properties as key-value pairs. Any object can be provided as the payload.
|
||||
The https://docs.spring.io/spring-integration/reference/html/message.html#message-builder[MessageBuilder] helps to create a Message instance from any payload content and assign any key/value as a header:
|
||||
|
||||
[source,java]
|
||||
----
|
||||
Message<String> myMessage = MessageBuilder
|
||||
.withPayload("My message text")
|
||||
.setHeader("kind", "CUSTOM")
|
||||
.setHeader("foo", "bar")
|
||||
.build();
|
||||
----
|
||||
|
||||
The `SpEL` expressions use the `headers` and `payload` keywords to access message’s headers and payload values. For example a counter metrics can have a value amount computed from the size of the input message payload add a `my_tag` tag, extracted from the `kind` header value:
|
||||
|
||||
[source]
|
||||
----
|
||||
counter.amount-expression=payload.lenght()
|
||||
counter.tag.expression.my_tag=headers['kind']
|
||||
----
|
||||
|
||||
Review the https://github.com/spring-cloud/stream-applications/blob/master/functions/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java[CounterConsumerProperties]'s javadocs for further details how to use the SpEL properties.
|
||||
|
||||
By default, Micrometer is packed with a SimpleMeterRegistry that holds the latest value of each meter in memory and doesn’t export the data anywhere.
|
||||
To enable support for another monitoring system you have to add the spring-boot-starter-actuator dependency and the micrometer dependency of the monitoring system of choice:
|
||||
|
||||
[source,xml]
|
||||
----
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
<version>CHANGE TO LATEST VERSION</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.micrometer</groupId>
|
||||
<artifactId>micrometer-registry-[MONITORING SYSTEM NAME]</artifactId>
|
||||
<version>${micrometer.version}</version>
|
||||
</dependency>
|
||||
----
|
||||
|
||||
Follow the https://docs.spring.io/spring-boot/docs/2.3.1.RELEASE/reference/html/production-ready-features.html#production-ready-metrics-export[configuration instructions] for the selected monitoring system. All monitoring configuration properties start with a prefix: `management.metrics.export`.
|
||||
|
||||
## Configuration Options
|
||||
|
||||
All configuration properties are prefixed with `counter`.
|
||||
All `counter-consumer` configuration properties use the `counter` prefix. For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java[CounterConsumerProperties].
|
||||
|
||||
All monitoring configuration properties start with a prefix `management.metrics.export`. For configuring a particular monitoring system follow the provided https://docs.spring.io/spring-boot/docs/2.3.1.RELEASE/reference/html/production-ready-features.html#production-ready-metrics-export[configuration instructions].
|
||||
|
||||
#### Sample Configuration
|
||||
|
||||
Following examples show how to configure counter and gauge metrics over a series of stock-exchange messages like this:
|
||||
|
||||
[source,json]
|
||||
----
|
||||
{
|
||||
"data": {
|
||||
"symbol": "AAPL",
|
||||
"exchange": "XNAS",
|
||||
"open": 318.66,
|
||||
"close": 316.85,
|
||||
"volume": 25672211.0
|
||||
}
|
||||
}
|
||||
----
|
||||
|
||||
The following configuration will create a counter metrics called `stockrates` with two tags: `symbol` and `exchange` computed from the json fields:
|
||||
|
||||
.Counter Metrcis Configuration - count stock transactions
|
||||
|===
|
||||
|Property |Description
|
||||
|
||||
|counter.meter-type=counter
|
||||
|Counter meter type (default)
|
||||
|
||||
|counter.name=stockrates
|
||||
|Metrics name
|
||||
|
||||
|counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
|
||||
|Add tag `symbol` equal to the `date.symbol` field in the json messages.
|
||||
|
||||
|counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
|
||||
|Add tag `exchange` equal to the `date.exchange` field in the json messages.
|
||||
|
||||
|===
|
||||
|
||||
Now you can use the `stockrates` metrics to measure the rates at which the stock transactions occur over a given time interval. Furthermore, you can aggregate those rates by the `symbol` and `exchange` tags.
|
||||
|
||||
To measure the transaction volumes contained in the data.volume JSON fields, you can build a GAUGE metrics like this:
|
||||
|
||||
.Gauge Metrcis Configuration - compute stock volumes
|
||||
|===
|
||||
|Property |Description
|
||||
|
||||
|counter.meter-type=gauge
|
||||
|Gauge meter type
|
||||
|
||||
|counter.name=stockvolumes
|
||||
|Metrics name
|
||||
|
||||
|counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
|
||||
|Add tag `symbol` equal to the `date.symbol` field in the json messages.
|
||||
|
||||
|counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
|
||||
|Add tag `exchange` equal to the `date.exchange` field in the json messages.
|
||||
|
||||
|counter.tag.amount-expression=#jsonPath(payload,'$.data.volume')
|
||||
|Set the Gauge to the `data/volume` field values.
|
||||
|===
|
||||
|
||||
Then use the `stockvolumes` metrics to graph, in real-time, the transaction volumes changes over time. You can aggregate those volumes by the `symbol` and `exchange` tags.
|
||||
|
||||
WARNING: Micrometer implements the Gauges for the purpose of data sampling! There is no information about what might have occurred between two consecutive samples. Any intermediate values set on a gauge are lost by the time the gauge value is reported to a metrics backend.
|
||||
|
||||
To enable one or more https://micrometer.io/docs[supported monitoring systems] you need to add a configuration like this:
|
||||
|
||||
.Wavefront Configuration.
|
||||
|===
|
||||
|Property |Description
|
||||
|
||||
|management.metrics.export.wavefront.enabled=true
|
||||
|Enable or disable the monitoring system. (enabled by default).
|
||||
|
||||
|management.metrics.export.wavefront.uri=YOUR_WAVEFRONT_SERVER_URI
|
||||
|UIR of your Wavefront server or Wavefront Proxy.
|
||||
|
||||
|management.metrics.export.wavefront.api-token=YOUR_API_TOKEN
|
||||
|Wavefront access token.
|
||||
|
||||
|management.metrics.export.wavefront.source=stock-exchange-demo
|
||||
|The `source` is used to distinct your metrics on the Wavefront server.
|
||||
|
||||
|===
|
||||
|
||||
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java[CounterConsumerProperties].
|
||||
|
||||
## Tests
|
||||
|
||||
@@ -25,4 +172,6 @@ See this link:src/test/java/org/springframework/cloud/fn/consumer/counter[test s
|
||||
|
||||
## Other usage
|
||||
|
||||
See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/counter-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a Counter sink.
|
||||
* See the https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/counter-sink/README.adoc[Counter Sink README] where this consumer is used to create a Spring Cloud Stream application where it makes a Counter sink.
|
||||
|
||||
* https://docs.google.com/document/d/1BHBjgMmg4a1ue2wr-dmPTfgaN0so4ufw2XkG541Ac9Q/edit?usp=sharing[Stock Exchange Sample].
|
||||
|
||||
@@ -43,6 +43,21 @@
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>io.micrometer</groupId>
|
||||
<artifactId>micrometer-registry-wavefront</artifactId>
|
||||
<version>1.5.1</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
||||
@@ -22,10 +22,14 @@ import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import io.micrometer.core.instrument.Meter;
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
import io.micrometer.core.instrument.Tag;
|
||||
import io.micrometer.core.instrument.Tags;
|
||||
@@ -51,9 +55,11 @@ import org.springframework.util.StringUtils;
|
||||
* @author Christian Tzolov
|
||||
*/
|
||||
@Configuration
|
||||
@EnableConfigurationProperties({CounterConsumerProperties.class})
|
||||
@EnableConfigurationProperties({ CounterConsumerProperties.class })
|
||||
public class CounterConsumerConfiguration {
|
||||
|
||||
private final Map<Meter.Id, AtomicLong> gaugeValues = new ConcurrentHashMap<>();
|
||||
|
||||
@Bean
|
||||
public Function<String, Expression> stringToSpelFunction(@Lazy EvaluationContext evaluationContext) {
|
||||
return new StringToSpelConversionFunction(evaluationContext);
|
||||
@@ -72,19 +78,19 @@ public class CounterConsumerConfiguration {
|
||||
|
||||
@Bean(name = "counterConsumer")
|
||||
public Consumer<Message<?>> counterConsumer(CounterConsumerProperties properties, MeterRegistry[] meterRegistries,
|
||||
@Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME) EvaluationContext context) {
|
||||
@Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME) EvaluationContext context) {
|
||||
|
||||
return message -> {
|
||||
|
||||
String counterName = properties.getComputedNameExpression().getValue(context, message, CharSequence.class).toString();
|
||||
String meterName = properties.getComputedNameExpression().getValue(context, message, CharSequence.class).toString();
|
||||
|
||||
// All fixed tags together are passed with every counter increment.
|
||||
// All fixed tags together are passed with every meter update.
|
||||
Tags fixedTags = this.toTags(properties.getTag().getFixed());
|
||||
|
||||
double amount = properties.getComputedAmountExpression().getValue(context, message, double.class);
|
||||
|
||||
Map<String, List<Tag>> allGroupedTags = new HashMap<>();
|
||||
// Tag Expressions Counter
|
||||
// Tag Expressions
|
||||
if (properties.getTag().getExpression() != null) {
|
||||
|
||||
Map<String, List<Tag>> groupedTags = properties.getTag().getExpression().entrySet().stream()
|
||||
@@ -93,11 +99,11 @@ public class CounterConsumerConfiguration {
|
||||
toList(namedExpression.getValue().getValue(context, message)).stream()
|
||||
.map(tagValue -> Tag.of(namedExpression.getKey(), tagValue))
|
||||
.collect(Collectors.toList())).flatMap(List::stream)
|
||||
.collect(Collectors.groupingBy(tag -> tag.getKey(), Collectors.toList()));
|
||||
.collect(Collectors.groupingBy(Tag::getKey, Collectors.toList()));
|
||||
allGroupedTags.putAll(groupedTags);
|
||||
}
|
||||
|
||||
this.count(meterRegistries, counterName, fixedTags, allGroupedTags, amount);
|
||||
this.recordMetrics(meterRegistries, meterName, fixedTags, allGroupedTags, amount, properties.getMeterType());
|
||||
};
|
||||
}
|
||||
|
||||
@@ -117,7 +123,7 @@ public class CounterConsumerConfiguration {
|
||||
|
||||
/**
|
||||
* Converts the input value into an list of values. If the value is not a collection/array type the result
|
||||
* is a single element list. For collection/array input value the result is the list of stringifie content of
|
||||
* is a single element list. For collection/array input value the result is the list of stringified content of
|
||||
* this collection.
|
||||
*
|
||||
* @param value input value can be array, collection or single value.
|
||||
@@ -133,17 +139,18 @@ public class CounterConsumerConfiguration {
|
||||
: Arrays.asList(ObjectUtils.toObjectArray(value));
|
||||
|
||||
return valueCollection.stream()
|
||||
.filter(v -> v != null)
|
||||
.filter(Objects::nonNull)
|
||||
.map(Object::toString)
|
||||
.filter(StringUtils::hasText)
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
else {
|
||||
return Arrays.asList(value.toString());
|
||||
return Collections.singletonList(value.toString());
|
||||
}
|
||||
}
|
||||
|
||||
private void count(MeterRegistry[] meterRegistries, String counterName, Tags fixedTags, Map<String, List<Tag>> groupedTags, double amount) {
|
||||
private void recordMetrics(MeterRegistry[] meterRegistries, String meterName, Tags fixedTags, Map<String,
|
||||
List<Tag>> groupedTags, double amount, CounterConsumerProperties.MeterType meterType) {
|
||||
if (!CollectionUtils.isEmpty(groupedTags)) {
|
||||
groupedTags.values().stream().map(List::size).max(Integer::compareTo).ifPresent(
|
||||
max -> {
|
||||
@@ -155,22 +162,48 @@ public class CounterConsumerConfiguration {
|
||||
currentTags.and(Tags.of(e.getKey(), ""));
|
||||
}
|
||||
|
||||
// Increment the counterName increment for every configured MaterRegistry.
|
||||
for (MeterRegistry meterRegistry : meterRegistries) {
|
||||
meterRegistry.counter(counterName, currentTags).increment(amount);
|
||||
}
|
||||
// Update the meterName for every configured MaterRegistry.
|
||||
record(meterRegistries, meterName, currentTags, amount, meterType);
|
||||
}
|
||||
}
|
||||
);
|
||||
}
|
||||
else {
|
||||
// Increment the counterName increment for every configured MaterRegistry.
|
||||
for (MeterRegistry meterRegistry : meterRegistries) {
|
||||
meterRegistry.counter(counterName, fixedTags).increment(amount);
|
||||
// Update the meterName for every configured MaterRegistry.
|
||||
record(meterRegistries, meterName, fixedTags, amount, meterType);
|
||||
}
|
||||
}
|
||||
|
||||
private void record(MeterRegistry[] meterRegistries, String meterName,
|
||||
Iterable<Tag> tags, double meterAmount, CounterConsumerProperties.MeterType meterType) {
|
||||
|
||||
for (MeterRegistry meterRegistry : meterRegistries) {
|
||||
if (meterType == CounterConsumerProperties.MeterType.gauge) {
|
||||
Meter.Id gaugeId = new Meter.Id(meterName, Tags.of(tags), null, null, Meter.Type.GAUGE);
|
||||
if (!this.gaugeValues.containsKey(gaugeId)) {
|
||||
this.gaugeValues.put(gaugeId, new AtomicLong((long) meterAmount));
|
||||
}
|
||||
else {
|
||||
this.gaugeValues.get(gaugeId).set((long) meterAmount);
|
||||
}
|
||||
if (!isMeterRegistryContainsGauge(meterRegistry, gaugeId)) {
|
||||
meterRegistry.gauge(meterName, tags, this.gaugeValues.get(gaugeId), AtomicLong::doubleValue);
|
||||
}
|
||||
}
|
||||
else if (meterType == CounterConsumerProperties.MeterType.counter) {
|
||||
meterRegistry.counter(meterName, tags).increment(meterAmount);
|
||||
}
|
||||
else {
|
||||
throw new RuntimeException("Unknown meter type:" + meterType);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isMeterRegistryContainsGauge(MeterRegistry meterRegistry, Meter.Id gaugeId) {
|
||||
return meterRegistry.find(gaugeId.getName()).gauges().stream()
|
||||
.anyMatch(gauge -> gauge.getId().equals(gaugeId));
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean
|
||||
public SimpleMeterRegistry simpleMeterRegistry() {
|
||||
|
||||
@@ -33,6 +33,21 @@ import org.springframework.validation.annotation.Validated;
|
||||
@Validated
|
||||
public class CounterConsumerProperties {
|
||||
|
||||
enum MeterType {
|
||||
/** Uses the Micrometer Counter meter type. It accumulates intermediate counts toward the point where
|
||||
* the data is sent to the metrics backend.*/
|
||||
counter,
|
||||
/** Uses the Micrometer Gauge meter type. Gauges sample the input time series. Any intermediate values set on
|
||||
* a gauge are lost by the time the gauge value is reported to a metrics backend.
|
||||
* TIP: Never gauge something you can count with a Counter!*/
|
||||
gauge
|
||||
}
|
||||
|
||||
/**
|
||||
* Micrometer meter type used to report the metrics to the backend.
|
||||
*/
|
||||
private MeterType meterType = MeterType.counter;
|
||||
|
||||
/**
|
||||
* The default name of the increment.
|
||||
*/
|
||||
@@ -66,12 +81,20 @@ public class CounterConsumerProperties {
|
||||
/**
|
||||
* Fixed and computed tags to be assignee with the counter increment measurement.
|
||||
*/
|
||||
private MetricsTag tag = new MetricsTag();
|
||||
private final MetricsTag tag = new MetricsTag();
|
||||
|
||||
public MetricsTag getTag() {
|
||||
return tag;
|
||||
}
|
||||
|
||||
public MeterType getMeterType() {
|
||||
return meterType;
|
||||
}
|
||||
|
||||
public void setMeterType(MeterType meterType) {
|
||||
this.meterType = meterType;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
if (name == null && nameExpression == null) {
|
||||
return defaultName;
|
||||
|
||||
@@ -27,7 +27,8 @@ import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.test.annotation.DirtiesContext;
|
||||
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
|
||||
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
|
||||
properties = { "management.metrics.export.wavefront.enabled=false" })
|
||||
@DirtiesContext
|
||||
public class CounterConsumerParentTest {
|
||||
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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
|
||||
*
|
||||
* https://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.fn.consumer.counter;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Christian Tzolov
|
||||
*/
|
||||
@TestPropertySource(properties = {
|
||||
"counter.meter-type=gauge",
|
||||
"counter.name=myGauge",
|
||||
"counter.tag.expression.foo='bar'",
|
||||
"counter.amount-expression=payload.length()"
|
||||
})
|
||||
class GaugeWithAmountTest extends CounterConsumerParentTest {
|
||||
|
||||
@Test
|
||||
void testCounterSink() {
|
||||
String messageSmall = "hello";
|
||||
counterConsumer.accept(new GenericMessage(messageSmall));
|
||||
assertThat(meterRegistry.find("myGauge").gauge().value()).isEqualTo(size(messageSmall));
|
||||
|
||||
assertThat(meterRegistry.find("myGauge").gauge().getId().getTags()).hasSize(1);
|
||||
assertThat(meterRegistry.find("myGauge").gauge().getId().getTag("foo")).isEqualTo("bar");
|
||||
|
||||
String messageMiddle = "hello world";
|
||||
counterConsumer.accept(new GenericMessage(messageMiddle));
|
||||
assertThat(meterRegistry.find("myGauge").gauge().value()).isEqualTo(size(messageMiddle));
|
||||
|
||||
String messageLarge = "hello world, hello people!";
|
||||
counterConsumer.accept(new GenericMessage(messageLarge));
|
||||
assertThat(meterRegistry.find("myGauge").gauge().value()).isEqualTo(size(messageLarge));
|
||||
|
||||
}
|
||||
|
||||
private double size(String msg) {
|
||||
return Long.valueOf(msg.length()).doubleValue();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,76 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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
|
||||
*
|
||||
* https://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.fn.consumer.counter;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collection;
|
||||
import java.util.Iterator;
|
||||
|
||||
import io.micrometer.core.instrument.Counter;
|
||||
import io.micrometer.core.instrument.Tag;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.core.io.DefaultResourceLoader;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.test.context.TestPropertySource;
|
||||
import org.springframework.util.StreamUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Christian Tzolov
|
||||
*/
|
||||
|
||||
@TestPropertySource(properties = {
|
||||
"counter.meter-type=counter",
|
||||
"counter.name=stocks",
|
||||
"counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')",
|
||||
"counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')"
|
||||
})
|
||||
public class StockExchangeAnalyticsTests extends CounterConsumerParentTest {
|
||||
|
||||
@Test
|
||||
public void testCounter() throws IOException {
|
||||
byte[] messageAppl = StreamUtils.copyToByteArray(
|
||||
new DefaultResourceLoader().getResource("classpath:/data/stock_appl.json").getInputStream());
|
||||
|
||||
counterConsumer.accept(MessageBuilder.withPayload(messageAppl).build());
|
||||
counterConsumer.accept(MessageBuilder.withPayload(messageAppl).build());
|
||||
counterConsumer.accept(MessageBuilder.withPayload(messageAppl).build());
|
||||
|
||||
byte[] messageVmw = StreamUtils.copyToByteArray(
|
||||
new DefaultResourceLoader().getResource("classpath:/data/stock_vmw.json").getInputStream());
|
||||
|
||||
counterConsumer.accept(MessageBuilder.withPayload(messageVmw).build());
|
||||
counterConsumer.accept(MessageBuilder.withPayload(messageVmw).build());
|
||||
|
||||
Collection<Counter> counters = meterRegistry.find("stocks").counters();
|
||||
|
||||
assertThat(counters).hasSize(2);
|
||||
|
||||
Iterator<Counter> itr = counters.iterator();
|
||||
|
||||
Counter applCounter = itr.next();
|
||||
assertThat(applCounter.count()).isEqualTo(3);
|
||||
assertThat(applCounter.getId().getTags()).contains(Tag.of("symbol", "AAPL"), Tag.of("exchange", "XNAS"));
|
||||
|
||||
Counter vmwCounter = itr.next();
|
||||
assertThat(vmwCounter.count()).isEqualTo(2);
|
||||
assertThat(vmwCounter.getId().getTags()).contains(Tag.of("symbol", "VMW"), Tag.of("exchange", "NYSE"));
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
/*
|
||||
* Copyright 2020-2020 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
|
||||
*
|
||||
* https://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.fn.consumer.demo;
|
||||
|
||||
import java.util.Random;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Supplier;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import io.micrometer.core.instrument.MeterRegistry;
|
||||
|
||||
import org.springframework.boot.CommandLineRunner;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.fn.consumer.counter.CounterConsumerConfiguration;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
|
||||
/**
|
||||
* Sample Spring Boot Application that uses the counterConsumer to compute running stats from
|
||||
* stock exchange messages.
|
||||
*
|
||||
* Counter configuration:
|
||||
* <code>
|
||||
* --counter.meter-type=counter
|
||||
* --counter.name=stocks
|
||||
* --counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
|
||||
* --counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
|
||||
* </code>
|
||||
*
|
||||
* Gauge configuration:
|
||||
* <code>
|
||||
* --counter.meter-type=gauge
|
||||
* --counter.name=stocks
|
||||
* --counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')
|
||||
* --counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')
|
||||
* --counter.amount-expression=#jsonPath(payload,'$.data.volume')
|
||||
* </code>
|
||||
*
|
||||
* Sample Wavefront configuration:
|
||||
* <code>
|
||||
* --management.metrics.export.wavefront.enabled=true
|
||||
* --management.metrics.export.wavefront.uri=YOUR_WAVEFRONT_SERVER_URI
|
||||
* --management.metrics.export.wavefront.api-token=YOUR_API_TOKEN
|
||||
* --management.metrics.export.wavefront.source=stock-exchange-demo
|
||||
* </code>
|
||||
*
|
||||
* @author Christian Tzolov
|
||||
*/
|
||||
@Import(CounterConsumerConfiguration.class)
|
||||
@SpringBootApplication
|
||||
public class StockExchangeAnalyticsExample {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(StockExchangeAnalyticsExample.class, args);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public CommandLineRunner commandLineRunner(Consumer<Message<?>> counterConsumer,
|
||||
MeterRegistry meterRegistry, Supplier<String> stockMessageGenerator) {
|
||||
|
||||
// Run every second.
|
||||
return args -> Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
|
||||
|
||||
String message = stockMessageGenerator.get();
|
||||
|
||||
// Submit new message using the stockMessageGenerator to generate random stock messages.
|
||||
counterConsumer.accept(MessageBuilder.withPayload(message).build());
|
||||
|
||||
// Print current stock meters
|
||||
System.out.println(meterRegistry.getMeters().stream()
|
||||
.filter(meter -> meter.getId().getName().contains("stocks"))
|
||||
.map(meter -> meter.getId().getType() + " | " + meter.getId() + " | " + meter.measure())
|
||||
.collect(Collectors.joining("\n")) +
|
||||
"\n=========================================================================");
|
||||
|
||||
}, 0, 1000, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Supplier<String> stockMessageGenerator() {
|
||||
final Random random = new Random();
|
||||
final String[][] STOCKS = new String[][] { { "NASDAQ", "GOOGL" }, { "AMS", "TOM2" }, { "NYSE", "CLDR" },
|
||||
{ "NYSE", "VMW" }, { "NYSE", "IBM" }, { "NASDAQ", "MSFT" }, { "NASDAQ", "AAPL" } };
|
||||
|
||||
return () -> {
|
||||
int stockIndex = random.nextInt(STOCKS.length);
|
||||
return "{\n" +
|
||||
" \"data\": {\n" +
|
||||
" \"symbol\": \"" + STOCKS[stockIndex][1] + "\",\n" +
|
||||
" \"exchange\": \"" + STOCKS[stockIndex][0] + "\",\n" +
|
||||
" \"open\": " + (1 + 10 * random.nextDouble()) + ",\n" +
|
||||
" \"close\": " + (1 + 10 * random.nextDouble()) + ",\n" +
|
||||
" \"volume\": " + (1000 + 100000 * random.nextDouble()) + "\n" +
|
||||
" }\n" +
|
||||
"}";
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"data": {
|
||||
"date": "2020-05-21T00:00:00+0000",
|
||||
"symbol": "AAPL",
|
||||
"exchange": "XNAS",
|
||||
"open": 318.66,
|
||||
"high": 320.89,
|
||||
"low": 315.87,
|
||||
"close": 316.85,
|
||||
"volume": 25672211.0
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"data": {
|
||||
"date": "2020-05-21T00:00:00+0000",
|
||||
"symbol": "VMW",
|
||||
"exchange": "NYSE",
|
||||
"open": 160.16,
|
||||
"high": 180.91,
|
||||
"low": 165.87,
|
||||
"close": 170.85,
|
||||
"volume": 5672211.0
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user