-
Notifications
You must be signed in to change notification settings - Fork 3
Reef service and agent #251
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.
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
Open
khluu
wants to merge
9
commits into
main
Choose a base branch
from
reef-service
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
Open
Changes from all commits
Commits
Show all changes
9 commits
Select commit
Hold shift + click to select a range
0285374
Create push.sh
khluu fd0ebca
Create .gitignore
khluu 884cf56
p
khluu e0d23de
Service with launch request
khluu 5dfe132
p
khluu 5952623
full setup
khluu 96a5e95
p
khluu 2c997ed
add instance config
khluu e31c70f
simplify serve
khluu File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -5,3 +5,5 @@ | |
|
|
||
| __pycache__ | ||
| .hypothesis | ||
| *.log | ||
| *.db | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| package reefagent | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
| "log" | ||
| "net/http" | ||
| "time" | ||
| ) | ||
|
|
||
| type Agent struct { | ||
| Id string | ||
| Queue string | ||
| Job *Job | ||
| ServiceHost string | ||
| } | ||
|
|
||
| func (a *Agent) Start() { | ||
| // start a loop to ping the server every 1 second on a goroutine until ping returns a job | ||
| for { | ||
| success, job := a.Ping() | ||
| if success { | ||
| a.AcquireAndRunJob(job) | ||
| return | ||
| } | ||
| time.Sleep(1 * time.Second) | ||
| } | ||
| } | ||
|
|
||
| func (a *Agent) Ping() (bool, *Job) { | ||
| // call POST /ping endpoint with agent ID and queue | ||
| url := fmt.Sprintf("%s/ping?agentId=%s&queue=%s", a.ServiceHost, a.Id, a.Queue) | ||
| resp, err := http.Get(url) | ||
|
|
||
| if err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| var response struct { | ||
| JobId string `json:"jobId"` | ||
| Commands []string `json:"commands"` | ||
| } | ||
| if err := json.NewDecoder(resp.Body).Decode(&response); err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| job := &Job{ | ||
| Id: response.JobId, | ||
| Commands: response.Commands, | ||
| } | ||
| defer resp.Body.Close() | ||
| // if response is 200, return the job and true | ||
| // otherwise return nil and false | ||
| if resp.StatusCode == 200 { | ||
| return true, job | ||
| } | ||
| return false, nil | ||
| } | ||
|
|
||
| func (a *Agent) AcquireAndRunJob(job *Job) { | ||
| success := a.AcquireJob(job.Id) | ||
| if !success { | ||
| fmt.Println("Failed to acquire job") | ||
| return | ||
| } | ||
| a.Job = job | ||
| a.RunJob() | ||
| } | ||
|
|
||
| func (a *Agent) RunJob() { | ||
| jr := NewJobRunner(a.Job, a.ServiceHost) | ||
| jr.Run() | ||
| } | ||
|
|
||
| func (a *Agent) AcquireJob(jobId string) bool { | ||
| // Send POST request to acquire the job | ||
| url := fmt.Sprintf("%s/job/acquire?jobId=%s&agentId=%s", a.ServiceHost, jobId, a.Id) | ||
| resp, err := http.Post(url, "application/json", nil) | ||
| if err != nil { | ||
| log.Printf("Error acquiring job: %v", err) | ||
| return false | ||
| } | ||
| defer resp.Body.Close() | ||
|
|
||
| return resp.StatusCode == 200 | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,6 @@ | ||
| package reefagent | ||
|
|
||
| type Job struct { | ||
| Id string | ||
| Commands []string | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| package reefagent | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "io" | ||
| "os" | ||
| "os/exec" | ||
| "strings" | ||
| ) | ||
|
|
||
| type JobRunner struct { | ||
| logStreamer *LogStreamer | ||
| job *Job | ||
| } | ||
|
|
||
| func (jr *JobRunner) Run() { | ||
| jr.logStreamer.Start() | ||
| for _, command := range jr.job.Commands { | ||
| parts := strings.Split(command, " ") | ||
| cmd := exec.Command(parts[0], parts[1:]...) | ||
| cmd.Stdout = jr.logStreamer.logsWriter | ||
| cmd.Stderr = jr.logStreamer.logsWriter | ||
| if err := cmd.Run(); err != nil { | ||
| fmt.Printf("Error running command %s: %v\n", command, err) | ||
| } | ||
| } | ||
| jr.logStreamer.Stop() | ||
| } | ||
|
|
||
| func NewJobRunner(job *Job, serviceHost string) *JobRunner { | ||
| jr := &JobRunner{ | ||
| job: job, | ||
| logStreamer: NewLogStreamer(job.Id, serviceHost), | ||
| } | ||
| jr.logStreamer.logsWriter = io.MultiWriter(jr.logStreamer.logs, os.Stdout) | ||
| return jr | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,149 @@ | ||
| package reefagent | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "context" | ||
| "fmt" | ||
| "io" | ||
| "net/http" | ||
| "os" | ||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| type LogChunk struct { | ||
| Data []byte | ||
| Sequence int | ||
| } | ||
|
|
||
| type Buffer struct { | ||
| buf []byte | ||
| } | ||
|
|
||
| func (l *Buffer) Write(b []byte) (int, error) { | ||
| l.buf = append(l.buf, b...) | ||
| return len(b), nil | ||
| } | ||
|
|
||
| func (l *Buffer) ReadAndFlush() []byte { | ||
| buf := l.buf | ||
| l.buf = []byte{} | ||
| return buf | ||
| } | ||
|
|
||
| type LogStreamer struct { | ||
| jobId string | ||
| logsWriter io.Writer | ||
| logs *Buffer | ||
| logOrder int | ||
| queue chan LogChunk | ||
| maxSize int | ||
| active bool | ||
| wg sync.WaitGroup | ||
| ctx context.Context | ||
| cancel context.CancelFunc | ||
| serviceHost string | ||
| } | ||
|
|
||
| func NewLogStreamer(jobId string) *LogStreamer { | ||
| ctx, cancel := context.WithCancel(context.Background()) | ||
| return &LogStreamer{ | ||
| jobId: jobId, | ||
| logs: &Buffer{}, | ||
| logOrder: 0, | ||
| maxSize: 10 * 1024 * 1024, // 10MB | ||
| active: false, | ||
| queue: make(chan LogChunk, 100), | ||
| ctx: ctx, | ||
| cancel: cancel, | ||
| } | ||
| } | ||
|
|
||
| func (ls *LogStreamer) Start() { | ||
| ls.active = true | ||
| ls.wg.Add(2) | ||
| go ls.StreamLogs() | ||
| go ls.RetrieveAndUploadChunk() | ||
| } | ||
|
|
||
| func (ls *LogStreamer) Stop() { | ||
| ls.ChunkLogs(ls.logs.ReadAndFlush()) | ||
| ls.cancel() | ||
| ls.wg.Wait() | ||
| close(ls.queue) | ||
| } | ||
|
|
||
| // StreamLogs streams the result from the logs and chunks them into the queue | ||
| func (ls *LogStreamer) StreamLogs() { | ||
| defer ls.wg.Done() | ||
| for { | ||
| // read the logs and chunk them then push into the queue | ||
| logs := ls.logs.ReadAndFlush() | ||
| if len(logs) > 0 { | ||
| ls.ChunkLogs(logs) | ||
| } | ||
| select { | ||
| case <-time.After(1 * time.Second): | ||
| case <-ls.ctx.Done(): | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (ls *LogStreamer) ChunkLogs(data []byte) { | ||
| chunkSize := ls.maxSize | ||
| for i := 0; i < len(data); i += chunkSize { | ||
| chunkData := data[i:min(i+chunkSize, len(data))] | ||
| logChunk := LogChunk{Data: chunkData, Sequence: ls.logOrder} | ||
| ls.queue <- logChunk | ||
| ls.logOrder++ | ||
| } | ||
| } | ||
|
|
||
| // function that takes chunk from the queue and write it to file | ||
| func (ls *LogStreamer) RetrieveAndUploadChunk() { | ||
| defer ls.wg.Done() | ||
| for { | ||
| select { | ||
| case chunk, ok := <-ls.queue: | ||
| if !ok { | ||
| fmt.Println("Queue closed.. exiting") | ||
| return | ||
| } | ||
| // write the chunk to file | ||
| ls.WriteToFile(chunk) | ||
| // send the chunk to server | ||
| ls.UploadChunk(chunk) | ||
| case <-ls.ctx.Done(): | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (ls *LogStreamer) WriteToFile(logChunk LogChunk) { | ||
| fileName := fmt.Sprintf("logs/%s-%d.log", ls.jobId, logChunk.Sequence) | ||
| file, err := os.OpenFile(fileName, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644) | ||
| if err != nil { | ||
| fmt.Println("Error writing to file", err) | ||
| return | ||
| } | ||
| defer file.Close() | ||
| file.Write(logChunk.Data) | ||
| } | ||
|
|
||
| func (ls *LogStreamer) UploadChunk(logChunk LogChunk) { | ||
| // send the chunk to server | ||
| url := fmt.Sprintf("%s/job/logs?jobId=%s&sequence=%d", ls.serviceHost, ls.jobId, logChunk.Sequence) | ||
| req, err := http.NewRequest("POST", url, bytes.NewBuffer(logChunk.Data)) | ||
| if err != nil { | ||
| fmt.Println("Error creating request", err) | ||
| return | ||
| } | ||
| req.Header.Set("Content-Type", "text/plain") | ||
| client := &http.Client{} | ||
| resp, err := client.Do(req) | ||
| if err != nil { | ||
| fmt.Println("Error uploading chunk", err) | ||
| } | ||
| defer resp.Body.Close() | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
you should use sdk v2