func (c *Client) Shutdown(termTimeout time.Duration) error {
	term, ticker := time.After(termTimeout), time.NewTicker(100*time.Millisecond)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			if atomic.LoadUint64(&c.ConnectionsCounter) > 0 {
				continue
			}
		case <-term:
		}
		break
	}
	active := atomic.LoadUint64(&c.ConnectionsCounter)
	if active == 0 {
		return nil
	}
	return fmt.Errorf("%d active connections still exist after waiting for %v", active, termTimeout)
}
Shutdown waits up to a given amount of time for all active connections to
close. Returns an error if there are still active connections after waiting
for the whole length of the timeout.
package proxy
import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/GoogleCloudPlatform/cloudsql-proxy/logging"
	"github.com/GoogleCloudPlatform/cloudsql-proxy/proxy/util"
	"golang.org/x/net/proxy"
	"golang.org/x/time/rate"
)
const (
	DefaultRefreshCfgThrottle = time.Minute
	IAMLoginRefreshThrottle = 30 * time.Second
	keepAlivePeriod         = time.Minute
	DefaultRefreshCfgBuffer = 5 * time.Minute
	IAMLoginRefreshCfgBuffer = 55 * time.Second
)
var (
	errNotCached = errors.New("instance was not found in cache")
)
type Conn struct {
	Instance string
	Conn     net.Conn
}
type CertSource interface {
	Local(instance string) (tls.Certificate, error)
	Remote(instance string) (cert *x509.Certificate, addr, name, version string, err error)
}
type Client struct {
	ConnectionsCounter uint64
	MaxConnections uint64
	Port int
	Certs CertSource
	Conns *ConnSet
	ContextDialer func(ctx context.Context, net, addr string) (net.Conn, error)
	Dialer func(net, addr string) (net.Conn, error)
	cfgCache map[string]cacheEntry
	cacheL   sync.RWMutex
	limiters map[string]*rate.Limiter
	refreshCfgL sync.Mutex
	RefreshCfgThrottle time.Duration
	RefreshCfgBuffer time.Duration
}
type cacheEntry struct {
	lastRefreshed time.Time
	err     error
	addr    string
	version string
	cfg     *tls.Config
	done chan struct{}
}
func (c *Client) Run(connSrc <-chan Conn) {
	c.RunContext(context.Background(), connSrc)
}
func (c *Client) run(ctx context.Context, connSrc <-chan Conn) {
	for {
		select {
		case conn, ok := <-connSrc:
			if !ok {
				return
			}
			go c.handleConn(ctx, conn)
		case <-ctx.Done():
			return
		}
	}
}
func (c *Client) RunContext(ctx context.Context, connSrc <-chan Conn) {
	c.run(ctx, connSrc)
	if err := c.Conns.Close(); err != nil {
		logging.Errorf("closing client had error: %v", err)
	}
}
func (c *Client) handleConn(ctx context.Context, conn Conn) {
	active := atomic.AddUint64(&c.ConnectionsCounter, 1)
	defer atomic.AddUint64(&c.ConnectionsCounter, ^uint64(0))
	if c.MaxConnections > 0 && active > c.MaxConnections {
		logging.Errorf("too many open connections (max %d)", c.MaxConnections)
		conn.Conn.Close()
		return
	}
	server, err := c.DialContext(ctx, conn.Instance)
	if err != nil {
		logging.Errorf("couldn't connect to %q: %v", conn.Instance, err)
		conn.Conn.Close()
		return
	}
	c.Conns.Add(conn.Instance, conn.Conn)
	copyThenClose(server, conn.Conn, conn.Instance, "local connection on "+conn.Conn.LocalAddr().String())
	if err := c.Conns.Remove(conn.Instance, conn.Conn); err != nil {
		logging.Errorf("%s", err)
	}
}
func (c *Client) refreshCfg(instance string) (addr string, cfg *tls.Config, version string, err error) {
	c.refreshCfgL.Lock()
	defer c.refreshCfgL.Unlock()
	logging.Verbosef("refreshing ephemeral certificate for instance %s", instance)
	mycert, err := c.Certs.Local(instance)
	if err != nil {
		return "", nil, "", err
	}
	scert, addr, name, version, err := c.Certs.Remote(instance)
	if err != nil {
		return "", nil, "", err
	}
	certs := x509.NewCertPool()
	certs.AddCert(scert)
	cfg = &tls.Config{
		ServerName:   name,
		Certificates: []tls.Certificate{mycert},
		RootCAs:      certs,
		InsecureSkipVerify:    true,
		VerifyPeerCertificate: genVerifyPeerCertificateFunc(name, certs),
		MinVersion:            tls.VersionTLS13,
	}
	return fmt.Sprintf("%s:%d", addr, c.Port), cfg, version, nil
}
func (c *Client) refreshCertAfter(instance string, timeToRefresh time.Duration) {
	<-time.After(timeToRefresh)
	logging.Verbosef("ephemeral certificate for instance %s will expire soon, refreshing now.", instance)
	if _, _, _, err := c.cachedCfg(context.Background(), instance); err != nil {
		logging.Errorf("failed to refresh the ephemeral certificate for %s before expiring: %v", instance, err)
	}
}
func genVerifyPeerCertificateFunc(instanceName string, pool *x509.CertPool) func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
	return func(rawCerts [][]byte, verifiedChains [][]*x509.Certificate) error {
		if len(rawCerts) == 0 {
			return fmt.Errorf("no certificate to verify")
		}
		cert, err := x509.ParseCertificate(rawCerts[0])
		if err != nil {
			return fmt.Errorf("x509.ParseCertificate(rawCerts[0]) returned error: %v", err)
		}
		opts := x509.VerifyOptions{Roots: pool}
		if _, err = cert.Verify(opts); err != nil {
			return err
		}
		if cert.Subject.CommonName != instanceName {
			return fmt.Errorf("certificate had CN %q, expected %q", cert.Subject.CommonName, instanceName)
		}
		return nil
	}
}
func isExpired(cfg *tls.Config) bool {
	if cfg == nil {
		return true
	}
	return time.Now().After(cfg.Certificates[0].Leaf.NotAfter)
}
func (c *Client) startRefresh(instance string, refreshCfgBuffer time.Duration) chan struct{} {
	done := make(chan struct{})
	go func() {
		defer close(done)
		addr, cfg, ver, err := c.refreshCfg(instance)
		c.cacheL.Lock()
		old := c.cfgCache[instance]
		if err != nil && !isExpired(old.cfg) {
			logging.Errorf("failed to refresh the ephemeral certificate for %s, returning previous cert instead: %v", instance, err)
			addr, cfg, ver, err = old.addr, old.cfg, old.version, old.err
		}
		e := cacheEntry{
			lastRefreshed: time.Now(),
			err:           err,
			addr:          addr,
			version:       ver,
			cfg:           cfg,
			done:          done,
		}
		c.cfgCache[instance] = e
		c.cacheL.Unlock()
		if !isValid(e) {
			logging.Errorf("failed to refresh the ephemeral certificate for %v: %v", instance, err)
			return
		}
		certExpiration := cfg.Certificates[0].Leaf.NotAfter
		now := time.Now()
		timeToRefresh := certExpiration.Sub(now) - refreshCfgBuffer
		if timeToRefresh <= 0 {
			timeToRefresh = certExpiration.Sub(now) - (5 * time.Second)
			logging.Errorf("new ephemeral certificate expires sooner than expected (adjusting refresh time to compensate): current time: %v, certificate expires: %v", now, certExpiration)
		}
		logging.Infof("Scheduling refresh of ephemeral certificate in %s", timeToRefresh)
		go c.refreshCertAfter(instance, timeToRefresh)
	}()
	return done
}
func isValid(c cacheEntry) bool {
	return c.err == nil && c.cfg != nil
}
type InvalidError struct {
	instance string
	err error
	hasTLS bool
}
func (e *InvalidError) Error() string {
	if e.hasTLS {
		return e.instance + ": " + e.err.Error()
	}
	return e.instance + ": missing TLS config, " + e.err.Error()
}
func (c *Client) InvalidInstances() []*InvalidError {
	c.cacheL.RLock()
	defer c.cacheL.RUnlock()
	var invalid []*InvalidError
	for instance, entry := range c.cfgCache {
		var refreshInProgress bool
		select {
		case <-entry.done:
		default:
			refreshInProgress = true
		}
		if !isValid(entry) && !refreshInProgress {
			invalid = append(invalid, &InvalidError{
				instance: instance,
				err:      entry.err,
				hasTLS:   entry.cfg != nil,
			})
		}
	}
	return invalid
}
func needsRefresh(e cacheEntry, refreshCfgBuffer time.Duration) bool {
	if e.done == nil {  
		return true
	}
	if !isValid(e) || e.cfg.Certificates[0].Leaf.NotAfter.Sub(time.Now()) <= refreshCfgBuffer {
		select {
		case <-e.done:  
			return true
		default:  
			return false
		}
	}
	return false
}
func (c *Client) cachedCfg(ctx context.Context, instance string) (string, *tls.Config, string, error) {
	c.cacheL.RLock()
	throttle := c.RefreshCfgThrottle
	if throttle == 0 {
		throttle = DefaultRefreshCfgThrottle
	}
	refreshCfgBuffer := c.RefreshCfgBuffer
	if refreshCfgBuffer == 0 {
		refreshCfgBuffer = DefaultRefreshCfgBuffer
	}
	e := c.cfgCache[instance]
	c.cacheL.RUnlock()
	if needsRefresh(e, refreshCfgBuffer) {
		c.cacheL.Lock()
		if c.cfgCache == nil {
			c.cfgCache = make(map[string]cacheEntry)
		}
		if c.limiters == nil {
			c.limiters = make(map[string]*rate.Limiter)
		}
		e = c.cfgCache[instance]
		limiter := c.limiters[instance]
		if limiter == nil {
			limiter = rate.NewLimiter(rate.Every(throttle), 1)
			c.limiters[instance] = limiter
		}
		if needsRefresh(e, refreshCfgBuffer) {
			if limiter.Allow() {
				e.done = c.startRefresh(instance, refreshCfgBuffer)
				e.lastRefreshed = time.Now()
				c.cfgCache[instance] = e
			} else {
				logging.Infof("refresh operation throttled for %s: reusing config from last refresh (%s ago)", instance, time.Since(e.lastRefreshed))
			}
		}
		c.cacheL.Unlock()
	}
	if !isValid(e) {
		select {
		case <-ctx.Done():
			return "", nil, "", ctx.Err()
		case <-e.done:
		}
		c.cacheL.RLock()
		e = c.cfgCache[instance]
		c.cacheL.RUnlock()
	}
	return e.addr, e.cfg, e.version, e.err
}
func (c *Client) DialContext(ctx context.Context, instance string) (net.Conn, error) {
	addr, cfg, _, err := c.cachedCfg(ctx, instance)
	if err != nil {
		return nil, err
	}
	return c.tryConnect(ctx, addr, instance, cfg)
}
func (c *Client) Dial(instance string) (net.Conn, error) {
	return c.DialContext(context.Background(), instance)
}
var ErrUnexpectedFailure = errors.New("ErrUnexpectedFailure")
func (c *Client) tryConnect(ctx context.Context, addr, instance string, cfg *tls.Config) (net.Conn, error) {
	if addr == "" {
		return nil, ErrUnexpectedFailure
	}
	dial := c.selectDialer()
	conn, err := dial(ctx, "tcp", addr)
	if err != nil {
		return nil, err
	}
	type setKeepAliver interface {
		SetKeepAlive(keepalive bool) error
		SetKeepAlivePeriod(d time.Duration) error
	}
	if s, ok := conn.(setKeepAliver); ok {
		if err := s.SetKeepAlive(true); err != nil {
			logging.Verbosef("Couldn't set KeepAlive to true: %v", err)
		} else if err := s.SetKeepAlivePeriod(keepAlivePeriod); err != nil {
			logging.Verbosef("Couldn't set KeepAlivePeriod to %v", keepAlivePeriod)
		}
	} else {
		logging.Verbosef("KeepAlive not supported: long-running tcp connections may be killed by the OS.")
	}
	return c.connectTLS(ctx, conn, instance, cfg)
}
func (c *Client) selectDialer() func(context.Context, string, string) (net.Conn, error) {
	if c.ContextDialer != nil {
		return c.ContextDialer
	}
	if c.Dialer != nil {
		return func(_ context.Context, net, addr string) (net.Conn, error) {
			return c.Dialer(net, addr)
		}
	}
	dialer := proxy.FromEnvironment()
	if ctxDialer, ok := dialer.(proxy.ContextDialer); ok {
		return ctxDialer.DialContext
	}
	return func(_ context.Context, net, addr string) (net.Conn, error) {
		return dialer.Dial(net, addr)
	}
}
func (c *Client) invalidateCfg(cfg *tls.Config, instance string) {
	c.cacheL.RLock()
	e := c.cfgCache[instance]
	c.cacheL.RUnlock()
	if e.cfg != cfg {
		return
	}
	c.cacheL.Lock()
	defer c.cacheL.Unlock()
	e = c.cfgCache[instance]
	if e.cfg != cfg {
		return
	}
	c.cfgCache[instance] = cacheEntry{
		done:          e.done,
		lastRefreshed: e.lastRefreshed,
	}
}
func NewConnSrc(instance string, l net.Listener) <-chan Conn {
	ch := make(chan Conn)
	go func() {
		for {
			start := time.Now()
			c, err := l.Accept()
			if err != nil {
				logging.Errorf("listener (%#v) had error: %v", l, err)
				if nerr, ok := err.(net.Error); ok && nerr.Temporary() {
					d := 10*time.Millisecond - time.Since(start)
					if d > 0 {
						time.Sleep(d)
					}
					continue
				}
				l.Close()
				close(ch)
				return
			}
			ch <- Conn{instance, c}
		}
	}()
	return ch
}
func (c *Client) InstanceVersion(instance string) (string, error) {
	return c.InstanceVersionContext(context.Background(), instance)
}
func (c *Client) InstanceVersionContext(ctx context.Context, instance string) (string, error) {
	_, _, version, err := c.cachedCfg(ctx, instance)
	if err != nil {
		return "", nil
	}
	return version, nil
}
func ParseInstanceConnectionName(instance string) (string, string, string, []string, error) {
	args := strings.Split(instance, "=")
	if len(args) > 2 {
		return "", "", "", nil, fmt.Errorf("invalid instance argument: must be either form - `<instance_connection_string>` or `<instance_connection_string>=<options>`; invalid arg was %q", instance)
	}
	proj, region, name := util.SplitName(args[0])
	if proj == "" || region == "" || name == "" {
		return "", "", "", nil, fmt.Errorf("invalid instance connection string: must be in the form `project:region:instance-name`; invalid name was %q", args[0])
	}
	return proj, region, name, args, nil
}
func (c *Client) GetInstances() []string {
	var insts []string
	c.cacheL.Lock()
	cfgCache := c.cfgCache
	c.cacheL.Unlock()
	for i := range cfgCache {
		insts = append(insts, i)
	}
	return insts
}
func (c *Client) AvailableConn() bool {
	return c.MaxConnections == 0 || atomic.LoadUint64(&c.ConnectionsCounter) < c.MaxConnections
}