Repository: pinpoint-apm/pinpoint
Issue: Can we trace the standlone thrift RPC server?
Labels: help wanted

Hi All,

First of all, Thank you so much for the lovely tool. I have gone through it and found really interesting.

I have checked this by configuring agent on Tomcat and it's showing more than I expected.

Now, The question is :- Can we trace the  thrift RPC client and server? 

I look forward for your response.

Cheers,
duwupeng

2016-09-26 17:49:03 [DEBUG](c.n.p.b.i.s.ScopedInterceptor      :298) tryBefore() returns false: interceptorScopeTransaction: InterceptorScopeInvocation(ThriftClientScope)[depth=0], executionPoint: INTERNAL. Skip interceptor class com.navercorp.pinpoint.plugin.thrift.interceptor.tprotocol.client.TProtocolWriteFieldStopInterceptor
2016-09-26 17:49:03 [DEBUG](c.n.p.b.i.s.ScopedInterceptor      :298) tryAfter() returns false: interceptorScopeTransaction: InterceptorScopeInvocation(ThriftClientScope)[depth=0], executionPoint: INTERNAL. Skip interceptor class com.navercorp.pinpoint.plugin.thrift.interceptor.tprotocol.client.TProtocolWriteFieldStopInterceptor


Issue comments:
**The thrift plugin has been enabled , while it can not trace the data.** 

profiler.thrift.client=true
profiler.thrift.client.async=true
profiler.thrift.processor=true
profiler.thrift.processor.async=true
profiler.thrift.service.args=true
profiler.thrift.service.result=true

Hi duwupeng,
Could you give a short explanation on how you defined your processors? Are you using the default processor auto-generated from the IDL?

Hell Xylus ,
Thanks for four reply。
 As you said , we defined the processors using the thrift Compiler (0.9.2) from IDL, and the services has been brought up by the TThreadedSelectorServer in a java main process.  

Hmm..that's weird, will definitely need to look into it further.
Could you post the agent log please? It should be under '$AGENT_DIR/logs' directory.
Thanks!

sure.  The  the agent log and  processor are as below. 

**1. The agent log:**

```
2016-09-26 18:55:56 [DEBUG](c.n.p.p.t.i.s.ProcessFunctionProcessInterceptor:96 ) BEFORE com.kuaisu.infra.thrift.service.HeartBeatService$Processor$hello  args:(1, StoredMessageProtocol, TCompactProtocol, HeartBeatServiceImpl)
2016-09-26 18:55:56 [DEBUG](c.n.p.p.t.i.s.ProcessFunctionProcessInterceptor:288) Invalid target object. Need field accessor(com.navercorp.pinpoint.plugin.thrift.field.accessor.ServerMarkerFlagFieldAccessor).
2016-09-26 18:55:56 [DEBUG](c.n.p.p.t.i.t.s.TProtocolReadFieldBeginInterceptor:121) AFTER org.apache.thrift.protocol.TCompactProtocol  args:() result:TField
2016-09-26 18:55:56 [DEBUG](c.n.p.p.t.i.t.s.TProtocolReadMessageEndInterceptor:121) AFTER org.apache.thrift.protocol.TCompactProtocol  args:() result:null
2016-09-26 18:55:56 [DEBUG](c.n.p.b.i.s.ScopedInterceptor      :298) tryBefore() returns false: interceptorScopeTransaction: InterceptorScopeInvocation(ThriftClientScope)[depth=0], executionPoint: INTERNAL. Skip interceptor class com.navercorp.pinpoint.plugin.thrift.interceptor.tprotocol.client.TProtocolWriteFieldStopInterceptor
2016-09-26 18:55:56 [DEBUG](c.n.p.b.i.s.ScopedInterceptor      :298) tryAfter() returns false: interceptorScopeTransaction: InterceptorScopeInvocation(ThriftClientScope)[depth=0], executionPoint: INTERNAL. Skip interceptor class com.navercorp.pinpoint.plugin.thrift.interceptor.tprotocol.client.TProtocolWriteFieldStopInterceptor
2016-09-26 18:55:56 [DEBUG](c.n.p.p.t.i.s.ProcessFunctionProcessInterceptor:121) AFTER com.kuaisu.infra.thrift.service.HeartBeatService$Processor$hello  args:(1, StoredMessageProtocol, TCompactProtocol, HeartBeatServiceImpl) result:null
2016-09-26 18:55:56 [DEBUG](c.n.p.b.i.s.ScopedInterceptor      :298) tryBefore() returns false: interceptorScopeTransaction: InterceptorScopeInvocation(ThriftServerScope)[depth=0], executionPoint: INTERNAL. Skip interceptor class com.navercorp.pinpoint.plugin.thrift.interceptor.tprotocol.server.TProtocolReadMessageBeginInterceptor
2016-09-26 18:55:56 [DEBUG](c.n.p.b.i.s.ScopedInterceptor      :298) tryAfter() returns false: interceptorScopeTransaction: InterceptorScopeInvocation(ThriftServerScope)[depth=0], executionPoint: INTERNAL. Skip interceptor class com.navercorp.pinpoint.plugin.thrift.interceptor.tprotocol.server.TProtocolReadMessageBeginInterceptor
```

**2.  The  processor ：**
 /**
- Autogenerated by Thrift Compiler (0.9.2)
  *
- DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
-  @generated
  */
  package com.kuaisu.infra.thrift.service;

import org.apache.thrift.TException;
import org.apache.thrift.async.AsyncMethodCallback;
import org.apache.thrift.protocol.TTupleProtocol;
import org.apache.thrift.scheme.IScheme;
import org.apache.thrift.scheme.SchemeFactory;
import org.apache.thrift.scheme.StandardScheme;
import org.apache.thrift.scheme.TupleScheme;
import org.apache.thrift.server.AbstractNonblockingServer.AsyncFrameBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Generated;
import java.util.*;

@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
@Generated(value = "Autogenerated by Thrift Compiler (0.9.2)", date = "2015-5-27")
public class HeartBeatService {

  /**
- Common interface
  */
  public interface Iface {
  
    /**
  - Heart Beart
    */
    public void hello() throws org.apache.thrift.TException;
  
  }
  
  public interface AsyncIface {
  
    public void hello(org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException;
  
  }
  
  public static class Client extends org.apache.thrift.TServiceClient implements Iface {
    public static class Factory implements org.apache.thrift.TServiceClientFactory<Client> {
      public Factory() {}
      public Client getClient(org.apache.thrift.protocol.TProtocol prot) {
        return new Client(prot);
      }
      public Client getClient(org.apache.thrift.protocol.TProtocol iprot, org.apache.thrift.protocol.TProtocol oprot) {
        return new Client(iprot, oprot);
      }
    }
  ...

@duwupeng It looks like you're using `TTupleProtocol`, and unfortunately I don't think the Thrift plugin supports tracing it (I think only the `TBinaryProtocol`, `TCompactProtocol`, and `TJSONProtocol` are supported currently). There probably was a reason why I left `TTupleProtocol` out but I don't quite remember exactly why as it's been a while since I developed it. I'll look into this so that I can give you a better answer.

In the mean time, you could try adding `TTupleProtocol` support yourself if you're building from source (master).
Simply add `addTProtocolInterceptors(config, "org.apache.thrift.protocol.TTupleProtocol");` to [ThriftPlugin.java](https://github.com/naver/pinpoint/blob/master/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftPlugin.java#L420) and build the _plugins_ and _agent_ module. Then use this agent and see if it works.

@Xylus  ，  Thanks a lot for you investigation . and I am afraid that I am using the TCompactProtocol
The following is my app server . 

```
import org.apache.thrift.TMultiplexedProcessor;
import org.apache.thrift.TProcessor;
import org.apache.thrift.protocol.TCompactProtocol;
import org.apache.thrift.protocol.TProtocolFactory;
import org.apache.thrift.server.TServer;
import org.apache.thrift.server.TThreadedSelectorServer;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TNonblockingServerSocket;
import org.apache.thrift.transport.TNonblockingServerTransport;
import org.apache.thrift.transport.TTransportFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class AppServer {
    protected static ClassPathXmlApplicationContext context;
    public static void main(String[] args) {
        String logDir = System.getProperty("log.dir");
        if (logDir == null) {
            System.setProperty("log.dir", ".");
        }
        final Logger logger = LoggerFactory.getLogger(AppServer.class);
        PropertyConfigurator.configure(AppServer.class.getClassLoader().getResource("logback.xml"));
        context = new ClassPathXmlApplicationContext("spring/root.xml");
        SpringInstanceProvider provider = new SpringInstanceProvider(context);
        InstanceFactory.setInstanceProvider(provider);

        Integer port = (Integer) context.getBean("port");
        System.setProperty("serverPort", String.valueOf(port));

        logger.info("serverPort:{}", port);

        try {
            TNonblockingServerTransport serverTransport = new TNonblockingServerSocket(port);

            TTransportFactory transportFactory = new TFramedTransport.Factory();

            TProtocolFactory proFactory = new TCompactProtocol.Factory();

            TMultiplexedProcessor processor = new TMultiplexedProcessor();

            String[] strings = context.getBeanNamesForType(org.apache.thrift.TProcessor.class);
            for (String beanName : strings) {
                TProcessor bean = (TProcessor) context.getBean(beanName);
                if (bean == null) {
                    logger.warn(">>>>{}<<<<Register failed", beanName);
                    continue;
                }
                beanName = bean.toString().split("\\$")[0];
                beanName = beanName.substring(beanName.lastIndexOf(".") + 1);

                logger.info("registerProcessor:{}", beanName);
                processor.registerProcessor(beanName, bean);
            }

            int corePoolSize = 100;
            int maximumPoolSize = 200;
            long keepAliveTime = 60;
            TimeUnit unit = TimeUnit.SECONDS;
            BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(50);
            ThreadFactory threadFactory = new DefaultThreadFactory();

            ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);

            TThreadedSelectorServer.Args args1 = new TThreadedSelectorServer.Args(serverTransport)
                    .protocolFactory(proFactory)
                    .transportFactory(transportFactory)
                    .processor(processor)
                    //.workerThreads(35)
                    //.acceptQueueSizePerThread(30)
                    .executorService(poolExecutor);

            args1.maxReadBufferBytes = 1024 * 1024 * 5; // 5M

            //Create the server
            TServer server = new TThreadedSelectorServer(args1);

            logger.info("Start server on port {}...", port);
            RunModeHelper.initStart();
            server.serve();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
    static class DefaultThreadFactory implements ThreadFactory {
        final AtomicInteger threadNumber = new AtomicInteger(1);

        public Thread newThread(Runnable r) {
            return new Thread(r, "ks-trans-" + threadNumber.getAndIncrement());
        }
    }
}
```

@duwupeng Ah I see! Thanks for the code example.
I'll test it out and figure out why it's not being traced.

@Xylus ， Thanks a lot,  looking forward to your further  reply
And I would like to provide the URL that wish you can accesss.
http://120.25.79.51:10128/stat.html.
I am with the trouble of tracing  the **sandbox4-ksservices** which is the thrift Server. 

@duwupeng 
Looks like it's because we don't have support for `TMultiplexedProcessor` yet.
We'll take a look at adding support for `TMultiplexedProcessor` but it'll take some time.

If you can, could you try testing it by registering a `TProcessor` object directly to the server and see if that works?

@duwupeng So upon further inspection, the actual culprit of all this was that when using `TMultiplexedProcessor`, the actual input protocol is wrapped by `TMultiplexedProcessor$StoredMessageProtocol`.
So when `ProcessFunction.process()` is invoked, the interceptor injects the server marker into this decorated object and not the actual input protocol, stopping any trace from being generated.

You can test this out once #2110 is merged if you're building from source.
Let me know if anything looks weird.

Thanks!

@duwupeng With the latest source, you should now be able to trace your application.
Please reopen the issue if you have more questions.

Yep I have tried 1.6.0, it worked.
@DevotedTangLiu Awesome, thanks for testing this out! Please let us know if there's anything broken.

Pull request review comments:

Patch that resolved the issue:
diff --git a/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftConstants.java b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftConstants.java
index 6f4c7fdf3dd2..ca70c0df91b0 100644
--- a/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftConstants.java
+++ b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftConstants.java
@@ -67,4 +67,5 @@ private ThriftConstants() {
     private static final String FIELD_GETTER_BASE = "com.navercorp.pinpoint.plugin.thrift.field.getter.";
     public static final String FIELD_GETTER_T_NON_BLOCKING_TRANSPORT = FIELD_GETTER_BASE + "TNonblockingTransportFieldGetter";
     public static final String FIELD_GETTER_T_TRANSPORT = FIELD_GETTER_BASE + "TTransportFieldGetter";
+    public static final String FIELD_GETTER_T_PROTOCOL = FIELD_GETTER_BASE + "TProtocolFieldGetter";
 }
diff --git a/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftPlugin.java b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftPlugin.java
index 47afb2694ec3..e8a24dbfefab 100644
--- a/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftPlugin.java
+++ b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/ThriftPlugin.java
@@ -418,6 +418,7 @@ private void addTProtocolEditors(ThriftPluginConfig config) {
         addTProtocolInterceptors(config, "org.apache.thrift.protocol.TBinaryProtocol");
         addTProtocolInterceptors(config, "org.apache.thrift.protocol.TCompactProtocol");
         addTProtocolInterceptors(config, "org.apache.thrift.protocol.TJSONProtocol");
+        addTProtocolDecoratorEditor();
     }
 
     private void addTProtocolInterceptors(ThriftPluginConfig config, String tProtocolClassName) {
@@ -485,6 +486,20 @@ public byte[] doInTransform(Instrumentor instrumentor, ClassLoader loader, Strin
         });
     }
 
+    private void addTProtocolDecoratorEditor() {
+        transformTemplate.transform("org.apache.thrift.protocol.TProtocolDecorator", new TransformCallback() {
+            @Override
+            public byte[] doInTransform(Instrumentor instrumentor, ClassLoader loader, String className, Class<?> classBeingRedefined,
+                                        ProtectionDomain protectionDomain, byte[] classfileBuffer) throws InstrumentException {
+
+                final InstrumentClass target = instrumentor.getInstrumentClass(loader, className, classfileBuffer);
+
+                target.addGetter(ThriftConstants.FIELD_GETTER_T_PROTOCOL, "concreteProtocol");
+                return target.toBytecode();
+            }
+        });
+    }
+
     @Override
     public void setTransformTemplate(TransformTemplate transformTemplate) {
         this.transformTemplate = transformTemplate;
diff --git a/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/field/getter/TProtocolFieldGetter.java b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/field/getter/TProtocolFieldGetter.java
new file mode 100644
index 000000000000..224af69a451a
--- /dev/null
+++ b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/field/getter/TProtocolFieldGetter.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright 2016 Naver Corp.
+ *
+ * Licensed 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 com.navercorp.pinpoint.plugin.thrift.field.getter;
+
+import org.apache.thrift.protocol.TProtocol;
+
+/**
+ * @author HyunGil Jeong
+ */
+public interface TProtocolFieldGetter {
+    TProtocol _$PINPOINT$_getTProtocol();
+}
diff --git a/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/interceptor/server/ProcessFunctionProcessInterceptor.java b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/interceptor/server/ProcessFunctionProcessInterceptor.java
index 102a3b0d1328..c4c85fce2f88 100644
--- a/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/interceptor/server/ProcessFunctionProcessInterceptor.java
+++ b/plugins/thrift/src/main/java/com/navercorp/pinpoint/plugin/thrift/interceptor/server/ProcessFunctionProcessInterceptor.java
@@ -19,6 +19,7 @@
 import static com.navercorp.pinpoint.plugin.thrift.ThriftScope.THRIFT_SERVER_SCOPE;
 
 import com.navercorp.pinpoint.bootstrap.interceptor.scope.InterceptorScope;
+import com.navercorp.pinpoint.plugin.thrift.field.getter.TProtocolFieldGetter;
 import org.apache.thrift.ProcessFunction;
 import org.apache.thrift.protocol.TProtocol;
 
@@ -82,8 +83,10 @@ public void before(Object target, Object[] args) {
         // When this happens, TProtocol interceptors for clients are triggered since technically they're still within THRIFT_SERVER_SCOPE.
         // We set the marker inside server's input protocol to safeguard against such cases.
         Object iprot = args[1];
-        if (validateInputProtocol(iprot)) {
-            ((ServerMarkerFlagFieldAccessor)iprot)._$PINPOINT$_setServerMarkerFlag(true);
+        // With the addition of TProtocolDecorator, iprot may actually be a wrapper around the actual input protocol
+        Object rootInputProtocol = getRootInputProtocol(iprot);
+        if (validateInputProtocol(rootInputProtocol)) {
+            ((ServerMarkerFlagFieldAccessor)rootInputProtocol)._$PINPOINT$_setServerMarkerFlag(true);
         }
     }
 
@@ -101,6 +104,14 @@ public void after(Object target, Object[] args, Object result, Throwable throwab
         }
     }
 
+    private Object getRootInputProtocol(Object iprot) {
+        if (iprot instanceof TProtocolFieldGetter) {
+            return getRootInputProtocol(((TProtocolFieldGetter) iprot)._$PINPOINT$_getTProtocol());
+        } else {
+            return iprot;
+        }
+    }
+
     private boolean validateInputProtocol(Object iprot) {
         if (iprot instanceof TProtocol) {
             if (iprot instanceof ServerMarkerFlagFieldAccessor) {


Test patch: