X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/16905dd94cb77336351a7a6d2e6ece957f8f3368..ec17f6971109186961283443f2df6d5802bea401:/lib/controller/rpc/conn.go diff --git a/lib/controller/rpc/conn.go b/lib/controller/rpc/conn.go index 9d64dc539a..3621f42d0e 100644 --- a/lib/controller/rpc/conn.go +++ b/lib/controller/rpc/conn.go @@ -16,6 +16,7 @@ import ( "io/ioutil" "net" "net/http" + "net/http/httputil" "net/url" "strconv" "strings" @@ -40,7 +41,9 @@ func PassthroughTokenProvider(ctx context.Context) ([]string, error) { } type Conn struct { - SendHeader http.Header + SendHeader http.Header + RedactHostInErrors bool + clusterID string httpClient http.Client baseURL url.URL @@ -86,6 +89,8 @@ func (conn *Conn) requestAndDecode(ctx context.Context, dst interface{}, ep arva Scheme: conn.baseURL.Scheme, APIHost: conn.baseURL.Host, SendHeader: conn.SendHeader, + // Disable auto-retry + Timeout: 0, } tokens, err := conn.tokenProvider(ctx) if err != nil { @@ -149,7 +154,21 @@ func (conn *Conn) requestAndDecode(ctx context.Context, dst interface{}, ep arva path = strings.Replace(path, "/{uuid}", "/"+uuid, 1) delete(params, "uuid") } - return aClient.RequestAndDecodeContext(ctx, dst, ep.Method, path, body, params) + err = aClient.RequestAndDecodeContext(ctx, dst, ep.Method, path, body, params) + if err != nil && conn.RedactHostInErrors { + redacted := strings.Replace(err.Error(), strings.TrimSuffix(conn.baseURL.String(), "/"), "//railsapi.internal", -1) + if strings.HasPrefix(redacted, "request failed: ") { + redacted = strings.Replace(redacted, "request failed: ", "", -1) + } + if redacted != err.Error() { + if err, ok := err.(httpStatusError); ok { + return wrapHTTPStatusError(err, redacted) + } else { + return errors.New(redacted) + } + } + } + return err } func (conn *Conn) BaseURL() url.URL { @@ -163,6 +182,13 @@ func (conn *Conn) ConfigGet(ctx context.Context) (json.RawMessage, error) { return resp, err } +func (conn *Conn) VocabularyGet(ctx context.Context) (arvados.Vocabulary, error) { + ep := arvados.EndpointVocabularyGet + var resp arvados.Vocabulary + err := conn.requestAndDecode(ctx, &resp, ep, nil, nil) + return resp, err +} + func (conn *Conn) Login(ctx context.Context, options arvados.LoginOptions) (arvados.LoginResponse, error) { ep := arvados.EndpointLogin var resp arvados.LoginResponse @@ -194,6 +220,41 @@ func (conn *Conn) relativeToBaseURL(location string) string { return location } +func (conn *Conn) AuthorizedKeyCreate(ctx context.Context, options arvados.CreateOptions) (arvados.AuthorizedKey, error) { + ep := arvados.EndpointAuthorizedKeyCreate + var resp arvados.AuthorizedKey + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) AuthorizedKeyUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.AuthorizedKey, error) { + ep := arvados.EndpointAuthorizedKeyUpdate + var resp arvados.AuthorizedKey + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) AuthorizedKeyGet(ctx context.Context, options arvados.GetOptions) (arvados.AuthorizedKey, error) { + ep := arvados.EndpointAuthorizedKeyGet + var resp arvados.AuthorizedKey + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) AuthorizedKeyList(ctx context.Context, options arvados.ListOptions) (arvados.AuthorizedKeyList, error) { + ep := arvados.EndpointAuthorizedKeyList + var resp arvados.AuthorizedKeyList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) AuthorizedKeyDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.AuthorizedKey, error) { + ep := arvados.EndpointAuthorizedKeyDelete + var resp arvados.AuthorizedKey + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + func (conn *Conn) CollectionCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Collection, error) { ep := arvados.EndpointCollectionCreate var resp arvados.Collection @@ -271,6 +332,13 @@ func (conn *Conn) ContainerUpdate(ctx context.Context, options arvados.UpdateOpt return resp, err } +func (conn *Conn) ContainerPriorityUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Container, error) { + ep := arvados.EndpointContainerPriorityUpdate + var resp arvados.Container + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + func (conn *Conn) ContainerGet(ctx context.Context, options arvados.GetOptions) (arvados.Container, error) { ep := arvados.EndpointContainerGet var resp arvados.Container @@ -306,10 +374,52 @@ func (conn *Conn) ContainerUnlock(ctx context.Context, options arvados.GetOption return resp, err } +func (conn *Conn) ContainerLog(ctx context.Context, options arvados.ContainerLogOptions) (resp http.Handler, err error) { + proxy := &httputil.ReverseProxy{ + Transport: conn.httpClient.Transport, + Director: func(r *http.Request) { + u := conn.baseURL + u.Path = r.URL.Path + u.RawQuery = fmt.Sprintf("no_forward=%v", options.NoForward) + r.URL = &u + }, + } + return proxy, nil +} + // ContainerSSH returns a connection to the out-of-band SSH server for // a running container. If the returned error is nil, the caller is // responsible for closing sshconn.Conn. -func (conn *Conn) ContainerSSH(ctx context.Context, options arvados.ContainerSSHOptions) (sshconn arvados.ContainerSSHConnection, err error) { +func (conn *Conn) ContainerSSH(ctx context.Context, options arvados.ContainerSSHOptions) (sshconn arvados.ConnectionResponse, err error) { + u, err := conn.baseURL.Parse("/" + strings.Replace(arvados.EndpointContainerSSHCompat.Path, "{uuid}", options.UUID, -1)) + if err != nil { + err = fmt.Errorf("url.Parse: %w", err) + return + } + return conn.socket(ctx, u, "ssh", url.Values{ + "detach_keys": {options.DetachKeys}, + "login_username": {options.LoginUsername}, + "no_forward": {fmt.Sprintf("%v", options.NoForward)}, + }) +} + +// ContainerGatewayTunnel returns a connection to a yamux session on +// the controller. The caller should connect the returned resp.Conn to +// a client-side yamux session. +func (conn *Conn) ContainerGatewayTunnel(ctx context.Context, options arvados.ContainerGatewayTunnelOptions) (tunnelconn arvados.ConnectionResponse, err error) { + u, err := conn.baseURL.Parse("/" + strings.Replace(arvados.EndpointContainerGatewayTunnelCompat.Path, "{uuid}", options.UUID, -1)) + if err != nil { + err = fmt.Errorf("url.Parse: %w", err) + return + } + return conn.socket(ctx, u, "tunnel", url.Values{ + "auth_secret": {options.AuthSecret}, + }) +} + +// socket sets up a socket using the specified API endpoint and +// upgrade header. +func (conn *Conn) socket(ctx context.Context, u *url.URL, upgradeHeader string, postform url.Values) (connresp arvados.ConnectionResponse, err error) { addr := conn.baseURL.Host if strings.Index(addr, ":") < 1 || (strings.Contains(addr, "::") && addr[0] != '[') { // hostname or ::1 or 1::1 @@ -321,8 +431,7 @@ func (conn *Conn) ContainerSSH(ctx context.Context, options arvados.ContainerSSH } netconn, err := tls.Dial("tcp", addr, &tls.Config{InsecureSkipVerify: insecure}) if err != nil { - err = fmt.Errorf("tls.Dial: %w", err) - return + return connresp, fmt.Errorf("tls.Dial: %w", err) } defer func() { if err != nil { @@ -332,36 +441,30 @@ func (conn *Conn) ContainerSSH(ctx context.Context, options arvados.ContainerSSH bufr := bufio.NewReader(netconn) bufw := bufio.NewWriter(netconn) - u, err := conn.baseURL.Parse("/" + strings.Replace(arvados.EndpointContainerSSH.Path, "{uuid}", options.UUID, -1)) - if err != nil { - err = fmt.Errorf("tls.Dial: %w", err) - return - } - u.RawQuery = url.Values{ - "detach_keys": {options.DetachKeys}, - "login_username": {options.LoginUsername}, - }.Encode() tokens, err := conn.tokenProvider(ctx) if err != nil { - return + return connresp, err } else if len(tokens) < 1 { - err = httpserver.ErrorWithStatus(errors.New("unauthorized"), http.StatusUnauthorized) - return + return connresp, httpserver.ErrorWithStatus(errors.New("unauthorized"), http.StatusUnauthorized) } - bufw.WriteString("GET " + u.String() + " HTTP/1.1\r\n") + postdata := postform.Encode() + bufw.WriteString("POST " + u.String() + " HTTP/1.1\r\n") bufw.WriteString("Authorization: Bearer " + tokens[0] + "\r\n") bufw.WriteString("Host: " + u.Host + "\r\n") - bufw.WriteString("Upgrade: ssh\r\n") + bufw.WriteString("Upgrade: " + upgradeHeader + "\r\n") + bufw.WriteString("Content-Type: application/x-www-form-urlencoded\r\n") + fmt.Fprintf(bufw, "Content-Length: %d\r\n", len(postdata)) bufw.WriteString("\r\n") + bufw.WriteString(postdata) bufw.Flush() - resp, err := http.ReadResponse(bufr, &http.Request{Method: "GET"}) + resp, err := http.ReadResponse(bufr, &http.Request{Method: "POST"}) if err != nil { - err = fmt.Errorf("http.ReadResponse: %w", err) - return + return connresp, fmt.Errorf("http.ReadResponse: %w", err) } + defer resp.Body.Close() if resp.StatusCode != http.StatusSwitchingProtocols { - defer resp.Body.Close() - body, _ := ioutil.ReadAll(resp.Body) + ctxlog.FromContext(ctx).Infof("rpc.Conn.socket: server %s did not switch protocols, got status %s", u.String(), resp.Status) + body, _ := ioutil.ReadAll(io.LimitReader(resp.Body, 10000)) var message string var errDoc httpserver.ErrorResponse if err := json.Unmarshal(body, &errDoc); err == nil { @@ -369,17 +472,16 @@ func (conn *Conn) ContainerSSH(ctx context.Context, options arvados.ContainerSSH } else { message = fmt.Sprintf("%q", body) } - err = fmt.Errorf("server did not provide a tunnel: %s (HTTP %d)", message, resp.StatusCode) - return + return connresp, fmt.Errorf("server did not provide a tunnel: %s: %s", resp.Status, message) } - if strings.ToLower(resp.Header.Get("Upgrade")) != "ssh" || + if strings.ToLower(resp.Header.Get("Upgrade")) != upgradeHeader || strings.ToLower(resp.Header.Get("Connection")) != "upgrade" { - err = fmt.Errorf("bad response from server: Upgrade %q Connection %q", resp.Header.Get("Upgrade"), resp.Header.Get("Connection")) - return + return connresp, fmt.Errorf("bad response from server: Upgrade %q Connection %q", resp.Header.Get("Upgrade"), resp.Header.Get("Connection")) } - sshconn.Conn = netconn - sshconn.Bufrw = &bufio.ReadWriter{Reader: bufr, Writer: bufw} - return + connresp.Conn = netconn + connresp.Bufrw = &bufio.ReadWriter{Reader: bufr, Writer: bufw} + connresp.Header = resp.Header + return connresp, nil } func (conn *Conn) ContainerRequestCreate(ctx context.Context, options arvados.CreateOptions) (arvados.ContainerRequest, error) { @@ -446,49 +548,9 @@ func (conn *Conn) GroupList(ctx context.Context, options arvados.ListOptions) (a } func (conn *Conn) GroupContents(ctx context.Context, options arvados.GroupContentsOptions) (arvados.ObjectList, error) { - // The requested UUID can be a user (virtual home project), which we just pass on to - // the API server. - if strings.Index(options.UUID, "j7d0g") != 6 { - ep := arvados.EndpointGroupContents - var resp arvados.ObjectList - err := conn.requestAndDecode(ctx, &resp, ep, nil, options) - return resp, err - } - - log := ctxlog.FromContext(ctx) - var resp arvados.ObjectList - - // Get the group object - epGet := arvados.EndpointGroupGet - var respGroup arvados.Group - err := conn.requestAndDecode(ctx, &respGroup, epGet, nil, options) - if err != nil { - return resp, err - } - - // If the group has groupClass 'filter', apply the filters before getting the contents. - if respGroup.GroupClass == "filter" { - if filters, ok := respGroup.Properties["filters"]; ok { - for _, f := range filters.([]interface{}) { - // f is supposed to be a []string - tmp, ok2 := f.([]interface{}) - if !ok2 || len(tmp) < 3 { - log.Warnf("filter unparsable: %T, %+v, original field: %T, %+v\n", tmp, tmp, f, f) - continue - } - var filter arvados.Filter - filter.Attr = tmp[0].(string) - filter.Operator = tmp[1].(string) - filter.Operand = tmp[2] - options.Filters = append(options.Filters, filter) - } - } - // Use the generic /groups/contents endpoint for filter groups - options.UUID = "" - } - ep := arvados.EndpointGroupContents - err = conn.requestAndDecode(ctx, &resp, ep, nil, options) + var resp arvados.ObjectList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } @@ -520,6 +582,76 @@ func (conn *Conn) GroupUntrash(ctx context.Context, options arvados.UntrashOptio return resp, err } +func (conn *Conn) LinkCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Link, error) { + ep := arvados.EndpointLinkCreate + var resp arvados.Link + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LinkUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Link, error) { + ep := arvados.EndpointLinkUpdate + var resp arvados.Link + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LinkGet(ctx context.Context, options arvados.GetOptions) (arvados.Link, error) { + ep := arvados.EndpointLinkGet + var resp arvados.Link + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LinkList(ctx context.Context, options arvados.ListOptions) (arvados.LinkList, error) { + ep := arvados.EndpointLinkList + var resp arvados.LinkList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LinkDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Link, error) { + ep := arvados.EndpointLinkDelete + var resp arvados.Link + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LogCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Log, error) { + ep := arvados.EndpointLogCreate + var resp arvados.Log + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LogUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.Log, error) { + ep := arvados.EndpointLogUpdate + var resp arvados.Log + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LogGet(ctx context.Context, options arvados.GetOptions) (arvados.Log, error) { + ep := arvados.EndpointLogGet + var resp arvados.Log + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LogList(ctx context.Context, options arvados.ListOptions) (arvados.LogList, error) { + ep := arvados.EndpointLogList + var resp arvados.LogList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + +func (conn *Conn) LogDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.Log, error) { + ep := arvados.EndpointLogDelete + var resp arvados.Log + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + func (conn *Conn) SpecimenCreate(ctx context.Context, options arvados.CreateOptions) (arvados.Specimen, error) { ep := arvados.EndpointSpecimenCreate var resp arvados.Specimen @@ -555,6 +687,13 @@ func (conn *Conn) SpecimenDelete(ctx context.Context, options arvados.DeleteOpti return resp, err } +func (conn *Conn) SysTrashSweep(ctx context.Context, options struct{}) (struct{}, error) { + ep := arvados.EndpointSysTrashSweep + var resp struct{} + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} + func (conn *Conn) UserCreate(ctx context.Context, options arvados.CreateOptions) (arvados.User, error) { ep := arvados.EndpointUserCreate var resp arvados.User @@ -567,12 +706,6 @@ func (conn *Conn) UserUpdate(ctx context.Context, options arvados.UpdateOptions) err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } -func (conn *Conn) UserUpdateUUID(ctx context.Context, options arvados.UpdateUUIDOptions) (arvados.User, error) { - ep := arvados.EndpointUserUpdateUUID - var resp arvados.User - err := conn.requestAndDecode(ctx, &resp, ep, nil, options) - return resp, err -} func (conn *Conn) UserMerge(ctx context.Context, options arvados.UserMergeOptions) (arvados.User, error) { ep := arvados.EndpointUserMerge var resp arvados.User @@ -634,6 +767,36 @@ func (conn *Conn) APIClientAuthorizationCurrent(ctx context.Context, options arv err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } +func (conn *Conn) APIClientAuthorizationCreate(ctx context.Context, options arvados.CreateOptions) (arvados.APIClientAuthorization, error) { + ep := arvados.EndpointAPIClientAuthorizationCreate + var resp arvados.APIClientAuthorization + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} +func (conn *Conn) APIClientAuthorizationUpdate(ctx context.Context, options arvados.UpdateOptions) (arvados.APIClientAuthorization, error) { + ep := arvados.EndpointAPIClientAuthorizationUpdate + var resp arvados.APIClientAuthorization + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} +func (conn *Conn) APIClientAuthorizationDelete(ctx context.Context, options arvados.DeleteOptions) (arvados.APIClientAuthorization, error) { + ep := arvados.EndpointAPIClientAuthorizationDelete + var resp arvados.APIClientAuthorization + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} +func (conn *Conn) APIClientAuthorizationList(ctx context.Context, options arvados.ListOptions) (arvados.APIClientAuthorizationList, error) { + ep := arvados.EndpointAPIClientAuthorizationList + var resp arvados.APIClientAuthorizationList + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} +func (conn *Conn) APIClientAuthorizationGet(ctx context.Context, options arvados.GetOptions) (arvados.APIClientAuthorization, error) { + ep := arvados.EndpointAPIClientAuthorizationGet + var resp arvados.APIClientAuthorization + err := conn.requestAndDecode(ctx, &resp, ep, nil, options) + return resp, err +} type UserSessionAuthInfo struct { UserUUID string `json:"user_uuid"` @@ -670,3 +833,26 @@ func (conn *Conn) UserAuthenticate(ctx context.Context, options arvados.UserAuth err := conn.requestAndDecode(ctx, &resp, ep, nil, options) return resp, err } + +// httpStatusError is an error with an HTTP status code that can be +// propagated by lib/controller/router, etc. +type httpStatusError interface { + error + HTTPStatus() int +} + +// wrappedHTTPStatusError is used to augment/replace an error message +// while preserving the HTTP status code indicated by the original +// error. +type wrappedHTTPStatusError struct { + httpStatusError + message string +} + +func wrapHTTPStatusError(err httpStatusError, message string) httpStatusError { + return wrappedHTTPStatusError{err, message} +} + +func (err wrappedHTTPStatusError) Error() string { + return err.message +}