LogDog: Support per-namespace expired archival.

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"
+        "sort"
         "testing"
         "time"
 
         "github.com/julienschmidt/httprouter"
         ds "github.com/luci/gae/service/datastore"
+        tq "github.com/luci/gae/service/taskqueue"
         "github.com/luci/luci-go/appengine/gaetesting"
         "github.com/luci/luci-go/appengine/logdog/coordinator"
         ct "github.com/luci/luci-go/appengine/logdog/coordinator/coordinatorTest"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/clock/testclock"
+        memoryConfig "github.com/luci/luci-go/common/config/impl/memory"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/proto/google"
 
         . "github.com/smartystreets/goconvey/convey"
 )
 
 func TestHandleArchiveCron(t *testing.T) {
         t.Parallel()
 
+        const queueName = "test-archive-scan-queue"
+
+        Convey(`A testing environment`, t, func() {
+                c := gaetesting.TestingContext()
+
+                mcfg := map[string]memoryConfig.ConfigSet{}
+                c = memoryConfig.Use(c, mcfg)
+
+                svcStub := ct.Services{}
+                svcStub.InitConfig()
+                svcStub.ServiceConfig.Coordinator.ArchiveScanProjectQueueName = queueName
+                c = coordinator.WithServices(c, &svcStub)
+
+                b := Backend{}
+
+                tb := testBase{Context: c}
+                r := httprouter.New()
+                b.InstallHandlers(r, tb.base)
+
+                s := httptest.NewServer(r)
+                defer s.Close()
+
+                endpoint := fmt.Sprintf("%s/archive/cron", s.URL)
+
+                Convey(`When projects are registered`, func() {
+                        mcfg["projects/foo"] = memoryConfig.ConfigSet{}
+                        mcfg["projects/bar"] = memoryConfig.ConfigSet{}
+                        mcfg["projects/baz"] = memoryConfig.ConfigSet{}
+
+                        Convey(`When the testing queue exists`, func() {
+                                tq.Get(c).Testable().CreateQueue(queueName)
+
+                                Convey(`Will dispatch archive tasks for each registered project.`, func() {
+                                        resp, err := http.Get(endpoint)
+                                        So(err, ShouldBeNil)
+                                        So(resp.StatusCode, ShouldEqual, http.StatusOK)
+
+                                        var namespaces []string
+                                        for _, task := range tq.Get(c).Testable().GetScheduledTasks()[queueName] {
+                                                namespaces = append(namespaces, task.Header["X-Appengine-Current-Namespace"]...)
+                                        }
+                                        sort.Strings(namespaces)
+
+                                        pns := coordinator.ProjectNamespace
+                                        So(namespaces, ShouldResemble, []string{pns("bar"), pns("baz"), pns("foo")})
+                                })
+                        })
+                })
+        })
+}
+
+func TestHandleArchiveScanTask(t *testing.T) {
+        t.Parallel()
+
         Convey(`A testing environment`, t, func() {
                 c := gaetesting.TestingContext()
                 c, _ = testclock.UseTime(c, testclock.TestTimeUTC)
 
+                // In production, this task will be called with the namespace of is task
+                // queue entry installed. However, since we're calling it manually, we'll
+                // just fix the namespace.
+                if err := coordinator.WithProjectNamespace(&c, "test"); err != nil {
+                        panic(err)
+                }
+
                 // Add the archival index from "index.yaml".
                 ds.Get(c).Testable().AddIndexes(
                         &ds.IndexDefinition{
                                 Kind: "LogStream",
                                 SortBy: []ds.IndexColumn{
                                         {Property: "State"},
                                         {Property: "Created", Descending: true},
                                 },
                         },
                 )
@@ skipping 38 matching lines @@
 
                 b := Backend{}
 
                 tb := testBase{Context: c}
                 r := httprouter.New()
                 b.InstallHandlers(r, tb.base)
 
                 s := httptest.NewServer(r)
                 defer s.Close()
 
-                endpoint := fmt.Sprintf("%s/archive/cron", s.URL)
+                endpoint := fmt.Sprintf("%s/archive/cron/scan", s.URL)
 
                 Convey(`With no configuration loaded, the endpoint will fail.`, func() {
-                        resp, err := http.Get(endpoint)
+                        resp, err := http.Post(endpoint, "", nil)
                         So(err, ShouldBeNil)
                         So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
                 })
 
                 Convey(`With no task topic configured, the endpoint will fail.`, func() {
                         svcStub.InitConfig()
 
-                        resp, err := http.Get(endpoint)
+                        resp, err := http.Post(endpoint, "", nil)
                         So(err, ShouldBeNil)
                         So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
                 })
 
                 Convey(`With a configuration loaded`, func() {
                         svcStub.InitConfig()
                         svcStub.ServiceConfig.Coordinator.ArchiveTopic = "projects/test/topics/archive-test-topic"
                         svcStub.ServiceConfig.Coordinator.ArchiveSettleDelay = google.NewDuration(10 * time.Second)
                         svcStub.ServiceConfig.Coordinator.ArchiveDelayMax = google.NewDuration(10 * time.Minute)
 
                         Convey(`A request to the endpoint will be successful.`, func() {
-                                resp, err := http.Get(endpoint)
+                                resp, err := http.Post(endpoint, "", nil)
                                 So(err, ShouldBeNil)
                                 So(resp.StatusCode, ShouldEqual, http.StatusOK)
 
                                 // Candidate tasks should be scheduled.
                                 So(tap.StreamNames(), ShouldResemble, []string{"bar", "quux"})
 
                                 // 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)
+                                        resp, err = http.Post(endpoint, "", nil)
                                         So(err, ShouldBeNil)
                                         So(resp.StatusCode, ShouldEqual, http.StatusOK)
                                         So(tap.StreamNames(), ShouldResemble, []string{"bar", "quux"})
                                 })
                         })
 
                         Convey(`When scheduling multiple tasks`, func() {
                                 b.multiTaskBatchSize = 5
 
                                 // Create a lot of archival candidate tasks to schedule.
@@ skipping 11 matching lines @@
                                         if err := ds.Get(c).Put(ls); err != nil {
                                                 panic(err)
                                         }
 
                                         names = append(names, ls.Name)
                                 }
                                 ds.Get(c).Testable().CatchupIndexes()
 
                                 Convey(`Will schedule all pages properly.`, func() {
                                         // Ensure that all of these tasks get added to the task queue.
-                                        resp, err := http.Get(endpoint)
+                                        resp, err := http.Post(endpoint, "", nil)
                                         So(err, ShouldBeNil)
                                         So(resp.StatusCode, ShouldEqual, http.StatusOK)
                                         So(tap.StreamNames(), ShouldResemble, names)
 
                                         // Use this opportunity to assert that none of the scheduled streams
                                         // have any settle or completion delay.
                                         for _, at := range tap.Tasks() {
                                                 So(at.SettleDelay.Duration(), ShouldEqual, 0)
                                                 So(at.CompletePeriod.Duration(), ShouldEqual, 0)
                                         }
 
                                         Convey(`Will not schedule additional tasks on the next run.`, func() {
-                                                resp, err := http.Get(endpoint)
+                                                resp, err := http.Post(endpoint, "", nil)
                                                 So(err, ShouldBeNil)
                                                 So(resp.StatusCode, ShouldEqual, http.StatusOK)
                                                 So(tap.StreamNames(), ShouldResemble, names)
                                         })
                                 })
 
                                 Convey(`Will not schedule tasks if task scheduling fails.`, func() {
                                         tap.Err = errors.New("test error")
 
                                         // Ensure that all of these tasks get added to the task queue.
-                                        resp, err := http.Get(endpoint)
+                                        resp, err := http.Post(endpoint, "", nil)
                                         So(err, ShouldBeNil)
                                         So(resp.StatusCode, ShouldEqual, http.StatusInternalServerError)
                                         So(tap.StreamNames(), ShouldResemble, []string{})
 
                                         Convey(`And will schedule streams next run.`, func() {
                                                 tap.Err = nil
 
-                                                resp, err := http.Get(endpoint)
+                                                resp, err := http.Post(endpoint, "", nil)
                                                 So(err, ShouldBeNil)
                                                 So(resp.StatusCode, ShouldEqual, http.StatusOK)
                                                 So(tap.StreamNames(), ShouldResemble, names)
                                         })
                                 })
                         })
                 })
         })
 }