Change transport format in zrepl config & parse it.

cmd/config.go

@@ -3,17 +3,37 @@ package main
 import (
 	"errors"
 	"fmt"
+	"github.com/jinzhu/copier"
 	"github.com/mitchellh/mapstructure"
+	"github.com/zrepl/zrepl/rpc"
+	"github.com/zrepl/zrepl/sshbytestream"
 	"github.com/zrepl/zrepl/zfs"
 	yaml "gopkg.in/yaml.v2"
+	"io"
 	"io/ioutil"
 	"strings"
 )
 
 type Pool struct {
-	Name string
+	Name      string
-	Url  string
+	Transport Transport
 }
+
+type Transport interface {
+	Connect() (rpc.RPCRequester, error)
+}
+type LocalTransport struct {
+	Pool string
+}
+type SSHTransport struct {
+	ZreplIdentity        string
+	Host                 string
+	User                 string
+	Port                 uint16
+	TransportOpenCommand []string
+	Options              []string
+}
+
 type Push struct {
 	To       *Pool
 	Datasets []zfs.DatasetPath
@@ -77,12 +97,63 @@ func parseMain(root map[string]interface{}) (c Config, err error) {
 	return
 }
 
-func parsePools(v interface{}) (p []Pool, err error) {
+func parsePools(v interface{}) (pools []Pool, err error) {
-	p = make([]Pool, 0)
+
-	err = mapstructure.Decode(v, &p)
+	asList := make([]struct {
+		Name      string
+		Transport map[string]interface{}
+	}, 0)
+	if err = mapstructure.Decode(v, &asList); err != nil {
+		return
+	}
+
+	pools = make([]Pool, len(asList))
+	for i, p := range asList {
+		var transport Transport
+		if transport, err = parseTransport(p.Transport); err != nil {
+			return
+		}
+		pools[i] = Pool{
+			Name:      p.Name,
+			Transport: transport,
+		}
+	}
+
 	return
 }
 
+func parseTransport(it map[string]interface{}) (t Transport, err error) {
+
+	if len(it) != 1 {
+		err = errors.New("ambiguous transport type")
+		return
+	}
+
+	for key, val := range it {
+		switch key {
+		case "ssh":
+			t := SSHTransport{}
+			if err = mapstructure.Decode(val, &t); err != nil {
+				err = errors.New(fmt.Sprintf("could not parse ssh transport: %s", err))
+				return nil, err
+			}
+			return t, nil
+		case "local":
+			t := LocalTransport{}
+			if err = mapstructure.Decode(val, &t); err != nil {
+				err = errors.New(fmt.Sprintf("could not parse local transport: %s", err))
+				return nil, err
+			}
+			return t, nil
+		default:
+			return nil, errors.New(fmt.Sprintf("unknown transport type '%s'\n", key))
+		}
+	}
+
+	return // unreachable
+
+}
+
 type poolLookup func(name string) (*Pool, error)
 
 func parsePushs(v interface{}, pl poolLookup) (p []Push, err error) {
@@ -244,3 +315,18 @@ func parseComboMapping(m map[string]string) (c zfs.ComboMapping, err error) {
 	return
 
 }
+
+func (t SSHTransport) Connect() (r rpc.RPCRequester, err error) {
+	var stream io.ReadWriteCloser
+	var rpcTransport sshbytestream.SSHTransport
+	copier.Copy(rpcTransport, t)
+	if stream, err = sshbytestream.Outgoing(rpcTransport); err != nil {
+		return
+	}
+	return rpc.ConnectByteStreamRPC(stream)
+}
+
+func (t LocalTransport) Connect() (r rpc.RPCRequester, err error) {
+	// TODO ugly hidden global variable reference
+	return rpc.ConnectLocalRPC(handler), nil
+}

cmd/sampleconf/zrepl.yml

@@ -1,9 +1,18 @@
 pools:
   - name: offsite_backups
-    url: ssh://db2@backups1/db2/
+    transport:
-  - name:
+      ssh:
+        zrepl_identity: db2
+        host: backups1.example.com
+        user: root
+        port: 22
+        command: ssh
+        args: []
+
   - name: local_mirror
-    url: local://mirrorpool/mirrors/tank
+    transport:
+      local:
+        pool: mirrorpool
 
 pushs:
   - to: offsite_backups
@@ -11,16 +20,17 @@ pushs:
       - tank/var/db
       - tank/usr/home
 
-  - to: local_mirror
-    datasets:
-      - tank
-
 pulls:
-  - from: offsite
+  - from: offsite_backups
     mapping: {
       # like in sinks
     }
 
+  - from: local_mirror
+    mapping: {
+      "tank/usr/home":"mirrorpool/foo/bar"
+    }
+
 sinks:
 
   # direct mapping