HDFSDATANODE数据传输详解

本文主要阐述datanode中一个socket连接接收字节流的构成，帮助datanode的接收与处理数据。注意hadoop版本为3.1.1。

写在前面

Datanode本质上也是TCPServer，一般的TCPServer接到客户端请求以后会分配一个线程处理，对于Datanode而言，这个线程可以叫做Op处理连接。每个OP连接会多次和客户端交互，中间涉及多种packet。

关于proto writeDelimitedTo方法

在整个处理流程中，会非常频繁的使用的到proto.writeDelimitedTo来传递相关proto，简单理解就是要写入proto时，写入总长度，再写入proto。读取proto时，先读取长度，在解析proto。

DataNode连接数据流说明

OpPacket的接收

Op连接第一个包总是来定义Op连接处理那种Op，例如读块op，写块op。这种包简单命名为OpPacket。Packet的结构如下，可以根据下图读取OpPacket。

DataTransferProtocol.DATA_TRANSFER_VERSION：short，3.1.1版本默认为28。

OpCode：

OpProto：定义Op是哪种Op。Op在datatransfer.proto定义，包含OpReadBlockProto，OpWriteBlockProto，OpTransferBlockProto，OpReplaceBlockProto，OpCopyBlockProto，OpBlockChecksumProto，OpRequestShortCircuitAccessProto，ReleaseShortCircuitAccessRequestProto。

接下来的是否回应或者直接发数据，都要根据不同的op来处理，后续介绍了write和read。

WriteBlock

当接收完OpPacket以后，需要写入一个BlockOpResponseProto应答。

当客户端接受BlockOpResponseProto应答后，就会发送数据包，数据包的格式如下

PktLen：数据包长度，不同于字面意思，这个数值并不是包的总长度，而是4(pktLen所占字节数)+chunkchecksums字节数+chunkdatas字节数。
HeadLen：short，PacketHeaderProto的长度。不同于writeDelimitedTo，这边使用的proto.getSerializedSize。
PktHeadProto：PacketHeaderProto.writeTo。
Chunkchecksums：chunk校验数据。
ChunkData：实际数据。
DataNode接受到数据以后，完成checksum后就把Status.success放入Responder的处理队列。Responder最终会返回PipelineAckProto(PipelineAckProto.writeDelimitedTo)给客户端。

ReadBlock

当接收完OpPacket以后，需要写入一个BlockOpResponseProto应答。

写完应答以后，立马会发送Block的数据包，数据包的结果如下：

PktLen：数据包长度，不同于字面意思，这个数值并不是包的总长度，而是4(pktLen所占字节数)+chunkchecksums字节数+chunkdatas字节数。
HeadLen：short，PacketHeaderProto的长度。不同于writeDelimitedTo，这边使用的proto.getSerializedSize。
PktHeadProto：PacketHeaderProto.writeTo。
Chunkchecksums：chunk校验数据。
ChunkData：实际数据。
数据会被分成多个数据包发送，发送完最后一个数据包以后，会发送一个空包（没有数据只有header），空包的PacketHeaderProto会有LastPackctInBlock的标记。空包发送完成后，会接受一个ClientReadStatusProto的包（客户端使用ClientReadStatusProto.writeDelimitedTo写入）。

TransferBlock、ReplaceBlock未分析。

数据流中的sasl

Hadoop使用dfs.data.transfer.protection参数来保证数据流的安全。dfs.data.transfer.protection有三种模式authentication,integrity,privacy，分别对于sasl qop中的auth，auth-int，auth-conf。Auth：流建立需要握手，握手成功以后，后续流就是正常使用。
Auth-int：流握手+后续流都需要通过sasl的wrap unwarp加密解密。
Auth-conf：流握手+后续流都需要通过协商的算法来数据加密解密。

Sasl握手：
Sasl的mech为DIGEST-MD5，serviceName为0，protocol（c中的username）为hdfs。通过此信息可以创建saslclient，saslserver。DIGEST-MD5的callback的本质上就是验证用户名密码。数据流的用户密码来源与blocktoken。
关于sasl中协商的包结构为DataTransferEncryptorMessageProto，写入使用writeDelimitedTo。

message DataTransferEncryptorMessageProto {
  enum DataTransferEncryptorStatus {
    SUCCESS = 0;
    ERROR_UNKNOWN_KEY = 1;
    ERROR = 2;
  }
  required DataTransferEncryptorStatus status = 1;
  optional bytes payload = 2;
  optional string message = 3;
  repeated CipherOptionProto cipherOption = 4;
}

Payload就是token，message只有status为error才使用，为errmsg。Server发生异常都会发送错误，并关闭这个流。

流程图：

1. client发送sasl_Version，为4byte

SASL_TRANSFER_MAGIC_NUMBER = 0xDEADBEEF;server接收并验证。

2. client发送第一个saslMessage，Status为success，payload为byte[0]。

3. Server就收到包以后使用saslserver.evaluateResponse（c中为gsasl_setup)来处理payload。

Server发送应答saslMessage，Status为success，payload为算出来的token。

4. client接收到包以后，使用saslclient.evaluateChallenge（c中为gsasl_setup)来出来payload。

client发送第二个saslMessage，Status为success，payload为算出来的token。

5. Server就收到包以后使用saslserver.evaluateResponse（c中为gsasl_setup)来处理payload。

这时候如果payload没问题，saslserver会complete。Server发送应答saslMessage，Status为success，payload为算出来的token。

6. client接收到包以后，使用saslclient.evaluateChallenge（c中为gsasl_setup)来出来payload。

这时候如果payload没问题，saslclient会complete。

独立站原文