[task scheduler] Add ModifiedComments
Adds full support for GetModifiedComments, both in-memory and pubsub.
Pass the unified ModifiedTasks+ModifiedJobs+ModifiedComments into the DB
constructors rather than individually.
Bug: skia:
Change-Id: I68386757fd95db81aeaa16d9ac8b34ec928c9b12
Reviewed-on: https://skia-review.googlesource.com/c/177722
Commit-Queue: Eric Boren <borenet@google.com>
Reviewed-by: Ben Wagner <benjaminwagner@google.com>
diff --git a/datahopper/go/datahopper/main.go b/datahopper/go/datahopper/main.go
index 9ea7ef0..b13cf27 100644
--- a/datahopper/go/datahopper/main.go
+++ b/datahopper/go/datahopper/main.go
@@ -7,6 +7,7 @@
import (
"context"
"flag"
+ "fmt"
"os"
"path"
"path/filepath"
@@ -42,10 +43,9 @@
taskSchedulerDbUrl = flag.String("task_db_url", "http://skia-task-scheduler:8008/db/", "Where the Skia task scheduler database is hosted.")
workdir = flag.String("workdir", ".", "Working directory used by data processors.")
- perfBucket = flag.String("perf_bucket", "skia-perf", "The GCS bucket that should be used for writing into perf")
- perfPrefix = flag.String("perf_duration_prefix", "task-duration", "The folder name in the bucket that task duration metric shoudl be written.")
- tasksPubsubTopic = flag.String("pubsub_topic_tasks", pubsub.TOPIC_TASKS, "Pubsub topic for tasks.")
- jobsPubsubTopic = flag.String("pubsub_topic_jobs", pubsub.TOPIC_JOBS, "Pubsub topic for jobs.")
+ perfBucket = flag.String("perf_bucket", "skia-perf", "The GCS bucket that should be used for writing into perf")
+ perfPrefix = flag.String("perf_duration_prefix", "task-duration", "The folder name in the bucket that task duration metric shoudl be written.")
+ pubsubTopicSet = flag.String("pubsub_topic_set", "", fmt.Sprintf("Pubsub topic set; one of: %v", pubsub.VALID_TOPIC_SETS))
)
var (
@@ -166,20 +166,16 @@
var d db.RemoteDB
if *firestoreInstance != "" {
label := "datahopper"
- modTasks, err := pubsub.NewModifiedTasks(*tasksPubsubTopic, label, newTs)
+ mod, err := pubsub.NewModifiedData(*pubsubTopicSet, label, newTs)
if err != nil {
sklog.Fatal(err)
}
- modJobs, err := pubsub.NewModifiedJobs(*jobsPubsubTopic, label, newTs)
- if err != nil {
- sklog.Fatal(err)
- }
- d, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, newTs, modTasks, modJobs)
+ d, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, newTs, mod)
if err != nil {
sklog.Fatalf("Failed to create Firestore DB client: %s", err)
}
} else {
- d, err = remote_db.NewClient(*taskSchedulerDbUrl, *tasksPubsubTopic, *jobsPubsubTopic, "datahopper", newTs)
+ d, err = remote_db.NewClient(*taskSchedulerDbUrl, *pubsubTopicSet, "datahopper", newTs)
if err != nil {
sklog.Fatal(err)
}
diff --git a/go/util/gob.go b/go/util/gob.go
new file mode 100644
index 0000000..207f93e
--- /dev/null
+++ b/go/util/gob.go
@@ -0,0 +1,222 @@
+package util
+
+import (
+ "bytes"
+ "encoding/gob"
+ "sync"
+)
+
+const kNumDecoderGoroutines = 10
+
+// GobEncoder encodes structs into bytes via GOB encoding. Not safe for
+// concurrent use.
+//
+// Here's a template for writing a type-specific encoder:
+//
+// // FooEncoder encodes Foos into bytes via GOB encoding. Not safe for
+// // concurrent use.
+// type FooEncoder {
+// util.GobEncoder
+// }
+//
+// // Next returns one of the Foox provided to Process (in arbitrary order) and
+// // its serialized bytes. If any items remain, returns the item, the
+// // serialized bytes, nil. If all items have been returned, returns nil, nil,
+// // nil. If an error is encountered, returns nil, nil, error.
+// func (e *FooEncoder) Next() (*Foo, []byte, error) {
+// item, serialized, err := e.GobEncoder.Next()
+// if err != nil {
+// return nil, nil, err
+// } else if item == nil {
+// return nil, nil, nil
+// }
+// return item.(*Foo), serialized, nil
+// }
+type GobEncoder struct {
+ err error
+ items []interface{}
+ result [][]byte
+}
+
+// Process encodes the item into a byte slice that will be returned from
+// Next() (in arbitrary order). Returns false if Next is certain to return an
+// error. Caller must ensure item does not change until after the first call to
+// Next(). May not be called after calling Next().
+func (e *GobEncoder) Process(item interface{}) bool {
+ if e.err != nil {
+ return false
+ }
+ var buf bytes.Buffer
+ if err := gob.NewEncoder(&buf).Encode(item); err != nil {
+ e.err = err
+ e.items = nil
+ e.result = nil
+ return false
+ }
+ e.items = append(e.items, item)
+ e.result = append(e.result, buf.Bytes())
+ return true
+}
+
+// Next returns one of the items provided to Process (in arbitrary order) and
+// its serialized bytes. If any items remain, returns the item, the serialized
+// bytes, nil. If all items have been returned, returns nil, nil, nil. If an
+// error is encountered, returns nil, nil, error.
+func (e *GobEncoder) Next() (interface{}, []byte, error) {
+ if e.err != nil {
+ return nil, nil, e.err
+ }
+ if len(e.items) == 0 {
+ return nil, nil, nil
+ }
+ c := e.items[0]
+ e.items = e.items[1:]
+ serialized := e.result[0]
+ e.result = e.result[1:]
+ return c, serialized, nil
+}
+
+// GobDecoder decodes bytes into structs via GOB decoding. Not safe for
+// concurrent use.
+//
+// Here's a template for writing a type-specific decoder:
+//
+// FooDecoder decodes bytes into Foos via GOB decoding. Not safe for
+// concurrent use.
+// type FooDecoder struct {
+// *util.GobDecoder
+// }
+//
+// // NewFooDecoder returns a FooDecoder instance.
+// func NewFooDecoder() *FooDecoder {
+// return &FooDecoder{
+// GobDecoder: util.NewGobDecoder(func() interface{} {
+// return &Foo{}
+// }, func(ch <-chan interface{}) interface{} {
+// items := []*Foo{}
+// for item := range ch {
+// items = append(items, item.(*Foo))
+// }
+// return items
+// }),
+// }
+// }
+//
+// // Result returns all decoded Foos provided to Process (in arbitrary order), or
+// // any error encountered.
+// func (d *FooDecoder) Result() ([]*Foo, error) {
+// res, err := d.GobDecoder.Result()
+// if err != nil {
+// return nil, err
+// }
+// return res.([]*Foo), nil
+// }
+type GobDecoder struct {
+ // input contains the incoming byte slices. Process() sends on this
+ // channel, decode() receives from it, and Result() closes it.
+ input chan []byte
+ // output contains decoded items. decode() sends on this channel,
+ // collect() receives from it, and run() closes it when all decode()
+ // goroutines have finished.
+ output chan interface{}
+ // result contains the return value of Result(). collect() sends a single
+ // value on this channel and closes it. Result() receives from it.
+ result chan interface{}
+ // errors contains the first error from any goroutine. It's a channel in
+ // case multiple goroutines experience an error at the same time.
+ errors chan error
+
+ newItem func() interface{}
+ collectImpl func(<-chan interface{}) interface{}
+}
+
+// NewGobDecoder returns a GobDecoder instance. The first argument is a
+// goroutine-safe function which returns a zero-valued instance of the type
+// being decoded, eg.
+//
+// func() interface{} {
+// return &MyType{}
+// }
+//
+// The second argument is a function which collects decoded instances of that
+// type from a channel and returns a slice, eg.
+//
+// func(ch <-chan interface{}) interface{} {
+// items := []*MyType{}
+// for item := range ch {
+// items = append(items, item.(*MyType))
+// }
+// return items
+// }
+func NewGobDecoder(newItem func() interface{}, collect func(<-chan interface{}) interface{}) *GobDecoder {
+ d := &GobDecoder{
+ input: make(chan []byte, kNumDecoderGoroutines*2),
+ output: make(chan interface{}, kNumDecoderGoroutines),
+ result: make(chan interface{}, 1),
+ errors: make(chan error, kNumDecoderGoroutines),
+ newItem: newItem,
+ collectImpl: collect,
+ }
+ go d.run()
+ go d.collect()
+ return d
+}
+
+// run starts the decode goroutines and closes d.output when they finish.
+func (d *GobDecoder) run() {
+ // Start decoders.
+ wg := sync.WaitGroup{}
+ for i := 0; i < kNumDecoderGoroutines; i++ {
+ wg.Add(1)
+ go d.decode(&wg)
+ }
+ // Wait for decoders to exit.
+ wg.Wait()
+ // Drain d.input in the case that errors were encountered, to avoid deadlock.
+ for range d.input {
+ }
+ close(d.output)
+}
+
+// decode receives from d.input and sends to d.output until d.input is closed or
+// d.errors is non-empty. Decrements wg when done.
+func (d *GobDecoder) decode(wg *sync.WaitGroup) {
+ for b := range d.input {
+ item := d.newItem()
+ if err := gob.NewDecoder(bytes.NewReader(b)).Decode(item); err != nil {
+ d.errors <- err
+ break
+ }
+ d.output <- item
+ if len(d.errors) > 0 {
+ break
+ }
+ }
+ wg.Done()
+}
+
+// collect receives from d.output until it is closed, then sends on d.result.
+func (d *GobDecoder) collect() {
+ d.result <- d.collectImpl(d.output)
+ close(d.result)
+}
+
+// Process decodes the byte slice and includes it in Result() (in arbitrary
+// order). Returns false if Result is certain to return an error. Caller must
+// ensure b does not change until after Result() returns.
+func (d *GobDecoder) Process(b []byte) bool {
+ d.input <- b
+ return len(d.errors) == 0
+}
+
+// Result returns all decoded items provided to Process (in arbitrary order), or
+// any error encountered.
+func (d *GobDecoder) Result() (interface{}, error) {
+ close(d.input)
+ select {
+ case err := <-d.errors:
+ return nil, err
+ case result := <-d.result:
+ return result, nil
+ }
+}
diff --git a/go/util/gob_test.go b/go/util/gob_test.go
new file mode 100644
index 0000000..ec63312
--- /dev/null
+++ b/go/util/gob_test.go
@@ -0,0 +1,138 @@
+package util
+
+import (
+ "bytes"
+ "encoding/gob"
+ "fmt"
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "go.skia.org/infra/go/deepequal"
+ "go.skia.org/infra/go/testutils"
+)
+
+type Item struct {
+ Id string
+ Map map[string]string
+ Slice []string
+}
+
+func TestGobEncoder(t *testing.T) {
+ testutils.SmallTest(t)
+ // TODO(benjaminwagner): Is there any way to cause an error?
+ e := GobEncoder{}
+ expectedItems := map[*Item][]byte{}
+ for i := 0; i < 25; i++ {
+ item := &Item{}
+ item.Id = fmt.Sprintf("Id-%d", i)
+ item.Map = map[string]string{"PointA": "PointB"}
+ item.Slice = []string{"bread"}
+ var buf bytes.Buffer
+ err := gob.NewEncoder(&buf).Encode(item)
+ assert.NoError(t, err)
+ expectedItems[item] = buf.Bytes()
+ assert.True(t, e.Process(item))
+ }
+
+ actualItems := map[*Item][]byte{}
+ for item, serialized, err := e.Next(); item != nil; item, serialized, err = e.Next() {
+ assert.NoError(t, err)
+ actualItems[item.(*Item)] = serialized
+ }
+
+ deepequal.AssertDeepEqual(t, expectedItems, actualItems)
+}
+
+func TestGobEncoderNoItems(t *testing.T) {
+ testutils.SmallTest(t)
+ e := GobEncoder{}
+ item, serialized, err := e.Next()
+ assert.NoError(t, err)
+ assert.Nil(t, item)
+ assert.Nil(t, serialized)
+}
+
+func TestGobDecoder(t *testing.T) {
+ testutils.SmallTest(t)
+ d := NewGobDecoder(func() interface{} {
+ return &Item{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*Item{}
+ for item := range ch {
+ items = append(items, item.(*Item))
+ }
+ return items
+ })
+ expectedItems := map[string]*Item{}
+ for i := 0; i < 250; i++ {
+ item := &Item{}
+ item.Id = fmt.Sprintf("Id-%d", i)
+ item.Map = map[string]string{"PointA": "PointB"}
+ item.Slice = []string{"bread"}
+ var buf bytes.Buffer
+ err := gob.NewEncoder(&buf).Encode(item)
+ assert.NoError(t, err)
+ expectedItems[item.Id] = item
+ assert.True(t, d.Process(buf.Bytes()))
+ }
+
+ actualItems := map[string]*Item{}
+ iResult, err := d.Result()
+ assert.NoError(t, err)
+ result := iResult.([]*Item)
+ assert.Equal(t, len(expectedItems), len(result))
+ for _, item := range result {
+ actualItems[item.Id] = item
+ }
+ deepequal.AssertDeepEqual(t, expectedItems, actualItems)
+}
+
+func TestGobDecoderNoItems(t *testing.T) {
+ testutils.SmallTest(t)
+ d := NewGobDecoder(func() interface{} {
+ return &Item{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*Item{}
+ for item := range ch {
+ items = append(items, item.(*Item))
+ }
+ return items
+ })
+ result, err := d.Result()
+ assert.NoError(t, err)
+ assert.Equal(t, 0, len(result.([]*Item)))
+}
+
+func TestGobDecoderError(t *testing.T) {
+ testutils.SmallTest(t)
+ item := &Item{}
+ item.Id = "Id"
+ var buf bytes.Buffer
+ err := gob.NewEncoder(&buf).Encode(item)
+ assert.NoError(t, err)
+ serialized := buf.Bytes()
+ invalid := append([]byte("Hi Mom!"), serialized...)
+
+ d := NewGobDecoder(func() interface{} {
+ return &Item{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*Item{}
+ for item := range ch {
+ items = append(items, item.(*Item))
+ }
+ return items
+ })
+ // Process should return true before it encounters an invalid result.
+ assert.True(t, d.Process(serialized))
+ assert.True(t, d.Process(serialized))
+ // Process may return true or false after encountering an invalid value.
+ _ = d.Process(invalid)
+ for i := 0; i < 250; i++ {
+ _ = d.Process(serialized)
+ }
+
+ // Result should return error.
+ result, err := d.Result()
+ assert.Error(t, err)
+ assert.Nil(t, result)
+}
diff --git a/status/go/incremental/incremental_test.go b/status/go/incremental/incremental_test.go
index ad977cd..2cf6955 100644
--- a/status/go/incremental/incremental_test.go
+++ b/status/go/incremental/incremental_test.go
@@ -21,9 +21,7 @@
func setup(t *testing.T) (context.Context, string, *IncrementalCache, repograph.Map, db.DB, *git_testutils.GitBuilder, func()) {
testutils.LargeTest(t)
- taskDb := memory.NewInMemoryTaskDB(nil)
- commentDb := &db.CommentBox{}
- d := db.NewDB(taskDb, nil, commentDb)
+ d := memory.NewInMemoryDB(nil)
ctx := context.Background()
gb := git_testutils.GitInit(t, ctx)
@@ -48,7 +46,7 @@
Name: "DummyTask",
},
}
- assert.NoError(t, taskDb.PutTask(initialTask))
+ assert.NoError(t, d.PutTask(initialTask))
w, err := window.New(24*time.Hour, 100, repos)
assert.NoError(t, err)
diff --git a/status/go/status/main.go b/status/go/status/main.go
index 7a10288..03f0100 100644
--- a/status/go/status/main.go
+++ b/status/go/status/main.go
@@ -101,8 +101,7 @@
testing = flag.Bool("testing", false, "Set to true for locally testing rules. No email will be sent.")
useMetadata = flag.Bool("use_metadata", true, "Load sensitive values from metadata not from flags.")
workdir = flag.String("workdir", ".", "Directory to use for scratch work.")
- pubsubTopicTasks = flag.String("pubsub_topic_tasks", pubsub.TOPIC_TASKS, "Pubsub topic for tasks.")
- pubsubTopicJobs = flag.String("pubsub_topic_jobs", pubsub.TOPIC_JOBS, "Pubsub topic for jobs.")
+ pubsubTopicSet = flag.String("pubsub_topic_set", "", fmt.Sprintf("Pubsub topic set; one of: %v", pubsub.VALID_TOPIC_SETS))
repos repograph.Map
)
@@ -729,7 +728,7 @@
// Create remote Tasks DB.
if *testing {
- taskDb, err = local_db.NewDB("status-testing", path.Join(*workdir, "status-testing.bdb"), nil, nil)
+ taskDb, err = local_db.NewDB("status-testing", path.Join(*workdir, "status-testing.bdb"), nil)
if err != nil {
sklog.Fatalf("Failed to create local task DB: %s", err)
}
@@ -737,22 +736,18 @@
} else if *firestoreInstance != "" {
sklog.Infof("Creating firestore DB.")
label := *host
- modTasks, err := pubsub.NewModifiedTasks(*pubsubTopicTasks, label, ts)
+ mod, err := pubsub.NewModifiedData(*pubsubTopicSet, label, ts)
if err != nil {
sklog.Fatal(err)
}
- modJobs, err := pubsub.NewModifiedJobs(*pubsubTopicJobs, label, ts)
- if err != nil {
- sklog.Fatal(err)
- }
- taskDb, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, ts, modTasks, modJobs)
+ taskDb, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, ts, mod)
if err != nil {
sklog.Fatalf("Failed to create Firestore DB client: %s", err)
}
} else {
sklog.Infof("Creating remote DB.")
label := *host
- taskDb, err = remote_db.NewClient(*taskSchedulerDbUrl, *pubsubTopicTasks, *pubsubTopicJobs, label, ts)
+ taskDb, err = remote_db.NewClient(*taskSchedulerDbUrl, *pubsubTopicSet, label, ts)
if err != nil {
sklog.Fatalf("Failed to create remote task DB: %s", err)
}
diff --git a/status/go/statusk/main.go b/status/go/statusk/main.go
index 780d280..9334ff1 100644
--- a/status/go/statusk/main.go
+++ b/status/go/statusk/main.go
@@ -109,8 +109,7 @@
testing = flag.Bool("testing", false, "Set to true for locally testing rules. No email will be sent.")
useMetadata = flag.Bool("use_metadata", true, "Load sensitive values from metadata not from flags.")
workdir = flag.String("workdir", ".", "Directory to use for scratch work.")
- pubsubTopicTasks = flag.String("pubsub_topic_tasks", pubsub.TOPIC_TASKS, "Pubsub topic for tasks.")
- pubsubTopicJobs = flag.String("pubsub_topic_jobs", pubsub.TOPIC_JOBS, "Pubsub topic for jobs.")
+ pubsubTopicSet = flag.String("pubsub_topic_set", "", fmt.Sprintf("Pubsub topic set; one of: %v", pubsub.VALID_TOPIC_SETS))
repos repograph.Map
)
@@ -716,7 +715,7 @@
// Create remote Tasks DB.
if *testing {
- taskDb, err = local_db.NewDB("status-testing", path.Join(*workdir, "status-testing.bdb"), nil, nil)
+ taskDb, err = local_db.NewDB("status-testing", path.Join(*workdir, "status-testing.bdb"), nil)
if err != nil {
sklog.Fatalf("Failed to create local task DB: %s", err)
}
@@ -729,21 +728,17 @@
if *firestoreInstance != "" {
label := *host
- modTasks, err := pubsub.NewModifiedTasks(*pubsubTopicTasks, label, ts)
+ mod, err := pubsub.NewModifiedData(*pubsubTopicSet, label, ts)
if err != nil {
sklog.Fatal(err)
}
- modJobs, err := pubsub.NewModifiedJobs(*pubsubTopicJobs, label, ts)
- if err != nil {
- sklog.Fatal(err)
- }
- taskDb, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, ts, modTasks, modJobs)
+ taskDb, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, ts, mod)
if err != nil {
sklog.Fatalf("Failed to create Firestore DB client: %s", err)
}
} else {
label := *host
- taskDb, err = remote_db.NewClient(*taskSchedulerDbUrl, *pubsubTopicTasks, *pubsubTopicJobs, label, ts)
+ taskDb, err = remote_db.NewClient(*taskSchedulerDbUrl, *pubsubTopicSet, label, ts)
if err != nil {
sklog.Fatalf("Failed to create remote task DB: %s", err)
}
diff --git a/status/sys/status-internal.service b/status/sys/status-internal.service
index 91e59fa..571d63a 100644
--- a/status/sys/status-internal.service
+++ b/status/sys/status-internal.service
@@ -12,8 +12,7 @@
--host=status-internal.skia.org \
--resources_dir=/usr/local/share/status \
--capacity_recalculate_interval=30m \
- --pubsub_topic_tasks=task-scheduler-modified-tasks-internal \
- --pubsub_topic_jobs=task-scheduler-modified-jobs-internal \
+ --pubsub_topic_set=internal \
--repo=https://skia.googlesource.com/skia_internal.git \
--repo=https://skia.googlesource.com/internal_test.git \
--swarming_url=https://chrome-swarming.appspot.com \
diff --git a/status/sys/status-staging.service b/status/sys/status-staging.service
index 61077ee..708ac21 100644
--- a/status/sys/status-staging.service
+++ b/status/sys/status-staging.service
@@ -13,8 +13,7 @@
--resources_dir=/usr/local/share/status \
--capacity_recalculate_interval=30m \
--firestore_instance=staging \
- --pubsub_topic_tasks=task-scheduler-modified-tasks-staging \
- --pubsub_topic_jobs=task-scheduler-modified-jobs-staging \
+ --pubsub_topic_set=staging \
--repo=https://skia.googlesource.com/skiabot-test.git \
--swarming_url=https://chromium-swarm-dev.appspot.com \
--task_scheduler_url=https://task-scheduler-staging.skia.org \
diff --git a/status/sys/statusd.service b/status/sys/statusd.service
index 45ec057..8488151 100644
--- a/status/sys/statusd.service
+++ b/status/sys/statusd.service
@@ -12,8 +12,7 @@
--host=status.skia.org \
--resources_dir=/usr/local/share/status \
--capacity_recalculate_interval=30m \
- --pubsub_topic_tasks=task-scheduler-modified-tasks \
- --pubsub_topic_jobs=task-scheduler-modified-jobs \
+ --pubsub_topic_set=prod \
--task_db_url=http://skia-task-scheduler:8008/db/
Restart=always
User=default
diff --git a/task_scheduler/go/db/comments.go b/task_scheduler/go/db/comments.go
index 8123c53..8885f2b 100644
--- a/task_scheduler/go/db/comments.go
+++ b/task_scheduler/go/db/comments.go
@@ -1,20 +1,56 @@
package db
import (
- "fmt"
- "sync"
"time"
- "go.skia.org/infra/go/sklog"
- "go.skia.org/infra/go/util"
"go.skia.org/infra/task_scheduler/go/types"
)
+// ModifiedCommentsReader tracks which comments have been added or deleted and
+// returns results to subscribers based on what has changed since the last call
+// to GetModifiedComments.
+type ModifiedComments interface {
+ // GetModifiedComments returns all comments added or deleted since the
+ // last time GetModifiedComments was run with the given id. The returned
+ // comments are sorted by timestamp. If GetModifiedComments returns an
+ // error, the caller should call StopTrackingModifiedComments and
+ // StartTrackingModifiedComments again, and load all data from scratch
+ // to be sure that no comments were missed.
+ GetModifiedComments(string) ([]*types.TaskComment, []*types.TaskSpecComment, []*types.CommitComment, error)
+
+ // StartTrackingModifiedComments initiates tracking of modified comments
+ // for the current caller. Returns a unique ID which can be used by the
+ // caller to retrieve comments which have been added or deleted since
+ // the last query. The ID expires after a period of inactivity.
+ StartTrackingModifiedComments() (string, error)
+
+ // StopTrackingModifiedComments cancels tracking of modified comments
+ // for the provided ID.
+ StopTrackingModifiedComments(string)
+
+ // TrackModifiedTaskComment indicates the given comment should be
+ // returned from the next call to GetModifiedComments from each
+ // subscriber.
+ TrackModifiedTaskComment(*types.TaskComment)
+
+ // TrackModifiedTaskSpecComment indicates the given comment should be
+ // returned from the next call to GetModifiedComments from each
+ // subscriber.
+ TrackModifiedTaskSpecComment(*types.TaskSpecComment)
+
+ // TrackModifiedCommitComment indicates the given comment should be
+ // returned from the next call to GetModifiedComments from each
+ // subscriber.
+ TrackModifiedCommitComment(*types.CommitComment)
+}
+
// CommentDB stores comments on Tasks, TaskSpecs, and commits.
//
// Clients must be tolerant of comments that refer to nonexistent Tasks,
// TaskSpecs, or commits.
type CommentDB interface {
+ ModifiedComments
+
// GetComments returns all comments for the given repos.
//
// If from is specified, it is a hint that TaskComments and CommitComments
@@ -45,373 +81,3 @@
// Non-ID fields of the argument are ignored.
DeleteCommitComment(*types.CommitComment) error
}
-
-// CommentBox implements CommentDB with in-memory storage.
-//
-// When created via NewCommentBoxWithPersistence, CommentBox will persist the
-// in-memory representation on every change using the provided writer function.
-//
-// CommentBox can be default-initialized if only in-memory storage is desired.
-type CommentBox struct {
- // mtx protects comments.
- mtx sync.RWMutex
- // comments is map[repo_name]*types.RepoComments.
- comments map[string]*types.RepoComments
- // writer is called to persist comments after every change.
- writer func(map[string]*types.RepoComments) error
-}
-
-// NewCommentBoxWithPersistence creates a CommentBox that is initialized with
-// init and sends the updated in-memory representation to writer after each
-// change. The value of init and the argument to writer is
-// map[repo_name]*types.RepoComments. init must not be modified by the caller. writer
-// must not call any methods of CommentBox. writer may return an error to
-// prevent a change from taking effect.
-func NewCommentBoxWithPersistence(init map[string]*types.RepoComments, writer func(map[string]*types.RepoComments) error) *CommentBox {
- return &CommentBox{
- comments: init,
- writer: writer,
- }
-}
-
-// See documentation for CommentDB.GetCommentsForRepos.
-func (b *CommentBox) GetCommentsForRepos(repos []string, from time.Time) ([]*types.RepoComments, error) {
- b.mtx.RLock()
- defer b.mtx.RUnlock()
- rv := make([]*types.RepoComments, len(repos))
- for i, repo := range repos {
- if rc, ok := b.comments[repo]; ok {
- rv[i] = rc.Copy()
- } else {
- rv[i] = &types.RepoComments{Repo: repo}
- }
- }
- return rv, nil
-}
-
-// write calls b.writer with comments if non-null.
-func (b *CommentBox) write() error {
- if b.writer == nil {
- return nil
- }
- return b.writer(b.comments)
-}
-
-// getRepoComments returns the initialized *types.RepoComments for the given repo.
-func (b *CommentBox) getRepoComments(repo string) *types.RepoComments {
- if b.comments == nil {
- b.comments = make(map[string]*types.RepoComments, 1)
- }
- rc, ok := b.comments[repo]
- if !ok {
- rc = &types.RepoComments{
- Repo: repo,
- TaskComments: map[string]map[string][]*types.TaskComment{},
- TaskSpecComments: map[string][]*types.TaskSpecComment{},
- CommitComments: map[string][]*types.CommitComment{},
- }
- b.comments[repo] = rc
- }
- return rc
-}
-
-// putTaskComment validates c and adds c to b.comments, or returns
-// ErrAlreadyExists if a different comment has the same ID fields. Assumes b.mtx
-// is write-locked.
-func (b *CommentBox) putTaskComment(c *types.TaskComment) error {
- if c.Repo == "" || c.Revision == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
- return fmt.Errorf("TaskComment missing required fields. %#v", c)
- }
- rc := b.getRepoComments(c.Repo)
- nameMap, ok := rc.TaskComments[c.Revision]
- if !ok {
- nameMap = map[string][]*types.TaskComment{}
- rc.TaskComments[c.Revision] = nameMap
- }
- cSlice := nameMap[c.Name]
- // TODO(benjaminwagner): Would using utilities in the sort package make this
- // cleaner?
- if len(cSlice) > 0 {
- // Assume comments normally inserted at the end.
- insert := 0
- for i := len(cSlice) - 1; i >= 0; i-- {
- if cSlice[i].Timestamp.Equal(c.Timestamp) {
- if *cSlice[i] == *c {
- return nil
- } else {
- return ErrAlreadyExists
- }
- } else if cSlice[i].Timestamp.Before(c.Timestamp) {
- insert = i + 1
- break
- }
- }
- // Ensure capacity for another comment and move any comments after the
- // insertion point.
- cSlice = append(cSlice, nil)
- copy(cSlice[insert+1:], cSlice[insert:])
- cSlice[insert] = c.Copy()
- } else {
- cSlice = []*types.TaskComment{c.Copy()}
- }
- nameMap[c.Name] = cSlice
- return nil
-}
-
-// deleteTaskComment validates c, then finds and removes a comment matching c's
-// ID fields, returning the comment if found. Assumes b.mtx is write-locked.
-func (b *CommentBox) deleteTaskComment(c *types.TaskComment) (*types.TaskComment, error) {
- if c.Repo == "" || c.Revision == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
- return nil, fmt.Errorf("TaskComment missing required fields. %#v", c)
- }
- if rc, ok := b.comments[c.Repo]; ok {
- if cSlice, ok := rc.TaskComments[c.Revision][c.Name]; ok {
- // Assume linear search is fast.
- for i, existing := range cSlice {
- if existing.Timestamp.Equal(c.Timestamp) {
- if len(cSlice) > 1 {
- rc.TaskComments[c.Revision][c.Name] = append(cSlice[:i], cSlice[i+1:]...)
- } else {
- delete(rc.TaskComments[c.Revision], c.Name)
- if len(rc.TaskComments[c.Revision]) == 0 {
- delete(rc.TaskComments, c.Revision)
- }
- }
- return existing, nil
- }
- }
- }
- }
- return nil, nil
-}
-
-// See documentation for CommentDB.PutTaskComment.
-func (b *CommentBox) PutTaskComment(c *types.TaskComment) error {
- b.mtx.Lock()
- defer b.mtx.Unlock()
- if err := b.putTaskComment(c); err != nil {
- return err
- }
- if err := b.write(); err != nil {
- // If write returns an error, we must revert to previous.
- if _, delErr := b.deleteTaskComment(c); delErr != nil {
- sklog.Warningf("Unexpected error: %s", delErr)
- }
- return err
- }
- return nil
-}
-
-// See documentation for CommentDB.DeleteTaskComment.
-func (b *CommentBox) DeleteTaskComment(c *types.TaskComment) error {
- b.mtx.Lock()
- defer b.mtx.Unlock()
- existing, err := b.deleteTaskComment(c)
- if err != nil {
- return err
- }
- if existing != nil {
- if err := b.write(); err != nil {
- // If write returns an error, we must revert to previous.
- if putErr := b.putTaskComment(existing); putErr != nil {
- sklog.Warningf("Unexpected error: %s", putErr)
- }
- return err
- }
- }
- return nil
-}
-
-// putTaskSpecComment validates c and adds c to b.comments, or returns
-// ErrAlreadyExists if a different comment has the same ID fields. Assumes b.mtx
-// is write-locked.
-func (b *CommentBox) putTaskSpecComment(c *types.TaskSpecComment) error {
- if c.Repo == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
- return fmt.Errorf("TaskSpecComment missing required fields. %#v", c)
- }
- rc := b.getRepoComments(c.Repo)
- cSlice := rc.TaskSpecComments[c.Name]
- if len(cSlice) > 0 {
- // Assume comments normally inserted at the end.
- insert := 0
- for i := len(cSlice) - 1; i >= 0; i-- {
- if cSlice[i].Timestamp.Equal(c.Timestamp) {
- if *cSlice[i] == *c {
- return nil
- } else {
- return ErrAlreadyExists
- }
- } else if cSlice[i].Timestamp.Before(c.Timestamp) {
- insert = i + 1
- break
- }
- }
- // Ensure capacity for another comment and move any comments after the
- // insertion point.
- cSlice = append(cSlice, nil)
- copy(cSlice[insert+1:], cSlice[insert:])
- cSlice[insert] = c.Copy()
- } else {
- cSlice = []*types.TaskSpecComment{c.Copy()}
- }
- rc.TaskSpecComments[c.Name] = cSlice
- return nil
-}
-
-// deleteTaskSpecComment validates c, then finds and removes a comment matching
-// c's ID fields, returning the comment if found. Assumes b.mtx is write-locked.
-func (b *CommentBox) deleteTaskSpecComment(c *types.TaskSpecComment) (*types.TaskSpecComment, error) {
- if c.Repo == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
- return nil, fmt.Errorf("TaskSpecComment missing required fields. %#v", c)
- }
- if rc, ok := b.comments[c.Repo]; ok {
- if cSlice, ok := rc.TaskSpecComments[c.Name]; ok {
- // Assume linear search is fast.
- for i, existing := range cSlice {
- if existing.Timestamp.Equal(c.Timestamp) {
- if len(cSlice) > 1 {
- rc.TaskSpecComments[c.Name] = append(cSlice[:i], cSlice[i+1:]...)
- } else {
- delete(rc.TaskSpecComments, c.Name)
- }
- return existing, nil
- }
- }
- }
- }
- return nil, nil
-}
-
-// See documentation for CommentDB.PutTaskSpecComment.
-func (b *CommentBox) PutTaskSpecComment(c *types.TaskSpecComment) error {
- b.mtx.Lock()
- defer b.mtx.Unlock()
- if err := b.putTaskSpecComment(c); err != nil {
- return err
- }
- if err := b.write(); err != nil {
- // If write returns an error, we must revert to previous.
- if _, delErr := b.deleteTaskSpecComment(c); delErr != nil {
- sklog.Warningf("Unexpected error: %s", delErr)
- }
- return err
- }
- return nil
-}
-
-// See documentation for CommentDB.DeleteTaskSpecComment.
-func (b *CommentBox) DeleteTaskSpecComment(c *types.TaskSpecComment) error {
- b.mtx.Lock()
- defer b.mtx.Unlock()
- existing, err := b.deleteTaskSpecComment(c)
- if err != nil {
- return err
- }
- if existing != nil {
- if err := b.write(); err != nil {
- // If write returns an error, we must revert to previous.
- if putErr := b.putTaskSpecComment(existing); putErr != nil {
- sklog.Warningf("Unexpected error: %s", putErr)
- }
- return err
- }
- }
- return nil
-}
-
-// putCommitComment validates c and adds c to b.comments, or returns
-// ErrAlreadyExists if a different comment has the same ID fields. Assumes b.mtx
-// is write-locked.
-func (b *CommentBox) putCommitComment(c *types.CommitComment) error {
- if c.Repo == "" || c.Revision == "" || util.TimeIsZero(c.Timestamp) {
- return fmt.Errorf("CommitComment missing required fields. %#v", c)
- }
- rc := b.getRepoComments(c.Repo)
- cSlice := rc.CommitComments[c.Revision]
- if len(cSlice) > 0 {
- // Assume comments normally inserted at the end.
- insert := 0
- for i := len(cSlice) - 1; i >= 0; i-- {
- if cSlice[i].Timestamp.Equal(c.Timestamp) {
- if *cSlice[i] == *c {
- return nil
- } else {
- return ErrAlreadyExists
- }
- } else if cSlice[i].Timestamp.Before(c.Timestamp) {
- insert = i + 1
- break
- }
- }
- // Ensure capacity for another comment and move any comments after the
- // insertion point.
- cSlice = append(cSlice, nil)
- copy(cSlice[insert+1:], cSlice[insert:])
- cSlice[insert] = c.Copy()
- } else {
- cSlice = []*types.CommitComment{c.Copy()}
- }
- rc.CommitComments[c.Revision] = cSlice
- return nil
-}
-
-// deleteCommitComment validates c, then finds and removes a comment matching
-// c's ID fields, returning the comment if found. Assumes b.mtx is write-locked.
-func (b *CommentBox) deleteCommitComment(c *types.CommitComment) (*types.CommitComment, error) {
- if c.Repo == "" || c.Revision == "" || util.TimeIsZero(c.Timestamp) {
- return nil, fmt.Errorf("CommitComment missing required fields. %#v", c)
- }
- if rc, ok := b.comments[c.Repo]; ok {
- if cSlice, ok := rc.CommitComments[c.Revision]; ok {
- // Assume linear search is fast.
- for i, existing := range cSlice {
- if existing.Timestamp.Equal(c.Timestamp) {
- if len(cSlice) > 1 {
- rc.CommitComments[c.Revision] = append(cSlice[:i], cSlice[i+1:]...)
- } else {
- delete(rc.CommitComments, c.Revision)
- }
- return existing, nil
- }
- }
- }
- }
- return nil, nil
-}
-
-// See documentation for CommentDB.PutCommitComment.
-func (b *CommentBox) PutCommitComment(c *types.CommitComment) error {
- b.mtx.Lock()
- defer b.mtx.Unlock()
- if err := b.putCommitComment(c); err != nil {
- return err
- }
- if err := b.write(); err != nil {
- // If write returns an error, we must revert to previous.
- if _, delErr := b.deleteCommitComment(c); delErr != nil {
- sklog.Warningf("Unexpected error: %s", delErr)
- }
- return err
- }
- return nil
-}
-
-// See documentation for CommentDB.DeleteCommitComment.
-func (b *CommentBox) DeleteCommitComment(c *types.CommitComment) error {
- b.mtx.Lock()
- defer b.mtx.Unlock()
- existing, err := b.deleteCommitComment(c)
- if err != nil {
- return err
- }
- if existing != nil {
- if err := b.write(); err != nil {
- // If write returns an error, we must revert to previous.
- if putErr := b.putCommitComment(existing); putErr != nil {
- sklog.Warningf("Unexpected error: %s", putErr)
- }
- return err
- }
- }
- return nil
-}
diff --git a/task_scheduler/go/db/db.go b/task_scheduler/go/db/db.go
index ce5e259..dd7215f 100644
--- a/task_scheduler/go/db/db.go
+++ b/task_scheduler/go/db/db.go
@@ -265,6 +265,30 @@
PutJobs([]*types.Job) error
}
+// ModifiedData combines ModifiedTasks, ModifiedJobs, and ModifiedComments.
+type ModifiedData interface {
+ ModifiedTasks
+ ModifiedJobs
+ ModifiedComments
+}
+
+// modifiedData implements ModifiedData.
+type modifiedData struct {
+ ModifiedTasks
+ ModifiedJobs
+ ModifiedComments
+}
+
+// NewModifiedData returns a ModifiedData which combines the given
+// ModifiedTasks, ModifiedJobs, and ModifiedComments.
+func NewModifiedData(t ModifiedTasks, j ModifiedJobs, c ModifiedComments) ModifiedData {
+ return &modifiedData{
+ ModifiedTasks: t,
+ ModifiedJobs: j,
+ ModifiedComments: c,
+ }
+}
+
// UpdateJobsWithRetries wraps a call to db.PutJobs with retries. It calls
// db.PutJobs(f()) repeatedly until one of the following happen:
// - f or db.PutJobs returns an error, which is then returned from
diff --git a/task_scheduler/go/db/firestore/comments.go b/task_scheduler/go/db/firestore/comments.go
index 4fc0234..befd147 100644
--- a/task_scheduler/go/db/firestore/comments.go
+++ b/task_scheduler/go/db/firestore/comments.go
@@ -123,14 +123,34 @@
if st, ok := status.FromError(err); ok && st.Code() == codes.AlreadyExists {
return db.ErrAlreadyExists
}
- return err
+ if err != nil {
+ return err
+ }
+ d.TrackModifiedTaskComment(c)
+ return nil
}
// See documentation for db.CommentDB interface.
func (d *firestoreDB) DeleteTaskComment(c *types.TaskComment) error {
id := taskCommentId(c)
- _, err := firestore.Delete(d.taskComments().Doc(id), DEFAULT_ATTEMPTS, PUT_SINGLE_TIMEOUT)
- return err
+ ref := d.taskComments().Doc(id)
+ exists := true
+ if _, err := firestore.Get(ref, DEFAULT_ATTEMPTS, GET_SINGLE_TIMEOUT); err != nil {
+ if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
+ exists = false
+ } else {
+ return err
+ }
+ }
+ if exists {
+ if _, err := firestore.Delete(ref, DEFAULT_ATTEMPTS, PUT_SINGLE_TIMEOUT); err != nil {
+ return err
+ }
+ deleted := true
+ c.Deleted = &deleted
+ d.TrackModifiedTaskComment(c)
+ }
+ return nil
}
// taskSpecCommentId returns an ID for the TaskSpecComment.
@@ -146,14 +166,34 @@
if st, ok := status.FromError(err); ok && st.Code() == codes.AlreadyExists {
return db.ErrAlreadyExists
}
- return err
+ if err != nil {
+ return err
+ }
+ d.TrackModifiedTaskSpecComment(c)
+ return nil
}
// See documentation for db.CommentDB interface.
func (d *firestoreDB) DeleteTaskSpecComment(c *types.TaskSpecComment) error {
id := taskSpecCommentId(c)
- _, err := firestore.Delete(d.taskSpecComments().Doc(id), DEFAULT_ATTEMPTS, PUT_SINGLE_TIMEOUT)
- return err
+ ref := d.taskSpecComments().Doc(id)
+ exists := true
+ if _, err := firestore.Get(ref, DEFAULT_ATTEMPTS, GET_SINGLE_TIMEOUT); err != nil {
+ if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
+ exists = false
+ } else {
+ return err
+ }
+ }
+ if exists {
+ if _, err := firestore.Delete(ref, DEFAULT_ATTEMPTS, PUT_SINGLE_TIMEOUT); err != nil {
+ return err
+ }
+ deleted := true
+ c.Deleted = &deleted
+ d.TrackModifiedTaskSpecComment(c)
+ }
+ return nil
}
// commitCommentId returns an ID for the CommitComment.
@@ -169,12 +209,32 @@
if st, ok := status.FromError(err); ok && st.Code() == codes.AlreadyExists {
return db.ErrAlreadyExists
}
- return err
+ if err != nil {
+ return err
+ }
+ d.TrackModifiedCommitComment(c)
+ return nil
}
// See documentation for db.CommentDB interface.
func (d *firestoreDB) DeleteCommitComment(c *types.CommitComment) error {
id := commitCommentId(c)
- _, err := firestore.Delete(d.commitComments().Doc(id), DEFAULT_ATTEMPTS, PUT_SINGLE_TIMEOUT)
- return err
+ ref := d.commitComments().Doc(id)
+ exists := true
+ if _, err := firestore.Get(ref, DEFAULT_ATTEMPTS, GET_SINGLE_TIMEOUT); err != nil {
+ if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
+ exists = false
+ } else {
+ return err
+ }
+ }
+ if exists {
+ if _, err := firestore.Delete(d.commitComments().Doc(id), DEFAULT_ATTEMPTS, PUT_SINGLE_TIMEOUT); err != nil {
+ return err
+ }
+ deleted := true
+ c.Deleted = &deleted
+ d.TrackModifiedCommitComment(c)
+ }
+ return nil
}
diff --git a/task_scheduler/go/db/firestore/firestore.go b/task_scheduler/go/db/firestore/firestore.go
index ca9f388..5607ca8 100644
--- a/task_scheduler/go/db/firestore/firestore.go
+++ b/task_scheduler/go/db/firestore/firestore.go
@@ -56,31 +56,24 @@
client *firestore.Client
parentDoc string
- // ModifiedTasksImpl and ModifiedJobsImpl are embedded in order to
- // implement db.ModifiedTasksReader and db.ModifiedJobsReader.
- db.ModifiedTasks
- db.ModifiedJobs
+ db.ModifiedData
}
// NewDB returns a db.DB which uses Cloud Firestore for storage. The parentDoc
// parameter is optional and indicates the path of a parent document to which
// all collections within the DB will belong. If it is not supplied, then the
// collections will be at the top level.
-func NewDB(ctx context.Context, project, instance string, ts oauth2.TokenSource, modTasks db.ModifiedTasks, modJobs db.ModifiedJobs) (db.BackupDBCloser, error) {
+func NewDB(ctx context.Context, project, instance string, ts oauth2.TokenSource, mod db.ModifiedData) (db.BackupDBCloser, error) {
client, err := firestore.NewClient(ctx, project, firestore.APP_TASK_SCHEDULER, instance, ts)
if err != nil {
return nil, err
}
- if modTasks == nil {
- modTasks = &modified.ModifiedTasksImpl{}
- }
- if modJobs == nil {
- modJobs = &modified.ModifiedJobsImpl{}
+ if mod == nil {
+ mod = modified.NewModifiedData()
}
return &firestoreDB{
- client: client,
- ModifiedTasks: modTasks,
- ModifiedJobs: modJobs,
+ client: client,
+ ModifiedData: mod,
}, nil
}
diff --git a/task_scheduler/go/db/firestore/firestore_test.go b/task_scheduler/go/db/firestore/firestore_test.go
index be7579e..4002133 100644
--- a/task_scheduler/go/db/firestore/firestore_test.go
+++ b/task_scheduler/go/db/firestore/firestore_test.go
@@ -24,7 +24,7 @@
testutils.MediumTest(t)
testutils.ManualTest(t)
instance := fmt.Sprintf("test-%s", uuid.New())
- d, err := NewDB(context.Background(), "skia-firestore", instance, nil, nil, nil)
+ d, err := NewDB(context.Background(), "skia-firestore", instance, nil, nil)
assert.NoError(t, err)
cleanup := func() {
c := d.(*firestoreDB).client
diff --git a/task_scheduler/go/db/local_db/busywork/main.go b/task_scheduler/go/db/local_db/busywork/main.go
index 01844df..41b083c 100644
--- a/task_scheduler/go/db/local_db/busywork/main.go
+++ b/task_scheduler/go/db/local_db/busywork/main.go
@@ -520,7 +520,7 @@
sklog.Fatal(err)
}
id := fmt.Sprintf("busywork_%s", hostname)
- d, err := firestore.NewDB(context.Background(), "skia-firestore", id, ts, nil, nil)
+ d, err := firestore.NewDB(context.Background(), "skia-firestore", id, ts, nil)
if err != nil {
sklog.Fatal(err)
}
diff --git a/task_scheduler/go/db/local_db/local_db.go b/task_scheduler/go/db/local_db/local_db.go
index 3109a63..689ecfe 100644
--- a/task_scheduler/go/db/local_db/local_db.go
+++ b/task_scheduler/go/db/local_db/local_db.go
@@ -18,6 +18,7 @@
"go.skia.org/infra/go/sklog"
"go.skia.org/infra/go/util"
"go.skia.org/infra/task_scheduler/go/db"
+ "go.skia.org/infra/task_scheduler/go/db/memory"
"go.skia.org/infra/task_scheduler/go/db/modified"
"go.skia.org/infra/task_scheduler/go/types"
)
@@ -230,12 +231,11 @@
// ModifiedTasksImpl and ModifiedJobsImpl are embedded in order to
// implement db.ModifiedTasksReader and db.ModifiedJobsReader.
- db.ModifiedTasks
- db.ModifiedJobs
+ db.ModifiedData
// CommentBox is embedded in order to implement db.CommentDB. CommentBox uses
// this localDB to persist the comments.
- *db.CommentBox
+ *memory.CommentBox
// Close will send on each of these channels to indicate goroutines should
// stop.
@@ -316,16 +316,13 @@
}
// NewDB returns a local DB instance.
-func NewDB(name, filename string, modTasks db.ModifiedTasks, modJobs db.ModifiedJobs) (db.BackupDBCloser, error) {
+func NewDB(name, filename string, mod db.ModifiedData) (db.BackupDBCloser, error) {
boltdb, err := bolt.Open(filename, 0600, nil)
if err != nil {
return nil, err
}
- if modTasks == nil {
- modTasks = &modified.ModifiedTasksImpl{}
- }
- if modJobs == nil {
- modJobs = &modified.ModifiedJobsImpl{}
+ if mod == nil {
+ mod = modified.NewModifiedData()
}
d := &localDB{
name: name,
@@ -384,8 +381,7 @@
"bucket": BUCKET_JOBS,
"count": "rows",
}),
- ModifiedTasks: modTasks,
- ModifiedJobs: modJobs,
+ ModifiedData: mod,
}
stopReportActiveTx := make(chan bool)
@@ -432,7 +428,7 @@
return nil, err
}
- d.CommentBox = db.NewCommentBoxWithPersistence(comments, d.writeCommentsMap)
+ d.CommentBox = memory.NewCommentBoxWithPersistence(mod, comments, d.writeCommentsMap)
if dbMetric, err := boltutil.NewDbMetric(boltdb, []string{BUCKET_TASKS, BUCKET_JOBS, BUCKET_COMMENTS}, map[string]string{"database": name}); err != nil {
return nil, err
@@ -533,7 +529,7 @@
d.metricReadTaskQueries.Inc(1)
min := []byte(start.Add(-MAX_CREATED_TIME_SKEW).UTC().Format(TIMESTAMP_FORMAT))
max := []byte(end.Add(MAX_CREATED_TIME_SKEW).UTC().Format(TIMESTAMP_FORMAT))
- decoder := types.TaskDecoder{}
+ decoder := types.NewTaskDecoder()
if err := d.view("GetTasksFromDateRange", func(tx *bolt.Tx) error {
c := tasksBucket(tx).Cursor()
for k, v := c.Seek(min); k != nil && bytes.Compare(k, max) <= 0; k, v = c.Next() {
@@ -740,7 +736,7 @@
d.metricReadJobQueries.Inc(1)
min := []byte(start.UTC().Format(TIMESTAMP_FORMAT))
max := []byte(end.UTC().Format(TIMESTAMP_FORMAT))
- decoder := types.JobDecoder{}
+ decoder := types.NewJobDecoder()
if err := d.view("GetJobsFromDateRange", func(tx *bolt.Tx) error {
c := jobsBucket(tx).Cursor()
for k, v := c.Seek(min); k != nil && bytes.Compare(k, max) <= 0; k, v = c.Next() {
diff --git a/task_scheduler/go/db/local_db/local_db_test.go b/task_scheduler/go/db/local_db/local_db_test.go
index 22b9649..ee2f1e9 100644
--- a/task_scheduler/go/db/local_db/local_db_test.go
+++ b/task_scheduler/go/db/local_db/local_db_test.go
@@ -150,7 +150,7 @@
func makeDB(t *testing.T, name string) (db.BackupDBCloser, string) {
tmpdir, err := ioutil.TempDir("", name)
assert.NoError(t, err)
- d, err := NewDB(name, filepath.Join(tmpdir, "task.db"), nil, nil)
+ d, err := NewDB(name, filepath.Join(tmpdir, "task.db"), nil)
assert.NoError(t, err)
return d, tmpdir
}
diff --git a/task_scheduler/go/db/local_db/ts_local_db_viewer/main.go b/task_scheduler/go/db/local_db/ts_local_db_viewer/main.go
index 87c44aa..00f7412 100644
--- a/task_scheduler/go/db/local_db/ts_local_db_viewer/main.go
+++ b/task_scheduler/go/db/local_db/ts_local_db_viewer/main.go
@@ -28,7 +28,7 @@
// Global init.
common.Init()
- d, err := local_db.NewDB(local_db.DB_NAME, *dbfile, nil, nil)
+ d, err := local_db.NewDB(local_db.DB_NAME, *dbfile, nil)
if err != nil {
sklog.Fatal(err)
}
diff --git a/task_scheduler/go/db/memory/comments.go b/task_scheduler/go/db/memory/comments.go
new file mode 100644
index 0000000..122f7bc
--- /dev/null
+++ b/task_scheduler/go/db/memory/comments.go
@@ -0,0 +1,399 @@
+package memory
+
+import (
+ "fmt"
+ "sync"
+ "time"
+
+ "go.skia.org/infra/go/sklog"
+ "go.skia.org/infra/go/util"
+ "go.skia.org/infra/task_scheduler/go/db"
+ "go.skia.org/infra/task_scheduler/go/db/modified"
+ "go.skia.org/infra/task_scheduler/go/types"
+)
+
+// CommentBox implements CommentDB with in-memory storage.
+//
+// When created via NewCommentBoxWithPersistence, CommentBox will persist the
+// in-memory representation on every change using the provided writer function.
+type CommentBox struct {
+ db.ModifiedComments
+
+ // mtx protects comments.
+ mtx sync.RWMutex
+ // comments is map[repo_name]*types.RepoComments.
+ comments map[string]*types.RepoComments
+ // writer is called to persist comments after every change.
+ writer func(map[string]*types.RepoComments) error
+}
+
+// NewCommentBoxWithPersistence creates a CommentBox that is initialized with
+// init and sends the updated in-memory representation to writer after each
+// change. The value of init and the argument to writer is
+// map[repo_name]*types.RepoComments. init must not be modified by the caller. writer
+// must not call any methods of CommentBox. writer may return an error to
+// prevent a change from taking effect.
+func NewCommentBoxWithPersistence(modComments db.ModifiedComments, init map[string]*types.RepoComments, writer func(map[string]*types.RepoComments) error) *CommentBox {
+ if modComments == nil {
+ modComments = &modified.ModifiedCommentsImpl{}
+ }
+ return &CommentBox{
+ ModifiedComments: modComments,
+ comments: init,
+ writer: writer,
+ }
+}
+
+// See documentation for CommentDB.GetCommentsForRepos.
+func (b *CommentBox) GetCommentsForRepos(repos []string, from time.Time) ([]*types.RepoComments, error) {
+ b.mtx.RLock()
+ defer b.mtx.RUnlock()
+ rv := make([]*types.RepoComments, len(repos))
+ for i, repo := range repos {
+ if rc, ok := b.comments[repo]; ok {
+ rv[i] = rc.Copy()
+ } else {
+ rv[i] = &types.RepoComments{Repo: repo}
+ }
+ }
+ return rv, nil
+}
+
+// write calls b.writer with comments if non-null.
+func (b *CommentBox) write() error {
+ if b.writer == nil {
+ return nil
+ }
+ return b.writer(b.comments)
+}
+
+// getRepoComments returns the initialized *types.RepoComments for the given repo.
+func (b *CommentBox) getRepoComments(repo string) *types.RepoComments {
+ if b.comments == nil {
+ b.comments = make(map[string]*types.RepoComments, 1)
+ }
+ rc, ok := b.comments[repo]
+ if !ok {
+ rc = &types.RepoComments{
+ Repo: repo,
+ TaskComments: map[string]map[string][]*types.TaskComment{},
+ TaskSpecComments: map[string][]*types.TaskSpecComment{},
+ CommitComments: map[string][]*types.CommitComment{},
+ }
+ b.comments[repo] = rc
+ }
+ return rc
+}
+
+// putTaskComment validates c and adds c to b.comments, or returns
+// db.ErrAlreadyExists if a different comment has the same ID fields. Assumes b.mtx
+// is write-locked.
+func (b *CommentBox) putTaskComment(c *types.TaskComment) error {
+ if c.Repo == "" || c.Revision == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
+ return fmt.Errorf("TaskComment missing required fields. %#v", c)
+ }
+ rc := b.getRepoComments(c.Repo)
+ nameMap, ok := rc.TaskComments[c.Revision]
+ if !ok {
+ nameMap = map[string][]*types.TaskComment{}
+ rc.TaskComments[c.Revision] = nameMap
+ }
+ cSlice := nameMap[c.Name]
+ // TODO(benjaminwagner): Would using utilities in the sort package make this
+ // cleaner?
+ if len(cSlice) > 0 {
+ // Assume comments normally inserted at the end.
+ insert := 0
+ for i := len(cSlice) - 1; i >= 0; i-- {
+ if cSlice[i].Timestamp.Equal(c.Timestamp) {
+ if *cSlice[i] == *c {
+ return nil
+ } else {
+ return db.ErrAlreadyExists
+ }
+ } else if cSlice[i].Timestamp.Before(c.Timestamp) {
+ insert = i + 1
+ break
+ }
+ }
+ // Ensure capacity for another comment and move any comments after the
+ // insertion point.
+ cSlice = append(cSlice, nil)
+ copy(cSlice[insert+1:], cSlice[insert:])
+ cSlice[insert] = c.Copy()
+ } else {
+ cSlice = []*types.TaskComment{c.Copy()}
+ }
+ nameMap[c.Name] = cSlice
+ return nil
+}
+
+// deleteTaskComment validates c, then finds and removes a comment matching c's
+// ID fields, returning the comment if found. Assumes b.mtx is write-locked.
+func (b *CommentBox) deleteTaskComment(c *types.TaskComment) (*types.TaskComment, error) {
+ if c.Repo == "" || c.Revision == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
+ return nil, fmt.Errorf("TaskComment missing required fields. %#v", c)
+ }
+ if rc, ok := b.comments[c.Repo]; ok {
+ if cSlice, ok := rc.TaskComments[c.Revision][c.Name]; ok {
+ // Assume linear search is fast.
+ for i, existing := range cSlice {
+ if existing.Timestamp.Equal(c.Timestamp) {
+ if len(cSlice) > 1 {
+ rc.TaskComments[c.Revision][c.Name] = append(cSlice[:i], cSlice[i+1:]...)
+ } else {
+ delete(rc.TaskComments[c.Revision], c.Name)
+ if len(rc.TaskComments[c.Revision]) == 0 {
+ delete(rc.TaskComments, c.Revision)
+ }
+ }
+ return existing, nil
+ }
+ }
+ }
+ }
+ return nil, nil
+}
+
+// See documentation for CommentDB.PutTaskComment.
+func (b *CommentBox) PutTaskComment(c *types.TaskComment) error {
+ b.mtx.Lock()
+ defer b.mtx.Unlock()
+ if err := b.putTaskComment(c); err != nil {
+ return err
+ }
+ if err := b.write(); err != nil {
+ // If write returns an error, we must revert to previous.
+ if _, delErr := b.deleteTaskComment(c); delErr != nil {
+ sklog.Warningf("Unexpected error: %s", delErr)
+ }
+ return err
+ }
+ b.TrackModifiedTaskComment(c)
+ return nil
+}
+
+// See documentation for CommentDB.DeleteTaskComment.
+func (b *CommentBox) DeleteTaskComment(c *types.TaskComment) error {
+ b.mtx.Lock()
+ defer b.mtx.Unlock()
+ existing, err := b.deleteTaskComment(c)
+ if err != nil {
+ return err
+ }
+ if existing != nil {
+ if err := b.write(); err != nil {
+ // If write returns an error, we must revert to previous.
+ if putErr := b.putTaskComment(existing); putErr != nil {
+ sklog.Warningf("Unexpected error: %s", putErr)
+ }
+ return err
+ }
+ deleted := true
+ c.Deleted = &deleted
+ b.TrackModifiedTaskComment(c)
+ }
+ return nil
+}
+
+// putTaskSpecComment validates c and adds c to b.comments, or returns
+// db.ErrAlreadyExists if a different comment has the same ID fields. Assumes b.mtx
+// is write-locked.
+func (b *CommentBox) putTaskSpecComment(c *types.TaskSpecComment) error {
+ if c.Repo == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
+ return fmt.Errorf("TaskSpecComment missing required fields. %#v", c)
+ }
+ rc := b.getRepoComments(c.Repo)
+ cSlice := rc.TaskSpecComments[c.Name]
+ if len(cSlice) > 0 {
+ // Assume comments normally inserted at the end.
+ insert := 0
+ for i := len(cSlice) - 1; i >= 0; i-- {
+ if cSlice[i].Timestamp.Equal(c.Timestamp) {
+ if *cSlice[i] == *c {
+ return nil
+ } else {
+ return db.ErrAlreadyExists
+ }
+ } else if cSlice[i].Timestamp.Before(c.Timestamp) {
+ insert = i + 1
+ break
+ }
+ }
+ // Ensure capacity for another comment and move any comments after the
+ // insertion point.
+ cSlice = append(cSlice, nil)
+ copy(cSlice[insert+1:], cSlice[insert:])
+ cSlice[insert] = c.Copy()
+ } else {
+ cSlice = []*types.TaskSpecComment{c.Copy()}
+ }
+ rc.TaskSpecComments[c.Name] = cSlice
+ return nil
+}
+
+// deleteTaskSpecComment validates c, then finds and removes a comment matching
+// c's ID fields, returning the comment if found. Assumes b.mtx is write-locked.
+func (b *CommentBox) deleteTaskSpecComment(c *types.TaskSpecComment) (*types.TaskSpecComment, error) {
+ if c.Repo == "" || c.Name == "" || util.TimeIsZero(c.Timestamp) {
+ return nil, fmt.Errorf("TaskSpecComment missing required fields. %#v", c)
+ }
+ if rc, ok := b.comments[c.Repo]; ok {
+ if cSlice, ok := rc.TaskSpecComments[c.Name]; ok {
+ // Assume linear search is fast.
+ for i, existing := range cSlice {
+ if existing.Timestamp.Equal(c.Timestamp) {
+ if len(cSlice) > 1 {
+ rc.TaskSpecComments[c.Name] = append(cSlice[:i], cSlice[i+1:]...)
+ } else {
+ delete(rc.TaskSpecComments, c.Name)
+ }
+ return existing, nil
+ }
+ }
+ }
+ }
+ return nil, nil
+}
+
+// See documentation for CommentDB.PutTaskSpecComment.
+func (b *CommentBox) PutTaskSpecComment(c *types.TaskSpecComment) error {
+ b.mtx.Lock()
+ defer b.mtx.Unlock()
+ if err := b.putTaskSpecComment(c); err != nil {
+ return err
+ }
+ if err := b.write(); err != nil {
+ // If write returns an error, we must revert to previous.
+ if _, delErr := b.deleteTaskSpecComment(c); delErr != nil {
+ sklog.Warningf("Unexpected error: %s", delErr)
+ }
+ return err
+ }
+ b.TrackModifiedTaskSpecComment(c)
+ return nil
+}
+
+// See documentation for CommentDB.DeleteTaskSpecComment.
+func (b *CommentBox) DeleteTaskSpecComment(c *types.TaskSpecComment) error {
+ b.mtx.Lock()
+ defer b.mtx.Unlock()
+ existing, err := b.deleteTaskSpecComment(c)
+ if err != nil {
+ return err
+ }
+ if existing != nil {
+ if err := b.write(); err != nil {
+ // If write returns an error, we must revert to previous.
+ if putErr := b.putTaskSpecComment(existing); putErr != nil {
+ sklog.Warningf("Unexpected error: %s", putErr)
+ }
+ return err
+ }
+ deleted := true
+ c.Deleted = &deleted
+ b.TrackModifiedTaskSpecComment(c)
+ }
+ return nil
+}
+
+// putCommitComment validates c and adds c to b.comments, or returns
+// db.ErrAlreadyExists if a different comment has the same ID fields. Assumes b.mtx
+// is write-locked.
+func (b *CommentBox) putCommitComment(c *types.CommitComment) error {
+ if c.Repo == "" || c.Revision == "" || util.TimeIsZero(c.Timestamp) {
+ return fmt.Errorf("CommitComment missing required fields. %#v", c)
+ }
+ rc := b.getRepoComments(c.Repo)
+ cSlice := rc.CommitComments[c.Revision]
+ if len(cSlice) > 0 {
+ // Assume comments normally inserted at the end.
+ insert := 0
+ for i := len(cSlice) - 1; i >= 0; i-- {
+ if cSlice[i].Timestamp.Equal(c.Timestamp) {
+ if *cSlice[i] == *c {
+ return nil
+ } else {
+ return db.ErrAlreadyExists
+ }
+ } else if cSlice[i].Timestamp.Before(c.Timestamp) {
+ insert = i + 1
+ break
+ }
+ }
+ // Ensure capacity for another comment and move any comments after the
+ // insertion point.
+ cSlice = append(cSlice, nil)
+ copy(cSlice[insert+1:], cSlice[insert:])
+ cSlice[insert] = c.Copy()
+ } else {
+ cSlice = []*types.CommitComment{c.Copy()}
+ }
+ rc.CommitComments[c.Revision] = cSlice
+ return nil
+}
+
+// deleteCommitComment validates c, then finds and removes a comment matching
+// c's ID fields, returning the comment if found. Assumes b.mtx is write-locked.
+func (b *CommentBox) deleteCommitComment(c *types.CommitComment) (*types.CommitComment, error) {
+ if c.Repo == "" || c.Revision == "" || util.TimeIsZero(c.Timestamp) {
+ return nil, fmt.Errorf("CommitComment missing required fields. %#v", c)
+ }
+ if rc, ok := b.comments[c.Repo]; ok {
+ if cSlice, ok := rc.CommitComments[c.Revision]; ok {
+ // Assume linear search is fast.
+ for i, existing := range cSlice {
+ if existing.Timestamp.Equal(c.Timestamp) {
+ if len(cSlice) > 1 {
+ rc.CommitComments[c.Revision] = append(cSlice[:i], cSlice[i+1:]...)
+ } else {
+ delete(rc.CommitComments, c.Revision)
+ }
+ return existing, nil
+ }
+ }
+ }
+ }
+ return nil, nil
+}
+
+// See documentation for CommentDB.PutCommitComment.
+func (b *CommentBox) PutCommitComment(c *types.CommitComment) error {
+ b.mtx.Lock()
+ defer b.mtx.Unlock()
+ if err := b.putCommitComment(c); err != nil {
+ return err
+ }
+ if err := b.write(); err != nil {
+ // If write returns an error, we must revert to previous.
+ if _, delErr := b.deleteCommitComment(c); delErr != nil {
+ sklog.Warningf("Unexpected error: %s", delErr)
+ }
+ return err
+ }
+ b.TrackModifiedCommitComment(c)
+ return nil
+}
+
+// See documentation for CommentDB.DeleteCommitComment.
+func (b *CommentBox) DeleteCommitComment(c *types.CommitComment) error {
+ b.mtx.Lock()
+ defer b.mtx.Unlock()
+ existing, err := b.deleteCommitComment(c)
+ if err != nil {
+ return err
+ }
+ if existing != nil {
+ if err := b.write(); err != nil {
+ // If write returns an error, we must revert to previous.
+ if putErr := b.putCommitComment(existing); putErr != nil {
+ sklog.Warningf("Unexpected error: %s", putErr)
+ }
+ return err
+ }
+ deleted := true
+ c.Deleted = &deleted
+ b.TrackModifiedCommitComment(c)
+ }
+ return nil
+}
diff --git a/task_scheduler/go/db/comments_test.go b/task_scheduler/go/db/memory/comments_test.go
similarity index 63%
rename from task_scheduler/go/db/comments_test.go
rename to task_scheduler/go/db/memory/comments_test.go
index a00e224..54f1914 100644
--- a/task_scheduler/go/db/comments_test.go
+++ b/task_scheduler/go/db/memory/comments_test.go
@@ -1,26 +1,22 @@
-package db
+package memory
import (
"fmt"
- "os"
"testing"
"time"
assert "github.com/stretchr/testify/require"
"go.skia.org/infra/go/deepequal"
"go.skia.org/infra/go/testutils"
+ "go.skia.org/infra/task_scheduler/go/db"
+ "go.skia.org/infra/task_scheduler/go/db/modified"
"go.skia.org/infra/task_scheduler/go/types"
)
-func TestMain(m *testing.M) {
- AssertDeepEqual = deepequal.AssertDeepEqual
- os.Exit(m.Run())
-}
-
// TestCommentBox checks that CommentBox correctly implements CommentDB.
func TestCommentBox(t *testing.T) {
testutils.SmallTest(t)
- TestCommentDB(t, &CommentBox{})
+ db.TestCommentDB(t, &CommentBox{ModifiedComments: &modified.ModifiedCommentsImpl{}})
}
// TestCommentBoxWithPersistence checks that NewCommentBoxWithPersistence can be
@@ -36,7 +32,7 @@
return nil
}
- db := NewCommentBoxWithPersistence(nil, testWriter)
+ d := NewCommentBoxWithPersistence(nil, nil, testWriter)
now := time.Now()
@@ -51,23 +47,23 @@
TaskSpecComments: map[string][]*types.TaskSpecComment{},
CommitComments: map[string][]*types.CommitComment{},
}
- assert.NoError(t, db.PutTaskComment(tc1))
+ assert.NoError(t, d.PutTaskComment(tc1))
tc2 := types.MakeTaskComment(2, 1, 1, 1, now.Add(2*time.Second))
expected[r1].TaskComments["c1"]["n1"] = []*types.TaskComment{tc1, tc2}
- assert.NoError(t, db.PutTaskComment(tc2))
+ assert.NoError(t, d.PutTaskComment(tc2))
tc3 := types.MakeTaskComment(3, 1, 1, 1, now.Add(time.Second))
expected[r1].TaskComments["c1"]["n1"] = []*types.TaskComment{tc1, tc3, tc2}
- assert.NoError(t, db.PutTaskComment(tc3))
+ assert.NoError(t, d.PutTaskComment(tc3))
tc4 := types.MakeTaskComment(4, 1, 1, 2, now)
expected[r1].TaskComments["c2"] = map[string][]*types.TaskComment{"n1": {tc4}}
- assert.NoError(t, db.PutTaskComment(tc4))
+ assert.NoError(t, d.PutTaskComment(tc4))
tc5 := types.MakeTaskComment(5, 1, 2, 2, now)
expected[r1].TaskComments["c2"]["n2"] = []*types.TaskComment{tc5}
- assert.NoError(t, db.PutTaskComment(tc5))
+ assert.NoError(t, d.PutTaskComment(tc5))
tc6 := types.MakeTaskComment(6, 2, 3, 3, now)
r2 := tc6.Repo
@@ -77,21 +73,21 @@
TaskSpecComments: map[string][]*types.TaskSpecComment{},
CommitComments: map[string][]*types.CommitComment{},
}
- assert.NoError(t, db.PutTaskComment(tc6))
+ assert.NoError(t, d.PutTaskComment(tc6))
tc6copy := tc6.Copy() // Adding identical comment should be ignored.
- assert.NoError(t, db.PutTaskComment(tc6copy))
+ assert.NoError(t, d.PutTaskComment(tc6copy))
tc6.Message = "modifying after Put shouldn't affect stored comment"
assert.True(t, callCount >= 6)
sc1 := types.MakeTaskSpecComment(1, 1, 1, now)
expected[r1].TaskSpecComments["n1"] = []*types.TaskSpecComment{sc1}
- assert.NoError(t, db.PutTaskSpecComment(sc1))
+ assert.NoError(t, d.PutTaskSpecComment(sc1))
cc1 := types.MakeCommitComment(1, 1, 1, now)
expected[r1].CommitComments["c1"] = []*types.CommitComment{cc1}
- assert.NoError(t, db.PutCommitComment(cc1))
+ assert.NoError(t, d.PutCommitComment(cc1))
assert.True(t, callCount >= 8)
callCount = 0
@@ -99,13 +95,13 @@
// Check that if there's an error adding, testWriter is not called.
tc1different := tc1.Copy()
tc1different.Message = "not the same"
- assert.True(t, IsAlreadyExists(db.PutTaskComment(tc1different)))
+ assert.True(t, db.IsAlreadyExists(d.PutTaskComment(tc1different)))
sc1different := sc1.Copy()
sc1different.Message = "not the same"
- assert.True(t, IsAlreadyExists(db.PutTaskSpecComment(sc1different)))
+ assert.True(t, db.IsAlreadyExists(d.PutTaskSpecComment(sc1different)))
cc1different := cc1.Copy()
cc1different.Message = "not the same"
- assert.True(t, IsAlreadyExists(db.PutCommitComment(cc1different)))
+ assert.True(t, db.IsAlreadyExists(d.PutCommitComment(cc1different)))
assert.Equal(t, 0, callCount)
@@ -114,10 +110,10 @@
r1: expected[r1].Copy(),
r2: expected[r2].Copy(),
}
- db = NewCommentBoxWithPersistence(init, testWriter)
+ d = NewCommentBoxWithPersistence(nil, init, testWriter)
{
- actual, err := db.GetCommentsForRepos([]string{"r0", r1, r2}, now.Add(-10000*time.Hour))
+ actual, err := d.GetCommentsForRepos([]string{"r0", r1, r2}, now.Add(-10000*time.Hour))
assert.NoError(t, err)
expectedSlice := []*types.RepoComments{
{Repo: "r0"},
@@ -131,28 +127,28 @@
// Delete some comments.
expected[r1].TaskComments["c1"]["n1"] = []*types.TaskComment{tc1, tc2}
- assert.NoError(t, db.DeleteTaskComment(tc3))
+ assert.NoError(t, d.DeleteTaskComment(tc3))
expected[r1].TaskSpecComments = map[string][]*types.TaskSpecComment{}
- assert.NoError(t, db.DeleteTaskSpecComment(sc1))
+ assert.NoError(t, d.DeleteTaskSpecComment(sc1))
expected[r1].CommitComments = map[string][]*types.CommitComment{}
- assert.NoError(t, db.DeleteCommitComment(cc1))
+ assert.NoError(t, d.DeleteCommitComment(cc1))
assert.Equal(t, 3, callCount)
// Delete of nonexistent task should succeed.
- assert.NoError(t, db.DeleteTaskComment(types.MakeTaskComment(99, 1, 1, 1, now.Add(99*time.Second))))
- assert.NoError(t, db.DeleteTaskComment(types.MakeTaskComment(99, 1, 1, 99, now)))
- assert.NoError(t, db.DeleteTaskComment(types.MakeTaskComment(99, 1, 99, 1, now)))
- assert.NoError(t, db.DeleteTaskComment(types.MakeTaskComment(99, 99, 1, 1, now)))
- assert.NoError(t, db.DeleteTaskSpecComment(types.MakeTaskSpecComment(99, 1, 1, now.Add(99*time.Second))))
- assert.NoError(t, db.DeleteTaskSpecComment(types.MakeTaskSpecComment(99, 1, 99, now)))
- assert.NoError(t, db.DeleteTaskSpecComment(types.MakeTaskSpecComment(99, 99, 1, now)))
- assert.NoError(t, db.DeleteCommitComment(types.MakeCommitComment(99, 1, 1, now.Add(99*time.Second))))
- assert.NoError(t, db.DeleteCommitComment(types.MakeCommitComment(99, 1, 99, now)))
- assert.NoError(t, db.DeleteCommitComment(types.MakeCommitComment(99, 99, 1, now)))
+ assert.NoError(t, d.DeleteTaskComment(types.MakeTaskComment(99, 1, 1, 1, now.Add(99*time.Second))))
+ assert.NoError(t, d.DeleteTaskComment(types.MakeTaskComment(99, 1, 1, 99, now)))
+ assert.NoError(t, d.DeleteTaskComment(types.MakeTaskComment(99, 1, 99, 1, now)))
+ assert.NoError(t, d.DeleteTaskComment(types.MakeTaskComment(99, 99, 1, 1, now)))
+ assert.NoError(t, d.DeleteTaskSpecComment(types.MakeTaskSpecComment(99, 1, 1, now.Add(99*time.Second))))
+ assert.NoError(t, d.DeleteTaskSpecComment(types.MakeTaskSpecComment(99, 1, 99, now)))
+ assert.NoError(t, d.DeleteTaskSpecComment(types.MakeTaskSpecComment(99, 99, 1, now)))
+ assert.NoError(t, d.DeleteCommitComment(types.MakeCommitComment(99, 1, 1, now.Add(99*time.Second))))
+ assert.NoError(t, d.DeleteCommitComment(types.MakeCommitComment(99, 1, 99, now)))
+ assert.NoError(t, d.DeleteCommitComment(types.MakeCommitComment(99, 99, 1, now)))
{
- actual, err := db.GetCommentsForRepos([]string{"r0", r1, r2}, now.Add(-10000*time.Hour))
+ actual, err := d.GetCommentsForRepos([]string{"r0", r1, r2}, now.Add(-10000*time.Hour))
assert.NoError(t, err)
expectedSlice := []*types.RepoComments{
{Repo: "r0"},
@@ -167,10 +163,10 @@
r1: expected[r1].Copy(),
r2: expected[r2].Copy(),
}
- db = NewCommentBoxWithPersistence(init, testWriter)
+ d = NewCommentBoxWithPersistence(nil, init, testWriter)
{
- actual, err := db.GetCommentsForRepos([]string{"r0", r1, r2}, now.Add(-10000*time.Hour))
+ actual, err := d.GetCommentsForRepos([]string{"r0", r1, r2}, now.Add(-10000*time.Hour))
assert.NoError(t, err)
expectedSlice := []*types.RepoComments{
{Repo: "r0"},
@@ -193,7 +189,7 @@
return injectedError
}
- db := NewCommentBoxWithPersistence(nil, testWriter)
+ d := NewCommentBoxWithPersistence(nil, nil, testWriter)
now := time.Now()
@@ -205,16 +201,16 @@
tc5 := types.MakeTaskComment(5, 1, 2, 2, now)
tc6 := types.MakeTaskComment(6, 2, 3, 3, now)
for _, c := range []*types.TaskComment{tc1, tc2, tc3, tc4, tc5, tc6} {
- assert.NoError(t, db.PutTaskComment(c))
+ assert.NoError(t, d.PutTaskComment(c))
}
r1 := tc1.Repo
r2 := tc6.Repo
sc1 := types.MakeTaskSpecComment(1, 1, 1, now)
- assert.NoError(t, db.PutTaskSpecComment(sc1))
+ assert.NoError(t, d.PutTaskSpecComment(sc1))
cc1 := types.MakeCommitComment(1, 1, 1, now)
- assert.NoError(t, db.PutCommitComment(cc1))
+ assert.NoError(t, d.PutCommitComment(cc1))
expected := []*types.RepoComments{
{
@@ -248,7 +244,7 @@
}
{
- actual, err := db.GetCommentsForRepos([]string{r1, r2}, now.Add(-10000*time.Hour))
+ actual, err := d.GetCommentsForRepos([]string{r1, r2}, now.Add(-10000*time.Hour))
assert.NoError(t, err)
deepequal.AssertDeepEqual(t, expected, actual)
}
@@ -257,33 +253,33 @@
injectedError = fmt.Errorf("No comments from the peanut gallery.")
- assert.Error(t, db.PutTaskComment(types.MakeTaskComment(99, 1, 1, 1, now.Add(99*time.Second))))
- assert.Error(t, db.PutTaskComment(types.MakeTaskComment(99, 1, 1, 99, now)))
- assert.Error(t, db.PutTaskComment(types.MakeTaskComment(99, 1, 99, 1, now)))
- assert.Error(t, db.PutTaskComment(types.MakeTaskComment(99, 99, 1, 1, now)))
- assert.Error(t, db.PutTaskSpecComment(types.MakeTaskSpecComment(99, 1, 1, now.Add(99*time.Second))))
- assert.Error(t, db.PutTaskSpecComment(types.MakeTaskSpecComment(99, 1, 99, now)))
- assert.Error(t, db.PutTaskSpecComment(types.MakeTaskSpecComment(99, 99, 1, now)))
- assert.Error(t, db.PutCommitComment(types.MakeCommitComment(99, 1, 1, now.Add(99*time.Second))))
- assert.Error(t, db.PutCommitComment(types.MakeCommitComment(99, 1, 99, now)))
- assert.Error(t, db.PutCommitComment(types.MakeCommitComment(99, 99, 1, now)))
+ assert.Error(t, d.PutTaskComment(types.MakeTaskComment(99, 1, 1, 1, now.Add(99*time.Second))))
+ assert.Error(t, d.PutTaskComment(types.MakeTaskComment(99, 1, 1, 99, now)))
+ assert.Error(t, d.PutTaskComment(types.MakeTaskComment(99, 1, 99, 1, now)))
+ assert.Error(t, d.PutTaskComment(types.MakeTaskComment(99, 99, 1, 1, now)))
+ assert.Error(t, d.PutTaskSpecComment(types.MakeTaskSpecComment(99, 1, 1, now.Add(99*time.Second))))
+ assert.Error(t, d.PutTaskSpecComment(types.MakeTaskSpecComment(99, 1, 99, now)))
+ assert.Error(t, d.PutTaskSpecComment(types.MakeTaskSpecComment(99, 99, 1, now)))
+ assert.Error(t, d.PutCommitComment(types.MakeCommitComment(99, 1, 1, now.Add(99*time.Second))))
+ assert.Error(t, d.PutCommitComment(types.MakeCommitComment(99, 1, 99, now)))
+ assert.Error(t, d.PutCommitComment(types.MakeCommitComment(99, 99, 1, now)))
assert.Equal(t, 10, callCount)
// Assert nothing has changed.
{
- actual, err := db.GetCommentsForRepos([]string{r1, r2}, now.Add(-10000*time.Hour))
+ actual, err := d.GetCommentsForRepos([]string{r1, r2}, now.Add(-10000*time.Hour))
assert.NoError(t, err)
deepequal.AssertDeepEqual(t, expected, actual)
}
- assert.Error(t, db.DeleteTaskComment(tc1))
- assert.Error(t, db.DeleteTaskSpecComment(sc1))
- assert.Error(t, db.DeleteCommitComment(cc1))
+ assert.Error(t, d.DeleteTaskComment(tc1))
+ assert.Error(t, d.DeleteTaskSpecComment(sc1))
+ assert.Error(t, d.DeleteCommitComment(cc1))
// Assert nothing has changed.
{
- actual, err := db.GetCommentsForRepos([]string{r1, r2}, now.Add(-10000*time.Hour))
+ actual, err := d.GetCommentsForRepos([]string{r1, r2}, now.Add(-10000*time.Hour))
assert.NoError(t, err)
deepequal.AssertDeepEqual(t, expected, actual)
}
diff --git a/task_scheduler/go/db/memory/memory.go b/task_scheduler/go/db/memory/memory.go
index 0a9700b..7f6b84d 100644
--- a/task_scheduler/go/db/memory/memory.go
+++ b/task_scheduler/go/db/memory/memory.go
@@ -208,6 +208,9 @@
// NewInMemoryDB returns an extremely simple, inefficient, in-memory DB
// implementation.
-func NewInMemoryDB(modTasks db.ModifiedTasks, modJobs db.ModifiedJobs) db.DB {
- return db.NewDB(NewInMemoryTaskDB(modTasks), NewInMemoryJobDB(modJobs), &db.CommentBox{})
+func NewInMemoryDB(mod db.ModifiedData) db.DB {
+ if mod == nil {
+ mod = modified.NewModifiedData()
+ }
+ return db.NewDB(NewInMemoryTaskDB(mod), NewInMemoryJobDB(mod), &CommentBox{ModifiedComments: mod})
}
diff --git a/task_scheduler/go/db/migrate/migrate.go b/task_scheduler/go/db/migrate/migrate.go
index 33b6cd0..7c3651e 100644
--- a/task_scheduler/go/db/migrate/migrate.go
+++ b/task_scheduler/go/db/migrate/migrate.go
@@ -121,7 +121,7 @@
}
ctx := context.Background()
- oldDB, err := local_db.NewDB(local_db.DB_NAME, *boltDB, nil, nil)
+ oldDB, err := local_db.NewDB(local_db.DB_NAME, *boltDB, nil)
if err != nil {
sklog.Fatal(err)
}
@@ -131,7 +131,7 @@
if err != nil {
sklog.Fatal(err)
}
- newDB, err := firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *fsInstance, ts, nil, nil)
+ newDB, err := firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *fsInstance, ts, nil)
if err != nil {
sklog.Fatal(err)
}
diff --git a/task_scheduler/go/db/modified/modified_data.go b/task_scheduler/go/db/modified/modified_data.go
index 21b3f6f..471355c 100644
--- a/task_scheduler/go/db/modified/modified_data.go
+++ b/task_scheduler/go/db/modified/modified_data.go
@@ -4,6 +4,7 @@
"bytes"
"encoding/gob"
"sort"
+ "strings"
"sync"
"time"
@@ -13,6 +14,12 @@
"go.skia.org/infra/task_scheduler/go/types"
)
+// NewModifiedData returns a db.ModifiedData instance which tracks modifications
+// in memory.
+func NewModifiedData() db.ModifiedData {
+ return db.NewModifiedData(&ModifiedTasksImpl{}, &ModifiedJobsImpl{}, &ModifiedCommentsImpl{})
+}
+
// modifiedData allows subscribers to keep track of DB entries that have been
// modified. It is designed to be used with wrappers in order to store a desired
// type of data.
@@ -125,7 +132,7 @@
if err != nil {
return nil, err
}
- d := types.TaskDecoder{}
+ d := types.NewTaskDecoder()
for _, g := range tasks {
if !d.Process(g) {
break
@@ -178,7 +185,7 @@
if err != nil {
return nil, err
}
- d := types.JobDecoder{}
+ d := types.NewJobDecoder()
for _, g := range jobs {
if !d.Process(g) {
break
@@ -221,5 +228,128 @@
m.m.StopTrackingModifiedEntries(id)
}
+type ModifiedCommentsImpl struct {
+ tasks modifiedData
+ taskSpecs modifiedData
+ commits modifiedData
+}
+
+// See docs for ModifiedComments interface.
+func (m *ModifiedCommentsImpl) GetModifiedComments(id string) ([]*types.TaskComment, []*types.TaskSpecComment, []*types.CommitComment, error) {
+ ids := strings.Split(id, "#")
+ if len(ids) != 3 {
+ return nil, nil, nil, db.ErrUnknownId
+ }
+ tasks, err := m.tasks.GetModifiedEntries(ids[0])
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ d1 := types.NewTaskCommentDecoder()
+ for _, g := range tasks {
+ if !d1.Process(g) {
+ break
+ }
+ }
+ rv1, err := d1.Result()
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ sort.Sort(types.TaskCommentSlice(rv1))
+
+ taskSpecs, err := m.taskSpecs.GetModifiedEntries(ids[1])
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ d2 := types.NewTaskSpecCommentDecoder()
+ for _, g := range taskSpecs {
+ if !d2.Process(g) {
+ break
+ }
+ }
+ rv2, err := d2.Result()
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ sort.Sort(types.TaskSpecCommentSlice(rv2))
+
+ commits, err := m.commits.GetModifiedEntries(ids[2])
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ d3 := types.NewCommitCommentDecoder()
+ for _, g := range commits {
+ if !d3.Process(g) {
+ break
+ }
+ }
+ rv3, err := d3.Result()
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ sort.Sort(types.CommitCommentSlice(rv3))
+
+ return rv1, rv2, rv3, nil
+}
+
+// See docs for ModifiedComments interface.
+func (m *ModifiedCommentsImpl) TrackModifiedTaskComment(c *types.TaskComment) {
+ var buf bytes.Buffer
+ if err := gob.NewEncoder(&buf).Encode(c); err != nil {
+ sklog.Fatal(err)
+ }
+ m.tasks.TrackModifiedEntries(map[string][]byte{c.Id(): buf.Bytes()})
+}
+
+// See docs for ModifiedComments interface.
+func (m *ModifiedCommentsImpl) TrackModifiedTaskSpecComment(c *types.TaskSpecComment) {
+ var buf bytes.Buffer
+ if err := gob.NewEncoder(&buf).Encode(c); err != nil {
+ sklog.Fatal(err)
+ }
+ m.taskSpecs.TrackModifiedEntries(map[string][]byte{c.Id(): buf.Bytes()})
+}
+
+// See docs for ModifiedComments interface.
+func (m *ModifiedCommentsImpl) TrackModifiedCommitComment(c *types.CommitComment) {
+ var buf bytes.Buffer
+ if err := gob.NewEncoder(&buf).Encode(c); err != nil {
+ sklog.Fatal(err)
+ }
+ m.commits.TrackModifiedEntries(map[string][]byte{c.Id(): buf.Bytes()})
+}
+
+// See docs for ModifiedComments interface.
+func (m *ModifiedCommentsImpl) StartTrackingModifiedComments() (string, error) {
+ id1, err := m.tasks.StartTrackingModifiedEntries()
+ if err != nil {
+ return "", err
+ }
+ id2, err := m.taskSpecs.StartTrackingModifiedEntries()
+ if err != nil {
+ m.tasks.StopTrackingModifiedEntries(id1)
+ return "", err
+ }
+ id3, err := m.commits.StartTrackingModifiedEntries()
+ if err != nil {
+ m.tasks.StopTrackingModifiedEntries(id1)
+ m.taskSpecs.StopTrackingModifiedEntries(id2)
+ return "", err
+ }
+ return id1 + "#" + id2 + "#" + id3, nil
+}
+
+// See docs for ModifiedComments interface.
+func (m *ModifiedCommentsImpl) StopTrackingModifiedComments(id string) {
+ ids := strings.Split(id, "#")
+ if len(ids) != 3 {
+ sklog.Errorf("Invalid id %q", id)
+ return
+ }
+ m.tasks.StopTrackingModifiedEntries(ids[0])
+ m.taskSpecs.StopTrackingModifiedEntries(ids[1])
+ m.commits.StopTrackingModifiedEntries(ids[2])
+}
+
var _ db.ModifiedTasks = &ModifiedTasksImpl{}
var _ db.ModifiedJobs = &ModifiedJobsImpl{}
+var _ db.ModifiedComments = &ModifiedCommentsImpl{}
diff --git a/task_scheduler/go/db/modified/modified_data_test.go b/task_scheduler/go/db/modified/modified_data_test.go
index 7d1a222..cb771ce 100644
--- a/task_scheduler/go/db/modified/modified_data_test.go
+++ b/task_scheduler/go/db/modified/modified_data_test.go
@@ -67,3 +67,34 @@
_, err = m.StartTrackingModifiedJobs()
assert.NoError(t, err)
}
+
+func TestDefaultModifiedComments(t *testing.T) {
+ testutils.MediumTest(t)
+ m := &ModifiedCommentsImpl{}
+ db.TestModifiedComments(t, m)
+}
+
+func TestDefaultMultipleCommentModifications(t *testing.T) {
+ testutils.MediumTest(t)
+ m := &ModifiedCommentsImpl{}
+ db.TestMultipleCommentModifications(t, m)
+}
+
+func TestDefaultModifiedCommentsTooManyUsers(t *testing.T) {
+ testutils.MediumTest(t)
+ m := ModifiedCommentsImpl{}
+
+ var oneId string
+ // Max out the number of modified-tasks users; ensure that we error out.
+ for i := 0; i < db.MAX_MODIFIED_DATA_USERS; i++ {
+ id, err := m.StartTrackingModifiedComments()
+ assert.NoError(t, err)
+ oneId = id
+ }
+ _, err := m.StartTrackingModifiedComments()
+ assert.True(t, db.IsTooManyUsers(err))
+
+ m.StopTrackingModifiedComments(oneId)
+ _, err = m.StartTrackingModifiedComments()
+ assert.NoError(t, err)
+}
diff --git a/task_scheduler/go/db/modified/mux.go b/task_scheduler/go/db/modified/mux.go
index 7d25457..d89aa68 100644
--- a/task_scheduler/go/db/modified/mux.go
+++ b/task_scheduler/go/db/modified/mux.go
@@ -7,6 +7,23 @@
"go.skia.org/infra/task_scheduler/go/types"
)
+// NewMuxModifiedData returns a db.ModifiedData implementation which writes to
+// multiple ModifiedData instances but only reads from one.
+func NewMuxModifiedData(readWrite db.ModifiedData, writeOnly ...db.ModifiedData) db.ModifiedData {
+ tWriteOnly := make([]db.ModifiedTasks, 0, len(writeOnly))
+ jWriteOnly := make([]db.ModifiedJobs, 0, len(writeOnly))
+ cWriteOnly := make([]db.ModifiedComments, 0, len(writeOnly))
+ for _, wo := range writeOnly {
+ tWriteOnly = append(tWriteOnly, wo)
+ jWriteOnly = append(jWriteOnly, wo)
+ cWriteOnly = append(cWriteOnly, wo)
+ }
+ t := NewMuxModifiedTasks(readWrite, tWriteOnly...)
+ j := NewMuxModifiedJobs(readWrite, jWriteOnly...)
+ c := NewMuxModifiedComments(readWrite, cWriteOnly...)
+ return db.NewModifiedData(t, j, c)
+}
+
// MuxModifiedTasks is an implementation of db.ModifiedTasks which writes to
// multiple ModifiedTasks instances but only reads from one.
type MuxModifiedTasks struct {
@@ -14,7 +31,7 @@
writeOnly []db.ModifiedTasks
}
-// New MuxModifiedTasks returns an implementation of db.ModifiedTasks which
+// NewMuxModifiedTasks returns an implementation of db.ModifiedTasks which
// writes to multiple ModifiedTasks instances but only reads from one.
func NewMuxModifiedTasks(readWrite db.ModifiedTasks, writeOnly ...db.ModifiedTasks) db.ModifiedTasks {
return &MuxModifiedTasks{
@@ -70,3 +87,43 @@
wo.TrackModifiedJobsGOB(dbModified, gobs)
}
}
+
+// MuxModifiedComments is an implementation of db.ModifiedComments which writes
+// to multiple ModifiedJobs instances but only reads from one.
+type MuxModifiedComments struct {
+ db.ModifiedComments
+ writeOnly []db.ModifiedComments
+}
+
+// New MuxModifiedComments returns an implementation of db.ModifiedComments
+// which writes to multiple ModifiedJobs instances but only reads from one.
+func NewMuxModifiedComments(readWrite db.ModifiedComments, writeOnly ...db.ModifiedComments) db.ModifiedComments {
+ return &MuxModifiedComments{
+ ModifiedComments: readWrite,
+ writeOnly: writeOnly,
+ }
+}
+
+// See documentation for db.ModifiedComments interface.
+func (m *MuxModifiedComments) TrackModifiedTaskComment(c *types.TaskComment) {
+ m.ModifiedComments.TrackModifiedTaskComment(c)
+ for _, wo := range m.writeOnly {
+ wo.TrackModifiedTaskComment(c)
+ }
+}
+
+// See documentation for db.ModifiedComments interface.
+func (m *MuxModifiedComments) TrackModifiedTaskSpecComment(c *types.TaskSpecComment) {
+ m.ModifiedComments.TrackModifiedTaskSpecComment(c)
+ for _, wo := range m.writeOnly {
+ wo.TrackModifiedTaskSpecComment(c)
+ }
+}
+
+// See documentation for db.ModifiedComments interface.
+func (m *MuxModifiedComments) TrackModifiedCommitComment(c *types.CommitComment) {
+ m.ModifiedComments.TrackModifiedCommitComment(c)
+ for _, wo := range m.writeOnly {
+ wo.TrackModifiedCommitComment(c)
+ }
+}
diff --git a/task_scheduler/go/db/modified/mux_test.go b/task_scheduler/go/db/modified/mux_test.go
index fc63641..a09bd10 100644
--- a/task_scheduler/go/db/modified/mux_test.go
+++ b/task_scheduler/go/db/modified/mux_test.go
@@ -104,3 +104,53 @@
m.TrackModifiedJob(t1)
check(t1)
}
+
+func TestMuxModifiedComments(t *testing.T) {
+ testutils.MediumTest(t)
+ m := NewMuxModifiedComments(&ModifiedCommentsImpl{}, &ModifiedCommentsImpl{})
+ db.TestModifiedComments(t, m)
+}
+
+func TestMuxMultipleCommentModifications(t *testing.T) {
+ testutils.MediumTest(t)
+ m := NewMuxModifiedComments(&ModifiedCommentsImpl{}, &ModifiedCommentsImpl{})
+ db.TestMultipleCommentModifications(t, m)
+}
+
+// Simple test to verify that we actually write to the write-only ModifiedComments.
+func TestMuxModifiedCommentsWriteOnly(t *testing.T) {
+ testutils.MediumTest(t)
+ rw := &ModifiedCommentsImpl{}
+ w1 := &ModifiedCommentsImpl{}
+ w2 := &ModifiedCommentsImpl{}
+ w3 := &ModifiedCommentsImpl{}
+ m := NewMuxModifiedComments(rw, w1, w2, w3)
+ rwId, err := m.StartTrackingModifiedComments()
+ assert.NoError(t, err)
+ wo := []db.ModifiedComments{w1, w2, w3}
+ ids := []string{}
+ for _, w := range wo {
+ id, err := w.StartTrackingModifiedComments()
+ assert.NoError(t, err)
+ ids = append(ids, id)
+ }
+
+ check := func(e1 []*types.TaskComment, e2 []*types.TaskSpecComment, e3 []*types.CommitComment) {
+ a1, a2, a3, err := m.GetModifiedComments(rwId)
+ assert.NoError(t, err)
+ deepequal.AssertDeepEqual(t, e1, a1)
+ deepequal.AssertDeepEqual(t, e2, a2)
+ deepequal.AssertDeepEqual(t, e3, a3)
+ for idx, w := range []db.ModifiedComments{w1, w2, w3} {
+ a1, a2, a3, err := w.GetModifiedComments(ids[idx])
+ assert.NoError(t, err)
+ deepequal.AssertDeepEqual(t, e1, a1)
+ deepequal.AssertDeepEqual(t, e2, a2)
+ deepequal.AssertDeepEqual(t, e3, a3)
+ }
+ }
+ check([]*types.TaskComment{}, []*types.TaskSpecComment{}, []*types.CommitComment{})
+ c1 := types.MakeTaskComment(1, 1, 1, 1, time.Now())
+ m.TrackModifiedTaskComment(c1)
+ check([]*types.TaskComment{c1}, []*types.TaskSpecComment{}, []*types.CommitComment{})
+}
diff --git a/task_scheduler/go/db/pubsub/modified.go b/task_scheduler/go/db/pubsub/modified.go
index 442ea85..0d95504 100644
--- a/task_scheduler/go/db/pubsub/modified.go
+++ b/task_scheduler/go/db/pubsub/modified.go
@@ -6,6 +6,7 @@
"encoding/gob"
"fmt"
"sort"
+ "strings"
"sync"
"time"
@@ -18,6 +19,27 @@
"google.golang.org/api/option"
)
+// NewModifiedData returns a db.ModifiedData instance which uses pubsub.
+func NewModifiedData(topicSet, label string, ts oauth2.TokenSource) (db.ModifiedData, error) {
+ topicSetObj, ok := topics[topicSet]
+ if !ok {
+ return nil, fmt.Errorf("Topic must be one of %v, not %q", VALID_TOPIC_SETS, topicSet)
+ }
+ t, err := NewModifiedTasks(topicSetObj.tasks, label, ts)
+ if err != nil {
+ return nil, err
+ }
+ j, err := NewModifiedJobs(topicSetObj.jobs, label, ts)
+ if err != nil {
+ return nil, err
+ }
+ c, err := NewModifiedComments(topicSetObj.taskComments, topicSetObj.taskSpecComments, topicSetObj.commitComments, label, ts)
+ if err != nil {
+ return nil, err
+ }
+ return db.NewModifiedData(t, j, c), nil
+}
+
type entry struct {
ts time.Time
data []byte
@@ -133,7 +155,7 @@
data: m.Data,
}
} else {
- sklog.Debugf("Received duplicate or outdated message for %s", dataId)
+ sklog.Debugf("Received duplicate or outdated message (%s vs %s) for %s", prev.ts, dbModified, dataId)
}
return nil
})
@@ -346,3 +368,164 @@
func (c *jobClient) TrackModifiedJobsGOB(ts time.Time, jobsById map[string][]byte) {
c.publisher.publishGOB(ts, jobsById)
}
+
+// commentClient implements db.ModifiedComments using pubsub.
+type commentClient struct {
+ tasks *modifiedClient
+ taskSpecs *modifiedClient
+ commits *modifiedClient
+}
+
+// NewModifiedComments returns a db.ModifiedComments which uses pubsub. The
+// topics should be one of the sets of TOPIC_* constants defined in this
+// package. The subscriberLabel is included in the subscription ID, along with a
+// timestamp; this should help to debug zombie subscriptions. It should be
+// descriptive and unique to this process, or if the process uses multiple
+// instances of ModifiedJobs, unique to each instance.
+func NewModifiedComments(taskCommentsTopic, taskSpecCommentsTopic, commitCommentsTopic string, label string, ts oauth2.TokenSource) (db.ModifiedComments, error) {
+ c, err := pubsub.NewClient(context.Background(), PROJECT_ID, option.WithTokenSource(ts))
+ if err != nil {
+ return nil, err
+ }
+ tasks, err := newModifiedClient(c, taskCommentsTopic, label)
+ if err != nil {
+ return nil, err
+ }
+ taskSpecs, err := newModifiedClient(c, taskSpecCommentsTopic, label)
+ if err != nil {
+ return nil, err
+ }
+ commits, err := newModifiedClient(c, commitCommentsTopic, label)
+ if err != nil {
+ return nil, err
+ }
+ return &commentClient{
+ tasks: tasks,
+ taskSpecs: taskSpecs,
+ commits: commits,
+ }, nil
+}
+
+// See documentation for db.ModifiedComments interface.
+func (c *commentClient) GetModifiedComments(id string) ([]*types.TaskComment, []*types.TaskSpecComment, []*types.CommitComment, error) {
+ ids := strings.Split(id, "#")
+ if len(ids) != 3 {
+ return nil, nil, nil, db.ErrUnknownId
+ }
+ gobs, err := c.tasks.getModifiedData(ids[0])
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ rv1 := make([]*types.TaskComment, 0, len(gobs))
+ for _, g := range gobs {
+ var c types.TaskComment
+ if err := gob.NewDecoder(bytes.NewReader(g)).Decode(&c); err != nil {
+ // We didn't attempt to decode the blob in the pubsub
+ // message when we received it. Ignore this job.
+ sklog.Errorf("Failed to decode job from pubsub message: %s", err)
+ } else {
+ rv1 = append(rv1, &c)
+ }
+ }
+ sort.Sort(types.TaskCommentSlice(rv1))
+
+ gobs, err = c.taskSpecs.getModifiedData(ids[1])
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ rv2 := make([]*types.TaskSpecComment, 0, len(gobs))
+ for _, g := range gobs {
+ var c types.TaskSpecComment
+ if err := gob.NewDecoder(bytes.NewReader(g)).Decode(&c); err != nil {
+ // We didn't attempt to decode the blob in the pubsub
+ // message when we received it. Ignore this job.
+ sklog.Errorf("Failed to decode job from pubsub message: %s", err)
+ } else {
+ rv2 = append(rv2, &c)
+ }
+ }
+ sort.Sort(types.TaskSpecCommentSlice(rv2))
+
+ gobs, err = c.commits.getModifiedData(ids[2])
+ if err != nil {
+ return nil, nil, nil, err
+ }
+ rv3 := make([]*types.CommitComment, 0, len(gobs))
+ for _, g := range gobs {
+ var c types.CommitComment
+ if err := gob.NewDecoder(bytes.NewReader(g)).Decode(&c); err != nil {
+ // We didn't attempt to decode the blob in the pubsub
+ // message when we received it. Ignore this job.
+ sklog.Errorf("Failed to decode job from pubsub message: %s", err)
+ } else {
+ rv3 = append(rv3, &c)
+ }
+ }
+ sort.Sort(types.CommitCommentSlice(rv3))
+ return rv1, rv2, rv3, nil
+}
+
+// See documentation for db.ModifiedComments interface.
+func (c *commentClient) StartTrackingModifiedComments() (string, error) {
+ id1, err := c.tasks.startTrackingModifiedData()
+ if err != nil {
+ return "", err
+ }
+ id2, err := c.taskSpecs.startTrackingModifiedData()
+ if err != nil {
+ return "", err
+ }
+ id3, err := c.commits.startTrackingModifiedData()
+ if err != nil {
+ return "", err
+ }
+ return id1 + "#" + id2 + "#" + id3, nil
+}
+
+// See documentation for db.ModifiedComments interface.
+func (c *commentClient) StopTrackingModifiedComments(id string) {
+ ids := strings.Split(id, "#")
+ if len(ids) != 3 {
+ sklog.Errorf("Invalid ID %q", id)
+ return
+ }
+ c.tasks.stopTrackingModifiedData(ids[0])
+ c.taskSpecs.stopTrackingModifiedData(ids[1])
+ c.commits.stopTrackingModifiedData(ids[2])
+}
+
+// See documentation for db.ModifiedComments interface.
+func (c *commentClient) TrackModifiedTaskComment(tc *types.TaskComment) {
+ // Hack: since the timestamp is part of the ID, we can't change it. But,
+ // we have to provide a different timestamp from the one we sent when
+ // the comment was created, or else it'll get de-duplicated.
+ ts := tc.Timestamp
+ if tc.Deleted != nil && *tc.Deleted {
+ ts = time.Now()
+ }
+ c.tasks.publisher.publish(tc.Id(), ts, tc)
+}
+
+// See documentation for db.ModifiedComments interface.
+func (c *commentClient) TrackModifiedTaskSpecComment(tc *types.TaskSpecComment) {
+ // Hack: since the timestamp is part of the ID, we can't change it. But,
+ // we have to provide a different timestamp from the one we sent when
+ // the comment was created, or else it'll get de-duplicated.
+ ts := tc.Timestamp
+ if tc.Deleted != nil && *tc.Deleted {
+ ts = time.Now()
+ }
+ c.taskSpecs.publisher.publish(tc.Id(), ts, tc)
+}
+
+// See documentation for db.ModifiedComments interface.
+func (c *commentClient) TrackModifiedCommitComment(cc *types.CommitComment) {
+ // Hack: since the timestamp is part of the ID, we can't change it. But,
+ // we have to provide a different timestamp from the one we sent when
+ // the comment was created, or else it'll get de-duplicated.
+ ts := cc.Timestamp
+ if cc.Deleted != nil && *cc.Deleted {
+ ts = time.Now()
+ }
+ c.commits.publisher.publish(cc.Id(), ts, cc)
+}
diff --git a/task_scheduler/go/db/pubsub/modified_test.go b/task_scheduler/go/db/pubsub/modified_test.go
index 9a5014a..c87ce0f 100644
--- a/task_scheduler/go/db/pubsub/modified_test.go
+++ b/task_scheduler/go/db/pubsub/modified_test.go
@@ -45,3 +45,23 @@
m := setupJobs(t)
db.TestMultipleJobModifications(t, m)
}
+
+func setupComments(t *testing.T) db.ModifiedComments {
+ testutils.LargeTest(t)
+ topic1 := fmt.Sprintf("modified-comments-test-tasks-%s", uuid.New())
+ topic2 := fmt.Sprintf("modified-comments-test-taskspecs-%s", uuid.New())
+ topic3 := fmt.Sprintf("modified-comments-test-commits-%s", uuid.New())
+ m, err := NewModifiedComments(topic1, topic2, topic3, "fake-label", nil)
+ assert.NoError(t, err)
+ return m
+}
+
+func TestPubsubModifiedComments(t *testing.T) {
+ m := setupComments(t)
+ db.TestModifiedComments(t, m)
+}
+
+func TestPubsubMultipleCommentModifications(t *testing.T) {
+ m := setupComments(t)
+ db.TestMultipleCommentModifications(t, m)
+}
diff --git a/task_scheduler/go/db/pubsub/pubsub.go b/task_scheduler/go/db/pubsub/pubsub.go
index e5efb42..18e6ca0 100644
--- a/task_scheduler/go/db/pubsub/pubsub.go
+++ b/task_scheduler/go/db/pubsub/pubsub.go
@@ -24,13 +24,27 @@
// Default project ID.
PROJECT_ID = "skia-public"
+ // Sets of topic, based on scheduler instance.
+ TOPIC_SET_PROD = "prod"
+ TOPIC_SET_INTERNAL = "internal"
+ TOPIC_SET_STAGING = "staging"
+
// Known topic names.
- TOPIC_TASKS = "task-scheduler-modified-tasks"
- TOPIC_TASKS_INTERNAL = "task-scheduler-modified-tasks-internal"
- TOPIC_TASKS_STAGING = "task-scheduler-modified-tasks-staging"
- TOPIC_JOBS = "task-scheduler-modified-jobs"
- TOPIC_JOBS_INTERNAL = "task-scheduler-modified-jobs-internal"
- TOPIC_JOBS_STAGING = "task-scheduler-modified-jobs-staging"
+ TOPIC_TASKS = "task-scheduler-modified-tasks"
+ TOPIC_TASKS_INTERNAL = "task-scheduler-modified-tasks-internal"
+ TOPIC_TASKS_STAGING = "task-scheduler-modified-tasks-staging"
+ TOPIC_JOBS = "task-scheduler-modified-jobs"
+ TOPIC_JOBS_INTERNAL = "task-scheduler-modified-jobs-internal"
+ TOPIC_JOBS_STAGING = "task-scheduler-modified-jobs-staging"
+ TOPIC_TASK_COMMENTS = "task-scheduler-modified-task-comments"
+ TOPIC_TASK_COMMENTS_INTERNAL = "task-scheduler-modified-task-comments-internal"
+ TOPIC_TASK_COMMENTS_STAGING = "task-scheduler-modified-task-comments-staging"
+ TOPIC_TASKSPEC_COMMENTS = "task-scheduler-modified-taskspec-comments"
+ TOPIC_TASKSPEC_COMMENTS_INTERNAL = "task-scheduler-modified-taskspec-comments-internal"
+ TOPIC_TASKSPEC_COMMENTS_STAGING = "task-scheduler-modified-taskspec-comments-staging"
+ TOPIC_COMMIT_COMMENTS = "task-scheduler-modified-commit-comments"
+ TOPIC_COMMIT_COMMENTS_INTERNAL = "task-scheduler-modified-commit-comments-internal"
+ TOPIC_COMMIT_COMMENTS_STAGING = "task-scheduler-modified-commit-comments-staging"
// Attributes sent with all pubsub messages.
@@ -42,6 +56,47 @@
ATTR_SENDER_ID = "sender"
)
+var (
+ VALID_TOPIC_SETS = []string{
+ TOPIC_SET_PROD,
+ TOPIC_SET_INTERNAL,
+ TOPIC_SET_STAGING,
+ }
+
+ topics = map[string]topicSet{
+ TOPIC_SET_PROD: topicSet{
+ tasks: TOPIC_TASKS,
+ jobs: TOPIC_JOBS,
+ taskComments: TOPIC_TASK_COMMENTS,
+ taskSpecComments: TOPIC_TASKSPEC_COMMENTS,
+ commitComments: TOPIC_COMMIT_COMMENTS,
+ },
+ TOPIC_SET_INTERNAL: topicSet{
+ tasks: TOPIC_TASKS_INTERNAL,
+ jobs: TOPIC_JOBS_INTERNAL,
+ taskComments: TOPIC_TASK_COMMENTS_INTERNAL,
+ taskSpecComments: TOPIC_TASKSPEC_COMMENTS_INTERNAL,
+ commitComments: TOPIC_COMMIT_COMMENTS_INTERNAL,
+ },
+ TOPIC_SET_STAGING: topicSet{
+ tasks: TOPIC_TASKS_STAGING,
+ jobs: TOPIC_JOBS_STAGING,
+ taskComments: TOPIC_TASK_COMMENTS_STAGING,
+ taskSpecComments: TOPIC_TASKSPEC_COMMENTS_STAGING,
+ commitComments: TOPIC_COMMIT_COMMENTS_STAGING,
+ },
+ }
+)
+
+// topicSet is used for organizing sets of pubsub topics.
+type topicSet struct {
+ tasks string
+ jobs string
+ taskComments string
+ taskSpecComments string
+ commitComments string
+}
+
// publisher sends pubsub messages for modified tasks and jobs.
type publisher struct {
senderId string
@@ -157,6 +212,78 @@
p.publish(j.Id, j.DbModified, j)
}
+// TaskSpecCommentPublisher sends pubsub messages for comments.
+type TaskSpecCommentPublisher struct {
+ *publisher
+}
+
+// NewTaskSpecCommentPublisher creates a TaskSpecCommentPublisher instance. It
+// creates the given topic if it does not already exist.
+func NewTaskSpecCommentPublisher(topic string, ts oauth2.TokenSource) (*TaskSpecCommentPublisher, error) {
+ c, err := pubsub.NewClient(context.Background(), PROJECT_ID, option.WithTokenSource(ts))
+ if err != nil {
+ return nil, err
+ }
+ pub, err := newPublisher(c, topic)
+ if err != nil {
+ return nil, err
+ }
+ return &TaskSpecCommentPublisher{pub}, nil
+}
+
+// Publish publishes a pubsub message for the given comment.
+func (p *TaskSpecCommentPublisher) Publish(t *types.TaskSpecComment) {
+ p.publish(t.Id(), t.Timestamp, t)
+}
+
+// TaskCommentPublisher sends pubsub messages for comments.
+type TaskCommentPublisher struct {
+ *publisher
+}
+
+// NewTaskCommentPublisher creates a TaskCommentPublisher instance. It creates the given
+// topic if it does not already exist.
+func NewTaskCommentPublisher(topic string, ts oauth2.TokenSource) (*TaskCommentPublisher, error) {
+ c, err := pubsub.NewClient(context.Background(), PROJECT_ID, option.WithTokenSource(ts))
+ if err != nil {
+ return nil, err
+ }
+ pub, err := newPublisher(c, topic)
+ if err != nil {
+ return nil, err
+ }
+ return &TaskCommentPublisher{pub}, nil
+}
+
+// Publish publishes a pubsub message for the given comment.
+func (p *TaskCommentPublisher) Publish(t *types.TaskComment) {
+ p.publish(t.Id(), t.Timestamp, t)
+}
+
+// CommitCommentPublisher sends pubsub messages for comments.
+type CommitCommentPublisher struct {
+ *publisher
+}
+
+// NewCommitCommentPublisher creates a CommitCommentPublisher instance. It creates the given
+// topic if it does not already exist.
+func NewCommitCommentPublisher(topic string, ts oauth2.TokenSource) (*CommitCommentPublisher, error) {
+ c, err := pubsub.NewClient(context.Background(), PROJECT_ID, option.WithTokenSource(ts))
+ if err != nil {
+ return nil, err
+ }
+ pub, err := newPublisher(c, topic)
+ if err != nil {
+ return nil, err
+ }
+ return &CommitCommentPublisher{pub}, nil
+}
+
+// Publish publishes a pubsub message for the given comment.
+func (p *CommitCommentPublisher) Publish(t *types.CommitComment) {
+ p.publish(t.Id(), t.Timestamp, t)
+}
+
// subscriber uses pubsub to watch for modified data.
type subscriber struct {
client *pubsub.Client
@@ -296,3 +423,90 @@
}
return s.start()
}
+
+// NewTaskCommentSubscriber creates a subscriber which calls the given callback
+// function for every pubsub message. The topic should be one of the TOPIC_*
+// constants defined in this package. The subscriberLabel is included in the
+// subscription ID, along with a timestamp; this should help to debug zombie
+// subscriptions. Acknowledgement of the message is done automatically based on
+// the return value of the callback: if the callback returns an error, the
+// message is Nack'd and will be re-sent at a later time, otherwise the message
+// is Ack'd and will not be re-sent. Therefore, if the comment is not valid or
+// otherwise cannot ever be processed, the callback should return nil to prevent
+// the message from being re-sent.
+func NewTaskCommentSubscriber(topic, subscriberLabel string, ts oauth2.TokenSource, callback func(*types.TaskComment) error) (context.CancelFunc, error) {
+ c, err := pubsub.NewClient(context.Background(), PROJECT_ID, option.WithTokenSource(ts))
+ if err != nil {
+ return nil, err
+ }
+ s, err := newSubscriber(c, topic, subscriberLabel, func(m *pubsub.Message) error {
+ var c types.TaskComment
+ if err := gob.NewDecoder(bytes.NewReader(m.Data)).Decode(&c); err != nil {
+ sklog.Errorf("Failed to decode TaskComment from pubsub message: %s", err)
+ return nil // We will never be able to process this message.
+ }
+ return callback(&c)
+ })
+ if err != nil {
+ return nil, err
+ }
+ return s.start()
+}
+
+// NewTaskSpecCommentSubscriber creates a subscriber which calls the given
+// callback function for every pubsub message. The topic should be one of the
+// TOPIC_* constants defined in this package. The subscriberLabel is included in
+// the subscription ID, along with a timestamp; this should help to debug zombie
+// subscriptions. Acknowledgement of the message is done automatically based on
+// the return value of the callback: if the callback returns an error, the
+// message is Nack'd and will be re-sent at a later time, otherwise the message
+// is Ack'd and will not be re-sent. Therefore, if the comment is not valid or
+// otherwise cannot ever be processed, the callback should return nil to prevent
+// the message from being re-sent.
+func NewTaskSpecCommentSubscriber(topic, subscriberLabel string, ts oauth2.TokenSource, callback func(*types.TaskSpecComment) error) (context.CancelFunc, error) {
+ c, err := pubsub.NewClient(context.Background(), PROJECT_ID, option.WithTokenSource(ts))
+ if err != nil {
+ return nil, err
+ }
+ s, err := newSubscriber(c, topic, subscriberLabel, func(m *pubsub.Message) error {
+ var c types.TaskSpecComment
+ if err := gob.NewDecoder(bytes.NewReader(m.Data)).Decode(&c); err != nil {
+ sklog.Errorf("Failed to decode TaskSpecComment from pubsub message: %s", err)
+ return nil // We will never be able to process this message.
+ }
+ return callback(&c)
+ })
+ if err != nil {
+ return nil, err
+ }
+ return s.start()
+}
+
+// NewCommitCommentSubscriber creates a subscriber which calls the given
+// callback function for every pubsub message. The topic should be one of the
+// TOPIC_* constants defined in this package. The subscriberLabel is included in
+// the subscription ID, along with a timestamp; this should help to debug zombie
+// subscriptions. Acknowledgement of the message is done automatically based on
+// the return value of the callback: if the callback returns an error, the
+// message is Nack'd and will be re-sent at a later time, otherwise the message
+// is Ack'd and will not be re-sent. Therefore, if the comment is not valid or
+// otherwise cannot ever be processed, the callback should return nil to prevent
+// the message from being re-sent.
+func NewCommitCommentSubscriber(topic, subscriberLabel string, ts oauth2.TokenSource, callback func(*types.CommitComment) error) (context.CancelFunc, error) {
+ c, err := pubsub.NewClient(context.Background(), PROJECT_ID, option.WithTokenSource(ts))
+ if err != nil {
+ return nil, err
+ }
+ s, err := newSubscriber(c, topic, subscriberLabel, func(m *pubsub.Message) error {
+ var c types.CommitComment
+ if err := gob.NewDecoder(bytes.NewReader(m.Data)).Decode(&c); err != nil {
+ sklog.Errorf("Failed to decode CommitComment from pubsub message: %s", err)
+ return nil // We will never be able to process this message.
+ }
+ return callback(&c)
+ })
+ if err != nil {
+ return nil, err
+ }
+ return s.start()
+}
diff --git a/task_scheduler/go/db/recovery/backups_test.go b/task_scheduler/go/db/recovery/backups_test.go
index 0cd38e4..eda14e3 100644
--- a/task_scheduler/go/db/recovery/backups_test.go
+++ b/task_scheduler/go/db/recovery/backups_test.go
@@ -136,7 +136,7 @@
assert.NoError(t, os.MkdirAll(path.Join(dir, TRIGGER_DIRNAME), os.ModePerm))
db := &testDB{
- DB: memory.NewInMemoryDB(nil, nil),
+ DB: memory.NewInMemoryDB(nil),
content: content,
ts: time.Unix(TEST_DB_TIME, 0),
}
diff --git a/task_scheduler/go/db/remote_db/remote_db.go b/task_scheduler/go/db/remote_db/remote_db.go
index 8b79c3f..7c20be5 100644
--- a/task_scheduler/go/db/remote_db/remote_db.go
+++ b/task_scheduler/go/db/remote_db/remote_db.go
@@ -80,27 +80,21 @@
type client struct {
serverRoot string
client *http.Client
- db.ModifiedTasks
- db.ModifiedJobs
+ db.ModifiedData
}
// NewClient returns a db.RemoteDB that connects to the server created by
// NewServer. serverRoot should end with a slash.
-func NewClient(serverRoot, tasksTopic, jobsTopic, label string, ts oauth2.TokenSource) (db.RemoteDB, error) {
+func NewClient(serverRoot, topicSet, label string, ts oauth2.TokenSource) (db.RemoteDB, error) {
c := httputils.DefaultClientConfig().WithTokenSource(ts).Client()
- modTasks, err := pubsub.NewModifiedTasks(tasksTopic, label, ts)
- if err != nil {
- return nil, err
- }
- modJobs, err := pubsub.NewModifiedJobs(jobsTopic, label, ts)
+ mod, err := pubsub.NewModifiedData(topicSet, label, ts)
if err != nil {
return nil, err
}
return &client{
- serverRoot: serverRoot,
- client: c,
- ModifiedTasks: modTasks,
- ModifiedJobs: modJobs,
+ serverRoot: serverRoot,
+ client: c,
+ ModifiedData: mod,
}, nil
}
diff --git a/task_scheduler/go/db/remote_db/remote_db_test.go b/task_scheduler/go/db/remote_db/remote_db_test.go
index 926ecb8..f5b9b31 100644
--- a/task_scheduler/go/db/remote_db/remote_db_test.go
+++ b/task_scheduler/go/db/remote_db/remote_db_test.go
@@ -97,17 +97,15 @@
// makeDB sets up a client/server pair wrapped in a clientWithBackdoor.
func makeDB(t *testing.T) db.DBCloser {
serverLabel := fmt.Sprintf("remote-db-test-%s", uuid.New())
- modTasks, err := pubsub.NewModifiedTasks(pubsub.TOPIC_TASKS, serverLabel, nil)
+ mod, err := pubsub.NewModifiedData(pubsub.TOPIC_SET_PROD, serverLabel, nil)
assert.NoError(t, err)
- modJobs, err := pubsub.NewModifiedJobs(pubsub.TOPIC_JOBS, serverLabel, nil)
- assert.NoError(t, err)
- baseDB := memory.NewInMemoryDB(modTasks, modJobs)
+ baseDB := memory.NewInMemoryDB(mod)
r := mux.NewRouter()
err = RegisterServer(baseDB, r.PathPrefix("/db").Subrouter())
assert.NoError(t, err)
ts := httptest.NewServer(r)
clientLabel := fmt.Sprintf("remote-db-test-%s", uuid.New())
- dbclient, err := NewClient(ts.URL+"/db/", pubsub.TOPIC_TASKS, pubsub.TOPIC_JOBS, clientLabel, nil)
+ dbclient, err := NewClient(ts.URL+"/db/", pubsub.TOPIC_SET_PROD, clientLabel, nil)
assert.NoError(t, err)
dbclient.(*client).client.Transport = newReqCountingTransport(dbclient.(*client).client.Transport)
return &clientWithBackdoor{
diff --git a/task_scheduler/go/db/testutil.go b/task_scheduler/go/db/testutil.go
index e80d30b..bf650f1 100644
--- a/task_scheduler/go/db/testutil.go
+++ b/task_scheduler/go/db/testutil.go
@@ -72,6 +72,33 @@
}))
}
+func findModifiedComments(t testutils.TestingT, m ModifiedComments, id string, e1 []*types.TaskComment, e2 []*types.TaskSpecComment, e3 []*types.CommitComment) {
+ // Note that the slice only works because we don't call
+ // TrackModifiedJob more than once for any given job, otherwise
+ // we'd have to use a map and compare DbModified.
+ a1 := []*types.TaskComment{}
+ a2 := []*types.TaskSpecComment{}
+ a3 := []*types.CommitComment{}
+ assert.NoError(t, testutils.EventuallyConsistent(10*time.Second, func() error {
+ c1, c2, c3, err := m.GetModifiedComments(id)
+ assert.NoError(t, err)
+ a1 = append(a1, c1...)
+ a2 = append(a2, c2...)
+ a3 = append(a3, c3...)
+ if len(a1) != len(e1) || len(a2) != len(e2) || len(a3) != len(e3) {
+ time.Sleep(100 * time.Millisecond)
+ return testutils.TryAgainErr
+ }
+ sort.Sort(types.TaskCommentSlice(a1))
+ sort.Sort(types.TaskSpecCommentSlice(a2))
+ sort.Sort(types.CommitCommentSlice(a3))
+ deepequal.AssertDeepEqual(t, e1, a1)
+ deepequal.AssertDeepEqual(t, e2, a2)
+ deepequal.AssertDeepEqual(t, e3, a3)
+ return nil
+ }))
+}
+
// TestTaskDB performs basic tests for an implementation of TaskDB.
func TestTaskDB(t testutils.TestingT, db TaskDB) {
_, err := db.GetModifiedTasks("dummy-id")
@@ -1174,7 +1201,19 @@
// TestCommentDB validates that db correctly implements the CommentDB interface.
func TestCommentDB(t testutils.TestingT, db CommentDB) {
- now := time.Now()
+ _, _, _, err := db.GetModifiedComments("dummy-id")
+ assert.True(t, IsUnknownId(err))
+
+ id, err := db.StartTrackingModifiedComments()
+ assert.NoError(t, err)
+
+ c1, c2, c3, err := db.GetModifiedComments(id)
+ assert.NoError(t, err)
+ assert.Equal(t, 0, len(c1))
+ assert.Equal(t, 0, len(c2))
+ assert.Equal(t, 0, len(c3))
+
+ now := time.Now().Truncate(TS_RESOLUTION)
// Empty db.
r0 := fmt.Sprintf("%s%d", common.REPO_SKIA, 0)
@@ -1195,40 +1234,43 @@
}
// Add some comments.
- tc1 := types.MakeTaskComment(1, 1, 1, 1, now)
- tc2 := types.MakeTaskComment(2, 1, 1, 1, now.Add(2*time.Second))
- tc3 := types.MakeTaskComment(3, 1, 1, 1, now.Add(time.Second))
- tc4 := types.MakeTaskComment(4, 1, 1, 2, now)
- tc5 := types.MakeTaskComment(5, 1, 2, 2, now)
- tc6 := types.MakeTaskComment(6, 2, 3, 3, now)
+ tc1 := types.MakeTaskComment(1, 1, 1, 1, now.Add(-2*time.Second))
+ tc2 := types.MakeTaskComment(2, 1, 1, 1, now)
+ tc3 := types.MakeTaskComment(3, 1, 1, 1, now.Add(-time.Second))
+ tc4 := types.MakeTaskComment(4, 1, 1, 2, now.Add(-2*time.Second+time.Millisecond))
+ tc5 := types.MakeTaskComment(5, 1, 2, 2, now.Add(-2*time.Second+2*time.Millisecond))
+ tc6 := types.MakeTaskComment(6, 2, 3, 3, now.Add(-2*time.Second+3*time.Millisecond))
for _, c := range []*types.TaskComment{tc1, tc2, tc3, tc4, tc5, tc6} {
assert.NoError(t, db.PutTaskComment(c))
}
tc6copy := tc6.Copy()
tc6.Message = "modifying after Put shouldn't affect stored comment"
- sc1 := types.MakeTaskSpecComment(1, 1, 1, now)
- sc2 := types.MakeTaskSpecComment(2, 1, 1, now.Add(2*time.Second))
- sc3 := types.MakeTaskSpecComment(3, 1, 1, now.Add(time.Second))
- sc4 := types.MakeTaskSpecComment(4, 1, 2, now)
- sc5 := types.MakeTaskSpecComment(5, 2, 3, now)
+ sc1 := types.MakeTaskSpecComment(1, 1, 1, now.Add(-2*time.Second))
+ sc2 := types.MakeTaskSpecComment(2, 1, 1, now)
+ sc3 := types.MakeTaskSpecComment(3, 1, 1, now.Add(-time.Second))
+ sc4 := types.MakeTaskSpecComment(4, 1, 2, now.Add(-2*time.Second+time.Millisecond))
+ sc5 := types.MakeTaskSpecComment(5, 2, 3, now.Add(-2*time.Second+2*time.Millisecond))
for _, c := range []*types.TaskSpecComment{sc1, sc2, sc3, sc4, sc5} {
assert.NoError(t, db.PutTaskSpecComment(c))
}
sc5copy := sc5.Copy()
sc5.Message = "modifying after Put shouldn't affect stored comment"
- cc1 := types.MakeCommitComment(1, 1, 1, now)
- cc2 := types.MakeCommitComment(2, 1, 1, now.Add(2*time.Second))
- cc3 := types.MakeCommitComment(3, 1, 1, now.Add(time.Second))
- cc4 := types.MakeCommitComment(4, 1, 2, now)
- cc5 := types.MakeCommitComment(5, 2, 3, now)
+ cc1 := types.MakeCommitComment(1, 1, 1, now.Add(-2*time.Second))
+ cc2 := types.MakeCommitComment(2, 1, 1, now)
+ cc3 := types.MakeCommitComment(3, 1, 1, now.Add(-time.Second))
+ cc4 := types.MakeCommitComment(4, 1, 2, now.Add(-2*time.Second+time.Millisecond))
+ cc5 := types.MakeCommitComment(5, 2, 3, now.Add(-2*time.Second+2*time.Millisecond))
for _, c := range []*types.CommitComment{cc1, cc2, cc3, cc4, cc5} {
assert.NoError(t, db.PutCommitComment(c))
}
cc5copy := cc5.Copy()
cc5.Message = "modifying after Put shouldn't affect stored comment"
+ // Ensure that all comments show up in the modified list.
+ findModifiedComments(t, db, id, []*types.TaskComment{tc1, tc4, tc5, tc6copy, tc3, tc2}, []*types.TaskSpecComment{sc1, sc4, sc5copy, sc3, sc2}, []*types.CommitComment{cc1, cc4, cc5copy, cc3, cc2})
+
// Check that adding duplicate non-identical comment gives an error.
tc1different := tc1.Copy()
tc1different.Message = "not the same"
@@ -1285,7 +1327,7 @@
// Specifying a cutoff time shouldn't drop required comments.
{
- actual, err := db.GetCommentsForRepos([]string{r1}, now.Add(time.Second))
+ actual, err := db.GetCommentsForRepos([]string{r1}, now.Add(-time.Second))
assert.NoError(t, err)
assert.Equal(t, 1, len(actual))
{
@@ -1319,6 +1361,9 @@
AssertDeepEqual(t, cc2, ccs[offset+1])
}
}
+ // Clear out modified comments.
+ _, _, _, err = db.GetModifiedComments(id)
+ assert.NoError(t, err)
// Delete some comments.
assert.NoError(t, db.DeleteTaskComment(tc3))
@@ -1347,6 +1392,14 @@
actual, err := db.GetCommentsForRepos([]string{r0, r1, r2}, now.Add(-10000*time.Hour))
assert.NoError(t, err)
AssertDeepEqual(t, expected, actual)
+ deleted := true
+ tc1different.Deleted = &deleted
+ sc1different.Deleted = &deleted
+ cc1different.Deleted = &deleted
+ tc3.Deleted = &deleted
+ sc3.Deleted = &deleted
+ cc3.Deleted = &deleted
+ findModifiedComments(t, db, id, []*types.TaskComment{tc1different, tc3}, []*types.TaskSpecComment{sc1different, sc3}, []*types.CommitComment{cc1different, cc3})
}
// Delete all the comments.
@@ -1664,6 +1717,93 @@
}))
}
+func TestModifiedComments(t testutils.TestingT, m ModifiedComments) {
+ _, _, _, err := m.GetModifiedComments("dummy-id")
+ assert.True(t, IsUnknownId(err))
+
+ id, err := m.StartTrackingModifiedComments()
+ assert.NoError(t, err)
+
+ time.Sleep(time.Second)
+ a1, a2, a3, err := m.GetModifiedComments(id)
+ assert.NoError(t, err)
+ assert.Equal(t, 0, len(a1))
+ assert.Equal(t, 0, len(a2))
+ assert.Equal(t, 0, len(a3))
+
+ c1 := types.MakeTaskComment(1, 1, 1, 1, time.Now())
+
+ // Insert the comment.
+ m.TrackModifiedTaskComment(c1)
+
+ // Ensure that the comment shows up in the modified list.
+ findModifiedComments(t, m, id, []*types.TaskComment{c1}, []*types.TaskSpecComment{}, []*types.CommitComment{})
+
+ // Insert two more comments.
+ c2 := types.MakeTaskComment(2, 2, 2, 2, time.Now())
+ m.TrackModifiedTaskComment(c2)
+ c3 := types.MakeTaskComment(3, 3, 3, 3, time.Now())
+ m.TrackModifiedTaskComment(c3)
+
+ // Ensure that both comments show up in the modified list.
+ findModifiedComments(t, m, id, []*types.TaskComment{c2, c3}, []*types.TaskSpecComment{}, []*types.CommitComment{})
+
+ // Insert two more comments.
+ c4 := types.MakeTaskSpecComment(4, 4, 4, time.Now())
+ m.TrackModifiedTaskSpecComment(c4)
+ c5 := types.MakeCommitComment(5, 5, 5, time.Now())
+ m.TrackModifiedCommitComment(c5)
+
+ // Ensure that both comments show up in the modified list.
+ findModifiedComments(t, m, id, []*types.TaskComment{}, []*types.TaskSpecComment{c4}, []*types.CommitComment{c5})
+
+ // Check StopTrackingModifiedComments.
+ m.StopTrackingModifiedComments(id)
+ err = testutils.EventuallyConsistent(10*time.Second, func() error {
+ _, _, _, err := m.GetModifiedComments(id)
+ if err == nil {
+ return testutils.TryAgainErr
+ }
+ return err
+ })
+ assert.True(t, IsUnknownId(err))
+}
+
+func TestMultipleCommentModifications(t testutils.TestingT, m ModifiedComments) {
+ id, err := m.StartTrackingModifiedComments()
+ assert.NoError(t, err)
+
+ c1 := types.MakeTaskComment(1, 1, 1, 1, time.Now())
+
+ // Insert the comment.
+ m.TrackModifiedTaskComment(c1)
+
+ // Delete the comment.
+ deleted := true
+ c1.Deleted = &deleted
+ m.TrackModifiedTaskComment(c1)
+
+ // Ensure that the comment shows up only once in the modified list and
+ // is the most recent value.
+ var actual *types.TaskComment
+ assert.NoError(t, testutils.EventuallyConsistent(10*time.Second, func() error {
+ a1, _, _, err := m.GetModifiedComments(id)
+ if err != nil {
+ return err
+ }
+ if len(a1) == 1 {
+ if actual == nil || actual.Deleted == nil {
+ actual = a1[0]
+ }
+ }
+ if deepequal.DeepEqual(c1, actual) {
+ return nil
+ }
+ time.Sleep(100 * time.Millisecond)
+ return testutils.TryAgainErr
+ }))
+}
+
func TestUpdateDBFromSwarmingTask(t testutils.TestingT, db TaskDB) {
// Create task, initialize from swarming, and save.
now := time.Now().UTC().Round(time.Microsecond)
diff --git a/task_scheduler/go/scheduling/perftest/perftest.go b/task_scheduler/go/scheduling/perftest/perftest.go
index aa1a1bc..96b710b 100644
--- a/task_scheduler/go/scheduling/perftest/perftest.go
+++ b/task_scheduler/go/scheduling/perftest/perftest.go
@@ -265,7 +265,7 @@
assertNoError(err)
assertDeepEqual([]string{head}, commits)
- d, err := local_db.NewDB("testdb", path.Join(workdir, "tasks.db"), nil, nil)
+ d, err := local_db.NewDB("testdb", path.Join(workdir, "tasks.db"), nil)
assertNoError(err)
w, err := window.New(time.Hour, 0, nil)
assertNoError(err)
diff --git a/task_scheduler/go/scheduling/task_scheduler_test.go b/task_scheduler/go/scheduling/task_scheduler_test.go
index 642945a..6ea2f69 100644
--- a/task_scheduler/go/scheduling/task_scheduler_test.go
+++ b/task_scheduler/go/scheduling/task_scheduler_test.go
@@ -194,7 +194,7 @@
assert.NoError(t, err)
assert.NoError(t, os.Mkdir(path.Join(tmp, periodic_triggers.TRIGGER_DIRNAME), os.ModePerm))
- d := memory.NewInMemoryDB(nil, nil)
+ d := memory.NewInMemoryDB(nil)
isolateClient, err := isolate.NewClient(tmp, isolate.ISOLATE_SERVER_URL_FAKE)
assert.NoError(t, err)
swarmingClient := swarming_testutils.NewTestClient()
@@ -2008,7 +2008,7 @@
gb.Commit(ctx)
// Setup the scheduler.
- d := memory.NewInMemoryDB(nil, nil)
+ d := memory.NewInMemoryDB(nil)
isolateClient, err := isolate.NewClient(workdir, isolate.ISOLATE_SERVER_URL_FAKE)
assert.NoError(t, err)
swarmingClient := swarming_testutils.NewTestClient()
@@ -3383,7 +3383,7 @@
func TestValidateTaskForUpdate(t *testing.T) {
testutils.SmallTest(t)
- d := memory.NewInMemoryDB(nil, nil)
+ d := memory.NewInMemoryDB(nil)
c1 := "abc123"
c2 := "def456"
@@ -3461,7 +3461,7 @@
func TestUpdateTask(t *testing.T) {
testutils.SmallTest(t)
- d := memory.NewInMemoryDB(nil, nil)
+ d := memory.NewInMemoryDB(nil)
c1 := "abc123"
c2 := "def456"
diff --git a/task_scheduler/go/task_scheduler/main.go b/task_scheduler/go/task_scheduler/main.go
index dce39f1..1c195ec 100644
--- a/task_scheduler/go/task_scheduler/main.go
+++ b/task_scheduler/go/task_scheduler/main.go
@@ -87,8 +87,7 @@
firestoreInstance = flag.String("firestore_instance", "", "Firestore instance to use, eg. \"prod\"")
isolateServer = flag.String("isolate_server", isolate.ISOLATE_SERVER_URL, "Which Isolate server to use.")
local = flag.Bool("local", false, "Whether we're running on a dev machine vs in production.")
- pubsubTopicTasks = flag.String("pubsub_topic_tasks", "", "Pubsub topic for tasks.")
- pubsubTopicJobs = flag.String("pubsub_topic_jobs", "", "Pubsub topic for jobs.")
+ pubsubTopicSet = flag.String("pubsub_topic_set", "", fmt.Sprintf("Pubsub topic set; one of: %v", pubsub.VALID_TOPIC_SETS))
repoUrls = common.NewMultiStringFlag("repo", nil, "Repositories for which to schedule tasks.")
recipesCfgFile = flag.String("recipes_cfg", "", "Path to the recipes.cfg file.")
resourcesDir = flag.String("resources_dir", "", "The directory to find templates, JS, and CSS files. If blank, assumes you're running inside a checkout and will attempt to find the resources relative to this source file.")
@@ -656,23 +655,18 @@
// Initialize the database.
label := *host
- modTasks, err := pubsub.NewModifiedTasks(*pubsubTopicTasks, label, tokenSource)
+ mod, err := pubsub.NewModifiedData(*pubsubTopicSet, label, tokenSource)
if err != nil {
sklog.Fatal(err)
}
- modTasks = modified.NewMuxModifiedTasks(&modified.ModifiedTasksImpl{}, modTasks)
- modJobs, err := pubsub.NewModifiedJobs(*pubsubTopicJobs, label, tokenSource)
- if err != nil {
- sklog.Fatal(err)
- }
- modJobs = modified.NewMuxModifiedJobs(&modified.ModifiedJobsImpl{}, modJobs)
+ mod = modified.NewMuxModifiedData(modified.NewModifiedData(), mod)
if *firestoreInstance != "" {
- tsDb, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, tokenSource, modTasks, modJobs)
+ tsDb, err = firestore.NewDB(ctx, firestore.FIRESTORE_PROJECT, *firestoreInstance, tokenSource, mod)
if err != nil {
sklog.Fatalf("Failed to create Firestore DB client: %s", err)
}
} else {
- tsDb, err = local_db.NewDB(local_db.DB_NAME, path.Join(wdAbs, local_db.DB_FILENAME), modTasks, modJobs)
+ tsDb, err = local_db.NewDB(local_db.DB_NAME, path.Join(wdAbs, local_db.DB_FILENAME), mod)
if err != nil {
sklog.Fatal(err)
}
diff --git a/task_scheduler/go/tryjobs/testutils.go b/task_scheduler/go/tryjobs/testutils.go
index ae3a409..4f29a5f 100644
--- a/task_scheduler/go/tryjobs/testutils.go
+++ b/task_scheduler/go/tryjobs/testutils.go
@@ -111,7 +111,7 @@
assert.NoError(t, err)
taskCfgCache, err := specs.NewTaskCfgCache(ctx, rm, depot_tools_testutils.GetDepotTools(t, ctx), path.Join(tmpDir, "cache"), specs.DEFAULT_NUM_WORKERS)
assert.NoError(t, err)
- d, err := local_db.NewDB("tasks_db", path.Join(tmpDir, "tasks.db"), nil, nil)
+ d, err := local_db.NewDB("tasks_db", path.Join(tmpDir, "tasks.db"), nil)
assert.NoError(t, err)
mock := mockhttpclient.NewURLMock()
projectRepoMapping := map[string]string{
diff --git a/task_scheduler/go/types/comments.go b/task_scheduler/go/types/comments.go
index d0af194..05302c7 100644
--- a/task_scheduler/go/types/comments.go
+++ b/task_scheduler/go/types/comments.go
@@ -6,6 +6,7 @@
"time"
"go.skia.org/infra/go/sklog"
+ "go.skia.org/infra/go/util"
)
// TaskComment contains a comment about a Task. {Repo, Revision, Name,
@@ -18,13 +19,23 @@
// Timestamp is compared ignoring timezone. The timezone reflects User's
// location.
Timestamp time.Time `json:"time"`
- TaskId string `json:"taskId"`
- User string `json:"user"`
- Message string `json:"message"`
+ TaskId string `json:"taskId,omitempty"`
+ User string `json:"user,omitempty"`
+ Message string `json:"message,omitempty"`
+ Deleted *bool `json:"deleted,omitempty"`
}
func (c TaskComment) Copy() *TaskComment {
- return &c
+ rv := &c
+ if c.Deleted != nil {
+ v := *c.Deleted
+ rv.Deleted = &v
+ }
+ return rv
+}
+
+func (c *TaskComment) Id() string {
+ return c.Repo + "#" + c.Revision + "#" + c.Name + "#" + c.Timestamp.Format(util.SAFE_TIMESTAMP_FORMAT)
}
// TaskSpecComment contains a comment about a TaskSpec. {Repo, Name, Timestamp}
@@ -35,14 +46,24 @@
// Timestamp is compared ignoring timezone. The timezone reflects User's
// location.
Timestamp time.Time `json:"time"`
- User string `json:"user"`
+ User string `json:"user,omitempty"`
Flaky bool `json:"flaky"`
IgnoreFailure bool `json:"ignoreFailure"`
- Message string `json:"message"`
+ Message string `json:"message,omitempty"`
+ Deleted *bool `json:"deleted,omitempty"`
}
func (c TaskSpecComment) Copy() *TaskSpecComment {
- return &c
+ rv := &c
+ if c.Deleted != nil {
+ v := *c.Deleted
+ rv.Deleted = &v
+ }
+ return rv
+}
+
+func (c *TaskSpecComment) Id() string {
+ return c.Repo + "#" + c.Name + "#" + c.Timestamp.Format(util.SAFE_TIMESTAMP_FORMAT)
}
// CommitComment contains a comment about a commit. {Repo, Revision, Timestamp}
@@ -53,13 +74,23 @@
// Timestamp is compared ignoring timezone. The timezone reflects User's
// location.
Timestamp time.Time `json:"time"`
- User string `json:"user"`
+ User string `json:"user,omitempty"`
IgnoreFailure bool `json:"ignoreFailure"`
- Message string `json:"message"`
+ Message string `json:"message,omitempty"`
+ Deleted *bool `json:"deleted,omitempty"`
}
func (c CommitComment) Copy() *CommitComment {
- return &c
+ rv := &c
+ if c.Deleted != nil {
+ v := *c.Deleted
+ rv.Deleted = &v
+ }
+ return rv
+}
+
+func (c *CommitComment) Id() string {
+ return c.Repo + "#" + c.Revision + "#" + c.Timestamp.Format(util.SAFE_TIMESTAMP_FORMAT)
}
// RepoComments contains comments that all pertain to the same repository.
@@ -90,3 +121,199 @@
}
return ©
}
+
+// TaskCommentSlice implements sort.Interface. To sort taskComments
+// []*TaskComment, use sort.Sort(TaskCommentSlice(taskComments)).
+type TaskCommentSlice []*TaskComment
+
+func (s TaskCommentSlice) Len() int { return len(s) }
+
+func (s TaskCommentSlice) Less(i, j int) bool {
+ return s[i].Timestamp.Before(s[j].Timestamp)
+}
+
+func (s TaskCommentSlice) Swap(i, j int) {
+ s[i], s[j] = s[j], s[i]
+}
+
+// TaskSpecCommentSlice implements sort.Interface. To sort taskSpecComments
+// []*TaskSpecComment, use sort.Sort(TaskSpecCommentSlice(taskSpecComments)).
+type TaskSpecCommentSlice []*TaskSpecComment
+
+func (s TaskSpecCommentSlice) Len() int { return len(s) }
+
+func (s TaskSpecCommentSlice) Less(i, j int) bool {
+ return s[i].Timestamp.Before(s[j].Timestamp)
+}
+
+func (s TaskSpecCommentSlice) Swap(i, j int) {
+ s[i], s[j] = s[j], s[i]
+}
+
+// CommitCommentSlice implements sort.Interface. To sort commitComments
+// []*CommitComment, use sort.Sort(CommitCommentSlice(commitComments)).
+type CommitCommentSlice []*CommitComment
+
+func (s CommitCommentSlice) Len() int { return len(s) }
+
+func (s CommitCommentSlice) Less(i, j int) bool {
+ return s[i].Timestamp.Before(s[j].Timestamp)
+}
+
+func (s CommitCommentSlice) Swap(i, j int) {
+ s[i], s[j] = s[j], s[i]
+}
+
+// TaskCommentEncoder encodes TaskComments into bytes via GOB encoding. Not
+// safe for concurrent use.
+type TaskCommentEncoder struct {
+ util.GobEncoder
+}
+
+// Next returns one of the TaskComments provided to Process (in arbitrary order)
+// and its serialized bytes. If any comments remain, returns the TaskComment,
+// the serialized bytes, nil. If all comments have been returned, returns nil,
+// nil, nil. If an error is encountered, returns nil, nil, error.
+func (e *TaskCommentEncoder) Next() (*TaskComment, []byte, error) {
+ item, serialized, err := e.GobEncoder.Next()
+ if err != nil {
+ return nil, nil, err
+ } else if item == nil {
+ return nil, nil, nil
+ }
+ return item.(*TaskComment), serialized, err
+}
+
+// TaskCommentDecoder decodes bytes into TaskComments via GOB decoding. Not safe
+// for concurrent use.
+type TaskCommentDecoder struct {
+ *util.GobDecoder
+}
+
+// NewTaskCommentDecoder returns a TaskCommentDecoder instance.
+func NewTaskCommentDecoder() *TaskCommentDecoder {
+ return &TaskCommentDecoder{
+ GobDecoder: util.NewGobDecoder(func() interface{} {
+ return &TaskComment{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*TaskComment{}
+ for item := range ch {
+ items = append(items, item.(*TaskComment))
+ }
+ return items
+ }),
+ }
+}
+
+// Result returns all decoded TaskComments provided to Process (in arbitrary
+// order), or any error encountered.
+func (d *TaskCommentDecoder) Result() ([]*TaskComment, error) {
+ res, err := d.GobDecoder.Result()
+ if err != nil {
+ return nil, err
+ }
+ return res.([]*TaskComment), nil
+}
+
+// TaskSpecCommentEncoder encodes TaskSpecComments into bytes via GOB encoding.
+// Not safe for concurrent use.
+type TaskSpecCommentEncoder struct {
+ util.GobEncoder
+}
+
+// Next returns one of the TaskSpecComments provided to Process (in arbitrary
+// order) and its serialized bytes. If any comments remain, returns the
+// TaskSpecComment, the serialized bytes, nil. If all comments have been
+// returned, returns nil, nil, nil. If an error is encountered, returns nil,
+// nil, error.
+func (e *TaskSpecCommentEncoder) Next() (*TaskSpecComment, []byte, error) {
+ item, serialized, err := e.GobEncoder.Next()
+ if err != nil {
+ return nil, nil, err
+ } else if item == nil {
+ return nil, nil, nil
+ }
+ return item.(*TaskSpecComment), serialized, err
+}
+
+// TaskSpecCommentDecoder decodes bytes into TaskSpecComments via GOB decoding.
+// Not safe for concurrent use.
+type TaskSpecCommentDecoder struct {
+ *util.GobDecoder
+}
+
+// NewTaskSpecCommentDecoder returns a TaskSpecCommentDecoder instance.
+func NewTaskSpecCommentDecoder() *TaskSpecCommentDecoder {
+ return &TaskSpecCommentDecoder{
+ GobDecoder: util.NewGobDecoder(func() interface{} {
+ return &TaskSpecComment{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*TaskSpecComment{}
+ for item := range ch {
+ items = append(items, item.(*TaskSpecComment))
+ }
+ return items
+ }),
+ }
+}
+
+// Result returns all decoded TaskSpecComments provided to Process (in arbitrary
+// order), or any error encountered.
+func (d *TaskSpecCommentDecoder) Result() ([]*TaskSpecComment, error) {
+ res, err := d.GobDecoder.Result()
+ if err != nil {
+ return nil, err
+ }
+ return res.([]*TaskSpecComment), nil
+}
+
+// CommitCommentEncoder encodes CommitComments into bytes via GOB encoding. Not
+// safe for concurrent use.
+type CommitCommentEncoder struct {
+ util.GobEncoder
+}
+
+// Next returns one of the CommitComments provided to Process (in arbitrary
+// order) and its serialized bytes. If any comments remain, returns the
+// CommitComment, the serialized bytes, nil. If all comments have been returned,
+// returns nil, nil, nil. If an error is encountered, returns nil, nil, error.
+func (e *CommitCommentEncoder) Next() (*CommitComment, []byte, error) {
+ item, serialized, err := e.GobEncoder.Next()
+ if err != nil {
+ return nil, nil, err
+ } else if item == nil {
+ return nil, nil, nil
+ }
+ return item.(*CommitComment), serialized, err
+}
+
+// CommitCommentDecoder decodes bytes into CommitComments via GOB decoding.
+// Not safe for concurrent use.
+type CommitCommentDecoder struct {
+ *util.GobDecoder
+}
+
+// NewCommitCommentDecoder returns a CommitCommentDecoder instance.
+func NewCommitCommentDecoder() *CommitCommentDecoder {
+ return &CommitCommentDecoder{
+ GobDecoder: util.NewGobDecoder(func() interface{} {
+ return &CommitComment{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*CommitComment{}
+ for item := range ch {
+ items = append(items, item.(*CommitComment))
+ }
+ return items
+ }),
+ }
+}
+
+// Result returns all decoded CommitComments provided to Process (in arbitrary
+// order), or any error encountered.
+func (d *CommitCommentDecoder) Result() ([]*CommitComment, error) {
+ res, err := d.GobDecoder.Result()
+ if err != nil {
+ return nil, err
+ }
+ return res.([]*CommitComment), nil
+}
diff --git a/task_scheduler/go/types/comments_test.go b/task_scheduler/go/types/comments_test.go
index 19fb7d1..ccf6708 100644
--- a/task_scheduler/go/types/comments_test.go
+++ b/task_scheduler/go/types/comments_test.go
@@ -11,6 +11,8 @@
func TestCopyTaskComment(t *testing.T) {
testutils.SmallTest(t)
v := MakeTaskComment(1, 1, 1, 1, time.Now())
+ deleted := true
+ v.Deleted = &deleted
deepequal.AssertCopy(t, v, v.Copy())
}
@@ -19,6 +21,8 @@
v := MakeTaskSpecComment(1, 1, 1, time.Now())
v.Flaky = true
v.IgnoreFailure = true
+ deleted := true
+ v.Deleted = &deleted
deepequal.AssertCopy(t, v, v.Copy())
}
@@ -26,6 +30,8 @@
testutils.SmallTest(t)
v := MakeCommitComment(1, 1, 1, time.Now())
v.IgnoreFailure = true
+ deleted := true
+ v.Deleted = &deleted
deepequal.AssertCopy(t, v, v.Copy())
}
diff --git a/task_scheduler/go/types/job.go b/task_scheduler/go/types/job.go
index b4bd6ca..1dac7ad 100644
--- a/task_scheduler/go/types/job.go
+++ b/task_scheduler/go/types/job.go
@@ -1,13 +1,11 @@
package types
import (
- "bytes"
- "encoding/gob"
"fmt"
- "sync"
"time"
"go.skia.org/infra/go/sklog"
+ "go.skia.org/infra/go/util"
)
const (
@@ -308,29 +306,7 @@
// concurrent use.
// TODO(benjaminwagner): Encode in parallel.
type JobEncoder struct {
- err error
- jobs []*Job
- result [][]byte
-}
-
-// Process encodes the Job into a byte slice that will be returned from Next()
-// (in arbitrary order). Returns false if Next is certain to return an error.
-// Caller must ensure j does not change until after the first call to Next().
-// May not be called after calling Next().
-func (e *JobEncoder) Process(j *Job) bool {
- if e.err != nil {
- return false
- }
- var buf bytes.Buffer
- if err := gob.NewEncoder(&buf).Encode(j); err != nil {
- e.err = err
- e.jobs = nil
- e.result = nil
- return false
- }
- e.jobs = append(e.jobs, j)
- e.result = append(e.result, buf.Bytes())
- return true
+ util.GobEncoder
}
// Next returns one of the Jobs provided to Process (in arbitrary order) and
@@ -338,113 +314,42 @@
// bytes, nil. If all jobs have been returned, returns nil, nil, nil. If an
// error is encountered, returns nil, nil, error.
func (e *JobEncoder) Next() (*Job, []byte, error) {
- if e.err != nil {
- return nil, nil, e.err
- }
- if len(e.jobs) == 0 {
+ item, serialized, err := e.GobEncoder.Next()
+ if err != nil {
+ return nil, nil, err
+ } else if item == nil {
return nil, nil, nil
}
- j := e.jobs[0]
- e.jobs = e.jobs[1:]
- serialized := e.result[0]
- e.result = e.result[1:]
- return j, serialized, nil
+ return item.(*Job), serialized, nil
}
// JobDecoder decodes bytes into Jobs via GOB decoding. Not safe for
// concurrent use.
type JobDecoder struct {
- // input contains the incoming byte slices. Process() sends on this channel,
- // decode() receives from it, and Result() closes it.
- input chan []byte
- // output contains decoded Jobs. decode() sends on this channel, collect()
- // receives from it, and run() closes it when all decode() goroutines have
- // finished.
- output chan *Job
- // result contains the return value of Result(). collect() sends a single
- // value on this channel and closes it. Result() receives from it.
- result chan []*Job
- // errors contains the first error from any goroutine. It's a channel in case
- // multiple goroutines experience an error at the same time.
- errors chan error
+ *util.GobDecoder
}
-// init initializes d if it has not been initialized. May not be called concurrently.
-func (d *JobDecoder) init() {
- if d.input == nil {
- d.input = make(chan []byte, kNumDecoderGoroutines*2)
- d.output = make(chan *Job, kNumDecoderGoroutines)
- d.result = make(chan []*Job, 1)
- d.errors = make(chan error, kNumDecoderGoroutines)
- go d.run()
- go d.collect()
+// NewJobDecoder returns a JobDecoder instance.
+func NewJobDecoder() *JobDecoder {
+ return &JobDecoder{
+ GobDecoder: util.NewGobDecoder(func() interface{} {
+ return &Job{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*Job{}
+ for item := range ch {
+ items = append(items, item.(*Job))
+ }
+ return items
+ }),
}
}
-// run starts the decode goroutines and closes d.output when they finish.
-func (d *JobDecoder) run() {
- // Start decoders.
- wg := sync.WaitGroup{}
- for i := 0; i < kNumDecoderGoroutines; i++ {
- wg.Add(1)
- go d.decode(&wg)
- }
- // Wait for decoders to exit.
- wg.Wait()
- // Drain d.input in the case that errors were encountered, to avoid deadlock.
- for range d.input {
- }
- close(d.output)
-}
-
-// decode receives from d.input and sends to d.output until d.input is closed or
-// d.errors is non-empty. Decrements wg when done.
-func (d *JobDecoder) decode(wg *sync.WaitGroup) {
- for b := range d.input {
- var j Job
- if err := gob.NewDecoder(bytes.NewReader(b)).Decode(&j); err != nil {
- d.errors <- err
- break
- }
- d.output <- &j
- if len(d.errors) > 0 {
- break
- }
- }
- wg.Done()
-}
-
-// collect receives from d.output until it is closed, then sends on d.result.
-func (d *JobDecoder) collect() {
- result := []*Job{}
- for j := range d.output {
- result = append(result, j)
- }
- d.result <- result
- close(d.result)
-}
-
-// Process decodes the byte slice into a Job and includes it in Result() (in
-// arbitrary order). Returns false if Result is certain to return an error.
-// Caller must ensure b does not change until after Result() returns.
-func (d *JobDecoder) Process(b []byte) bool {
- d.init()
- d.input <- b
- return len(d.errors) == 0
-}
-
// Result returns all decoded Jobs provided to Process (in arbitrary order), or
// any error encountered.
func (d *JobDecoder) Result() ([]*Job, error) {
- // Allow JobDecoder to be used without initialization.
- if d.result == nil {
- return []*Job{}, nil
- }
- close(d.input)
- select {
- case err := <-d.errors:
+ res, err := d.GobDecoder.Result()
+ if err != nil {
return nil, err
- case result := <-d.result:
- return result, nil
}
+ return res.([]*Job), nil
}
diff --git a/task_scheduler/go/types/job_test.go b/task_scheduler/go/types/job_test.go
index 22f67e9..b2ecb32 100644
--- a/task_scheduler/go/types/job_test.go
+++ b/task_scheduler/go/types/job_test.go
@@ -82,7 +82,7 @@
func TestJobDecoder(t *testing.T) {
testutils.SmallTest(t)
- d := JobDecoder{}
+ d := NewJobDecoder()
expectedJobs := map[string]*Job{}
for i := 0; i < 250; i++ {
job := &Job{}
@@ -109,7 +109,7 @@
func TestJobDecoderNoJobs(t *testing.T) {
testutils.SmallTest(t)
- d := JobDecoder{}
+ d := NewJobDecoder()
result, err := d.Result()
assert.NoError(t, err)
assert.Equal(t, 0, len(result))
@@ -125,7 +125,7 @@
serialized := buf.Bytes()
invalid := append([]byte("Hi Mom!"), serialized...)
- d := JobDecoder{}
+ d := NewJobDecoder()
// Process should return true before it encounters an invalid result.
assert.True(t, d.Process(serialized))
assert.True(t, d.Process(serialized))
diff --git a/task_scheduler/go/types/task.go b/task_scheduler/go/types/task.go
index e8dfed4..d49e394 100644
--- a/task_scheduler/go/types/task.go
+++ b/task_scheduler/go/types/task.go
@@ -1,14 +1,11 @@
package types
import (
- "bytes"
- "encoding/gob"
"fmt"
"reflect"
"sort"
"strconv"
"strings"
- "sync"
"time"
"unicode/utf8"
@@ -447,31 +444,8 @@
// TaskEncoder encodes Tasks into bytes via GOB encoding. Not safe for
// concurrent use.
-// TODO(benjaminwagner): Encode in parallel.
type TaskEncoder struct {
- err error
- tasks []*Task
- result [][]byte
-}
-
-// Process encodes the Task into a byte slice that will be returned from Next()
-// (in arbitrary order). Returns false if Next is certain to return an error.
-// Caller must ensure t does not change until after the first call to Next().
-// May not be called after calling Next().
-func (e *TaskEncoder) Process(t *Task) bool {
- if e.err != nil {
- return false
- }
- var buf bytes.Buffer
- if err := gob.NewEncoder(&buf).Encode(t); err != nil {
- e.err = err
- e.tasks = nil
- e.result = nil
- return false
- }
- e.tasks = append(e.tasks, t)
- e.result = append(e.result, buf.Bytes())
- return true
+ util.GobEncoder
}
// Next returns one of the Tasks provided to Process (in arbitrary order) and
@@ -479,117 +453,44 @@
// bytes, nil. If all tasks have been returned, returns nil, nil, nil. If an
// error is encountered, returns nil, nil, error.
func (e *TaskEncoder) Next() (*Task, []byte, error) {
- if e.err != nil {
- return nil, nil, e.err
- }
- if len(e.tasks) == 0 {
+ item, serialized, err := e.GobEncoder.Next()
+ if err != nil {
+ return nil, nil, err
+ } else if item == nil {
return nil, nil, nil
}
- t := e.tasks[0]
- e.tasks = e.tasks[1:]
- serialized := e.result[0]
- e.result = e.result[1:]
- return t, serialized, nil
+ return item.(*Task), serialized, nil
}
// TaskDecoder decodes bytes into Tasks via GOB decoding. Not safe for
// concurrent use.
type TaskDecoder struct {
- // input contains the incoming byte slices. Process() sends on this channel,
- // decode() receives from it, and Result() closes it.
- input chan []byte
- // output contains decoded Tasks. decode() sends on this channel, collect()
- // receives from it, and run() closes it when all decode() goroutines have
- // finished.
- output chan *Task
- // result contains the return value of Result(). collect() sends a single
- // value on this channel and closes it. Result() receives from it.
- result chan []*Task
- // errors contains the first error from any goroutine. It's a channel in case
- // multiple goroutines experience an error at the same time.
- errors chan error
+ *util.GobDecoder
}
-const kNumDecoderGoroutines = 10
-
-// init initializes d if it has not been initialized. May not be called concurrently.
-func (d *TaskDecoder) init() {
- if d.input == nil {
- d.input = make(chan []byte, kNumDecoderGoroutines*2)
- d.output = make(chan *Task, kNumDecoderGoroutines)
- d.result = make(chan []*Task, 1)
- d.errors = make(chan error, kNumDecoderGoroutines)
- go d.run()
- go d.collect()
+// NewTaskDecoder returns a TaskDecoder instance.
+func NewTaskDecoder() *TaskDecoder {
+ return &TaskDecoder{
+ GobDecoder: util.NewGobDecoder(func() interface{} {
+ return &Task{}
+ }, func(ch <-chan interface{}) interface{} {
+ items := []*Task{}
+ for item := range ch {
+ items = append(items, item.(*Task))
+ }
+ return items
+ }),
}
}
-// run starts the decode goroutines and closes d.output when they finish.
-func (d *TaskDecoder) run() {
- // Start decoders.
- wg := sync.WaitGroup{}
- for i := 0; i < kNumDecoderGoroutines; i++ {
- wg.Add(1)
- go d.decode(&wg)
- }
- // Wait for decoders to exit.
- wg.Wait()
- // Drain d.input in the case that errors were encountered, to avoid deadlock.
- for range d.input {
- }
- close(d.output)
-}
-
-// decode receives from d.input and sends to d.output until d.input is closed or
-// d.errors is non-empty. Decrements wg when done.
-func (d *TaskDecoder) decode(wg *sync.WaitGroup) {
- for b := range d.input {
- var t Task
- if err := gob.NewDecoder(bytes.NewReader(b)).Decode(&t); err != nil {
- d.errors <- err
- break
- }
- d.output <- &t
- if len(d.errors) > 0 {
- break
- }
- }
- wg.Done()
-}
-
-// collect receives from d.output until it is closed, then sends on d.result.
-func (d *TaskDecoder) collect() {
- result := []*Task{}
- for t := range d.output {
- result = append(result, t)
- }
- d.result <- result
- close(d.result)
-}
-
-// Process decodes the byte slice into a Task and includes it in Result() (in
-// arbitrary order). Returns false if Result is certain to return an error.
-// Caller must ensure b does not change until after Result() returns.
-func (d *TaskDecoder) Process(b []byte) bool {
- d.init()
- d.input <- b
- return len(d.errors) == 0
-}
-
// Result returns all decoded Tasks provided to Process (in arbitrary order), or
// any error encountered.
func (d *TaskDecoder) Result() ([]*Task, error) {
- // Allow TaskDecoder to be used without initialization.
- if d.result == nil {
- return []*Task{}, nil
- }
- close(d.input)
- select {
- case err := <-d.errors:
+ res, err := d.GobDecoder.Result()
+ if err != nil {
return nil, err
- case result := <-d.result:
- return result, nil
}
+ return res.([]*Task), nil
}
// TagsForTask returns the tags which should be set for a Task.
diff --git a/task_scheduler/go/types/task_test.go b/task_scheduler/go/types/task_test.go
index e6bcbe4..c1e53fd 100644
--- a/task_scheduler/go/types/task_test.go
+++ b/task_scheduler/go/types/task_test.go
@@ -590,7 +590,7 @@
func TestTaskDecoder(t *testing.T) {
testutils.SmallTest(t)
- d := TaskDecoder{}
+ d := NewTaskDecoder()
expectedTasks := map[string]*Task{}
for i := 0; i < 250; i++ {
task := &Task{}
@@ -617,7 +617,7 @@
func TestTaskDecoderNoTasks(t *testing.T) {
testutils.SmallTest(t)
- d := TaskDecoder{}
+ d := NewTaskDecoder()
result, err := d.Result()
assert.NoError(t, err)
assert.Equal(t, 0, len(result))
@@ -633,7 +633,7 @@
serialized := buf.Bytes()
invalid := append([]byte("Hi Mom!"), serialized...)
- d := TaskDecoder{}
+ d := NewTaskDecoder()
// Process should return true before it encounters an invalid result.
assert.True(t, d.Process(serialized))
assert.True(t, d.Process(serialized))
diff --git a/task_scheduler/sys/task-scheduler-internal.service b/task_scheduler/sys/task-scheduler-internal.service
index 24724fe..e13856a 100644
--- a/task_scheduler/sys/task-scheduler-internal.service
+++ b/task_scheduler/sys/task-scheduler-internal.service
@@ -11,8 +11,7 @@
--isolate_server=https://chrome-isolated.appspot.com \
--logtostderr \
--pool=SkiaInternal \
- --pubsub_topic_tasks=task-scheduler-modified-tasks-internal \
- --pubsub_topic_jobs=task-scheduler-modified-jobs-internal \
+ --pubsub_topic_set=internal \
--pubsub_topic=swarming-tasks-internal \
--pubsub_subscriber=task-scheduler-internal \
--repo=https://skia.googlesource.com/internal_test.git \
diff --git a/task_scheduler/sys/task-scheduler-staging.service b/task_scheduler/sys/task-scheduler-staging.service
index 51e7b30..a322539 100644
--- a/task_scheduler/sys/task-scheduler-staging.service
+++ b/task_scheduler/sys/task-scheduler-staging.service
@@ -12,8 +12,7 @@
--isolate_server=https://isolateserver-dev.appspot.com \
--logtostderr \
--pool=Skia \
- --pubsub_topic_tasks=task-scheduler-modified-tasks-staging \
- --pubsub_topic_jobs=task-scheduler-modified-jobs-staging \
+ --pubsub_topic_set=staging \
--pubsub_topic=swarming-tasks-staging \
--pubsub_subscriber=task-scheduler-staging \
--repo=https://skia.googlesource.com/skiabot-test.git \
diff --git a/task_scheduler/sys/task-scheduler.service b/task_scheduler/sys/task-scheduler.service
index 4ba663e..782014c 100644
--- a/task_scheduler/sys/task-scheduler.service
+++ b/task_scheduler/sys/task-scheduler.service
@@ -8,8 +8,7 @@
ExecStart=/usr/local/bin/task_scheduler \
--host=task-scheduler.skia.org \
--logtostderr \
- --pubsub_topic_tasks=task-scheduler-modified-tasks \
- --pubsub_topic_jobs=task-scheduler-modified-jobs \
+ --pubsub_topic_set=prod \
--workdir=/mnt/pd0/task_scheduler_workdir \
--resources_dir=/usr/local/share/task-scheduler/
Restart=always
