LogDog: Support per-namespace expired archival.

appengine/logdog/coordinator/endpoints/services/registerStream_test.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 services
 
 import (
         "bytes"
         "errors"
         "testing"
+        "time"
 
         "github.com/luci/gae/filter/featureBreaker"
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         ct "github.com/luci/luci-go/appengine/logdog/coordinator/coordinatorTest"
         "github.com/luci/luci-go/appengine/logdog/coordinator/hierarchy"
         "github.com/luci/luci-go/appengine/tumble"
         "github.com/luci/luci-go/common/api/logdog_coordinator/services/v1"
+        "github.com/luci/luci-go/common/clock"
+        "github.com/luci/luci-go/common/clock/testclock"
         "github.com/luci/luci-go/common/logdog/types"
+        "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/proto/logdog/logpb"
         "github.com/luci/luci-go/server/auth"
         "github.com/luci/luci-go/server/auth/authtest"
 
         . "github.com/luci/luci-go/common/testing/assertions"
         . "github.com/smartystreets/goconvey/convey"
 )
 
 func TestRegisterStream(t *testing.T) {
         t.Parallel()
 
         Convey(`With a testing configuration`, t, func() {
                 tt := &tumble.Testing{}
                 c := tt.Context()
+                tc := clock.Get(c).(testclock.TestClock)
+
+                // Add Tumble delayed mutation index.
+                tt.EnableDelayedMutations(c)
                 ds.Get(c).Testable().Consistent(true)
 
-                svcStub := ct.Services{}
+                var tap ct.ArchivalPublisher
+                svcStub := ct.Services{
+                        AP: func() (coordinator.ArchivalPublisher, error) {
+                                return &tap, nil
+                        },
+                }
                 svcStub.InitConfig()
                 svcStub.ServiceConfig.Coordinator.ServiceAuthGroup = "test-services"
+                svcStub.ServiceConfig.Coordinator.ArchiveDelayMax = google.NewDuration(time.Hour)
                 c = coordinator.WithServices(c, &svcStub)
 
                 svr := New()
 
                 fs := authtest.FakeState{}
                 c = auth.WithState(c, &fs)
 
                 Convey(`Returns Forbidden error if not a service.`, func() {
                         _, err := svr.RegisterStream(c, &logdog.RegisterStreamRequest{})
                         So(err, ShouldBeRPCPermissionDenied)
@@ skipping 16 matching lines @@
                                 expResp := &logdog.RegisterStreamResponse{
                                         State: &logdog.LogStreamState{
                                                 Path:          "testing/+/foo/bar",
                                                 ProtoVersion:  logpb.Version,
                                                 TerminalIndex: -1,
                                         },
                                         Secret: secret,
                                 }
 
                                 Convey(`Can register the stream.`, func() {
+                                        created := ds.RoundTime(tc.Now())
+
                                         resp, err := svr.RegisterStream(c, &req)
                                         So(err, ShouldBeRPCOK)
                                         So(resp, ShouldResemble, expResp)
                                         ds.Get(c).Testable().CatchupIndexes()
                                         tt.Drain(c)
 
+                                        ls := coordinator.LogStreamFromPath(types.StreamPath(req.Path))
+                                        So(ds.Get(c).Get(ls), ShouldBeNil)
+                                        So(ls.Created, ShouldResemble, created)
+
+                                        // No archival request yet.
+                                        So(tap.StreamNames(), ShouldResemble, []string{})
+
                                         // Should have name components.
                                         getNameComponents := func(b string) []string {
                                                 l, err := hierarchy.Get(ds.Get(c), hierarchy.Request{Base: b})
                                                 if err != nil {
                                                         panic(err)
                                                 }
                                                 names := make([]string, len(l.Comp))
                                                 for i, e := range l.Comp {
                                                         names[i] = e.Name
                                                         if e.Stream != "" {
                                                                 names[i] += "$"
                                                         }
                                                 }
                                                 return names
                                         }
                                         So(getNameComponents(""), ShouldResemble, []string{"testing"})
                                         So(getNameComponents("testing"), ShouldResemble, []string{"+"})
                                         So(getNameComponents("testing/+"), ShouldResemble, []string{"foo"})
                                         So(getNameComponents("testing/+/foo"), ShouldResemble, []string{"bar$"})
 
                                         Convey(`Can register the stream again (idempotent).`, func() {
+                                                tc.Set(created.Add(10 * time.Minute))
+
                                                 resp, err := svr.RegisterStream(c, &req)
                                                 So(err, ShouldBeRPCOK)
                                                 So(resp, ShouldResemble, expResp)
+
+                                                ls := coordinator.LogStreamFromPath(types.StreamPath(req.Path))
+                                                So(ds.Get(c).Get(ls), ShouldBeNil)
+                                                So(ls.Created, ShouldResemble, created)
+
+                                                // No archival request yet.
+                                                So(tap.StreamNames(), ShouldResemble, []string{})
+
+                                                Convey(`Forces an archival request after first archive expiration.`, func() {
+                                                        tc.Set(created.Add(time.Hour)) // 1 hour after initial registration.
+                                                        tt.Drain(c)
+
+                                                        So(tap.StreamNames(), ShouldResemble, []string{ls.Name})
+                                                })
                                         })
 
                                         Convey(`Will not re-register if scerets don't match.`, func() {
                                                 req.Secret[0] = 0xAB
                                                 _, err := svr.RegisterStream(c, &req)
                                                 So(err, ShouldBeRPCAlreadyExists, "Log stream is already incompatibly registered")
                                         })
 
                                         Convey(`Will not re-register if descriptor data differs.`, func() {
                                                 req.Desc.Tags = map[string]string{
@@ skipping 68 matching lines @@
                                                 So(req.Desc.Validate(true), ShouldNotBeNil)
 
                                                 _, err := svr.RegisterStream(c, &req)
                                                 So(err, ShouldBeRPCInvalidArgument, "Invalid log stream descriptor")
                                         })
                                 })
                         })
                 })
         })
 }