drpc

package
v0.8.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Dec 17, 2019 License: Apache-2.0 Imports: 11 Imported by: 0

README

dRPC

dRPC is a means of communication between processes local to the same physical system, via a Unix Domain Socket. At any given time a process may act as a client, a server, or both, though each listening dRPC server needs its own Unix Domain Socket.

The server will fail to create the socket if something already exists at that location in the filesystem, even if it is an older incarnation of the socket. Optionally, your application may wish to unlink that filesystem location before creating the socket.

dRPC calls are defined by module and method identifiers. The dRPC module can be thought of as a package of related functions. The dRPC method indicates a specific function to be executed by the server. If the method requires an input, it should be marshalled in the body of the dRPC call. The server will respond with a dRPC response structure, which may include a method-specific response in the body.

The DAOS dRPC implementation is dependent on Protocol Buffers to define the structures passed over the dRPC channel. Any structure to be sent via dRPC as part of a call or response must be defined in a .proto file.

Go API

In Go, the drpc package includes both client and server functionality, which is outlined below. For documentation of the C API, see here.

The dRPC call and response are represented by the Protobuf-generated drpc.Call and drpc.Response structures.

Go Client

The dRPC client is represented by the drpc.ClientConnection object.

Basic Client Workflow
  1. Create a new client connection with the path to the dRPC server's Unix Domain Socket:
    conn := drpc.NewClientConnection("/var/run/my_socket.sock")
    
  2. Connect to the dRPC server:
    err := conn.Connect()
    
  3. Create your drpc.Call and send it to the server:
    call := drpc.Call{}
    // Set up the Call with module, method, and body
    resp, err := drpc.SendMsg(call)
    
    An error indicates that the drpc.Call couldn't be sent, or an invalid drpc.Response was received. If there is no error returned, the content of the drpc.Response should still be checked for errors reported by the server.
  4. Send as many calls as desired.
  5. Close the connection when finished:
    conn.Close()
    
Go Server

The dRPC server is represented by the drpc.DomainSocketServer object.

Individual dRPC modules must be registered with the server in order to handle incoming dRPC calls for that module. To create a dRPC module, create an object that implements the drpc.Module interface. The module ID must be unique.

Basic Server Workflow
  1. Create the new DomainSocketServer with the server's Unix Domain Socket:
    drpcServer, err := drpc.NewDomainSocketServer("/var/run/my_socket.sock")
    
  2. Register the dRPC modules that the server needs to handle:
    drpcServer.RegisterRPCModule(&MyExampleModule{})
    drpcServer.RegisterRPCModule(&AnotherExampleModule{})
    
  3. Start the server to kick off the Goroutine to start listening for and handling incoming connections:
    err = drpc.Start()
    
  4. When it is time to shut down the server, close down the listening Goroutine:
    drpcServer.Shutdown()
    

Documentation

Index

Constants

View Source
const (
	// ModuleSecurityAgent is the dRPC module for security tasks in the
	// DAOS agent
	ModuleSecurityAgent = C.DRPC_MODULE_SEC_AGENT
	// ModuleMgmt is the dRPC module for management service tasks
	ModuleMgmt = C.DRPC_MODULE_MGMT
	// ModuleSrv is the dRPC module for tasks relating to server setup
	ModuleSrv = C.DRPC_MODULE_SRV
	// ModuleSecurity is the dRPC module for security tasks in the DAOS
	// server
	ModuleSecurity = C.DRPC_MODULE_SEC
)
View Source
const (
	// MethodPrepShutdown is a ModuleMgmt method
	MethodPrepShutdown = C.DRPC_METHOD_MGMT_PREP_SHUTDOWN
	// MethodKillRank is a ModuleMgmt method
	MethodKillRank = C.DRPC_METHOD_MGMT_KILL_RANK
	// MethodSetRank is a ModuleMgmt method
	MethodSetRank = C.DRPC_METHOD_MGMT_SET_RANK
	// MethodCreateMS is a ModuleMgmt method
	MethodCreateMS = C.DRPC_METHOD_MGMT_CREATE_MS
	// MethodStartMS is a ModuleMgmt method
	MethodStartMS = C.DRPC_METHOD_MGMT_START_MS
	// MethodJoin is a ModuleMgmt method
	MethodJoin = C.DRPC_METHOD_MGMT_JOIN
	// MethodGetAttachInfo is a ModuleMgmt method
	MethodGetAttachInfo = C.DRPC_METHOD_MGMT_GET_ATTACH_INFO
	// MethodPoolCreate is a ModuleMgmt method
	MethodPoolCreate = C.DRPC_METHOD_MGMT_POOL_CREATE
	// MethodPoolDestroy is a ModuleMgmt method
	MethodPoolDestroy = C.DRPC_METHOD_MGMT_POOL_DESTROY
	// MethodBioHealth is a ModuleMgmt method
	MethodBioHealth = C.DRPC_METHOD_MGMT_BIO_HEALTH_QUERY
	// MethodSetUp is a ModuleMgmt method
	MethodSetUp = C.DRPC_METHOD_MGMT_SET_UP
	// MethodSmdDevs is a ModuleMgmt method
	MethodSmdDevs = C.DRPC_METHOD_MGMT_SMD_LIST_DEVS
	// MethodSmdPools is a ModuleMgmt method
	MethodSmdPools = C.DRPC_METHOD_MGMT_SMD_LIST_POOLS
	// MethodPoolGetACL is a ModuleMgmt method
	MethodPoolGetACL = C.DRPC_METHOD_MGMT_POOL_GET_ACL
	// MethodListPools is a ModuleMgmt method
	MethodListPools = C.DRPC_METHOD_MGMT_LIST_POOLS
	// MethodPoolOverwriteACL is a ModuleMgmt method
	MethodPoolOverwriteACL = C.DRPC_METHOD_MGMT_POOL_OVERWRITE_ACL
	// MethodPoolUpdateACL is a ModuleMgmt method
	MethodPoolUpdateACL = C.DRPC_METHOD_MGMT_POOL_UPDATE_ACL
	// MethodPoolDeleteACL is a ModuleMgmt method
	MethodPoolDeleteACL = C.DRPC_METHOD_MGMT_POOL_DELETE_ACL
)
View Source
const (
	// MethodNotifyReady is a ModuleSrv method
	MethodNotifyReady = C.DRPC_METHOD_SRV_NOTIFY_READY
	// MethodBIOError is a ModuleSrv method
	MethodBIOError = C.DRPC_METHOD_SRV_BIO_ERR
)
View Source
const MaxMsgSize = 16384

MaxMsgSize is the maximum drpc message size that may be sent. Using a packetsocket over the unix domain socket means that we receive a whole message at a time without knowing its size. So for this reason we need to restrict the maximum message size so we can preallocate a buffer to put all of the information in. Corresponding C definition is found in include/daos/drpc.h

View Source
const (
	// MethodRequestCredentials is a ModuleSecurityAgent method
	MethodRequestCredentials = C.DRPC_METHOD_SEC_AGENT_REQUEST_CREDS
)
View Source
const (
	// MethodValidateCredentials is a ModuleSecurity method
	MethodValidateCredentials = C.DRPC_METHOD_SEC_VALIDATE_CREDS
)

Variables

View Source
var Status_name = map[int32]string{
	0: "SUCCESS",
	1: "SUBMITTED",
	2: "FAILURE",
	3: "UNKNOWN_MODULE",
	4: "UNKNOWN_METHOD",
	5: "FAILED_UNMARSHAL_CALL",
	6: "FAILED_UNMARSHAL_PAYLOAD",
	7: "FAILED_MARSHAL",
}
View Source
var Status_value = map[string]int32{
	"SUCCESS":                  0,
	"SUBMITTED":                1,
	"FAILURE":                  2,
	"UNKNOWN_MODULE":           3,
	"UNKNOWN_METHOD":           4,
	"FAILED_UNMARSHAL_CALL":    5,
	"FAILED_UNMARSHAL_PAYLOAD": 6,
	"FAILED_MARSHAL":           7,
}

Functions

This section is empty.

Types

type Call

type Call struct {
	Module               int32    `protobuf:"varint,1,opt,name=module,proto3" json:"module,omitempty"`
	Method               int32    `protobuf:"varint,2,opt,name=method,proto3" json:"method,omitempty"`
	Sequence             int64    `protobuf:"varint,3,opt,name=sequence,proto3" json:"sequence,omitempty"`
	Body                 []byte   `protobuf:"bytes,4,opt,name=body,proto3" json:"body,omitempty"`
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

* Call describes a function call to be executed over the dRPC channel.

func (*Call) Descriptor

func (*Call) Descriptor() ([]byte, []int)

func (*Call) GetBody

func (m *Call) GetBody() []byte

func (*Call) GetMethod

func (m *Call) GetMethod() int32

func (*Call) GetModule

func (m *Call) GetModule() int32

func (*Call) GetSequence

func (m *Call) GetSequence() int64

func (*Call) ProtoMessage

func (*Call) ProtoMessage()

func (*Call) Reset

func (m *Call) Reset()

func (*Call) String

func (m *Call) String() string

func (*Call) XXX_DiscardUnknown

func (m *Call) XXX_DiscardUnknown()

func (*Call) XXX_Marshal

func (m *Call) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Call) XXX_Merge

func (dst *Call) XXX_Merge(src proto.Message)

func (*Call) XXX_Size

func (m *Call) XXX_Size() int

func (*Call) XXX_Unmarshal

func (m *Call) XXX_Unmarshal(b []byte) error

type ClientConnection

type ClientConnection struct {
	sync.Mutex
	// contains filtered or unexported fields
}

ClientConnection represents a client connection to a dRPC server

func NewClientConnection

func NewClientConnection(socket string) *ClientConnection

NewClientConnection creates a new dRPC client

func (*ClientConnection) Close

func (c *ClientConnection) Close() error

Close shuts down the connection to the Unix Domain Socket

func (*ClientConnection) Connect

func (c *ClientConnection) Connect() error

Connect opens a connection to the internal Unix Domain Socket path

func (*ClientConnection) IsConnected

func (c *ClientConnection) IsConnected() bool

IsConnected indicates whether the client connection is currently active

func (*ClientConnection) SendMsg

func (c *ClientConnection) SendMsg(msg *Call) (*Response, error)

SendMsg sends a message to the connected dRPC server, and returns the response to the caller.

type DomainSocketClient

type DomainSocketClient interface {
	sync.Locker
	IsConnected() bool
	Connect() error
	Close() error
	SendMsg(call *Call) (*Response, error)
}

DomainSocketClient is the interface to a dRPC client communicating over a Unix Domain Socket

type DomainSocketServer

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

DomainSocketServer is the object that listens for incoming dRPC connections, maintains the connections for sessions, and manages the message processing.

func NewDomainSocketServer

func NewDomainSocketServer(ctx context.Context, log logging.Logger, sock string) (*DomainSocketServer, error)

NewDomainSocketServer returns a new unstarted instance of a DomainSocketServer for the specified unix domain socket path.

func (*DomainSocketServer) Listen

func (d *DomainSocketServer) Listen()

Listen listens for incoming connections on the UNIX domain socket and creates individual sessions for each one.

func (*DomainSocketServer) RegisterRPCModule

func (d *DomainSocketServer) RegisterRPCModule(mod Module)

RegisterRPCModule takes a Module and associates it with the given DomainSocketServer so it can be used to process incoming dRPC calls.

func (*DomainSocketServer) Shutdown

func (d *DomainSocketServer) Shutdown()

Shutdown places the state of the server to shutdown which terminates the Listen go routine and starts the cleanup of all open connections.

func (*DomainSocketServer) Start

func (d *DomainSocketServer) Start() error

Start sets up the dRPC server socket and kicks off the listener goroutine.

type Failure added in v0.8.0

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

Failure represents a dRPC protocol failure.

func MarshalingFailure added in v0.8.0

func MarshalingFailure() Failure

MarshalingFailure creates a Failure for a failed attempt at marshaling a response.

func NewFailure added in v0.8.0

func NewFailure(status Status) Failure

NewFailure returns a Failure with the given status and a corresponding message.

func NewFailureWithMessage added in v0.8.0

func NewFailureWithMessage(message string) Failure

NewFailureWithMessage returns a generic failure with a custom message

func UnknownMethodFailure added in v0.8.0

func UnknownMethodFailure() Failure

UnknownMethodFailure creates a Failure for unknown dRPC method.

func UnknownModuleFailure added in v0.8.0

func UnknownModuleFailure() Failure

UnknownModuleFailure creates a Failure for unknown dRPC module.

func UnmarshalingCallFailure added in v0.8.0

func UnmarshalingCallFailure() Failure

UnmarshalingCallFailure creates a Failure for a failed attempt to unmarshal an incoming call.

func UnmarshalingPayloadFailure added in v0.8.0

func UnmarshalingPayloadFailure() Failure

UnmarshalingPayloadFailure creates a Failure for a failed attempt to unmarshal a call payload.

func (Failure) Error added in v0.8.0

func (e Failure) Error() string

Error provides a descriptive string associated with the failure.

func (Failure) GetStatus added in v0.8.0

func (e Failure) GetStatus() Status

GetStatus provides a dRPC status code associated with the failure.

type Module

type Module interface {
	HandleCall(*Session, int32, []byte) ([]byte, error)
	ID() int32
}

Module is an interface that a type must implement to provide the functionality needed by the ModuleService to process dRPC requests.

type ModuleService

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

ModuleService is the collection of Modules used by DomainSocketServer to be used to process messages.

func NewModuleService

func NewModuleService(log logging.Logger) *ModuleService

NewModuleService creates an initialized ModuleService instance

func (*ModuleService) GetModule

func (r *ModuleService) GetModule(id int32) (Module, bool)

GetModule fetches the module for the given ID. Returns true if found, false otherwise.

func (*ModuleService) ProcessMessage

func (r *ModuleService) ProcessMessage(session *Session, msgBytes []byte) ([]byte, error)

ProcessMessage is the main entry point into the ModuleService. It accepts a marshaled drpc.Call instance, processes it, calls the handler in the appropriate Module, and marshals the result into the body of a drpc.Response.

func (*ModuleService) RegisterModule

func (r *ModuleService) RegisterModule(mod Module) error

RegisterModule will take in a type that implements the Module interface and ensure that no other module is already registered with that module identifier.

type Response

type Response struct {
	Sequence             int64    `protobuf:"varint,1,opt,name=sequence,proto3" json:"sequence,omitempty"`
	Status               Status   `protobuf:"varint,2,opt,name=status,proto3,enum=drpc.Status" json:"status,omitempty"`
	Body                 []byte   `protobuf:"bytes,3,opt,name=body,proto3" json:"body,omitempty"`
	XXX_NoUnkeyedLiteral struct{} `json:"-"`
	XXX_unrecognized     []byte   `json:"-"`
	XXX_sizecache        int32    `json:"-"`
}

* Response describes the result of a dRPC call.

func (*Response) Descriptor

func (*Response) Descriptor() ([]byte, []int)

func (*Response) GetBody

func (m *Response) GetBody() []byte

func (*Response) GetSequence

func (m *Response) GetSequence() int64

func (*Response) GetStatus

func (m *Response) GetStatus() Status

func (*Response) ProtoMessage

func (*Response) ProtoMessage()

func (*Response) Reset

func (m *Response) Reset()

func (*Response) String

func (m *Response) String() string

func (*Response) XXX_DiscardUnknown

func (m *Response) XXX_DiscardUnknown()

func (*Response) XXX_Marshal

func (m *Response) XXX_Marshal(b []byte, deterministic bool) ([]byte, error)

func (*Response) XXX_Merge

func (dst *Response) XXX_Merge(src proto.Message)

func (*Response) XXX_Size

func (m *Response) XXX_Size() int

func (*Response) XXX_Unmarshal

func (m *Response) XXX_Unmarshal(b []byte) error

type Session

type Session struct {
	Conn net.Conn
	// contains filtered or unexported fields
}

Session represents an individual client connection to the Domain Socket Server.

func NewSession

func NewSession(conn net.Conn, svc *ModuleService) *Session

NewSession creates a new dRPC Session object

func (*Session) Close

func (s *Session) Close()

Close closes the session

func (*Session) ProcessIncomingMessage

func (s *Session) ProcessIncomingMessage() error

ProcessIncomingMessage listens for an incoming message on the session, calls its handler, and sends the response.

type Status

type Status int32

* Status represents the valid values for a response status.

const (
	Status_SUCCESS                  Status = 0
	Status_SUBMITTED                Status = 1
	Status_FAILURE                  Status = 2
	Status_UNKNOWN_MODULE           Status = 3
	Status_UNKNOWN_METHOD           Status = 4
	Status_FAILED_UNMARSHAL_CALL    Status = 5
	Status_FAILED_UNMARSHAL_PAYLOAD Status = 6
	Status_FAILED_MARSHAL           Status = 7
)

func ErrorToStatus added in v0.8.0

func ErrorToStatus(err error) Status

ErrorToStatus translates an error to a dRPC Status. In practice it checks to see if it was a dRPC Failure error, and uses the Status if so. Otherwise it is assumed to be a generic failure.

func (Status) EnumDescriptor

func (Status) EnumDescriptor() ([]byte, []int)

func (Status) String

func (x Status) String() string

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL