Export metrics to Wavefront using WavefrontSender

This commit upgrades the Wavefront metrics export auto-configuration to
provide a `WavefrontSender` if necessary and use that to export metrics
rather than the http client Micrometer used previously.

As a result, the "read-timeout" and "connect-timeout" properties are no
longer honoured.

Closes gh-20810

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/metrics/export/wavefront/WavefrontMetricsExportAutoConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-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.
@@ -16,14 +16,19 @@
 
 package org.springframework.boot.actuate.autoconfigure.metrics.export.wavefront;
 
+import java.time.Duration;
+
+import com.wavefront.sdk.common.WavefrontSender;
+import com.wavefront.sdk.direct.ingestion.WavefrontDirectIngestionClient.Builder;
 import io.micrometer.core.instrument.Clock;
-import io.micrometer.core.ipc.http.HttpUrlConnectionSender;
+import io.micrometer.core.instrument.config.MissingRequiredConfigurationException;
 import io.micrometer.wavefront.WavefrontConfig;
 import io.micrometer.wavefront.WavefrontMeterRegistry;
 
 import org.springframework.boot.actuate.autoconfigure.metrics.CompositeMeterRegistryAutoConfiguration;
 import org.springframework.boot.actuate.autoconfigure.metrics.MetricsAutoConfiguration;
 import org.springframework.boot.actuate.autoconfigure.metrics.export.simple.SimpleMetricsExportAutoConfiguration;
+import org.springframework.boot.actuate.autoconfigure.metrics.export.wavefront.WavefrontProperties.Sender;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -32,21 +37,25 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.util.StringUtils;
+import org.springframework.util.unit.DataSize;
 
 /**
  * {@link EnableAutoConfiguration Auto-configuration} for exporting metrics to Wavefront.
  *
  * @author Jon Schneider
  * @author Artsiom Yudovin
+ * @author Stephane Nicoll
  * @since 2.0.0
  */
 @Configuration(proxyBeanMethods = false)
 @AutoConfigureBefore({ CompositeMeterRegistryAutoConfiguration.class, SimpleMetricsExportAutoConfiguration.class })
 @AutoConfigureAfter(MetricsAutoConfiguration.class)
 @ConditionalOnBean(Clock.class)
-@ConditionalOnClass(WavefrontMeterRegistry.class)
+@ConditionalOnClass({ WavefrontMeterRegistry.class, WavefrontSender.class })
 @ConditionalOnProperty(prefix = "management.metrics.export.wavefront", name = "enabled", havingValue = "true",
 		matchIfMissing = true)
 @EnableConfigurationProperties(WavefrontProperties.class)
@@ -66,10 +75,38 @@ public class WavefrontMetricsExportAutoConfiguration {
 
 	@Bean
 	@ConditionalOnMissingBean
-	public WavefrontMeterRegistry wavefrontMeterRegistry(WavefrontConfig wavefrontConfig, Clock clock) {
+	public WavefrontSender wavefrontSender(WavefrontConfig wavefrontConfig) {
-		return WavefrontMeterRegistry.builder(wavefrontConfig).clock(clock).httpClient(
+		if (!StringUtils.hasText(wavefrontConfig.apiToken())) {
-				new HttpUrlConnectionSender(this.properties.getConnectTimeout(), this.properties.getReadTimeout()))
+			throw new MissingRequiredConfigurationException(
-				.build();
+					"apiToken must be set whenever publishing directly to the Wavefront API");
+		}
+		return createWavefrontSender(wavefrontConfig);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	public WavefrontMeterRegistry wavefrontMeterRegistry(WavefrontConfig wavefrontConfig, Clock clock,
+			WavefrontSender wavefrontSender) {
+		return WavefrontMeterRegistry.builder(wavefrontConfig).clock(clock).wavefrontSender(wavefrontSender).build();
+	}
+
+	private WavefrontSender createWavefrontSender(WavefrontConfig wavefrontConfig) {
+		Builder builder = new Builder(getWavefrontReportingUri(wavefrontConfig.uri()), wavefrontConfig.apiToken());
+		PropertyMapper mapper = PropertyMapper.get().alwaysApplyingWhenNonNull();
+		Sender sender = this.properties.getSender();
+		mapper.from(sender.getMaxQueueSize()).to(builder::maxQueueSize);
+		mapper.from(sender.getBatchSize()).to(builder::batchSize);
+		mapper.from(sender.getFlushInterval()).asInt(Duration::getSeconds).to(builder::flushIntervalSeconds);
+		mapper.from(sender.getMessageSize()).asInt(DataSize::toBytes).to(builder::messageSizeBytes);
+		return builder.build();
+	}
+
+	private String getWavefrontReportingUri(String uri) {
+		// proxy reporting is now http reporting on newer wavefront proxies.
+		if (uri.startsWith("proxy")) {
+			return "http" + uri.substring(5);
+		}
+		return uri;
 	}
 
 }

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/java/org/springframework/boot/actuate/autoconfigure/metrics/export/wavefront/WavefrontProperties.java

@@ -17,15 +17,18 @@
 package org.springframework.boot.actuate.autoconfigure.metrics.export.wavefront;
 
 import java.net.URI;
+import java.time.Duration;
 
 import org.springframework.boot.actuate.autoconfigure.metrics.export.properties.PushRegistryProperties;
 import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.util.unit.DataSize;
 
 /**
  * {@link ConfigurationProperties @ConfigurationProperties} for configuring Wavefront
  * metrics export.
  *
  * @author Jon Schneider
+ * @author Stephane Nicoll
  * @since 2.0.0
  */
 @ConfigurationProperties("management.metrics.export.wavefront")
@@ -54,6 +57,8 @@ public class WavefrontProperties extends PushRegistryProperties {
 	 */
 	private String globalPrefix;
 
+	private final Sender sender = new Sender();
+
 	public URI getUri() {
 		return this.uri;
 	}
@@ -86,4 +91,52 @@ public class WavefrontProperties extends PushRegistryProperties {
 		this.globalPrefix = globalPrefix;
 	}
 
+	public Sender getSender() {
+		return this.sender;
+	}
+
+	public static class Sender {
+
+		private int maxQueueSize = 50000;
+
+		private int batchSize = 10000;
+
+		private Duration flushInterval = Duration.ofSeconds(1);
+
+		private DataSize messageSize = DataSize.ofBytes(Integer.MAX_VALUE);
+
+		public int getMaxQueueSize() {
+			return this.maxQueueSize;
+		}
+
+		public void setMaxQueueSize(int maxQueueSize) {
+			this.maxQueueSize = maxQueueSize;
+		}
+
+		public int getBatchSize() {
+			return this.batchSize;
+		}
+
+		public void setBatchSize(int batchSize) {
+			this.batchSize = batchSize;
+		}
+
+		public Duration getFlushInterval() {
+			return this.flushInterval;
+		}
+
+		public void setFlushInterval(Duration flushInterval) {
+			this.flushInterval = flushInterval;
+		}
+
+		public DataSize getMessageSize() {
+			return this.messageSize;
+		}
+
+		public void setMessageSize(DataSize messageSize) {
+			this.messageSize = messageSize;
+		}
+
+	}
+
 }

spring-boot-project/spring-boot-actuator-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -380,6 +380,18 @@
         "level": "error"
       }
     },
+    {
+      "name": "management.metrics.export.wavefront.connect-timeout",
+      "deprecation": {
+        "level": "error"
+      }
+    },
+    {
+      "name": "management.metrics.export.wavefront.read-timeout",
+      "deprecation": {
+        "level": "error"
+      }
+    },
     {
       "name": "management.metrics.web.client.request.autotime.enabled",
       "description": "Whether to automatically time web client requests.",

spring-boot-project/spring-boot-actuator-autoconfigure/src/test/java/org/springframework/boot/actuate/autoconfigure/metrics/export/wavefront/WavefrontMetricsExportAutoConfigurationTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-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.
@@ -16,6 +16,7 @@
 
 package org.springframework.boot.actuate.autoconfigure.metrics.export.wavefront;
 
+import com.wavefront.sdk.common.WavefrontSender;
 import io.micrometer.core.instrument.Clock;
 import io.micrometer.wavefront.WavefrontConfig;
 import io.micrometer.wavefront.WavefrontMeterRegistry;
@@ -28,6 +29,7 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
 
 /**
  * Tests for {@link WavefrontMetricsExportAutoConfiguration}.
@@ -53,9 +55,10 @@ class WavefrontMetricsExportAutoConfigurationTests {
 	@Test
 	void autoConfigurationCanBeDisabled() {
 		this.contextRunner.withUserConfiguration(BaseConfiguration.class)
-				.withPropertyValues("management.metrics.export.wavefront.enabled=false")
+				.withPropertyValues("management.metrics.export.wavefront.api-token=abcde",
+						"management.metrics.export.wavefront.enabled=false")
 				.run((context) -> assertThat(context).doesNotHaveBean(WavefrontMeterRegistry.class)
-						.doesNotHaveBean(WavefrontConfig.class));
+						.doesNotHaveBean(WavefrontConfig.class).doesNotHaveBean(WavefrontSender.class));
 	}
 
 	@Test
@@ -63,7 +66,30 @@ class WavefrontMetricsExportAutoConfigurationTests {
 		this.contextRunner.withUserConfiguration(CustomConfigConfiguration.class)
 				.run((context) -> assertThat(context).hasSingleBean(Clock.class)
 						.hasSingleBean(WavefrontMeterRegistry.class).hasSingleBean(WavefrontConfig.class)
-						.hasBean("customConfig"));
+						.hasSingleBean(WavefrontSender.class).hasBean("customConfig"));
+	}
+
+	@Test
+	void configureWavefrontSender() {
+		this.contextRunner.withUserConfiguration(BaseConfiguration.class)
+				.withPropertyValues("management.metrics.export.wavefront.api-token=abcde",
+						"management.metrics.export.wavefront.sender.max-queue-size=100",
+						"management.metrics.export.wavefront.sender.batch-size=200",
+						"management.metrics.export.wavefront.sender.message-size=1KB")
+				.run((context) -> {
+					WavefrontSender sender = context.getBean(WavefrontSender.class);
+					assertThat(sender).extracting("metricsBuffer").hasFieldOrPropertyWithValue("capacity", 100);
+					assertThat(sender).hasFieldOrPropertyWithValue("batchSize", 200);
+					assertThat(sender).hasFieldOrPropertyWithValue("messageSizeBytes", 1024);
+				});
+	}
+
+	@Test
+	void allowsWavefrontSenderToBeCustomized() {
+		this.contextRunner.withUserConfiguration(CustomSenderConfiguration.class)
+				.run((context) -> assertThat(context).hasSingleBean(Clock.class)
+						.hasSingleBean(WavefrontMeterRegistry.class).hasSingleBean(WavefrontConfig.class)
+						.hasSingleBean(WavefrontSender.class).hasBean("customSender"));
 	}
 
 	@Test
@@ -109,13 +135,29 @@ class WavefrontMetricsExportAutoConfigurationTests {
 
 				@Override
 				public String uri() {
-					return WavefrontConfig.DEFAULT_PROXY.uri();
+					return WavefrontConfig.DEFAULT_DIRECT.uri();
+				}
+
+				@Override
+				public String apiToken() {
+					return "abc-def";
 				}
 			};
 		}
 
 	}
 
+	@Configuration(proxyBeanMethods = false)
+	@Import(BaseConfiguration.class)
+	static class CustomSenderConfiguration {
+
+		@Bean
+		WavefrontSender customSender() {
+			return mock(WavefrontSender.class);
+		}
+
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	@Import(BaseConfiguration.class)
 	static class CustomRegistryConfiguration {