LogDog: Use per-project settings for archival.

server/internal/logdog/archivist/archivist.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 archivist
 
 import (
         "bytes"
         "encoding/hex"
         "fmt"
@@ skipping 86 matching lines @@
         // consumed. This may be called multiple times with no additional effect.
         Consume()
 
         // AssertLease asserts that the lease for this Task is still held.
         //
         // On failure, it will return an error. If successful, the Archivist may
         // assume that it holds the lease longer.
         AssertLease(context.Context) error
 }
 
+// Settings retrieve the settings for an archival.

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

s/retrieve/store
SettingsLoader retrieves them; Settings does not have methods

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

Done.

+type Settings struct {
+        // GSBase is the base Google Storage path. This includes the bucket name
+        // and any associated path.
+        GSBase gs.Path
+        // GSStagingBase is the base Google Storage path for archive staging. This
+        // includes the bucket name and any associated path.
+        GSStagingBase gs.Path

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

rebase?

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

Acknowledged.

+
+        // AlwaysCreateBinary if true, means that a binary should be archived

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

nit: comma before if?

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

Done.

+        // regardless of whether a specific binary file extension has been supplied
+        // with the log stream.
+        AlwaysCreateBinary bool
+
+        // IndexStreamRange is the maximum number of stream indexes in between index
+        // entries. See archive.Manifest for more information.
+        IndexStreamRange int
+        // IndexPrefixRange is the maximum number of prefix indexes in between index
+        // entries. See archive.Manifest for more information.
+        IndexPrefixRange int
+        // IndexByteRange is the maximum number of stream data bytes in between index
+        // entries. See archive.Manifest for more information.
+        IndexByteRange int
+}
+
+// SettingsLoader returns archival Settings for a given project.
+type SettingsLoader func(context.Context, config.ProjectName) (*Settings, error)
+
 // Archivist is a stateless configuration capable of archiving individual log
 // streams.
 type Archivist struct {
         // Service is the client to use to communicate with Coordinator's Services
         // endpoint.
         Service logdog.ServicesClient
 
+        // SettingsLoader loads archival settings for a specific project.
+        SettingsLoader SettingsLoader
+
         // Storage is the archival source Storage instance.
         Storage storage.Storage
         // GSClient is the Google Storage client to for archive generation.
         GSClient gs.Client
-
-        // GSBase is the base Google Storage path. This includes the bucket name
-        // and any associated path.
-        GSBase gs.Path
-        // GSStagingBase is the base Google Storage path for archive staging. This
-        // includes the bucket name and any associated path.
-        GSStagingBase gs.Path
-
-        // PrefixIndexRange is the maximum number of stream indexes in between index
-        // entries. See archive.Manifest for more information.
-        StreamIndexRange int
-        // PrefixIndexRange is the maximum number of prefix indexes in between index
-        // entries. See archive.Manifest for more information.
-        PrefixIndexRange int
-        // ByteRange is the maximum number of stream data bytes in between index
-        // entries. See archive.Manifest for more information.
-        ByteRange int
 }
 
 // storageBufferSize is the size, in bytes, of the LogEntry buffer that is used
 // to during archival. This should be greater than the maximum LogEntry size.
 const storageBufferSize = types.MaxLogEntryDataSize * 64
 
 // ArchiveTask processes and executes a single log stream archive task.
 //
 // During processing, the Task's Consume method may be called to indicate that
 // it should be consumed.
@@ skipping 100 matching lines @@
         // retry later.
         age := ls.Age.Duration()
         if age < at.SettleDelay.Duration() {
                 log.Fields{
                         "age":         age,
                         "settleDelay": at.SettleDelay.Duration(),
                 }.Infof(c, "Log stream is younger than the settle delay. Returning task to queue.")
                 return statusErr(errors.New("log stream is within settle delay"))
         }
 
+        // Load archival settings for this project.
+        settings, err := a.loadSettings(c, config.ProjectName(at.Project))
+        if err != nil {
+                log.Fields{
+                        log.ErrorKey: err,
+                        "project":    at.Project,
+                }.Errorf(c, "Failed to load settings for project.")
+                return err
+        }
+
         ar := logdog.ArchiveStreamRequest{
                 Project: at.Project,
                 Id:      at.Id,
         }
 
         // Build our staged archival plan. This doesn't actually do any archiving.
-        staged, err := a.makeStagedArchival(c, config.ProjectName(at.Project), ls, task.UniqueID())
+        staged, err := a.makeStagedArchival(c, config.ProjectName(at.Project), settings, ls, task.UniqueID())
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create staged archival plan.")
                 return err
         }
 
         // Are we required to archive a complete log stream?
         if age <= at.CompletePeriod.Duration() {
                 // If we're requiring completeness, perform a keys-only scan of intermediate
                 // storage to ensure that we have all of the records before we bother
                 // streaming to storage only to find that we are missing data.
@@ skipping 76 matching lines @@
                 log.WithError(err).Errorf(c, "Failed to report archive state.")
                 return err
         }
 
         // Archival is complete and acknowledged by Coordinator. Consume the archival
         // task.
         task.Consume()
         return nil
 }
 
+// loadSettings loads and validates archival settings.
+func (a *Archivist) loadSettings(c context.Context, project config.ProjectName) (*Settings, error) {
+        if a.SettingsLoader == nil {
+                panic("no settings loader configured")

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

I am glad to see you didn't ban panics altogether ;)

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

IMO panics have their place. I think explicit error handling is generally
better, though.

+        }
+
+        st, err := a.SettingsLoader(c, project)
+        switch {
+        case err != nil:
+                return nil, err
+
+        case st.GSBase.Bucket() == "":
+                log.Fields{
+                        log.ErrorKey: err,
+                        "gsBase":     st.GSBase,
+                }.Errorf(c, "Invalid storage base.")
+                return nil, errors.New("invalid storage base")
+
+        case st.GSStagingBase.Bucket() == "":
+                log.Fields{
+                        log.ErrorKey:    err,
+                        "gsStagingBase": st.GSStagingBase,
+                }.Errorf(c, "Invalid storage staging base.")
+                return nil, errors.New("invalid storage staging base")
+
+        default:
+                return st, nil
+        }
+}
+
 func (a *Archivist) makeStagedArchival(c context.Context, project config.ProjectName,
-        ls *logdog.LoadStreamResponse, uid string) (*stagedArchival, error) {
+        st *Settings, ls *logdog.LoadStreamResponse, uid string) (*stagedArchival, error) {
         sa := stagedArchival{
                 Archivist: a,
+                Settings:  st,
                 project:   project,
 
                 terminalIndex: types.MessageIndex(ls.State.TerminalIndex),
         }
 
         // Deserialize and validate the descriptor protobuf. If this fails, it is a
         // non-transient error.
         if err := proto.Unmarshal(ls.Desc, &sa.desc); err != nil {
                 log.Fields{
                         log.ErrorKey:   err,
@@ skipping 10 matching lines @@
 
         // Construct our staged archival paths.
         sa.stream = sa.makeStagingPaths("logstream.entries", uid)
         sa.index = sa.makeStagingPaths("logstream.index", uid)
         sa.data = sa.makeStagingPaths(fmt.Sprintf("data.%s", bext), uid)
         return &sa, nil
 }
 
 type stagedArchival struct {
         *Archivist
+        *Settings
 
         project config.ProjectName
         path    types.StreamPath
         desc    logpb.LogStreamDescriptor
 
         stream stagingPaths
         index  stagingPaths
         data   stagingPaths
 
         finalized     bool
         terminalIndex types.MessageIndex
         logEntryCount int64
 }
 
 // makeStagingPaths returns a stagingPaths instance for the given path and
 // file name. It incorporates a unique ID into the staging name to differentiate
 // it from other staging paths for the same path/name.
 func (sa *stagedArchival) makeStagingPaths(name, uid string) stagingPaths {
-        // TODO(dnj): This won't be necessary when empty project is invalid.
-        project := string(sa.project)
-        if project == "" {
-                project = "_"
+        proj := string(sa.project)
+        // TODO (dnj): When empty projects are disallowed, remove this.
+        if proj == "" {
+                proj = "_"
         }
 
+        // Either of these paths may be shared between projects. To enforce
+        // an absence of conflicts, we will insert the project name as part of the
+        // path.
         return stagingPaths{
-                staged: sa.GSStagingBase.Concat(project, string(sa.path), uid, name),
-                final:  sa.GSBase.Concat(project, string(sa.path), name),
+                staged: sa.GSStagingBase.Concat(proj, string(sa.path), uid, name),
+                final:  sa.GSBase.Concat(proj, string(sa.path), name),
         }
 }
 
 // checkComplete performs a quick scan of intermediate storage to ensure that
 // all of the log stream's records are available.
 func (sa *stagedArchival) checkComplete(c context.Context) error {
         if sa.terminalIndex < 0 {
                 log.Warningf(c, "Cannot archive complete stream with no terminal index.")
                 return statusErr(errors.New("completeness required, but stream has no terminal index"))
         }
@@ skipping 110 matching lines @@
                 terminalIndex: sa.terminalIndex,
                 lastIndex:     -1,
         }
 
         m := archive.Manifest{
                 Desc:             &sa.desc,
                 Source:           &ss,
                 LogWriter:        streamWriter,
                 IndexWriter:      indexWriter,
                 DataWriter:       dataWriter,
-                StreamIndexRange: sa.StreamIndexRange,
-                PrefixIndexRange: sa.PrefixIndexRange,
-                ByteRange:        sa.ByteRange,
+                StreamIndexRange: sa.IndexStreamRange,
+                PrefixIndexRange: sa.IndexPrefixRange,
+                ByteRange:        sa.IndexByteRange,
 
                 Logger: log.Get(c),
         }
         if err = archive.Archive(m); err != nil {
                 log.WithError(err).Errorf(c, "Failed to archive log stream.")
                 return
         }
 
         switch {
         case ss.logEntryCount == 0:
@@ skipping 125 matching lines @@
         return e.inner
 }
 
 func isFailure(err error) bool {
         if err == nil {
                 return false
         }
         _, ok := err.(*statusErrorWrapper)
         return !ok
 }