Thrift RPC添加access log

前言：

当我们在部署web服务的时候，web容器通常都会记录来自客户端的访问日志。而当我们使用Thrift RPC服务的时候，Thrift服务则不会给我们自动记录客户端的访问日志。

通过这篇文章，你可以学习到如何使用在Thrift服务器端添加客户端的访问日志。

面临的问题：

要在Thrift服务器端添加客户端的访问日志，我们需要解决两个问题：

找到合适的拦截点记录信息
收集访问日志中需要的信息

寻找合适的拦截点：

我们都知道，Thrift协议为我们提供了thrift文件向各种编程语言转换的程序。通过观察，我们会发现Thrift将IDL中定义的每个方法抽象为一个类，即ProcessFunction类。

该类负责从输入中读取参数，调用用户编写的服务将响应写回到输出中。该类是如何发挥作用的，下面这张类图可以比较清晰地说明。

当我们使用Thrift.exe可执行程序处理IDL文件的时候，Processor会被自动创建出来。它负责把实际的方法实现和方法的key关联起来，放到Map中维护。

以TMultiplexedProcessor为例，TMultiplexedProcessor会将所有注册的Processor都存储到SERVICE_PROCESSOR_MAP中。

public boolean process(TProtocol iprot, TProtocol oprot) throws TException {

    /*

     先读取消息头

    */

    TMessage message = iprot.readMessageBegin();

    if (message.type != TMessageType.CALL && message.type != TMessageType.ONEWAY) {

        // TODO Apache Guys - Can the server ever get an EXCEPTION or REPLY?

        // TODO Should we check for this here?

        throw new TException("This should not have happened!?");

    }

    // Extract the service name

    int index = message.name.indexOf(TMultiplexedProtocol.SEPARATOR);

    if (index < 0) {

        throw new TException("Service name not found in message name: " + message.name + ".  Did you " +

                "forget to use a TMultiplexProtocol in your client?");

    }

    // 从message中读取serviceName

    String serviceName = message.name.substring(0, index);

    TProcessor actualProcessor = SERVICE_PROCESSOR_MAP.get(serviceName);

    if (actualProcessor == null) {

        throw new TException("Service name not found: " + serviceName + ".  Did you forget " +

                "to call registerProcessor()?");

    }

    // Create a new TMessage, removing the service name

    TMessage standardMessage = new TMessage(

            message.name.substring(serviceName.length()+TMultiplexedProtocol.SEPARATOR.length()),

            message.type,

            message.seqid

    );

    //由真实的处理器对输入信息进行处理

    return actualProcessor.process(new StoredMessageProtocol(iprot, standardMessage), oprot);

}

actualProcessor的process过程如下，其具体的实现逻辑在TBaseProcessor中。

public boolean process(TProtocol in, TProtocol out) throws TException {

  //读取消息头

  TMessage msg = in.readMessageBegin();

  //从方法集合中获取对应的方法处理类

  ProcessFunction fn = processMap.get(msg.name);

  if (fn == null) {

    TProtocolUtil.skip(in, TType.STRUCT);

    in.readMessageEnd();

    TApplicationException x = new TApplicationException(TApplicationException.UNKNOWN_METHOD, "Invalid method name: '"+msg.name+"'");

    out.writeMessageBegin(new TMessage(msg.name, TMessageType.EXCEPTION, msg.seqid));

    x.write(out);

    out.writeMessageEnd();

    out.getTransport().flush();

    return true;

  }

  //进行具体的处理，ProcessFunction对象是实际方法的装饰器，

  //process内部会调用实际方法的处理逻辑

  fn.process(msg.seqid, in, out, iface);

  return true;

}

通过上面的分析，我们可以在ProcessFunction中添加有关的access日志。但是这其中有一个问题，就是经过ThriftServer对thrift请求的解析以及消息内容处理，在到达ProcessFunction::process方法的时候，我们已经无法获取到客户端的远程IP地址了。

接下来，我们就要考虑如何收集访问日志需要的信息了。

如何收集访问日志需要的信息：

从上面ProcessFunction中的process方法中，我们可以看出将客户端的IP地址保存到iprot中，是一个不错的选择。

那么，接下来我们需要找到iprot这个对象参数是在什么地方被创建的，以及在合适的地方将客户端的IP地址写入到这个对象中。

经过分析，我们会发现TNonblockingServer是NIO服务器的实现，它通过Selector来检查IO就绪状态，进而调用相关的Channel。

就方法调用而言，它处理的是读事件，用AbstractNonblockingServer的handelRead()来进一步处理。

protected void handleRead(SelectionKey key) throws IOException {

  FrameBuffer buffer = (FrameBuffer) key.attachment();

  if (!buffer.read()) {

    cleanupSelectionKey(key);

    return;

  }

  // if the buffer's frame read is complete, invoke the method.

  if (buffer.isFrameFullyRead()) {

    if (!requestInvoke(buffer)) {

      cleanupSelectionKey(key);

    }

  }

}

SelectionKey中有客户端的IP地址，FrameBuffer则是处理方法调用的缓冲区对象，其内部的invoke方法会对Processor中的方法进行实际调用。

因此，在handleRead方法中添加两行代码将客户端的IP地址写入inProt_中就可以带入ProcessFunction中了。

SocketChannel socketChannel = (SocketChannel)key.channel();

buffer.inProt_.setClientAddr(socketChannel.getRemoteAddress().toString());

巴特西

Thrift RPC添加access log

前言：

面临的问题：

最新文章

热门文章