lib/controller/rpc/conn.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package rpc
   6
   7 import (
   8         "context"
   9         "crypto/tls"
  10         "encoding/json"
  11         "fmt"
  12         "io"
  13         "net"
  14         "net/http"
  15         "net/url"
  16         "strings"
  17         "time"
  18
  19         "git.curoverse.com/arvados.git/sdk/go/arvados"
  20 )
  21
  22 type contextKey string
  23
  24 const ContextKeyCredentials contextKey = "credentials"
  25
  26 type TokenProvider func(context.Context) ([]string, error)
  27
  28 type Conn struct {
  29         clusterID     string
  30         httpClient    http.Client
  31         baseURL       url.URL
  32         tokenProvider TokenProvider
  33 }
  34
  35 func NewConn(clusterID string, url *url.URL, insecure bool, tp TokenProvider) *Conn {
  36         transport := http.DefaultTransport
  37         if insecure {
  38                 // It's not safe to copy *http.DefaultTransport
  39                 // because it has a mutex (which might be locked)
  40                 // protecting a private map (which might not be nil).
  41                 // So we build our own, using the Go 1.12 default
  42                 // values, ignoring any changes the application has
  43                 // made to http.DefaultTransport.
  44                 transport = &http.Transport{
  45                         DialContext: (&net.Dialer{
  46                                 Timeout:   30 * time.Second,
  47                                 KeepAlive: 30 * time.Second,
  48                                 DualStack: true,
  49                         }).DialContext,
  50                         MaxIdleConns:          100,
  51                         IdleConnTimeout:       90 * time.Second,
  52                         TLSHandshakeTimeout:   10 * time.Second,
  53                         ExpectContinueTimeout: 1 * time.Second,
  54                         TLSClientConfig:       &tls.Config{InsecureSkipVerify: true},
  55                 }
  56         }
  57         return &Conn{
  58                 clusterID:     clusterID,
  59                 httpClient:    http.Client{Transport: transport},
  60                 baseURL:       *url,
  61                 tokenProvider: tp,
  62         }
  63 }
  64
  65 func (conn *Conn) requestAndDecode(ctx context.Context, dst interface{}, ep arvados.APIEndpoint, body io.Reader, opts interface{}) error {
  66         aClient := arvados.Client{
  67                 Client:  &conn.httpClient,
  68                 Scheme:  conn.baseURL.Scheme,
  69                 APIHost: conn.baseURL.Host,
  70         }
  71         tokens, err := conn.tokenProvider(ctx)
  72         if err != nil {
  73                 return err
  74         } else if len(tokens) > 0 {
  75                 ctx = context.WithValue(ctx, "Authorization", "Bearer "+tokens[0])
  76         } else {
  77                 // Use a non-empty auth string to ensure we override
  78                 // any default token set on aClient -- and to avoid
  79                 // having the remote prompt us to send a token by
  80                 // responding 401.
  81                 ctx = context.WithValue(ctx, "Authorization", "Bearer -")
  82         }
  83
  84         // Encode opts to JSON and decode from there to a
  85         // map[string]interface{}, so we can munge the query params
  86         // using the JSON key names specified by opts' struct tags.
  87         j, err := json.Marshal(opts)
  88         if err != nil {
  89                 return fmt.Errorf("%T: requestAndDecode: Marshal opts: %s", conn, err)
  90         }
  91         var params map[string]interface{}
  92         err = json.Unmarshal(j, &params)
  93         if err != nil {
  94                 return fmt.Errorf("%T: requestAndDecode: Unmarshal opts: %s", conn, err)
  95         }
  96         if attrs, ok := params["attrs"]; ok && ep.AttrsKey != "" {
  97                 params[ep.AttrsKey] = attrs
  98                 delete(params, "attrs")
  99         }
 100         if limit, ok := params["limit"].(float64); ok && limit < 0 {
 101                 // Negative limit means "not specified" here, but some
 102                 // servers/versions do not accept that, so we need to
 103                 // remove it entirely.
 104                 delete(params, "limit")
 105         }
 106         if len(tokens) > 1 {
 107                 params["reader_tokens"] = tokens[1:]
 108         }
 109         path := ep.Path
 110         if strings.Contains(ep.Path, "/:uuid") {
 111                 uuid, _ := params["uuid"].(string)
 112                 path = strings.Replace(path, "/:uuid", "/"+uuid, 1)
 113                 delete(params, "uuid")
 114         }
 115         return aClient.RequestAndDecodeContext(ctx, dst, ep.Method, path, body, params)
 116 }
 117
 118 func (conn *Conn) CollectionCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Collection, error) {
 119         ep := arvados.EndpointCollectionCreate
 120         var resp arvados.Collection
 121         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 122         return resp, err
 123 }
 124
 125 func (conn *Conn) CollectionUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Collection, error) {
 126         ep := arvados.EndpointCollectionUpdate
 127         var resp arvados.Collection
 128         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 129         return resp, err
 130 }
 131
 132 func (conn *Conn) CollectionGet(ctx context.Context, options arvados.GetOptions) (arvados.Collection, error) {
 133         ep := arvados.EndpointCollectionGet
 134         var resp arvados.Collection
 135         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 136         return resp, err
 137 }
 138
 139 func (conn *Conn) CollectionList(ctx context.Context, options arvados.ListOptions) (arvados.CollectionList, error) {
 140         ep := arvados.EndpointCollectionList
 141         var resp arvados.CollectionList
 142         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 143         return resp, err
 144 }
 145
 146 func (conn *Conn) CollectionProvenance(ctx context.Context, options arvados.GetOptions) (map[string]interface{}, error) {
 147         ep := arvados.EndpointCollectionProvenance
 148         var resp map[string]interface{}
 149         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 150         return resp, err
 151 }
 152
 153 func (conn *Conn) CollectionUsedBy(ctx context.Context, options arvados.GetOptions) (map[string]interface{}, error) {
 154         ep := arvados.EndpointCollectionUsedBy
 155         var resp map[string]interface{}
 156         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 157         return resp, err
 158 }
 159
 160 func (conn *Conn) CollectionDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Collection, error) {
 161         ep := arvados.EndpointCollectionDelete
 162         var resp arvados.Collection
 163         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 164         return resp, err
 165 }
 166
 167 func (conn *Conn) ContainerCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Container, error) {
 168         ep := arvados.EndpointContainerCreate
 169         var resp arvados.Container
 170         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 171         return resp, err
 172 }
 173
 174 func (conn *Conn) ContainerUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Container, error) {
 175         ep := arvados.EndpointContainerUpdate
 176         var resp arvados.Container
 177         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 178         return resp, err
 179 }
 180
 181 func (conn *Conn) ContainerGet(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
 182         ep := arvados.EndpointContainerGet
 183         var resp arvados.Container
 184         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 185         return resp, err
 186 }
 187
 188 func (conn *Conn) ContainerList(ctx context.Context, options arvados.ListOptions) (arvados.ContainerList, error) {
 189         ep := arvados.EndpointContainerList
 190         var resp arvados.ContainerList
 191         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 192         return resp, err
 193 }
 194
 195 func (conn *Conn) ContainerDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Container, error) {
 196         ep := arvados.EndpointContainerDelete
 197         var resp arvados.Container
 198         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 199         return resp, err
 200 }
 201
 202 func (conn *Conn) ContainerLock(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
 203         ep := arvados.EndpointContainerLock
 204         var resp arvados.Container
 205         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 206         return resp, err
 207 }
 208
 209 func (conn *Conn) ContainerUnlock(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) {
 210         ep := arvados.EndpointContainerUnlock
 211         var resp arvados.Container
 212         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 213         return resp, err
 214 }
 215
 216 func (conn *Conn) SpecimenCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Specimen, error) {
 217         ep := arvados.EndpointSpecimenCreate
 218         var resp arvados.Specimen
 219         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 220         return resp, err
 221 }
 222
 223 func (conn *Conn) SpecimenUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Specimen, error) {
 224         ep := arvados.EndpointSpecimenUpdate
 225         var resp arvados.Specimen
 226         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 227         return resp, err
 228 }
 229
 230 func (conn *Conn) SpecimenGet(ctx context.Context, options arvados.GetOptions) (arvados.Specimen, error) {
 231         ep := arvados.EndpointSpecimenGet
 232         var resp arvados.Specimen
 233         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 234         return resp, err
 235 }
 236
 237 func (conn *Conn) SpecimenList(ctx context.Context, options arvados.ListOptions) (arvados.SpecimenList, error) {
 238         ep := arvados.EndpointSpecimenList
 239         var resp arvados.SpecimenList
 240         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 241         return resp, err
 242 }
 243
 244 func (conn *Conn) SpecimenDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Specimen, error) {
 245         ep := arvados.EndpointSpecimenDelete
 246         var resp arvados.Specimen
 247         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 248         return resp, err
 249 }
 250
 251 func (conn *Conn) APIClientAuthorizationCurrent(ctx context.Context, options arvados.GetOptions) (arvados.APIClientAuthorization, error) {
 252         ep := arvados.EndpointAPIClientAuthorizationCurrent
 253         var resp arvados.APIClientAuthorization
 254         err := conn.requestAndDecode(ctx, &resp, ep, nil, options)
 255         return resp, err
 256 }