moby/pkg/plugins/client.go
Daniel Nephin 4f0d95fa6e Add canonical import comment
Signed-off-by: Daniel Nephin <dnephin@docker.com>
2018-02-05 16:51:57 -05:00

242 lines
6.1 KiB
Go

package plugins // import "github.com/docker/docker/pkg/plugins"
import (
"bytes"
"context"
"encoding/json"
"io"
"io/ioutil"
"net/http"
"net/url"
"time"
"github.com/docker/docker/pkg/ioutils"
"github.com/docker/docker/pkg/plugins/transport"
"github.com/docker/go-connections/sockets"
"github.com/docker/go-connections/tlsconfig"
"github.com/sirupsen/logrus"
)
const (
defaultTimeOut = 30
)
func newTransport(addr string, tlsConfig *tlsconfig.Options) (transport.Transport, error) {
tr := &http.Transport{}
if tlsConfig != nil {
c, err := tlsconfig.Client(*tlsConfig)
if err != nil {
return nil, err
}
tr.TLSClientConfig = c
}
u, err := url.Parse(addr)
if err != nil {
return nil, err
}
socket := u.Host
if socket == "" {
// valid local socket addresses have the host empty.
socket = u.Path
}
if err := sockets.ConfigureTransport(tr, u.Scheme, socket); err != nil {
return nil, err
}
scheme := httpScheme(u)
return transport.NewHTTPTransport(tr, scheme, socket), nil
}
// NewClient creates a new plugin client (http).
func NewClient(addr string, tlsConfig *tlsconfig.Options) (*Client, error) {
clientTransport, err := newTransport(addr, tlsConfig)
if err != nil {
return nil, err
}
return newClientWithTransport(clientTransport, 0), nil
}
// NewClientWithTimeout creates a new plugin client (http).
func NewClientWithTimeout(addr string, tlsConfig *tlsconfig.Options, timeout time.Duration) (*Client, error) {
clientTransport, err := newTransport(addr, tlsConfig)
if err != nil {
return nil, err
}
return newClientWithTransport(clientTransport, timeout), nil
}
// newClientWithTransport creates a new plugin client with a given transport.
func newClientWithTransport(tr transport.Transport, timeout time.Duration) *Client {
return &Client{
http: &http.Client{
Transport: tr,
Timeout: timeout,
},
requestFactory: tr,
}
}
// Client represents a plugin client.
type Client struct {
http *http.Client // http client to use
requestFactory transport.RequestFactory
}
// RequestOpts is the set of options that can be passed into a request
type RequestOpts struct {
Timeout time.Duration
}
// WithRequestTimeout sets a timeout duration for plugin requests
func WithRequestTimeout(t time.Duration) func(*RequestOpts) {
return func(o *RequestOpts) {
o.Timeout = t
}
}
// Call calls the specified method with the specified arguments for the plugin.
// It will retry for 30 seconds if a failure occurs when calling.
func (c *Client) Call(serviceMethod string, args, ret interface{}) error {
return c.CallWithOptions(serviceMethod, args, ret)
}
// CallWithOptions is just like call except it takes options
func (c *Client) CallWithOptions(serviceMethod string, args interface{}, ret interface{}, opts ...func(*RequestOpts)) error {
var buf bytes.Buffer
if args != nil {
if err := json.NewEncoder(&buf).Encode(args); err != nil {
return err
}
}
body, err := c.callWithRetry(serviceMethod, &buf, true, opts...)
if err != nil {
return err
}
defer body.Close()
if ret != nil {
if err := json.NewDecoder(body).Decode(&ret); err != nil {
logrus.Errorf("%s: error reading plugin resp: %v", serviceMethod, err)
return err
}
}
return nil
}
// Stream calls the specified method with the specified arguments for the plugin and returns the response body
func (c *Client) Stream(serviceMethod string, args interface{}) (io.ReadCloser, error) {
var buf bytes.Buffer
if err := json.NewEncoder(&buf).Encode(args); err != nil {
return nil, err
}
return c.callWithRetry(serviceMethod, &buf, true)
}
// SendFile calls the specified method, and passes through the IO stream
func (c *Client) SendFile(serviceMethod string, data io.Reader, ret interface{}) error {
body, err := c.callWithRetry(serviceMethod, data, true)
if err != nil {
return err
}
defer body.Close()
if err := json.NewDecoder(body).Decode(&ret); err != nil {
logrus.Errorf("%s: error reading plugin resp: %v", serviceMethod, err)
return err
}
return nil
}
func (c *Client) callWithRetry(serviceMethod string, data io.Reader, retry bool, reqOpts ...func(*RequestOpts)) (io.ReadCloser, error) {
var retries int
start := time.Now()
var opts RequestOpts
for _, o := range reqOpts {
o(&opts)
}
for {
req, err := c.requestFactory.NewRequest(serviceMethod, data)
if err != nil {
return nil, err
}
cancelRequest := func() {}
if opts.Timeout > 0 {
var ctx context.Context
ctx, cancelRequest = context.WithTimeout(req.Context(), opts.Timeout)
req = req.WithContext(ctx)
}
resp, err := c.http.Do(req)
if err != nil {
cancelRequest()
if !retry {
return nil, err
}
timeOff := backoff(retries)
if abort(start, timeOff) {
return nil, err
}
retries++
logrus.Warnf("Unable to connect to plugin: %s%s: %v, retrying in %v", req.URL.Host, req.URL.Path, err, timeOff)
time.Sleep(timeOff)
continue
}
if resp.StatusCode != http.StatusOK {
b, err := ioutil.ReadAll(resp.Body)
resp.Body.Close()
cancelRequest()
if err != nil {
return nil, &statusError{resp.StatusCode, serviceMethod, err.Error()}
}
// Plugins' Response(s) should have an Err field indicating what went
// wrong. Try to unmarshal into ResponseErr. Otherwise fallback to just
// return the string(body)
type responseErr struct {
Err string
}
remoteErr := responseErr{}
if err := json.Unmarshal(b, &remoteErr); err == nil {
if remoteErr.Err != "" {
return nil, &statusError{resp.StatusCode, serviceMethod, remoteErr.Err}
}
}
// old way...
return nil, &statusError{resp.StatusCode, serviceMethod, string(b)}
}
return ioutils.NewReadCloserWrapper(resp.Body, func() error {
err := resp.Body.Close()
cancelRequest()
return err
}), nil
}
}
func backoff(retries int) time.Duration {
b, max := 1, defaultTimeOut
for b < max && retries > 0 {
b *= 2
retries--
}
if b > max {
b = max
}
return time.Duration(b) * time.Second
}
func abort(start time.Time, timeOff time.Duration) bool {
return timeOff+time.Since(start) >= time.Duration(defaultTimeOut)*time.Second
}
func httpScheme(u *url.URL) string {
scheme := u.Scheme
if scheme != "https" {
scheme = "http"
}
return scheme
}