diff --git a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserver.java b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserver.java index c189616f..42bc9225 100644 --- a/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserver.java +++ b/spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserver.java @@ -26,7 +26,9 @@ import org.apache.commons.logging.LogFactory; import org.springframework.http.HttpEntity; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; import org.springframework.web.client.HttpClientErrorException; import org.springframework.web.client.RestClientException; import org.springframework.web.client.RestTemplate; @@ -127,7 +129,11 @@ public class AtlasMetricObserver implements MetricObserver { } } - private void sendMetricsBatch(List metrics) { + enum PublishMetricsBatchStatus { + NothingToDo, Success, PartialSuccess, Failure + } + + PublishMetricsBatchStatus sendMetricsBatch(List metrics) { try { ByteArrayOutputStream output = new ByteArrayOutputStream(); JsonGenerator gen = smileFactory.createGenerator(output, JsonEncoding.UTF8); @@ -136,7 +142,7 @@ public class AtlasMetricObserver implements MetricObserver { writeCommonTags(gen); if (writeMetrics(gen, metrics) == 0) - return; // short circuit this batch if no valid/numeric metrics existed + return PublishMetricsBatchStatus.NothingToDo; // short circuit this batch if no valid/numeric metrics existed gen.writeEndObject(); gen.flush(); @@ -145,22 +151,32 @@ public class AtlasMetricObserver implements MetricObserver { headers.setContentType(MediaType.valueOf("application/x-jackson-smile")); HttpEntity entity = new HttpEntity<>(output.toByteArray(), headers); try { - restTemplate.exchange(uri, HttpMethod.POST, entity, Map.class); + ResponseEntity response = restTemplate.exchange(uri, HttpMethod.POST, entity, Map.class); + if(response.getStatusCode() == HttpStatus.ACCEPTED) { + // partial success processing the metrics batch + List messages = (List) response.getBody().get("message"); + if(messages != null) { + for (String message : messages) { + logger.error("Failed to write metric to atlas: " + message); + } + } + return PublishMetricsBatchStatus.PartialSuccess; + } } catch (HttpClientErrorException e) { - logger.error( - "Failed to write metrics to atlas: " - + e.getResponseBodyAsString(), e); + logger.error("Failed to write metrics to atlas: " + e.getResponseBodyAsString()); + return PublishMetricsBatchStatus.Failure; } catch (RestClientException e) { logger.error("Failed to write metrics to atlas", e); + return PublishMetricsBatchStatus.Failure; } } catch (IOException e) { - // an IOException stemming from the generator writing to a - // ByteArrayOutputStream is impossible - throw new RuntimeException(e); + return PublishMetricsBatchStatus.Failure; } + + return PublishMetricsBatchStatus.Success; } private void writeCommonTags(JsonGenerator gen) throws IOException { @@ -208,8 +224,8 @@ public class AtlasMetricObserver implements MetricObserver { Metric transformed; // Atlas will not normalize metrics tagged with atlas.dstype=gauge. Since - // these metric types are - // pre-normalized, we do not want Atlas to touch the value + // these metric types are pre-normalized, we do not want Atlas to touch the + // value if (DataSourceType.GAUGE.name().equals(value) || DataSourceType.RATE.name().equals(value) || DataSourceType.NORMALIZED.name().equals(value)) { @@ -218,10 +234,8 @@ public class AtlasMetricObserver implements MetricObserver { } // atlas.dstype=counter means you're sending the absolute value of the counter - // (a monotonically - // increasing value), and Atlas will keep the previous value and convert it to - // a rate per second - // when the metric is received + // (a monotonically increasing value), and Atlas will keep the previous value + // and convert it to a rate per second when the metric is received else if (DataSourceType.COUNTER.name().equals(value)) { transformed = new Metric( m.getConfig().withAdditionalTag(atlasCounterTag), diff --git a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserverTests.java b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserverTests.java index b1c4d3d8..5c1a4464 100644 --- a/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserverTests.java +++ b/spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserverTests.java @@ -19,6 +19,7 @@ import java.util.List; import org.junit.Test; import org.springframework.http.HttpMethod; +import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.test.web.client.MockRestServiceServer; import org.springframework.test.web.client.match.MockRestRequestMatchers; @@ -30,14 +31,11 @@ import com.netflix.servo.annotations.DataSourceType; import com.netflix.servo.monitor.MonitorConfig; import com.netflix.servo.tag.BasicTagList; -import static com.netflix.servo.annotations.DataSourceType.COUNTER; -import static com.netflix.servo.annotations.DataSourceType.GAUGE; -import static com.netflix.servo.annotations.DataSourceType.INFORMATIONAL; -import static com.netflix.servo.annotations.DataSourceType.KEY; -import static com.netflix.servo.annotations.DataSourceType.NORMALIZED; -import static com.netflix.servo.annotations.DataSourceType.RATE; -import static org.junit.Assert.assertEquals; +import static com.netflix.servo.annotations.DataSourceType.*; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.core.Is.is; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; /** @@ -47,10 +45,10 @@ public class AtlasMetricObserverTests { @Test public void normalizeAtlasUri() { String normalized = "http://localhost:7001/api/v1/publish"; - assertEquals(normalized, AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001")); - assertEquals(normalized, AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001/")); - assertEquals(normalized, AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001/api/v1/publish")); - assertEquals(normalized, AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001/api/v1/publish/")); + assertThat(AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001"), is(equalTo(normalized))); + assertThat(AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001/"), is(equalTo(normalized))); + assertThat(AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001/api/v1/publish"), is(equalTo(normalized))); + assertThat(AtlasMetricObserver.normalizeAtlasUri("http://localhost:7001/api/v1/publish/"), is(equalTo(normalized))); } @Test(expected = IllegalStateException.class) @@ -80,25 +78,24 @@ public class AtlasMetricObserverTests { assertHasAtlasType("rate", metricWithType("foo", INFORMATIONAL)); assertHasAtlasType("rate", new Metric(new MonitorConfig.Builder("foo").build(), - 0, "bar")); + System.currentTimeMillis(), "bar")); // already has type Metric m = new Metric(new MonitorConfig.Builder("foo") .withTag(KEY, COUNTER.name()).withTag("atlas.dstype", "counter").build(), - 0, "bar"); + System.currentTimeMillis(), "bar"); assertHasAtlasType("counter", m); - assertEquals(2, m.getConfig().getTags().size()); + assertThat(m.getConfig().getTags().size(), is(equalTo(2))); } private void assertHasAtlasType(String atlasType, Metric m) { - assertEquals(atlasType, - AtlasMetricObserver.addTypeTagsAsNecessary(Collections.singletonList(m)) - .get(0).getConfig().getTags().getValue("atlas.dstype")); + assertThat(AtlasMetricObserver.addTypeTagsAsNecessary(Collections.singletonList(m)) + .get(0).getConfig().getTags().getValue("atlas.dstype"), is(equalTo(atlasType))); } private Metric metricWithType(String key, DataSourceType type) { return new Metric(new MonitorConfig.Builder(key).withTag(KEY, type.name()) - .build(), 0, 1); + .build(), System.currentTimeMillis(), 1); } @Test @@ -145,6 +142,56 @@ public class AtlasMetricObserverTests { mockServer.verify(); } + /** + * If ALL of the metrics in a batch fail, Atlas will return a 400 with a String body indicating why. + */ + @Test + public void failingMetricsBatch() { + RestTemplate restTemplate = new RestTemplate(); + + AtlasMetricObserverConfigBean config = new AtlasMetricObserverConfigBean(); + config.setBatchSize(1); + config.setUri("atlas"); + + MockRestServiceServer mockServer = MockRestServiceServer.createServer(restTemplate); + mockServer + .expect(MockRestRequestMatchers.requestTo("atlas/api/v1/publish")) + .andExpect(MockRestRequestMatchers.method(HttpMethod.POST)) + .andRespond(MockRestResponseCreators.withBadRequest().body("foo0 is bad for some reason")); + + AtlasMetricObserver obs = new AtlasMetricObserver(config, restTemplate, BasicTagList.EMPTY); + + assertThat(obs.sendMetricsBatch(generateMetrics(1)), + is(equalTo(AtlasMetricObserver.PublishMetricsBatchStatus.Failure))); + } + + /** + * If SOME metrics in a batch fail, Atlas will return a 202 with a JSON body with a message for each + * failing metric. + */ + @Test + public void partialSuccessMetricsBatch() { + RestTemplate restTemplate = new RestTemplate(); + + AtlasMetricObserverConfigBean config = new AtlasMetricObserverConfigBean(); + config.setBatchSize(2); + config.setUri("atlas"); + + MockRestServiceServer mockServer = MockRestServiceServer.createServer(restTemplate); + mockServer + .expect(MockRestRequestMatchers.requestTo("atlas/api/v1/publish")) + .andExpect(MockRestRequestMatchers.method(HttpMethod.POST)) + .andRespond( + MockRestResponseCreators.withStatus(HttpStatus.ACCEPTED) + .body("{\"message\" : [\"foo1 is bad for some reason\"]}") + .contentType(MediaType.APPLICATION_JSON)); + + AtlasMetricObserver obs = new AtlasMetricObserver(config, restTemplate, BasicTagList.EMPTY); + + assertThat(obs.sendMetricsBatch(generateMetrics(2)), + is(equalTo(AtlasMetricObserver.PublishMetricsBatchStatus.PartialSuccess))); + } + private List generateMetrics(int numberOfMetrics) { List metrics = new ArrayList<>(); for (int i = 0; i < numberOfMetrics; i++)