[Refactor] DCR Monitor Refactoring (3/3) (#29)

2/3 PR for a major refactoring of DCR Monitor. See
#15.

1/3: #16
2/3: #19

This pretty much concludes the major refactoring, resolving #15

# Changes

* Added `Created` job status, because the reconciler needs to handle
newly created jobs.
* Move KanikoService functionality into the monitor (=reconciler)
* Remove unused config `Cluster.PodServiceAccount`
* Clean up ARG/ENV of TEE dockerfile
* In-memory Dockerfile injection in API. See `addDockerfileToTarGz`.
This resolves #3. Also Kaniko can just refer to the remote object path.
* Removed most of shallow functions in `Config`
* Completely removed `CloudProvider`
* Removed shallow / no-longer used modules in `pkg`:
`pkg/utils/constant.go`, `pkg/utils/file.go`, and `pkg/utils/k8s.go`.

# Testing

Manual end-to-end testing.

Author: dayeol

BUILD

@@ -17,9 +17,7 @@ echo 'CloudProvider:
     Zone: "{zone}"
     Region: "{region}"
     Debug: false
-    Env: {env}
-Cluster:
-  PodServiceAccount: "dcr-k8s-pod-sa"' > $@
+    Env: {env}' > $@
     """.format(
         env = env,
         project_id = project_id,

app/dcr_api/Dockerfile

@@ -1,26 +1,18 @@
 ARG BASE_IMAGE
 FROM $BASE_IMAGE
 ARG OUTPUTPATH
-ARG ENCRYPTED_FILENAME
-ARG ENCRYPTED_CLOUDSTORAGE_PATH
-ARG CREATOR
-ARG IMPERSONATION_SERVICE_ACCOUNT
 ARG JUPYTER_FILENAME
 ARG USER_WORKSPACE
 ARG CUSTOMTOKEN_CLOUDSTORAGE_PATH 
 
 ENV OUTPUTPATH=$OUTPUTPATH
-ENV ENCRYPTED_FILENAME=$ENCRYPTED_FILENAME
-ENV ENCRYPTED_CLOUDSTORAGE_PATH=$ENCRYPTED_CLOUDSTORAGE_PATH
-ENV IMPERSONATION_SERVICE_ACCOUNT=$IMPERSONATION_SERVICE_ACCOUNT
-ENV CREATOR=$CREATOR
 ENV JUPYTER_FILENAME=$JUPYTER_FILENAME
 ENV CUSTOMTOKEN_CLOUDSTORAGE_PATH=$CUSTOMTOKEN_CLOUDSTORAGE_PATH
 
 WORKDIR /home/jovyan
 COPY $USER_WORKSAPCE/* ./
 
-LABEL "tee.launch_policy.allow_env_override"="USER_TOKEN,EXECUTION_STAGE,DEPLOYMENT_ENV,PROJECT_ID,KEY_LOCATION"
+LABEL "tee.launch_policy.allow_env_override"="USER_TOKEN,EXECUTION_STAGE,DEPLOYMENT_ENV,PROJECT_ID"
 
 ENTRYPOINT jupyter nbconvert --execute --to notebook --inplace $JUPYTER_FILENAME --ExecutePreprocessor.timeout=-1 --allow-errors \
     && hash=$(md5sum $JUPYTER_FILENAME | awk '{ print $1 }') \

app/dcr_api/biz/dal/db/job.go

@@ -27,6 +27,7 @@ type Job struct {
 	UUID              string `gorm:"uuid" json:"uuid"`
 	Creator           string `gorm:"creator" json:"creator"`
 	JupyterFileName   string `gorm:"jupyter_file_name" json:"jupyter_file_name"`
+	BuildContextPath  string `gorm:"build_context_path" json:"build_context_path"`
 	DockerImage       string `gorm:"docker_image" json:"docker_image"`
 	DockerImageDigest string `gorm:"docker_image_digest" json:"docker_image_digest"`
 	AttestationReport string `gorm:"attestation_report" json:"attestation_report"`
@@ -50,7 +51,7 @@ func CreateJob(job *Job) error {
 }
 
 func UpdateJob(j *Job) error {
-	result := DB.Model(j).Updates(Job{JobStatus: j.JobStatus, DockerImageDigest: j.DockerImageDigest, DockerImage: j.DockerImage, AttestationReport: j.AttestationReport, InstanceName: j.InstanceName})
+	result := DB.Model(j).Updates(Job{JobStatus: j.JobStatus, BuildContextPath: j.BuildContextPath, DockerImageDigest: j.DockerImageDigest, DockerImage: j.DockerImage, AttestationReport: j.AttestationReport, InstanceName: j.InstanceName})
 	if result.Error != nil {
 		return errors.Wrap(result.Error, "failed to update job %v")
 	}
@@ -98,15 +99,15 @@ func QueryJobByUUIDAndCreator(creator string, uuid string) (*Job, error) {
 
 func GetInProgressJobs(creator string) ([]*Job, error) {
 	var res []*Job
-	if err := DB.Model(Job{}).Where("creator = ? AND job_status in (1, 3, 4)", creator).Find(&res).Error; err != nil {
+	if err := DB.Model(Job{}).Where("creator = ? AND job_status in (0, 1, 3, 4)", creator).Find(&res).Error; err != nil {
 		return nil, errors.Wrap(err, "failed to find finished job status")
 	}
 	return res, nil
 }
 
 func GetAllInProgressJobs() ([]*Job, error) {
 	var res []*Job
-	if err := DB.Model(Job{}).Where("job_status in (1, 3, 4)").Find(&res).Error; err != nil {
+	if err := DB.Model(Job{}).Where("job_status in (0, 1, 3, 4)").Find(&res).Error; err != nil {
 		return nil, errors.Wrap(err, "failed to find finished job status")
 	}
 	return res, nil

app/dcr_api/biz/model/job/job.go

@@ -2,29 +2,17 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
 go_library(
     name = "service",
-    srcs = [
-        "build_service.go",
-        "job_service.go",
-        "kaniko_service.go",
-    ],
+    srcs = ["job_service.go"],
     importpath = "github.com/manatee-project/manatee/app/dcr_api/biz/service",
     visibility = ["//visibility:public"],
     deps = [
         "//app/dcr_api/biz/dal/db",
         "//app/dcr_api/biz/model/job",
-        "//pkg/cloud",
         "//pkg/config",
         "//pkg/errno",
-        "//pkg/utils",
         "@com_github_cloudwego_hertz//pkg/common/hlog",
-        "@com_github_docker_docker//pkg/archive",
         "@com_github_google_uuid//:uuid",
         "@com_github_pkg_errors//:errors",
-        "@io_k8s_api//batch/v1:batch",
-        "@io_k8s_api//core/v1:core",
-        "@io_k8s_apimachinery//pkg/api/resource",
-        "@io_k8s_apimachinery//pkg/apis/meta/v1:meta",
-        "@io_k8s_client_go//kubernetes",
-        "@io_k8s_client_go//rest",
+        "@com_google_cloud_go_storage//:storage",
     ],
 )

app/dcr_api/biz/service/BUILD.bazel

@@ -14,19 +14,22 @@
 package service
 
 import (
+	"archive/tar"
+	"bytes"
+	"compress/gzip"
 	"context"
 	"encoding/base64"
 	"fmt"
 	"io"
+	"os"
 
+	"cloud.google.com/go/storage"
 	"github.com/cloudwego/hertz/pkg/common/hlog"
 	"github.com/google/uuid"
 	"github.com/manatee-project/manatee/app/dcr_api/biz/dal/db"
 	"github.com/manatee-project/manatee/app/dcr_api/biz/model/job"
-	"github.com/manatee-project/manatee/pkg/cloud"
 	"github.com/manatee-project/manatee/pkg/config"
 	"github.com/manatee-project/manatee/pkg/errno"
-	"github.com/manatee-project/manatee/pkg/utils"
 	"github.com/pkg/errors"
 )
 
@@ -49,23 +52,33 @@ func (js *JobService) SubmitJob(req *job.SubmitJobRequest, userWorkspace io.Read
 		return "", errors.Wrap(fmt.Errorf(errno.ReachJobLimitErrMsg), "")
 	}
 
-	provider := cloud.GetCloudProvider(js.ctx)
-	err = provider.UploadFile(userWorkspace, config.GetUserWorkSpacePath(creator), false)
+	// inject Dockerfile in /usr/local/dcr_conf/Dockerfile into the build context
+	// FIXME: avoid reading the file system, use a string instead in the future.
+	dockerFileContent, err := os.ReadFile("/usr/local/dcr_conf/Dockerfile")
 	if err != nil {
 		return "", err
 	}
+	buildctx, err := js.addDockerfileToTarGz(userWorkspace, string(dockerFileContent))
+	if err != nil {
+		return "", err
+	}
+	err = js.uploadFile(buildctx, fmt.Sprintf("%s/%s-workspace.tar.gz", creator, creator), false)
+	if err != nil {
+		return "", err
+	}
+	buildctxpath := fmt.Sprintf("gs://%s/%s/%s-workspace.tar.gz", config.GetBucket(), creator, creator)
 
 	uuidStr, err := uuid.NewUUID()
 	if err != nil {
 		return "", errors.Wrap(err, "failed to generate uuid")
 	}
 	t := db.Job{
-		UUID:            uuidStr.String(),
-		Creator:         req.Creator,
-		JupyterFileName: req.JupyterFileName,
-		JobStatus:       int(job.JobStatus_ImageBuilding),
+		UUID:             uuidStr.String(),
+		Creator:          req.Creator,
+		JupyterFileName:  req.JupyterFileName,
+		JobStatus:        int(job.JobStatus_Created),
+		BuildContextPath: buildctxpath,
 	}
-	err = BuildImage(js.ctx, t, req.AccessToken)
 	if err != nil {
 		return "", err
 	}
@@ -78,15 +91,69 @@ func (js *JobService) SubmitJob(req *job.SubmitJobRequest, userWorkspace io.Read
 	return uuidStr.String(), nil
 }
 
+func (js *JobService) addDockerfileToTarGz(input io.Reader, dockerfileContent string) (io.Reader, error) {
+	gzReader, err := gzip.NewReader(input)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create gzip reader: %w", err)
+	}
+	defer gzReader.Close()
+
+	var buffer bytes.Buffer
+	gzWriter := gzip.NewWriter(&buffer)
+	defer gzWriter.Close()
+
+	tarReader := tar.NewReader(gzReader)
+	tarWriter := tar.NewWriter(gzWriter)
+	defer tarWriter.Close()
+	for {
+		header, err := tarReader.Next()
+		if err == io.EOF {
+			break
+		}
+		if err != nil {
+			return nil, fmt.Errorf("failed to read tar entry: %w", err)
+		}
+
+		// Write the existing header and file content to the new tar archive
+		if err := tarWriter.WriteHeader(header); err != nil {
+			return nil, fmt.Errorf("failed to write tar header: %w", err)
+		}
+
+		if _, err := io.Copy(tarWriter, tarReader); err != nil {
+			return nil, fmt.Errorf("failed to write tar entry: %w", err)
+		}
+	}
+	// Add the Dockerfile as a new entry
+	dockerfileHeader := &tar.Header{
+		Name: "Dockerfile",
+		Size: int64(len(dockerfileContent)),
+		Mode: 0600,
+	}
+	if err := tarWriter.WriteHeader(dockerfileHeader); err != nil {
+		return nil, fmt.Errorf("failed to write Dockerfile header: %w", err)
+	}
+	if _, err := tarWriter.Write([]byte(dockerfileContent)); err != nil {
+		return nil, fmt.Errorf("failed to write Dockerfile content: %w", err)
+	}
+	// Close the tar and gzip writers
+	if err := tarWriter.Close(); err != nil {
+		return nil, fmt.Errorf("failed to close tar writer: %w", err)
+	}
+	if err := gzWriter.Close(); err != nil {
+		return nil, fmt.Errorf("failed to close gzip writer: %w", err)
+	}
+	return &buffer, nil
+}
+
 func convertEntityToModel(j *db.Job) *job.Job {
 	return &job.Job{
 		ID:              int64(j.ID),
 		UUID:            j.UUID,
 		Creator:         j.Creator,
 		JobStatus:       job.JobStatus(j.JobStatus),
 		JupyterFileName: j.JupyterFileName,
-		CreatedAt:       j.CreatedAt.Format(utils.Layout),
-		UpdatedAt:       j.UpdatedAt.Format(utils.Layout),
+		CreatedAt:       j.CreatedAt.Format("2006-01-02 15:04:05"),
+		UpdatedAt:       j.UpdatedAt.Format("2006-01-02 15:04:05"),
 	}
 }
 
@@ -107,24 +174,22 @@ func (js *JobService) GetJobOutputAttrs(req *job.QueryJobOutputRequest) (string,
 	if err != nil {
 		return "", 0, err
 	}
-	outputPath := config.GetJobOutputPath(j.Creator, j.UUID, j.JupyterFileName)
+	outputPath := js.getJobOutputPath(j.Creator, j.UUID, j.JupyterFileName)
 
-	provider := cloud.GetCloudProvider(js.ctx)
-	size, err := provider.GetFileSize(outputPath)
+	size, err := js.getFileSize(outputPath)
 	if err != nil {
 		return "", 0, err
 	}
-	return config.GetJobOutputFilename(fmt.Sprintf("%v", j.ID), j.JupyterFileName), size, nil
+	return js.getJobOutputFilename(fmt.Sprintf("%v", j.ID), j.JupyterFileName), size, nil
 }
 
 func (js *JobService) DownloadJobOutput(req *job.DownloadJobOutputRequest) (string, error) {
 	j, err := db.QueryJobByIdAndCreator(req.ID, req.Creator)
 	if err != nil {
 		return "", err
 	}
-	outputPath := config.GetJobOutputPath(j.Creator, j.UUID, j.JupyterFileName)
-	provider := cloud.GetCloudProvider(js.ctx)
-	datg, err := provider.GetFilebyChunk(outputPath, req.Offset, req.Chunk)
+	outputPath := js.getJobOutputPath(j.Creator, j.UUID, j.JupyterFileName)
+	datg, err := js.getFilebyChunk(outputPath, req.Offset, req.Chunk)
 	if err != nil {
 		return "", err
 	}
@@ -146,3 +211,70 @@ func (js *JobService) GetJobAttestationReport(req *job.QueryJobAttestationReques
 	}
 	return j.AttestationReport, nil
 }
+
+func (js *JobService) getJobOutputFilename(UUID string, originName string) string {
+	return fmt.Sprintf("out-%s-%s", UUID, originName)
+}
+
+func (js *JobService) getJobOutputPath(creator string, UUID string, originName string) string {
+	return fmt.Sprintf("%s/output/%s", creator, js.getJobOutputFilename(UUID, originName))
+}
+
+func (g *JobService) getFileSize(remotePath string) (int64, error) {
+	client, err := storage.NewClient(g.ctx)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to create gcp storage client")
+	}
+	defer client.Close()
+	bucket := config.GetBucket()
+	attr, err := client.Bucket(bucket).Object(remotePath).Attrs(g.ctx)
+	if err != nil {
+		return 0, errors.Wrap(err, "failed to get file attributes, or it doesn't exist")
+	}
+	return attr.Size, nil
+}
+
+func (g *JobService) getFilebyChunk(remotePath string, offset int64, chunkSize int64) ([]byte, error) {
+	client, err := storage.NewClient(g.ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create gcp storage client")
+	}
+	defer client.Close()
+	bucket := config.GetBucket()
+	objectHandle := client.Bucket(bucket).Object(remotePath)
+	objectReader, err := objectHandle.NewRangeReader(g.ctx, offset, chunkSize)
+	if err != nil {
+		return nil, errors.Wrap(err, fmt.Sprintf("failed to create reader on %s", remotePath))
+	}
+	defer objectReader.Close()
+	data := make([]byte, chunkSize)
+	n, err := objectReader.Read(data)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to read cloud storage object")
+	}
+	data = data[:n]
+	return data, nil
+}
+
+func (g *JobService) uploadFile(reader io.Reader, remotePath string, compress bool) error {
+	client, err := storage.NewClient(g.ctx)
+	if err != nil {
+		return errors.Wrap(err, "failed to create storage client")
+	}
+	defer client.Close()
+	bucket := config.GetBucket()
+	writer := client.Bucket(bucket).Object(remotePath).NewWriter(g.ctx)
+	defer writer.Close()
+	if compress {
+		gzipWriter := gzip.NewWriter(writer)
+		if _, err = io.Copy(gzipWriter, reader); err != nil {
+			return errors.Wrap(err, "failed to copy content to gzip writer")
+		}
+		defer gzipWriter.Close()
+	} else {
+		if _, err = io.Copy(writer, reader); err != nil {
+			return errors.Wrap(err, "failed to copy content to writer")
+		}
+	}
+	return nil
+}