LogDog/Archivist: Conditionally render data.

logdog/server/archivist/archivist.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 archivist
 
 import (
         "bytes"
         "encoding/hex"
         "fmt"
@@ skipping 430 matching lines @@
                 }.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,
         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,
                         "protoVersion": ls.State.ProtoVersion,
                 }.Errorf(c, "Failed to unmarshal descriptor protobuf.")
                 return nil, err
         }
         sa.path = sa.desc.Path()
 
-        bext := sa.desc.BinaryFileExt
-        if bext == "" {
-                bext = "bin"
-        }
-
         // 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)
+
+        // If we're emitting binary files, construct that too.
+        bext := sa.desc.BinaryFileExt
+        if bext != "" || sa.AlwaysRender {
+                // If no binary file extension was supplied, choose a default.
+                if bext == "" {
+                        bext = "bin"
+                }
+
+                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
@@ skipping 121 matching lines @@
         if streamWriter, err = createWriter(sa.stream.staged); err != nil {
                 return
         }
         defer closeWriter(streamWriter, sa.stream.staged)
 
         if indexWriter, err = createWriter(sa.index.staged); err != nil {
                 return err
         }
         defer closeWriter(indexWriter, sa.index.staged)
 
-        if dataWriter, err = createWriter(sa.data.staged); err != nil {
-                return err
+        if sa.data.enabled() {
+                // Only emit a data stream if we are configured to do so.
+                if dataWriter, err = createWriter(sa.data.staged); err != nil {
+                        return err
+                }
+                defer closeWriter(dataWriter, sa.data.staged)
         }
-        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: sa.terminalIndex,
                 lastIndex:     -1,
         }
@@ skipping 26 matching lines @@
                         "terminalIndex": ss.lastIndex,
                         "logEntryCount": ss.logEntryCount,
                 }.Debugf(c, "Finished archiving log stream.")
         }
 
         // Update our state with archival results.
         sa.terminalIndex = ss.lastIndex
         sa.logEntryCount = ss.logEntryCount
         sa.stream.bytesWritten = streamWriter.Count()
         sa.index.bytesWritten = indexWriter.Count()
-        sa.data.bytesWritten = dataWriter.Count()
+        if dataWriter != nil {
+                sa.data.bytesWritten = dataWriter.Count()
+        }
         return
 }
 
 type stagingPaths struct {
         staged       gs.Path
         final        gs.Path
         bytesWritten int64
 }
 
-func (d *stagingPaths) clearStaged() {
-        d.staged = ""
-}
+func (d *stagingPaths) clearStaged() { d.staged = "" }
+
+func (d *stagingPaths) enabled() bool { return d.final != "" }
 
 func (d *stagingPaths) addMetrics(c context.Context, archiveField, streamField string) {
         tsSize.Add(c, float64(d.bytesWritten), archiveField, streamField)
         tsTotalBytes.Add(c, d.bytesWritten, archiveField, streamField)
 }
 
 func (sa *stagedArchival) finalize(c context.Context, client gs.Client, ar *logdog.ArchiveStreamRequest) error {
         err := parallel.FanOutIn(func(taskC chan<- func() error) {
                 for _, d := range sa.getStagingPaths() {
                         d := d
 
-                        // Don't copy zero-sized streams.
-                        if d.bytesWritten == 0 {
+                        // Don't finalize zero-sized streams.
+                        if !d.enabled() || d.bytesWritten == 0 {
                                 continue
                         }
 
                         taskC <- func() error {
                                 if err := client.Rename(d.staged, d.final); err != nil {
                                         log.Fields{
                                                 log.ErrorKey: err,
                                                 "stagedPath": d.staged,
                                                 "finalPath":  d.final,
                                         }.Errorf(c, "Failed to rename GS object.")
                                         return err
                                 }
 
                                 // Clear the staged value to indicate that it no longer exists.
                                 d.clearStaged()
                                 return nil
                         }
                 }
         })
         if err != nil {
                 return err
         }
 
         ar.TerminalIndex = int64(sa.terminalIndex)
         ar.LogEntryCount = sa.logEntryCount
         ar.StreamUrl = string(sa.stream.final)
         ar.StreamSize = sa.stream.bytesWritten
         ar.IndexUrl = string(sa.index.final)
         ar.IndexSize = sa.index.bytesWritten
-        ar.DataUrl = string(sa.data.final)
-        ar.DataSize = sa.data.bytesWritten
+        if sa.data.enabled() {
+                ar.DataUrl = string(sa.data.final)
+                ar.DataSize = sa.data.bytesWritten
+        }
         return nil
 }
 
 func (sa *stagedArchival) cleanup(c context.Context) {
         for _, d := range sa.getStagingPaths() {
                 if d.staged == "" {
                         continue
                 }
 
                 if err := sa.GSClient.Delete(d.staged); err != nil {
@@ skipping 42 matching lines @@
         return e.inner
 }
 
 func isFailure(err error) bool {
         if err == nil {
                 return false
         }
         _, ok := err.(*statusErrorWrapper)
         return !ok
 }