diff --git a/candyfloss/build.gradle b/candyfloss/build.gradle index d58186b..9f2d624 100644 --- a/candyfloss/build.gradle +++ b/candyfloss/build.gradle @@ -57,6 +57,8 @@ dependencies { testImplementation libs.junit.jupiter.params testImplementation libs.jackson.databind testImplementation libs.jsonassert + + testImplementation project(':test-utils') } test { diff --git a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/TimeExtractorConfig.java b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/TimeExtractorConfig.java index f130972..d2669a7 100644 --- a/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/TimeExtractorConfig.java +++ b/candyfloss/src/main/java/com/swisscom/daisy/cosmos/candyfloss/config/TimeExtractorConfig.java @@ -12,7 +12,8 @@ public class TimeExtractorConfig { public enum TimestampType { RFC2822, EpochMilli, - EpochSeconds + EpochSeconds, + ISO8601 } private final JsonPath jsonPath; @@ -25,6 +26,8 @@ public static TimeExtractorConfig fromJson(Map configs, String n final TimestampType timestampType; if (timestampTypeString.equals("RFC2822")) { timestampType = TimestampType.RFC2822; + } else if (timestampTypeString.equals("ISO8601")) { + timestampType = TimestampType.ISO8601; } else if (timestampTypeString.equals("EpochMilli")) { timestampType = TimestampType.EpochMilli; } else if (timestampTypeString.equals("EpochSeconds")) { 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 5f4f8af..dddbc31 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 @@ -270,6 +270,9 @@ private Instant getTimestamp(String tag, DocumentContext context) { var time = LocalDateTime.parse(dateTimeString, formatter); var zoned = time.atZone(ZoneId.of(ZoneId.SHORT_IDS.get("ECT"))); return zoned.toInstant(); + } else if (timeExtractorConfig.getValueType() == TimeExtractorConfig.TimestampType.ISO8601) { + var dateTimeString = (String) extractedTimestamp; + return Instant.parse(dateTimeString); } else { Long longTimestamp; if (extractedTimestamp instanceof String) { diff --git a/candyfloss/src/main/resources/application.ietf.dev.conf b/candyfloss/src/main/resources/application.ietf.dev.conf new file mode 100644 index 0000000..c22705d --- /dev/null +++ b/candyfloss/src/main/resources/application.ietf.dev.conf @@ -0,0 +1,37 @@ +kafka { + application.id = "daisy.dev.candyfloss" + group.id = "daisy.dev.candyfloss-1" + bootstrap.servers = "kafka.sbd.corproot.net:9093" + acks = 1 + linger.ms = 5 + processing.guarantee = at_least_once + enable.idempotence = false + schema.registry.url = "https://schema-registry.zhh.sbd.corproot.net" + key.serializer = "org.apache.kafka.common.serialization.StringSerializer" + value.serializer = "org.apache.kafka.common.serialization.StringSerializer" + metrics.recording.level = DEBUG + compression.type = zstd + replication.factor = 2 + num.stream.threads = 2 + auto.offset.reset = latest + probing.rebalance.interval.ms = 120000 +} +kstream { + input.topic.name = daisy.dev.device-json-raw + discard.topic.name = daisy.dev.candyfloss-discard + dlq.topic.name = daisy.dev.candyfloss-dlq + state.store.name = candyfloss-counters-store + state.store.max.counter.cache.age = 900000 // 15 minutes + state.store.int.counter.wrap.limit = 10000 + state.store.long.counter.wrap.limit = 10000000 + state.store.long.counter.time.ms = 300000 # allow max 5 min for counter wrap around otherwise it's a reset + state.store.delete.scan.frequency.days = 7 # how often to trigger scanning for old unused counters and delete them + pre.transform = pre-transform.ietf.dev.json + pipeline = { + # todo: customize ietf-telemetry-message pipeline here + fano-interfaces { + output.topic.name = output-topic-ietf-telemetry-message + file = pipeline-ietf-telemetry-message/fano-interfaces.json + } + } +} diff --git a/candyfloss/src/main/resources/pipeline-ietf-telemetry-message/fano-interfaces.json b/candyfloss/src/main/resources/pipeline-ietf-telemetry-message/fano-interfaces.json new file mode 100644 index 0000000..3070cda --- /dev/null +++ b/candyfloss/src/main/resources/pipeline-ietf-telemetry-message/fano-interfaces.json @@ -0,0 +1,13 @@ +{ + "match": { + "jsonpath": "$.ietf-telemetry-message:message" + }, + "transform": [ + { + "operation": "shift", + "spec": { + "*": "&" + } + } + ] +} \ No newline at end of file diff --git a/candyfloss/src/main/resources/pre-transform.ietf.dev.json b/candyfloss/src/main/resources/pre-transform.ietf.dev.json new file mode 100644 index 0000000..49524dd --- /dev/null +++ b/candyfloss/src/main/resources/pre-transform.ietf.dev.json @@ -0,0 +1,22 @@ +[ + { + /* + This transformation makes sure (door keeper) that we only process valid messages containing the leaf serialisation + with value containing "ietf-subscribed-notifications:encode-json" + */ + "operation": "shift", + "spec": { + "ietf-telemetry-message:message": { + "telemetry-message-metadata": { + "ietf-yang-push-telemetry-message:yang-push-subscription": { + "encoding":{ + "ietf-subscribed-notifications:encode-json": { + "@(5,ietf-telemetry-message:message)": "ietf-telemetry-message:message" + } + } + } + } + } + } + } +] \ No newline at end of file diff --git a/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/AbstractDeploymentTest.java b/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/AbstractDeploymentTest.java index 609c4a7..75d89c9 100644 --- a/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/AbstractDeploymentTest.java +++ b/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/AbstractDeploymentTest.java @@ -1,9 +1,11 @@ package com.swisscom.daisy.cosmos.candyfloss; +import static com.swisscom.daisy.cosmos.candyfloss.testutils.JsonUtil.getValueByJsonPath; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.fail; import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.swisscom.daisy.cosmos.candyfloss.config.JsonKStreamApplicationConfig; import com.swisscom.daisy.cosmos.candyfloss.config.PipelineStepConfig; @@ -164,10 +166,25 @@ protected void testCase( final String testKey = "testKey"; var dateTimeString = (String) inputMap.get("timestamp"); + + if (dateTimeString == null && inputMap.containsKey("ietf-telemetry-message:message")) { + String inputString = objectMapper.writeValueAsString(inputMap); + JsonNode rootNode = objectMapper.readTree(inputString); + String path = + "ietf-telemetry-message:message." + + "payload.ietf-yp-notification:envelope." + + "contents." + + "ietf-yang-push:push-update." + + "ietf-yp-observation:timestamp"; + dateTimeString = getValueByJsonPath(rootNode, path); + } + final ZonedDateTime zoned; if (dateTimeString != null) { DateTimeFormatter formatter; - if (dateTimeString.contains("-")) { + if (dateTimeString.contains("T")) { + zoned = ZonedDateTime.parse(dateTimeString, DateTimeFormatter.ISO_OFFSET_DATE_TIME); + } else if (dateTimeString.contains("-")) { formatter = new DateTimeFormatterBuilder() .append(DateTimeFormatter.ISO_LOCAL_DATE) diff --git a/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/DeploymentTest.java b/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/DeploymentTest.java index 59e9b8a..8f78d22 100644 --- a/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/DeploymentTest.java +++ b/candyfloss/src/test/java/com/swisscom/daisy/cosmos/candyfloss/DeploymentTest.java @@ -16,21 +16,31 @@ class DeploymentTest extends AbstractDeploymentTest { @ParameterizedTest - @MethodSource("providerForYangModels") + @MethodSource("providerForAllYangModels") protected void test(String applicationConfigFileName, String env, String name, Path testCase) throws InvalidConfigurations, IOException, JSONException, InvalidMatchConfiguration { testImpl(applicationConfigFileName, env, name, testCase); } + protected static Stream providerForAllYangModels() + throws IOException, URISyntaxException { + var outLegacy = providerForYangModels("legacy"); + var outIetf = providerForYangModels("ietf"); + + return Stream.concat(outLegacy, outIetf); + } + /*** Discover YANG model test cases and provide them as arguments to the parametrized test */ - protected static Stream providerForYangModels() + protected static Stream providerForYangModels(String profile) throws IOException, URISyntaxException { ClassLoader loader = Thread.currentThread().getContextClassLoader(); // Create a map from the discovered folders in the `test/resources/deployment` folder // The key is the sub-folder name and the value is a full path to the sub folder + String deploymentFolder = + profile.equals("ietf") ? "deployment-ietf-telemetry-message" : "deployment"; var yangModels = - Files.list(Path.of(Objects.requireNonNull(loader.getResource("deployment")).toURI())) + Files.list(Path.of(Objects.requireNonNull(loader.getResource(deploymentFolder)).toURI())) .filter(Files::isDirectory) .collect(Collectors.toMap(k -> k.getFileName().toString(), v -> v)); @@ -47,14 +57,11 @@ protected static Stream providerForYangModels() var testFixtures = Files.list(path).filter(Files::isDirectory).toList(); for (var testPath : testFixtures) { for (var env : List.of("dev", "test", "prod")) { - String applicationConfig = "application." + env + ".conf"; - try (var stream = - DeploymentTest.class.getClassLoader().getResourceAsStream(applicationConfig)) { - if (stream == null) { - continue; - } - } - testCases.add(Arguments.of(applicationConfig, env, name, testPath)); + String fileNamePrefix = profile.equals("ietf") ? "application.ietf." : "application."; + + String applicationConfig = fileNamePrefix + env + ".conf"; + if (loader.getResource(applicationConfig) != null) + testCases.add(Arguments.of(applicationConfig, env, name, testPath)); } } } diff --git a/candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-input.json b/candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-input.json new file mode 100644 index 0000000..9e26dfe --- /dev/null +++ b/candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-input.json @@ -0,0 +1 @@ +{} \ No newline at end of file diff --git a/candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-output.json b/candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-output.json new file mode 100644 index 0000000..0637a08 --- /dev/null +++ b/candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-output.json @@ -0,0 +1 @@ +[] \ No newline at end of file diff --git a/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/CustomFunctions.java b/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/CustomFunctions.java index 1275353..62cf6da 100644 --- a/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/CustomFunctions.java +++ b/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/CustomFunctions.java @@ -42,6 +42,14 @@ protected Optional applySingle(Object arg) { } } + public static final class getSysProperty extends Function.SingleFunction { + @Override + protected Optional applySingle(Object arg) { + String prop = (String) arg; + return Optional.of(System.getProperty(prop)); + } + } + public static final class multiply extends Function.ListFunction { @Override protected Optional applyList(List argList) { diff --git a/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/DaisyModifier.java b/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/DaisyModifier.java index 45858ab..c9420f9 100644 --- a/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/DaisyModifier.java +++ b/json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/DaisyModifier.java @@ -47,6 +47,7 @@ public class DaisyModifier implements ContextualTransform, SpecDriven { static { STOCK_FUNCTIONS.put("jsonStringToJson", new CustomFunctions.jsonStringToJson()); STOCK_FUNCTIONS.put("multiply", new CustomFunctions.multiply()); + STOCK_FUNCTIONS.put("getSysProperty", new CustomFunctions.getSysProperty()); } private DaisyModifier(Object spec, OpMode opMode, Map functionsMap) { diff --git a/json-transformation/src/test/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/GetSysPropertyTest.java b/json-transformation/src/test/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/GetSysPropertyTest.java new file mode 100644 index 0000000..b8bfbed --- /dev/null +++ b/json-transformation/src/test/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/GetSysPropertyTest.java @@ -0,0 +1,53 @@ +package com.swisscom.daisy.cosmos.candyfloss.transformations.jolt; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import com.bazaarvoice.jolt.Chainr; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +public class GetSysPropertyTest { + private static final String operation = + "com.swisscom.daisy.cosmos.candyfloss.transformations.jolt.DaisyModifier$Overwritr"; + static Map input = new HashMap<>(Map.of("k1", "to_be_changed")); + + @BeforeEach + public void setUp() { + System.setProperty("ENV_VAR", "xx"); + } + + @Test + void testNull() { + var spec = + Chainr.fromSpec( + List.of( + Map.of( + "operation", + operation, + "spec", + Map.of("k1", "=getSysProperty(ENV_VAR_NON_EXISTING)")))); + var output = spec.transform(input); + + Map outExpected = new HashMap<>(); + outExpected.put("k1", null); + + assertEquals(outExpected, output); + } + + @Test + void testNotNull() { + var spec = + Chainr.fromSpec( + List.of( + Map.of("operation", operation, "spec", Map.of("k1", "=getSysProperty(ENV_VAR)")))); + var output = spec.transform(input); + + Map outExpected = new HashMap<>(); + outExpected.put("k1", "xx"); + + assertEquals(outExpected, output); + } +} diff --git a/test-utils/src/main/java/com/swisscom/daisy/cosmos/candyfloss/testutils/JsonUtil.java b/test-utils/src/main/java/com/swisscom/daisy/cosmos/candyfloss/testutils/JsonUtil.java index 714edb8..5191993 100644 --- a/test-utils/src/main/java/com/swisscom/daisy/cosmos/candyfloss/testutils/JsonUtil.java +++ b/test-utils/src/main/java/com/swisscom/daisy/cosmos/candyfloss/testutils/JsonUtil.java @@ -2,6 +2,7 @@ import static com.swisscom.daisy.cosmos.candyfloss.transformations.jolt.CustomFunctions.factory; +import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import java.io.BufferedReader; import java.io.IOException; @@ -59,4 +60,17 @@ public static List> readJsonArray(URL url) throws IOExceptio } return readJsonArray(url.openStream()); } + + public static String getValueByJsonPath(JsonNode rootNode, String path) { + String[] keys = path.split("\\."); + JsonNode currentNode = rootNode; + + for (String k : keys) { + currentNode = currentNode.get(k); + if (currentNode == null) { + return null; + } + } + return currentNode.asText(); + } }