Merge remote-tracking branch 'origin/master' into InsanePrawn-master

daemon/job/active.go

@@ -2,9 +2,12 @@ package job
 
 import (
 	"context"
+	"sync"
+	"time"
+
 	"github.com/pkg/errors"
-	"github.com/problame/go-streamrpc"
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/job/reset"
@@ -12,32 +15,30 @@ import (
 	"github.com/zrepl/zrepl/daemon/logging"
 	"github.com/zrepl/zrepl/daemon/pruner"
 	"github.com/zrepl/zrepl/daemon/snapper"
-	"github.com/zrepl/zrepl/daemon/transport/connecter"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/replication"
+	"github.com/zrepl/zrepl/rpc"
+	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/transport/fromconfig"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/zfs"
-	"sync"
-	"time"
 )
 
 type ActiveSide struct {
-	mode          activeMode
-	name          string
-	clientFactory *connecter.ClientFactory
+	mode      activeMode
+	name      string
+	connecter transport.Connecter
 
 	prunerFactory *pruner.PrunerFactory
 
-
-	promRepStateSecs *prometheus.HistogramVec // labels: state
-	promPruneSecs *prometheus.HistogramVec // labels: prune_side
-	promBytesReplicated *prometheus.CounterVec // labels: filesystem
+	promRepStateSecs    *prometheus.HistogramVec // labels: state
+	promPruneSecs       *prometheus.HistogramVec // labels: prune_side
+	promBytesReplicated *prometheus.CounterVec   // labels: filesystem
 
 	tasksMtx sync.Mutex
 	tasks    activeSideTasks
 }
 
-
 //go:generate enumer -type=ActiveSideState
 type ActiveSideState int
 
@@ -48,12 +49,11 @@ const (
 	ActiveSideDone // also errors
 )
 
-
 type activeSideTasks struct {
 	state ActiveSideState
 
 	// valid for state ActiveSideReplicating, ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
-	replication *replication.Replication
+	replication       *replication.Replication
 	replicationCancel context.CancelFunc
 
 	// valid for state ActiveSidePruneSender, ActiveSidePruneReceiver, ActiveSideDone
@@ -77,28 +77,59 @@ func (a *ActiveSide) updateTasks(u func(*activeSideTasks)) activeSideTasks {
 }
 
 type activeMode interface {
-	SenderReceiver(client *streamrpc.Client) (replication.Sender, replication.Receiver, error)
+	ConnectEndpoints(rpcLoggers rpc.Loggers, connecter transport.Connecter)
+	DisconnectEndpoints()
+	SenderReceiver() (replication.Sender, replication.Receiver)
 	Type() Type
 	RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{})
+	ResetConnectBackoff()
 }
 
 type modePush struct {
-	fsfilter         endpoint.FSFilter
-	snapper *snapper.PeriodicOrManual
+	setupMtx sync.Mutex
+	sender   *endpoint.Sender
+	receiver *rpc.Client
+	fsfilter endpoint.FSFilter
+	snapper  *snapper.PeriodicOrManual
 }
 
-func (m *modePush) SenderReceiver(client *streamrpc.Client) (replication.Sender, replication.Receiver, error) {
-	sender := endpoint.NewSender(m.fsfilter)
-	receiver := endpoint.NewRemote(client)
-	return sender, receiver, nil
+func (m *modePush) ConnectEndpoints(loggers rpc.Loggers, connecter transport.Connecter) {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	if m.receiver != nil || m.sender != nil {
+		panic("inconsistent use of ConnectEndpoints and DisconnectEndpoints")
+	}
+	m.sender = endpoint.NewSender(m.fsfilter)
+	m.receiver = rpc.NewClient(connecter, loggers)
+}
+
+func (m *modePush) DisconnectEndpoints() {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	m.receiver.Close()
+	m.sender = nil
+	m.receiver = nil
+}
+
+func (m *modePush) SenderReceiver() (replication.Sender, replication.Receiver) {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	return m.sender, m.receiver
 }
 
 func (m *modePush) Type() Type { return TypePush }
 
-func (m *modePush) RunPeriodic(ctx context.Context, wakeUpCommon chan <- struct{}) {
+func (m *modePush) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}) {
 	m.snapper.Run(ctx, wakeUpCommon)
 }
 
+func (m *modePush) ResetConnectBackoff() {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	if m.receiver != nil {
+		m.receiver.ResetConnectBackoff()
+	}
+}
 
 func modePushFromConfig(g *config.Global, in *config.PushJob) (*modePush, error) {
 	m := &modePush{}
@@ -116,14 +147,35 @@ func modePushFromConfig(g *config.Global, in *config.PushJob) (*modePush, error)
 }
 
 type modePull struct {
+	setupMtx sync.Mutex
+	receiver *endpoint.Receiver
+	sender   *rpc.Client
 	rootFS   *zfs.DatasetPath
 	interval time.Duration
 }
 
-func (m *modePull) SenderReceiver(client *streamrpc.Client) (replication.Sender, replication.Receiver, error) {
-	sender := endpoint.NewRemote(client)
-	receiver, err := endpoint.NewReceiver(m.rootFS)
-	return sender, receiver, err
+func (m *modePull) ConnectEndpoints(loggers rpc.Loggers, connecter transport.Connecter) {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	if m.receiver != nil || m.sender != nil {
+		panic("inconsistent use of ConnectEndpoints and DisconnectEndpoints")
+	}
+	m.receiver = endpoint.NewReceiver(m.rootFS, false)
+	m.sender = rpc.NewClient(connecter, loggers)
+}
+
+func (m *modePull) DisconnectEndpoints() {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	m.sender.Close()
+	m.sender = nil
+	m.receiver = nil
+}
+
+func (m *modePull) SenderReceiver() (replication.Sender, replication.Receiver) {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	return m.sender, m.receiver
 }
 
 func (*modePull) Type() Type { return TypePull }
@@ -148,6 +200,14 @@ func (m *modePull) RunPeriodic(ctx context.Context, wakeUpCommon chan<- struct{}
 	}
 }
 
+func (m *modePull) ResetConnectBackoff() {
+	m.setupMtx.Lock()
+	defer m.setupMtx.Unlock()
+	if m.sender != nil {
+		m.sender.ResetConnectBackoff()
+	}
+}
+
 func modePullFromConfig(g *config.Global, in *config.PullJob) (m *modePull, err error) {
 	m = &modePull{}
 	if in.Interval <= 0 {
@@ -175,17 +235,17 @@ func activeSide(g *config.Global, in *config.ActiveJob, mode activeMode) (j *Act
 		Subsystem:   "replication",
 		Name:        "state_time",
 		Help:        "seconds spent during replication",
-		ConstLabels: prometheus.Labels{"zrepl_job":j.name},
+		ConstLabels: prometheus.Labels{"zrepl_job": j.name},
 	}, []string{"state"})
 	j.promBytesReplicated = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace:   "zrepl",
 		Subsystem:   "replication",
 		Name:        "bytes_replicated",
 		Help:        "number of bytes replicated from sender to receiver per filesystem",
-		ConstLabels: prometheus.Labels{"zrepl_job":j.name},
+		ConstLabels: prometheus.Labels{"zrepl_job": j.name},
 	}, []string{"filesystem"})
 
-	j.clientFactory, err = connecter.FromConfig(g, in.Connect)
+	j.connecter, err = fromconfig.ConnecterFromConfig(g, in.Connect)
 	if err != nil {
 		return nil, errors.Wrap(err, "cannot build client")
 	}
@@ -195,7 +255,7 @@ func activeSide(g *config.Global, in *config.ActiveJob, mode activeMode) (j *Act
 		Subsystem:   "pruning",
 		Name:        "time",
 		Help:        "seconds spent in pruner",
-		ConstLabels: prometheus.Labels{"zrepl_job":j.name},
+		ConstLabels: prometheus.Labels{"zrepl_job": j.name},
 	}, []string{"prune_side"})
 	j.prunerFactory, err = pruner.NewPrunerFactory(in.Pruning, j.promPruneSecs)
 	if err != nil {
@@ -214,7 +274,7 @@ func (j *ActiveSide) RegisterMetrics(registerer prometheus.Registerer) {
 func (j *ActiveSide) Name() string { return j.name }
 
 type ActiveSideStatus struct {
-	Replication *replication.Report
+	Replication                    *replication.Report
 	PruningSender, PruningReceiver *pruner.Report
 }
 
@@ -256,6 +316,7 @@ outer:
 			break outer
 
 		case <-wakeup.Wait(ctx):
+			j.mode.ResetConnectBackoff()
 		case <-periodicDone:
 		}
 		invocationCount++
@@ -268,6 +329,9 @@ func (j *ActiveSide) do(ctx context.Context) {
 
 	log := GetLogger(ctx)
 	ctx = logging.WithSubsystemLoggers(ctx, log)
+	loggers := rpc.GetLoggersOrPanic(ctx) // filled by WithSubsystemLoggers
+	j.mode.ConnectEndpoints(loggers, j.connecter)
+	defer j.mode.DisconnectEndpoints()
 
 	// allow cancellation of an invocation (this function)
 	ctx, cancelThisRun := context.WithCancel(ctx)
@@ -353,13 +417,7 @@ func (j *ActiveSide) do(ctx context.Context) {
 		}
 	}()
 
-	client, err := j.clientFactory.NewClient()
-	if err != nil {
-		log.WithError(err).Error("factory cannot instantiate streamrpc client")
-	}
-	defer client.Close(ctx)
-
-	sender, receiver, err := j.mode.SenderReceiver(client)
+	sender, receiver := j.mode.SenderReceiver()
 
 	{
 		select {

daemon/job/passive.go

@@ -2,28 +2,30 @@ package job
 
 import (
 	"context"
+	"fmt"
+
 	"github.com/pkg/errors"
-	"github.com/problame/go-streamrpc"
 	"github.com/prometheus/client_golang/prometheus"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/filters"
 	"github.com/zrepl/zrepl/daemon/logging"
-	"github.com/zrepl/zrepl/daemon/transport/serve"
 	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/rpc"
+	"github.com/zrepl/zrepl/transport"
+	"github.com/zrepl/zrepl/transport/fromconfig"
 	"github.com/zrepl/zrepl/zfs"
-	"path"
 )
 
 type PassiveSide struct {
-	mode passiveMode
-	name     string
-	l        serve.ListenerFactory
-	rpcConf  *streamrpc.ConnConfig
+	mode   passiveMode
+	name   string
+	listen transport.AuthenticatedListenerFactory
 }
 
 type passiveMode interface {
-	ConnHandleFunc(ctx context.Context, conn serve.AuthenticatedConn) streamrpc.HandlerFunc
+	Handler() rpc.Handler
 	RunPeriodic(ctx context.Context)
 	Type() Type
 }
@@ -34,26 +36,8 @@ type modeSink struct {
 
 func (m *modeSink) Type() Type { return TypeSink }
 
-func (m *modeSink) ConnHandleFunc(ctx context.Context, conn serve.AuthenticatedConn) streamrpc.HandlerFunc {
-	log := GetLogger(ctx)
-
-	clientRootStr := path.Join(m.rootDataset.ToString(), conn.ClientIdentity())
-	clientRoot, err := zfs.NewDatasetPath(clientRootStr)
-	if err != nil {
-		log.WithError(err).
-			WithField("client_identity", conn.ClientIdentity()).
-			Error("cannot build client filesystem map (client identity must be a valid ZFS FS name")
-	}
-	log.WithField("client_root", clientRoot).Debug("client root")
-
-	local, err := endpoint.NewReceiver(clientRoot)
-	if err != nil {
-		log.WithError(err).Error("unexpected error: cannot convert mapping to filter")
-		return nil
-	}
-
-	h := endpoint.NewHandler(local)
-	return h.Handle
+func (m *modeSink) Handler() rpc.Handler {
+	return endpoint.NewReceiver(m.rootDataset, true)
 }
 
 func (m *modeSink) RunPeriodic(_ context.Context) {}
@@ -72,7 +56,7 @@ func modeSinkFromConfig(g *config.Global, in *config.SinkJob) (m *modeSink, err
 
 type modeSource struct {
 	fsfilter zfs.DatasetFilter
-	snapper *snapper.PeriodicOrManual
+	snapper  *snapper.PeriodicOrManual
 }
 
 func modeSourceFromConfig(g *config.Global, in *config.SourceJob) (m *modeSource, err error) {
@@ -93,10 +77,8 @@ func modeSourceFromConfig(g *config.Global, in *config.SourceJob) (m *modeSource
 
 func (m *modeSource) Type() Type { return TypeSource }
 
-func (m *modeSource) ConnHandleFunc(ctx context.Context, conn serve.AuthenticatedConn) streamrpc.HandlerFunc {
-	sender := endpoint.NewSender(m.fsfilter)
-	h := endpoint.NewHandler(sender)
-	return h.Handle
+func (m *modeSource) Handler() rpc.Handler {
+	return endpoint.NewSender(m.fsfilter)
 }
 
 func (m *modeSource) RunPeriodic(ctx context.Context) {
@@ -106,8 +88,8 @@ func (m *modeSource) RunPeriodic(ctx context.Context) {
 func passiveSideFromConfig(g *config.Global, in *config.PassiveJob, mode passiveMode) (s *PassiveSide, err error) {
 
 	s = &PassiveSide{mode: mode, name: in.Name}
-	if s.l, s.rpcConf, err = serve.FromConfig(g, in.Serve); err != nil {
-		return nil, errors.Wrap(err, "cannot build server")
+	if s.listen, err = fromconfig.ListenerFactoryFromConfig(g, in.Serve); err != nil {
+		return nil, errors.Wrap(err, "cannot build listener factory")
 	}
 
 	return s, nil
@@ -115,7 +97,7 @@ func passiveSideFromConfig(g *config.Global, in *config.PassiveJob, mode passive
 
 func (j *PassiveSide) Name() string { return j.name }
 
-type PassiveStatus struct {}
+type PassiveStatus struct{}
 
 func (s *PassiveSide) Status() *Status {
 	return &Status{Type: s.mode.Type()} // FIXME PassiveStatus
@@ -127,70 +109,30 @@ func (j *PassiveSide) Run(ctx context.Context) {
 
 	log := GetLogger(ctx)
 	defer log.Info("job exiting")
-
-	l, err := j.l.Listen()
-	if err != nil {
-		log.WithError(err).Error("cannot listen")
-		return
-	}
-	defer l.Close()
-
+	ctx = logging.WithSubsystemLoggers(ctx, log)
 	{
-		ctx, cancel := context.WithCancel(logging.WithSubsystemLoggers(ctx, log)) // shadowing
+		ctx, cancel := context.WithCancel(ctx) // shadowing
 		defer cancel()
 		go j.mode.RunPeriodic(ctx)
 	}
 
-	log.WithField("addr", l.Addr()).Debug("accepting connections")
-	var connId int
-outer:
-	for {
-
-		select {
-		case res := <-accept(ctx, l):
-			if res.err != nil {
-				log.WithError(res.err).Info("accept error")
-				continue
-			}
-			conn := res.conn
-			connId++
-			connLog := log.
-				WithField("connID", connId)
-			connLog.
-				WithField("addr", conn.RemoteAddr()).
-				WithField("client_identity", conn.ClientIdentity()).
-				Info("handling connection")
-			go func() {
-				defer connLog.Info("finished handling connection")
-				defer conn.Close()
-				ctx := logging.WithSubsystemLoggers(ctx, connLog)
-				handleFunc := j.mode.ConnHandleFunc(ctx, conn)
-				if handleFunc == nil {
-					return
-				}
-				if err := streamrpc.ServeConn(ctx, conn, j.rpcConf, handleFunc); err != nil {
-					log.WithError(err).Error("error serving client")
-				}
-			}()
-
-		case <-ctx.Done():
-			break outer
-		}
-
+	handler := j.mode.Handler()
+	if handler == nil {
+		panic(fmt.Sprintf("implementation error: j.mode.Handler() returned nil: %#v", j))
 	}
 
-}
+	ctxInterceptor := func(handlerCtx context.Context) context.Context {
+		return logging.WithSubsystemLoggers(handlerCtx, log)
+	}
 
-type acceptResult struct {
-	conn serve.AuthenticatedConn
-	err  error
-}
+	rpcLoggers := rpc.GetLoggersOrPanic(ctx) // WithSubsystemLoggers above
+	server := rpc.NewServer(handler, rpcLoggers, ctxInterceptor)
 
-func accept(ctx context.Context, listener serve.AuthenticatedListener) <-chan acceptResult {
-	c := make(chan acceptResult, 1)
-	go func() {
-		conn, err := listener.Accept(ctx)
-		c <- acceptResult{conn, err}
-	}()
-	return c
+	listener, err := j.listen()
+	if err != nil {
+		log.WithError(err).Error("cannot listen")
+		return
+	}
+
+	server.Serve(ctx, listener)
 }

daemon/logging/adaptors.go

@@ -1,32 +0,0 @@
-package logging
-
-import (
-	"fmt"
-	"github.com/problame/go-streamrpc"
-	"github.com/zrepl/zrepl/logger"
-	"strings"
-)
-
-type streamrpcLogAdaptor = twoClassLogAdaptor
-
-type twoClassLogAdaptor struct {
-	logger.Logger
-}
-
-var _ streamrpc.Logger = twoClassLogAdaptor{}
-
-func (a twoClassLogAdaptor) Errorf(fmtStr string, args ...interface{}) {
-	const errorSuffix = ": %s"
-	if len(args) == 1 {
-		if err, ok := args[0].(error); ok && strings.HasSuffix(fmtStr, errorSuffix) {
-			msg := strings.TrimSuffix(fmtStr, errorSuffix)
-			a.WithError(err).Error(msg)
-			return
-		}
-	}
-	a.Logger.Error(fmt.Sprintf(fmtStr, args...))
-}
-
-func (a twoClassLogAdaptor) Infof(fmtStr string, args ...interface{}) {
-	a.Logger.Debug(fmt.Sprintf(fmtStr, args...))
-}

daemon/logging/build_logging.go

@@ -4,18 +4,22 @@ import (
 	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"log/syslog"
+	"os"
+
 	"github.com/mattn/go-isatty"
 	"github.com/pkg/errors"
-	"github.com/problame/go-streamrpc"
+
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/pruner"
+	"github.com/zrepl/zrepl/daemon/snapper"
 	"github.com/zrepl/zrepl/endpoint"
 	"github.com/zrepl/zrepl/logger"
 	"github.com/zrepl/zrepl/replication"
+	"github.com/zrepl/zrepl/rpc"
+	"github.com/zrepl/zrepl/rpc/transportmux"
 	"github.com/zrepl/zrepl/tlsconf"
-	"os"
-	"github.com/zrepl/zrepl/daemon/snapper"
-	"github.com/zrepl/zrepl/daemon/transport/serve"
+	"github.com/zrepl/zrepl/transport"
 )
 
 func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error) {
@@ -60,22 +64,41 @@ func OutletsFromConfig(in config.LoggingOutletEnumList) (*logger.Outlets, error)
 
 }
 
+type Subsystem string
+
 const (
-	SubsysReplication = "repl"
-	SubsysStreamrpc   = "rpc"
-	SubsyEndpoint     = "endpoint"
+	SubsysReplication  Subsystem = "repl"
+	SubsyEndpoint      Subsystem = "endpoint"
+	SubsysPruning      Subsystem = "pruning"
+	SubsysSnapshot     Subsystem = "snapshot"
+	SubsysTransport    Subsystem = "transport"
+	SubsysTransportMux Subsystem = "transportmux"
+	SubsysRPC          Subsystem = "rpc"
+	SubsysRPCControl   Subsystem = "rpc.ctrl"
+	SubsysRPCData      Subsystem = "rpc.data"
 )
 
 func WithSubsystemLoggers(ctx context.Context, log logger.Logger) context.Context {
-	ctx = replication.WithLogger(ctx, log.WithField(SubsysField, "repl"))
-	ctx = streamrpc.ContextWithLogger(ctx, streamrpcLogAdaptor{log.WithField(SubsysField, "rpc")})
-	ctx = endpoint.WithLogger(ctx, log.WithField(SubsysField, "endpoint"))
-	ctx = pruner.WithLogger(ctx, log.WithField(SubsysField, "pruning"))
-	ctx = snapper.WithLogger(ctx, log.WithField(SubsysField, "snapshot"))
-	ctx = serve.WithLogger(ctx, log.WithField(SubsysField, "serve"))
+	ctx = replication.WithLogger(ctx, log.WithField(SubsysField, SubsysReplication))
+	ctx = endpoint.WithLogger(ctx, log.WithField(SubsysField, SubsyEndpoint))
+	ctx = pruner.WithLogger(ctx, log.WithField(SubsysField, SubsysPruning))
+	ctx = snapper.WithLogger(ctx, log.WithField(SubsysField, SubsysSnapshot))
+	ctx = transport.WithLogger(ctx, log.WithField(SubsysField, SubsysTransport))
+	ctx = transportmux.WithLogger(ctx, log.WithField(SubsysField, SubsysTransportMux))
+	ctx = rpc.WithLoggers(ctx,
+		rpc.Loggers{
+			General: log.WithField(SubsysField, SubsysRPC),
+			Control: log.WithField(SubsysField, SubsysRPCControl),
+			Data:    log.WithField(SubsysField, SubsysRPCData),
+		},
+	)
 	return ctx
 }
 
+func LogSubsystem(log logger.Logger, subsys Subsystem) logger.Logger {
+	return log.ReplaceField(SubsysField, subsys)
+}
+
 func parseLogFormat(i interface{}) (f EntryFormatter, err error) {
 	var is string
 	switch j := i.(type) {
@@ -200,6 +223,7 @@ func parseSyslogOutlet(in *config.SyslogLoggingOutlet, formatter EntryFormatter)
 	out = &SyslogOutlet{}
 	out.Formatter = formatter
 	out.Formatter.SetMetadataFlags(MetadataNone)
+	out.Facility = syslog.Priority(*in.Facility)
 	out.RetryInterval = in.RetryInterval
 	return out, nil
 }

daemon/logging/logging_outlets.go

@@ -124,6 +124,7 @@ func (h *TCPOutlet) WriteEntry(e logger.Entry) error {
 type SyslogOutlet struct {
 	Formatter          EntryFormatter
 	RetryInterval      time.Duration
+	Facility           syslog.Priority
 	writer             *syslog.Writer
 	lastConnectAttempt time.Time
 }
@@ -142,7 +143,7 @@ func (o *SyslogOutlet) WriteEntry(entry logger.Entry) error {
 		if now.Sub(o.lastConnectAttempt) < o.RetryInterval {
 			return nil // not an error toward logger
 		}
-		o.writer, err = syslog.New(syslog.LOG_LOCAL0, "zrepl")
+		o.writer, err = syslog.New(o.Facility, "zrepl")
 		o.lastConnectAttempt = time.Now()
 		if err != nil {
 			o.writer = nil

daemon/prometheus.go

@@ -7,6 +7,7 @@ import (
 	"github.com/zrepl/zrepl/config"
 	"github.com/zrepl/zrepl/daemon/job"
 	"github.com/zrepl/zrepl/logger"
+	"github.com/zrepl/zrepl/rpc/dataconn/frameconn"
 	"github.com/zrepl/zrepl/zfs"
 	"net"
 	"net/http"
@@ -49,6 +50,10 @@ func (j *prometheusJob) Run(ctx context.Context) {
 		panic(err)
 	}
 
+	if err := frameconn.PrometheusRegister(prometheus.DefaultRegisterer); err != nil {
+		panic(err)
+	}
+
 	log := job.GetLogger(ctx)
 
 	l, err := net.Listen("tcp", j.listen)

daemon/pruner/pruner.go

@@ -11,7 +11,6 @@ import (
 	"github.com/zrepl/zrepl/replication/pdu"
 	"github.com/zrepl/zrepl/util/envconst"
 	"github.com/zrepl/zrepl/util/watchdog"
-	"github.com/problame/go-streamrpc"
 	"net"
 	"sort"
 	"strings"
@@ -19,14 +18,15 @@ import (
 	"time"
 )
 
-// Try to keep it compatible with gitub.com/zrepl/zrepl/replication.Endpoint
+// Try to keep it compatible with gitub.com/zrepl/zrepl/endpoint.Endpoint
 type History interface {
 	ReplicationCursor(ctx context.Context, req *pdu.ReplicationCursorReq) (*pdu.ReplicationCursorRes, error)
 }
 
+// Try to keep it compatible with gitub.com/zrepl/zrepl/endpoint.Endpoint
 type Target interface {
-	ListFilesystems(ctx context.Context) ([]*pdu.Filesystem, error)
-	ListFilesystemVersions(ctx context.Context, fs string) ([]*pdu.FilesystemVersion, error) // fix depS
+	ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error)
+	ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error)
 	DestroySnapshots(ctx context.Context, req *pdu.DestroySnapshotsReq) (*pdu.DestroySnapshotsRes, error)
 }
 
@@ -391,7 +391,6 @@ type Error interface {
 }
 
 var _ Error = net.Error(nil)
-var _ Error = streamrpc.Error(nil)
 
 func shouldRetry(e error) bool {
 	if neterr, ok := e.(net.Error); ok {
@@ -426,10 +425,11 @@ func statePlan(a *args, u updater) state {
 		ka = &pruner.Progress
 	})
 
-	tfss, err := target.ListFilesystems(ctx)
+	tfssres, err := target.ListFilesystems(ctx, &pdu.ListFilesystemReq{})
 	if err != nil {
 		return onErr(u, err)
 	}
+	tfss := tfssres.GetFilesystems()
 
 	pfss := make([]*fs, len(tfss))
 	for i, tfs := range tfss {
@@ -443,11 +443,12 @@ func statePlan(a *args, u updater) state {
 		}
 		pfss[i] = pfs
 
-		tfsvs, err := target.ListFilesystemVersions(ctx, tfs.Path)
+		tfsvsres, err := target.ListFilesystemVersions(ctx, &pdu.ListFilesystemVersionsReq{Filesystem: tfs.Path})
 		if err != nil {
 			l.WithError(err).Error("cannot list filesystem versions")
 			return onErr(u, err)
 		}
+		tfsvs := tfsvsres.GetVersions()
 		// no progress here since we could run in a live-lock (must have used target AND receiver before progress)
 
 		pfs.snaps = make([]pruning.Snapshot, 0, len(tfsvs))

daemon/pruner/pruner_test.go

@@ -44,7 +44,7 @@ type mockTarget struct {
 	destroyErrs        map[string][]error
 }
 
-func (t *mockTarget) ListFilesystems(ctx context.Context) ([]*pdu.Filesystem, error) {
+func (t *mockTarget) ListFilesystems(ctx context.Context, req *pdu.ListFilesystemReq) (*pdu.ListFilesystemRes, error) {
 	if len(t.listFilesystemsErr) > 0 {
 		e := t.listFilesystemsErr[0]
 		t.listFilesystemsErr = t.listFilesystemsErr[1:]
@@ -54,10 +54,11 @@ func (t *mockTarget) ListFilesystems(ctx context.Context) ([]*pdu.Filesystem, er
 	for i := range fss {
 		fss[i] = t.fss[i].Filesystem()
 	}
-	return fss, nil
+	return &pdu.ListFilesystemRes{Filesystems: fss}, nil
 }
 
-func (t *mockTarget) ListFilesystemVersions(ctx context.Context, fs string) ([]*pdu.FilesystemVersion, error) {
+func (t *mockTarget) ListFilesystemVersions(ctx context.Context, req *pdu.ListFilesystemVersionsReq) (*pdu.ListFilesystemVersionsRes, error) {
+	fs := req.Filesystem
 	if len(t.listVersionsErrs[fs]) != 0 {
 		e := t.listVersionsErrs[fs][0]
 		t.listVersionsErrs[fs] = t.listVersionsErrs[fs][1:]
@@ -68,7 +69,7 @@ func (t *mockTarget) ListFilesystemVersions(ctx context.Context, fs string) ([]*
 		if mfs.path != fs {
 			continue
 		}
-		return mfs.FilesystemVersions(), nil
+		return &pdu.ListFilesystemVersionsRes{Versions: mfs.FilesystemVersions()}, nil
 	}
 	return nil, fmt.Errorf("filesystem %s does not exist", fs)
 }

daemon/snapper/snapper.go

@@ -177,7 +177,7 @@ func onMainCtxDone(ctx context.Context, u updater) state {
 }
 
 func syncUp(a args, u updater) state {
-	fss, err := listFSes(a.fsf)
+	fss, err := listFSes(a.ctx, a.fsf)
 	if err != nil {
 		return onErr(err, u)
 	}
@@ -204,7 +204,7 @@ func plan(a args, u updater) state {
 	u(func(snapper *Snapper) {
 		snapper.lastInvocation = time.Now()
 	})
-	fss, err := listFSes(a.fsf)
+	fss, err := listFSes(a.ctx, a.fsf)
 	if err != nil {
 		return onErr(err, u)
 	}
@@ -299,8 +299,8 @@ func wait(a args, u updater) state {
 	}
 }
 
-func listFSes(mf *filters.DatasetMapFilter) (fss []*zfs.DatasetPath, err error) {
-	return zfs.ZFSListMapping(mf)
+func listFSes(ctx context.Context, mf *filters.DatasetMapFilter) (fss []*zfs.DatasetPath, err error) {
+	return zfs.ZFSListMapping(ctx, mf)
 }
 
 func findSyncPoint(log Logger, fss []*zfs.DatasetPath, prefix string, interval time.Duration) (syncPoint time.Time, err error) {

daemon/streamrpcconfig/streamrpcconfig.go

@@ -1,25 +0,0 @@
-package streamrpcconfig
-
-import (
-	"github.com/problame/go-streamrpc"
-	"github.com/zrepl/zrepl/config"
-)
-
-func FromDaemonConfig(g *config.Global, in *config.RPCConfig) (*streamrpc.ConnConfig, error) {
-	conf := in
-	if conf == nil {
-		conf = g.RPC
-	}
-	srpcConf := &streamrpc.ConnConfig{
-		RxHeaderMaxLen:       conf.RxHeaderMaxLen,
-		RxStructuredMaxLen:   conf.RxStructuredMaxLen,
-		RxStreamMaxChunkSize: conf.RxStreamChunkMaxLen,
-		TxChunkSize:          conf.TxChunkSize,
-		Timeout: 			  conf.Timeout,
-		SendHeartbeatInterval: conf.SendHeartbeatInterval,
-	}
-	if err := srpcConf.Validate(); err != nil {
-		return nil, err
-	}
-	return srpcConf, nil
-}

daemon/transport/connecter/connect_local.go

@@ -1,30 +0,0 @@
-package connecter
-
-import (
-	"context"
-	"fmt"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/transport/serve"
-	"net"
-)
-
-type LocalConnecter struct {
-	listenerName string
-	clientIdentity string
-}
-
-func LocalConnecterFromConfig(in *config.LocalConnect) (*LocalConnecter, error) {
-	if in.ClientIdentity == "" {
-		return nil, fmt.Errorf("ClientIdentity must not be empty")
-	}
-	if in.ListenerName == "" {
-		return nil, fmt.Errorf("ListenerName must not be empty")
-	}
-	return &LocalConnecter{listenerName: in.ListenerName, clientIdentity: in.ClientIdentity}, nil
-}
-
-func (c *LocalConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
-	l := serve.GetLocalListener(c.listenerName)
-	return l.Connect(dialCtx, c.clientIdentity)
-}
-

daemon/transport/connecter/connect_ssh.go

@@ -1,66 +0,0 @@
-package connecter
-
-import (
-	"context"
-	"github.com/jinzhu/copier"
-	"github.com/pkg/errors"
-	"github.com/problame/go-netssh"
-	"github.com/problame/go-streamrpc"
-	"github.com/zrepl/zrepl/config"
-	"net"
-	"time"
-)
-
-type SSHStdinserverConnecter struct {
-	Host                 string
-	User                 string
-	Port                 uint16
-	IdentityFile         string
-	TransportOpenCommand []string
-	SSHCommand           string
-	Options              []string
-	dialTimeout          time.Duration
-}
-
-var _ streamrpc.Connecter = &SSHStdinserverConnecter{}
-
-func SSHStdinserverConnecterFromConfig(in *config.SSHStdinserverConnect) (c *SSHStdinserverConnecter, err error) {
-
-	c = &SSHStdinserverConnecter{
-		Host:         in.Host,
-		User:         in.User,
-		Port:         in.Port,
-		IdentityFile: in.IdentityFile,
-		SSHCommand:   in.SSHCommand,
-		Options:      in.Options,
-		dialTimeout:  in.DialTimeout,
-	}
-	return
-
-}
-
-type netsshConnToConn struct{ *netssh.SSHConn }
-
-var _ net.Conn = netsshConnToConn{}
-
-func (netsshConnToConn) SetDeadline(dl time.Time) error      { return nil }
-func (netsshConnToConn) SetReadDeadline(dl time.Time) error  { return nil }
-func (netsshConnToConn) SetWriteDeadline(dl time.Time) error { return nil }
-
-func (c *SSHStdinserverConnecter) Connect(dialCtx context.Context) (net.Conn, error) {
-
-	var endpoint netssh.Endpoint
-	if err := copier.Copy(&endpoint, c); err != nil {
-		return nil, errors.WithStack(err)
-	}
-	dialCtx, dialCancel := context.WithTimeout(dialCtx, c.dialTimeout) // context.TODO tied to error handling below
-	defer dialCancel()
-	nconn, err := netssh.Dial(dialCtx, endpoint)
-	if err != nil {
-		if err == context.DeadlineExceeded {
-			err = errors.Errorf("dial_timeout of %s exceeded", c.dialTimeout)
-		}
-		return nil, err
-	}
-	return netsshConnToConn{nconn}, nil
-}

daemon/transport/connecter/connect_tcp.go

@@ -1,24 +0,0 @@
-package connecter
-
-import (
-	"context"
-	"github.com/zrepl/zrepl/config"
-	"net"
-)
-
-type TCPConnecter struct {
-	Address string
-	dialer  net.Dialer
-}
-
-func TCPConnecterFromConfig(in *config.TCPConnect) (*TCPConnecter, error) {
-	dialer := net.Dialer{
-		Timeout: in.DialTimeout,
-	}
-
-	return &TCPConnecter{in.Address, dialer}, nil
-}
-
-func (c *TCPConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
-	return c.dialer.DialContext(dialCtx, "tcp", c.Address)
-}

daemon/transport/connecter/connect_tls.go

@@ -1,47 +0,0 @@
-package connecter
-
-import (
-	"context"
-	"crypto/tls"
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/tlsconf"
-	"net"
-)
-
-type TLSConnecter struct {
-	Address   string
-	dialer    net.Dialer
-	tlsConfig *tls.Config
-}
-
-func TLSConnecterFromConfig(in *config.TLSConnect) (*TLSConnecter, error) {
-	dialer := net.Dialer{
-		Timeout: in.DialTimeout,
-	}
-
-	ca, err := tlsconf.ParseCAFile(in.Ca)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse ca file")
-	}
-
-	cert, err := tls.LoadX509KeyPair(in.Cert, in.Key)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse cert/key pair")
-	}
-
-	tlsConfig, err := tlsconf.ClientAuthClient(in.ServerCN, ca, cert)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot build tls config")
-	}
-
-	return &TLSConnecter{in.Address, dialer, tlsConfig}, nil
-}
-
-func (c *TLSConnecter) Connect(dialCtx context.Context) (conn net.Conn, err error) {
-	conn, err = c.dialer.DialContext(dialCtx, "tcp", c.Address)
-	if err != nil {
-		return nil, err
-	}
-	return tls.Client(conn, c.tlsConfig), nil
-}

daemon/transport/connecter/connecter.go

@@ -1,84 +0,0 @@
-package connecter
-
-import (
-	"context"
-	"fmt"
-	"github.com/problame/go-streamrpc"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/streamrpcconfig"
-	"github.com/zrepl/zrepl/daemon/transport"
-	"net"
-	"time"
-)
-
-
-type HandshakeConnecter struct {
-	connecter streamrpc.Connecter
-}
-
-func (c HandshakeConnecter) Connect(ctx context.Context) (net.Conn, error) {
-	conn, err := c.connecter.Connect(ctx)
-	if err != nil {
-		return nil, err
-	}
-	dl, ok := ctx.Deadline()
-	if !ok {
-		dl = time.Now().Add(10 * time.Second) // FIXME constant
-	}
-	if err := transport.DoHandshakeCurrentVersion(conn, dl); err != nil {
-		conn.Close()
-		return nil, err
-	}
-	return conn, nil
-}
-
-
-
-func FromConfig(g *config.Global, in config.ConnectEnum) (*ClientFactory, error) {
-	var (
-		connecter            streamrpc.Connecter
-		errConnecter, errRPC error
-		connConf             *streamrpc.ConnConfig
-	)
-	switch v := in.Ret.(type) {
-	case *config.SSHStdinserverConnect:
-		connecter, errConnecter = SSHStdinserverConnecterFromConfig(v)
-		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	case *config.TCPConnect:
-		connecter, errConnecter = TCPConnecterFromConfig(v)
-		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	case *config.TLSConnect:
-		connecter, errConnecter = TLSConnecterFromConfig(v)
-		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	case *config.LocalConnect:
-		connecter, errConnecter = LocalConnecterFromConfig(v)
-		connConf, errRPC = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	default:
-		panic(fmt.Sprintf("implementation error: unknown connecter type %T", v))
-	}
-
-	if errConnecter != nil {
-		return nil, errConnecter
-	}
-	if errRPC != nil {
-		return nil, errRPC
-	}
-
-	config := streamrpc.ClientConfig{ConnConfig: connConf}
-	if err := config.Validate(); err != nil {
-		return nil, err
-	}
-
-	connecter = HandshakeConnecter{connecter}
-
-	return &ClientFactory{connecter: connecter, config: &config}, nil
-}
-
-type ClientFactory struct {
-	connecter streamrpc.Connecter
-	config    *streamrpc.ClientConfig
-}
-
-func (f ClientFactory) NewClient() (*streamrpc.Client, error) {
-	return streamrpc.NewClient(f.connecter, f.config)
-}

daemon/transport/handshake.go

@@ -1,136 +0,0 @@
-package transport
-
-import (
-	"bytes"
-	"fmt"
-	"io"
-	"net"
-	"strings"
-	"time"
-	"unicode/utf8"
-)
-
-type HandshakeMessage struct {
-	ProtocolVersion int
-	Extensions []string
-}
-
-func (m *HandshakeMessage) Encode() ([]byte, error) {
-	if m.ProtocolVersion <= 0 || m.ProtocolVersion > 9999 {
-		return nil, fmt.Errorf("protocol version must be in [1, 9999]")
-	}
-	if len(m.Extensions) >= 9999 {
-		return nil, fmt.Errorf("protocol only supports [0, 9999] extensions")
-	}
-	// EXTENSIONS is a count of subsequent \n separated lines that contain protocol extensions
-	var extensions strings.Builder
-	for i, ext := range m.Extensions {
-		if strings.ContainsAny(ext, "\n") {
-			return nil, fmt.Errorf("Extension #%d contains forbidden newline character", i)
-		}
-		if !utf8.ValidString(ext) {
-			return nil, fmt.Errorf("Extension #%d is not valid UTF-8", i)
-		}
-		extensions.WriteString(ext)
-		extensions.WriteString("\n")
-	}
-	withoutLen := fmt.Sprintf("ZREPL_ZFS_REPLICATION PROTOVERSION=%04d EXTENSIONS=%04d\n%s",
-		m.ProtocolVersion, len(m.Extensions), extensions.String())
-	withLen := fmt.Sprintf("%010d %s", len(withoutLen), withoutLen)
-	return []byte(withLen), nil
-}
-
-func (m *HandshakeMessage) DecodeReader(r io.Reader, maxLen int) error {
-	var lenAndSpace [11]byte
-	if _, err := io.ReadFull(r, lenAndSpace[:]); err != nil {
-		return err
-	}
-	if !utf8.Valid(lenAndSpace[:]) {
-		return fmt.Errorf("invalid start of handshake message: not valid UTF-8")
-	}
-	var followLen int
-	n, err := fmt.Sscanf(string(lenAndSpace[:]), "%010d ", &followLen)
-	if n != 1 || err != nil {
-		return fmt.Errorf("could not parse handshake message length")
-	}
-	if followLen > maxLen {
-		return fmt.Errorf("handshake message length exceeds max length (%d vs %d)",
-			followLen, maxLen)
-	}
-
-	var buf bytes.Buffer
-	_, err = io.Copy(&buf, io.LimitReader(r, int64(followLen)))
-	if err != nil {
-		return err
-	}
-
-	var (
-		protoVersion, extensionCount int
-	)
-	n, err = fmt.Fscanf(&buf, "ZREPL_ZFS_REPLICATION PROTOVERSION=%04d EXTENSIONS=%4d\n",
-		&protoVersion, &extensionCount)
-	if n != 2 || err != nil {
-		return fmt.Errorf("could not parse handshake message: %s", err)
-	}
-	if protoVersion < 1 {
-		return fmt.Errorf("invalid protocol version %q", protoVersion)
-	}
-	m.ProtocolVersion = protoVersion
-
-	if extensionCount < 0 {
-		return fmt.Errorf("invalid extension count %q", extensionCount)
-	}
-	if extensionCount == 0 {
-		if buf.Len() != 0 {
-			return fmt.Errorf("unexpected data trailing after header")
-		}
-		m.Extensions = nil
-		return nil
-	}
-	s := buf.String()
-	if strings.Count(s, "\n") != extensionCount {
-		return fmt.Errorf("inconsistent extension count: found %d, header says %d", len(m.Extensions), extensionCount)
-	}
-	exts := strings.Split(s, "\n")
-	if exts[len(exts)-1] != "" {
-		return fmt.Errorf("unexpected data trailing after last extension newline")
-	}
-	m.Extensions = exts[0:len(exts)-1]
-
-	return nil
-}
-
-func DoHandshakeCurrentVersion(conn net.Conn, deadline time.Time) error {
-	// current protocol version is hardcoded here
-	return DoHandshakeVersion(conn, deadline, 1)
-}
-
-func DoHandshakeVersion(conn net.Conn, deadline time.Time, version int) error {
-	ours := HandshakeMessage{
-		ProtocolVersion: version,
-		Extensions: nil,
-	}
-	hsb, err := ours.Encode()
-	if err != nil {
-		return fmt.Errorf("could not encode protocol banner: %s", err)
-	}
-
-	conn.SetDeadline(deadline)
-	_, err = io.Copy(conn, bytes.NewBuffer(hsb))
-	if err != nil {
-		return fmt.Errorf("could not send protocol banner: %s", err)
-	}
-
-	theirs := HandshakeMessage{}
-	if err := theirs.DecodeReader(conn, 16 * 4096); err != nil { // FIXME constant
-		return fmt.Errorf("could not decode protocol banner: %s", err)
-	}
-
-	if theirs.ProtocolVersion != ours.ProtocolVersion {
-		return fmt.Errorf("protocol versions do not match: ours is %d, theirs is %d",
-			ours.ProtocolVersion, theirs.ProtocolVersion)
-	}
-	// ignore extensions, we don't use them
-
-	return nil
-}

daemon/transport/handshake_test.go

@@ -1,119 +0,0 @@
-package transport
-
-import (
-	"bytes"
-	"fmt"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/zrepl/zrepl/util/socketpair"
-	"io"
-	"strings"
-	"testing"
-	"time"
-)
-
-func TestHandshakeMessage_Encode(t *testing.T) {
-
-	msg := HandshakeMessage{
-		ProtocolVersion: 2342,
-	}
-
-	encB, err := msg.Encode()
-	require.NoError(t, err)
-	enc := string(encB)
-	t.Logf("enc: %s", enc)
-
-
-
-	assert.False(t, strings.ContainsAny(enc[0:10], " "))
-	assert.True(t, enc[10] == ' ')
-
-	var (
-		headerlen, protoversion, extensionCount int
-	)
-	n, err := fmt.Sscanf(enc, "%010d ZREPL_ZFS_REPLICATION PROTOVERSION=%04d EXTENSIONS=%04d\n",
-		&headerlen, &protoversion, &extensionCount)
-	if n != 3 || (err != nil && err != io.EOF) {
-		t.Fatalf("%v %v", n, err)
-	}
-
-	assert.Equal(t, 2342, protoversion)
-	assert.Equal(t, 0, extensionCount)
-	assert.Equal(t, len(enc)-11, headerlen)
-
-}
-
-func TestHandshakeMessage_Encode_InvalidProtocolVersion(t *testing.T) {
-
-	for _, pv := range []int{-1, 0,  10000, 10001} {
-		t.Logf("testing invalid protocol version = %v", pv)
-		msg := HandshakeMessage{
-			ProtocolVersion: pv,
-		}
-		b, err := msg.Encode()
-		assert.Error(t, err)
-		assert.Nil(t, b)
-	}
-
-}
-
-func TestHandshakeMessage_DecodeReader(t *testing.T) {
-
-	in := HandshakeMessage{
-		2342,
-		[]string{"foo", "bar 2342"},
-	}
-
-	enc, err := in.Encode()
-	require.NoError(t, err)
-
-	out := HandshakeMessage{}
-	err = out.DecodeReader(bytes.NewReader([]byte(enc)), 4 * 4096)
-	assert.NoError(t, err)
-	assert.Equal(t, 2342, out.ProtocolVersion)
-	assert.Equal(t, 2, len(out.Extensions))
-	assert.Equal(t, "foo", out.Extensions[0])
-	assert.Equal(t, "bar 2342", out.Extensions[1])
-
-}
-
-func TestDoHandshakeVersion_ErrorOnDifferentVersions(t *testing.T) {
-	srv, client, err := socketpair.SocketPair()
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer srv.Close()
-	defer client.Close()
-
-	srvErrCh := make(chan error)
-	go func() {
-		srvErrCh <- DoHandshakeVersion(srv, time.Now().Add(2*time.Second), 1)
-	}()
-	err = DoHandshakeVersion(client, time.Now().Add(2*time.Second), 2)
-	t.Log(err)
-	assert.Error(t, err)
-	assert.True(t, strings.Contains(err.Error(), "version"))
-
-	srvErr := <-srvErrCh
-	t.Log(srvErr)
-	assert.Error(t, srvErr)
-	assert.True(t, strings.Contains(srvErr.Error(), "version"))
-}
-
-func TestDoHandshakeCurrentVersion(t *testing.T) {
-	srv, client, err := socketpair.SocketPair()
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer srv.Close()
-	defer client.Close()
-
-	srvErrCh := make(chan error)
-	go func() {
-		srvErrCh <- DoHandshakeVersion(srv, time.Now().Add(2*time.Second), 1)
-	}()
-	err = DoHandshakeVersion(client, time.Now().Add(2*time.Second), 1)
-	assert.Nil(t, err)
-	assert.Nil(t, <-srvErrCh)
-
-}

daemon/transport/serve/serve.go

@@ -1,147 +0,0 @@
-package serve
-
-import (
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/transport"
-	"net"
-	"github.com/zrepl/zrepl/daemon/streamrpcconfig"
-	"github.com/problame/go-streamrpc"
-	"context"
-	"github.com/zrepl/zrepl/logger"
-	"github.com/zrepl/zrepl/zfs"
-	"time"
-)
-
-type contextKey int
-
-const contextKeyLog contextKey = 0
-
-type Logger = logger.Logger
-
-func WithLogger(ctx context.Context, log Logger) context.Context {
-	return context.WithValue(ctx, contextKeyLog, log)
-}
-
-func getLogger(ctx context.Context) Logger {
-	if log, ok := ctx.Value(contextKeyLog).(Logger); ok {
-		return log
-	}
-	return logger.NewNullLogger()
-}
-
-type AuthenticatedConn interface {
-	net.Conn
-	// ClientIdentity must be a string that satisfies ValidateClientIdentity
-	ClientIdentity() string
-}
-
-// A client identity must be a single component in a ZFS filesystem path
-func ValidateClientIdentity(in string) (err error) {
-	path, err := zfs.NewDatasetPath(in)
-	if err != nil {
-		return err
-	}
-	if path.Length() != 1 {
-		return errors.New("client identity must be a single path comonent (not empty, no '/')")
-	}
-	return nil
-}
-
-type authConn struct {
-	net.Conn
-	clientIdentity string
-}
-
-var _ AuthenticatedConn = authConn{}
-
-func (c authConn) ClientIdentity() string {
-	if err := ValidateClientIdentity(c.clientIdentity); err != nil {
-		panic(err)
-	}
-	return c.clientIdentity
-}
-
-// like net.Listener, but with an AuthenticatedConn instead of net.Conn
-type AuthenticatedListener interface {
-	Addr() (net.Addr)
-	Accept(ctx context.Context) (AuthenticatedConn, error)
-	Close() error
-}
-
-type ListenerFactory interface {
-	Listen() (AuthenticatedListener, error)
-}
-
-type HandshakeListenerFactory struct {
-	lf ListenerFactory
-}
-
-func (lf HandshakeListenerFactory) Listen() (AuthenticatedListener, error) {
-	l, err := lf.lf.Listen()
-	if err != nil {
-		return nil, err
-	}
-	return HandshakeListener{l}, nil
-}
-
-type HandshakeListener struct {
-	l AuthenticatedListener
-}
-
-func (l HandshakeListener) Addr() (net.Addr) { return l.l.Addr() }
-
-func (l HandshakeListener) Close() error { return l.l.Close() }
-
-func (l HandshakeListener) Accept(ctx context.Context) (AuthenticatedConn, error) {
-	conn, err := l.l.Accept(ctx)
-	if err != nil {
-		return nil, err
-	}
-	dl, ok := ctx.Deadline()
-	if !ok {
-		dl = time.Now().Add(10*time.Second) // FIXME constant
-	}
-	if err := transport.DoHandshakeCurrentVersion(conn, dl); err != nil {
-		conn.Close()
-		return nil, err
-	}
-	return conn, nil
-}
-
-func FromConfig(g *config.Global, in config.ServeEnum) (lf ListenerFactory, conf *streamrpc.ConnConfig, _ error) {
-
-	var (
-		lfError, rpcErr error
-	)
-	switch v := in.Ret.(type) {
-	case *config.TCPServe:
-		lf, lfError = TCPListenerFactoryFromConfig(g, v)
-		conf, rpcErr = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	case *config.TLSServe:
-		lf, lfError = TLSListenerFactoryFromConfig(g, v)
-		conf, rpcErr = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	case *config.StdinserverServer:
-		lf, lfError = MultiStdinserverListenerFactoryFromConfig(g, v)
-		conf, rpcErr = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	case *config.LocalServe:
-		lf, lfError = LocalListenerFactoryFromConfig(g, v)
-		conf, rpcErr = streamrpcconfig.FromDaemonConfig(g, v.RPC)
-	default:
-		return nil, nil, errors.Errorf("internal error: unknown serve type %T", v)
-	}
-
-	if lfError != nil {
-		return nil, nil, lfError
-	}
-	if rpcErr != nil {
-		return nil, nil, rpcErr
-	}
-
-	lf = HandshakeListenerFactory{lf}
-
-	return lf, conf, nil
-
-}
-
-

daemon/transport/serve/serve_local.go

@@ -1,187 +0,0 @@
-package serve
-
-import (
-	"context"
-	"fmt"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/util/socketpair"
-	"net"
-	"sync"
-)
-
-var localListeners struct {
-	m map[string]*LocalListener // listenerName -> listener
-	init sync.Once
-	mtx sync.Mutex
-}
-
-func GetLocalListener(listenerName string) (*LocalListener) {
-
-	localListeners.init.Do(func() {
-		localListeners.m = make(map[string]*LocalListener)
-	})
-
-	localListeners.mtx.Lock()
-	defer localListeners.mtx.Unlock()
-
-	l, ok := localListeners.m[listenerName]
-	if !ok {
-		l = newLocalListener()
-		localListeners.m[listenerName] = l
-	}
-	return l
-
-}
-
-type connectRequest struct {
-	clientIdentity string
-	callback chan connectResult
-}
-
-type connectResult struct {
-	conn net.Conn
-	err error
-}
-
-type LocalListener struct {
-	connects chan connectRequest
-}
-
-func newLocalListener() *LocalListener {
-	return &LocalListener{
-		connects: make(chan connectRequest),
-	}
-}
-
-// Connect to the LocalListener from a client with identity clientIdentity
-func (l *LocalListener) Connect(dialCtx context.Context, clientIdentity string) (conn net.Conn, err error) {
-
-	// place request
-	req := connectRequest{
-		clientIdentity: clientIdentity,
-		callback: make(chan connectResult),
-	}
-	select {
-	case l.connects <- req:
-	case <-dialCtx.Done():
-		return nil, dialCtx.Err()
-	}
-
-	// wait for listener response
-	select {
-	case connRes := <- req.callback:
-		conn, err = connRes.conn, connRes.err
-	case <-dialCtx.Done():
-		close(req.callback) // sending to the channel afterwards will panic, the listener has to catch this
-		conn, err = nil, dialCtx.Err()
-	}
-
-	return conn, err
-}
-
-type localAddr struct {
-	S string
-}
-
-func (localAddr) Network() string { return "local" }
-
-func (a localAddr) String() string { return a.S }
-
-func (l *LocalListener) Addr() (net.Addr) { return localAddr{"<listening>"} }
-
-type localConn struct {
-	net.Conn
-	clientIdentity string
-}
-
-func (l localConn) ClientIdentity() string { return l.clientIdentity }
-
-func (l *LocalListener) Accept(ctx context.Context) (AuthenticatedConn, error) {
-	respondToRequest := func(req connectRequest, res connectResult) (err error) {
-		getLogger(ctx).
-			WithField("res.conn", res.conn).WithField("res.err", res.err).
-			Debug("responding to client request")
-		defer func() {
-			errv := recover()
-			getLogger(ctx).WithField("recover_err", errv).
-				Debug("panic on send to client callback, likely a legitimate client-side timeout")
-		}()
-		select {
-		case req.callback <- res:
-			err = nil
-		default:
-			err = fmt.Errorf("client-provided callback did block on send")
-		}
-		close(req.callback)
-		return err
-	}
-
-	getLogger(ctx).Debug("waiting for local client connect requests")
-	var req connectRequest
-	select {
-	case req = <-l.connects:
-	case <-ctx.Done():
-		return nil, ctx.Err()
-	}
-
-	getLogger(ctx).WithField("client_identity", req.clientIdentity).Debug("got connect request")
-	if req.clientIdentity == "" {
-		res := connectResult{nil, fmt.Errorf("client identity must not be empty")}
-		if err := respondToRequest(req, res); err != nil {
-			return nil, err
-		}
-		return nil, fmt.Errorf("client connected with empty client identity")
-	}
-
-	getLogger(ctx).Debug("creating socketpair")
-	left, right, err := socketpair.SocketPair()
-	if err != nil {
-		res := connectResult{nil, fmt.Errorf("server error: %s", err)}
-		if respErr := respondToRequest(req, res); respErr != nil {
-			// returning the socketpair error properly is more important than the error sent to the client
-			getLogger(ctx).WithError(respErr).Error("error responding to client")
-		}
-		return nil, err
-	}
-
-	getLogger(ctx).Debug("responding with left side of socketpair")
-	res := connectResult{left, nil}
-	if err := respondToRequest(req, res); err != nil {
-		getLogger(ctx).WithError(err).Error("error responding to client")
-		if err := left.Close(); err != nil {
-			getLogger(ctx).WithError(err).Error("cannot close left side of socketpair")
-		}
-		if err := right.Close(); err != nil {
-			getLogger(ctx).WithError(err).Error("cannot close right side of socketpair")
-		}
-		return nil, err
-	}
-
-	return localConn{right, req.clientIdentity}, nil
-}
-
-func (l *LocalListener) Close() error {
-	// FIXME: make sure concurrent Accepts return with error, and further Accepts return that error, too
-	// Example impl: for each accept, do context.WithCancel, and store the cancel in a list
-	// When closing, set a member variable to state=closed, make sure accept will exit early
-	// and then call all cancels in the list
-	// The code path from Accept entry over check if state=closed to list entry must be protected by a mutex.
-	return nil
-}
-
-type LocalListenerFactory struct {
-	listenerName string
-}
-
-func LocalListenerFactoryFromConfig(g *config.Global, in *config.LocalServe) (f *LocalListenerFactory, err error) {
-	if in.ListenerName == "" {
-		return nil, fmt.Errorf("ListenerName must not be empty")
-	}
-	return &LocalListenerFactory{listenerName: in.ListenerName}, nil
-}
-
-
-func (lf *LocalListenerFactory) Listen() (AuthenticatedListener, error) {
-	return GetLocalListener(lf.listenerName), nil
-}
-

daemon/transport/serve/serve_stdinserver.go

@@ -1,158 +0,0 @@
-package serve
-
-import (
-	"github.com/problame/go-netssh"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/daemon/nethelpers"
-	"io"
-	"net"
-	"path"
-	"time"
-	"context"
-	"github.com/pkg/errors"
-	"sync/atomic"
-)
-
-type StdinserverListenerFactory struct {
-	ClientIdentities []string
-	Sockdir string
-}
-
-func MultiStdinserverListenerFactoryFromConfig(g *config.Global, in *config.StdinserverServer) (f *multiStdinserverListenerFactory, err error) {
-
-	for _, ci := range in.ClientIdentities {
-		if err := ValidateClientIdentity(ci); err != nil {
-			return nil, errors.Wrapf(err, "invalid client identity %q", ci)
-		}
-	}
-
-	f = &multiStdinserverListenerFactory{
-		ClientIdentities: in.ClientIdentities,
-		Sockdir: g.Serve.StdinServer.SockDir,
-	}
-
-	return
-}
-
-type multiStdinserverListenerFactory struct {
-	ClientIdentities []string
-	Sockdir string
-}
-
-func (f *multiStdinserverListenerFactory) Listen() (AuthenticatedListener, error) {
-	return multiStdinserverListenerFromClientIdentities(f.Sockdir, f.ClientIdentities)
-}
-
-type multiStdinserverAcceptRes struct {
-	conn AuthenticatedConn
-	err error
-}
-
-type MultiStdinserverListener struct {
-	listeners []*stdinserverListener
-	accepts chan multiStdinserverAcceptRes
-	closed int32
-}
-
-// client identities must be validated
-func multiStdinserverListenerFromClientIdentities(sockdir string, cis []string) (*MultiStdinserverListener, error) {
-	listeners := make([]*stdinserverListener, 0, len(cis))
-	var err error
-	for _, ci := range cis {
-		sockpath := path.Join(sockdir, ci)
-		l  := &stdinserverListener{clientIdentity: ci}
-		if err = nethelpers.PreparePrivateSockpath(sockpath); err != nil {
-			break
-		}
-		if l.l, err = netssh.Listen(sockpath); err != nil {
-			break
-		}
-		listeners = append(listeners, l)
-	}
-	if err != nil {
-		for _, l := range listeners {
-			l.Close() // FIXME error reporting?
-		}
-		return nil, err
-	}
-	return &MultiStdinserverListener{listeners: listeners}, nil
-}
-
-func (m *MultiStdinserverListener) Accept(ctx context.Context) (AuthenticatedConn, error){
-
-	if m.accepts == nil {
-		m.accepts = make(chan multiStdinserverAcceptRes, len(m.listeners))
-		for i := range m.listeners {
-			go func(i int) {
-				for atomic.LoadInt32(&m.closed) == 0 {
-					conn, err := m.listeners[i].Accept(context.TODO())
-					m.accepts <- multiStdinserverAcceptRes{conn, err}
-				}
-			}(i)
-		}
-	}
-
-	res := <- m.accepts
-	return res.conn, res.err
-
-}
-
-func (m *MultiStdinserverListener) Addr() (net.Addr) {
-	return netsshAddr{}
-}
-
-func (m *MultiStdinserverListener) Close() error {
-	atomic.StoreInt32(&m.closed, 1)
-	var oneErr error
-	for _, l := range m.listeners {
-		if err := l.Close(); err != nil && oneErr == nil {
-			oneErr = err
-		}
-	}
-	return oneErr
-}
-
-// a single stdinserverListener (part of multiStinserverListener)
-type stdinserverListener struct {
-	l *netssh.Listener
-	clientIdentity string
-}
-
-func (l stdinserverListener) Addr() net.Addr {
-	return netsshAddr{}
-}
-
-func (l stdinserverListener) Accept(ctx context.Context) (AuthenticatedConn, error) {
-	c, err := l.l.Accept()
-	if err != nil {
-		return nil, err
-	}
-	return netsshConnToNetConnAdatper{c, l.clientIdentity}, nil
-}
-
-func (l stdinserverListener) Close() (err error) {
-	return l.l.Close()
-}
-
-type netsshAddr struct{}
-
-func (netsshAddr) Network() string { return "netssh" }
-func (netsshAddr) String() string  { return "???" }
-
-type netsshConnToNetConnAdatper struct {
-	io.ReadWriteCloser // works for both netssh.SSHConn and netssh.ServeConn
-	clientIdentity string
-}
-
-func (a netsshConnToNetConnAdatper) ClientIdentity() string { return a.clientIdentity }
-
-func (netsshConnToNetConnAdatper) LocalAddr() net.Addr { return netsshAddr{} }
-
-func (netsshConnToNetConnAdatper) RemoteAddr() net.Addr { return netsshAddr{} }
-
-// FIXME log warning once!
-func (netsshConnToNetConnAdatper) SetDeadline(t time.Time) error { return nil }
-
-func (netsshConnToNetConnAdatper) SetReadDeadline(t time.Time) error { return nil }
-
-func (netsshConnToNetConnAdatper) SetWriteDeadline(t time.Time) error { return nil }

daemon/transport/serve/serve_tcp.go

@@ -1,91 +0,0 @@
-package serve
-
-import (
-	"github.com/zrepl/zrepl/config"
-	"net"
-	"github.com/pkg/errors"
-	"context"
-)
-
-type TCPListenerFactory struct {
-	address *net.TCPAddr
-	clientMap *ipMap
-}
-
-type ipMapEntry struct {
-	ip net.IP
-	ident string
-}
-
-type ipMap struct {
-	entries []ipMapEntry
-}
-
-func ipMapFromConfig(clients map[string]string) (*ipMap, error) {
-	entries := make([]ipMapEntry, 0, len(clients))
-	for clientIPString, clientIdent := range clients {
-		clientIP := net.ParseIP(clientIPString)
-		if clientIP == nil {
-			return nil, errors.Errorf("cannot parse client IP %q", clientIPString)
-		}
-		if err := ValidateClientIdentity(clientIdent); err != nil {
-			return nil, errors.Wrapf(err,"invalid client identity for IP %q", clientIPString)
-		}
-		entries = append(entries, ipMapEntry{clientIP, clientIdent})
-	}
-	return &ipMap{entries: entries}, nil
-}
-
-func (m *ipMap) Get(ip net.IP) (string, error) {
-	for _, e := range m.entries {
-		if e.ip.Equal(ip) {
-			return e.ident, nil
-		}
-	}
-	return "", errors.Errorf("no identity mapping for client IP %s", ip)
-}
-
-func TCPListenerFactoryFromConfig(c *config.Global, in *config.TCPServe) (*TCPListenerFactory, error) {
-	addr, err := net.ResolveTCPAddr("tcp", in.Listen)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse listen address")
-	}
-	clientMap, err := ipMapFromConfig(in.Clients)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse client IP map")
-	}
-	lf := &TCPListenerFactory{
-		address: addr,
-		clientMap: clientMap,
-	}
-	return lf, nil
-}
-
-func (f *TCPListenerFactory) Listen() (AuthenticatedListener, error) {
-	l, err := net.ListenTCP("tcp", f.address)
-	if err != nil {
-		return nil, err
-	}
-	return &TCPAuthListener{l, f.clientMap}, nil
-}
-
-type TCPAuthListener struct {
-	*net.TCPListener
-	clientMap *ipMap
-}
-
-func (f *TCPAuthListener) Accept(ctx context.Context) (AuthenticatedConn, error) {
-	nc, err := f.TCPListener.Accept()
-	if err != nil {
-		return nil, err
-	}
-	clientIP := nc.RemoteAddr().(*net.TCPAddr).IP
-	clientIdent, err := f.clientMap.Get(clientIP)
-	if err != nil {
-		getLogger(ctx).WithField("ip", clientIP).Error("client IP not in client map")
-		nc.Close()
-		return nil, err
-	}
-	return authConn{nc, clientIdent}, nil
-}
-

daemon/transport/serve/serve_tls.go

@@ -1,83 +0,0 @@
-package serve
-
-import (
-	"crypto/tls"
-	"crypto/x509"
-	"fmt"
-	"github.com/pkg/errors"
-	"github.com/zrepl/zrepl/config"
-	"github.com/zrepl/zrepl/tlsconf"
-	"net"
-	"time"
-	"context"
-)
-
-type TLSListenerFactory struct {
-	address          string
-	clientCA         *x509.CertPool
-	serverCert       tls.Certificate
-	handshakeTimeout time.Duration
-	clientCNs map[string]struct{}
-}
-
-func TLSListenerFactoryFromConfig(c *config.Global, in *config.TLSServe) (lf *TLSListenerFactory, err error) {
-	lf = &TLSListenerFactory{
-		address: in.Listen,
-		handshakeTimeout: in.HandshakeTimeout,
-	}
-
-	if in.Ca == "" || in.Cert == "" || in.Key == "" {
-		return nil, errors.New("fields 'ca', 'cert' and 'key'must be specified")
-	}
-
-	lf.clientCA, err = tlsconf.ParseCAFile(in.Ca)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse ca file")
-	}
-
-	lf.serverCert, err = tls.LoadX509KeyPair(in.Cert, in.Key)
-	if err != nil {
-		return nil, errors.Wrap(err, "cannot parse cer/key pair")
-	}
-
-	lf.clientCNs = make(map[string]struct{}, len(in.ClientCNs))
-	for i, cn := range in.ClientCNs {
-		if err := ValidateClientIdentity(cn); err != nil {
-			return nil, errors.Wrapf(err, "unsuitable client_cn #%d %q", i, cn)
-		}
-		// dupes are ok fr now
-		lf.clientCNs[cn] = struct{}{}
-	}
-
-	return lf, nil
-}
-
-func (f *TLSListenerFactory) Listen() (AuthenticatedListener, error) {
-	l, err := net.Listen("tcp", f.address)
-	if err != nil {
-		return nil, err
-	}
-	tl := tlsconf.NewClientAuthListener(l, f.clientCA, f.serverCert, f.handshakeTimeout)
-	return tlsAuthListener{tl, f.clientCNs}, nil
-}
-
-type tlsAuthListener struct {
-	*tlsconf.ClientAuthListener
-	clientCNs map[string]struct{}
-}
-
-func (l tlsAuthListener) Accept(ctx context.Context) (AuthenticatedConn, error) {
-	c, cn, err := l.ClientAuthListener.Accept()
-	if err != nil {
-		return nil, err
-	}
-	if _, ok := l.clientCNs[cn]; !ok {
-		if err := c.Close(); err != nil {
-			getLogger(ctx).WithError(err).Error("error closing connection with unauthorized common name")
-		}
-		return nil, fmt.Errorf("unauthorized client common name %q from %s", cn, c.RemoteAddr())
-	}
-	return authConn{c, cn}, nil
-}
-
-