Upstream/harbor
1
0
Fork 0
mirror of https://github.com/goharbor/harbor.git synced 2024-06-29 00:05:09 +02:00
Code Issues Projects Releases Wiki Activity

Merge branch 'job-service' of https://github.com/vmware/harbor into job-service

Browse Source
This commit is contained in:
Tan Jiang 2016-05-24 13:16:24 +08:00
commit 6fcb532da3
9 changed files with 75 additions and 60 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
Deploy/db/registry.sql
View File

@ -138,17 +138,6 @@ create table replication_job (
update_time timestamp default CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP,
PRIMARY KEY (id)
);
create table job_log (
log_id int NOT NULL AUTO_INCREMENT,
job_id int NOT NULL,
level varchar(64) NOT NULL,
message text,
creation_time timestamp,
update_time timestamp,
PRIMARY KEY (log_id),
FOREIGN KEY (job_id) REFERENCES replication_job (id)
);
create table properties (
k varchar(64) NOT NULL,

13
api/jobs/replication.go
View File

@ -6,6 +6,7 @@ import (
"github.com/vmware/harbor/api"
"github.com/vmware/harbor/dao"
"github.com/vmware/harbor/job"
"github.com/vmware/harbor/job/utils"
"github.com/vmware/harbor/models"
"github.com/vmware/harbor/utils/log"
"io/ioutil"
@ -91,6 +92,18 @@ func (rj *ReplicationJob) HandleAction() {
job.WorkerPool.StopJobs(jobIDList)
}
func (rj *ReplicationJob) GetLog() {
idStr := rj.Ctx.Input.Param(":id")
jid, err := strconv.ParseInt(idStr, 10, 64)
if err != nil {
log.Errorf("Error parsing job id: %s, error: %v", idStr, err)
rj.RenderError(http.StatusBadRequest, "Invalid job id")
return
}
logFile := utils.GetJobLogPath(jid)
rj.Ctx.Output.Download(logFile)
}
// calls the api from UI to get repo list
func getRepoList(projectID int64) ([]string, error) {
uiURL := os.Getenv("UI_URL")

25
job/config/config.go
View File

@ -1,6 +1,7 @@
package config
import (
"fmt"
"os"
"strconv"
@ -11,6 +12,7 @@ const defaultMaxWorkers int = 10
var maxJobWorkers int
var localRegURL string
var logDir string
func init() {
maxWorkersEnv := os.Getenv("MAX_JOB_WORKERS")
@ -26,8 +28,27 @@ func init() {
localRegURL = "http://registry:5000/"
}
logDir = os.Getenv("LOG_DIR")
if len(logDir) == 0 {
logDir = "/var/log"
}
f, err := os.Open(logDir)
defer f.Close()
if err != nil {
panic(err)
}
finfo, err := f.Stat()
if err != nil {
panic(err)
}
if !finfo.IsDir() {
panic(fmt.Sprintf("%s is not a direcotry", logDir))
}
log.Debugf("config: maxJobWorkers: %d", maxJobWorkers)
log.Debugf("config: localRegURL: %s", localRegURL)
log.Debugf("config: logDir: %s", logDir)
}
func MaxJobWorkers() int {
@ -37,3 +58,7 @@ func MaxJobWorkers() int {
func LocalRegURL() string {
return localRegURL
}
func LogDir() string {
return logDir
}

6
job/replication/statehandlers.go
View File

@ -26,8 +26,8 @@ import (
"github.com/docker/distribution"
"github.com/docker/distribution/manifest/schema1"
"github.com/docker/distribution/manifest/schema2"
"github.com/vmware/harbor/job/utils"
"github.com/vmware/harbor/models"
"github.com/vmware/harbor/utils/log"
"github.com/vmware/harbor/utils/registry"
"github.com/vmware/harbor/utils/registry/auth"
)
@ -64,13 +64,13 @@ type BaseHandler struct {
blobsExistence map[string]bool //key: digest of blob, value: existence
logger *utils.Logger
logger *log.Logger
}
// InitBaseHandler initializes a BaseHandler: creating clients for source and destination registry,
// listing tags of the repository if parameter tags is nil.
func InitBaseHandler(repository, srcURL, srcSecretKey,
dstURL, dstUsr, dstPwd string, tags []string, logger *utils.Logger) (*BaseHandler, error) {
dstURL, dstUsr, dstPwd string, tags []string, logger *log.Logger) (*BaseHandler, error) {
logger.Infof("initializing: repository: %s, tags: %v, source URL: %s, destination URL: %s, destination user: %s",
repository, tags, srcURL, dstURL, dstUsr)

5
job/statehandlers.go
View File

@ -4,7 +4,6 @@ import (
"time"
"github.com/vmware/harbor/dao"
"github.com/vmware/harbor/job/utils"
"github.com/vmware/harbor/models"
"github.com/vmware/harbor/utils/log"
)
@ -51,7 +50,7 @@ func (su StatusUpdater) Enter() (string, error) {
type ImgPuller struct {
DummyHandler
img string
logger utils.Logger
logger *log.Logger
}
func (ip ImgPuller) Enter() (string, error) {
@ -64,7 +63,7 @@ func (ip ImgPuller) Enter() (string, error) {
type ImgPusher struct {
DummyHandler
targetURL string
logger utils.Logger
logger *log.Logger
}
func (ip ImgPusher) Enter() (string, error) {

6
job/statemachine.go
View File

@ -31,7 +31,7 @@ type JobSM struct {
Transitions map[string]map[string]struct{}
Handlers map[string]StateHandler
desiredState string
Logger utils.Logger
Logger *log.Logger
Parms *RepJobParm
lock *sync.Mutex
}
@ -200,7 +200,7 @@ func (sm *JobSM) Reset(jid int64) error {
sm.Parms.TargetUsername = target.Username
sm.Parms.TargetPassword = target.Password
//init states handlers
sm.Logger = utils.Logger{sm.JobID}
sm.Logger = utils.NewLogger(sm.JobID)
sm.CurrentState = models.JobPending
sm.AddTransition(models.JobPending, models.JobRunning, StatusUpdater{DummyHandler{JobID: sm.JobID}, models.JobRunning})
@ -226,7 +226,7 @@ func (sm *JobSM) Reset(jid int64) error {
func addImgOutTransition(sm *JobSM) error {
base, err := replication.InitBaseHandler(sm.Parms.Repository, sm.Parms.LocalRegURL, "",
sm.Parms.TargetURL, sm.Parms.TargetUsername, sm.Parms.TargetPassword,
nil, &sm.Logger)
nil, sm.Logger)
if err != nil {
return err
}

38
job/utils/logger.go
View File

@ -3,37 +3,23 @@ package utils
import (
"fmt"
"github.com/vmware/harbor/dao"
"github.com/vmware/harbor/job/config"
"github.com/vmware/harbor/utils/log"
"os"
"path/filepath"
)
const (
INFO = "info"
WARN = "warning"
ERR = "error"
)
type Logger struct {
ID int64
}
func (l *Logger) Infof(format string, v ...interface{}) {
err := dao.AddJobLog(l.ID, INFO, fmt.Sprintf(format, v...))
func NewLogger(jobID int64) *log.Logger {
logFile := GetJobLogPath(jobID)
f, err := os.OpenFile(logFile, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0660)
if err != nil {
log.Warningf("Failed to add job log, id: %d, error: %v", l.ID, err)
log.Errorf("Failed to open log file %s, the log of job %d will be printed to standard output", logFile, jobID)
f = os.Stdout
}
return log.New(f, log.NewTextFormatter(), log.InfoLevel)
}
func (l *Logger) Warningf(format string, v ...interface{}) {
err := dao.AddJobLog(l.ID, WARN, fmt.Sprintf(format, v...))
if err != nil {
log.Warningf("Failed to add job log, id: %d, error: %v", l.ID, err)
}
}
func (l *Logger) Errorf(format string, v ...interface{}) {
err := dao.AddJobLog(l.ID, ERR, fmt.Sprintf(format, v...))
if err != nil {
log.Warningf("Failed to add job log, id: %d, error: %v", l.ID, err)
}
func GetJobLogPath(jobID int64) string {
fn := fmt.Sprintf("job_%d.log", jobID)
return filepath.Join(config.LogDir(), fn)
}

1
jobservice/router.go
View File

@ -8,5 +8,6 @@ import (
func initRouters() {
beego.Router("/api/jobs/replication", &api.ReplicationJob{})
beego.Router("/api/jobs/replication/:id/log", &api.ReplicationJob{}, "get:GetLog")
beego.Router("/api/jobs/replication/actions", &api.ReplicationJob{}, "post:HandleAction")
}

30
utils/log/logger.go
View File

@ -27,7 +27,7 @@ import (
var logger = New(os.Stdout, NewTextFormatter(), WarningLevel)
func init() {
logger.callDepth = 3
logger.callDepth = 4
// TODO add item in configuaration file
lvl := os.Getenv("LOG_LEVEL")
@ -52,6 +52,7 @@ type Logger struct {
fmtter Formatter
lvl Level
callDepth int
skipLine bool
mu sync.Mutex
}
@ -61,7 +62,7 @@ func New(out io.Writer, fmtter Formatter, lvl Level) *Logger {
out: out,
fmtter: fmtter,
lvl: lvl,
callDepth: 2,
callDepth: 3,
}
}
@ -121,8 +122,7 @@ func (l *Logger) output(record *Record) (err error) {
// Debug ...
func (l *Logger) Debug(v ...interface{}) {
if l.lvl <= DebugLevel {
line := line(l.callDepth)
record := NewRecord(time.Now(), fmt.Sprint(v...), line, DebugLevel)
record := NewRecord(time.Now(), fmt.Sprint(v...), l.getLine(), DebugLevel)
l.output(record)
}
}
@ -130,8 +130,7 @@ func (l *Logger) Debug(v ...interface{}) {
// Debugf ...
func (l *Logger) Debugf(format string, v ...interface{}) {
if l.lvl <= DebugLevel {
line := line(l.callDepth)
record := NewRecord(time.Now(), fmt.Sprintf(format, v...), line, DebugLevel)
record := NewRecord(time.Now(), fmt.Sprintf(format, v...), l.getLine(), DebugLevel)
l.output(record)
}
}
@ -171,8 +170,7 @@ func (l *Logger) Warningf(format string, v ...interface{}) {
// Error ...
func (l *Logger) Error(v ...interface{}) {
if l.lvl <= ErrorLevel {
line := line(l.callDepth)
record := NewRecord(time.Now(), fmt.Sprint(v...), line, ErrorLevel)
record := NewRecord(time.Now(), fmt.Sprint(v...), l.getLine(), ErrorLevel)
l.output(record)
}
}
@ -180,8 +178,7 @@ func (l *Logger) Error(v ...interface{}) {
// Errorf ...
func (l *Logger) Errorf(format string, v ...interface{}) {
if l.lvl <= ErrorLevel {
line := line(l.callDepth)
record := NewRecord(time.Now(), fmt.Sprintf(format, v...), line, ErrorLevel)
record := NewRecord(time.Now(), fmt.Sprintf(format, v...), l.getLine(), ErrorLevel)
l.output(record)
}
}
@ -189,8 +186,7 @@ func (l *Logger) Errorf(format string, v ...interface{}) {
// Fatal ...
func (l *Logger) Fatal(v ...interface{}) {
if l.lvl <= FatalLevel {
line := line(l.callDepth)
record := NewRecord(time.Now(), fmt.Sprint(v...), line, FatalLevel)
record := NewRecord(time.Now(), fmt.Sprint(v...), l.getLine(), FatalLevel)
l.output(record)
}
os.Exit(1)
@ -199,13 +195,19 @@ func (l *Logger) Fatal(v ...interface{}) {
// Fatalf ...
func (l *Logger) Fatalf(format string, v ...interface{}) {
if l.lvl <= FatalLevel {
line := line(l.callDepth)
record := NewRecord(time.Now(), fmt.Sprintf(format, v...), line, FatalLevel)
record := NewRecord(time.Now(), fmt.Sprintf(format, v...), l.getLine(), FatalLevel)
l.output(record)
}
os.Exit(1)
}
func (l *Logger) getLine() string {
if l.skipLine {
return ""
}
return line(l.callDepth)
}
// Debug ...
func Debug(v ...interface{}) {
logger.Debug(v...)