diff --git a/README.adoc b/README.adoc index 821d9184..6768cf5d 100644 --- a/README.adoc +++ b/README.adoc @@ -69,6 +69,9 @@ The following are the four major components of this repository. | |link:functions/consumer/websocket-consumer/README.adoc[Websocker] | +| +|link:functions/consumer/wavefront-consumer/README.adoc[Wavefront] +| |=== === Available Spring Cloud Stream Applications @@ -130,6 +133,9 @@ The following are the four major components of this repository. | |link:applications/sink/twitter-update-sink/README.adoc[Twitter Update] | +|link:applications/sink/wavefront-sink/README.adoc[Wavefront] +| +| |=== === Composite Function Support diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index c7e002c4..29868e94 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -30,5 +30,6 @@ websocket-sink twitter-update-sink twitter-message-sink + wavefront-sink diff --git a/applications/sink/wavefront-sink/README.adoc b/applications/sink/wavefront-sink/README.adoc new file mode 100644 index 00000000..c337700c --- /dev/null +++ b/applications/sink/wavefront-sink/README.adoc @@ -0,0 +1,30 @@ +//tag::ref-doc[] += Wavefront Sink + +The *Wavefront* sink consumes Messages, coverts it into a metric in +https://docs.wavefront.com/wavefront_data_format.html[Wavefront data format] and sends the metric directly to Wavefront or a Wavefront proxy. + +Supports common ETL use-cases, where existing (historical) metrics data has to be cleaned, transformed and stored in Wavefront for further analysis. + +== Options + +The *Wavefront* $$sink$$ has the following options: + +//tag::configuration-properties[] +$$wavefront.api-token$$:: $$Wavefront API access token.$$ *($$String$$, default: `$$$$`)* +$$wavefront.metric-expression$$:: $$A SpEL expression that evaluates to a metric value.$$ *($$Expression$$, default: `$$$$`)* +$$wavefront.metric-name$$:: $$The name of the metric.Defaults to the application name.$$ *($$String$$, default: `$$$$`)* +$$wavefront.proxy-uri$$:: $$The URL of the Wavefront proxy.$$ *($$String$$, default: `$$$$`)* +$$wavefront.source$$:: $$Unique application, host, container, or instance that emits metrics.$$ *($$String$$, default: `$$$$`)* +$$wavefront.tag-expression$$:: $$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$$ *($$Map$$, default: `$$$$`)* +$$wavefront.timestamp-expression$$:: $$A SpEL expression that evaluates to a timestamp of the metric (optional).$$ *($$Expression$$, default: `$$$$`)* +$$wavefront.uri$$:: $$The URL of the Wavefront environment.$$ *($$String$$, default: `$$$$`)* +//end::configuration-properties[] + +//end::ref-doc[] diff --git a/applications/sink/wavefront-sink/pom.xml b/applications/sink/wavefront-sink/pom.xml new file mode 100644 index 00000000..b542a1c5 --- /dev/null +++ b/applications/sink/wavefront-sink/pom.xml @@ -0,0 +1,90 @@ + + + 4.0.0 + wavefront-sink + 3.0.0-SNAPSHOT + wavefront-sink + Wavefront sink app + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + + org.springframework.cloud.fn + wavefront-consumer + ${java-functions.version} + + + + org.springframework.boot + spring-boot-starter-test + test + + + + org.springframework.cloud.fn + function-test-support + ${java-functions.version} + test + + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + wavefront + sink + ${project.version} + org.springframework.cloud.fn.consumer.wavefront.WavefrontConsumerConfiguration.class + + wavefrontConsumer + + + + org.springframework.cloud.fn + wavefront-consumer + ${java-functions.version} + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/sink/wavefront-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/wavefront-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..42098591 --- /dev/null +++ b/applications/sink/wavefront-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.wavefront.WavefrontConsumerProperties diff --git a/applications/sink/wavefront-sink/src/test/java/org/springframework/cloud/stream/app/sink/wavefront/WavefrontSinkTests.java b/applications/sink/wavefront-sink/src/test/java/org/springframework/cloud/stream/app/sink/wavefront/WavefrontSinkTests.java new file mode 100644 index 00000000..30e77f6d --- /dev/null +++ b/applications/sink/wavefront-sink/src/test/java/org/springframework/cloud/stream/app/sink/wavefront/WavefrontSinkTests.java @@ -0,0 +1,80 @@ +/* + * 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.stream.app.sink.wavefront; + +import java.util.Date; +import java.util.Locale; + +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.WavefrontConsumerConfiguration; +import org.springframework.cloud.fn.consumer.wavefront.service.WavefrontService; +import org.springframework.cloud.stream.binder.test.InputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.annotation.Import; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Timo Salm + */ +@Import(TestChannelBinderConfiguration.class) +@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 WavefrontSinkTests { + + @Autowired + private InputDestination input; + + @MockBean + private WavefrontService wavefrontServiceMock; + + @BeforeEach + public void init() { + Locale.setDefault(Locale.US); + } + + @Test + void testWavefrontSink() { + final long timestamp = new Date().getTime(); + final String dataJsonString = "{ \"mileage\": 1.5, \"receivedAt\": " + timestamp + ", \"vin\": \"test-vin\", " + + "\"location\": {\"latitude\": 4.53, \"longitude\": 2.89 }}"; + + input.send(MessageBuilder.withPayload(dataJsonString).build()); + + 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 + @Import(WavefrontConsumerConfiguration.class) + static class WavefrontSinkTestApplication { + } +} diff --git a/functions/consumer/wavefront-consumer/README.adoc b/functions/consumer/wavefront-consumer/README.adoc new file mode 100644 index 00000000..3de258ce --- /dev/null +++ b/functions/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>` + +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. diff --git a/functions/consumer/wavefront-consumer/pom.xml b/functions/consumer/wavefront-consumer/pom.xml new file mode 100644 index 00000000..11d6b56e --- /dev/null +++ b/functions/consumer/wavefront-consumer/pom.xml @@ -0,0 +1,65 @@ + + + 4.0.0 + wavefront-consumer + 1.0.0-SNAPSHOT + wavefront-consumer + Wavefront Consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + 2.4.0 + + + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework.cloud.fn + config-common + ${spring-cloud-fn.version} + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.junit.vintage + junit-vintage-engine + + + + + junit + junit + 4.8.2 + test + + + + diff --git a/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerConfiguration.java b/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerConfiguration.java new file mode 100644 index 00000000..bcd35214 --- /dev/null +++ b/functions/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> 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()); + } + } +} diff --git a/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerProperties.java b/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerProperties.java new file mode 100644 index 00000000..3d33c039 --- /dev/null +++ b/functions/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 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 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 getTagExpression() { + return tagExpression; + } + + public void setTagExpression(Map 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())); + } +} diff --git a/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontFormat.java b/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontFormat.java new file mode 100644 index 00000000..6bacd711 --- /dev/null +++ b/functions/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 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 pointTagsMap) { + return pointTagsMap.entrySet().stream() + .map(it -> String.format("%s=\"%s\"", it.getKey(), it.getValue())) + .collect(Collectors.joining(" ")); + } + + private Map extractPointTagsMapFromPayload(Map 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 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 + "."); + } + }); + } +} diff --git a/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/DirectConnectionWavefrontService.java b/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/DirectConnectionWavefrontService.java new file mode 100644 index 00000000..d2860d3a --- /dev/null +++ b/functions/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 httpEntity = new HttpEntity<>(metricInWavefrontFormat, headers); + restTemplate.exchange(wavefrontDomain + "/report", HttpMethod.POST, httpEntity, Void.class); + } +} diff --git a/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/ProxyConnectionWavefrontService.java b/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/ProxyConnectionWavefrontService.java new file mode 100644 index 00000000..1b61aebb --- /dev/null +++ b/functions/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); + } +} diff --git a/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/WavefrontService.java b/functions/consumer/wavefront-consumer/src/main/java/org/springframework/cloud/fn/consumer/wavefront/service/WavefrontService.java new file mode 100644 index 00000000..2d75f1d9 --- /dev/null +++ b/functions/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); +} diff --git a/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerConfigurationTest.java b/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerConfigurationTest.java new file mode 100644 index 00000000..10e32b8a --- /dev/null +++ b/functions/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> 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 { + } +} diff --git a/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerPropertiesTest.java b/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontConsumerPropertiesTest.java new file mode 100644 index 00000000..d2b8dab2 --- /dev/null +++ b/functions/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 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 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 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 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 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); + } +} diff --git a/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontFormatTest.java b/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/WavefrontFormatTest.java new file mode 100644 index 00000000..8983f267 --- /dev/null +++ b/functions/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 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 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 listAppender = new ListAppender<>(); + listAppender.start(); + logger.addAppender(listAppender); + + final String testPointTagKey = createStringOfLength(127); + + final Map 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 validPointTagKeys = Arrays.asList("b", "B", "2", ".", "_", "-", "c.8W-2h_dE_J-h"); + for (String validPointTagKey : validPointTagKeys) { + final Map 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 invalidPointTagKeys = Arrays.asList(" ", ":", "a B", "#", "/", ","); + + invalidPointTagKeys.forEach(invalidPointTagKey -> { + final Map 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; + } +} diff --git a/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/DirectConnectionWavefrontServiceTest.java b/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/DirectConnectionWavefrontServiceTest.java new file mode 100644 index 00000000..968ccd58 --- /dev/null +++ b/functions/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 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); + } +} diff --git a/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/ProxyConnectionWavefrontServiceTest.java b/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/ProxyConnectionWavefrontServiceTest.java new file mode 100644 index 00000000..3f8f60bd --- /dev/null +++ b/functions/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)); + } +} diff --git a/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/WavefrontServiceConditionTest.java b/functions/consumer/wavefront-consumer/src/test/java/org/springframework/cloud/fn/consumer/wavefront/service/WavefrontServiceConditionTest.java new file mode 100644 index 00000000..7cbb224e --- /dev/null +++ b/functions/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!"); + }); + } +} diff --git a/functions/pom.xml b/functions/pom.xml index 34569152..5b6156f5 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -68,6 +68,7 @@ consumer/websocket-consumer consumer/s3-consumer consumer/twitter-consumer + consumer/wavefront-consumer function/filter-function function/header-enricher-function