feat(manager): implemented restful API for updating mirror size

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"`

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

manager/server.go

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

manager/server_test.go

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