LogDog: Update to archival V2.

appengine/logdog/coordinator/backend/archiveCron_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 backend
 
 import (
         "fmt"
         "net/http"
         "net/http/httptest"
@@ skipping 22 matching lines @@
         Convey(`A testing environment`, t, func() {
                 c := gaetesting.TestingContext()
                 c, _ = testclock.UseTime(c, testclock.TestTimeUTC)
 
                 // Add the archival index from "index.yaml".
                 ds.Get(c).Testable().AddIndexes(
                         &ds.IndexDefinition{
                                 Kind: "LogStream",
                                 SortBy: []ds.IndexColumn{
                                         {Property: "State"},
-                                        {Property: "Updated"},
+                                        {Property: "Created", Descending: true},
                                 },
                         },
                 )
 
                 // Install a base set of streams.
+                //
+                // If the stream was created >= 10min ago, it will be candidate for
+                // archival.
+                now := clock.Now(c)
                 for _, v := range []struct {
-                        name string
-                        d    time.Duration
-                        term bool
+                        name  string
+                        d     time.Duration // Time before "now" for this streams Created.
+                        state coordinator.LogStreamState
                 }{
-                        {"foo", 0, true},                 // Not candidate for archival (too soon).
-                        {"bar", 10 * time.Minute, true},  // Candidate for archival.
-                        {"baz", 10 * time.Minute, false}, // Not candidate for archival (not terminal).
-                        {"qux", 24 * time.Hour, false},   // Candidate for non-terminal archival.
+                        {"foo", 0, coordinator.LSStreaming},                    // Not candidate for archival (too soon).
+                        {"bar", 10 * time.Minute, coordinator.LSStreaming},     // Candidate for archival.
+                        {"baz", 10 * time.Minute, coordinator.LSArchiveTasked}, // Not candidate for archival (archive tasked).
+                        {"qux", 10 * time.Hour, coordinator.LSArchived},        // Not candidate for archival (archived).
+                        {"quux", 10 * time.Hour, coordinator.LSStreaming},      // Candidate for archival.
                 } {
                         ls := ct.TestLogStream(c, ct.TestLogStreamDescriptor(c, v.name))
 
                         // The entry was created a week ago.
-                        ls.Created = clock.Now(c).Add(-(7 * 24 * time.Hour))
-                        ls.Updated = clock.Now(c).Add(-v.d)
-                        if v.term {
-                                ls.State = coordinator.LSTerminated
-                        }
-                        if err := ls.Put(ds.Get(c)); err != nil {
+                        ls.Created = now.Add(-v.d)
+                        ls.State = v.state
+                        ls.TerminatedTime = now // Doesn't matter, but required to Put the stream.
+                        ls.ArchivedTime = now   // Doesn't matter, but required to Put the stream.
+                        if err := ds.Get(c).Put(ls); err != nil {
                                 panic(err)
                         }
                 }
                 ds.Get(c).Testable().CatchupIndexes()
 
                 // This allows us to update our Context in test setup.
-                b := Backend{
-                        multiTaskBatchSize: 5,
-                }
+                b := Backend{}
 
                 tb := testBase{Context: c}
                 r := httprouter.New()
                 b.InstallHandlers(r, tb.base)
 
                 s := httptest.NewServer(r)
                 defer s.Close()
 
-                Convey(`With no configuration loaded`, func() {
-                        for _, ep := range []string{"terminal", "nonterminal"} {
-                                Convey(fmt.Sprintf(`A %q endpoint hit will fail.`, ep), func() {
-                                        resp, err := http.Get(fmt.Sprintf("%s/archive/cron/%s", s.URL, ep))
-                                        So(err, ShouldBeNil)
-                                        So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
-                                })
-                        }
+                endpoint := fmt.Sprintf("%s/archive/cron", s.URL)
+
+                Convey(`With no configuration loaded, the endpoint will fail.`, func() {
+                        resp, err := http.Get(endpoint)
+                        So(err, ShouldBeNil)
+                        So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
                 })
 
                 Convey(`With a configuration loaded`, func() {
                         qName := "archive-test-queue"
                         c = ct.UseConfig(c, &svcconfig.Coordinator{
-                                ArchiveDelay:     google.NewDuration(10 * time.Minute),
-                                ArchiveDelayMax:  google.NewDuration(24 * time.Hour),
                                 ArchiveTaskQueue: qName,
+                                ArchiveDelayMax:  google.NewDuration(10 * time.Minute),
                         })
                         tb.Context = c
 
-                        Convey(`With no task queue configured`, func() {
-                                for _, ep := range []string{"terminal", "nonterminal", "purge"} {
-                                        Convey(fmt.Sprintf(`A %q endpoint hit will fail.`, ep), func() {
-                                                resp, err := http.Get(fmt.Sprintf("%s/archive/cron/%s", s.URL, ep))
-                                                So(err, ShouldBeNil)
-                                                So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
-                                        })
-                                }
+                        Convey(`With no task queue configured, the endpoint will fail.`, func() {
+                                resp, err := http.Get(endpoint)
+                                So(err, ShouldBeNil)
+                                So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
                         })
 
                         Convey(`With a task queue configured`, func() {
                                 tq.Get(c).Testable().CreateQueue(qName)
 
-                                Convey(`A terminal endpoint hit will be successful and idempotent.`, func() {
-                                        resp, err := http.Get(fmt.Sprintf("%s/archive/cron/terminal", s.URL))
+                                streams := func() []string {
+                                        s, err := ct.GetArchiveTaskStreams(tq.Get(c), qName)
+                                        if err != nil {
+                                                panic(err)
+                                        }
+                                        return s
+                                }
+
+                                Convey(`A request to the endpoint will be successful.`, func() {
+                                        resp, err := http.Get(endpoint)
                                         So(err, ShouldBeNil)
                                         So(resp.StatusCode, ShouldEqual, http.StatusOK)
 
                                         // Candidate tasks should be scheduled.
-                                        tasks := tq.Get(c).Testable().GetScheduledTasks()[qName]
-                                        So(tasks, shouldHaveTasks, archiveTaskName("testing/+/bar"))
+                                        So(streams(), ShouldResemble, []string{"bar", "quux"})
 
-                                        // Hit the endpoint again, the same tasks should be scheduled.
-                                        resp, err = http.Get(fmt.Sprintf("%s/archive/cron/terminal", s.URL))
-                                        So(err, ShouldBeNil)
-                                        So(resp.StatusCode, ShouldEqual, http.StatusOK)
-
-                                        tasks2 := tq.Get(c).Testable().GetScheduledTasks()[qName]
-                                        So(tasks2, ShouldResemble, tasks)
-                                })
-
-                                Convey(`A non-terminal endpoint hit will be successful and idempotent.`, func() {
-                                        resp, err := http.Get(fmt.Sprintf("%s/archive/cron/nonterminal", s.URL))
-                                        So(err, ShouldBeNil)
-                                        So(resp.StatusCode, ShouldEqual, http.StatusOK)
-
-                                        // Candidate tasks should be scheduled.
-                                        tasks := tq.Get(c).Testable().GetScheduledTasks()[qName]
-                                        So(tasks, shouldHaveTasks, archiveTaskName("testing/+/qux"))
-
-                                        // Hit the endpoint again, the same tasks should be scheduled.
-                                        resp, err = http.Get(fmt.Sprintf("%s/archive/cron/nonterminal", s.URL))
-                                        So(err, ShouldBeNil)
-                                        So(resp.StatusCode, ShouldEqual, http.StatusOK)
-
-                                        tasks2 := tq.Get(c).Testable().GetScheduledTasks()[qName]
-                                        So(tasks2, ShouldResemble, tasks)
-                                })
-
-                                Convey(`A terminal endpoint hit followed by a non-terminal endpoint hit will be successful.`, func() {
-                                        resp, err := http.Get(fmt.Sprintf("%s/archive/cron/terminal", s.URL))
-                                        So(err, ShouldBeNil)
-                                        So(resp.StatusCode, ShouldEqual, http.StatusOK)
-                                        So(tq.Get(c).Testable().GetScheduledTasks()[qName], shouldHaveTasks, archiveTaskName("testing/+/bar"))
-
-                                        resp, err = http.Get(fmt.Sprintf("%s/archive/cron/nonterminal", s.URL))
-                                        So(err, ShouldBeNil)
-                                        So(resp.StatusCode, ShouldEqual, http.StatusOK)
-                                        So(tq.Get(c).Testable().GetScheduledTasks()[qName], shouldHaveTasks,
-                                                archiveTaskName("testing/+/bar"), archiveTaskName("testing/+/qux"))
+                                        // Hit the endpoint again, no additional tasks should be scheduled.
+                                        Convey(`A subsequent endpoint hit will not schedule any additional tasks.`, func() {
+                                                resp, err = http.Get(endpoint)
+                                                So(err, ShouldBeNil)
+                                                So(resp.StatusCode, ShouldEqual, http.StatusOK)
+                                                So(streams(), ShouldResemble, []string{"bar", "quux"})
+                                        })
                                 })
 
                                 Convey(`When scheduling multiple tasks`, func() {
+                                        b.multiTaskBatchSize = 5
+
                                         // Create a lot of archival candidate tasks to schedule.
-                                        var names []string
+                                        //
+                                        // Note that since task queue names are returned sorted, we want to
+                                        // name these streams greater than our stock stream names.
+                                        names := []string{"bar", "quux"}
                                         for i := 0; i < 11; i++ {
-                                                ls := ct.TestLogStream(c, ct.TestLogStreamDescriptor(c, fmt.Sprintf("stream-%d", i)))
+                                                ls := ct.TestLogStream(c, ct.TestLogStreamDescriptor(c, fmt.Sprintf("stream-%02d", i)))
 
-                                                ls.Created = clock.Now(c).Add(-(10 * time.Minute))
-                                                ls.Updated = ls.Created
-                                                ls.State = coordinator.LSTerminated
+                                                ls.Created = now.Add(-(10 * time.Minute))
+                                                ls.State = coordinator.LSStreaming
                                                 ls.TerminalIndex = 1337
-                                                if err := ls.Put(ds.Get(c)); err != nil {
+                                                ls.TerminatedTime = now
+                                                if err := ds.Get(c).Put(ls); err != nil {
                                                         panic(err)
                                                 }
 
                                                 names = append(names, ls.Name)
                                         }
-                                        names = append(names, "bar")
                                         ds.Get(c).Testable().CatchupIndexes()
 
                                         Convey(`Will schedule all pages properly.`, func() {
-                                                taskNames := make([]interface{}, len(names))
-                                                for i, n := range names {
-                                                        taskNames[i] = archiveTaskName(fmt.Sprintf("testing/+/%s", n))
+                                                // Ensure that all of these tasks get added to the task queue.
+                                                resp, err := http.Get(endpoint)
+                                                So(err, ShouldBeNil)
+                                                So(resp.StatusCode, ShouldEqual, http.StatusOK)
+                                                So(streams(), ShouldResemble, names)
+
+                                                // Verify that all tasks are configured to execute immediately.
+                                                for _, t := range tq.Get(c).Testable().GetScheduledTasks()[qName] {
+                                                        So(t.ETA, ShouldResemble, now)
                                                 }
 
-                                                // Ensure that all of these tasks get added to the task queue.
-                                                resp, err := http.Get(fmt.Sprintf("%s/archive/cron/terminal", s.URL))
-                                                So(err, ShouldBeNil)
-                                                So(resp.StatusCode, ShouldEqual, http.StatusOK)
-                                                So(tq.Get(c).Testable().GetScheduledTasks()[qName], shouldHaveTasks, taskNames...)
-
-                                                Convey(`Will be successful when rescheduling the same tasks.`, func() {
-                                                        resp, err := http.Get(fmt.Sprintf("%s/archive/cron/terminal", s.URL))
+                                                Convey(`Will not schedule additional tasks on the next run.`, func() {
+                                                        resp, err := http.Get(endpoint)
                                                         So(err, ShouldBeNil)
                                                         So(resp.StatusCode, ShouldEqual, http.StatusOK)
-                                                        So(tq.Get(c).Testable().GetScheduledTasks()[qName], shouldHaveTasks, taskNames...)
+                                                        So(streams(), ShouldResemble, names)
                                                 })
                                         })
 
-                                        Convey(`Will return an error if task scheduling fails.`, func() {
+                                        Convey(`Will not schedule tasks if task scheduling fails.`, func() {
                                                 c, fb := featureBreaker.FilterTQ(c, nil)
                                                 tb.Context = c
                                                 fb.BreakFeatures(errors.New("test error"), "AddMulti")
 
                                                 // Ensure that all of these tasks get added to the task queue.
-                                                resp, err := http.Get(fmt.Sprintf("%s/archive/cron/terminal", s.URL))
+                                                resp, err := http.Get(endpoint)
                                                 So(err, ShouldBeNil)
                                                 So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
-                                        })
+                                                So(streams(), ShouldResemble, []string{})
 
-                                        Convey(`Will return an error if a single task scheduling fails.`, func() {
-                                                merr := make(errors.MultiError, len(names))
-                                                merr[0] = errors.New("test error")
+                                                Convey(`And will schedule streams next run.`, func() {
+                                                        fb.UnbreakFeatures("AddMulti")
 
-                                                c, fb := featureBreaker.FilterTQ(c, nil)
-                                                tb.Context = c
-                                                fb.BreakFeatures(merr, "AddMulti")
-
-                                                // Ensure that all of these tasks get added to the task queue.
-                                                resp, err := http.Get(fmt.Sprintf("%s/archive/cron/terminal", s.URL))
-                                                So(err, ShouldBeNil)
-                                                So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
-                                        })
-
-                                        Convey(`And a purge endpoint hit will purge the tasks.`, func() {
-                                                resp, err := http.Get(fmt.Sprintf("%s/archive/cron/purge", s.URL))
-                                                So(err, ShouldBeNil)
-                                                So(resp.StatusCode, ShouldEqual, http.StatusOK)
-                                                So(tq.Get(c).Testable().GetScheduledTasks()[qName], shouldHaveTasks)
+                                                        resp, err := http.Get(endpoint)
+                                                        So(err, ShouldBeNil)
+                                                        So(resp.StatusCode, ShouldEqual, http.StatusOK)
+                                                        So(streams(), ShouldResemble, names)
+                                                })
                                         })
                                 })
                         })
                 })
         })
 }