Skip to content

Commit 0f17b68

Browse files
BrandonArpBrandonArp
committed
extract and inject tags
1 parent e44ffa2 commit 0f17b68

File tree

4 files changed

+136
-34
lines changed

4 files changed

+136
-34
lines changed

‎src/main/java/com/arpnetworking/metrics/mad/sources/MappingSource.java‎

Lines changed: 33 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import com.arpnetworking.steno.LoggerFactory;
3131
import com.arpnetworking.tsdcore.model.MetricType;
3232
import com.arpnetworking.tsdcore.model.Quantity;
33+
import com.arpnetworking.utility.RegexAndMapReplacer;
3334
import com.google.common.base.MoreObjects;
3435
import com.google.common.base.Splitter;
3536
import com.google.common.collect.ImmutableList;
@@ -41,7 +42,6 @@
4142
import java.util.Map;
4243
import java.util.regex.Matcher;
4344
import java.util.regex.Pattern;
44-
import java.util.stream.Collectors;
4545

4646
/**
4747
* Implementation of <code>Source</code> which wraps another <code>Source</code>
@@ -119,67 +119,68 @@ public void notify(final Observable observable, final Object event) {
119119

120120
// Merge the metrics in the record together
121121
final Record record = (Record) event;
122-
final Map<String, MergingMetric> mergedMetrics = Maps.newHashMap();
122+
final Map<ImmutableMap<String, String>, Map<String, MergingMetric>> mergedMetrics = Maps.newHashMap();
123123
for (final Map.Entry<String, ? extends Metric> metric : record.getMetrics().entrySet()) {
124124
boolean found = false;
125125
for (final Map.Entry<Pattern, List<String>> findAndReplace : _findAndReplace.entrySet()) {
126126
final Matcher matcher = findAndReplace.getKey().matcher(metric.getKey());
127127
if (matcher.find()) {
128128
for (final String replacement : findAndReplace.getValue()) {
129-
final String replacedString = matcher.replaceAll(replacement);
129+
final String replacedString =
130+
RegexAndMapReplacer.replaceAll(matcher, metric.getKey(), replacement, record.getDimensions());
130131

131132
final int tagsStart = replacedString.indexOf(';');
132133
if (tagsStart == -1) {
133134
// We just have a metric name. Optimize for this common case
134-
merge(metric.getValue(), replacedString, mergedMetrics);
135+
merge(metric.getValue(), replacedString, mergedMetrics, record.getDimensions());
135136
} else {
137+
final String metricName = replacedString.substring(0, tagsStart);
136138
final Map<String, String> parsedTags = TAG_SPLITTER.split(replacedString.substring(tagsStart + 1));
137-
final Map<String, String> finalTags = Maps.newTreeMap();
139+
final ImmutableMap.Builder<String, String> finalTags = ImmutableMap.builder();
138140
finalTags.putAll(record.getDimensions());
139141
finalTags.putAll(parsedTags);
140-
final StringBuilder keyBuilder = new StringBuilder();
141-
keyBuilder.append(replacedString.substring(0, tagsStart + 1));
142-
keyBuilder.append(
143-
finalTags.entrySet()
144-
.stream()
145-
.map(entry -> String.format("%s=%s", entry.getKey(), entry.getValue()))
146-
.collect(Collectors.joining(";")));
147-
148-
merge(metric.getValue(), keyBuilder.toString(), mergedMetrics);
142+
143+
merge(metric.getValue(), metricName, mergedMetrics, finalTags.build());
149144
}
150145
}
151146
//Having "found" set here means that mapping a metric to an empty list suppresses that metric
152147
found = true;
153148
}
154149
}
155150
if (!found) {
156-
merge(metric.getValue(), metric.getKey(), mergedMetrics);
151+
merge(metric.getValue(), metric.getKey(), mergedMetrics, record.getDimensions());
157152
}
158153
}
159154

160155
// Raise the merged record event with this source's observers
161156
// NOTE: Do not leak instances of MergingMetric since it is mutable
162-
_source.notify(
163-
ThreadLocalBuilder.build(
164-
DefaultRecord.Builder.class,
165-
b1 -> b1.setMetrics(
166-
mergedMetrics.entrySet().stream().collect(
167-
ImmutableMap.toImmutableMap(
168-
Map.Entry::getKey,
169-
e -> ThreadLocalBuilder.clone(
170-
e.getValue(),
171-
DefaultMetric.Builder.class))))
172-
.setId(record.getId())
173-
.setTime(record.getTime())
174-
.setAnnotations(record.getAnnotations())
175-
.setDimensions(record.getDimensions())));
157+
for (Map.Entry<ImmutableMap<String, String>, Map<String, MergingMetric>> entry : mergedMetrics.entrySet()) {
158+
_source.notify(
159+
ThreadLocalBuilder.build(
160+
DefaultRecord.Builder.class,
161+
b1 -> b1.setMetrics(
162+
entry.getValue().entrySet().stream().collect(
163+
ImmutableMap.toImmutableMap(
164+
Map.Entry::getKey,
165+
e -> ThreadLocalBuilder.clone(
166+
e.getValue(),
167+
DefaultMetric.Builder.class))))
168+
.setId(record.getId())
169+
.setTime(record.getTime())
170+
.setAnnotations(record.getAnnotations())
171+
.setDimensions(entry.getKey())));
172+
}
176173
}
177174

178-
private void merge(final Metric metric, final String key, final Map<String, MergingMetric> mergedMetrics) {
179-
final MergingMetric mergedMetric = mergedMetrics.get(key);
175+
private void merge(final Metric metric, final String key,
176+
final Map<ImmutableMap<String, String>, Map<String, MergingMetric>> mergedMetrics,
177+
final ImmutableMap<String, String> dimensions) {
178+
179+
final Map<String, MergingMetric> mergedMetricsForDimensions = mergedMetrics.computeIfAbsent(dimensions, k -> Maps.newHashMap());
180+
final MergingMetric mergedMetric = mergedMetricsForDimensions.get(key);
180181
if (mergedMetric == null) {
181182
// This is the first time this metric is being merged into
182-
mergedMetrics.put(key, new MergingMetric(metric));
183+
mergedMetricsForDimensions.put(key, new MergingMetric(metric));
183184
} else if (!mergedMetric.isMergable(metric)) {
184185
// This instance of the metric is not mergable with previous
185186
LOGGER.error()

‎src/main/java/com/arpnetworking/utility/RegexAndMapReplacer.java‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -135,6 +135,13 @@ private static int writeReplacementToken(final String replacement, final int off
135135
}
136136
x++;
137137
}
138+
if (tokenBuilder.length() == 0) {
139+
throw new IllegalArgumentException(
140+
String.format(
141+
"Invalid replacement token, non-numeric tokens must be surrounded by { } at col %d: %s",
142+
x,
143+
replacement));
144+
}
138145
output.append(getReplacement(matcher, tokenBuilder.toString(), true, variables));
139146
}
140147
return x - offset - 1;

‎src/test/java/com/arpnetworking/metrics/mad/sources/MappingSourceTest.java‎

Lines changed: 94 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import com.arpnetworking.metrics.mad.sources.MappingSource.MergingMetric;
2323
import com.arpnetworking.test.TestBeanFactory;
2424
import com.arpnetworking.test.UnorderedRecordEquality;
25+
import com.arpnetworking.tsdcore.model.Key;
2526
import com.arpnetworking.tsdcore.model.MetricType;
2627
import com.arpnetworking.tsdcore.model.Quantity;
2728
import com.arpnetworking.tsdcore.model.Unit;
@@ -54,7 +55,7 @@ public void setUp() {
5455
"foo/([^/]*)/bar", ImmutableList.of("foo/bar"),
5556
"cat/([^/]*)/dog", ImmutableList.of("cat/dog", "cat/dog/$1"),
5657
"tagged/([^/]*)/dog", ImmutableList.of("tagged/dog;animal=$1"),
57-
"tagged/([^/]*)/animal", ImmutableList.of("tagged/$animal/animal")))
58+
"tagged/([^/]*)/animal", ImmutableList.of("tagged/${animal}/animal")))
5859
.setSource(_mockSource);
5960
}
6061

@@ -280,6 +281,98 @@ public void testReplaceWithCapture() {
280281
UnorderedRecordEquality.equals(expectedRecord, actualRecord));
281282
}
282283

284+
@Test
285+
public void testExtractTagWithCapture() {
286+
final Record matchingRecord = TestBeanFactory.createRecordBuilder()
287+
.setMetrics(ImmutableMap.of(
288+
"tagged/sheep/dog",
289+
TestBeanFactory.createMetricBuilder()
290+
.setType(MetricType.GAUGE)
291+
.setValues(ImmutableList.of(
292+
new Quantity.Builder()
293+
.setValue(1.23d)
294+
.setUnit(Unit.BYTE)
295+
.build()))
296+
.build()))
297+
.build();
298+
299+
final Source mergingSource = _mappingSourceBuilder.build();
300+
mergingSource.attach(_mockObserver);
301+
notify(_mockSource, matchingRecord);
302+
303+
final ArgumentCaptor<Record> argument = ArgumentCaptor.forClass(Record.class);
304+
Mockito.verify(_mockObserver).notify(Mockito.same(mergingSource), argument.capture());
305+
final Record actualRecord = argument.getValue();
306+
307+
final Record expectedRecord = TestBeanFactory.createRecordBuilder()
308+
.setAnnotations(matchingRecord.getAnnotations())
309+
.setTime(matchingRecord.getTime())
310+
.setDimensions(ImmutableMap.<String, String>builder().putAll(matchingRecord.getDimensions()).put("animal", "sheep").build())
311+
.setMetrics(ImmutableMap.of(
312+
"tagged/dog",
313+
TestBeanFactory.createMetricBuilder()
314+
.setType(MetricType.GAUGE)
315+
.setValues(ImmutableList.of(
316+
new Quantity.Builder()
317+
.setValue(1.23d)
318+
.setUnit(Unit.BYTE)
319+
.build()))
320+
.build()))
321+
.build();
322+
Assert.assertTrue(
323+
String.format("expected=%s, actual=%s", expectedRecord, actualRecord),
324+
UnorderedRecordEquality.equals(expectedRecord, actualRecord));
325+
}
326+
@Test
327+
328+
public void testInjectTagWithCapture() {
329+
final Record matchingRecord = TestBeanFactory.createRecordBuilder()
330+
.setMetrics(ImmutableMap.of(
331+
"tagged/foo/animal",
332+
TestBeanFactory.createMetricBuilder()
333+
.setType(MetricType.GAUGE)
334+
.setValues(ImmutableList.of(
335+
new Quantity.Builder()
336+
.setValue(1.23d)
337+
.setUnit(Unit.BYTE)
338+
.build()))
339+
.build()))
340+
.setDimensions(
341+
ImmutableMap.of(
342+
Key.HOST_DIMENSION_KEY, "MyHost",
343+
Key.SERVICE_DIMENSION_KEY, "MyService",
344+
Key.CLUSTER_DIMENSION_KEY, "MyCluster",
345+
"animal", "frog"))
346+
.build();
347+
348+
final Source mergingSource = _mappingSourceBuilder.build();
349+
mergingSource.attach(_mockObserver);
350+
notify(_mockSource, matchingRecord);
351+
352+
final ArgumentCaptor<Record> argument = ArgumentCaptor.forClass(Record.class);
353+
Mockito.verify(_mockObserver).notify(Mockito.same(mergingSource), argument.capture());
354+
final Record actualRecord = argument.getValue();
355+
356+
final Record expectedRecord = TestBeanFactory.createRecordBuilder()
357+
.setAnnotations(matchingRecord.getAnnotations())
358+
.setTime(matchingRecord.getTime())
359+
.setDimensions(matchingRecord.getDimensions())
360+
.setMetrics(ImmutableMap.of(
361+
"tagged/frog/animal",
362+
TestBeanFactory.createMetricBuilder()
363+
.setType(MetricType.GAUGE)
364+
.setValues(ImmutableList.of(
365+
new Quantity.Builder()
366+
.setValue(1.23d)
367+
.setUnit(Unit.BYTE)
368+
.build()))
369+
.build()))
370+
.build();
371+
Assert.assertTrue(
372+
String.format("expected=%s, actual=%s", expectedRecord, actualRecord),
373+
UnorderedRecordEquality.equals(expectedRecord, actualRecord));
374+
}
375+
283376
@Test
284377
public void testReplaceWithCaptureWithTags() {
285378
final Record matchingRecord = TestBeanFactory.createRecordBuilder()

‎src/test/java/com/arpnetworking/test/UnorderedRecordEquality.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,8 @@ public final class UnorderedRecordEquality {
4242
*/
4343
public static boolean equals(final Record r1, final Record r2) {
4444
if (!r1.getTime().equals(r2.getTime())
45-
|| !r1.getAnnotations().equals(r2.getAnnotations())) {
45+
|| !r1.getAnnotations().equals(r2.getAnnotations())
46+
|| !r1.getDimensions().equals(r2.getDimensions())) {
4647
return false;
4748
}
4849

0 commit comments

Comments
 (0)