Fix latency/throughput tracker issues

siddhi-io · Mar 28, 2022 · aa1c1dc · aa1c1dc
1 parent 8330c51
commit aa1c1dc
Show file tree

Hide file tree

Showing 4 changed files with 76 additions and 56 deletions.
diff --git a/modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/AggregationRuntime.java b/modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/AggregationRuntime.java
@@ -341,10 +341,13 @@ public StreamEvent find(StateEvent matchingEvent, CompiledCondition compiledCond
                             SiddhiQueryContext siddhiQueryContext) {
         try {
             SnapshotService.getSkipStateStorageThreadLocal().set(true);
+            if (throughputTrackerFind != null &&
+                    Level.BASIC.compareTo(siddhiQueryContext.getSiddhiAppContext().getRootMetricsLevel()) <= 0) {
+                throughputTrackerFind.eventIn();
+            }
             if (latencyTrackerFind != null &&
                     Level.BASIC.compareTo(siddhiQueryContext.getSiddhiAppContext().getRootMetricsLevel()) <= 0) {
                 latencyTrackerFind.markIn();
-                throughputTrackerFind.eventIn();
             }
             if (!isDistributed && !isFirstEventArrived) {
                 // No need to initialise executors if it is distributed

diff --git a/modules/siddhi-core/src/main/java/io/siddhi/core/stream/input/source/SourceMapper.java b/modules/siddhi-core/src/main/java/io/siddhi/core/stream/input/source/SourceMapper.java
@@ -194,7 +194,7 @@ public final void onEvent(Object eventObject, Object[] transportProperties, Stri
                             Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                         throughputTracker.eventIn();
                     }
-                    if (throughputTracker != null &&
+                    if (mapperLatencyTracker != null &&
                             Level.DETAIL.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                         mapperLatencyTracker.markIn();
                     }
@@ -207,7 +207,7 @@ public final void onEvent(Object eventObject, Object[] transportProperties, Stri
                         receivedEventCounter.countEvents(eventObject);
                     }
                 } finally {
-                    if (throughputTracker != null &&
+                    if (mapperLatencyTracker != null &&
                             Level.DETAIL.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                         mapperLatencyTracker.markOut();
                     }

diff --git a/modules/siddhi-core/src/main/java/io/siddhi/core/table/Table.java b/modules/siddhi-core/src/main/java/io/siddhi/core/table/Table.java
@@ -216,19 +216,22 @@ private void handleStoreAddError(ComplexEventChunk addingEventChunk, boolean isF
     }
 
     public void addEvents(ComplexEventChunk<StreamEvent> addingEventChunk, int noOfEvents) {
-        if (latencyTrackerInsert != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerInsert.markIn();
-        }
-        addingEventChunk.reset();
-        add(addingEventChunk);
-        if (latencyTrackerInsert != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerInsert.markOut();
-        }
-        if (throughputTrackerInsert != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            throughputTrackerInsert.eventsIn(noOfEvents);
+        try {
+            if (throughputTrackerInsert != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                throughputTrackerInsert.eventsIn(noOfEvents);
+            }
+            if (latencyTrackerInsert != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerInsert.markIn();
+            }
+            addingEventChunk.reset();
+            add(addingEventChunk);
+        } finally {
+            if (latencyTrackerInsert != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerInsert.markOut();
+            }
         }
     }
 
@@ -335,18 +338,21 @@ protected void onDeleteError(ComplexEventChunk<StateEvent> deletingEventChunk, C
 
     public void deleteEvents(ComplexEventChunk<StateEvent> deletingEventChunk, CompiledCondition compiledCondition,
                              int noOfEvents) {
-        if (latencyTrackerDelete != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerDelete.markIn();
-        }
-        delete(deletingEventChunk, compiledCondition);
-        if (latencyTrackerDelete != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerDelete.markOut();
-        }
-        if (throughputTrackerDelete != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            throughputTrackerDelete.eventsIn(noOfEvents);
+        try {
+            if (throughputTrackerDelete != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                throughputTrackerDelete.eventsIn(noOfEvents);
+            }
+            if (latencyTrackerDelete != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerDelete.markIn();
+            }
+            delete(deletingEventChunk, compiledCondition);
+        } finally {
+            if (latencyTrackerDelete != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerDelete.markOut();
+            }
         }
     }
 
@@ -418,18 +424,21 @@ protected void onUpdateError(ComplexEventChunk<StateEvent> updatingEventChunk, C
     public void updateEvents(ComplexEventChunk<StateEvent> updatingEventChunk,
                              CompiledCondition compiledCondition,
                              CompiledUpdateSet compiledUpdateSet, int noOfEvents) {
-        if (latencyTrackerUpdate != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerUpdate.markIn();
-        }
-        update(updatingEventChunk, compiledCondition, compiledUpdateSet);
-        if (latencyTrackerUpdate != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerUpdate.markOut();
-        }
-        if (throughputTrackerUpdate != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            throughputTrackerUpdate.eventsIn(noOfEvents);
+        try {
+            if (throughputTrackerUpdate != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                throughputTrackerUpdate.eventsIn(noOfEvents);
+            }
+            if (latencyTrackerUpdate != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerUpdate.markIn();
+            }
+            update(updatingEventChunk, compiledCondition, compiledUpdateSet);
+        } finally {
+            if (latencyTrackerUpdate != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerUpdate.markOut();
+            }
         }
     }
 
@@ -506,19 +515,22 @@ public void updateOrAddEvents(ComplexEventChunk<StateEvent> updateOrAddingEventC
                                   CompiledCondition compiledCondition,
                                   CompiledUpdateSet compiledUpdateSet,
                                   AddingStreamEventExtractor addingStreamEventExtractor, int noOfEvents) {
-        if (latencyTrackerUpdateOrInsert != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerUpdateOrInsert.markIn();
-        }
-        updateOrAdd(updateOrAddingEventChunk, compiledCondition, compiledUpdateSet,
-                addingStreamEventExtractor);
-        if (latencyTrackerUpdateOrInsert != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            latencyTrackerUpdateOrInsert.markOut();
-        }
-        if (throughputTrackerUpdateOrInsert != null &&
-                Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
-            throughputTrackerUpdateOrInsert.eventsIn(noOfEvents);
+        try {
+            if (throughputTrackerUpdateOrInsert != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                throughputTrackerUpdateOrInsert.eventsIn(noOfEvents);
+            }
+            if (latencyTrackerUpdateOrInsert != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerUpdateOrInsert.markIn();
+            }
+            updateOrAdd(updateOrAddingEventChunk, compiledCondition, compiledUpdateSet,
+                    addingStreamEventExtractor);
+        } finally {
+            if (latencyTrackerUpdateOrInsert != null &&
+                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+                latencyTrackerUpdateOrInsert.markOut();
+            }
         }
     }
 

diff --git a/modules/siddhi-core/src/main/java/io/siddhi/core/window/Window.java b/modules/siddhi-core/src/main/java/io/siddhi/core/window/Window.java
@@ -238,12 +238,15 @@ public void add(ComplexEventChunk complexEventChunk) {
                 if (throughputTrackerInsert != null &&
                         Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                     throughputTrackerInsert.eventsIn(numberOfEvents);
+                }
+                if (latencyTrackerInsert != null &&
+                        Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                     latencyTrackerInsert.markIn();
                 }
                 // Send to the window windowProcessor
                 windowProcessor.process(new ComplexEventChunk<>(firstEvent, currentEvent));
             } finally {
-                if (throughputTrackerInsert != null &&
+                if (latencyTrackerInsert != null &&
                         Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                     latencyTrackerInsert.markOut();
                 }
@@ -261,11 +264,13 @@ public StreamEvent find(StateEvent matchingEvent, CompiledCondition compiledCond
         try {
             if (throughputTrackerFind != null && Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                 throughputTrackerFind.eventIn();
+            }
+            if (latencyTrackerFind != null && Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                 latencyTrackerFind.markIn();
             }
             return ((FindableProcessor) this.internalWindowProcessor).find(matchingEvent, compiledCondition);
         } finally {
-            if (throughputTrackerFind != null && Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
+            if (latencyTrackerFind != null && Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                 latencyTrackerFind.markOut();
             }
         }
@@ -330,7 +335,7 @@ private class StreamPublishProcessor implements Processor {
         }
 
         public void process(ComplexEventChunk complexEventChunk) {
-            if (throughputTrackerInsert != null &&
+            if (latencyTrackerInsert != null &&
                     Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                 latencyTrackerInsert.markOut();
             }