module

Documentation

Overview

Package module provides services for external resource and logic modules.

Module Resource System Overview

The module system allows a user to build an external binary, either in Golang, using this package and any others from the RDK ecosystem, or in any other language, provided it can properly support protobuf/grpc. The path to the binary (the module) and a name for it must be given in the Modules section of the robot config. The normal viam-server (rdk) process will then start this binary, and query it via GRPC for what protocols (protobuf described APIs) and models it supports. Then, any components or services that match will be handled seamlessly by the module, including reconfiguration, shutdown, and dependency management. Modular components may depend on others from either the parent (aka built-in resources) or other modules, and vice versa. Modular resources should behave identically to built-in resources from a user perspective.

Startup

The module manager (modmanager) integrates with the robot and resource manager. During startup, a dedicated GRPC module service is started, listening on a unix socket in a temporary directory (ex: /tmp/viam-modules-893893/parent.sock) and then individual modules are executed. These are each passed dedicated socket address of their own in the same directory, and based on the module name. (ex: /tmp/viam-modules-893893/acme.sock) The parent then queries this address with Ready() and waits for confirmation. The ready response also includes a HandlerMap that defines which protocols and models the module provides support for. The parent then registers these APIs and models, with creator functions that call the manager's AddResource() method. Once all modules are started, normal robot loading continues.

When resources or components are attempting to load that are not built in, their creator method calls AddResource() and a request is built and sent to the module. The entire config is sent as part of this, as are dependencies. Dependencies are passed by name only through GRPC, and the module library on the module side automatically creates grpc clients for each resource, before calling the component/service constructor. In this way, fully usable dependencies are provided, just as they would be during built-in resource creation.

Back on the parent side, once the AddResource() call completes, the modmanager then establishes an rpc client for the resource, and returns that to the resource manager, which inserts it into the resource graph. For built-in protocols (arm, motor, base, etc.) this rpc client is cast to the expected interface, and is functionally identical to a built-in component. For new protocols, the client created is wrapped as a ForeignResource, which (along with the reflection service in the module) allows it to be used normally by external clients that are also aware of the new protocol in question.

Reconfiguration

The reconfiguration process is handled as transparently as possible to the end user. When a resource would be reconfigured by the resource manager, it is checked if it belongs to a module. If true, then a ReconfigureResource() request is sent to the module instead. (The existing grpc client object on the parent side is untouched.) In the module, the receiving method attempts to cast the real resource to registry.ReconfigurableComponent/Service. If successful, the Reconfigure() method is called on the resource. This method receives the full new config (and dependencies) just as AddResource would. It's then up to the resource itself to reconfigure itself accordingly. If the cast fails (e.g. the resource doesn't have the Reconfigure method.) then the existing resource is closed, and a new one created in its place. Note that unlike built-in resources, no proxy resource is used, since the real client is in the parent, and will automatically get the new resource, since it is looked up by name on each function call.

For removal (during shutdown) RemoveResource() is called, and only passes the resource.Name to the module.

Shutdown

The shutdown process is hooked so that during the Close() of the resource manager, resources are checked if they are modular, and if so, RemoveResource() is called after the parent-side rpc client is closed. The grpc module service is also kept open as late as possible. Otherwise, shutdown happens as normal, including the closing of components in topological (dependency) order.

Module Protocol Requirements

A module can technically be built in any language, with or without support of this RDK or other Viam SDKs. From a technical point of view, all that's required is that the module:

• Is an executable file by unix standards. This can be a compiled binary, or a script with the proper shebang to its interpreter, such as python.
• Looks at the first argument passed to it at execution, and uses that as it's grpc socket path.
• Listens with plaintext GRPC on that socket.
• GRPC must provide the Module service (https://github.com/viamrobotics/api/tree/main/proto/viam/module/v1/module.proto), a reflection service, and any APIs needed for the resources it intends to serve. Note that the "robot" service itself is NOT required.
• Handles the Module service's calls for Ready(), and Add/Remove/ReconfigureResource()
• Cleanly exits when sent a SIGINT or SIGTERM signal.

Module Creation Considerations

Under Golang, the module side of things tries to use as much of the "RDK" idioms as possible. Most notably, this includes the registry. So when creating modular components with this package, resources (and protocols) register their "Creator" methods and such during init() or during main(). They then are explicitly added via AddModelFromRegistry() so that merely importing a module doesn't add unneeded/unused grpc services.

In other languages, and for small modules not part of a larger code ecosystem, the registry concept may not make as much sense, and foregoing the registry step in favor of some more direct AddModel() call (which takes the creation handler func directly) may be better.

Example

package main

import (
	"context"
	"fmt"
	"sync/atomic"

	"github.com/pkg/errors"
	pb "go.viam.com/api/module/v1"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"

	"go.viam.com/rdk/components/generic"
	"go.viam.com/rdk/logging"
	"go.viam.com/rdk/module"
	"go.viam.com/rdk/resource"
)

var (
	logger     = logging.NewLogger("SimpleModule")
	ctx        = context.Background()
	myModel    = resource.NewModel("acme", "demo", "mycounter")
	socketPath = "/tmp/viam-module-example.socket"
)

func main() {
	// Normally we're passed a socket path as the first argument.
	// socketPath := args[1]
	// For this example though, socketPath is hardcoded above.

	// Instantiate the module itself
	myMod, err := module.NewModule(ctx, socketPath, logger)
	if err != nil {
		logger.Error(err)
	}

	// We first put our component's constructor in the registry, then tell the module to load it
	// Note that all resources must be added before the module is started.
	resource.RegisterComponent(
		generic.API,
		myModel,
		resource.Registration[resource.Resource, resource.NoNativeConfig]{Constructor: newCounter})
	myMod.AddModelFromRegistry(ctx, generic.API, myModel)

	// The module is started.
	err = myMod.Start(ctx)
	// Close is deferred and will run automatically when this function returns.
	defer myMod.Close(ctx)
	if err != nil {
		logger.Error(err)
	}

	// Normally a module would then wait for a signal to exit.
	// sigChan := make(chan os.Signal)
	// signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
	// <-sigChan

	// For this example, we'll instead make a quick connection and check things.
	checkReady()

	// The deferred myMod.Close() will now run as the function returns.

}

func checkReady() {
	conn, err := grpc.Dial(
		"unix://"+socketPath,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
	)
	if err != nil {
		logger.Error(err)
	}
	client := pb.NewModuleServiceClient(conn)

	resp, err := client.Ready(ctx, &pb.ReadyRequest{})
	if err != nil {
		logger.Error(err)
	}

	api := resp.Handlermap.GetHandlers()[0].Subtype.Subtype

	fmt.Printf("Ready: %t, ", resp.Ready)
	fmt.Printf("API: %s:%s:%s, ", api.Namespace, api.Type, api.Subtype)
	fmt.Printf("Model: %s\n", resp.Handlermap.GetHandlers()[0].GetModels()[0])
}

// newCounter is used to create a new instance of our specific model. It is called for each component in the robot's config with this model.
func newCounter(
	ctx context.Context,
	deps resource.Dependencies,
	conf resource.Config,
	logger logging.Logger,
) (resource.Resource, error) {
	return &counter{
		name: conf.ResourceName(),
	}, nil
}

// counter is the representation of this model. It holds only a "total" count.
type counter struct {
	resource.TriviallyCloseable
	name  resource.Name
	total int64
}

func (c *counter) Name() resource.Name {
	return c.name
}

func (c *counter) Reconfigure(ctx context.Context, deps resource.Dependencies, conf resource.Config) error {
	return nil
}

// DoCommand is the only method of this component. It looks up the "real" command from the map it's passed.
// Because of this, any arbitrary commands can be received, and any data returned.
func (c *counter) DoCommand(ctx context.Context, req map[string]interface{}) (map[string]interface{}, error) {
	// We look for a map key called "command"
	cmd, ok := req["command"]
	if !ok {
		return nil, errors.New("missing 'command' string")
	}

	// If it's "get" we return the current total.
	if cmd == "get" {
		return map[string]interface{}{"total": atomic.LoadInt64(&c.total)}, nil
	}

	// If it's "add" we atomically add a second key "value" to the total.
	if cmd == "add" {
		_, ok := req["value"]
		if !ok {
			return nil, errors.New("value must exist")
		}
		val, ok := req["value"].(float64)
		if !ok {
			return nil, errors.New("value must be a number")
		}
		atomic.AddInt64(&c.total, int64(val))
		// We return the new total after the addition.
		return map[string]interface{}{"total": atomic.LoadInt64(&c.total)}, nil
	}
	// The command must've been something else.
	return nil, fmt.Errorf("unknown command string %s", cmd)
}

Output:

Ready: true, API: rdk:component:generic, Model: acme:demo:mycounter

Index

• func CheckSocketOwner(address string) error
• func CreateSocketAddress(parentDir, desiredName string) (string, error)
• func MakeSelfOwnedFilesFunc(f func() error) error
• func NewLoggerFromArgs(moduleName string) logging.Logger
• func NewServer(unary []grpc.UnaryServerInterceptor, stream []grpc.StreamServerInterceptor) rpc.Server
• type HandlerMap
  • func NewHandlerMapFromProto(ctx context.Context, pMap *pb.HandlerMap, conn *grpc.ClientConn) (HandlerMap, error)
  • func (h HandlerMap) ToProto() *pb.HandlerMap
• type Module
  • func NewModule(ctx context.Context, address string, logger logging.Logger) (*Module, error)
  • func NewModuleFromArgs(ctx context.Context, logger logging.Logger) (*Module, error)
  • func (m *Module) AddModelFromRegistry(ctx context.Context, api resource.API, model resource.Model) error
  • func (m *Module) AddResource(ctx context.Context, req *pb.AddResourceRequest) (*pb.AddResourceResponse, error)
  • func (m *Module) Close(ctx context.Context)
  • func (m *Module) GetParentResource(ctx context.Context, name resource.Name) (resource.Resource, error)
  • func (m *Module) OperationManager() *operation.Manager
  • func (m *Module) Ready(ctx context.Context, req *pb.ReadyRequest) (*pb.ReadyResponse, error)
  • func (m *Module) ReconfigureResource(ctx context.Context, req *pb.ReconfigureResourceRequest) (*pb.ReconfigureResourceResponse, error)
  • func (m *Module) RemoveResource(ctx context.Context, req *pb.RemoveResourceRequest) (*pb.RemoveResourceResponse, error)
  • func (m *Module) SetReady(ready bool)
  • func (m *Module) Start(ctx context.Context) error
  • func (m *Module) ValidateConfig(ctx context.Context, req *pb.ValidateConfigRequest) (*pb.ValidateConfigResponse, error)
• type Server
  • func (s *Server) EnsureAuthed(ctx context.Context) (context.Context, error)
  • func (s *Server) GRPCHandler() http.Handler
  • func (s *Server) GatewayHandler() http.Handler
  • func (s *Server) InstanceNames() []string
  • func (s *Server) InternalAddr() net.Addr
  • func (s *Server) RegisterServiceServer(ctx context.Context, svcDesc *grpc.ServiceDesc, svcServer interface{}, ...) error
  • func (s *Server) Serve(listener net.Listener) error
  • func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)
  • func (s *Server) ServeTLS(listener net.Listener, certFile, keyFile string, tlsConfig *tls.Config) error
  • func (s *Server) Start() error
  • func (s *Server) Stop() error

Examples

• Package

Functions

func CheckSocketOwner

func CheckSocketOwner(address string) error

CheckSocketOwner verifies that UID of a filepath/socket matches the current process's UID.

func CreateSocketAddress

func CreateSocketAddress(parentDir, desiredName string) (string, error)

CreateSocketAddress returns a socket address of the form parentDir/desiredName.sock if it is shorter than the socketMaxAddressLength. If this path would be too long, this function truncates desiredName and returns parentDir/truncatedName-hashOfDesiredName.sock.

Importantly, this function will return the same socket address as long as the desiredName doesn't change.

func MakeSelfOwnedFilesFunc

func MakeSelfOwnedFilesFunc(f func() error) error

MakeSelfOwnedFilesFunc calls the given function such that any files made will be self owned.

func NewLoggerFromArgs

func NewLoggerFromArgs(moduleName string) logging.Logger

NewLoggerFromArgs can be used to create a logging.Logger at "DebugLevel" if "--log-level=debug" is the third argument in os.Args and at "InfoLevel" otherwise. See config.Module.LogLevel documentation for more info on how to start modules with a "log-level" commandline argument.

func NewServer

func NewServer(unary []grpc.UnaryServerInterceptor, stream []grpc.StreamServerInterceptor) rpc.Server

NewServer returns a new (module specific) rpc.Server.

Types

type HandlerMap

type HandlerMap map[resource.RPCAPI][]resource.Model

HandlerMap is the format for api->model pairs that the module will service. Ex: mymap["rdk:component:motor"] = ["acme:marine:thruster", "acme:marine:outboard"].

func NewHandlerMapFromProto

func NewHandlerMapFromProto(ctx context.Context, pMap *pb.HandlerMap, conn *grpc.ClientConn) (HandlerMap, error)

NewHandlerMapFromProto converts protobuf to HandlerMap.

func (HandlerMap) ToProto

func (h HandlerMap) ToProto() *pb.HandlerMap

ToProto converts the HandlerMap to a protobuf representation.

type Module

type Module struct {
	pb.UnimplementedModuleServiceServer
	// contains filtered or unexported fields
}

Module represents an external resource module that services components/services.

func NewModule

func NewModule(ctx context.Context, address string, logger logging.Logger) (*Module, error)

NewModule returns the basic module framework/structure.

func NewModuleFromArgs

func NewModuleFromArgs(ctx context.Context, logger logging.Logger) (*Module, error)

NewModuleFromArgs directly parses the command line argument to get its address.

func (*Module) AddModelFromRegistry

func (m *Module) AddModelFromRegistry(ctx context.Context, api resource.API, model resource.Model) error

AddModelFromRegistry adds a preregistered component or service model to the module's services.

func (*Module) AddResource

func (m *Module) AddResource(ctx context.Context, req *pb.AddResourceRequest) (*pb.AddResourceResponse, error)

AddResource receives the component/service configuration from the parent.

func (*Module) Close

func (m *Module) Close(ctx context.Context)

Close shuts down the module and grpc server.

func (*Module) GetParentResource

func (m *Module) GetParentResource(ctx context.Context, name resource.Name) (resource.Resource, error)

GetParentResource returns a resource from the parent robot by name.

func (*Module) OperationManager

func (m *Module) OperationManager() *operation.Manager

OperationManager returns the operation manager for the module.

func (*Module) Ready

func (m *Module) Ready(ctx context.Context, req *pb.ReadyRequest) (*pb.ReadyResponse, error)

Ready receives the parent address and reports api/model combos the module is ready to service.

func (*Module) ReconfigureResource

func (m *Module) ReconfigureResource(ctx context.Context, req *pb.ReconfigureResourceRequest) (*pb.ReconfigureResourceResponse, error)

ReconfigureResource receives the component/service configuration from the parent.

func (*Module) RemoveResource

func (m *Module) RemoveResource(ctx context.Context, req *pb.RemoveResourceRequest) (*pb.RemoveResourceResponse, error)

RemoveResource receives the request for resource removal.

func (*Module) SetReady

func (m *Module) SetReady(ready bool)

SetReady can be set to false if the module is not ready (ex. waiting on hardware).

func (*Module) Start

func (m *Module) Start(ctx context.Context) error

Start starts the module service and grpc server.

func (*Module) ValidateConfig

func (m *Module) ValidateConfig(ctx context.Context,
	req *pb.ValidateConfigRequest,
) (*pb.ValidateConfigResponse, error)

ValidateConfig receives the validation request for a resource from the parent.

type Server

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

Server provides an rpc.Server wrapper around a grpc.Server.

func (*Server) EnsureAuthed

func (s *Server) EnsureAuthed(ctx context.Context) (context.Context, error)

EnsureAuthed is unsupported.

func (*Server) GRPCHandler

func (s *Server) GRPCHandler() http.Handler

GRPCHandler is unsupported.

func (*Server) GatewayHandler

func (s *Server) GatewayHandler() http.Handler

GatewayHandler is unsupported.

func (*Server) InstanceNames

func (s *Server) InstanceNames() []string

InstanceNames is unsupported.

func (*Server) InternalAddr

func (s *Server) InternalAddr() net.Addr

InternalAddr returns the internal address of the server.

func (*Server) RegisterServiceServer

func (s *Server) RegisterServiceServer(
	ctx context.Context,
	svcDesc *grpc.ServiceDesc,
	svcServer interface{},
	svcHandlers ...rpc.RegisterServiceHandlerFromEndpointFunc,
) error

RegisterServiceServer associates a service description with its implementation along with any gateway handlers.

func (*Server) Serve

func (s *Server) Serve(listener net.Listener) error

Serve begins listening/serving grpc.

func (*Server) ServeHTTP

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

ServeHTTP is unsupported.

func (*Server) ServeTLS

func (s *Server) ServeTLS(listener net.Listener, certFile, keyFile string, tlsConfig *tls.Config) error

ServeTLS is unsupported.

func (*Server) Start

func (s *Server) Start() error

Start is unsupported.

func (*Server) Stop

func (s *Server) Stop() error

Stop performs a GracefulStop() on the underlying grpc server.