We must handle all exceptions while intercepting ES requests (#99)

* Making sure that we don't throw exceptions while intercepting ES requests PerformanceAnalyzer intercepts various ES request paths toget detailed metrics. But today if we throw an exception, then it will bubble all the way upto ES and fail the request. * Addressing the PR comments * Updating the .gitignore * style changes
opendistro-for-elasticsearch · May 15, 2020 · 8b6e446 · 8b6e446
1 parent 5db7fc8
commit 8b6e446
Show file tree

Hide file tree

Showing 5 changed files with 67 additions and 13 deletions.
diff --git a/.gitignore b/.gitignore
@@ -1,2 +1,3 @@
 .idea/
 *.iml
+licenses/performanceanalyzer-rca-1.3.jar.sha1s
diff --git a/licenses/performanceanalyzer-rca-1.3.jar.sha1 b/licenses/performanceanalyzer-rca-1.3.jar.sha1
@@ -1 +1 @@
-67a91489b55ec6cd563c823e6790a8909bd334c3
+81381772c2fea1f20c8d4050b70e11c36e881752
diff --git a/...n/opendistro/elasticsearch/performanceanalyzer/config/setting/ClusterSettingsManager.java b/...n/opendistro/elasticsearch/performanceanalyzer/config/setting/ClusterSettingsManager.java
@@ -1,5 +1,7 @@
 package com.amazon.opendistro.elasticsearch.performanceanalyzer.config.setting;
 
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.collectors.StatExceptionCode;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.collectors.StatsCollector;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -165,11 +167,16 @@ public void clusterChanged(final ClusterChangedEvent event) {
      * @param settingValue The new value for the setting.
      */
     private void callListeners(final Setting<Integer> setting, int settingValue) {
-        final List<ClusterSettingListener<Integer>> listeners = listenerMap.get(setting);
-        if (listeners != null) {
-            for (ClusterSettingListener<Integer> listener : listeners) {
-                listener.onSettingUpdate(settingValue);
+        try {
+            final List<ClusterSettingListener<Integer>> listeners = listenerMap.get(setting);
+            if (listeners != null) {
+                for (ClusterSettingListener<Integer> listener : listeners) {
+                    listener.onSettingUpdate(settingValue);
+                }
             }
+        } catch(Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
         }
     }
 

diff --git a/...ndistro/elasticsearch/performanceanalyzer/listener/PerformanceAnalyzerSearchListener.java b/...ndistro/elasticsearch/performanceanalyzer/listener/PerformanceAnalyzerSearchListener.java
@@ -15,6 +15,8 @@
 
 package com.amazon.opendistro.elasticsearch.performanceanalyzer.listener;
 
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.collectors.StatExceptionCode;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.collectors.StatsCollector;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.index.shard.SearchOperationListener;
@@ -51,32 +53,62 @@ private SearchListener getSearchListener() {
 
     @Override
     public void onPreQueryPhase(SearchContext searchContext) {
-        getSearchListener().preQueryPhase(searchContext);
+        try {
+            getSearchListener().preQueryPhase(searchContext);
+        } catch (Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
+        }
     }
 
     @Override
     public void onQueryPhase(SearchContext searchContext, long tookInNanos) {
-        getSearchListener().queryPhase(searchContext, tookInNanos);
+        try {
+            getSearchListener().queryPhase(searchContext, tookInNanos);
+        } catch (Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
+        }
     }
 
     @Override
     public void onFailedQueryPhase(SearchContext searchContext) {
-        getSearchListener().failedQueryPhase(searchContext);
+        try {
+            getSearchListener().failedQueryPhase(searchContext);
+        } catch (Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
+        }
     }
 
     @Override
     public void onPreFetchPhase(SearchContext searchContext) {
-        getSearchListener().preFetchPhase(searchContext);
+        try {
+            getSearchListener().preFetchPhase(searchContext);
+        } catch (Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
+        }
     }
 
     @Override
     public void onFetchPhase(SearchContext searchContext, long tookInNanos) {
-        getSearchListener().fetchPhase(searchContext, tookInNanos);
+        try {
+            getSearchListener().fetchPhase(searchContext, tookInNanos);
+        } catch (Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
+        }
     }
 
     @Override
     public void onFailedFetchPhase(SearchContext searchContext) {
-        getSearchListener().failedFetchPhase(searchContext);
+        try {
+            getSearchListener().failedFetchPhase(searchContext);
+        } catch (Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
+        }
     }
 
     @Override

diff --git a/...asticsearch/performanceanalyzer/transport/PerformanceAnalyzerTransportRequestHandler.java b/...asticsearch/performanceanalyzer/transport/PerformanceAnalyzerTransportRequestHandler.java
@@ -15,6 +15,8 @@
 
 package com.amazon.opendistro.elasticsearch.performanceanalyzer.transport;
 
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.collectors.StatExceptionCode;
+import com.amazon.opendistro.elasticsearch.performanceanalyzer.collectors.StatsCollector;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.action.bulk.BulkShardRequest;
@@ -43,7 +45,7 @@ PerformanceAnalyzerTransportRequestHandler<T> set(TransportRequestHandler<T> act
 
     @Override
     public void messageReceived(T request, TransportChannel channel, Task task) throws Exception {
-        actualHandler.messageReceived(request, getChannel(request, channel, task) , task);
+        actualHandler.messageReceived(request, getChannel(request, channel, task), task);
     }
 
     private TransportChannel getChannel(T request, TransportChannel channel, Task task) {
@@ -78,7 +80,19 @@ private TransportChannel getShardBulkChannel(T request, TransportChannel channel
 
         BulkShardRequest bsr = (BulkShardRequest) transportRequest;
         PerformanceAnalyzerTransportChannel performanceanalyzerChannel = new PerformanceAnalyzerTransportChannel();
-        performanceanalyzerChannel.set(channel, System.currentTimeMillis(), bsr.index(), bsr.shardId().id(), bsr.items().length, bPrimary);
+
+        try {
+            performanceanalyzerChannel.set(
+                    channel,
+                    System.currentTimeMillis(),
+                    bsr.index(),
+                    bsr.shardId().id(),
+                    bsr.items().length,
+                    bPrimary);
+        } catch (Exception ex) {
+            LOG.error(ex);
+            StatsCollector.instance().logException(StatExceptionCode.ES_REQUEST_INTERCEPTOR_ERROR);
+        }
 
         return performanceanalyzerChannel;
     }
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		67a91489b55ec6cd563c823e6790a8909bd334c3
		81381772c2fea1f20c8d4050b70e11c36e881752