-
Notifications
You must be signed in to change notification settings - Fork 21
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Support v3io stream consumer groups (#48)
- Loading branch information
Showing
23 changed files
with
2,140 additions
and
96 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "math" | ||
| "math/rand" | ||
| "sync/atomic" | ||
| "time" | ||
| ) | ||
|
|
||
| // Backoff is a time.Duration counter, starting at Min. After every call to | ||
| // the Duration method the current timing is multiplied by Factor, but it | ||
| // never exceeds Max. | ||
| // | ||
| // Backoff is not generally concurrent-safe, but the ForAttempt method can | ||
| // be used concurrently. | ||
| type Backoff struct { | ||
| attempt uint64 | ||
| // Factor is the multiplying factor for each increment step | ||
| Factor float64 | ||
| // Jitter eases contention by randomizing backoff steps | ||
| Jitter bool | ||
| // Min and Max are the minimum and maximum values of the counter | ||
| Min, Max time.Duration | ||
| } | ||
|
|
||
| // Duration returns the duration for the current attempt before incrementing | ||
| // the attempt counter. See ForAttempt. | ||
| func (b *Backoff) Duration() time.Duration { | ||
| d := b.ForAttempt(float64(atomic.AddUint64(&b.attempt, 1) - 1)) | ||
| return d | ||
| } | ||
|
|
||
| const maxInt64 = float64(math.MaxInt64 - 512) | ||
|
|
||
| // ForAttempt returns the duration for a specific attempt. This is useful if | ||
| // you have a large number of independent Backoffs, but don't want use | ||
| // unnecessary memory storing the Backoff parameters per Backoff. The first | ||
| // attempt should be 0. | ||
| // | ||
| // ForAttempt is concurrent-safe. | ||
| func (b *Backoff) ForAttempt(attempt float64) time.Duration { | ||
| // Zero-values are nonsensical, so we use | ||
| // them to apply defaults | ||
| min := b.Min | ||
| if min <= 0 { | ||
| min = 100 * time.Millisecond | ||
| } | ||
| max := b.Max | ||
| if max <= 0 { | ||
| max = 10 * time.Second | ||
| } | ||
| if min >= max { | ||
| // short-circuit | ||
| return max | ||
| } | ||
| factor := b.Factor | ||
| if factor <= 0 { | ||
| factor = 2 | ||
| } | ||
| //calculate this duration | ||
| minf := float64(min) | ||
| durf := minf * math.Pow(factor, attempt) | ||
| if b.Jitter { | ||
| durf = rand.Float64()*(durf-minf) + minf | ||
| } | ||
| //ensure float64 wont overflow int64 | ||
| if durf > maxInt64 { | ||
| return max | ||
| } | ||
| dur := time.Duration(durf) | ||
| //keep within bounds | ||
| if dur < min { | ||
| return min | ||
| } | ||
| if dur > max { | ||
| return max | ||
| } | ||
| return dur | ||
| } | ||
|
|
||
| // Reset restarts the current attempt counter at zero. | ||
| func (b *Backoff) Reset() { | ||
| atomic.StoreUint64(&b.attempt, 0) | ||
| } | ||
|
|
||
| // Attempt returns the current attempt counter value. | ||
| func (b *Backoff) Attempt() float64 { | ||
| return float64(atomic.LoadUint64(&b.attempt)) | ||
| } | ||
|
|
||
| // Copy returns a backoff with equals constraints as the original | ||
| func (b *Backoff) Copy() *Backoff { | ||
| return &Backoff{ | ||
| Factor: b.Factor, | ||
| Jitter: b.Jitter, | ||
| Min: b.Min, | ||
| Max: b.Max, | ||
| } | ||
| } |
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,142 @@ | ||
| package common | ||
|
|
||
| import ( | ||
| "context" | ||
| "reflect" | ||
| "runtime" | ||
| "time" | ||
|
|
||
| "github.com/nuclio/errors" | ||
| "github.com/nuclio/logger" | ||
| ) | ||
|
|
||
| func getFunctionName(fn interface{}) string { | ||
| return runtime.FuncForPC(reflect.ValueOf(fn).Pointer()).Name() | ||
| } | ||
|
|
||
| // give either retryInterval or backoff | ||
| func RetryFunc(ctx context.Context, | ||
| loggerInstance logger.Logger, | ||
| attempts int, | ||
| retryInterval *time.Duration, | ||
| backoff *Backoff, | ||
| fn func(int) (bool, error)) error { | ||
|
|
||
| var err error | ||
| var retry bool | ||
|
|
||
| for attempt := 1; attempt <= attempts; attempt++ { | ||
| retry, err = fn(attempt) | ||
|
|
||
| // if there's no need to retry - we're done | ||
| if !retry { | ||
| return err | ||
| } | ||
|
|
||
| // are we out of time? | ||
| if ctx.Err() != nil { | ||
|
|
||
| loggerInstance.WarnWithCtx(ctx, | ||
| "Context error detected during retries", | ||
| "ctxErr", ctx.Err(), | ||
| "previousErr", err, | ||
| "function", getFunctionName(fn), | ||
| "attempt", attempt) | ||
|
|
||
| // return the error if one was provided | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return ctx.Err() | ||
| } | ||
|
|
||
| if backoff != nil { | ||
| time.Sleep(backoff.Duration()) | ||
| } else { | ||
| if retryInterval == nil { | ||
| return errors.New("Either retry interval or backoff must be given") | ||
| } | ||
| time.Sleep(*retryInterval) | ||
| } | ||
| } | ||
|
|
||
| // attempts exhausted and we're unsuccessful | ||
| // Return the original error for later checking | ||
| loggerInstance.WarnWithCtx(ctx, | ||
| "Failed final attempt to invoke function", | ||
| "function", getFunctionName(fn), | ||
| "err", err, | ||
| "attempts", attempts) | ||
|
|
||
| // this shouldn't happen | ||
| if err == nil { | ||
| loggerInstance.ErrorWithCtx(ctx, | ||
| "Failed final attempt to invoke function, but error is nil. This shouldn't happen", | ||
| "function", getFunctionName(fn), | ||
| "err", err, | ||
| "attempts", attempts) | ||
| return errors.New("Failed final attempt to invoke function without proper error supplied") | ||
| } | ||
| return err | ||
| } | ||
|
|
||
| func MakeRange(min, max int) []int { | ||
| a := make([]int, max-min+1) | ||
| for i := range a { | ||
| a[i] = min + i | ||
| } | ||
| return a | ||
| } | ||
|
|
||
| func IntSliceContainsInt(slice []int, number int) bool { | ||
| for _, intInSlice := range slice { | ||
| if intInSlice == number { | ||
| return true | ||
| } | ||
| } | ||
|
|
||
| return false | ||
| } | ||
|
|
||
| func IntSlicesEqual(slice1 []int, slice2 []int) bool { | ||
| if len(slice1) != len(slice2) { | ||
| return false | ||
| } | ||
|
|
||
| for intIndex := 0; intIndex < len(slice1); intIndex++ { | ||
| if slice1[intIndex] != slice2[intIndex] { | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| return true | ||
| } | ||
|
|
||
| func Uint64SlicesEqual(slice1 []uint64, slice2 []uint64) bool { | ||
| if len(slice1) != len(slice2) { | ||
| return false | ||
| } | ||
|
|
||
| for intIndex := 0; intIndex < len(slice1); intIndex++ { | ||
| if slice1[intIndex] != slice2[intIndex] { | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| return true | ||
| } | ||
|
|
||
| func StringSlicesEqual(slice1 []string, slice2 []string) bool { | ||
| if len(slice1) != len(slice2) { | ||
| return false | ||
| } | ||
|
|
||
| for stringIndex := 0; stringIndex < len(slice1); stringIndex++ { | ||
| if slice1[stringIndex] != slice2[stringIndex] { | ||
| return false | ||
| } | ||
| } | ||
|
|
||
| return true | ||
| } |
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.