diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java index f8857fd..5f555d0 100644 --- a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java +++ b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/CandyflossKStreamsApplication.java @@ -221,6 +221,7 @@ public Topology buildTopology() { new CounterNormalizationProcessor( config.getPipeline(), config.getStateStoreName(), + config.getStateStoreCutoffTime(), config.getMaxCounterCacheAge(), config.getIntCounterWrapAroundLimit(), config.getLongCounterWrapAroundLimit(), diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java index 9b367e4..cae722a 100644 --- a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java +++ b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/JsonKStreamApplicationConfig.java @@ -32,6 +32,7 @@ public enum InputType { private final String discardTopicName; private final String dlqTopicName; private final String stateStoreName; + private final long stateStoreCutoffTime; private final long maxCounterCacheAge; private final int intCounterWrapAroundLimit; private final long longCounterWrapAroundLimit; @@ -58,6 +59,8 @@ public static JsonKStreamApplicationConfig fromConfig(Config config) } final String discardTopicName = config.getConfig("kstream").getString("discard.topic.name"); final String dlqTopicName = config.getConfig("kstream").getString("dlq.topic.name"); + final long stateStoreCutoffTime = + config.getConfig("kstream").getLong("normalization.statestore.cutoff.time.ms"); final String stateStoreName = config.getConfig("kstream").getString("state.store.name"); final long maxCounterCacheAge = config.getConfig("kstream").getLong("state.store.max.counter.cache.age"); @@ -90,6 +93,7 @@ public static JsonKStreamApplicationConfig fromConfig(Config config) discardTopicName, dlqTopicName, stateStoreName, + stateStoreCutoffTime, maxCounterCacheAge, intCounterWrapAroundLimit, longCounterWrapAroundLimit, diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessor.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessor.java index dddbc31..9f80102 100644 --- a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessor.java +++ b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessor.java @@ -41,6 +41,7 @@ public class CounterNormalizationProcessor .description("Number of error messages that are discarded to dlq topic") .register(Metrics.globalRegistry); private final String stateStoreName; + private final long stateStoreCutoffTime; private final PipelineConfig pipelineConfig; private TimestampedKeyValueStore stateStore; private final long maxCounterCacheAge; @@ -67,6 +68,7 @@ public class CounterNormalizationProcessor public CounterNormalizationProcessor( PipelineConfig pipelineConfig, String stateStoreName, + long stateStoreCutoffTime, long maxCounterCacheAge, int intCounterWrapAroundLimit, long longCounterWrapAroundLimit, @@ -75,6 +77,7 @@ public CounterNormalizationProcessor( Duration scanFrequency) { this.pipelineConfig = pipelineConfig; this.stateStoreName = stateStoreName; + this.stateStoreCutoffTime = stateStoreCutoffTime; this.maxCounterCacheAge = maxCounterCacheAge; this.intCounterWrapAroundLimit = intCounterWrapAroundLimit; this.longCounterWrapAroundLimit = longCounterWrapAroundLimit; @@ -175,10 +178,17 @@ private void normalizeCounter( } if (timestamp.toEpochMilli() < savedCounterState.timestamp()) { logger.warn( - "Received a message with a timestamp {} older than saved in the counter store {}. Message: {}", + "Received a message with a timestamp {} older than saved in the counter store {} " + + "(timestamp difference: {} ms). Message: {}", timestamp.toEpochMilli(), savedCounterState.timestamp(), + (timestamp.toEpochMilli() - savedCounterState.timestamp()), flattenedMessage.getValue().json()); + + if (savedCounterState.timestamp() - timestamp.toEpochMilli() > stateStoreCutoffTime) { + stateStore.delete(counterKey); + } + savedCounterState = null; } else { // Save the new counterValue to the store diff --git a/candyfloss/src/main/resources/application.dev.conf b/candyfloss/src/main/resources/application.dev.conf index 883165f..88c5f29 100644 --- a/candyfloss/src/main/resources/application.dev.conf +++ b/candyfloss/src/main/resources/application.dev.conf @@ -18,6 +18,7 @@ kstream { input.topic.name = dev.device-json-raw-input discard.topic.name = dev.candyfloss-processing-discard # All messages that didn't match any step in the pipeline dlq.topic.name = dev.candyfloss-processing--dlq # Messages that encountered a Java exception during the processing + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = dev.candyfloss-counters-store # Kafka state store name to save the counter values state.store.max.counter.cache.age = 900000 // 15 minutes state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/main/resources/application.ietf.dev.conf b/candyfloss/src/main/resources/application.ietf.dev.conf index c22705d..34e5099 100644 --- a/candyfloss/src/main/resources/application.ietf.dev.conf +++ b/candyfloss/src/main/resources/application.ietf.dev.conf @@ -20,6 +20,7 @@ kstream { input.topic.name = daisy.dev.device-json-raw discard.topic.name = daisy.dev.candyfloss-discard dlq.topic.name = daisy.dev.candyfloss-dlq + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = candyfloss-counters-store state.store.max.counter.cache.age = 900000 // 15 minutes state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/main/resources/application.prod.conf b/candyfloss/src/main/resources/application.prod.conf index 4960756..137a632 100644 --- a/candyfloss/src/main/resources/application.prod.conf +++ b/candyfloss/src/main/resources/application.prod.conf @@ -18,6 +18,7 @@ kstream { input.topic.name = prod.device-json-raw-input discard.topic.name = prod.candyfloss-processing-discard # All messages that didn't match any step in the pipeline dlq.topic.name = prod.candyfloss-processing-dlq # Messages that encountered a Java exception during the processing + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = prod.candyfloss-counters-store # Kafka state store name to save the counter values state.store.max.counter.cache.age = 900000 // 15 minutes state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/main/resources/application.test.conf b/candyfloss/src/main/resources/application.test.conf index cf2ef28..0f7f052 100644 --- a/candyfloss/src/main/resources/application.test.conf +++ b/candyfloss/src/main/resources/application.test.conf @@ -19,6 +19,7 @@ kstream { discard.topic.name = test.candyfloss-processing-discard # All messages that didn't match any step in the pipeline dlq.topic.name = test.candyfloss-processing-dlq # Messages that encountered a Java exception during the processing state.store.name = test.candyfloss-counters-store # Kafka state store name to save the counter values + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.max.counter.cache.age = 900000 // 15 minutes state.store.int.counter.wrap.limit = 10000 state.store.long.counter.wrap.limit = 10000000 diff --git a/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessorTest.java b/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessorTest.java index c91ccd8..346708f 100644 --- a/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessorTest.java +++ b/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/processors/CounterNormalizationProcessorTest.java @@ -96,6 +96,7 @@ private Topology getTestTopology(JsonKStreamApplicationConfig appConf) { appConf.getPipeline(), appConf.getStateStoreName(), appConf.getMaxCounterCacheAge(), + appConf.getStateStoreCutoffTime(), appConf.getIntCounterWrapAroundLimit(), appConf.getLongCounterWrapAroundLimit(), appConf.getCounterWrapAroundTimeMs(), @@ -429,33 +430,33 @@ void transformCounterNegative() void transformCounterOlderMilli() throws IOException, JSONException, InvalidConfigurations, InvalidMatchConfiguration { setup("counter-normalization/application-milli.conf"); - var input1 = - JsonUtil.readFromInputStream( - getClass() - .getClassLoader() - .getResourceAsStream("counter-normalization/older/input1.json")); - var input2 = - JsonUtil.readFromInputStream( - getClass() - .getClassLoader() - .getResourceAsStream("counter-normalization/older/input2.json")); - var expected1 = - JsonUtil.readFromInputStream( - getClass() - .getClassLoader() - .getResourceAsStream("counter-normalization/older/output1.json")); - var expected2 = - JsonUtil.readFromInputStream( - getClass() - .getClassLoader() - .getResourceAsStream("counter-normalization/older/output2.json")); + String resourcePath = "counter-normalization/older/"; + List inputFiles = retrieveSortedFilenames(resourcePath, "input"); + List outputFiles = retrieveSortedFilenames(resourcePath, "output"); - inputTopic.pipeInput("k1", input1); - inputTopic.pipeInput("k1", input2); + assertEquals(inputFiles.size(), outputFiles.size()); + for (var i = 0; i < inputFiles.size(); i++) { + var input = inputFiles.get(i).getFileName(); + var inputString = + JsonUtil.readFromInputStream( + getClass() + .getClassLoader() + .getResourceAsStream(String.format("%s/%s", resourcePath, input))); + + inputTopic.pipeInput("k1", inputString); + } var processed = outputTopic.readValuesToList(); - assertEquals(2, processed.size()); - JSONAssert.assertEquals(expected1, processed.get(0), true); - JSONAssert.assertEquals(expected2, processed.get(1), true); + assertEquals(inputFiles.size(), processed.size()); + for (var i = 0; i < outputFiles.size(); i++) { + var output = outputFiles.get(i).getFileName(); + var expectedString = + JsonUtil.readFromInputStream( + getClass() + .getClassLoader() + .getResourceAsStream(String.format("%s/%s", resourcePath, output))); + + JSONAssert.assertEquals(expectedString, processed.get(i), true); + } } } diff --git a/candyfloss/src/test/resources/avro-input/application.conf b/candyfloss/src/test/resources/avro-input/application.conf index c37be86..a2aad60 100644 --- a/candyfloss/src/test/resources/avro-input/application.conf +++ b/candyfloss/src/test/resources/avro-input/application.conf @@ -16,6 +16,7 @@ kstream { discard.topic.name = daisy.dev.avro-input-discard dlq.topic.name =daisy.dev.avro-input-dlq state.store.name = daisy.dev.avro-input-store + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.max.counter.cache.age = 1000 state.store.int.counter.wrap.limit = 10000 state.store.long.counter.wrap.limit = 10000000 diff --git a/candyfloss/src/test/resources/counter-normalization/application-milli.conf b/candyfloss/src/test/resources/counter-normalization/application-milli.conf index 8e6b431..725bd12 100644 --- a/candyfloss/src/test/resources/counter-normalization/application-milli.conf +++ b/candyfloss/src/test/resources/counter-normalization/application-milli.conf @@ -13,6 +13,7 @@ kstream { input.topic.name = daisy.prod.device-json-raw discard.topic.name = daisy.dev.device-json-flat dlq.topic.name = daisy.dev.device-json-flat-dlq + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = daisy.dev.device-counters-store state.store.max.counter.cache.age = 10000 state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/test/resources/counter-normalization/application-seconds.conf b/candyfloss/src/test/resources/counter-normalization/application-seconds.conf index 77a4641..abe8515 100644 --- a/candyfloss/src/test/resources/counter-normalization/application-seconds.conf +++ b/candyfloss/src/test/resources/counter-normalization/application-seconds.conf @@ -13,6 +13,7 @@ kstream { input.topic.name = daisy.prod.device-json-raw discard.topic.name = daisy.dev.device-json-flat dlq.topic.name = daisy.dev.device-json-flat-dlq + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = daisy.dev.device-counters-store state.store.max.counter.cache.age = 10000 state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/test/resources/counter-normalization/application.conf b/candyfloss/src/test/resources/counter-normalization/application.conf index eb53999..b408d74 100644 --- a/candyfloss/src/test/resources/counter-normalization/application.conf +++ b/candyfloss/src/test/resources/counter-normalization/application.conf @@ -13,6 +13,7 @@ kstream { input.topic.name = daisy.prod.device-json-raw discard.topic.name = daisy.dev.device-json-flat dlq.topic.name = daisy.dev.device-json-flat-dlq + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = daisy.dev.device-counters-store state.store.max.counter.cache.age = 10000 state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/test/resources/counter-normalization/older/input3.json b/candyfloss/src/test/resources/counter-normalization/older/input3.json new file mode 100644 index 0000000..788e38b --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/input3.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244800004, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244767814, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244767278, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": 3300, + "u32-counter": 330, + "u32-value": 22, + "u64-value": 11 + } +} \ No newline at end of file diff --git a/candyfloss/src/test/resources/counter-normalization/older/input4.json b/candyfloss/src/test/resources/counter-normalization/older/input4.json new file mode 100644 index 0000000..b1a37ca --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/input4.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 32498808463000, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244767814, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244767278, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": 3800, + "u32-counter": 380, + "u32-value": 28, + "u64-value": 18 + } +} diff --git a/candyfloss/src/test/resources/counter-normalization/older/input5.json b/candyfloss/src/test/resources/counter-normalization/older/input5.json new file mode 100644 index 0000000..bff250c --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/input5.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244810011, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244800008, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244800008, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": 3810, + "u32-counter": 383, + "u32-value": 28, + "u64-value": 18 + } +} \ No newline at end of file diff --git a/candyfloss/src/test/resources/counter-normalization/older/input6.json b/candyfloss/src/test/resources/counter-normalization/older/input6.json new file mode 100644 index 0000000..90030d9 --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/input6.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244810021, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244800008, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244800008, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": 3820, + "u32-counter": 384, + "u32-value": 28, + "u64-value": 18 + } +} diff --git a/candyfloss/src/test/resources/counter-normalization/older/input7.json b/candyfloss/src/test/resources/counter-normalization/older/input7.json new file mode 100644 index 0000000..adb3152 --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/input7.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244810031, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244800008, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244800008, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": 3830, + "u32-counter": 387, + "u32-value": 28, + "u64-value": 18 + } +} diff --git a/candyfloss/src/test/resources/counter-normalization/older/output3.json b/candyfloss/src/test/resources/counter-normalization/older/output3.json new file mode 100644 index 0000000..3fec892 --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/output3.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244800004, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244767814, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244767278, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": 2300, // state is still from input 1 + "u32-counter": 230, + "u32-value": 22, + "u64-value": 11 + } +} diff --git a/candyfloss/src/test/resources/counter-normalization/older/output4.json b/candyfloss/src/test/resources/counter-normalization/older/output4.json new file mode 100644 index 0000000..bf796c0 --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/output4.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 32498808463000, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244767814, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244767278, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": null, // we still do normal normalization (input4-input3), stateStore is by input4 + "u32-counter": null, + "u32-value": 28, // this is Value directly from input4 + "u64-value": 18 + } +} \ No newline at end of file diff --git a/candyfloss/src/test/resources/counter-normalization/older/output5.json b/candyfloss/src/test/resources/counter-normalization/older/output5.json new file mode 100644 index 0000000..c42ce29 --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/output5.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244810011, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244800008, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244800008, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": null, // due to state store is deleted + "u32-counter": null, + "u32-value": 28, + "u64-value": 18 + } +} diff --git a/candyfloss/src/test/resources/counter-normalization/older/output6.json b/candyfloss/src/test/resources/counter-normalization/older/output6.json new file mode 100644 index 0000000..3a73413 --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/output6.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244810021, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244800008, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244800008, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": null, + "u32-counter": null, + "u32-value": 28, + "u64-value": 18 + } +} \ No newline at end of file diff --git a/candyfloss/src/test/resources/counter-normalization/older/output7.json b/candyfloss/src/test/resources/counter-normalization/older/output7.json new file mode 100644 index 0000000..bc7092c --- /dev/null +++ b/candyfloss/src/test/resources/counter-normalization/older/output7.json @@ -0,0 +1,32 @@ +{ + "msg_metadata": { + "event_type": "dump", + "seq": 21196988, + "milli-timestamp": 1647244810031, + "telemetry_node": "192.168.1.22", + "telemetry_port": 55286, + "serialization": "gpb", + "writer_id": "daisy64stcisco1v01c/2080144", + "label": { + "nkey": "unknown", + "pkey": "unknown" + } + }, + "telemetry_metadata": { + "collection_id": 910867, + "collection_end_time": 1647244800008, + "collection_start_time": 0, + "encoding_path": "openconfig-interfaces:interfaces", + "subscription": "DAISY-SUBSCRIPTION", + "msg_timestamp": 1647244767813, + "node_id": "hoa01ro1010zoi" + }, + "timestamp": 1647244800008, + "name": "TenGigE0/0/0/0", + "state": { + "u64-counter": 10, + "u32-counter": 3, + "u32-value": 28, + "u64-value": 18 + } +} diff --git a/candyfloss/src/test/resources/transformers/application-multi-match.conf b/candyfloss/src/test/resources/transformers/application-multi-match.conf index 1108863..18c3841 100644 --- a/candyfloss/src/test/resources/transformers/application-multi-match.conf +++ b/candyfloss/src/test/resources/transformers/application-multi-match.conf @@ -12,6 +12,7 @@ kstream { input.topic.name = daisy.prod.device-json-raw discard.topic.name = daisy.dev.device-json-flat dlq.topic.name = daisy.dev.device-json-flat-dlq + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = daisy.dev.device-counters-store state.store.max.counter.cache.age = 1000 state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/test/resources/transformers/application-pre-transform.conf b/candyfloss/src/test/resources/transformers/application-pre-transform.conf index 9d33e30..36dc199 100644 --- a/candyfloss/src/test/resources/transformers/application-pre-transform.conf +++ b/candyfloss/src/test/resources/transformers/application-pre-transform.conf @@ -13,6 +13,7 @@ kstream { input.topic.name = daisy.prod.device-json-raw discard.topic.name = daisy.dev.device-json-flat dlq.topic.name = daisy.dev.device-json-flat-dlq + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = daisy.dev.device-counters-store state.store.max.counter.cache.age = 1000 state.store.int.counter.wrap.limit = 10000 diff --git a/candyfloss/src/test/resources/transformers/application.conf b/candyfloss/src/test/resources/transformers/application.conf index d06ad55..adc235a 100644 --- a/candyfloss/src/test/resources/transformers/application.conf +++ b/candyfloss/src/test/resources/transformers/application.conf @@ -13,6 +13,7 @@ kstream { input.topic.name = daisy.prod.device-json-raw discard.topic.name = daisy.dev.device-json-flat dlq.topic.name = daisy.dev.device-json-flat-dlq + normalization.statestore.cutoff.time.ms = 1800000 # 30 minutes state.store.name = daisy.dev.device-counters-store state.store.max.counter.cache.age = 1000 state.store.int.counter.wrap.limit = 10000