Thrift RPC添加access log

前言：

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

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

面临的问题：

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

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

我们都知道，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());