diff --git a/api/breaker/breaker.go b/api/breaker/breaker.go new file mode 100644 index 00000000000..177f87cfb39 --- /dev/null +++ b/api/breaker/breaker.go @@ -0,0 +1,435 @@ +// Copyright 2022 Gravitational, 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 breaker + +import ( + "fmt" + "net/http" + "sync" + "time" + + "github.com/gravitational/teleport/api/defaults" + "github.com/gravitational/teleport/api/utils" + + "github.com/gravitational/trace" + "github.com/jonboulle/clockwork" + "github.com/sirupsen/logrus" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// Metrics tallies success and failure counts +// for all executions performed by a CircuitBreaker +type Metrics struct { + // Executions the total number of times the breaker has executed within the interval + Executions uint32 + // Successes the number of successful executions + Successes uint32 + // Failures the total number of failed executions + Failures uint32 + // ConsecutiveSuccesses the number of consecutive successful executions + ConsecutiveSuccesses uint32 + // ConsecutiveFailures the number of consecutive failed executions + ConsecutiveFailures uint32 +} + +func (m *Metrics) String() string { + return fmt.Sprintf("Metrics(executions=%d, successes=%d, failures=%d, consecutiveSuccesses=%d, consecutiveFailures=%d)", m.Executions, m.Successes, m.Failures, m.ConsecutiveSuccesses, m.ConsecutiveFailures) +} + +// reset restores all counts to zero +func (m *Metrics) reset() { + *m = Metrics{} +} + +// success increments the counters tracking successful executions +// and resets the ConsecutiveFailures count +func (m *Metrics) success() { + m.Successes++ + m.ConsecutiveSuccesses++ + m.ConsecutiveFailures = 0 +} + +// failure increments the counters tracking failed executions +// and resets the ConsecutiveSuccesses count +func (m *Metrics) failure() { + m.Failures++ + m.ConsecutiveFailures++ + m.ConsecutiveSuccesses = 0 +} + +// execute increments Executions +func (m *Metrics) execute() { + m.Executions++ +} + +// State represents an operating state that a CircuitBreaker may be in. +type State int + +const ( + // StateStandby indicates the breaker is passing all requests and watching stats + StateStandby State = iota + // StateTripped indicates too many errors have occurred and requests are actively being rejected + StateTripped + // StateRecovering indicates the breaker is allowing some requests to go through and rejecting others + StateRecovering +) + +// String returns the string representation of a State +func (s State) String() string { + switch s { + case StateStandby: + return "standby" + case StateTripped: + return "tripped" + case StateRecovering: + return "recovering" + default: + return fmt.Sprintf("undefined(%v)", int(s)) + } +} + +// ErrStateTripped will be returned from executions performed while the CircuitBreaker +// is in StateTripped +var ErrStateTripped = trace.ConnectionProblem(nil, "breaker is tripped") + +// Config contains configuration of the CircuitBreaker +type Config struct { + // Clock is used to control time - mainly used for testing + Clock clockwork.Clock + // Interval is the period of time that execution metrics will be collected for within StateStandby before + // transitioning to the next generation. + Interval time.Duration + // TrippedPeriod is the amount of time to remain in StateTripped before transitioning + // into StateRecovering + TrippedPeriod time.Duration + // Recover specifies the TripFn that will be used to determine if the CircuitBreaker should transition from + // StateRecovering to StateTripped. This is required to be supplied, failure to do so will result in an error + // creating the CircuitBreaker. + Recover TripFn + // RecoveryLimit is the number on consecutive successful executions required to transition from + // StateRecovering to StateStandby + RecoveryLimit uint32 + // Trip specifies the TripFn that will be used to determine if the CircuitBreaker should transition from + // StateStandby to StateTripped. This is required to be supplied, failure to do so will result in an error + // creating the CircuitBreaker. + Trip TripFn + // OnTripped will be called when the CircuitBreaker enters the StateTripped state + OnTripped func() + // OnStandby will be called when the CircuitBreaker returns to the StateStandby state + OnStandBy func() + // IsSuccessful is used by the CircuitBreaker to determine if the executed function was successful or not + IsSuccessful func(v interface{}, err error) bool + // Logger is the logger + Logger logrus.FieldLogger +} + +// TripFn determines if the CircuitBreaker should be tripped based +// on the state of the provided Metrics. A return value of true will +// cause the CircuitBreaker to transition into the StateTripped state +type TripFn = func(m Metrics) bool + +// StaticTripper is a TripFn that always returns the provided value +// regardless of the Metrics. Useful for testing. +func StaticTripper(b bool) TripFn { + return func(m Metrics) bool { + return b + } +} + +// RatioTripper is a TripFn that returns true it the error ratio +// is greater than the provided ratio and there have been at least +// minExecutions performed. +func RatioTripper(ratio float64, minExecutions uint32) TripFn { + return func(m Metrics) bool { + if m.Executions < minExecutions { + return false + } + + r := float64(m.Failures) / float64(m.Executions) + return r >= ratio + } +} + +// ConsecutiveFailureTripper is a TripFn that will return true if +// Metrics.ConsecutiveFailures is greater than the provided value. +func ConsecutiveFailureTripper(max uint32) TripFn { + return func(m Metrics) bool { + return m.ConsecutiveFailures > max + } +} + +// NonNilErrorIsSuccess returns true if the provided error is non nil. This +// is the default value for Config.IsSuccessful if not provided. +func NonNilErrorIsSuccess(_ interface{}, err error) bool { + return err == nil +} + +// IsResponseSuccessful determines whether the error provided should be ignored by the circuit breaker. This checks +// for http status codes < 500 and a few unsuccessful grpc status codes. +func IsResponseSuccessful(v interface{}, err error) bool { + switch t := v.(type) { + case nil: + break + case *http.Response: + if t == nil { + break + } + return t.StatusCode < http.StatusInternalServerError + } + + code := status.Code(err) + switch { + case err == nil: + return true + case code == codes.Canceled || code == codes.Unknown || code == codes.Unavailable || code == codes.DeadlineExceeded: + return false + default: + return true + } +} + +func DefaultBreakerConfig(clock clockwork.Clock) Config { + return Config{ + Clock: clock, + Interval: defaults.BreakerInterval, + Trip: RatioTripper(defaults.BreakerRatio, defaults.BreakerRatioMinExecutions), + Recover: RatioTripper(defaults.BreakerRatio/2, defaults.BreakerRatioMinExecutions/3), + IsSuccessful: IsResponseSuccessful, + } +} + +func NoopBreakerConfig() Config { + return Config{ + Interval: defaults.BreakerInterval, + Trip: StaticTripper(false), + Recover: StaticTripper(false), + IsSuccessful: func(v interface{}, err error) bool { return true }, + } +} + +// CheckAndSetDefaults checks and sets default config values. +func (c *Config) CheckAndSetDefaults() error { + if c.Clock == nil { + c.Clock = clockwork.NewRealClock() + } + + if c.Interval <= 0 { + return trace.BadParameter("CircuitBreaker Interval must be set") + } + + if c.Trip == nil { + return trace.BadParameter("CircuitBreaker Trip must be set") + } + if c.Recover == nil { + return trace.BadParameter("CircuitBreaker Recover must be set") + } + + if c.TrippedPeriod <= 0 { + c.TrippedPeriod = defaults.TrippedPeriod + } + + if c.RecoveryLimit <= 0 { + c.RecoveryLimit = defaults.RecoveryLimit + } + + if c.OnTripped == nil { + c.OnTripped = func() {} + } + + if c.OnStandBy == nil { + c.OnStandBy = func() {} + } + + if c.IsSuccessful == nil { + c.IsSuccessful = NonNilErrorIsSuccess + } + + if c.Logger == nil { + c.Logger = logrus.New().WithFields(logrus.Fields{ + trace.Component: "breaker", + }) + } + + c.TrippedPeriod = utils.NewSeventhJitter()(c.TrippedPeriod) + + return nil +} + +// CircuitBreaker implements the circuit breaker pattern +type CircuitBreaker struct { + cfg Config + + mu sync.Mutex + state State + generation uint64 + metrics Metrics + expiry time.Time +} + +func NewNoop() *CircuitBreaker { + return &CircuitBreaker{ + cfg: NoopBreakerConfig(), + } +} + +// New returns a CircuitBreaker configured with the provided Config +func New(cfg Config) (*CircuitBreaker, error) { + if err := cfg.CheckAndSetDefaults(); err != nil { + return nil, err + } + + cb := CircuitBreaker{cfg: cfg} + cb.nextGeneration(cfg.Clock.Now()) + + return &cb, nil +} + +// Execute calls the provided function depending on the CircuitBreaker state. +// - StateStandby: all functions are executed. +// - StateTripped: no functions are executed and ErrStateTripped is returned. +// - StateRecovering: some functions are executed, some functions are not, +// when not executed ErrLimitExceeded is returned. +// +// The CircuitBreaker state is updated according to the outcome of executing the +// provided function and the current state. See package docs for a more detailed +// explanation of state transitions. +func (c *CircuitBreaker) Execute(f func() (interface{}, error)) (interface{}, error) { + generation, err := c.beforeExecution() + if err != nil { + return nil, err + } + + v, err := f() + + c.afterExecution(generation, v, err) + + return v, err +} + +// beforeExecution checks the current state to determine if a new generation +// should be created and whether Execute is allowed to proceed. +func (c *CircuitBreaker) beforeExecution() (uint64, error) { + c.mu.Lock() + defer c.mu.Unlock() + + now := c.cfg.Clock.Now() + + generation, state := c.currentState(now) + + switch { + case state == StateTripped: + return generation, ErrStateTripped + } + + c.metrics.execute() + return generation, nil +} + +// afterExecution updates the CircuitBreaker state based on the outcome of +// processing the fn in Execute. +func (c *CircuitBreaker) afterExecution(prior uint64, v interface{}, err error) { + c.mu.Lock() + defer c.mu.Unlock() + + now := c.cfg.Clock.Now() + + generation, state := c.currentState(now) + if generation != prior { + return + } + + if c.cfg.IsSuccessful(v, err) { + c.cfg.Logger.Debugf("successful execution, %s", c.metrics.String()) + c.success(state, now) + } else { + c.cfg.Logger.Debugf("failed execution, %s", c.metrics.String()) + c.failure(state, now) + } +} + +// success tallies a successful execution and migrates to StateStandby +// if in another state and criteria has been met to transition +func (c *CircuitBreaker) success(state State, t time.Time) { + switch state { + case StateStandby: + c.metrics.success() + case StateRecovering: + c.metrics.success() + if c.metrics.ConsecutiveSuccesses >= c.cfg.RecoveryLimit { + c.setState(StateStandby, t) + go c.cfg.OnStandBy() + } + } +} + +// failure tallies a failed execution and migrate to StateTripped +// if in another state and criteria has been met to transition +func (c *CircuitBreaker) failure(state State, t time.Time) { + c.metrics.failure() + + switch state { + case StateRecovering: + if c.cfg.Recover(c.metrics) { + c.setState(StateTripped, t) + } + case StateStandby: + if c.cfg.Trip(c.metrics) { + c.setState(StateTripped, t) + go c.cfg.OnTripped() + } + } +} + +// setState updates the state and creates a new generation if the +// provided state is different from the CircuitBreakers current state +func (c *CircuitBreaker) setState(s State, t time.Time) { + if c.state == s { + return + } + + c.cfg.Logger.Debugf("state is transition from %s -> %s", c.state, s) + + c.state = s + c.nextGeneration(t) +} + +// currentState returns the state of the CircuitBreaker +func (c *CircuitBreaker) currentState(t time.Time) (uint64, State) { + switch { + case c.state == StateTripped && c.expiry.Before(t): + c.setState(StateRecovering, t) + case c.state == StateStandby && !c.expiry.IsZero() && c.expiry.Before(t): + c.nextGeneration(t) + } + + return c.generation, c.state +} + +// nextGeneration creates a new generation and adjusts its expiration +// based on the current state +func (c *CircuitBreaker) nextGeneration(t time.Time) { + c.metrics.reset() + c.generation++ + + switch c.state { + case StateRecovering: + c.expiry = time.Time{} + case StateTripped: + c.expiry = t.Add(c.cfg.TrippedPeriod) + case StateStandby: + c.expiry = t.Add(c.cfg.Interval) + } +} diff --git a/api/breaker/breaker_test.go b/api/breaker/breaker_test.go new file mode 100644 index 00000000000..d76a426aa63 --- /dev/null +++ b/api/breaker/breaker_test.go @@ -0,0 +1,618 @@ +// Copyright 2022 Gravitational, 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 breaker + +import ( + "errors" + "net/http" + "testing" + "time" + + "github.com/gravitational/trace" + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/require" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func TestCircuitBreaker_generation(t *testing.T) { + t.Parallel() + clock := clockwork.NewFakeClock() + + cb, err := New(Config{ + Clock: clock, + Interval: time.Second, + Trip: StaticTripper(false), + Recover: StaticTripper(false), + }) + require.NoError(t, err) + + generation, state := cb.currentState(clock.Now().UTC()) + require.Equal(t, uint64(1), generation) + require.Equal(t, StateStandby, state) + require.Equal(t, clock.Now().UTC().Add(time.Second), cb.expiry) + + clock.Advance(500 * time.Millisecond) + generation, state = cb.currentState(clock.Now().UTC()) + require.Equal(t, uint64(1), generation) + require.Equal(t, StateStandby, state) + clock.Advance(501 * time.Millisecond) + generation, state = cb.currentState(clock.Now().UTC()) + require.Equal(t, uint64(2), generation) + require.Equal(t, StateStandby, state) + require.Equal(t, clock.Now().UTC().Add(time.Second), cb.expiry) + + for i := 0; i < 1000; i++ { + prevGeneration, prevState := cb.currentState(clock.Now().UTC()) + cb.nextGeneration(clock.Now().UTC()) + generation, state := cb.currentState(clock.Now().UTC()) + require.NotEqual(t, prevGeneration, generation) + require.Equal(t, prevState, state) + } + + generation, state = cb.currentState(clock.Now().UTC()) + require.Equal(t, uint64(1002), generation) + require.Equal(t, StateStandby, state) +} + +func TestCircuitBreaker_beforeRequest(t *testing.T) { + t.Parallel() + cases := []struct { + desc string + generation uint64 + executions uint32 + advance time.Duration + state State + errorCheck require.ErrorAssertionFunc + }{ + { + desc: "standby allows execution", + generation: 1, + executions: 1, + state: StateStandby, + errorCheck: require.NoError, + }, + { + desc: "tripped prevents executions", + generation: 1, + executions: 0, + state: StateTripped, + errorCheck: func(t require.TestingT, err error, i ...interface{}) { + require.Error(t, err) + require.ErrorIs(t, ErrStateTripped, err) + }, + }, + { + desc: "recovering after allows executions", + generation: 1, + executions: 1, + state: StateRecovering, + advance: 3 * time.Second, + errorCheck: require.NoError, + }, + } + + for _, tt := range cases { + t.Run(tt.desc, func(t *testing.T) { + clock := clockwork.NewFakeClock() + + cb, err := New(Config{ + Clock: clock, + Interval: time.Second, + Trip: StaticTripper(false), + Recover: StaticTripper(false), + RecoveryLimit: 1, + }) + require.NoError(t, err) + cb.state = tt.state + + clock.Advance(tt.advance) + + generation, err := cb.beforeExecution() + tt.errorCheck(t, err) + require.Equal(t, tt.generation, generation) + require.Equal(t, tt.executions, cb.metrics.Executions) + + }) + } +} + +func TestCircuitBreaker_afterExecution(t *testing.T) { + t.Parallel() + cases := []struct { + desc string + err error + priorGeneration uint64 + checkMetrics require.ValueAssertionFunc + trip TripFn + recover TripFn + expectedState State + }{ + { + desc: "successful execution", + priorGeneration: 1, + checkMetrics: func(t require.TestingT, i interface{}, i2 ...interface{}) { + m, ok := i.(Metrics) + require.True(t, ok) + require.Equal(t, uint32(1), m.Successes) + require.Equal(t, uint32(0), m.Failures) + }, + trip: StaticTripper(false), + recover: StaticTripper(false), + expectedState: StateStandby, + }, + { + desc: "generation change", + priorGeneration: 0, + trip: StaticTripper(false), + recover: StaticTripper(false), + checkMetrics: func(t require.TestingT, i interface{}, i2 ...interface{}) { + m, ok := i.(Metrics) + require.True(t, ok) + require.Equal(t, uint32(0), m.Successes) + require.Equal(t, uint32(0), m.Failures) + }, + expectedState: StateStandby, + }, + { + desc: "failed execution with out tripping", + priorGeneration: 1, + err: errors.New("failure"), + trip: StaticTripper(false), + recover: StaticTripper(false), + checkMetrics: func(t require.TestingT, i interface{}, i2 ...interface{}) { + m, ok := i.(Metrics) + require.True(t, ok) + require.Equal(t, uint32(0), m.Successes) + require.Equal(t, uint32(1), m.Failures) + }, + expectedState: StateStandby, + }, + { + desc: "failed execution causing a trip", + priorGeneration: 1, + err: errors.New("failure"), + trip: StaticTripper(true), + recover: StaticTripper(false), + checkMetrics: func(t require.TestingT, i interface{}, i2 ...interface{}) { + m, ok := i.(Metrics) + require.True(t, ok) + require.Equal(t, uint32(0), m.Successes) + require.Equal(t, uint32(0), m.Failures) + }, + expectedState: StateTripped, + }, + } + + for _, tt := range cases { + t.Run(tt.desc, func(t *testing.T) { + clock := clockwork.NewFakeClock() + cb, err := New(Config{ + Clock: clock, + Interval: time.Second, + Trip: tt.trip, + Recover: tt.recover, + }) + require.NoError(t, err) + + cb.afterExecution(tt.priorGeneration, nil, tt.err) + tt.checkMetrics(t, cb.metrics) + require.Equal(t, tt.expectedState, cb.state) + }) + } +} + +func TestCircuitBreaker_success(t *testing.T) { + t.Parallel() + cases := []struct { + desc string + initialState State + successState State + expectedState State + recoveryLimit uint32 + }{ + { + desc: "success in standby", + initialState: StateStandby, + successState: StateStandby, + expectedState: StateStandby, + }, + { + desc: "success in recovery below limit", + initialState: StateRecovering, + successState: StateRecovering, + expectedState: StateRecovering, + recoveryLimit: 2, + }, + { + desc: "success in recovery above limit", + initialState: StateRecovering, + successState: StateRecovering, + expectedState: StateStandby, + recoveryLimit: 1, + }, + } + + for _, tt := range cases { + t.Run(tt.desc, func(t *testing.T) { + clock := clockwork.NewFakeClock() + cb, err := New(Config{ + Clock: clock, + Interval: time.Second, + RecoveryLimit: tt.recoveryLimit, + Trip: StaticTripper(false), + Recover: StaticTripper(false), + }) + require.NoError(t, err) + cb.state = tt.initialState + + generation, state := cb.currentState(clock.Now().UTC()) + cb.success(tt.successState, clock.Now().UTC()) + require.Equal(t, tt.expectedState, cb.state) + if tt.expectedState != state { + require.NotEqual(t, generation, cb.generation) + } + }) + } +} + +func TestCircuitBreaker_failure(t *testing.T) { + t.Parallel() + cases := []struct { + desc string + initialState State + failureState State + expectedState State + tripFn TripFn + recover TripFn + onTrip func(ch chan bool) func() + tripped bool + requireTripped require.BoolAssertionFunc + }{ + { + desc: "failure in recovering transitions to tripped", + initialState: StateRecovering, + failureState: StateRecovering, + expectedState: StateTripped, + tripFn: StaticTripper(false), + recover: StaticTripper(true), + requireTripped: require.False, + }, + { + desc: "failure in standby without tripping", + initialState: StateStandby, + failureState: StateStandby, + expectedState: StateStandby, + tripFn: StaticTripper(false), + recover: StaticTripper(false), + requireTripped: require.False, + }, + { + desc: "failure in standby causes tripping", + initialState: StateStandby, + failureState: StateStandby, + expectedState: StateTripped, + tripFn: StaticTripper(true), + recover: StaticTripper(false), + requireTripped: require.True, + onTrip: func(ch chan bool) func() { + return func() { + ch <- true + } + }, + }, + } + + for _, tt := range cases { + tt := tt + t.Run(tt.desc, func(t *testing.T) { + t.Parallel() + clock := clockwork.NewFakeClock() + + if tt.onTrip == nil { + tt.onTrip = func(ch chan bool) func() { + ch <- false + return func() {} + } + } + + trippedCh := make(chan bool, 1) + + cb, err := New(Config{ + Clock: clock, + Interval: time.Second, + Trip: tt.tripFn, + OnTripped: tt.onTrip(trippedCh), + Recover: tt.recover, + }) + require.NoError(t, err) + cb.state = tt.initialState + + generation, state := cb.currentState(clock.Now().UTC()) + cb.failure(tt.failureState, clock.Now().UTC()) + require.Equal(t, tt.expectedState, cb.state) + if tt.expectedState != state { + require.NotEqual(t, generation, cb.generation) + } + + tripped := <-trippedCh + + tt.requireTripped(t, tripped) + }) + } +} + +func TestCircuitBreaker_Execute(t *testing.T) { + t.Parallel() + + clock := clockwork.NewFakeClock() + + trippedCh := make(chan struct{}) + onTripped := func(ch chan struct{}) func() { + return func() { + ch <- struct{}{} + } + } + + cb, err := New(Config{ + Clock: clock, + Interval: time.Second, + Trip: ConsecutiveFailureTripper(3), + Recover: ConsecutiveFailureTripper(1), + OnTripped: onTripped(trippedCh), + TrippedPeriod: 2 * time.Second, + RecoveryLimit: 2, + }) + require.NoError(t, err) + + testErr := errors.New("failure") + errorFn := func() (interface{}, error) { return nil, testErr } + noErrorFn := func() (interface{}, error) { return nil, nil } + cases := []struct { + desc string + exec func() (interface{}, error) + advance time.Duration + errorAssertion require.ErrorAssertionFunc + expectedState State + expectedGeneration uint64 + }{ + { + desc: "no errors remain in standby", + exec: noErrorFn, + errorAssertion: require.NoError, + expectedState: StateStandby, + expectedGeneration: 1, + }, + { + desc: "error below limit remain in standby", + exec: errorFn, + errorAssertion: require.Error, + expectedState: StateStandby, + expectedGeneration: 1, + }, + { + desc: "another error below limit remain in standby", + exec: errorFn, + errorAssertion: require.Error, + expectedState: StateStandby, + expectedGeneration: 1, + }, + { + desc: "last error below limit remain in standby", + exec: errorFn, + errorAssertion: require.Error, + expectedState: StateStandby, + expectedGeneration: 1, + }, + { + desc: "transition from standby to tripped", + exec: errorFn, + errorAssertion: require.Error, + expectedState: StateTripped, + expectedGeneration: 2, + }, + { + desc: "error remain tripped", + exec: errorFn, + errorAssertion: require.Error, + expectedState: StateTripped, + expectedGeneration: 2, + }, + { + desc: "no error remain tripped", + exec: noErrorFn, + errorAssertion: require.Error, + expectedState: StateTripped, + expectedGeneration: 2, + }, + { + desc: "transition from tripped to recovering", + exec: noErrorFn, + errorAssertion: require.NoError, + expectedState: StateRecovering, + expectedGeneration: 3, + advance: 3 * time.Second, + }, + { + desc: "first failed execution recovering remains in recovering", + exec: errorFn, + errorAssertion: require.Error, + expectedState: StateRecovering, + expectedGeneration: 3, + advance: 250 * time.Millisecond, + }, + { + desc: "second failed execution recovering transitions to tripped", + exec: errorFn, + errorAssertion: require.Error, + expectedState: StateTripped, + expectedGeneration: 4, + advance: 450 * time.Millisecond, + }, + { + desc: "transition from tripped to recovering", + exec: noErrorFn, + errorAssertion: require.NoError, + expectedState: StateRecovering, + expectedGeneration: 5, + advance: 3 * time.Second, + }, + { + desc: "transition from recovering to standby", + exec: noErrorFn, + errorAssertion: require.NoError, + expectedState: StateStandby, + expectedGeneration: 6, + advance: 450 * time.Millisecond, + }, + { + desc: "remain in standby while in new generation", + exec: noErrorFn, + errorAssertion: require.NoError, + expectedState: StateStandby, + expectedGeneration: 7, + advance: time.Minute, + }, + } + + for i, tt := range cases { + t.Run(tt.desc, func(t *testing.T) { + clock.Advance(tt.advance) + _, err := cb.Execute(tt.exec) + tt.errorAssertion(t, err) + generation, state := cb.currentState(clock.Now().UTC()) + require.Equal(t, tt.expectedGeneration, generation, "incorrect generation") + require.Equal(t, tt.expectedState, state, "incorrect state") + + if state != StateTripped && tt.expectedState == StateTripped { + select { + case <-trippedCh: + default: + t.Fatalf("step %d expected to get tripped, but wasn't", i) + } + } + }) + } + +} + +func TestMetrics(t *testing.T) { + m := Metrics{} + + zero := uint32(0) + one := uint32(1) + require.Equal(t, zero, m.Executions) + require.Equal(t, zero, m.Successes) + require.Equal(t, zero, m.Failures) + require.Equal(t, zero, m.ConsecutiveSuccesses) + require.Equal(t, zero, m.ConsecutiveFailures) + + m.success() + + require.Equal(t, zero, m.Executions) + require.Equal(t, one, m.Successes) + require.Equal(t, zero, m.Failures) + require.Equal(t, one, m.ConsecutiveSuccesses) + require.Equal(t, zero, m.ConsecutiveFailures) + + m.execute() + + require.Equal(t, one, m.Executions) + require.Equal(t, one, m.Successes) + require.Equal(t, zero, m.Failures) + require.Equal(t, one, m.ConsecutiveSuccesses) + require.Equal(t, zero, m.ConsecutiveFailures) + + m.failure() + + require.Equal(t, one, m.Executions) + require.Equal(t, one, m.Successes) + require.Equal(t, one, m.Failures) + require.Equal(t, zero, m.ConsecutiveSuccesses) + require.Equal(t, one, m.ConsecutiveFailures) + + m.reset() + + require.Equal(t, zero, m.Executions) + require.Equal(t, zero, m.Successes) + require.Equal(t, zero, m.Failures) + require.Equal(t, zero, m.ConsecutiveSuccesses) + require.Equal(t, zero, m.ConsecutiveFailures) +} + +func TestIsResponseSuccessful(t *testing.T) { + cases := []struct { + name string + err error + response *http.Response + assertion require.BoolAssertionFunc + }{ + { + name: "nil error", + assertion: require.True, + }, + { + name: "codes.Canceled error", + err: status.Error(codes.Canceled, ""), + assertion: require.False, + }, + { + name: "codes.Unknown error", + err: status.Error(codes.Unknown, ""), + assertion: require.False, + }, + { + name: "codes.Unavailable error", + err: status.Error(codes.Unavailable, ""), + assertion: require.False, + }, + { + name: "codes.Unavailable error", + err: status.Error(codes.DeadlineExceeded, ""), + assertion: require.False, + }, + { + name: "other error", + err: trace.NotFound("not found"), + assertion: require.False, + }, + { + name: "error", + err: trace.NotFound(""), + assertion: require.False, + }, + { + name: "200", + response: &http.Response{StatusCode: http.StatusOK}, + assertion: require.True, + }, + { + name: "500", + response: &http.Response{StatusCode: http.StatusBadGateway}, + assertion: require.False, + }, + { + name: "404", + response: &http.Response{StatusCode: http.StatusNotFound}, + assertion: require.True, + }, + } + + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + tt.assertion(t, IsResponseSuccessful(tt.response, tt.err)) + }) + } +} diff --git a/api/breaker/doc.go b/api/breaker/doc.go new file mode 100644 index 00000000000..203d3356590 --- /dev/null +++ b/api/breaker/doc.go @@ -0,0 +1,34 @@ +// Copyright 2022 Gravitational, 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 breaker implements a circuit breaker. +// +// Circuit breaker watches the error from executed functions and according to the configured TripFn will +// allow requests for a period of time before slowly allowing a few. +// +// Circuit breakers start in StateStandby first, observing errors and watching Metrics. +// +// Once the Circuit breaker TripFn returns true, it enters the StateTripped, where it blocks all traffic and returns +// ErrStateTripped for the configured Config.TrippedPeriod. +// +// After the Config.TrippedPeriod passes, Circuit breaker enters StateRecovering, during that state it will +// either transition to StateStandby after the Config.RecoveryLimit is met or StateTripped if the Config.Recover TripFn +// is satisfied. +// +// It is possible to define actions on transitions between states: +// +// * Config.OnTripped is called on transition (StateStandby -> StateTripped) +// * Config.OnStandBy is called on transition (StateRecovering -> StateStandby) +// +package breaker diff --git a/api/breaker/interceptors.go b/api/breaker/interceptors.go new file mode 100644 index 00000000000..af50d2b3a76 --- /dev/null +++ b/api/breaker/interceptors.go @@ -0,0 +1,53 @@ +// Copyright 2022 Gravitational, 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 breaker + +import ( + "context" + + "google.golang.org/grpc" +) + +// UnaryClientInterceptor is a unary gRPC client interceptor that uses the provided CircuitBreaker to track errors +// returned from the outgoing calls. +func UnaryClientInterceptor(cb *CircuitBreaker) grpc.UnaryClientInterceptor { + return func(ctx context.Context, method string, req, reply interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error { + _, err := cb.Execute(func() (interface{}, error) { + err := invoker(ctx, method, req, reply, cc, opts...) + return nil, err + }) + return err + } +} + +// StreamClientInterceptor is a stream gRPC client interceptor that uses the provided CircuitBreaker to track errors +// returned from the outgoing calls. +func StreamClientInterceptor(cb *CircuitBreaker) grpc.StreamClientInterceptor { + return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string, streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) { + stream, err := cb.Execute(func() (interface{}, error) { + return streamer(ctx, desc, cc, method, opts...) + }) + + if stream == nil { + return nil, err + } + + if cs, ok := stream.(grpc.ClientStream); ok { + return cs, err + } + + return nil, err + } +} diff --git a/api/breaker/round_tripper.go b/api/breaker/round_tripper.go new file mode 100644 index 00000000000..c133c58758a --- /dev/null +++ b/api/breaker/round_tripper.go @@ -0,0 +1,52 @@ +// Copyright 2022 Gravitational, 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 breaker + +import ( + "net/http" +) + +// RoundTripper wraps a http.RoundTripper with a CircuitBreaker +type RoundTripper struct { + tripper http.RoundTripper + cb *CircuitBreaker +} + +// NewRoundTripper returns a RoundTripper +func NewRoundTripper(cb *CircuitBreaker, tripper http.RoundTripper) *RoundTripper { + return &RoundTripper{ + tripper: tripper, + cb: cb, + } +} + +// RoundTrip forwards the request on to the provided http.RoundTripper if +// the CircuitBreaker allows it +// +// nolint:bodyclose +// The interface{} conversion to *http.Response trips the linter even though this +// is merely a pass through function. Closing the body here would prevent the actual +// consumer to not be able to read it. Copying here to satisfy the linter seems wasteful. +func (t *RoundTripper) RoundTrip(request *http.Request) (*http.Response, error) { + v, err := t.cb.Execute(func() (interface{}, error) { + return t.tripper.RoundTrip(request) + }) + + if v == nil { + return nil, err + } + + return v.(*http.Response), err +} diff --git a/api/breaker/round_tripper_test.go b/api/breaker/round_tripper_test.go new file mode 100644 index 00000000000..c6fdeb64583 --- /dev/null +++ b/api/breaker/round_tripper_test.go @@ -0,0 +1,137 @@ +// Copyright 2022 Gravitational, 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 breaker + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/jonboulle/clockwork" + "github.com/stretchr/testify/require" +) + +func TestRoundTripper_RoundTrip(t *testing.T) { + t.Parallel() + + clock := clockwork.NewFakeClock() + cb, err := New(Config{ + Clock: clock, + RecoveryLimit: 1, + Interval: time.Second, + TrippedPeriod: time.Second, + IsSuccessful: func(v interface{}, err error) bool { + if err != nil { + return false + } + + if v == nil { + return false + } + + switch t := v.(type) { + case *http.Response: + return t.StatusCode < http.StatusInternalServerError + } + + return true + }, + Trip: ConsecutiveFailureTripper(1), + Recover: ConsecutiveFailureTripper(0), + }) + require.NoError(t, err) + + mux := http.NewServeMux() + mux.HandleFunc("/success", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("/fail", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + }) + srv := httptest.NewServer(mux) + + clt := srv.Client() + clt.Transport = NewRoundTripper(cb, clt.Transport) + + ctx := context.Background() + + cases := []struct { + desc string + url string + state State + advance time.Duration + errorAssertion require.ErrorAssertionFunc + responseAssertion require.ValueAssertionFunc + }{ + { + desc: "success in standby", + url: "/success", + state: StateStandby, + errorAssertion: require.NoError, + responseAssertion: func(t require.TestingT, i interface{}, i2 ...interface{}) { + require.Equal(t, http.StatusOK, i.(*http.Response).StatusCode) + }, + }, + { + desc: "error when tripped", + url: "/success", + state: StateTripped, + errorAssertion: func(t require.TestingT, err error, i ...interface{}) { + require.Error(t, err) + require.ErrorIs(t, err, ErrStateTripped) + }, + responseAssertion: require.Nil, + }, + { + desc: "allowed request when recovery progresses", + url: "/success", + state: StateRecovering, + advance: time.Minute, + errorAssertion: require.NoError, + responseAssertion: func(t require.TestingT, i interface{}, i2 ...interface{}) { + require.Equal(t, http.StatusOK, i.(*http.Response).StatusCode) + }, + }, + { + desc: "failure in standby", + url: "/fail", + state: StateStandby, + errorAssertion: require.NoError, + responseAssertion: func(t require.TestingT, i interface{}, i2 ...interface{}) { + require.Equal(t, http.StatusBadGateway, i.(*http.Response).StatusCode) + }, + }, + } + + for _, tt := range cases { + t.Run(tt.desc, func(t *testing.T) { + cb.setState(tt.state, clock.Now()) + clock.Advance(tt.advance) + + r, err := http.NewRequestWithContext(ctx, http.MethodGet, srv.URL+tt.url, nil) + require.NoError(t, err) + resp, err := clt.Do(r) + if resp != nil { + t.Cleanup(func() { + require.NoError(t, resp.Body.Close()) + }) + } + tt.errorAssertion(t, err) + tt.responseAssertion(t, resp) + }) + } +} diff --git a/api/client/client.go b/api/client/client.go index e786cbf2cc3..82418dffb83 100644 --- a/api/client/client.go +++ b/api/client/client.go @@ -20,6 +20,7 @@ import ( "compress/gzip" "context" "crypto/tls" + "errors" "fmt" "io" "net" @@ -27,6 +28,7 @@ import ( "sync/atomic" "time" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/client/proto" "github.com/gravitational/teleport/api/constants" "github.com/gravitational/teleport/api/defaults" @@ -39,6 +41,7 @@ import ( "github.com/golang/protobuf/ptypes/empty" "github.com/gravitational/trace" "github.com/gravitational/trace/trail" + "github.com/jonboulle/clockwork" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "golang.org/x/crypto/ssh" "google.golang.org/grpc" @@ -76,7 +79,7 @@ type Client struct { // JoinServiceClient is a client for the JoinService, which runs on both the // auth and proxy. *JoinServiceClient - // closedFlag is set to indicate that the connnection is closed. + // closedFlag is set to indicate that the connection is closed. // It's a pointer to allow the Client struct to be copied. closedFlag *int32 // callOpts configure calls made by this client. @@ -377,29 +380,40 @@ func (c *Client) dialGRPC(ctx context.Context, addr string) error { dialContext, cancel := context.WithTimeout(ctx, c.c.DialTimeout) defer cancel() + cb, err := breaker.New(c.c.CircuitBreakerConfig) + if err != nil { + return trace.Wrap(err) + } + var dialOpts []grpc.DialOption dialOpts = append(dialOpts, grpc.WithContextDialer(c.grpcDialer())) dialOpts = append(dialOpts, grpc.WithChainUnaryInterceptor( otelgrpc.UnaryClientInterceptor(), metadata.UnaryClientInterceptor, + breaker.UnaryClientInterceptor(cb), ), grpc.WithChainStreamInterceptor( otelgrpc.StreamClientInterceptor(), - metadata.StreamClientInterceptor), + metadata.StreamClientInterceptor, + breaker.StreamClientInterceptor(cb), + ), ) // Only set transportCredentials if tlsConfig is set. This makes it possible - // to explicitly provide gprc.WithInsecure in the client's dial options. + // to explicitly provide grpc.WithTransportCredentials(insecure.NewCredentials()) + // in the client's dial options. if c.tlsConfig != nil { dialOpts = append(dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(c.tlsConfig))) } // must come last, otherwise provided opts may get clobbered by defaults above dialOpts = append(dialOpts, c.c.DialOpts...) - var err error - if c.conn, err = grpc.DialContext(dialContext, addr, dialOpts...); err != nil { + conn, err := grpc.DialContext(dialContext, addr, dialOpts...) + if err != nil { return trace.Wrap(err) } + + c.conn = conn c.grpc = proto.NewAuthServiceClient(c.conn) c.JoinServiceClient = NewJoinServiceClient(proto.NewJoinServiceClient(c.conn)) @@ -435,11 +449,14 @@ func (c *Client) grpcDialer() func(ctx context.Context, addr string) (net.Conn, } } -// waitForConnectionReady waits for the client's grpc connection finish dialing, returning an errror +// waitForConnectionReady waits for the client's grpc connection finish dialing, returning an error // if the ctx is canceled or the client's gRPC connection enters an unexpected state. This can be used // alongside the DialInBackground client config option to wait until background dialing has completed. func (c *Client) waitForConnectionReady(ctx context.Context) error { for { + if c.conn == nil { + return errors.New("conn was closed") + } switch state := c.conn.GetState(); state { case connectivity.Ready: return nil @@ -489,6 +506,8 @@ type Config struct { ALPNSNIAuthDialClusterName string // IgnoreHTTPProxy disables support for HTTP proxying when true. IgnoreHTTPProxy bool + // CircuitBreakerConfig defines how the circuit breaker should behave. + CircuitBreakerConfig breaker.Config } // CheckAndSetDefaults checks and sets default config values. @@ -506,6 +525,9 @@ func (c *Config) CheckAndSetDefaults() error { if c.DialTimeout == 0 { c.DialTimeout = defaults.DefaultDialTimeout } + if c.CircuitBreakerConfig.Trip == nil || c.CircuitBreakerConfig.IsSuccessful == nil { + c.CircuitBreakerConfig = breaker.DefaultBreakerConfig(clockwork.NewRealClock()) + } c.DialOpts = append(c.DialOpts, grpc.WithKeepaliveParams(keepalive.ClientParameters{ Time: c.KeepAlivePeriod, diff --git a/api/client/client_test.go b/api/client/client_test.go index e3e33709506..e4b182307a0 100644 --- a/api/client/client_test.go +++ b/api/client/client_test.go @@ -24,28 +24,32 @@ import ( "testing" "time" - "github.com/golang/protobuf/ptypes/empty" - "github.com/google/go-cmp/cmp" "github.com/gravitational/teleport/api/client/proto" "github.com/gravitational/teleport/api/defaults" "github.com/gravitational/teleport/api/types" - "google.golang.org/grpc/credentials/insecure" + "github.com/golang/protobuf/ptypes/empty" + "github.com/google/go-cmp/cmp" "github.com/gravitational/trace" "github.com/gravitational/trace/trail" "github.com/stretchr/testify/require" "golang.org/x/crypto/ssh" "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" ) // mockServer mocks an Auth Server. type mockServer struct { + addr string grpc *grpc.Server *proto.UnimplementedAuthServiceServer } -func newMockServer() *mockServer { +func newMockServer(addr string) *mockServer { m := &mockServer{ + addr: addr, grpc: grpc.NewServer(), UnimplementedAuthServiceServer: &proto.UnimplementedAuthServiceServer{}, } @@ -53,13 +57,56 @@ func newMockServer() *mockServer { return m } +func (m *mockServer) Stop() { + m.grpc.Stop() +} + +func (m *mockServer) Addr() string { + return m.addr +} + +type ConfigOpt func(*Config) + +func WithConfig(cfg Config) ConfigOpt { + return func(config *Config) { + *config = cfg + } +} + +func (m *mockServer) NewClient(ctx context.Context, opts ...ConfigOpt) (*Client, error) { + cfg := Config{ + Addrs: []string{m.addr}, + Credentials: []Credentials{ + &mockInsecureTLSCredentials{}, // TODO(Joerger) replace insecure credentials + }, + DialOpts: []grpc.DialOption{ + grpc.WithTransportCredentials(insecure.NewCredentials()), // TODO(Joerger) remove insecure dial option + }, + } + + for _, opt := range opts { + opt(&cfg) + } + + return New(ctx, cfg) +} + // startMockServer starts a new mock server. Parallel tests cannot use the same addr. -func startMockServer(t *testing.T) string { +func startMockServer(t *testing.T) *mockServer { l, err := net.Listen("tcp", "") require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, l.Close()) }) - go newMockServer().grpc.Serve(l) - return l.Addr().String() + return startMockServerWithListener(t, l) +} + +// startMockServerWithListener starts a new mock server with the provided listener +func startMockServerWithListener(t *testing.T, l net.Listener) *mockServer { + srv := newMockServer(l.Addr().String()) + t.Cleanup(srv.grpc.Stop) + + go func() { + require.NoError(t, srv.grpc.Serve(l)) + }() + return srv } func (m *mockServer) Ping(ctx context.Context, req *proto.PingRequest) (*proto.PingResponse, error) { @@ -144,6 +191,10 @@ func (m *mockServer) ListResources(ctx context.Context, req *proto.ListResources return resp, nil } +func (m *mockServer) AddMFADeviceSync(ctx context.Context, req *proto.AddMFADeviceSyncRequest) (*proto.AddMFADeviceSyncResponse, error) { + return nil, status.Error(codes.AlreadyExists, "Already Exists") +} + const fiveMBNode = "fiveMBNode" func testResources(resourceType, namespace string) ([]types.ResourceWithLabels, error) { @@ -276,7 +327,7 @@ func (mc *mockInsecureTLSCredentials) SSHClientConfig() (*ssh.ClientConfig, erro func TestNew(t *testing.T) { t.Parallel() ctx := context.Background() - addr := startMockServer(t) + srv := startMockServer(t) tests := []struct { desc string @@ -285,7 +336,7 @@ func TestNew(t *testing.T) { }{{ desc: "successfully dial tcp address.", config: Config{ - Addrs: []string{addr}, + Addrs: []string{srv.Addr()}, Credentials: []Credentials{ &mockInsecureTLSCredentials{}, // TODO(Joerger) replace insecure credentials }, @@ -297,7 +348,7 @@ func TestNew(t *testing.T) { }, { desc: "synchronously dial addr/cred pairs and succeed with the 1 good pair.", config: Config{ - Addrs: []string{"bad addr", addr, "bad addr"}, + Addrs: []string{"bad addr", srv.Addr(), "bad addr"}, Credentials: []Credentials{ &tlsConfigCreds{nil}, &mockInsecureTLSCredentials{}, // TODO(Joerger) replace insecure credentials @@ -343,11 +394,11 @@ func TestNew(t *testing.T) { for _, tt := range tests { t.Run(tt.desc, func(t *testing.T) { - clt, err := New(ctx, tt.config) + clt, err := srv.NewClient(ctx, WithConfig(tt.config)) tt.assertErr(t, err) if err == nil { - defer clt.Close() + t.Cleanup(func() { require.NoError(t, clt.Close()) }) // requests to the server should succeed. _, err = clt.Ping(ctx) require.NoError(t, err) @@ -363,7 +414,6 @@ func TestNewDialBackground(t *testing.T) { // get listener but don't serve it yet. l, err := net.Listen("tcp", "") require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, l.Close()) }) addr := l.Addr().String() // Create client before the server is listening. @@ -378,7 +428,7 @@ func TestNewDialBackground(t *testing.T) { }, }) require.NoError(t, err) - defer clt.Close() + t.Cleanup(func() { require.NoError(t, clt.Close()) }) // requests to the server will result in a connection error. cancelCtx, cancel := context.WithTimeout(ctx, time.Second*3) @@ -387,7 +437,7 @@ func TestNewDialBackground(t *testing.T) { require.Error(t, err) // Start the server and wait for the client connection to be ready. - go newMockServer().grpc.Serve(l) + startMockServerWithListener(t, l) require.NoError(t, clt.waitForConnectionReady(ctx)) // requests to the server should succeed. @@ -401,7 +451,6 @@ func TestWaitForConnectionReady(t *testing.T) { l, err := net.Listen("tcp", "") require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, l.Close()) }) addr := l.Addr().String() // Create client before the server is listening. @@ -416,7 +465,7 @@ func TestWaitForConnectionReady(t *testing.T) { }, }) require.NoError(t, err) - defer clt.Close() + t.Cleanup(func() { require.NoError(t, clt.Close()) }) // WaitForConnectionReady should return false once the // context is canceled if the server isn't open to connections. @@ -425,29 +474,21 @@ func TestWaitForConnectionReady(t *testing.T) { require.Error(t, clt.waitForConnectionReady(cancelCtx)) // WaitForConnectionReady should return nil if the server is open to connections. - go newMockServer().grpc.Serve(l) + startMockServerWithListener(t, l) require.NoError(t, clt.waitForConnectionReady(ctx)) // WaitForConnectionReady should return an error if the grpc connection is closed. - require.NoError(t, clt.GetConnection().Close()) + require.NoError(t, clt.Close()) require.Error(t, clt.waitForConnectionReady(ctx)) } func TestLimitExceeded(t *testing.T) { t.Parallel() ctx := context.Background() - addr := startMockServer(t) + srv := startMockServer(t) // Create client - clt, err := New(ctx, Config{ - Addrs: []string{addr}, - Credentials: []Credentials{ - &mockInsecureTLSCredentials{}, // TODO(Joerger) replace insecure credentials - }, - DialOpts: []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), // TODO(Joerger) remove insecure dial option - }, - }) + clt, err := srv.NewClient(ctx) require.NoError(t, err) // ListNodes should return a limit exceeded error when exceeding gRPC message size limit. @@ -481,7 +522,7 @@ func TestLimitExceeded(t *testing.T) { func TestListResources(t *testing.T) { t.Parallel() ctx := context.Background() - addr := startMockServer(t) + srv := startMockServer(t) testCases := map[string]struct { resourceType string @@ -510,15 +551,7 @@ func TestListResources(t *testing.T) { } // Create client - clt, err := New(ctx, Config{ - Addrs: []string{addr}, - Credentials: []Credentials{ - &mockInsecureTLSCredentials{}, // TODO(Joerger) replace insecure credentials - }, - DialOpts: []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), // TODO(Joerger) remove insecure dial option - }, - }) + clt, err := srv.NewClient(ctx) require.NoError(t, err) for name, test := range testCases { @@ -557,18 +590,10 @@ func TestListResources(t *testing.T) { func TestGetResources(t *testing.T) { t.Parallel() ctx := context.Background() - addr := startMockServer(t) + srv := startMockServer(t) // Create client - clt, err := New(ctx, Config{ - Addrs: []string{addr}, - Credentials: []Credentials{ - &mockInsecureTLSCredentials{}, // TODO(Joerger) replace insecure credentials - }, - DialOpts: []grpc.DialOption{ - grpc.WithTransportCredentials(insecure.NewCredentials()), // TODO(Joerger) remove insecure dial option - }, - }) + clt, err := srv.NewClient(ctx) require.NoError(t, err) testCases := map[string]struct { diff --git a/api/defaults/defaults.go b/api/defaults/defaults.go index 3b7d7f3264c..b38fe58707c 100644 --- a/api/defaults/defaults.go +++ b/api/defaults/defaults.go @@ -51,6 +51,27 @@ const ( // SessionTrackerTTL defines the default base ttl of a session tracker. SessionTrackerTTL = time.Hour + + // BreakerInterval is the period in time the circuit breaker will + // tally metrics for + BreakerInterval = time.Minute + + // TrippedPeriod is the default period of time the circuit breaker will + // remain in breaker.StateTripped before transitioning to breaker.StateRecovering. No + // outbound requests are allowed for the duration of this period. + TrippedPeriod = 60 * time.Second + + // RecoveryLimit is the default number of consecutive successful requests needed to transition + // from breaker.StateRecovering to breaker.StateStandby + RecoveryLimit = 3 + + // BreakerRatio is the default ratio of failed requests to successful requests that will + // result in the circuit breaker transitioning to breaker.StateTripped + BreakerRatio = 0.9 + + // BreakerRatioMinExecutions is the minimum number of requests before the ratio tripper + // will consider examining the request pass rate + BreakerRatioMinExecutions = 10 ) var ( diff --git a/api/go.mod b/api/go.mod index 5b0eb211b1f..084ea4a210b 100644 --- a/api/go.mod +++ b/api/go.mod @@ -7,6 +7,7 @@ require ( github.com/golang/protobuf v1.5.2 github.com/google/go-cmp v0.5.7 github.com/gravitational/trace v1.1.17 + github.com/jonboulle/clockwork v0.2.2 github.com/russellhaering/gosaml2 v0.6.1-0.20210916051624-757d23f1bc28 github.com/sirupsen/logrus v1.8.1 github.com/stretchr/testify v1.7.1 diff --git a/api/utils/jitter.go b/api/utils/jitter.go new file mode 100644 index 00000000000..f52d34ea481 --- /dev/null +++ b/api/utils/jitter.go @@ -0,0 +1,41 @@ +// Copyright 2022 Gravitational, 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 utils + +import ( + "math/rand" + "sync" + "time" +) + +type Jitter func(time.Duration) time.Duration + +// NewSeventhJitter builds a new jitter on the range [6n/7,n). Prefer smaller +// jitters such as this when jittering periodic operations (e.g. cert rotation +// checks) since large jitters result in significantly increased load. +func NewSeventhJitter() Jitter { + var mu sync.Mutex + rng := rand.New(rand.NewSource(time.Now().UnixNano())) + return func(d time.Duration) time.Duration { + // values less than 1 cause rng to panic, and some logic + // relies on treating zero duration as non-blocking case. + if d < 1 { + return 0 + } + mu.Lock() + defer mu.Unlock() + return (6 * d / 7) + time.Duration(rng.Int63n(int64(d))/7) + } +} diff --git a/integration/app_integration_test.go b/integration/app_integration_test.go index 3b70cf374f7..4a42cbc22f8 100644 --- a/integration/app_integration_test.go +++ b/integration/app_integration_test.go @@ -37,6 +37,7 @@ import ( "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" apidefaults "github.com/gravitational/teleport/api/defaults" "github.com/gravitational/teleport/api/types" apievents "github.com/gravitational/teleport/api/types/events" @@ -1086,6 +1087,7 @@ func setupWithOptions(t *testing.T, opts appTestOptions) *pack { rcConf.Proxy.DisableWebInterface = true rcConf.SSH.Enabled = false rcConf.Apps.Enabled = false + rcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() if opts.rootConfig != nil { opts.rootConfig(rcConf) } @@ -1101,6 +1103,7 @@ func setupWithOptions(t *testing.T, opts appTestOptions) *pack { lcConf.Proxy.DisableWebInterface = true lcConf.SSH.Enabled = false lcConf.Apps.Enabled = false + lcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() if opts.rootConfig != nil { opts.rootConfig(lcConf) } @@ -1538,6 +1541,7 @@ func (p *pack) startRootAppServers(t *testing.T, count int, extraApps []service. raConf.Proxy.Enabled = false raConf.SSH.Enabled = false raConf.Apps.Enabled = true + raConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() raConf.Apps.Apps = append([]service.App{ { Name: p.rootAppName, @@ -1608,6 +1612,7 @@ func (p *pack) startLeafAppServers(t *testing.T, count int, extraApps []service. laConf.Proxy.Enabled = false laConf.SSH.Enabled = false laConf.Apps.Enabled = true + laConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() laConf.Apps.Apps = append([]service.App{ { Name: p.leafAppName, diff --git a/integration/db_integration_test.go b/integration/db_integration_test.go index d3f7ac7c82e..3ef378e8e98 100644 --- a/integration/db_integration_test.go +++ b/integration/db_integration_test.go @@ -34,6 +34,7 @@ import ( "go.mongodb.org/mongo-driver/bson" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" apidefaults "github.com/gravitational/teleport/api/defaults" "github.com/gravitational/teleport/api/types" apievents "github.com/gravitational/teleport/api/types/events" @@ -1001,6 +1002,7 @@ func setupDatabaseTest(t *testing.T, options ...testOptionFunc) *databasePack { rcConf.Proxy.Enabled = true rcConf.Proxy.DisableWebInterface = true rcConf.Clock = p.clock + rcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() if opts.rootConfig != nil { opts.rootConfig(rcConf) } @@ -1013,6 +1015,7 @@ func setupDatabaseTest(t *testing.T, options ...testOptionFunc) *databasePack { lcConf.Proxy.Enabled = true lcConf.Proxy.DisableWebInterface = true lcConf.Clock = p.clock + lcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() if opts.leafConfig != nil { opts.rootConfig(lcConf) } @@ -1083,12 +1086,11 @@ func setupDatabaseTest(t *testing.T, options ...testOptionFunc) *databasePack { p.root.mongoService, } rdConf.Clock = p.clock + rdConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() p.root.dbProcess, p.root.dbAuthClient, err = p.root.cluster.StartDatabase(rdConf) require.NoError(t, err) - t.Cleanup(func() { - p.root.dbProcess.Close() - }) + t.Cleanup(func() { require.NoError(t, p.root.dbProcess.Close()) }) // Create and start database services in the leaf cluster. p.leaf.postgresService = service.Database{ @@ -1122,6 +1124,7 @@ func setupDatabaseTest(t *testing.T, options ...testOptionFunc) *databasePack { p.leaf.mongoService, } ldConf.Clock = p.clock + ldConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() p.leaf.dbProcess, p.leaf.dbAuthClient, err = p.leaf.cluster.StartDatabase(ldConf) require.NoError(t, err) t.Cleanup(func() { @@ -1281,6 +1284,7 @@ func (p *databasePack) startRootDatabaseAgent(t *testing.T, params databaseAgent conf.Databases.Enabled = true conf.Databases.Databases = params.databases conf.Databases.ResourceMatchers = params.resourceMatchers + conf.CircuitBreakerConfig = breaker.NoopBreakerConfig() server, authClient, err := p.root.cluster.StartDatabase(conf) require.NoError(t, err) diff --git a/integration/ec2_test.go b/integration/ec2_test.go index 364c3535baf..eb0d4d30690 100644 --- a/integration/ec2_test.go +++ b/integration/ec2_test.go @@ -25,6 +25,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" apidefaults "github.com/gravitational/teleport/api/defaults" "github.com/gravitational/teleport/api/types" "github.com/gravitational/teleport/lib" @@ -64,6 +65,7 @@ func newNodeConfig(t *testing.T, authAddr utils.NetAddr, tokenName string, joinM config.DataDir = t.TempDir() config.AuthServers = append(config.AuthServers, authAddr) config.Log = newSilentLogger() + config.CircuitBreakerConfig = breaker.NoopBreakerConfig() return config } @@ -84,6 +86,7 @@ func newProxyConfig(t *testing.T, authAddr utils.NetAddr, tokenName string, join config.DataDir = t.TempDir() config.AuthServers = append(config.AuthServers, authAddr) config.Log = newSilentLogger() + config.CircuitBreakerConfig = breaker.NoopBreakerConfig() return config } @@ -115,6 +118,7 @@ func newAuthConfig(t *testing.T, clock clockwork.Clock) *service.Config { config.SSH.Enabled = false config.Clock = clock config.Log = newSilentLogger() + config.CircuitBreakerConfig = breaker.NoopBreakerConfig() return config } diff --git a/integration/helpers.go b/integration/helpers.go index 55993df1c81..98854949c99 100644 --- a/integration/helpers.go +++ b/integration/helpers.go @@ -36,6 +36,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "golang.org/x/crypto/ssh" "golang.org/x/crypto/ssh/agent" @@ -411,6 +412,7 @@ func (i *TeleInstance) Create(t *testing.T, trustedSecrets []*InstanceSecrets, e tconf.Log = i.log tconf.Proxy.DisableWebService = true tconf.Proxy.DisableWebInterface = true + tconf.CircuitBreakerConfig = breaker.NoopBreakerConfig() return i.CreateEx(t, trustedSecrets, tconf) } @@ -650,6 +652,8 @@ func (i *TeleInstance) GenerateConfig(t *testing.T, trustedSecrets []*InstanceSe tconf.Keygen = testauthority.New() tconf.MaxRetryPeriod = defaults.HighResPollingPeriod + tconf.CircuitBreakerConfig = breaker.NoopBreakerConfig() + i.Config = tconf return tconf, nil } @@ -1010,6 +1014,7 @@ func (i *TeleInstance) StartNodeAndProxy(name string, sshPort, proxyWebPort, pro Addr: Host, }, } + tconf.CircuitBreakerConfig = breaker.NoopBreakerConfig() // Create a new Teleport process and add it to the list of nodes that // compose this "cluster". @@ -1097,6 +1102,7 @@ func (i *TeleInstance) StartProxy(cfg ProxyConfig) (reversetunnel.Server, error) tconf.Proxy.DisableWebService = cfg.DisableWebService tconf.Proxy.DisableWebInterface = cfg.DisableWebInterface tconf.Proxy.DisableALPNSNIListener = cfg.DisableALPNSNIListener + tconf.CircuitBreakerConfig = breaker.NoopBreakerConfig() // Create a new Teleport process and add it to the list of nodes that // compose this "cluster". diff --git a/integration/hsm/hsm_test.go b/integration/hsm/hsm_test.go index 58e041746a7..edf208ca1d3 100644 --- a/integration/hsm/hsm_test.go +++ b/integration/hsm/hsm_test.go @@ -23,6 +23,7 @@ import ( "time" "github.com/google/uuid" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/trace" "github.com/jonboulle/clockwork" "github.com/stretchr/testify/require" @@ -274,6 +275,7 @@ func newHSMAuthConfig(ctx context.Context, t *testing.T, storageConfig backend.C }() config.Auth.KeyStore = keystore.SetupSoftHSMTest(t) config.Log = log + config.CircuitBreakerConfig = breaker.NoopBreakerConfig() return config } @@ -302,6 +304,7 @@ func newProxyConfig(ctx context.Context, t *testing.T, authAddr utils.NetAddr, l config.AuthServers = append(config.AuthServers, authAddr) fakeClock := clockwork.NewFakeClock() config.Clock = fakeClock + config.CircuitBreakerConfig = breaker.NoopBreakerConfig() go func() { for { select { @@ -487,6 +490,7 @@ func TestHSMDualAuthRotation(t *testing.T) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) require.NoError(t, err) return clt @@ -565,6 +569,7 @@ func TestHSMDualAuthRotation(t *testing.T) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) require.NoError(t, err) return clt @@ -790,6 +795,7 @@ func TestHSMMigrate(t *testing.T) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) require.NoError(t, err) return clt diff --git a/integration/integration_test.go b/integration/integration_test.go index a14dfa80cbf..7616dcea0ce 100644 --- a/integration/integration_test.go +++ b/integration/integration_test.go @@ -41,6 +41,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "golang.org/x/crypto/ssh" "github.com/gravitational/teleport" @@ -3841,7 +3842,7 @@ func testRotateSuccess(t *testing.T, suite *integrationTestSuite) { // shut down the service cancel() // close the service without waiting for the connections to drain - svc.Close() + require.NoError(t, svc.Close()) select { case err := <-runErrCh: @@ -4106,17 +4107,6 @@ func testRotateTrustedClusters(t *testing.T, suite *integrationTestSuite) { err = waitForProcessEvent(svc, service.TeleportPhaseChangeEvent, 10*time.Second) require.NoError(t, err) - watcher, err := services.NewCertAuthorityWatcher(ctx, services.CertAuthorityWatcherConfig{ - ResourceWatcherConfig: services.ResourceWatcherConfig{ - Component: teleport.ComponentProxy, - Clock: tconf.Clock, - Client: aux.GetSiteAPI(clusterAux), - }, - Types: []types.CertAuthType{types.HostCA}, - }) - require.NoError(t, err) - t.Cleanup(watcher.Close) - // waitForPhase waits until aux cluster detects the rotation waitForPhase := func(phase string) { require.Eventually(t, func() bool { @@ -4135,7 +4125,7 @@ func testRotateTrustedClusters(t *testing.T, suite *integrationTestSuite) { } return false - }, tconf.PollingPeriod*10, tconf.PollingPeriod/2, "failed to converge to phase %q", phase) + }, 30*time.Second, 250*time.Millisecond, "failed to converge to phase %q", phase) } waitForPhase(types.RotationPhaseInit) @@ -4374,6 +4364,8 @@ func (s *integrationTestSuite) rotationConfig(disableWebService bool) *service.C tconf.SSH.Enabled = true tconf.Proxy.DisableWebService = disableWebService tconf.Proxy.DisableWebInterface = true + tconf.Proxy.DisableDatabaseProxy = true + tconf.Proxy.DisableALPNSNIListener = true tconf.PollingPeriod = time.Second tconf.ClientTimeout = time.Second tconf.ShutdownTimeout = 2 * tconf.ClientTimeout @@ -5725,6 +5717,7 @@ func (s *integrationTestSuite) defaultServiceConfig() *service.Config { cfg := service.MakeDefaultConfig() cfg.Console = nil cfg.Log = s.log + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() return cfg } @@ -5816,6 +5809,7 @@ func TestWebProxyInsecure(t *testing.T) { rcConf.Proxy.DisableWebInterface = true // DisableTLS flag should turn off TLS termination and multiplexing. rcConf.Proxy.DisableTLS = true + rcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() err = rc.CreateEx(t, nil, rcConf) require.NoError(t, err) @@ -5873,6 +5867,7 @@ func TestTraitsPropagation(t *testing.T) { rcConf.SSH.Enabled = true rcConf.SSH.Addr.Addr = net.JoinHostPort(rc.Hostname, rc.GetPortSSH()) rcConf.SSH.Labels = map[string]string{"env": "integration"} + rcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() // Make leaf cluster config. lcConf := service.MakeDefaultConfig() @@ -5884,6 +5879,7 @@ func TestTraitsPropagation(t *testing.T) { lcConf.SSH.Enabled = true lcConf.SSH.Addr.Addr = net.JoinHostPort(lc.Hostname, lc.GetPortSSH()) lcConf.SSH.Labels = map[string]string{"env": "integration"} + lcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() // Create identical user/role in both clusters. me, err := user.Current() diff --git a/integration/kube_integration_test.go b/integration/kube_integration_test.go index 8e92c249dae..2496711e712 100644 --- a/integration/kube_integration_test.go +++ b/integration/kube_integration_test.go @@ -33,6 +33,7 @@ import ( "time" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/profile" "github.com/gravitational/teleport/api/types" "github.com/gravitational/teleport/lib" @@ -1200,6 +1201,7 @@ func (s *KubeSuite) teleKubeConfig(hostname string) *service.Config { tconf.Proxy.Kube.ListenAddr.Addr = net.JoinHostPort(hostname, ports.Pop()) tconf.Proxy.Kube.KubeconfigPath = s.kubeConfigPath tconf.Proxy.Kube.LegacyKubeProxy = true + tconf.CircuitBreakerConfig = breaker.NoopBreakerConfig() return tconf } diff --git a/integration/proxy_helpers_test.go b/integration/proxy_helpers_test.go index 85da7a1651b..e8bf54b7dd9 100644 --- a/integration/proxy_helpers_test.go +++ b/integration/proxy_helpers_test.go @@ -28,6 +28,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/trace" v1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -183,6 +184,7 @@ func (p *ProxySuite) addNodeToLeafCluster(t *testing.T, tunnelNodeHostname strin tconf.Auth.Enabled = false tconf.Proxy.Enabled = false tconf.SSH.Enabled = true + tconf.CircuitBreakerConfig = breaker.NoopBreakerConfig() return tconf } _, err := p.leaf.StartNode(nodeConfig()) @@ -318,6 +320,7 @@ func rootClusterStandardConfig(t *testing.T) func(suite *ProxySuite) *service.Co config.SSH.Enabled = true config.SSH.Addr.Addr = net.JoinHostPort(rc.Hostname, rc.GetPortSSH()) config.SSH.Labels = map[string]string{"env": "integration"} + config.CircuitBreakerConfig = breaker.NoopBreakerConfig() return config } } @@ -337,6 +340,7 @@ func leafClusterStandardConfig(t *testing.T) func(suite *ProxySuite) *service.Co config.SSH.Enabled = true config.SSH.Addr.Addr = net.JoinHostPort(lc.Hostname, lc.GetPortSSH()) config.SSH.Labels = map[string]string{"env": "integration"} + config.CircuitBreakerConfig = breaker.NoopBreakerConfig() return config } } @@ -508,6 +512,7 @@ func makeNodeConfig(nodeName, authAddr string) *service.Config { nodeConfig.Auth.Enabled = false nodeConfig.Proxy.Enabled = false nodeConfig.SSH.Enabled = true + nodeConfig.CircuitBreakerConfig = breaker.NoopBreakerConfig() return nodeConfig } diff --git a/integration/proxy_test.go b/integration/proxy_test.go index 3a74f7a0c10..660fadf145d 100644 --- a/integration/proxy_test.go +++ b/integration/proxy_test.go @@ -29,6 +29,7 @@ import ( "time" "github.com/google/uuid" + "github.com/gravitational/teleport/api/breaker" "github.com/stretchr/testify/require" "go.mongodb.org/mongo-driver/bson" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -631,6 +632,7 @@ func TestALPNProxyAuthClientConnectWithUserIdentity(t *testing.T) { rcConf.Proxy.DisableWebInterface = true rcConf.SSH.Enabled = false rcConf.Version = "v2" + rcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() username := mustGetCurrentUser(t).Username rc.AddUser(username, []string{username}) @@ -686,6 +688,7 @@ func TestALPNProxyDialProxySSHWithoutInsecureMode(t *testing.T) { rcConf.Auth.Preference.SetSecondFactor("off") rcConf.Proxy.Enabled = true rcConf.Proxy.DisableWebInterface = true + rcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() err = rc.CreateEx(t, nil, rcConf) require.NoError(t, err) @@ -752,6 +755,7 @@ func TestALPNProxyHTTPProxyNoProxyDial(t *testing.T) { rcConf.Proxy.Enabled = true rcConf.Proxy.DisableWebInterface = true rcConf.SSH.Enabled = false + rcConf.CircuitBreakerConfig = breaker.NoopBreakerConfig() err = rc.CreateEx(t, nil, rcConf) require.NoError(t, err) diff --git a/lib/auth/authclient/authclient.go b/lib/auth/authclient/authclient.go index 58a6e26019b..b3f33b9f2aa 100644 --- a/lib/auth/authclient/authclient.go +++ b/lib/auth/authclient/authclient.go @@ -22,6 +22,7 @@ import ( "context" "crypto/tls" + "github.com/gravitational/teleport/api/breaker" apiclient "github.com/gravitational/teleport/api/client" "github.com/gravitational/teleport/lib" "github.com/gravitational/teleport/lib/auth" @@ -42,6 +43,8 @@ type Config struct { AuthServers []utils.NetAddr // Log sets the logger for the client to use. Log logrus.FieldLogger + // CircuitBreakerConfig is the configuration for the auth client circuit breaker. + CircuitBreakerConfig breaker.Config } // Connect creates a valid client connection to the auth service. It may @@ -56,7 +59,8 @@ func Connect(ctx context.Context, cfg *Config) (auth.ClientI, error) { apiclient.LoadTLS(cfg.TLS), }, // Deliberately ignore HTTP proxies for backwards compatibility. - IgnoreHTTPProxy: true, + IgnoreHTTPProxy: true, + CircuitBreakerConfig: cfg.CircuitBreakerConfig, }) if err != nil { return nil, trace.Wrap(err, "failed direct dial to auth server: %v", err) diff --git a/lib/auth/clt.go b/lib/auth/clt.go index 9973d0d49b7..6c92705d1fe 100644 --- a/lib/auth/clt.go +++ b/lib/auth/clt.go @@ -29,6 +29,7 @@ import ( "time" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/client" "github.com/gravitational/teleport/api/client/proto" "github.com/gravitational/teleport/api/constants" @@ -186,9 +187,14 @@ func NewHTTPClient(cfg client.Config, tls *tls.Config, params ...roundtrip.Clien IdleConnTimeout: defaults.HTTPIdleTimeout, } + cb, err := breaker.New(cfg.CircuitBreakerConfig) + if err != nil { + return nil, trace.Wrap(err) + } + clientParams := append( []roundtrip.ClientParam{ - roundtrip.HTTPClient(&http.Client{Transport: otelhttp.NewTransport(transport)}), + roundtrip.HTTPClient(&http.Client{Transport: otelhttp.NewTransport(breaker.NewRoundTripper(cb, transport))}), roundtrip.SanitizerEnabled(true), }, params..., @@ -1595,7 +1601,7 @@ type IdentityService interface { UpsertOIDCConnector(ctx context.Context, connector types.OIDCConnector) error // GetOIDCConnector returns OIDC connector information by id GetOIDCConnector(ctx context.Context, id string, withSecrets bool) (types.OIDCConnector, error) - // GetOIDCConnector gets OIDC connectors list + // GetOIDCConnectors gets OIDC connectors list GetOIDCConnectors(ctx context.Context, withSecrets bool) ([]types.OIDCConnector, error) // DeleteOIDCConnector deletes OIDC connector by ID DeleteOIDCConnector(ctx context.Context, connectorID string) error @@ -1696,7 +1702,7 @@ type IdentityService interface { // (https://github.com/gravitational/teleport/blob/3a1cf9111c2698aede2056513337f32bfc16f1f1/rfd/0014-session-2FA.md#sessions). GenerateUserSingleUseCerts(ctx context.Context) (proto.AuthService_GenerateUserSingleUseCertsClient, error) - // IsMFARequiredRequest is a request to check whether MFA is required to + // IsMFARequired is a request to check whether MFA is required to // access the Target. IsMFARequired(ctx context.Context, req *proto.IsMFARequiredRequest) (*proto.IsMFARequiredResponse, error) diff --git a/lib/auth/clt_test.go b/lib/auth/clt_test.go index 2656a592cc6..997c551f19b 100644 --- a/lib/auth/clt_test.go +++ b/lib/auth/clt_test.go @@ -25,6 +25,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" apiclient "github.com/gravitational/teleport/api/client" "github.com/gravitational/teleport/api/types" "github.com/gravitational/teleport/lib/auth/testauthority" @@ -63,6 +64,7 @@ func TestClient_DialTimeout(t *testing.T) { Credentials: []apiclient.Credentials{ apiclient.LoadTLS(&tls.Config{}), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), } clt, err := NewClient(cfg) require.NoError(t, err) @@ -118,6 +120,7 @@ func TestClient_RequestTimeout(t *testing.T) { Credentials: []apiclient.Credentials{ apiclient.LoadTLS(srv.TLS), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), } clt, err := NewClient(cfg) require.NoError(t, err) diff --git a/lib/auth/helpers.go b/lib/auth/helpers.go index 5b90be31b43..42db54063ae 100644 --- a/lib/auth/helpers.go +++ b/lib/auth/helpers.go @@ -26,6 +26,7 @@ import ( "time" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/client" "github.com/gravitational/teleport/api/client/proto" "github.com/gravitational/teleport/api/constants" @@ -566,6 +567,7 @@ func (a *TestAuthServer) NewRemoteClient(identity TestIdentity, addr net.Addr, p Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) } @@ -770,6 +772,7 @@ func (t *TestTLSServer) NewClientFromWebSession(sess types.WebSession) (*Client, Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) } @@ -811,6 +814,7 @@ func (t *TestTLSServer) CloneClient(clt *Client) *Client { Credentials: []client.Credentials{ client.LoadTLS(clt.Config()), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) if err != nil { panic(err) @@ -831,6 +835,7 @@ func (t *TestTLSServer) NewClientWithCert(clientCert tls.Certificate) *Client { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) if err != nil { panic(err) @@ -851,6 +856,7 @@ func (t *TestTLSServer) NewClient(identity TestIdentity) (*Client, error) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) if err != nil { return nil, trace.Wrap(err) diff --git a/lib/auth/register.go b/lib/auth/register.go index 609523c7881..48b709fe78c 100644 --- a/lib/auth/register.go +++ b/lib/auth/register.go @@ -21,6 +21,7 @@ import ( "crypto/x509" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/client" "github.com/gravitational/teleport/api/client/proto" "github.com/gravitational/teleport/api/metadata" @@ -116,6 +117,8 @@ type RegisterParams struct { // ec2IdentityDocument is used for Simplified Node Joining to prove the // identity of a joining EC2 instance. ec2IdentityDocument []byte + // CircuitBreakerConfig defines how the circuit breaker should behave. + CircuitBreakerConfig breaker.Config } func (r *RegisterParams) setDefaults() { @@ -344,6 +347,7 @@ func insecureRegisterClient(params RegisterParams) (*Client, error) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: params.CircuitBreakerConfig, }) if err != nil { return nil, trace.Wrap(err) @@ -382,6 +386,7 @@ func pinRegisterClient(params RegisterParams) (*Client, error) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: params.CircuitBreakerConfig, }) if err != nil { return nil, trace.Wrap(err) @@ -432,6 +437,7 @@ func pinRegisterClient(params RegisterParams) (*Client, error) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: params.CircuitBreakerConfig, }) if err != nil { return nil, trace.Wrap(err) diff --git a/lib/auth/tls_test.go b/lib/auth/tls_test.go index db211e8e8ba..f11d0385689 100644 --- a/lib/auth/tls_test.go +++ b/lib/auth/tls_test.go @@ -28,6 +28,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "golang.org/x/crypto/ssh" "github.com/stretchr/testify/require" @@ -2246,6 +2247,7 @@ func (s *TLSSuite) TestCipherSuites(c *check.C) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) c.Assert(err, check.IsNil) @@ -2273,6 +2275,7 @@ func (s *TLSSuite) TestTLSFailover(c *check.C) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) c.Assert(err, check.IsNil) diff --git a/lib/client/api_login_test.go b/lib/client/api_login_test.go index 5ebb175d6c2..a7a5dfa218b 100644 --- a/lib/client/api_login_test.go +++ b/lib/client/api_login_test.go @@ -27,6 +27,7 @@ import ( "time" "github.com/google/uuid" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/client/proto" "github.com/gravitational/teleport/api/constants" "github.com/gravitational/teleport/api/types" @@ -278,6 +279,7 @@ func newStandaloneTeleport(t *testing.T, clock clockwork.Clock) *standaloneBundl cfg.Auth.SSHAddr = randomAddr cfg.Proxy.Enabled = false cfg.SSH.Enabled = false + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() authProcess := startAndWait(t, cfg, service.AuthTLSReady) t.Cleanup(func() { authProcess.Close() }) authAddr, err := authProcess.AuthSSHAddr() @@ -344,6 +346,7 @@ func newStandaloneTeleport(t *testing.T, clock clockwork.Clock) *standaloneBundl cfg.Proxy.ReverseTunnelListenAddr = randomAddr cfg.Proxy.DisableWebInterface = true cfg.SSH.Enabled = false + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() proxyProcess := startAndWait(t, cfg, service.ProxyWebServerReady) t.Cleanup(func() { proxyProcess.Close() }) proxyWebAddr, err := proxyProcess.ProxyWebAddr() diff --git a/lib/client/client.go b/lib/client/client.go index 6c05a0d9565..baff7e89f22 100644 --- a/lib/client/client.go +++ b/lib/client/client.go @@ -31,6 +31,7 @@ import ( "time" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/client" "github.com/gravitational/teleport/api/client/proto" apidefaults "github.com/gravitational/teleport/api/defaults" @@ -848,6 +849,7 @@ func (proxy *ProxyClient) ConnectToAuthServiceThroughALPNSNIProxy(ctx context.Co client.LoadTLS(tlsConfig), }, ALPNSNIAuthDialClusterName: clusterName, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) if err != nil { return nil, trace.Wrap(err) @@ -881,6 +883,7 @@ func (proxy *ProxyClient) ConnectToCluster(ctx context.Context, clusterName stri Credentials: []client.Credentials{ client.LoadTLS(proxy.teleportClient.TLS), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) } @@ -898,6 +901,7 @@ func (proxy *ProxyClient) ConnectToCluster(ctx context.Context, clusterName stri Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) if err != nil { return nil, trace.Wrap(err) diff --git a/lib/reversetunnel/remotesite.go b/lib/reversetunnel/remotesite.go index fcda0836773..1678b66132b 100644 --- a/lib/reversetunnel/remotesite.go +++ b/lib/reversetunnel/remotesite.go @@ -118,6 +118,7 @@ func (s *remoteSite) getRemoteClient() (auth.ClientI, bool, error) { Credentials: []client.Credentials{ client.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: s.srv.CircuitBreakerConfig, }) if err != nil { return nil, false, trace.Wrap(err) diff --git a/lib/reversetunnel/srv.go b/lib/reversetunnel/srv.go index be2fc124c85..134d84ac332 100644 --- a/lib/reversetunnel/srv.go +++ b/lib/reversetunnel/srv.go @@ -27,6 +27,7 @@ import ( "time" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/constants" "github.com/gravitational/teleport/api/types" apisshutils "github.com/gravitational/teleport/api/utils/sshutils" @@ -212,6 +213,9 @@ type Config struct { // CertAuthorityWatcher is a cert authority watcher. CertAuthorityWatcher *services.CertAuthorityWatcher + + // CircuitBreakerConfig configures the auth client circuit breaker + CircuitBreakerConfig breaker.Config } // CheckAndSetDefaults checks parameters and sets default values diff --git a/lib/service/cfg.go b/lib/service/cfg.go index 7408e68f01a..43da660f588 100644 --- a/lib/service/cfg.go +++ b/lib/service/cfg.go @@ -32,6 +32,7 @@ import ( "time" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/types" apiutils "github.com/gravitational/teleport/api/utils" "github.com/gravitational/teleport/lib/auth" @@ -268,6 +269,9 @@ type Config struct { // TeleportHome is the path to tsh configuration and data, used // for loading profiles when TELEPORT_HOME is set TeleportHome string + + // CircuitBreakerConfig configures the auth client circuit breaker. + CircuitBreakerConfig breaker.Config } // ApplyToken assigns a given token to all internal services but only if token @@ -1293,6 +1297,7 @@ func ApplyDefaults(cfg *Config) { } cfg.MaxRetryPeriod = defaults.MaxWatcherBackoff cfg.ConnectFailureC = make(chan time.Duration, 1) + cfg.CircuitBreakerConfig = breaker.DefaultBreakerConfig(cfg.Clock) } // ApplyFIPSDefaults updates default configuration to be FedRAMP/FIPS 140-2 diff --git a/lib/service/connect.go b/lib/service/connect.go index d0a4f95d7d6..cdbff0d8d3f 100644 --- a/lib/service/connect.go +++ b/lib/service/connect.go @@ -429,6 +429,7 @@ func (process *TeleportProcess) firstTimeConnect(role types.SystemRole) (*Connec GetHostCredentials: client.HostCredentials, Clock: process.Clock, JoinMethod: process.Config.JoinMethod, + CircuitBreakerConfig: process.Config.CircuitBreakerConfig, }) if err != nil { return nil, trace.Wrap(err) @@ -933,6 +934,7 @@ func (process *TeleportProcess) newClientThroughTunnel(authServers []utils.NetAd Credentials: []apiclient.Credentials{ apiclient.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: process.Config.CircuitBreakerConfig, }) if err != nil { return nil, trace.Wrap(err) @@ -974,7 +976,8 @@ func (process *TeleportProcess) newClientDirect(authServers []utils.NetAddr, tls Credentials: []apiclient.Credentials{ apiclient.LoadTLS(tlsConfig), }, - DialOpts: dialOpts, + CircuitBreakerConfig: process.Config.CircuitBreakerConfig, + DialOpts: dialOpts, // Deliberately ignore HTTP proxies for backwards compatibility. IgnoreHTTPProxy: true, }, cltParams...) diff --git a/lib/service/service.go b/lib/service/service.go index f9ca40de640..02fd592e550 100644 --- a/lib/service/service.go +++ b/lib/service/service.go @@ -3097,6 +3097,7 @@ func (process *TeleportProcess) initProxyEndpoint(conn *Connector) error { PeerClient: peerClient, NodeWatcher: nodeWatcher, CertAuthorityWatcher: caWatcher, + CircuitBreakerConfig: process.Config.CircuitBreakerConfig, }) if err != nil { return trace.Wrap(err) diff --git a/lib/service/service_test.go b/lib/service/service_test.go index 72ebceef65e..5e13fabf998 100644 --- a/lib/service/service_test.go +++ b/lib/service/service_test.go @@ -30,6 +30,7 @@ import ( "github.com/coreos/go-semver/semver" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" + "github.com/gravitational/teleport/api/breaker" "golang.org/x/crypto/ssh" "github.com/gravitational/teleport" @@ -100,6 +101,7 @@ func TestMonitor(t *testing.T) { cfg.Auth.SSHAddr = utils.NetAddr{AddrNetwork: "tcp", Addr: "127.0.0.1:0"} cfg.Proxy.Enabled = false cfg.SSH.Enabled = false + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() process, err := NewTeleport(cfg) require.NoError(t, err) @@ -526,6 +528,7 @@ func TestSetupProxyTLSConfig(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { cfg := MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() cfg.Proxy.ACME.Enabled = tc.acmeEnabled cfg.DataDir = t.TempDir() cfg.Proxy.PublicAddrs = utils.MustParseAddrList("localhost") @@ -565,6 +568,7 @@ func TestTeleportProcess_reconnectToAuth(t *testing.T) { cfg.Proxy.Enabled = false cfg.SSH.Enabled = true cfg.MaxRetryPeriod = defaults.MaxWatcherBackoff + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() cfg.ConnectFailureC = make(chan time.Duration, 5) process, err := NewTeleport(cfg) require.NoError(t, err) diff --git a/lib/srv/regular/sshserver_test.go b/lib/srv/regular/sshserver_test.go index 5679734949d..1c5f41b3b6d 100644 --- a/lib/srv/regular/sshserver_test.go +++ b/lib/srv/regular/sshserver_test.go @@ -33,6 +33,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "golang.org/x/crypto/ssh" "golang.org/x/crypto/ssh/agent" @@ -1145,6 +1146,7 @@ func TestProxyRoundRobin(t *testing.T) { LockWatcher: lockWatcher, NodeWatcher: nodeWatcher, CertAuthorityWatcher: caWatcher, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) require.NoError(t, err) logger.WithField("tun-addr", reverseTunnelAddress.String()).Info("Created reverse tunnel server.") @@ -1265,6 +1267,7 @@ func TestProxyDirectAccess(t *testing.T) { LockWatcher: lockWatcher, NodeWatcher: nodeWatcher, CertAuthorityWatcher: caWatcher, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) require.NoError(t, err) diff --git a/lib/web/apiserver_test.go b/lib/web/apiserver_test.go index 38454d0f680..a379fe2d682 100644 --- a/lib/web/apiserver_test.go +++ b/lib/web/apiserver_test.go @@ -60,6 +60,7 @@ import ( kyaml "k8s.io/apimachinery/pkg/util/yaml" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" apiProto "github.com/gravitational/teleport/api/client/proto" authproto "github.com/gravitational/teleport/api/client/proto" "github.com/gravitational/teleport/api/client/webclient" @@ -300,6 +301,7 @@ func newWebSuite(t *testing.T) *WebSuite { LockWatcher: proxyLockWatcher, NodeWatcher: proxyNodeWatcher, CertAuthorityWatcher: caWatcher, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) require.NoError(t, err) s.proxyTunnel = revTunServer @@ -3840,6 +3842,7 @@ func createProxy(ctx context.Context, t *testing.T, proxyID string, node *regula LockWatcher: proxyLockWatcher, NodeWatcher: proxyNodeWatcher, CertAuthorityWatcher: proxyCAWatcher, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) require.NoError(t, err) t.Cleanup(func() { require.NoError(t, revTunServer.Close()) }) diff --git a/lib/web/sessions.go b/lib/web/sessions.go index daf0259d412..31aa251c828 100644 --- a/lib/web/sessions.go +++ b/lib/web/sessions.go @@ -26,6 +26,7 @@ import ( "sync" "time" + "github.com/gravitational/teleport/api/breaker" "golang.org/x/crypto/ssh" "golang.org/x/crypto/ssh/agent" @@ -256,6 +257,7 @@ func (c *SessionContext) newRemoteTLSClient(cluster reversetunnel.RemoteSite) (a Credentials: []apiclient.Credentials{ apiclient.LoadTLS(tlsConfig), }, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) } @@ -818,8 +820,9 @@ func (s *sessionCache) newSessionContextFromSession(session types.WebSession) (* return nil, trace.Wrap(err) } userClient, err := auth.NewClient(apiclient.Config{ - Addrs: utils.NetAddrsToStrings(s.authServers), - Credentials: []apiclient.Credentials{apiclient.LoadTLS(tlsConfig)}, + Addrs: utils.NetAddrsToStrings(s.authServers), + Credentials: []apiclient.Credentials{apiclient.LoadTLS(tlsConfig)}, + CircuitBreakerConfig: breaker.NoopBreakerConfig(), }) if err != nil { return nil, trace.Wrap(err) diff --git a/tool/tctl/common/helpers_test.go b/tool/tctl/common/helpers_test.go index b3c32e580dc..1b7007028dd 100644 --- a/tool/tctl/common/helpers_test.go +++ b/tool/tctl/common/helpers_test.go @@ -27,6 +27,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/defaults" "github.com/gravitational/teleport/lib/auth" "github.com/gravitational/teleport/lib/auth/authclient" @@ -68,6 +69,7 @@ func runResourceCommand(t *testing.T, fc *config.FileConfig, args []string, opts stdout: &stdoutBuff, } cfg := service.MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() app := utils.InitCLIParser("tctl", GlobalHelpString) command.Initialize(app, cfg) @@ -181,6 +183,7 @@ func makeAndRunTestAuthServer(t *testing.T, opts ...testServerOptionFunc) (auth var err error cfg := service.MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() if options.fileConfig != nil { err = config.ApplyFileConfig(options.fileConfig, cfg) require.NoError(t, err) diff --git a/tool/tctl/common/tctl.go b/tool/tctl/common/tctl.go index cc7275ba002..371c1efe3c8 100644 --- a/tool/tctl/common/tctl.go +++ b/tool/tctl/common/tctl.go @@ -24,6 +24,7 @@ import ( "path/filepath" "syscall" + "github.com/gravitational/teleport/api/breaker" "golang.org/x/crypto/ssh" "github.com/gravitational/teleport" @@ -95,6 +96,7 @@ func Run(commands []CLICommand) { // cfg (teleport auth server configuration) is going to be shared by all // commands cfg := service.MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() // each command will add itself to the CLI parser: for i := range commands { diff --git a/tool/tsh/db_test.go b/tool/tsh/db_test.go index c08a1ed8e85..7ab34533b8e 100644 --- a/tool/tsh/db_test.go +++ b/tool/tsh/db_test.go @@ -26,6 +26,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/constants" apidefaults "github.com/gravitational/teleport/api/defaults" "github.com/gravitational/teleport/api/types" @@ -247,6 +248,7 @@ func makeTestDatabaseServer(t *testing.T, auth *service.TeleportProcess, proxy * cfg := service.MakeDefaultConfig() cfg.Hostname = "localhost" cfg.DataDir = t.TempDir() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() proxyAddr, err := proxy.ProxyWebAddr() require.NoError(t, err) diff --git a/tool/tsh/tsh_helper_test.go b/tool/tsh/tsh_helper_test.go index f03e7b533ee..0afd3f5b7fa 100644 --- a/tool/tsh/tsh_helper_test.go +++ b/tool/tsh/tsh_helper_test.go @@ -23,6 +23,7 @@ import ( "testing" "time" + "github.com/gravitational/teleport/api/breaker" "github.com/stretchr/testify/require" apiclient "github.com/gravitational/teleport/api/client" @@ -69,6 +70,7 @@ func (s *suite) setupRootCluster(t *testing.T, options testSuiteOptions) { } cfg := service.MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() err := config.ApplyFileConfig(fileConfig, cfg) require.NoError(t, err) @@ -138,6 +140,7 @@ func (s *suite) setupLeafCluster(t *testing.T, options testSuiteOptions) { } cfg := service.MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() err := config.ApplyFileConfig(fileConfig, cfg) require.NoError(t, err) diff --git a/tool/tsh/tsh_test.go b/tool/tsh/tsh_test.go index c8e8fb76629..ae2eb8a5df7 100644 --- a/tool/tsh/tsh_test.go +++ b/tool/tsh/tsh_test.go @@ -31,6 +31,7 @@ import ( "time" "github.com/gravitational/teleport" + "github.com/gravitational/teleport/api/breaker" "github.com/gravitational/teleport/api/constants" apidefaults "github.com/gravitational/teleport/api/defaults" "github.com/gravitational/teleport/api/profile" @@ -146,10 +147,10 @@ func TestFailedLogin(t *testing.T) { "--debug", "--auth", connector.GetName(), "--proxy", proxyAddr.String(), - }, setHomePath(tmpHomePath), cliOption(func(cf *CLIConf) error { - cf.mockSSOLogin = client.SSOLoginFunc(ssoLogin) + }, setHomePath(tmpHomePath), func(cf *CLIConf) error { + cf.mockSSOLogin = ssoLogin return nil - })) + }) require.ErrorIs(t, err, loginFailed) } @@ -236,12 +237,12 @@ func TestOIDCLogin(t *testing.T) { "--auth", connector.GetName(), "--proxy", proxyAddr.String(), "--user", "alice", // explicitly use wrong name - }, setHomePath(tmpHomePath), cliOption(func(cf *CLIConf) error { + }, setHomePath(tmpHomePath), func(cf *CLIConf) error { cf.mockSSOLogin = mockSSOLogin(t, authServer, alice) cf.SiteName = "localhost" cf.overrideStderr = buf return nil - })) + }) require.NoError(t, err) @@ -292,10 +293,10 @@ func TestLoginIdentityOut(t *testing.T) { "--auth", connector.GetName(), "--proxy", proxyAddr.String(), "--out", identPath, - }, setHomePath(tmpHomePath), cliOption(func(cf *CLIConf) error { + }, setHomePath(tmpHomePath), func(cf *CLIConf) error { cf.mockSSOLogin = mockSSOLogin(t, authServer, alice) return nil - })) + }) require.NoError(t, err) _, err = client.KeyFromIdentityFile(identPath) @@ -331,11 +332,11 @@ func TestRelogin(t *testing.T) { "--debug", "--auth", connector.GetName(), "--proxy", proxyAddr.String(), - }, setHomePath(tmpHomePath), cliOption(func(cf *CLIConf) error { + }, setHomePath(tmpHomePath), func(cf *CLIConf) error { cf.mockSSOLogin = mockSSOLogin(t, authServer, alice) cf.overrideStderr = buf return nil - })) + }) require.NoError(t, err) findMOTD(t, sc, motd) @@ -346,19 +347,19 @@ func TestRelogin(t *testing.T) { "--proxy", proxyAddr.String(), "localhost", }, setHomePath(tmpHomePath), - cliOption(func(cf *CLIConf) error { + func(cf *CLIConf) error { cf.mockSSOLogin = mockSSOLogin(t, authServer, alice) cf.overrideStderr = buf return nil - })) + }) require.NoError(t, err) findMOTD(t, sc, motd) err = Run(context.Background(), []string{"logout"}, setHomePath(tmpHomePath), - cliOption(func(cf *CLIConf) error { + func(cf *CLIConf) error { cf.overrideStderr = buf return nil - })) + }) require.NoError(t, err) err = Run(context.Background(), []string{ @@ -368,11 +369,11 @@ func TestRelogin(t *testing.T) { "--auth", connector.GetName(), "--proxy", proxyAddr.String(), "localhost", - }, setHomePath(tmpHomePath), cliOption(func(cf *CLIConf) error { + }, setHomePath(tmpHomePath), func(cf *CLIConf) error { cf.mockSSOLogin = mockSSOLogin(t, authServer, alice) cf.overrideStderr = buf return nil - })) + }) findMOTD(t, sc, motd) require.NoError(t, err) } @@ -551,10 +552,10 @@ func TestAccessRequestOnLeaf(t *testing.T) { "--debug", "--auth", connector.GetName(), "--proxy", rootProxyAddr.String(), - }, setHomePath(tmpHomePath), cliOption(func(cf *CLIConf) error { + }, setHomePath(tmpHomePath), func(cf *CLIConf) error { cf.mockSSOLogin = mockSSOLogin(t, rootAuthServer, alice) return nil - })) + }) require.NoError(t, err) err = Run(context.Background(), []string{ @@ -1404,6 +1405,7 @@ func makeTestServers(t *testing.T, opts ...testServerOptFunc) (auth *service.Tel // We need this to get a random port assigned to it and allow parallel // execution of this test. cfg := service.MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() cfg.Hostname = "localhost" cfg.DataDir = t.TempDir() @@ -1457,6 +1459,7 @@ func makeTestServers(t *testing.T, opts ...testServerOptFunc) (auth *service.Tel // Set up a test proxy service. cfg = service.MakeDefaultConfig() + cfg.CircuitBreakerConfig = breaker.NoopBreakerConfig() cfg.Hostname = "localhost" cfg.DataDir = t.TempDir() diff --git a/trace.out b/trace.out new file mode 100644 index 00000000000..bbb077e130c Binary files /dev/null and b/trace.out differ