Merge branch 'master' into release-0.5

.gitlab-ci.yml

@@ -314,6 +314,7 @@ linux_kubernetes_operator_PVC:
         when: always
     except:
         - schedules
+    allow_failure: true
 
 linux_deletion_simple:
     stage: test
@@ -479,8 +480,9 @@ trigger_e2e_test:
     script:
         - echo - curl -X POST -F "token=$E2E_TRIGGER_TOKEN" -F "ref=master" -F "variables[DOTMESH_CI_COMMIT_SHA]=$CI_COMMIT_SHA" -F "variables[DOTMESH_CI_BUILD_REF_NAME]=$CI_BUILD_REF_NAME" http://gitlab.dotmesh.io:9999/api/v4/projects/31/trigger/pipeline
         - curl -X POST -F "token=$E2E_TRIGGER_TOKEN" -F "ref=master" -F "variables[DOTMESH_CI_COMMIT_SHA]=$CI_COMMIT_SHA" -F "variables[DOTMESH_CI_BUILD_REF_NAME]=$CI_BUILD_REF_NAME" http://gitlab.dotmesh.io:9999/api/v4/projects/31/trigger/pipeline
+    only:
+        - master
     except:
-        - /^release-.*$/
         - schedules
 
 deploy_unstable_build:

cmd/dm/pkg/commands/cluster.go

@@ -67,6 +67,7 @@ var (
 	usePoolDir         string
 	usePoolName        string
 	discoveryUrl       string
+	port               int
 )
 
 // names of environment variables we pass from the content of `dm cluster {init,join}`
@@ -250,6 +251,11 @@ func NewCmdClusterInit(out io.Writer) *cobra.Command {
 		&serverCount, "count", 1,
 		"Initial cluster size",
 	)
+	// TODO: need to block using 32608, and probably block anything lower than 32xx for host port reasons?
+	cmd.Flags().IntVar(
+		&port, "port", 0,
+		"Port to run cluster on",
+	)
 	return cmd
 }
 
@@ -663,6 +669,10 @@ func startDotmeshContainer(pkiPath string) error {
 		"-e", fmt.Sprintf("DOTMESH_UPGRADES_URL=%s", checkpointUrl),
 		"-e", fmt.Sprintf("DOTMESH_UPGRADES_INTERVAL_SECONDS=%d", checkpointInterval),
 	}
+	if port != 0 {
+		args = append(args, "-e")
+		args = append(args, fmt.Sprintf("DOTMESH_SERVER_PORT=%d", port))
+	}
 
 	// inject the inherited env variables from the context of the dm binary into require_zfs.sh
 	for _, envName := range inheritedEnvironment {
@@ -877,7 +887,7 @@ func clusterCommonSetup(clusterUrl, adminPassword, adminKey, pkiPath string) err
 			return err
 		}
 	}
-	err = config.AddRemote("local", "admin", getHostFromEnv(), 0, adminKey)
+	err = config.AddRemote("local", "admin", getHostFromEnv(), port, adminKey)
 	if err != nil {
 		return err
 	}
@@ -913,6 +923,7 @@ func clusterCommonSetup(clusterUrl, adminPassword, adminKey, pkiPath string) err
 				time.Sleep(250 * time.Millisecond)
 			}
 			if err != nil {
+				fmt.Printf("Errored creating api")
 				e()
 				return false
 			}

cmd/dm/pkg/commands/dot.go

@@ -238,17 +238,17 @@ func dotShow(cmd *cobra.Command, args []string, out io.Writer) error {
 		return err
 	}
 
-	var localDot string
+	var qualifiedDotName string
 	if len(args) == 1 {
-		localDot = args[0]
+		qualifiedDotName = args[0]
 	} else {
-		localDot, err = dm.CurrentVolume()
+		qualifiedDotName, err = dm.CurrentVolume()
 		if err != nil {
 			return err
 		}
 	}
 
-	namespace, dot, err := remotes.ParseNamespacedVolume(localDot)
+	namespace, dot, err := remotes.ParseNamespacedVolume(qualifiedDotName)
 	if err != nil {
 		return err
 	}
@@ -307,12 +307,12 @@ func dotShow(cmd *cobra.Command, args []string, out io.Writer) error {
 		}
 	}
 
-	currentBranch, err := dm.CurrentBranch(localDot)
+	currentBranch, err := dm.CurrentBranch(qualifiedDotName)
 	if err != nil {
 		return err
 	}
 
-	bs, err := dm.AllBranches(localDot)
+	bs, err := dm.AllBranches(qualifiedDotName)
 	if err != nil {
 		return err
 	}
@@ -329,7 +329,7 @@ func dotShow(cmd *cobra.Command, args []string, out io.Writer) error {
 		if branch == "master" {
 			branchDot = masterDot
 		} else {
-			branchDot, err = dm.BranchInfo(namespace, localDot, branch)
+			branchDot, err = dm.BranchInfo(namespace, dot, branch)
 			if err != nil {
 				return err
 			}
@@ -371,7 +371,7 @@ func dotShow(cmd *cobra.Command, args []string, out io.Writer) error {
 			branchInternalName = ""
 		}
 
-		latency, err := dm.GetReplicationLatencyForBranch(localDot, branchInternalName)
+		latency, err := dm.GetReplicationLatencyForBranch(qualifiedDotName, branchInternalName)
 		if err != nil {
 			fmt.Fprintf(out, "unable to fetch replication status (%s), proceeding...\n", err)
 		} else {

cmd/dm/pkg/commands/main.go

@@ -165,7 +165,6 @@ func NewCmdRemote(out io.Writer) *cobra.Command {
 				if err != nil {
 					return err
 				}
-				fmt.Printf("dm client %#v\n", dm)
 				// allow this to be used be a script
 				apiKey := os.Getenv("DOTMESH_PASSWORD")
 				if apiKey == "" {

cmd/dm/pkg/remotes/api.go

@@ -2,16 +2,16 @@ package remotes
 
 import (
 	"fmt"
-	"golang.org/x/net/context"
-	"gopkg.in/cheggaaa/pb.v1"
 	"io"
-	"log"
 	"os"
 	"reflect"
 	"regexp"
 	"sort"
 	"strings"
 	"time"
+
+	"golang.org/x/net/context"
+	"gopkg.in/cheggaaa/pb.v1"
 )
 
 const DEFAULT_BRANCH string = "master"
@@ -761,9 +761,6 @@ func (dm *DotmeshAPI) PollTransfer(transferId string, out io.Writer) error {
 			}
 		}
 
-		if debugMode {
-			log.Printf("\nGot DotmeshRPC.GetTransfer response:  %+v", result)
-		}
 		if !started {
 			bar = pb.New64(result.Size)
 			bar.ShowFinalTime = false
@@ -772,6 +769,10 @@ func (dm *DotmeshAPI) PollTransfer(transferId string, out io.Writer) error {
 			bar.Start()
 			started = true
 		}
+
+		if result.Size != 0 {
+			bar.Total = result.Size
+		}
 		// Numbers reported by data transferred thru dotmesh versus size
 		// of stream reported by 'zfs send -nP' are off by a few kilobytes,
 		// fudge it (maybe no one will notice).
@@ -843,6 +844,7 @@ push
 type TransferRequest struct {
 	Peer             string
 	User             string
+	Port             int
 	ApiKey           string
 	Direction        string
 	LocalNamespace   string
@@ -986,6 +988,7 @@ func (dm *DotmeshAPI) RequestTransfer(
 		"DotmeshRPC.Transfer", TransferRequest{
 			Peer:             remote.Hostname,
 			User:             remote.User,
+			Port:             remote.Port,
 			ApiKey:           remote.ApiKey,
 			Direction:        direction,
 			LocalNamespace:   localNamespace,

cmd/dotmesh-server/pkg/dind-flexvolume/flexvolume.go

@@ -38,7 +38,7 @@ var flexVolumeDebug = false
 const DIND_SHARED_FOLDER = "/dotmesh-test-pools/dind-flexvolume"
 
 func System(cmd string, args ...string) error {
-	log.Printf("[system] running %s %s", cmd, args)
+	logger.Printf("[system] running %s %s", cmd, args)
 	c := exec.Command(cmd, args...)
 	c.Stdout = os.Stdout
 	c.Stderr = os.Stderr

cmd/dotmesh-server/pkg/main/client.go

@@ -19,13 +19,15 @@ type JsonRpcClient struct {
 	User     string
 	Hostname string
 	ApiKey   string
+	Port     int
 }
 
-func NewJsonRpcClient(user, hostname, apiKey string) *JsonRpcClient {
+func NewJsonRpcClient(user, hostname, apiKey string, port int) *JsonRpcClient {
 	return &JsonRpcClient{
 		User:     user,
 		Hostname: hostname,
 		ApiKey:   apiKey,
+		Port:     port,
 	}
 }
 
@@ -35,9 +37,15 @@ func (j *JsonRpcClient) CallRemote(
 	ctx context.Context, method string, args interface{}, result interface{},
 ) error {
 	// RPCs are always between clusters, so "external"
-	url, err := deduceUrl(ctx, []string{j.Hostname}, "external", j.User, j.ApiKey)
-	if err != nil {
-		return err
+	var url string
+	var err error
+	if j.Port == 0 {
+		url, err = deduceUrl(ctx, []string{j.Hostname}, "external", j.User, j.ApiKey)
+		if err != nil {
+			return err
+		}
+	} else {
+		url = fmt.Sprintf("http://%s:%d", j.Hostname, j.Port)
 	}
 	url = fmt.Sprintf("%s/rpc", url)
 	return j.reallyCallRemote(ctx, method, args, result, url)

cmd/dotmesh-server/pkg/main/controller.go

@@ -50,8 +50,8 @@ func NewInMemoryState(localPoolId string, config Config) *InMemoryState {
 		// a sort of global event bus for filesystems getting new snapshots on
 		// their masters, keyed on filesystem name, which interested parties
 		// such as slaves for that filesystem may subscribe to
-		newSnapsOnMaster:     NewObserver(),
-		localReceiveProgress: NewObserver(),
+		newSnapsOnMaster:     NewObserver("newSnapsOnMaster"),
+		localReceiveProgress: NewObserver("localReceiveProgress"),
 		// containers that are running with dotmesh volumes by filesystem id
 		containers:     d,
 		containersLock: &sync.Mutex{},
@@ -139,13 +139,13 @@ func (s *InMemoryState) alignMountStateWithMasters(filesystemId string) error {
 		fs, ok := (*s.filesystems)[filesystemId]
 		if !ok {
 			log.Printf(
-				"[maybeMountFilesystem] not doing anything - cannot find %v in fsMachines",
+				"[alignMountStateWithMasters] not doing anything - cannot find %v in fsMachines",
 				filesystemId,
 			)
 			return nil, false, fmt.Errorf("cannot find %v in fsMachines", filesystemId)
 		}
 		log.Printf(
-			"[maybeMountFilesystem] called for %v; masterFor=%v, myNodeId=%v; mounted=%b",
+			"[alignMountStateWithMasters] called for %v; masterFor=%v, myNodeId=%v; mounted=%b",
 			filesystemId,
 			s.masterFor(filesystemId),
 			s.myNodeId,