WIP rewrite the daemon

cmd subdir does not build on purpose, it's only left in tree to grab old code and move it to github.com/zrepl/zrepl/daemon
2018-08-27 22:21:45 +02:00
parent df6e1bc64d
commit c69ebd3806
55 changed files with 1133 additions and 84 deletions
@@ -1,8 +1,8 @@
 package job

 import (
-	"github.com/zrepl/zrepl/cmd/config"
 	"fmt"
+	"github.com/zrepl/zrepl/config"
 )

 func JobsFromConfig(c config.Config) ([]Job, error) {
@@ -20,8 +20,12 @@ func JobsFromConfig(c config.Config) ([]Job, error) {
 func buildJob(c config.Global, in config.JobEnum) (j Job, err error) {

 	switch v := in.Ret.(type) {
+	case *config.SinkJob:
+		return SinkFromConfig(c, v)
+	case *config.PushJob:
+		return PushFromConfig(c, v)
 	default:
-		panic(fmt.Sprintf("implementation error: unknown job type %s", v))
+		panic(fmt.Sprintf("implementation error: unknown job type %T", v))
 	}

-}
+}
@@ -2,7 +2,10 @@ package job

 import (
 	"context"
+	"errors"
+	"github.com/problame/go-streamrpc"
 	"github.com/zrepl/zrepl/logger"
+	"time"
 )

 type Logger = logger.Logger
@@ -25,9 +28,21 @@ func WithLogger(ctx context.Context, l Logger) context.Context {
 	return context.WithValue(ctx, contextKeyLog, l)
 }

-func WithWakeup(ctx context.Context) (context.Context, WakeupChan) {
-	wc := make(chan struct{}, 1)
-	return context.WithValue(ctx, contextKeyWakeup, wc), wc
+type WakeupFunc func() error
+
+var AlreadyWokenUp = errors.New("already woken up")
+
+func WithWakeup(ctx context.Context) (context.Context, WakeupFunc) {
+	wc := make(chan struct{})
+	wuf := func() error {
+		select {
+		case wc <- struct{}{}:
+			return nil
+		default:
+			return AlreadyWokenUp
+		}
+	}
+	return context.WithValue(ctx, contextKeyWakeup, wc), wuf
 }

 type Job interface {
@@ -36,12 +51,23 @@ type Job interface {
 	Status() interface{}
 }

-type WakeupChan <-chan struct{}
-
-func WaitWakeup(ctx context.Context) WakeupChan {
-	wc, ok := ctx.Value(contextKeyWakeup).(WakeupChan)
+func WaitWakeup(ctx context.Context) <-chan struct{} {
+	wc, ok := ctx.Value(contextKeyWakeup).(chan struct{})
 	if !ok {
 		wc = make(chan struct{})
 	}
 	return wc
 }
+
+var STREAMRPC_CONFIG = &streamrpc.ConnConfig{ // FIXME oversight and configurability
+	RxHeaderMaxLen:       4096,
+	RxStructuredMaxLen:   4096 * 4096,
+	RxStreamMaxChunkSize: 4096 * 4096,
+	TxChunkSize:          4096 * 4096,
+	RxTimeout: streamrpc.Timeout{
+		Progress: 10 * time.Second,
+	},
+	TxTimeout: streamrpc.Timeout{
+		Progress: 10 * time.Second,
+	},
+}
@@ -0,0 +1,86 @@
+package job
+
+import (
+	"context"
+	"github.com/pkg/errors"
+	"github.com/problame/go-streamrpc"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/connecter"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/logging"
+	"github.com/zrepl/zrepl/endpoint"
+	"github.com/zrepl/zrepl/replication"
+	"sync"
+)
+
+type Push struct {
+	name      string
+	connecter streamrpc.Connecter
+	fsfilter  endpoint.FSFilter
+
+	mtx         sync.Mutex
+	replication *replication.Replication
+}
+
+func PushFromConfig(g config.Global, in *config.PushJob) (j *Push, err error) {
+
+	j = &Push{}
+	j.name = in.Name
+
+	j.connecter, err = connecter.FromConfig(g, in.Replication.Connect)
+
+	if j.fsfilter, err = filters.DatasetMapFilterFromConfig(in.Replication.Filesystems); err != nil {
+		return nil, errors.Wrap(err, "cannnot build filesystem filter")
+	}
+
+	return j, nil
+}
+
+func (j *Push) Name() string { return j.name }
+
+func (j *Push) Status() interface{} {
+	return nil // FIXME
+}
+
+func (j *Push) Run(ctx context.Context) {
+	log := GetLogger(ctx)
+
+	defer log.Info("job exiting")
+
+	log.Debug("wait for wakeups")
+
+	invocationCount := 0
+outer:
+	for {
+		select {
+		case <-ctx.Done():
+			log.WithError(ctx.Err()).Info("context")
+			break outer
+		case <-WaitWakeup(ctx):
+			invocationCount++
+			invLog := log.WithField("invocation", invocationCount)
+			j.do(WithLogger(ctx, invLog))
+		}
+	}
+}
+
+func (j *Push) do(ctx context.Context) {
+
+	log := GetLogger(ctx)
+
+	client, err := streamrpc.NewClient(j.connecter, &streamrpc.ClientConfig{STREAMRPC_CONFIG})
+	if err != nil {
+		log.WithError(err).Error("cannot create streamrpc client")
+	}
+	defer client.Close()
+
+	sender := endpoint.NewSender(j.fsfilter, filters.NewAnyFSVFilter())
+	receiver := endpoint.NewRemote(client)
+
+	j.mtx.Lock()
+	rep := replication.NewReplication()
+	j.mtx.Unlock()
+
+	ctx = logging.WithSubsystemLoggers(ctx, log)
+	rep.Drive(ctx, sender, receiver)
+}
@@ -0,0 +1,115 @@
+package job
+
+import (
+	"context"
+	"github.com/pkg/errors"
+	"github.com/problame/go-streamrpc"
+	"github.com/zrepl/zrepl/config"
+	"github.com/zrepl/zrepl/daemon/filters"
+	"github.com/zrepl/zrepl/daemon/logging"
+	"github.com/zrepl/zrepl/daemon/serve"
+	"github.com/zrepl/zrepl/endpoint"
+	"net"
+)
+
+type Sink struct {
+	name     string
+	l        serve.ListenerFactory
+	fsmap    endpoint.FSMap
+	fsmapInv endpoint.FSFilter
+}
+
+func SinkFromConfig(g config.Global, in *config.SinkJob) (s *Sink, err error) {
+
+	// FIXME multi client support
+
+	s = &Sink{name: in.Name}
+	if s.l, err = serve.FromConfig(g, in.Replication.Serve); err != nil {
+		return nil, errors.Wrap(err, "cannot build server")
+	}
+
+	fsmap := filters.NewDatasetMapFilter(1, false) // FIXME multi-client support
+	if err := fsmap.Add("<", in.Replication.RootDataset); err != nil {
+		return nil, errors.Wrap(err, "unexpected error: cannot build filesystem mapping")
+	}
+	s.fsmap = fsmap
+
+	return s, nil
+}
+
+func (j *Sink) Name() string { return j.name }
+
+func (*Sink) Status() interface{} {
+	// FIXME
+	return nil
+}
+
+func (j *Sink) 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
+	}
+
+	log.WithField("addr", l.Addr()).Debug("accepting connections")
+
+	var connId int
+
+outer:
+	for {
+
+		select {
+		case res := <-accept(l):
+			if res.err != nil {
+				log.WithError(err).Info("accept error")
+				break outer
+			}
+			connId++
+			connLog := log.
+				WithField("connID", connId)
+			j.handleConnection(WithLogger(ctx, connLog), res.conn)
+
+		case <-ctx.Done():
+			break outer
+		}
+
+	}
+
+}
+
+func (j *Sink) handleConnection(ctx context.Context, conn net.Conn) {
+	log := GetLogger(ctx)
+	log.WithField("addr", conn.RemoteAddr()).Info("handling connection")
+	defer log.Info("finished handling connection")
+
+	logging.WithSubsystemLoggers(ctx, log)
+
+	local, err := endpoint.NewReceiver(j.fsmap, filters.NewAnyFSVFilter())
+	if err != nil {
+		log.WithError(err).Error("unexpected error: cannot convert mapping to filter")
+		return
+	}
+
+	handler := endpoint.NewHandler(local)
+	if err := streamrpc.ServeConn(ctx, conn, STREAMRPC_CONFIG, handler.Handle); err != nil {
+		log.WithError(err).Error("error serving client")
+	}
+}
+
+type acceptResult struct {
+	conn net.Conn
+	err  error
+}
+
+func accept(listener net.Listener) <-chan acceptResult {
+	c := make(chan acceptResult, 1)
+	go func() {
+		conn, err := listener.Accept()
+		c <- acceptResult{conn, err}
+	}()
+	return c
+}