Skip to content

Commit ead941d

Browse files
vjkoskelaBrandonArp
vjkoskela
authored and
BrandonArp
committed
Reduce unnecessary MetricReport instance creations. (#74)
1 parent 655f627 commit ead941d

File tree

6 files changed

+149
-172
lines changed

6 files changed

+149
-172
lines changed

‎src/main/java/com/arpnetworking/metrics/proxy/actors/Connection.java‎

Lines changed: 113 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,17 +26,25 @@
2626
import com.arpnetworking.metrics.incubator.PeriodicMetrics;
2727
import com.arpnetworking.metrics.proxy.models.messages.Command;
2828
import com.arpnetworking.metrics.proxy.models.messages.Connect;
29+
import com.arpnetworking.metrics.proxy.models.messages.MetricReport;
2930
import com.arpnetworking.metrics.proxy.models.protocol.MessageProcessorsFactory;
3031
import com.arpnetworking.metrics.proxy.models.protocol.MessagesProcessor;
3132
import com.arpnetworking.steno.LogValueMapFactory;
3233
import com.arpnetworking.steno.Logger;
3334
import com.arpnetworking.steno.LoggerFactory;
35+
import com.arpnetworking.tsdcore.model.AggregatedData;
36+
import com.arpnetworking.tsdcore.model.Key;
37+
import com.arpnetworking.tsdcore.model.PeriodicData;
3438
import com.fasterxml.jackson.core.JsonProcessingException;
3539
import com.fasterxml.jackson.databind.ObjectMapper;
3640
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
3741
import com.fasterxml.jackson.databind.node.ObjectNode;
42+
import com.google.common.collect.Maps;
43+
import com.google.common.collect.Sets;
3844

3945
import java.util.List;
46+
import java.util.Map;
47+
import java.util.Set;
4048

4149
/**
4250
* Actor class to hold the state for a single connection.
@@ -96,6 +104,9 @@ public Receive createReceive() {
96104
.log();
97105
getSelf().tell(PoisonPill.getInstance(), getSelf());
98106
})
107+
.match(PeriodicData.class, message -> {
108+
processPeriodicData(message);
109+
})
99110
.matchAny(message -> {
100111
if (_channel == null) {
101112
LOGGER.warn()
@@ -179,6 +190,52 @@ public void sendCommand(final String command, final ObjectNode data) {
179190
send(message);
180191
}
181192

193+
/**
194+
* Subscribe the connection to a stream.
195+
*
196+
* @param service The service to subscribe to.
197+
* @param metric The metric to subscribe to.
198+
* @param statistic The statistic to subscribe to.
199+
*/
200+
public void subscribe(final String service, final String metric, final String statistic) {
201+
if (!_subscriptions.containsKey(service)) {
202+
_subscriptions.put(service, Maps.newHashMap());
203+
}
204+
205+
final Map<String, Set<String>> metrics = _subscriptions.get(service);
206+
if (!metrics.containsKey(metric)) {
207+
metrics.put(metric, Sets.newHashSet());
208+
}
209+
210+
final Set<String> statistics = metrics.get(metric);
211+
if (!statistics.contains(statistic)) {
212+
statistics.add(statistic);
213+
}
214+
}
215+
216+
/**
217+
* Unsubscribe the connection from a stream.
218+
*
219+
* @param service The service to unsubscribe from.
220+
* @param metric The metric to unsubscribe from.
221+
* @param statistic The statistic to unsubscribe from.
222+
*/
223+
public void unsubscribe(final String service, final String metric, final String statistic) {
224+
if (!_subscriptions.containsKey(service)) {
225+
return;
226+
}
227+
228+
final Map<String, Set<String>> metrics = _subscriptions.get(service);
229+
if (!metrics.containsKey(metric)) {
230+
return;
231+
}
232+
233+
final Set<String> statistics = metrics.get(metric);
234+
if (statistics.contains(statistic)) {
235+
statistics.remove(statistic);
236+
}
237+
}
238+
182239
/**
183240
* Accessor to this Connection's Telemetry actor.
184241
*
@@ -198,6 +255,7 @@ public Object toLogValue() {
198255
return LogValueMapFactory.builder(this)
199256
.put("connection", _channel)
200257
.put("messageProcessors", _messageProcessors)
258+
.put("subscriptions", _subscriptions)
201259
.build();
202260
}
203261

@@ -206,11 +264,66 @@ public String toString() {
206264
return toLogValue().toString();
207265
}
208266

267+
private void processPeriodicData(final PeriodicData message) {
268+
final Key dimensions = message.getDimensions();
269+
final String host = dimensions.getHost();
270+
final String service = dimensions.getService();
271+
final Map<String, Set<String>> metrics = _subscriptions.get(service);
272+
if (metrics == null) {
273+
LOGGER.trace()
274+
.setMessage("Not sending MetricReport")
275+
.addData("reason", "service not found in subscriptions")
276+
.addData("service", service)
277+
.log();
278+
return;
279+
}
280+
281+
for (final Map.Entry<String, AggregatedData> entry : message.getData().entries()) {
282+
final String metric = entry.getKey();
283+
final AggregatedData datum = entry.getValue();
284+
285+
final Set<String> stats = metrics.get(metric);
286+
if (stats == null) {
287+
LOGGER.trace()
288+
.setMessage("Not sending MetricReport")
289+
.addData("reason", "metric not found in subscriptions")
290+
.addData("metric", metric)
291+
.log();
292+
continue;
293+
}
294+
295+
final String statisticName = datum.getStatistic().getName();
296+
if (!stats.contains(statisticName)) {
297+
LOGGER.trace()
298+
.setMessage("Not sending MetricReport")
299+
.addData("reason", "statistic not found in subscriptions")
300+
.addData("statistic", statisticName)
301+
.log();
302+
continue;
303+
}
304+
305+
final MetricReport metricReport = new MetricReport(
306+
service,
307+
host,
308+
statisticName,
309+
metric,
310+
datum.getValue().getValue(),
311+
datum.getValue().getUnit(),
312+
message.getStart());
313+
for (final MessagesProcessor messagesProcessor : _messageProcessors) {
314+
if (messagesProcessor.handleMessage(metricReport)) {
315+
break;
316+
}
317+
}
318+
}
319+
}
320+
209321
private ActorRef _telemetry;
210322
private ActorRef _channel;
211323

212324
private final PeriodicMetrics _metrics;
213325
private final List<MessagesProcessor> _messageProcessors;
326+
private final Map<String, Map<String, Set<String>>> _subscriptions = Maps.newHashMap();
214327

215328
private static final String METRICS_PREFIX = "actors/connection/";
216329
private static final String UNKNOWN_COMMAND_COUNTER = METRICS_PREFIX + "command/UNKNOWN";

‎src/main/java/com/arpnetworking/metrics/proxy/actors/Telemetry.java‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -29,14 +29,16 @@
2929
import com.arpnetworking.metrics.proxy.models.messages.LogLine;
3030
import com.arpnetworking.metrics.proxy.models.messages.LogsList;
3131
import com.arpnetworking.metrics.proxy.models.messages.LogsListRequest;
32-
import com.arpnetworking.metrics.proxy.models.messages.MetricReport;
3332
import com.arpnetworking.metrics.proxy.models.messages.MetricsList;
3433
import com.arpnetworking.metrics.proxy.models.messages.MetricsListRequest;
3534
import com.arpnetworking.metrics.proxy.models.messages.NewLog;
3635
import com.arpnetworking.metrics.proxy.models.messages.NewMetric;
3736
import com.arpnetworking.steno.LogValueMapFactory;
3837
import com.arpnetworking.steno.Logger;
3938
import com.arpnetworking.steno.LoggerFactory;
39+
import com.arpnetworking.tsdcore.model.AggregatedData;
40+
import com.arpnetworking.tsdcore.model.Key;
41+
import com.arpnetworking.tsdcore.model.PeriodicData;
4042
import com.google.common.collect.ImmutableMap;
4143
import com.google.common.collect.Maps;
4244
import com.google.common.collect.Sets;
@@ -79,8 +81,8 @@ public Telemetry(final MetricsFactory metricsFactory) {
7981
public Receive createReceive() {
8082
return receiveBuilder()
8183
.matchEquals("instrument", message -> periodicInstrumentation())
84+
.match(PeriodicData.class, this::executePeriodicData)
8285
.match(Connect.class, this::executeConnect)
83-
.match(MetricReport.class, this::executeMetricReport)
8486
.match(LogLine.class, this::executeLogLine)
8587
.match(MetricsListRequest.class, ignored -> executeMetricsListRequest())
8688
.match(LogsListRequest.class, ignored -> executeLogsListRequest())
@@ -165,13 +167,16 @@ private void executeConnect(final Connect message) {
165167
.log();
166168
}
167169

168-
private void executeMetricReport(final MetricReport message) {
169-
_metrics.incrementCounter(METRIC_REPORT_COUNTER);
170+
private void executePeriodicData(final PeriodicData message) {
171+
_metrics.incrementCounter(PERIODIC_DATA_COUNTER);
170172

171-
// Ensure the metric is in the registry
172-
registerMetric(message.getService(), message.getMetric(), message.getStatistic());
173+
// Ensure all the metrics are in the registry
174+
final Key dimensions = message.getDimensions();
175+
for (final Map.Entry<String, AggregatedData> entry : message.getData().entries()) {
176+
registerMetric(dimensions.getService(), entry.getKey(), entry.getValue().getStatistic().getName());
177+
}
173178

174-
// Transmit the report to all members
179+
// Transmit the data to all members
175180
broadcast(message);
176181
}
177182

@@ -208,12 +213,12 @@ private void broadcast(final Object message) {
208213

209214
private void registerMetric(final String service, final String metric, final String statistic) {
210215
if (!_serviceMetrics.containsKey(service)) {
211-
_serviceMetrics.put(service, Maps.<String, Set<String>>newHashMap());
216+
_serviceMetrics.put(service, Maps.newHashMap());
212217
}
213218
final Map<String, Set<String>> serviceMap = _serviceMetrics.get(service);
214219

215220
if (!serviceMap.containsKey(metric)) {
216-
serviceMap.put(metric, Sets.<String>newHashSet());
221+
serviceMap.put(metric, Sets.newHashSet());
217222
}
218223
final Set<String> statistics = serviceMap.get(metric);
219224

@@ -245,7 +250,7 @@ private void periodicInstrumentation() {
245250
private static final String METRIC_PREFIX = "actors/stream/";
246251
private static final String METRICS_LIST_REQUEST = METRIC_PREFIX + "metrics_list_request";
247252
private static final String QUIT_COUNTER = METRIC_PREFIX + "quit";
248-
private static final String METRIC_REPORT_COUNTER = METRIC_PREFIX + "metric_report";
253+
private static final String PERIODIC_DATA_COUNTER = METRIC_PREFIX + "periodic_data";
249254
private static final String CONNECT_COUNTER = METRIC_PREFIX + "connect";
250255
private static final String LOG_LINE_COUNTER = METRIC_PREFIX + "log_line";
251256
private static final String METRICS_LIST_COUNTER = METRIC_PREFIX + "metrics_list";

‎src/main/java/com/arpnetworking/metrics/proxy/models/protocol/v1/MetricMessagesProcessor.java‎

Lines changed: 3 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -26,14 +26,10 @@
2626
import com.arpnetworking.metrics.proxy.models.messages.NewMetric;
2727
import com.arpnetworking.metrics.proxy.models.protocol.MessagesProcessor;
2828
import com.arpnetworking.steno.LogValueMapFactory;
29-
import com.arpnetworking.steno.Logger;
30-
import com.arpnetworking.steno.LoggerFactory;
3129
import com.fasterxml.jackson.databind.ObjectMapper;
3230
import com.fasterxml.jackson.databind.node.ArrayNode;
3331
import com.fasterxml.jackson.databind.node.JsonNodeFactory;
3432
import com.fasterxml.jackson.databind.node.ObjectNode;
35-
import com.google.common.collect.Maps;
36-
import com.google.common.collect.Sets;
3733

3834
import java.util.Map;
3935
import java.util.Set;
@@ -72,15 +68,15 @@ public boolean handleMessage(final Object message) {
7268
final String service = commandNode.get("service").asText();
7369
final String metric = commandNode.get("metric").asText();
7470
final String statistic = commandNode.get("statistic").asText();
75-
subscribe(service, metric, statistic);
71+
_connection.subscribe(service, metric, statistic);
7672
break;
7773
}
7874
case COMMAND_UNSUBSCRIBE_METRIC: {
7975
_metrics.recordCounter(UNSUBSCRIBE_COUNTER, 1);
8076
final String service = commandNode.get("service").asText();
8177
final String metric = commandNode.get("metric").asText();
8278
final String statistic = commandNode.get("statistic").asText();
83-
unsubscribe(service, metric, statistic);
79+
_connection.unsubscribe(service, metric, statistic);
8480
break;
8581
}
8682
default:
@@ -114,7 +110,7 @@ public boolean handleMessage(final Object message) {
114110
public Object toLogValue() {
115111
// NOTE: Do not log connection context as this creates a circular reference
116112
return LogValueMapFactory.builder(this)
117-
.put("subscriptions", _subscriptions)
113+
.put("connection", _connection)
118114
.build();
119115
}
120116

@@ -132,33 +128,6 @@ private void processNewMetric(final NewMetric newMetric) {
132128
}
133129

134130
private void processMetricReport(final MetricReport report) {
135-
final Map<String, Set<String>> metrics = _subscriptions.get(report.getService());
136-
if (metrics == null) {
137-
LOGGER.trace()
138-
.setMessage("Not sending MetricReport")
139-
.addData("reason", "service not found in subscriptions")
140-
.addData("service", report.getService())
141-
.log();
142-
return;
143-
}
144-
final Set<String> stats = metrics.get(report.getMetric());
145-
if (stats == null) {
146-
LOGGER.trace()
147-
.setMessage("Not sending MetricReport")
148-
.addData("reason", "metric not found in subscriptions")
149-
.addData("metric", report.getMetric())
150-
.log();
151-
return;
152-
}
153-
if (!stats.contains(report.getStatistic())) {
154-
LOGGER.trace()
155-
.setMessage("Not sending MetricReport")
156-
.addData("reason", "statistic not found in subscriptions")
157-
.addData("statistic", report.getStatistic())
158-
.log();
159-
return;
160-
}
161-
162131
//TODO(barp): Map with a POJO mapper [MAI-184]
163132
final ObjectNode event = new ObjectNode(OBJECT_MAPPER.getNodeFactory());
164133
event.put("server", report.getHost());
@@ -199,41 +168,6 @@ private void processMetricsList(final MetricsList metricsList) {
199168
_connection.sendCommand(COMMAND_METRICS_LIST, dataNode);
200169
}
201170

202-
private void subscribe(final String service, final String metric, final String statistic) {
203-
if (!_subscriptions.containsKey(service)) {
204-
_subscriptions.put(service, Maps.<String, Set<String>>newHashMap());
205-
}
206-
207-
final Map<String, Set<String>> metrics = _subscriptions.get(service);
208-
209-
if (!metrics.containsKey(metric)) {
210-
metrics.put(metric, Sets.<String>newHashSet());
211-
}
212-
213-
final Set<String> statistics = metrics.get(metric);
214-
215-
if (!statistics.contains(statistic)) {
216-
statistics.add(statistic);
217-
}
218-
}
219-
220-
private void unsubscribe(final String service, final String metric, final String statistic) {
221-
if (!_subscriptions.containsKey(service)) {
222-
return;
223-
}
224-
225-
final Map<String, Set<String>> metrics = _subscriptions.get(service);
226-
if (!metrics.containsKey(metric)) {
227-
return;
228-
}
229-
230-
final Set<String> statistics = metrics.get(metric);
231-
if (statistics.contains(statistic)) {
232-
statistics.remove(statistic);
233-
}
234-
}
235-
236-
private final Map<String, Map<String, Set<String>>> _subscriptions = Maps.newHashMap();
237171
private final Connection _connection;
238172
private PeriodicMetrics _metrics;
239173

@@ -253,5 +187,4 @@ private void unsubscribe(final String service, final String metric, final String
253187
private static final String GET_METRICS_COUNTER = METRICS_PREFIX + "command/get_metrics";
254188

255189
private static final ObjectMapper OBJECT_MAPPER = ObjectMapperFactory.getInstance();
256-
private static final Logger LOGGER = LoggerFactory.getLogger(MetricMessagesProcessor.class);
257190
}

0 commit comments

Comments
 (0)