17944: Adds /_health/vocabulary health endpoint. Improves cache refreshing.

[arvados.git] / lib / controller / localdb / conn.go

diff --git a/lib/controller/localdb/conn.go b/lib/controller/localdb/conn.go
index 0fae35e7d36a0d2515e6dd97f70e3080aa0948a0..f515673154b60da2d1522371de0f98c6fb8c0e7d 100644 (file)
--- a/lib/controller/localdb/conn.go
+++ b/lib/controller/localdb/conn.go
@@ -8,24 +8,27 @@ import (
        "context"
        "encoding/json"
        "fmt"
+       "net/http"
        "os"
        "strings"
+       "time"
 
        "git.arvados.org/arvados.git/lib/controller/railsproxy"
        "git.arvados.org/arvados.git/lib/controller/rpc"
        "git.arvados.org/arvados.git/sdk/go/arvados"
        "git.arvados.org/arvados.git/sdk/go/ctxlog"
-       "github.com/fsnotify/fsnotify"
-       "github.com/sirupsen/logrus"
+       "git.arvados.org/arvados.git/sdk/go/httpserver"
 )
 
 type railsProxy = rpc.Conn
 
 type Conn struct {
-       cluster          *arvados.Cluster
-       *railsProxy      // handles API methods that aren't defined on Conn itself
-       vocabularyCache  *arvados.Vocabulary
-       reloadVocabulary bool
+       cluster                    *arvados.Cluster
+       *railsProxy                // handles API methods that aren't defined on Conn itself
+       vocabularyCache            *arvados.Vocabulary
+       vocabularyFileModTime      time.Time
+       lastVocabularyRefreshCheck time.Time
+       lastVocabularyError        error
        loginController
 }
 
@@ -60,46 +63,41 @@ func (conn *Conn) checkProperties(ctx context.Context, properties interface{}) e
        if err != nil {
                return err
        }
-       return voc.Check(props)
-}
-
-func watchVocabulary(logger logrus.FieldLogger, vocPath string, fn func()) {
-       watcher, err := fsnotify.NewWatcher()
+       err = voc.Check(props)
        if err != nil {
-               logger.WithError(err).Error("vocabulary fsnotify setup failed")
-               return
+               return httpErrorf(http.StatusBadRequest, voc.Check(props).Error())
        }
-       defer watcher.Close()
+       return nil
+}
 
-       err = watcher.Add(vocPath)
+func (conn *Conn) maybeRefreshVocabularyCache() error {
+       if conn.lastVocabularyRefreshCheck.Add(time.Second).After(time.Now()) {
+               // Throttle the access to disk to at most once per second.
+               return nil
+       }
+       conn.lastVocabularyRefreshCheck = time.Now()
+       fi, err := os.Stat(conn.cluster.API.VocabularyPath)
        if err != nil {
-               logger.WithError(err).Error("vocabulary file watcher failed")
-               return
+               err = fmt.Errorf("couldn't stat vocabulary file %q: %v", conn.cluster.API.VocabularyPath, err)
+               conn.lastVocabularyError = err
+               return err
        }
-
-       for {
-               select {
-               case err, ok := <-watcher.Errors:
-                       if !ok {
-                               return
-                       }
-                       logger.WithError(err).Warn("vocabulary file watcher error")
-               case _, ok := <-watcher.Events:
-                       if !ok {
-                               return
-                       }
-                       for len(watcher.Events) > 0 {
-                               <-watcher.Events
-                       }
-                       fn()
+       if fi.ModTime().After(conn.vocabularyFileModTime) {
+               err = conn.loadVocabularyFile()
+               if err != nil {
+                       conn.lastVocabularyError = err
+                       return err
                }
+               conn.vocabularyFileModTime = fi.ModTime()
+               conn.lastVocabularyError = nil
        }
+       return nil
 }
 
 func (conn *Conn) loadVocabularyFile() error {
        vf, err := os.ReadFile(conn.cluster.API.VocabularyPath)
        if err != nil {
-               return fmt.Errorf("couldn't read vocabulary file %q: %v", conn.cluster.API.VocabularyPath, err)
+               return fmt.Errorf("couldn't reading the vocabulary file: %v", err)
        }
        mk := make([]string, 0, len(conn.cluster.Collections.ManagedProperties))
        for k := range conn.cluster.Collections.ManagedProperties {
@@ -117,10 +115,19 @@ func (conn *Conn) loadVocabularyFile() error {
        return nil
 }
 
+// LastVocabularyError returns the last error encountered while loading the
+// vocabulary file.
+func (conn *Conn) LastVocabularyError() error {
+       conn.maybeRefreshVocabularyCache()
+       return conn.lastVocabularyError
+}
+
 // VocabularyGet refreshes the vocabulary cache if necessary and returns it.
 func (conn *Conn) VocabularyGet(ctx context.Context) (arvados.Vocabulary, error) {
        if conn.cluster.API.VocabularyPath == "" {
-               return arvados.Vocabulary{}, nil
+               return arvados.Vocabulary{
+                       Tags: map[string]arvados.VocabularyTag{},
+               }, nil
        }
        logger := ctxlog.FromContext(ctx)
        if conn.vocabularyCache == nil {
@@ -130,19 +137,12 @@ func (conn *Conn) VocabularyGet(ctx context.Context) (arvados.Vocabulary, error)
                        logger.WithError(err).Error("error loading vocabulary file")
                        return arvados.Vocabulary{}, err
                }
-               go watchVocabulary(logger, conn.cluster.API.VocabularyPath, func() {
-                       logger.Info("vocabulary file changed, it'll be reloaded next time it's needed")
-                       conn.reloadVocabulary = true
-               })
-       } else if conn.reloadVocabulary {
-               // Requested reload of vocabulary file.
-               conn.reloadVocabulary = false
-               err := conn.loadVocabularyFile()
-               if err != nil {
-                       logger.WithError(err).Error("error reloading vocabulary file - ignoring")
-               } else {
-                       logger.Info("vocabulary file reloaded successfully")
-               }
+       }
+       err := conn.maybeRefreshVocabularyCache()
+       if err != nil {
+               logger.WithError(err).Error("error reloading vocabulary file - ignoring")
+       } else {
+               logger.Info("vocabulary file reloaded successfully")
        }
        return *conn.vocabularyCache, nil
 }
@@ -209,3 +209,7 @@ func (conn *Conn) GroupContents(ctx context.Context, options arvados.GroupConten
 
        return conn.railsProxy.GroupContents(ctx, options)
 }
+
+func httpErrorf(code int, format string, args ...interface{}) error {
+       return httpserver.ErrorWithStatus(fmt.Errorf(format, args...), code)
+}