mirror of
https://github.com/42wim/matterbridge.git
synced 2024-12-30 07:02:37 +01:00
324 lines
7.6 KiB
Go
324 lines
7.6 KiB
Go
package consulapi
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"time"
|
|
)
|
|
|
|
// QueryOptions are used to parameterize a query
|
|
type QueryOptions struct {
|
|
// Providing a datacenter overwrites the DC provided
|
|
// by the Config
|
|
Datacenter string
|
|
|
|
// AllowStale allows any Consul server (non-leader) to service
|
|
// a read. This allows for lower latency and higher throughput
|
|
AllowStale bool
|
|
|
|
// RequireConsistent forces the read to be fully consistent.
|
|
// This is more expensive but prevents ever performing a stale
|
|
// read.
|
|
RequireConsistent bool
|
|
|
|
// WaitIndex is used to enable a blocking query. Waits
|
|
// until the timeout or the next index is reached
|
|
WaitIndex uint64
|
|
|
|
// WaitTime is used to bound the duration of a wait.
|
|
// Defaults to that of the Config, but can be overriden.
|
|
WaitTime time.Duration
|
|
|
|
// Token is used to provide a per-request ACL token
|
|
// which overrides the agent's default token.
|
|
Token string
|
|
}
|
|
|
|
// WriteOptions are used to parameterize a write
|
|
type WriteOptions struct {
|
|
// Providing a datacenter overwrites the DC provided
|
|
// by the Config
|
|
Datacenter string
|
|
|
|
// Token is used to provide a per-request ACL token
|
|
// which overrides the agent's default token.
|
|
Token string
|
|
}
|
|
|
|
// QueryMeta is used to return meta data about a query
|
|
type QueryMeta struct {
|
|
// LastIndex. This can be used as a WaitIndex to perform
|
|
// a blocking query
|
|
LastIndex uint64
|
|
|
|
// Time of last contact from the leader for the
|
|
// server servicing the request
|
|
LastContact time.Duration
|
|
|
|
// Is there a known leader
|
|
KnownLeader bool
|
|
|
|
// How long did the request take
|
|
RequestTime time.Duration
|
|
}
|
|
|
|
// WriteMeta is used to return meta data about a write
|
|
type WriteMeta struct {
|
|
// How long did the request take
|
|
RequestTime time.Duration
|
|
}
|
|
|
|
// HttpBasicAuth is used to authenticate http client with HTTP Basic Authentication
|
|
type HttpBasicAuth struct {
|
|
// Username to use for HTTP Basic Authentication
|
|
Username string
|
|
|
|
// Password to use for HTTP Basic Authentication
|
|
Password string
|
|
}
|
|
|
|
// Config is used to configure the creation of a client
|
|
type Config struct {
|
|
// Address is the address of the Consul server
|
|
Address string
|
|
|
|
// Scheme is the URI scheme for the Consul server
|
|
Scheme string
|
|
|
|
// Datacenter to use. If not provided, the default agent datacenter is used.
|
|
Datacenter string
|
|
|
|
// HttpClient is the client to use. Default will be
|
|
// used if not provided.
|
|
HttpClient *http.Client
|
|
|
|
// HttpAuth is the auth info to use for http access.
|
|
HttpAuth *HttpBasicAuth
|
|
|
|
// WaitTime limits how long a Watch will block. If not provided,
|
|
// the agent default values will be used.
|
|
WaitTime time.Duration
|
|
|
|
// Token is used to provide a per-request ACL token
|
|
// which overrides the agent's default token.
|
|
Token string
|
|
}
|
|
|
|
// DefaultConfig returns a default configuration for the client
|
|
func DefaultConfig() *Config {
|
|
return &Config{
|
|
Address: "127.0.0.1:8500",
|
|
Scheme: "http",
|
|
HttpClient: http.DefaultClient,
|
|
}
|
|
}
|
|
|
|
// Client provides a client to the Consul API
|
|
type Client struct {
|
|
config Config
|
|
}
|
|
|
|
// NewClient returns a new client
|
|
func NewClient(config *Config) (*Client, error) {
|
|
// bootstrap the config
|
|
defConfig := DefaultConfig()
|
|
|
|
if len(config.Address) == 0 {
|
|
config.Address = defConfig.Address
|
|
}
|
|
|
|
if len(config.Scheme) == 0 {
|
|
config.Scheme = defConfig.Scheme
|
|
}
|
|
|
|
if config.HttpClient == nil {
|
|
config.HttpClient = defConfig.HttpClient
|
|
}
|
|
|
|
client := &Client{
|
|
config: *config,
|
|
}
|
|
return client, nil
|
|
}
|
|
|
|
// request is used to help build up a request
|
|
type request struct {
|
|
config *Config
|
|
method string
|
|
url *url.URL
|
|
params url.Values
|
|
body io.Reader
|
|
obj interface{}
|
|
}
|
|
|
|
// setQueryOptions is used to annotate the request with
|
|
// additional query options
|
|
func (r *request) setQueryOptions(q *QueryOptions) {
|
|
if q == nil {
|
|
return
|
|
}
|
|
if q.Datacenter != "" {
|
|
r.params.Set("dc", q.Datacenter)
|
|
}
|
|
if q.AllowStale {
|
|
r.params.Set("stale", "")
|
|
}
|
|
if q.RequireConsistent {
|
|
r.params.Set("consistent", "")
|
|
}
|
|
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.Token != "" {
|
|
r.params.Set("token", q.Token)
|
|
}
|
|
}
|
|
|
|
// durToMsec converts a duration to a millisecond specified string
|
|
func durToMsec(dur time.Duration) string {
|
|
return fmt.Sprintf("%dms", dur/time.Millisecond)
|
|
}
|
|
|
|
// setWriteOptions is used to annotate the request with
|
|
// additional write options
|
|
func (r *request) setWriteOptions(q *WriteOptions) {
|
|
if q == nil {
|
|
return
|
|
}
|
|
if q.Datacenter != "" {
|
|
r.params.Set("dc", q.Datacenter)
|
|
}
|
|
if q.Token != "" {
|
|
r.params.Set("token", q.Token)
|
|
}
|
|
}
|
|
|
|
// toHTTP converts the request to an HTTP request
|
|
func (r *request) toHTTP() (*http.Request, error) {
|
|
// Encode the query parameters
|
|
r.url.RawQuery = r.params.Encode()
|
|
|
|
// Get the url sring
|
|
urlRaw := r.url.String()
|
|
|
|
// Check if we should encode the body
|
|
if r.body == nil && r.obj != nil {
|
|
if b, err := encodeBody(r.obj); err != nil {
|
|
return nil, err
|
|
} else {
|
|
r.body = b
|
|
}
|
|
}
|
|
|
|
// Create the HTTP request
|
|
req, err := http.NewRequest(r.method, urlRaw, r.body)
|
|
|
|
// Setup auth
|
|
if err == nil && r.config.HttpAuth != nil {
|
|
req.SetBasicAuth(r.config.HttpAuth.Username, r.config.HttpAuth.Password)
|
|
}
|
|
|
|
return req, err
|
|
}
|
|
|
|
// newRequest is used to create a new request
|
|
func (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),
|
|
}
|
|
if c.config.Datacenter != "" {
|
|
r.params.Set("dc", c.config.Datacenter)
|
|
}
|
|
if c.config.WaitTime != 0 {
|
|
r.params.Set("wait", durToMsec(r.config.WaitTime))
|
|
}
|
|
if c.config.Token != "" {
|
|
r.params.Set("token", r.config.Token)
|
|
}
|
|
return r
|
|
}
|
|
|
|
// doRequest runs a request with our client
|
|
func (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.Now().Sub(start)
|
|
return diff, resp, err
|
|
}
|
|
|
|
// parseQueryMeta is used to help parse query meta-data
|
|
func parseQueryMeta(resp *http.Response, q *QueryMeta) error {
|
|
header := resp.Header
|
|
|
|
// Parse the X-Consul-Index
|
|
index, err := strconv.ParseUint(header.Get("X-Consul-Index"), 10, 64)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to parse X-Consul-Index: %v", err)
|
|
}
|
|
q.LastIndex = index
|
|
|
|
// Parse the X-Consul-LastContact
|
|
last, err := strconv.ParseUint(header.Get("X-Consul-LastContact"), 10, 64)
|
|
if err != nil {
|
|
return fmt.Errorf("Failed to parse X-Consul-LastContact: %v", err)
|
|
}
|
|
q.LastContact = time.Duration(last) * time.Millisecond
|
|
|
|
// Parse the X-Consul-KnownLeader
|
|
switch header.Get("X-Consul-KnownLeader") {
|
|
case "true":
|
|
q.KnownLeader = true
|
|
default:
|
|
q.KnownLeader = false
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// decodeBody is used to JSON decode a body
|
|
func decodeBody(resp *http.Response, out interface{}) error {
|
|
dec := json.NewDecoder(resp.Body)
|
|
return dec.Decode(out)
|
|
}
|
|
|
|
// encodeBody is used to encode a request body
|
|
func 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 200
|
|
func requireOK(d time.Duration, resp *http.Response, e error) (time.Duration, *http.Response, error) {
|
|
if e != nil {
|
|
return d, resp, e
|
|
}
|
|
if resp.StatusCode != 200 {
|
|
var buf bytes.Buffer
|
|
io.Copy(&buf, resp.Body)
|
|
return d, resp, fmt.Errorf("Unexpected response code: %d (%s)", resp.StatusCode, buf.Bytes())
|
|
}
|
|
return d, resp, e
|
|
}
|