LogDog: Add project namespacing to Archivist.

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"
         "io"
 
         "github.com/golang/protobuf/proto"
         "github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
+        "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/logdog/types"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/parallel"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/server/logdog/archive"
         "github.com/luci/luci-go/server/logdog/storage"
         "golang.org/x/net/context"
 )
@@ skipping 54 matching lines @@
 // delete the task). The return value of true should only be used if the task
 // is truly complete and acknowledged by the Coordinator.
 //
 // If the supplied Context is Done, operation may terminate before completion,
 // returning the Context's error.
 func (a *Archivist) ArchiveTask(c context.Context, task Task) bool {
         delete, err := a.archiveTaskImpl(c, task)
         log.Fields{
                 log.ErrorKey: err,
                 "delete":     delete,
+                "project":    task.Task().Project,
                 "path":       task.Task().Path,
         }.Infof(c, "Finished archive task.")
         return delete
 }
 
 // archiveTaskImpl returns the same boolean value as ArchiveTask, but includes
 // an error. The error is useful for testing to assert that certain conditions
 // were hit.
 func (a *Archivist) archiveTaskImpl(c context.Context, task Task) (bool, error) {
         at := task.Task()
         log.Fields{
-                "path": at.Path,
+                "project": at.Project,
+                "path":    at.Path,
         }.Debugf(c, "Received archival task.")
 
+        if err := types.StreamPath(at.Path).Validate(); err != nil {
+                return true, fmt.Errorf("invalid path %q: %s", at.Path, err)
+        }
+
+        // TODO(dnj): Remove empty project exemption, make empty project invalid.
+        if at.Project != "" {
+                if err := config.ProjectName(at.Project).Validate(); err != nil {
+                        return true, fmt.Errorf("invalid project name %q: %s", at.Project, err)
+                }
+        }
+
         // Load the log stream's current state. If it is already archived, we will
         // return an immediate success.
         ls, err := a.Service.LoadStream(c, &logdog.LoadStreamRequest{
-                Path: at.Path,
-                Desc: true,
+                Project: at.Project,
+                Path:    at.Path,
+                Desc:    true,
         })
         switch {
         case err != nil:
                 log.WithError(err).Errorf(c, "Failed to load log stream.")
                 return false, err
 
         case ls.State == nil:
                 log.Errorf(c, "Log stream did not include state.")
                 return false, errors.New("log stream did not include state")
 
@@ skipping 58 matching lines @@
                 tidx := ls.State.TerminalIndex
 
                 if tidx < 0 {
                         log.Warningf(c, "Cannot archive complete stream with no terminal index.")
                         return false, errors.New("completeness required, but stream has no terminal index")
                 }
 
                 // 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.
-                if err := a.checkComplete(types.StreamPath(at.Path), types.MessageIndex(tidx)); err != nil {
+                if err := a.checkComplete(config.ProjectName(at.Project), types.StreamPath(at.Path), types.MessageIndex(tidx)); err != nil {
                         return false, err
                 }
         }
 
         ar := logdog.ArchiveStreamRequest{
-                Path: at.Path,
+                Project: at.Project,
+                Path:    at.Path,
         }
 
         // Archive to staging.
         //
         // If a non-transient failure occurs here, we will report it to the Archivist
         // under the assumption that it will continue occurring.
         //
         // We will handle error creating the plan and executing the plan in the same
         // switch statement below.
-        staged, err := a.makeStagedArchival(c, types.StreamPath(at.Path), ls, task.UniqueID())
+        staged, err := a.makeStagedArchival(c, config.ProjectName(at.Project), types.StreamPath(at.Path), ls, task.UniqueID())
         if err != nil {
                 log.WithError(err).Errorf(c, "Failed to create staged archival plan.")
         } else {
                 err = staged.stage(c)
         }
 
         switch {
         case errors.IsTransient(err):
                 // If this is a transient error, exit immediately and do not delete the
                 // archival task.
@@ skipping 51 matching lines @@
                 return false, err
         }
 
         // Archival is complete and acknowledged by Coordinator. Consume the archival
         // task.
         return true, nil
 }
 
 // checkComplete performs a quick scan of intermediate storage to ensure that
 // all of the log stream's records are available.
-func (a *Archivist) checkComplete(path types.StreamPath, tidx types.MessageIndex) error {
+func (a *Archivist) checkComplete(project config.ProjectName, path types.StreamPath, tidx types.MessageIndex) error {
         sreq := storage.GetRequest{
+                Project:  project,
                 Path:     path,
                 KeysOnly: true,
         }
 
         nextIndex := types.MessageIndex(0)
         var ierr error
         err := a.Storage.Get(sreq, func(idx types.MessageIndex, d []byte) bool {
                 switch {
                 case idx != nextIndex:
                         ierr = fmt.Errorf("missing log entry index %d (next %d)", nextIndex, idx)
@@ skipping 10 matching lines @@
         })
         if ierr != nil {
                 return ierr
         }
         if err != nil {
                 return err
         }
         return nil
 }
 
-func (a *Archivist) makeStagedArchival(c context.Context, path types.StreamPath, ls *logdog.LoadStreamResponse, uid string) (
-        *stagedArchival, error) {
+func (a *Archivist) makeStagedArchival(c context.Context, project config.ProjectName, path types.StreamPath,
+        ls *logdog.LoadStreamResponse, uid string) (*stagedArchival, error) {
         sa := stagedArchival{
                 Archivist: a,
+                project:   project,
                 path:      path,
 
                 terminalIndex: 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,
                         "protoVersion": ls.State.ProtoVersion,
                 }.Errorf(c, "Failed to unmarshal descriptor protobuf.")
                 return nil, err
         }
 
         bext := sa.desc.BinaryFileExt
         if bext == "" {
                 bext = "bin"
         }
 
         // Construct our staged archival paths.
-        sa.stream = a.makeStagingPaths(path, "logstream.entries", uid)
-        sa.index = a.makeStagingPaths(path, "logstream.index", uid)
-        sa.data = a.makeStagingPaths(path, fmt.Sprintf("data.%s", bext), uid)
+        sa.stream = a.makeStagingPaths(project, path, "logstream.entries", uid)
+        sa.index = a.makeStagingPaths(project, path, "logstream.index", uid)
+        sa.data = a.makeStagingPaths(project, path, fmt.Sprintf("data.%s", bext), uid)
         return &sa, nil
 }
 
 // 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 (a *Archivist) makeStagingPaths(path types.StreamPath, name, uid string) stagingPaths {
+func (a *Archivist) makeStagingPaths(project config.ProjectName, path types.StreamPath, name, uid string) stagingPaths {
+        // TODO(dnj): This won't be necessary when empty project is invalid.
+        if project == "" {
+                project = "_"
+        }
+
         return stagingPaths{
-                staged: a.GSStagingBase.Concat(string(path), uid, name),
-                final:  a.GSBase.Concat(string(path), name),
+                staged: a.GSStagingBase.Concat(string(project), string(path), uid, name),
+                final:  a.GSBase.Concat(string(project), string(path), name),
         }
 }
 
 type stagedArchival struct {
         *Archivist
 
-        path types.StreamPath
-        desc logpb.LogStreamDescriptor
+        project config.ProjectName
+        path    types.StreamPath
+        desc    logpb.LogStreamDescriptor
 
         stream stagingPaths
         index  stagingPaths
         data   stagingPaths
 
         finalized     bool
         terminalIndex int64
         logEntryCount int64
 }
 
@@ skipping 65 matching lines @@
 
         if dataWriter, err = createWriter(sa.data.staged); err != nil {
                 return err
         }
         defer closeWriter(dataWriter, sa.data.staged)
 
         // Read our log entries from intermediate storage.
         ss := storageSource{
                 Context:       c,
                 st:            sa.Storage,
+                project:       sa.project,
                 path:          sa.path,
                 terminalIndex: types.MessageIndex(sa.terminalIndex),
                 lastIndex:     -1,
         }
 
         m := archive.Manifest{
                 Desc:             &sa.desc,
                 Source:           &ss,
                 LogWriter:        streamWriter,
                 IndexWriter:      indexWriter,
@@ skipping 99 matching lines @@
         }
 }
 
 func (sa *stagedArchival) getStagingPaths() []*stagingPaths {
         return []*stagingPaths{
                 &sa.stream,
                 &sa.index,
                 &sa.data,
         }
 }