Skip to content

Fix job head renaming #296

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
r4victor merged 2 commits into from
Apr 3, 2023
Merged

Fix job head renaming #296

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c2760c2
Rename job heads for aws
r4victor Mar 31, 2023
2232ad7
Factor out local runner storage methods
r4victor Apr 3, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
290 changes: 86 additions & 204 deletions runner/internal/backend/local/backend.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,19 @@ import (
"gopkg.in/yaml.v3"
)

var _ backend.Backend = (*Local)(nil)
type File struct {
Path string `yaml:"path"`
}

func init() {
type Local struct {
path string
runnerID string
state *models.State
storage *LocalStorage
cliSecret *ClientSecret
}

func init() {
backend.DefaultBackend = New()
backend.RegisterBackend("local", func(ctx context.Context, pathConfig string) (backend.Backend, error) {
file := File{}
Expand All @@ -43,254 +52,84 @@ func init() {
})
}

type Local struct {
path string
runnerID string
state *models.State
cliSecret *ClientSecret
}

func New() *Local {
path := filepath.Join(common.HomeDir(), consts.DSTACK_DIR_PATH)
return &Local{
path: path,
storage: NewLocalStorage(path),
cliSecret: NewClientSecret(path),
}
}

func (l Local) GetJobByPath(ctx context.Context, path string) (*models.Job, error) {
if l.state == nil {
log.Trace(ctx, "State not exist")
return nil, gerrors.Wrap(backend.ErrLoadStateFile)
}
theFile, err := os.Open(filepath.Join(l.path, path))
if err != nil {
log.Error(ctx, "Failed to open file", "err", err)
return nil, gerrors.Wrap(err)
}
bodyFile, err := io.ReadAll(theFile)
if err != nil {
log.Error(ctx, "Failed to read file", "err", err)
return nil, gerrors.Wrap(err)
}
job := new(models.Job)
if err = yaml.Unmarshal(bodyFile, job); err != nil {
return nil, gerrors.Wrap(err)
}
return job, nil
}

func (l *Local) GitCredentials(ctx context.Context) *models.GitCredentials {
log.Trace(ctx, "Getting credentials")
if l == nil {
log.Error(ctx, "Backend is empty")
return nil
}
if l.state == nil {
log.Error(ctx, "State is empty")
return nil
}
if l.state.Job == nil {
log.Error(ctx, "Job is empty")
return nil
}
return l.cliSecret.fetchCredentials(ctx, l.state.Job.RepoHostNameWithPort(), l.state.Job.RepoUserName, l.state.Job.RepoName)
}

func (l *Local) Secrets(ctx context.Context) (map[string]string, error) {
log.Trace(ctx, "Getting secrets")
if l == nil {
return nil, gerrors.New("Backend is nil")
}
if l.state == nil {
return nil, gerrors.New("State is empty")
}
templatePath := fmt.Sprintf("secrets/%s/%s/%s", l.state.Job.RepoHostNameWithPort(), l.state.Job.RepoUserName, l.state.Job.RepoName)
if _, err := os.Stat(filepath.Join(l.path, templatePath)); err != nil {
return map[string]string{}, nil
}
listSecrets, err := os.ReadDir(filepath.Join(l.path, templatePath))
if err != nil {
return nil, gerrors.Wrap(err)
}

secrets := make(map[string]string, 0)
for _, file := range listSecrets {
if file.IsDir() {
continue
}
if strings.HasPrefix(file.Name(), "l;") {
clearName := strings.ReplaceAll(file.Name(), "l;", "")
secrets[clearName] = fmt.Sprintf("%s/%s/%s/%s",
l.state.Job.RepoHostNameWithPort(),
l.state.Job.RepoUserName,
l.state.Job.RepoName,
clearName)
}
}
return l.cliSecret.fetchSecret(ctx, templatePath, secrets)
}

func (l Local) Bucket(ctx context.Context) string {
return ""
}

func (l Local) ListSubDir(ctx context.Context, dir string) ([]string, error) {
log.Trace(ctx, "Fetching list sub dir")
listDir := strings.Split(dir, "/")
root := strings.Join(listDir[:len(listDir)-1], string(filepath.Separator))
prefix := listDir[len(listDir)-1]
list, err := os.ReadDir(filepath.Join(l.path, root))
if err != nil {
return nil, gerrors.Wrap(err)
}
listJob := make([]string, 0, 5)
for _, value := range list {
if strings.HasPrefix(value.Name(), prefix) {
listJob = append(listJob, filepath.Join(root, value.Name()))
}
}
return listJob, nil
}

func (l *Local) Init(ctx context.Context, ID string) error {
log.Trace(ctx, "Initialize backend with ID runner", "runner ID", ID)
l.runnerID = ID
pathRunner := filepath.Join(l.path, "runners", fmt.Sprintf("%s.yaml", ID))
log.Trace(ctx, "Fetch runner state from S3",
"path", pathRunner)
theFile, err := os.Open(pathRunner)
pathRunner := filepath.Join("runners", fmt.Sprintf("%s.yaml", ID))
log.Trace(ctx, "Fetch local runner state", "path", pathRunner)
contents, err := l.storage.GetFile(pathRunner)
if err != nil {
return gerrors.Wrap(err)
}
log.Trace(ctx, "Unmarshal state")
bodyFile, err := io.ReadAll(theFile)
if err != nil {
return gerrors.Wrap(err)
}
err = yaml.Unmarshal(bodyFile, &l.state)
err = yaml.Unmarshal(contents, &l.state)
if err != nil {
return gerrors.Wrap(err)
}
return nil
}

func (l Local) Job(ctx context.Context) *models.Job {
log.Trace(ctx, "Getting job from state")
if l.state == nil {
log.Trace(ctx, "State not exist")
return new(models.Job)
}
log.Trace(ctx, "Get job", "ID", l.state.Job.JobID)
log.Trace(ctx, "Getting job from state", "ID", l.state.Job.JobID)
return l.state.Job
}

func (l Local) MasterJob(ctx context.Context) *models.Job {
if l.state == nil {
log.Trace(ctx, "State not exist")
return nil
}
theFile, err := os.Open(filepath.Join(l.path, "jobs", l.state.Job.RepoUserName, l.state.Job.RepoName, fmt.Sprintf("%s.yaml", l.state.Job.MasterJobID)))
contents, err := l.storage.GetFile(filepath.Join("jobs", l.state.Job.RepoUserName, l.state.Job.RepoName, fmt.Sprintf("%s.yaml", l.state.Job.MasterJobID)))
if err != nil {
log.Error(ctx, "Failed to open file", "err", err)
return nil
}
bodyFile, err := io.ReadAll(theFile)
if err != nil {
log.Error(ctx, "Failed to read file", "err", err)
return nil
}
masterJob := new(models.Job)
err = yaml.Unmarshal(bodyFile, masterJob)
err = yaml.Unmarshal(contents, masterJob)
if err != nil {
return nil
}
return masterJob
}

func (l Local) Requirements(ctx context.Context) models.Requirements {
if l.state == nil {
log.Trace(ctx, "State not exist")
return models.Requirements{}
}
log.Trace(ctx, "Return model resource")
log.Trace(ctx, "Getting requirements")
return l.state.Job.Requirements
}

func (l Local) UpdateState(ctx context.Context) error {
log.Trace(ctx, "Start update state")
if l.state == nil {
log.Trace(ctx, "State not exist")
return gerrors.Wrap(backend.ErrLoadStateFile)
}
log.Trace(ctx, "Fetching list jobs", "Repo username", l.state.Job.RepoUserName, "Repo name", l.state.Job.RepoName, "Job ID", l.state.Job.JobID)
pathDir := filepath.Join(l.path, fmt.Sprintf("jobs/%s/%s/%s", l.state.Job.RepoHostNameWithPort(), l.state.Job.RepoUserName, l.state.Job.RepoName))
list, err := os.ReadDir(pathDir)
if err != nil {
return gerrors.Wrap(err)
}

for _, value := range list {
if !value.IsDir() {
if strings.HasPrefix(value.Name(), fmt.Sprintf("l;%s;", l.state.Job.JobID)) {
if err = os.Remove(filepath.Join(pathDir, value.Name())); err != nil {
return gerrors.Wrap(err)
}
}
}
}
log.Trace(ctx, "Marshaling job")
theFileBody, err := yaml.Marshal(&l.state.Job)
contents, err := yaml.Marshal(&l.state.Job)
if err != nil {
return gerrors.Wrap(err)
}
pathJob := fmt.Sprintf("jobs/%s/%s/%s/%s.yaml", l.state.Job.RepoHostNameWithPort(), l.state.Job.RepoUserName, l.state.Job.RepoName, l.state.Job.JobID)
log.Trace(ctx, "Write to file job", "Path", pathJob)

theFile, err := os.OpenFile(filepath.Join(l.path, pathJob), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0777)
jobPath := l.state.Job.JobFilepath()
log.Trace(ctx, "Write to file job", "Path", jobPath)
err = l.storage.PutFile(jobPath, contents)
if err != nil {
return gerrors.Wrap(err)
}
defer theFile.Close()

if _, err = theFile.Write(theFileBody); err != nil {
log.Trace(ctx, "Fetching list jobs", "Repo username", l.state.Job.RepoUserName, "Repo name", l.state.Job.RepoName, "Job ID", l.state.Job.JobID)
files, err := l.storage.ListFile(l.state.Job.JobHeadFilepathPrefix())
if err != nil {
return gerrors.Wrap(err)
}
appString := make([]string, 0, len(l.state.Job.Apps))
for _, app := range l.state.Job.Apps {
appString = append(appString, app.Name)
}

artifactSlice := make([]string, 0, 5)
for _, art := range l.state.Job.Artifacts {
artifactSlice = append(artifactSlice, strings.ReplaceAll(art.Path, `/`, "_"))
}

pathLockJob := fmt.Sprintf("jobs/%s/%s/%s/l;%s;%s;%s;%d;%s;%s;%s;%s",
l.state.Job.RepoHostNameWithPort(),
l.state.Job.RepoUserName,
l.state.Job.RepoName,
l.state.Job.JobID,
l.state.Job.ProviderName,
l.state.Job.LocalRepoUserName,
l.state.Job.SubmittedAt,
l.state.Job.Status,
strings.Join(artifactSlice, ","),
strings.Join(appString, ","),
l.state.Job.TagName)
log.Trace(ctx, "Write to file lock job", "Path", pathLockJob)

if _, err = os.OpenFile(filepath.Join(l.path, pathLockJob), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0777); err != nil {
return gerrors.Wrap(err)
jobHeadFilepath := l.state.Job.JobHeadFilepath()
for _, file := range files[:1] {
log.Trace(ctx, "Renaming file job", "From", file, "To", jobHeadFilepath)
err = l.storage.RenameFile(file, jobHeadFilepath)
if err != nil {
return gerrors.Wrap(err)
}
}
return nil
}

func (l Local) CheckStop(ctx context.Context) (bool, error) {
if l.state == nil {
log.Trace(ctx, "State not exist")
return false, gerrors.Wrap(backend.ErrLoadStateFile)
}
pathStateFile := fmt.Sprintf("runners/m;%s;status", l.runnerID)
log.Trace(ctx, "Reading metadata from state file", "path", pathStateFile)
if _, err := os.Stat(filepath.Join(l.path, pathStateFile)); err == nil {
Expand All @@ -313,13 +152,11 @@ func (l Local) CheckStop(ctx context.Context) (bool, error) {
}

func (l Local) Shutdown(ctx context.Context) error {
log.Trace(ctx, "Start shutdown")
return nil
}

func (l *Local) GetArtifact(ctx context.Context, runName, localPath, remotePath string, _ bool) artifacts.Artifacter {
if l == nil {
return nil
}
rootPath := path.Join(common.HomeDir(), consts.USER_ARTIFACTS_PATH, runName)
log.Trace(ctx, "Create simple artifact's engine. Local", "Root path", rootPath)
art, err := local.NewLocal(l.path, rootPath, localPath, remotePath)
Expand All @@ -331,10 +168,6 @@ func (l *Local) GetArtifact(ctx context.Context, runName, localPath, remotePath
}

func (l Local) CreateLogger(ctx context.Context, logGroup, logName string) io.Writer {
if l.state == nil {
log.Trace(ctx, "State not exist")
return nil
}
log.Trace(ctx, "Build logger", "LogGroup", logGroup, "LogName", logName)
logger, err := NewLogger(logGroup, logName)
if err != nil {
Expand All @@ -344,6 +177,55 @@ func (l Local) CreateLogger(ctx context.Context, logGroup, logName string) io.Wr
return logger
}

type File struct {
Path string `yaml:"path"`
func (l Local) GetJobByPath(ctx context.Context, path string) (*models.Job, error) {
contents, err := l.storage.GetFile(path)
if err != nil {
return nil, gerrors.Wrap(err)
}
job := new(models.Job)
if err = yaml.Unmarshal(contents, job); err != nil {
return nil, gerrors.Wrap(err)
}
return job, nil
}

func (l *Local) GitCredentials(ctx context.Context) *models.GitCredentials {
log.Trace(ctx, "Getting credentials")
return l.cliSecret.fetchCredentials(ctx, l.state.Job.RepoHostNameWithPort(), l.state.Job.RepoUserName, l.state.Job.RepoName)
}

func (l *Local) Secrets(ctx context.Context) (map[string]string, error) {
log.Trace(ctx, "Getting secrets")
templatePath := fmt.Sprintf("secrets/%s/%s/%s", l.state.Job.RepoHostNameWithPort(), l.state.Job.RepoUserName, l.state.Job.RepoName)
if _, err := os.Stat(filepath.Join(l.path, templatePath)); err != nil {
return map[string]string{}, nil
}
listSecrets, err := os.ReadDir(filepath.Join(l.path, templatePath))
if err != nil {
return nil, gerrors.Wrap(err)
}
secrets := make(map[string]string, 0)
for _, file := range listSecrets {
if file.IsDir() {
continue
}
if strings.HasPrefix(file.Name(), "l;") {
clearName := strings.ReplaceAll(file.Name(), "l;", "")
secrets[clearName] = fmt.Sprintf("%s/%s/%s/%s",
l.state.Job.RepoHostNameWithPort(),
l.state.Job.RepoUserName,
l.state.Job.RepoName,
clearName)
}
}
return l.cliSecret.fetchSecret(ctx, templatePath, secrets)
}

func (l Local) Bucket(ctx context.Context) string {
return ""
}

func (l Local) ListSubDir(ctx context.Context, dir string) ([]string, error) {
log.Trace(ctx, "Fetching list sub dir")
return l.storage.ListFile(dir)
}
1 change: 0 additions & 1 deletion runner/internal/backend/local/secret.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ type ClientSecret struct {
}

func NewClientSecret(path string) *ClientSecret {

return &ClientSecret{path: path}
}

Expand Down
Loading