Add support for asynchronous dispatch #37
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,246 @@ | ||
| // Copyright 2020 Palantir Technologies, Inc. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package githubapp | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sync/atomic" | ||
|
|
||
| "github.com/pkg/errors" | ||
| "github.com/rcrowley/go-metrics" | ||
| "github.com/rs/zerolog" | ||
| ) | ||
|
|
||
| const ( | ||
| MetricsKeyQueueLength = "github.event.queued" | ||
| MetricsKeyActiveWorkers = "github.event.workers" | ||
| ) | ||
|
|
||
| var ( | ||
| ErrCapacityExceeded = errors.New("scheduler: capacity exceeded") | ||
| ) | ||
|
|
||
| // Dispatch is a webhook payload and the handler that handles it. | ||
| type Dispatch struct { | ||
| Handler EventHandler | ||
|
|
||
| EventType string | ||
| DeliveryID string | ||
| Payload []byte | ||
| } | ||
|
|
||
| // Execute calls the Dispatch's handler with the stored arguments. | ||
| func (d Dispatch) Execute(ctx context.Context) error { | ||
| return d.Handler.Handle(ctx, d.EventType, d.DeliveryID, d.Payload) | ||
| } | ||
|
|
||
| // AsyncErrorCallback is called by an asynchronous scheduler when an event | ||
| // handler returns an error. The error from the handler is passed directly as | ||
| // the final argument. | ||
| type AsyncErrorCallback func(ctx context.Context, err error) | ||
|
|
||
| // DefaultAsyncErrorCallback logs errors. | ||
| func DefaultAsyncErrorCallback(ctx context.Context, err error) { | ||
| zerolog.Ctx(ctx).Error().Err(err).Msg("Unexpected error handling webhook") | ||
| } | ||
|
|
||
| // ContextDeriver creates a new independent context from a request's context. | ||
| // The new context must be based on context.Background(), not the input. | ||
| type ContextDeriver func(context.Context) context.Context | ||
|
There was a problem hiding this comment. why not have the Scheduler decide how to derive the context? then we could have it feels like context deriving should be tied explicitly to schedule logic, rather than be swappable There was a problem hiding this comment. Context derivation is customizable because only the end user can know the context values that are relevant to copy to the new context for the handler, particularly if they've added custom middleware or other stuff before the event dispatcher. This felt like the right tradeoff to me, rather than having user implement their own schedulers to change the context, but maybe there are more elegant primitives to expose. Regarding the signature, I was lazy: |
||
|
|
||
| // DefaultContextDeriver copies the logger from the request's context to a new | ||
| // context. | ||
| func DefaultContextDeriver(ctx context.Context) context.Context { | ||
| newCtx := context.Background() | ||
|
|
||
| // this value is always unused by async schedulers, but is set for | ||
| // compatibility with existing handlers that call SetResponder | ||
| newCtx = InitializeResponder(newCtx) | ||
|
|
||
| return zerolog.Ctx(ctx).WithContext(newCtx) | ||
| } | ||
|
|
||
| // Scheduler is a strategy for executing event handlers. | ||
| // | ||
| // The Schedule method takes a Dispatch and executes it by calling the handler | ||
| // for the payload. The execution may be asynchronous, but the scheduler must | ||
| // create a new context in this case. The dispatcher waits for Schedule to | ||
| // return before responding to GitHub, so asynchronous schedulers should only | ||
| // return errors that happen during scheduling, not during execution. | ||
| // | ||
| // Schedule may return ErrCapacityExceeded if it cannot schedule or queue new | ||
| // events at the time of the call. | ||
| type Scheduler interface { | ||
| Schedule(ctx context.Context, d Dispatch) error | ||
| } | ||
|
|
||
| // SchedulerOption configures properties of a scheduler. | ||
| type SchedulerOption func(*scheduler) | ||
|
|
||
| // WithAsyncErrorCallback sets the error callback for an asynchronous | ||
| // scheduler. If not set, the scheduler uses DefaultAsyncErrorCallback. | ||
| func WithAsyncErrorCallback(onError AsyncErrorCallback) SchedulerOption { | ||
| return func(s *scheduler) { | ||
| if onError != nil { | ||
| s.onError = onError | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // WithContextDeriver sets the context deriver for an asynchronous scheduler. | ||
| // If not set, the scheduler uses DefaultContextDeriver. | ||
| func WithContextDeriver(deriver ContextDeriver) SchedulerOption { | ||
| return func(s *scheduler) { | ||
| if deriver != nil { | ||
| s.deriver = deriver | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // WithSchedulingMetrics enables metrics reporting for schedulers. | ||
| func WithSchedulingMetrics(r metrics.Registry) SchedulerOption { | ||
| return func(s *scheduler) { | ||
| metrics.NewRegisteredFunctionalGauge(MetricsKeyQueueLength, r, func() int64 { | ||
| return int64(len(s.queue)) | ||
| }) | ||
| metrics.NewRegisteredFunctionalGauge(MetricsKeyActiveWorkers, r, func() int64 { | ||
| return atomic.LoadInt64(&s.activeWorkers) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| type queueDispatch struct { | ||
| ctx context.Context | ||
| d Dispatch | ||
| } | ||
|
|
||
| // core functionality and options for (async) schedulers | ||
| type scheduler struct { | ||
| onError AsyncErrorCallback | ||
| deriver ContextDeriver | ||
|
|
||
| activeWorkers int64 | ||
| queue chan queueDispatch | ||
| } | ||
|
|
||
| func (s *scheduler) safeExecute(ctx context.Context, d Dispatch) { | ||
| var err error | ||
| defer func() { | ||
| if r := recover(); r != nil { | ||
| if rerr, ok := r.(error); ok { | ||
| err = rerr | ||
| } else { | ||
| err = fmt.Errorf("%v", r) | ||
| } | ||
| } | ||
| if err != nil && s.onError != nil { | ||
| s.onError(ctx, err) | ||
| } | ||
| atomic.AddInt64(&s.activeWorkers, -1) | ||
| }() | ||
|
|
||
| atomic.AddInt64(&s.activeWorkers, 1) | ||
| err = d.Execute(ctx) | ||
| } | ||
|
|
||
| func (s *scheduler) derive(ctx context.Context) context.Context { | ||
| if s.deriver == nil { | ||
| return ctx | ||
| } | ||
| return s.deriver(ctx) | ||
| } | ||
|
|
||
| // DefaultScheduler returns a scheduler that executes handlers in the go | ||
| // routine of the caller and returns any error. | ||
| func DefaultScheduler() Scheduler { | ||
| return &defaultScheduler{} | ||
| } | ||
|
|
||
| type defaultScheduler struct{} | ||
|
|
||
| func (s *defaultScheduler) Schedule(ctx context.Context, d Dispatch) error { | ||
| return d.Execute(ctx) | ||
| } | ||
|
|
||
| // AsyncScheduler returns a scheduler that executes handlers in new goroutines. | ||
| // Goroutines are not reused and there is no limit on the number created. | ||
| func AsyncScheduler(opts ...SchedulerOption) Scheduler { | ||
| s := &asyncScheduler{ | ||
| scheduler: scheduler{ | ||
| deriver: DefaultContextDeriver, | ||
| onError: DefaultAsyncErrorCallback, | ||
| }, | ||
| } | ||
| for _, opt := range opts { | ||
| opt(&s.scheduler) | ||
| } | ||
| return s | ||
| } | ||
|
|
||
| type asyncScheduler struct { | ||
| scheduler | ||
| } | ||
|
|
||
| func (s *asyncScheduler) Schedule(ctx context.Context, d Dispatch) error { | ||
| go s.safeExecute(s.derive(ctx), d) | ||
| return nil | ||
| } | ||
|
|
||
| // QueueAsyncScheduler returns a scheduler that executes handlers in a fixed | ||
| // number of worker goroutines. If no workers are available, events queue until | ||
| // the queue is full. | ||
| func QueueAsyncScheduler(queueSize int, workers int, opts ...SchedulerOption) Scheduler { | ||
| if queueSize < 0 { | ||
| panic("QueueAsyncScheduler: queue size must be non-negative") | ||
| } | ||
| if workers < 1 { | ||
| panic("QueueAsyncScheduler: worker count must be positive") | ||
| } | ||
|
|
||
| s := &queueScheduler{ | ||
| scheduler: scheduler{ | ||
| deriver: DefaultContextDeriver, | ||
| onError: DefaultAsyncErrorCallback, | ||
| queue: make(chan queueDispatch, queueSize), | ||
| }, | ||
| } | ||
| for _, opt := range opts { | ||
| opt(&s.scheduler) | ||
| } | ||
|
|
||
| for i := 0; i < workers; i++ { | ||
| go func() { | ||
| for d := range s.queue { | ||
| s.safeExecute(d.ctx, d.d) | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| return s | ||
| } | ||
|
|
||
| type queueScheduler struct { | ||
| scheduler | ||
| } | ||
|
|
||
| func (s *queueScheduler) Schedule(ctx context.Context, d Dispatch) error { | ||
| select { | ||
| case s.queue <- queueDispatch{ctx: s.derive(ctx), d: d}: | ||
| default: | ||
| return ErrCapacityExceeded | ||
| } | ||
| return nil | ||
| } | ||
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.
is it worth naming this something that's more descriptive of the function?
InlineSchedulerorBlockingSchedulerdont feel right... i can't come up with a better name, maybe default works.the question applies to most things named default below, and also in those cases,
Defaultmay be the better prefix.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.
I'm using default because these are the functions/types that are used when the caller does not override them via options. In this sense, it matches
DefaultErrorCallback,DefaultResponseCallback, andNewDefaultEventDispatcher, which already exist.I'm open to renaming the scheduler if you have suggestions and then documenting the default-ness in comments. I'm less convinced that renaming the other functions will improve clarity.