yacloud/tunasync
1
0
Fork 0
mirror of https://github.com/tuna/tunasync.git synced 2025-04-20 20:22:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #60 from tuna/dev

Browse Source 
Dev
This commit is contained in:
bigeagle 2016-12-19 01:10:38 +08:00 committed by GitHub
commit 93194cde2e
9 changed files with 173 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
cmd/tunasynctl/tunasynctl.go
View File

@ -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 <worker-id> <mirror> <size>", 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",

2
internal/msg.go
View File

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

2
manager/db.go
View File

@ -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

64
manager/server.go
View File

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

43
manager/server_test.go
View File

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

1
worker/cmd_provider.go
View File

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

11
worker/docker.go
View File

@ -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

1
worker/exec_post_hook.go
View File

@ -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,
}

6
worker/runner.go
View File

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