LogDog: Use per-project settings for archival.

server/cmd/logdog_archivist/main.go

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package main
 
 import (
         "io"
         "time"
 
         "github.com/luci/luci-go/common/auth"
         "github.com/luci/luci-go/common/clock"
+        "github.com/luci/luci-go/common/config"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/gcloud/gs"
         "github.com/luci/luci-go/common/gcloud/pubsub"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/parallel"
+        "github.com/luci/luci-go/common/proto/logdog/svcconfig"
         "github.com/luci/luci-go/common/tsmon/distribution"
         "github.com/luci/luci-go/common/tsmon/field"
         "github.com/luci/luci-go/common/tsmon/metric"
         "github.com/luci/luci-go/server/internal/logdog/archivist"
         "github.com/luci/luci-go/server/internal/logdog/service"
         "golang.org/x/net/context"
         "google.golang.org/cloud"
         gcps "google.golang.org/cloud/pubsub"
 )
 
@@ skipping 26 matching lines @@
 func (a *application) runArchivist(c context.Context) error {
         cfg := a.Config()
 
         coordCfg, acfg := cfg.GetCoordinator(), cfg.GetArchivist()
         switch {
         case coordCfg == nil:
                 fallthrough
 
         case acfg == nil:
                 return errors.New("missing Archivist configuration")
-        case acfg.GsBase == "":
-                return errors.New("missing archive GS bucket")
         case acfg.GsStagingBase == "":
                 return errors.New("missing archive staging GS bucket")
         }
 
-        // Construct and validate our GS bases.
-        gsBase := gs.Path(acfg.GsBase)
-        if gsBase.Bucket() == "" {
-                log.Fields{
-                        "value": gsBase,
-                }.Errorf(c, "Google Storage base does not include a bucket name.")
-                return errors.New("invalid Google Storage base")
-        }
-
-        gsStagingBase := gs.Path(acfg.GsStagingBase)
-        if gsStagingBase.Bucket() == "" {
-                log.Fields{
-                        "value": gsStagingBase,
-                }.Errorf(c, "Google Storage staging base does not include a bucket name.")
-                return errors.New("invalid Google Storage staging base")
-        }
-
         // Initialize Pub/Sub client.
         //
         // We will initialize both an authenticated Client instance and an
         // authenticated Context, since we need the latter for raw ACK deadline
         // updates.
         taskSub := pubsub.Subscription(acfg.Subscription)
         if err := taskSub.Validate(); err != nil {
                 log.Fields{
                         log.ErrorKey: err,
                         "value":      taskSub,
@@ skipping 39 matching lines @@
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to get Google Storage client.")
                 return err
         }
         defer gsClient.Close()
 
         ar := archivist.Archivist{
                 Service:  a.Coordinator(),
                 Storage:  st,
                 GSClient: gsClient,
-
-                GSBase:           gsBase,
-                GSStagingBase:    gsStagingBase,
-                StreamIndexRange: int(acfg.IndexStreamRange),
-                PrefixIndexRange: int(acfg.IndexPrefixRange),
-                ByteRange:        int(acfg.IndexByteRange),
         }
 
         tasks := int(acfg.Tasks)
         if tasks <= 0 {
                 tasks = 1
         }
 
         log.Fields{
                 "subscription": taskSub,
                 "tasks":        tasks,
@@ skipping 70 matching lines @@
                                 clock.Sleep(c, subscriptionErrorDelay)
                                 continue
                         }
                 }
         }))
 
         log.Debugf(c, "Archivist finished.")
         return nil
 }
 
+// GetSettingsLoader is an archivist.SettingsLoader implementation that merges
+// global and project-specific settings.
+//
+// The resulting settings object will be verified by the Archivist.
+func (a *application) GetSettingsLoader(acfg *svcconfig.Archivist) archivist.SettingsLoader {
+        return func(c context.Context, proj config.ProjectName) (*archivist.Settings, error) {
+                // Load our Archivist-wide settings.
+                st := archivist.Settings{
+                        GSBase:        gs.Path(acfg.GsBase),
+                        GSStagingBase: gs.Path(acfg.GsStagingBase),

[nodir, 2016/05/19 15:56:29]

I assume this will be rebased to be use GsStagingBucket or something like that

[dnj (Google), 2016/05/19 16:34:51]

Yeah. This is weird, I did a "dependency upload" and it should have uploaded the
rebased version. I uploaded again; it is using that in the current patch set.

+
+                        IndexStreamRange: int(acfg.IndexStreamRange),
+                        IndexPrefixRange: int(acfg.IndexPrefixRange),
+                        IndexByteRange:   int(acfg.IndexByteRange),

[nodir, 2016/05/19 15:56:29]

I assume this will be rebased to be use GsStagingBucket and ArchiveIndexConfig
fields

[dnj (Google), 2016/05/19 16:34:51]

Acknowledged.

+                }
+
+                // Fold in our project-specific configuration, if valid.
+                pcfg, err := a.ProjectConfig(c, proj)
+                if err != nil {
+                        log.Fields{
+                                log.ErrorKey: err,
+                                "project":    proj,
+                        }.Errorf(c, "Failed to fetch project configuration.")
+                }

[nodir, 2016/05/19 15:56:29]

return err?

[dnj (Google), 2016/05/19 16:34:51]

er um ... yes.

+
+                // Fold project settings into loaded ones.
+                if pcfg.ArchiveGsBase != "" {
+                        st.GSBase = gs.Path(pcfg.ArchiveGsBase)
+                }
+                st.AlwaysCreateBinary = (acfg.AlwaysCreateBinary || pcfg.AlwaysCreateBinary)
+                if r := pcfg.ArchiveIndexStreamRange; r >= 0 {

[nodir, 2016/05/19 15:56:29]

I still don't get this obsession to save a field value to a variable. It is more
code and ~same speed (maybe more). I am under impression that you are doing this
only because Golang has this assignment-in-if feature, it is abuse

[dnj (Google), 2016/05/19 16:34:51]

In this case, I think it is less wordy and therefore more readable. I would not
be doing this if Go didn't have assignment-in-if, since it would be less
readable. I don't think this is abuse at all.

+                        st.IndexStreamRange = int(r)
+                }
+                if r := pcfg.ArchiveIndexPrefixRange; r >= 0 {
+                        st.IndexPrefixRange = int(r)
+                }
+                if r := pcfg.ArchiveIndexByteRange; r >= 0 {
+                        st.IndexByteRange = int(r)
+                }
+                return &st, nil
+        }
+}
+
 // Entry point.
 func main() {
         a := application{
                 Service: service.Service{
                         Name: "archivist",
                 },
         }
         a.Run(context.Background(), a.runArchivist)
 }