Something went wrong. Try again.
leaderless gossip-based service discovery w/ consul compatibility
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415package api
import ( "bytes" "context" "encoding/json" "fmt" "io" "io/ioutil" "net" "net/http" "net/url" "runtime" "strconv" "strings" "sync" "time")
// Config is used to configure the creation of a clienttype Config struct { // Address is the address of the Consul server Address string
// Scheme is the URI scheme for the Consul server Scheme string
// Transport is the Transport to use for the http client. Transport *http.Transport
// HttpClient is the client to use. Default will be // used if not provided. HttpClient *http.Client
// WaitTime limits how long a Watch will block. If not provided, // the agent default values will be used. WaitTime time.Duration}
// DefaultConfig returns a default configuration for the client. By default this// will pool and reuse idle connections to cascade.func DefaultConfig() *Config { transport := &http.Transport{ Proxy: http.ProxyFromEnvironment, DialContext: (&net.Dialer{ Timeout: 30 * time.Second, KeepAlive: 30 * time.Second, DualStack: true, }).DialContext, MaxIdleConns: 100, IdleConnTimeout: 90 * time.Second, ExpectContinueTimeout: 1 * time.Second, ForceAttemptHTTP2: true, MaxIdleConnsPerHost: runtime.GOMAXPROCS(0) + 1, }
config := &Config{ Address: "127.0.0.1:8500", Scheme: "http", Transport: transport, }
return config}
// Client provides a client to the cascade APItype Client struct { modifyLock sync.RWMutex headers http.Header
config Config}
// Headers gets the current set of headers used for requests. This returns a// copy; to modify it call AddHeader or SetHeaders.func (c *Client) Headers() http.Header { c.modifyLock.RLock() defer c.modifyLock.RUnlock()
if c.headers == nil { return nil }
ret := make(http.Header) for k, v := range c.headers { for _, val := range v { ret[k] = append(ret[k], val) } }
return ret}
// NewClient returns a new clientfunc NewClient(config *Config) (*Client, error) { // bootstrap the config defConfig := DefaultConfig()
if config.Address == "" { config.Address = defConfig.Address }
if config.Scheme == "" { config.Scheme = defConfig.Scheme }
if config.Transport == nil { config.Transport = defConfig.Transport }
if config.HttpClient == nil { config.HttpClient = NewHttpClient(config.Transport) }
parts := strings.SplitN(config.Address, "://", 2) if len(parts) == 2 { switch parts[0] { case "http": config.Scheme = "http" // TODO: unix socket support? do i care? no??? default: return nil, fmt.Errorf("Unknown protocol scheme: %s", parts[0]) } config.Address = parts[1] }
return &Client{config: *config, headers: make(http.Header)}, nil}
// NewHttpClient returns an http client configured with the given Transportfunc NewHttpClient(transport *http.Transport) *http.Client { client := &http.Client{ Transport: transport, } return client}
// request is used to help build up a requesttype request struct { config *Config method string url *url.URL params url.Values body io.Reader header http.Header obj interface{} ctx context.Context}
// QueryMeta is used to return meta data about a querytype QueryMeta struct { // LastIndex. This can be used as a WaitIndex to perform // a blocking query LastIndex uint64
// LastContentHash. This can be used as a WaitHash to perform a blocking query // for endpoints that support hash-based blocking. Endpoints that do not // support it will return an empty hash. LastContentHash string
// How long did the request take RequestTime time.Duration
// Is address translation enabled for HTTP responses on this agent AddressTranslationEnabled bool}
// QueryOptions are used to parameterize a querytype QueryOptions struct { // WaitIndex is used to enable a blocking query. Waits // until the timeout or the next index is reached WaitIndex uint64
// WaitHash is used by some endpoints instead of WaitIndex to perform blocking // on state based on a hash of the response rather than a monotonic index. // This is required when the state being blocked on is not stored in Raft, for // example agent-local proxy configuration. WaitHash string
// WaitTime is used to bound the duration of a wait. // Defaults to that of the Config, but can be overridden. WaitTime time.Duration
// Near is used to provide a node name that will sort the results // in ascending order based on the estimated round trip time from // that node. Setting this to "_agent" will use the agent's node // for the sort. Near string
// NodeMeta is used to filter results by nodes with the given // metadata key/value pairs. Currently, only one key/value pair can // be provided for filtering. NodeMeta map[string]string
// ctx is an optional context pass through to the underlying HTTP // request layer. Use Context() and WithContext() to manage this. ctx context.Context
// Filter requests filtering data prior to it being returned. The string // is a go-bexpr compatible expression. Filter string}
// setQueryOptions is used to annotate the request with// additional query optionsfunc (r *request) setQueryOptions(q *QueryOptions) { if q == nil { return } if q.WaitIndex != 0 { r.params.Set("index", strconv.FormatUint(q.WaitIndex, 10)) } if q.WaitTime != 0 { r.params.Set("wait", durToMsec(q.WaitTime)) } if q.WaitHash != "" { r.params.Set("hash", q.WaitHash) } if q.Near != "" { r.params.Set("near", q.Near) } if q.Filter != "" { r.params.Set("filter", q.Filter) } if len(q.NodeMeta) > 0 { for key, value := range q.NodeMeta { r.params.Add("node-meta", key+":"+value) } }
r.ctx = q.ctx}
// durToMsec converts a duration to a millisecond specified string. If the// user selected a positive value that rounds to 0 ms, then we will use 1 ms// so they get a short delay, otherwise Consul will translate the 0 ms into// a huge default delay.func durToMsec(dur time.Duration) string { ms := dur / time.Millisecond if dur > 0 && ms == 0 { ms = 1 } return fmt.Sprintf("%dms", ms)}
// serverError is a string we look for to detect 500 errors.const serverError = "Unexpected response code: 500"
// IsRetryableError returns true for 500 errors from the Consul servers, and// network connection errors. These are usually retryable at a later time.// This applies to reads but NOT to writes. This may return true for errors// on writes that may have still gone through, so do not use this to retry// any write operations.func IsRetryableError(err error) bool { if err == nil { return false }
if _, ok := err.(net.Error); ok { return true }
// TODO (slackpad) - Make a real error type here instead of using // a string check. return strings.Contains(err.Error(), serverError)}
// toHTTP converts the request to an HTTP requestfunc (r *request) toHTTP() (*http.Request, error) { // Encode the query parameters r.url.RawQuery = r.params.Encode()
// Check if we should encode the body if r.body == nil && r.obj != nil { b, err := encodeBody(r.obj) if err != nil { return nil, err } r.body = b }
// Create the HTTP request req, err := http.NewRequest(r.method, r.url.RequestURI(), r.body) if err != nil { return nil, err }
req.URL.Host = r.url.Host req.URL.Scheme = r.url.Scheme req.Host = r.url.Host req.Header = r.header
// Content-Type must always be set when a body is present // See https://github.com/hashicorp/consul/issues/10011 if req.Body != nil && req.Header.Get("Content-Type") == "" { req.Header.Set("Content-Type", "application/json") }
if r.ctx != nil { return req.WithContext(r.ctx), nil }
return req, nil}
// newRequest is used to create a new requestfunc (c *Client) newRequest(method, path string) *request { r := &request{ config: &c.config, method: method, url: &url.URL{ Scheme: c.config.Scheme, Host: c.config.Address, Path: path, }, params: make(map[string][]string), header: c.Headers(), }
if c.config.WaitTime != 0 { r.params.Set("wait", durToMsec(r.config.WaitTime)) }
return r}
// doRequest runs a request with our clientfunc (c *Client) doRequest(r *request) (time.Duration, *http.Response, error) { req, err := r.toHTTP() if err != nil { return 0, nil, err } start := time.Now() resp, err := c.config.HttpClient.Do(req) diff := time.Since(start) return diff, resp, err}
// Query is used to do a GET request against an endpoint// and deserialize the response into an interface using// standard cascade conventions.func (c *Client) query(endpoint string, out interface{}, q *QueryOptions) (*QueryMeta, error) { r := c.newRequest("GET", endpoint) r.setQueryOptions(q) rtt, resp, err := c.doRequest(r) if err != nil { return nil, err } defer closeResponseBody(resp)
qm := &QueryMeta{} qm.RequestTime = rtt
if err := decodeBody(resp, out); err != nil { return nil, err } return qm, nil}
// decodeBody is used to JSON decode a bodyfunc decodeBody(resp *http.Response, out interface{}) error { dec := json.NewDecoder(resp.Body) return dec.Decode(out)}
// encodeBody is used to encode a request bodyfunc encodeBody(obj interface{}) (io.Reader, error) { buf := bytes.NewBuffer(nil) enc := json.NewEncoder(buf) if err := enc.Encode(obj); err != nil { return nil, err } return buf, nil}
// requireOK is used to wrap doRequest and check for a 200func requireOK(d time.Duration, resp *http.Response, e error) (time.Duration, *http.Response, error) { if e != nil { if resp != nil { closeResponseBody(resp) } return d, nil, e } if resp.StatusCode != 200 { return d, nil, generateUnexpectedResponseCodeError(resp) } return d, resp, nil}
// closeResponseBody reads resp.Body until EOF, and then closes it. The read// is necessary to ensure that the http.Client's underlying RoundTripper is able// to re-use the TCP connection. See godoc on net/http.Client.Do.func closeResponseBody(resp *http.Response) error { _, _ = io.Copy(ioutil.Discard, resp.Body) return resp.Body.Close()}
func (req *request) filterQuery(filter string) { if filter == "" { return }
req.params.Set("filter", filter)}
// generateUnexpectedResponseCodeError consumes the rest of the body, closes// the body stream and generates an error indicating the status code was// unexpected.func generateUnexpectedResponseCodeError(resp *http.Response) error { var buf bytes.Buffer io.Copy(&buf, resp.Body) closeResponseBody(resp) return fmt.Errorf("Unexpected response code: %d (%s)", resp.StatusCode, buf.Bytes())}