Skip to content

Commit 6def901

Browse files
Heng CuiHeng Cui
Heng Cui
authored and
Heng Cui
committed
update normalization counter bound
1 parent dd3e387 commit 6def901

17 files changed

Lines changed: 474 additions & 57 deletions

File tree

‎candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessor.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@ public class CounterNormalizationProcessor
4949
private final long counterWrapAroundTimeMs;
5050
private final BigInteger maxUnsignedInt = new BigInteger(Integer.toUnsignedString(-1));
5151
private final BigInteger maxUnsignedLong = new BigInteger(Long.toUnsignedString(-1L));
52+
private final BigInteger maxSignedLong = new BigInteger(String.valueOf(Long.MAX_VALUE));
5253

5354
private final Timer timer =
5455
Timer.builder("json_streams_old_counter_cleaner_duration")
@@ -213,7 +214,12 @@ private void normalizeCounter(
213214
inWrapAroundRange =
214215
maxDiff.intValue() >= 0 && maxDiff.intValue() < intCounterWrapAroundLimit;
215216
} else if (counterConfig.getCounterType() == NormalizeCounterConfig.CounterType.U64) {
216-
maxDiff = maxUnsignedLong.subtract(savedCounterValueBigInt).add(counterValue);
217+
BigInteger upperLimit =
218+
(savedCounterValueBigInt.compareTo(maxSignedLong) > 0)
219+
? maxUnsignedLong
220+
: maxSignedLong;
221+
222+
maxDiff = upperLimit.subtract(savedCounterValueBigInt).add(counterValue);
217223
inWrapAroundRange =
218224
maxDiff.longValue() >= 0 && maxDiff.longValue() < longCounterWrapAroundLimit;
219225
} else {

‎candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessorTest.java‎

Lines changed: 80 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
package com.swisscom.daisy.cosmos.candyfloss.processors;
22

3-
import static org.junit.jupiter.api.Assertions.*;
3+
import static org.junit.jupiter.api.Assertions.assertEquals;
44

55
import com.fasterxml.jackson.databind.DeserializationFeature;
66
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -14,7 +14,12 @@
1414
import com.typesafe.config.ConfigFactory;
1515
import java.io.IOException;
1616
import java.nio.charset.StandardCharsets;
17+
import java.nio.file.*;
18+
import java.nio.file.attribute.BasicFileAttributes;
1719
import java.time.Duration;
20+
import java.util.ArrayList;
21+
import java.util.Comparator;
22+
import java.util.List;
1823
import org.apache.kafka.common.serialization.Serde;
1924
import org.apache.kafka.common.serialization.Serdes;
2025
import org.apache.kafka.common.utils.Bytes;
@@ -119,6 +124,26 @@ private Topology getTestTopology(JsonKStreamApplicationConfig appConf) {
119124
return builder.build();
120125
}
121126

127+
List<Path> retrieveSortedFilenames(String resourcePath, String fileKeyword) throws IOException {
128+
Path targetDir = Paths.get(getClass().getResource("/" + resourcePath).getPath());
129+
List<Path> out = new ArrayList<>();
130+
131+
Files.walkFileTree(
132+
targetDir,
133+
new SimpleFileVisitor<>() {
134+
@Override
135+
public FileVisitResult visitFile(Path f, BasicFileAttributes attrs) {
136+
if (f.getFileName().toString().startsWith(fileKeyword)) {
137+
out.add(f);
138+
}
139+
return FileVisitResult.CONTINUE;
140+
}
141+
});
142+
out.sort(Comparator.comparing(f -> f.getFileName().toString()));
143+
144+
return out;
145+
}
146+
122147
@Test
123148
@DisplayName("Test that previously unseen counters will produce zero values")
124149
void transformCounterInit()
@@ -287,69 +312,71 @@ void transformCounterOldReset()
287312
void transformCounterUnexpectedReset()
288313
throws IOException, JSONException, InvalidConfigurations, InvalidMatchConfiguration {
289314
setup("counter-normalization/application.conf");
290-
var input1String =
291-
JsonUtil.readFromInputStream(
292-
getClass()
293-
.getClassLoader()
294-
.getResourceAsStream("counter-normalization/reset/input1.json"));
295-
var input2String =
296-
JsonUtil.readFromInputStream(
297-
getClass()
298-
.getClassLoader()
299-
.getResourceAsStream("counter-normalization/reset/input2.json"));
300-
var expected1String =
301-
JsonUtil.readFromInputStream(
302-
getClass()
303-
.getClassLoader()
304-
.getResourceAsStream("counter-normalization/reset/output1.json"));
305-
var expected2String =
306-
JsonUtil.readFromInputStream(
307-
getClass()
308-
.getClassLoader()
309-
.getResourceAsStream("counter-normalization/reset/output2.json"));
315+
String resourcePath = "counter-normalization/reset/";
316+
List<Path> inputFiles = retrieveSortedFilenames(resourcePath, "input");
317+
List<Path> outputFiles = retrieveSortedFilenames(resourcePath, "output");
318+
319+
assertEquals(inputFiles.size(), outputFiles.size());
320+
for (var i = 0; i < inputFiles.size(); i++) {
321+
var input = inputFiles.get(i).getFileName();
322+
var inputString =
323+
JsonUtil.readFromInputStream(
324+
getClass()
325+
.getClassLoader()
326+
.getResourceAsStream(String.format("%s/%s", resourcePath, input)));
327+
328+
inputTopic.pipeInput("k1", inputString, 1647244800000L + 60000L * i);
329+
}
310330

311-
inputTopic.pipeInput("k1", input1String, 1647244800000L);
312-
inputTopic.pipeInput("k1", input2String, 1647244860000L);
313331
var processed = outputTopic.readValuesToList();
314-
315-
assertEquals(2, processed.size());
316-
JSONAssert.assertEquals(expected1String, processed.get(0), true);
317-
JSONAssert.assertEquals(expected2String, processed.get(1), true);
332+
assertEquals(inputFiles.size(), processed.size());
333+
334+
for (var i = 0; i < outputFiles.size(); i++) {
335+
var output = outputFiles.get(i).getFileName();
336+
var expectedString =
337+
JsonUtil.readFromInputStream(
338+
getClass()
339+
.getClassLoader()
340+
.getResourceAsStream(String.format("%s/%s", resourcePath, output)));
341+
342+
JSONAssert.assertEquals(expectedString, processed.get(i), true);
343+
}
318344
}
319345

320346
@Test
321347
@DisplayName("Test normalization works correctly after counters wrap around max u32 and u64")
322348
void transformCounterWrap()
323349
throws IOException, JSONException, InvalidConfigurations, InvalidMatchConfiguration {
324350
setup("counter-normalization/application.conf");
325-
var input1String =
326-
JsonUtil.readFromInputStream(
327-
getClass()
328-
.getClassLoader()
329-
.getResourceAsStream("counter-normalization/wrap/input1.json"));
330-
var input2String =
331-
JsonUtil.readFromInputStream(
332-
getClass()
333-
.getClassLoader()
334-
.getResourceAsStream("counter-normalization/wrap/input2.json"));
335-
var expected1String =
336-
JsonUtil.readFromInputStream(
337-
getClass()
338-
.getClassLoader()
339-
.getResourceAsStream("counter-normalization/wrap/output1.json"));
340-
var expected2String =
341-
JsonUtil.readFromInputStream(
342-
getClass()
343-
.getClassLoader()
344-
.getResourceAsStream("counter-normalization/wrap/output2.json"));
351+
String resourcePath = "counter-normalization/wrap/";
352+
List<Path> inputFiles = retrieveSortedFilenames(resourcePath, "input");
353+
List<Path> outputFiles = retrieveSortedFilenames(resourcePath, "output");
354+
355+
assertEquals(inputFiles.size(), outputFiles.size());
356+
for (var i = 0; i < inputFiles.size(); i++) {
357+
var input = inputFiles.get(i).getFileName();
358+
var inputString =
359+
JsonUtil.readFromInputStream(
360+
getClass()
361+
.getClassLoader()
362+
.getResourceAsStream(String.format("%s/%s", resourcePath, input)));
363+
364+
inputTopic.pipeInput("k1", inputString, 1647244800000L + 60000L * i);
365+
}
345366

346-
inputTopic.pipeInput("k1", input1String, 1647244800000L);
347-
inputTopic.pipeInput("k1", input2String, 1647244860000L);
348367
var processed = outputTopic.readValuesToList();
349-
350-
assertEquals(2, processed.size());
351-
JSONAssert.assertEquals(expected1String, processed.get(0), true);
352-
JSONAssert.assertEquals(expected2String, processed.get(1), true);
368+
assertEquals(inputFiles.size(), processed.size());
369+
370+
for (var i = 0; i < outputFiles.size(); i++) {
371+
var output = outputFiles.get(i).getFileName();
372+
var expectedString =
373+
JsonUtil.readFromInputStream(
374+
getClass()
375+
.getClassLoader()
376+
.getResourceAsStream(String.format("%s/%s", resourcePath, output)));
377+
378+
JSONAssert.assertEquals(expectedString, processed.get(i), true);
379+
}
353380
}
354381

355382
@Test

‎candyfloss/src/test/resources/counter-normalization/reset/input3.json‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
{
2+
"msg_metadata": {
3+
"event_type": "dump",
4+
"seq": 21196988,
5+
"timestamp": "2022-03-14 09:00:02",
6+
"telemetry_node": "138.190.128.22",
7+
"telemetry_port": 55286,
8+
"serialization": "gpb",
9+
"writer_id": "daisy64stcisco1v01c/2080144",
10+
"label": {
11+
"nkey": "unknown",
12+
"pkey": "unknown"
13+
}
14+
},
15+
"telemetry_metadata": {
16+
"collection_id": 910867,
17+
"collection_end_time": 1647244767814,
18+
"collection_start_time": 0,
19+
"encoding_path": "openconfig-interfaces:interfaces",
20+
"subscription": "DAISY-SUBSCRIPTION",
21+
"msg_timestamp": 1647244767813,
22+
"node_id": "hoa01ro1010zoi"
23+
},
24+
"timestamp": 1647244767278,
25+
"name": "TenGigE0/0/0/0",
26+
"state": {
27+
"u64-counter": 9223372036854774705, // this is slightly lower than u64/2
28+
"u32-counter": null, // put these to null, input3-4 focus on test for u64-counter
29+
"u32-value": null,
30+
"u64-value": null
31+
}
32+
}

‎candyfloss/src/test/resources/counter-normalization/reset/input4.json‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
{
2+
"msg_metadata": {
3+
"event_type": "dump",
4+
"seq": 21196988,
5+
"timestamp": "2022-03-14 09:00:03",
6+
"telemetry_node": "138.190.128.22",
7+
"telemetry_port": 55286,
8+
"serialization": "gpb",
9+
"writer_id": "daisy64stcisco1v01c/2080144",
10+
"label": {
11+
"nkey": "unknown",
12+
"pkey": "unknown"
13+
}
14+
},
15+
"telemetry_metadata": {
16+
"collection_id": 910867,
17+
"collection_end_time": 1647244767814,
18+
"collection_start_time": 0,
19+
"encoding_path": "openconfig-interfaces:interfaces",
20+
"subscription": "DAISY-SUBSCRIPTION",
21+
"msg_timestamp": 1647244767813,
22+
"node_id": "hoa01ro1010zoi"
23+
},
24+
"timestamp": 1647244767278,
25+
"name": "TenGigE0/0/0/0",
26+
"state": {
27+
"u64-counter": 2000000, // this one is reset comparing to previous slot, then output 4 is null
28+
"u32-counter": null,
29+
"u32-value": null,
30+
"u64-value": null
31+
}
32+
}

‎candyfloss/src/test/resources/counter-normalization/reset/input5.json‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
{
2+
"msg_metadata": {
3+
"event_type": "dump",
4+
"seq": 21196988,
5+
"timestamp": "2022-03-14 09:00:03",
6+
"telemetry_node": "138.190.128.22",
7+
"telemetry_port": 55286,
8+
"serialization": "gpb",
9+
"writer_id": "daisy64stcisco1v01c/2080144",
10+
"label": {
11+
"nkey": "unknown",
12+
"pkey": "unknown"
13+
}
14+
},
15+
"telemetry_metadata": {
16+
"collection_id": 910867,
17+
"collection_end_time": 1647244767814,
18+
"collection_start_time": 0,
19+
"encoding_path": "openconfig-interfaces:interfaces",
20+
"subscription": "DAISY-SUBSCRIPTION",
21+
"msg_timestamp": 1647244767813,
22+
"node_id": "hoa01ro1010zoi"
23+
},
24+
"timestamp": 1647244767278,
25+
"name": "TenGigE0/0/0/0",
26+
"state": {
27+
"u64-counter": 15223372036854774705, // this one between u64 and u64/2
28+
"u32-counter": null,
29+
"u32-value": null,
30+
"u64-value": null
31+
}
32+
}

‎candyfloss/src/test/resources/counter-normalization/reset/input6.json‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
{
2+
"msg_metadata": {
3+
"event_type": "dump",
4+
"seq": 21196988,
5+
"timestamp": "2022-03-14 09:00:03",
6+
"telemetry_node": "138.190.128.22",
7+
"telemetry_port": 55286,
8+
"serialization": "gpb",
9+
"writer_id": "daisy64stcisco1v01c/2080144",
10+
"label": {
11+
"nkey": "unknown",
12+
"pkey": "unknown"
13+
}
14+
},
15+
"telemetry_metadata": {
16+
"collection_id": 910867,
17+
"collection_end_time": 1647244767814,
18+
"collection_start_time": 0,
19+
"encoding_path": "openconfig-interfaces:interfaces",
20+
"subscription": "DAISY-SUBSCRIPTION",
21+
"msg_timestamp": 1647244767813,
22+
"node_id": "hoa01ro1010zoi"
23+
},
24+
"timestamp": 1647244767278,
25+
"name": "TenGigE0/0/0/0",
26+
"state": {
27+
"u64-counter": 2000000, // this one is reset, output should be null
28+
"u32-counter": null,
29+
"u32-value": null,
30+
"u64-value": null
31+
}
32+
}

‎candyfloss/src/test/resources/counter-normalization/reset/output3.json‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
{
2+
"msg_metadata": {
3+
"event_type": "dump",
4+
"seq": 21196988,
5+
"timestamp": "2022-03-14 09:00:02",
6+
"telemetry_node": "138.190.128.22",
7+
"telemetry_port": 55286,
8+
"serialization": "gpb",
9+
"writer_id": "daisy64stcisco1v01c/2080144",
10+
"label": {
11+
"nkey": "unknown",
12+
"pkey": "unknown"
13+
}
14+
},
15+
"telemetry_metadata": {
16+
"collection_id": 910867,
17+
"collection_end_time": 1647244767814,
18+
"collection_start_time": 0,
19+
"encoding_path": "openconfig-interfaces:interfaces",
20+
"subscription": "DAISY-SUBSCRIPTION",
21+
"msg_timestamp": 1647244767813,
22+
"node_id": "hoa01ro1010zoi"
23+
},
24+
"timestamp": 1647244767278,
25+
"name": "TenGigE0/0/0/0",
26+
"state": {
27+
"u64-counter": 9223372036851774705, // previous state is still ok, do a normal normalization, but next slot reset
28+
"u32-counter": null,
29+
"u32-value": null,
30+
"u64-value": null
31+
}
32+
}

‎candyfloss/src/test/resources/counter-normalization/reset/output4.json‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
1+
{
2+
"msg_metadata": {
3+
"event_type": "dump",
4+
"seq": 21196988,
5+
"timestamp": "2022-03-14 09:00:03",
6+
"telemetry_node": "138.190.128.22",
7+
"telemetry_port": 55286,
8+
"serialization": "gpb",
9+
"writer_id": "daisy64stcisco1v01c/2080144",
10+
"label": {
11+
"nkey": "unknown",
12+
"pkey": "unknown"
13+
}
14+
},
15+
"telemetry_metadata": {
16+
"collection_id": 910867,
17+
"collection_end_time": 1647244767814,
18+
"collection_start_time": 0,
19+
"encoding_path": "openconfig-interfaces:interfaces",
20+
"subscription": "DAISY-SUBSCRIPTION",
21+
"msg_timestamp": 1647244767813,
22+
"node_id": "hoa01ro1010zoi"
23+
},
24+
"timestamp": 1647244767278,
25+
"name": "TenGigE0/0/0/0",
26+
"state": {
27+
"u64-counter": null,
28+
"u32-counter": null,
29+
"u32-value": null,
30+
"u64-value": null
31+
}
32+
}

0 commit comments

Comments
 (0)