func (c varbitChunk) firstTimeDelta() model.Time {
	// Only the first 3 bytes are actually the timestamp, so get rid of the
	// last one by bitshifting.
	return model.Time(c[varbitFirstTimeDeltaOffset+2]) |
		model.Time(c[varbitFirstTimeDeltaOffset+1])<<8 |
		model.Time(c[varbitFirstTimeDeltaOffset])<<16
}

// UnmarshalBinary implements encoding.BinaryUnmarshaler.
func (tr *TimeRange) UnmarshalBinary(buf []byte) error {
	r := bytes.NewReader(buf)
	first, err := binary.ReadVarint(r)
	if err != nil {
		return err
	}
	last, err := binary.ReadVarint(r)
	if err != nil {
		return err
	}
	tr.First = model.Time(first)
	tr.Last = model.Time(last)
	return nil
}

func (c doubleDeltaEncodedChunk) baseTime() model.Time {
	return model.Time(
		binary.LittleEndian.Uint64(
			c[doubleDeltaHeaderBaseTimeOffset:],
		),
	)
}

func benchmarkRangeValues(b *testing.B, encoding chunkEncoding) {
	samples := make(model.Samples, 10000)
	for i := range samples {
		samples[i] = &model.Sample{
			Timestamp: model.Time(2 * i),
			Value:     model.SampleValue(float64(i) * 0.2),
		}
	}
	s, closer := NewTestStorage(b, encoding)
	defer closer.Close()

	for _, sample := range samples {
		s.Append(sample)
	}
	s.WaitForIndexing()

	fp := model.Metric{}.FastFingerprint()

	_, it := s.preloadChunksForRange(fp, model.Earliest, model.Latest)

	b.ResetTimer()

	for i := 0; i < b.N; i++ {
		for _, sample := range samples {
			actual := it.RangeValues(metric.Interval{
				OldestInclusive: sample.Timestamp - 20,
				NewestInclusive: sample.Timestamp + 20,
			})

			if len(actual) < 10 {
				b.Fatalf("not enough samples found")
			}
		}
	}
}

func buildTestChunks(t *testing.T, encoding chunk.Encoding) map[model.Fingerprint][]chunk.Chunk {
	fps := model.Fingerprints{
		m1.FastFingerprint(),
		m2.FastFingerprint(),
		m3.FastFingerprint(),
	}
	fpToChunks := map[model.Fingerprint][]chunk.Chunk{}

	for _, fp := range fps {
		fpToChunks[fp] = make([]chunk.Chunk, 0, 10)
		for i := 0; i < 10; i++ {
			ch, err := chunk.NewForEncoding(encoding)
			if err != nil {
				t.Fatal(err)
			}
			chs, err := ch.Add(model.SamplePair{
				Timestamp: model.Time(i),
				Value:     model.SampleValue(fp),
			})
			if err != nil {
				t.Fatal(err)
			}
			fpToChunks[fp] = append(fpToChunks[fp], chs[0])
		}
	}
	return fpToChunks
}

func TestToReader(t *testing.T) {
	cntVec := prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name:        "name",
			Help:        "docstring",
			ConstLabels: prometheus.Labels{"constname": "constvalue"},
		},
		[]string{"labelname"},
	)
	cntVec.WithLabelValues("val1").Inc()
	cntVec.WithLabelValues("val2").Inc()

	reg := prometheus.NewRegistry()
	reg.MustRegister(cntVec)

	want := `prefix.name.constname.constvalue.labelname.val1 1 1477043
prefix.name.constname.constvalue.labelname.val2 1 1477043
`
	mfs, err := reg.Gather()
	if err != nil {
		t.Fatalf("error: %v", err)
	}

	now := model.Time(1477043083)
	var buf bytes.Buffer
	err = writeMetrics(&buf, mfs, "prefix", now)
	if err != nil {
		t.Fatalf("error: %v", err)
	}

	if got := buf.String(); want != got {
		t.Fatalf("wanted \n%s\n, got \n%s\n", want, got)
	}
}

func (c varbitChunk) lastTime() model.Time {
	return model.Time(
		binary.BigEndian.Uint64(
			c[varbitLastTimeOffset:],
		),
	)
}

func TestSampleDeliveryOrder(t *testing.T) {
	cfg := defaultConfig
	ts := 10
	n := cfg.MaxSamplesPerSend * ts
	// Ensure we don't drop samples in this test.
	cfg.QueueCapacity = n

	samples := make(model.Samples, 0, n)
	for i := 0; i < n; i++ {
		name := model.LabelValue(fmt.Sprintf("test_metric_%d", i%ts))
		samples = append(samples, &model.Sample{
			Metric: model.Metric{
				model.MetricNameLabel: name,
			},
			Value:     model.SampleValue(i),
			Timestamp: model.Time(i),
		})
	}

	c := NewTestStorageClient()
	c.expectSamples(samples)
	m := NewStorageQueueManager(c, &cfg)

	// These should be received by the client.
	for _, s := range samples {
		m.Append(s)
	}
	go m.Run()
	defer m.Stop()

	c.waitForExpectedSamples(t)
}

// timestampAtIndex implements chunkIterator.
func (it *deltaEncodedChunkIterator) timestampAtIndex(idx int) model.Time {
	offset := deltaHeaderBytes + idx*int(it.tBytes+it.vBytes)

	switch it.tBytes {
	case d1:
		return it.baseT + model.Time(uint8(it.c[offset]))
	case d2:
		return it.baseT + model.Time(binary.LittleEndian.Uint16(it.c[offset:]))
	case d4:
		return it.baseT + model.Time(binary.LittleEndian.Uint32(it.c[offset:]))
	case d8:
		// Take absolute value for d8.
		return model.Time(binary.LittleEndian.Uint64(it.c[offset:]))
	default:
		panic("invalid number of bytes for time delta")
	}
}

// timestampAtIndex implements chunkIterator.
func (it *doubleDeltaEncodedChunkIterator) timestampAtIndex(idx int) model.Time {
	if idx == 0 {
		return it.baseT
	}
	if idx == 1 {
		// If time bytes are at d8, the time is saved directly rather
		// than as a difference.
		if it.tBytes == d8 {
			return it.baseΔT
		}
		return it.baseT + it.baseΔT
	}

	offset := doubleDeltaHeaderBytes + (idx-2)*int(it.tBytes+it.vBytes)

	switch it.tBytes {
	case d1:
		return it.baseT +
			model.Time(idx)*it.baseΔT +
			model.Time(int8(it.c[offset]))
	case d2:
		return it.baseT +
			model.Time(idx)*it.baseΔT +
			model.Time(int16(binary.LittleEndian.Uint16(it.c[offset:])))
	case d4:
		return it.baseT +
			model.Time(idx)*it.baseΔT +
			model.Time(int32(binary.LittleEndian.Uint32(it.c[offset:])))
	case d8:
		// Take absolute value for d8.
		return model.Time(binary.LittleEndian.Uint64(it.c[offset:]))
	default:
		panic("invalid number of bytes for time delta")
	}
}

func (acc *deltaEncodedIndexAccessor) timestampAtIndex(idx int) model.Time {
	offset := deltaHeaderBytes + idx*int(acc.tBytes+acc.vBytes)

	switch acc.tBytes {
	case d1:
		return acc.baseT + model.Time(uint8(acc.c[offset]))
	case d2:
		return acc.baseT + model.Time(binary.LittleEndian.Uint16(acc.c[offset:]))
	case d4:
		return acc.baseT + model.Time(binary.LittleEndian.Uint32(acc.c[offset:]))
	case d8:
		// Take absolute value for d8.
		return model.Time(binary.LittleEndian.Uint64(acc.c[offset:]))
	default:
		acc.lastErr = fmt.Errorf("invalid number of bytes for time delta: %d", acc.tBytes)
		return model.Earliest
	}
}

// loadChunkDescs loads the chunkDescs for a series from disk. offsetFromEnd is
// the number of chunkDescs to skip from the end of the series file. It is the
// caller's responsibility to not persist or drop anything for the same
// fingerprint concurrently.
func (p *persistence) loadChunkDescs(fp model.Fingerprint, offsetFromEnd int) ([]*chunkDesc, error) {
	f, err := p.openChunkFileForReading(fp)
	if os.IsNotExist(err) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	defer f.Close()

	fi, err := f.Stat()
	if err != nil {
		return nil, err
	}
	if fi.Size()%int64(chunkLenWithHeader) != 0 {
		// The returned error will bubble up and lead to quarantining of the whole series.
		return nil, fmt.Errorf(
			"size of series file for fingerprint %v is %d, which is not a multiple of the chunk length %d",
			fp, fi.Size(), chunkLenWithHeader,
		)
	}

	numChunks := int(fi.Size())/chunkLenWithHeader - offsetFromEnd
	cds := make([]*chunkDesc, numChunks)
	chunkTimesBuf := make([]byte, 16)
	for i := 0; i < numChunks; i++ {
		_, err := f.Seek(offsetForChunkIndex(i)+chunkHeaderFirstTimeOffset, os.SEEK_SET)
		if err != nil {
			return nil, err
		}

		_, err = io.ReadAtLeast(f, chunkTimesBuf, 16)
		if err != nil {
			return nil, err
		}
		cds[i] = &chunkDesc{
			chunkFirstTime: model.Time(binary.LittleEndian.Uint64(chunkTimesBuf)),
			chunkLastTime:  model.Time(binary.LittleEndian.Uint64(chunkTimesBuf[8:])),
		}
	}
	chunkDescOps.WithLabelValues(load).Add(float64(len(cds)))
	numMemChunkDescs.Add(float64(len(cds)))
	return cds, nil
}

func (c doubleDeltaEncodedChunk) baseTimeDelta() model.Time {
	if len(c) < doubleDeltaHeaderBaseTimeDeltaOffset+8 {
		return 0
	}
	return model.Time(
		binary.LittleEndian.Uint64(
			c[doubleDeltaHeaderBaseTimeDeltaOffset:],
		),
	)
}

func (it *varbitChunkIterator) readDDT() {
	if it.repeats > 0 {
		it.repeats--
	} else {
		switch it.readControlBits(3) {
		case 0:
			it.repeats = byte(it.readBitPattern(7))
		case 1:
			it.dT += model.Time(it.readSignedInt(6))
		case 2:
			it.dT += model.Time(it.readSignedInt(17))
		case 3:
			it.dT += model.Time(it.readSignedInt(23))
		default:
			panic("unexpected number of control bits")
		}
	}
	it.t += it.dT
}

func TestWriteHistogram(t *testing.T) {
	histVec := prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:        "name",
			Help:        "docstring",
			ConstLabels: prometheus.Labels{"constname": "constvalue"},
			Buckets:     []float64{0.01, 0.02, 0.05, 0.1},
		},
		[]string{"labelname"},
	)

	histVec.WithLabelValues("val1").Observe(float64(10))
	histVec.WithLabelValues("val1").Observe(float64(20))
	histVec.WithLabelValues("val1").Observe(float64(30))
	histVec.WithLabelValues("val2").Observe(float64(20))
	histVec.WithLabelValues("val2").Observe(float64(30))
	histVec.WithLabelValues("val2").Observe(float64(40))

	reg := prometheus.NewRegistry()
	reg.MustRegister(histVec)

	mfs, err := reg.Gather()
	if err != nil {
		t.Fatalf("error: %v", err)
	}

	now := model.Time(1477043083)
	var buf bytes.Buffer
	err = writeMetrics(&buf, mfs, "prefix", now)
	if err != nil {
		t.Fatalf("error: %v", err)
	}

	want := `prefix.name_bucket.constname.constvalue.labelname.val1.le.0_01 0 1477043
prefix.name_bucket.constname.constvalue.labelname.val1.le.0_02 0 1477043
prefix.name_bucket.constname.constvalue.labelname.val1.le.0_05 0 1477043
prefix.name_bucket.constname.constvalue.labelname.val1.le.0_1 0 1477043
prefix.name_sum.constname.constvalue.labelname.val1 60 1477043
prefix.name_count.constname.constvalue.labelname.val1 3 1477043
prefix.name_bucket.constname.constvalue.labelname.val1.le._Inf 3 1477043
prefix.name_bucket.constname.constvalue.labelname.val2.le.0_01 0 1477043
prefix.name_bucket.constname.constvalue.labelname.val2.le.0_02 0 1477043
prefix.name_bucket.constname.constvalue.labelname.val2.le.0_05 0 1477043
prefix.name_bucket.constname.constvalue.labelname.val2.le.0_1 0 1477043
prefix.name_sum.constname.constvalue.labelname.val2 90 1477043
prefix.name_count.constname.constvalue.labelname.val2 3 1477043
prefix.name_bucket.constname.constvalue.labelname.val2.le._Inf 3 1477043
`
	if got := buf.String(); want != got {
		t.Fatalf("wanted \n%s\n, got \n%s\n", want, got)
	}
}

func TestWriteSummary(t *testing.T) {
	sumVec := prometheus.NewSummaryVec(
		prometheus.SummaryOpts{
			Name:        "name",
			Help:        "docstring",
			ConstLabels: prometheus.Labels{"constname": "constvalue"},
			Objectives:  map[float64]float64{0.5: 0.05, 0.9: 0.01, 0.99: 0.001},
		},
		[]string{"labelname"},
	)

	sumVec.WithLabelValues("val1").Observe(float64(10))
	sumVec.WithLabelValues("val1").Observe(float64(20))
	sumVec.WithLabelValues("val1").Observe(float64(30))
	sumVec.WithLabelValues("val2").Observe(float64(20))
	sumVec.WithLabelValues("val2").Observe(float64(30))
	sumVec.WithLabelValues("val2").Observe(float64(40))

	reg := prometheus.NewRegistry()
	reg.MustRegister(sumVec)

	mfs, err := reg.Gather()
	if err != nil {
		t.Fatalf("error: %v", err)
	}

	now := model.Time(1477043083)
	var buf bytes.Buffer
	err = writeMetrics(&buf, mfs, "prefix", now)
	if err != nil {
		t.Fatalf("error: %v", err)
	}

	want := `prefix.name.constname.constvalue.labelname.val1.quantile.0_5 20 1477043
prefix.name.constname.constvalue.labelname.val1.quantile.0_9 30 1477043
prefix.name.constname.constvalue.labelname.val1.quantile.0_99 30 1477043
prefix.name_sum.constname.constvalue.labelname.val1 60 1477043
prefix.name_count.constname.constvalue.labelname.val1 3 1477043
prefix.name.constname.constvalue.labelname.val2.quantile.0_5 30 1477043
prefix.name.constname.constvalue.labelname.val2.quantile.0_9 40 1477043
prefix.name.constname.constvalue.labelname.val2.quantile.0_99 40 1477043
prefix.name_sum.constname.constvalue.labelname.val2 90 1477043
prefix.name_count.constname.constvalue.labelname.val2 3 1477043
`

	if got := buf.String(); want != got {
		t.Fatalf("wanted \n%s\n, got \n%s\n", want, got)
	}
}

func TestAppendOutOfOrder(t *testing.T) {
	s, closer := NewTestStorage(t, 1)
	defer closer.Close()

	m := model.Metric{
		model.MetricNameLabel: "out_of_order",
	}

	for i, t := range []int{0, 2, 2, 1} {
		s.Append(&model.Sample{
			Metric:    m,
			Timestamp: model.Time(t),
			Value:     model.SampleValue(i),
		})
	}

	fp, err := s.mapper.mapFP(m.FastFingerprint(), m)
	if err != nil {
		t.Fatal(err)
	}

	pl := s.NewPreloader()
	defer pl.Close()

	err = pl.PreloadRange(fp, 0, 2, 5*time.Minute)
	if err != nil {
		t.Fatalf("Error preloading chunks: %s", err)
	}

	it := s.NewIterator(fp)

	want := []model.SamplePair{
		{
			Timestamp: 0,
			Value:     0,
		},
		{
			Timestamp: 2,
			Value:     1,
		},
	}
	got := it.RangeValues(metric.Interval{OldestInclusive: 0, NewestInclusive: 2})
	if !reflect.DeepEqual(want, got) {
		t.Fatalf("want %v, got %v", want, got)
	}
}

func testChunk(t *testing.T, encoding chunkEncoding) {
	samples := make(model.Samples, 500000)
	for i := range samples {
		samples[i] = &model.Sample{
			Timestamp: model.Time(i),
			Value:     model.SampleValue(float64(i) * 0.2),
		}
	}
	s, closer := NewTestStorage(t, encoding)
	defer closer.Close()

	for _, sample := range samples {
		s.Append(sample)
	}
	s.WaitForIndexing()

	for m := range s.fpToSeries.iter() {
		s.fpLocker.Lock(m.fp)
		defer s.fpLocker.Unlock(m.fp) // TODO remove, see below
		var values []model.SamplePair
		for _, cd := range m.series.chunkDescs {
			if cd.isEvicted() {
				continue
			}
			it := cd.c.newIterator()
			for it.scan() {
				values = append(values, it.value())
			}
			if it.err() != nil {
				t.Error(it.err())
			}
		}

		for i, v := range values {
			if samples[i].Timestamp != v.Timestamp {
				t.Errorf("%d. Got %v; want %v", i, v.Timestamp, samples[i].Timestamp)
			}
			if samples[i].Value != v.Value {
				t.Errorf("%d. Got %v; want %v", i, v.Value, samples[i].Value)
			}
		}
		//s.fpLocker.Unlock(m.fp)
	}
	log.Info("test done, closing")
}

func buildTestChunks(encoding chunkEncoding) map[model.Fingerprint][]chunk {
	fps := model.Fingerprints{
		m1.FastFingerprint(),
		m2.FastFingerprint(),
		m3.FastFingerprint(),
	}
	fpToChunks := map[model.Fingerprint][]chunk{}

	for _, fp := range fps {
		fpToChunks[fp] = make([]chunk, 0, 10)
		for i := 0; i < 10; i++ {
			fpToChunks[fp] = append(fpToChunks[fp], newChunkForEncoding(encoding).add(&model.SamplePair{
				Timestamp: model.Time(i),
				Value:     model.SampleValue(fp),
			})[0])
		}
	}
	return fpToChunks
}

// TestLoop is just a smoke test for the loop method, if we can switch it on and
// off without disaster.
func TestLoop(t *testing.T) {
	if testing.Short() {
		t.Skip("Skipping test in short mode.")
	}
	samples := make(model.Samples, 1000)
	for i := range samples {
		samples[i] = &model.Sample{
			Timestamp: model.Time(2 * i),
			Value:     model.SampleValue(float64(i) * 0.2),
		}
	}
	directory := testutil.NewTemporaryDirectory("test_storage", t)
	defer directory.Close()
	o := &MemorySeriesStorageOptions{
		MemoryChunks:               50,
		MaxChunksToPersist:         1000000,
		PersistenceRetentionPeriod: 24 * 7 * time.Hour,
		PersistenceStoragePath:     directory.Path(),
		CheckpointInterval:         250 * time.Millisecond,
		SyncStrategy:               Adaptive,
		MinShrinkRatio:             0.1,
	}
	storage := NewMemorySeriesStorage(o)
	if err := storage.Start(); err != nil {
		t.Errorf("Error starting storage: %s", err)
	}
	for _, s := range samples {
		storage.Append(s)
	}
	storage.WaitForIndexing()
	series, _ := storage.(*memorySeriesStorage).fpToSeries.get(model.Metric{}.FastFingerprint())
	cdsBefore := len(series.chunkDescs)
	time.Sleep(fpMaxWaitDuration + time.Second) // TODO(beorn7): Ugh, need to wait for maintenance to kick in.
	cdsAfter := len(series.chunkDescs)
	storage.Stop()
	if cdsBefore <= cdsAfter {
		t.Errorf(
			"Number of chunk descriptors should have gone down by now. Got before %d, after %d.",
			cdsBefore, cdsAfter,
		)
	}
}