Bump version to 0.7.0

Add redis db backend

.github/workflows/release.yml

@@ -21,16 +21,12 @@ jobs:
     - 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
+        for i in linux-amd64 linux-arm64; do
-        make tunasynctl
+          make ARCH=$i all
-        tar -jcf build/tunasync-linux-bin.tar.bz2 -C build tunasync tunasynctl
+          tar -cz --numeric-owner --owner root --group root -f tunasync-$i-bin.tar.gz -C build-$i tunasync tunasynctl 
+        done
 
     - name: Create Release
       id: create_release
@@ -42,13 +38,9 @@ jobs:
         release_name: Release ${{ github.ref }}
         draft: false
         prerelease: false
-    - name: Upload Release Asset
+    - name: Upload Release Assets
-      id: upload-release-asset 
-      uses: actions/upload-release-asset@v1
       env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+        GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      with:
+        TAG_NAME: ${{ github.ref }}
-          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 
+      run: |
-          asset_path: ./build/tunasync-linux-bin.tar.bz2
+        hub release edit $(find . -type f -name "tunasync-*.tar.gz" -printf "-a %p ") -m "" "${TAG_NAME##*/}"
-          asset_name: tunasync-linux-bin.tar.bz2
-          asset_content_type: application/x-bzip2

.github/workflows/tunasync.yml

@@ -32,7 +32,7 @@ jobs:
       uses: actions/upload-artifact@v1
       with:
         name: tunasync-bin
-        path: build/
+        path: build-linux-amd64/
 
   test:
     name: Test

.gitignore

@@ -1 +1,2 @@
 /build
+/build-*

Makefile

@@ -1,19 +1,22 @@
 LDFLAGS="-X main.buildstamp=`date -u '+%s'` -X main.githash=`git rev-parse HEAD`"
+ARCH ?= linux-amd64
+ARCH_LIST = $(subst -, ,$(ARCH))
+GOOS = $(word 1, $(ARCH_LIST))
+GOARCH = $(word 2, $(ARCH_LIST))
+BUILDBIN = tunasync tunasynctl
 
-all: get tunasync tunasynctl
+all: $(BUILDBIN)
 
-get: 
+build-$(ARCH):
-	go get ./cmd/tunasync
+	mkdir -p $@
-	go get ./cmd/tunasynctl
 
-build:
+$(BUILDBIN): % : build-$(ARCH) build-$(ARCH)/%
-	mkdir -p build
 
-tunasync: build
+$(BUILDBIN:%=build-$(ARCH)/%) : build-$(ARCH)/% : cmd/%
-	go build -o build/tunasync -ldflags ${LDFLAGS} github.com/tuna/tunasync/cmd/tunasync
+	GOOS=$(GOOS) GOARCH=$(GOARCH) go get ./$<
-
+	GOOS=$(GOOS) GOARCH=$(GOARCH) go build -o $@ -ldflags ${LDFLAGS} github.com/tuna/tunasync/$<
-tunasynctl: build
-	go build -o build/tunasynctl -ldflags ${LDFLAGS} github.com/tuna/tunasync/cmd/tunasynctl
 
 test:
 	go test -v -covermode=count -coverprofile=profile.cov ./...
+
+.PHONY: all test $(BUILDBIN)

README.md

@@ -1,5 +1,4 @@
-tunasync
+# tunasync
-========
 
 ![Build Status](https://github.com/tuna/tunasync/workflows/tunasync/badge.svg)
 [![Coverage Status](https://coveralls.io/repos/github/tuna/tunasync/badge.svg?branch=master)](https://coveralls.io/github/tuna/tunasync?branch=master)
@@ -12,11 +11,11 @@ tunasync
 
 ## Download
 
-Pre-built binary for Linux x86_64 is available at [Github releases](https://github.com/tuna/tunasync/releases/latest).
+Pre-built binary for Linux x86_64 and ARM64 is available at [Github releases](https://github.com/tuna/tunasync/releases/latest).
 
 ## Design
 
-```
+```text
 # Architecture
 
 - Manager: Central instance for status and job management
@@ -50,13 +49,12 @@ PreSyncing                           Syncing                               Succe
                                 +-----------------+
 ```
 
-
 ## Building
 
 Go version: 1.13
 
-```
+```shell
-make all
+> make all
 ```
 
-Binaries in the `build/`.
+Binaries in the `build-linux-amd64/`.

cmd/tunasynctl/tunasynctl.go

@@ -8,6 +8,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"text/template"
 	"time"
 
 	"github.com/BurntSushi/toml"
@@ -160,8 +161,31 @@ func listJobs(c *cli.Context) error {
 					"of all jobs from manager server: %s", err.Error()),
 				1)
 		}
-		genericJobs = jobs
+		if statusStr := c.String("status"); statusStr != "" {
-
+			filteredJobs := make([]tunasync.WebMirrorStatus, 0, len(jobs))
+			var statuses []tunasync.SyncStatus
+			for _, s := range strings.Split(statusStr, ",") {
+				var status tunasync.SyncStatus
+				err = status.UnmarshalJSON([]byte("\"" + strings.TrimSpace(s) + "\""))
+				if err != nil {
+					return cli.NewExitError(
+						fmt.Sprintf("Error parsing status: %s", err.Error()),
+						1)
+				}
+				statuses = append(statuses, status)
+			}
+			for _, job := range jobs {
+				for _, s := range statuses {
+					if job.Status == s {
+						filteredJobs = append(filteredJobs, job)
+						break
+					}
+				}
+			}
+			genericJobs = filteredJobs
+		} else {
+			genericJobs = jobs
+		}
 	} else {
 		var jobs []tunasync.MirrorStatus
 		args := c.Args()
@@ -196,13 +220,46 @@ func listJobs(c *cli.Context) error {
 		genericJobs = jobs
 	}
 
-	b, err := json.MarshalIndent(genericJobs, "", "  ")
+	if format := c.String("format"); format != "" {
-	if err != nil {
+		tpl := template.New("")
-		return cli.NewExitError(
+		_, err := tpl.Parse(format)
-			fmt.Sprintf("Error printing out information: %s", err.Error()),
+		if err != nil {
-			1)
+			return cli.NewExitError(
+				fmt.Sprintf("Error parsing format template: %s", err.Error()),
+				1)
+		}
+		switch jobs := genericJobs.(type) {
+		case []tunasync.WebMirrorStatus:
+			for _, job := range jobs {
+				err = tpl.Execute(os.Stdout, job)
+				if err != nil {
+					return cli.NewExitError(
+						fmt.Sprintf("Error printing out information: %s", err.Error()),
+						1)
+				}
+				fmt.Println()
+			}
+		case []tunasync.MirrorStatus:
+			for _, job := range jobs {
+				err = tpl.Execute(os.Stdout, job)
+				if err != nil {
+					return cli.NewExitError(
+						fmt.Sprintf("Error printing out information: %s", err.Error()),
+						1)
+				}
+				fmt.Println()
+			}
+		}
+	} else {
+		b, err := json.MarshalIndent(genericJobs, "", "  ")
+		if err != nil {
+			return cli.NewExitError(
+				fmt.Sprintf("Error printing out information: %s", err.Error()),
+				1)
+		}
+		fmt.Println(string(b))
 	}
-	fmt.Println(string(b))
+
 	return nil
 }
 
@@ -506,6 +563,14 @@ func main() {
 						Name:  "all, a",
 						Usage: "List all jobs of all workers",
 					},
+					cli.StringFlag{
+						Name:  "status, s",
+						Usage: "Filter output based on status provided",
+					},
+					cli.StringFlag{
+						Name:  "format, f",
+						Usage: "Pretty-print containers using a Go template",
+					},
 				}...),
 			Action: initializeWrapper(listJobs),
 		},

docs/zh_CN/get_started.md

@@ -1,4 +1,5 @@
 # tunasync 上手指南
+
 date: 2016-10-31 00:50:00
 
 [tunasync](https://github.com/tuna/tunasync) 是[清华大学 TUNA 镜像源](https://mirrors.tuna.tsinghua.edu.cn)目前使用的镜像方案。
@@ -7,32 +8,32 @@ date: 2016-10-31 00:50:00
 
 本例中:
 
- - 只镜像[elvish](https://elvish.io)项目
+- 只镜像[elvish](https://elvish.io)项目
- - 禁用了https
+- 禁用了https
- - 禁用了cgroup支持
+- 禁用了cgroup支持
 
 ## 获得tunasync
 
 ### 二进制包
 
-到 [Github Releases](https://github.com/tuna/tunasync/releases/latest) 下载 `tunasync-linux-bin.tar.gz` 即可。
+到 [Github Releases](https://github.com/tuna/tunasync/releases/latest) 下载 `tunasync-linux-amd64-bin.tar.gz` 即可。
 
 ### 自行编译
 
-```
+```shell
-$ make
+> make
 ```
 
 ## 配置
 
-```
+```shell
-$ mkdir ~/tunasync_demo
+> mkdir ~/tunasync_demo
-$ mkdir /tmp/tunasync
+> mkdir /tmp/tunasync
 ```
 
-`~/tunasync_demo/worker.conf`:
+编辑 `~/tunasync_demo/worker.conf`:
 
-```
+```conf
 [global]
 name = "test_worker"
 log_dir = "/tmp/tunasync/log/tunasync/{{.Name}}"
@@ -64,9 +65,9 @@ upstream = "rsync://rsync.elvish.io/elvish/"
 use_ipv6 = false
 ```
 
-`~/tunasync_demo/manager.conf`:
+编辑 `~/tunasync_demo/manager.conf`:
 
-```
+```conf
 debug = false
 
 [server]
@@ -81,28 +82,30 @@ db_file = "/tmp/tunasync/manager.db"
 ca_cert = ""
 ```
 
+如果使用 redis 作为数据库后端，把 db_type 改为 redis，下面的 db_file 设为 redis 服务器的地址： `redis://user:password@host:port/db_number`。
+
 ### 运行
 
-```
+```shell
-$ tunasync manager --config ~/tunasync_demo/manager.conf
+> tunasync manager --config ~/tunasync_demo/manager.conf
-$ tunasync worker --config ~/tunasync_demo/worker.conf
+> tunasync worker --config ~/tunasync_demo/worker.conf
 ```
 
-本例中，镜像的数据在`/tmp/tunasync/`
+本例中，镜像的数据在 `/tmp/tunasync/`。
 
 ### 控制
 
 查看同步状态
 
-```
+```shell
-$ tunasynctl list -p 12345 --all
+> tunasynctl list -p 12345 --all
 ```
 
 tunasynctl 也支持配置文件。配置文件可以放在 `/etc/tunasync/ctl.conf` 或者 `~/.config/tunasync/ctl.conf` 两个位置，后者可以覆盖前者的配置值。
 
 配置文件内容为：
 
-```
+```conf
 manager_addr = "127.0.0.1"
 manager_port = 12345
 ca_cert = ""
@@ -118,13 +121,13 @@ worker 和 manager 之间用 http(s) 通信，如果你 worker 和 manager 都
 
 可以参看
 
-```
+```shell
-$ tunasync manager --help
+> tunasync manager --help
-$ tunasync worker --help
+> tunasync worker --help
 ```
 
 可以看一下 log 目录
 
-一些 worker 配置文件示例 [workers.conf](workers.conf) 
+一些 worker 配置文件示例 [workers.conf](workers.conf)。
 
-你可能会用到的操作 [tips.md](tips.md) 
+你可能会用到的操作 [tips.md](tips.md)。

docs/zh_CN/workers.conf

@@ -7,6 +7,11 @@ mirror_dir = "/srv/tunasync"
 concurrent = 10
 interval = 1
 
+# ensure the exec user be add into `docker` group
+[docker]
+# in `command provider` can use docker_image and docker_volumes
+enable = true
+
 [manager]
 api_base = "http://localhost:12345"
 token = "some_token"

go.mod

@@ -4,18 +4,22 @@ go 1.13
 
 require (
 	github.com/BurntSushi/toml v0.3.1
+	github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a // indirect
+	github.com/alicebob/miniredis v2.5.0+incompatible
 	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/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568 // indirect
 	github.com/gin-gonic/gin v1.5.0
+	github.com/go-redis/redis/v8 v8.3.0
+	github.com/gomodule/redigo v1.8.2 // indirect
 	github.com/imdario/mergo v0.3.9
-	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
+	github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb // indirect
-	golang.org/x/tools v0.0.0-20200312194400-c312e98713c2 // indirect
+	golang.org/x/sys v0.0.0-20200519105757-fe76b779f299
 	gopkg.in/op/go-logging.v1 v1.0.0-20160211212156-b2cb9fa56473
 )

go.sum

@@ -1,9 +1,18 @@
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a h1:HbKu58rmZpUGpz5+4FfNmIU+FmZg2P3Xaj2v2bfNWmk=
+github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
+github.com/alicebob/miniredis v2.5.0+incompatible h1:yBHoLpsyjupjz3NL3MhKMVkR41j82Yjf3KFv7ApYzUI=
+github.com/alicebob/miniredis v2.5.0+incompatible/go.mod h1:8HZjEj4yU0dwhYHky+DxYx+6BMjkBbe5ONFIF1MXffk=
 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/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
+github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
+github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
+github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 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=
@@ -17,6 +26,13 @@ github.com/dennwc/btrfs v0.0.0-20190517175702-d917b30ff035 h1:4e+UEZaKPx0ZEiCMPU
 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/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568 h1:BHsljHzVlRcyQhjrss6TZTdY2VfCqZPbv5k3iBFa2ZQ=
+github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc=
+github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
+github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 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=
@@ -25,10 +41,29 @@ github.com/go-playground/locales v0.12.1 h1:2FITxuFt/xuCNP1Acdhv62OzaCiviiE4kotf
 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/go-redis/redis/v8 v8.3.0 h1:Xrwvn8+QqUYD1MbQmda3cVR2U9li5XbtRFkKZN5Y0hk=
+github.com/go-redis/redis/v8 v8.3.0/go.mod h1:a2xkpBM7NJUN5V5kiF46X5Ltx4WeXJ9757X/ScKUBdE=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 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/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8=
+github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA=
+github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs=
+github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w=
+github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0=
+github.com/golang/protobuf v1.4.2 h1:+Z5KGCizgyZCbGh1KZqA0fcLLkwbsjIzS4aV2v7wJX0=
+github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI=
+github.com/gomodule/redigo v1.8.2 h1:H5XSIre1MB5NbPYFp+i1NBbb5qN1W8Y8YAQoAYbkm8k=
+github.com/gomodule/redigo v1.8.2/go.mod h1:P9dn9mFrCBvWhGE1wpxx6fgq7BAeLBk+UUUzlpkBYO0=
+github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
+github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.5.2 h1:X2ev0eStA3AbceY54o37/0PQ/UWqKEiiO2dKL5OPaFM=
+github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1 h1:EGx4pi6eqNxGaHF6qqu48+N2wcFQ5qg5FXgOdqsJ5d8=
 github.com/gopherjs/gopherjs v0.0.0-20181017120253-0766667cb4d1/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/imdario/mergo v0.3.9 h1:UauaLniWCFHWd+Jp9oCEkTBj8VO/9DKg3PV3VCNMDIg=
 github.com/imdario/mergo v0.3.9/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/json-iterator/go v1.1.7 h1:KfgG9LzI+pYjr4xvmz/5H4FXjokeP+rlHLhv3iH62Fo=
@@ -39,12 +74,20 @@ 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 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742 h1:Esafd1046DLDQ0W1YjYsBW+p8U2u7vzgW2SQVmlNazg=
 github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/nxadm/tail v1.4.4 h1:DQuhQpB1tVlglWS2hLQ5OV6B5r8aGxSrPc5Qo6uTN78=
+github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
+github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk=
+github.com/onsi/ginkgo v1.14.1 h1:jMU0WaQrP0a/YAEq8eJmJKjBoMs+pClEr1vDMlM/Do4=
+github.com/onsi/ginkgo v1.14.1/go.mod h1:iSB4RoI2tjJc9BBv4NKIKWKya62Rps+oPG/Lv9klQyY=
+github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY=
+github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
+github.com/onsi/gomega v1.10.2 h1:aY/nuoWlKJud2J6U0E3NWsjlg+0GtwXxgEqthRdzlcs=
+github.com/onsi/gomega v1.10.2/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 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=
@@ -63,42 +106,65 @@ github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+
 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/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
+github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=
+github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 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=
+github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb h1:ZkM6LRnq40pR1Ox0hTHlnpkcOTuFIDQpZ1IN8rKKhX0=
+github.com/yuin/gopher-lua v0.0.0-20191220021717-ab39c6098bdb/go.mod h1:gqRgreBUhTSL0GeU64rtZ3Uq3wtjOa/TB2YfrtkCbVQ=
+go.opentelemetry.io/otel v0.13.0 h1:2isEnyzjjJZq6r2EKMsFj4TxiQiexsM04AVhwbR/oBA=
+go.opentelemetry.io/otel v0.13.0/go.mod h1:dlSNewoRYikTkotEnxdmuBHgzT+k/idJSfDv/FxEnOY=
 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/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
-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-20200520004742-59133d7f0dd7 h1:AeiKBIuRw3UomYXSbLy0Mc2dDLfdtbT/IVn4keq83P0=
-golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
-golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20190204203706-41f3e6584952/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 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-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20200302150141-5c8b2ff67527/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20200519105757-fe76b779f299 h1:DYfZAGf2WMFjMxbgTjaC+2HC7NkNAQs+6Q8b9WEB/F4=
+golang.org/x/sys v0.0.0-20200519105757-fe76b779f299/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs=
+golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
+golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 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/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
-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=
+google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
+google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
+google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM=
+google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE=
+google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo=
+google.golang.org/protobuf v1.23.0 h1:4MY060fB1DLGMB/7MBTLnwQUY6+F09GEiz6SsrNqyzM=
+google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/go-playground/assert.v1 v1.2.1 h1:xoYuJVE7KT85PYWrN730RguIQO0ePzVRfFMXadIrXTM=
 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/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v2 v2.3.0 h1:clyUAQHOM3G0M3f5vQj7LuJrETvjVot3Z5el9nffUtU=
+gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

internal/msg.go

@@ -1,6 +1,8 @@
 package internal
 
 import (
+	"bytes"
+	"encoding/json"
 	"fmt"
 	"time"
 )
@@ -24,10 +26,11 @@ type MirrorStatus struct {
 // A WorkerStatus is the information struct that describe
 // a worker, and sent from the manager to clients.
 type WorkerStatus struct {
-	ID         string    `json:"id"`
+	ID           string    `json:"id"`
-	URL        string    `json:"url"`         // worker url
+	URL          string    `json:"url"`           // worker url
-	Token      string    `json:"token"`       // session token
+	Token        string    `json:"token"`         // session token
-	LastOnline time.Time `json:"last_online"` // last seen
+	LastOnline   time.Time `json:"last_online"`   // last seen
+	LastRegister time.Time `json:"last_register"` // last register time
 }
 
 type MirrorSchedules struct {
@@ -59,21 +62,45 @@ const (
 )
 
 func (c CmdVerb) String() string {
-	switch c {
+	mapping := map[CmdVerb]string{
-	case CmdStart:
+		CmdStart:   "start",
-		return "start"
+		CmdStop:    "stop",
-	case CmdStop:
+		CmdDisable: "disable",
-		return "stop"
+		CmdRestart: "restart",
-	case CmdDisable:
+		CmdPing:    "ping",
-		return "disable"
+		CmdReload:  "reload",
-	case CmdRestart:
-		return "restart"
-	case CmdPing:
-		return "ping"
-	case CmdReload:
-		return "reload"
 	}
-	return "unknown"
+	return mapping[c]
+}
+
+func NewCmdVerbFromString(s string) CmdVerb {
+	mapping := map[string]CmdVerb{
+		"start":   CmdStart,
+		"stop":    CmdStop,
+		"disable": CmdDisable,
+		"restart": CmdRestart,
+		"ping":    CmdPing,
+		"reload":  CmdReload,
+	}
+	return mapping[s]
+}
+
+// Marshal and Unmarshal for CmdVerb
+func (s CmdVerb) MarshalJSON() ([]byte, error) {
+	buffer := bytes.NewBufferString(`"`)
+	buffer.WriteString(s.String())
+	buffer.WriteString(`"`)
+	return buffer.Bytes(), nil
+}
+
+func (s *CmdVerb) UnmarshalJSON(b []byte) error {
+	var j string
+	err := json.Unmarshal(b, &j)
+	if err != nil {
+		return err
+	}
+	*s = NewCmdVerbFromString(j)
+	return nil
 }
 
 // A WorkerCmd is the command message send from the

internal/version.go

@@ -1,4 +1,4 @@
 package internal
 
 // Version of the program
-const Version string = "0.6.5"
+const Version string = "0.7.0"

manager/config.go

@@ -29,6 +29,7 @@ type FileConfig struct {
 	CACert string `toml:"ca_cert"`
 }
 
+// LoadConfig loads config from specified file
 func LoadConfig(cfgFile string, c *cli.Context) (*Config, error) {
 
 	cfg := new(Config)

manager/db.go

@@ -4,8 +4,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/boltdb/bolt"
+	"github.com/go-redis/redis/v8"
 
 	. "github.com/tuna/tunasync/internal"
 )
@@ -16,6 +18,7 @@ type dbAdapter interface {
 	GetWorker(workerID string) (WorkerStatus, error)
 	DeleteWorker(workerID string) error
 	CreateWorker(w WorkerStatus) (WorkerStatus, error)
+	RefreshWorker(workerID string) (WorkerStatus, error)
 	UpdateMirrorStatus(workerID, mirrorID string, status MirrorStatus) (MirrorStatus, error)
 	GetMirrorStatus(workerID, mirrorID string) (MirrorStatus, error)
 	ListMirrorStatus(workerID string) ([]MirrorStatus, error)
@@ -24,21 +27,14 @@ type dbAdapter interface {
 	Close() error
 }
 
-func makeDBAdapter(dbType string, dbFile string) (dbAdapter, error) {
+// interface for a kv database
-	if dbType == "bolt" {
+type kvAdapter interface {
-		innerDB, err := bolt.Open(dbFile, 0600, nil)
+	InitBucket(bucket string) error
-		if err != nil {
+	Get(bucket string, key string) ([]byte, error)
-			return nil, err
+	GetAll(bucket string) (map[string][]byte, error)
-		}
+	Put(bucket string, key string, value []byte) error
-		db := boltAdapter{
+	Delete(bucket string, key string) error
-			db:     innerDB,
+	Close() error
-			dbFile: dbFile,
-		}
-		err = db.Init()
-		return &db, err
-	}
-	// unsupported db-type
-	return nil, fmt.Errorf("unsupported db-type: %s", dbType)
 }
 
 const (
@@ -46,133 +42,141 @@ const (
 	_statusBucketKey = "mirror_status"
 )
 
-type boltAdapter struct {
+func makeDBAdapter(dbType string, dbFile string) (dbAdapter, error) {
-	db     *bolt.DB
+	if dbType == "bolt" {
-	dbFile string
+		innerDB, err := bolt.Open(dbFile, 0600, &bolt.Options{
+			Timeout: 5 * time.Second,
+		})
+		if err != nil {
+			return nil, err
+		}
+		db := boltAdapter{
+			db: innerDB,
+		}
+		kv := kvDBAdapter{
+			db: &db,
+		}
+		err = kv.Init()
+		return &kv, err
+	} else if dbType == "redis" {
+		opt, err := redis.ParseURL(dbFile)
+		if err != nil {
+			return nil, fmt.Errorf("bad redis url: %s", err)
+		}
+		innerDB := redis.NewClient(opt)
+		db := redisAdapter{
+			db: innerDB,
+		}
+		kv := kvDBAdapter{
+			db: &db,
+		}
+		err = kv.Init()
+		return &kv, err
+	}
+	// unsupported db-type
+	return nil, fmt.Errorf("unsupported db-type: %s", dbType)
 }
 
-func (b *boltAdapter) Init() (err error) {
+// use the underlying kv database to store data
-	return b.db.Update(func(tx *bolt.Tx) error {
+type kvDBAdapter struct {
-		_, err = tx.CreateBucketIfNotExists([]byte(_workerBucketKey))
+	db kvAdapter
-		if err != nil {
-			return fmt.Errorf("create bucket %s error: %s", _workerBucketKey, err.Error())
-		}
-		_, err = tx.CreateBucketIfNotExists([]byte(_statusBucketKey))
-		if err != nil {
-			return fmt.Errorf("create bucket %s error: %s", _statusBucketKey, err.Error())
-		}
-		return nil
-	})
 }
 
-func (b *boltAdapter) ListWorkers() (ws []WorkerStatus, err error) {
+func (b *kvDBAdapter) Init() error {
-	err = b.db.View(func(tx *bolt.Tx) error {
+	err := b.db.InitBucket(_workerBucketKey)
-		bucket := tx.Bucket([]byte(_workerBucketKey))
+	if err != nil {
-		c := bucket.Cursor()
+		return fmt.Errorf("create bucket %s error: %s", _workerBucketKey, err.Error())
-		var w WorkerStatus
+	}
-		for k, v := c.First(); k != nil; k, v = c.Next() {
+	err = b.db.InitBucket(_statusBucketKey)
-			jsonErr := json.Unmarshal(v, &w)
+	if err != nil {
-			if jsonErr != nil {
+		return fmt.Errorf("create bucket %s error: %s", _workerBucketKey, err.Error())
-				err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
+	}
-				continue
+	return err
-			}
+}
-			ws = append(ws, w)
+
+func (b *kvDBAdapter) ListWorkers() (ws []WorkerStatus, err error) {
+	var workers map[string][]byte
+	workers, err = b.db.GetAll(_workerBucketKey)
+
+	var w WorkerStatus
+	for _, v := range workers {
+		jsonErr := json.Unmarshal(v, &w)
+		if jsonErr != nil {
+			err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
+			continue
 		}
-		return err
+		ws = append(ws, w)
-	})
+	}
 	return
 }
 
-func (b *boltAdapter) GetWorker(workerID string) (w WorkerStatus, err error) {
+func (b *kvDBAdapter) GetWorker(workerID string) (w WorkerStatus, err error) {
-	err = b.db.View(func(tx *bolt.Tx) error {
+	var v []byte
-		bucket := tx.Bucket([]byte(_workerBucketKey))
+	v, err = b.db.Get(_workerBucketKey, workerID)
-		v := bucket.Get([]byte(workerID))
+	if v == nil {
-		if v == nil {
+		err = fmt.Errorf("invalid workerID %s", workerID)
-			return fmt.Errorf("invalid workerID %s", workerID)
+	} else {
-		}
+		err = json.Unmarshal(v, &w)
-		err := json.Unmarshal(v, &w)
+	}
-		return err
-	})
 	return
 }
 
-func (b *boltAdapter) DeleteWorker(workerID string) (err error) {
+func (b *kvDBAdapter) DeleteWorker(workerID string) error {
-	err = b.db.Update(func(tx *bolt.Tx) error {
+	v, _ := b.db.Get(_workerBucketKey, workerID)
-		bucket := tx.Bucket([]byte(_workerBucketKey))
+	if v == nil {
-		v := bucket.Get([]byte(workerID))
+		return fmt.Errorf("invalid workerID %s", workerID)
-		if v == nil {
+	}
-			return fmt.Errorf("invalid workerID %s", workerID)
+	return b.db.Delete(_workerBucketKey, workerID)
-		}
-		err := bucket.Delete([]byte(workerID))
-		return err
-	})
-	return
 }
 
-func (b *boltAdapter) CreateWorker(w WorkerStatus) (WorkerStatus, error) {
+func (b *kvDBAdapter) CreateWorker(w WorkerStatus) (WorkerStatus, error) {
-	err := b.db.Update(func(tx *bolt.Tx) error {
+	v, err := json.Marshal(w)
-		bucket := tx.Bucket([]byte(_workerBucketKey))
+	if err == nil {
-		v, err := json.Marshal(w)
+		err = b.db.Put(_workerBucketKey, w.ID, v)
-		if err != nil {
+	}
-			return err
-		}
-		err = bucket.Put([]byte(w.ID), v)
-		return err
-	})
 	return w, err
 }
 
-func (b *boltAdapter) UpdateMirrorStatus(workerID, mirrorID string, status MirrorStatus) (MirrorStatus, error) {
+func (b *kvDBAdapter) RefreshWorker(workerID string) (w WorkerStatus, err error) {
+	w, err = b.GetWorker(workerID)
+	if err == nil {
+		w.LastOnline = time.Now()
+		w, err = b.CreateWorker(w)
+	}
+	return w, err
+}
+
+func (b *kvDBAdapter) UpdateMirrorStatus(workerID, mirrorID string, status MirrorStatus) (MirrorStatus, error) {
 	id := mirrorID + "/" + workerID
-	err := b.db.Update(func(tx *bolt.Tx) error {
+	v, err := json.Marshal(status)
-		bucket := tx.Bucket([]byte(_statusBucketKey))
+	if err == nil {
-		v, err := json.Marshal(status)
+		err = b.db.Put(_statusBucketKey, id, v)
-		err = bucket.Put([]byte(id), v)
+	}
-		return err
-	})
 	return status, err
 }
 
-func (b *boltAdapter) GetMirrorStatus(workerID, mirrorID string) (m MirrorStatus, err error) {
+func (b *kvDBAdapter) GetMirrorStatus(workerID, mirrorID string) (m MirrorStatus, err error) {
 	id := mirrorID + "/" + workerID
-	err = b.db.Update(func(tx *bolt.Tx) error {
+	var v []byte
-		bucket := tx.Bucket([]byte(_statusBucketKey))
+	v, err = b.db.Get(_statusBucketKey, id)
-		v := bucket.Get([]byte(id))
+	if v == nil {
-		if v == nil {
+		err = fmt.Errorf("no mirror '%s' exists in worker '%s'", mirrorID, workerID)
-			return fmt.Errorf("no mirror '%s' exists in worker '%s'", mirrorID, workerID)
+	} else if err == nil {
-		}
+		err = json.Unmarshal(v, &m)
-		err := json.Unmarshal(v, &m)
+	}
-		return err
-	})
 	return
 }
 
-func (b *boltAdapter) ListMirrorStatus(workerID string) (ms []MirrorStatus, err error) {
+func (b *kvDBAdapter) ListMirrorStatus(workerID string) (ms []MirrorStatus, err error) {
-	err = b.db.View(func(tx *bolt.Tx) error {
+	var vals map[string][]byte
-		bucket := tx.Bucket([]byte(_statusBucketKey))
+	vals, err = b.db.GetAll(_statusBucketKey)
-		c := bucket.Cursor()
+	if err != nil {
-		var m MirrorStatus
+		return
-		for k, v := c.First(); k != nil; k, v = c.Next() {
+	}
-			if wID := strings.Split(string(k), "/")[1]; wID == workerID {
-				jsonErr := json.Unmarshal(v, &m)
-				if jsonErr != nil {
-					err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
-					continue
-				}
-				ms = append(ms, m)
-			}
-		}
-		return err
-	})
-	return
-}
 
-func (b *boltAdapter) ListAllMirrorStatus() (ms []MirrorStatus, err error) {
+	for k, v := range vals {
-	err = b.db.View(func(tx *bolt.Tx) error {
+		if wID := strings.Split(k, "/")[1]; wID == workerID {
-		bucket := tx.Bucket([]byte(_statusBucketKey))
+			var m MirrorStatus
-		c := bucket.Cursor()
-		var m MirrorStatus
-		for k, v := c.First(); k != nil; k, v = c.Next() {
 			jsonErr := json.Unmarshal(v, &m)
 			if jsonErr != nil {
 				err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
@@ -180,32 +184,54 @@ func (b *boltAdapter) ListAllMirrorStatus() (ms []MirrorStatus, err error) {
 			}
 			ms = append(ms, m)
 		}
-		return err
+	}
-	})
 	return
 }
 
-func (b *boltAdapter) FlushDisabledJobs() (err error) {
+func (b *kvDBAdapter) ListAllMirrorStatus() (ms []MirrorStatus, err error) {
-	err = b.db.Update(func(tx *bolt.Tx) error {
+	var vals map[string][]byte
-		bucket := tx.Bucket([]byte(_statusBucketKey))
+	vals, err = b.db.GetAll(_statusBucketKey)
-		c := bucket.Cursor()
+	if err != nil {
+		return
+	}
+
+	for _, v := range vals {
 		var m MirrorStatus
-		for k, v := c.First(); k != nil; k, v = c.Next() {
+		jsonErr := json.Unmarshal(v, &m)
-			jsonErr := json.Unmarshal(v, &m)
+		if jsonErr != nil {
-			if jsonErr != nil {
+			err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
-				err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
+			continue
-				continue
+		}
-			}
+		ms = append(ms, m)
-			if m.Status == Disabled || len(m.Name) == 0 {
+	}
-				err = c.Delete()
+	return
+}
+
+func (b *kvDBAdapter) FlushDisabledJobs() (err error) {
+	var vals map[string][]byte
+	vals, err = b.db.GetAll(_statusBucketKey)
+	if err != nil {
+		return
+	}
+
+	for k, v := range vals {
+		var m MirrorStatus
+		jsonErr := json.Unmarshal(v, &m)
+		if jsonErr != nil {
+			err = fmt.Errorf("%s; %s", err.Error(), jsonErr)
+			continue
+		}
+		if m.Status == Disabled || len(m.Name) == 0 {
+			deleteErr := b.db.Delete(_statusBucketKey, k)
+			if deleteErr != nil {
+				err = fmt.Errorf("%s; %s", err.Error(), deleteErr)
 			}
 		}
-		return err
+	}
-	})
 	return
 }
 
-func (b *boltAdapter) Close() error {
+func (b *kvDBAdapter) Close() error {
 	if b.db != nil {
 		return b.db.Close()
 	}

manager/db_bolt.go

@@ -0,0 +1,66 @@
+package manager
+
+import (
+	"fmt"
+
+	"github.com/boltdb/bolt"
+)
+
+// implement kv interface backed by boltdb
+type boltAdapter struct {
+	db *bolt.DB
+}
+
+func (b *boltAdapter) InitBucket(bucket string) (err error) {
+	return b.db.Update(func(tx *bolt.Tx) error {
+		_, err = tx.CreateBucketIfNotExists([]byte(bucket))
+		if err != nil {
+			return fmt.Errorf("create bucket %s error: %s", _workerBucketKey, err.Error())
+		}
+		return nil
+	})
+}
+
+func (b *boltAdapter) Get(bucket string, key string) (v []byte, err error) {
+	err = b.db.View(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(bucket))
+		v = bucket.Get([]byte(key))
+		return nil
+	})
+	return
+}
+
+func (b *boltAdapter) GetAll(bucket string) (m map[string][]byte, err error) {
+	err = b.db.View(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(bucket))
+		c := bucket.Cursor()
+		m = make(map[string][]byte)
+		for k, v := c.First(); k != nil; k, v = c.Next() {
+			m[string(k)] = v
+		}
+		return nil
+	})
+	return
+}
+
+func (b *boltAdapter) Put(bucket string, key string, value []byte) error {
+	err := b.db.Update(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(bucket))
+		err := bucket.Put([]byte(key), value)
+		return err
+	})
+	return err
+}
+
+func (b *boltAdapter) Delete(bucket string, key string) error {
+	err := b.db.Update(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(bucket))
+		err := bucket.Delete([]byte(key))
+		return err
+	})
+	return err
+}
+
+func (b *boltAdapter) Close() error {
+	return b.db.Close()
+}

manager/db_redis.go

@@ -0,0 +1,54 @@
+package manager
+
+import (
+	"context"
+
+	"github.com/go-redis/redis/v8"
+)
+
+// implement kv interface backed by redis
+type redisAdapter struct {
+	db *redis.Client
+}
+
+var ctx = context.Background()
+
+func (b *redisAdapter) InitBucket(bucket string) (err error) {
+	// no-op
+	return
+}
+
+func (b *redisAdapter) Get(bucket string, key string) (v []byte, err error) {
+	var val string
+	val, err = b.db.HGet(ctx, bucket, key).Result()
+	if err == nil {
+		v = []byte(val)
+	}
+	return
+}
+
+func (b *redisAdapter) GetAll(bucket string) (m map[string][]byte, err error) {
+	var val map[string]string
+	val, err = b.db.HGetAll(ctx, bucket).Result()
+	if err == nil && val != nil {
+		m = make(map[string][]byte)
+		for k, v := range val {
+			m[k] = []byte(v)
+		}
+	}
+	return
+}
+
+func (b *redisAdapter) Put(bucket string, key string, value []byte) error {
+	_, err := b.db.HSet(ctx, bucket, key, string(value)).Result()
+	return err
+}
+
+func (b *redisAdapter) Delete(bucket string, key string) error {
+	_, err := b.db.HDel(ctx, bucket, key).Result()
+	return err
+}
+
+func (b *redisAdapter) Close() error {
+	return b.db.Close()
+}

manager/db_test.go

@@ -2,16 +2,166 @@ package manager
 
 import (
 	"encoding/json"
+	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
+	"sort"
 	"testing"
 	"time"
 
+	"github.com/alicebob/miniredis"
 	. "github.com/smartystreets/goconvey/convey"
 	. "github.com/tuna/tunasync/internal"
 )
 
+func SortMirrorStatus(status []MirrorStatus) {
+	sort.Slice(status, func(l, r int) bool {
+		return status[l].Name < status[r].Name
+	})
+}
+
+func DBAdapterTest(db dbAdapter) {
+	var err error
+	testWorkerIDs := []string{"test_worker1", "test_worker2"}
+	Convey("create worker", func() {
+		for _, id := range testWorkerIDs {
+			w := WorkerStatus{
+				ID:           id,
+				Token:        "token_" + id,
+				LastOnline:   time.Now(),
+				LastRegister: time.Now(),
+			}
+			w, err = db.CreateWorker(w)
+			So(err, ShouldBeNil)
+		}
+
+		Convey("get existent worker", func() {
+			_, err := db.GetWorker(testWorkerIDs[0])
+			So(err, ShouldBeNil)
+		})
+
+		Convey("list existent workers", func() {
+			ws, err := db.ListWorkers()
+			So(err, ShouldBeNil)
+			So(len(ws), ShouldEqual, 2)
+		})
+
+		Convey("get non-existent worker", func() {
+			_, err := db.GetWorker("invalid workerID")
+			So(err, ShouldNotBeNil)
+		})
+
+		Convey("delete existent worker", func() {
+			err := db.DeleteWorker(testWorkerIDs[0])
+			So(err, ShouldBeNil)
+			_, err = db.GetWorker(testWorkerIDs[0])
+			So(err, ShouldNotBeNil)
+			ws, err := db.ListWorkers()
+			So(err, ShouldBeNil)
+			So(len(ws), ShouldEqual, 1)
+		})
+
+		Convey("delete non-existent worker", func() {
+			err := db.DeleteWorker("invalid workerID")
+			So(err, ShouldNotBeNil)
+			ws, err := db.ListWorkers()
+			So(err, ShouldBeNil)
+			So(len(ws), ShouldEqual, 2)
+		})
+	})
+
+	Convey("update mirror status", func() {
+		status := []MirrorStatus{
+			MirrorStatus{
+				Name:        "arch-sync1",
+				Worker:      testWorkerIDs[0],
+				IsMaster:    true,
+				Status:      Success,
+				LastUpdate:  time.Now(),
+				LastStarted: time.Now().Add(-time.Minute),
+				LastEnded:   time.Now(),
+				Upstream:    "mirrors.tuna.tsinghua.edu.cn",
+				Size:        "3GB",
+			},
+			MirrorStatus{
+				Name:        "arch-sync2",
+				Worker:      testWorkerIDs[1],
+				IsMaster:    true,
+				Status:      Disabled,
+				LastUpdate:  time.Now().Add(-time.Hour),
+				LastStarted: time.Now().Add(-time.Minute),
+				LastEnded:   time.Now(),
+				Upstream:    "mirrors.tuna.tsinghua.edu.cn",
+				Size:        "4GB",
+			},
+			MirrorStatus{
+				Name:        "arch-sync3",
+				Worker:      testWorkerIDs[1],
+				IsMaster:    true,
+				Status:      Success,
+				LastUpdate:  time.Now().Add(-time.Minute),
+				LastStarted: time.Now().Add(-time.Second),
+				LastEnded:   time.Now(),
+				Upstream:    "mirrors.tuna.tsinghua.edu.cn",
+				Size:        "4GB",
+			},
+		}
+		SortMirrorStatus(status)
+
+		for _, s := range status {
+			_, err := db.UpdateMirrorStatus(s.Worker, s.Name, s)
+			So(err, ShouldBeNil)
+
+		}
+
+		Convey("get mirror status", func() {
+			m, err := db.GetMirrorStatus(testWorkerIDs[0], status[0].Name)
+			So(err, ShouldBeNil)
+			expectedJSON, err := json.Marshal(status[0])
+			So(err, ShouldBeNil)
+			actualJSON, err := json.Marshal(m)
+			So(err, ShouldBeNil)
+			So(string(actualJSON), ShouldEqual, string(expectedJSON))
+		})
+
+		Convey("list mirror status", func() {
+			ms, err := db.ListMirrorStatus(testWorkerIDs[0])
+			So(err, ShouldBeNil)
+			expectedJSON, err := json.Marshal([]MirrorStatus{status[0]})
+			So(err, ShouldBeNil)
+			actualJSON, err := json.Marshal(ms)
+			So(err, ShouldBeNil)
+			So(string(actualJSON), ShouldEqual, string(expectedJSON))
+		})
+
+		Convey("list all mirror status", func() {
+			ms, err := db.ListAllMirrorStatus()
+			So(err, ShouldBeNil)
+			SortMirrorStatus(ms)
+
+			expectedJSON, err := json.Marshal(status)
+			So(err, ShouldBeNil)
+			actualJSON, err := json.Marshal(ms)
+			So(err, ShouldBeNil)
+			So(string(actualJSON), ShouldEqual, string(expectedJSON))
+		})
+
+		Convey("flush disabled jobs", func() {
+			ms, err := db.ListAllMirrorStatus()
+			So(err, ShouldBeNil)
+			So(len(ms), ShouldEqual, 3)
+			err = db.FlushDisabledJobs()
+			So(err, ShouldBeNil)
+			ms, err = db.ListAllMirrorStatus()
+			So(err, ShouldBeNil)
+			So(len(ms), ShouldEqual, 2)
+		})
+
+	})
+	return
+}
+
 func TestBoltAdapter(t *testing.T) {
 	Convey("boltAdapter should work", t, func() {
 		tmpDir, err := ioutil.TempDir("", "tunasync")
@@ -28,138 +178,24 @@ func TestBoltAdapter(t *testing.T) {
 			So(err, ShouldBeNil)
 		}()
 
-		testWorkerIDs := []string{"test_worker1", "test_worker2"}
+		DBAdapterTest(boltDB)
-		Convey("create worker", func() {
+	})
-			for _, id := range testWorkerIDs {
-				w := WorkerStatus{
-					ID:         id,
-					Token:      "token_" + id,
-					LastOnline: time.Now(),
-				}
-				w, err = boltDB.CreateWorker(w)
-				So(err, ShouldBeNil)
-			}
 
-			Convey("get existent worker", func() {
+	Convey("redisAdapter should work", t, func() {
-				_, err := boltDB.GetWorker(testWorkerIDs[0])
+		mr, err := miniredis.Run()
-				So(err, ShouldBeNil)
+		So(err, ShouldBeNil)
-			})
 
-			Convey("list existent workers", func() {
+		addr := fmt.Sprintf("redis://%s", mr.Addr())
-				ws, err := boltDB.ListWorkers()
+		redisDB, err := makeDBAdapter("redis", addr)
-				So(err, ShouldBeNil)
+		So(err, ShouldBeNil)
-				So(len(ws), ShouldEqual, 2)
-			})
 
-			Convey("get non-existent worker", func() {
+		defer func() {
-				_, err := boltDB.GetWorker("invalid workerID")
+			// close redisDB
-				So(err, ShouldNotBeNil)
+			err := redisDB.Close()
-			})
+			So(err, ShouldBeNil)
-
+			mr.Close()
-			Convey("delete existent worker", func() {
+		}()
-				err := boltDB.DeleteWorker(testWorkerIDs[0])
-				So(err, ShouldBeNil)
-				_, err = boltDB.GetWorker(testWorkerIDs[0])
-				So(err, ShouldNotBeNil)
-				ws, err := boltDB.ListWorkers()
-				So(err, ShouldBeNil)
-				So(len(ws), ShouldEqual, 1)
-			})
-
-			Convey("delete non-existent worker", func() {
-				err := boltDB.DeleteWorker("invalid workerID")
-				So(err, ShouldNotBeNil)
-				ws, err := boltDB.ListWorkers()
-				So(err, ShouldBeNil)
-				So(len(ws), ShouldEqual, 2)
-			})
-		})
-
-		Convey("update mirror status", func() {
-			status := []MirrorStatus{
-				MirrorStatus{
-					Name:        "arch-sync1",
-					Worker:      testWorkerIDs[0],
-					IsMaster:    true,
-					Status:      Success,
-					LastUpdate:  time.Now(),
-					LastStarted: time.Now().Add(-time.Minute),
-					LastEnded:   time.Now(),
-					Upstream:    "mirrors.tuna.tsinghua.edu.cn",
-					Size:        "3GB",
-				},
-				MirrorStatus{
-					Name:        "arch-sync2",
-					Worker:      testWorkerIDs[1],
-					IsMaster:    true,
-					Status:      Disabled,
-					LastUpdate:  time.Now().Add(-time.Hour),
-					LastStarted: time.Now().Add(-time.Minute),
-					LastEnded:   time.Now(),
-					Upstream:    "mirrors.tuna.tsinghua.edu.cn",
-					Size:        "4GB",
-				},
-				MirrorStatus{
-					Name:        "arch-sync3",
-					Worker:      testWorkerIDs[1],
-					IsMaster:    true,
-					Status:      Success,
-					LastUpdate:  time.Now().Add(-time.Minute),
-					LastStarted: time.Now().Add(-time.Second),
-					LastEnded:   time.Now(),
-					Upstream:    "mirrors.tuna.tsinghua.edu.cn",
-					Size:        "4GB",
-				},
-			}
-
-			for _, s := range status {
-				_, err := boltDB.UpdateMirrorStatus(s.Worker, s.Name, s)
-				So(err, ShouldBeNil)
-
-			}
-
-			Convey("get mirror status", func() {
-				m, err := boltDB.GetMirrorStatus(testWorkerIDs[0], status[0].Name)
-				So(err, ShouldBeNil)
-				expectedJSON, err := json.Marshal(status[0])
-				So(err, ShouldBeNil)
-				actualJSON, err := json.Marshal(m)
-				So(err, ShouldBeNil)
-				So(string(actualJSON), ShouldEqual, string(expectedJSON))
-			})
-
-			Convey("list mirror status", func() {
-				ms, err := boltDB.ListMirrorStatus(testWorkerIDs[0])
-				So(err, ShouldBeNil)
-				expectedJSON, err := json.Marshal([]MirrorStatus{status[0]})
-				So(err, ShouldBeNil)
-				actualJSON, err := json.Marshal(ms)
-				So(err, ShouldBeNil)
-				So(string(actualJSON), ShouldEqual, string(expectedJSON))
-			})
-
-			Convey("list all mirror status", func() {
-				ms, err := boltDB.ListAllMirrorStatus()
-				So(err, ShouldBeNil)
-				expectedJSON, err := json.Marshal(status)
-				So(err, ShouldBeNil)
-				actualJSON, err := json.Marshal(ms)
-				So(err, ShouldBeNil)
-				So(string(actualJSON), ShouldEqual, string(expectedJSON))
-			})
-
-			Convey("flush disabled jobs", func() {
-				ms, err := boltDB.ListAllMirrorStatus()
-				So(err, ShouldBeNil)
-				So(len(ms), ShouldEqual, 3)
-				err = boltDB.FlushDisabledJobs()
-				So(err, ShouldBeNil)
-				ms, err = boltDB.ListAllMirrorStatus()
-				So(err, ShouldBeNil)
-				So(len(ms), ShouldEqual, 2)
-			})
-
-		})
 
+		DBAdapterTest(redisDB)
 	})
 }

manager/server.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"sync"
 	"time"
 
 	"github.com/gin-gonic/gin"
@@ -23,6 +24,7 @@ type Manager struct {
 	cfg        *Config
 	engine     *gin.Engine
 	adapter    dbAdapter
+	rwmu       sync.RWMutex
 	httpClient *http.Client
 }
 
@@ -127,9 +129,11 @@ func (s *Manager) Run() {
 	}
 }
 
-// listAllJobs repond with all jobs of specified workers
+// listAllJobs respond with all jobs of specified workers
 func (s *Manager) listAllJobs(c *gin.Context) {
+	s.rwmu.RLock()
 	mirrorStatusList, err := s.adapter.ListAllMirrorStatus()
+	s.rwmu.RUnlock()
 	if err != nil {
 		err := fmt.Errorf("failed to list all mirror status: %s",
 			err.Error(),
@@ -150,7 +154,9 @@ func (s *Manager) listAllJobs(c *gin.Context) {
 
 // flushDisabledJobs deletes all jobs that marks as deleted
 func (s *Manager) flushDisabledJobs(c *gin.Context) {
+	s.rwmu.Lock()
 	err := s.adapter.FlushDisabledJobs()
+	s.rwmu.Unlock()
 	if err != nil {
 		err := fmt.Errorf("failed to flush disabled jobs: %s",
 			err.Error(),
@@ -165,7 +171,9 @@ func (s *Manager) flushDisabledJobs(c *gin.Context) {
 // deleteWorker deletes one worker by id
 func (s *Manager) deleteWorker(c *gin.Context) {
 	workerID := c.Param("id")
+	s.rwmu.Lock()
 	err := s.adapter.DeleteWorker(workerID)
+	s.rwmu.Unlock()
 	if err != nil {
 		err := fmt.Errorf("failed to delete worker: %s",
 			err.Error(),
@@ -178,10 +186,12 @@ func (s *Manager) deleteWorker(c *gin.Context) {
 	c.JSON(http.StatusOK, gin.H{_infoKey: "deleted"})
 }
 
-// listWrokers respond with informations of all the workers
+// listWorkers respond with information of all the workers
 func (s *Manager) listWorkers(c *gin.Context) {
 	var workerInfos []WorkerStatus
+	s.rwmu.RLock()
 	workers, err := s.adapter.ListWorkers()
+	s.rwmu.RUnlock()
 	if err != nil {
 		err := fmt.Errorf("failed to list workers: %s",
 			err.Error(),
@@ -193,8 +203,11 @@ func (s *Manager) listWorkers(c *gin.Context) {
 	for _, w := range workers {
 		workerInfos = append(workerInfos,
 			WorkerStatus{
-				ID:         w.ID,
+				ID:           w.ID,
-				LastOnline: w.LastOnline,
+				URL:          w.URL,
+				Token:        "REDACTED",
+				LastOnline:   w.LastOnline,
+				LastRegister: w.LastRegister,
 			})
 	}
 	c.JSON(http.StatusOK, workerInfos)
@@ -205,6 +218,7 @@ func (s *Manager) registerWorker(c *gin.Context) {
 	var _worker WorkerStatus
 	c.BindJSON(&_worker)
 	_worker.LastOnline = time.Now()
+	_worker.LastRegister = time.Now()
 	newWorker, err := s.adapter.CreateWorker(_worker)
 	if err != nil {
 		err := fmt.Errorf("failed to register worker: %s",
@@ -223,7 +237,9 @@ func (s *Manager) registerWorker(c *gin.Context) {
 // listJobsOfWorker respond with all the jobs of the specified worker
 func (s *Manager) listJobsOfWorker(c *gin.Context) {
 	workerID := c.Param("id")
+	s.rwmu.RLock()
 	mirrorStatusList, err := s.adapter.ListMirrorStatus(workerID)
+	s.rwmu.RUnlock()
 	if err != nil {
 		err := fmt.Errorf("failed to list jobs of worker %s: %s",
 			workerID, err.Error(),
@@ -255,9 +271,12 @@ func (s *Manager) updateSchedulesOfWorker(c *gin.Context) {
 			)
 		}
 
+		s.rwmu.RLock()
+		s.adapter.RefreshWorker(workerID)
 		curStatus, err := s.adapter.GetMirrorStatus(workerID, mirrorName)
+		s.rwmu.RUnlock()
 		if err != nil {
-			fmt.Errorf("failed to get job %s of worker %s: %s",
+			logger.Errorf("failed to get job %s of worker %s: %s",
 				mirrorName, workerID, err.Error(),
 			)
 			continue
@@ -269,7 +288,9 @@ func (s *Manager) updateSchedulesOfWorker(c *gin.Context) {
 		}
 
 		curStatus.Scheduled = schedule.NextSchedule
+		s.rwmu.Lock()
 		_, err = s.adapter.UpdateMirrorStatus(workerID, mirrorName, curStatus)
+		s.rwmu.Unlock()
 		if err != nil {
 			err := fmt.Errorf("failed to update job %s of worker %s: %s",
 				mirrorName, workerID, err.Error(),
@@ -295,7 +316,10 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) {
 		)
 	}
 
+	s.rwmu.RLock()
+	s.adapter.RefreshWorker(workerID)
 	curStatus, _ := s.adapter.GetMirrorStatus(workerID, mirrorName)
+	s.rwmu.RUnlock()
 
 	curTime := time.Now()
 
@@ -331,7 +355,9 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) {
 		logger.Noticef("Job [%s] @<%s> %s", status.Name, status.Worker, status.Status)
 	}
 
+	s.rwmu.Lock()
 	newStatus, err := s.adapter.UpdateMirrorStatus(workerID, mirrorName, status)
+	s.rwmu.Unlock()
 	if err != nil {
 		err := fmt.Errorf("failed to update job %s of worker %s: %s",
 			mirrorName, workerID, err.Error(),
@@ -353,7 +379,10 @@ func (s *Manager) updateMirrorSize(c *gin.Context) {
 	c.BindJSON(&msg)
 
 	mirrorName := msg.Name
+	s.rwmu.RLock()
+	s.adapter.RefreshWorker(workerID)
 	status, err := s.adapter.GetMirrorStatus(workerID, mirrorName)
+	s.rwmu.RUnlock()
 	if err != nil {
 		logger.Errorf(
 			"Failed to get status of mirror %s @<%s>: %s",
@@ -370,7 +399,9 @@ func (s *Manager) updateMirrorSize(c *gin.Context) {
 
 	logger.Noticef("Mirror size of [%s] @<%s>: %s", status.Name, status.Worker, status.Size)
 
+	s.rwmu.Lock()
 	newStatus, err := s.adapter.UpdateMirrorStatus(workerID, mirrorName, status)
+	s.rwmu.Unlock()
 	if err != nil {
 		err := fmt.Errorf("failed to update job %s of worker %s: %s",
 			mirrorName, workerID, err.Error(),
@@ -393,7 +424,9 @@ func (s *Manager) handleClientCmd(c *gin.Context) {
 		return
 	}
 
+	s.rwmu.RLock()
 	w, err := s.adapter.GetWorker(workerID)
+	s.rwmu.RUnlock()
 	if err != nil {
 		err := fmt.Errorf("worker %s is not registered yet", workerID)
 		s.returnErrJSON(c, http.StatusBadRequest, err)
@@ -410,7 +443,9 @@ func (s *Manager) handleClientCmd(c *gin.Context) {
 
 	// update job status, even if the job did not disable successfully,
 	// this status should be set as disabled
+	s.rwmu.RLock()
 	curStat, _ := s.adapter.GetMirrorStatus(clientCmd.WorkerID, clientCmd.MirrorID)
+	s.rwmu.RUnlock()
 	changed := false
 	switch clientCmd.Cmd {
 	case CmdDisable:
@@ -421,7 +456,9 @@ func (s *Manager) handleClientCmd(c *gin.Context) {
 		changed = true
 	}
 	if changed {
+		s.rwmu.Lock()
 		s.adapter.UpdateMirrorStatus(clientCmd.WorkerID, clientCmd.MirrorID, curStat)
+		s.rwmu.Unlock()
 	}
 
 	logger.Noticef("Posting command '%s %s' to <%s>", clientCmd.Cmd, clientCmd.MirrorID, clientCmd.WorkerID)

manager/server_test.go

@@ -7,6 +7,7 @@ import (
 	"math/rand"
 	"net/http"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -64,6 +65,34 @@ func TestHTTPServer(t *testing.T) {
 			So(msg[_errorKey], ShouldEqual, fmt.Sprintf("failed to list jobs of worker %s: %s", _magicBadWorkerID, "database fail"))
 		})
 
+		Convey("when register multiple workers", func(ctx C) {
+			N := 10
+			var cnt uint32
+			for i := 0; i < N; i++ {
+				go func(id int) {
+					w := WorkerStatus{
+						ID: fmt.Sprintf("worker%d", id),
+					}
+					resp, err := PostJSON(baseURL+"/workers", w, nil)
+					ctx.So(err, ShouldBeNil)
+					ctx.So(resp.StatusCode, ShouldEqual, http.StatusOK)
+					atomic.AddUint32(&cnt, 1)
+				}(i)
+			}
+			time.Sleep(2 * time.Second)
+			So(cnt, ShouldEqual, N)
+
+			Convey("list all workers", func(ctx C) {
+				resp, err := http.Get(baseURL + "/workers")
+				So(err, ShouldBeNil)
+				defer resp.Body.Close()
+				var actualResponseObj []WorkerStatus
+				err = json.NewDecoder(resp.Body).Decode(&actualResponseObj)
+				So(err, ShouldBeNil)
+				So(len(actualResponseObj), ShouldEqual, N+1)
+			})
+		})
+
 		Convey("when register a worker", func(ctx C) {
 			w := WorkerStatus{
 				ID: "test_worker1",
@@ -433,6 +462,15 @@ func (b *mockDBAdapter) CreateWorker(w WorkerStatus) (WorkerStatus, error) {
 	return w, nil
 }
 
+func (b *mockDBAdapter) RefreshWorker(workerID string) (w WorkerStatus, err error) {
+	w, err = b.GetWorker(workerID)
+	if err == nil {
+		w.LastOnline = time.Now()
+		w, err = b.CreateWorker(w)
+	}
+	return w, err
+}
+
 func (b *mockDBAdapter) GetMirrorStatus(workerID, mirrorID string) (MirrorStatus, error) {
 	id := mirrorID + "/" + workerID
 	status, ok := b.statusStore[id]

worker/btrfs_snapshot_hook.go

@@ -1,3 +1,5 @@
+// +build linux
+
 package worker
 
 import (

worker/btrfs_snapshot_hook_nolinux.go

@@ -0,0 +1,30 @@
+// +build !linux
+
+package worker
+
+type btrfsSnapshotHook struct {
+}
+
+func newBtrfsSnapshotHook(provider mirrorProvider, snapshotPath string, mirror mirrorConfig) *btrfsSnapshotHook {
+	return &btrfsSnapshotHook{}
+}
+
+func (h *btrfsSnapshotHook) postExec() error {
+	return nil
+}
+
+func (h *btrfsSnapshotHook) postFail() error {
+	return nil
+}
+
+func (h *btrfsSnapshotHook) postSuccess() error {
+	return nil
+}
+
+func (h *btrfsSnapshotHook) preExec() error {
+	return nil
+}
+
+func (h *btrfsSnapshotHook) preJob() error {
+	return nil
+}

worker/provider_test.go

@@ -91,7 +91,7 @@ exit 0
 					"Done\n",
 				targetDir,
 				fmt.Sprintf(
-					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
+					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --filter risk .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
 						"--timeout=120 -6 %s %s",
 					provider.upstreamURL, provider.WorkingDir(),
@@ -190,7 +190,7 @@ exit 0
 					"Done\n",
 				targetDir,
 				fmt.Sprintf(
-					"%s %s %s -aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
+					"%s %s %s -aHvh --no-o --no-g --stats --exclude .~tmp~/ --filter risk .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
 						"--timeout=30 -4 --delete-excluded %s %s",
 					provider.username, provider.password, proxyAddr,
@@ -613,14 +613,14 @@ exit 0
 					"Done\n",
 				targetDir,
 				fmt.Sprintf(
-					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --safe-links "+
+					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --filter risk .~tmp~/ --safe-links "+
-						"--exclude dists/ --timeout=30 -6 "+
+						"--include=*.diff/ --exclude=*.diff/Index --exclude=Packages* --exclude=Sources* --exclude=Release* --exclude=InRelease --include=i18n/by-hash --exclude=i18n/* --exclude=ls-lR* --timeout=30 -6 "+
 						"--exclude-from %s %s %s",
 					provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 				),
 				targetDir,
 				fmt.Sprintf(
-					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ "+
+					"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --filter risk .~tmp~/ "+
 						"--delete --delete-after --delay-updates --safe-links "+
 						"--delete-excluded --cache --timeout=30 -6 --exclude-from %s %s %s",
 					provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
@@ -655,8 +655,8 @@ exit 0
 			So(err, ShouldBeNil)
 
 			expectedOutput := fmt.Sprintf(
-				"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --safe-links "+
+				"-aHvh --no-o --no-g --stats --exclude .~tmp~/ --filter risk .~tmp~/ --safe-links "+
-					"--exclude dists/ --timeout=30 -6 "+
+					"--include=*.diff/ --exclude=*.diff/Index --exclude=Packages* --exclude=Sources* --exclude=Release* --exclude=InRelease --include=i18n/by-hash --exclude=i18n/* --exclude=ls-lR* --timeout=30 -6 "+
 					"--exclude-from %s %s %s\n",
 				provider.excludeFile, provider.upstreamURL, provider.WorkingDir(),
 			)

worker/rsync_provider.go

@@ -67,7 +67,7 @@ func newRsyncProvider(c rsyncConfig) (*rsyncProvider, error) {
 
 	options := []string{
 		"-aHvh", "--no-o", "--no-g", "--stats",
-		"--exclude", ".~tmp~/",
+		"--exclude", ".~tmp~/", "--filter" , "risk .~tmp~/",
 		"--delete", "--delete-after", "--delay-updates",
 		"--safe-links",
 	}

worker/runner.go

@@ -149,10 +149,10 @@ func (c *cmdJob) Terminate() error {
 	select {
 	case <-time.After(2 * time.Second):
 		unix.Kill(c.cmd.Process.Pid, syscall.SIGKILL)
-		return errors.New("SIGTERM failed to kill the job")
+		logger.Warningf("SIGTERM failed to kill the job in 2s. SIGKILL sent")
 	case <-c.finished:
-		return nil
 	}
+	return nil
 }
 
 // Copied from go-sh

worker/two_stage_rsync_provider.go

@@ -34,11 +34,12 @@ type twoStageRsyncProvider struct {
 	dataSize      string
 }
 
+// ref: https://salsa.debian.org/mirror-team/archvsync/-/blob/master/bin/ftpsync#L431
 var rsyncStage1Profiles = map[string]([]string){
-	"debian": []string{"dists/"},
+	"debian": []string{"--include=*.diff/", "--exclude=*.diff/Index", "--exclude=Packages*", "--exclude=Sources*", "--exclude=Release*", "--exclude=InRelease", "--include=i18n/by-hash", "--exclude=i18n/*", "--exclude=ls-lR*"},
 	"debian-oldstyle": []string{
-		"Packages*", "Sources*", "Release*",
+		"--exclude=Packages*", "--exclude=Sources*", "--exclude=Release*",
-		"InRelease", "i18n/*", "ls-lR*", "dep11/*",
+		"--exclude=InRelease", "--exclude=i18n/*", "--exclude=ls-lR*", "--exclude=dep11/*",
 	},
 }
 
@@ -62,12 +63,12 @@ func newTwoStageRsyncProvider(c twoStageRsyncConfig) (*twoStageRsyncProvider, er
 		twoStageRsyncConfig: c,
 		stage1Options: []string{
 			"-aHvh", "--no-o", "--no-g", "--stats",
-			"--exclude", ".~tmp~/",
+			"--exclude", ".~tmp~/", "--filter", "risk .~tmp~/",
 			"--safe-links",
 		},
 		stage2Options: []string{
 			"-aHvh", "--no-o", "--no-g", "--stats",
-			"--exclude", ".~tmp~/",
+			"--exclude", ".~tmp~/", "--filter", "risk .~tmp~/",
 			"--delete", "--delete-after", "--delay-updates",
 			"--safe-links",
 		},
@@ -109,12 +110,12 @@ func (p *twoStageRsyncProvider) Options(stage int) ([]string, error) {
 	var options []string
 	if stage == 1 {
 		options = append(options, p.stage1Options...)
-		stage1Excludes, ok := rsyncStage1Profiles[p.stage1Profile]
+		stage1Profile, ok := rsyncStage1Profiles[p.stage1Profile]
 		if !ok {
 			return nil, errors.New("Invalid Stage 1 Profile")
 		}
-		for _, exc := range stage1Excludes {
+		for _, exc := range stage1Profile {
-			options = append(options, "--exclude", exc)
+			options = append(options, exc)
 		}
 
 	} else if stage == 2 {

worker/worker.go

@@ -61,7 +61,7 @@ func NewTUNASyncWorker(cfg *Config) *Worker {
 
 // Run runs worker forever
 func (w *Worker) Run() {
-	w.registorWorker()
+	w.registerWorker()
 	go w.runHTTPServer()
 	w.runSchedule()
 }
@@ -393,7 +393,7 @@ func (w *Worker) URL() string {
 	return fmt.Sprintf("%s://%s:%d/", proto, w.cfg.Server.Hostname, w.cfg.Server.Port)
 }
 
-func (w *Worker) registorWorker() {
+func (w *Worker) registerWorker() {
 	msg := WorkerStatus{
 		ID:  w.Name(),
 		URL: w.URL(),
@@ -402,8 +402,17 @@ func (w *Worker) registorWorker() {
 	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 {
+		for retry := 10; retry > 0; {
-			logger.Errorf("Failed to register worker")
+			if _, err := PostJSON(url, msg, w.httpClient); err != nil {
+				logger.Errorf("Failed to register worker")
+				retry--
+				if retry > 0 {
+					time.Sleep(1 * time.Second)
+					logger.Noticef("Retrying... (%d)", retry)
+				}
+			} else {
+				break
+			}
 		}
 	}
 }

worker/worker_test.go

@@ -25,6 +25,7 @@ func makeMockManagerServer(recvData chan interface{}) *gin.Engine {
 		var _worker WorkerStatus
 		c.BindJSON(&_worker)
 		_worker.LastOnline = time.Now()
+		_worker.LastRegister = time.Now()
 		recvData <- _worker
 		c.JSON(http.StatusOK, _worker)
 	})