hystrix

Documentation

Overview

Package hystrix is a latency and fault tolerance library designed to isolate points of access to remote systems, services and 3rd party libraries, stop cascading failure and enable resilience in complex distributed systems where failure is inevitable.

Based on the java project of the same name, by Netflix. https://github.com/Netflix/Hystrix

Execute code as a Hystrix command

Define your application logic which relies on external systems, passing your function to Go. When that system is healthy this will be the only thing which executes.

hystrix.Go("my_command", func() error {
	// talk to other services
	return nil
}, nil)

Defining fallback behavior

If you want code to execute during a service outage, pass in a second function to Go. Ideally, the logic here will allow your application to gracefully handle external services being unavailable.

This triggers when your code returns an error, or whenever it is unable to complete based on a variety of health checks https://github.com/Netflix/Hystrix/wiki/How-it-Works.

hystrix.Go("my_command", func() error {
	// talk to other services
	return nil
}, func(err error) error {
	// do this when services are down
	return nil
})

Waiting for output

Calling Go is like launching a goroutine, except you receive a channel of errors you can choose to monitor.

output := make(chan bool, 1)
errors := hystrix.Go("my_command", func() error {
	// talk to other services
	output <- true
	return nil
}, nil)

select {
case out := <-output:
	// success
case err := <-errors:
	// failure
}

Synchronous API

Since calling a command and immediately waiting for it to finish is a common pattern, a synchronous API is available with the Do function which returns a single error.

err := hystrix.Do("my_command", func() error {
	// talk to other services
	return nil
}, nil)

Configure settings

During application boot, you can call ConfigureCommand to tweak the settings for each command.

hystrix.ConfigureCommand("my_command", hystrix.CommandConfig{
	Timeout:               1000,
	MaxConcurrentRequests: 100,
	ErrorPercentThreshold: 25,
})

You can also use Configure which accepts a map[string]CommandConfig.

Enable dashboard metrics

In your main.go, register the event stream HTTP handler on a port and launch it in a goroutine. Once you configure turbine for your Hystrix Dashboard https://github.com/Netflix/Hystrix/tree/master/hystrix-dashboard to start streaming events, your commands will automatically begin appearing.

hystrixStreamHandler := hystrix.NewStreamHandler()
hystrixStreamHandler.Start()
go http.ListenAndServe(net.JoinHostPort("", "81"), hystrixStreamHandler)

Index

• Constants
• Variables
• func Configure(cmds map[string]CommandConfig)
• func ConfigureCommand(name string, config CommandConfig)
• func Do(name string, run runFunc, fallback fallbackFunc) error
• func Flush()
• func FlushByName(name string)
• func GetCircuitSettings() map[string]*Settings
• func Go(name string, run runFunc, fallback fallbackFunc) chan error
• type CircuitBreaker
  • func GetCircuit(name string) (*CircuitBreaker, bool, error)
  • func (circuit *CircuitBreaker) AllowRequest() bool
  • func (circuit *CircuitBreaker) IsOpen() bool
  • func (circuit *CircuitBreaker) ReportEvent(eventTypes []string, start time.Time, runDuration time.Duration) error
  • func (circuit *CircuitBreaker) ToggleForceOpen(toggle bool) error
• type CircuitError
  • func (e CircuitError) Error() string
• type CommandConfig
  • func NewCommandConfig(opt ...CommandConfigOption) CommandConfig
• type CommandConfigOption
  • func WithErrorPercent(errorpercent int) CommandConfigOption
  • func WithMaxRequests(maxrequests int) CommandConfigOption
  • func WithSleepWindow(sleepwindow int) CommandConfigOption
  • func WithTimeOut(timeout int) CommandConfigOption
  • func WithVolumeThreshold(volumethreshold int) CommandConfigOption
• type FallbackNullError
  • func (e FallbackNullError) Error() string
• type Settings
• type StreamHandler
  • func NewStreamHandler() *StreamHandler
  • func (sh *StreamHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request)
  • func (sh *StreamHandler) Start()
  • func (sh *StreamHandler) Stop()

Constants

const ConcurrentRequestsLimit = 1000

Variables

var (
	// ErrMaxConcurrency occurs when too many of the same named command are executed at the same time.
	ErrMaxConcurrency = CircuitError{Message: "max concurrency"}
	// ErrCircuitOpen returns when an execution attempt "short circuits". This happens due to the circuit being measured as unhealthy.
	ErrCircuitOpen = CircuitError{Message: "circuit open"}
	// ErrTimeout occurs when the provided function takes too long to execute.
	ErrTimeout = CircuitError{Message: "timeout"}
	// ErrForceFallback occurs when force fallback is true
	ErrForceFallback = CircuitError{Message: "force fallback"}
)

var (
	// DefaultTimeout is how long to wait for command to complete, in milliseconds
	DefaultTimeout = 1000

	// DefaultMaxConcurrent is how many commands of the same type can run at the same time
	DefaultMaxConcurrent = 10

	// DefaultVolumeThreshold is the minimum number of requests needed before a circuit can be tripped due to health
	DefaultVolumeThreshold = 20

	// DefaultSleepWindow is how long, in milliseconds, to wait after a circuit opens before testing for recovery
	DefaultSleepWindow = 5000

	// DefaultErrorPercentThreshold causes circuits to open once the rolling measure of errors exceeds this percent of requests
	DefaultErrorPercentThreshold = 50
)

Functions

func Configure

func Configure(cmds map[string]CommandConfig)

Configure applies settings for a set of circuits

func ConfigureCommand

func ConfigureCommand(name string, config CommandConfig)

ConfigureCommand applies settings for a circuit

func Do

func Do(name string, run runFunc, fallback fallbackFunc) error

Do runs your function in a synchronous manner, blocking until either your function succeeds or an error is returned, including hystrix circuit errors

func Flush

func Flush()

Flush purges all circuit and metric information from memory.

func FlushByName

func FlushByName(name string)

func GetCircuitSettings

func GetCircuitSettings() map[string]*Settings

func Go

func Go(name string, run runFunc, fallback fallbackFunc) chan error

Go runs your function while tracking the health of previous calls to it. If your function begins slowing down or failing repeatedly, we will block new calls to it for you to give the dependent service time to repair.

Define a fallback function if you want to define some code to execute during outages.

Types

type CircuitBreaker

type CircuitBreaker struct {
	Name string
	// contains filtered or unexported fields
}

CircuitBreaker is created for each ExecutorPool to track whether requests should be attempted, or rejected if the Health of the circuit is too low.

func GetCircuit

func GetCircuit(name string) (*CircuitBreaker, bool, error)

GetCircuit returns the circuit for the given command and whether this call created it.

func (*CircuitBreaker) AllowRequest

func (circuit *CircuitBreaker) AllowRequest() bool

AllowRequest is checked before a command executes, ensuring that circuit state and metric health allow it. When the circuit is open, this call will occasionally return true to measure whether the external service has recovered.

func (*CircuitBreaker) IsOpen

func (circuit *CircuitBreaker) IsOpen() bool

IsOpen is called before any Command execution to check whether or not it should be attempted. An "open" circuit means it is disabled.

func (*CircuitBreaker) ReportEvent

func (circuit *CircuitBreaker) ReportEvent(eventTypes []string, start time.Time, runDuration time.Duration) error

ReportEvent records command metrics for tracking recent error rates and exposing data to the dashboard.

func (*CircuitBreaker) ToggleForceOpen

func (circuit *CircuitBreaker) ToggleForceOpen(toggle bool) error

toggleForceOpen allows manually causing the fallback logic for all instances of a given command.

type CircuitError

type CircuitError struct {
	Message string
}

A CircuitError is an error which models various failure states of execution, such as the circuit being open or a timeout.

func (CircuitError) Error

func (e CircuitError) Error() string

type CommandConfig

type CommandConfig struct {
	TimeoutEnabled         bool
	Timeout                int `json:"timeout"`
	MaxConcurrentRequests  int `json:"max_concurrent_requests"`
	RequestVolumeThreshold int `json:"request_volume_threshold"`
	SleepWindow            int `json:"sleep_window"`
	ErrorPercentThreshold  int `json:"error_percent_threshold"`
	//动态治理
	ForceFallback         bool
	CircuitBreakerEnabled bool
	ForceOpen             bool
	ForceClose            bool
}

CommandConfig is used to tune circuit settings at runtime

func NewCommandConfig

func NewCommandConfig(opt ...CommandConfigOption) CommandConfig

新建一个CommandConfig返回

type CommandConfigOption

type CommandConfigOption func(*CommandConfig)

func WithErrorPercent

func WithErrorPercent(errorpercent int) CommandConfigOption

func WithMaxRequests

func WithMaxRequests(maxrequests int) CommandConfigOption

func WithSleepWindow

func WithSleepWindow(sleepwindow int) CommandConfigOption

func WithTimeOut

func WithTimeOut(timeout int) CommandConfigOption

func WithVolumeThreshold

func WithVolumeThreshold(volumethreshold int) CommandConfigOption

type FallbackNullError

type FallbackNullError struct {
	Message string
}

func (FallbackNullError) Error

func (e FallbackNullError) Error() string

type Settings

type Settings struct {
	// isolation 属性
	Timeout               time.Duration
	TimeoutEnabled        bool
	MaxConcurrentRequests int

	// circuit break 属性
	CircuitBreakerEnabled  bool
	RequestVolumeThreshold uint64
	SleepWindow            time.Duration
	ErrorPercentThreshold  int

	//动态治理
	ForceFallback bool
	ForceOpen     bool
	ForceClose    bool
}

type StreamHandler

type StreamHandler struct {
	// contains filtered or unexported fields
}

StreamHandler publishes metrics for each command and each pool once a second to all connected HTTP client.

func NewStreamHandler

func NewStreamHandler() *StreamHandler

NewStreamHandler returns a server capable of exposing dashboard metrics via HTTP.

func (*StreamHandler) ServeHTTP

func (sh *StreamHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request)

func (*StreamHandler) Start

func (sh *StreamHandler) Start()

Start begins watching the in-memory circuit breakers for metrics

func (*StreamHandler) Stop

func (sh *StreamHandler) Stop()

Stop shuts down the metric collection routine