func getKey(ts *remote.TimeSeries) (string, error) {
	var (
		jobName, inst string
		instFound     bool
	)
	for _, l := range ts.Labels {
		if jobName != "" && instFound {
			break
		}
		switch l.Name {
		case model.JobLabel:
			jobName = l.Value
		case model.InstanceLabel:
			inst = l.Value
			instFound = true
		case model.AddressLabel:
			if inst == "" {
				inst = l.Value
			}
		}
	}
	if jobName == "" {
		return "", errors.New("missing job label")
	}
	if inst == "" {
		return "", errors.New("missing instance label and address label")
	}
	return fmt.Sprintf("%s-%s", jobName, inst), nil
}
getKey formulates the instance key based on Prometheus metric labels.
package forwarder
import (
	"fmt"
	"sync"
	"time"
	"golang.org/x/net/context"
	"github.com/digitalocean/vulcan/bus"
	log "github.com/Sirupsen/logrus"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/common/model"
	"github.com/prometheus/prometheus/storage/remote"
)
const (
	namespace = "vulcan"
	subsystem = "forwarder"
)
var ErrServiceUnavailable = errors.New("Forwarder Service Unavailable")
type Forwarder struct {
	writer bus.Writer
	wg   *sync.WaitGroup
	once *sync.Once
	done chan struct{}
	writeBatchDuration prometheus.Histogram
	reqTimeseriesCount prometheus.Histogram
}
type Config struct {
	Writer bus.Writer
}
func NewForwarder(config *Config) *Forwarder {
	f := &Forwarder{
		writer: config.Writer,
		wg:   new(sync.WaitGroup),
		once: new(sync.Once),
		done: make(chan struct{}),
		writeBatchDuration: prometheus.NewHistogram(
			prometheus.HistogramOpts{
				Namespace: namespace,
				Subsystem: subsystem,
				Name:      "write_batch_duration_seconds",
				Help:      "Duration of write batch calls.",
				Buckets:   prometheus.DefBuckets,
			},
		),
		reqTimeseriesCount: prometheus.NewHistogram(
			prometheus.HistogramOpts{
				Namespace: namespace,
				Subsystem: subsystem,
				Name:      "request_timeseries_length",
				Help:      "Number of timeseries objects in the write request.",
				Buckets:   []float64{0, 10, 20, 30, 40, 50, 60, 70, 80, 90, 100},
			},
		),
	}
	f.wg.Add(1)
	return f
}
func (f *Forwarder) Describe(ch chan<- *prometheus.Desc) {
	f.writeBatchDuration.Describe(ch)
	f.reqTimeseriesCount.Describe(ch)
}
func (f *Forwarder) Collect(ch chan<- prometheus.Metric) {
	f.writeBatchDuration.Collect(ch)
	f.reqTimeseriesCount.Collect(ch)
}
func (f *Forwarder) Write(ctx context.Context, req *remote.WriteRequest) error {
	var (
		toWrite = map[string]*remote.WriteRequest{}
		ll      = log.WithFields(log.Fields{"source": "forwarder.Write"})
	)
	select {
	case <-f.done:
		return ErrServiceUnavailable
	default:
	}
	for _, ts := range req.Timeseries {
		key, err := getKey(ts)
		if err != nil {
			ll.WithFields(log.Fields{
				"timeseries": ts,
			}).WithError(err).Error("could not formulate key from labels")
			continue
		}
		if _, ok := toWrite[key]; !ok {
			toWrite[key] = &remote.WriteRequest{
				Timeseries: []*remote.TimeSeries{ts},
			}
		} else {
			toWrite[key].Timeseries = append(toWrite[key].Timeseries, ts)
		}
	}
	for key, wr := range toWrite {
		f.wg.Add(1)
		go func(key string, wr *remote.WriteRequest) {
			defer f.wg.Done()
			f.reqTimeseriesCount.Observe(float64(len(wr.Timeseries)))
			t0 := time.Now()
			if err := f.writer.Write(key, wr); err != nil {
				ll.WithError(err).Error("failed to write to bus")
			}
			f.writeBatchDuration.Observe(time.Since(t0).Seconds())
		}(key, wr)
	}
	return nil
}