-
Notifications
You must be signed in to change notification settings - Fork 18
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- use new rabbitmq pubsub code. fixes #34 fixes #27 - remove executor goroutines in favor of just spawing as many goroutines as are needed to handle alerting load. - dont retry failed jobs. The code that was in place didnt work due to the lru cache, retried jobs would just be skipped due to being in the cache already. issue #14
- Loading branch information
Showing
11 changed files
with
275 additions
and
273 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,69 @@ | ||
| package alerting | ||
|
|
||
| type JobQueue interface { | ||
| Put(job *Job) | ||
| import ( | ||
| "encoding/json" | ||
| "strconv" | ||
|
|
||
| "github.com/grafana/grafana/pkg/log" | ||
| "github.com/raintank/worldping-api/pkg/alerting/jobqueue" | ||
| "github.com/raintank/worldping-api/pkg/setting" | ||
| ) | ||
|
|
||
| var ( | ||
| pubChan chan jobqueue.Message | ||
| subChan chan jobqueue.Message | ||
| ) | ||
|
|
||
| func InitJobQueue(jobQueue chan<- *Job) { | ||
|
|
||
| if setting.Rabbitmq.Enabled { | ||
| pubChan = make(chan jobqueue.Message, setting.PreAMQPJobQueueSize) | ||
| // use rabbitmq for message distribution. | ||
|
|
||
| //subchan is unbuffered as the consumer creates a goroutine for | ||
| // every message recieved. | ||
| subChan = make(chan jobqueue.Message) | ||
| go jobqueue.Run(setting.Rabbitmq.Url, "alertingJobs", pubChan, subChan) | ||
| go handleJobs(subChan, jobQueue) | ||
| } else { | ||
| pubChan = make(chan jobqueue.Message, setting.InternalJobQueueSize) | ||
| // handle all message written to the publish chan. | ||
| go handleJobs(pubChan, jobQueue) | ||
| } | ||
| return | ||
| } | ||
|
|
||
| func PublishJob(job *Job) error { | ||
| body, err := json.Marshal(job) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| msg := jobqueue.Message{ | ||
| RoutingKey: strconv.FormatInt(job.CheckId, 10), | ||
| Payload: body, | ||
| } | ||
| if setting.Rabbitmq.Enabled { | ||
| jobQueuePreAMQPItems.Value(int64(len(pubChan))) | ||
| jobQueuePreAMQPSize.Value(int64(setting.PreAMQPJobQueueSize)) | ||
| } else { | ||
| jobQueueInternalItems.Value(int64(len(pubChan))) | ||
| jobQueueInternalSize.Value(int64(setting.InternalJobQueueSize)) | ||
| } | ||
|
|
||
| pubChan <- msg | ||
| return nil | ||
| } | ||
|
|
||
| func handleJobs(c chan jobqueue.Message, jobQueue chan<- *Job) { | ||
| for m := range c { | ||
| go func(msg jobqueue.Message) { | ||
| j := &Job{} | ||
| err := json.Unmarshal(msg.Payload, j) | ||
| if err != nil { | ||
| log.Error(3, "unable to unmarshal Job. %s", err) | ||
| return | ||
| } | ||
| jobQueue <- j | ||
| }(m) | ||
| } | ||
| } |
Oops, something went wrong.