|
|
@@ -128,9 +128,19 @@ func (c *cowHostList) remove(ip net.IP) bool {
|
|
|
// exposes the correct functions for the retry policy logic to evaluate correctly.
|
|
|
type RetryableQuery interface {
|
|
|
Attempts() int
|
|
|
+ SetConsistency(c Consistency)
|
|
|
GetConsistency() Consistency
|
|
|
}
|
|
|
|
|
|
+type RetryType uint16
|
|
|
+
|
|
|
+const (
|
|
|
+ Retry RetryType = 0x00 // retry on same connection
|
|
|
+ RetryNextHost RetryType = 0x01 // retry on another connection
|
|
|
+ Ignore RetryType = 0x02 // ignore error and return result
|
|
|
+ Rethrow RetryType = 0x03 // raise error and stop retrying
|
|
|
+)
|
|
|
+
|
|
|
// RetryPolicy interface is used by gocql to determine if a query can be attempted
|
|
|
// again after a retryable error has been received. The interface allows gocql
|
|
|
// users to implement their own logic to determine if a query can be attempted
|
|
|
@@ -140,6 +150,7 @@ type RetryableQuery interface {
|
|
|
// interface.
|
|
|
type RetryPolicy interface {
|
|
|
Attempt(RetryableQuery) bool
|
|
|
+ GetRetryType(error) RetryType
|
|
|
}
|
|
|
|
|
|
// SimpleRetryPolicy has simple logic for attempting a query a fixed number of times.
|
|
|
@@ -162,6 +173,10 @@ func (s *SimpleRetryPolicy) Attempt(q RetryableQuery) bool {
|
|
|
return q.Attempts() <= s.NumRetries
|
|
|
}
|
|
|
|
|
|
+func (s *SimpleRetryPolicy) GetRetryType(err error) RetryType {
|
|
|
+ return RetryNextHost
|
|
|
+}
|
|
|
+
|
|
|
// ExponentialBackoffRetryPolicy sleeps between attempts
|
|
|
type ExponentialBackoffRetryPolicy struct {
|
|
|
NumRetries int
|
|
|
@@ -176,23 +191,92 @@ func (e *ExponentialBackoffRetryPolicy) Attempt(q RetryableQuery) bool {
|
|
|
return true
|
|
|
}
|
|
|
|
|
|
-func (e *ExponentialBackoffRetryPolicy) napTime(attempts int) time.Duration {
|
|
|
- if e.Min <= 0 {
|
|
|
- e.Min = 100 * time.Millisecond
|
|
|
+// used to calculate exponentially growing time
|
|
|
+func getExponentialTime(min time.Duration, max time.Duration, attempts int) time.Duration {
|
|
|
+ if min <= 0 {
|
|
|
+ min = 100 * time.Millisecond
|
|
|
}
|
|
|
- if e.Max <= 0 {
|
|
|
- e.Max = 10 * time.Second
|
|
|
+ if max <= 0 {
|
|
|
+ max = 10 * time.Second
|
|
|
}
|
|
|
- minFloat := float64(e.Min)
|
|
|
+ minFloat := float64(min)
|
|
|
napDuration := minFloat * math.Pow(2, float64(attempts-1))
|
|
|
// add some jitter
|
|
|
napDuration += rand.Float64()*minFloat - (minFloat / 2)
|
|
|
- if napDuration > float64(e.Max) {
|
|
|
- return time.Duration(e.Max)
|
|
|
+ if napDuration > float64(max) {
|
|
|
+ return time.Duration(max)
|
|
|
}
|
|
|
return time.Duration(napDuration)
|
|
|
}
|
|
|
|
|
|
+func (e *ExponentialBackoffRetryPolicy) GetRetryType(err error) RetryType {
|
|
|
+ return RetryNextHost
|
|
|
+}
|
|
|
+
|
|
|
+// DowngradingConsistencyRetryPolicy: Next retry will be with the next consistency level
|
|
|
+// provided in the slice
|
|
|
+//
|
|
|
+// On a read timeout: the operation is retried with the next provided consistency
|
|
|
+// level.
|
|
|
+//
|
|
|
+// On a write timeout: if the operation is an :attr:`~.UNLOGGED_BATCH`
|
|
|
+// and at least one replica acknowledged the write, the operation is
|
|
|
+// retried with the next consistency level. Furthermore, for other
|
|
|
+// write types, if at least one replica acknowledged the write, the
|
|
|
+// timeout is ignored.
|
|
|
+//
|
|
|
+// On an unavailable exception: if at least one replica is alive, the
|
|
|
+// operation is retried with the next provided consistency level.
|
|
|
+
|
|
|
+type DowngradingConsistencyRetryPolicy struct {
|
|
|
+ ConsistencyLevelsToTry []Consistency
|
|
|
+}
|
|
|
+
|
|
|
+func (d *DowngradingConsistencyRetryPolicy) Attempt(q RetryableQuery) bool {
|
|
|
+ currentAttempt := q.Attempts()
|
|
|
+
|
|
|
+ if currentAttempt > len(d.ConsistencyLevelsToTry) {
|
|
|
+ return false
|
|
|
+ } else if currentAttempt > 0 {
|
|
|
+ q.SetConsistency(d.ConsistencyLevelsToTry[currentAttempt-1])
|
|
|
+ if gocqlDebug {
|
|
|
+ Logger.Printf("%T: set consistency to %q\n",
|
|
|
+ d,
|
|
|
+ d.ConsistencyLevelsToTry[currentAttempt-1])
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return true
|
|
|
+}
|
|
|
+
|
|
|
+func (d *DowngradingConsistencyRetryPolicy) GetRetryType(err error) RetryType {
|
|
|
+ switch t := err.(type) {
|
|
|
+ case *RequestErrUnavailable:
|
|
|
+ if t.Alive > 0 {
|
|
|
+ return Retry
|
|
|
+ }
|
|
|
+ return Rethrow
|
|
|
+ case *RequestErrWriteTimeout:
|
|
|
+ if t.WriteType == "SIMPLE" || t.WriteType == "BATCH" || t.WriteType == "COUNTER" {
|
|
|
+ if t.Received > 0 {
|
|
|
+ return Ignore
|
|
|
+ }
|
|
|
+ return Rethrow
|
|
|
+ }
|
|
|
+ if t.WriteType == "UNLOGGED_BATCH" {
|
|
|
+ return Retry
|
|
|
+ }
|
|
|
+ return Rethrow
|
|
|
+ case *RequestErrReadTimeout:
|
|
|
+ return Retry
|
|
|
+ default:
|
|
|
+ return RetryNextHost
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (e *ExponentialBackoffRetryPolicy) napTime(attempts int) time.Duration {
|
|
|
+ return getExponentialTime(e.Min, e.Max, attempts)
|
|
|
+}
|
|
|
+
|
|
|
type HostStateNotifier interface {
|
|
|
AddHost(host *HostInfo)
|
|
|
RemoveHost(host *HostInfo)
|
|
|
@@ -706,3 +790,45 @@ func (d *dcAwareRR) Pick(q ExecutableQuery) NextHost {
|
|
|
return (*selectedHost)(host)
|
|
|
}
|
|
|
}
|
|
|
+
|
|
|
+// ReconnectionPolicy interface is used by gocql to determine if reconnection
|
|
|
+// can be attempted after connection error. The interface allows gocql users
|
|
|
+// to implement their own logic to determine how to attempt reconnection.
|
|
|
+//
|
|
|
+type ReconnectionPolicy interface {
|
|
|
+ GetInterval(currentRetry int) time.Duration
|
|
|
+ GetMaxRetries() int
|
|
|
+}
|
|
|
+
|
|
|
+// ConstantReconnectionPolicy has simple logic for returning a fixed reconnection interval.
|
|
|
+//
|
|
|
+// Examples of usage:
|
|
|
+//
|
|
|
+// cluster.ReconnectionPolicy = &gocql.ConstantReconnectionPolicy{MaxRetries: 10, Interval: 8 * time.Second}
|
|
|
+//
|
|
|
+type ConstantReconnectionPolicy struct {
|
|
|
+ MaxRetries int
|
|
|
+ Interval time.Duration
|
|
|
+}
|
|
|
+
|
|
|
+func (c *ConstantReconnectionPolicy) GetInterval(currentRetry int) time.Duration {
|
|
|
+ return c.Interval
|
|
|
+}
|
|
|
+
|
|
|
+func (c *ConstantReconnectionPolicy) GetMaxRetries() int {
|
|
|
+ return c.MaxRetries
|
|
|
+}
|
|
|
+
|
|
|
+// ExponentialReconnectionPolicy returns a growing reconnection interval.
|
|
|
+type ExponentialReconnectionPolicy struct {
|
|
|
+ MaxRetries int
|
|
|
+ InitialInterval time.Duration
|
|
|
+}
|
|
|
+
|
|
|
+func (e *ExponentialReconnectionPolicy) GetInterval(currentRetry int) time.Duration {
|
|
|
+ return getExponentialTime(e.InitialInterval, math.MaxInt16*time.Second, e.GetMaxRetries())
|
|
|
+}
|
|
|
+
|
|
|
+func (e *ExponentialReconnectionPolicy) GetMaxRetries() int {
|
|
|
+ return e.MaxRetries
|
|
|
+}
|
