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) {