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 (
         "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"
         gcps "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"
         "google.golang.org/cloud/pubsub"
 )
 
@@ skipping 25 matching lines @@
 // run is the main execution function.
 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")
+                return errors.New("missing required config: archivist")
         case acfg.GsStagingBucket == "":
-                return errors.New("missing archive staging GS bucket")
-        }
-
-        // Construct and validate our GS bases.
-        gsBase := gs.Path(acfg.GsStagingBucket)
-        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.GsStagingBucket).Concat("staging")
-        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")
+                return errors.New("missing required config: archivist.gs_staging_bucket")
         }
 
         // 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 := gcps.Subscription(acfg.Subscription)
         if err := taskSub.Validate(); err != nil {
                 log.Fields{
@@ skipping 38 matching lines @@
 
         // Initialize our Google Storage client.
         gsClient, err := a.GSClient(c)
         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,
-        }
-        if ic := acfg.ArchiveIndexConfig; ic != nil {
-                ar.StreamIndexRange = int(ic.StreamRange)
-                ar.PrefixIndexRange = int(ic.PrefixRange)
-                ar.ByteRange = int(ic.ByteRange)
+                Service:        a.Coordinator(),
+                SettingsLoader: a.GetSettingsLoader(acfg),
+                Storage:        st,
+                GSClient:       gsClient,
         }
 
         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 {
+        serviceID := a.ServiceID()
+
+        return func(c context.Context, proj config.ProjectName) (*archivist.Settings, error) {
+                // 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.")
+                        return nil, err
+                }
+
+                indexParam := func(get func(ic *svcconfig.ArchiveIndexConfig) int32) int {
+                        if ic := pcfg.ArchiveIndexConfig; ic != nil {
+                                if v := get(ic); v > 0 {
+                                        return int(v)
+                                }
+                        }
+
+                        if ic := acfg.ArchiveIndexConfig; ic != nil {
+                                if v := get(ic); v > 0 {
+                                        return int(v)
+                                }
+                        }
+
+                        return 0
+                }
+
+                // Load our base settings.
+                //
+                // Archival bases are:
+                // Staging: gs://<services:gs_staging_bucket>/<project-id>/...
+                // Archive: gs://<project:archive_gs_bucket>/<project-id>/...
+                st := archivist.Settings{
+                        GSBase:        gs.MakePath(pcfg.ArchiveGsBucket, "").Concat(serviceID),
+                        GSStagingBase: gs.MakePath(acfg.GsStagingBucket, "").Concat(serviceID),
+
+                        IndexStreamRange: indexParam(func(ic *svcconfig.ArchiveIndexConfig) int32 { return ic.StreamRange }),
+                        IndexPrefixRange: indexParam(func(ic *svcconfig.ArchiveIndexConfig) int32 { return ic.PrefixRange }),
+                        IndexByteRange:   indexParam(func(ic *svcconfig.ArchiveIndexConfig) int32 { return ic.ByteRange }),
+                        AlwaysRender:     (acfg.RenderAllStreams || pcfg.RenderAllStreams),
+                }
+
+                // Fold project settings into loaded ones.
+                return &st, nil
+        }
+}
+
 // Entry point.
 func main() {
         a := application{
                 Service: service.Service{
                         Name: "archivist",
                 },
         }
         a.Run(context.Background(), a.runArchivist)
 }