Fix custom HystrixConcurrencyStrategy/CommandExecutionHook doesn't wo… (#1901)

* Fix custom HystrixConcurrencyStrategy/CommandExecutionHook doesn't work when using hystrix plugin

* add unit-tests

* add licenses

apm-sniffer/apm-sdk-plugin/hystrix-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/hystrix/v1/HystrixConcurrencyStrategyInterceptor.java

@@ -19,8 +19,14 @@
 
 package org.apache.skywalking.apm.plugin.hystrix.v1;
 
+import com.netflix.hystrix.strategy.HystrixPlugins;
 import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy;
 import java.lang.reflect.Method;
+
+import com.netflix.hystrix.strategy.eventnotifier.HystrixEventNotifier;
+import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook;
+import com.netflix.hystrix.strategy.metrics.HystrixMetricsPublisher;
+import com.netflix.hystrix.strategy.properties.HystrixPropertiesStrategy;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
@@ -43,12 +49,37 @@ public class HystrixConcurrencyStrategyInterceptor implements InstanceMethodsAro
                     objInst.setSkyWalkingDynamicField(wrapperCache);
                 }
                 if (wrapperCache.getSwHystrixConcurrencyStrategyWrapper() == null) {
-                    wrapperCache.setSwHystrixConcurrencyStrategyWrapper(new SWHystrixConcurrencyStrategyWrapper((HystrixConcurrencyStrategy) ret));
+                    // Return and register wrapper only for the first time
+                    // Try to believe that all other strategies will use the their delegates
+                    SWHystrixConcurrencyStrategyWrapper wrapper = new SWHystrixConcurrencyStrategyWrapper((HystrixConcurrencyStrategy) ret);
+                    wrapperCache.setSwHystrixConcurrencyStrategyWrapper(wrapper);
+
+                    registerSWHystrixConcurrencyStrategyWrapper(wrapper);
+
+                    return wrapper;
+                }
             }
         }
+        return ret;
     }
 
-        return wrapperCache.getSwHystrixConcurrencyStrategyWrapper();
+    private void registerSWHystrixConcurrencyStrategyWrapper(SWHystrixConcurrencyStrategyWrapper wrapper) {
+        // Copy from Spring Cloud Sleuth
+        HystrixCommandExecutionHook commandExecutionHook = HystrixPlugins
+                .getInstance().getCommandExecutionHook();
+        HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance()
+                .getEventNotifier();
+        HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance()
+                .getMetricsPublisher();
+        HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance()
+                .getPropertiesStrategy();
+        HystrixPlugins.reset();
+        HystrixPlugins.getInstance().registerConcurrencyStrategy(wrapper);
+        HystrixPlugins.getInstance()
+                .registerCommandExecutionHook(commandExecutionHook);
+        HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
+        HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
+        HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
     }
 
     @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,

apm-sniffer/apm-sdk-plugin/hystrix-1.x-plugin/src/main/java/org/apache/skywalking/apm/plugin/hystrix/v1/HystrixPluginsInterceptor.java

@@ -19,8 +19,13 @@
 package org.apache.skywalking.apm.plugin.hystrix.v1;
 
 import com.netflix.hystrix.strategy.HystrixPlugins;
+import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy;
+import com.netflix.hystrix.strategy.eventnotifier.HystrixEventNotifier;
 import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook;
 import java.lang.reflect.Method;
+
+import com.netflix.hystrix.strategy.metrics.HystrixMetricsPublisher;
+import com.netflix.hystrix.strategy.properties.HystrixPropertiesStrategy;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.InstanceMethodsAroundInterceptor;
 import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInterceptResult;
@@ -32,6 +37,9 @@ import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.MethodInt
  * @author zhang xin
  */
 public class HystrixPluginsInterceptor implements InstanceMethodsAroundInterceptor {
+
+
+
     @Override
     public void beforeMethod(EnhancedInstance objInst, Method method, Object[] allArguments, Class<?>[] argumentsTypes,
         MethodInterceptResult result) throws Throwable {
@@ -48,12 +56,32 @@ public class HystrixPluginsInterceptor implements InstanceMethodsAroundIntercept
                     objInst.setSkyWalkingDynamicField(wrapperCache);
                 }
                 if (wrapperCache.getSwExecutionHookWrapper() == null) {
-                    wrapperCache.setSwExecutionHookWrapper(new SWExecutionHookWrapper((HystrixCommandExecutionHook) ret));
+                    // Return and register wrapper only for the first time
+                    // Try to believe that all other hooks will use the their delegates
+                    SWExecutionHookWrapper wrapper = new SWExecutionHookWrapper((HystrixCommandExecutionHook) ret);
+                    wrapperCache.setSwExecutionHookWrapper(wrapper);
+
+                    registerSWExecutionHookWrapper(wrapper);
+
+                    return wrapper;
                 }
             }
         }
 
-        return wrapperCache.getSwExecutionHookWrapper();
+        return ret;
+    }
+
+    private static void registerSWExecutionHookWrapper(SWExecutionHookWrapper wrapper) {
+        HystrixConcurrencyStrategy concurrencyStrategy = HystrixPlugins.getInstance().getConcurrencyStrategy();
+        HystrixEventNotifier eventNotifier = HystrixPlugins.getInstance().getEventNotifier();
+        HystrixMetricsPublisher metricsPublisher = HystrixPlugins.getInstance().getMetricsPublisher();
+        HystrixPropertiesStrategy propertiesStrategy = HystrixPlugins.getInstance().getPropertiesStrategy();
+        HystrixPlugins.reset();
+        HystrixPlugins.getInstance().registerConcurrencyStrategy(concurrencyStrategy);
+        HystrixPlugins.getInstance().registerCommandExecutionHook(wrapper);
+        HystrixPlugins.getInstance().registerEventNotifier(eventNotifier);
+        HystrixPlugins.getInstance().registerMetricsPublisher(metricsPublisher);
+        HystrixPlugins.getInstance().registerPropertiesStrategy(propertiesStrategy);
     }
 
     @Override public void handleMethodException(EnhancedInstance objInst, Method method, Object[] allArguments,

apm-sniffer/apm-sdk-plugin/hystrix-1.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/hystrix/v1/HystrixConcurrencyStrategyInterceptorTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.hystrix.v1;
+
+import com.netflix.hystrix.strategy.HystrixPlugins;
+import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.concurrent.Callable;
+
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public class HystrixConcurrencyStrategyInterceptorTest {
+
+    private HystrixConcurrencyStrategyInterceptor hystrixConcurrencyStrategyInterceptor;
+
+    private EnhancedInstance enhancedInstance;
+
+    @Before
+    public void setUp() {
+        hystrixConcurrencyStrategyInterceptor = new HystrixConcurrencyStrategyInterceptor();
+        enhancedInstance = new EnhancedInstance() {
+
+            private SWHystrixPluginsWrapperCache cache;
+
+            @Override
+            public Object getSkyWalkingDynamicField() {
+                return cache;
+            }
+
+            @Override public void setSkyWalkingDynamicField(Object cache) {
+                this.cache = (SWHystrixPluginsWrapperCache) cache;
+            }
+        };
+        HystrixPlugins.reset();
+    }
+
+    @Test
+    public void testSWHystrixConcurrencyStrategyWrapperWillBeRegistered() throws Throwable {
+        Object wrapperResult = getConcurrencyStrategyByInterceptor();
+        assertTrue(wrapperResult instanceof SWHystrixConcurrencyStrategyWrapper);
+        assertSame(wrapperResult, HystrixPlugins.getInstance().getConcurrencyStrategy());
+    }
+
+
+    @Test
+    public void testInterceptorWithCustomHystrixConcurrencyStrategy() throws Throwable {
+        Object wrapperResult = getConcurrencyStrategyByInterceptor();
+        assertTrue(wrapperResult instanceof SWHystrixConcurrencyStrategyWrapper);
+        assertSame(HystrixPlugins.getInstance().getConcurrencyStrategy(), wrapperResult);
+
+        // register custom HystrixConcurrencyStrategy
+        final HystrixConcurrencyStrategy delegate = getConcurrencyStrategyByInterceptor();
+        HystrixConcurrencyStrategy customConcurrencyStrategy = new CustomConcurrencyStrategy(delegate);
+        HystrixPlugins.reset();
+        HystrixPlugins.getInstance().registerConcurrencyStrategy(customConcurrencyStrategy);
+
+        // custom HystrixConcurrencyStrategy can be consumed
+        wrapperResult = getConcurrencyStrategyByInterceptor();
+        assertSame(customConcurrencyStrategy, wrapperResult);
+        assertSame(HystrixPlugins.getInstance().getConcurrencyStrategy(), wrapperResult);
+    }
+
+    private HystrixConcurrencyStrategy getConcurrencyStrategyByInterceptor() throws Throwable {
+        return (HystrixConcurrencyStrategy) hystrixConcurrencyStrategyInterceptor.afterMethod(enhancedInstance, null, null, null, HystrixPlugins.getInstance().getConcurrencyStrategy());
+    }
+
+    static class CustomConcurrencyStrategy extends HystrixConcurrencyStrategy {
+
+        private HystrixConcurrencyStrategy delegate;
+
+        public CustomConcurrencyStrategy(HystrixConcurrencyStrategy delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public <T> Callable<T> wrapCallable(Callable<T> callable) {
+            return delegate.wrapCallable(callable);
+        }
+    }
+}
\ No newline at end of file

apm-sniffer/apm-sdk-plugin/hystrix-1.x-plugin/src/test/java/org/apache/skywalking/apm/plugin/hystrix/v1/HystrixPluginsInterceptorTest.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.plugin.hystrix.v1;
+
+import com.netflix.hystrix.HystrixInvokable;
+import com.netflix.hystrix.strategy.HystrixPlugins;
+import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook;
+import org.apache.skywalking.apm.agent.core.plugin.interceptor.enhance.EnhancedInstance;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public class HystrixPluginsInterceptorTest {
+
+    private HystrixPluginsInterceptor hystrixPluginsInterceptor;
+
+    private EnhancedInstance enhancedInstance;
+
+    @Before
+    public void setUp() {
+        hystrixPluginsInterceptor = new HystrixPluginsInterceptor();
+        enhancedInstance = new EnhancedInstance() {
+
+            private SWHystrixPluginsWrapperCache cache;
+
+            @Override
+            public Object getSkyWalkingDynamicField() {
+                return cache;
+            }
+
+            @Override public void setSkyWalkingDynamicField(Object cache) {
+                this.cache = (SWHystrixPluginsWrapperCache) cache;
+            }
+        };
+        HystrixPlugins.reset();
+    }
+
+    @Test
+    public void testSWExecutionHookWrapperWillBeRegistered() throws Throwable {
+        Object wrapperResult = getCommandExecutionHookByInterceptor();
+        assertTrue(wrapperResult instanceof SWExecutionHookWrapper);
+        assertSame(wrapperResult, HystrixPlugins.getInstance().getCommandExecutionHook());
+    }
+
+
+    @Test
+    public void testInterceptorWithCustomHystrixCommandExecutionHook() throws Throwable {
+        Object wrapperResult = getCommandExecutionHookByInterceptor();
+        assertTrue(wrapperResult instanceof SWExecutionHookWrapper);
+        assertSame(HystrixPlugins.getInstance().getCommandExecutionHook(), wrapperResult);
+
+        // register custom HystrixCommandExecutionHook
+        HystrixCommandExecutionHook delegate = getCommandExecutionHookByInterceptor();
+        HystrixCommandExecutionHook customCommandExecutionHook = new CustomHystrixCommandExecutionHook(delegate);
+        HystrixPlugins.reset();
+        HystrixPlugins.getInstance().registerCommandExecutionHook(customCommandExecutionHook);
+
+        // custom HystrixCommandExecutionHook can be consumed
+        wrapperResult = getCommandExecutionHookByInterceptor();
+        assertSame(customCommandExecutionHook, wrapperResult);
+        assertSame(HystrixPlugins.getInstance().getCommandExecutionHook(), wrapperResult);
+    }
+
+    private HystrixCommandExecutionHook getCommandExecutionHookByInterceptor() throws Throwable {
+        return (HystrixCommandExecutionHook) hystrixPluginsInterceptor.afterMethod(enhancedInstance, null, null, null, HystrixPlugins.getInstance().getCommandExecutionHook());
+    }
+
+    static class CustomHystrixCommandExecutionHook extends HystrixCommandExecutionHook {
+
+        private HystrixCommandExecutionHook delegate;
+
+        public CustomHystrixCommandExecutionHook(HystrixCommandExecutionHook delegate) {
+            this.delegate = delegate;
+        }
+
+        @Override
+        public <T> void onStart(HystrixInvokable<T> commandInstance) {
+            delegate.onStart(commandInstance);
+        }
+    }
+
+}
\ No newline at end of file