Merge branch 'master' into np1

* master: (22 commits)
  Enable travis build validation (google#1400)
  Use ProtoEqual for gomock reflection equality (google#1401)
  Fixups for deleted test constructor in trillian repo (google#1398)
  Fix the kubernetes configs (google#1397)
  Upgrade dependencies (google#1387)
  Update install instructions for go 1.13 (google#1388)
  Emit metric with the right dimensions (google#1383)
  Define watermarks as micros (google#1384)
  Library for converting time.Time to sequencer watermarks (google#1381)
  Reduce log spam (google#1382)
  Support and test mutation log queries at intermediate timestamps (google#1380)
  In memory logs implementation (google#1375)
  Fix generic comparisons on protobuf messages (google#1379)
  Do pagination the right way (google#1378)
  Move the responsibility to pick an input log from storage to the frontend (google#1376)
  Init metrics for whole test file (google#1373)
  Break layering violation by using native types (google#1374)
  Switch Timestamp storage to mysql DATETIME (google#1369)
  Use testdb in integration tests (google#1371)
  Use and test the same MySQL connection that main.go uses  (google#1370)
  ...

.travis.yml

@@ -1,6 +1,7 @@
+version: ~> 1.0
 language: go
 go:
-- 1.x
+- "1.13.x"
 dist: bionic
 
 go_import_path: github.com/google/keytransparency

README.md

@@ -32,8 +32,8 @@ development.
 ## Key Transparency Client
 
 ### Setup
-1. Install [Go 1.10](https://golang.org/doc/install).
-2. `go get -u github.com/google/keytransparency/cmd/keytransparency-client `
+1. Install [Go 1.13](https://golang.org/doc/install).
+2. `go get github.com/google/keytransparency/cmd/keytransparency-client `
 
 ### Client operations
 
@@ -96,8 +96,8 @@ NB A default for the Key Transparency server URL is being used here. The default
    - Docker Compose 1.11.0+ `docker-compose --version`
 
 ```sh
-go get -u github.com/google/keytransparency/...
-go get -u github.com/google/trillian/...
+go get github.com/google/keytransparency/...
+go get github.com/google/trillian/...
 cd $(go env GOPATH)/src/github.com/google/keytransparency
 ./scripts/prepare_server.sh -f
 docker-compose -f docker-compose.yml docker-compose.prod.yml up

cmd/keytransparency-monitor/main.go

@@ -16,13 +16,14 @@ package main
 
 import (
 	"context"
+	"crypto"
 	"crypto/tls"
 	"flag"
 	"net"
 	"net/http"
 
 	"github.com/golang/glog"
-	"github.com/google/trillian/crypto"
+	tcrypto "github.com/google/trillian/crypto"
 	"github.com/google/trillian/crypto/keys/pem"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
@@ -75,7 +76,7 @@ func main() {
 	if err != nil {
 		glog.Exitf("Could not create signer from %v: %v", *signingKey, err)
 	}
-	signer := crypto.NewSHA256Signer(key)
+	signer := tcrypto.NewSigner(0, key, crypto.SHA256)
 	store := fake.NewMonitorStorage()
 
 	// Create monitoring background process.

cmd/keytransparency-sequencer/main.go

@@ -16,7 +16,6 @@ package main
 
 import (
 	"context"
-	"database/sql"
 	"flag"
 	"fmt"
 	"net"
@@ -44,10 +43,10 @@ import (
 	pb "github.com/google/keytransparency/core/api/v1/keytransparency_go_proto"
 	dir "github.com/google/keytransparency/core/directory"
 	spb "github.com/google/keytransparency/core/sequencer/sequencer_go_proto"
+	ktsql "github.com/google/keytransparency/impl/sql"
 	etcdelect "github.com/google/trillian/util/election2/etcd"
 	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 
-	_ "github.com/go-sql-driver/mysql" // Set database engine.
 	_ "github.com/google/trillian/crypto/keys/der/proto"
 	_ "github.com/google/trillian/merkle/coniks"  // Register hasher
 	_ "github.com/google/trillian/merkle/rfc6962" // Register hasher
@@ -74,17 +73,6 @@ var (
 	batchSize  = flag.Int("batch-size", 100, "Maximum number of mutations to process per map revision")
 )
 
-func openDB() *sql.DB {
-	db, err := sql.Open("mysql", *serverDBPath)
-	if err != nil {
-		glog.Exitf("sql.Open(): %v", err)
-	}
-	if err := db.Ping(); err != nil {
-		glog.Exitf("db.Ping(): %v", err)
-	}
-	return db
-}
-
 // getElectionFactory returns an election factory based on flags, and a
 // function which releases the resources associated with the factory.
 func getElectionFactory() (election2.Factory, func()) {
@@ -130,7 +118,10 @@ func main() {
 	}
 
 	// Database tables
-	sqldb := openDB()
+	sqldb, err := ktsql.Open(*serverDBPath)
+	if err != nil {
+		glog.Exit(err)
+	}
 	defer sqldb.Close()
 
 	mutations, err := mutationstorage.New(sqldb)

cmd/keytransparency-server/main.go

@@ -16,7 +16,6 @@ package main
 
 import (
 	"context"
-	"database/sql"
 	"flag"
 	"net/http"
 
@@ -37,11 +36,11 @@ import (
 	"github.com/google/keytransparency/impl/sql/mutationstorage"
 
 	pb "github.com/google/keytransparency/core/api/v1/keytransparency_go_proto"
+	ktsql "github.com/google/keytransparency/impl/sql"
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
 	grpc_auth "github.com/grpc-ecosystem/go-grpc-middleware/auth"
 	grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
 
-	_ "github.com/go-sql-driver/mysql" // Set database engine.
 	_ "github.com/google/trillian/crypto/keys/der/proto"
 )
 
@@ -58,23 +57,15 @@ var (
 	revisionPageSize = flag.Int("revision-page-size", 10, "Max number of revisions to return at once")
 )
 
-func openDB() *sql.DB {
-	db, err := sql.Open("mysql", *serverDBPath)
-	if err != nil {
-		glog.Exitf("sql.Open(): %v", err)
-	}
-	if err := db.Ping(); err != nil {
-		glog.Exitf("db.Ping(): %v", err)
-	}
-	return db
-}
-
 func main() {
 	flag.Parse()
 	ctx := context.Background()
 
 	// Open Resources.
-	sqldb := openDB()
+	sqldb, err := ktsql.Open(*serverDBPath)
+	if err != nil {
+		glog.Exit(err)
+	}
 	defer sqldb.Close()
 
 	creds, err := credentials.NewServerTLSFromFile(*certFile, *keyFile)

core/integration/monitor_tests.go

@@ -16,6 +16,7 @@ package integration
 
 import (
 	"context"
+	"crypto"
 	"sync"
 	"testing"
 	"time"
@@ -28,7 +29,7 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 
-	"github.com/google/trillian/crypto"
+	tcrypto "github.com/google/trillian/crypto"
 	"github.com/google/trillian/crypto/keys/pem"
 
 	tpb "github.com/google/keytransparency/core/testdata/transcript_go_proto"
@@ -49,7 +50,7 @@ func TestMonitor(ctx context.Context, env *Env, t *testing.T) []*tpb.Action {
 	if err != nil {
 		t.Fatalf("Couldn't create signer: %v", err)
 	}
-	signer := crypto.NewSHA256Signer(privKey)
+	signer := tcrypto.NewSigner(0, privKey, crypto.SHA256)
 	store := fake.NewMonitorStorage()
 	mon, err := monitor.NewFromDirectory(env.Cli, env.Directory, signer, store)
 	if err != nil {

core/integration/storagetest/batch.go

@@ -27,15 +27,13 @@ import (
 	spb "github.com/google/keytransparency/core/sequencer/sequencer_go_proto"
 )
 
-// Batcher writes batch definitions to storage.
-type Batcher = sequencer.Batcher
+// batchStorageFactory returns a new database object, and a function for cleaning it up.
+type batchStorageFactory func(ctx context.Context, t *testing.T, dirID string) (sequencer.Batcher, func(context.Context))
 
-type BatchStorageFactory func(ctx context.Context, t *testing.T, dirID string) Batcher
-
-type BatchStorageTest func(ctx context.Context, t *testing.T, f BatchStorageFactory)
+type BatchStorageTest func(ctx context.Context, t *testing.T, f batchStorageFactory)
 
 // RunBatchStorageTests runs all the batch storage tests against the provided map storage implementation.
-func RunBatchStorageTests(t *testing.T, factory BatchStorageFactory) {
+func RunBatchStorageTests(t *testing.T, factory batchStorageFactory) {
 	ctx := context.Background()
 	b := &BatchTests{}
 	for name, f := range map[string]BatchStorageTest{
@@ -52,19 +50,21 @@ func RunBatchStorageTests(t *testing.T, factory BatchStorageFactory) {
 // BatchTests is a suite of tests to run against
 type BatchTests struct{}
 
-func (*BatchTests) TestNotFound(ctx context.Context, t *testing.T, f BatchStorageFactory) {
+func (*BatchTests) TestNotFound(ctx context.Context, t *testing.T, f batchStorageFactory) {
 	domainID := "testnotfounddir"
-	b := f(ctx, t, domainID)
+	b, done := f(ctx, t, domainID)
+	defer done(ctx)
 	_, err := b.ReadBatch(ctx, domainID, 0)
 	st := status.Convert(err)
 	if got, want := st.Code(), codes.NotFound; got != want {
 		t.Errorf("ReadBatch(): %v, want %v", err, want)
 	}
 }
 
-func (*BatchTests) TestWriteBatch(ctx context.Context, t *testing.T, f BatchStorageFactory) {
+func (*BatchTests) TestWriteBatch(ctx context.Context, t *testing.T, f batchStorageFactory) {
 	domainID := "writebatchtest"
-	b := f(ctx, t, domainID)
+	b, done := f(ctx, t, domainID)
+	defer done(ctx)
 	for _, tc := range []struct {
 		rev     int64
 		wantErr bool
@@ -86,9 +86,10 @@ func (*BatchTests) TestWriteBatch(ctx context.Context, t *testing.T, f BatchStor
 	}
 }
 
-func (*BatchTests) TestReadBatch(ctx context.Context, t *testing.T, f BatchStorageFactory) {
+func (*BatchTests) TestReadBatch(ctx context.Context, t *testing.T, f batchStorageFactory) {
 	domainID := "readbatchtest"
-	b := f(ctx, t, domainID)
+	b, done := f(ctx, t, domainID)
+	defer done(ctx)
 	for _, tc := range []struct {
 		rev  int64
 		want *spb.MapMetadata
@@ -115,9 +116,10 @@ func (*BatchTests) TestReadBatch(ctx context.Context, t *testing.T, f BatchStora
 	}
 }
 
-func (*BatchTests) TestHighestRev(ctx context.Context, t *testing.T, f BatchStorageFactory) {
+func (*BatchTests) TestHighestRev(ctx context.Context, t *testing.T, f batchStorageFactory) {
 	domainID := "writebatchtest"
-	b := f(ctx, t, domainID)
+	b, done := f(ctx, t, domainID)
+	defer done(ctx)
 	for _, tc := range []struct {
 		rev     int64
 		sources []*spb.MapMetadata_SourceSlice

core/integration/storagetest/mutation_logs.go

@@ -16,24 +16,28 @@ package storagetest
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
 	"github.com/golang/protobuf/proto"
+	"github.com/google/go-cmp/cmp"
 	"github.com/google/keytransparency/core/keyserver"
 
 	pb "github.com/google/keytransparency/core/api/v1/keytransparency_go_proto"
 )
 
-type MutationLogsFactory func(ctx context.Context, t *testing.T, dirID string, logIDs ...int64) keyserver.MutationLogs
+// mutationLogsFactory returns a new database object, and a function for cleaning it up.
+type mutationLogsFactory func(ctx context.Context, t *testing.T, dirID string, logIDs ...int64) (keyserver.MutationLogs, func(context.Context))
 
 // RunMutationLogsTests runs all the tests against the provided storage implementation.
-func RunMutationLogsTests(t *testing.T, factory MutationLogsFactory) {
+func RunMutationLogsTests(t *testing.T, factory mutationLogsFactory) {
 	ctx := context.Background()
 	b := &mutationLogsTests{}
-	for name, f := range map[string]func(ctx context.Context, t *testing.T, f MutationLogsFactory){
+	for name, f := range map[string]func(ctx context.Context, t *testing.T, f mutationLogsFactory){
 		// TODO(gbelvin): Discover test methods via reflection.
-		"TestReadLog": b.TestReadLog,
+		"TestReadLog":      b.TestReadLog,
+		"TestReadLogExact": b.TestReadLogExact,
 	} {
 		t.Run(name, func(t *testing.T) { f(ctx, t, factory) })
 	}
@@ -50,35 +54,87 @@ func mustMarshal(t *testing.T, p proto.Message) []byte {
 	return b
 }
 
+// https://dev.mysql.com/doc/refman/8.0/en/datetime.html
+var minWatermark = time.Date(1000, 1, 1, 0, 0, 0, 0, time.UTC)
+
 // TestReadLog ensures that reads happen in atomic units of batch size.
-func (mutationLogsTests) TestReadLog(ctx context.Context, t *testing.T, newForTest MutationLogsFactory) {
+func (mutationLogsTests) TestReadLog(ctx context.Context, t *testing.T, newForTest mutationLogsFactory) {
 	directoryID := "TestReadLog"
 	logID := int64(5) // Any log ID.
-	m := newForTest(ctx, t, directoryID, logID)
-	// Write ten batches, three entries each.
+	m, done := newForTest(ctx, t, directoryID, logID)
+	defer done(ctx)
+	// Write ten batches.
 	for i := byte(0); i < 10; i++ {
 		entry := &pb.EntryUpdate{Mutation: &pb.SignedEntry{Entry: mustMarshal(t, &pb.Entry{Index: []byte{i}})}}
-		if _, err := m.Send(ctx, directoryID, entry, entry, entry); err != nil {
+		if _, err := m.Send(ctx, directoryID, logID, entry, entry, entry); err != nil {
 			t.Fatalf("Send(): %v", err)
 		}
 	}
 
-	for _, tc := range []struct {
+	for i, tc := range []struct {
 		limit int32
-		count int
+		want  int
 	}{
-		{limit: 0, count: 0},
-		{limit: 1, count: 3},    // We asked for 1 item, which gets us into the first batch, so we return 3 items.
-		{limit: 3, count: 3},    // We asked for 3 items, which gets us through the first batch, so we return 3 items.
-		{limit: 4, count: 6},    // Reading 4 items gets us into the second batch of size 3.
-		{limit: 100, count: 30}, // Reading all the items gets us the 30 items we wrote.
+		{limit: 0, want: 0},
+		{limit: 1, want: 3},    // We asked for 1 item, which gets us into the first batch, so we return 3 items.
+		{limit: 3, want: 3},    // We asked for 3 items, which gets us through the first batch, so we return 3 items.
+		{limit: 4, want: 6},    // Reading 4 items gets us into the second batch of size 3.
+		{limit: 100, want: 30}, // Reading all the items gets us the 30 items we wrote.
 	} {
-		rows, err := m.ReadLog(ctx, directoryID, logID, 0, time.Now().UnixNano(), tc.limit)
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			rows, err := m.ReadLog(ctx, directoryID, logID, minWatermark, time.Now(), tc.limit)
+			if err != nil {
+				t.Fatalf("ReadLog(%v): %v", tc.limit, err)
+			}
+			if got := len(rows); got != tc.want {
+				t.Fatalf("ReadLog(%v): len: %v, want %v", tc.limit, got, tc.want)
+			}
+		})
+	}
+}
+
+// TestReadLogExact ensures that reads respect the low inclusive, high exclusive API.
+func (mutationLogsTests) TestReadLogExact(ctx context.Context, t *testing.T, newForTest mutationLogsFactory) {
+	directoryID := "TestReadLogExact"
+	logID := int64(5) // Any log ID.
+	m, done := newForTest(ctx, t, directoryID, logID)
+	defer done(ctx)
+	// Write ten batches.
+	idx := make([]time.Time, 0, 10)
+	for i := byte(0); i < 10; i++ {
+		entry := &pb.EntryUpdate{Mutation: &pb.SignedEntry{Entry: []byte{i}}}
+		ts, err := m.Send(ctx, directoryID, logID, entry)
 		if err != nil {
-			t.Fatalf("ReadLog(%v): %v", tc.limit, err)
-		}
-		if got, want := len(rows), tc.count; got != want {
-			t.Fatalf("ReadLog(%v): len: %v, want %v", tc.limit, got, want)
+			t.Fatalf("Send(): %v", err)
 		}
+		idx = append(idx, ts)
+	}
+
+	for i, tc := range []struct {
+		low, high time.Time
+		want      []byte
+	}{
+		{low: idx[0], high: idx[0], want: []byte{}},
+		{low: idx[0], high: idx[1], want: []byte{0}},
+		{low: idx[0], high: idx[9], want: []byte{0, 1, 2, 3, 4, 5, 6, 7, 8}},
+		{low: idx[1], high: idx[9], want: []byte{1, 2, 3, 4, 5, 6, 7, 8}},
+		// Ensure that adding 1 correctly modifies the range semantics.
+		{low: idx[0].Add(1), high: idx[9], want: []byte{1, 2, 3, 4, 5, 6, 7, 8}},
+		{low: idx[0].Add(1), high: idx[9].Add(1), want: []byte{1, 2, 3, 4, 5, 6, 7, 8, 9}},
+	} {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			rows, err := m.ReadLog(ctx, directoryID, logID, tc.low, tc.high, 100)
+			if err != nil {
+				t.Fatalf("ReadLog(): %v", err)
+			}
+			got := make([]byte, 0, len(rows))
+			for _, r := range rows {
+				i := r.Mutation.Entry[0]
+				got = append(got, i)
+			}
+			if !cmp.Equal(got, tc.want) {
+				t.Fatalf("ReadLog(%v,%v): got %v, want %v", tc.low, tc.high, got, tc.want)
+			}
+		})
 	}
 }