func (b *ConsumerWithMetricsBuilder) SetRollUpSpan(dur time.Duration) {
	if dur < 0 {
		panic("Non-negative duration required.")
	}
	b.c.attributes.RollUpSpan = dur
}
SetRollUpSpan sets the length of time periods for rolled up values
Other than setting RollUpSpan consumer attribute, this method has
no effect on built consumer.
package pstore
import (
	"bytes"
	"errors"
	"fmt"
	"github.com/Symantec/Dominator/lib/log"
	"github.com/Symantec/Dominator/lib/log/nulllogger"
	"github.com/Symantec/scotty/lib/gate"
	"github.com/Symantec/scotty/store"
	"github.com/Symantec/tricorder/go/tricorder"
	"github.com/Symantec/tricorder/go/tricorder/types"
	"github.com/Symantec/tricorder/go/tricorder/units"
	"regexp"
	"sync"
	"time"
)
const (
	TagAppName    = "appname"
	TagRegionName = "region"
	TagIpAddress  = "ipaddress"
)
var (
	ErrDisabled = errors.New("pstore: Writer disabled.")
)
type TagGroup map[string]string
func (t TagGroup) String() string {
	buffer := &bytes.Buffer{}
	fmt.Fprintf(buffer, "{")
	firstTime := true
	for k, v := range t {
		if firstTime {
			fmt.Fprintf(buffer, "%s: %s", k, v)
			firstTime = false
		} else {
			fmt.Fprintf(buffer, ", %s: %s", k, v)
		}
	}
	fmt.Fprintf(buffer, "}")
	return buffer.String()
}
type Record struct {
	HostName string
	Path string
	Tags TagGroup
	Kind types.Type
	SubType types.Type
	Unit units.Unit
	Value interface{}
	Timestamp time.Time
}
func (r Record) String() string {
	return fmt.Sprintf("{HostName: %s, Path: %s, Tags: %v, Kind: %v, SubType: %v, Unit: %v, Value: %v, Timestamp: %v}", r.HostName, r.Path, r.Tags, r.Kind, r.SubType, r.Unit, r.Value, r.Timestamp)
}
type RecordWriter interface {
	Write(records []Record) error
}
type LimitedRecordWriter interface {
	RecordWriter
	IsTypeSupported(kind types.Type) bool
}
type LimitedBySubTypeRecordWriter interface {
	LimitedRecordWriter
	IsTypeAndSubTypeSupported(kind, subType types.Type) bool
}
type RecordWriterMetrics struct {
	ValuesWritten         uint64
	WriteAttempts         uint64
	SuccessfulWrites      uint64
	LastWriteError        string
	LastWriteErrorTS      time.Time
	LastSuccessfulWriteTS time.Time
	TimeSpentWriting      time.Duration
	Paused                bool
	Disabled              bool
}
func (w *RecordWriterMetrics) SuccessfulWriteRatio() float64 {
	if w.WriteAttempts == 0 {
		return 0.0
	}
	return float64(w.SuccessfulWrites) / float64(w.WriteAttempts)
}
type RecordWriterWithMetrics struct {
	W RecordWriter
	PerMetricWriteTimes *tricorder.CumulativeDistribution
	BatchSizes      *tricorder.CumulativeDistribution
	Logger          log.Logger
	criticalSection *gate.Gate
	lock            sync.Mutex
	metrics         RecordWriterMetrics
}
func NewRecordWriterWithMetrics(writer RecordWriter) *RecordWriterWithMetrics {
	return &RecordWriterWithMetrics{
		W:               writer,
		Logger:          nulllogger.New(),
		criticalSection: gate.New(),
	}
}
func (w *RecordWriterWithMetrics) SetMetrics(m *RecordWriterMetrics) {
	w.setMetrics(m)
}
func (w *RecordWriterWithMetrics) Write(records []Record) error {
	return w.write(records)
}
func (w *RecordWriterWithMetrics) Pause() {
	w.criticalSection.Pause()
	w.setPauseMetric(true)
}
func (w *RecordWriterWithMetrics) Resume() {
	w.setPauseMetric(false)
	w.criticalSection.Resume()
}
func (w *RecordWriterWithMetrics) Disable() {
	w.criticalSection.End()
	w.setDisabledMetric(true)
}
func (w *RecordWriterWithMetrics) Metrics(m *RecordWriterMetrics) {
	w._metrics(m)
}
type AsyncConsumer struct {
	requests     chan consumerRequestType
	flushBarrier *barrier
	concurrency  uint
}
func NewAsyncConsumer(
	w RecordWriter, bufferSize, concurrency uint) *AsyncConsumer {
	return newAsyncConsumer(w, bufferSize, concurrency)
}
func (a *AsyncConsumer) WriteAsync(
	n store.NamedIterator, host string, tagGroup TagGroup) {
	a.writeAsync(n, host, tagGroup)
}
func (a *AsyncConsumer) Flush() {
	a.flush()
}
type Consumer struct {
	w             RecordWriter
	buffer        []Record
	toBeCommitted []store.NamedIterator
	idx           int
}
func NewConsumer(w RecordWriter, bufferSize uint) *Consumer {
	return newConsumer(w, bufferSize)
}
func (c *Consumer) Write(
	n store.NamedIterator, host string, tagGroup TagGroup) error {
	return c.write(n, host, tagGroup)
}
func (c *Consumer) Flush() error {
	return c.flush()
}
type ConsumerMetrics struct {
	RecordWriterMetrics
	TotalValues uint64
	SkippedValues uint64
}
func (c *ConsumerMetrics) ZeroValuesNotWritten() {
	c.SkippedValues = 0
	c.TotalValues = c.ValuesWritten
}
func (c *ConsumerMetrics) ValuesNotWritten() uint64 {
	if c.TotalValues <= c.ValuesWritten+c.SkippedValues {
		return 0
	}
	return c.TotalValues - c.SkippedValues - c.ValuesWritten
}
type ConsumerMetricsStore struct {
	w                  *RecordWriterWithMetrics
	filterer           store.TypeFilterer
	lock               sync.Mutex
	recordCount        uint64
	removedRecordCount uint64
}
func (s *ConsumerMetricsStore) Pause() {
	s.w.Pause()
}
func (s *ConsumerMetricsStore) Resume() {
	s.w.Resume()
}
func (s *ConsumerMetricsStore) DisableWrites() {
	s.w.Disable()
}
func (s *ConsumerMetricsStore) AddToRecordCount(count uint64) {
	s.addToRecordCount(count)
}
func (s *ConsumerMetricsStore) RemoveFromRecordCount(count uint64) {
	s.removeFromRecordCount(count)
}
func (s *ConsumerMetricsStore) Filter(r *store.Record) bool {
	return s.filterer.Filter(r)
}
func (s *ConsumerMetricsStore) Metrics(m *ConsumerMetrics) {
	s.metrics(m)
}
func (s *ConsumerMetricsStore) SetMetrics(m *ConsumerMetrics) {
	s.setMetrics(m)
}
type ConsumerAttributes struct {
	Concurrency uint
	BatchSize uint
	RecordsPerSecond uint
	RollUpSpan time.Duration
	MetricsToExclude    []*regexp.Regexp
	BatchSizes          *tricorder.CumulativeDistribution
	PerMetricWriteTimes *tricorder.CumulativeDistribution
}
func (c *ConsumerAttributes) WritesSameAs(other *ConsumerAttributes) bool {
	if c.RollUpSpan != other.RollUpSpan {
		return false
	}
	if len(c.MetricsToExclude) != len(other.MetricsToExclude) {
		return false
	}
	for i := range c.MetricsToExclude {
		if c.MetricsToExclude[i].String() != other.MetricsToExclude[i].String() {
			return false
		}
	}
	return true
}
func (c *ConsumerAttributes) TotalRecordsPerSecond() uint {
	return c.RecordsPerSecond * c.Concurrency
}
type ConsumerWithMetrics struct {
	name         string
	attributes   ConsumerAttributes
	metricsStore *ConsumerMetricsStore
	consumer     consumerType
}
func (c *ConsumerWithMetrics) Name() string {
	return c.name
}
func (c *ConsumerWithMetrics) Attributes(a *ConsumerAttributes) {
	*a = c.attributes
}
func (c *ConsumerWithMetrics) MetricsStore() *ConsumerMetricsStore {
	return c.metricsStore
}
func (c *ConsumerWithMetrics) Write(
	n store.NamedIterator, host string, tagGroup TagGroup) {
	c.consumer.Write(
		store.NamedIteratorFilter(n, c.metricsStore.filterer),
		host,
		tagGroup)
}
func (c *ConsumerWithMetrics) Flush() {
	c.consumer.Flush()
}
type RecordWriteHooker interface {
	WriteHook(records []Record, result error)
}
type ConsumerWithMetricsBuilder struct {
	c       *ConsumerWithMetrics
	hooks   []RecordWriteHooker
	metrics ConsumerMetrics
	filter  func(*store.MetricInfo) bool
	paused  bool
	logger  log.Logger
}
func NewConsumerWithMetricsBuilder(
	w LimitedRecordWriter) *ConsumerWithMetricsBuilder {
	return newConsumerWithMetricsBuilder(w)
}
func (b *ConsumerWithMetricsBuilder) SetLogger(logger log.Logger) {
	b.logger = logger
}
func (b *ConsumerWithMetricsBuilder) SetConsumerMetrics(m *ConsumerMetrics) {
	b.metrics = *m
}
func (b *ConsumerWithMetricsBuilder) AddHook(hook RecordWriteHooker) {
	if hook == nil {
		panic("nil RecordWriteHooker")
	}
	b.hooks = append(b.hooks, hook)
}
func (b *ConsumerWithMetricsBuilder) SetRecordsPerSecond(
	recordsPerSecond uint) {
	b.c.attributes.RecordsPerSecond = recordsPerSecond
}
func (b *ConsumerWithMetricsBuilder) SetBufferSize(size uint) {
	if size < 1 {
		panic("non-zero size required.")
	}
	b.c.attributes.BatchSize = size
}
func (b *ConsumerWithMetricsBuilder) SetConcurrency(concurrency uint) {
	if concurrency < 1 {
		panic("non-zero concurrency required.")
	}
	b.c.attributes.Concurrency = concurrency
}