subservers: add new package to manage the lit subservers

subservers/config.go

@@ -0,0 +1,35 @@
+package subservers
+
+// RemoteConfig holds the configuration parameters that are needed when running
+// LiT in the "remote" lnd mode.
+type RemoteConfig struct {
+	LitLogDir         string `long:"lit-logdir" description:"For lnd remote mode only: Directory to log output."`
+	LitMaxLogFiles    int    `long:"lit-maxlogfiles" description:"For lnd remote mode only: Maximum logfiles to keep (0 for no rotation)"`
+	LitMaxLogFileSize int    `long:"lit-maxlogfilesize" description:"For lnd remote mode only: Maximum logfile size in MB"`
+
+	LitDebugLevel string `long:"lit-debuglevel" description:"For lnd remote mode only: Logging level for all subsystems {trace, debug, info, warn, error, critical} -- You may also specify <subsystem>=<level>,<subsystem2>=<level>,... to set the log level for individual subsystems."`
+
+	Lnd     *RemoteDaemonConfig `group:"Remote lnd (use when lnd-mode=remote)" namespace:"lnd"`
+	Faraday *RemoteDaemonConfig `group:"Remote faraday (use when faraday-mode=remote)" namespace:"faraday"`
+	Loop    *RemoteDaemonConfig `group:"Remote loop (use when loop-mode=remote)" namespace:"loop"`
+	Pool    *RemoteDaemonConfig `group:"Remote pool (use when pool-mode=remote)" namespace:"pool"`
+}
+
+// RemoteDaemonConfig holds the configuration parameters that are needed to
+// connect to a remote daemon like lnd for example.
+type RemoteDaemonConfig struct {
+	// RPCServer is host:port that the remote daemon's RPC server is
+	// listening on.
+	RPCServer string `long:"rpcserver" description:"The host:port that the remote daemon is listening for RPC connections on."`
+
+	// MacaroonPath is the path to the single macaroon that should be used
+	// instead of needing to specify the macaroon directory that contains
+	// all of the daemon's macaroons. The specified macaroon MUST have all
+	// permissions that all the subservers use, otherwise permission errors
+	// will occur.
+	MacaroonPath string `long:"macaroonpath" description:"The full path to the single macaroon to use, either the main (admin.macaroon in lnd's case) or a custom baked one. A custom macaroon must contain ALL permissions required for all subservers to work, otherwise permission errors will occur."`
+
+	// TLSCertPath is the path to the tls cert of the remote daemon that
+	// should be used to verify the TLS identity of the remote RPC server.
+	TLSCertPath string `long:"tlscertpath" description:"The full path to the remote daemon's TLS cert to use for RPC connection verification."`
+}

subservers/interface.go

@@ -0,0 +1,48 @@
+package subservers
+
+import (
+	"github.com/lightninglabs/lndclient"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"github.com/lightningnetwork/lnd/macaroons"
+	"google.golang.org/grpc"
+)
+
+// SubServer defines an interface that should be implemented by any sub-server
+// that the subServer manager should manage. A sub-server can be run in either
+// integrated or remote mode. A sub-server is considered non-fatal to LiT
+// meaning that if a sub-server fails to start, LiT can safely continue with its
+// operations and other sub-servers can too.
+type SubServer interface {
+	macaroons.MacaroonValidator
+
+	// Name returns the name of the sub-server.
+	Name() string
+
+	// Remote returns true if the sub-server is running remotely and so
+	// should be connected to instead of spinning up an integrated server.
+	Remote() bool
+
+	// RemoteConfig returns the config required to connect to the sub-server
+	// if it is running in remote mode.
+	RemoteConfig() *RemoteDaemonConfig
+
+	// Start starts the sub-server in integrated mode.
+	Start(lnrpc.LightningClient, *lndclient.GrpcLndServices, bool) error
+
+	// Stop stops the sub-server in integrated mode.
+	Stop() error
+
+	// RegisterGrpcService must register the sub-server's GRPC server with
+	// the given registrar.
+	RegisterGrpcService(grpc.ServiceRegistrar)
+
+	// ServerErrChan returns an error channel that should be listened on
+	// after starting the sub-server to listen for any runtime errors. It
+	// is optional and may be set to nil. This only applies in integrated
+	// mode.
+	ServerErrChan() chan error
+
+	// MacPath returns the path to the sub-server's macaroon if it is not
+	// running in remote mode.
+	MacPath() string
+}

subservers/log.go

@@ -0,0 +1,25 @@
+package subservers
+
+import (
+	"github.com/btcsuite/btclog"
+	"github.com/lightningnetwork/lnd/build"
+)
+
+const Subsystem = "SSVR"
+
+// log is a logger that is initialized with no output filters.  This
+// means the package will not perform any logging by default until the caller
+// requests it.
+var log btclog.Logger
+
+// The default amount of logging is none.
+func init() {
+	UseLogger(build.NewSubLogger(Subsystem, nil))
+}
+
+// UseLogger uses a specified Logger to output package logging info.
+// This should be used in preference to SetLogWriter if the caller is also
+// using btclog.
+func UseLogger(logger btclog.Logger) {
+	log = logger
+}

subservers/manager.go

@@ -0,0 +1,214 @@
+package subservers
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/lightninglabs/lndclient"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	grpcProxy "github.com/mwitkow/grpc-proxy/proxy"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/backoff"
+	"google.golang.org/grpc/credentials"
+	"gopkg.in/macaroon-bakery.v2/bakery"
+)
+
+var (
+	// maxMsgRecvSize is the largest message our REST proxy will receive. We
+	// set this to 200MiB atm.
+	maxMsgRecvSize = grpc.MaxCallRecvMsgSize(1 * 1024 * 1024 * 200)
+
+	// defaultConnectTimeout is the default timeout for connecting to the
+	// backend.
+	defaultConnectTimeout = 15 * time.Second
+)
+
+// Manager manages a set of subServer objects.
+type Manager struct {
+	servers []*subServerWrapper
+	mu      sync.RWMutex
+}
+
+// NewManager constructs a new subServerMgr.
+func NewManager() *Manager {
+	return &Manager{}
+}
+
+// AddServer adds a new subServer to the manager's set.
+func (s *Manager) AddServer(ss SubServer) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	s.servers = append(s.servers, &subServerWrapper{
+		subServer: ss,
+		quit:      make(chan struct{}),
+	})
+}
+
+// StartIntegratedServers starts all the manager's sub-servers that should be
+// started in integrated mode.
+func (s *Manager) StartIntegratedServers(lndClient lnrpc.LightningClient,
+	lndGrpc *lndclient.GrpcLndServices, withMacaroonService bool) error {
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for _, ss := range s.servers {
+		if ss.subServer.Remote() {
+			continue
+		}
+
+		err := ss.startIntegrated(
+			lndClient, lndGrpc, withMacaroonService,
+		)
+		if err != nil {
+			return fmt.Errorf("Unable to start %v in integrated "+
+				"mode: %v", ss.subServer.Name(), err)
+		}
+	}
+
+	return nil
+}
+
+// ConnectRemoteSubServers creates connections to all the manager's sub-servers
+// that are running remotely.
+func (s *Manager) ConnectRemoteSubServers() {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	for _, ss := range s.servers {
+		if !ss.subServer.Remote() {
+			continue
+		}
+
+		err := ss.connectRemote()
+		if err != nil {
+			log.Errorf("Failed to connect to remote %s: %v",
+				ss.subServer.Name(), err)
+
+			continue
+		}
+	}
+}
+
+// RegisterRPCServices registers all the manager's sub-servers with the given
+// grpc registrar.
+func (s *Manager) RegisterRPCServices(server grpc.ServiceRegistrar) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	for _, ss := range s.servers {
+		// In remote mode the "director" of the RPC proxy will act as
+		// a catch-all for any gRPC request that isn't known because we
+		// didn't register any server for it. The director will then
+		// forward the request to the remote service.
+		if ss.subServer.Remote() {
+			continue
+		}
+
+		ss.subServer.RegisterGrpcService(server)
+	}
+}
+
+// ValidateMacaroon checks if any of the manager's sub-servers owns the given
+// uri and if so, if it is running in remote mode, then true is returned since
+// the macaroon will be validated by the remote subserver itself when the
+// request arrives. Otherwise, the integrated sub-server's validator validates
+// the macaroon.
+func (s *Manager) ValidateMacaroon(ctx context.Context,
+	requiredPermissions []bakery.Op, uri string) (bool, error) {
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	for _, ss := range s.servers {
+		// TODO(positiveblue): check subserver permissions.
+
+		// If the sub-server is running in remote mode, then we don't
+		// need to validate the macaroon here since the remote server
+		// will do it when the request arrives. But we have handled the
+		// request, as we were able to identify it.
+		if ss.subServer.Remote() {
+			return true, nil
+		}
+
+		// If the sub-server hasn't started yet, then we can't validate
+		// the macaroon. But we know that we can handle the request, as
+		// we were able to identify it.
+		if !ss.started() {
+			return true, fmt.Errorf("%s is not yet ready for "+
+				"requests, the subserver has not started or "+
+				"lnd still starting/syncing",
+				ss.subServer.Name())
+		}
+
+		// Validate the macaroon with the integrated sub-server's own
+		// validator.
+		err := ss.subServer.ValidateMacaroon(
+			ctx, requiredPermissions, uri,
+		)
+		if err != nil {
+			return true, fmt.Errorf("invalid macaroon: %v", err)
+		}
+
+		// The macaroon is valid for this sub-server, we can return
+		// early.
+		return true, nil
+	}
+
+	// No sub-server owns the given uri, so we haven't handled this call.
+	return false, nil
+}
+
+// Stop stops all the manager's sub-servers
+func (s *Manager) Stop() error {
+	var returnErr error
+
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	for _, ss := range s.servers {
+		if ss.subServer.Remote() {
+			continue
+		}
+
+		err := ss.stop()
+		if err != nil {
+			log.Errorf("Error stopping %s: %v", ss.subServer.Name(),
+				err)
+			returnErr = err
+		}
+	}
+
+	return returnErr
+}
+
+func dialBackend(name, dialAddr, tlsCertPath string) (*grpc.ClientConn, error) {
+	tlsConfig, err := credentials.NewClientTLSFromFile(tlsCertPath, "")
+	if err != nil {
+		return nil, fmt.Errorf("could not read %s TLS cert %s: %v",
+			name, tlsCertPath, err)
+	}
+
+	opts := []grpc.DialOption{
+		// From the grpcProxy doc: This codec is *crucial* to the
+		// functioning of the proxy.
+		grpc.WithCodec(grpcProxy.Codec()), // nolint
+		grpc.WithTransportCredentials(tlsConfig),
+		grpc.WithDefaultCallOptions(maxMsgRecvSize),
+		grpc.WithConnectParams(grpc.ConnectParams{
+			Backoff:           backoff.DefaultConfig,
+			MinConnectTimeout: defaultConnectTimeout,
+		}),
+	}
+
+	log.Infof("Dialing %s gRPC server at %s", name, dialAddr)
+	cc, err := grpc.Dial(dialAddr, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("failed dialing %s backend: %v", name,
+			err)
+	}
+	return cc, nil
+}

subservers/subserver.go

@@ -0,0 +1,146 @@
+package subservers
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/lightninglabs/lndclient"
+	"github.com/lightningnetwork/lnd/lncfg"
+	"github.com/lightningnetwork/lnd/lnrpc"
+	"google.golang.org/grpc"
+)
+
+const (
+	LND string = "lnd"
+	LIT string = "lit"
+)
+
+// subServerWrapper is a wrapper around the SubServer interface and is used by
+// the subServerMgr to manage a SubServer.
+type subServerWrapper struct {
+	integratedStarted bool
+	startedMu         sync.RWMutex
+
+	stopped sync.Once
+
+	subServer SubServer
+
+	remoteConn *grpc.ClientConn
+
+	wg   sync.WaitGroup
+	quit chan struct{}
+}
+
+// started returns true if the subServer has been started. This only applies if
+// the subServer is running in integrated mode.
+func (s *subServerWrapper) started() bool {
+	s.startedMu.RLock()
+	defer s.startedMu.RUnlock()
+
+	return s.integratedStarted
+}
+
+// setStarted sets the subServer as started or not. This only applies if the
+// subServer is running in integrated mode.
+func (s *subServerWrapper) setStarted(started bool) {
+	s.startedMu.Lock()
+	defer s.startedMu.Unlock()
+
+	s.integratedStarted = started
+}
+
+// stop the subServer by closing the connection to it if it is remote or by
+// stopping the integrated process.
+func (s *subServerWrapper) stop() error {
+	// If the sub-server has not yet started, then we can exit early.
+	if !s.started() {
+		return nil
+	}
+
+	var returnErr error
+	s.stopped.Do(func() {
+		close(s.quit)
+		s.wg.Wait()
+
+		// If running in remote mode, close the connection.
+		if s.subServer.Remote() && s.remoteConn != nil {
+			err := s.remoteConn.Close()
+			if err != nil {
+				returnErr = fmt.Errorf("could not close "+
+					"remote connection: %v", err)
+			}
+			return
+		}
+
+		// Else, stop the integrated sub-server process.
+		err := s.subServer.Stop()
+		if err != nil {
+			returnErr = fmt.Errorf("could not close "+
+				"integrated connection: %v", err)
+			return
+		}
+
+		if s.subServer.ServerErrChan() == nil {
+			return
+		}
+
+		select {
+		case returnErr = <-s.subServer.ServerErrChan():
+		default:
+		}
+	})
+
+	return returnErr
+}
+
+// startIntegrated starts the subServer in integrated mode.
+func (s *subServerWrapper) startIntegrated(lndClient lnrpc.LightningClient,
+	lndGrpc *lndclient.GrpcLndServices, withMacaroonService bool) error {
+
+	err := s.subServer.Start(lndClient, lndGrpc, withMacaroonService)
+	if err != nil {
+		return err
+	}
+	s.setStarted(true)
+
+	if s.subServer.ServerErrChan() == nil {
+		return nil
+	}
+
+	s.wg.Add(1)
+	go func() {
+		defer s.wg.Done()
+
+		select {
+		case err := <-s.subServer.ServerErrChan():
+			// The sub server should shut itself down if an error
+			// happens. We don't need to try to stop it again.
+			s.setStarted(false)
+
+			err = fmt.Errorf("received critical error from "+
+				"sub-server (%s), shutting down: %v",
+				s.subServer.Name(), err)
+
+			log.Error(err)
+
+		case <-s.quit:
+		}
+	}()
+
+	return nil
+}
+
+// connectRemote attempts to make a connection to the remote sub-server.
+func (s *subServerWrapper) connectRemote() error {
+	cfg := s.subServer.RemoteConfig()
+	certPath := lncfg.CleanAndExpandPath(cfg.TLSCertPath)
+	name := s.subServer.Name()
+	conn, err := dialBackend(name, cfg.RPCServer, certPath)
+	if err != nil {
+		return fmt.Errorf("remote dial error: %v", err)
+	}
+
+	s.remoteConn = conn
+
+	return nil
+}