rpc

Documentation

Overview

Package rpc implements some of the lower-level functionality required to communicate with the namenode and datanodes.

Index

• Variables
• type BlockReader
  • func NewBlockReader(block *hdfs.LocatedBlockProto, offset int64, clientName string) *BlockReader
  • func (br *BlockReader) Close() error
  • func (br *BlockReader) Read(b []byte) (int, error)
• type BlockWriter
  • func NewBlockWriter(block *hdfs.LocatedBlockProto, namenode *NamenodeConnection, blockSize int64) *BlockWriter
  • func (bw *BlockWriter) Close() error
  • func (bw *BlockWriter) Flush() error
  • func (bw *BlockWriter) Write(b []byte) (int, error)
• type ChecksumReader
  • func NewChecksumReader(block *hdfs.LocatedBlockProto) *ChecksumReader
  • func (cr *ChecksumReader) ReadChecksum() ([]byte, error)
• type NamenodeConnection
  • func NewNamenodeConnection(address string, user string) (*NamenodeConnection, error)
  • func NewNamenodeConnectionWithOptions(options NamenodeConnectionOptions) (*NamenodeConnection, error)
  • func WrapNamenodeConnection(conn net.Conn, user string) (*NamenodeConnection, error)
  • func (c *NamenodeConnection) ClientName() string
  • func (c *NamenodeConnection) Close() error
  • func (c *NamenodeConnection) Execute(method string, req proto.Message, resp proto.Message) error
• type NamenodeConnectionOptions
• type NamenodeError
  • func (err *NamenodeError) Desc() string
  • func (err *NamenodeError) Error() string

Variables

var ErrEndOfBlock = errors.New("The amount of data to be written is more than is left in the block.")

var ErrInvalidSeqno = errors.New("Invalid ack sequence number")

Types

type BlockReader

type BlockReader struct {
	// contains filtered or unexported fields
}

BlockReader implements io.ReadCloser, for reading a block. It abstracts over reading from multiple datanodes, in order to be robust to connection failures, timeouts, and other shenanigans.

func NewBlockReader

func NewBlockReader(block *hdfs.LocatedBlockProto, offset int64, clientName string) *BlockReader

NewBlockReader returns a new BlockReader, given the block information and security token from the namenode. It will connect (lazily) to one of the provided datanode locations based on which datanodes have seen failures.

func (*BlockReader) Close

func (br *BlockReader) Close() error

Close implements io.Closer.

func (*BlockReader) Read

func (br *BlockReader) Read(b []byte) (int, error)

Read implements io.Reader.

In the case that a failure (such as a disconnect) occurs while reading, the BlockReader will failover to another datanode and continue reading transparently. In the case that all the datanodes fail, the error from the most recent attempt will be returned.

Any datanode failures are recorded in a global cache, so subsequent reads, even reads for different blocks, will prioritize them lower.

type BlockWriter

type BlockWriter struct {
	// contains filtered or unexported fields
}

BlockWriter implements io.WriteCloser for writing a block to a datanode. Given a block location, it handles pipeline construction and failures, including communicating with the namenode if need be.

func NewBlockWriter

func NewBlockWriter(block *hdfs.LocatedBlockProto, namenode *NamenodeConnection, blockSize int64) *BlockWriter

NewBlockWriter returns a BlockWriter for the given block. It will lazily set up a replication pipeline, and connect to the "best" datanode based on any previously seen failures.

func (*BlockWriter) Close

func (bw *BlockWriter) Close() error

Close implements io.Closer. It flushes any unwritten packets out to the datanode, and sends a final packet indicating the end of the block.

func (*BlockWriter) Flush

func (bw *BlockWriter) Flush() error

Flush flushes any unwritten packets out to the datanode.

func (*BlockWriter) Write

func (bw *BlockWriter) Write(b []byte) (int, error)

Write implements io.Writer.

Unlike BlockReader, BlockWriter currently has no ability to recover from write failures (timeouts, datanode failure, etc). Once it returns an error from Write or Close, it may be in an invalid state.

This will hopefully be fixed in a future release.

type ChecksumReader

type ChecksumReader struct {
	// contains filtered or unexported fields
}

ChecksumReader provides an interface for reading the "MD5CRC32" checksums of individual blocks. It abstracts over reading from multiple datanodes, in order to be robust to failures.

func NewChecksumReader

func NewChecksumReader(block *hdfs.LocatedBlockProto) *ChecksumReader

NewChecksumReader creates a new ChecksumReader for the given block.

func (*ChecksumReader) ReadChecksum

func (cr *ChecksumReader) ReadChecksum() ([]byte, error)

ReadChecksum returns the checksum of the block.

type NamenodeConnection

type NamenodeConnection struct {
	// contains filtered or unexported fields
}

NamenodeConnection represents an open connection to a namenode.

func NewNamenodeConnection

func NewNamenodeConnection(address string, user string) (*NamenodeConnection, error)

NewNamenodeConnection creates a new connection to a namenode and performs an initial handshake.

You probably want to use hdfs.New instead, which provides a higher-level interface.

func NewNamenodeConnectionWithOptions

func NewNamenodeConnectionWithOptions(options NamenodeConnectionOptions) (*NamenodeConnection, error)

NewNamenodeConnectionWithOptions creates a new connection to a namenode with the given options and performs an initial handshake.

func WrapNamenodeConnection

func WrapNamenodeConnection(conn net.Conn, user string) (*NamenodeConnection, error)

WrapNamenodeConnection wraps an existing net.Conn to a Namenode, and preforms an initial handshake.

Deprecated: use the higher-level hdfs.New or NewNamenodeConnection instead.

func (*NamenodeConnection) ClientName

func (c *NamenodeConnection) ClientName() string

ClientName provides a unique identifier for this client, which is required for various RPC calls. Confusingly, it's separate from clientID, which is used in the RPC header; to make things simpler, it reuses the random bytes from that, but adds a prefix to make it human-readable.

func (*NamenodeConnection) Close

func (c *NamenodeConnection) Close() error

Close terminates all underlying socket connections to remote server.

func (*NamenodeConnection) Execute

func (c *NamenodeConnection) Execute(method string, req proto.Message, resp proto.Message) error

Execute performs an rpc call. It does this by sending req over the wire and unmarshaling the result into resp.

type NamenodeConnectionOptions

type NamenodeConnectionOptions struct {
	Addresses []string
	User      string
}

NamenodeConnectionOptions represents the configurable options available for a NamenodeConnection.

type NamenodeError

type NamenodeError struct {
	Method    string
	Message   string
	Code      int
	Exception string
}

NamenodeError represents an interepreted error from the Namenode, including the error code and the java backtrace.

func (*NamenodeError) Desc

func (err *NamenodeError) Desc() string

Desc returns the long form of the error code, as defined in the RpcErrorCodeProto in RpcHeader.proto

func (*NamenodeError) Error

func (err *NamenodeError) Error() string