Merge pull request #60 from tuna/dev

Dev

.travis.yml

@@ -1,3 +1,5 @@
+sudo: required
+
 language: go
 go:
     - 1.6
@@ -11,8 +13,14 @@ before_install:
 os:
     - linux
 
+services:
+    - docker
+
 before_script:
-    - sudo cgcreate -t travis -a travis -g memory:tunasync
+    - lssubsys -am
+    - sudo cgcreate -a $USER -t $USER -g cpu:tunasync
+    - sudo cgcreate -a $USER -t $USER -g memory:tunasync
+    - docker pull alpine
 
 script:
     - ./.testandcover.bash

cmd/tunasync/tunasync.go

@@ -134,7 +134,7 @@ func main() {
 	app.Name = "tunasync"
 	app.Usage = "tunasync mirror job management tool"
 	app.EnableBashCompletion = true
-	app.Version = "0.1"
+	app.Version = tunasync.Version
 	app.Commands = []cli.Command{
 		{
 			Name:    "manager",

cmd/tunasynctl/tunasynctl.go

@@ -99,8 +99,11 @@ func initialize(c *cli.Context) error {
 	}
 
 	// parse base url of the manager server
-	baseURL = fmt.Sprintf("https://%s:%d",
-		cfg.ManagerAddr, cfg.ManagerPort)
+	if cfg.CACert != "" {
+		baseURL = fmt.Sprintf("https://%s:%d", cfg.ManagerAddr, cfg.ManagerPort)
+	} else {
+		baseURL = fmt.Sprintf("http://%s:%d", cfg.ManagerAddr, cfg.ManagerPort)
+	}
 
 	logger.Infof("Use manager address: %s", baseURL)
 
@@ -183,6 +186,57 @@ func listJobs(c *cli.Context) error {
 	return nil
 }
 
+func updateMirrorSize(c *cli.Context) error {
+	args := c.Args()
+	if len(args) != 2 {
+		return cli.NewExitError("Usage: tunasynctl -w <worker-id> <mirror> <size>", 1)
+	}
+	workerID := c.String("worker")
+	mirrorID := args.Get(0)
+	mirrorSize := args.Get(1)
+
+	msg := struct {
+		Name string `json:"name"`
+		Size string `json:"size"`
+	}{
+		Name: mirrorID,
+		Size: mirrorSize,
+	}
+
+	url := fmt.Sprintf(
+		"%s/workers/%s/jobs/%s/size", baseURL, workerID, mirrorID,
+	)
+
+	resp, err := tunasync.PostJSON(url, msg, client)
+	if err != nil {
+		return cli.NewExitError(
+			fmt.Sprintf("Failed to send request to manager: %s",
+				err.Error()),
+			1)
+	}
+	defer resp.Body.Close()
+	body, _ := ioutil.ReadAll(resp.Body)
+	if resp.StatusCode != http.StatusOK {
+		return cli.NewExitError(
+			fmt.Sprintf("Manager failed to update mirror size: %s", body), 1,
+		)
+	}
+
+	var status tunasync.MirrorStatus
+	json.Unmarshal(body, &status)
+	if status.Size != mirrorSize {
+		return cli.NewExitError(
+			fmt.Sprintf(
+				"Mirror size error, expecting %s, manager returned %s",
+				mirrorSize, status.Size,
+			), 1,
+		)
+	}
+
+	logger.Infof("Successfully updated mirror size to %s", mirrorSize)
+	return nil
+}
+
 func flushDisabledJobs(c *cli.Context) error {
 	req, err := http.NewRequest("DELETE", baseURL+flushDisabledPath, nil)
 	if err != nil {
@@ -322,7 +376,7 @@ func main() {
 
 	app := cli.NewApp()
 	app.EnableBashCompletion = true
-	app.Version = "0.1"
+	app.Version = tunasync.Version
 	app.Name = "tunasynctl"
 	app.Usage = "control client for tunasync manager"
 
@@ -382,6 +436,18 @@ func main() {
 			Flags:  commonFlags,
 			Action: initializeWrapper(listWorkers),
 		},
+		{
+			Name:  "set-size",
+			Usage: "Set mirror size",
+			Flags: append(
+				commonFlags,
+				cli.StringFlag{
+					Name:  "worker, w",
+					Usage: "specify worker-id of the mirror job",
+				},
+			),
+			Action: initializeWrapper(updateMirrorSize),
+		},
 		{
 			Name:   "start",
 			Usage:  "Start a job",

internal/msg.go

@@ -5,7 +5,7 @@ import (
 	"time"
 )
 
-// A StatusUpdateMsg represents a msg when
+// A MirrorStatus represents a msg when
 // a worker has done syncing
 type MirrorStatus struct {
 	Name       string     `json:"name"`

internal/version.go

@@ -0,0 +1,3 @@
+package internal
+
+const Version string = "0.2-dev"

manager/db.go

@@ -125,7 +125,7 @@ func (b *boltAdapter) GetMirrorStatus(workerID, mirrorID string) (m MirrorStatus
 		bucket := tx.Bucket([]byte(_statusBucketKey))
 		v := bucket.Get([]byte(id))
 		if v == nil {
-			return fmt.Errorf("no mirror %s exists in worker %s", mirrorID, workerID)
+			return fmt.Errorf("no mirror '%s' exists in worker '%s'", mirrorID, workerID)
 		}
 		err := json.Unmarshal(v, &m)
 		return err
@@ -182,7 +182,7 @@ func (b *boltAdapter) FlushDisabledJobs() (err error) {
 				err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
 				continue
 			}
-			if m.Status == Disabled {
+			if m.Status == Disabled || len(m.Name) == 0 {
 				err = c.Delete()
 			}
 		}

manager/server.go

@@ -1,6 +1,7 @@
 package manager
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"time"
@@ -87,6 +88,7 @@ func GetTUNASyncManager(cfg *Config) *Manager {
 		workerValidateGroup.GET(":id/jobs", s.listJobsOfWorker)
 		// post job status
 		workerValidateGroup.POST(":id/jobs/:job", s.updateJobOfWorker)
+		workerValidateGroup.POST(":id/jobs/:job/size", s.updateMirrorSize)
 	}
 
 	// for tunasynctl to post commands
@@ -225,6 +227,12 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) {
 	var status MirrorStatus
 	c.BindJSON(&status)
 	mirrorName := status.Name
+	if len(mirrorName) == 0 {
+		s.returnErrJSON(
+			c, http.StatusBadRequest,
+			errors.New("Mirror Name should not be empty"),
+		)
+	}
 
 	curStatus, _ := s.adapter.GetMirrorStatus(workerID, mirrorName)
 
@@ -235,20 +243,19 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) {
 		status.LastUpdate = curStatus.LastUpdate
 	}
 
+	// Only message with meaningful size updates the mirror size
+	if len(curStatus.Size) > 0 && curStatus.Size != "unknown" {
+		if len(status.Size) == 0 || status.Size == "unknown" {
+			status.Size = curStatus.Size
+		}
+	}
+
 	// for logging
 	switch status.Status {
-	case Success:
-		logger.Noticef("Job [%s] @<%s> success", status.Name, status.Worker)
-	case Failed:
-		logger.Warningf("Job [%s] @<%s> failed", status.Name, status.Worker)
 	case Syncing:
 		logger.Noticef("Job [%s] @<%s> starts syncing", status.Name, status.Worker)
-	case Disabled:
-		logger.Noticef("Job [%s] @<%s> disabled", status.Name, status.Worker)
-	case Paused:
-		logger.Noticef("Job [%s] @<%s> paused", status.Name, status.Worker)
 	default:
-		logger.Infof("Job [%s] @<%s> status: %s", status.Name, status.Worker, status.Status)
+		logger.Noticef("Job [%s] @<%s> %s", status.Name, status.Worker, status.Status)
 	}
 
 	newStatus, err := s.adapter.UpdateMirrorStatus(workerID, mirrorName, status)
@@ -263,6 +270,45 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) {
 	c.JSON(http.StatusOK, newStatus)
 }
 
+func (s *Manager) updateMirrorSize(c *gin.Context) {
+	workerID := c.Param("id")
+	type SizeMsg struct {
+		Name string `json:"name"`
+		Size string `json:"size"`
+	}
+	var msg SizeMsg
+	c.BindJSON(&msg)
+
+	mirrorName := msg.Name
+	status, err := s.adapter.GetMirrorStatus(workerID, mirrorName)
+	if err != nil {
+		logger.Errorf(
+			"Failed to get status of mirror %s @<%s>: %s",
+			mirrorName, workerID, err.Error(),
+		)
+		s.returnErrJSON(c, http.StatusInternalServerError, err)
+		return
+	}
+
+	// Only message with meaningful size updates the mirror size
+	if len(msg.Size) > 0 || msg.Size != "unknown" {
+		status.Size = msg.Size
+	}
+
+	logger.Noticef("Mirror size of [%s] @<%s>: %s", status.Name, status.Worker, status.Size)
+
+	newStatus, err := s.adapter.UpdateMirrorStatus(workerID, mirrorName, status)
+	if err != nil {
+		err := fmt.Errorf("failed to update job %s of worker %s: %s",
+			mirrorName, workerID, err.Error(),
+		)
+		c.Error(err)
+		s.returnErrJSON(c, http.StatusInternalServerError, err)
+		return
+	}
+	c.JSON(http.StatusOK, newStatus)
+}
+
 func (s *Manager) handleClientCmd(c *gin.Context) {
 	var clientCmd ClientCmd
 	c.BindJSON(&clientCmd)

manager/server_test.go

@@ -99,7 +99,7 @@ func TestHTTPServer(t *testing.T) {
 					IsMaster: true,
 					Status:   Success,
 					Upstream: "mirrors.tuna.tsinghua.edu.cn",
-					Size:     "3GB",
+					Size:     "unknown",
 				}
 				resp, err := PostJSON(fmt.Sprintf("%s/workers/%s/jobs/%s", baseURL, status.Worker, status.Name), status, nil)
 				defer resp.Body.Close()
@@ -139,6 +139,47 @@ func TestHTTPServer(t *testing.T) {
 					So(time.Now().Sub(m.LastUpdate.Time), ShouldBeLessThan, 1*time.Second)
 
 				})
+
+				Convey("Update size of a valid mirror", func(ctx C) {
+					msg := struct {
+						Name string `json:"name"`
+						Size string `json:"size"`
+					}{status.Name, "5GB"}
+
+					url := fmt.Sprintf("%s/workers/%s/jobs/%s/size", baseURL, status.Worker, status.Name)
+					resp, err := PostJSON(url, msg, nil)
+					So(err, ShouldBeNil)
+					So(resp.StatusCode, ShouldEqual, http.StatusOK)
+
+					Convey("Get new size of a mirror", func(ctx C) {
+						var ms []MirrorStatus
+						resp, err := GetJSON(baseURL+"/workers/test_worker1/jobs", &ms, nil)
+
+						So(err, ShouldBeNil)
+						So(resp.StatusCode, ShouldEqual, http.StatusOK)
+						// err = json.NewDecoder(resp.Body).Decode(&mirrorStatusList)
+						m := ms[0]
+						So(m.Name, ShouldEqual, status.Name)
+						So(m.Worker, ShouldEqual, status.Worker)
+						So(m.Status, ShouldEqual, status.Status)
+						So(m.Upstream, ShouldEqual, status.Upstream)
+						So(m.Size, ShouldEqual, "5GB")
+						So(m.IsMaster, ShouldEqual, status.IsMaster)
+						So(time.Now().Sub(m.LastUpdate), ShouldBeLessThan, 1*time.Second)
+					})
+				})
+
+				Convey("Update size of an invalid mirror", func(ctx C) {
+					msg := struct {
+						Name string `json:"name"`
+						Size string `json:"size"`
+					}{"Invalid mirror", "5GB"}
+
+					url := fmt.Sprintf("%s/workers/%s/jobs/%s/size", baseURL, status.Worker, status.Name)
+					resp, err := PostJSON(url, msg, nil)
+					So(err, ShouldBeNil)
+					So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
+				})
 			})
 
 			Convey("update mirror status of an inexisted worker", func(ctx C) {

worker/base_provider.go

@@ -23,7 +23,10 @@ type baseProvider struct {
 	logFile *os.File
 
 	cgroup *cgroupHook
-	hooks  []jobHook
+	zfs    *zfsHook
+	docker *dockerHook
+
+	hooks []jobHook
 }
 
 func (p *baseProvider) Name() string {
@@ -77,12 +80,17 @@ func (p *baseProvider) LogFile() string {
 			return s
 		}
 	}
-	panic("log dir is impossible to be unavailable")
+	panic("log file is impossible to be unavailable")
 }
 
 func (p *baseProvider) AddHook(hook jobHook) {
-	if cg, ok := hook.(*cgroupHook); ok {
-		p.cgroup = cg
+	switch v := hook.(type) {
+	case *cgroupHook:
+		p.cgroup = v
+	case *zfsHook:
+		p.zfs = v
+	case *dockerHook:
+		p.docker = v
 	}
 	p.hooks = append(p.hooks, hook)
 }
@@ -95,6 +103,14 @@ func (p *baseProvider) Cgroup() *cgroupHook {
 	return p.cgroup
 }
 
+func (p *baseProvider) ZFS() *zfsHook {
+	return p.zfs
+}
+
+func (p *baseProvider) Docker() *dockerHook {
+	return p.docker
+}
+
 func (p *baseProvider) prepareLogFile() error {
 	if p.LogFile() == "/dev/null" {
 		p.cmd.SetLogFile(nil)

worker/cgroup.go

@@ -15,35 +15,31 @@ import (
 	"github.com/codeskyblue/go-sh"
 )
 
-var cgSubsystem = "cpu"
-
 type cgroupHook struct {
 	emptyHook
 	provider  mirrorProvider
 	basePath  string
 	baseGroup string
 	created   bool
+	subsystem string
+	memLimit  string
 }
 
-func initCgroup(basePath string) {
-	if _, err := os.Stat(filepath.Join(basePath, "memory")); err == nil {
-		cgSubsystem = "memory"
-		return
-	}
-	logger.Warning("Memory subsystem of cgroup not enabled, fallback to cpu")
-}
-
-func newCgroupHook(p mirrorProvider, basePath, baseGroup string) *cgroupHook {
+func newCgroupHook(p mirrorProvider, basePath, baseGroup, subsystem, memLimit string) *cgroupHook {
 	if basePath == "" {
 		basePath = "/sys/fs/cgroup"
 	}
 	if baseGroup == "" {
 		baseGroup = "tunasync"
 	}
+	if subsystem == "" {
+		subsystem = "cpu"
+	}
 	return &cgroupHook{
 		provider:  p,
 		basePath:  basePath,
 		baseGroup: baseGroup,
+		subsystem: subsystem,
 	}
 }
 
@@ -52,13 +48,15 @@ func (c *cgroupHook) preExec() error {
 	if err := sh.Command("cgcreate", "-g", c.Cgroup()).Run(); err != nil {
 		return err
 	}
-	if cgSubsystem != "memory" {
+	if c.subsystem != "memory" {
 		return nil
 	}
-	if c.provider.Type() == provRsync || c.provider.Type() == provTwoStageRsync {
+	if c.memLimit != "" {
 		gname := fmt.Sprintf("%s/%s", c.baseGroup, c.provider.Name())
 		return sh.Command(
-			"cgset", "-r", "memory.limit_in_bytes=128M", gname,
+			"cgset", "-r",
+			fmt.Sprintf("memory.limit_in_bytes=%s", c.memLimit),
+			gname,
 		).Run()
 	}
 	return nil
@@ -76,7 +74,7 @@ func (c *cgroupHook) postExec() error {
 
 func (c *cgroupHook) Cgroup() string {
 	name := c.provider.Name()
-	return fmt.Sprintf("%s:%s/%s", cgSubsystem, c.baseGroup, name)
+	return fmt.Sprintf("%s:%s/%s", c.subsystem, c.baseGroup, name)
 }
 
 func (c *cgroupHook) killAll() error {
@@ -87,7 +85,7 @@ func (c *cgroupHook) killAll() error {
 
 	readTaskList := func() ([]int, error) {
 		taskList := []int{}
-		taskFile, err := os.Open(filepath.Join(c.basePath, cgSubsystem, c.baseGroup, name, "tasks"))
+		taskFile, err := os.Open(filepath.Join(c.basePath, c.subsystem, c.baseGroup, name, "tasks"))
 		if err != nil {
 			return taskList, err
 		}

worker/cgroup_test.go

@@ -72,11 +72,14 @@ sleep 30
 		provider, err := newCmdProvider(c)
 		So(err, ShouldBeNil)
 
-		initCgroup("/sys/fs/cgroup")
-		cg := newCgroupHook(provider, "/sys/fs/cgroup", "tunasync")
+		cg := newCgroupHook(provider, "/sys/fs/cgroup", "tunasync", "cpu", "")
 		provider.AddHook(cg)
 
 		err = cg.preExec()
+		if err != nil {
+			logger.Errorf("Failed to create cgroup")
+			return
+		}
 		So(err, ShouldBeNil)
 
 		go func() {
@@ -129,15 +132,18 @@ sleep 30
 		provider, err := newRsyncProvider(c)
 		So(err, ShouldBeNil)
 
-		initCgroup("/sys/fs/cgroup")
-		cg := newCgroupHook(provider, "/sys/fs/cgroup", "tunasync")
+		cg := newCgroupHook(provider, "/sys/fs/cgroup", "tunasync", "cpu", "512M")
 		provider.AddHook(cg)
 
-		cg.preExec()
-		if cgSubsystem == "memory" {
+		err = cg.preExec()
+		if err != nil {
+			logger.Errorf("Failed to create cgroup")
+			return
+		}
+		if cg.subsystem == "memory" {
 			memoLimit, err := ioutil.ReadFile(filepath.Join(cg.basePath, "memory", cg.baseGroup, provider.Name(), "memory.limit_in_bytes"))
 			So(err, ShouldBeNil)
-			So(strings.Trim(string(memoLimit), "\n"), ShouldEqual, strconv.Itoa(128*1024*1024))
+			So(strings.Trim(string(memoLimit), "\n"), ShouldEqual, strconv.Itoa(512*1024*1024))
 		}
 		cg.postExec()
 	})

worker/cmd_provider.go

@@ -64,6 +64,7 @@ func (p *cmdProvider) Start() error {
 		"TUNASYNC_MIRROR_NAME":  p.Name(),
 		"TUNASYNC_WORKING_DIR":  p.WorkingDir(),
 		"TUNASYNC_UPSTREAM_URL": p.upstreamURL,
+		"TUNASYNC_LOG_DIR":      p.LogDir(),
 		"TUNASYNC_LOG_FILE":     p.LogFile(),
 	}
 	for k, v := range p.env {

worker/config.go

@@ -37,6 +37,8 @@ type Config struct {
 	Manager managerConfig  `toml:"manager"`
 	Server  serverConfig   `toml:"server"`
 	Cgroup  cgroupConfig   `toml:"cgroup"`
+	ZFS     zfsConfig      `toml:"zfs"`
+	Docker  dockerConfig   `toml:"docker"`
 	Include includeConfig  `toml:"include"`
 	Mirrors []mirrorConfig `toml:"mirrors"`
 }
@@ -54,8 +56,17 @@ type globalConfig struct {
 
 type managerConfig struct {
 	APIBase string `toml:"api_base"`
-	CACert  string `toml:"ca_cert"`
-	Token   string `toml:"token"`
+	// this option overrides the APIBase
+	APIList []string `toml:"api_base_list"`
+	CACert  string   `toml:"ca_cert"`
+	// Token   string `toml:"token"`
+}
+
+func (mc managerConfig) APIBaseList() []string {
+	if len(mc.APIList) > 0 {
+		return mc.APIList
+	}
+	return []string{mc.APIBase}
 }
 
 type serverConfig struct {
@@ -67,9 +78,21 @@ type serverConfig struct {
 }
 
 type cgroupConfig struct {
-	Enable   bool   `toml:"enable"`
-	BasePath string `toml:"base_path"`
-	Group    string `toml:"group"`
+	Enable    bool   `toml:"enable"`
+	BasePath  string `toml:"base_path"`
+	Group     string `toml:"group"`
+	Subsystem string `toml:"subsystem"`
+}
+
+type dockerConfig struct {
+	Enable  bool     `toml:"enable"`
+	Volumes []string `toml:"volumes"`
+	Options []string `toml:"options"`
+}
+
+type zfsConfig struct {
+	Enable bool   `toml:"enable"`
+	Zpool  string `toml:"zpool"`
 }
 
 type includeConfig struct {
@@ -104,6 +127,12 @@ type mirrorConfig struct {
 	Username      string `toml:"username"`
 	Password      string `toml:"password"`
 	Stage1Profile string `toml:"stage1_profile"`
+
+	MemoryLimit string `toml:"memory_limit"`
+
+	DockerImage   string   `toml:"docker_image"`
+	DockerVolumes []string `toml:"docker_volumes"`
+	DockerOptions []string `toml:"docker_options"`
 }
 
 // LoadConfig loads configuration

worker/docker.go

@@ -0,0 +1,95 @@
+package worker
+
+import (
+	"fmt"
+	"os"
+)
+
+type dockerHook struct {
+	emptyHook
+	provider mirrorProvider
+	image    string
+	volumes  []string
+	options  []string
+}
+
+func newDockerHook(p mirrorProvider, gCfg dockerConfig, mCfg mirrorConfig) *dockerHook {
+	volumes := []string{}
+	volumes = append(volumes, gCfg.Volumes...)
+	volumes = append(volumes, mCfg.DockerVolumes...)
+
+	options := []string{}
+	options = append(options, gCfg.Options...)
+	options = append(options, mCfg.DockerOptions...)
+
+	return &dockerHook{
+		provider: p,
+		image:    mCfg.DockerImage,
+		volumes:  volumes,
+		options:  options,
+	}
+}
+
+func (d *dockerHook) preExec() error {
+	p := d.provider
+	logDir := p.LogDir()
+	logFile := p.LogFile()
+	workingDir := p.WorkingDir()
+
+	if _, err := os.Stat(workingDir); os.IsNotExist(err) {
+		logger.Debugf("Making dir %s", workingDir)
+		if err = os.MkdirAll(workingDir, 0755); err != nil {
+			return fmt.Errorf("Error making dir %s: %s", workingDir, err.Error())
+		}
+	}
+
+	// Override workingDir
+	ctx := p.EnterContext()
+	ctx.Set(
+		"volumes", []string{
+			fmt.Sprintf("%s:%s", logDir, logDir),
+			fmt.Sprintf("%s:%s", logFile, logFile),
+			fmt.Sprintf("%s:%s", workingDir, workingDir),
+		},
+	)
+	return nil
+}
+
+func (d *dockerHook) postExec() error {
+	// sh.Command(
+	// 	"docker", "rm", "-f", d.Name(),
+	// ).Run()
+	d.provider.ExitContext()
+	return nil
+}
+
+// Volumes returns the configured volumes and
+// runtime-needed volumes, including mirror dirs
+// and log files
+func (d *dockerHook) Volumes() []string {
+	vols := make([]string, len(d.volumes))
+	copy(vols, d.volumes)
+
+	p := d.provider
+	ctx := p.Context()
+	if ivs, ok := ctx.Get("volumes"); ok {
+		vs := ivs.([]string)
+		vols = append(vols, vs...)
+	}
+	return vols
+}
+
+func (d *dockerHook) LogFile() string {
+	p := d.provider
+	ctx := p.Context()
+	if iv, ok := ctx.Get(_LogFileKey + ":docker"); ok {
+		v := iv.(string)
+		return v
+	}
+	return p.LogFile()
+}
+
+func (d *dockerHook) Name() string {
+	p := d.provider
+	return "tunasync-job-" + p.Name()
+}

worker/docker_test.go

@@ -0,0 +1,97 @@
+package worker
+
+import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/codeskyblue/go-sh"
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func getDockerByName(name string) (string, error) {
+	// docker ps -f 'name=$name' --format '{{.Names}}'
+	out, err := sh.Command(
+		"docker", "ps",
+		"--filter", "name="+name,
+		"--format", "{{.Names}}",
+	).Output()
+	return string(out), err
+}
+
+func TestDocker(t *testing.T) {
+	Convey("Docker Should Work", t, func(ctx C) {
+		tmpDir, err := ioutil.TempDir("", "tunasync")
+		defer os.RemoveAll(tmpDir)
+		So(err, ShouldBeNil)
+		cmdScript := filepath.Join(tmpDir, "cmd.sh")
+		tmpFile := filepath.Join(tmpDir, "log_file")
+		expectedOutput := "HELLO_WORLD"
+
+		c := cmdConfig{
+			name:        "tuna-docker",
+			upstreamURL: "http://mirrors.tuna.moe/",
+			command:     "/bin/cmd.sh",
+			workingDir:  tmpDir,
+			logDir:      tmpDir,
+			logFile:     tmpFile,
+			interval:    600 * time.Second,
+			env: map[string]string{
+				"TEST_CONTENT": expectedOutput,
+			},
+		}
+
+		cmdScriptContent := `#!/bin/sh
+echo ${TEST_CONTENT}
+sleep 10
+`
+		err = ioutil.WriteFile(cmdScript, []byte(cmdScriptContent), 0755)
+		So(err, ShouldBeNil)
+
+		provider, err := newCmdProvider(c)
+		So(err, ShouldBeNil)
+
+		d := &dockerHook{
+			provider: provider,
+			image:    "alpine",
+			volumes: []string{
+				fmt.Sprintf("%s:%s", cmdScript, "/bin/cmd.sh"),
+			},
+		}
+		provider.AddHook(d)
+		So(provider.Docker(), ShouldNotBeNil)
+
+		err = d.preExec()
+		So(err, ShouldBeNil)
+
+		go func() {
+			err = provider.Run()
+			ctx.So(err, ShouldNotBeNil)
+		}()
+
+		time.Sleep(1 * time.Second)
+
+		// assert container running
+		names, err := getDockerByName(d.Name())
+		So(err, ShouldBeNil)
+		So(names, ShouldEqual, d.Name()+"\n")
+
+		err = provider.Terminate()
+		So(err, ShouldBeNil)
+
+		// container should be terminated and removed
+		names, err = getDockerByName(d.Name())
+		So(err, ShouldBeNil)
+		So(names, ShouldEqual, "")
+
+		// check log content
+		loggedContent, err := ioutil.ReadFile(provider.LogFile())
+		So(err, ShouldBeNil)
+		So(string(loggedContent), ShouldEqual, expectedOutput+"\n")
+
+		d.postExec()
+	})
+}

worker/exec_post_hook.go

@@ -71,6 +71,7 @@ func (h *execPostHook) Do() error {
 		"TUNASYNC_MIRROR_NAME":     p.Name(),
 		"TUNASYNC_WORKING_DIR":     p.WorkingDir(),
 		"TUNASYNC_UPSTREAM_URL":    p.Upstream(),
+		"TUNASYNC_LOG_DIR":         p.LogDir(),
 		"TUNASYNC_LOG_FILE":        p.LogFile(),
 		"TUNASYNC_JOB_EXIT_STATUS": exitStatus,
 	}

worker/provider.go

@@ -36,6 +36,10 @@ type mirrorProvider interface {
 	IsRunning() bool
 	// Cgroup
 	Cgroup() *cgroupHook
+	// ZFS
+	ZFS() *zfsHook
+	// Docker
+	Docker() *dockerHook
 
 	AddHook(hook jobHook)
 	Hooks() []jobHook
@@ -162,10 +166,22 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 	// Add Logging Hook
 	provider.AddHook(newLogLimiter(provider))
 
-	// Add Cgroup Hook
-	if cfg.Cgroup.Enable {
+	// Add ZFS Hook
+	if cfg.ZFS.Enable {
+		provider.AddHook(newZfsHook(provider, cfg.ZFS.Zpool))
+	}
+
+	// Add Docker Hook
+	if cfg.Docker.Enable && len(mirror.DockerImage) > 0 {
+		provider.AddHook(newDockerHook(provider, cfg.Docker, mirror))
+
+	} else if cfg.Cgroup.Enable {
+		// Add Cgroup Hook
 		provider.AddHook(
-			newCgroupHook(provider, cfg.Cgroup.BasePath, cfg.Cgroup.Group),
+			newCgroupHook(
+				provider, cfg.Cgroup.BasePath, cfg.Cgroup.Group,
+				cfg.Cgroup.Subsystem, mirror.MemoryLimit,
+			),
 		)
 	}
 

worker/runner.go

@@ -2,6 +2,7 @@ package worker
 
 import (
 	"errors"
+	"fmt"
 	"os"
 	"os/exec"
 	"strings"
@@ -9,6 +10,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/codeskyblue/go-sh"
 	"golang.org/x/sys/unix"
 )
 
@@ -31,11 +33,44 @@ type cmdJob struct {
 func newCmdJob(provider mirrorProvider, cmdAndArgs []string, workingDir string, env map[string]string) *cmdJob {
 	var cmd *exec.Cmd
 
-	if provider.Cgroup() != nil {
+	if d := provider.Docker(); d != nil {
+		c := "docker"
+		args := []string{
+			"run", "--rm",
+			"-a", "STDOUT", "-a", "STDERR",
+			"--name", d.Name(),
+			"-w", workingDir,
+		}
+		// specify user
+		args = append(
+			args, "-u",
+			fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid()),
+		)
+		// add volumes
+		for _, vol := range d.Volumes() {
+			logger.Debugf("volume: %s", vol)
+			args = append(args, "-v", vol)
+		}
+		// set env
+		for k, v := range env {
+			kv := fmt.Sprintf("%s=%s", k, v)
+			args = append(args, "-e", kv)
+		}
+		// apply options
+		args = append(args, d.options...)
+		// apply image and command
+		args = append(args, d.image)
+		// apply command
+		args = append(args, cmdAndArgs...)
+
+		cmd = exec.Command(c, args...)
+
+	} else if provider.Cgroup() != nil {
 		c := "cgexec"
 		args := []string{"-g", provider.Cgroup().Cgroup()}
 		args = append(args, cmdAndArgs...)
 		cmd = exec.Command(c, args...)
+
 	} else {
 		if len(cmdAndArgs) == 1 {
 			cmd = exec.Command(cmdAndArgs[0])
@@ -48,25 +83,28 @@ func newCmdJob(provider mirrorProvider, cmdAndArgs []string, workingDir string,
 		}
 	}
 
-	logger.Debugf("Executing command %s at %s", cmdAndArgs[0], workingDir)
-	if _, err := os.Stat(workingDir); os.IsNotExist(err) {
-		logger.Debugf("Making dir %s", workingDir)
-		if err = os.MkdirAll(workingDir, 0755); err != nil {
-			logger.Errorf("Error making dir %s", workingDir)
+	if provider.Docker() == nil {
+		logger.Debugf("Executing command %s at %s", cmdAndArgs[0], workingDir)
+		if _, err := os.Stat(workingDir); os.IsNotExist(err) {
+			logger.Debugf("Making dir %s", workingDir)
+			if err = os.MkdirAll(workingDir, 0755); err != nil {
+				logger.Errorf("Error making dir %s: %s", workingDir, err.Error())
+			}
 		}
+		cmd.Dir = workingDir
+		cmd.Env = newEnviron(env, true)
 	}
 
-	cmd.Dir = workingDir
-	cmd.Env = newEnviron(env, true)
-
 	return &cmdJob{
 		cmd:        cmd,
 		workingDir: workingDir,
 		env:        env,
+		provider:   provider,
 	}
 }
 
 func (c *cmdJob) Start() error {
+	// logger.Debugf("Command start: %v", c.cmd.Args)
 	c.finished = make(chan empty, 1)
 	return c.cmd.Start()
 }
@@ -95,6 +133,14 @@ func (c *cmdJob) Terminate() error {
 	if c.cmd == nil || c.cmd.Process == nil {
 		return errProcessNotStarted
 	}
+
+	if d := c.provider.Docker(); d != nil {
+		sh.Command(
+			"docker", "stop", "-t", "2", d.Name(),
+		).Run()
+		return nil
+	}
+
 	err := unix.Kill(c.cmd.Process.Pid, syscall.SIGTERM)
 	if err != nil {
 		return err

worker/worker.go

@@ -55,9 +55,6 @@ func GetTUNASyncWorker(cfg *Config) *Worker {
 		w.httpClient = httpClient
 	}
 
-	if cfg.Cgroup.Enable {
-		initCgroup(cfg.Cgroup.BasePath)
-	}
 	w.initJobs()
 	w.makeHTTPServer()
 	tunasyncWorker = w
@@ -389,28 +386,21 @@ func (w *Worker) URL() string {
 }
 
 func (w *Worker) registorWorker() {
-	url := fmt.Sprintf(
-		"%s/workers",
-		w.cfg.Manager.APIBase,
-	)
-
 	msg := WorkerStatus{
 		ID:  w.Name(),
 		URL: w.URL(),
 	}
 
-	if _, err := PostJSON(url, msg, w.httpClient); err != nil {
-		logger.Errorf("Failed to register worker")
+	for _, root := range w.cfg.Manager.APIBaseList() {
+		url := fmt.Sprintf("%s/workers", root)
+		logger.Debugf("register on manager url: %s", url)
+		if _, err := PostJSON(url, msg, w.httpClient); err != nil {
+			logger.Errorf("Failed to register worker")
+		}
 	}
 }
 
 func (w *Worker) updateStatus(job *mirrorJob, jobMsg jobMessage) {
-	url := fmt.Sprintf(
-		"%s/workers/%s/jobs/%s",
-		w.cfg.Manager.APIBase,
-		w.Name(),
-		jobMsg.name,
-	)
 	p := job.provider
 	smsg := MirrorStatus{
 		Name:     jobMsg.name,
@@ -422,19 +412,22 @@ func (w *Worker) updateStatus(job *mirrorJob, jobMsg jobMessage) {
 		ErrorMsg: jobMsg.msg,
 	}
 
-	if _, err := PostJSON(url, smsg, w.httpClient); err != nil {
-		logger.Errorf("Failed to update mirror(%s) status: %s", jobMsg.name, err.Error())
+	for _, root := range w.cfg.Manager.APIBaseList() {
+		url := fmt.Sprintf(
+			"%s/workers/%s/jobs/%s", root, w.Name(), jobMsg.name,
+		)
+		logger.Debugf("reporting on manager url: %s", url)
+		if _, err := PostJSON(url, smsg, w.httpClient); err != nil {
+			logger.Errorf("Failed to update mirror(%s) status: %s", jobMsg.name, err.Error())
+		}
 	}
 }
 
 func (w *Worker) fetchJobStatus() []MirrorStatus {
 	var mirrorList []MirrorStatus
+	apiBase := w.cfg.Manager.APIBaseList()[0]
 
-	url := fmt.Sprintf(
-		"%s/workers/%s/jobs",
-		w.cfg.Manager.APIBase,
-		w.Name(),
-	)
+	url := fmt.Sprintf("%s/workers/%s/jobs", apiBase, w.Name())
 
 	if _, err := GetJSON(url, &mirrorList, w.httpClient); err != nil {
 		logger.Errorf("Failed to fetch job status: %s", err.Error())

worker/zfs_hook.go

@@ -0,0 +1,45 @@
+package worker
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/codeskyblue/go-sh"
+)
+
+type zfsHook struct {
+	emptyHook
+	provider mirrorProvider
+	zpool    string
+}
+
+func newZfsHook(provider mirrorProvider, zpool string) *zfsHook {
+	return &zfsHook{
+		provider: provider,
+		zpool:    zpool,
+	}
+}
+
+// create zfs dataset for a new mirror
+func (z *zfsHook) preJob() error {
+	workingDir := z.provider.WorkingDir()
+	if _, err := os.Stat(workingDir); os.IsNotExist(err) {
+		// sudo zfs create $zfsDataset
+		// sudo zfs set mountpoint=${absPath} ${zfsDataset}
+
+		zfsDataset := fmt.Sprintf("%s/%s", z.zpool, z.provider.Name())
+		// Unknown issue of ZFS:
+		// dataset name should not contain upper case letters
+		zfsDataset = strings.ToLower(zfsDataset)
+		logger.Infof("Creating ZFS dataset %s", zfsDataset)
+		if err := sh.Command("sudo", "zfs", "create", zfsDataset).Run(); err != nil {
+			return err
+		}
+		logger.Infof("Mount ZFS dataset %s to %s", zfsDataset, workingDir)
+		if err := sh.Command("sudo", "zfs", "set", "mountpoint="+workingDir, zfsDataset).Run(); err != nil {
+			return err
+		}
+	}
+	return nil
+}