diff --git a/datahopper/go/datahopper/jobs.go b/datahopper/go/datahopper/jobs.go
index 61c50a8..065618d 100644
--- a/datahopper/go/datahopper/jobs.go
+++ b/datahopper/go/datahopper/jobs.go
@@ -29,11 +29,11 @@
 )
 
 const (
-	STREAM = "job_metrics"
+	JOB_STREAM = "job_metrics"
 )
 
-// jobTypeString is an enum of the types of Jobs that computeAvgDuration will aggregate separately.
-// The string value is the same as the job_type tag value returned by computeAvgDuration.
+// jobTypeString is an enum of the types of Jobs that computeAvgJobDuration will aggregate separately.
+// The string value is the same as the job_type tag value returned by computeAvgJobDuration.
 type jobTypeString string
 
 const (
@@ -54,7 +54,7 @@
 	// 3. Thread 1: EventManager.updateMetrics calls jobEventDB.Range, which waits
 	//    to lock jobEventDB.mtx
 	// 4. Thread 2: jobEventDB.update calls EventMetrics.AggregateMetric (by way
-	//    of addAggregates and EventStream.AggregateMetric), which waits to lock
+	//    of addJobAggregates and EventStream.AggregateMetric), which waits to lock
 	//    EventMetrics.mtx
 	mtx sync.Mutex
 }
@@ -118,7 +118,7 @@
 			return fmt.Errorf("Failed to encode %#v to GOB: %s", job, err)
 		}
 		ev := &events.Event{
-			Stream:    STREAM,
+			Stream:    JOB_STREAM,
 			Timestamp: job.Created,
 			Data:      buf.Bytes(),
 		}
@@ -130,12 +130,12 @@
 	return nil
 }
 
-// computeAvgDuration is an events.DynamicAggregateFn that returns metrics for average Job duration
+// computeAvgJobDuration is an events.DynamicAggregateFn that returns metrics for average Job duration
 // for Jobs with status SUCCESS or FAILURE, given a slice of Events created by jobEventDB.update.
 // The first return value will contain the tags "job_name" (db.Job.Name) and "job_type" (one of
 // "normal", "tryjob", "forced"), and the second return value will be the corresponding average Job
 // duration. Returns an error if Event.Data can't be GOB-decoded as a db.Job.
-func computeAvgDuration(ev []*events.Event) ([]map[string]string, []float64, error) {
+func computeAvgJobDuration(ev []*events.Event) ([]map[string]string, []float64, error) {
 	if len(ev) > 0 {
 		// ev should be ordered by timestamp
 		sklog.Debugf("Calculating avg-duration for %d jobs since %s.", len(ev), ev[0].Timestamp)
@@ -196,12 +196,12 @@
 	return rvTags, rvVals, nil
 }
 
-// computeAvgDuration is an events.DynamicAggregateFn that returns metrics for Job failure rate and
+// computeJobFailureMishapRate is an events.DynamicAggregateFn that returns metrics for Job failure rate and
 // mishap rate, given a slice of Events created by jobEventDB.update. The first return value will
 // contain the tags "job_name" (db.Job.Name) and "metric" (one of "failure-rate", "mishap-rate"),
 // and the second return value will be the corresponding ratio of failed/mishap Jobs to all
 // completed Jobs. Returns an error if Event.Data can't be GOB-decoded as a db.Job.
-func computeFailureMishapRate(ev []*events.Event) ([]map[string]string, []float64, error) {
+func computeJobFailureMishapRate(ev []*events.Event) ([]map[string]string, []float64, error) {
 	if len(ev) > 0 {
 		// ev should be ordered by timestamp
 		sklog.Debugf("Calculating failure-rate for %d jobs since %s.", len(ev), ev[0].Timestamp)
@@ -250,15 +250,15 @@
 	return rvTags, rvVals, nil
 }
 
-// addAggregates adds aggregation functions for job events to the EventStream.
-func addAggregates(s *events.EventStream) error {
+// addJobAggregates adds aggregation functions for job events to the EventStream.
+func addJobAggregates(s *events.EventStream) error {
 	for _, period := range TIME_PERIODS {
-		if err := s.DynamicMetric(map[string]string{"metric": "avg-duration"}, period, computeAvgDuration); err != nil {
+		if err := s.DynamicMetric(map[string]string{"metric": "avg-duration"}, period, computeAvgJobDuration); err != nil {
 			return err
 		}
 
 		// Job failure/mishap rate.
-		if err := s.DynamicMetric(nil, period, computeFailureMishapRate); err != nil {
+		if err := s.DynamicMetric(nil, period, computeJobFailureMishapRate); err != nil {
 			return err
 		}
 	}
@@ -281,8 +281,8 @@
 	}
 	edb.em = em
 
-	s := em.GetEventStream(STREAM)
-	if err := addAggregates(s); err != nil {
+	s := em.GetEventStream(JOB_STREAM)
+	if err := addJobAggregates(s); err != nil {
 		return err
 	}
 
diff --git a/datahopper/go/datahopper/jobs_test.go b/datahopper/go/datahopper/jobs_test.go
index 7c4818a..be342f4 100644
--- a/datahopper/go/datahopper/jobs_test.go
+++ b/datahopper/go/datahopper/jobs_test.go
@@ -15,7 +15,7 @@
 )
 
 // Create a db.JobDB and jobEventDB.
-func setup(t *testing.T, now time.Time) (*jobEventDB, db.JobDB) {
+func setupJobs(t *testing.T, now time.Time) (*jobEventDB, db.JobDB) {
 	jdb := db.NewInMemoryJobDB()
 	edb := &jobEventDB{
 		cached: []*events.Event{},
@@ -45,20 +45,20 @@
 	return j
 }
 
-// assertEvent checks that ev.Data contains j.
-func assertEvent(t *testing.T, ev *events.Event, j *db.Job) {
-	assert.Equal(t, STREAM, ev.Stream)
+// assertJobEvent checks that ev.Data contains j.
+func assertJobEvent(t *testing.T, ev *events.Event, j *db.Job) {
+	assert.Equal(t, JOB_STREAM, ev.Stream)
 	var job db.Job
 	assert.NoError(t, gob.NewDecoder(bytes.NewReader(ev.Data)).Decode(&job))
 	deepequal.AssertDeepEqual(t, j, &job)
 	assert.True(t, j.Created.Equal(ev.Timestamp))
 }
 
-// TestUpdate checks that jobEventDB.update creates the correct Events from Jobs in the DB.
-func TestUpdate(t *testing.T) {
+// TestJobUpdate checks that jobEventDB.update creates the correct Events from Jobs in the DB.
+func TestJobUpdate(t *testing.T) {
 	testutils.SmallTest(t)
 	now := time.Now()
-	edb, jdb := setup(t, now)
+	edb, jdb := setupJobs(t, now)
 	start := now.Add(-TIME_PERIODS[len(TIME_PERIODS)-1])
 	jobs := []*db.Job{
 		// 0: Filtered out -- too early.
@@ -74,21 +74,21 @@
 	}
 	assert.NoError(t, jdb.PutJobs(jobs))
 	assert.NoError(t, edb.update())
-	evs, err := edb.Range(STREAM, start.Add(-time.Hour), start.Add(time.Hour))
+	evs, err := edb.Range(JOB_STREAM, start.Add(-time.Hour), start.Add(time.Hour))
 	assert.NoError(t, err)
 
 	expected := append(jobs[1:3], jobs[4:8]...)
 	assert.Len(t, evs, len(expected))
 	for i, ev := range evs {
-		assertEvent(t, ev, expected[i])
+		assertJobEvent(t, ev, expected[i])
 	}
 }
 
-// TestRange checks that jobEventDB.Range returns Events within the given range.
-func TestRange(t *testing.T) {
+// TestJobRange checks that jobEventDB.Range returns Events within the given range.
+func TestJobRange(t *testing.T) {
 	testutils.SmallTest(t)
 	now := time.Now()
-	edb, jdb := setup(t, now)
+	edb, jdb := setupJobs(t, now)
 	base := now.Add(-time.Hour)
 	jobs := []*db.Job{
 		makeJob(base.Add(-time.Nanosecond), "A", db.JOB_STATUS_SUCCESS, NORMAL, time.Minute),
@@ -100,11 +100,11 @@
 	assert.NoError(t, edb.update())
 
 	test := func(start, end time.Time, startIdx, count int) {
-		evs, err := edb.Range(STREAM, start, end)
+		evs, err := edb.Range(JOB_STREAM, start, end)
 		assert.NoError(t, err)
 		assert.Len(t, evs, count)
 		for i, ev := range evs {
-			assertEvent(t, ev, jobs[startIdx+i])
+			assertJobEvent(t, ev, jobs[startIdx+i])
 		}
 	}
 	before := base.Add(-time.Hour)
@@ -177,13 +177,13 @@
 	}
 }
 
-func TestComputeAvgDuration(t *testing.T) {
+func TestComputeAvgJobDuration(t *testing.T) {
 	testutils.SmallTest(t)
 	now := time.Now()
-	edb, jdb := setup(t, now)
+	edb, jdb := setupJobs(t, now)
 	created := now.Add(-time.Hour)
 
-	tester := newDynamicAggregateFnTester(t, computeAvgDuration)
+	tester := newDynamicAggregateFnTester(t, computeAvgJobDuration)
 	expect := func(jobName string, jobType jobTypeString, jobs []*db.Job) {
 		var totalDur float64 = 0
 		for _, j := range jobs {
@@ -230,20 +230,20 @@
 	expect("AllTypes", FORCED, jobsType[8:9])
 
 	assert.NoError(t, edb.update())
-	evs, err := edb.Range(STREAM, created.Add(-time.Hour), created.Add(time.Hour))
+	evs, err := edb.Range(JOB_STREAM, created.Add(-time.Hour), created.Add(time.Hour))
 	assert.NoError(t, err)
 	assert.Len(t, evs, len(jobsStatus)+len(jobsType))
 
 	tester.Run(evs)
 }
 
-func TestComputeFailureMishapRate(t *testing.T) {
+func TestComputeJobFailureMishapRate(t *testing.T) {
 	testutils.SmallTest(t)
 	now := time.Now()
-	edb, jdb := setup(t, now)
+	edb, jdb := setupJobs(t, now)
 	created := now.Add(-time.Hour)
 
-	tester := newDynamicAggregateFnTester(t, computeFailureMishapRate)
+	tester := newDynamicAggregateFnTester(t, computeJobFailureMishapRate)
 	expect := func(jobName string, metric string, numer, denom int) {
 		tester.AddAssert(map[string]string{
 			"job_name": jobName,
@@ -322,7 +322,7 @@
 	}
 
 	assert.NoError(t, edb.update())
-	evs, err := edb.Range(STREAM, created.Add(-time.Hour), created.Add(time.Hour))
+	evs, err := edb.Range(JOB_STREAM, created.Add(-time.Hour), created.Add(time.Hour))
 	assert.NoError(t, err)
 	assert.Len(t, evs, jobCount)
 
diff --git a/datahopper/go/datahopper/tasks.go b/datahopper/go/datahopper/tasks.go
new file mode 100644
index 0000000..65e79cb
--- /dev/null
+++ b/datahopper/go/datahopper/tasks.go
@@ -0,0 +1,207 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/gob"
+	"fmt"
+	"sort"
+	"sync"
+	"time"
+
+	"go.skia.org/infra/go/httputils"
+	"go.skia.org/infra/go/metrics2"
+	"go.skia.org/infra/go/metrics2/events"
+	"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/remote_db"
+	"go.skia.org/infra/task_scheduler/go/flakes"
+)
+
+const (
+	TASK_STREAM = "task_metrics"
+)
+
+// taskEventDB implements the events.EventDB interface.
+type taskEventDB struct {
+	cached []*events.Event
+	db     db.TaskReader
+	em     *events.EventMetrics
+	// Do not lock mtx when calling methods on em. Otherwise deadlock can occur.
+	// E.g. (now fixed):
+	// 1. Thread 1: EventManager.updateMetrics locks EventMetrics.mtx
+	// 2. Thread 2: taskEventDB.update locks taskEventDB.mtx
+	// 3. Thread 1: EventManager.updateMetrics calls taskEventDB.Range, which waits
+	//    to lock taskEventDB.mtx
+	// 4. Thread 2: taskEventDB.update calls EventMetrics.AggregateMetric (by way
+	//    of addTaskAggregates and EventStream.AggregateMetric), which waits to lock
+	//    EventMetrics.mtx
+	mtx sync.Mutex
+}
+
+// See docs for events.EventDB interface.
+func (t *taskEventDB) Append(string, []byte) error {
+	return fmt.Errorf("taskEventDB is read-only!")
+}
+
+// See docs for events.EventDB interface.
+func (t *taskEventDB) Close() error {
+	return nil
+}
+
+// See docs for events.EventDB interface.
+func (t *taskEventDB) Insert(*events.Event) error {
+	return fmt.Errorf("taskEventDB is read-only!")
+}
+
+// See docs for events.EventDB interface.
+func (t *taskEventDB) Range(stream string, start, end time.Time) ([]*events.Event, error) {
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+
+	n := len(t.cached)
+	if n == 0 {
+		return []*events.Event{}, nil
+	}
+
+	first := sort.Search(n, func(i int) bool {
+		return !t.cached[i].Timestamp.Before(start)
+	})
+
+	last := first + sort.Search(n-first, func(i int) bool {
+		return !t.cached[i+first].Timestamp.Before(end)
+	})
+
+	rv := make([]*events.Event, last-first, last-first)
+	copy(rv[:], t.cached[first:last])
+	return rv, nil
+}
+
+// update updates the cached tasks in the taskEventDB. Only a single thread may
+// call this method, but it can be called concurrently with other methods.
+func (t *taskEventDB) update() error {
+	defer metrics2.FuncTimer().Stop()
+	now := time.Now()
+	longestPeriod := TIME_PERIODS[len(TIME_PERIODS)-1]
+	tasks, err := t.db.GetTasksFromDateRange(now.Add(-longestPeriod), now, "")
+	if err != nil {
+		return fmt.Errorf("Failed to load tasks from %s to %s: %s", now.Add(-longestPeriod), now, err)
+	}
+	sklog.Debugf("taskEventDB.update: Processing %d tasks for time range %s to %s.", len(tasks), now.Add(-longestPeriod), now)
+	cached := make([]*events.Event, 0, len(tasks))
+	for _, task := range tasks {
+		if !task.Done() {
+			continue
+		}
+		var buf bytes.Buffer
+		if err := gob.NewEncoder(&buf).Encode(task); err != nil {
+			return fmt.Errorf("Failed to encode %#v to GOB: %s", task, err)
+		}
+		ev := &events.Event{
+			Stream:    TASK_STREAM,
+			Timestamp: task.Created,
+			Data:      buf.Bytes(),
+		}
+		cached = append(cached, ev)
+	}
+	t.mtx.Lock()
+	defer t.mtx.Unlock()
+	t.cached = cached
+	return nil
+}
+
+// computeTaskFlakeRate is an events.DynamicAggregateFn that returns metrics for Task flake rate, given
+// a slice of Events created by taskEventDB.update. The first return value will contain the tags
+// "task_name" (db.Task.Name) and "metric" (one of "failure-rate", "mishap-rate"),
+// and the second return value will be the corresponding ratio of failed/mishap Task to all
+// completed Tasks. Returns an error if Event.Data can't be GOB-decoded as a db.Task.
+func computeTaskFlakeRate(ev []*events.Event) ([]map[string]string, []float64, error) {
+	if len(ev) > 0 {
+		// ev should be ordered by timestamp
+		sklog.Debugf("Calculating flake-rate for %d tasks since %s.", len(ev), ev[0].Timestamp)
+	}
+	type taskSum struct {
+		flakes int
+		count  int
+	}
+	byTask := map[string]*taskSum{}
+	tasks := make([]*db.Task, 0, len(ev))
+	for _, e := range ev {
+		var task db.Task
+		if err := gob.NewDecoder(bytes.NewReader(e.Data)).Decode(&task); err != nil {
+			return nil, nil, err
+		}
+		tasks = append(tasks, &task)
+		entry, ok := byTask[task.Name]
+		if !ok {
+			entry = &taskSum{}
+			byTask[task.Name] = entry
+		}
+		entry.count++
+	}
+	flaky := flakes.FindFlakes(tasks)
+	for _, task := range flaky {
+		byTask[task.Name].flakes++
+	}
+	rvTags := make([]map[string]string, 0, len(byTask)*2)
+	rvVals := make([]float64, 0, len(byTask)*2)
+	add := func(taskName, metric string, value float64) {
+		rvTags = append(rvTags, map[string]string{
+			"task_name": taskName,
+			"metric":    metric,
+		})
+		rvVals = append(rvVals, value)
+	}
+	for taskName, taskSum := range byTask {
+		if taskSum.count == 0 {
+			continue
+		}
+		add(taskName, "flake-rate", float64(taskSum.flakes)/float64(taskSum.count))
+	}
+	return rvTags, rvVals, nil
+}
+
+// addTaskAggregates adds aggregation functions for job events to the EventStream.
+func addTaskAggregates(s *events.EventStream) error {
+	for _, period := range TIME_PERIODS {
+		// Flake rate.
+		if err := s.DynamicMetric(nil, period, computeTaskFlakeRate); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// StartTaskMetrics starts a goroutine which ingests metrics data based on Tasks.
+func StartTaskMetrics(taskSchedulerDbUrl string, ctx context.Context) error {
+	db, err := remote_db.NewClient(taskSchedulerDbUrl, httputils.NewTimeoutClient())
+	if err != nil {
+		return err
+	}
+	edb := &taskEventDB{
+		cached: []*events.Event{},
+		db:     db,
+	}
+	em, err := events.NewEventMetrics(edb, "task_metrics")
+	if err != nil {
+		return err
+	}
+	edb.em = em
+
+	s := em.GetEventStream(TASK_STREAM)
+	if err := addTaskAggregates(s); err != nil {
+		return err
+	}
+
+	lv := metrics2.NewLiveness("last_successful_task_metrics_update")
+	go util.RepeatCtx(5*time.Minute, ctx, func() {
+		if err := edb.update(); err != nil {
+			sklog.Errorf("Failed to update task data: %s", err)
+		} else {
+			lv.Reset()
+		}
+	})
+	em.Start(ctx)
+	return nil
+}
diff --git a/datahopper/go/datahopper/tasks_test.go b/datahopper/go/datahopper/tasks_test.go
new file mode 100644
index 0000000..cabd391
--- /dev/null
+++ b/datahopper/go/datahopper/tasks_test.go
@@ -0,0 +1,207 @@
+package main
+
+import (
+	"bytes"
+	"encoding/gob"
+	"testing"
+	"time"
+
+	assert "github.com/stretchr/testify/require"
+	"go.skia.org/infra/go/deepequal"
+	"go.skia.org/infra/go/metrics2/events"
+	"go.skia.org/infra/go/testutils"
+	"go.skia.org/infra/task_scheduler/go/db"
+)
+
+// Create a db.TaskDB and taskEventDB.
+func setupTasks(t *testing.T, now time.Time) (*taskEventDB, db.TaskDB) {
+	tdb := db.NewInMemoryTaskDB()
+	edb := &taskEventDB{
+		cached: []*events.Event{},
+		db:     tdb,
+	}
+	return edb, tdb
+}
+
+// makeTask returns a fake task with only the fields relevant to this test set.
+func makeTask(created time.Time, name string, status db.TaskStatus) *db.Task {
+	task := &db.Task{
+		Created: created,
+		TaskKey: db.TaskKey{
+			Name: name,
+		},
+		Status: status,
+	}
+	return task
+}
+
+// assertTaskEvent checks that ev.Data contains task.
+func assertTaskEvent(t *testing.T, ev *events.Event, task *db.Task) {
+	assert.Equal(t, TASK_STREAM, ev.Stream)
+	var other db.Task
+	assert.NoError(t, gob.NewDecoder(bytes.NewReader(ev.Data)).Decode(&other))
+	deepequal.AssertDeepEqual(t, task, &other)
+	assert.True(t, task.Created.Equal(ev.Timestamp))
+}
+
+// TestTaskUpdate checks that taskEventDB.update creates the correct Events from Tasks in the DB.
+func TestTaskUpdate(t *testing.T) {
+	testutils.SmallTest(t)
+	now := time.Now()
+	edb, tdb := setupTasks(t, now)
+	start := now.Add(-TIME_PERIODS[len(TIME_PERIODS)-1])
+	tasks := []*db.Task{
+		// 0: Filtered out -- too early.
+		makeTask(start.Add(-time.Minute), "A", db.TASK_STATUS_SUCCESS),
+		makeTask(start.Add(time.Minute), "A", db.TASK_STATUS_SUCCESS),
+		makeTask(start.Add(2*time.Minute), "A", db.TASK_STATUS_FAILURE),
+		// 3: Filtered out -- not Done.
+		makeTask(start.Add(3*time.Minute), "A", db.TASK_STATUS_RUNNING),
+		makeTask(start.Add(4*time.Minute), "A", db.TASK_STATUS_MISHAP),
+		makeTask(start.Add(5*time.Minute), "A", db.TASK_STATUS_FAILURE),
+		makeTask(start.Add(6*time.Minute), "B", db.TASK_STATUS_SUCCESS),
+		makeTask(start.Add(7*time.Minute), "A", db.TASK_STATUS_SUCCESS),
+	}
+	assert.NoError(t, tdb.PutTasks(tasks))
+	assert.NoError(t, edb.update())
+	evs, err := edb.Range(TASK_STREAM, start.Add(-time.Hour), start.Add(time.Hour))
+	assert.NoError(t, err)
+
+	expected := append(tasks[1:3], tasks[4:8]...)
+	assert.Len(t, evs, len(expected))
+	for i, ev := range evs {
+		assertTaskEvent(t, ev, expected[i])
+	}
+}
+
+// TestTaskRange checks that taskEventDB.Range returns Events within the given range.
+func TestTaskRange(t *testing.T) {
+	testutils.SmallTest(t)
+	now := time.Now()
+	edb, tdb := setupTasks(t, now)
+	base := now.Add(-time.Hour)
+	tasks := []*db.Task{
+		makeTask(base.Add(-time.Nanosecond), "A", db.TASK_STATUS_SUCCESS),
+		makeTask(base, "A", db.TASK_STATUS_SUCCESS),
+		makeTask(base.Add(time.Nanosecond), "A", db.TASK_STATUS_SUCCESS),
+		makeTask(base.Add(time.Minute), "A", db.TASK_STATUS_SUCCESS),
+	}
+	assert.NoError(t, tdb.PutTasks(tasks))
+	assert.NoError(t, edb.update())
+
+	test := func(start, end time.Time, startIdx, count int) {
+		evs, err := edb.Range(TASK_STREAM, start, end)
+		assert.NoError(t, err)
+		assert.Len(t, evs, count)
+		for i, ev := range evs {
+			assertTaskEvent(t, ev, tasks[startIdx+i])
+		}
+	}
+	before := base.Add(-time.Hour)
+	after := base.Add(time.Hour)
+	test(before, before, -1, 0)
+	test(before, tasks[0].Created, -1, 0)
+	test(before, tasks[1].Created, 0, 1)
+	test(before, tasks[2].Created, 0, 2)
+	test(before, tasks[3].Created, 0, 3)
+	test(before, after, 0, 4)
+	test(tasks[0].Created, before, -1, 0)
+	test(tasks[0].Created, tasks[0].Created, -1, 0)
+	test(tasks[0].Created, tasks[1].Created, 0, 1)
+	test(tasks[0].Created, tasks[2].Created, 0, 2)
+	test(tasks[0].Created, tasks[3].Created, 0, 3)
+	test(tasks[0].Created, after, 0, 4)
+	test(tasks[1].Created, tasks[0].Created, -1, 0)
+	test(tasks[1].Created, tasks[1].Created, -1, 0)
+	test(tasks[1].Created, tasks[2].Created, 1, 1)
+	test(tasks[1].Created, tasks[3].Created, 1, 2)
+	test(tasks[1].Created, after, 1, 3)
+	test(tasks[2].Created, tasks[2].Created, -1, 0)
+	test(tasks[2].Created, tasks[3].Created, 2, 1)
+	test(tasks[2].Created, after, 2, 2)
+	test(tasks[3].Created, tasks[3].Created, -1, 0)
+	test(tasks[3].Created, after, 3, 1)
+	test(after, after, -1, 0)
+}
+
+func TestComputeTaskFlakeRate(t *testing.T) {
+	testutils.SmallTest(t)
+	now := time.Now()
+	edb, tdb := setupTasks(t, now)
+	created := now.Add(-time.Hour)
+
+	tester := newDynamicAggregateFnTester(t, computeTaskFlakeRate)
+	expect := func(taskName string, metric string, numer, denom int) {
+		tester.AddAssert(map[string]string{
+			"task_name": taskName,
+			"metric":    metric,
+		}, float64(numer)/float64(denom))
+	}
+
+	taskCount := 0
+	addTask := func(name, commit string, status db.TaskStatus) {
+		taskCount++
+		task := makeTask(created, name, status)
+		task.Revision = commit
+		assert.NoError(t, tdb.PutTask(task))
+	}
+	{
+		name := "NoFlakes"
+		addTask(name, "a", db.TASK_STATUS_SUCCESS)
+		addTask(name, "b", db.TASK_STATUS_SUCCESS)
+		addTask(name, "c", db.TASK_STATUS_SUCCESS)
+		addTask(name, "d", db.TASK_STATUS_FAILURE)
+		addTask(name, "d", db.TASK_STATUS_FAILURE)
+		expect(name, "flake-rate", 0, 5)
+	}
+	{
+		name := "Mishaps"
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "b", db.TASK_STATUS_FAILURE)
+		addTask(name, "c", db.TASK_STATUS_FAILURE)
+		addTask(name, "c", db.TASK_STATUS_MISHAP)
+		expect(name, "flake-rate", 1, 4)
+	}
+	{
+		name := "RetrySucceeded"
+		addTask(name, "a", db.TASK_STATUS_SUCCESS)
+		addTask(name, "b", db.TASK_STATUS_FAILURE)
+		addTask(name, "b", db.TASK_STATUS_SUCCESS)
+		expect(name, "flake-rate", 1, 3)
+	}
+	{
+		name := "RetryFailed"
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		expect(name, "flake-rate", 0, 2)
+	}
+	{
+		name := "Mix"
+		addTask(name, "a", db.TASK_STATUS_SUCCESS)
+		addTask(name, "b", db.TASK_STATUS_FAILURE)
+		addTask(name, "c", db.TASK_STATUS_FAILURE)
+		addTask(name, "b", db.TASK_STATUS_FAILURE)
+		addTask(name, "c", db.TASK_STATUS_SUCCESS)
+		addTask(name, "d", db.TASK_STATUS_MISHAP)
+		addTask(name, "d", db.TASK_STATUS_SUCCESS)
+		expect(name, "flake-rate", 2, 7)
+	}
+	{
+		name := "LongRetryChain"
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "a", db.TASK_STATUS_FAILURE)
+		addTask(name, "a", db.TASK_STATUS_SUCCESS)
+		expect(name, "flake-rate", 6, 7)
+	}
+
+	assert.NoError(t, edb.update())
+	evs, err := edb.Range(TASK_STREAM, created.Add(-time.Hour), created.Add(time.Hour))
+	assert.NoError(t, err)
+	assert.Len(t, evs, taskCount)
+
+	tester.Run(evs)
+}
diff --git a/task_scheduler/go/flakes/flakes.go b/task_scheduler/go/flakes/flakes.go
new file mode 100644
index 0000000..f447d61
--- /dev/null
+++ b/task_scheduler/go/flakes/flakes.go
@@ -0,0 +1,44 @@
+package flakes
+
+/*
+   Find flakily-failed tasks in a time window.
+*/
+
+import (
+	"go.skia.org/infra/task_scheduler/go/db"
+)
+
+// Find flakily-failed tasks in the given slice of tasks.
+func FindFlakes(tasks []*db.Task) []*db.Task {
+	tasksMap := map[db.TaskKey][]*db.Task{}
+	for _, task := range tasks {
+		if task.Done() {
+			tasksMap[task.TaskKey] = append(tasksMap[task.TaskKey], task)
+		}
+	}
+	flaky := []*db.Task{}
+	for _, tasks := range tasksMap {
+		// If one or more tasks succeeded and failed, then all failures
+		// are flakes.
+		success := 0
+		failure := 0
+		for _, task := range tasks {
+			if task.Status == db.TASK_STATUS_SUCCESS {
+				success++
+			} else if task.Status == db.TASK_STATUS_FAILURE {
+				failure++
+			} else if task.Status == db.TASK_STATUS_MISHAP {
+				// Mishaps are flakes by definition.
+				flaky = append(flaky, task)
+			}
+		}
+		if success > 0 && failure > 0 {
+			for _, task := range tasks {
+				if task.Status == db.TASK_STATUS_FAILURE {
+					flaky = append(flaky, task)
+				}
+			}
+		}
+	}
+	return flaky
+}