SpringCloud/spring-cloud-openfeign
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Log partial failure reasons for metrics batches published to Atlas (#1314)

Browse Source
This commit is contained in:
Jon Schneider
2016-09-01 11:57:52 -07:00
committed by Spencer Gibb
parent b4284d18d5
commit 8f7ea41fe8
2 changed files with 94 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
spring-cloud-netflix-core/src/main/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserver.java
View File

@@ -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<Metric> metrics) {
enum PublishMetricsBatchStatus {
NothingToDo, Success, PartialSuccess, Failure
}
PublishMetricsBatchStatus sendMetricsBatch(List<Metric> 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<byte[]> entity = new HttpEntity<>(output.toByteArray(), headers);
try {
restTemplate.exchange(uri, HttpMethod.POST, entity, Map.class);
ResponseEntity<Map> response = restTemplate.exchange(uri, HttpMethod.POST, entity, Map.class);
if(response.getStatusCode() == HttpStatus.ACCEPTED) {
// partial success processing the metrics batch
List<String> messages = (List<String>) 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),

83
spring-cloud-netflix-core/src/test/java/org/springframework/cloud/netflix/metrics/atlas/AtlasMetricObserverTests.java
View File

@@ -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<Metric> generateMetrics(int numberOfMetrics) {
List<Metric> metrics = new ArrayList<>();
for (int i = 0; i < numberOfMetrics; i++)