forked from RedHatInsights/sources-api-go
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
move job engine to a new worker pod, using redis as the backend
- Loading branch information
1 parent
527a42e
commit 2b8b93b
Showing
21 changed files
with
290 additions
and
91 deletions.
There are no files selected for viewing
This file contains 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 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 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 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 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 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 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 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,16 @@ | ||
| package mocks | ||
|
|
||
| import "github.com/RedHatInsights/sources-api-go/kafka" | ||
|
|
||
| type MockSender struct { | ||
| Hit int | ||
| Headers []kafka.Header | ||
| Body string | ||
| } | ||
|
|
||
| func (m *MockSender) RaiseEvent(_ string, b []byte, headers []kafka.Header) error { | ||
| m.Headers = headers | ||
| m.Body = string(b) | ||
| m.Hit++ | ||
| return nil | ||
| } |
This file contains 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 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,67 @@ | ||
| package jobs | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "fmt" | ||
|
|
||
| l "github.com/RedHatInsights/sources-api-go/logger" | ||
| "github.com/RedHatInsights/sources-api-go/redis" | ||
| ) | ||
|
|
||
| type JobRequest struct { | ||
| JobName string | ||
| JobRaw []byte | ||
| Job Job | ||
| } | ||
|
|
||
| // implementing binary mashaler/unmarshaler interfaces for redis encoding/decoding. | ||
| func (jr JobRequest) MarshalBinary() (data []byte, err error) { | ||
| return json.Marshal(&jr) | ||
| } | ||
| func (jr *JobRequest) UnmarshalBinary(data []byte) error { | ||
| return json.Unmarshal(data, jr) | ||
| } | ||
|
|
||
| func (jr *JobRequest) Parse() error { | ||
| switch jr.JobName { | ||
| case "SuperkeyDestroyJob": | ||
| sdj := SuperkeyDestroyJob{} | ||
| err := json.Unmarshal([]byte(jr.JobRaw), &sdj) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| jr.Job = &sdj | ||
| case "AsyncDestroyJob": | ||
| adj := AsyncDestroyJob{} | ||
| err := json.Unmarshal(jr.JobRaw, &adj) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| jr.Job = &adj | ||
| default: | ||
| l.Log.Warnf("Unsupported job: %v", jr.JobName) | ||
| return fmt.Errorf("unsupported job %v", jr.JobName) | ||
| } | ||
|
|
||
| l.Log.Debugf("Successfully parsed job %v, args %v", jr.Job.Name(), jr.Job.Arguments()) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // Throws a `job` on the redis list to be picked up by the worker | ||
| func Enqueue(j Job) { | ||
| l.Log.Infof("Submitting job %v to redis with %v", j.Name(), j.Arguments()) | ||
|
|
||
| req := JobRequest{ | ||
| JobName: j.Name(), | ||
| JobRaw: j.ToJSON(), | ||
| } | ||
|
|
||
| err := redis.Client.RPush(context.Background(), workQueue, req).Err() | ||
| if err != nil { | ||
| l.Log.Warnf("Failed to submit job: %v", err) | ||
| } | ||
| } |
This file contains 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 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
Oops, something went wrong.