-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
12 changed files
with
549 additions
and
25 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| package postq | ||
|
|
||
| import ( | ||
| "container/ring" | ||
| gocontext "context" | ||
| "fmt" | ||
|
|
||
| "github.com/flanksource/duty/context" | ||
| "github.com/flanksource/duty/models" | ||
| ) | ||
|
|
||
| // AsyncEventHandlerFunc processes multiple events and returns the failed ones | ||
| type AsyncEventHandlerFunc func(context.Context, models.Events) models.Events | ||
|
|
||
| type AsyncEventConsumer struct { | ||
| eventLog *ring.Ring | ||
|
|
||
| // Name of the events in the push queue to watch for. | ||
| WatchEvents []string | ||
|
|
||
| // Number of events to be fetched and processed at a time. | ||
| BatchSize int | ||
|
|
||
| // An async event handler that consumes events. | ||
| Consumer AsyncEventHandlerFunc | ||
|
|
||
| // ConsumerOption is the configuration for the PGConsumer. | ||
| ConsumerOption *ConsumerOption | ||
|
|
||
| // EventFetcherOption contains configuration on how the events should be fetched. | ||
| EventFetcherOption *EventFetcherOption | ||
| } | ||
|
|
||
| // RecordEvents will record all the events fetched by the consumer in a ring buffer. | ||
| func (t *AsyncEventConsumer) RecordEvents(size int) { | ||
| t.eventLog = ring.New(size) | ||
| } | ||
|
|
||
| func (t AsyncEventConsumer) GetRecords() ([]models.Event, error) { | ||
| if t.eventLog == nil { | ||
| return nil, fmt.Errorf("event log is not initialized") | ||
| } | ||
|
|
||
| return getRecords(t.eventLog), nil | ||
| } | ||
|
|
||
| func (t *AsyncEventConsumer) Handle(ctx context.Context) (int, error) { | ||
| ctx = ctx.WithName("postq").Fast() | ||
| tx := ctx.DB().Begin() | ||
| defer tx.Rollback() //nolint:errcheck | ||
|
|
||
| events, err := fetchEvents(ctx, tx, t.WatchEvents, t.BatchSize, t.EventFetcherOption) | ||
| if err != nil { | ||
| return 0, fmt.Errorf("error fetching events: %w", err) | ||
| } | ||
|
|
||
| if t.eventLog != nil { | ||
| for _, event := range events { | ||
| t.eventLog.Value = event | ||
| t.eventLog = t.eventLog.Next() | ||
| } | ||
| } | ||
| c := ctx.Wrap(gocontext.Background()) | ||
| failedEvents := t.Consumer(c, events) | ||
| if err := failedEvents.Recreate(ctx, tx); err != nil { | ||
| ctx.Debugf("error saving event attempt updates to event_queue: %v\n", err) | ||
| } | ||
|
|
||
| return len(events), tx.Commit().Error | ||
| } | ||
|
|
||
| func (t AsyncEventConsumer) EventConsumer() (*PGConsumer, error) { | ||
| return NewPGConsumer(t.Handle, t.ConsumerOption) | ||
| } | ||
|
|
||
| // AsyncHandler converts the given user defined handler into a async event handler. | ||
| func AsyncHandler(fn func(ctx context.Context, e models.Events) models.Events) AsyncEventHandlerFunc { | ||
| return func(ctx context.Context, e models.Events) models.Events { | ||
| return fn(ctx, e) | ||
| } | ||
| } |
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,87 @@ | ||
| package postq | ||
|
|
||
| import ( | ||
| "strings" | ||
|
|
||
| "github.com/flanksource/duty/context" | ||
| "github.com/flanksource/duty/models" | ||
| "github.com/lib/pq" | ||
| "github.com/samber/oops" | ||
| "gorm.io/gorm" | ||
| ) | ||
|
|
||
| type EventFetcherOption struct { | ||
| // MaxAttempts is the number of times an event is attempted to process | ||
| // default: 3 | ||
| MaxAttempts int | ||
|
|
||
| // BaseDelay is the base delay between retries | ||
| // default: 60 seconds | ||
| BaseDelay int | ||
|
|
||
| // Exponent is the exponent of the base delay | ||
| // default: 5 (along with baseDelay = 60, the retries are 1, 6, 31, 156 (in minutes)) | ||
| Exponent int | ||
| } | ||
|
|
||
| // fetchEvents fetches given watch events from the `event_queue` table. | ||
| func fetchEvents(ctx context.Context, tx *gorm.DB, watchEvents []string, batchSize int, opts *EventFetcherOption) ([]models.Event, error) { | ||
| if batchSize == 0 { | ||
| batchSize = 1 | ||
| } | ||
|
|
||
| const selectEventsQuery = ` | ||
| DELETE FROM event_queue | ||
| WHERE id IN ( | ||
| SELECT id FROM event_queue | ||
| WHERE | ||
| attempts <= @MaxAttempts AND | ||
| name = ANY(@Events) AND | ||
| (last_attempt IS NULL OR last_attempt <= NOW() - INTERVAL '1 SECOND' * @BaseDelay * POWER(attempts, @Exponent)) | ||
| ORDER BY priority DESC, created_at ASC | ||
| FOR UPDATE SKIP LOCKED | ||
| LIMIT @BatchSize | ||
| ) | ||
| RETURNING * | ||
| ` | ||
|
|
||
| type EventArgs struct { | ||
| Events pq.StringArray | ||
| BatchSize int | ||
| MaxAttempts int | ||
| BaseDelay int | ||
| Exponent int | ||
| } | ||
|
|
||
| args := EventArgs{ | ||
| Events: watchEvents, | ||
| BatchSize: batchSize, | ||
| MaxAttempts: 3, | ||
| BaseDelay: 60, | ||
| Exponent: 5, | ||
| } | ||
|
|
||
| if opts != nil { | ||
| if opts.MaxAttempts > 0 { | ||
| args.MaxAttempts = opts.MaxAttempts | ||
| } | ||
|
|
||
| if opts.BaseDelay > 0 { | ||
| args.BaseDelay = opts.BaseDelay | ||
| } | ||
|
|
||
| if opts.Exponent > 0 { | ||
| args.Exponent = opts.Exponent | ||
| } | ||
| } | ||
| var events []models.Event | ||
|
|
||
| if err := tx.Raw(selectEventsQuery, args).Scan(&events).Error; err != nil { | ||
| return nil, oops.Tags("db").Wrap(err) | ||
| } | ||
|
|
||
| if len(events) > 0 { | ||
| ctx.Tracef("queue=%s fetched=%d", strings.Join(watchEvents, ","), len(events)) | ||
| } | ||
| return events, 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| package pg | ||
|
|
||
| import ( | ||
| gocontext "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/flanksource/duty/context" | ||
| "github.com/sethvargo/go-retry" | ||
| ) | ||
|
|
||
| // Defaults ... | ||
| var ( | ||
| DBReconnectMaxDuration = time.Minute * 5 | ||
| DBReconnectBackoffBaseDuration = time.Second | ||
| ) | ||
|
|
||
| // Listen listens to postgres notifications. | ||
| // On failure, it'll keep retrying with backoff | ||
| func Listen(ctx context.Context, channel string, pgNotify chan<- string) { | ||
| var listen = func(ctx context.Context, pgNotify chan<- string) error { | ||
| conn, err := ctx.Pool().Acquire(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("error acquiring database connection: %v", err) | ||
| } | ||
| defer conn.Release() | ||
|
|
||
| _, err = conn.Exec(ctx, fmt.Sprintf("LISTEN %s", channel)) | ||
| if err != nil { | ||
| return fmt.Errorf("error listening to database notifications: %v", err) | ||
| } | ||
|
|
||
| for { | ||
| n, err := conn.Conn().WaitForNotification(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("error listening to database notifications: %v", err) | ||
| } | ||
|
|
||
| pgNotify <- n.Payload | ||
| } | ||
| } | ||
|
|
||
| // retry on failure. | ||
| for { | ||
| backoff := retry.WithMaxDuration(DBReconnectMaxDuration, retry.NewExponential(DBReconnectBackoffBaseDuration)) | ||
| err := retry.Do(ctx, backoff, func(retryContext gocontext.Context) error { | ||
| ctx := retryContext.(context.Context) | ||
| if err := listen(ctx, pgNotify); err != nil { | ||
| return retry.RetryableError(err) | ||
| } | ||
|
|
||
| return nil | ||
| }) | ||
|
|
||
| fmt.Printf("failed to connect to database: %v\n", 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,50 @@ | ||
| package pg | ||
|
|
||
| import ( | ||
| "strings" | ||
|
|
||
| "github.com/flanksource/duty/context" | ||
| ) | ||
|
|
||
| // notifyRouter distributes the pgNotify event to multiple channels | ||
| // based on the payload. | ||
| type notifyRouter struct { | ||
| registry map[string]chan string | ||
| } | ||
|
|
||
| func NewNotifyRouter() *notifyRouter { | ||
| return ¬ifyRouter{ | ||
| registry: make(map[string]chan string), | ||
| } | ||
| } | ||
|
|
||
| // RegisterRoutes creates a single channel for the given routes and returns it. | ||
| func (t *notifyRouter) RegisterRoutes(routes ...string) <-chan string { | ||
| pgNotifyChannel := make(chan string) | ||
| for _, we := range routes { | ||
| t.registry[we] = pgNotifyChannel | ||
| } | ||
|
|
||
| return pgNotifyChannel | ||
| } | ||
|
|
||
| func (t *notifyRouter) Run(ctx context.Context, channel string) { | ||
| eventQueueNotifyChannel := make(chan string) | ||
| go Listen(ctx, channel, eventQueueNotifyChannel) | ||
|
|
||
| for payload := range eventQueueNotifyChannel { | ||
| if _, ok := t.registry[payload]; !ok || payload == "" { | ||
| continue | ||
| } | ||
|
|
||
| // The original payload is expected to be in the form of | ||
| // <route> <...optional payload> | ||
| fields := strings.Fields(payload) | ||
| route := fields[0] | ||
| derivedPayload := strings.Join(fields[1:], " ") | ||
|
|
||
| if ch, ok := t.registry[route]; ok { | ||
| ch <- derivedPayload | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.