Skip to content

Commit 4f321bc

Browse files
vjkoskelaBrandonArp
vjkoskela
authored and
BrandonArp
committed
Add support for dedicated HTTP endpoints (paths) which either persist or reaggregate. (#84)
1 parent 592f4e6 commit 4f321bc

File tree

8 files changed

+261
-51
lines changed

8 files changed

+261
-51
lines changed

‎src/main/java/com/arpnetworking/clusteraggregator/client/AggClientConnection.java‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import akka.io.Tcp;
2424
import akka.io.TcpMessage;
2525
import akka.util.ByteString;
26+
import com.arpnetworking.clusteraggregator.models.AggregationMode;
2627
import com.arpnetworking.clusteraggregator.models.CombinedMetricData;
2728
import com.arpnetworking.metrics.aggregation.protocol.Messages;
2829
import com.arpnetworking.steno.Logger;
@@ -137,6 +138,8 @@ public Receive createReceive() {
137138
}
138139

139140
private void processMessages() {
141+
final AggregationMode aggregationMode =
142+
_calculateAggregates ? AggregationMode.PERSIST_AND_REAGGREGATE : AggregationMode.PERSIST;
140143
ByteString current = _buffer;
141144
Optional<AggregationMessage> messageOptional = AggregationMessage.deserialize(current);
142145
while (messageOptional.isPresent()) {
@@ -162,13 +165,15 @@ private void processMessages() {
162165
.log();
163166
// StatisticSetRecords get forwarded to the parent, who then forwards them to the shard for cluster aggregating
164167
// If we aren't doing shard aggregating, don't forward it
165-
if (_calculateAggregates) {
168+
if (aggregationMode.shouldReaggregate()) {
166169
getContext().parent().tell(setRecord, getSelf());
167170
}
168-
if (setRecord.getStatisticsCount() > 0) {
169-
final Optional<PeriodicData> periodicData = buildPeriodicData(setRecord);
170-
if (periodicData.isPresent()) {
171-
getContext().parent().tell(periodicData.get(), self());
171+
if (aggregationMode.shouldPersist()) {
172+
if (setRecord.getStatisticsCount() > 0) {
173+
final Optional<PeriodicData> periodicData = buildPeriodicData(setRecord);
174+
if (periodicData.isPresent()) {
175+
getContext().parent().tell(periodicData.get(), self());
176+
}
172177
}
173178
}
174179
} else if (gm instanceof Messages.HeartbeatRecord) {

‎src/main/java/com/arpnetworking/clusteraggregator/client/HttpSourceActor.java‎

Lines changed: 53 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -42,12 +42,15 @@
4242
import akka.stream.javadsl.Zip;
4343
import akka.util.ByteString;
4444
import com.arpnetworking.clusteraggregator.configuration.ClusterAggregatorConfiguration;
45+
import com.arpnetworking.clusteraggregator.http.Routes;
46+
import com.arpnetworking.clusteraggregator.models.AggregationMode;
4547
import com.arpnetworking.clusteraggregator.models.CombinedMetricData;
4648
import com.arpnetworking.metrics.aggregation.protocol.Messages;
4749
import com.arpnetworking.steno.Logger;
4850
import com.arpnetworking.steno.LoggerFactory;
4951
import com.arpnetworking.tsdcore.model.AggregatedData;
5052
import com.arpnetworking.tsdcore.model.AggregationMessage;
53+
import com.arpnetworking.tsdcore.model.AggregationRequest;
5154
import com.arpnetworking.tsdcore.model.FQDSN;
5255
import com.arpnetworking.tsdcore.model.PeriodicData;
5356
import com.arpnetworking.tsdcore.statistics.Statistic;
@@ -61,7 +64,6 @@
6164
import java.io.IOException;
6265
import java.time.Duration;
6366
import java.util.Collections;
64-
import java.util.List;
6567
import java.util.Map;
6668
import java.util.Optional;
6769
import java.util.concurrent.CompletionStage;
@@ -100,18 +102,27 @@ public HttpSourceActor(
100102
@Named("host-emitter") final ActorRef emitter,
101103
final ClusterAggregatorConfiguration configuration) {
102104

105+
_calculateAggregates = configuration.getCalculateClusterAggregations();
106+
103107
final ActorRef self = self();
104-
_sink = Sink.foreach(msg -> {
105-
final GeneratedMessageV3 generatedMessageV3 = msg.getMessage();
106-
if (generatedMessageV3 instanceof Messages.StatisticSetRecord) {
107-
final Messages.StatisticSetRecord statisticSetRecord =
108-
(Messages.StatisticSetRecord) msg.getMessage();
109-
if (configuration.getCalculateClusterAggregations()) {
110-
shardRegion.tell(statisticSetRecord, self);
111-
}
112-
final Optional<PeriodicData> periodicData = buildPeriodicData(statisticSetRecord);
113-
if (periodicData.isPresent()) {
114-
emitter.tell(periodicData.get(), self);
108+
_sink = Sink.foreach(aggregationRequest -> {
109+
final AggregationMode aggregationMode = aggregationRequest.getAggregationMode();
110+
for (final AggregationMessage aggregationMessage : aggregationRequest.getAggregationMessages()) {
111+
final GeneratedMessageV3 generatedMessageV3 = aggregationMessage.getMessage();
112+
if (generatedMessageV3 instanceof Messages.StatisticSetRecord) {
113+
final Messages.StatisticSetRecord statisticSetRecord =
114+
(Messages.StatisticSetRecord) aggregationMessage.getMessage();
115+
116+
if (aggregationMode.shouldReaggregate()) {
117+
shardRegion.tell(statisticSetRecord, self);
118+
}
119+
120+
if (aggregationMode.shouldPersist()) {
121+
final Optional<PeriodicData> periodicData = buildPeriodicData(statisticSetRecord);
122+
if (periodicData.isPresent()) {
123+
emitter.tell(periodicData.get(), self);
124+
}
125+
}
115126
}
116127
}
117128
});
@@ -130,33 +141,27 @@ public HttpSourceActor(
130141
.reduce(ByteString::concat)
131142
.named("getBody");
132143

133-
final Flow<HttpRequest, ImmutableMultimap<String, String>, NotUsed> getHeadersFlow = Flow.<HttpRequest>create()
134-
.map(HttpRequest::getHeaders)
135-
.map(HttpSourceActor::createHeaderMultimap) // Transform to array form
136-
.named("getHeaders");
137-
138-
final Flow<Pair<ByteString, ImmutableMultimap<String, String>>, AggregationMessage, NotUsed> createAndParseFlow =
139-
Flow.<Pair<ByteString, ImmutableMultimap<String, String>>>create()
144+
final Flow<Pair<ByteString, HttpRequest>, AggregationRequest, NotUsed> createAndParseFlow =
145+
Flow.<Pair<ByteString, HttpRequest>>create()
140146
.map(HttpSourceActor::mapModel)
141-
.mapConcat(this::parseRecords) // Parse the json string into a record builder
147+
.map(this::parseRecords) // Parse the json string into a record builder
142148
// NOTE: this should be _parser::parse, but aspectj NPEs with that currently
143149
.named("createAndParseRequest");
144150

145151
// Shapes
146152
final UniformFanOutShape<HttpRequest, HttpRequest> split = builder.add(Broadcast.create(2));
147153

148154
final FlowShape<HttpRequest, ByteString> getBody = builder.add(getBodyFlow);
149-
final FlowShape<HttpRequest, ImmutableMultimap<String, String>> getHeaders = builder.add(getHeadersFlow);
150155
final FanInShape2<
151156
ByteString,
152-
ImmutableMultimap<String, String>,
153-
Pair<ByteString, ImmutableMultimap<String, String>>> join = builder.add(Zip.create());
154-
final FlowShape<Pair<ByteString, ImmutableMultimap<String, String>>, AggregationMessage> createRequest =
157+
HttpRequest,
158+
Pair<ByteString, HttpRequest>> join = builder.add(Zip.create());
159+
final FlowShape<Pair<ByteString, HttpRequest>, AggregationRequest> createRequest =
155160
builder.add(createAndParseFlow);
156161

157162
// Wire the shapes
158163
builder.from(split.out(0)).via(getBody).toInlet(join.in0()); // Split to get the body bytes
159-
builder.from(split.out(1)).via(getHeaders).toInlet(join.in1()); // Split to get the headers
164+
builder.from(split.out(1)).toInlet(join.in1()); // Pass the Akka HTTP request through
160165
builder.from(join.out()).toInlet(createRequest.in()); // Join to create the Request and parse it
161166

162167
return FlowShape.of(split.in(), createRequest.out());
@@ -201,22 +206,30 @@ public Receive createReceive() {
201206
.build();
202207
}
203208

204-
private static ImmutableMultimap<String, String> createHeaderMultimap(final Iterable<HttpHeader> headers) {
209+
private static com.arpnetworking.clusteraggregator.models.HttpRequest mapModel(
210+
final Pair<ByteString, HttpRequest> pair) {
205211
final ImmutableMultimap.Builder<String, String> headersBuilder = ImmutableMultimap.builder();
206212

207-
for (final HttpHeader httpHeader : headers) {
213+
for (final HttpHeader httpHeader : pair.second().getHeaders()) {
208214
headersBuilder.put(httpHeader.lowercaseName(), httpHeader.value());
209215
}
210216

211-
return headersBuilder.build();
217+
return new com.arpnetworking.clusteraggregator.models.HttpRequest(
218+
pair.second().getUri().path(),
219+
headersBuilder.build(),
220+
pair.first());
212221
}
213222

214-
private static com.arpnetworking.clusteraggregator.models.HttpRequest mapModel(
215-
final Pair<ByteString, ImmutableMultimap<String, String>> pair) {
216-
return new com.arpnetworking.clusteraggregator.models.HttpRequest(pair.second(), pair.first());
217-
}
223+
private AggregationRequest parseRecords(final com.arpnetworking.clusteraggregator.models.HttpRequest request) throws IOException {
224+
final AggregationMode aggregationMode;
225+
if (Routes.INCOMING_DATA_REAGGREGATE_V1_PATH.equals(request.getPath())) {
226+
aggregationMode = AggregationMode.REAGGREGATE;
227+
} else if (Routes.INCOMING_DATA_PERSIST_V1_PATH.equals(request.getPath())) {
228+
aggregationMode = AggregationMode.PERSIST;
229+
} else {
230+
aggregationMode = _calculateAggregates ? AggregationMode.PERSIST_AND_REAGGREGATE : AggregationMode.PERSIST;
231+
}
218232

219-
private List<AggregationMessage> parseRecords(final com.arpnetworking.clusteraggregator.models.HttpRequest request) throws IOException {
220233
final ImmutableList.Builder<AggregationMessage> recordsBuilder = ImmutableList.builder();
221234
ByteString current = request.getBody();
222235
Optional<AggregationMessage> messageOptional = AggregationMessage.deserialize(current);
@@ -234,7 +247,10 @@ private List<AggregationMessage> parseRecords(final com.arpnetworking.clusteragg
234247
if (records.size() == 0) {
235248
throw new NoRecordsException();
236249
}
237-
return records;
250+
return new AggregationRequest.Builder()
251+
.setAggregationMode(aggregationMode)
252+
.setAggregationMessages(recordsBuilder.build())
253+
.build();
238254
}
239255

240256
private Optional<PeriodicData> buildPeriodicData(final Messages.StatisticSetRecord setRecord) {
@@ -299,13 +315,13 @@ private Optional<PeriodicData> buildPeriodicData(final Messages.StatisticSetReco
299315
.build());
300316
}
301317

318+
private final boolean _calculateAggregates;
302319
private final Materializer _materializer;
303-
private final Sink<AggregationMessage, CompletionStage<Done>> _sink;
304-
private final Graph<FlowShape<HttpRequest, AggregationMessage>, NotUsed> _processGraph;
320+
private final Sink<AggregationRequest, CompletionStage<Done>> _sink;
321+
private final Graph<FlowShape<HttpRequest, AggregationRequest>, NotUsed> _processGraph;
305322

306323
private static final Logger BAD_REQUEST_LOGGER =
307324
LoggerFactory.getRateLimitLogger(HttpSourceActor.class, Duration.ofSeconds(30));
308-
private static final Logger LOGGER = LoggerFactory.getLogger(HttpSourceActor.class);
309325

310326

311327
private static class NoRecordsException extends IOException {

‎src/main/java/com/arpnetworking/clusteraggregator/http/Routes.java‎

Lines changed: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -161,7 +161,9 @@ private CompletionStage<HttpResponse> process(final HttpRequest request) {
161161
});
162162
}
163163
} else if (HttpMethods.POST.equals(request.method())) {
164-
if (INCOMING_DATA_V1_PATH.equals(request.getUri().path())) {
164+
if (INCOMING_DATA_V1_PATH.equals(request.getUri().path())
165+
|| INCOMING_DATA_PERSIST_V1_PATH.equals(request.getUri().path())
166+
|| INCOMING_DATA_REAGGREGATE_V1_PATH.equals(request.getUri().path())) {
165167
return ask("/user/http-ingest-v1", request, HttpResponse.create().withStatus(500));
166168
}
167169
}
@@ -210,15 +212,13 @@ private String createMetricName(final HttpRequest request, final String actionPa
210212

211213
private static final Logger LOGGER = LoggerFactory.getLogger(Routes.class);
212214

213-
// Ping
214215
private static final HttpHeader PING_CACHE_CONTROL_HEADER = CacheControl.create(
215216
CacheDirectives.PRIVATE(),
216217
CacheDirectives.NO_CACHE,
217218
CacheDirectives.NO_STORE,
218219
CacheDirectives.MUST_REVALIDATE);
219220
private static final String UNHEALTHY_STATE = "UNHEALTHY";
220221
private static final String HEALTHY_STATE = "HEALTHY";
221-
private static final String INCOMING_DATA_V1_PATH = "/metrics/v1/data";
222222
private static final String REST_SERVICE_METRIC_ROOT = "rest_service/";
223223
private static final String BODY_SIZE_METRIC = "body_size";
224224
private static final String REQUEST_METRIC = "request";
@@ -228,6 +228,29 @@ private String createMetricName(final HttpRequest request, final String actionPa
228228
private static final ContentType JSON_CONTENT_TYPE = ContentTypes.APPLICATION_JSON;
229229

230230
private static final long serialVersionUID = -1573473630801540757L;
231+
232+
/**
233+
* The legacy path for incoming data over HTTP. The reaggregation behavior
234+
* on this path is controlled by the the {@code calculateClusterAggregations}
235+
* value. If this is set to {@code False} this endpoint behaves with
236+
* {@link com.arpnetworking.clusteraggregator.models.AggregationMode} set to
237+
* {@code PERSIST}. If the configuration key is set to {@code True} this endpoint
238+
* behaves with {@link com.arpnetworking.clusteraggregator.models.AggregationMode}
239+
* set to {@code PERSIST_AND_REAGGREGATE}.
240+
*/
241+
public static final String INCOMING_DATA_V1_PATH = "/metrics/v1/data";
242+
243+
/**
244+
* This HTTP endpoint is for only persisting data directly and through the
245+
* host emitter.
246+
*/
247+
public static final String INCOMING_DATA_PERSIST_V1_PATH = "/metrics/v1/data/persist";
248+
249+
/**
250+
* This HTTP endpoint is for only reaggregating data and then persisting
251+
* through the cluster emitter.
252+
*/
253+
public static final String INCOMING_DATA_REAGGREGATE_V1_PATH = "/metrics/v1/data/reaggregate";
231254

232255
private static class MemberSerializer extends JsonSerializer<Member> {
233256
@Override

‎src/main/java/com/arpnetworking/clusteraggregator/models/AggregationMode.java‎

Lines changed: 68 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
1+
/*
2+
* Copyright 2018 Dropbox
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.arpnetworking.clusteraggregator.models;
17+
18+
/**
19+
* The aggregation mode applied to data.
20+
*
21+
* @author Ville Koskela (vkoskela at dropbox dot com)
22+
*/
23+
public enum AggregationMode {
24+
25+
/**
26+
* Persist the statistics data as provided by the client.
27+
*/
28+
PERSIST(true, false),
29+
30+
/**
31+
* Reaggregate the statistics data provided by the client and persist only
32+
* the reaggregated data.
33+
*/
34+
REAGGREGATE(false, true),
35+
36+
/**
37+
* Persist the statistics data as provided by the client <b>AND</b> reaggregate
38+
* the statistics data provided by the client persisting the reaggregated data
39+
* as well.
40+
*/
41+
PERSIST_AND_REAGGREGATE(true, true);
42+
43+
/**
44+
* Whether this {@link AggregationMode} should persist client statistics.
45+
*
46+
* @return True if and only if client statistics should be persisted
47+
*/
48+
public boolean shouldPersist() {
49+
return _shouldPersist;
50+
}
51+
52+
/**
53+
* Whether this {@link AggregationMode} should reaggregate client statistics.
54+
*
55+
* @return True if and only if client statistics should be reaggregated
56+
*/
57+
public boolean shouldReaggregate() {
58+
return _shouldReaggregate;
59+
}
60+
61+
AggregationMode(final boolean shouldPersist, final boolean shouldReaggregate) {
62+
_shouldPersist = shouldPersist;
63+
_shouldReaggregate = shouldReaggregate;
64+
}
65+
66+
private final boolean _shouldPersist;
67+
private final boolean _shouldReaggregate;
68+
}

‎src/main/java/com/arpnetworking/clusteraggregator/models/CombinedMetricData.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -207,9 +207,9 @@ public static Builder fromStatisticSetRecord(final Messages.StatisticSetRecord r
207207

208208
final HistogramStatistic.HistogramSupportingData histogramSupportingData =
209209
new HistogramStatistic.HistogramSupportingData.Builder()
210-
.setHistogramSnapshot(histogram.getSnapshot())
211-
.setUnit(getUnitFromName(supportingData.getUnit()))
212-
.build();
210+
.setHistogramSnapshot(histogram.getSnapshot())
211+
.setUnit(getUnitFromName(supportingData.getUnit()))
212+
.build();
213213

214214
calculatedValueBuilder = new CalculatedValue.Builder<HistogramStatistic.HistogramSupportingData>()
215215
.setData(histogramSupportingData);

‎src/main/java/com/arpnetworking/clusteraggregator/models/HttpRequest.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,11 @@
2525
* @author Brandon Arp (brandon dot arp at smartsheet dot com)
2626
*/
2727
public final class HttpRequest {
28+
29+
public String getPath() {
30+
return _path;
31+
}
32+
2833
public Multimap<String, String> getHeaders() {
2934
return _headers;
3035
}
@@ -36,14 +41,20 @@ public ByteString getBody() {
3641
/**
3742
* Public constructor.
3843
*
44+
* @param path The path.
3945
* @param headers The headers.
4046
* @param body The body of the request.
4147
*/
42-
public HttpRequest(final ImmutableMultimap<String, String> headers, final ByteString body) {
48+
public HttpRequest(
49+
final String path,
50+
final ImmutableMultimap<String, String> headers,
51+
final ByteString body) {
52+
_path = path;
4353
_headers = headers;
4454
_body = body;
4555
}
4656

57+
private final String _path;
4758
private final ImmutableMultimap<String, String> _headers;
4859
private final ByteString _body;
4960
}

‎src/main/java/com/arpnetworking/tsdcore/model/AggregationMessage.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,7 @@ public static Optional<AggregationMessage> deserialize(final ByteString data) {
121121
}
122122
} catch (final InvalidProtocolBufferException e) {
123123
LOGGER.warn(
124-
String.format("Invalid protocol buffer; type=%s bytes=%s", type, Hex.encodeHexString(payloadBytes)), e);
124+
String.format("Invalid protocol buffer; type=%s bytes=%s", type, Hex.encodeHexString(payloadBytes)), e);
125125
return Optional.empty();
126126
}
127127
}
@@ -190,5 +190,4 @@ private AggregationMessage(final GeneratedMessageV3 message, final int length) {
190190
private static final int INTEGER_SIZE_IN_BYTES = Integer.SIZE / 8;
191191
private static final int HEADER_SIZE_IN_BYTES = INTEGER_SIZE_IN_BYTES + 1;
192192
private static final Logger LOGGER = LoggerFactory.getLogger(AggregationMessage.class);
193-
194193
}

0 commit comments

Comments
 (0)