From 7f19cfff226ee7a33ee926a2a40804ffa6c9741f Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Sun, 28 Jun 2020 14:47:38 +0200 Subject: [PATCH] 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 --- consumer/counter-consumer/README.adoc | 167 +++++++++++++++++- consumer/counter-consumer/pom.xml | 15 ++ .../counter/CounterConsumerConfiguration.java | 69 ++++++-- .../counter/CounterConsumerProperties.java | 25 ++- .../counter/CounterConsumerParentTest.java | 3 +- .../consumer/counter/GaugeWithAmountTest.java | 59 +++++++ .../counter/StockExchangeAnalyticsTests.java | 76 ++++++++ .../demo/StockExchangeAnalyticsExample.java | 118 +++++++++++++ .../src/test/resources/data/stock_appl.json | 12 ++ .../src/test/resources/data/stock_vmw.json | 12 ++ 10 files changed, 527 insertions(+), 29 deletions(-) create mode 100644 consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/GaugeWithAmountTest.java create mode 100644 consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/StockExchangeAnalyticsTests.java create mode 100644 consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/demo/StockExchangeAnalyticsExample.java create mode 100644 consumer/counter-consumer/src/test/resources/data/stock_appl.json create mode 100644 consumer/counter-consumer/src/test/resources/data/stock_vmw.json diff --git a/consumer/counter-consumer/README.adoc b/consumer/counter-consumer/README.adoc index 463570f7..200e2cb9 100644 --- a/consumer/counter-consumer/README.adoc +++ b/consumer/counter-consumer/README.adoc @@ -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>] 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> counterConsumer` +[source,xml] +---- + + org.springframework.cloud.fn + counter-consumer + 1.0.0-SNAPSHOT + +---- -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> 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 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] +---- + + org.springframework.boot + spring-boot-starter-actuator + CHANGE TO LATEST VERSION + + + + io.micrometer + micrometer-registry-[MONITORING SYSTEM NAME] + ${micrometer.version} + +---- + +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. \ No newline at end of file +* 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]. diff --git a/consumer/counter-consumer/pom.xml b/consumer/counter-consumer/pom.xml index 4f2784d2..f8b1879f 100644 --- a/consumer/counter-consumer/pom.xml +++ b/consumer/counter-consumer/pom.xml @@ -43,6 +43,21 @@ spring-boot-starter-test test + + + org.springframework.boot + spring-boot-starter-actuator + test + + + + io.micrometer + micrometer-registry-wavefront + 1.5.1 + test + + + diff --git a/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerConfiguration.java b/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerConfiguration.java index 08ea8ad2..72f3733a 100644 --- a/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerConfiguration.java +++ b/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerConfiguration.java @@ -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 gaugeValues = new ConcurrentHashMap<>(); + @Bean public Function stringToSpelFunction(@Lazy EvaluationContext evaluationContext) { return new StringToSpelConversionFunction(evaluationContext); @@ -72,19 +78,19 @@ public class CounterConsumerConfiguration { @Bean(name = "counterConsumer") public Consumer> 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> allGroupedTags = new HashMap<>(); - // Tag Expressions Counter + // Tag Expressions if (properties.getTag().getExpression() != null) { Map> 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> groupedTags, double amount) { + private void recordMetrics(MeterRegistry[] meterRegistries, String meterName, Tags fixedTags, Map> 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 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() { diff --git a/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java b/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java index 52de6eef..afbc223d 100644 --- a/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java +++ b/consumer/counter-consumer/src/main/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerProperties.java @@ -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; diff --git a/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerParentTest.java b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerParentTest.java index eb20a78f..e84dcfa0 100644 --- a/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerParentTest.java +++ b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/CounterConsumerParentTest.java @@ -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 { diff --git a/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/GaugeWithAmountTest.java b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/GaugeWithAmountTest.java new file mode 100644 index 00000000..0bc9e4f3 --- /dev/null +++ b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/GaugeWithAmountTest.java @@ -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(); + } +} diff --git a/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/StockExchangeAnalyticsTests.java b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/StockExchangeAnalyticsTests.java new file mode 100644 index 00000000..563b173a --- /dev/null +++ b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/counter/StockExchangeAnalyticsTests.java @@ -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 counters = meterRegistry.find("stocks").counters(); + + assertThat(counters).hasSize(2); + + Iterator 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")); + + } +} diff --git a/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/demo/StockExchangeAnalyticsExample.java b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/demo/StockExchangeAnalyticsExample.java new file mode 100644 index 00000000..a372c5ee --- /dev/null +++ b/consumer/counter-consumer/src/test/java/org/springframework/cloud/fn/consumer/demo/StockExchangeAnalyticsExample.java @@ -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: + * + * --counter.meter-type=counter + * --counter.name=stocks + * --counter.tag.expression.symbol=#jsonPath(payload,'$.data.symbol') + * --counter.tag.expression.exchange=#jsonPath(payload,'$.data.exchange') + * + * + * Gauge configuration: + * + * --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') + * + * + * Sample Wavefront configuration: + * + * --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 + * + * + * @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> counterConsumer, + MeterRegistry meterRegistry, Supplier 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 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" + + "}"; + }; + } +} + diff --git a/consumer/counter-consumer/src/test/resources/data/stock_appl.json b/consumer/counter-consumer/src/test/resources/data/stock_appl.json new file mode 100644 index 00000000..887a0adf --- /dev/null +++ b/consumer/counter-consumer/src/test/resources/data/stock_appl.json @@ -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 + } +} diff --git a/consumer/counter-consumer/src/test/resources/data/stock_vmw.json b/consumer/counter-consumer/src/test/resources/data/stock_vmw.json new file mode 100644 index 00000000..752d43e6 --- /dev/null +++ b/consumer/counter-consumer/src/test/resources/data/stock_vmw.json @@ -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 + } +}