func newConnectionWithTransportAndProtocols(handler ConnectionHandler,
	transport ConnectionTransport, errorUnwrapper ErrorUnwrapper,
	connectNow bool, wef WrapErrorFunc, log LogOutput,
	tagsFunc LogTagsFromContext, protocols []Protocol) *Connection {
	// retry w/exponential backoff
	reconnectBackoff := backoff.NewExponentialBackOff()
	// never give up reconnecting
	reconnectBackoff.MaxElapsedTime = 0
	randBytes := make([]byte, 4)
	rand.Read(randBytes)
	connectionPrefix := fmt.Sprintf("CONN %s %x", handler.HandlerName(),
		randBytes)
	connection := &Connection{
		handler:          handler,
		transport:        transport,
		errorUnwrapper:   errorUnwrapper,
		reconnectBackoff: reconnectBackoff,
		doCommandBackoff: backoff.NewExponentialBackOff(),
		wef:              wef,
		tagsFunc:         tagsFunc,
		log: connectionLog{
			LogOutput: log,
			logPrefix: connectionPrefix,
		},
		protocols: protocols,
	}
	if connectNow {
		// start connecting now
		connection.getReconnectChan()
	}
	return connection
}

func (d *Decoder) backOff() backoff.BackOff {
	result := backoff.NewExponentialBackOff()
	result.InitialInterval = 1 * time.Second
	result.MaxInterval = 10 * time.Second
	result.MaxElapsedTime = 0
	return result
}

func refreshFromSource() {
	certData := bytes.NewBuffer([]byte{})

	bo := backoff.NewExponentialBackOff()
	bo.MaxElapsedTime = time.Minute

	err := backoff.Retry(func() error {
		src, err := http.Get(certDataSource)
		if err != nil {
			return err
		}
		defer src.Body.Close()

		license, cvsID, objects := cert.ParseInput(src.Body)

		fmt.Fprint(certData, license)
		if len(cvsID) > 0 {
			fmt.Fprint(certData, "CVS_ID "+cvsID+"\n")
		}

		cert.OutputTrustedCerts(certData, objects)

		return nil
	}, bo)

	if err != nil {
		log.Fatal(err)
	}

	saveToCache(strconv.FormatInt(time.Now().UTC().Unix(), 10), certData.Bytes())
	latestCertificates = certData.Bytes()
}

func (api cvedictClient) httpGet(key, url string, resChan chan<- response, errChan chan<- error) {
	var body string
	var errs []error
	var resp *http.Response
	f := func() (err error) {
		//  resp, body, errs = gorequest.New().SetDebug(config.Conf.Debug).Get(url).End()
		resp, body, errs = gorequest.New().Get(url).End()
		if 0 < len(errs) || resp == nil || resp.StatusCode != 200 {
			return fmt.Errorf("HTTP GET error: %v, url: %s, resp: %v", errs, url, resp)
		}
		return nil
	}
	notify := func(err error, t time.Duration) {
		log.Warnf("Failed to HTTP GET. retrying in %s seconds. err: %s", t, err)
	}
	err := backoff.RetryNotify(f, backoff.NewExponentialBackOff(), notify)
	if err != nil {
		errChan <- fmt.Errorf("HTTP Error %s", err)
	}
	cveDetail := cve.CveDetail{}
	if err := json.Unmarshal([]byte(body), &cveDetail); err != nil {
		errChan <- fmt.Errorf("Failed to Unmarshall. body: %s, err: %s", body, err)
	}
	resChan <- response{
		key,
		cveDetail,
	}
}

func (provider *Kv) watchKv(configurationChan chan<- types.ConfigMessage, prefix string, stop chan bool) {
	operation := func() error {
		events, err := provider.kvclient.WatchTree(provider.Prefix, make(chan struct{}) /* stop chan */)
		if err != nil {
			log.Errorf("Failed to WatchTree %s", err)
			return err
		}
		for {
			select {
			case <-stop:
				return nil
			case _, ok := <-events:
				if !ok {
					return errors.New("watchtree channel closed")
				}
				configuration := provider.loadConfig()
				if configuration != nil {
					configurationChan <- types.ConfigMessage{
						ProviderName:  string(provider.storeType),
						Configuration: configuration,
					}
				}
			}
		}
	}

	notify := func(err error, time time.Duration) {
		log.Errorf("KV connection error %+v, retrying in %s", err, time)
	}
	err := backoff.RetryNotify(operation, backoff.NewExponentialBackOff(), notify)
	if err != nil {
		log.Fatalf("Cannot connect to KV server %+v", err)
	}
}

func (as *APISelector) makeImageRequest(urlString string, bodyLines []string) (*apiSelectorImageResponse, error) {
	var responseBody []byte

	b := backoff.NewExponentialBackOff()
	b.MaxInterval = 10 * time.Second
	b.MaxElapsedTime = time.Minute

	err := backoff.Retry(func() (err error) {
		resp, err := http.Post(urlString, imageAPIRequestContentType,
			strings.NewReader(strings.Join(bodyLines, "\n")+"\n"))

		if err != nil {
			return err
		}
		defer resp.Body.Close()
		responseBody, err = ioutil.ReadAll(resp.Body)
		return
	}, b)

	if err != nil {
		return nil, err
	}

	imageResp := &apiSelectorImageResponse{
		Data: []*apiSelectorImageRef{},
	}

	err = json.Unmarshal(responseBody, imageResp)
	if err != nil {
		return nil, err
	}

	return imageResp, nil
}

func portUpdate(c *config.Config, ctx context.Context) error {
	ip, er := getIP(c.OpenVPN.Tun, c.Timeout.Duration, ctx)
	if er != nil || ctx.Err() != nil {
		return er
	}
	logger.Infof("New bind ip: (%s) %s", c.OpenVPN.Tun, ip)

	port, er := getPort(ip, c.PIA.User, c.PIA.Pass, c.PIA.ClientID, c.Timeout.Duration, ctx)
	if er != nil || ctx.Err() != nil {
		return er
	}

	logger.Infof("New peer port: %d", port)
	notify := func(e error, w time.Duration) {
		logger.Debugf("Failed to update transmission port: %v", er)
	}
	operation := func() error {
		select {
		default:
			return transmission.
				NewRawClient(c.Transmission.URL.String(), c.Transmission.User, c.Transmission.Pass).
				UpdatePort(port)
		case <-ctx.Done():
			return nil
		}
	}
	b := backoff.NewExponentialBackOff()
	b.MaxElapsedTime = c.Timeout.Duration
	return backoff.RetryNotify(operation, b, notify)
}

func (i *gceInstance) isPreempted(ctx gocontext.Context) (bool, error) {
	if !i.ic.Preemptible {
		return false, nil
	}

	listOpCall := i.provider.client.GlobalOperations.AggregatedList(i.provider.projectID).
		Filter(fmt.Sprintf("targetId eq %d", i.instance.Id))

	b := backoff.NewExponentialBackOff()
	b.InitialInterval = 1 * time.Second
	b.MaxElapsedTime = 1 * time.Minute

	var preempted bool
	err := backoff.Retry(func() error {
		i.provider.apiRateLimit(ctx)
		list, err := listOpCall.Do()
		if err != nil {
			return err
		}

		for _, item := range list.Items {
			for _, op := range item.Operations {
				if op.Kind == "compute#operation" && op.OperationType == "compute.instances.preempted" {
					preempted = true
					return nil
				}
			}
		}

		return nil
	}, b)

	return preempted, err
}

func (c *consulCoordinator) Start(addr net.Addr, errCh chan error) error {
	if addr == nil {
		addr = &net.TCPAddr{}
	}
	c.addr = addr
	session := c.client.Session()
	// set session to delete our keys on invalidation
	sessionOptions := &api.SessionEntry{
		Behavior:  api.SessionBehaviorDelete,
		LockDelay: c.config.LockDelay,
		TTL:       c.config.TTL,
	}
	var sessionID string
	var err error
	err = backoff.RetryNotify(func() error {
		sessionID, _, err = session.Create(sessionOptions, nil)
		return err
	}, backoff.NewExponentialBackOff(), func(err error, t time.Duration) {
		log.Println("Cannot create session, retrying in", t, ". Error:", err)
	})
	if err != nil {
		return fmt.Errorf("failed to create consul session: %v", err)
	}

	// set up a long-running goroutine for renewing the session
	c.sessionRenew = make(chan struct{})
	c.sessionID = sessionID
	log.Println("[coordinator] Coordinator ready")

	go func() {
		errCh <- session.RenewPeriodic("5s", sessionID, nil, c.sessionRenew)
	}()
	return nil
}

func sshConnect(c conf.ServerInfo) (client *ssh.Client, err error) {
	if client = tryAgentConnect(c); client != nil {
		return client, nil
	}

	var auths = []ssh.AuthMethod{}
	if auths, err = addKeyAuth(auths, c.KeyPath, c.KeyPassword); err != nil {
		return nil, err
	}

	// http://blog.ralch.com/tutorial/golang-ssh-connection/
	config := &ssh.ClientConfig{
		User: c.User,
		Auth: auths,
	}

	notifyFunc := func(e error, t time.Duration) {
		logger := getSSHLogger()
		logger.Debugf("Failed to Dial to %s, err: %s, Retrying in %s...",
			c.ServerName, e, t)
	}
	err = backoff.RetryNotify(func() error {
		if client, err = ssh.Dial("tcp", c.Host+":"+c.Port, config); err != nil {
			return err
		}
		return nil
	}, backoff.NewExponentialBackOff(), notifyFunc)

	return
}

func watch(src string) error {
	loc, err := storage.ParseLocation(src)
	if err != nil {
		return err
	}

	watcher, err := sync.Watch(loc)
	if err != nil {
		return err
	}
	defer watcher.Stop()

	// TODO: a better approach here would be to use a channel to retry on,
	//       then if you jacked up the config, it would pick up the change
	//       in the middle of all the retries. As it stands now it would take a
	//       minute to fix itself.
	eb := backoff.NewExponentialBackOff()
	eb.MaxElapsedTime = time.Minute

	for range watcher.C {
		log.Println("[watch] new version")
		backoff.Retry(func() error {
			err := apply(src)
			if err != nil {
				log.Printf("[watch] error installing: %v\n", err)
			}
			return err
		}, eb)
	}

	return nil
}

func main() {
	flag.Parse()
	args := flag.Args()
	if len(args) == 0 {
		usage()
	}

	var b []byte
	operation := func() error {
		var err error
		b, err = exec.Command(flag.Arg(0), args[1:]...).Output()
		if err != nil {
			log.Printf("err: %s", err)
		}
		return err
	}

	bf := backoff.NewExponentialBackOff()
	second := func(i int) time.Duration {
		return time.Duration(i) * time.Second
	}

	bf.MaxElapsedTime = second(*flagMaxElapsedTime)
	bf.MaxInterval = second(*flagMaxInterval)
	bf.InitialInterval = second(*flagInitialInterval)

	err := backoff.Retry(operation, bf)
	if err != nil {
		fmt.Fprintf(os.Stderr, "operation failed: %s\n", err)
		os.Exit(1)
	}

	fmt.Fprint(os.Stdout, string(b))
	os.Exit(0)
}

// NewClient creates a new tunnel that is established between the serverAddr
// and localAddr. It exits if it can't create a new control connection to the
// server. If localAddr is empty client will always try to proxy to a local
// port.
func NewClient(cfg *ClientConfig) (*Client, error) {
	yamuxConfig := yamux.DefaultConfig()
	if cfg.YamuxConfig != nil {
		yamuxConfig = cfg.YamuxConfig
	}

	log := newLogger("tunnel-client", cfg.Debug)
	if cfg.Log != nil {
		log = cfg.Log
	}

	if err := cfg.verify(); err != nil {
		return nil, err
	}

	forever := backoff.NewExponentialBackOff()
	forever.MaxElapsedTime = 365 * 24 * time.Hour // 1 year

	client := &Client{
		config:        cfg,
		log:           log,
		yamuxConfig:   yamuxConfig,
		redialBackoff: forever,
		startNotify:   make(chan bool, 1),
	}

	return client, nil
}

// NewBreakerWithOptions creates a base breaker with a specified backoff, clock and TripFunc
func NewBreakerWithOptions(options *Options) *Breaker {
	if options == nil {
		options = &Options{}
	}

	if options.Clock == nil {
		options.Clock = clock.New()
	}

	if options.BackOff == nil {
		b := backoff.NewExponentialBackOff()
		b.InitialInterval = defaultInitialBackOffInterval
		b.Clock = options.Clock
		b.Reset()
		options.BackOff = b
	}

	return &Breaker{
		BackOff:     options.BackOff,
		Clock:       options.Clock,
		ShouldTrip:  options.ShouldTrip,
		nextBackOff: options.BackOff.NextBackOff(),
		counts:      newWindow(DefaultWindowTime, DefaultWindowBuckets),
	}
}

func (s *stepGenerateScript) Run(state multistep.StateBag) multistep.StepAction {
	buildJob := state.Get("buildJob").(Job)
	ctx := state.Get("ctx").(gocontext.Context)

	b := backoff.NewExponentialBackOff()
	b.MaxInterval = 10 * time.Second
	b.MaxElapsedTime = time.Minute

	var script []byte
	err := backoff.Retry(func() (err error) {
		script, err = s.generator.Generate(ctx, buildJob.RawPayload())
		return
	}, b)

	if err != nil {
		context.LoggerFromContext(ctx).WithField("err", err).Error("couldn't generate build script, erroring job")
		err := buildJob.Error(ctx, "An error occurred while generating the build script.")
		if err != nil {
			context.LoggerFromContext(ctx).WithField("err", err).Error("couldn't requeue job")
		}

		return multistep.ActionHalt
	}

	context.LoggerFromContext(ctx).Info("generated script")

	state.Put("script", script)

	return multistep.ActionContinue
}

func newForeverBackoff() *expBackoff {
	eb := &expBackoff{
		bk: backoff.NewExponentialBackOff(),
	}
	eb.bk.MaxElapsedTime = 0 // never stops
	return eb
}

func (r *Route53) wait(timeout time.Duration, comment, id string) error {
	if timeout == 0 {
		return nil
	}

	retry := backoff.NewExponentialBackOff()
	retry.MaxElapsedTime = timeout
	retry.Reset()

	change := &route53.GetChangeInput{
		Id: aws.String(id),
	}
	for {
		resp, err := r.GetChange(change)
		var status string
		if err == nil && resp.ChangeInfo != nil {
			status = strings.ToLower(aws.StringValue(resp.ChangeInfo.Status))
		}

		r.opts.Log.Debug("%s: checking %s status=%s, err=%v", comment, id, status, err)

		if status == "insync" {
			return nil
		}

		next := retry.NextBackOff()
		if next == backoff.Stop {
			return fmt.Errorf("waiting for %s status to be insync timed out", id)
		}
		time.Sleep(next)
	}
}

// listenForNodeChanges listens for changes to node status using change feeds.
// This function will block until the query fails
func (c *Cluster) listenForNodeChanges() error {
	// Start listening to changes from a random active node
	node, hpr, err := c.GetNextNode()
	if err != nil {
		return err
	}

	q, err := newQuery(
		DB("rethinkdb").Table("server_status").Changes(),
		map[string]interface{}{},
		c.opts,
	)
	if err != nil {
		return fmt.Errorf("Error building query: %s", err)
	}

	cursor, err := node.Query(q)
	if err != nil {
		hpr.Mark(err)
		return err
	}

	// Keep reading node status updates from changefeed
	var result struct {
		NewVal nodeStatus `gorethink:"new_val"`
		OldVal nodeStatus `gorethink:"old_val"`
	}
	for cursor.Next(&result) {
		addr := fmt.Sprintf("%s:%d", result.NewVal.Network.Hostname, result.NewVal.Network.ReqlPort)
		addr = strings.ToLower(addr)

		switch result.NewVal.Status {
		case "connected":
			// Connect to node using exponential backoff (give up after waiting 5s)
			// to give the node time to start-up.
			b := backoff.NewExponentialBackOff()
			b.MaxElapsedTime = time.Second * 5

			backoff.Retry(func() error {
				node, err := c.connectNodeWithStatus(result.NewVal)
				if err == nil {
					if !c.nodeExists(node) {
						c.addNode(node)

						Log.WithFields(logrus.Fields{
							"id":   node.ID,
							"host": node.Host.String(),
						}).Debug("Connected to node")
					}
				}

				return err
			}, b)
		}
	}

	err = cursor.Err()
	hpr.Mark(err)
	return err
}

// doReconnect attempts a reconnection.  It assumes that reconnectChan
// and reconnectErrPtr are the same ones in c, but are passed in to
// avoid having to take the mutex at the beginning of the method.
func (c *Connection) doReconnect(ctx context.Context,
	reconnectChan chan struct{}, reconnectErrPtr *error) {
	// retry w/exponential backoff
	backoff.RetryNotify(func() error {
		// try to connect
		err := c.connect(ctx)
		select {
		case <-ctx.Done():
			// context was canceled by Shutdown() or a user action
			*reconnectErrPtr = ctx.Err()
			// short-circuit Retry
			return nil
		default:
		}
		if dontRetryOnConnect(err) {
			// A fatal error happened.
			*reconnectErrPtr = err
			// short-circuit Retry
			return nil
		}
		return err
	}, backoff.NewExponentialBackOff(),
		// give the caller a chance to log any other error or adjust state
		c.handler.OnConnectError)

	// close the reconnect channel to signal we're connected.
	c.mutex.Lock()
	defer c.mutex.Unlock()
	close(reconnectChan)
	c.reconnectChan = nil
	c.cancelFunc = nil
	c.reconnectErrPtr = nil
}

func backOffTick(initial, max time.Duration) *backoff.Ticker {
	b := backoff.NewExponentialBackOff()
	b.InitialInterval = initial
	b.MaxInterval = max
	b.MaxElapsedTime = 0 // infinite
	return backoff.NewTicker(b)
}