LogDog: Update to archival V2.

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"
         "fmt"
         "strings"
         "time"
 
         "github.com/golang/protobuf/proto"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/common/logdog/types"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
 )
 
+// currentLogStreamSchema is the current schema version of the LogStream.
+// Changes that are not backward-compatible should update this field so
+// migration logic and scripts can translate appropriately.
+const currentLogStreamSchema = "1"
+
 // LogStreamState is the archival state of the log stream.
 type LogStreamState int
 
 const (
-        // 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.
+        // LSStreaming indicates that the log stream is still streaming. This implies
+        // that no terminal index has been identified yet.
+        LSStreaming LogStreamState = iota
+        // LSArchiveTasked indicates that the log stream has had an archival task
+        // generated for it and is awaiting archival.
+        LSArchiveTasked
+        // LSArchived indicates that the log stream has been successfully archived.
         LSArchived
 )
 
+func (s LogStreamState) String() string {
+        switch s {
+        case LSStreaming:
+                return "STREAMING"
+        case LSArchiveTasked:
+                return "ARCHIVE_TASKED"
+        case LSArchived:
+                return "ARCHIVED"
+        default:
+                return fmt.Sprintf("UNKNOWN(%d)", s)
+        }
+}
+
+// Archived returns true if this LogStreamState represents a finished archival.
+func (s LogStreamState) Archived() bool {
+        return s >= LSArchived
+}
+
 // 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 11 matching lines @@
 //
 //        - _Tags is a string slice containing:
 //          - 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.
 type LogStream struct {
+        // Schema is the datastore schema version for this object. This can be used
+        // to facilitate schema migrations.
+        //
+        // The current schema is currentLogStreamSchema.
+        Schema string
+
         // 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
 
         // Purged, if true, indicates that this log stream has been marked as purged.
         // Non-administrative queries and requests for this stream will operate as
         // if this entry doesn't exist.
         Purged bool
 
         // Secret is the Butler secret value for this stream.
         //
         // This value may only be returned to LogDog services; it is not user-visible.
         Secret []byte `gae:",noindex"`
 
         // Created is the time when this stream was created.
         Created time.Time
-        // Updated is the Coordinator's record of when this log stream was last
-        // updated.
-        Updated time.Time
+        // TerminatedTime is the Coordinator's record of when this log stream was
+        // terminated.
+        TerminatedTime time.Time `gae:",noindex"`
+        // ArchivedTime is the Coordinator's record of when this log stream was
+        // archived.
+        ArchivedTime time.Time `gae:",noindex"`
 
         // ProtoVersion is the version string of the protobuf, as reported by the
         // Collector (and ultimately self-identified by the Butler).
         ProtoVersion string
         // Descriptor is the binary protobuf data LogStreamDescriptor.
         Descriptor []byte `gae:",noindex"`
         // ContentType is the MIME-style content type string for this stream.
         ContentType string
         // StreamType is the data type of the stream.
         StreamType logpb.StreamType
         // Timestamp is the Descriptor's recorded client-side timestamp.
         Timestamp time.Time
 
         // Tags is a set of arbitrary key/value tags associated with this stream. Tags
         // 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.
+        // TerminalIndex is the index of the last log entry in the stream.
+        //
+        // If this is <0, the log stream is either still streaming or has been
+        // archived with no log entries.
         TerminalIndex int64 `gae:",noindex"`
 
+        // ArchiveLogEntryCount is the number of LogEntry records that were archived
+        // for this log stream.
+        //
+        // This is valid only if the log stream is Archived.
+        ArchiveLogEntryCount int64 `gae:",noindex"`
+        // ArchiveTaskName is the name of this LogStream's archival task.
+        ArchiveTaskName string `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.
-        ArchiveWhole bool
+        // 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"`
+        // extra causes datastore to ignore unrecognized fields and strip them in
+        // future writes.
+        extra 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
+
+        // noDatastoreValidate is a testing parameter to instruct the LogStream not to

[Vadim Sh., 2016/04/07 01:21:32]

noDSValidate

[dnj, 2016/04/11 17:20:03]

Done.

+        // validate before reading/writing to datastore. It can be controlled by
+        // calling SetDSValidate().
+        noDSValidate bool
 }
 
 var _ interface {
         ds.PropertyLoadSaver
         ds.MetaGetterSetter
 } = (*LogStream)(nil)
 
 // NewLogStream returns a LogStream instance with its ID field initialized based
 // on the supplied path.
 //
@@ skipping 52 matching lines @@
 
                 switch k {
                 case "_Tags":
                         // Load the tag map. Ignore errors.
                         tm, _ := tagMapFromProperties(v)
                         s.Tags = tm
                 }
                 delete(pmap, k)
         }
 
-        return ds.GetPLS(s).Load(pmap)
+        if err := ds.GetPLS(s).Load(pmap); err != nil {
+                return err
+        }
+
+        // Migrate schema (if needed), then validate.
+        if err := s.migrateSchema(); err != nil {
+                return err
+        }
+
+        if !s.noDSValidate {
+                if err := s.Validate(); err != nil {
+                        return err
+                }
+        }
+        return nil
 }
 
 // Save implements ds.PropertyLoadSaver.
 func (s *LogStream) Save(withMeta bool) (ds.PropertyMap, error) {
+        if !s.noDSValidate {
+                if err := s.Validate(); err != nil {
+                        return nil, err
+                }
+                s.Schema = currentLogStreamSchema
+        }
+
+        // Save default struct fields.
         pmap, err := ds.GetPLS(s).Save(withMeta)
         if err != nil {
                 return nil, err
         }
 
         // Encode _Tags.
         pmap["_Tags"], err = s.Tags.toProperties()
         if err != nil {
                 return nil, fmt.Errorf("failed to encode tags: %v", err)
         }
@@ skipping 38 matching lines @@
                 if s.Prefix == "" || s.Name == "" {
                         panic("cannot generate ID hash: Prefix and Name are not populated")
                 }
 
                 hash := sha256.Sum256([]byte(s.Path()))
                 s.hashID = hex.EncodeToString(hash[:])
         }
         return s.hashID
 }
 
-// Put writes this LogStream to the Datastore. Before writing, it validates that
-// LogStream is complete.
-func (s *LogStream) Put(di ds.Interface) error {
-        if err := s.Validate(); err != nil {
-                return err
-        }
-        return di.Put(s)
-}
-
 // Validate evaluates the state and data contents of the LogStream and returns
 // an error if it is invalid.
 func (s *LogStream) Validate() error {
         if err := types.StreamName(s.Prefix).Validate(); err != nil {
                 return fmt.Errorf("invalid prefix: %s", err)
         }
         if err := types.StreamName(s.Name).Validate(); err != nil {
                 return fmt.Errorf("invalid name: %s", err)
         }
         if len(s.Secret) != types.StreamSecretLength {
                 return fmt.Errorf("invalid secret length (%d != %d)", len(s.Secret), types.StreamSecretLength)
         }
         if s.ContentType == "" {
                 return errors.New("empty content type")
         }
         if s.Created.IsZero() {
                 return errors.New("created time is not set")
         }
-        if s.Updated.IsZero() {
-                return errors.New("updated time is not set")
+
+        if s.Terminated() && s.TerminatedTime.IsZero() {
+                return errors.New("log stream is terminated, but missing terminated time")
         }
-        if s.Updated.Before(s.Created) {
-                return fmt.Errorf("updated time must be >= created time (%s < %s)", s.Updated, s.Created)
+        if s.Archived() && s.ArchivedTime.IsZero() {
+                return errors.New("log stream is archived, but missing archived time")
         }
 
         switch s.StreamType {
         case logpb.StreamType_TEXT, logpb.StreamType_BINARY, logpb.StreamType_DATAGRAM:
                 break
 
         default:
                 return fmt.Errorf("unsupported stream type: %v", s.StreamType)
         }
 
@@ skipping 14 matching lines @@
 func (s *LogStream) DescriptorValue() (*logpb.LogStreamDescriptor, error) {
         pb := logpb.LogStreamDescriptor{}
         if err := proto.Unmarshal(s.Descriptor, &pb); err != nil {
                 return nil, err
         }
         return &pb, nil
 }
 
 // Terminated returns true if this stream has been terminated.
 func (s *LogStream) Terminated() bool {
-        return s.State >= LSTerminated
+        if s.Archived() {
+                return true
+        }
+        return s.TerminalIndex >= 0
 }
 
-// Archived returns true if this stream has been archived. A stream is archived
-// if it has any of its archival properties set.
+// Archived returns true if this stream has been archived.
 func (s *LogStream) Archived() bool {
-        return s.State >= LSArchived
+        return s.State.Archived()
 }
 
-// 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)
+// ArchiveComplete returns true if this stream has been archived and all of its
+// log entries were present.
+func (s *LogStream) ArchiveComplete() bool {
+        return (s.Archived() && s.ArchiveLogEntryCount == (s.TerminalIndex+1))
 }
 
 // LoadDescriptor loads the fields in the log stream descriptor into this
 // LogStream entry. These fields are:
 //   - Prefix
 //   - Name
 //   - ContentType
 //   - StreamType
 //   - Descriptor
 //   - Timestamp
@@ skipping 27 matching lines @@
 // DescriptorProto unmarshals a LogStreamDescriptor from the stream's Descriptor
 // field. It will return an error if the unmarshalling fails.
 func (s *LogStream) DescriptorProto() (*logpb.LogStreamDescriptor, error) {
         desc := logpb.LogStreamDescriptor{}
         if err := proto.Unmarshal(s.Descriptor, &desc); err != nil {
                 return nil, err
         }
         return &desc, nil
 }
 
+// SetDSValidate controls whether this LogStream is validated prior to being
+// read from or written to datastore.
+//
+// This is a testing parameter, and should NOT be used in production code.
+func (s *LogStream) SetDSValidate(v bool) {
+        s.noDSValidate = !v
+}
+
 // normalizeHash takes a SHA256 hexadecimal string as input. It validates that
 // it is a valid SHA256 hash and, if so, returns a normalized version that can
 // be used as a log stream key.
 func normalizeHash(v string) (string, error) {
         if decodeSize := hex.DecodedLen(len(v)); decodeSize != sha256.Size {
                 return "", fmt.Errorf("invalid SHA256 hash size (%d != %d)", decodeSize, sha256.Size)
         }
         b, err := hex.DecodeString(v)
         if err != nil {
                 return "", err
@@ skipping 151 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")
 }