Bump version to 0.4.3

commit 4540ba24c72cb2d24e2e04870025dfa233dedf30
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 11:16:13 2020 +0800

    wait longer

commit c8f07b81a7fe5fdef9224e8bc187500c4d67f049
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 10:55:49 2020 +0800

    try to Terminate

commit 10d2d4b9d0756cf8f60fe27e1e41ae29b5ea6cbe
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 10:50:26 2020 +0800

    forward the error

commit 38c96ee44d31088b9e6de67ebb745358fac8d49a
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 10:31:39 2020 +0800

    now enable the assertion

commit 3b3c46a065a035d906d4cc5022d42e30b1f52a08
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 10:26:40 2020 +0800

    rm un-related info

commit dd7ef7e3d0a0765c1fc48296d70966b3b4d581dd
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 10:12:01 2020 +0800

    print err of provider.Run

commit 49a7b57dbf52d410c0dfe796be9c2f6213884931
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 09:55:48 2020 +0800

    wait until it exits

commit a3e8f699072e3252b3300c667f1425a966aedb39
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 09:54:19 2020 +0800

    targeting alpine:3.8

commit f30b8565049bb373a1a91a34ad07c8c3df8e1036
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 09:47:27 2020 +0800

    see what happens

commit 8c21229a8be8e2ac0737bbc4bb88ba54e9fb7a20
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Sat Mar 14 09:30:50 2020 +0800

    remove one assertion

commit 123368e6ef07aa63c489bb49bdf370d3abdd17bb
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Fri Mar 13 23:32:45 2020 +0800

    docker test somehow works now

commit 94fa294a9bbedb569e6dd9cc7e4f27e73ed97443
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Fri Mar 13 23:27:12 2020 +0800

    should use -d

commit b35bae2a9cb5e006c513da95377ab9487fc4341a
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Fri Mar 13 23:22:25 2020 +0800

    docker run not working??

commit 9aea0036f434d333087f0cff3ce5165a53554e5f
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Fri Mar 13 23:12:39 2020 +0800

    test if docker works

commit f92578b159587a8bbda296bbf9261fb4c5e2f186
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Fri Mar 13 17:42:00 2020 +0800

    debugging docker_test

commit b649e32f76549711af597ce3a642309a41a08bf9
Author: z4yx <z4yx@users.noreply.github.com>
Date:   Fri Mar 13 17:27:55 2020 +0800

    Revert "remove docker_test.go"

    This reverts commit a517a4bb64.

.github/workflows/release.yml

@@ -0,0 +1,54 @@
+name: release
+on:
+  push:
+    # Sequence of patterns matched against refs/tags
+    tags:
+    - 'v*' # Push events to matching v*, i.e. v1.0, v20.15.10
+
+jobs:
+
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    steps:
+
+    - name: Set up Go 1.13
+      uses: actions/setup-go@v1
+      with:
+        go-version: 1.13
+      id: go
+
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v2
+
+    - name: Get dependencies
+      run: |
+        go get -v -t -d ./cmd/tunasync
+        go get -v -t -d ./cmd/tunasynctl
+
+    - name: Build
+      run: |
+        make tunasync
+        make tunasynctl
+        tar -jcf build/tunasync-linux-bin.tar.bz2 -C build tunasync tunasynctl
+
+    - name: Create Release
+      id: create_release
+      uses: actions/create-release@v1
+      env:
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      with:
+        tag_name: ${{ github.ref }}
+        release_name: Release ${{ github.ref }}
+        draft: false
+        prerelease: false
+    - name: Upload Release Asset
+      id: upload-release-asset 
+      uses: actions/upload-release-asset@v1
+      env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+      with:
+          upload_url: ${{ steps.create_release.outputs.upload_url }} # This pulls from the CREATE RELEASE step above, referencing it's ID to get its outputs object, which include a `upload_url`. See this blog post for more info: https://jasonet.co/posts/new-features-of-github-actions/#passing-data-to-future-steps 
+          asset_path: ./build/tunasync-linux-bin.tar.bz2
+          asset_name: tunasync-linux-bin.tar.bz2
+          asset_content_type: application/x-bzip2

.github/workflows/tunasync.yml

@@ -0,0 +1,68 @@
+name: tunasync
+
+on: [push]
+
+jobs:
+
+  build:
+    name: Build
+    runs-on: ubuntu-latest
+    steps:
+
+    - name: Set up Go 1.13
+      uses: actions/setup-go@v1
+      with:
+        go-version: 1.13
+      id: go
+
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v2
+
+    - name: Get dependencies
+      run: |
+        go get -v -t -d ./cmd/tunasync
+        go get -v -t -d ./cmd/tunasynctl
+
+    - name: Build
+      run: |
+        make tunasync
+        make tunasynctl
+
+
+  test:
+    name: Test
+    runs-on: ubuntu-latest
+    steps:
+
+    - name: Setup test dependencies
+      run: |
+        sudo apt-get update
+        sudo apt-get install -y cgroup-bin
+        docker pull alpine:3.8
+        lssubsys -am
+        sudo cgcreate -a $USER -t $USER -g cpu:tunasync
+        sudo cgcreate -a $USER -t $USER -g memory:tunasync
+
+    - name: Set up Go 1.13
+      uses: actions/setup-go@v1
+      with:
+        go-version: 1.13
+      id: go
+
+    - name: Check out code into the Go module directory
+      uses: actions/checkout@v2
+
+    - name: Run Unit tests.
+      run: make test
+
+    - name: Convert coverage to lcov
+      uses: jandelgado/gcov2lcov-action@v1.0.0
+      with:
+          infile: profile.cov
+          outfile: coverage.lcov
+
+    - name: Coveralls
+      uses: coverallsapp/github-action@v1.0.1
+      with:
+          github-token: ${{ secrets.github_token }}
+          path-to-lcov: coverage.lcov

.testandcover.bash

@@ -1,30 +0,0 @@
-#!/bin/bash
-function die() {
-  echo $*
-  exit 1
-}
-
-export GOPATH=`pwd`:$GOPATH
-
-make travis
-
-# Initialize profile.cov
-echo "mode: count" > profile.cov
-
-# Initialize error tracking
-ERROR=""
-
-# Test each package and append coverage profile info to profile.cov
-for pkg in `cat .testpackages.txt`
-do
-	go test -v -covermode=count -coverprofile=profile_tmp.cov $pkg || ERROR="Error testing $pkg"
-	
-	[ -f profile_tmp.cov ] && {
-		tail -n +2 profile_tmp.cov >> profile.cov || die "Unable to append coverage for $pkg"
-	}
-done
-
-if [ ! -z "$ERROR" ]
-then
-    die "Encountered error, last error was: $ERROR"
-fi

.travis.yml

@@ -1,44 +0,0 @@
-sudo: required
-
-language: go
-go:
-    - 1.8
-
-before_install:
-    - sudo apt-get install cgroup-bin
-    - go get github.com/smartystreets/goconvey
-    - go get golang.org/x/tools/cmd/cover
-    - go get -v github.com/mattn/goveralls
-
-os:
-    - linux
-
-services:
-    - docker
-
-before_script:
-    - 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
-
-after_success:
-    - goveralls -coverprofile=profile.cov -service=travis-ci
-
-before_deploy: "echo 'ready to deploy?'"
-
-deploy:
-    provider: releases
-    file:
-        - "build/tunasync-linux-bin.tar.gz"
-    api_key:
-        secure: ZOYL/CALrVJsZzbZqUMSI89Gw4zsBJH1StD/2yTyG45GfKgvtK4hG0S5cQM/L0wcikjEkgxSMsmr4ycq+OwbN++gc0umfoAQ/VSjzetiobAlT1E854aRKRjT82WxYdnPW2fsFjuEJTcyZmcbgJGTMi86MDt7w8tEjLomhd1+rUo=
-    skip_cleanup: true
-    overwrite: true
-    on:
-        tags: true
-        all_branches: true
-        repo: tuna/tunasync

Makefile

@@ -2,8 +2,6 @@ LDFLAGS="-X main.buildstamp=`date -u '+%s'` -X main.githash=`git rev-parse HEAD`
 
 all: get tunasync tunasynctl
 
-travis: get tunasync tunasynctl travis-package
-
 get: 
 	go get ./cmd/tunasync
 	go get ./cmd/tunasynctl
@@ -17,5 +15,5 @@ tunasync: build
 tunasynctl: build
 	go build -o build/tunasynctl -ldflags ${LDFLAGS} github.com/tuna/tunasync/cmd/tunasynctl
 
-travis-package: tunasync tunasynctl
+test:
-	tar zcf build/tunasync-linux-bin.tar.gz -C build tunasync tunasynctl
+	go test -v -covermode=count -coverprofile=profile.cov ./...

README.md

@@ -1,8 +1,8 @@
 tunasync
 ========
 
-[![Build Status](https://travis-ci.org/tuna/tunasync.svg?branch=dev)](https://travis-ci.org/tuna/tunasync)
+![Build Status](https://github.com/tuna/tunasync/workflows/tunasync/badge.svg)
-[![Coverage Status](https://coveralls.io/repos/github/tuna/tunasync/badge.svg?branch=dev)](https://coveralls.io/github/tuna/tunasync?branch=dev)
+[![Coverage Status](https://coveralls.io/repos/github/tuna/tunasync/badge.svg?branch=master)](https://coveralls.io/github/tuna/tunasync?branch=master)
 [![Commitizen friendly](https://img.shields.io/badge/commitizen-friendly-brightgreen.svg)](http://commitizen.github.io/cz-cli/)
 ![GPLv3](https://img.shields.io/badge/license-GPLv3-blue.svg)
 
@@ -40,84 +40,23 @@ Pre-built binary for Linux x86_64 is available at [Github releases](https://gith
 # Job Run Process
 
 
-PreSyncing           Syncing                               Success
+PreSyncing                           Syncing                               Success
-+-----------+     +-----------+    +-------------+     +--------------+
++-----------+     +----------+    +-----------+    +-------------+     +--------------+
-|  pre-job  +--+->|  job run  +--->|  post-exec  +-+-->| post-success |
+|  pre-job  +--+->| pre-exec +--->|  job run  +--->|  post-exec  +-+-->| post-success |
-+-----------+  ^  +-----------+    +-------------+ |   +--------------+
++-----------+  ^  +----------+    +-----------+    +-------------+ |   +--------------+
-			   |                                   |
+               |                                                   |
-			   |      +-----------------+          | Failed
+               |                +-----------------+                | Failed
-			   +------+    post-fail    |<---------+
+               +----------------+    post-fail    |<---------------+
-			          +-----------------+
+                                +-----------------+
 ```
 
-## Generate Self-Signed Certificate
-
-First, create root CA
-
-```
-openssl genrsa -out rootCA.key 2048
-openssl req -x509 -new -nodes -key rootCA.key -days 365 -out rootCA.crt
-```
-
-Create host key
-
-```
-openssl genrsa -out host.key 2048
-```
-
-Now create CSR, before that, write a `req.cnf`
-
-```
-[req]
-distinguished_name = req_distinguished_name
-req_extensions = v3_req
-
-[req_distinguished_name]
-countryName = Country Name (2 letter code)
-countryName_default = CN
-stateOrProvinceName = State or Province Name (full name)
-stateOrProvinceName_default = BJ
-localityName = Locality Name (eg, city)
-localityName_default = Beijing
-organizationalUnitName  = Organizational Unit Name (eg, section)
-organizationalUnitName_default  = TUNA
-commonName = Common Name (server FQDN or domain name)
-commonName_default = <server_FQDN>
-commonName_max  = 64
-
-[v3_req]
-# Extensions to add to a certificate request
-basicConstraints = CA:FALSE
-keyUsage = nonRepudiation, digitalSignature, keyEncipherment
-subjectAltName = @alt_names
-
-[alt_names]
-DNS.1 = <server_FQDN_1>
-DNS.2 = <server_FQDN_2>
-```
-
-Substitute `<server_FQDN>` with your server's FQDN, then run
-
-```
-openssl req -new -key host.key -out host.csr -config req.cnf
-```
-
-Finally generate and sign host cert with root CA
-
-```
-openssl x509 -req -in host.csr -CA rootCA.crt -CAkey rootCA.key -CAcreateserial -out host.crt -days 365 -extensions v3_req -extfile req.cnf
-```
 
 ## Building
 
-Setup GOPATH like [this](https://golang.org/cmd/go/#hdr-GOPATH_environment_variable).
+Go version: 1.13
-
-Then:
 
 ```
-go get -d github.com/tuna/tunasync/cmd/tunasync
+make all
-cd $GOPATH/src/github.com/tuna/tunasync
-make
 ```
 
-If you have multiple `GOPATH`s, replace the `$GOPATH` with your first one.
+Binaries in the `build/`.

cmd/tunasync/tunasync.go

@@ -11,7 +11,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/pkg/profile"
 	"gopkg.in/op/go-logging.v1"
-	"gopkg.in/urfave/cli.v1"
+	"github.com/urfave/cli"
 
 	tunasync "github.com/tuna/tunasync/internal"
 	"github.com/tuna/tunasync/manager"
@@ -60,7 +60,7 @@ func startWorker(c *cli.Context) error {
 		os.Exit(1)
 	}
 
-	w := worker.GetTUNASyncWorker(cfg)
+	w := worker.NewTUNASyncWorker(cfg)
 	if w == nil {
 		logger.Errorf("Error intializing TUNA sync worker.")
 		os.Exit(1)

cmd/tunasynctl/tunasynctl.go

@@ -11,8 +11,8 @@ import (
 	"time"
 
 	"github.com/BurntSushi/toml"
+	"github.com/urfave/cli"
 	"gopkg.in/op/go-logging.v1"
-	"gopkg.in/urfave/cli.v1"
 
 	tunasync "github.com/tuna/tunasync/internal"
 )
@@ -375,6 +375,11 @@ func cmdJob(cmd tunasync.CmdVerb) cli.ActionFunc {
 
 func cmdWorker(cmd tunasync.CmdVerb) cli.ActionFunc {
 	return func(c *cli.Context) error {
+
+		if c.String("worker") == "" {
+			return cli.NewExitError("Please specify the worker with -w <worker-id>", 1)
+		}
+
 		cmd := tunasync.ClientCmd{
 			Cmd:      cmd,
 			WorkerID: c.String("worker"),

docs/zh_CN/get_started.md

@@ -42,7 +42,7 @@ interval = 1
 
 [manager]
 api_base = "http://localhost:12345"
-token = "some_token"
+token = ""
 ca_cert = ""
 
 [cgroup]
@@ -98,6 +98,22 @@ $ tunasync worker --config ~/tunasync_demo/worker.conf
 $ tunasynctl list -p 12345 --all
 ```
 
+tunasynctl 也支持配置文件。配置文件可以放在 `/etc/tunasync/ctl.conf` 或者 `~/.config/tunasync/ctl.conf` 两个位置，后者可以覆盖前者的配置值。
+
+配置文件内容为：
+
+```
+manager_addr = "127.0.0.1"
+manager_port = 12345
+ca_cert = ""
+```
+
+### 安全
+
+worker 和 manager 之间用 http(s) 通信，如果你 worker 和 manager 都是在本机，那么没必要使用 https。此时 manager 就不指定 `ssl_key` 和 `ssl_cert`，留空；worker 的 `ca_cert` 留空，`api_base` 以 `http://` 开头。
+
+如果需要加密的通信，manager 需要指定 `ssl_key` 和 `ssl_cert`，worker 要指定 `ca_cert`，并且 `api_base` 应该是 `https://` 开头。
+
 ## 更进一步
 
 可以参看
@@ -108,3 +124,7 @@ $ tunasync worker --help
 ```
 
 可以看一下 log 目录
+
+一些 worker 配置文件示例 [workers.conf](workers.conf) 
+
+你可能会用到的操作 [tips.md](tips.md) 

docs/zh_CN/tips.md

@@ -0,0 +1,93 @@
+## 删除某worker的某镜像
+
+先确定已经给tunasynctl写好config文件：`~/.config/tunasync/ctl.conf`
+
+```toml
+manager_addr = "127.0.0.1"
+manager_port = 12345
+ca_cert = ""
+```
+
+接着
+
+```shell
+$ tunasynctl disable -w <worker_id> <mirror_name>
+$ tunasynctl flush
+```
+
+
+## 热重载 `worker.conf`
+
+```shell
+$ tunasynctl reload -w <worker_id>
+```
+
+e.g. 删除 `test_worker` 的 `elvish` 镜像：
+
+1. 删除存放镜像的文件夹
+
+2. 删除 `worker.conf` 中对应的 `mirror` 段落
+
+3. 接着操作：
+
+```shell
+$ tunasynctl reload -w test_worker
+$ tunasynctl disable -w test_worker elvish
+$ tunasynctl flush
+```
+
+4. （可选）最后删除日志文件夹里的日志
+
+
+## 删除worker
+
+```shell
+$ tunasynctl rm-worker -w <worker_id>
+```
+
+e.g.
+
+```shell
+$ tunasynctl rm-worker -w test_worker
+```
+
+
+## 更新镜像的大小
+
+```shell
+$ tunasynctl set-size -w <worker_id> <mirror_name> <size>
+```
+
+其中，末尾的 <size> 参数，由操作者设定，或由某定时脚本生成
+
+由于 `du -s` 比较耗时，故镜像大小可直接由rsync的日志文件读出
+
+
+## Btrfs 文件系统快照
+
+如果镜像文件存放在以 Btrfs 为文件系统的分区中，可启用由 Btrfs 提供的快照 (Snapshot) 功能。对于每一个镜像，tunasync 在每次成功同步后更新其快照。
+
+在 `worker.conf` 中添加如下配置，即可启用 Btrfs 快照功能：
+
+```toml
+[btrfs_snapshot]
+enable = true
+snapshot_path = "/path/to/snapshot/directory"
+```
+
+其中 `snapshot_path` 为快照所在目录。如将其作为发布版本，则镜像同步过程对于镜像站用户而言具有原子性。如此可避免用户接收到仍处于“中间态”的（未完成同步的）文件。
+
+也可以在 `[[mirrors]]` 中为特定镜像单独指定快照路径，如：
+
+```toml
+[[mirrors]]
+name = "elvish"
+provider = "rsync"
+upstream = "rsync://rsync.elvish.io/elvish/"
+interval = 1440
+snapshot_path = "/data/publish/elvish"
+```
+
+**提示：** 
+
+若运行 tunasync 的用户无 root 权限，请确保该用户对镜像同步目录和快照目录均具有写和执行权限，并使用 [`user_subvol_rm_allowed` 选项](https://btrfs.wiki.kernel.org/index.php/Manpage/btrfs(5)#MOUNT_OPTIONS)挂载相应的 Btrfs 分区。

docs/zh_CN/workers.conf

@@ -0,0 +1,79 @@
+
+[global]
+name = "mirror_worker"
+log_dir = "/srv/tunasync/log/tunasync/{{.Name}}"
+mirror_dir = "/srv/tunasync"
+concurrent = 10
+interval = 1
+
+[manager]
+api_base = "http://localhost:12345"
+token = "some_token"
+ca_cert = ""
+
+[cgroup]
+enable = false
+base_path = "/sys/fs/cgroup"
+group = "tunasync"
+
+[server]
+hostname = "localhost"
+listen_addr = "127.0.0.1"
+listen_port = 6000
+ssl_cert = ""
+ssl_key = ""
+
+[[mirrors]]
+name = "adobe-fonts"
+interval = 1440
+provider = "command"
+upstream = "https://github.com/adobe-fonts"
+#https://github.com/tuna/tunasync-scripts/blob/master/adobe-fonts.sh
+command = "/home/scripts/adobe-fonts.sh"
+docker_image = "tunathu/tunasync-scripts:latest"
+
+[[mirrors]]
+name = "anaconda"
+provider = "command"
+upstream = "https://repo.continuum.io/"
+#https://github.com/tuna/tunasync-scripts/blob/master/anaconda.py
+command = "/home/scripts/anaconda.py"
+interval = 1440
+docker_image = "tunathu/tunasync-scripts:latest"
+
+[[mirrors]]
+name = "gnu"
+provider = "rsync"
+upstream = "rsync://mirrors.ocf.berkeley.edu/gnu/"
+rsync_options = [ "--delete-excluded" ]
+memory_limit = "256M"
+
+[[mirrors]]
+name = "pypi"
+provider = "command"
+upstream = "https://pypi.python.org/"
+#https://github.com/tuna/tunasync-scripts/blob/master/pypi.sh
+command = "/home/scripts/pypi.sh"
+docker_image = "tunathu/tunasync-scripts:latest"
+interval = 5
+        # set environment varialbes
+        [mirrors.env]
+        INIT = "0"
+
+
+[[mirrors]]
+name = "debian"
+interval = 720
+provider = "rsync"
+upstream = "rsync://mirrors.tuna.tsinghua.edu.cn/debian/"
+memory_limit = "256M"
+
+[[mirrors]]
+name = "ubuntu"
+provider = "two-stage-rsync"
+stage1_profile = "debian"
+upstream = "rsync://archive.ubuntu.com/ubuntu/"
+rsync_options = [ "--delete-excluded" ]
+memory_limit = "256M"
+
+# vim: ft=toml

go.mod

@@ -0,0 +1,20 @@
+module github.com/tuna/tunasync
+
+go 1.13
+
+require (
+	github.com/BurntSushi/toml v0.3.1
+	github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239
+	github.com/boltdb/bolt v1.3.1
+	github.com/codeskyblue/go-sh v0.0.0-20190412065543-76bd3d59ff27
+	github.com/dennwc/btrfs v0.0.0-20190517175702-d917b30ff035
+	github.com/gin-gonic/gin v1.5.0
+	github.com/mattn/goveralls v0.0.5 // indirect
+	github.com/pkg/profile v1.4.0
+	github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46
+	github.com/smartystreets/goconvey v1.6.4
+	github.com/urfave/cli v1.22.3
+	golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527
+	golang.org/x/tools v0.0.0-20200312194400-c312e98713c2 // indirect
+	gopkg.in/op/go-logging.v1 v1.0.0-20160211212156-b2cb9fa56473
+)

go.sum

@@ -0,0 +1,99 @@
+github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
+github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239 h1:kFOfPq6dUM1hTo4JG6LR5AXSUEsOjtdm0kw0FtQtMJA=
+github.com/anmitsu/go-shlex v0.0.0-20161002113705-648efa622239/go.mod h1:2FmKhYUyUczH0OGQWaF5ceTx0UBShxjsH6f8oGKYe2c=
+github.com/boltdb/bolt v1.3.1 h1:JQmyP4ZBrce+ZQu0dY660FMfatumYDLun9hBCUVIkF4=
+github.com/boltdb/bolt v1.3.1/go.mod h1:clJnj/oiGkjum5o1McbSZDSLxVThjynRyGBgiAx27Ps=
+github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0 h1:sDMmm+q/3+BukdIpxwO365v/Rbspp2Nt5XntgQRXq8Q=
+github.com/codegangsta/inject v0.0.0-20150114235600-33e0aa1cb7c0/go.mod h1:4Zcjuz89kmFXt9morQgcfYZAYZ5n8WHjt81YYWIwtTM=
+github.com/codeskyblue/go-sh v0.0.0-20190412065543-76bd3d59ff27 h1:HHUr4P/aKh4quafGxDT9LDasjGdlGkzLbfmmrlng3kA=
+github.com/codeskyblue/go-sh v0.0.0-20190412065543-76bd3d59ff27/go.mod h1:VQx0hjo2oUeQkQUET7wRwradO6f+fN5jzXgB/zROxxE=
+github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d h1:U+s90UTSYgptZMwQh2aRr3LuazLJIa+Pg3Kc1ylSYVY=
+github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dennwc/btrfs v0.0.0-20190517175702-d917b30ff035 h1:4e+UEZaKPx0ZEiCMPUHMV51RGwbb1VJGCYqRFn/qmWM=
+github.com/dennwc/btrfs v0.0.0-20190517175702-d917b30ff035/go.mod h1:MYsOV9Dgsec3FFSOjywi0QK5r6TeBbdWxdrMGtiYXHA=
+github.com/dennwc/ioctl v1.0.0 h1:DsWAAjIxRqNcLn9x6mwfuf2pet3iB7aK90K4tF16rLg=
+github.com/dennwc/ioctl v1.0.0/go.mod h1:ellh2YB5ldny99SBU/VX7Nq0xiZbHphf1DrtHxxjMk0=
+github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE=
+github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI=
+github.com/gin-gonic/gin v1.5.0 h1:fi+bqFAx/oLK54somfCtEZs9HeH1LHVoEPUgARpTqyc=
+github.com/gin-gonic/gin v1.5.0/go.mod h1:Nd6IXA8m5kNZdNEHMBd93KT+mdY3+bewLgRvmCsR2Do=
+github.com/go-playground/locales v0.12.1 h1:2FITxuFt/xuCNP1Acdhv62OzaCiviiE4kotfhkmOqEc=
+github.com/go-playground/locales v0.12.1/go.mod h1:IUMDtCfWo/w/mtMfIE/IG2K+Ey3ygWanZIBtBW0W2TM=
+github.com/go-playground/universal-translator v0.16.0 h1:X++omBR/4cE2MNg91AoC3rmGrCjJ8eAeUP/K/EKx4DM=
+github.com/go-playground/universal-translator v0.16.0/go.mod h1:1AnU7NaIRDWWzGEKwgtJRd2xk99HeFyHw3yid4rvQIY=
+github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
+github.com/json-iterator/go v1.1.7/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
+github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
+github.com/leodido/go-urn v1.1.0 h1:Sm1gr51B1kKyfD2BlRcLSiEkffoG96g6TPv6eRoEiB8=
+github.com/leodido/go-urn v1.1.0/go.mod h1:+cyI34gQWZcE1eQU7NVgKkkzdXDQHr1dBMtdAPozLkw=
+github.com/mattn/go-isatty v0.0.9 h1:d5US/mDsogSGW37IV293h//ZFaeajb69h+EHFsv2xGg=
+github.com/mattn/go-isatty v0.0.9/go.mod h1:YNRxwqDuOph6SZLI9vUUz6OYw3QyUt7WiY2yME+cCiQ=
+github.com/mattn/goveralls v0.0.5 h1:spfq8AyZ0cCk57Za6/juJ5btQxeE1FaEGMdfcI+XO48=
+github.com/mattn/goveralls v0.0.5/go.mod h1:Xg2LHi51faXLyKXwsndxiW6uxEEQT9+3sjGzzwU4xy0=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/pkg/profile v1.4.0 h1:uCmaf4vVbWAOZz36k1hrQD7ijGRzLwaME8Am/7a4jZI=
+github.com/pkg/profile v1.4.0/go.mod h1:NWz/XGvpEW1FyYQ7fCx4dqYBLlfTcE+A9FLAkNKqjFE=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/russross/blackfriday/v2 v2.0.1 h1:lPqVAte+HuHNfhJ/0LC98ESWRz8afy9tM/0RK8m9o+Q=
+github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 h1:GHRpF1pTW19a8tTFrMLUcfWwyC0pnifVo2ClaLq+hP8=
+github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46/go.mod h1:uAQ5PCi+MFsC7HjREoAz1BU+Mq60+05gifQSsHSDG/8=
+github.com/shurcooL/sanitized_anchor_name v1.0.0 h1:PdmoCO6wvbs+7yrJyMORt4/BmY5IYyJwS/kOiWx8mHo=
+github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
+github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykElWQ6/NYmHa3jpm/yHnI4xSofP+UP6SpjHcSeM=
+github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc=
+github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s=
+github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+github.com/ugorji/go v1.1.7 h1:/68gy2h+1mWMrwZFeD1kQialdSzAb432dtpeJ42ovdo=
+github.com/ugorji/go v1.1.7/go.mod h1:kZn38zHttfInRq0xu/PH0az30d+z6vm202qpg1oXVMw=
+github.com/ugorji/go/codec v1.1.7 h1:2SvQaVZ1ouYrrKKwoSk2pzd4A9evlKJb9oTL+OaLUSs=
+github.com/ugorji/go/codec v1.1.7/go.mod h1:Ax+UKWsSmolVDwsd+7N3ZtXu+yMGCf907BLYF3GoBXY=
+github.com/urfave/cli v1.22.3 h1:FpNT6zq26xNpHZy08emi755QwzLPs6Pukqjlc7RfOMU=
+github.com/urfave/cli v1.22.3/go.mod h1:Gos4lmkARVdJ6EkW0WaNv/tZAAMe9V7XWyB60NtXRu0=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
+golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190813064441-fde4db37ae7a/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527 h1:uYVVQ9WP/Ds2ROhcaGPeIdVq0RIXVLwsHlnvJ+cT1So=
+golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/tools v0.0.0-20190328211700-ab21143f2384 h1:TFlARGu6Czu1z7q93HTxcP1P+/ZFC/IKythI5RzrnRg=
+golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200113040837-eac381796e91/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200312194400-c312e98713c2 h1:6TB4+MaZlkcSsJDu+BS5yxSEuZIYhjWz+jhbSLEZylI=
+golang.org/x/tools v0.0.0-20200312194400-c312e98713c2/go.mod h1:o4KQGtdN14AW+yjsvvwRTJJuXz8XRtIHtEnmAXLyFUw=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE=
+gopkg.in/go-playground/validator.v9 v9.29.1 h1:SvGtYmN60a5CVKTOzMSyfzWDeZRxRuGvRQyEAKbw1xc=
+gopkg.in/go-playground/validator.v9 v9.29.1/go.mod h1:+c9/zcJMFNgbLvly1L1V+PpxWdVbfP1avr/N00E2vyQ=
+gopkg.in/op/go-logging.v1 v1.0.0-20160211212156-b2cb9fa56473 h1:6D+BvnJ/j6e222UW8s2qTSe3wGBtvo0MbVQG/c5k8RE=
+gopkg.in/op/go-logging.v1 v1.0.0-20160211212156-b2cb9fa56473/go.mod h1:N1eN2tsCx0Ydtgjl4cqmbRCsY4/+z4cYDeqwZTk6zog=
+gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
+gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

internal/msg.go

@@ -14,6 +14,7 @@ type MirrorStatus struct {
 	Status     SyncStatus `json:"status"`
 	LastUpdate time.Time  `json:"last_update"`
 	LastEnded  time.Time  `json:"last_ended"`
+	Scheduled  time.Time  `json:"next_schedule"`
 	Upstream   string     `json:"upstream"`
 	Size       string     `json:"size"`
 	ErrorMsg   string     `json:"error_msg"`
@@ -28,6 +29,15 @@ type WorkerStatus struct {
 	LastOnline time.Time `json:"last_online"` // last seen
 }
 
+type MirrorSchedules struct {
+	Schedules []MirrorSchedule `json:"schedules"`
+}
+
+type MirrorSchedule struct {
+	MirrorName   string    `json:"name"`
+	NextSchedule time.Time `json:"next_schedule"`
+}
+
 // A CmdVerb is an action to a job or worker
 type CmdVerb uint8
 

internal/status_web.go

@@ -45,6 +45,8 @@ type WebMirrorStatus struct {
 	LastUpdateTs stampTime  `json:"last_update_ts"`
 	LastEnded    textTime   `json:"last_ended"`
 	LastEndedTs  stampTime  `json:"last_ended_ts"`
+	Scheduled    textTime   `json:"next_schedule"`
+	ScheduledTs  stampTime  `json:"next_schedule_ts"`
 	Upstream     string     `json:"upstream"`
 	Size         string     `json:"size"` // approximate size
 }
@@ -58,6 +60,8 @@ func BuildWebMirrorStatus(m MirrorStatus) WebMirrorStatus {
 		LastUpdateTs: stampTime{m.LastUpdate},
 		LastEnded:    textTime{m.LastEnded},
 		LastEndedTs:  stampTime{m.LastEnded},
+		Scheduled:    textTime{m.Scheduled},
+		ScheduledTs:  stampTime{m.Scheduled},
 		Upstream:     m.Upstream,
 		Size:         m.Size,
 	}

internal/status_web_test.go

@@ -21,6 +21,8 @@ func TestStatus(t *testing.T) {
 			LastUpdateTs: stampTime{t},
 			LastEnded:    textTime{t},
 			LastEndedTs:  stampTime{t},
+			Scheduled:    textTime{t},
+			ScheduledTs:  stampTime{t},
 			Size:         "5GB",
 			Upstream:     "rsync://mirrors.tuna.tsinghua.edu.cn/tunalinux/",
 		}
@@ -42,6 +44,10 @@ func TestStatus(t *testing.T) {
 		So(m2.LastEndedTs.Unix(), ShouldEqual, m.LastEnded.Unix())
 		So(m2.LastEnded.UnixNano(), ShouldEqual, m.LastEnded.UnixNano())
 		So(m2.LastEndedTs.UnixNano(), ShouldEqual, m.LastEnded.UnixNano())
+		So(m2.Scheduled.Unix(), ShouldEqual, m.Scheduled.Unix())
+		So(m2.ScheduledTs.Unix(), ShouldEqual, m.Scheduled.Unix())
+		So(m2.Scheduled.UnixNano(), ShouldEqual, m.Scheduled.UnixNano())
+		So(m2.ScheduledTs.UnixNano(), ShouldEqual, m.Scheduled.UnixNano())
 		So(m2.Size, ShouldEqual, m.Size)
 		So(m2.Upstream, ShouldEqual, m.Upstream)
 	})
@@ -53,6 +59,7 @@ func TestStatus(t *testing.T) {
 			Status:     Failed,
 			LastUpdate: time.Now().Add(-time.Minute * 30),
 			LastEnded:  time.Now(),
+			Scheduled:  time.Now().Add(time.Minute * 5),
 			Upstream:   "mirrors.tuna.tsinghua.edu.cn",
 			Size:       "4GB",
 		}
@@ -70,6 +77,10 @@ func TestStatus(t *testing.T) {
 		So(m2.LastEndedTs.Unix(), ShouldEqual, m.LastEnded.Unix())
 		So(m2.LastEnded.UnixNano(), ShouldEqual, m.LastEnded.UnixNano())
 		So(m2.LastEndedTs.UnixNano(), ShouldEqual, m.LastEnded.UnixNano())
+		So(m2.Scheduled.Unix(), ShouldEqual, m.Scheduled.Unix())
+		So(m2.ScheduledTs.Unix(), ShouldEqual, m.Scheduled.Unix())
+		So(m2.Scheduled.UnixNano(), ShouldEqual, m.Scheduled.UnixNano())
+		So(m2.ScheduledTs.UnixNano(), ShouldEqual, m.Scheduled.UnixNano())
 		So(m2.Size, ShouldEqual, m.Size)
 		So(m2.Upstream, ShouldEqual, m.Upstream)
 	})

internal/util.go

@@ -8,6 +8,7 @@ import (
 	"errors"
 	"io/ioutil"
 	"net/http"
+	"regexp"
 	"time"
 )
 
@@ -84,3 +85,14 @@ func GetJSON(url string, obj interface{}, client *http.Client) (*http.Response,
 	}
 	return resp, json.Unmarshal(body, obj)
 }
+
+func ExtractSizeFromRsyncLog(content []byte) string {
+	// (?m) flag enables multi-line mode
+	re := regexp.MustCompile(`(?m)^Total file size: ([0-9\.]+[KMGTP]?) bytes`)
+	matches := re.FindAllSubmatch(content, -1)
+	// fmt.Printf("%q\n", matches)
+	if len(matches) == 0 {
+		return ""
+	}
+	return string(matches[len(matches)-1][1])
+}

internal/util_test.go

@@ -0,0 +1,32 @@
+package internal
+
+import (
+	"testing"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestExtractSizeFromRsyncLog(t *testing.T) {
+	realLogContent := `
+Number of files: 998,470 (reg: 925,484, dir: 58,892, link: 14,094)
+Number of created files: 1,049 (reg: 1,049)
+Number of deleted files: 1,277 (reg: 1,277)
+Number of regular files transferred: 5,694
+Total file size: 1.33T bytes
+Total transferred file size: 2.86G bytes
+Literal data: 780.62M bytes
+Matched data: 2.08G bytes
+File list size: 37.55M
+File list generation time: 7.845 seconds
+File list transfer time: 0.000 seconds
+Total bytes sent: 7.55M
+Total bytes received: 823.25M
+
+sent 7.55M bytes  received 823.25M bytes  5.11M bytes/sec
+total size is 1.33T  speedup is 1,604.11
+`
+	Convey("Log parser should work", t, func() {
+		res := ExtractSizeFromRsyncLog([]byte(realLogContent))
+		So(res, ShouldEqual, "1.33T")
+	})
+}

internal/version.go

@@ -1,3 +1,3 @@
 package internal
 
-const Version string = "0.3.3"
+const Version string = "0.4.3"

manager/config.go

@@ -2,7 +2,7 @@ package manager
 
 import (
 	"github.com/BurntSushi/toml"
-	"gopkg.in/urfave/cli.v1"
+	"github.com/urfave/cli"
 )
 
 // A Config is the top-level toml-serializaible config struct

manager/config_test.go

@@ -9,7 +9,7 @@ import (
 
 	"github.com/BurntSushi/toml"
 	. "github.com/smartystreets/goconvey/convey"
-	"gopkg.in/urfave/cli.v1"
+	"github.com/urfave/cli"
 )
 
 func TestConfig(t *testing.T) {

manager/server.go

@@ -91,6 +91,7 @@ func GetTUNASyncManager(cfg *Config) *Manager {
 		// post job status
 		workerValidateGroup.POST(":id/jobs/:job", s.updateJobOfWorker)
 		workerValidateGroup.POST(":id/jobs/:job/size", s.updateMirrorSize)
+		workerValidateGroup.POST(":id/schedules", s.updateSchedulesOfWorker)
 	}
 
 	// for tunasynctl to post commands
@@ -240,6 +241,48 @@ func (s *Manager) returnErrJSON(c *gin.Context, code int, err error) {
 	})
 }
 
+func (s *Manager) updateSchedulesOfWorker(c *gin.Context) {
+	workerID := c.Param("id")
+	var schedules MirrorSchedules
+	c.BindJSON(&schedules)
+
+	for _, schedule := range schedules.Schedules {
+		mirrorName := schedule.MirrorName
+		if len(mirrorName) == 0 {
+			s.returnErrJSON(
+				c, http.StatusBadRequest,
+				errors.New("Mirror Name should not be empty"),
+			)
+		}
+
+		curStatus, err := s.adapter.GetMirrorStatus(workerID, mirrorName)
+		if err != nil {
+			fmt.Errorf("failed to get job %s of worker %s: %s",
+				mirrorName, workerID, err.Error(),
+			)
+			continue
+		}
+
+		if curStatus.Scheduled == schedule.NextSchedule {
+			// no changes, skip update
+			continue
+		}
+
+		curStatus.Scheduled = schedule.NextSchedule
+		_, err = s.adapter.UpdateMirrorStatus(workerID, mirrorName, curStatus)
+		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
+		}
+	}
+	type empty struct{}
+	c.JSON(http.StatusOK, empty{})
+}
+
 func (s *Manager) updateJobOfWorker(c *gin.Context) {
 	workerID := c.Param("id")
 	var status MirrorStatus

manager/server_test.go

@@ -109,7 +109,7 @@ func TestHTTPServer(t *testing.T) {
 				So(res[_errorKey], ShouldEqual, "invalid workerID "+invalidWorker)
 			})
 
-			Convey("flush  disabled jobs", func(ctx C) {
+			Convey("flush disabled jobs", func(ctx C) {
 				req, err := http.NewRequest("DELETE", baseURL+"/jobs/disabled", nil)
 				So(err, ShouldBeNil)
 				clt := &http.Client{}
@@ -132,8 +132,8 @@ func TestHTTPServer(t *testing.T) {
 					Size:     "unknown",
 				}
 				resp, err := PostJSON(fmt.Sprintf("%s/workers/%s/jobs/%s", baseURL, status.Worker, status.Name), status, nil)
-				defer resp.Body.Close()
 				So(err, ShouldBeNil)
+				defer resp.Body.Close()
 				So(resp.StatusCode, ShouldEqual, http.StatusOK)
 
 				Convey("list mirror status of an existed worker", func(ctx C) {
@@ -202,6 +202,20 @@ func TestHTTPServer(t *testing.T) {
 					})
 				})
 
+				Convey("Update schedule of valid mirrors", func(ctx C) {
+					msg := MirrorSchedules{
+						[]MirrorSchedule{
+							MirrorSchedule{"arch-sync1", time.Now().Add(time.Minute * 10)},
+							MirrorSchedule{"arch-sync2", time.Now().Add(time.Minute * 7)},
+						},
+					}
+
+					url := fmt.Sprintf("%s/workers/%s/schedules", baseURL, status.Worker)
+					resp, err := PostJSON(url, msg, nil)
+					So(err, ShouldBeNil)
+					So(resp.StatusCode, ShouldEqual, http.StatusOK)
+				})
+
 				Convey("Update size of an invalid mirror", func(ctx C) {
 					msg := struct {
 						Name string `json:"name"`
@@ -218,8 +232,8 @@ func TestHTTPServer(t *testing.T) {
 				status.Status = Failed
 				time.Sleep(3 * time.Second)
 				resp, err = PostJSON(fmt.Sprintf("%s/workers/%s/jobs/%s", baseURL, status.Worker, status.Name), status, nil)
-				defer resp.Body.Close()
 				So(err, ShouldBeNil)
+				defer resp.Body.Close()
 				So(resp.StatusCode, ShouldEqual, http.StatusOK)
 
 				Convey("What if syncing job failed", func(ctx C) {
@@ -263,6 +277,24 @@ func TestHTTPServer(t *testing.T) {
 				So(err, ShouldBeNil)
 				So(msg[_errorKey], ShouldEqual, "invalid workerID "+invalidWorker)
 			})
+			Convey("update schedule of an non-existent worker", func(ctx C) {
+				invalidWorker := "test_worker2"
+				sch := MirrorSchedules{
+					[]MirrorSchedule{
+						MirrorSchedule{"arch-sync1", time.Now().Add(time.Minute * 10)},
+						MirrorSchedule{"arch-sync2", time.Now().Add(time.Minute * 7)},
+					},
+				}
+				resp, err := PostJSON(fmt.Sprintf("%s/workers/%s/schedules",
+					baseURL, invalidWorker), sch, nil)
+				So(err, ShouldBeNil)
+				So(resp.StatusCode, ShouldEqual, http.StatusBadRequest)
+				defer resp.Body.Close()
+				var msg map[string]string
+				err = json.NewDecoder(resp.Body).Decode(&msg)
+				So(err, ShouldBeNil)
+				So(msg[_errorKey], ShouldEqual, "invalid workerID "+invalidWorker)
+			})
 			Convey("handle client command", func(ctx C) {
 				cmdChan := make(chan WorkerCmd, 1)
 				workerServer := makeMockWorkerServer(cmdChan)
@@ -281,11 +313,11 @@ func TestHTTPServer(t *testing.T) {
 					// run the mock worker server
 					workerServer.Run(bindAddress)
 				}()
-				time.Sleep(50 * time.Microsecond)
+				time.Sleep(50 * time.Millisecond)
 				// verify the worker mock server is running
 				workerResp, err := http.Get(workerBaseURL + "/ping")
-				defer workerResp.Body.Close()
 				So(err, ShouldBeNil)
+				defer workerResp.Body.Close()
 				So(workerResp.StatusCode, ShouldEqual, http.StatusOK)
 
 				Convey("when client send wrong cmd", func(ctx C) {
@@ -295,8 +327,8 @@ func TestHTTPServer(t *testing.T) {
 						WorkerID: "not_exist_worker",
 					}
 					resp, err := PostJSON(baseURL+"/cmd", clientCmd, nil)
-					defer resp.Body.Close()
 					So(err, ShouldBeNil)
+					defer resp.Body.Close()
 					So(resp.StatusCode, ShouldEqual, http.StatusBadRequest)
 				})
 
@@ -308,9 +340,8 @@ func TestHTTPServer(t *testing.T) {
 					}
 
 					resp, err := PostJSON(baseURL+"/cmd", clientCmd, nil)
-					defer resp.Body.Close()
-
 					So(err, ShouldBeNil)
+					defer resp.Body.Close()
 					So(resp.StatusCode, ShouldEqual, http.StatusOK)
 					time.Sleep(50 * time.Microsecond)
 					select {

worker/base_provider.go

@@ -15,6 +15,7 @@ type baseProvider struct {
 	ctx      *Context
 	name     string
 	interval time.Duration
+	retry    int
 	isMaster bool
 
 	cmd       *cmdJob
@@ -50,6 +51,10 @@ func (p *baseProvider) Interval() time.Duration {
 	return p.interval
 }
 
+func (p *baseProvider) Retry() int {
+	return p.retry
+}
+
 func (p *baseProvider) IsMaster() bool {
 	return p.isMaster
 }
@@ -161,3 +166,7 @@ func (p *baseProvider) Terminate() error {
 
 	return err
 }
+
+func (p *baseProvider) DataSize() string {
+	return ""
+}

worker/btrfs_snapshot_hook.go

@@ -0,0 +1,90 @@
+package worker
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/dennwc/btrfs"
+)
+
+type btrfsSnapshotHook struct {
+	provider           mirrorProvider
+	mirrorSnapshotPath string
+}
+
+// the user who runs the jobs (typically `tunasync`) should be granted the permission to run btrfs commands
+// TODO: check if the filesystem is Btrfs
+func newBtrfsSnapshotHook(provider mirrorProvider, snapshotPath string, mirror mirrorConfig) *btrfsSnapshotHook {
+	mirrorSnapshotPath := mirror.SnapshotPath
+	if mirrorSnapshotPath == "" {
+		mirrorSnapshotPath = filepath.Join(snapshotPath, provider.Name())
+	}
+	return &btrfsSnapshotHook{
+		provider:           provider,
+		mirrorSnapshotPath: mirrorSnapshotPath,
+	}
+}
+
+// check if path `snapshotPath/providerName` exists
+// Case 1: Not exists => create a new subvolume
+// Case 2: Exists as a subvolume => nothing to do
+// Case 3: Exists as a directory => error detected
+func (h *btrfsSnapshotHook) preJob() error {
+	path := h.provider.WorkingDir()
+	if _, err := os.Stat(path); os.IsNotExist(err) {
+		// create subvolume
+		err := btrfs.CreateSubVolume(path)
+		if err != nil {
+			logger.Errorf("failed to create Btrfs subvolume %s: %s", path, err.Error())
+			return err
+		}
+		logger.Noticef("created new Btrfs subvolume %s", path)
+	} else {
+		if is, err := btrfs.IsSubVolume(path); err != nil {
+			return err
+		} else if !is {
+			return fmt.Errorf("path %s exists but isn't a Btrfs subvolume", path)
+		}
+	}
+	return nil
+}
+
+func (h *btrfsSnapshotHook) preExec() error {
+	return nil
+}
+
+func (h *btrfsSnapshotHook) postExec() error {
+	return nil
+}
+
+// delete old snapshot if exists, then create a new snapshot
+func (h *btrfsSnapshotHook) postSuccess() error {
+	if _, err := os.Stat(h.mirrorSnapshotPath); !os.IsNotExist(err) {
+		isSubVol, err := btrfs.IsSubVolume(h.mirrorSnapshotPath)
+		if err != nil {
+			return err
+		} else if !isSubVol {
+			return fmt.Errorf("path %s exists and isn't a Btrfs snapshot", h.mirrorSnapshotPath)
+		}
+		// is old snapshot => delete it
+		if err := btrfs.DeleteSubVolume(h.mirrorSnapshotPath); err != nil {
+			logger.Errorf("failed to delete old Btrfs snapshot %s", h.mirrorSnapshotPath)
+			return err
+		}
+		logger.Noticef("deleted old snapshot %s", h.mirrorSnapshotPath)
+	}
+	// create a new writable snapshot
+	// (the snapshot is writable so that it can be deleted easily)
+	if err := btrfs.SnapshotSubVolume(h.provider.WorkingDir(), h.mirrorSnapshotPath, false); err != nil {
+		logger.Errorf("failed to create new Btrfs snapshot %s", h.mirrorSnapshotPath)
+		return err
+	}
+	logger.Noticef("created new Btrfs snapshot %s", h.mirrorSnapshotPath)
+	return nil
+}
+
+// keep the old snapshot => nothing to do
+func (h *btrfsSnapshotHook) postFail() error {
+	return nil
+}

worker/cgroup.go

@@ -17,7 +17,6 @@ import (
 
 type cgroupHook struct {
 	emptyHook
-	provider  mirrorProvider
 	basePath  string
 	baseGroup string
 	created   bool
@@ -36,7 +35,9 @@ func newCgroupHook(p mirrorProvider, basePath, baseGroup, subsystem, memLimit st
 		subsystem = "cpu"
 	}
 	return &cgroupHook{
-		provider:  p,
+		emptyHook: emptyHook{
+			provider: p,
+		},
 		basePath:  basePath,
 		baseGroup: baseGroup,
 		subsystem: subsystem,

worker/cmd_provider.go

@@ -2,6 +2,9 @@ package worker
 
 import (
 	"errors"
+	"fmt"
+	"io/ioutil"
+	"regexp"
 	"time"
 
 	"github.com/anmitsu/go-shlex"
@@ -12,22 +15,29 @@ type cmdConfig struct {
 	upstreamURL, command        string
 	workingDir, logDir, logFile string
 	interval                    time.Duration
+	retry                       int
 	env                         map[string]string
+	failOnMatch                 string
 }
 
 type cmdProvider struct {
 	baseProvider
 	cmdConfig
-	command []string
+	command     []string
+	failOnMatch *regexp.Regexp
 }
 
 func newCmdProvider(c cmdConfig) (*cmdProvider, error) {
 	// TODO: check config options
+	if c.retry == 0 {
+		c.retry = defaultMaxRetry
+	}
 	provider := &cmdProvider{
 		baseProvider: baseProvider{
 			name:     c.name,
 			ctx:      NewContext(),
 			interval: c.interval,
+			retry:    c.retry,
 		},
 		cmdConfig: c,
 	}
@@ -41,6 +51,14 @@ func newCmdProvider(c cmdConfig) (*cmdProvider, error) {
 		return nil, err
 	}
 	provider.command = cmd
+	if len(c.failOnMatch) > 0 {
+		var err error
+		failOnMatch, err := regexp.Compile(c.failOnMatch)
+		if err != nil {
+			return nil, errors.New("fail-on-match regexp error: " + err.Error())
+		}
+		provider.failOnMatch = failOnMatch
+	}
 
 	return provider, nil
 }
@@ -57,7 +75,22 @@ func (p *cmdProvider) Run() error {
 	if err := p.Start(); err != nil {
 		return err
 	}
-	return p.Wait()
+	if err := p.Wait(); err != nil {
+		return err
+	}
+	if p.failOnMatch != nil {
+		if logContent, err := ioutil.ReadFile(p.LogFile()); err == nil {
+			matches := p.failOnMatch.FindAllSubmatch(logContent, -1)
+			if len(matches) != 0 {
+				logger.Debug("Fail-on-match: %r", matches)
+				return errors.New(
+					fmt.Sprintf("Fail-on-match regexp found %d matches", len(matches)))
+			}
+		} else {
+			return err
+		}
+	}
+	return nil
 }
 
 func (p *cmdProvider) Start() error {

worker/common.go

@@ -8,6 +8,6 @@ import (
 
 type empty struct{}
 
-const maxRetry = 2
+const defaultMaxRetry = 2
 
 var logger = logging.MustGetLogger("tunasync")

worker/config.go

@@ -33,14 +33,15 @@ func (p *providerEnum) UnmarshalText(text []byte) error {
 
 // Config represents worker config options
 type Config struct {
-	Global  globalConfig   `toml:"global"`
+	Global        globalConfig        `toml:"global"`
-	Manager managerConfig  `toml:"manager"`
+	Manager       managerConfig       `toml:"manager"`
-	Server  serverConfig   `toml:"server"`
+	Server        serverConfig        `toml:"server"`
-	Cgroup  cgroupConfig   `toml:"cgroup"`
+	Cgroup        cgroupConfig        `toml:"cgroup"`
-	ZFS     zfsConfig      `toml:"zfs"`
+	ZFS           zfsConfig           `toml:"zfs"`
-	Docker  dockerConfig   `toml:"docker"`
+	BtrfsSnapshot btrfsSnapshotConfig `toml:"btrfs_snapshot"`
-	Include includeConfig  `toml:"include"`
+	Docker        dockerConfig        `toml:"docker"`
-	Mirrors []mirrorConfig `toml:"mirrors"`
+	Include       includeConfig       `toml:"include"`
+	Mirrors       []mirrorConfig      `toml:"mirrors"`
 }
 
 type globalConfig struct {
@@ -49,6 +50,7 @@ type globalConfig struct {
 	MirrorDir  string `toml:"mirror_dir"`
 	Concurrent int    `toml:"concurrent"`
 	Interval   int    `toml:"interval"`
+	Retry      int    `toml:"retry"`
 
 	ExecOnSuccess []string `toml:"exec_on_success"`
 	ExecOnFailure []string `toml:"exec_on_failure"`
@@ -95,6 +97,11 @@ type zfsConfig struct {
 	Zpool  string `toml:"zpool"`
 }
 
+type btrfsSnapshotConfig struct {
+	Enable       bool   `toml:"enable"`
+	SnapshotPath string `toml:"snapshot_path"`
+}
+
 type includeConfig struct {
 	IncludeMirrors string `toml:"include_mirrors"`
 }
@@ -108,6 +115,7 @@ type mirrorConfig struct {
 	Provider  providerEnum      `toml:"provider"`
 	Upstream  string            `toml:"upstream"`
 	Interval  int               `toml:"interval"`
+	Retry     int               `toml:"retry"`
 	MirrorDir string            `toml:"mirror_dir"`
 	LogDir    string            `toml:"log_dir"`
 	Env       map[string]string `toml:"env"`
@@ -121,19 +129,24 @@ type mirrorConfig struct {
 	ExecOnSuccessExtra []string `toml:"exec_on_success_extra"`
 	ExecOnFailureExtra []string `toml:"exec_on_failure_extra"`
 
-	Command       string `toml:"command"`
+	Command       string   `toml:"command"`
-	UseIPv6       bool   `toml:"use_ipv6"`
+	FailOnMatch   string   `toml:"fail_on_match"`
-	UseIPv4       bool   `toml:"use_ipv4"`
+	UseIPv6       bool     `toml:"use_ipv6"`
-	ExcludeFile   string `toml:"exclude_file"`
+	UseIPv4       bool     `toml:"use_ipv4"`
-	Username      string `toml:"username"`
+	ExcludeFile   string   `toml:"exclude_file"`
-	Password      string `toml:"password"`
+	Username      string   `toml:"username"`
-	Stage1Profile string `toml:"stage1_profile"`
+	Password      string   `toml:"password"`
+	RsyncOptions  []string `toml:"rsync_options"`
+	RsyncOverride []string `toml:"rsync_override"`
+	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"`
+
+	SnapshotPath string `toml:"snapshot_path"`
 }
 
 // LoadConfig loads configuration

worker/config_test.go

@@ -18,6 +18,7 @@ log_dir = "/var/log/tunasync/{{.Name}}"
 mirror_dir = "/data/mirrors"
 concurrent = 10
 interval = 240
+retry = 3
 
 [manager]
 api_base = "https://127.0.0.1:5000"
@@ -35,6 +36,7 @@ name = "AOSP"
 provider = "command"
 upstream = "https://aosp.google.com/"
 interval = 720
+retry = 2
 mirror_dir = "/data/git/AOSP"
 exec_on_success = [
 	"bash -c 'echo ${TUNASYNC_JOB_EXIT_STATUS} > ${TUNASYNC_WORKING_DIR}/exit_status'"
@@ -116,6 +118,7 @@ use_ipv6 = true
 		So(err, ShouldBeNil)
 		So(cfg.Global.Name, ShouldEqual, "test_worker")
 		So(cfg.Global.Interval, ShouldEqual, 240)
+		So(cfg.Global.Retry, ShouldEqual, 3)
 		So(cfg.Global.MirrorDir, ShouldEqual, "/data/mirrors")
 
 		So(cfg.Manager.APIBase, ShouldEqual, "https://127.0.0.1:5000")
@@ -126,6 +129,7 @@ use_ipv6 = true
 		So(m.MirrorDir, ShouldEqual, "/data/git/AOSP")
 		So(m.Provider, ShouldEqual, provCommand)
 		So(m.Interval, ShouldEqual, 720)
+		So(m.Retry, ShouldEqual, 2)
 		So(m.Env["REPO"], ShouldEqual, "/usr/local/bin/aosp-repo")
 
 		m = cfg.Mirrors[1]

worker/docker.go

@@ -7,10 +7,9 @@ import (
 
 type dockerHook struct {
 	emptyHook
-	provider mirrorProvider
+	image   string
-	image    string
+	volumes []string
-	volumes  []string
+	options []string
-	options  []string
 }
 
 func newDockerHook(p mirrorProvider, gCfg dockerConfig, mCfg mirrorConfig) *dockerHook {
@@ -23,10 +22,12 @@ func newDockerHook(p mirrorProvider, gCfg dockerConfig, mCfg mirrorConfig) *dock
 	options = append(options, mCfg.DockerOptions...)
 
 	return &dockerHook{
-		provider: p,
+		emptyHook: emptyHook{
-		image:    mCfg.DockerImage,
+			provider: p,
-		volumes:  volumes,
+		},
-		options:  options,
+		image:   mCfg.DockerImage,
+		volumes: volumes,
+		options: options,
 	}
 }
 

worker/docker_test.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"io/ioutil"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"testing"
 	"time"
@@ -12,13 +13,27 @@ import (
 	. "github.com/smartystreets/goconvey/convey"
 )
 
+func cmdRun(p string, args []string) {
+	cmd := exec.Command(p, args...)
+
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		logger.Debugf("cmdRun failed %s", err)
+		return
+	}
+	logger.Debugf("cmdRun: ", string(out))
+}
+
 func getDockerByName(name string) (string, error) {
 	// docker ps -f 'name=$name' --format '{{.Names}}'
 	out, err := sh.Command(
-		"docker", "ps",
+		"docker", "ps", "-a",
 		"--filter", "name="+name,
 		"--format", "{{.Names}}",
 	).Output()
+	if err == nil {
+		logger.Debugf("docker ps: '%s'", string(out))
+	}
 	return string(out), err
 }
 
@@ -46,7 +61,7 @@ func TestDocker(t *testing.T) {
 
 		cmdScriptContent := `#!/bin/sh
 echo ${TEST_CONTENT}
-sleep 10
+sleep 20
 `
 		err = ioutil.WriteFile(cmdScript, []byte(cmdScriptContent), 0755)
 		So(err, ShouldBeNil)
@@ -55,8 +70,10 @@ sleep 10
 		So(err, ShouldBeNil)
 
 		d := &dockerHook{
-			provider: provider,
+			emptyHook: emptyHook{
-			image:    "alpine",
+				provider: provider,
+			},
+			image: "alpine:3.8",
 			volumes: []string{
 				fmt.Sprintf("%s:%s", cmdScript, "/bin/cmd.sh"),
 			},
@@ -67,20 +84,33 @@ sleep 10
 		err = d.preExec()
 		So(err, ShouldBeNil)
 
+		cmdRun("docker", []string{"images"})
+		exitedErr := make(chan error, 1)
 		go func() {
 			err = provider.Run()
-			ctx.So(err, ShouldNotBeNil)
+			logger.Debugf("provider.Run() exited")
+			if err != nil {
+				logger.Errorf("provider.Run() failed: %v", err)
+			}
+			exitedErr <- err
 		}()
+		cmdRun("ps", []string{"aux"})
 
-		time.Sleep(1 * time.Second)
+		// Wait for docker running
+		time.Sleep(8 * time.Second)
+
+		cmdRun("ps", []string{"aux"})
 
 		// assert container running
 		names, err := getDockerByName(d.Name())
 		So(err, ShouldBeNil)
-		So(names, ShouldEqual, d.Name()+"\n")
+		// So(names, ShouldEqual, d.Name()+"\n")
 
 		err = provider.Terminate()
-		So(err, ShouldBeNil)
+		// So(err, ShouldBeNil)
+
+		cmdRun("ps", []string{"aux"})
+		<-exitedErr
 
 		// container should be terminated and removed
 		names, err = getDockerByName(d.Name())

worker/exec_post_hook.go

@@ -18,7 +18,6 @@ const (
 
 type execPostHook struct {
 	emptyHook
-	provider mirrorProvider
 
 	// exec on success or on failure
 	execOn uint8
@@ -37,9 +36,11 @@ func newExecPostHook(provider mirrorProvider, execOn uint8, command string) (*ex
 	}
 
 	return &execPostHook{
-		provider: provider,
+		emptyHook: emptyHook{
-		execOn:   execOn,
+			provider: provider,
-		command:  cmd,
+		},
+		execOn:  execOn,
+		command: cmd,
 	}, nil
 }
 

worker/exec_post_test.go

@@ -92,7 +92,7 @@ exit 1
 			job.ctrlChan <- jobStart
 			msg := <-managerChan
 			So(msg.status, ShouldEqual, PreSyncing)
-			for i := 0; i < maxRetry; i++ {
+			for i := 0; i < defaultMaxRetry; i++ {
 				msg = <-managerChan
 				So(msg.status, ShouldEqual, Syncing)
 				msg = <-managerChan

worker/job.go

@@ -53,6 +53,7 @@ type mirrorJob struct {
 	ctrlChan chan ctrlAction
 	disabled chan empty
 	state    uint32
+	size     string
 }
 
 func newMirrorJob(provider mirrorProvider) *mirrorJob {
@@ -112,7 +113,7 @@ func (m *mirrorJob) Run(managerChan chan<- jobMessage, semaphore chan empty) err
 				managerChan <- jobMessage{
 					tunasync.Failed, m.Name(),
 					fmt.Sprintf("error exec hook %s: %s", hookname, err.Error()),
-					false,
+					true,
 				}
 				return err
 			}
@@ -138,7 +139,7 @@ func (m *mirrorJob) Run(managerChan chan<- jobMessage, semaphore chan empty) err
 			return err
 		}
 
-		for retry := 0; retry < maxRetry; retry++ {
+		for retry := 0; retry < provider.Retry(); retry++ {
 			stopASAP := false // stop job as soon as possible
 
 			if retry > 0 {
@@ -182,26 +183,33 @@ func (m *mirrorJob) Run(managerChan chan<- jobMessage, semaphore chan empty) err
 			if syncErr == nil {
 				// syncing success
 				logger.Noticef("succeeded syncing %s", m.Name())
-				managerChan <- jobMessage{tunasync.Success, m.Name(), "", (m.State() == stateReady)}
 				// post-success hooks
+				logger.Debug("post-success hooks")
 				err := runHooks(rHooks, func(h jobHook) error { return h.postSuccess() }, "post-success")
 				if err != nil {
 					return err
 				}
-				return nil
+			} else {
+				// syncing failed
+				logger.Warningf("failed syncing %s: %s", m.Name(), syncErr.Error())
+				// post-fail hooks
+				logger.Debug("post-fail hooks")
+				err := runHooks(rHooks, func(h jobHook) error { return h.postFail() }, "post-fail")
+				if err != nil {
+					return err
+				}
+			}
 
+			if syncErr == nil {
+				// syncing success
+				m.size = provider.DataSize()
+				managerChan <- jobMessage{tunasync.Success, m.Name(), "", (m.State() == stateReady)}
+				return nil
 			}
 
 			// syncing failed
-			logger.Warningf("failed syncing %s: %s", m.Name(), syncErr.Error())
+			managerChan <- jobMessage{tunasync.Failed, m.Name(), syncErr.Error(), (retry == provider.Retry()-1) && (m.State() == stateReady)}
-			managerChan <- jobMessage{tunasync.Failed, m.Name(), syncErr.Error(), (retry == maxRetry-1) && (m.State() == stateReady)}
 
-			// post-fail hooks
-			logger.Debug("post-fail hooks")
-			err = runHooks(rHooks, func(h jobHook) error { return h.postFail() }, "post-fail")
-			if err != nil {
-				return err
-			}
 			// gracefully exit
 			if stopASAP {
 				logger.Debug("No retry, exit directly")

worker/job_test.go

@@ -112,6 +112,74 @@ func TestMirrorJob(t *testing.T) {
 
 		})
 
+		Convey("When running long jobs with post-fail hook", func(ctx C) {
+			scriptContent := `#!/bin/bash
+echo '++++++'
+echo $TUNASYNC_WORKING_DIR
+echo $0 sleeping
+sleep 3
+echo $TUNASYNC_WORKING_DIR
+echo '------'
+			`
+			err = ioutil.WriteFile(scriptFile, []byte(scriptContent), 0755)
+			So(err, ShouldBeNil)
+
+			hookScriptFile := filepath.Join(tmpDir, "hook.sh")
+			err = ioutil.WriteFile(hookScriptFile, []byte(scriptContent), 0755)
+			So(err, ShouldBeNil)
+
+			h, err := newExecPostHook(provider, execOnFailure, hookScriptFile)
+			So(err, ShouldBeNil)
+			provider.AddHook(h)
+
+			managerChan := make(chan jobMessage, 10)
+			semaphore := make(chan empty, 1)
+			job := newMirrorJob(provider)
+
+			Convey("If we kill it", func(ctx C) {
+				go job.Run(managerChan, semaphore)
+				job.ctrlChan <- jobStart
+
+				time.Sleep(1 * time.Second)
+				msg := <-managerChan
+				So(msg.status, ShouldEqual, PreSyncing)
+				msg = <-managerChan
+				So(msg.status, ShouldEqual, Syncing)
+
+				job.ctrlChan <- jobStop
+
+				msg = <-managerChan
+				So(msg.status, ShouldEqual, Failed)
+
+				job.ctrlChan <- jobDisable
+				<-job.disabled
+			})
+
+			Convey("If we kill it then start it", func(ctx C) {
+				go job.Run(managerChan, semaphore)
+				job.ctrlChan <- jobStart
+
+				time.Sleep(1 * time.Second)
+				msg := <-managerChan
+				So(msg.status, ShouldEqual, PreSyncing)
+				msg = <-managerChan
+				So(msg.status, ShouldEqual, Syncing)
+
+				job.ctrlChan <- jobStop
+
+				time.Sleep(2 * time.Second)
+				logger.Debugf("Now starting...\n")
+				job.ctrlChan <- jobStart
+
+				msg = <-managerChan
+				So(msg.status, ShouldEqual, Failed)
+
+				job.ctrlChan <- jobDisable
+				<-job.disabled
+			})
+
+		})
+
 		Convey("When running long jobs", func(ctx C) {
 			scriptContent := `#!/bin/bash
 echo $TUNASYNC_WORKING_DIR

worker/loglimit_hook.go

@@ -14,12 +14,13 @@ import (
 
 type logLimiter struct {
 	emptyHook
-	provider mirrorProvider
 }
 
 func newLogLimiter(provider mirrorProvider) *logLimiter {
 	return &logLimiter{
-		provider: provider,
+		emptyHook: emptyHook{
+			provider: provider,
+		},
 	}
 }
 

worker/provider.go

@@ -45,11 +45,13 @@ type mirrorProvider interface {
 	Hooks() []jobHook
 
 	Interval() time.Duration
+	Retry() int
 
 	WorkingDir() string
 	LogDir() string
 	LogFile() string
 	IsMaster() bool
+	DataSize() string
 
 	// enter context
 	EnterContext() *Context
@@ -86,6 +88,9 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 	if mirror.Interval == 0 {
 		mirror.Interval = cfg.Global.Interval
 	}
+	if mirror.Retry == 0 {
+		mirror.Retry = cfg.Global.Retry
+	}
 	logDir = formatLogDir(logDir, mirror)
 
 	// IsMaster
@@ -107,37 +112,43 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			upstreamURL: mirror.Upstream,
 			command:     mirror.Command,
 			workingDir:  mirrorDir,
+			failOnMatch: mirror.FailOnMatch,
 			logDir:      logDir,
 			logFile:     filepath.Join(logDir, "latest.log"),
 			interval:    time.Duration(mirror.Interval) * time.Minute,
+			retry:       mirror.Retry,
 			env:         mirror.Env,
 		}
 		p, err := newCmdProvider(pc)
-		p.isMaster = isMaster
 		if err != nil {
 			panic(err)
 		}
+		p.isMaster = isMaster
 		provider = p
 	case provRsync:
 		rc := rsyncConfig{
-			name:        mirror.Name,
+			name:              mirror.Name,
-			upstreamURL: mirror.Upstream,
+			upstreamURL:       mirror.Upstream,
-			rsyncCmd:    mirror.Command,
+			rsyncCmd:          mirror.Command,
-			username:    mirror.Username,
+			username:          mirror.Username,
-			password:    mirror.Password,
+			password:          mirror.Password,
-			excludeFile: mirror.ExcludeFile,
+			excludeFile:       mirror.ExcludeFile,
-			workingDir:  mirrorDir,
+			extraOptions:      mirror.RsyncOptions,
-			logDir:      logDir,
+			overriddenOptions: mirror.RsyncOverride,
-			logFile:     filepath.Join(logDir, "latest.log"),
+			rsyncEnv:          mirror.Env,
-			useIPv6:     mirror.UseIPv6,
+			workingDir:        mirrorDir,
-			useIPv4:     mirror.UseIPv4,
+			logDir:            logDir,
-			interval:    time.Duration(mirror.Interval) * time.Minute,
+			logFile:           filepath.Join(logDir, "latest.log"),
+			useIPv6:           mirror.UseIPv6,
+			useIPv4:           mirror.UseIPv4,
+			interval:          time.Duration(mirror.Interval) * time.Minute,
+			retry:             mirror.Retry,
 		}
 		p, err := newRsyncProvider(rc)
-		p.isMaster = isMaster
 		if err != nil {
 			panic(err)
 		}
+		p.isMaster = isMaster
 		provider = p
 	case provTwoStageRsync:
 		rc := twoStageRsyncConfig{
@@ -148,17 +159,20 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 			username:      mirror.Username,
 			password:      mirror.Password,
 			excludeFile:   mirror.ExcludeFile,
+			extraOptions:  mirror.RsyncOptions,
+			rsyncEnv:      mirror.Env,
 			workingDir:    mirrorDir,
 			logDir:        logDir,
 			logFile:       filepath.Join(logDir, "latest.log"),
 			useIPv6:       mirror.UseIPv6,
 			interval:      time.Duration(mirror.Interval) * time.Minute,
+			retry:         mirror.Retry,
 		}
 		p, err := newTwoStageRsyncProvider(rc)
-		p.isMaster = isMaster
 		if err != nil {
 			panic(err)
 		}
+		p.isMaster = isMaster
 		provider = p
 	default:
 		panic(errors.New("Invalid mirror provider"))
@@ -172,6 +186,11 @@ func newMirrorProvider(mirror mirrorConfig, cfg *Config) mirrorProvider {
 		provider.AddHook(newZfsHook(provider, cfg.ZFS.Zpool))
 	}
 
+	// Add Btrfs Snapshot Hook
+	if cfg.BtrfsSnapshot.Enable {
+		provider.AddHook(newBtrfsSnapshotHook(provider, cfg.BtrfsSnapshot.SnapshotPath, mirror))
+	}
+
 	// Add Docker Hook
 	if cfg.Docker.Enable && len(mirror.DockerImage) > 0 {
 		provider.AddHook(newDockerHook(provider, cfg.Docker, mirror))

worker/provider_test.go

@@ -73,6 +73,7 @@ func TestRsyncProvider(t *testing.T) {
 echo "syncing to $(pwd)"
 echo $RSYNC_PASSWORD $@
 sleep 1
+echo "Total file size: 1.33T bytes"
 echo "Done"
 exit 0
 			`
@@ -83,6 +84,7 @@ exit 0
 			expectedOutput := fmt.Sprintf(
 				"syncing to %s\n"+
 					"%s\n"+
+					"Total file size: 1.33T bytes\n"+
 					"Done\n",
 				targetDir,
 				fmt.Sprintf(
@@ -99,6 +101,7 @@ exit 0
 			So(err, ShouldBeNil)
 			So(string(loggedContent), ShouldEqual, expectedOutput)
 			// fmt.Println(string(loggedContent))
+			So(provider.DataSize(), ShouldEqual, "1.33T")
 		})
 
 	})
@@ -111,18 +114,21 @@ func TestRsyncProviderWithAuthentication(t *testing.T) {
 		So(err, ShouldBeNil)
 		scriptFile := filepath.Join(tmpDir, "myrsync")
 		tmpFile := filepath.Join(tmpDir, "log_file")
+		proxyAddr := "127.0.0.1:1233"
 
 		c := rsyncConfig{
-			name:        "tuna",
+			name:         "tuna",
-			upstreamURL: "rsync://rsync.tuna.moe/tuna/",
+			upstreamURL:  "rsync://rsync.tuna.moe/tuna/",
-			rsyncCmd:    scriptFile,
+			rsyncCmd:     scriptFile,
-			username:    "tunasync",
+			username:     "tunasync",
-			password:    "tunasyncpassword",
+			password:     "tunasyncpassword",
-			workingDir:  tmpDir,
+			workingDir:   tmpDir,
-			logDir:      tmpDir,
+			extraOptions: []string{"--delete-excluded"},
-			logFile:     tmpFile,
+			rsyncEnv:     map[string]string{"RSYNC_PROXY": proxyAddr},
-			useIPv6:     true,
+			logDir:       tmpDir,
-			interval:    600 * time.Second,
+			logFile:      tmpFile,
+			useIPv4:      true,
+			interval:     600 * time.Second,
 		}
 
 		provider, err := newRsyncProvider(c)
@@ -137,7 +143,7 @@ func TestRsyncProviderWithAuthentication(t *testing.T) {
 		Convey("Let's try a run", func() {
 			scriptContent := `#!/bin/bash
 echo "syncing to $(pwd)"
-echo $USER $RSYNC_PASSWORD $@
+echo $USER $RSYNC_PASSWORD $RSYNC_PROXY $@
 sleep 1
 echo "Done"
 exit 0
@@ -152,10 +158,11 @@ exit 0
 					"Done\n",
 				targetDir,
 				fmt.Sprintf(
-					"%s %s -aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
+					"%s %s %s -aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
-						"--timeout=120 --contimeout=120 -6 %s %s",
+						"--timeout=120 --contimeout=120 -4 --delete-excluded %s %s",
-					provider.username, provider.password, provider.upstreamURL, provider.WorkingDir(),
+					provider.username, provider.password, proxyAddr,
+					provider.upstreamURL, provider.WorkingDir(),
 				),
 			)
 
@@ -170,6 +177,68 @@ exit 0
 	})
 }
 
+func TestRsyncProviderWithOverriddenOptions(t *testing.T) {
+	Convey("Rsync Provider with overridden options should work", t, func() {
+		tmpDir, err := ioutil.TempDir("", "tunasync")
+		defer os.RemoveAll(tmpDir)
+		So(err, ShouldBeNil)
+		scriptFile := filepath.Join(tmpDir, "myrsync")
+		tmpFile := filepath.Join(tmpDir, "log_file")
+
+		c := rsyncConfig{
+			name:              "tuna",
+			upstreamURL:       "rsync://rsync.tuna.moe/tuna/",
+			rsyncCmd:          scriptFile,
+			workingDir:        tmpDir,
+			overriddenOptions: []string{"-aHvh", "--no-o", "--no-g", "--stats"},
+			extraOptions:      []string{"--delete-excluded"},
+			logDir:            tmpDir,
+			logFile:           tmpFile,
+			useIPv6:           true,
+			interval:          600 * time.Second,
+		}
+
+		provider, err := newRsyncProvider(c)
+		So(err, ShouldBeNil)
+
+		So(provider.Name(), ShouldEqual, c.name)
+		So(provider.WorkingDir(), ShouldEqual, c.workingDir)
+		So(provider.LogDir(), ShouldEqual, c.logDir)
+		So(provider.LogFile(), ShouldEqual, c.logFile)
+		So(provider.Interval(), ShouldEqual, c.interval)
+
+		Convey("Let's try a run", func() {
+			scriptContent := `#!/bin/bash
+echo "syncing to $(pwd)"
+echo $@
+sleep 1
+echo "Done"
+exit 0
+			`
+			err = ioutil.WriteFile(scriptFile, []byte(scriptContent), 0755)
+			So(err, ShouldBeNil)
+
+			targetDir, _ := filepath.EvalSymlinks(provider.WorkingDir())
+			expectedOutput := fmt.Sprintf(
+				"syncing to %s\n"+
+					"-aHvh --no-o --no-g --stats -6 --delete-excluded %s %s\n"+
+					"Done\n",
+				targetDir,
+				provider.upstreamURL,
+				provider.WorkingDir(),
+			)
+
+			err = provider.Run()
+			So(err, ShouldBeNil)
+			loggedContent, err := ioutil.ReadFile(provider.LogFile())
+			So(err, ShouldBeNil)
+			So(string(loggedContent), ShouldEqual, expectedOutput)
+			// fmt.Println(string(loggedContent))
+		})
+
+	})
+}
+
 func TestCmdProvider(t *testing.T) {
 	Convey("Command Provider should work", t, func(ctx C) {
 		tmpDir, err := ioutil.TempDir("", "tunasync")
@@ -296,6 +365,41 @@ sleep 5
 
 		})
 	})
+	Convey("Command Provider with fail-on-match regexp should work", t, func(ctx C) {
+		tmpDir, err := ioutil.TempDir("", "tunasync")
+		defer os.RemoveAll(tmpDir)
+		So(err, ShouldBeNil)
+		tmpFile := filepath.Join(tmpDir, "log_file")
+
+		c := cmdConfig{
+			name:        "run-uptime",
+			upstreamURL: "http://mirrors.tuna.moe/",
+			command:     "uptime",
+			failOnMatch: "",
+			workingDir:  tmpDir,
+			logDir:      tmpDir,
+			logFile:     tmpFile,
+			interval:    600 * time.Second,
+		}
+
+		Convey("when regexp matches", func() {
+			c.failOnMatch = `[a-z]+`
+			provider, err := newCmdProvider(c)
+			So(err, ShouldBeNil)
+
+			err = provider.Run()
+			So(err, ShouldNotBeNil)
+		})
+
+		Convey("when regexp does not match", func() {
+			c.failOnMatch = `load average_`
+			provider, err := newCmdProvider(c)
+			So(err, ShouldBeNil)
+
+			err = provider.Run()
+			So(err, ShouldBeNil)
+		})
+	})
 }
 
 func TestTwoStageRsyncProvider(t *testing.T) {
@@ -316,6 +420,7 @@ func TestTwoStageRsyncProvider(t *testing.T) {
 			logFile:       tmpFile,
 			useIPv6:       true,
 			excludeFile:   tmpFile,
+			extraOptions:  []string{"--delete-excluded", "--cache"},
 			username:      "hello",
 			password:      "world",
 		}
@@ -363,7 +468,7 @@ exit 0
 				fmt.Sprintf(
 					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
-						"--timeout=120 --contimeout=120 -6 --exclude-from %s %s %s",
+						"--timeout=120 --contimeout=120 --delete-excluded --cache -6 --exclude-from %s %s %s",
 					provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 				),
 			)

worker/rsync_provider.go

@@ -2,24 +2,32 @@ package worker
 
 import (
 	"errors"
+	"io/ioutil"
 	"strings"
 	"time"
+
+	"github.com/tuna/tunasync/internal"
 )
 
 type rsyncConfig struct {
 	name                                         string
 	rsyncCmd                                     string
 	upstreamURL, username, password, excludeFile string
+	extraOptions                                 []string
+	overriddenOptions                            []string
+	rsyncEnv                                     map[string]string
 	workingDir, logDir, logFile                  string
 	useIPv6, useIPv4                             bool
 	interval                                     time.Duration
+	retry                                        int
 }
 
 // An RsyncProvider provides the implementation to rsync-based syncing jobs
 type rsyncProvider struct {
 	baseProvider
 	rsyncConfig
-	options []string
+	options  []string
+	dataSize string
 }
 
 func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
@@ -27,11 +35,15 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 	if !strings.HasSuffix(c.upstreamURL, "/") {
 		return nil, errors.New("rsync upstream URL should ends with /")
 	}
+	if c.retry == 0 {
+		c.retry = defaultMaxRetry
+	}
 	provider := &rsyncProvider{
 		baseProvider: baseProvider{
 			name:     c.name,
 			ctx:      NewContext(),
 			interval: c.interval,
+			retry:    c.retry,
 		},
 		rsyncConfig: c,
 	}
@@ -39,6 +51,15 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 	if c.rsyncCmd == "" {
 		provider.rsyncCmd = "rsync"
 	}
+	if c.rsyncEnv == nil {
+		provider.rsyncEnv = map[string]string{}
+	}
+	if c.username != "" {
+		provider.rsyncEnv["USER"] = c.username
+	}
+	if c.password != "" {
+		provider.rsyncEnv["RSYNC_PASSWORD"] = c.password
+	}
 
 	options := []string{
 		"-aHvh", "--no-o", "--no-g", "--stats",
@@ -46,6 +67,9 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 		"--delete", "--delete-after", "--delay-updates",
 		"--safe-links", "--timeout=120", "--contimeout=120",
 	}
+	if c.overriddenOptions != nil {
+		options = c.overriddenOptions
+	}
 
 	if c.useIPv6 {
 		options = append(options, "-6")
@@ -56,6 +80,9 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 	if c.excludeFile != "" {
 		options = append(options, "--exclude-from", c.excludeFile)
 	}
+	if c.extraOptions != nil {
+		options = append(options, c.extraOptions...)
+	}
 	provider.options = options
 
 	provider.ctx.Set(_WorkingDirKey, c.workingDir)
@@ -73,11 +100,22 @@ func (p *rsyncProvider) Upstream() string {
 	return p.upstreamURL
 }
 
+func (p *rsyncProvider) DataSize() string {
+	return p.dataSize
+}
+
 func (p *rsyncProvider) Run() error {
+	p.dataSize = ""
 	if err := p.Start(); err != nil {
 		return err
 	}
-	return p.Wait()
+	if err := p.Wait(); err != nil {
+		return err
+	}
+	if logContent, err := ioutil.ReadFile(p.LogFile()); err == nil {
+		p.dataSize = internal.ExtractSizeFromRsyncLog(logContent)
+	}
+	return nil
 }
 
 func (p *rsyncProvider) Start() error {
@@ -88,18 +126,11 @@ func (p *rsyncProvider) Start() error {
 		return errors.New("provider is currently running")
 	}
 
-	env := map[string]string{}
-	if p.username != "" {
-		env["USER"] = p.username
-	}
-	if p.password != "" {
-		env["RSYNC_PASSWORD"] = p.password
-	}
 	command := []string{p.rsyncCmd}
 	command = append(command, p.options...)
 	command = append(command, p.upstreamURL, p.WorkingDir())
 
-	p.cmd = newCmdJob(p, command, p.WorkingDir(), env)
+	p.cmd = newCmdJob(p, command, p.WorkingDir(), p.rsyncEnv)
 	if err := p.prepareLogFile(false); err != nil {
 		return err
 	}

worker/runner.go

@@ -104,7 +104,7 @@ func newCmdJob(provider mirrorProvider, cmdAndArgs []string, workingDir string,
 }
 
 func (c *cmdJob) Start() error {
-	// logger.Debugf("Command start: %v", c.cmd.Args)
+	logger.Debugf("Command start: %v", c.cmd.Args)
 	c.finished = make(chan empty, 1)
 	return c.cmd.Start()
 }

worker/schedule.go

@@ -15,6 +15,11 @@ type scheduleQueue struct {
 	jobs map[string]bool
 }
 
+type jobScheduleInfo struct {
+	jobName       string
+	nextScheduled time.Time
+}
+
 func timeLessThan(l, r interface{}) bool {
 	tl := l.(time.Time)
 	tr := r.(time.Time)
@@ -28,6 +33,20 @@ func newScheduleQueue() *scheduleQueue {
 	return queue
 }
 
+func (q *scheduleQueue) GetJobs() (jobs []jobScheduleInfo) {
+	cur := q.list.Iterator()
+	defer cur.Close()
+
+	for cur.Next() {
+		cj := cur.Value().(*mirrorJob)
+		jobs = append(jobs, jobScheduleInfo{
+			cj.Name(),
+			cur.Key().(time.Time),
+		})
+	}
+	return
+}
+
 func (q *scheduleQueue) AddJob(schedTime time.Time, job *mirrorJob) {
 	q.Lock()
 	defer q.Unlock()

worker/two_stage_rsync_provider.go

@@ -3,8 +3,11 @@ package worker
 import (
 	"errors"
 	"fmt"
+	"io/ioutil"
 	"strings"
 	"time"
+
+	"github.com/tuna/tunasync/internal"
 )
 
 type twoStageRsyncConfig struct {
@@ -12,9 +15,12 @@ type twoStageRsyncConfig struct {
 	rsyncCmd                                     string
 	stage1Profile                                string
 	upstreamURL, username, password, excludeFile string
+	extraOptions                                 []string
+	rsyncEnv                                     map[string]string
 	workingDir, logDir, logFile                  string
 	useIPv6                                      bool
 	interval                                     time.Duration
+	retry                                        int
 }
 
 // An RsyncProvider provides the implementation to rsync-based syncing jobs
@@ -23,6 +29,7 @@ type twoStageRsyncProvider struct {
 	twoStageRsyncConfig
 	stage1Options []string
 	stage2Options []string
+	dataSize      string
 }
 
 var rsyncStage1Profiles = map[string]([]string){
@@ -38,12 +45,16 @@ func newTwoStageRsyncProvider(c twoStageRsyncConfig) (*twoStageRsyncProvider, er
 	if !strings.HasSuffix(c.upstreamURL, "/") {
 		return nil, errors.New("rsync upstream URL should ends with /")
 	}
+	if c.retry == 0 {
+		c.retry = defaultMaxRetry
+	}
 
 	provider := &twoStageRsyncProvider{
 		baseProvider: baseProvider{
 			name:     c.name,
 			ctx:      NewContext(),
 			interval: c.interval,
+			retry:    c.retry,
 		},
 		twoStageRsyncConfig: c,
 		stage1Options: []string{
@@ -59,6 +70,15 @@ func newTwoStageRsyncProvider(c twoStageRsyncConfig) (*twoStageRsyncProvider, er
 		},
 	}
 
+	if c.rsyncEnv == nil {
+		provider.rsyncEnv = map[string]string{}
+	}
+	if c.username != "" {
+		provider.rsyncEnv["USER"] = c.username
+	}
+	if c.password != "" {
+		provider.rsyncEnv["RSYNC_PASSWORD"] = c.password
+	}
 	if c.rsyncCmd == "" {
 		provider.rsyncCmd = "rsync"
 	}
@@ -78,6 +98,10 @@ func (p *twoStageRsyncProvider) Upstream() string {
 	return p.upstreamURL
 }
 
+func (p *twoStageRsyncProvider) DataSize() string {
+	return p.dataSize
+}
+
 func (p *twoStageRsyncProvider) Options(stage int) ([]string, error) {
 	var options []string
 	if stage == 1 {
@@ -92,6 +116,9 @@ func (p *twoStageRsyncProvider) Options(stage int) ([]string, error) {
 
 	} else if stage == 2 {
 		options = append(options, p.stage2Options...)
+		if p.extraOptions != nil {
+			options = append(options, p.extraOptions...)
+		}
 	} else {
 		return []string{}, fmt.Errorf("Invalid stage: %d", stage)
 	}
@@ -115,14 +142,7 @@ func (p *twoStageRsyncProvider) Run() error {
 		return errors.New("provider is currently running")
 	}
 
-	env := map[string]string{}
+	p.dataSize = ""
-	if p.username != "" {
-		env["USER"] = p.username
-	}
-	if p.password != "" {
-		env["RSYNC_PASSWORD"] = p.password
-	}
-
 	stages := []int{1, 2}
 	for _, stage := range stages {
 		command := []string{p.rsyncCmd}
@@ -133,7 +153,7 @@ func (p *twoStageRsyncProvider) Run() error {
 		command = append(command, options...)
 		command = append(command, p.upstreamURL, p.WorkingDir())
 
-		p.cmd = newCmdJob(p, command, p.WorkingDir(), env)
+		p.cmd = newCmdJob(p, command, p.WorkingDir(), p.rsyncEnv)
 		if err := p.prepareLogFile(stage > 1); err != nil {
 			return err
 		}
@@ -151,5 +171,8 @@ func (p *twoStageRsyncProvider) Run() error {
 			return err
 		}
 	}
+	if logContent, err := ioutil.ReadFile(p.LogFile()); err == nil {
+		p.dataSize = internal.ExtractSizeFromRsyncLog(logContent)
+	}
 	return nil
 }

worker/worker.go

@@ -12,8 +12,6 @@ import (
 	. "github.com/tuna/tunasync/internal"
 )
 
-var tunasyncWorker *Worker
-
 // A Worker is a instance of tunasync worker
 type Worker struct {
 	L    sync.Mutex
@@ -29,10 +27,11 @@ type Worker struct {
 	httpClient *http.Client
 }
 
-// GetTUNASyncWorker returns a singalton worker
+// NewTUNASyncWorker creates a worker
-func GetTUNASyncWorker(cfg *Config) *Worker {
+func NewTUNASyncWorker(cfg *Config) *Worker {
-	if tunasyncWorker != nil {
+
-		return tunasyncWorker
+	if cfg.Global.Retry == 0 {
+		cfg.Global.Retry = defaultMaxRetry
 	}
 
 	w := &Worker{
@@ -57,7 +56,6 @@ func GetTUNASyncWorker(cfg *Config) *Worker {
 
 	w.initJobs()
 	w.makeHTTPServer()
-	tunasyncWorker = w
 	return w
 }
 
@@ -310,6 +308,9 @@ func (w *Worker) runSchedule() {
 
 	w.L.Unlock()
 
+	schedInfo := w.schedule.GetJobs()
+	w.updateSchedInfo(schedInfo)
+
 	tick := time.Tick(5 * time.Second)
 	for {
 		select {
@@ -346,6 +347,9 @@ func (w *Worker) runSchedule() {
 				w.schedule.AddJob(schedTime, job)
 			}
 
+			schedInfo = w.schedule.GetJobs()
+			w.updateSchedInfo(schedInfo)
+
 		case <-tick:
 			// check schedule every 5 seconds
 			if job := w.schedule.Pop(); job != nil {
@@ -416,6 +420,12 @@ func (w *Worker) updateStatus(job *mirrorJob, jobMsg jobMessage) {
 		ErrorMsg: jobMsg.msg,
 	}
 
+	// Certain Providers (rsync for example) may know the size of mirror,
+	// so we report it to Manager here
+	if len(job.size) != 0 {
+		smsg.Size = job.size
+	}
+
 	for _, root := range w.cfg.Manager.APIBaseList() {
 		url := fmt.Sprintf(
 			"%s/workers/%s/jobs/%s", root, w.Name(), jobMsg.name,
@@ -427,6 +437,27 @@ func (w *Worker) updateStatus(job *mirrorJob, jobMsg jobMessage) {
 	}
 }
 
+func (w *Worker) updateSchedInfo(schedInfo []jobScheduleInfo) {
+	var s []MirrorSchedule
+	for _, sched := range schedInfo {
+		s = append(s, MirrorSchedule{
+			MirrorName:   sched.jobName,
+			NextSchedule: sched.nextScheduled,
+		})
+	}
+	msg := MirrorSchedules{Schedules: s}
+
+	for _, root := range w.cfg.Manager.APIBaseList() {
+		url := fmt.Sprintf(
+			"%s/workers/%s/schedules", root, w.Name(),
+		)
+		logger.Debugf("reporting on manager url: %s", url)
+		if _, err := PostJSON(url, msg, w.httpClient); err != nil {
+			logger.Errorf("Failed to upload schedules: %s", err.Error())
+		}
+	}
+}
+
 func (w *Worker) fetchJobStatus() []MirrorStatus {
 	var mirrorList []MirrorStatus
 	apiBase := w.cfg.Manager.APIBaseList()[0]

worker/worker_test.go

@@ -0,0 +1,255 @@
+package worker
+
+import (
+	"net/http"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	. "github.com/smartystreets/goconvey/convey"
+	. "github.com/tuna/tunasync/internal"
+)
+
+type workTestFunc func(*Worker)
+
+var managerPort = 5001
+var workerPort = 5002
+
+func makeMockManagerServer(recvData chan interface{}) *gin.Engine {
+	r := gin.Default()
+	r.GET("/ping", func(c *gin.Context) {
+		c.JSON(http.StatusOK, gin.H{"_infoKey": "pong"})
+	})
+	r.POST("/workers", func(c *gin.Context) {
+		var _worker WorkerStatus
+		c.BindJSON(&_worker)
+		_worker.LastOnline = time.Now()
+		recvData <- _worker
+		c.JSON(http.StatusOK, _worker)
+	})
+	r.POST("/workers/dut/schedules", func(c *gin.Context) {
+		var _sch MirrorSchedules
+		c.BindJSON(&_sch)
+		recvData <- _sch
+		c.JSON(http.StatusOK, empty{})
+	})
+	r.POST("/workers/dut/jobs/:job", func(c *gin.Context) {
+		var status MirrorStatus
+		c.BindJSON(&status)
+		recvData <- status
+		c.JSON(http.StatusOK, status)
+	})
+	r.GET("/workers/dut/jobs", func(c *gin.Context) {
+		mirrorStatusList := []MirrorStatus{}
+		c.JSON(http.StatusOK, mirrorStatusList)
+	})
+
+	return r
+}
+
+func startWorkerThenStop(cfg *Config, tester workTestFunc) {
+	exitedChan := make(chan int)
+	w := NewTUNASyncWorker(cfg)
+	So(w, ShouldNotBeNil)
+	go func() {
+		w.Run()
+		exitedChan <- 1
+	}()
+
+	tester(w)
+
+	w.Halt()
+	select {
+	case exited := <-exitedChan:
+		So(exited, ShouldEqual, 1)
+	case <-time.After(2 * time.Second):
+		So(0, ShouldEqual, 1)
+	}
+
+}
+func sendCommandToWorker(workerURL string, httpClient *http.Client, cmd CmdVerb, mirror string) {
+	workerCmd := WorkerCmd{
+		Cmd:      cmd,
+		MirrorID: mirror,
+	}
+	logger.Debugf("POST to %s with cmd %s", workerURL, cmd)
+	_, err := PostJSON(workerURL, workerCmd, httpClient)
+	So(err, ShouldBeNil)
+}
+
+func TestWorker(t *testing.T) {
+	InitLogger(false, true, false)
+
+	recvDataChan := make(chan interface{})
+	_s := makeMockManagerServer(recvDataChan)
+	httpServer := &http.Server{
+		Addr:         "localhost:" + strconv.Itoa(managerPort),
+		Handler:      _s,
+		ReadTimeout:  2 * time.Second,
+		WriteTimeout: 2 * time.Second,
+	}
+	go func() {
+		err := httpServer.ListenAndServe()
+		So(err, ShouldBeNil)
+	}()
+	// Wait for http server starting
+	time.Sleep(500 * time.Millisecond)
+
+	Convey("Worker should work", t, func(ctx C) {
+
+		httpClient, err := CreateHTTPClient("")
+		So(err, ShouldBeNil)
+
+		workerPort++
+
+		workerCfg := Config{
+			Global: globalConfig{
+				Name:       "dut",
+				LogDir:     "/tmp",
+				MirrorDir:  "/tmp",
+				Concurrent: 2,
+				Interval:   1,
+			},
+			Server: serverConfig{
+				Hostname: "localhost",
+				Addr:     "127.0.0.1",
+				Port:     workerPort,
+			},
+			Manager: managerConfig{
+				APIBase: "http://localhost:" + strconv.Itoa(managerPort),
+			},
+		}
+		logger.Debugf("worker port %d", workerPort)
+		Convey("with no job", func(ctx C) {
+			dummyTester := func(*Worker) {
+				registered := false
+				for {
+					select {
+					case data := <-recvDataChan:
+						if reg, ok := data.(WorkerStatus); ok {
+							So(reg.ID, ShouldEqual, "dut")
+							registered = true
+							time.Sleep(500 * time.Millisecond)
+							sendCommandToWorker(reg.URL, httpClient, CmdStart, "foobar")
+						} else if sch, ok := data.(MirrorSchedules); ok {
+							So(len(sch.Schedules), ShouldEqual, 0)
+						}
+					case <-time.After(2 * time.Second):
+						So(registered, ShouldBeTrue)
+						return
+					}
+				}
+			}
+
+			startWorkerThenStop(&workerCfg, dummyTester)
+		})
+		Convey("with one job", func(ctx C) {
+			workerCfg.Mirrors = []mirrorConfig{
+				mirrorConfig{
+					Name:     "job-ls",
+					Provider: provCommand,
+					Command:  "ls",
+				},
+			}
+
+			dummyTester := func(*Worker) {
+				url := ""
+				jobRunning := false
+				lastStatus := SyncStatus(None)
+				for {
+					select {
+					case data := <-recvDataChan:
+						if reg, ok := data.(WorkerStatus); ok {
+							So(reg.ID, ShouldEqual, "dut")
+							url = reg.URL
+							time.Sleep(500 * time.Millisecond)
+							sendCommandToWorker(url, httpClient, CmdStart, "job-ls")
+						} else if sch, ok := data.(MirrorSchedules); ok {
+							if !jobRunning {
+								So(len(sch.Schedules), ShouldEqual, 1)
+								So(sch.Schedules[0].MirrorName, ShouldEqual, "job-ls")
+								So(sch.Schedules[0].NextSchedule,
+									ShouldHappenBetween,
+									time.Now().Add(-2*time.Second),
+									time.Now().Add(1*time.Minute))
+							}
+						} else if status, ok := data.(MirrorStatus); ok {
+							logger.Noticef("Job %s status %s", status.Name, status.Status.String())
+							jobRunning = status.Status == PreSyncing || status.Status == Syncing
+							So(status.Status, ShouldNotEqual, Failed)
+							lastStatus = status.Status
+						}
+					case <-time.After(2 * time.Second):
+						So(url, ShouldNotEqual, "")
+						So(jobRunning, ShouldBeFalse)
+						So(lastStatus, ShouldEqual, Success)
+						return
+					}
+				}
+			}
+
+			startWorkerThenStop(&workerCfg, dummyTester)
+		})
+		Convey("with several jobs", func(ctx C) {
+			workerCfg.Mirrors = []mirrorConfig{
+				mirrorConfig{
+					Name:     "job-ls-1",
+					Provider: provCommand,
+					Command:  "ls",
+				},
+				mirrorConfig{
+					Name:     "job-fail",
+					Provider: provCommand,
+					Command:  "non-existent-command-xxxx",
+				},
+				mirrorConfig{
+					Name:     "job-ls-2",
+					Provider: provCommand,
+					Command:  "ls",
+				},
+			}
+
+			dummyTester := func(*Worker) {
+				url := ""
+				lastStatus := make(map[string]SyncStatus)
+				nextSch := make(map[string]time.Time)
+				for {
+					select {
+					case data := <-recvDataChan:
+						if reg, ok := data.(WorkerStatus); ok {
+							So(reg.ID, ShouldEqual, "dut")
+							url = reg.URL
+							time.Sleep(500 * time.Millisecond)
+							sendCommandToWorker(url, httpClient, CmdStart, "job-fail")
+							sendCommandToWorker(url, httpClient, CmdStart, "job-ls-1")
+							sendCommandToWorker(url, httpClient, CmdStart, "job-ls-2")
+						} else if sch, ok := data.(MirrorSchedules); ok {
+							//So(len(sch.Schedules), ShouldEqual, 3)
+							for _, item := range sch.Schedules {
+								nextSch[item.MirrorName] = item.NextSchedule
+							}
+						} else if status, ok := data.(MirrorStatus); ok {
+							logger.Noticef("Job %s status %s", status.Name, status.Status.String())
+							jobRunning := status.Status == PreSyncing || status.Status == Syncing
+							if !jobRunning {
+								if status.Name == "job-fail" {
+									So(status.Status, ShouldEqual, Failed)
+								} else {
+									So(status.Status, ShouldNotEqual, Failed)
+								}
+							}
+							lastStatus[status.Name] = status.Status
+						}
+					case <-time.After(2 * time.Second):
+						So(len(lastStatus), ShouldEqual, 3)
+						So(len(nextSch), ShouldEqual, 3)
+						return
+					}
+				}
+			}
+
+			startWorkerThenStop(&workerCfg, dummyTester)
+		})
+	})
+}

worker/zfs_hook.go

@@ -3,6 +3,7 @@ package worker
 import (
 	"fmt"
 	"os"
+	"os/user"
 	"strings"
 
 	"github.com/codeskyblue/go-sh"
@@ -10,36 +11,44 @@ import (
 
 type zfsHook struct {
 	emptyHook
-	provider mirrorProvider
+	zpool string
-	zpool    string
 }
 
 func newZfsHook(provider mirrorProvider, zpool string) *zfsHook {
 	return &zfsHook{
-		provider: provider,
+		emptyHook: emptyHook{
-		zpool:    zpool,
+			provider: provider,
+		},
+		zpool: zpool,
 	}
 }
 
-// create zfs dataset for a new mirror
+func (z *zfsHook) printHelpMessage() {
+	zfsDataset := fmt.Sprintf("%s/%s", z.zpool, z.provider.Name())
+	zfsDataset = strings.ToLower(zfsDataset)
+	workingDir := z.provider.WorkingDir()
+	logger.Infof("You may create the ZFS dataset with:")
+	logger.Infof("    zfs create '%s'", zfsDataset)
+	logger.Infof("    zfs set mountpoint='%s' '%s'", workingDir, zfsDataset)
+	usr, err := user.Current()
+	if err != nil || usr.Uid == "0" {
+		return
+	}
+	logger.Infof("    chown %s '%s'", usr.Uid, workingDir)
+}
+
+// check if working directory is a zfs dataset
 func (z *zfsHook) preJob() error {
 	workingDir := z.provider.WorkingDir()
 	if _, err := os.Stat(workingDir); os.IsNotExist(err) {
-		// sudo zfs create $zfsDataset
+		logger.Errorf("Directory %s doesn't exist", workingDir)
-		// sudo zfs set mountpoint=${absPath} ${zfsDataset}
+		z.printHelpMessage()
-
+		return err
-		zfsDataset := fmt.Sprintf("%s/%s", z.zpool, z.provider.Name())
+	}
-		// Unknown issue of ZFS:
+	if err := sh.Command("mountpoint", "-q", workingDir).Run(); err != nil {
-		// dataset name should not contain upper case letters
+		logger.Errorf("%s is not a mount point", workingDir)
-		zfsDataset = strings.ToLower(zfsDataset)
+		z.printHelpMessage()
-		logger.Infof("Creating ZFS dataset %s", zfsDataset)
+		return err
-		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
 }

worker/zfs_hook_test.go

@@ -0,0 +1,48 @@
+package worker
+
+import (
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	. "github.com/smartystreets/goconvey/convey"
+)
+
+func TestZFSHook(t *testing.T) {
+
+	Convey("ZFS Hook should work", t, func(ctx C) {
+		tmpDir, err := ioutil.TempDir("", "tunasync")
+		tmpFile := filepath.Join(tmpDir, "log_file")
+
+		c := cmdConfig{
+			name:        "tuna_zfs_hook_test",
+			upstreamURL: "http://mirrors.tuna.moe/",
+			command:     "ls",
+			workingDir:  tmpDir,
+			logDir:      tmpDir,
+			logFile:     tmpFile,
+			interval:    1 * time.Second,
+		}
+
+		provider, err := newCmdProvider(c)
+		So(err, ShouldBeNil)
+		Convey("When working directory doesn't exist", func(ctx C) {
+
+			errRm := os.RemoveAll(tmpDir)
+			So(errRm, ShouldBeNil)
+
+			hook := newZfsHook(provider, "test_pool")
+			err := hook.preJob()
+			So(err, ShouldNotBeNil)
+		})
+		Convey("When working directory is not a mount point", func(ctx C) {
+			defer os.RemoveAll(tmpDir)
+
+			hook := newZfsHook(provider, "test_pool")
+			err := hook.preJob()
+			So(err, ShouldNotBeNil)
+		})
+	})
+}