Introduce compression in job archive.

pkg/archive/fsBackend.go

@@ -7,7 +7,9 @@ package archive
 import (
 	"bufio"
 	"bytes"
+	"compress/gzip"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"path"
@@ -29,6 +31,11 @@ type FsArchive struct {
 	clusters []string
 }
 
+func checkFileExists(filePath string) bool {
+	_, err := os.Stat(filePath)
+	return !errors.Is(err, os.ErrNotExist)
+}
+
 func getPath(
 	job *schema.Job,
 	rootPath string,
@@ -82,16 +89,31 @@ func (fsa *FsArchive) Init(rawConfig json.RawMessage) error {
 }
 
 func (fsa *FsArchive) LoadJobData(job *schema.Job) (schema.JobData, error) {
+	var isCompressed bool = true
+	filename := getPath(job, fsa.path, "data.json.gz")
 
-	filename := getPath(job, fsa.path, "data.json")
+	if !checkFileExists(filename) {
+		filename = getPath(job, fsa.path, "data.json")
+		isCompressed = false
+	}
 	f, err := os.Open(filename)
+
 	if err != nil {
 		log.Errorf("fsBackend LoadJobData()- %v", err)
 		return nil, err
 	}
 	defer f.Close()
 
-	return DecodeJobData(bufio.NewReader(f), filename)
+	if isCompressed {
+		r, err := gzip.NewReader(f)
+		if err != nil {
+			log.Errorf(" %v", err)
+			return nil, err
+		}
+		return DecodeJobData(r, filename)
+	} else {
+		return DecodeJobData(bufio.NewReader(f), filename)
+	}
 }
 
 func (fsa *FsArchive) LoadJobMeta(job *schema.Job) (*schema.JobMeta, error) {
@@ -217,12 +239,32 @@ func (fsa *FsArchive) ImportJob(
 		return err
 	}
 
+	// var isCompressed bool = true
+	// // TODO Use shortJob Config for check
+	// if jobMeta.Duration < 300 {
+	// 	isCompressed = false
+	// 	f, err = os.Create(path.Join(dir, "data.json"))
+	// } else {
+	// 	f, err = os.Create(path.Join(dir, "data.json.gz"))
+	// }
+	// if err != nil {
+	// 	return err
+	// }
+	//
+	// if isCompressed {
+	// 	if err := EncodeJobData(gzip.NewWriter(f), jobData); err != nil {
+	// 		return err
+	// 	}
+	// } else {
+	// 	if err := EncodeJobData(f, jobData); err != nil {
+	// 		return err
+	// 	}
+	// }
+
 	f, err = os.Create(path.Join(dir, "data.json"))
-	if err != nil {
-		return err
-	}
 	if err := EncodeJobData(f, jobData); err != nil {
 		return err
 	}
+
 	return f.Close()
 }

test/integration_test.go

@@ -468,7 +468,7 @@ func TestRestApi(t *testing.T) {
 	}
 
 	const stopJobBody string = `{
-"jobId":     123,
+        "jobId":     123,
 		"startTime": 123456789,
 		"cluster":   "testcluster",
 
@@ -551,7 +551,7 @@ func TestRestApi(t *testing.T) {
 
 func subtestLetJobFail(t *testing.T, restapi *api.RestApi, r *mux.Router) {
 	const startJobBody string = `{
-"jobId":            12345,
+        "jobId":            12345,
 		"user":             "testuser",
 		"project":          "testproj",
 		"cluster":          "testcluster",
@@ -588,7 +588,7 @@ func subtestLetJobFail(t *testing.T, restapi *api.RestApi, r *mux.Router) {
 	}
 
 	const stopJobBody string = `{
-"jobId":     12345,
+        "jobId":     12345,
 		"cluster":   "testcluster",
 
 		"jobState": "failed",