ref: change AS_SET check and move "move_up" to python

Author: KryptosKid

ris_live.py

@@ -32,15 +32,14 @@ async def process(message):
     data = orjson.loads(message)
 
     # as `path` may contain mixed data types Array(UInt32 | Array(Uint32)) (see https://ris-live.ripe.net/manual/)
-    # check `path`` for AS_SET (the array in the array) and pop it into it's own key
-    if "path" in data["data"]:
-        for idx, i in enumerate(data["data"]["path"]):
-            if type(i) == type(list()):
-                data["data"]["path"].pop(int(idx))
-                data["data"]["as_set"] = i
+    # check `path` for AS_SET (the array in the array) and pop it into it's own key
+    if "path" in data["data"] and data["data"]["path"]:
+        if type(data["data"]["path"][-1]) == type(list()):
+            data["data"]["as_set"] = data["data"]["path"][-1].copy()
+            data["data"]["path"].pop()
 
     # print and pipe into whatever
-    print(orjson.dumps(data).decode("utf-8"))
+    print(orjson.dumps(data["data"]).decode("utf-8"))
 
 
 if __name__ == "__main__":

vector/vector.toml.example

@@ -5,18 +5,11 @@ decoding.codec = "json"
 [transforms.update_only]
 type = "filter"
 inputs = ["stdin"]
-condition = '.data.type == "UPDATE"'
-
-[transforms.move_up]
-type = "remap"
-inputs = ["update_only"]
-source = """
-. = .data
-"""
+condition = '.type == "UPDATE"'
 
 [sinks.clickhouse_http]
 type = "http"
-inputs = ["move_up"]
+inputs = ["update_only"]
 encoding.codec = "json"
 uri = "http://localhost:8123/?query=INSERT%20INTO%20bgp.ris_live_queue%20%28data%29%20FORMAT%20JSONAsString"
 auth.strategy = "basic"