From d4e07a7b29cd812f2840cdcf4b7d6988b72b3896 Mon Sep 17 00:00:00 2001 From: bigeagle Date: Sun, 18 Dec 2016 14:28:32 +0800 Subject: [PATCH 1/5] fix(worker): keep the same working dir inside and outside of docker --- worker/docker.go | 11 ++++------- 1 file changed, 4 insertions(+), 7 deletions(-) diff --git a/worker/docker.go b/worker/docker.go index a94cbe3..852e67e 100644 --- a/worker/docker.go +++ b/worker/docker.go @@ -32,6 +32,7 @@ func newDockerHook(p mirrorProvider, gCfg dockerConfig, mCfg mirrorConfig) *dock func (d *dockerHook) preExec() error { p := d.provider + logDir := p.LogDir() logFile := p.LogFile() workingDir := p.WorkingDir() @@ -42,17 +43,13 @@ func (d *dockerHook) preExec() error { } } - logFileNew := "/log_latest" - workingDirNew := "/data" - // Override workingDir ctx := p.EnterContext() - ctx.Set(_WorkingDirKey, workingDirNew) - ctx.Set(_LogFileKey+":docker", logFileNew) ctx.Set( "volumes", []string{ - fmt.Sprintf("%s:%s", logFile, logFileNew), - fmt.Sprintf("%s:%s", workingDir, workingDirNew), + fmt.Sprintf("%s:%s", logDir, logDir), + fmt.Sprintf("%s:%s", logFile, logFile), + fmt.Sprintf("%s:%s", workingDir, workingDir), }, ) return nil From d5a438462fe838bc4059b56963f5b2ab7dca2e50 Mon Sep 17 00:00:00 2001 From: bigeagle Date: Sun, 18 Dec 2016 14:28:48 +0800 Subject: [PATCH 2/5] feat(worker): map current uid and gid to docker --- worker/runner.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/worker/runner.go b/worker/runner.go index 8d45f99..05a6f6b 100644 --- a/worker/runner.go +++ b/worker/runner.go @@ -41,13 +41,17 @@ func newCmdJob(provider mirrorProvider, cmdAndArgs []string, workingDir string, "--name", d.Name(), "-w", workingDir, } + // specify user + args = append( + args, "-u", + fmt.Sprintf("%d:%d", os.Getuid(), os.Getgid()), + ) // add volumes for _, vol := range d.Volumes() { logger.Debugf("volume: %s", vol) args = append(args, "-v", vol) } // set env - env["TUNASYNC_LOG_FILE"] = d.LogFile() for k, v := range env { kv := fmt.Sprintf("%s=%s", k, v) args = append(args, "-e", kv) From 939abaef9b29f7d5713d5cd921128800d7b42056 Mon Sep 17 00:00:00 2001 From: bigeagle Date: Sun, 18 Dec 2016 20:41:26 +0800 Subject: [PATCH 3/5] feat(worker): TUNASYNC_LOG_DIR environment variable --- worker/cmd_provider.go | 1 + worker/exec_post_hook.go | 1 + 2 files changed, 2 insertions(+) diff --git a/worker/cmd_provider.go b/worker/cmd_provider.go index 8ce3d70..2274cec 100644 --- a/worker/cmd_provider.go +++ b/worker/cmd_provider.go @@ -64,6 +64,7 @@ func (p *cmdProvider) Start() error { "TUNASYNC_MIRROR_NAME": p.Name(), "TUNASYNC_WORKING_DIR": p.WorkingDir(), "TUNASYNC_UPSTREAM_URL": p.upstreamURL, + "TUNASYNC_LOG_DIR": p.LogDir(), "TUNASYNC_LOG_FILE": p.LogFile(), } for k, v := range p.env { diff --git a/worker/exec_post_hook.go b/worker/exec_post_hook.go index 16e5d16..22a9ec6 100644 --- a/worker/exec_post_hook.go +++ b/worker/exec_post_hook.go @@ -71,6 +71,7 @@ func (h *execPostHook) Do() error { "TUNASYNC_MIRROR_NAME": p.Name(), "TUNASYNC_WORKING_DIR": p.WorkingDir(), "TUNASYNC_UPSTREAM_URL": p.Upstream(), + "TUNASYNC_LOG_DIR": p.LogDir(), "TUNASYNC_LOG_FILE": p.LogFile(), "TUNASYNC_JOB_EXIT_STATUS": exitStatus, } From 4c6a407c176f788ca67ec09364ccccffb134b482 Mon Sep 17 00:00:00 2001 From: bigeagle Date: Sun, 18 Dec 2016 23:06:08 +0800 Subject: [PATCH 4/5] feat(manager): implemented restful API for updating mirror size --- internal/msg.go | 2 +- manager/db.go | 2 +- manager/server.go | 64 ++++++++++++++++++++++++++++++++++++------ manager/server_test.go | 43 +++++++++++++++++++++++++++- 4 files changed, 99 insertions(+), 12 deletions(-) diff --git a/internal/msg.go b/internal/msg.go index 7f9db65..15791b9 100644 --- a/internal/msg.go +++ b/internal/msg.go @@ -5,7 +5,7 @@ import ( "time" ) -// A StatusUpdateMsg represents a msg when +// A MirrorStatus represents a msg when // a worker has done syncing type MirrorStatus struct { Name string `json:"name"` diff --git a/manager/db.go b/manager/db.go index afbb894..a88f26e 100644 --- a/manager/db.go +++ b/manager/db.go @@ -125,7 +125,7 @@ func (b *boltAdapter) GetMirrorStatus(workerID, mirrorID string) (m MirrorStatus bucket := tx.Bucket([]byte(_statusBucketKey)) v := bucket.Get([]byte(id)) if v == nil { - return fmt.Errorf("no mirror %s exists in worker %s", mirrorID, workerID) + return fmt.Errorf("no mirror '%s' exists in worker '%s'", mirrorID, workerID) } err := json.Unmarshal(v, &m) return err diff --git a/manager/server.go b/manager/server.go index 917c58b..05c56bf 100644 --- a/manager/server.go +++ b/manager/server.go @@ -1,6 +1,7 @@ package manager import ( + "errors" "fmt" "net/http" "time" @@ -87,6 +88,7 @@ func GetTUNASyncManager(cfg *Config) *Manager { workerValidateGroup.GET(":id/jobs", s.listJobsOfWorker) // post job status workerValidateGroup.POST(":id/jobs/:job", s.updateJobOfWorker) + workerValidateGroup.POST(":id/jobs/:job/size", s.updateMirrorSize) } // for tunasynctl to post commands @@ -225,6 +227,12 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) { var status MirrorStatus c.BindJSON(&status) mirrorName := status.Name + if len(mirrorName) == 0 { + s.returnErrJSON( + c, http.StatusBadRequest, + errors.New("Mirror Name should not be empty"), + ) + } curStatus, _ := s.adapter.GetMirrorStatus(workerID, mirrorName) @@ -235,20 +243,19 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) { status.LastUpdate = curStatus.LastUpdate } + // Only message with meaningful size updates the mirror size + if len(curStatus.Size) > 0 && curStatus.Size != "unknown" { + if len(status.Size) == 0 || status.Size == "unknown" { + status.Size = curStatus.Size + } + } + // for logging switch status.Status { - case Success: - logger.Noticef("Job [%s] @<%s> success", status.Name, status.Worker) - case Failed: - logger.Warningf("Job [%s] @<%s> failed", status.Name, status.Worker) case Syncing: logger.Noticef("Job [%s] @<%s> starts syncing", status.Name, status.Worker) - case Disabled: - logger.Noticef("Job [%s] @<%s> disabled", status.Name, status.Worker) - case Paused: - logger.Noticef("Job [%s] @<%s> paused", status.Name, status.Worker) default: - logger.Infof("Job [%s] @<%s> status: %s", status.Name, status.Worker, status.Status) + logger.Noticef("Job [%s] @<%s> %s", status.Name, status.Worker, status.Status) } newStatus, err := s.adapter.UpdateMirrorStatus(workerID, mirrorName, status) @@ -263,6 +270,45 @@ func (s *Manager) updateJobOfWorker(c *gin.Context) { c.JSON(http.StatusOK, newStatus) } +func (s *Manager) updateMirrorSize(c *gin.Context) { + workerID := c.Param("id") + type SizeMsg struct { + Name string `json:"name"` + Size string `json:"size"` + } + var msg SizeMsg + c.BindJSON(&msg) + + mirrorName := msg.Name + status, err := s.adapter.GetMirrorStatus(workerID, mirrorName) + if err != nil { + logger.Errorf( + "Failed to get status of mirror %s @<%s>: %s", + mirrorName, workerID, err.Error(), + ) + s.returnErrJSON(c, http.StatusInternalServerError, err) + return + } + + // Only message with meaningful size updates the mirror size + if len(msg.Size) > 0 || msg.Size != "unknown" { + status.Size = msg.Size + } + + logger.Noticef("Mirror size of [%s] @<%s>: %s", status.Name, status.Worker, status.Size) + + newStatus, err := s.adapter.UpdateMirrorStatus(workerID, mirrorName, status) + if err != nil { + err := fmt.Errorf("failed to update job %s of worker %s: %s", + mirrorName, workerID, err.Error(), + ) + c.Error(err) + s.returnErrJSON(c, http.StatusInternalServerError, err) + return + } + c.JSON(http.StatusOK, newStatus) +} + func (s *Manager) handleClientCmd(c *gin.Context) { var clientCmd ClientCmd c.BindJSON(&clientCmd) diff --git a/manager/server_test.go b/manager/server_test.go index 748a323..c1122b0 100644 --- a/manager/server_test.go +++ b/manager/server_test.go @@ -99,7 +99,7 @@ func TestHTTPServer(t *testing.T) { IsMaster: true, Status: Success, Upstream: "mirrors.tuna.tsinghua.edu.cn", - Size: "3GB", + Size: "unknown", } resp, err := PostJSON(fmt.Sprintf("%s/workers/%s/jobs/%s", baseURL, status.Worker, status.Name), status, nil) defer resp.Body.Close() @@ -139,6 +139,47 @@ func TestHTTPServer(t *testing.T) { So(time.Now().Sub(m.LastUpdate.Time), ShouldBeLessThan, 1*time.Second) }) + + Convey("Update size of a valid mirror", func(ctx C) { + msg := struct { + Name string `json:"name"` + Size string `json:"size"` + }{status.Name, "5GB"} + + url := fmt.Sprintf("%s/workers/%s/jobs/%s/size", baseURL, status.Worker, status.Name) + resp, err := PostJSON(url, msg, nil) + So(err, ShouldBeNil) + So(resp.StatusCode, ShouldEqual, http.StatusOK) + + Convey("Get new size of a mirror", func(ctx C) { + var ms []MirrorStatus + resp, err := GetJSON(baseURL+"/workers/test_worker1/jobs", &ms, nil) + + So(err, ShouldBeNil) + So(resp.StatusCode, ShouldEqual, http.StatusOK) + // err = json.NewDecoder(resp.Body).Decode(&mirrorStatusList) + m := ms[0] + So(m.Name, ShouldEqual, status.Name) + So(m.Worker, ShouldEqual, status.Worker) + So(m.Status, ShouldEqual, status.Status) + So(m.Upstream, ShouldEqual, status.Upstream) + So(m.Size, ShouldEqual, "5GB") + So(m.IsMaster, ShouldEqual, status.IsMaster) + So(time.Now().Sub(m.LastUpdate), ShouldBeLessThan, 1*time.Second) + }) + }) + + Convey("Update size of an invalid mirror", func(ctx C) { + msg := struct { + Name string `json:"name"` + Size string `json:"size"` + }{"Invalid mirror", "5GB"} + + url := fmt.Sprintf("%s/workers/%s/jobs/%s/size", baseURL, status.Worker, status.Name) + resp, err := PostJSON(url, msg, nil) + So(err, ShouldBeNil) + So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError) + }) }) Convey("update mirror status of an inexisted worker", func(ctx C) { From aa4c31a32be8c8174144963d0db2d25dd52d13c5 Mon Sep 17 00:00:00 2001 From: bigeagle Date: Sun, 18 Dec 2016 23:30:41 +0800 Subject: [PATCH 5/5] feat(tunasynctl): implemented 'set-size' command to update a mirror size --- cmd/tunasynctl/tunasynctl.go | 63 ++++++++++++++++++++++++++++++++++++ 1 file changed, 63 insertions(+) diff --git a/cmd/tunasynctl/tunasynctl.go b/cmd/tunasynctl/tunasynctl.go index 36a18a6..ea12dcf 100644 --- a/cmd/tunasynctl/tunasynctl.go +++ b/cmd/tunasynctl/tunasynctl.go @@ -186,6 +186,57 @@ func listJobs(c *cli.Context) error { return nil } +func updateMirrorSize(c *cli.Context) error { + args := c.Args() + if len(args) != 2 { + return cli.NewExitError("Usage: tunasynctl -w ", 1) + } + workerID := c.String("worker") + mirrorID := args.Get(0) + mirrorSize := args.Get(1) + + msg := struct { + Name string `json:"name"` + Size string `json:"size"` + }{ + Name: mirrorID, + Size: mirrorSize, + } + + url := fmt.Sprintf( + "%s/workers/%s/jobs/%s/size", baseURL, workerID, mirrorID, + ) + + resp, err := tunasync.PostJSON(url, msg, client) + if err != nil { + return cli.NewExitError( + fmt.Sprintf("Failed to send request to manager: %s", + err.Error()), + 1) + } + defer resp.Body.Close() + body, _ := ioutil.ReadAll(resp.Body) + if resp.StatusCode != http.StatusOK { + return cli.NewExitError( + fmt.Sprintf("Manager failed to update mirror size: %s", body), 1, + ) + } + + var status tunasync.MirrorStatus + json.Unmarshal(body, &status) + if status.Size != mirrorSize { + return cli.NewExitError( + fmt.Sprintf( + "Mirror size error, expecting %s, manager returned %s", + mirrorSize, status.Size, + ), 1, + ) + } + + logger.Infof("Successfully updated mirror size to %s", mirrorSize) + return nil +} + func flushDisabledJobs(c *cli.Context) error { req, err := http.NewRequest("DELETE", baseURL+flushDisabledPath, nil) if err != nil { @@ -385,6 +436,18 @@ func main() { Flags: commonFlags, Action: initializeWrapper(listWorkers), }, + { + Name: "set-size", + Usage: "Set mirror size", + Flags: append( + commonFlags, + cli.StringFlag{ + Name: "worker, w", + Usage: "specify worker-id of the mirror job", + }, + ), + Action: initializeWrapper(updateMirrorSize), + }, { Name: "start", Usage: "Start a job",