Merge pull request #261 from reasonerjt/job-service

store log in fs, provide api for getting log of a job

Deploy/db/registry.sql

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

api/jobs/replication.go

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

job/config/config.go

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

job/replication/statehandlers.go

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

job/statehandlers.go

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

job/statemachine.go

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

job/utils/logger.go

@@ -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 (
+func NewLogger(jobID int64) *log.Logger {
-	INFO = "info"
+	logFile := GetJobLogPath(jobID)
-	WARN = "warning"
+	f, err := os.OpenFile(logFile, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0660)
-	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...))
 	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{}) {
+func GetJobLogPath(jobID int64) string {
-	err := dao.AddJobLog(l.ID, WARN, fmt.Sprintf(format, v...))
+	fn := fmt.Sprintf("job_%d.log", jobID)
-	if err != nil {
+	return filepath.Join(config.LogDir(), fn)
-		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)
-	}
 }

jobservice/router.go

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

utils/log/logger.go

@@ -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...), l.getLine(), DebugLevel)
-		record := NewRecord(time.Now(), fmt.Sprint(v...), line, 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...), l.getLine(), DebugLevel)
-		record := NewRecord(time.Now(), fmt.Sprintf(format, v...), line, 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...), l.getLine(), ErrorLevel)
-		record := NewRecord(time.Now(), fmt.Sprint(v...), line, 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...), l.getLine(), ErrorLevel)
-		record := NewRecord(time.Now(), fmt.Sprintf(format, v...), line, 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...), l.getLine(), FatalLevel)
-		record := NewRecord(time.Now(), fmt.Sprint(v...), line, 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...), l.getLine(), FatalLevel)
-		record := NewRecord(time.Now(), fmt.Sprintf(format, v...), line, 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...)