Get job log by job_id in worker (#18261)

Get job by job_id in redis

  Get the last 10MB of data if it exceeds 10MB

Signed-off-by: stonezdj <[email protected]>

api/v2.0/swagger.yaml

@@ -4714,6 +4714,39 @@ paths:
           $ref: '#/responses/404'
         '500':
           $ref: '#/responses/500'
+  /jobservice/jobs/{job_id}/log:
+    get:
+      operationId: actionGetJobLog
+      summary: Get job log by job id
+      description: Get job log by job id, it is only used by administrator
+      produces:
+        - text/plain  
+      tags:
+        - jobservice
+      parameters:
+        - $ref: '#/parameters/requestId'
+        - name: job_id
+          in: path
+          required: true
+          type: string
+          description: The id of the job.
+      responses:
+        '200':
+          description: Get job log successfully.
+          headers:
+            Content-Type:
+              description: The content type of response body
+              type: string
+          schema:
+            type: string
+        '401':
+          $ref: '#/responses/401'
+        '403':
+          $ref: '#/responses/403'
+        '404':
+          $ref: '#/responses/404'
+        '500':
+          $ref: '#/responses/500'
   /jobservice/queues:
     get:
       operationId: listJobQueues

make/photon/prepare/templates/jobservice/config.yml.jinja

@@ -51,3 +51,6 @@ metric:
   path: {{ metric.path }}
   port: {{ metric.port }}
 {% endif %}
+
+# the max size of job log returned by API, default is 10M
+max_retrieve_size_mb: 10

src/controller/jobmonitor/monitor.go

@@ -70,6 +70,7 @@ type MonitorController interface {
 	PauseJobQueues(ctx context.Context, jobType string) error
 	// ResumeJobQueues resume the job queue by type
 	ResumeJobQueues(ctx context.Context, jobType string) error
+	GetJobLog(ctx context.Context, jobID string) ([]byte, error)
 }
 
 type monitorController struct {
@@ -366,3 +367,7 @@ func (w *monitorController) resumeQueue(ctx context.Context, jobType string) err
 	}
 	return nil
 }
+
+func (w *monitorController) GetJobLog(ctx context.Context, jobID string) ([]byte, error) {
+	return w.taskManager.GetLogByJobID(ctx, jobID)
+}

src/jobservice/config/config.go

@@ -23,7 +23,7 @@ import (
 	"strconv"
 	"strings"
 
-	yaml "gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v2"
 
 	"github.com/goharbor/harbor/src/jobservice/common/utils"
 	"github.com/goharbor/harbor/src/lib/log"
@@ -82,6 +82,9 @@ type Configuration struct {
 
 	// Metric configurations
 	Metric *MetricConfig `yaml:"metric,omitempty"`
+
+	// MaxLogSizeReturnedMB is the max size of log returned by job log API
+	MaxLogSizeReturnedMB int `yaml:"max_retrieve_size_mb,omitempty"`
 }
 
 // HTTPSConfig keeps additional configurations when using https protocol

src/jobservice/config/config_test.go

@@ -4,7 +4,7 @@
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
-//    http://www.apache.org/licenses/LICENSE-2.0
+//	http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ -86,6 +86,7 @@ func (suite *ConfigurationTestSuite) TestDefaultConfig() {
 	err := DefaultConfig.Load("../config_test.yml", true)
 	require.Nil(suite.T(), err, "load config from yaml file, expect nil error but got error '%s'", err)
 
+	assert.Equal(suite.T(), 10, DefaultConfig.MaxLogSizeReturnedMB, "expect max log size returned 10MB but got %d", DefaultConfig.MaxLogSizeReturnedMB)
 	redisURL := DefaultConfig.PoolConfig.RedisPoolCfg.RedisURL
 	assert.Equal(suite.T(), "redis://localhost:6379", redisURL, "expect redisURL '%s' but got '%s'", "redis://localhost:6379", redisURL)
 

src/jobservice/config_test.yml

@@ -39,3 +39,5 @@ job_loggers:
 loggers:
   - name: "STD_OUTPUT" # Same with above
     level: "DEBUG"
+
+max_retrieve_size_mb: 10

src/jobservice/logger/getter/db_getter.go

@@ -30,5 +30,15 @@ func (dbg *DBGetter) Retrieve(logID string) ([]byte, error) {
 		return nil, errs.NoObjectFoundError(fmt.Sprintf("log entity: %s", logID))
 	}
 
-	return []byte(jobLog.Content), nil
+	sz := int64(len(jobLog.Content))
+	var buf []byte
+	sizeLimit := logSizeLimit()
+	if sizeLimit <= 0 {
+		buf = []byte(jobLog.Content)
+		return buf, nil
+	}
+	if sz > sizeLimit {
+		buf = []byte(jobLog.Content[sz-sizeLimit:])
+	}
+	return buf, nil
 }

src/jobservice/logger/getter/file_getter.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/goharbor/harbor/src/jobservice/common/utils"
+	"github.com/goharbor/harbor/src/jobservice/config"
 	"github.com/goharbor/harbor/src/jobservice/errs"
 )
 
@@ -21,6 +22,12 @@ type FileGetter struct {
 func NewFileGetter(baseDir string) *FileGetter {
 	return &FileGetter{baseDir}
 }
+func logSizeLimit() int64 {
+	if config.DefaultConfig == nil {
+		return int64(0)
+	}
+	return int64(config.DefaultConfig.MaxLogSizeReturnedMB * 1024 * 1024)
+}
 
 // Retrieve implements @Interface.Retrieve
 func (fg *FileGetter) Retrieve(logID string) ([]byte, error) {
@@ -34,7 +41,7 @@ func (fg *FileGetter) Retrieve(logID string) ([]byte, error) {
 		return nil, errs.NoObjectFoundError(logID)
 	}
 
-	return os.ReadFile(fPath)
+	return tailLogFile(fPath, logSizeLimit())
 }
 
 func isValidLogID(id string) error {
@@ -54,3 +61,42 @@ func isValidLogID(id string) error {
 
 	return nil
 }
+
+func tailLogFile(filename string, limit int64) ([]byte, error) {
+	fInfo, err := os.Stat(filename)
+	if err != nil {
+		return nil, err
+	}
+	size := fInfo.Size()
+
+	var sizeToRead int64
+	if limit <= 0 {
+		sizeToRead = size
+	} else {
+		sizeToRead = limit
+	}
+	if sizeToRead > size {
+		sizeToRead = size
+	}
+
+	fi, err := os.Open(filename)
+	if err != nil {
+		return nil, err
+	}
+	defer fi.Close()
+
+	pos := size - sizeToRead
+	if pos < 0 {
+		pos = 0
+	}
+	if pos != 0 {
+		_, err = fi.Seek(pos, 0)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	buf := make([]byte, sizeToRead)
+	_, err = fi.Read(buf)
+	return buf, err
+}

src/jobservice/logger/getter/file_getter_test.go

@@ -44,3 +44,32 @@ func TestLogDataGetter(t *testing.T) {
 		t.Errorf("expect reading 5 bytes but got %d bytes", len(data))
 	}
 }
+
+func Test_tailLogFile(t *testing.T) {
+	type args struct {
+		filename string
+		mbs      int64
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    int
+		wantErr bool
+	}{
+		{"normal test", args{"testdata/normal.log", 1000}, len(`hello world`), false},
+		{"truncated test", args{"testdata/truncated.log", 1000}, 1000, false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := tailLogFile(tt.args.filename, tt.args.mbs)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("tailLogFile() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			// result should always less than the size limit
+			if len(got) > tt.want {
+				t.Errorf("tailLogFile() got = %v, want %v", len(got), tt.want)
+			}
+		})
+	}
+}

src/jobservice/logger/getter/testdata/normal.log

@@ -0,0 +1 @@
+hello world

src/pkg/task/task.go

@@ -53,6 +53,8 @@ type Manager interface {
 	UpdateExtraAttrs(ctx context.Context, id int64, extraAttrs map[string]interface{}) (err error)
 	// Get the log of the specified task
 	GetLog(ctx context.Context, id int64) (log []byte, err error)
+	// GetLogByJobID get the log of specified job id
+	GetLogByJobID(ctx context.Context, jobID string) (log []byte, err error)
 	// Count counts total of tasks according to the query.
 	// Query the "ExtraAttrs" by setting 'query.Keywords["ExtraAttrs.key"]="value"'
 	Count(ctx context.Context, query *q.Query) (int64, error)
@@ -259,3 +261,7 @@ func (m *manager) UpdateStatusInBatch(ctx context.Context, jobIDs []string, stat
 func (m *manager) ExecutionIDsByVendorAndStatus(ctx context.Context, vendorType, status string) ([]int64, error) {
 	return m.dao.ExecutionIDsByVendorAndStatus(ctx, vendorType, status)
 }
+
+func (m *manager) GetLogByJobID(ctx context.Context, jobID string) (log []byte, err error) {
+	return m.jsClient.GetJobLog(jobID)
+}

src/server/v2.0/handler/jobservice.go

@@ -196,3 +196,14 @@ func (j *jobServiceAPI) ActionPendingJobs(ctx context.Context, params jobservice
 	}
 	return jobservice.NewActionPendingJobsOK()
 }
+
+func (j *jobServiceAPI) ActionGetJobLog(ctx context.Context, params jobservice.ActionGetJobLogParams) middleware.Responder {
+	if err := j.RequireSystemAccess(ctx, rbac.ActionList, rbac.ResourceJobServiceMonitor); err != nil {
+		return j.SendError(ctx, err)
+	}
+	log, err := j.jobCtr.GetJobLog(ctx, params.JobID)
+	if err != nil {
+		return j.SendError(ctx, err)
+	}
+	return jobservice.NewActionGetJobLogOK().WithContentType("text/plain").WithPayload(string(log))
+}