Add Wavefront Consumer and Wavefront Sink

- Supports direct and proxy access.
 - Uses SpEL expressions to compute output metrics value and tags from the input message.
 - Add tests.

consumer/wavefront-consumer/README.adoc

@@ -0,0 +1,29 @@
+# Wavefront Consumer
+
+This module provides a Wavefront Consumer that can be reused and composed in other applications.
+
+## Beans for injection
+
+You can import the `WavefrontConsumerConfiguration` in the application and then inject the following bean.
+
+`wavefrontConsumer`
+
+You can use `wavefrontConsumer` as a qualifier when injecting.
+
+Type for injection: `Consumer<Message<?>>`
+
+You can ignore the return value from the function as this is a consumer and simply will send the data to Wavefront.
+
+## Configuration Options
+
+All configuration properties are prefixed with `wavefront`.
+
+For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerProperties.java[WavefrontConsumerProperties].
+
+## Tests
+
+See this link:src/test/java/org/springframework/cloud/fn/consumer/wavefront[test suite] for the various ways, this consumer is used.
+
+## Other usage
+
+See this https://github.com/spring-cloud/stream-applications/blob/master/applications/sink/wavefront-sink/README.adoc[README] where this consumer is used to create a Spring Cloud Stream application where it makes a Wavefront sink.

consumer/wavefront-consumer/pom.xml

@@ -0,0 +1,65 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+	<artifactId>wavefront-consumer</artifactId>
+	<version>1.0.0-SNAPSHOT</version>
+	<name>wavefront-consumer</name>
+	<description>Wavefront Consumer</description>
+
+	<parent>
+		<groupId>org.springframework.cloud.fn</groupId>
+		<artifactId>spring-functions-parent</artifactId>
+		<version>1.0.0-SNAPSHOT</version>
+		<relativePath>../../spring-functions-parent</relativePath>
+	</parent>
+
+	<properties>
+		<json-path.version>2.4.0</json-path.version>
+	</properties>
+
+	<dependencies>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-configuration-processor</artifactId>
+			<scope>provided</scope>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-integration</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-validation</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.cloud.fn</groupId>
+			<artifactId>config-common</artifactId>
+			<version>${spring-cloud-fn.version}</version>
+		</dependency>
+
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-test</artifactId>
+			<scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.junit.vintage</groupId>
+					<artifactId>junit-vintage-engine</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<version>4.8.2</version>
+			<scope>test</scope>
+		</dependency>
+	</dependencies>
+
+</project>

consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerConfiguration.java

@@ -0,0 +1,72 @@
+/*
+ * 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.wavefront;
+
+import java.util.function.Consumer;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.boot.autoconfigure.web.client.RestTemplateAutoConfiguration;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.web.client.RestTemplateBuilder;
+import org.springframework.cloud.fn.consumer.wavefront.service.DirectConnectionWavefrontService;
+import org.springframework.cloud.fn.consumer.wavefront.service.ProxyConnectionWavefrontService;
+import org.springframework.cloud.fn.consumer.wavefront.service.WavefrontService;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.messaging.Message;
+import org.springframework.util.StringUtils;
+
+/**
+ * @author Timo Salm
+ */
+@Configuration
+@EnableConfigurationProperties(WavefrontConsumerProperties.class)
+@Import(RestTemplateAutoConfiguration.class)
+public class WavefrontConsumerConfiguration {
+
+	private static final Log logger = LogFactory.getLog(WavefrontConsumerConfiguration.class);
+
+	@Bean
+	public Consumer<Message<?>> wavefrontConsumer(final WavefrontConsumerProperties properties,
+			final WavefrontService service) {
+
+		return message -> {
+			final WavefrontFormat wavefrontFormat = new WavefrontFormat(properties, message);
+			final String formattedString = wavefrontFormat.getFormattedString();
+			service.send(formattedString);
+			if (logger.isDebugEnabled()) {
+				logger.debug(formattedString);
+			}
+		};
+	}
+
+	@Bean
+	public WavefrontService wavefrontService(final WavefrontConsumerProperties properties,
+			final RestTemplateBuilder restTemplateBuilder) {
+
+		if (!StringUtils.isEmpty(properties.getProxyUri())) {
+			return new ProxyConnectionWavefrontService(restTemplateBuilder, properties.getProxyUri());
+		}
+		else {
+			return new DirectConnectionWavefrontService(restTemplateBuilder, properties.getUri(),
+					properties.getApiToken());
+		}
+	}
+}

consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerProperties.java

@@ -0,0 +1,178 @@
+/*
+ * 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.wavefront;
+
+import java.util.Map;
+
+import javax.validation.constraints.AssertTrue;
+import javax.validation.constraints.NotEmpty;
+import javax.validation.constraints.NotNull;
+import javax.validation.constraints.Pattern;
+import javax.validation.constraints.Size;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.expression.Expression;
+import org.springframework.util.StringUtils;
+import org.springframework.validation.annotation.Validated;
+
+/**
+ * @author Timo Salm
+ */
+@ConfigurationProperties("wavefront")
+@Validated
+public class WavefrontConsumerProperties {
+
+	/**
+	 * The name of the metric.Defaults to the application name.
+	 */
+	@Value("${spring.application.name:wavefront.consumer}")
+	private String metricName;
+
+	/**
+	 * Unique application, host, container, or instance that emits metrics.
+	 */
+	private String source;
+
+	/**
+	 * A SpEL expression that evaluates to a metric value.
+	 */
+	private Expression metricExpression;
+
+	/**
+	 * A SpEL expression that evaluates to a timestamp of the metric (optional).
+	 */
+	private Expression timestampExpression;
+
+	/**
+	 * Collection of custom metadata associated with the metric.Point tags cannot be empty.
+	 * Valid characters for keys: alphanumeric, hyphen ("-"), underscore ("_"), dot (".").
+	 * For values any character is allowed, including spaces. To include a double quote, escape it with a backslash, for
+	 * example, \". A backslash cannot be the last character in the tag value.
+	 * Maximum allowed length for a combination of a point tag key and value is 254 characters
+	 * (255 including the "=" separating key and value).
+	 * If the value is longer, the point is rejected and logged
+	 */
+	private Map<String, Expression> tagExpression;
+
+	/**
+	 * The URL of the Wavefront environment.
+	 */
+	private String uri;
+
+	/**
+	 * Wavefront API access token.
+	 */
+	private String apiToken;
+
+	/**
+	 * The URL of the Wavefront proxy.
+	 */
+	private String proxyUri;
+
+	public WavefrontConsumerProperties() {
+	}
+
+	WavefrontConsumerProperties(final String metricName, final String source, final Expression metricExpression,
+			final Expression timestampExpression, final Map<String, Expression> pointTagExpressions,
+			final String wavefrontServerUri, final String wavefrontApiToken, final String wavefrontProxyUrl) {
+
+		setMetricName(metricName);
+		setSource(source);
+		setMetricExpression(metricExpression);
+		setTimestampExpression(timestampExpression);
+		setTagExpression(pointTagExpressions);
+		setUri(wavefrontServerUri);
+		setApiToken(wavefrontApiToken);
+		setProxyUri(wavefrontProxyUrl);
+	}
+
+	@NotEmpty
+	@Pattern(regexp = "^[a-zA-Z0-9./_,-]+")
+	public String getMetricName() {
+		return metricName;
+	}
+
+	public void setMetricName(String metricName) {
+		this.metricName = metricName;
+	}
+
+	@NotEmpty
+	@Size(max = 128)
+	@Pattern(regexp = "^[a-zA-Z0-9._-]+")
+	public String getSource() {
+		return source;
+	}
+
+	public void setSource(String source) {
+		this.source = source;
+	}
+
+	@NotNull
+	public Expression getMetricExpression() {
+		return metricExpression;
+	}
+
+	public void setMetricExpression(Expression metricExpression) {
+		this.metricExpression = metricExpression;
+	}
+
+	public Expression getTimestampExpression() {
+		return timestampExpression;
+	}
+
+	public void setTimestampExpression(Expression timestampExpression) {
+		this.timestampExpression = timestampExpression;
+	}
+
+	public Map<String, Expression> getTagExpression() {
+		return tagExpression;
+	}
+
+	public void setTagExpression(Map<String, Expression> tagExpression) {
+		this.tagExpression = tagExpression;
+	}
+
+	public String getUri() {
+		return uri;
+	}
+
+	public void setUri(String uri) {
+		this.uri = uri;
+	}
+
+	public String getApiToken() {
+		return apiToken;
+	}
+
+	public void setApiToken(String apiToken) {
+		this.apiToken = apiToken;
+	}
+
+	public String getProxyUri() {
+		return proxyUri;
+	}
+
+	public void setProxyUri(String proxyUri) {
+		this.proxyUri = proxyUri;
+	}
+
+	@AssertTrue(message = "Exactly one of 'proxy-uri' or the pair of ('uri' and 'api-token') must be set!")
+	public boolean isMutuallyExclusiveProxyAndDirectAccessWavefrontConfiguration() {
+		return StringUtils.isEmpty(getProxyUri()) ^ (StringUtils.isEmpty(getUri()) || StringUtils.isEmpty(getApiToken()));
+	}
+}

consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontFormat.java

@@ -0,0 +1,125 @@
+/*
+ * 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.wavefront;
+
+import java.util.AbstractMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+import javax.validation.ValidationException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.expression.EvaluationException;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.SpelEvaluationException;
+import org.springframework.messaging.Message;
+
+/**
+ * @author Timo Salm
+ */
+public class WavefrontFormat {
+
+	private static final Log logger = LogFactory.getLog(WavefrontFormat.class);
+
+	private final WavefrontConsumerProperties properties;
+	private final Message<?> message;
+
+	public WavefrontFormat(final WavefrontConsumerProperties properties, Message<?> message) {
+		this.properties = properties;
+		this.message = message;
+	}
+
+	public String getFormattedString() {
+		final Number metricValue = extractMetricValueFromPayload();
+
+		final Map<String, Object> pointTagsMap = extractPointTagsMapFromPayload(
+				properties.getTagExpression(), message);
+		validatePointTagsKeyValuePairs(pointTagsMap);
+		final String formattedPointTagsPart = getFormattedPointTags(pointTagsMap);
+
+		if (properties.getTimestampExpression() == null) {
+			return String.format("\"%s\" %s source=%s %s", properties.getMetricName(), metricValue,
+					properties.getSource(), formattedPointTagsPart).trim();
+		}
+
+		final Long timestamp = extractTimestampFromPayload();
+		return String.format("\"%s\" %s %d source=%s %s", properties.getMetricName(), metricValue, timestamp,
+				properties.getSource(), formattedPointTagsPart).trim();
+	}
+
+	private Long extractTimestampFromPayload() {
+		try {
+			return properties.getTimestampExpression().getValue(message, Long.class);
+		}
+		catch (SpelEvaluationException e) {
+			throw new ValidationException("The timestamp value has to be a number that reflects the epoch seconds of the " +
+					"metric (e.g. 1382754475).", e);
+		}
+	}
+
+	private Number extractMetricValueFromPayload() {
+		try {
+			return properties.getMetricExpression().getValue(message, Number.class);
+		}
+		catch (SpelEvaluationException e) {
+			throw new ValidationException("The metric value has to be a double-precision floating point number or a " +
+					"long integer. It can be positive, negative, or 0.", e);
+		}
+	}
+
+	private String getFormattedPointTags(Map<String, Object> pointTagsMap) {
+		return pointTagsMap.entrySet().stream()
+				.map(it -> String.format("%s=\"%s\"", it.getKey(), it.getValue()))
+				.collect(Collectors.joining(" "));
+	}
+
+	private Map<String, Object> extractPointTagsMapFromPayload(Map<String, Expression> pointTagsExpressionsPointValue, Message<?> message) {
+		return pointTagsExpressionsPointValue.entrySet().stream()
+				.map(it -> {
+					try {
+						final Object pointValue = it.getValue().getValue(message);
+						return new AbstractMap.SimpleEntry<>(it.getKey(), pointValue);
+					}
+					catch (EvaluationException e) {
+						logger.warn("Unable to extract point tag for key " + it.getKey() + " from payload", e);
+						return null;
+					}
+				})
+				.filter(Objects::nonNull)
+				.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+	}
+
+	private void validatePointTagsKeyValuePairs(Map<String, Object> pointTagsMap) {
+		pointTagsMap.forEach((key, value) -> {
+			if (!Pattern.matches("^[a-zA-Z0-9._-]+", key)) {
+				throw new ValidationException("Point tag key \"" + key + "\" contains invalid characters: Valid " +
+						"characters are alphanumeric, hyphen (\"-\"), underscore (\"_\"), dot (\".\")");
+			}
+
+			final int keyValueCombinationLength = key.length() + value.toString().length();
+			if (keyValueCombinationLength > 254) {
+				logger.warn("Maximum allowed length for a combination of a point tag key and value " +
+						"is 254 characters. The length of combination for key " + key + " is " +
+						keyValueCombinationLength + ".");
+			}
+		});
+	}
+}

consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/DirectConnectionWavefrontService.java

@@ -0,0 +1,56 @@
+/*
+ * 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.wavefront.service;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.boot.web.client.RestTemplateBuilder;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * @author Timo Salm
+ */
+public class DirectConnectionWavefrontService implements WavefrontService {
+
+	private static final Log logger = LogFactory.getLog(DirectConnectionWavefrontService.class);
+
+	private final RestTemplate restTemplate;
+	private final String wavefrontDomain;
+	private final String wavefrontToken;
+
+	public DirectConnectionWavefrontService(final RestTemplateBuilder restTemplateBuilder,
+			final String wavefrontServerUri, final String wavefrontApiToken) {
+		this.restTemplate = restTemplateBuilder.build();
+		this.wavefrontDomain = wavefrontServerUri;
+		this.wavefrontToken = wavefrontApiToken;
+	}
+
+	@Override
+	public void send(String metricInWavefrontFormat) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Send metric directly to Wavefront");
+		}
+		final HttpHeaders headers = new HttpHeaders();
+		headers.setBearerAuth(wavefrontToken);
+		final HttpEntity<String> httpEntity = new HttpEntity<>(metricInWavefrontFormat, headers);
+		restTemplate.exchange(wavefrontDomain + "/report", HttpMethod.POST, httpEntity, Void.class);
+	}
+}

consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/ProxyConnectionWavefrontService.java

@@ -0,0 +1,48 @@
+/*
+ * 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.wavefront.service;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.boot.web.client.RestTemplateBuilder;
+import org.springframework.web.client.RestTemplate;
+
+/**
+ * @author Timo Salm
+ */
+public class ProxyConnectionWavefrontService implements WavefrontService {
+
+	private static final Log logger = LogFactory.getLog(ProxyConnectionWavefrontService.class);
+
+	private final RestTemplate restTemplate;
+	private final String wavefrontProxyUrl;
+
+	public ProxyConnectionWavefrontService(final RestTemplateBuilder restTemplateBuilder,
+			final String wavefrontProxyUri) {
+		this.restTemplate = restTemplateBuilder.build();
+		this.wavefrontProxyUrl = wavefrontProxyUri;
+	}
+
+	@Override
+	public void send(String metricInWavefrontFormat) {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Send metric to Wavefront proxy");
+		}
+		restTemplate.postForEntity(wavefrontProxyUrl, metricInWavefrontFormat, Void.class);
+	}
+}

consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/WavefrontService.java

@@ -0,0 +1,24 @@
+/*
+ * 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.wavefront.service;
+
+/**
+ * @author Timo Salm
+ */
+public interface WavefrontService {
+	void send(String metricInWavefrontFormat);
+}

consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerConfigurationTest.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.wavefront;
+
+import java.util.Date;
+import java.util.Locale;
+import java.util.function.Consumer;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.mock.mockito.MockBean;
+import org.springframework.cloud.fn.consumer.wavefront.service.WavefrontService;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.GenericMessage;
+
+/**
+ * @author Timo Salm
+ */
+@SpringBootTest(properties = {
+		"wavefront.metric-name=vehicle-location",
+		"wavefront.source=vehicle-api",
+		"wavefront.metric-expression=#jsonPath(payload,'$.mileage')",
+		"wavefront.timestamp-expression=#jsonPath(payload,'$.receivedAt')",
+		"wavefront.tag-expression.vin=#jsonPath(payload,'$.vin')",
+		"wavefront.tag-expression.latitude=#jsonPath(payload,'$.location.latitude')",
+		"wavefront.proxy-uri=testUrl"
+})
+public class WavefrontConsumerConfigurationTest {
+
+	@Autowired
+	private Consumer<Message<?>> wavefrontConsumer;
+
+	@MockBean
+	private WavefrontService wavefrontServiceMock;
+
+	@BeforeEach
+	public void init() {
+		Locale.setDefault(Locale.US);
+	}
+
+	@Test
+	void testWavefrontConsumer() {
+		final long timestamp = new Date().getTime();
+		final String dataJsonString = "{ \"mileage\": 1.5, \"receivedAt\": " + timestamp + ", \"vin\": \"test-vin\", " +
+				"\"location\": {\"latitude\": 4.53, \"longitude\": 2.89 }}";
+
+		wavefrontConsumer.accept(new GenericMessage(dataJsonString.getBytes()));
+
+		final String formattedString = "\"vehicle-location\" 1.5 " + timestamp + " source=vehicle-api " +
+				"latitude=\"4.53\" vin=\"test-vin\"";
+		Mockito.verify(wavefrontServiceMock, Mockito.times(1)).send(formattedString);
+	}
+
+	@SpringBootApplication
+	static class WavefrontConsumerTestApplication {
+	}
+}

consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerPropertiesTest.java

@@ -0,0 +1,112 @@
+/*
+ * 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.wavefront;
+
+import java.util.Arrays;
+import java.util.List;
+
+import javax.validation.Validation;
+import javax.validation.Validator;
+import javax.validation.ValidatorFactory;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Timo Salm
+ */
+public class WavefrontConsumerPropertiesTest {
+
+	private final Expression testExpression = new SpelExpressionParser().parseExpression("#jsonPath(payload,'$')");
+	private Validator validator;
+
+	@BeforeEach
+	public void setUp() {
+		ValidatorFactory factory = Validation.buildDefaultValidatorFactory();
+		validator = factory.getValidator();
+	}
+
+	@Test
+	void testRequiredProperties() {
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("v", "v", testExpression, null, null, null, null, "proxy");
+		final List<String> emptyValues = Arrays.asList(null, "");
+
+		emptyValues.forEach(emptyValue -> {
+			assertThat(validator.validate(properties).isEmpty()).isTrue();
+			properties.setMetricName(emptyValue);
+			assertThat(validator.validate(properties).isEmpty()).isFalse();
+			properties.setMetricName("v");
+
+			assertThat(validator.validate(properties).isEmpty()).isTrue();
+			properties.setSource(emptyValue);
+			assertThat(validator.validate(properties).isEmpty()).isFalse();
+			properties.setSource("v");
+		});
+		assertThat(validator.validate(properties).isEmpty()).isTrue();
+		properties.setMetricExpression(null);
+		assertThat(validator.validate(properties).isEmpty()).isFalse();
+		properties.setMetricExpression(testExpression);
+	}
+
+	@Test
+	void testValidMetricNameValues() {
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("v", "v", testExpression, null, null, null, null, "proxy");
+		final List<String> validMetricNameValues = Arrays.asList("b", "B", "2", ".", "/", "_", ",", "-", "c.8W-2h_dE_,J-h/");
+		assertThat(validator.validate(properties).isEmpty()).isTrue();
+
+		validMetricNameValues.forEach(validMetricNameValue -> {
+			properties.setMetricName(validMetricNameValue);
+			assertThat(validator.validate(properties).isEmpty()).isTrue();
+		});
+
+		final List<String> invalidMetricNameValues = Arrays.asList(" ", ":", "a B", "#");
+		invalidMetricNameValues.forEach(invalidMetricNameValue -> {
+			properties.setMetricName(invalidMetricNameValue);
+			assertThat(validator.validate(properties).isEmpty()).isFalse();
+		});
+	}
+
+	@Test
+	void testValidSourceValues() {
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("v", "v", testExpression, null, null, null, null, "proxy");
+		final List<String> validSourceValues = Arrays.asList("b", "B", "2", ".", "_", "-", "c.8W-2h_dE_J-h",
+				createStringOfLength(128));
+		assertThat(validator.validate(properties).isEmpty()).isTrue();
+
+		validSourceValues.forEach(validSourceValue -> {
+			properties.setSource(validSourceValue);
+			assertThat(validator.validate(properties).isEmpty()).isTrue();
+		});
+
+		final List<String> invalidSourceValues = Arrays.asList(" ", ":", "a B", "#", "/", ",", createStringOfLength(129));
+		invalidSourceValues.forEach(invalidSourceValue -> {
+			properties.setSource(invalidSourceValue);
+			assertThat(validator.validate(properties).isEmpty()).isFalse();
+		});
+	}
+
+	private String createStringOfLength(int length) {
+		char[] chars = new char[length];
+		Arrays.fill(chars, 'a');
+		return new String(chars);
+	}
+}

consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontFormatTest.java

@@ -0,0 +1,200 @@
+/*
+ * 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.wavefront;
+
+import java.util.AbstractMap;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.BeanUtils;
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.expression.spel.support.StandardEvaluationContext;
+import org.springframework.integration.json.JsonPathUtils;
+import org.springframework.messaging.support.GenericMessage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Timo Salm
+ */
+public class WavefrontFormatTest {
+
+	private final SpelExpressionParser parser = new SpelExpressionParser();
+
+	@BeforeEach
+	public void init() {
+		Locale.setDefault(Locale.US);
+	}
+
+	@Test
+	void testGetFormattedString() {
+		final long timestamp = new Date().getTime();
+		final String dataJsonString = "{ \"value\": 1.5, \"timestamp\": " + timestamp + ", "
+				+ "\"testProp1\": \"testvalue1\", \"testProp2\": \"testvalue2\" }";
+
+		final Map<String, Expression> pointTagsExpressionsPointValueMap = Stream.of(
+				new AbstractMap.SimpleEntry<>("testpoint1", expression("$.testProp1")),
+				new AbstractMap.SimpleEntry<>("testpoint2", expression("$.testProp2"))
+		).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+				expression("$.value"), expression("$.timestamp"), pointTagsExpressionsPointValueMap, null, null, null);
+
+		final String result = new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString();
+		assertThat(result).isEqualTo("\"testMetricName\" 1.5 " + timestamp + " source=testSource testpoint2=\"testvalue2\""
+				+ " testpoint1=\"testvalue1\"", result);
+	}
+
+	@Test
+	void testGetFormattedStringWithoutTimeStamp() {
+		final String dataJsonString = "{ \"value\": 1.5, \"testProp1\": \"testvalue1\", \"testProp2\": \"testvalue2\" }";
+
+		final Map<String, Expression> pointTagsExpressionsPointValueMap =
+				Stream.of(new AbstractMap.SimpleEntry<>("testpoint1", expression("$.testProp1")))
+						.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+				expression("$.value"), null, pointTagsExpressionsPointValueMap, null, null, null);
+
+		final String result = new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString();
+		assertThat(result).isEqualTo("\"testMetricName\" 1.5 source=testSource testpoint1=\"testvalue1\"");
+	}
+
+	@Test
+	void testGetFormattedStringWithoutPointTags() {
+		final long timestamp = new Date().getTime();
+		final String dataJsonString = "{ \"value\": 1.5, \"timestamp\": " + timestamp + "}";
+
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+				expression("$.value"), expression("$.timestamp"), Collections.emptyMap(), null, null, null);
+
+		final String result = new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString();
+		assertThat(result).isEqualTo("\"testMetricName\" 1.5 " + timestamp + " source=testSource");
+	}
+
+	@Test
+	void testInvalidMetricValue() {
+		final String dataJsonString = "{ \"value\": a}";
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+				expression("$.value"), null, Collections.emptyMap(), null, null, null);
+		final Exception exception = Assertions.assertThrows(RuntimeException.class, () ->
+				new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString());
+		assertThat(exception.getLocalizedMessage().startsWith("The metric value has to be a double-precision floating"))
+				.isTrue();
+	}
+
+	@Test
+	void testInvalidTimestampValue() {
+		final String dataJsonString = "{ \"value\": 1.5, \"timestamp\": 2020-06-02T13:53:18+0000}";
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+				expression("$.value"), expression("$.timestamp"), Collections.emptyMap(), null, null, null);
+		final Exception exception = Assertions.assertThrows(RuntimeException.class,
+				() -> new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString());
+		assertThat(exception.getLocalizedMessage().startsWith("The timestamp value has to be a number")).isTrue();
+	}
+
+	@Test
+	void testInvalidPointTagsLengthWarning() {
+		final Logger logger = (Logger) LoggerFactory.getLogger(WavefrontFormat.class);
+		final ListAppender<ILoggingEvent> listAppender = new ListAppender<>();
+		listAppender.start();
+		logger.addAppender(listAppender);
+
+		final String testPointTagKey = createStringOfLength(127);
+
+		final Map<String, Expression> pointTagsExpressionsPointValueMap =
+				Stream.of(new AbstractMap.SimpleEntry<>(testPointTagKey, expression("$.testPoint1")))
+						.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+		final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+				expression("$.value"), null, pointTagsExpressionsPointValueMap, null, null, null);
+
+		final String dataJsonString = "{ \"value\": 1.5, \"testPoint1\": \"" +
+				createStringOfLength(254 - testPointTagKey.length()) + "\" }";
+		new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString();
+		assertThat(listAppender.list.stream()
+				.filter(event -> event.getMessage().startsWith("Maximum allowed length for a combination"))
+				.count()).isEqualTo(0);
+
+		final String dataJsonStringWithTooLongValue = "{ \"value\": 1.5, \"testPoint1\": \"" +
+				createStringOfLength(255 - testPointTagKey.length()) + "\" }";
+		new WavefrontFormat(properties, new GenericMessage<>(dataJsonStringWithTooLongValue)).getFormattedString();
+
+		assertThat(listAppender.list.stream()
+				.filter(event -> event.getMessage().startsWith("Maximum allowed length for a combination")
+						&& event.getLevel().equals(Level.WARN))
+				.count()).isEqualTo(1);
+	}
+
+	@Test
+	void testInvalidPointTagKeys() {
+		final String dataJsonString = "{ \"value\": 1.5, \"testPoint1\": \"testvalue1\" }";
+
+		final List<String> validPointTagKeys = Arrays.asList("b", "B", "2", ".", "_", "-", "c.8W-2h_dE_J-h");
+		for (String validPointTagKey : validPointTagKeys) {
+			final Map<String, Expression> pointTagsExpressionsPointValueMap =
+					Stream.of(new AbstractMap.SimpleEntry<>(validPointTagKey, expression("$.testPoint1")))
+							.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+			final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+					expression("$.value"), null, pointTagsExpressionsPointValueMap, null, null, null);
+			new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString();
+		}
+
+		final List<String> invalidPointTagKeys = Arrays.asList(" ", ":", "a B", "#", "/", ",");
+
+		invalidPointTagKeys.forEach(invalidPointTagKey -> {
+			final Map<String, Expression> pointTagsExpressionsPointValueMap =
+					Stream.of(new AbstractMap.SimpleEntry<>(invalidPointTagKey, expression("$.testPoint1")))
+							.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+			final WavefrontConsumerProperties properties = new WavefrontConsumerProperties("testMetricName", "testSource",
+					expression("$.value"), null, pointTagsExpressionsPointValueMap, null, null, null);
+
+			final Exception exception = Assertions.assertThrows(RuntimeException.class,
+					() -> new WavefrontFormat(properties, new GenericMessage<>(dataJsonString)).getFormattedString());
+			assertThat(exception.getLocalizedMessage()
+					.startsWith("Point tag key \"" + invalidPointTagKey + "\" contains invalid characters")).isTrue();
+		});
+	}
+
+	private String createStringOfLength(int length) {
+		char[] chars = new char[length];
+		Arrays.fill(chars, 'a');
+		return new String(chars);
+	}
+
+	private Expression expression(final String jsonPathExpression) {
+		final Expression expression = parser.parseExpression("#jsonPath(payload,'" + jsonPathExpression + "')");
+		final StandardEvaluationContext evaluationContext = (StandardEvaluationContext) ((SpelExpression) expression).getEvaluationContext();
+		evaluationContext.registerFunction("jsonPath", Objects.requireNonNull(BeanUtils.resolveSignature("evaluate", JsonPathUtils.class)));
+		return expression;
+	}
+}

consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/DirectConnectionWavefrontServiceTest.java

@@ -0,0 +1,64 @@
+/*
+ * 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.wavefront.service;
+
+import java.util.Objects;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import org.springframework.boot.web.client.RestTemplateBuilder;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpMethod;
+import org.springframework.web.client.RestTemplate;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+
+/**
+ * @author Timo Salm
+ */
+public class DirectConnectionWavefrontServiceTest {
+
+	@Test
+	void testSendMetricInWavefrontFormat() {
+		final RestTemplateBuilder restTemplateBuilderMock = mock(RestTemplateBuilder.class);
+		final RestTemplate restTemplateMock = mock(RestTemplate.class);
+		when(restTemplateBuilderMock.build()).thenReturn(restTemplateMock);
+
+		final String metricInWavefrontFormat = "testMetric";
+		final String wavefrontServerUri = "testWavefrontDomain";
+		final String wavefrontApiToken = "testWavefrontToken";
+
+		final WavefrontService service = new DirectConnectionWavefrontService(restTemplateBuilderMock,
+				wavefrontServerUri, wavefrontApiToken);
+		service.send(metricInWavefrontFormat);
+
+		final ArgumentCaptor<HttpEntity> argument = ArgumentCaptor.forClass(HttpEntity.class);
+		verify(restTemplateMock, Mockito.times(1))
+				.exchange(eq(wavefrontServerUri + "/report"), eq(HttpMethod.POST), argument.capture(),
+						eq(Void.class));
+		assertThat(Objects.requireNonNull(argument.getValue().getHeaders().get("Authorization")).get(0))
+				.isEqualTo("Bearer " + wavefrontApiToken);
+		assertThat(Objects.requireNonNull(argument.getValue().getBody())).isEqualTo(metricInWavefrontFormat);
+	}
+}

consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/ProxyConnectionWavefrontServiceTest.java

@@ -0,0 +1,50 @@
+/*
+ * 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.wavefront.service;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import org.springframework.boot.web.client.RestTemplateBuilder;
+import org.springframework.web.client.RestTemplate;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * @author Timo Salm
+ */
+public class ProxyConnectionWavefrontServiceTest {
+
+	@Test
+	void testSendMetricInWavefrontFormat() {
+		final RestTemplateBuilder restTemplateBuilderMock = mock(RestTemplateBuilder.class);
+		final RestTemplate restTemplateMock = mock(RestTemplate.class);
+		when(restTemplateBuilderMock.build()).thenReturn(restTemplateMock);
+
+		final String metricInWavefrontFormat = "testMetric";
+		final String wavefrontProxyUrl = "testWavefrontProxyUrl";
+
+		final WavefrontService service = new ProxyConnectionWavefrontService(restTemplateBuilderMock, wavefrontProxyUrl);
+		service.send(metricInWavefrontFormat);
+
+		verify(restTemplateMock, Mockito.times(1))
+				.postForEntity(eq(wavefrontProxyUrl), eq(metricInWavefrontFormat), eq(Void.class));
+	}
+}

consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/WavefrontServiceConditionTest.java

@@ -0,0 +1,104 @@
+/*
+ * 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.wavefront.service;
+
+import java.util.Objects;
+import java.util.UUID;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.context.annotation.UserConfigurations;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.cloud.fn.common.config.SpelExpressionConverterConfiguration;
+import org.springframework.cloud.fn.consumer.wavefront.WavefrontConsumerConfiguration;
+import org.springframework.core.NestedExceptionUtils;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author Timo Salm
+ */
+public class WavefrontServiceConditionTest {
+
+	private final ApplicationContextRunner runner = new ApplicationContextRunner()
+			.withConfiguration(UserConfigurations.of(WavefrontConsumerConfiguration.class,
+					SpelExpressionConverterConfiguration.class));
+
+	@Test
+	public void proxyConnectionShouldBeUsedIfWavefrontProxyAddressSet() {
+		runner.withPropertyValues(
+				"wavefront.metric-name=vehicle-location",
+				"wavefront.source=vehicle-api",
+				"wavefront.metric-expression=#jsonPath(payload,'$.mileage')",
+				"wavefront.proxy-uri=http://wavefront-proxy.internal:2878",
+				"wavefront.uri=",
+				"wavefront.api-token="
+		).run(context -> {
+			assertThat(context).hasSingleBean(ProxyConnectionWavefrontService.class);
+			assertThat(context).doesNotHaveBean(DirectConnectionWavefrontService.class);
+		});
+	}
+
+	@Test
+	public void proxyConnectionShouldBeUsedIfWavefrontProxyAddressAndDomainAndTokenSet() {
+		runner.withPropertyValues(
+				"wavefront.metric-name=vehicle-location",
+				"wavefront.source=vehicle-api",
+				"wavefront.metric-expression=#jsonPath(payload,'$.mileage')",
+				"wavefront.proxy-uri=http://wavefront-proxy.internal:2878",
+				"wavefront.uri=https://my.wavefront.com",
+				"wavefront.api-token=" + UUID.randomUUID()
+		).run(context -> {
+			assertThat(context).hasFailed();
+			final Throwable rootCause = NestedExceptionUtils.getRootCause(context.getStartupFailure());
+			assertThat(Objects.requireNonNull(rootCause).getLocalizedMessage())
+					.contains("Exactly one of 'proxy-uri' or the pair of ('uri' and 'api-token') must be set!");
+		});
+	}
+
+	@Test
+	public void directConnectionShouldBeUsedIfWavefrontDomainAndTokenSet() {
+		runner.withPropertyValues(
+				"wavefront.metric-name=vehicle-location",
+				"wavefront.source=vehicle-api",
+				"wavefront.metric-expression=#jsonPath(payload,'$.mileage')",
+				"wavefront.proxy-uri=",
+				"wavefront.uri=https://my.wavefront.com",
+				"wavefront.api-token=" + UUID.randomUUID()
+		).run(context -> {
+			assertThat(context).hasSingleBean(DirectConnectionWavefrontService.class);
+			assertThat(context).doesNotHaveBean(ProxyConnectionWavefrontService.class);
+		});
+	}
+
+	@Test
+	public void applicationStartupShouldFailWithMeaningfulErrorMessageIfWavefrontProxyAddressOrDomainAndTokenNotSet() {
+		runner.withPropertyValues(
+				"wavefront.metric-name=vehicle-location",
+				"wavefront.source=vehicle-api",
+				"wavefront.metric-expression=#jsonPath(payload,'$.mileage')",
+				"wavefront.proxy-uri=",
+				"wavefront.uri=",
+				"wavefront.api-token="
+		).run(context -> {
+			assertThat(context).hasFailed();
+			final Throwable rootCause = NestedExceptionUtils.getRootCause(context.getStartupFailure());
+			assertThat(Objects.requireNonNull(rootCause).getLocalizedMessage())
+					.contains("Exactly one of 'proxy-uri' or the pair of ('uri' and 'api-token') must be set!");
+		});
+	}
+}

pom.xml

@@ -68,6 +68,7 @@
 		<module>consumer/websocket-consumer</module>
 		<module>consumer/s3-consumer</module>
 		<module>consumer/twitter-consumer</module>
+		<module>consumer/wavefront-consumer</module>
 
 		<module>function/filter-function</module>
 		<module>function/header-enricher-function</module>