func (w *Writer) Write(tsb model.TimeSeriesBatch) error {
	t0 := time.Now()
	defer func() {
		w.batchWriteDuration.Observe(time.Since(t0).Seconds())
	}()
	wg := &sync.WaitGroup{}
	errch := make(chan error, 1)  
	wg.Add(len(tsb))
	for _, ts := range tsb {
		wp := &writerPayload{
			wg:    wg,
			ts:    ts,
			errch: errch,
		}
		w.ch <- wp
	}
	wg.Wait()
	select {
	case err := <-errch:
		return err
	default:
		return nil
	}
}
Write implements the ingester.Write interface and allows the
ingester to write TimeSeriesBatch to Cassandra.
package cassandra
import (
	"fmt"
	"sync"
	"time"
	"github.com/digitalocean/vulcan/model"
	"github.com/gocql/gocql"
	"github.com/prometheus/client_golang/prometheus"
)
const (
	namespace = "vulcan"
	subsystem = "cassandra"
)
var writeTTLSampleCQL = `UPDATE %s USING TTL ? SET value = ? WHERE fqmn = ? AND at = ?`
type Writer struct {
	prometheus.Collector
	writeTTLSampleCQL string
	s          *gocql.Session
	ch         chan *writerPayload
	ttlSeconds int64
	batchWriteDuration  prometheus.Histogram
	sampleWriteDuration prometheus.Histogram
	workerCount         *prometheus.GaugeVec
}
func (w *Writer) Describe(ch chan<- *prometheus.Desc) {
	w.batchWriteDuration.Describe(ch)
	w.sampleWriteDuration.Describe(ch)
	w.workerCount.Describe(ch)
}
func (w *Writer) Collect(ch chan<- prometheus.Metric) {
	w.batchWriteDuration.Collect(ch)
	w.sampleWriteDuration.Collect(ch)
	w.workerCount.Collect(ch)
}
type writerPayload struct {
	wg    *sync.WaitGroup
	ts    *model.TimeSeries
	errch chan error
}
type WriterConfig struct {
	NumWorkers int
	Session    *gocql.Session
	TTL        time.Duration
	TableName  string
	Keyspace   string
}
func NewWriter(config *WriterConfig) *Writer {
	w := &Writer{
		s: config.Session,
		writeTTLSampleCQL: fmt.Sprintf(
			writeTTLSampleCQL,
			fmt.Sprintf("%s.%s", config.Keyspace, config.TableName),
		),
		ch:         make(chan *writerPayload),
		ttlSeconds: int64(config.TTL.Seconds()),
		batchWriteDuration: prometheus.NewHistogram(
			prometheus.HistogramOpts{
				Namespace: namespace,
				Subsystem: subsystem,
				Name:      "batch_write_duration_seconds",
				Help:      "Histogram of seconds elapsed to write a batch.",
				Buckets:   prometheus.DefBuckets,
			},
		),
		sampleWriteDuration: prometheus.NewHistogram(
			prometheus.HistogramOpts{
				Namespace: namespace,
				Subsystem: subsystem,
				Name:      "sample_write_duration_seconds",
				Help:      "Histogram of seconds elapsed to write a sample.",
				Buckets:   prometheus.DefBuckets,
			},
		),
		workerCount: prometheus.NewGaugeVec(
			prometheus.GaugeOpts{
				Namespace: namespace,
				Subsystem: subsystem,
				Name:      "worker_count",
				Help:      "Count of workers.",
			},
			[]string{"mode"},
		),
	}
	for n := 0; n < config.NumWorkers; n++ {
		go w.worker()
	}
	return w
}