Support for tracing the callbacks of async methods in elastiicsearch-…

…6.x/7.x-plugin (#694)
apache · May 27, 2024 · f736b37 · f736b37
1 parent ffbd90c
commit f736b37
Show file tree

Hide file tree

Showing 32 changed files with 1,223 additions and 160 deletions.
diff --git a/CHANGES.md b/CHANGES.md
@@ -16,6 +16,7 @@ Release Notes.
 * Fix NPE in Redisson plugin since Redisson 3.20.0.
 * Support for showing batch command details and ignoring PING commands in Redisson plugin.
 * Fix peer value of Master-Slave mode in Redisson plugin.
+* Support for tracing the callbacks of asynchronous methods in elasticsearch-6.x-plugin/elasticsearch-7.x-plugin.
 
 All issues and pull requests are [here](https://github.com/apache/skywalking/milestone/213?closed=1)
 

diff --git a/...rg/apache/skywalking/apm/plugin/elasticsearch/v6/define/IndicesClientInstrumentation.java b/...rg/apache/skywalking/apm/plugin/elasticsearch/v6/define/IndicesClientInstrumentation.java
@@ -68,7 +68,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -85,7 +85,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -102,7 +102,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
                 new InstanceMethodsInterceptPoint() {
@@ -119,7 +119,7 @@ public String getMethodsInterceptor() {
 
                     @Override
                     public boolean isOverrideArgs() {
-                        return false;
+                        return true;
                     }
                 }
         };

diff --git a/...che/skywalking/apm/plugin/elasticsearch/v6/define/RestHighLevelClientInstrumentation.java b/...che/skywalking/apm/plugin/elasticsearch/v6/define/RestHighLevelClientInstrumentation.java
@@ -86,7 +86,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -102,7 +102,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -118,7 +118,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -134,7 +134,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -182,7 +182,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -198,7 +198,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -214,7 +214,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             },
             new InstanceMethodsInterceptPoint() {
@@ -230,7 +230,7 @@ public String getMethodsInterceptor() {
 
                 @Override
                 public boolean isOverrideArgs() {
-                    return false;
+                    return true;
                 }
             }
         };

diff --git a/...rc/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v6/interceptor/Constants.java b/...rc/main/java/org/apache/skywalking/apm/plugin/elasticsearch/v6/interceptor/Constants.java
@@ -79,4 +79,7 @@ public class Constants {
     public static final AbstractTag<String> ES_TOOK_MILLIS = Tags.ofKey("es.took_millis");
     public static final AbstractTag<String> ES_TOTAL_HITS = Tags.ofKey("es.total_hits");
     public static final AbstractTag<String> ES_INGEST_TOOK_MILLIS = Tags.ofKey("es.ingest_took_millis");
+
+    public static final String ON_RESPONSE_SUFFIX = "/onResponse";
+    public static final String ON_FAILURE_SUFFIX = "/onFailure";
 }
diff --git a/...lking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientAnalyzeMethodsInterceptor.java b/...lking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientAnalyzeMethodsInterceptor.java
@@ -30,6 +30,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.analyze.AnalyzeRequest;
 
 import java.lang.reflect.Method;
@@ -58,6 +60,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
                 Tags.DB_STATEMENT.set(span, analyzeRequest.text()[0]);
             }
             SpanLayer.asDB(span);
+            if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+                allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.ANALYZE_OPERATOR_NAME,
+                                                                 (ActionListener) allArguments[2]);
+            }
         }
     }
 

diff --git a/...alking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientCreateMethodsInterceptor.java b/...alking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientCreateMethodsInterceptor.java
@@ -28,6 +28,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.client.indices.CreateIndexRequest;
 
 import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.TRACE_DSL;
@@ -52,6 +54,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
                 Tags.DB_STATEMENT.set(span, createIndexRequest.mappings().utf8ToString());
             }
             SpanLayer.asDB(span);
+            if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+                allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.CREATE_OPERATOR_NAME,
+                                                                 (ActionListener) allArguments[2]);
+            }
         }
     }
 

diff --git a/...alking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientDeleteMethodsInterceptor.java b/...alking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientDeleteMethodsInterceptor.java
@@ -29,6 +29,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
 
 import static org.apache.skywalking.apm.plugin.elasticsearch.v6.interceptor.Constants.DB_TYPE;
@@ -48,6 +50,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
             Tags.DB_TYPE.set(span, DB_TYPE);
             Tags.DB_INSTANCE.set(span, Arrays.asList(deleteIndexRequest.indices()).toString());
             SpanLayer.asDB(span);
+            if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+                allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.DELETE_OPERATOR_NAME,
+                                                                 (ActionListener) allArguments[2]);
+            }
         }
     }
 

diff --git a/...lking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientRefreshMethodsInterceptor.java b/...lking/apm/plugin/elasticsearch/v6/interceptor/IndicesClientRefreshMethodsInterceptor.java
@@ -27,6 +27,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
 
 import java.lang.reflect.Method;
@@ -48,6 +50,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
             Tags.DB_TYPE.set(span, DB_TYPE);
             Tags.DB_INSTANCE.set(span, buildIndicesString(request.indices()));
             SpanLayer.asDB(span);
+            if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+                allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.REFRESH_OPERATOR_NAME,
+                                                                 (ActionListener) allArguments[2]);
+            }
         }
     }
 

diff --git a/...plugin/elasticsearch/v6/interceptor/RestHighLevelClientClearScrollMethodsInterceptor.java b/...plugin/elasticsearch/v6/interceptor/RestHighLevelClientClearScrollMethodsInterceptor.java
@@ -27,6 +27,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.ClearScrollRequest;
 
 import java.lang.reflect.Method;
@@ -52,6 +54,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.CLEAR_SCROLL_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override

diff --git a/...ugin/elasticsearch/v6/interceptor/RestHighLevelClientDeleteByQueryMethodsInterceptor.java b/...ugin/elasticsearch/v6/interceptor/RestHighLevelClientDeleteByQueryMethodsInterceptor.java
@@ -27,6 +27,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.index.reindex.DeleteByQueryRequest;
 
 import java.lang.reflect.Method;
@@ -58,6 +60,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.DELETE_BY_QUERY_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override

diff --git a/...ing/apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientGetMethodsInterceptor.java b/...ing/apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientGetMethodsInterceptor.java
@@ -28,6 +28,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.get.GetRequest;
 
 import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.TRACE_DSL;
@@ -51,6 +53,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.GET_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override

diff --git a/...g/apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientIndexMethodsInterceptor.java b/...g/apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientIndexMethodsInterceptor.java
@@ -28,6 +28,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.index.IndexRequest;
 
 import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.TRACE_DSL;
@@ -51,6 +53,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.INDEX_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override

diff --git a/.../apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientSearchMethodsInterceptor.java b/.../apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientSearchMethodsInterceptor.java
@@ -29,6 +29,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchRequest;
 
 import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.TRACE_DSL;
@@ -52,6 +54,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.SEARCH_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override

diff --git a/...lugin/elasticsearch/v6/interceptor/RestHighLevelClientSearchScrollMethodsInterceptor.java b/...lugin/elasticsearch/v6/interceptor/RestHighLevelClientSearchScrollMethodsInterceptor.java
@@ -27,6 +27,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.search.SearchScrollRequest;
 
 import java.lang.reflect.Method;
@@ -51,6 +53,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.SEARCH_SCROLL_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override

diff --git a/...gin/elasticsearch/v6/interceptor/RestHighLevelClientSearchTemplateMethodsInterceptor.java b/...gin/elasticsearch/v6/interceptor/RestHighLevelClientSearchTemplateMethodsInterceptor.java
@@ -27,6 +27,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.script.mustache.SearchTemplateRequest;
 
 import java.lang.reflect.Method;
@@ -55,6 +57,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.SEARCH_TEMPLATE_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override

diff --git a/.../apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientUpdateMethodsInterceptor.java b/.../apm/plugin/elasticsearch/v6/interceptor/RestHighLevelClientUpdateMethodsInterceptor.java
@@ -28,6 +28,8 @@
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
 import org.apache.skywalking.apm.network.trace.component.ComponentsDefine;
 import org.apache.skywalking.apm.plugin.elasticsearch.common.RestClientEnhanceInfo;
+import org.apache.skywalking.apm.plugin.elasticsearch.v6.support.AdapterUtil;
+import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.update.UpdateRequest;
 
 import static org.apache.skywalking.apm.plugin.elasticsearch.v6.ElasticsearchPluginConfig.Plugin.Elasticsearch.TRACE_DSL;
@@ -51,6 +53,10 @@ public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allAr
         }
 
         SpanLayer.asDB(span);
+        if (allArguments.length > 2 && allArguments[2] != null && allArguments[2] instanceof ActionListener) {
+            allArguments[2] = AdapterUtil.wrapActionListener(restClientEnhanceInfo, Constants.UPDATE_OPERATOR_NAME,
+                                                             (ActionListener) allArguments[2]);
+        }
     }
 
     @Override