fix(#2166): sanitize event keys in influx sink (#2172)

* implement new round processor * add English locale, icon, and documentation * fix checkstyle * support different rounding modes * add rounding mode in documentation * fix time display * let NaryMapping selection account for property scope * implement boolean filter unit tests * add common StoreEventCollector class and refactor TestChangedValueDetectionProcessor * add new class * show associated pipelines' names and allow one click deletion * center text * fix minor error * replace magic number * add timeout * restore newline * changeb baseurl * revert port * revert timeout * implement pipelines owner check * undo automatic changes * enable admin to delete pipelines no matter ownership * sanitize event * add newline back * fix iter is on a copy --------- Co-authored-by: bossenti <[email protected]>
apache · Nov 14, 2023 · 7830506 · 7830506
1 parent 18cdb2d
commit 7830506
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 0 deletions.
diff --git a/...commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java b/...commons/src/main/java/org/apache/streampipes/dataexplorer/commons/influx/InfluxStore.java
@@ -121,6 +121,13 @@ public void onEvent(Event event) throws SpRuntimeException {
       throw new SpRuntimeException("event is null");
     }
 
+    // sanitize event
+    for (String key : event.getRaw().keySet()) {
+      if (InfluxDbReservedKeywords.KEYWORD_LIST.stream().anyMatch(k -> k.equalsIgnoreCase(key))) {
+        event.renameFieldByRuntimeName(key, key + "_");
+      }
+    }
+
     Long timestampValue = event.getFieldBySelector(measure.getTimestampField()).getAsPrimitive().getAsLong();
     Point.Builder point =
         Point.measurement(measure.getMeasureName()).time((long) timestampValue, TimeUnit.MILLISECONDS);

diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/Event.java b/streampipes-model/src/main/java/org/apache/streampipes/model/runtime/Event.java
@@ -119,6 +119,14 @@ public void updateFieldBySelector(String selector, AbstractField field) {
     }
   }
 
+  public void renameFieldByRuntimeName(String oldRuntimeName, String newRuntimeName) {
+    AbstractField field = getFieldByRuntimeName(oldRuntimeName);
+    String selector = makeKey(field);
+    removeFieldBySelector(selector);
+    field.rename(newRuntimeName);
+    addField(field);
+  }
+
   private void updateFieldMap(Map<String, AbstractField> currentFieldMap,
                               String selector, Integer position,
                               AbstractField field) {