LogDog: BigTable batching schema.

server/internal/logdog/collector/collector_test.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 collector
 
 import (
         "bytes"
         "fmt"
         "sync/atomic"
         "testing"
 
         "github.com/luci/luci-go/common/clock/testclock"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logdog/butlerproto"
         "github.com/luci/luci-go/common/logdog/types"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/server/internal/logdog/collector/coordinator"
         "github.com/luci/luci-go/server/logdog/storage/memory"
         "golang.org/x/net/context"
 
+        . "github.com/luci/luci-go/common/testing/assertions"
         . "github.com/smartystreets/goconvey/convey"
 )
 
 // TestCollector runs through a series of end-to-end Collector workflows and
 // ensures that the Collector behaves appropriately.
 func TestCollector(t *testing.T) {
         t.Parallel()
 
         Convey(`Using a test configuration`, t, func() {
                 c, _ := testclock.UseTime(context.Background(), testclock.TestTimeLocal)
@@ skipping 67 matching lines @@
                                         tcc.errC = make(chan error, 2)
                                         tcc.errC <- nil                      // Register
                                         tcc.errC <- errors.New("test error") // Terminate
 
                                         bb.addFullStream("foo/+/bar", 128)
                                         err := coll.Process(c, bb.bundle())
                                         So(err, ShouldNotBeNil)
                                         So(errors.IsTransient(err), ShouldBeFalse)
                                 })
 
-                                Convey(`Will return a transient error if an error happened on storage.`, func() {
+                                Convey(`Will return a transient error if a transient error happened on storage.`, func() {
                                         // Single transient error.
                                         count := int32(0)
                                         st.err = func() error {
                                                 if atomic.AddInt32(&count, 1) == 1 {
-                                                        return errors.New("test error")
+                                                        return errors.WrapTransient(errors.New("test error"))
                                                 }
                                                 return nil
                                         }
 
                                         bb.addFullStream("foo/+/bar", 128)
                                         err := coll.Process(c, bb.bundle())
                                         So(err, ShouldNotBeNil)
                                         So(errors.IsTransient(err), ShouldBeTrue)
                                 })
 
                                 Convey(`Will drop invalid LogStreamDescriptor bundle entries and process the valid ones.`, func() {
-                                        be := bb.genBundleEntry("foo/+/trash", 1337, 4, 6, 7, 8)
-                                        be.Desc.ContentType = "" // Missing ContentType => invalid.
+                                        be := bb.genBundleEntry("foo/+/trash", 1337, 4, 5, 6, 7, 8)
+                                        bb.addBundleEntry(be)
 
-                                        bb.addStreamEntries("foo/+/trash", -1, 0, 1, 2, 3, 5)
-                                        bb.addBundleEntry(be)
+                                        bb.addStreamEntries("foo/+/trash", 0, 1, 3) // Invalid: non-contiguous
                                         bb.addFullStream("foo/+/bar", 32)
 
-                                        So(coll.Process(c, bb.bundle()), ShouldBeNil)
+                                        err := coll.Process(c, bb.bundle())
+                                        So(err, ShouldNotBeNil)
+                                        So(errors.IsTransient(err), ShouldBeFalse)
 
                                         So(tcc, shouldHaveRegisteredStream, "foo/+/bar", 32)
                                         So(st, shouldHaveStoredStream, "foo/+/bar", indexRange{0, 31})
 
-                                        So(tcc, shouldHaveRegisteredStream, "foo/+/trash", -1)
-                                        So(st, shouldHaveStoredStream, "foo/+/trash", 0, 1, 2, 3, 5)
+                                        So(tcc, shouldHaveRegisteredStream, "foo/+/trash", 1337)
+                                        So(st, shouldHaveStoredStream, "foo/+/trash", 4, 5, 6, 7, 8)
                                 })
 
                                 Convey(`Will drop streams with missing secrets.`, func() {
                                         be := bb.genBundleEntry("foo/+/trash", 2, 0, 1, 2)
                                         be.Secret = nil
                                         bb.addBundleEntry(be)
 
-                                        So(coll.Process(c, bb.bundle()), ShouldBeNil)
-                                        So(tcc, shouldNotHaveRegisteredStream, "foo/+/bar", 127)
+                                        err := coll.Process(c, bb.bundle())
+                                        So(err, ShouldErrLike, "missing stream secret")
+                                        So(errors.IsTransient(err), ShouldBeFalse)
+                                        So(tcc, shouldNotHaveRegisteredStream, "foo/+/bar")
                                 })
 
                                 Convey(`Will drop messages with mismatching secrets.`, func() {
                                         bb.addStreamEntries("foo/+/bar", -1, 0, 1, 2)
                                         So(coll.Process(c, bb.bundle()), ShouldBeNil)
 
                                         // Push another bundle with a different secret.
                                         be := bb.genBundleEntry("foo/+/bar", 4, 3, 4)
                                         be.Secret = bytes.Repeat([]byte{0xAA}, types.StreamSecretLength)
                                         be.TerminalIndex = 1337
@@ skipping 16 matching lines @@
                                 Convey(`Will drop bundles with unknown ProtoVersion string.`, func() {
                                         buf := bytes.Buffer{}
                                         w := butlerproto.Writer{ProtoVersion: "!!!invalid!!!"}
                                         w.Write(&buf, &logpb.ButlerLogBundle{})
 
                                         So(coll.Process(c, buf.Bytes()), ShouldBeNil)
 
                                         So(tcc, shouldNotHaveRegisteredStream, "foo/+/bar")
                                 })
 
-                                Convey(`Will drop records beyond a local terminal index.`, func() {
-                                        bb.addStreamEntries("foo/+/bar", 3, 0, 1, 2, 4)
-                                        So(coll.Process(c, bb.bundle()), ShouldBeNil)
-
-                                        bb.addStreamEntries("foo/+/bar", 3, 3, 5)
-                                        So(coll.Process(c, bb.bundle()), ShouldBeNil)
-
-                                        So(tcc, shouldHaveRegisteredStream, "foo/+/bar", 4)
-                                        So(st, shouldHaveStoredStream, "foo/+/bar", indexRange{0, 3})
-                                })
-
-                                Convey(`Will not ingest records beyond a remote terminal index.`, func() {
-                                        tcc.register(coordinator.LogStreamState{
-                                                Path:          "foo/+/bar",
-                                                Secret:        testSecret,
-                                                TerminalIndex: 3,
-                                        })
-
-                                        bb.addStreamEntries("foo/+/bar", 3, 0, 1, 2)
-                                        bb.addStreamEntries("foo/+/bar", 3, 3, 5)
-                                        So(coll.Process(c, bb.bundle()), ShouldBeNil)
-
-                                        So(tcc, shouldHaveRegisteredStream, "foo/+/bar", 3)
-                                        So(st, shouldHaveStoredStream, "foo/+/bar", indexRange{0, 3})
-                                })
-
                                 Convey(`Will not ingest records if the stream is archived.`, func() {
                                         tcc.register(coordinator.LogStreamState{
                                                 Path:          "foo/+/bar",
                                                 Secret:        testSecret,
                                                 TerminalIndex: -1,
                                                 Archived:      true,
                                         })
 
-                                        bb.addStreamEntries("foo/+/bar", 3, 0, 1, 2, 4)
+                                        bb.addStreamEntries("foo/+/bar", 3, 0, 1, 2, 3, 4)
                                         So(coll.Process(c, bb.bundle()), ShouldBeNil)
 
                                         So(tcc, shouldHaveRegisteredStream, "foo/+/bar", -1)
                                         So(st, shouldHaveStoredStream, "foo/+/bar")
                                 })
 
                                 Convey(`Will not ingest records if the stream is purged.`, func() {
                                         tcc.register(coordinator.LogStreamState{
                                                 Path:          "foo/+/bar",
                                                 Secret:        testSecret,
                                                 TerminalIndex: -1,
                                                 Purged:        true,
                                         })
 
                                         So(coll.Process(c, bb.bundle()), ShouldBeNil)
 
                                         So(tcc, shouldHaveRegisteredStream, "foo/+/bar", -1)
                                         So(st, shouldHaveStoredStream, "foo/+/bar")
                                 })
 
                                 Convey(`Will not ingest a bundle with no bundle entries.`, func() {
                                         So(coll.Process(c, bb.bundleWithEntries()), ShouldBeNil)
                                 })
 
                                 Convey(`Will not ingest a bundle whose log entries don't match their descriptor.`, func() {
-                                        be := bb.genBundleEntry("foo/+/bar", 4, 0, 1, 3, 4)
+                                        be := bb.genBundleEntry("foo/+/bar", 4, 0, 1, 2, 3, 4)
 
                                         // Add a binary log entry. This does NOT match the text descriptor, and
                                         // should fail validation.
                                         be.Logs = append(be.Logs, &logpb.LogEntry{
                                                 StreamIndex: 2,
                                                 Sequence:    2,
                                                 Content: &logpb.LogEntry_Binary{
                                                         &logpb.Binary{
                                                                 Data: []byte{0xd0, 0x6f, 0x00, 0xd5},
                                                         },
                                                 },
                                         })
                                         bb.addBundleEntry(be)
-                                        So(coll.Process(c, bb.bundle()), ShouldBeNil)
+                                        So(coll.Process(c, bb.bundle()), ShouldErrLike, "invalid log entry")
 
-                                        So(tcc, shouldHaveRegisteredStream, "foo/+/bar", 4)
-                                        So(st, shouldHaveStoredStream, "foo/+/bar", 0, 1, 3, 4)
+                                        So(tcc, shouldNotHaveRegisteredStream, "foo/+/bar")
                                 })
                         })
                 }
         })
 }