maintner/maintnerd/gcslog/gcslog.go - build - Git at Google

 // Copyright 2017 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.

 // Package gcslog is an implementation of maintner.MutationSource and Logger for Google Cloud Storage.
 package gcslog

 import (
 	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/json"
 	"fmt"
 	"hash"
 	"io"
 	"log"
 	"net/http"
 	"path"
 	"regexp"
 	"sort"
 	"strconv"
 	"strings"
 	"sync"
 	"time"

 	"cloud.google.com/go/storage"
 	"github.com/golang/protobuf/proto"
 	"golang.org/x/build/maintner"
 	"golang.org/x/build/maintner/maintpb"
 	"golang.org/x/build/maintner/reclog"
 	"google.golang.org/api/iterator"
 )

 // targetObjectSize is the goal maximum size for each log segment on
 // GCS. In the unlikely case that a single record is larger than this
 // then a segment would be bigger than this (records are never split
 // between log segments).  But otherwise this is the max size.
 const targetObjectSize = 16 << 20

 const flushInterval = 10 * time.Minute

 // GCSLog implements MutationLogger and MutationSource.
 var _ maintner.MutationLogger = &GCSLog{}
 var _ maintner.MutationSource = &GCSLog{}

 // GCSLog logs mutations to GCS.
 type GCSLog struct {
 	sc            *storage.Client
 	bucketName    string
 	bucket        *storage.BucketHandle
 	segmentPrefix string
 	debug         bool

 	mu         sync.Mutex // guards the following
 	cond       *sync.Cond
 	seg        map[int]gcsLogSegment
 	curNum     int
 	logBuf     bytes.Buffer
 	logSHA224  hash.Hash
 	flushTimer *time.Timer // non-nil if flush timer is active
 }

 type gcsLogSegment struct {
 	num     int // starting with 0
 	size    int64
 	sha224  string // in lowercase hex
 	created time.Time
 }

 func (s gcsLogSegment) ObjectName() string {
 	return fmt.Sprintf("%04d.%s.mutlog", s.num, s.sha224)
 }

 func (s gcsLogSegment) String() string {
 	return fmt.Sprintf("{gcsLogSegment num=%v, size=%v, sha=%v, created=%v}", s.num, s.size, s.sha224, s.created.Format(time.RFC3339))
 }

 // newGCSLogBase returns a new gcsLog instance without any association
 // with Google Cloud Storage.
 func newGCSLogBase() *GCSLog {
 	gl := &GCSLog{
 		seg: map[int]gcsLogSegment{},
 	}
 	gl.cond = sync.NewCond(&gl.mu)
 	return gl
 }

 // NewGCSLog creates a GCSLog that logs mutations to a given GCS bucket.
 // If the bucket name contains a "/", the part after the slash will be a
 // prefix for the segments.
 func NewGCSLog(ctx context.Context, bucketName string) (*GCSLog, error) {
 	sc, err := storage.NewClient(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("storage.NewClient: %v", err)
 	}
 	gl := newGCSLogBase()
 	gl.sc = sc

 	prefix := ""
 	if f := strings.SplitN(bucketName, "/", 2); len(f) > 1 {
 		bucketName, prefix = f[0], f[1]
 	}

 	gl.bucketName = bucketName
 	gl.segmentPrefix = prefix
 	gl.bucket = sc.Bucket(gl.bucketName)
 	if err := gl.initLoad(ctx); err != nil {
 		return nil, err
 	}
 	return gl, nil
 }

 // objNameRx is used to identify a mutation log file by suffix.
 var objnameRx = regexp.MustCompile(`(\d{4})\.([0-9a-f]{56})\.mutlog$`)

 func (gl *GCSLog) initLoad(ctx context.Context) error {
 	it := gl.bucket.Objects(ctx, nil)
 	maxNum := 0
 	for {
 		objAttrs, err := it.Next()
 		if err == iterator.Done {
 			break
 		}
 		if err != nil {
 			return fmt.Errorf("iterating over %s bucket: %v", gl.bucketName, err)
 		}
 		if !strings.HasPrefix(objAttrs.Name, gl.segmentPrefix) {
 			log.Printf("Ignoring GCS object with invalid prefix %q", objAttrs.Name)
 			continue
 		}
 		m := objnameRx.FindStringSubmatch(objAttrs.Name)
 		if m == nil {
 			log.Printf("Ignoring unrecognized GCS object %q", objAttrs.Name)
 			continue
 		}
 		n, _ := strconv.ParseInt(m[1], 10, 32)
 		seg := gcsLogSegment{
 			num:     int(n),
 			sha224:  m[2],
 			size:    objAttrs.Size,
 			created: objAttrs.Created,
 		}
 		if seg.num > maxNum {
 			maxNum = seg.num
 		}
 		if prevSeg, ok := gl.seg[int(n)]; !ok || prevSeg.created.Before(seg.created) {
 			gl.seg[int(n)] = seg
 			log.Printf("seg[%v] = %s", n, seg)
 			if ok {
 				gl.deleteOldSegment(ctx, gl.objectPath(prevSeg))
 			}
 		}
 	}
 	gl.curNum = maxNum

 	if len(gl.seg) == 0 {
 		return nil
 	}

 	// Check for any missing segments.
 	for i := 0; i < maxNum; i++ {
 		if _, ok := gl.seg[i]; !ok {
 			return fmt.Errorf("saw max segment number %d but missing segment %d", maxNum, i)
 		}
 	}

 	// Should we resume writing to the latest entry?
 	// If the latest one is big enough, leave it be.
 	// Otherwise slurp it in and we'll append to it.
 	if gl.seg[maxNum].size >= targetObjectSize-(4<<10) {
 		gl.curNum++
 		return nil
 	}

 	r, err := gl.bucket.Object(gl.objectPath(gl.seg[maxNum])).NewReader(ctx)
 	if err != nil {
 		return err
 	}
 	defer r.Close()
 	if _, err = io.Copy(gcsLogWriter{gl}, r); err != nil {
 		return err
 	}

 	return nil
 }

 func (gl *GCSLog) objectPath(seg gcsLogSegment) string {
 	return path.Join(gl.segmentPrefix, seg.ObjectName())
 }

 func (gl *GCSLog) serveLogFile(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "GET" && r.Method != "HEAD" {
 		http.Error(w, "bad method", http.StatusBadRequest)
 		return
 	}

 	num, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/logs/"))
 	if err != nil {
 		http.Error(w, "bad path", http.StatusBadRequest)
 		return
 	}

 	gl.mu.Lock()
 	if num > gl.curNum {
 		gl.mu.Unlock()
 		http.Error(w, "bad segment number", http.StatusBadRequest)
 		return
 	}
 	if num != gl.curNum {
 		obj := gl.objectPath(gl.seg[num])
 		gl.mu.Unlock()
 		http.Redirect(w, r, "https://storage.googleapis.com/"+gl.bucketName+"/"+obj, http.StatusFound)
 		return
 	}
 	content := gl.logBuf.String()

 	gl.mu.Unlock()
 	w.Header().Set("Content-Type", "application/octet-stream")
 	http.ServeContent(w, r, "", time.Time{}, strings.NewReader(content))
 }

 func (gl *GCSLog) serveJSONLogsIndex(w http.ResponseWriter, r *http.Request) {
 	if r.Method != "GET" && r.Method != "HEAD" {
 		http.Error(w, "bad method", http.StatusBadRequest)
 		return
 	}

 	startSeg, _ := strconv.Atoi(r.FormValue("startseg"))
 	if startSeg < 0 {
 		http.Error(w, "bad startseg", http.StatusBadRequest)
 		return
 	}

 	// Long poll if request contains non-zero waitsizenot parameter.
 	// The client's provided 'waitsizenot' value is the sum of the segment
 	// sizes they already know. They're waiting for something new.
 	if s := r.FormValue("waitsizenot"); s != "" {
 		oldSize, err := strconv.ParseInt(s, 10, 64)
 		if err != nil || oldSize < 0 {
 			http.Error(w, "bad waitsizenot", http.StatusBadRequest)
 			return
 		}
 		// Return a 304 if there's no activity in just under a minute.
 		// This keeps some occasional activity on the TCP connection
 		// so we (and any proxies) know it's alive, and can fit
 		// within reason read/write deadlines on either side.
 		ctx, cancel := context.WithTimeout(r.Context(), 55*time.Second)
 		defer cancel()
 		changed := gl.waitSizeNot(ctx, oldSize)
 		if !changed {
 			w.WriteHeader(http.StatusNotModified)
 			return
 		}
 	}

 	segs := gl.getJSONLogs(startSeg)

 	w.Header().Set("Content-Type", "application/json")
 	w.Header().Set("X-Sum-Segment-Size", fmt.Sprint(sumSegmentSizes(segs)))

 	body, _ := json.MarshalIndent(segs, "", "\t")
 	w.Write(body)
 }

 // sumSegmentSizes returns the sum of each seg.Size in segs.
 func sumSegmentSizes(segs []maintner.LogSegmentJSON) (sum int64) {
 	for _, seg := range segs {
 		sum += seg.Size
 	}
 	return sum
 }

 // waitSizeNot blocks until the sum of GCSLog is not v, or the context expires.
 // It reports whether the size changed.
 func (gl *GCSLog) waitSizeNot(ctx context.Context, v int64) (changed bool) {
 	returned := make(chan struct{})
 	defer close(returned)
 	go gl.waitSizeNotAwaitContextOrChange(ctx, returned)
 	gl.mu.Lock()
 	defer gl.mu.Unlock()
 	for {
 		if curSize := gl.sumSizeLocked(); curSize != v {
 			if gl.debug {
 				log.Printf("gcslog: waitSize fired. from %d => %d", v, curSize)
 			}
 			return true
 		}
 		select {
 		case <-ctx.Done():
 			return false
 		default:
 			gl.cond.Wait()
 		}
 	}
 }

 // waitSizeNotAwaitContextOrChange is part of waitSizeNot.
 // It's a goroutine that selects on two channels and calls
 // sync.Cond.Broadcast to wake up the waitSizeNot waiter if the
 // context expires.
 func (gl *GCSLog) waitSizeNotAwaitContextOrChange(ctx context.Context, returned <-chan struct{}) {
 	select {
 	case <-ctx.Done():
 		gl.cond.Broadcast()
 	case <-returned:
 		// No need to do a wakeup. Caller is already gone.
 	}
 }

 func (gl *GCSLog) sumSizeLocked() int64 {
 	var sum int64
 	for n, seg := range gl.seg {
 		if n != gl.curNum {
 			sum += seg.size
 		}
 	}
 	sum += int64(gl.logBuf.Len())
 	return sum
 }

 func (gl *GCSLog) getJSONLogs(startSeg int) (segs []maintner.LogSegmentJSON) {
 	gl.mu.Lock()
 	defer gl.mu.Unlock()
 	if startSeg > gl.curNum || startSeg < 0 {
 		startSeg = 0
 	}
 	segs = make([]maintner.LogSegmentJSON, 0, gl.curNum-startSeg)
 	for i := startSeg; i < gl.curNum; i++ {
 		seg := gl.seg[i]
 		segs = append(segs, maintner.LogSegmentJSON{
 			Number: i,
 			Size:   seg.size,
 			SHA224: seg.sha224,
 			URL:    fmt.Sprintf("https://storage.googleapis.com/%s/%s", gl.bucketName, gl.objectPath(seg)),
 		})
 	}
 	if gl.logBuf.Len() > 0 {
 		segs = append(segs, maintner.LogSegmentJSON{
 			Number: gl.curNum,
 			Size:   int64(gl.logBuf.Len()),
 			SHA224: fmt.Sprintf("%x", gl.logSHA224.Sum(nil)),
 			URL:    fmt.Sprintf("/logs/%d", gl.curNum),
 		})
 	}
 	return
 }

 // gcsLogWriter is the io.Writer used to write to GCSLog.logBuf. It
 // keeps the sha224 in sync. Caller must hold gl.mu.
 type gcsLogWriter struct{ gl *GCSLog }

 func (w gcsLogWriter) Write(p []byte) (n int, err error) {
 	gl := w.gl
 	if gl.logBuf.Len() == 0 {
 		gl.logSHA224 = sha256.New224()
 	}
 	n, err = gl.logSHA224.Write(p)
 	if n != len(p) || err != nil {
 		panic(fmt.Sprintf("unexpected write (%v, %v) for %v bytes", n, err, len(p)))
 	}
 	n, err = gl.logBuf.Write(p)
 	if n != len(p) || err != nil {
 		panic(fmt.Sprintf("unexpected write (%v, %v) for %v bytes", n, err, len(p)))
 	}
 	return len(p), nil
 }

 // SetDebug controls whether verbose debugging is enabled on this log.
 //
 // It must only be called before it's used.
 func (gl *GCSLog) SetDebug(v bool) { gl.debug = v }

 // Log writes m to GCS after the buffer is full or after a periodic flush.
 func (gl *GCSLog) Log(m *maintpb.Mutation) error {
 	data, err := proto.Marshal(m)
 	if err != nil {
 		return err
 	}

 	gl.mu.Lock()
 	defer gl.mu.Unlock()

 	// If we have some data and this item would push us over, flush.
 	if gl.logBuf.Len()+len(data) > targetObjectSize {
 		log.Printf("Log: record requires buffer flush.")
 		if err := gl.flushLocked(context.TODO()); err != nil {
 			return err
 		}
 		gl.curNum++
 		gl.logBuf.Reset()
 		log.Printf("cur log file now %d", gl.curNum)
 	}

 	if err := reclog.WriteRecord(gcsLogWriter{gl}, int64(gl.logBuf.Len()), data); err != nil {
 		return err
 	}
 	gl.cond.Broadcast() // wake any long-polling subscribers

 	// Otherwise schedule a periodic flush.
 	if gl.flushTimer == nil {
 		log.Printf("wrote record; flush timer registered.")
 		gl.flushTimer = time.AfterFunc(flushInterval, gl.onFlushTimer)
 	} else {
 		log.Printf("wrote record; using existing flush timer.")
 	}
 	return nil
 }

 func (gl *GCSLog) onFlushTimer() {
 	log.Printf("flush timer fired.")
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	gl.flush(ctx)
 }

 func (gl *GCSLog) flush(ctx context.Context) error {
 	gl.mu.Lock()

 	defer gl.mu.Unlock()
 	if gl.flushTimer != nil {
 		gl.flushTimer.Stop()
 		gl.flushTimer = nil
 	}

 	if err := gl.flushLocked(ctx); err != nil {
 		gl.flushTimer = time.AfterFunc(1*time.Minute, gl.onFlushTimer)
 		log.Printf("Error background flushing: %v", err)
 		return err
 	}

 	return nil
 }

 func (gl *GCSLog) flushLocked(ctx context.Context) error {
 	buf := gl.logBuf.Bytes()
 	if len(buf) == 0 {
 		return nil
 	}
 	seg := gcsLogSegment{
 		num:    gl.curNum,
 		sha224: fmt.Sprintf("%x", sha256.Sum224(buf)),
 		size:   int64(len(buf)),
 	}
 	objName := gl.objectPath(seg)
 	log.Printf("flushing %s (%d bytes)", objName, len(buf))
 	err := try(4, time.Second, func() error {
 		w := gl.bucket.Object(objName).NewWriter(ctx)
 		w.ContentType = "application/octet-stream"
 		if _, err := w.Write(buf); err != nil {
 			return err
 		}
 		if err := w.Close(); err != nil {
 			return err
 		}
 		attrs := w.Attrs()
 		seg.created = attrs.Created
 		return nil
 	})
 	if err != nil {
 		return err
 	}

 	old := gl.seg[seg.num]
 	gl.seg[seg.num] = seg

 	// Atomically update the manifest file. If we lose the CAS
 	// race, that means some other instance of this process is
 	// running at the same time (e.g. accidental replica count > 1
 	// in k8s config) and we should fail hard.
 	// TODO: that^

 	// Delete any old segment from the same position.
 	if old.sha224 != "" && old.sha224 != seg.sha224 {
 		gl.deleteOldSegment(ctx, gl.objectPath(old))
 	}
 	return nil
 }

 func (gl *GCSLog) deleteOldSegment(ctx context.Context, objName string) {
 	err := gl.bucket.Object(objName).Delete(ctx)
 	if err != nil {
 		// Can ignore, though. Probably emphemeral, and not critical.
 		// It'll be deleted by new versions or next start-up anyway.
 		log.Printf("Warning: error deleting old segment version %v: %v", objName, err)
 	} else {
 		log.Printf("deleted old segment version %v", objName)
 	}
 }

 func (gl *GCSLog) objectNames() (names []string) {
 	gl.mu.Lock()
 	defer gl.mu.Unlock()
 	for _, seg := range gl.seg {
 		names = append(names, gl.objectPath(seg))
 	}
 	sort.Strings(names)
 	return
 }

 func (gl *GCSLog) foreachSegmentReader(ctx context.Context, fn func(r io.Reader) error) error {
 	objs := gl.objectNames()
 	for i, obj := range objs {
 		log.Printf("Reading %d/%d: %s ...", i+1, len(objs), obj)
 		rd, err := gl.bucket.Object(obj).NewReader(ctx)
 		if err != nil {
 			return fmt.Errorf("failed to open %v: %v", obj, err)
 		}
 		err = fn(rd)
 		rd.Close()
 		if err != nil {
 			return fmt.Errorf("error processing %v: %v", obj, err)
 		}
 	}
 	return nil
 }

 // GetMutations returns a channel of mutations or related events.
 // The channel will never be closed.
 // All sends on the returned channel should select on the provided context.
 func (gl *GCSLog) GetMutations(ctx context.Context) <-chan maintner.MutationStreamEvent {
 	ch := make(chan maintner.MutationStreamEvent, 50) // buffered: overlap gunzip/unmarshal with loading
 	go func() {
 		err := gl.foreachSegmentReader(ctx, func(r io.Reader) error {
 			return reclog.ForeachRecord(r, 0, func(off int64, hdr, rec []byte) error {
 				m := new(maintpb.Mutation)
 				if err := proto.Unmarshal(rec, m); err != nil {
 					return err
 				}
 				select {
 				case ch <- maintner.MutationStreamEvent{Mutation: m}:
 					return nil
 				case <-ctx.Done():
 					return ctx.Err()
 				}
 			})
 		})
 		final := maintner.MutationStreamEvent{Err: err}
 		if err == nil {
 			final.End = true
 		}
 		select {
 		case ch <- final:
 		case <-ctx.Done():
 		}
 	}()
 	return ch
 }

 func try(tries int, firstDelay time.Duration, fn func() error) error {
 	var err error
 	delay := firstDelay
 	for i := 0; i < tries; i++ {
 		err = fn()
 		if err == nil {
 			return nil
 		}
 		time.Sleep(delay)
 		delay *= 2
 	}
 	return err
 }

 // CopyFrom is only used for the one-time migrate from disk-to-GCS code path.
 func (gl *GCSLog) CopyFrom(src maintner.MutationSource) error {
 	gl.curNum = 0
 	ctx := context.Background()
 	for e := range src.GetMutations(ctx) {
 		if e.Err != nil {
 			log.Printf("Corpus.Initialize: %v", e.Err)
 			return e.Err
 		}
 		if e.End {
 			log.Printf("reached end. flushing.")
 			err := gl.flush(ctx)
 			log.Printf("final flush = %v", err)
 			return nil
 		}
 		if err := gl.Log(e.Mutation); err != nil {
 			return err
 		}
 	}
 	panic("unexpected channel close")
 }

 // RegisterHandlers adds handlers for the default paths (/logs and /logs/).
 func (gl *GCSLog) RegisterHandlers(mux *http.ServeMux) {
 	mux.HandleFunc("/logs", gl.serveJSONLogsIndex)
 	mux.HandleFunc("/logs/", gl.serveLogFile)
 }
	// Copyright 2017 The Go Authors. All rights reserved.
	// Use of this source code is governed by a BSD-style
	// license that can be found in the LICENSE file.

	// Package gcslog is an implementation of maintner.MutationSource and Logger for Google Cloud Storage.
	package gcslog

	import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/json"
	"fmt"
	"hash"
	"io"
	"log"
	"net/http"
	"path"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"cloud.google.com/go/storage"
	"github.com/golang/protobuf/proto"
	"golang.org/x/build/maintner"
	"golang.org/x/build/maintner/maintpb"
	"golang.org/x/build/maintner/reclog"
	"google.golang.org/api/iterator"
	)

	// targetObjectSize is the goal maximum size for each log segment on
	// GCS. In the unlikely case that a single record is larger than this
	// then a segment would be bigger than this (records are never split
	// between log segments). But otherwise this is the max size.
	const targetObjectSize = 16 << 20

	const flushInterval = 10 * time.Minute

	// GCSLog implements MutationLogger and MutationSource.
	var _ maintner.MutationLogger = &GCSLog{}
	var _ maintner.MutationSource = &GCSLog{}

	// GCSLog logs mutations to GCS.
	type GCSLog struct {
	sc *storage.Client
	bucketName string
	bucket *storage.BucketHandle
	segmentPrefix string
	debug bool

	mu sync.Mutex // guards the following
	cond *sync.Cond
	seg map[int]gcsLogSegment
	curNum int
	logBuf bytes.Buffer
	logSHA224 hash.Hash
	flushTimer *time.Timer // non-nil if flush timer is active
	}

	type gcsLogSegment struct {
	num int // starting with 0
	size int64
	sha224 string // in lowercase hex
	created time.Time
	}

	func (s gcsLogSegment) ObjectName() string {
	return fmt.Sprintf("%04d.%s.mutlog", s.num, s.sha224)
	}

	func (s gcsLogSegment) String() string {
	return fmt.Sprintf("{gcsLogSegment num=%v, size=%v, sha=%v, created=%v}", s.num, s.size, s.sha224, s.created.Format(time.RFC3339))
	}

	// newGCSLogBase returns a new gcsLog instance without any association
	// with Google Cloud Storage.
	func newGCSLogBase() *GCSLog {
	gl := &GCSLog{
	seg: map[int]gcsLogSegment{},
	}
	gl.cond = sync.NewCond(&gl.mu)
	return gl
	}

	// NewGCSLog creates a GCSLog that logs mutations to a given GCS bucket.
	// If the bucket name contains a "/", the part after the slash will be a
	// prefix for the segments.
	func NewGCSLog(ctx context.Context, bucketName string) (*GCSLog, error) {
	sc, err := storage.NewClient(ctx)
	if err != nil {
	return nil, fmt.Errorf("storage.NewClient: %v", err)
	}
	gl := newGCSLogBase()
	gl.sc = sc

	prefix := ""
	if f := strings.SplitN(bucketName, "/", 2); len(f) > 1 {
	bucketName, prefix = f[0], f[1]
	}

	gl.bucketName = bucketName
	gl.segmentPrefix = prefix
	gl.bucket = sc.Bucket(gl.bucketName)
	if err := gl.initLoad(ctx); err != nil {
	return nil, err
	}
	return gl, nil
	}

	// objNameRx is used to identify a mutation log file by suffix.
	var objnameRx = regexp.MustCompile(`(\d{4})\.([0-9a-f]{56})\.mutlog$`)

	func (gl *GCSLog) initLoad(ctx context.Context) error {
	it := gl.bucket.Objects(ctx, nil)
	maxNum := 0
	for {
	objAttrs, err := it.Next()
	if err == iterator.Done {
	break
	}
	if err != nil {
	return fmt.Errorf("iterating over %s bucket: %v", gl.bucketName, err)
	}
	if !strings.HasPrefix(objAttrs.Name, gl.segmentPrefix) {
	log.Printf("Ignoring GCS object with invalid prefix %q", objAttrs.Name)
	continue
	}
	m := objnameRx.FindStringSubmatch(objAttrs.Name)
	if m == nil {
	log.Printf("Ignoring unrecognized GCS object %q", objAttrs.Name)
	continue
	}
	n, _ := strconv.ParseInt(m[1], 10, 32)
	seg := gcsLogSegment{
	num: int(n),
	sha224: m[2],
	size: objAttrs.Size,
	created: objAttrs.Created,
	}
	if seg.num > maxNum {
	maxNum = seg.num
	}
	if prevSeg, ok := gl.seg[int(n)]; !ok \|\| prevSeg.created.Before(seg.created) {
	gl.seg[int(n)] = seg
	log.Printf("seg[%v] = %s", n, seg)
	if ok {
	gl.deleteOldSegment(ctx, gl.objectPath(prevSeg))
	}
	}
	}
	gl.curNum = maxNum

	if len(gl.seg) == 0 {
	return nil
	}

	// Check for any missing segments.
	for i := 0; i < maxNum; i++ {
	if _, ok := gl.seg[i]; !ok {
	return fmt.Errorf("saw max segment number %d but missing segment %d", maxNum, i)
	}
	}

	// Should we resume writing to the latest entry?
	// If the latest one is big enough, leave it be.
	// Otherwise slurp it in and we'll append to it.
	if gl.seg[maxNum].size >= targetObjectSize-(4<<10) {
	gl.curNum++
	return nil
	}

	r, err := gl.bucket.Object(gl.objectPath(gl.seg[maxNum])).NewReader(ctx)
	if err != nil {
	return err
	}
	defer r.Close()
	if _, err = io.Copy(gcsLogWriter{gl}, r); err != nil {
	return err
	}

	return nil
	}

	func (gl *GCSLog) objectPath(seg gcsLogSegment) string {
	return path.Join(gl.segmentPrefix, seg.ObjectName())
	}

	func (gl GCSLog) serveLogFile(w http.ResponseWriter, r http.Request) {
	if r.Method != "GET" && r.Method != "HEAD" {
	http.Error(w, "bad method", http.StatusBadRequest)
	return
	}

	num, err := strconv.Atoi(strings.TrimPrefix(r.URL.Path, "/logs/"))
	if err != nil {
	http.Error(w, "bad path", http.StatusBadRequest)
	return
	}

	gl.mu.Lock()
	if num > gl.curNum {
	gl.mu.Unlock()
	http.Error(w, "bad segment number", http.StatusBadRequest)
	return
	}
	if num != gl.curNum {
	obj := gl.objectPath(gl.seg[num])
	gl.mu.Unlock()
	http.Redirect(w, r, "https://storage.googleapis.com/"+gl.bucketName+"/"+obj, http.StatusFound)
	return
	}
	content := gl.logBuf.String()

	gl.mu.Unlock()
	w.Header().Set("Content-Type", "application/octet-stream")
	http.ServeContent(w, r, "", time.Time{}, strings.NewReader(content))
	}

	func (gl GCSLog) serveJSONLogsIndex(w http.ResponseWriter, r http.Request) {
	if r.Method != "GET" && r.Method != "HEAD" {
	http.Error(w, "bad method", http.StatusBadRequest)
	return
	}

	startSeg, _ := strconv.Atoi(r.FormValue("startseg"))
	if startSeg < 0 {
	http.Error(w, "bad startseg", http.StatusBadRequest)
	return
	}

	// Long poll if request contains non-zero waitsizenot parameter.
	// The client's provided 'waitsizenot' value is the sum of the segment
	// sizes they already know. They're waiting for something new.
	if s := r.FormValue("waitsizenot"); s != "" {
	oldSize, err := strconv.ParseInt(s, 10, 64)
	if err != nil \|\| oldSize < 0 {
	http.Error(w, "bad waitsizenot", http.StatusBadRequest)
	return
	}
	// Return a 304 if there's no activity in just under a minute.
	// This keeps some occasional activity on the TCP connection
	// so we (and any proxies) know it's alive, and can fit
	// within reason read/write deadlines on either side.
	ctx, cancel := context.WithTimeout(r.Context(), 55*time.Second)
	defer cancel()
	changed := gl.waitSizeNot(ctx, oldSize)
	if !changed {
	w.WriteHeader(http.StatusNotModified)
	return
	}
	}

	segs := gl.getJSONLogs(startSeg)

	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("X-Sum-Segment-Size", fmt.Sprint(sumSegmentSizes(segs)))

	body, _ := json.MarshalIndent(segs, "", "\t")
	w.Write(body)
	}

	// sumSegmentSizes returns the sum of each seg.Size in segs.
	func sumSegmentSizes(segs []maintner.LogSegmentJSON) (sum int64) {
	for _, seg := range segs {
	sum += seg.Size
	}
	return sum
	}

	// waitSizeNot blocks until the sum of GCSLog is not v, or the context expires.
	// It reports whether the size changed.
	func (gl *GCSLog) waitSizeNot(ctx context.Context, v int64) (changed bool) {
	returned := make(chan struct{})
	defer close(returned)
	go gl.waitSizeNotAwaitContextOrChange(ctx, returned)
	gl.mu.Lock()
	defer gl.mu.Unlock()
	for {
	if curSize := gl.sumSizeLocked(); curSize != v {
	if gl.debug {
	log.Printf("gcslog: waitSize fired. from %d => %d", v, curSize)
	}
	return true
	}
	select {
	case <-ctx.Done():
	return false
	default:
	gl.cond.Wait()
	}
	}
	}

	// waitSizeNotAwaitContextOrChange is part of waitSizeNot.
	// It's a goroutine that selects on two channels and calls
	// sync.Cond.Broadcast to wake up the waitSizeNot waiter if the
	// context expires.
	func (gl *GCSLog) waitSizeNotAwaitContextOrChange(ctx context.Context, returned <-chan struct{}) {
	select {
	case <-ctx.Done():
	gl.cond.Broadcast()
	case <-returned:
	// No need to do a wakeup. Caller is already gone.
	}
	}

	func (gl *GCSLog) sumSizeLocked() int64 {
	var sum int64
	for n, seg := range gl.seg {
	if n != gl.curNum {
	sum += seg.size
	}
	}
	sum += int64(gl.logBuf.Len())
	return sum
	}

	func (gl *GCSLog) getJSONLogs(startSeg int) (segs []maintner.LogSegmentJSON) {
	gl.mu.Lock()
	defer gl.mu.Unlock()
	if startSeg > gl.curNum \|\| startSeg < 0 {
	startSeg = 0
	}
	segs = make([]maintner.LogSegmentJSON, 0, gl.curNum-startSeg)
	for i := startSeg; i < gl.curNum; i++ {
	seg := gl.seg[i]
	segs = append(segs, maintner.LogSegmentJSON{
	Number: i,
	Size: seg.size,
	SHA224: seg.sha224,
	URL: fmt.Sprintf("https://storage.googleapis.com/%s/%s", gl.bucketName, gl.objectPath(seg)),
	})
	}
	if gl.logBuf.Len() > 0 {
	segs = append(segs, maintner.LogSegmentJSON{
	Number: gl.curNum,
	Size: int64(gl.logBuf.Len()),
	SHA224: fmt.Sprintf("%x", gl.logSHA224.Sum(nil)),
	URL: fmt.Sprintf("/logs/%d", gl.curNum),
	})
	}
	return
	}

	// gcsLogWriter is the io.Writer used to write to GCSLog.logBuf. It
	// keeps the sha224 in sync. Caller must hold gl.mu.
	type gcsLogWriter struct{ gl *GCSLog }

	func (w gcsLogWriter) Write(p []byte) (n int, err error) {
	gl := w.gl
	if gl.logBuf.Len() == 0 {
	gl.logSHA224 = sha256.New224()
	}
	n, err = gl.logSHA224.Write(p)
	if n != len(p) \|\| err != nil {
	panic(fmt.Sprintf("unexpected write (%v, %v) for %v bytes", n, err, len(p)))
	}
	n, err = gl.logBuf.Write(p)
	if n != len(p) \|\| err != nil {
	panic(fmt.Sprintf("unexpected write (%v, %v) for %v bytes", n, err, len(p)))
	}
	return len(p), nil
	}

	// SetDebug controls whether verbose debugging is enabled on this log.
	//
	// It must only be called before it's used.
	func (gl *GCSLog) SetDebug(v bool) { gl.debug = v }

	// Log writes m to GCS after the buffer is full or after a periodic flush.
	func (gl GCSLog) Log(m maintpb.Mutation) error {
	data, err := proto.Marshal(m)
	if err != nil {
	return err
	}

	gl.mu.Lock()
	defer gl.mu.Unlock()

	// If we have some data and this item would push us over, flush.
	if gl.logBuf.Len()+len(data) > targetObjectSize {
	log.Printf("Log: record requires buffer flush.")
	if err := gl.flushLocked(context.TODO()); err != nil {
	return err
	}
	gl.curNum++
	gl.logBuf.Reset()
	log.Printf("cur log file now %d", gl.curNum)
	}

	if err := reclog.WriteRecord(gcsLogWriter{gl}, int64(gl.logBuf.Len()), data); err != nil {
	return err
	}
	gl.cond.Broadcast() // wake any long-polling subscribers

	// Otherwise schedule a periodic flush.
	if gl.flushTimer == nil {
	log.Printf("wrote record; flush timer registered.")
	gl.flushTimer = time.AfterFunc(flushInterval, gl.onFlushTimer)
	} else {
	log.Printf("wrote record; using existing flush timer.")
	}
	return nil
	}

	func (gl *GCSLog) onFlushTimer() {
	log.Printf("flush timer fired.")
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
	defer cancel()
	gl.flush(ctx)
	}

	func (gl *GCSLog) flush(ctx context.Context) error {
	gl.mu.Lock()

	defer gl.mu.Unlock()
	if gl.flushTimer != nil {
	gl.flushTimer.Stop()
	gl.flushTimer = nil
	}

	if err := gl.flushLocked(ctx); err != nil {
	gl.flushTimer = time.AfterFunc(1*time.Minute, gl.onFlushTimer)
	log.Printf("Error background flushing: %v", err)
	return err
	}

	return nil
	}

	func (gl *GCSLog) flushLocked(ctx context.Context) error {
	buf := gl.logBuf.Bytes()
	if len(buf) == 0 {
	return nil
	}
	seg := gcsLogSegment{
	num: gl.curNum,
	sha224: fmt.Sprintf("%x", sha256.Sum224(buf)),
	size: int64(len(buf)),
	}
	objName := gl.objectPath(seg)
	log.Printf("flushing %s (%d bytes)", objName, len(buf))
	err := try(4, time.Second, func() error {
	w := gl.bucket.Object(objName).NewWriter(ctx)
	w.ContentType = "application/octet-stream"
	if _, err := w.Write(buf); err != nil {
	return err
	}
	if err := w.Close(); err != nil {
	return err
	}
	attrs := w.Attrs()
	seg.created = attrs.Created
	return nil
	})
	if err != nil {
	return err
	}

	old := gl.seg[seg.num]
	gl.seg[seg.num] = seg

	// Atomically update the manifest file. If we lose the CAS
	// race, that means some other instance of this process is
	// running at the same time (e.g. accidental replica count > 1
	// in k8s config) and we should fail hard.
	// TODO: that^

	// Delete any old segment from the same position.
	if old.sha224 != "" && old.sha224 != seg.sha224 {
	gl.deleteOldSegment(ctx, gl.objectPath(old))
	}
	return nil
	}

	func (gl *GCSLog) deleteOldSegment(ctx context.Context, objName string) {
	err := gl.bucket.Object(objName).Delete(ctx)
	if err != nil {
	// Can ignore, though. Probably emphemeral, and not critical.
	// It'll be deleted by new versions or next start-up anyway.
	log.Printf("Warning: error deleting old segment version %v: %v", objName, err)
	} else {
	log.Printf("deleted old segment version %v", objName)
	}
	}

	func (gl *GCSLog) objectNames() (names []string) {
	gl.mu.Lock()
	defer gl.mu.Unlock()
	for _, seg := range gl.seg {
	names = append(names, gl.objectPath(seg))
	}
	sort.Strings(names)
	return
	}

	func (gl *GCSLog) foreachSegmentReader(ctx context.Context, fn func(r io.Reader) error) error {
	objs := gl.objectNames()
	for i, obj := range objs {
	log.Printf("Reading %d/%d: %s ...", i+1, len(objs), obj)
	rd, err := gl.bucket.Object(obj).NewReader(ctx)
	if err != nil {
	return fmt.Errorf("failed to open %v: %v", obj, err)
	}
	err = fn(rd)
	rd.Close()
	if err != nil {
	return fmt.Errorf("error processing %v: %v", obj, err)
	}
	}
	return nil
	}

	// GetMutations returns a channel of mutations or related events.
	// The channel will never be closed.
	// All sends on the returned channel should select on the provided context.
	func (gl *GCSLog) GetMutations(ctx context.Context) <-chan maintner.MutationStreamEvent {
	ch := make(chan maintner.MutationStreamEvent, 50) // buffered: overlap gunzip/unmarshal with loading
	go func() {
	err := gl.foreachSegmentReader(ctx, func(r io.Reader) error {
	return reclog.ForeachRecord(r, 0, func(off int64, hdr, rec []byte) error {
	m := new(maintpb.Mutation)
	if err := proto.Unmarshal(rec, m); err != nil {
	return err
	}
	select {
	case ch <- maintner.MutationStreamEvent{Mutation: m}:
	return nil
	case <-ctx.Done():
	return ctx.Err()
	}
	})
	})
	final := maintner.MutationStreamEvent{Err: err}
	if err == nil {
	final.End = true
	}
	select {
	case ch <- final:
	case <-ctx.Done():
	}
	}()
	return ch
	}

	func try(tries int, firstDelay time.Duration, fn func() error) error {
	var err error
	delay := firstDelay
	for i := 0; i < tries; i++ {
	err = fn()
	if err == nil {
	return nil
	}
	time.Sleep(delay)
	delay *= 2
	}
	return err
	}

	// CopyFrom is only used for the one-time migrate from disk-to-GCS code path.
	func (gl *GCSLog) CopyFrom(src maintner.MutationSource) error {
	gl.curNum = 0
	ctx := context.Background()
	for e := range src.GetMutations(ctx) {
	if e.Err != nil {
	log.Printf("Corpus.Initialize: %v", e.Err)
	return e.Err
	}
	if e.End {
	log.Printf("reached end. flushing.")
	err := gl.flush(ctx)
	log.Printf("final flush = %v", err)
	return nil
	}
	if err := gl.Log(e.Mutation); err != nil {
	return err
	}
	}
	panic("unexpected channel close")
	}

	// RegisterHandlers adds handlers for the default paths (/logs and /logs/).
	func (gl GCSLog) RegisterHandlers(mux http.ServeMux) {
	mux.HandleFunc("/logs", gl.serveJSONLogsIndex)
	mux.HandleFunc("/logs/", gl.serveLogFile)
	}