= Analytics Consumer
The `analytics-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.
The analytics-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 afterward!
Monitoring systems such as Prometheus will complain if a metric with the same name has different sets of tags.
== Beans for injection
Add the analytics-consumer dependency to your POM:
[source,xml]
----
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>analytics-consumer</artifactId>
<version>1.0.0-SNAPSHOT</version>
</dependency>
----
The `AnalyticsConsumerConfiguration` auto-configures the following consumer bean:
[source,java]
----
Consumer<Message<?>> analyticsConsumer
----
For every input https://docs.spring.io/spring-integration/reference/html/message.html[Message] the `analyticsConsumer` 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,properties]
----
analytics.amount-expression=payload.lenght()
analytics.tag.expression.my_tag=headers['kind']
----
Review the https://github.com/spring-cloud/stream-applications/blob/master/functions/consumer/analytics-consumer/src/main/java/org/springframework/cloud/fn/consumer/analytics/AnalyticsConsumerProperties.java[AnalyticsConsumerProperties] 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>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-[MONITORING SYSTEM NAME]</artifactId>
</dependency>
----
Follow the https://docs.spring.io/spring-boot/docs/current/reference/htmlsingle/#actuator.metrics[configuration instructions] for the selected monitoring system.
All monitoring configuration properties start with a prefix: `management.metrics.export`.
== Configuration Options
All `analytics-consumer` configuration properties use the `analytics` prefix.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/analytics/AnalyticsConsumerProperties.java[AnalyticsConsumerProperties].
All monitoring configuration properties start with a prefix `management.metrics.export`.
=== 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 Metrics Configuration - count stock transactions
|===
|Property |Description
|`analytics.meter-type=counter`
|Counter meter type (default)
|`analytics.name=stockrates`
|Metrics name
|`analytics.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')`
|Add tag `symbol` equal to the `date.symbol` field in the json messages.
|`analytics.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 Metrics Configuration - compute stock volumes
|===
|Property |Description
|`analytics.meter-type=gauge`
|Gauge meter type
|`analytics.name=stockvolumes`
|Metrics name
|`analytics.tag.expression.symbol=#jsonPath(payload,'$.data.symbol')`
|Add tag `symbol` equal to the `date.symbol` field in the json messages.
|`analytics.tag.expression.exchange=#jsonPath(payload,'$.data.exchange')`
|Add tag `exchange` equal to the `date.exchange` field in the json messages.
|`analytics.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.
|===
== Tests
See this link:src/test/java/org/springframework/cloud/fn/consumer/analytics[test suite] for the various ways, this consumer is used.
== Other usage
* See the https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/analytics-sink/README.adoc[Analytics 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].
550d665bd9