Add build scheduler DB interface, dumb in-memory impl, cache

build_scheduler/go/db/cache.go

+package db
+
+import (
+        "sync"
+        "time"
+)
+
+type BuildCache struct {
+        builds         map[string]*Build
+        buildsByCommit map[string]map[string]*Build
+        db             DB
+        lastUpdate     time.Time
+        mtx            sync.RWMutex
+        queryId        string
+}
+
+// GetBuildsForCommits retrieves all builds which first included each of the

[dogben, 2016/08/08 19:22:38]

There are a couple things that I don't understand about the documentation for
this method:
 - Shouldn't/Doesn't this return the most recent build rather than the first
build? Actually, I think BuildCache is agnostic as to time and revision history
-- it just finds builds where the blamelist includes the commits passed in.
 - If we run two builds for the same builder at the same commit, is the client
of BuildCache required to set Commits to the empty list for one of those builds?
If not, what does this method return in that case?

[borenet, 2016/08/08 19:51:17]

Historically, newer builds "steal" from the blamelist of previous ones.  In
other words, if a newer build runs at the same commit as a previous build, the
previous build's blamelist becomes empty, and the new build gets all the
commits. That's currently handled by datahopper as it ingests builds, but in the
new world it'll be handled by the build scheduler.

[dogben, 2016/08/08 20:16:07]

Is it worthwhile to mention here that we assume the blamelists are
non-overlapping?

[borenet, 2016/08/09 11:22:55]

Wrote some more doc, hopefully it's clearer now.

[dogben, 2016/08/09 13:56:06]

Awesome! Thanks!

+// given commits. Returns a map whose keys are commit hashes and values are
+// sub-maps whose keys are builder names and values are builds.
+func (c *BuildCache) GetBuildsForCommits(commits []string) (map[string]map[string]*Build, error) {
+        c.mtx.RLock()
+        defer c.mtx.RUnlock()
+
+        rv := make(map[string]map[string]*Build, len(commits))
+        for _, commit := range commits {
+                if builds, ok := c.buildsByCommit[commit]; ok {
+                        rv[commit] = make(map[string]*Build, len(builds))
+                        for k, v := range builds {
+                                rv[commit][k] = v.Copy()

[dogben, 2016/08/08 19:22:38]

Method doc says the sub-map key is the builder name (which I think makes sense),
but in update() below (and in tests), the sub-map key is the build ID.

[borenet, 2016/08/08 19:51:18]

Fixed.

+                        }
+                } else {
+                        rv[commit] = map[string]*Build{}
+                }
+        }
+        return rv, nil
+}
+
+// update inserts the new/updated builds into the cache. Assumes the caller
+// holds a lock.
+func (c *BuildCache) update(builds []*Build) error {
+        for _, b := range builds {
+                // If we already know about this build, the blamelist might,
+                // have changed, so we need to remove it from buildsByCommit
+                // and re-insert where needed.
+                if old, ok := c.builds[b.Id]; ok {

[dogben, 2016/08/08 19:22:38]

We should think about whether we want to use BuildBucket's ID or some other ID.

If we have a way to generate IDs for "fake" builds that won't conflict with
BuildBucket IDs, then it's not really a concern.

[borenet, 2016/08/08 19:51:17]

Agreed. I'm just using Buildbucket's IDs for now for simplicity.

+                        for _, commit := range old.Commits {
+                                delete(c.buildsByCommit[commit], b.Id)
+                        }
+                }
+
+                // Insert the new build into the main map.
+                c.builds[b.Id] = b.Copy()
+
+                // Insert the build into buildsByCommits.
+                for _, commit := range b.Commits {
+                        if _, ok := c.buildsByCommit[commit]; !ok {
+                                c.buildsByCommit[commit] = map[string]*Build{}
+                        }
+                        c.buildsByCommit[commit][b.Id] = b.Copy()

[dogben, 2016/08/08 19:22:38]

We should make a single copy and store it in both maps.

[borenet, 2016/08/08 19:51:18]

Done.

+                }
+        }
+        return nil
+}
+
+// Load new builds from the database.
+func (c *BuildCache) Update() error {
+        newBuilds, err := c.db.GetModifiedBuilds(c.queryId)

[dogben, 2016/08/08 19:22:38]

nit:

now := time.Now()
newBuilds, err := c.db.GetModifiedBuilds(c.queryId)
...
if err := c.update(newBuilds); err != nil {
  c.lastUpdate = now
}

[borenet, 2016/08/08 19:51:18]

Good catch, done.

+        c.mtx.Lock()
+        defer c.mtx.Unlock()
+        if err != nil {
+                if err.Error() == ErrUnknownId.Error() {
+                        // The database may have restarted. Attempt to re-establish connection.
+                        queryId, err := c.db.StartTrackingModifiedBuilds()
+                        if err != nil {
+                                return err
+                        }
+                        c.queryId = queryId
+                        // We may have missed something. Query for builds since the last
+                        // successful query.
+                        builds, err := c.db.GetBuildsFromDateRange(c.lastUpdate, time.Now())
+                        if err != nil {
+                                return err
+                        }
+                        return c.update(builds)
+                } else {
+                        return err
+                }
+        }
+        return c.update(newBuilds)
+}
+
+// NewBuildCache returns a local cache which provides more convenient views of
+// build data than the database can provide.
+func NewBuildCache(db DB, timePeriod time.Duration) (*BuildCache, error) {
+        queryId, err := db.StartTrackingModifiedBuilds()
+        if err != nil {
+                return nil, err
+        }
+        now := time.Now()
+        start := now.Add(-timePeriod)
+        builds, err := db.GetBuildsFromDateRange(start, now)
+        if err != nil {
+                return nil, err
+        }
+        bc := &BuildCache{
+                builds:         map[string]*Build{},
+                buildsByCommit: map[string]map[string]*Build{},
+                db:             db,
+                lastUpdate:     now,
+                queryId:        queryId,
+        }
+        bc.update(builds)

[dogben, 2016/08/08 19:22:38]

nit: check error (or change update to not return error)

[borenet, 2016/08/08 19:51:17]

Done.

+        return bc, nil
+}