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         "sync"
  14         "time"
  15
  16         "git.arvados.org/arvados.git/lib/controller/railsproxy"
  17         "git.arvados.org/arvados.git/lib/controller/rpc"
  18         "git.arvados.org/arvados.git/sdk/go/arvados"
  19         "git.arvados.org/arvados.git/sdk/go/ctxlog"
  20         "git.arvados.org/arvados.git/sdk/go/httpserver"
  21         "github.com/hashicorp/yamux"
  22         "github.com/sirupsen/logrus"
  23 )
  24
  25 type railsProxy = rpc.Conn
  26
  27 type Conn struct {
  28         cluster                    *arvados.Cluster
  29         *railsProxy                // handles API methods that aren't defined on Conn itself
  30         vocabularyCache            *arvados.Vocabulary
  31         vocabularyFileModTime      time.Time
  32         lastVocabularyRefreshCheck time.Time
  33         lastVocabularyError        error
  34         loginController
  35         gwTunnels        map[string]*yamux.Session
  36         gwTunnelsLock    sync.Mutex
  37         activeUsers      map[string]bool
  38         activeUsersLock  sync.Mutex
  39         activeUsersReset time.Time
  40 }
  41
  42 func NewConn(cluster *arvados.Cluster) *Conn {
  43         railsProxy := railsproxy.NewConn(cluster)
  44         railsProxy.RedactHostInErrors = true
  45         conn := Conn{
  46                 cluster:    cluster,
  47                 railsProxy: railsProxy,
  48         }
  49         conn.loginController = chooseLoginController(cluster, &conn)
  50         return &conn
  51 }
  52
  53 func (conn *Conn) checkProperties(ctx context.Context, properties interface{}) error {
  54         if properties == nil {
  55                 return nil
  56         }
  57         var props map[string]interface{}
  58         switch properties := properties.(type) {
  59         case string:
  60                 err := json.Unmarshal([]byte(properties), &props)
  61                 if err != nil {
  62                         return err
  63                 }
  64         case map[string]interface{}:
  65                 props = properties
  66         default:
  67                 return fmt.Errorf("unexpected properties type %T", properties)
  68         }
  69         voc, err := conn.VocabularyGet(ctx)
  70         if err != nil {
  71                 return err
  72         }
  73         err = voc.Check(props)
  74         if err != nil {
  75                 return httpErrorf(http.StatusBadRequest, voc.Check(props).Error())
  76         }
  77         return nil
  78 }
  79
  80 func (conn *Conn) maybeRefreshVocabularyCache(logger logrus.FieldLogger) error {
  81         if conn.lastVocabularyRefreshCheck.Add(time.Second).After(time.Now()) {
  82                 // Throttle the access to disk to at most once per second.
  83                 return nil
  84         }
  85         conn.lastVocabularyRefreshCheck = time.Now()
  86         fi, err := os.Stat(conn.cluster.API.VocabularyPath)
  87         if err != nil {
  88                 err = fmt.Errorf("couldn't stat vocabulary file %q: %v", conn.cluster.API.VocabularyPath, err)
  89                 conn.lastVocabularyError = err
  90                 return err
  91         }
  92         if fi.ModTime().After(conn.vocabularyFileModTime) {
  93                 err = conn.loadVocabularyFile()
  94                 if err != nil {
  95                         conn.lastVocabularyError = err
  96                         return err
  97                 }
  98                 conn.vocabularyFileModTime = fi.ModTime()
  99                 conn.lastVocabularyError = nil
 100                 logger.Info("vocabulary file reloaded successfully")
 101         }
 102         return nil
 103 }
 104
 105 func (conn *Conn) loadVocabularyFile() error {
 106         vf, err := os.ReadFile(conn.cluster.API.VocabularyPath)
 107         if err != nil {
 108                 return fmt.Errorf("while reading the vocabulary file: %v", err)
 109         }
 110         mk := make([]string, 0, len(conn.cluster.Collections.ManagedProperties))
 111         for k := range conn.cluster.Collections.ManagedProperties {
 112                 mk = append(mk, k)
 113         }
 114         voc, err := arvados.NewVocabulary(vf, mk)
 115         if err != nil {
 116                 return fmt.Errorf("while loading vocabulary file %q: %s", conn.cluster.API.VocabularyPath, err)
 117         }
 118         conn.vocabularyCache = voc
 119         return nil
 120 }
 121
 122 // LastVocabularyError returns the last error encountered while loading the
 123 // vocabulary file.
 124 // Implements health.Func
 125 func (conn *Conn) LastVocabularyError() error {
 126         conn.maybeRefreshVocabularyCache(ctxlog.FromContext(context.Background()))
 127         return conn.lastVocabularyError
 128 }
 129
 130 // VocabularyGet refreshes the vocabulary cache if necessary and returns it.
 131 func (conn *Conn) VocabularyGet(ctx context.Context) (arvados.Vocabulary, error) {
 132         if conn.cluster.API.VocabularyPath == "" {
 133                 return arvados.Vocabulary{
 134                         Tags: map[string]arvados.VocabularyTag{},
 135                 }, nil
 136         }
 137         logger := ctxlog.FromContext(ctx)
 138         if conn.vocabularyCache == nil {
 139                 // Initial load of vocabulary file.
 140                 err := conn.loadVocabularyFile()
 141                 if err != nil {
 142                         logger.WithError(err).Error("error loading vocabulary file")
 143                         return arvados.Vocabulary{}, err
 144                 }
 145         }
 146         err := conn.maybeRefreshVocabularyCache(logger)
 147         if err != nil {
 148                 logger.WithError(err).Error("error reloading vocabulary file - ignoring")
 149         }
 150         return *conn.vocabularyCache, nil
 151 }
 152
 153 // Logout handles the logout of conn giving to the appropriate loginController
 154 func (conn *Conn) Logout(ctx context.Context, opts arvados.LogoutOptions) (arvados.LogoutResponse, error) {
 155         return conn.loginController.Logout(ctx, opts)
 156 }
 157
 158 // Login handles the login of conn giving to the appropriate loginController
 159 func (conn *Conn) Login(ctx context.Context, opts arvados.LoginOptions) (arvados.LoginResponse, error) {
 160         return conn.loginController.Login(ctx, opts)
 161 }
 162
 163 // UserAuthenticate handles the User Authentication of conn giving to the appropriate loginController
 164 func (conn *Conn) UserAuthenticate(ctx context.Context, opts arvados.UserAuthenticateOptions) (arvados.APIClientAuthorization, error) {
 165         return conn.loginController.UserAuthenticate(ctx, opts)
 166 }
 167
 168 func httpErrorf(code int, format string, args ...interface{}) error {
 169         return httpserver.ErrorWithStatus(fmt.Errorf(format, args...), code)
 170 }