LogDog: Use datastore config cache when available.

logdog/server/service/service.go

 // Copyright 2016 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package service
 
 import (
         "flag"
         "fmt"
         "net/http"
         "net/url"
         "os"
         "os/signal"
         "path/filepath"
         "runtime/pprof"
         "sync/atomic"
         "time"
 
+        "github.com/luci/luci-go/appengine/gaesettings"
         "github.com/luci/luci-go/client/authcli"
         "github.com/luci/luci-go/common/auth"
         "github.com/luci/luci-go/common/clock/clockflag"
         "github.com/luci/luci-go/common/config/impl/filesystem"
         "github.com/luci/luci-go/common/data/caching/proccache"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/gcloud/gs"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/logging/gologger"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/tsmon"
         "github.com/luci/luci-go/common/tsmon/target"
         "github.com/luci/luci-go/grpc/prpc"
         "github.com/luci/luci-go/logdog/api/config/svcconfig"
         "github.com/luci/luci-go/logdog/api/endpoints/coordinator/services/v1"
         "github.com/luci/luci-go/logdog/common/storage"
         "github.com/luci/luci-go/logdog/common/storage/bigtable"
         "github.com/luci/luci-go/logdog/server/retryServicesClient"
         "github.com/luci/luci-go/logdog/server/service/config"
         "github.com/luci/luci-go/luci_config/common/cfgtypes"
         "github.com/luci/luci-go/luci_config/server/cfgclient"
         "github.com/luci/luci-go/luci_config/server/cfgclient/backend/client"
         "github.com/luci/luci-go/luci_config/server/cfgclient/backend/testconfig"
         "github.com/luci/luci-go/luci_config/server/cfgclient/textproto"
+        "github.com/luci/luci-go/server/settings"
+
+        "github.com/luci/gae/impl/cloud"
 
         "cloud.google.com/go/compute/metadata"
+        "cloud.google.com/go/datastore"
         "golang.org/x/net/context"
         "golang.org/x/oauth2"
         "google.golang.org/api/option"
 )
 
 var (
         // ErrInvalidConfig is an error that is returned when the supplied
         // configuration is invalid.
         ErrInvalidConfig = errors.New("invalid configuration")
 
@@ skipping 30 matching lines @@
         authFlags    authcli.Flags
         tsMonFlags   tsmon.Flags
 
         coordinatorHost           string
         coordinatorInsecure       bool
         storageCredentialJSONPath string
         cpuProfilePath            string
         heapProfilePath           string
 
         // onGCE is true if we're on GCE. We probe this once during Run.
-        onGCE bool
+        onGCE        bool
+        hasDatastore bool
 
         // killCheckInterval is the amount of time in between service configuration
         // checks. If set, this service will periodically reload its service
         // configuration. If that configuration has changed, the service will kill
         // itself.
         //
         // Since, in production, this is running under an execution harness such as
         // Kubernetes, the service will restart and load the new configuration. This
         // is easier than implementing in-process configuration updating.
         killCheckInterval clockflag.Duration
@@ skipping 25 matching lines @@
 
         rc := 0
         if err := s.runImpl(c, f); err != nil {
                 log.WithError(err).Errorf(c, "Application exiting with error.")
                 rc = 1
         }
         os.Exit(rc)
 }
 
 func (s *Service) runImpl(c context.Context, f func(context.Context) error) error {
+        // Probe our environment for default values.
+        s.probeGCEEnvironment(c)
+
+        // Install service flags and parse.
         s.addFlags(c, &s.Flags)
         if err := s.Flags.Parse(os.Args[1:]); err != nil {
                 log.WithError(err).Errorf(c, "Failed to parse command-line.")
                 return err
         }
 
         // Install logging configuration.
         c = s.loggingFlags.Set(c)
 
         if p := s.cpuProfilePath; p != "" {
@@ skipping 25 matching lines @@
 
                         if err := pprof.WriteHeapProfile(fd); err != nil {
                                 log.Fields{
                                         log.ErrorKey: err,
                                         "path":       p,
                                 }.Warningf(c, "Failed to write heap profile.")
                         }
                 }()
         }
 
+        // Cancel our Context after we're done our run loop.
+        c, cancelFunc := context.WithCancel(c)
+        defer cancelFunc()
+
         // Validate the runtime environment.
         if s.serviceID == "" {
                 return errors.New("no service ID was configured (-service-id)")
         }
 
+        // Install a cloud datastore client. This is non-fatal if it fails.
+        dsClient, err := s.initDatastoreClient(c)
+        if err == nil {
+                defer dsClient.Close()
+
+                ccfg := cloud.Config{
+                        DS: dsClient,
+                }
+                c = ccfg.Use(c)
+                c = settings.Use(c, settings.New(gaesettings.Storage{}))
+
+                s.hasDatastore = true
+                log.Debugf(c, "Enabled cloud datastore access.")
+        } else {
+                log.WithError(err).Warningf(c, "Failed to create cloud datastore client.")
+        }
+
         // Install a process-wide cache.
         c = proccache.Use(c, &proccache.Cache{})
 
         // Configure our signal handler. It will listen for terminating signals and
         // issue a shutdown signal if one is received.
         signalC := make(chan os.Signal)
         go func(c context.Context) {
                 hasShutdownAlready := false
                 for sig := range signalC {
                         if !hasShutdownAlready {
@@ skipping 17 matching lines @@
 
         // Initialize our tsmon library.
         c = tsmon.WithState(c, tsmon.NewState())
 
         if err := tsmon.InitializeFromFlags(c, &s.tsMonFlags); err != nil {
                 log.WithError(err).Warningf(c, "Failed to initialize monitoring; will continue without metrics.")
         }
         defer tsmon.Shutdown(c)
 
         // Initialize our Client instantiations.
-        var err error
         if s.coord, err = s.initCoordinatorClient(c); err != nil {
                 log.WithError(err).Errorf(c, "Failed to setup Coordinator client.")
                 return err
         }
 
         // Initialize and install our config service and caching layers, and load our
         // initial service config.
         if err := s.initConfig(&c); err != nil {
                 log.WithError(err).Errorf(c, "Failed to setup configuration.")
                 return err
@@ skipping 36 matching lines @@
                 "If supplied, enable CPU profiling and write the profile here.")
         fs.Var(&s.killCheckInterval, "config-kill-interval",
                 "If non-zero, poll for configuration changes and kill the application if one is detected.")
         fs.StringVar(&s.testConfigFilePath, "test-config-file-path", "",
                 "(Testing) If set, load configuration from a local filesystem rooted here.")
 }
 
 // probeGCEEnvironment fills in any parameters that can be probed from Google
 // Compute Engine metadata.
 //
-// If we're not running on GCE, this will return nil. An error will only be
-// returned if an operation that is expected to work fails.
-func (s *Service) probeGCEEnvironment(c context.Context) error {
+// If we're not running on GCE, this will do nothing. It is non-fatal if any
+// given GCE field fails to be probed.
+func (s *Service) probeGCEEnvironment(c context.Context) {
         s.onGCE = metadata.OnGCE()
         if !s.onGCE {
-                return nil
+                return
         }
 
         // Determine our service ID from metadata. The service ID will equal the cloud
         // project ID.
         if s.serviceID == "" {
                 var err error
                 if s.serviceID, err = metadata.ProjectID(); err != nil {
-                        log.WithError(err).Errorf(c, "Failed to probe GCE project ID.")
-                        return err
+                        log.WithError(err).Warningf(c, "Failed to probe GCE project ID.")
                 }
         }
-        return nil
+}
+
+func (s *Service) initDatastoreClient(c context.Context) (*datastore.Client, error) {
+        // Initialize Storage authentication.
+        tokenSource, err := s.TokenSource(c, func(o *auth.Options) {
+                o.Scopes = []string{datastore.ScopeDatastore}
+        })
+        if err != nil {
+                log.WithError(err).Errorf(c, "Failed to create datastore TokenSource.")
+                return nil, err
+        }
+
+        return datastore.NewClient(c, s.serviceID,
+                option.WithTokenSource(tokenSource))
 }
 
 func (s *Service) initCoordinatorClient(c context.Context) (logdog.ServicesClient, error) {
         if s.coordinatorHost == "" {
                 log.Errorf(c, "Missing Coordinator URL (-coordinator).")
                 return nil, ErrInvalidConfig
         }
 
         httpClient, err := s.AuthenticatedClient(c, func(o *auth.Options) {
                 o.Scopes = CoordinatorScopes
@@ skipping 15 matching lines @@
         sc := logdog.NewServicesPRPCClient(&prpcClient)
 
         // Wrap the resulting client in a retry harness.
         return retryServicesClient.New(sc, nil), nil
 }
 
 func (s *Service) initConfig(c *context.Context) error {
         // Set up our in-memory config object cache.
         s.configCache.Lifetime = projectConfigCacheDuration
 
+        // Start to build our backend caching options.
+        opts := config.CacheOptions{
+                CacheExpiration: projectConfigCacheDuration,
+        }
+
         // If a testConfigFilePath was specified, use a mock configuration service
         // that loads from a local file.
         var p client.Provider
         if s.testConfigFilePath == "" {
                 ccfg, err := s.coord.GetConfig(*c, &google.Empty{})
                 if err != nil {
                         return err
                 }
 
                 // Determine our config service host.
@@ skipping 15 matching lines @@
                 if ccfg.ConfigSet == "" {
                         return errors.New("coordinator does not specify a config set")
                 }
 
                 log.Fields{
                         "host": host,
                 }.Debugf(*c, "Using remote configuration service client.")
                 p = &client.RemoteProvider{
                         Host: host,
                 }
+
+                // If using a remote config provider, enable datastore access and caching.
+                opts.DatastoreCacheAvailable = s.hasDatastore
         } else {
                 // Test / Local: use filesystem config path.
                 ci, err := filesystem.New(s.testConfigFilePath)
                 if err != nil {
                         return err
                 }
                 p = &testconfig.Provider{Base: ci}
         }
 
         // Add config caching layers.
-        opts := config.CacheOptions{
-                CacheExpiration: projectConfigCacheDuration,
-        }
         *c = opts.WrapBackend(*c, &client.Backend{
                 Provider: p,
         })
 
         // Load our service configuration.
         var meta cfgclient.Meta
         cset, path := s.ServiceConfigPath()
         if err := cfgclient.Get(*c, cfgclient.AsService, cset, path, textproto.Message(&s.serviceConfig), &meta); err != nil {
                 return errors.Annotate(err).Reason("failed to load service config").Err()
         }
@@ skipping 183 matching lines @@
 //
 // An optional permutation function can be provided to modify those Options
 // before the Authenticator is created.
 func (s *Service) TokenSource(c context.Context, f func(o *auth.Options)) (oauth2.TokenSource, error) {
         a, err := s.Authenticator(c, f)
         if err != nil {
                 return nil, err
         }
         return a.TokenSource()
 }