kubernetes/vendor/github.com/storageos/go-api/client.go

package storageos

import (
	"bytes"
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/storageos/go-api/netutil"
	"github.com/storageos/go-api/serror"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"net/url"
	"reflect"
	"strconv"
	"strings"
	"time"
)

const (
	userAgent         = "go-storageosclient"
	DefaultVersionStr = "1"
	DefaultVersion    = 1
)

var (
	// ErrConnectionRefused is returned when the client cannot connect to the given endpoint.
	ErrConnectionRefused = errors.New("cannot connect to StorageOS API endpoint")

	// ErrInactivityTimeout is returned when a streamable call has been inactive for some time.
	ErrInactivityTimeout = errors.New("inactivity time exceeded timeout")

	// ErrInvalidVersion is returned when a versioned client was requested but no version specified.
	ErrInvalidVersion = errors.New("invalid version")

	// DefaultPort is the default API port
	DefaultPort = "5705"

	// DataplaneHealthPort is the the port used by the dataplane health-check service
	DataplaneHealthPort = "5704"

	// DefaultHost is the default API host
	DefaultHost = "tcp://localhost:" + DefaultPort
)

// APIVersion is an internal representation of a version of the Remote API.
type APIVersion int

// NewAPIVersion returns an instance of APIVersion for the given string.
//
// The given string must be in the form <major>
func NewAPIVersion(input string) (APIVersion, error) {
	if input == "" {
		return DefaultVersion, ErrInvalidVersion
	}
	version, err := strconv.Atoi(input)
	if err != nil {
		return 0, fmt.Errorf("Unable to parse version %q", input)
	}
	return APIVersion(version), nil
}

func (version APIVersion) String() string {
	return fmt.Sprintf("v%d", version)
}

// Client is the basic type of this package. It provides methods for
// interaction with the API.
type Client struct {
	SkipServerVersionCheck bool
	HTTPClient             *http.Client
	TLSConfig              *tls.Config
	username               string
	secret                 string
	requestedAPIVersion    APIVersion
	serverAPIVersion       APIVersion
	expectedAPIVersion     APIVersion
	nativeHTTPClient       *http.Client
	useTLS                 bool
}

// ClientVersion returns the API version of the client
func (c *Client) ClientVersion() string {
	return DefaultVersionStr
}

// Dialer is an interface that allows network connections to be dialed
// (net.Dialer fulfills this interface) and named pipes (a shim using
// winio.DialPipe)
type Dialer interface {
	Dial(network, address string) (net.Conn, error)
}

// NewClient returns a Client instance ready for communication with the given
// server endpoint. It will use the latest remote API version available in the
// server.
func NewClient(nodes string) (*Client, error) {
	client, err := NewVersionedClient(nodes, "")
	if err != nil {
		return nil, err
	}
	client.SkipServerVersionCheck = true
	return client, nil
}

// NewVersionedClient returns a Client instance ready for communication with
// the given server endpoint, using a specific remote API version.
func NewVersionedClient(nodestring string, apiVersionString string) (*Client, error) {
	nodes := strings.Split(nodestring, ",")

	d, err := netutil.NewMultiDialer(nodes, nil)
	if err != nil {
		return nil, err
	}

	var useTLS bool
	if len(nodes) > 0 {
		if u, err := url.Parse(nodes[0]); err != nil && u.Scheme == "https" {
			useTLS = true
		}
	}

	c := &Client{
		HTTPClient: defaultClient(d),
		useTLS:     useTLS,
	}

	if apiVersionString != "" {
		version, err := strconv.Atoi(apiVersionString)
		if err != nil {
			return nil, err
		}
		c.requestedAPIVersion = APIVersion(version)
	}

	return c, nil
}

// SetAuth sets the API username and secret to be used for all API requests.
// It should not be called concurrently with any other Client methods.
func (c *Client) SetAuth(username string, secret string) {
	if username != "" {
		c.username = username
	}
	if secret != "" {
		c.secret = secret
	}
}

// SetTimeout takes a timeout and applies it to both the HTTPClient and
// nativeHTTPClient. It should not be called concurrently with any other Client
// methods.
func (c *Client) SetTimeout(t time.Duration) {
	if c.HTTPClient != nil {
		c.HTTPClient.Timeout = t
	}
	if c.nativeHTTPClient != nil {
		c.nativeHTTPClient.Timeout = t
	}
}

func (c *Client) checkAPIVersion() error {
	serverAPIVersionString, err := c.getServerAPIVersionString()
	if err != nil {
		return err
	}
	c.serverAPIVersion, err = NewAPIVersion(serverAPIVersionString)
	if err != nil {
		return err
	}
	if c.requestedAPIVersion == 0 {
		c.expectedAPIVersion = c.serverAPIVersion
	} else {
		c.expectedAPIVersion = c.requestedAPIVersion
	}
	return nil
}

// Ping pings the API server
//
// See https://goo.gl/wYfgY1 for more details.
func (c *Client) Ping() error {
	urlpath := "/_ping"
	resp, err := c.do("GET", urlpath, doOptions{})
	if err != nil {
		return err
	}
	if resp.StatusCode != http.StatusOK {
		return newError(resp)
	}
	resp.Body.Close()
	return nil
}

func (c *Client) getServerAPIVersionString() (version string, err error) {
	v, err := c.ServerVersion(context.Background())
	if err != nil {
		return "", err
	}
	return v.APIVersion, nil
}

type doOptions struct {
	data          interface{}
	fieldSelector string
	labelSelector string
	namespace     string
	forceJSON     bool
	force         bool
	values        url.Values
	headers       map[string]string
	unversioned   bool
	context       context.Context
}

func (c *Client) do(method, urlpath string, doOptions doOptions) (*http.Response, error) {
	var params io.Reader
	if doOptions.data != nil || doOptions.forceJSON {
		buf, err := json.Marshal(doOptions.data)
		if err != nil {
			return nil, err
		}
		params = bytes.NewBuffer(buf)
	}

	// Prefix the path with the namespace if given.  The caller should only set
	// the namespace if this is desired.
	if doOptions.namespace != "" {
		urlpath = "/" + NamespaceAPIPrefix + "/" + doOptions.namespace + "/" + urlpath
	}

	if !c.SkipServerVersionCheck && !doOptions.unversioned {
		err := c.checkAPIVersion()
		if err != nil {
			return nil, err
		}
	}

	query := url.Values{}
	if doOptions.values != nil {
		query = doOptions.values
	}
	if doOptions.force {
		query.Add("force", "1")
	}

	httpClient := c.HTTPClient
	u := c.getAPIPath(urlpath, query, doOptions.unversioned)

	req, err := http.NewRequest(method, u, params)
	if err != nil {
		return nil, err
	}
	req.Header.Set("User-Agent", userAgent)
	if doOptions.data != nil {
		req.Header.Set("Content-Type", "application/json")
	} else if method == "POST" {
		req.Header.Set("Content-Type", "plain/text")
	}
	if c.username != "" && c.secret != "" {
		req.SetBasicAuth(c.username, c.secret)
	}

	for k, v := range doOptions.headers {
		req.Header.Set(k, v)
	}

	ctx := doOptions.context
	if ctx == nil {
		ctx = context.Background()
	}

	resp, err := httpClient.Do(req.WithContext(ctx))
	if err != nil {
		// If it is a custom error, return it. It probably knows more than us
		if serror.IsStorageOSError(err) {
			return nil, err
		}

		if strings.Contains(err.Error(), "connection refused") {
			return nil, ErrConnectionRefused
		}
		return nil, chooseError(ctx, err)
	}
	if resp.StatusCode < 200 || resp.StatusCode >= 400 {
		return nil, newError(resp)
	}
	return resp, nil
}

// if error in context, return that instead of generic http error
func chooseError(ctx context.Context, err error) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
		return err
	}
}

func (c *Client) getAPIPath(path string, query url.Values, unversioned bool) string {
	// The custom dialer contacts the hosts for us, making this hosname irrelevant
	var urlStr string
	if c.useTLS {
		urlStr = "https://storageos-cluster"
	} else {
		urlStr = "http://storageos-cluster"
	}

	var apiPath string

	path = strings.TrimLeft(path, "/")
	if unversioned {
		apiPath = fmt.Sprintf("%s/%s", urlStr, path)
	} else {
		apiPath = fmt.Sprintf("%s/%s/%s", urlStr, c.requestedAPIVersion, path)
	}

	if len(query) > 0 {
		apiPath = apiPath + "?" + query.Encode()
	}

	return apiPath
}

func queryString(opts interface{}) string {
	if opts == nil {
		return ""
	}
	value := reflect.ValueOf(opts)
	if value.Kind() == reflect.Ptr {
		value = value.Elem()
	}
	if value.Kind() != reflect.Struct {
		return ""
	}
	items := url.Values(map[string][]string{})
	for i := 0; i < value.NumField(); i++ {
		field := value.Type().Field(i)
		if field.PkgPath != "" {
			continue
		}
		key := field.Tag.Get("qs")
		if key == "" {
			key = strings.ToLower(field.Name)
		} else if key == "-" {
			continue
		}
		addQueryStringValue(items, key, value.Field(i))
	}
	return items.Encode()
}

func addQueryStringValue(items url.Values, key string, v reflect.Value) {
	switch v.Kind() {
	case reflect.Bool:
		if v.Bool() {
			items.Add(key, "1")
		}
	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
		if v.Int() > 0 {
			items.Add(key, strconv.FormatInt(v.Int(), 10))
		}
	case reflect.Float32, reflect.Float64:
		if v.Float() > 0 {
			items.Add(key, strconv.FormatFloat(v.Float(), 'f', -1, 64))
		}
	case reflect.String:
		if v.String() != "" {
			items.Add(key, v.String())
		}
	case reflect.Ptr:
		if !v.IsNil() {
			if b, err := json.Marshal(v.Interface()); err == nil {
				items.Add(key, string(b))
			}
		}
	case reflect.Map:
		if len(v.MapKeys()) > 0 {
			if b, err := json.Marshal(v.Interface()); err == nil {
				items.Add(key, string(b))
			}
		}
	case reflect.Array, reflect.Slice:
		vLen := v.Len()
		if vLen > 0 {
			for i := 0; i < vLen; i++ {
				addQueryStringValue(items, key, v.Index(i))
			}
		}
	}
}

// Error represents failures in the API. It represents a failure from the API.
type Error struct {
	Status  int
	Message string
}

func newError(resp *http.Response) *Error {
	type jsonError struct {
		Message string `json:"message"`
	}

	defer resp.Body.Close()
	data, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return &Error{Status: resp.StatusCode, Message: fmt.Sprintf("cannot read body, err: %v", err)}
	}

	// attempt to unmarshal the error if in json format
	jerr := &jsonError{}
	err = json.Unmarshal(data, jerr)
	if err != nil {
		return &Error{Status: resp.StatusCode, Message: string(data)} // Failed, just return string
	}

	return &Error{Status: resp.StatusCode, Message: jerr.Message}
}

func (e *Error) Error() string {
	var niceStatus string

	switch e.Status {
	case 400, 500:
		niceStatus = "Server failed to process your request. Was the data correct?"
	case 401:
		niceStatus = "Unauthenticated access of secure endpoint, please retry after authentication"
	case 403:
		niceStatus = "Forbidden request. Your user cannot perform this action"
	case 404:
		niceStatus = "Requested object not found. Does this item exist?"
	}

	if niceStatus != "" {
		return fmt.Sprintf("API error (%s): %s", niceStatus, e.Message)
	}
	return fmt.Sprintf("API error (%s): %s", http.StatusText(e.Status), e.Message)
}

// defaultTransport returns a new http.Transport with the same default values
// as http.DefaultTransport, but with idle connections and keepalives disabled.
func defaultTransport(d Dialer) *http.Transport {
	transport := defaultPooledTransport(d)
	transport.DisableKeepAlives = true
	transport.MaxIdleConnsPerHost = -1
	return transport
}

// defaultPooledTransport returns a new http.Transport with similar default
// values to http.DefaultTransport. Do not use this for transient transports as
// it can leak file descriptors over time. Only use this for transports that
// will be re-used for the same host(s).
func defaultPooledTransport(d Dialer) *http.Transport {
	transport := &http.Transport{
		Proxy:               http.ProxyFromEnvironment,
		Dial:                d.Dial,
		TLSHandshakeTimeout: 5 * time.Second,
		DisableKeepAlives:   false,
		MaxIdleConnsPerHost: 1,
	}
	return transport
}

// defaultClient returns a new http.Client with similar default values to
// http.Client, but with a non-shared Transport, idle connections disabled, and
// keepalives disabled.
// If a custom dialer is not provided, one with sane defaults will be created.
func defaultClient(d Dialer) *http.Client {
	if d == nil {
		d = &net.Dialer{
			Timeout:   5 * time.Second,
			KeepAlive: 5 * time.Second,
		}
	}

	return &http.Client{
		Transport: defaultTransport(d),
	}
}