client.go 5.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205
  1. package plugins
  2. import (
  3. "bytes"
  4. "encoding/json"
  5. "io"
  6. "io/ioutil"
  7. "net/http"
  8. "net/url"
  9. "time"
  10. "github.com/Sirupsen/logrus"
  11. "github.com/docker/docker/pkg/plugins/transport"
  12. "github.com/docker/go-connections/sockets"
  13. "github.com/docker/go-connections/tlsconfig"
  14. )
  15. const (
  16. defaultTimeOut = 30
  17. )
  18. func newTransport(addr string, tlsConfig *tlsconfig.Options) (transport.Transport, error) {
  19. tr := &http.Transport{}
  20. if tlsConfig != nil {
  21. c, err := tlsconfig.Client(*tlsConfig)
  22. if err != nil {
  23. return nil, err
  24. }
  25. tr.TLSClientConfig = c
  26. }
  27. u, err := url.Parse(addr)
  28. if err != nil {
  29. return nil, err
  30. }
  31. socket := u.Host
  32. if socket == "" {
  33. // valid local socket addresses have the host empty.
  34. socket = u.Path
  35. }
  36. if err := sockets.ConfigureTransport(tr, u.Scheme, socket); err != nil {
  37. return nil, err
  38. }
  39. scheme := httpScheme(u)
  40. return transport.NewHTTPTransport(tr, scheme, socket), nil
  41. }
  42. // NewClient creates a new plugin client (http).
  43. func NewClient(addr string, tlsConfig *tlsconfig.Options) (*Client, error) {
  44. clientTransport, err := newTransport(addr, tlsConfig)
  45. if err != nil {
  46. return nil, err
  47. }
  48. return newClientWithTransport(clientTransport, 0), nil
  49. }
  50. // NewClientWithTimeout creates a new plugin client (http).
  51. func NewClientWithTimeout(addr string, tlsConfig *tlsconfig.Options, timeoutInSecs int) (*Client, error) {
  52. clientTransport, err := newTransport(addr, tlsConfig)
  53. if err != nil {
  54. return nil, err
  55. }
  56. return newClientWithTransport(clientTransport, timeoutInSecs), nil
  57. }
  58. // newClientWithTransport creates a new plugin client with a given transport.
  59. func newClientWithTransport(tr transport.Transport, timeoutInSecs int) *Client {
  60. return &Client{
  61. http: &http.Client{
  62. Transport: tr,
  63. Timeout: time.Duration(timeoutInSecs) * time.Second,
  64. },
  65. requestFactory: tr,
  66. }
  67. }
  68. // Client represents a plugin client.
  69. type Client struct {
  70. http *http.Client // http client to use
  71. requestFactory transport.RequestFactory
  72. }
  73. // Call calls the specified method with the specified arguments for the plugin.
  74. // It will retry for 30 seconds if a failure occurs when calling.
  75. func (c *Client) Call(serviceMethod string, args interface{}, ret interface{}) error {
  76. var buf bytes.Buffer
  77. if args != nil {
  78. if err := json.NewEncoder(&buf).Encode(args); err != nil {
  79. return err
  80. }
  81. }
  82. body, err := c.callWithRetry(serviceMethod, &buf, true)
  83. if err != nil {
  84. return err
  85. }
  86. defer body.Close()
  87. if ret != nil {
  88. if err := json.NewDecoder(body).Decode(&ret); err != nil {
  89. logrus.Errorf("%s: error reading plugin resp: %v", serviceMethod, err)
  90. return err
  91. }
  92. }
  93. return nil
  94. }
  95. // Stream calls the specified method with the specified arguments for the plugin and returns the response body
  96. func (c *Client) Stream(serviceMethod string, args interface{}) (io.ReadCloser, error) {
  97. var buf bytes.Buffer
  98. if err := json.NewEncoder(&buf).Encode(args); err != nil {
  99. return nil, err
  100. }
  101. return c.callWithRetry(serviceMethod, &buf, true)
  102. }
  103. // SendFile calls the specified method, and passes through the IO stream
  104. func (c *Client) SendFile(serviceMethod string, data io.Reader, ret interface{}) error {
  105. body, err := c.callWithRetry(serviceMethod, data, true)
  106. if err != nil {
  107. return err
  108. }
  109. defer body.Close()
  110. if err := json.NewDecoder(body).Decode(&ret); err != nil {
  111. logrus.Errorf("%s: error reading plugin resp: %v", serviceMethod, err)
  112. return err
  113. }
  114. return nil
  115. }
  116. func (c *Client) callWithRetry(serviceMethod string, data io.Reader, retry bool) (io.ReadCloser, error) {
  117. var retries int
  118. start := time.Now()
  119. for {
  120. req, err := c.requestFactory.NewRequest(serviceMethod, data)
  121. if err != nil {
  122. return nil, err
  123. }
  124. resp, err := c.http.Do(req)
  125. if err != nil {
  126. if !retry {
  127. return nil, err
  128. }
  129. timeOff := backoff(retries)
  130. if abort(start, timeOff) {
  131. return nil, err
  132. }
  133. retries++
  134. logrus.Warnf("Unable to connect to plugin: %s%s: %v, retrying in %v", req.URL.Host, req.URL.Path, err, timeOff)
  135. time.Sleep(timeOff)
  136. continue
  137. }
  138. if resp.StatusCode != http.StatusOK {
  139. b, err := ioutil.ReadAll(resp.Body)
  140. resp.Body.Close()
  141. if err != nil {
  142. return nil, &statusError{resp.StatusCode, serviceMethod, err.Error()}
  143. }
  144. // Plugins' Response(s) should have an Err field indicating what went
  145. // wrong. Try to unmarshal into ResponseErr. Otherwise fallback to just
  146. // return the string(body)
  147. type responseErr struct {
  148. Err string
  149. }
  150. remoteErr := responseErr{}
  151. if err := json.Unmarshal(b, &remoteErr); err == nil {
  152. if remoteErr.Err != "" {
  153. return nil, &statusError{resp.StatusCode, serviceMethod, remoteErr.Err}
  154. }
  155. }
  156. // old way...
  157. return nil, &statusError{resp.StatusCode, serviceMethod, string(b)}
  158. }
  159. return resp.Body, nil
  160. }
  161. }
  162. func backoff(retries int) time.Duration {
  163. b, max := 1, defaultTimeOut
  164. for b < max && retries > 0 {
  165. b *= 2
  166. retries--
  167. }
  168. if b > max {
  169. b = max
  170. }
  171. return time.Duration(b) * time.Second
  172. }
  173. func abort(start time.Time, timeOff time.Duration) bool {
  174. return timeOff+time.Since(start) >= time.Duration(defaultTimeOut)*time.Second
  175. }
  176. func httpScheme(u *url.URL) string {
  177. scheme := u.Scheme
  178. if scheme != "https" {
  179. scheme = "http"
  180. }
  181. return scheme
  182. }