lib/controller/localdb/conn.go

   1 // Copyright (C) The Arvados Authors. All rights reserved.
   2 //
   3 // SPDX-License-Identifier: AGPL-3.0
   4
   5 package localdb
   6
   7 import (
   8         "context"
   9         "encoding/json"
  10         "fmt"
  11         "net/http"
  12         "os"
  13         "strings"
  14         "sync"
  15         "time"
  16
  17         "git.arvados.org/arvados.git/lib/controller/railsproxy"
  18         "git.arvados.org/arvados.git/lib/controller/rpc"
  19         "git.arvados.org/arvados.git/sdk/go/arvados"
  20         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  21         "git.arvados.org/arvados.git/sdk/go/httpserver"
  22         "github.com/hashicorp/yamux"
  23         "github.com/sirupsen/logrus"
  24 )
  25
  26 type railsProxy = rpc.Conn
  27
  28 type Conn struct {
  29         cluster                    *arvados.Cluster
  30         *railsProxy                // handles API methods that aren't defined on Conn itself
  31         vocabularyCache            *arvados.Vocabulary
  32         vocabularyFileModTime      time.Time
  33         lastVocabularyRefreshCheck time.Time
  34         lastVocabularyError        error
  35         loginController
  36         gwTunnels     map[string]*yamux.Session
  37         gwTunnelsLock sync.Mutex
  38 }
  39
  40 func NewConn(cluster *arvados.Cluster) *Conn {
  41         railsProxy := railsproxy.NewConn(cluster)
  42         railsProxy.RedactHostInErrors = true
  43         conn := Conn{
  44                 cluster:    cluster,
  45                 railsProxy: railsProxy,
  46         }
  47         conn.loginController = chooseLoginController(cluster, &conn)
  48         return &conn
  49 }
  50
  51 func (conn *Conn) checkProperties(ctx context.Context, properties interface{}) error {
  52         if properties == nil {
  53                 return nil
  54         }
  55         var props map[string]interface{}
  56         switch properties := properties.(type) {
  57         case string:
  58                 err := json.Unmarshal([]byte(properties), &props)
  59                 if err != nil {
  60                         return err
  61                 }
  62         case map[string]interface{}:
  63                 props = properties
  64         default:
  65                 return fmt.Errorf("unexpected properties type %T", properties)
  66         }
  67         voc, err := conn.VocabularyGet(ctx)
  68         if err != nil {
  69                 return err
  70         }
  71         err = voc.Check(props)
  72         if err != nil {
  73                 return httpErrorf(http.StatusBadRequest, voc.Check(props).Error())
  74         }
  75         return nil
  76 }
  77
  78 func (conn *Conn) maybeRefreshVocabularyCache(logger logrus.FieldLogger) error {
  79         if conn.lastVocabularyRefreshCheck.Add(time.Second).After(time.Now()) {
  80                 // Throttle the access to disk to at most once per second.
  81                 return nil
  82         }
  83         conn.lastVocabularyRefreshCheck = time.Now()
  84         fi, err := os.Stat(conn.cluster.API.VocabularyPath)
  85         if err != nil {
  86                 err = fmt.Errorf("couldn't stat vocabulary file %q: %v", conn.cluster.API.VocabularyPath, err)
  87                 conn.lastVocabularyError = err
  88                 return err
  89         }
  90         if fi.ModTime().After(conn.vocabularyFileModTime) {
  91                 err = conn.loadVocabularyFile()
  92                 if err != nil {
  93                         conn.lastVocabularyError = err
  94                         return err
  95                 }
  96                 conn.vocabularyFileModTime = fi.ModTime()
  97                 conn.lastVocabularyError = nil
  98                 logger.Info("vocabulary file reloaded successfully")
  99         }
 100         return nil
 101 }
 102
 103 func (conn *Conn) loadVocabularyFile() error {
 104         vf, err := os.ReadFile(conn.cluster.API.VocabularyPath)
 105         if err != nil {
 106                 return fmt.Errorf("while reading the vocabulary file: %v", err)
 107         }
 108         mk := make([]string, 0, len(conn.cluster.Collections.ManagedProperties))
 109         for k := range conn.cluster.Collections.ManagedProperties {
 110                 mk = append(mk, k)
 111         }
 112         voc, err := arvados.NewVocabulary(vf, mk)
 113         if err != nil {
 114                 return fmt.Errorf("while loading vocabulary file %q: %s", conn.cluster.API.VocabularyPath, err)
 115         }
 116         conn.vocabularyCache = voc
 117         return nil
 118 }
 119
 120 // LastVocabularyError returns the last error encountered while loading the
 121 // vocabulary file.
 122 // Implements health.Func
 123 func (conn *Conn) LastVocabularyError() error {
 124         conn.maybeRefreshVocabularyCache(ctxlog.FromContext(context.Background()))
 125         return conn.lastVocabularyError
 126 }
 127
 128 // VocabularyGet refreshes the vocabulary cache if necessary and returns it.
 129 func (conn *Conn) VocabularyGet(ctx context.Context) (arvados.Vocabulary, error) {
 130         if conn.cluster.API.VocabularyPath == "" {
 131                 return arvados.Vocabulary{
 132                         Tags: map[string]arvados.VocabularyTag{},
 133                 }, nil
 134         }
 135         logger := ctxlog.FromContext(ctx)
 136         if conn.vocabularyCache == nil {
 137                 // Initial load of vocabulary file.
 138                 err := conn.loadVocabularyFile()
 139                 if err != nil {
 140                         logger.WithError(err).Error("error loading vocabulary file")
 141                         return arvados.Vocabulary{}, err
 142                 }
 143         }
 144         err := conn.maybeRefreshVocabularyCache(logger)
 145         if err != nil {
 146                 logger.WithError(err).Error("error reloading vocabulary file - ignoring")
 147         }
 148         return *conn.vocabularyCache, nil
 149 }
 150
 151 // Logout handles the logout of conn giving to the appropriate loginController
 152 func (conn *Conn) Logout(ctx context.Context, opts arvados.LogoutOptions) (arvados.LogoutResponse, error) {
 153         return conn.loginController.Logout(ctx, opts)
 154 }
 155
 156 // Login handles the login of conn giving to the appropriate loginController
 157 func (conn *Conn) Login(ctx context.Context, opts arvados.LoginOptions) (arvados.LoginResponse, error) {
 158         return conn.loginController.Login(ctx, opts)
 159 }
 160
 161 // UserAuthenticate handles the User Authentication of conn giving to the appropriate loginController
 162 func (conn *Conn) UserAuthenticate(ctx context.Context, opts arvados.UserAuthenticateOptions) (arvados.APIClientAuthorization, error) {
 163         return conn.loginController.UserAuthenticate(ctx, opts)
 164 }
 165
 166 func (conn *Conn) GroupContents(ctx context.Context, options arvados.GroupContentsOptions) (arvados.ObjectList, error) {
 167         // The requested UUID can be a user (virtual home project), which we just pass on to
 168         // the API server.
 169         if strings.Index(options.UUID, "-j7d0g-") != 5 {
 170                 return conn.railsProxy.GroupContents(ctx, options)
 171         }
 172
 173         var resp arvados.ObjectList
 174
 175         // Get the group object
 176         respGroup, err := conn.GroupGet(ctx, arvados.GetOptions{UUID: options.UUID})
 177         if err != nil {
 178                 return resp, err
 179         }
 180
 181         // If the group has groupClass 'filter', apply the filters before getting the contents.
 182         if respGroup.GroupClass == "filter" {
 183                 if filters, ok := respGroup.Properties["filters"].([]interface{}); ok {
 184                         for _, f := range filters {
 185                                 // f is supposed to be a []string
 186                                 tmp, ok2 := f.([]interface{})
 187                                 if !ok2 || len(tmp) < 3 {
 188                                         return resp, fmt.Errorf("filter unparsable: %T, %+v, original field: %T, %+v\n", tmp, tmp, f, f)
 189                                 }
 190                                 var filter arvados.Filter
 191                                 if attr, ok2 := tmp[0].(string); ok2 {
 192                                         filter.Attr = attr
 193                                 } else {
 194                                         return resp, fmt.Errorf("filter unparsable: attribute must be string: %T, %+v, filter: %T, %+v\n", tmp[0], tmp[0], f, f)
 195                                 }
 196                                 if operator, ok2 := tmp[1].(string); ok2 {
 197                                         filter.Operator = operator
 198                                 } else {
 199                                         return resp, fmt.Errorf("filter unparsable: operator must be string: %T, %+v, filter: %T, %+v\n", tmp[1], tmp[1], f, f)
 200                                 }
 201                                 filter.Operand = tmp[2]
 202                                 options.Filters = append(options.Filters, filter)
 203                         }
 204                 } else {
 205                         return resp, fmt.Errorf("filter unparsable: not an array\n")
 206                 }
 207                 // Use the generic /groups/contents endpoint for filter groups
 208                 options.UUID = ""
 209         }
 210
 211         return conn.railsProxy.GroupContents(ctx, options)
 212 }
 213
 214 func httpErrorf(code int, format string, args ...interface{}) error {
 215         return httpserver.ErrorWithStatus(fmt.Errorf(format, args...), code)
 216 }