func (pb *Base) GetStats() Stats {
	return pb.stats.GetStats()
}
GetStats returns the buffer/sink stats.
package row
import (
	"context"
	"errors"
	"io"
	"log"
	"sync"
	"time"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/m-lab/annotation-service/api"
	v2as "github.com/m-lab/annotation-service/api/v2"
	"github.com/m-lab/go/logx"
	"github.com/m-lab/etl/metrics"
)
var (
	ErrAnnotationError = errors.New("Annotation error")
	ErrNotAnnotatable  = errors.New("object does not implement Annotatable")
	ErrBufferFull      = errors.New("Buffer full")
	ErrInvalidSink     = errors.New("Not a valid row.Sink")
)
type Annotatable interface {
	GetLogTime() time.Time
	GetClientIPs() []string  
	GetServerIP() string
	AnnotateClients(map[string]*api.Annotations) error  
	AnnotateServer(*api.Annotations) error              
}
type Stats struct {
	Buffered  int  
	Pending   int  
	Committed int
	Failed    int
}
func (s Stats) Total() int {
	return s.Buffered + s.Pending + s.Committed + s.Failed
}
type ActiveStats struct {
	lock sync.RWMutex  
	Stats
}
func (as *ActiveStats) GetStats() Stats {
	as.lock.RLock()
	defer as.lock.RUnlock()
	return as.Stats
}
func (as *ActiveStats) MoveToPending(n int) {
	as.lock.Lock()
	defer as.lock.Unlock()
	as.Buffered -= n
	if as.Buffered < 0 {
		log.Println("BROKEN - negative buffered")
	}
	as.Pending += n
}
func (as *ActiveStats) Inc() {
	logx.Debug.Println("IncPending")
	as.lock.Lock()
	defer as.lock.Unlock()
	as.Buffered++
}
func (as *ActiveStats) Done(n int, err error) {
	as.lock.Lock()
	defer as.lock.Unlock()
	as.Pending -= n
	if as.Pending < 0 {
		log.Println("BROKEN: negative Pending")
	}
	if err != nil {
		as.Failed += n
	} else {
		as.Committed += n
	}
	logx.Debug.Printf("Done %d->%d %v\n", as.Pending+n, as.Pending, err)
}
type HasStats interface {
	GetStats() Stats
}
type Sink interface {
	Commit(rows []interface{}, label string) (int, error)
	io.Closer
}
type Buffer struct {
	lock sync.Mutex
	size int  
	rows []interface{}
}
func NewBuffer(size int) *Buffer {
	return &Buffer{size: size, rows: make([]interface{}, 0, size)}
}
func (buf *Buffer) Append(row interface{}) []interface{} {
	buf.lock.Lock()
	defer buf.lock.Unlock()
	if len(buf.rows) < buf.size {
		buf.rows = append(buf.rows, row)
		return nil
	}
	rows := buf.rows
	buf.rows = make([]interface{}, 0, buf.size)
	buf.rows = append(buf.rows, row)
	return rows
}
func (buf *Buffer) Reset() []interface{} {
	buf.lock.Lock()
	defer buf.lock.Unlock()
	res := buf.rows
	buf.rows = make([]interface{}, 0, buf.size)
	return res
}
type annotator struct {
	v2 v2as.Annotator
}
func (ann *annotator) annotateServers(rows []interface{}, label string) error {
	serverIPs := make(map[string]struct{})
	logTime := time.Time{}
	for i := range rows {
		r, ok := rows[i].(Annotatable)
		if !ok {
			return ErrNotAnnotatable
		}
		ip := r.GetServerIP()
		if ip != "" {
			serverIPs[ip] = struct{}{}
		}
		if (logTime == time.Time{}) {
			logTime = r.GetLogTime()
		}
	}
	ipSlice := make([]string, 0, len(rows))
	for ip := range serverIPs {
		ipSlice = append(ipSlice, ip)
	}
	if len(ipSlice) == 0 {
		return nil
	}
	response, err := ann.v2.GetAnnotations(context.Background(), logTime, ipSlice, label)
	if err != nil {
		log.Println("error in server GetAnnotations: ", err)
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "Server IP: RPC err in GetAnnotations."}).Inc()
		return err
	}
	annMap := response.Annotations
	if annMap == nil {
		log.Println("empty server annotation response")
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "Server IP: empty response"}).Inc()
		return ErrAnnotationError
	}
	for i := range rows {
		r, ok := rows[i].(Annotatable)
		if !ok {
			err = ErrNotAnnotatable
		} else {
			ip := r.GetServerIP()
			if ip != "" {
				ann, ok := annMap[ip]
				if ok {
					r.AnnotateServer(ann)
				}
			}
		}
	}
	return err
}
var logEmptyAnn = logx.NewLogEvery(nil, 60*time.Second)
func (ann *annotator) annotateClients(rows []interface{}, label string) error {
	ipSlice := make([]string, 0, 2*len(rows))  
	logTime := time.Time{}
	for i := range rows {
		r, ok := rows[i].(Annotatable)
		if !ok {
			return ErrNotAnnotatable
		}
		ipSlice = append(ipSlice, r.GetClientIPs()...)
		if (logTime == time.Time{}) {
			logTime = r.GetLogTime()
		}
	}
	response, err := ann.v2.GetAnnotations(context.Background(), logTime, ipSlice, label)
	if err != nil {
		log.Println("error in client GetAnnotations: ", err)
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "Client IP: RPC err in GetAnnotations."}).Inc()
		return err
	}
	annMap := response.Annotations
	if annMap == nil {
		logEmptyAnn.Println("empty client annotation response")
		metrics.AnnotationErrorCount.With(prometheus.
			Labels{"source": "Client IP: empty response"}).Inc()
		return ErrAnnotationError
	}
	for i := range rows {
		r, ok := rows[i].(Annotatable)
		if !ok {
			err = ErrNotAnnotatable
		} else {
			r.AnnotateClients(annMap)
		}
	}
	return err
}
func (ann *annotator) Annotate(rows []interface{}, metricLabel string) error {
	metrics.WorkerState.WithLabelValues(metricLabel, "annotate").Inc()
	defer metrics.WorkerState.WithLabelValues(metricLabel, "annotate").Dec()
	if len(rows) == 0 {
		return nil
	}
	defer func(label string, start time.Time) {
		metrics.AnnotationTimeSummary.With(prometheus.Labels{"test_type": label}).Observe(float64(time.Since(start).Nanoseconds()))
	}(metricLabel, time.Now())
	clientErr := ann.annotateClients(rows, metricLabel)
	serverErr := ann.annotateServers(rows, metricLabel)
	if clientErr != nil {
		return clientErr
	}
	if serverErr != nil {
		return serverErr
	}
	return nil
}
type Base struct {
	sink  Sink
	ann   annotator
	buf   *Buffer
	label string  
	stats ActiveStats
}
func NewBase(label string, sink Sink, bufSize int, ann v2as.Annotator) *Base {
	buf := NewBuffer(bufSize)
	return &Base{sink: sink, ann: annotator{ann}, buf: buf, label: label}
}