Thrift RPC添加access log

时间：2022-12-06 16:06:12

前言：

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

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

面临的问题：

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

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

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

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

Thrift RPC添加access log

当我们使用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());

相关文章