Feature/ietf-telemetry-message-format-adapt

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 {

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<String, Object> 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")) {

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) {

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
+    }
+  }
+}

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": {
+        "*": "&"
+      }
+    }
+  ]
+}

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"
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+]

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)

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<Arguments> 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<Arguments> providerForYangModels()
+  protected static Stream<Arguments> 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<Arguments> 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));
         }
       }
     }

candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-input.json

@@ -0,0 +1 @@
+{}

candyfloss/src/test/resources/deployment-ietf-telemetry-message/fano-interfaces/statistics/01-output.json

@@ -0,0 +1 @@
+[]

json-transformation/src/main/java/com/swisscom/daisy/cosmos/candyfloss/transformations/jolt/CustomFunctions.java

@@ -42,6 +42,14 @@ protected Optional<Object> applySingle(Object arg) {
     }
   }
 
+  public static final class getSysProperty extends Function.SingleFunction<String> {
+    @Override
+    protected Optional<String> applySingle(Object arg) {
+      String prop = (String) arg;
+      return Optional.of(System.getProperty(prop));
+    }
+  }
+
   public static final class multiply extends Function.ListFunction {
     @Override
     protected Optional<Object> applyList(List<Object> argList) {

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<String, Function> functionsMap) {

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<String, Object> 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<String, Object> 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<String, Object> outExpected = new HashMap<>();
+    outExpected.put("k1", "xx");
+
+    assertEquals(outExpected, output);
+  }
+}

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<Map<String, Object>> 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();
+  }
 }