LogDog: Handle archive failures.

appengine/logdog/coordinator/logStream.go

 // Copyright 2015 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 coordinator
 
 import (
         "crypto/sha256"
         "encoding/hex"
         "errors"
@@ skipping 14 matching lines @@
         // LSPending indicates that no archival has occurred yet.
         LSPending LogStreamState = iota
         // LSTerminated indicates that the log stream has received a terminal index
         // and is awaiting archival.
         LSTerminated
         // LSArchived indicates that the log stream has been successfully archived but
         // has not yet been cleaned up.
         LSArchived
 )
 
+// ArchiveState is the state of archival. It is valid on log streams whose
+// log stream state is LSArchived.
+type ArchiveState int
+
+const (
+        // NotArchived is the default archive state, indicating that the log stream is
+        // not archived.
+        NotArchived ArchiveState = iota
+        // Archived indicates that the log stream was archived completely and
+        // successfully.
+        Archived
+        // ArchivedPartially indicates that the log stream has been archived, but that
+        // there are missing log stream entries. This will happen when the stream has
+        // exceeded an archival threshold without all of its log entries available
+        // in intermediate storage.
+        ArchivedPartially
+        // ArchivedWithErrors indicates that the log stream is in an archived state,
+        // but that there were errors. Archive files may be present in any state.
+        ArchivedWithErrors
+)
+
 // LogStream is the primary datastore model containing information and state of
 // an individual log stream.
 //
 // This structure contains the standard queryable fields, and is the source of
 // truth for log stream state. Writes to LogStream should be done via Put, which
 // will ensure that the LogStream's related query objects are kept in sync.
 //
 // This structure has additional datastore fields imposed by the
 // PropertyLoadSaver. These fields enable querying against some of the complex
 // data types:
@@ skipping 13 matching lines @@
 //          - KEY=[VALUE] key/value tags.
 //          - KEY key presence tags.
 //
 //        - _Terminated is true if the LogStream has been terminated.
 //        - _Archived is true if the LogStream has been archived.
 //
 // Most of the values in QueryBase are static. Those that change can only be
 // changed through service endpoint methods.
 //
 // LogStream's QueryBase is authortative.
+//
+// TODO(dnj): Migration schema checklist. Next migration, we will want to...
+//      - Add schema version.
+//      - Remove State, ArchiveWhole in favor of "_Terminated" and "ArchiveState".
 type LogStream struct {
         // Prefix is this log stream's prefix value. Log streams with the same prefix
         // are logically grouped.
         Prefix string
         // Name is the unique name of this log stream within the Prefix scope.
         Name string
 
         // State is the log stream's current state.
         State LogStreamState
 
@@ skipping 29 matching lines @@
         // can be queried against.
         //
         // The serialization/deserialization is handled manually in order to enable
         // key/value queries.
         Tags TagMap `gae:"-"`
 
         // Source is the set of source strings sent by the Butler.
         Source []string
 
         // TerminalIndex is the log stream index of the last log entry in the stream.
-        // If the value is -1, the log is still streaming.
+        //
+        // When the value is <0:
+        //      - If the log is not archived, this means that the stream is still
+        //        streaming without a known terminal index.
+        //      - If the log is archived, it means that the stream's archival has
+        //        completed, but had no log entries.
         TerminalIndex int64 `gae:",noindex"`
 
         // ArchiveIndexURL is the Google Storage URL where the log stream's index is
         // archived.
         ArchiveIndexURL string `gae:",noindex"`
         // ArchiveIndexSize is the size, in bytes, of the archived Index. It will be
         // zero if the file is not archived.
         ArchiveIndexSize int64 `gae:",noindex"`
         // ArchiveStreamURL is the Google Storage URL where the log stream's raw
         // stream data is archived. If this is not empty, the log stream is considered
         // archived.
         ArchiveStreamURL string `gae:",noindex"`
         // ArchiveStreamSize is the size, in bytes, of the archived stream. It will be
         // zero if the file is not archived.
         ArchiveStreamSize int64 `gae:",noindex"`
         // ArchiveDataURL is the Google Storage URL where the log stream's assembled
         // data is archived. If this is not empty, the log stream is considered
         // archived.
         ArchiveDataURL string `gae:",noindex"`
         // ArchiveDataSize is the size, in bytes, of the archived data. It will be
         // zero if the file is not archived.
         ArchiveDataSize int64 `gae:",noindex"`
         // ArchiveWhole is true if archival is complete and the archived log stream
         // was not missing any entries.
+        //
+        // TODO(dnj): Next migration, deprecate this in favor of ArchiveState. For
+        // now, we will parallel it.
         ArchiveWhole bool
+        // ArchiveState is the state of log stream archival.
+        ArchiveState ArchiveState
+        // ArchiveErrors is the number of errors encountered during archival. This
+        // will be incremented when incomplete archival is permitted, but the archival
+        // still failed due to an error (e.g., data corruption). If this exceeds a
+        // threshold, the stream may be marked archived even if the archival failed.
+        ArchiveErrors int
 
         // _ causes datastore to ignore unrecognized fields and strip them in future
         // writes.
         _ ds.PropertyMap `gae:"-,extra"`
 
         // hashID is the cached generated ID from the stream's Prefix/Name fields. If
         // this is populated, ID metadata will be retrieved from this field instead of
         // generated.
         hashID string
 }
@@ skipping 193 matching lines @@
                 return nil, err
         }
         return &pb, nil
 }
 
 // Terminated returns true if this stream has been terminated.
 func (s *LogStream) Terminated() bool {
         return s.State >= LSTerminated
 }
 
-// Archived returns true if this stream has been archived. A stream is archived
-// if it has any of its archival properties set.
+// Streaming returns true if this log stream is still streaming without a known
+// terminal index.
+func (s *LogStream) Streaming() bool {
+        return (s.State == LSPending && s.TerminalIndex < 0)
+}
+
+// Archived returns true if this stream has been archived.
 func (s *LogStream) Archived() bool {
-        return s.State >= LSArchived
+        return (s.ArchiveState != NotArchived) || (s.State >= LSArchived)
 }
 
 // ArchiveMatches tests if the supplied Stream, Index, and Data archival URLs
 // match the current values.
 func (s *LogStream) ArchiveMatches(sURL, iURL, dURL string) bool {
         return (s.ArchiveStreamURL == sURL && s.ArchiveIndexURL == iURL && s.ArchiveDataURL == dURL)
 }
 
 // LoadDescriptor loads the fields in the log stream descriptor into this
 // LogStream entry. These fields are:
@@ skipping 204 matching lines @@
 // were created before the supplied time.
 func AddOlderFilter(q *ds.Query, t time.Time) *ds.Query {
         return q.Lt("Created", t.UTC()).Order("-Created")
 }
 
 // AddNewerFilter adds a filter to queries that restricts them to results that
 // were created after the supplied time.
 func AddNewerFilter(q *ds.Query, t time.Time) *ds.Query {
         return q.Gt("Created", t.UTC()).Order("-Created")
 }