Use custom timestamp type for sample timestamps and related code.

So far we've been using Go's native time.Time for anything related to sample
timestamps. Since the range of time.Time is much bigger than what we need, this
has created two problems:

- there could be time.Time values which were out of the range/precision of the
  time type that we persist to disk, therefore causing incorrectly ordered keys.
  One bug caused by this was:

  #367

  It would be good to use a timestamp type that's more closely aligned with
  what the underlying storage supports.

- sizeof(time.Time) is 192, while Prometheus should be ok with a single 64-bit
  Unix timestamp (possibly even a 32-bit one). Since we store samples in large
  numbers, this seriously affects memory usage. Furthermore, copying/working
  with the data will be faster if it's smaller.

*MEMORY USAGE RESULTS*
Initial memory usage comparisons for a running Prometheus with 1 timeseries and
100,000 samples show roughly a 13% decrease in total (VIRT) memory usage. In my
tests, this advantage for some reason decreased a bit the more samples the
timeseries had (to 5-7% for millions of samples). This I can't fully explain,
but perhaps garbage collection issues were involved.

*WHEN TO USE THE NEW TIMESTAMP TYPE*
The new clientmodel.Timestamp type should be used whenever time
calculations are either directly or indirectly related to sample
timestamps.

For example:
- the timestamp of a sample itself
- all kinds of watermarks
- anything that may become or is compared to a sample timestamp (like the timestamp
  passed into Target.Scrape()).

When to still use time.Time:
- for measuring durations/times not related to sample timestamps, like duration
  telemetry exporting, timers that indicate how frequently to execute some
  action, etc.

*NOTE ON OPERATOR OPTIMIZATION TESTS*
We don't use operator optimization code anymore, but it still lives in
the code as dead code. It still has tests, but I couldn't get all of them to
pass with the new timestamp format. I commented out the failing cases for now,
but we should probably remove the dead code soon. I just didn't want to do that
in the same change as this.

Change-Id: I821787414b0debe85c9fffaeb57abd453727af0f

coding/indexable/time.go

@@ -15,18 +15,19 @@ package indexable
 
 import (
 	"encoding/binary"
-	"time"
+
+	clientmodel "github.com/prometheus/client_golang/model"
 )
 
 // EncodeTimeInto writes the provided time into the specified buffer subject
 // to the LevelDB big endian key sort order requirement.
-func EncodeTimeInto(dst []byte, t time.Time) {
+func EncodeTimeInto(dst []byte, t clientmodel.Timestamp) {
 	binary.BigEndian.PutUint64(dst, uint64(t.Unix()))
 }
 
 // EncodeTime converts the provided time into a byte buffer subject to the
 // LevelDB big endian key sort order requirement.
-func EncodeTime(t time.Time) []byte {
+func EncodeTime(t clientmodel.Timestamp) []byte {
 	buffer := make([]byte, 8)
 
 	EncodeTimeInto(buffer, t)
@@ -36,6 +37,6 @@ func EncodeTime(t time.Time) []byte {
 
 // DecodeTime deserializes a big endian byte array into a Unix time in UTC,
 // omitting granularity precision less than a second.
-func DecodeTime(src []byte) time.Time {
-	return time.Unix(int64(binary.BigEndian.Uint64(src)), 0).UTC()
+func DecodeTime(src []byte) clientmodel.Timestamp {
+	return clientmodel.TimestampFromUnix(int64(binary.BigEndian.Uint64(src)))
 }

coding/indexable/time_test.go

@@ -17,14 +17,15 @@ import (
 	"math/rand"
 	"testing"
 	"testing/quick"
-	"time"
+
+	clientmodel "github.com/prometheus/client_golang/model"
 )
 
 func TestTimeEndToEnd(t *testing.T) {
 	tester := func(x int) bool {
 		random := rand.New(rand.NewSource(int64(x)))
 		buffer := make([]byte, 8)
-		incoming := time.Unix(random.Int63(), 0)
+		incoming := clientmodel.TimestampFromUnix(random.Int63())
 
 		EncodeTimeInto(buffer, incoming)
 		outgoing := DecodeTime(buffer)

main.go

@@ -128,7 +128,7 @@ func (p *prometheus) compact(olderThan time.Duration, groupSize int) error {
 	})
 	defer curator.Close()
 
-	return curator.Run(olderThan, time.Now(), processor, p.storage.DiskStorage.CurationRemarks, p.storage.DiskStorage.MetricSamples, p.storage.DiskStorage.MetricHighWatermarks, p.curationState)
+	return curator.Run(olderThan, clientmodel.Now(), processor, p.storage.DiskStorage.CurationRemarks, p.storage.DiskStorage.MetricSamples, p.storage.DiskStorage.MetricHighWatermarks, p.curationState)
 }
 
 func (p *prometheus) delete(olderThan time.Duration, batchSize int) error {
@@ -152,7 +152,7 @@ func (p *prometheus) delete(olderThan time.Duration, batchSize int) error {
 	})
 	defer curator.Close()
 
-	return curator.Run(olderThan, time.Now(), processor, p.storage.DiskStorage.CurationRemarks, p.storage.DiskStorage.MetricSamples, p.storage.DiskStorage.MetricHighWatermarks, p.curationState)
+	return curator.Run(olderThan, clientmodel.Now(), processor, p.storage.DiskStorage.CurationRemarks, p.storage.DiskStorage.MetricSamples, p.storage.DiskStorage.MetricHighWatermarks, p.curationState)
 }
 
 func (p *prometheus) close() {

retrieval/target.go

@@ -156,7 +156,7 @@ func NewTarget(address string, deadline time.Duration, baseLabels clientmodel.La
 	return target
 }
 
-func (t *target) recordScrapeHealth(ingester extraction.Ingester, timestamp time.Time, healthy bool) {
+func (t *target) recordScrapeHealth(ingester extraction.Ingester, timestamp clientmodel.Timestamp, healthy bool) {
 	metric := clientmodel.Metric{}
 	for label, value := range t.baseLabels {
 		metric[label] = value
@@ -182,7 +182,7 @@ func (t *target) recordScrapeHealth(ingester extraction.Ingester, timestamp time
 }
 
 func (t *target) Scrape(earliest time.Time, ingester extraction.Ingester) error {
-	now := time.Now()
+	now := clientmodel.Now()
 	futureState := t.state
 	err := t.scrape(now, ingester)
 	if err != nil {
@@ -202,7 +202,7 @@ func (t *target) Scrape(earliest time.Time, ingester extraction.Ingester) error
 
 const acceptHeader = `application/vnd.google.protobuf;proto=io.prometheus.client.MetricFamily;encoding=delimited;q=0.7,application/json;schema=prometheus/telemetry;version=0.0.2;q=0.2,*/*;q=0.1`
 
-func (t *target) scrape(timestamp time.Time, ingester extraction.Ingester) (err error) {
+func (t *target) scrape(timestamp clientmodel.Timestamp, ingester extraction.Ingester) (err error) {
 	defer func(start time.Time) {
 		ms := float64(time.Since(start)) / float64(time.Millisecond)
 		labels := map[string]string{address: t.Address(), outcome: success}

retrieval/target_test.go

@@ -56,7 +56,7 @@ func TestTargetRecordScrapeHealth(t *testing.T) {
 		httpClient: utility.NewDeadlineClient(0),
 	}
 
-	now := time.Now()
+	now := clientmodel.Now()
 	ingester := &collectResultIngester{}
 	testTarget.recordScrapeHealth(ingester, now, true)
 

rules/alerting.go

@@ -68,13 +68,13 @@ type Alert struct {
 	// The state of the alert (PENDING or FIRING).
 	State AlertState
 	// The time when the alert first transitioned into PENDING state.
-	ActiveSince time.Time
+	ActiveSince clientmodel.Timestamp
 	// The value of the alert expression for this vector element.
 	Value clientmodel.SampleValue
 }
 
 // sample returns a Sample suitable for recording the alert.
-func (a Alert) sample(timestamp time.Time, value clientmodel.SampleValue) *clientmodel.Sample {
+func (a Alert) sample(timestamp clientmodel.Timestamp, value clientmodel.SampleValue) *clientmodel.Sample {
 	recordedMetric := clientmodel.Metric{}
 	for label, value := range a.Labels {
 		recordedMetric[label] = value
@@ -118,11 +118,11 @@ func (rule *AlertingRule) Name() string {
 	return rule.name
 }
 
-func (rule *AlertingRule) EvalRaw(timestamp time.Time, storage *metric.TieredStorage) (ast.Vector, error) {
+func (rule *AlertingRule) EvalRaw(timestamp clientmodel.Timestamp, storage *metric.TieredStorage) (ast.Vector, error) {
 	return ast.EvalVectorInstant(rule.vector, timestamp, storage, stats.NewTimerGroup())
 }
 
-func (rule *AlertingRule) Eval(timestamp time.Time, storage *metric.TieredStorage) (ast.Vector, error) {
+func (rule *AlertingRule) Eval(timestamp clientmodel.Timestamp, storage *metric.TieredStorage) (ast.Vector, error) {
 	// Get the raw value of the rule expression.
 	exprResult, err := rule.EvalRaw(timestamp, storage)
 	if err != nil {

rules/ast/ast.go

@@ -103,23 +103,23 @@ type Node interface {
 // interface represents the type returned to the parent node.
 type ScalarNode interface {
 	Node
-	Eval(timestamp time.Time, view *viewAdapter) clientmodel.SampleValue
+	Eval(timestamp clientmodel.Timestamp, view *viewAdapter) clientmodel.SampleValue
 }
 
 type VectorNode interface {
 	Node
-	Eval(timestamp time.Time, view *viewAdapter) Vector
+	Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Vector
 }
 
 type MatrixNode interface {
 	Node
-	Eval(timestamp time.Time, view *viewAdapter) Matrix
-	EvalBoundaries(timestamp time.Time, view *viewAdapter) Matrix
+	Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Matrix
+	EvalBoundaries(timestamp clientmodel.Timestamp, view *viewAdapter) Matrix
 }
 
 type StringNode interface {
 	Node
-	Eval(timestamp time.Time, view *viewAdapter) string
+	Eval(timestamp clientmodel.Timestamp, view *viewAdapter) string
 }
 
 // ----------------------------------------------------------------------------
@@ -233,17 +233,17 @@ func (node MatrixLiteral) Children() Nodes      { return Nodes{} }
 func (node StringLiteral) Children() Nodes      { return Nodes{} }
 func (node StringFunctionCall) Children() Nodes { return node.args }
 
-func (node *ScalarLiteral) Eval(timestamp time.Time, view *viewAdapter) clientmodel.SampleValue {
+func (node *ScalarLiteral) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) clientmodel.SampleValue {
 	return node.value
 }
 
-func (node *ScalarArithExpr) Eval(timestamp time.Time, view *viewAdapter) clientmodel.SampleValue {
+func (node *ScalarArithExpr) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) clientmodel.SampleValue {
 	lhs := node.lhs.Eval(timestamp, view)
 	rhs := node.rhs.Eval(timestamp, view)
 	return evalScalarBinop(node.opType, lhs, rhs)
 }
 
-func (node *ScalarFunctionCall) Eval(timestamp time.Time, view *viewAdapter) clientmodel.SampleValue {
+func (node *ScalarFunctionCall) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) clientmodel.SampleValue {
 	return node.function.callFn(timestamp, view, node.args).(clientmodel.SampleValue)
 }
 
@@ -277,7 +277,7 @@ func labelsToKey(labels clientmodel.Metric) uint64 {
 	return summer.Sum64()
 }
 
-func EvalVectorInstant(node VectorNode, timestamp time.Time, storage *metric.TieredStorage, queryStats *stats.TimerGroup) (vector Vector, err error) {
+func EvalVectorInstant(node VectorNode, timestamp clientmodel.Timestamp, storage *metric.TieredStorage, queryStats *stats.TimerGroup) (vector Vector, err error) {
 	viewAdapter, err := viewAdapterForInstantQuery(node, timestamp, storage, queryStats)
 	if err != nil {
 		return
@@ -286,7 +286,7 @@ func EvalVectorInstant(node VectorNode, timestamp time.Time, storage *metric.Tie
 	return
 }
 
-func EvalVectorRange(node VectorNode, start time.Time, end time.Time, interval time.Duration, storage *metric.TieredStorage, queryStats *stats.TimerGroup) (Matrix, error) {
+func EvalVectorRange(node VectorNode, start clientmodel.Timestamp, end clientmodel.Timestamp, interval time.Duration, storage *metric.TieredStorage, queryStats *stats.TimerGroup) (Matrix, error) {
 	// Explicitly initialize to an empty matrix since a nil Matrix encodes to
 	// null in JSON.
 	matrix := Matrix{}
@@ -340,7 +340,7 @@ func labelIntersection(metric1, metric2 clientmodel.Metric) clientmodel.Metric {
 	return intersection
 }
 
-func (node *VectorAggregation) groupedAggregationsToVector(aggregations map[uint64]*groupedAggregation, timestamp time.Time) Vector {
+func (node *VectorAggregation) groupedAggregationsToVector(aggregations map[uint64]*groupedAggregation, timestamp clientmodel.Timestamp) Vector {
 	vector := Vector{}
 	for _, aggregation := range aggregations {
 		switch node.aggrType {
@@ -361,7 +361,7 @@ func (node *VectorAggregation) groupedAggregationsToVector(aggregations map[uint
 	return vector
 }
 
-func (node *VectorAggregation) Eval(timestamp time.Time, view *viewAdapter) Vector {
+func (node *VectorAggregation) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Vector {
 	vector := node.vector.Eval(timestamp, view)
 	result := map[uint64]*groupedAggregation{}
 	for _, sample := range vector {
@@ -399,7 +399,7 @@ func (node *VectorAggregation) Eval(timestamp time.Time, view *viewAdapter) Vect
 	return node.groupedAggregationsToVector(result, timestamp)
 }
 
-func (node *VectorLiteral) Eval(timestamp time.Time, view *viewAdapter) Vector {
+func (node *VectorLiteral) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Vector {
 	values, err := view.GetValueAtTime(node.fingerprints, timestamp)
 	if err != nil {
 		glog.Error("Unable to get vector values: ", err)
@@ -408,7 +408,7 @@ func (node *VectorLiteral) Eval(timestamp time.Time, view *viewAdapter) Vector {
 	return values
 }
 
-func (node *VectorFunctionCall) Eval(timestamp time.Time, view *viewAdapter) Vector {
+func (node *VectorFunctionCall) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Vector {
 	return node.function.callFn(timestamp, view, node.args).(Vector)
 }
 
@@ -552,7 +552,7 @@ func labelsEqual(labels1, labels2 clientmodel.Metric) bool {
 	return true
 }
 
-func (node *VectorArithExpr) Eval(timestamp time.Time, view *viewAdapter) Vector {
+func (node *VectorArithExpr) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Vector {
 	lhs := node.lhs.Eval(timestamp, view)
 	result := Vector{}
 	if node.rhs.Type() == SCALAR {
@@ -583,7 +583,7 @@ func (node *VectorArithExpr) Eval(timestamp time.Time, view *viewAdapter) Vector
 	panic("Invalid vector arithmetic expression operands")
 }
 
-func (node *MatrixLiteral) Eval(timestamp time.Time, view *viewAdapter) Matrix {
+func (node *MatrixLiteral) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Matrix {
 	interval := &metric.Interval{
 		OldestInclusive: timestamp.Add(-node.interval),
 		NewestInclusive: timestamp,
@@ -596,7 +596,7 @@ func (node *MatrixLiteral) Eval(timestamp time.Time, view *viewAdapter) Matrix {
 	return values
 }
 
-func (node *MatrixLiteral) EvalBoundaries(timestamp time.Time, view *viewAdapter) Matrix {
+func (node *MatrixLiteral) EvalBoundaries(timestamp clientmodel.Timestamp, view *viewAdapter) Matrix {
 	interval := &metric.Interval{
 		OldestInclusive: timestamp.Add(-node.interval),
 		NewestInclusive: timestamp,
@@ -621,11 +621,11 @@ func (matrix Matrix) Swap(i, j int) {
 	matrix[i], matrix[j] = matrix[j], matrix[i]
 }
 
-func (node *StringLiteral) Eval(timestamp time.Time, view *viewAdapter) string {
+func (node *StringLiteral) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) string {
 	return node.str
 }
 
-func (node *StringFunctionCall) Eval(timestamp time.Time, view *viewAdapter) string {
+func (node *StringFunctionCall) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) string {
 	return node.function.callFn(timestamp, view, node.args).(string)
 }
 

rules/ast/functions.go

@@ -29,7 +29,7 @@ type Function struct {
 	name       string
 	argTypes   []ExprType
 	returnType ExprType
-	callFn     func(timestamp time.Time, view *viewAdapter, args []Node) interface{}
+	callFn     func(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{}
 }
 
 func (function *Function) CheckArgTypes(args []Node) error {
@@ -68,12 +68,12 @@ func (function *Function) CheckArgTypes(args []Node) error {
 }
 
 // === time() clientmodel.SampleValue ===
-func timeImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
+func timeImpl(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{} {
 	return clientmodel.SampleValue(time.Now().Unix())
 }
 
 // === delta(matrix MatrixNode, isCounter ScalarNode) Vector ===
-func deltaImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
+func deltaImpl(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{} {
 	matrixNode := args[0].(MatrixNode)
 	isCounter := args[1].(ScalarNode).Eval(timestamp, view) > 0
 	resultVector := Vector{}
@@ -133,7 +133,7 @@ func deltaImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{}
 }
 
 // === rate(node *MatrixNode) Vector ===
-func rateImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
+func rateImpl(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{} {
 	args = append(args, &ScalarLiteral{value: 1})
 	vector := deltaImpl(timestamp, view, args).(Vector)
 
@@ -164,7 +164,7 @@ func (sorter vectorByValueSorter) Swap(i, j int) {
 }
 
 // === sort(node *VectorNode) Vector ===
-func sortImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
+func sortImpl(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{} {
 	byValueSorter := vectorByValueSorter{
 		vector: args[0].(VectorNode).Eval(timestamp, view),
 	}
@@ -173,7 +173,7 @@ func sortImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
 }
 
 // === sortDesc(node *VectorNode) Vector ===
-func sortDescImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
+func sortDescImpl(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{} {
 	descByValueSorter := utility.ReverseSorter{
 		vectorByValueSorter{
 			vector: args[0].(VectorNode).Eval(timestamp, view),
@@ -184,7 +184,7 @@ func sortDescImpl(timestamp time.Time, view *viewAdapter, args []Node) interface
 }
 
 // === sampleVectorImpl() Vector ===
-func sampleVectorImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
+func sampleVectorImpl(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{} {
 	return Vector{
 		&clientmodel.Sample{
 			Metric: clientmodel.Metric{
@@ -257,7 +257,7 @@ func sampleVectorImpl(timestamp time.Time, view *viewAdapter, args []Node) inter
 }
 
 // === scalar(node *VectorNode) Scalar ===
-func scalarImpl(timestamp time.Time, view *viewAdapter, args []Node) interface{} {
+func scalarImpl(timestamp clientmodel.Timestamp, view *viewAdapter, args []Node) interface{} {
 	v := args[0].(VectorNode).Eval(timestamp, view)
 	if len(v) != 1 {
 		return clientmodel.SampleValue(math.NaN())

rules/ast/functions_test.go

@@ -15,7 +15,6 @@ package ast
 
 import (
 	"testing"
-	"time"
 
 	clientmodel "github.com/prometheus/client_golang/model"
 
@@ -29,7 +28,7 @@ func (node emptyRangeNode) NodeTreeToDotGraph() string { return "" }
 func (node emptyRangeNode) String() string             { return "" }
 func (node emptyRangeNode) Children() Nodes            { return Nodes{} }
 
-func (node emptyRangeNode) Eval(timestamp time.Time, view *viewAdapter) Matrix {
+func (node emptyRangeNode) Eval(timestamp clientmodel.Timestamp, view *viewAdapter) Matrix {
 	return Matrix{
 		metric.SampleSet{
 			Metric: clientmodel.Metric{clientmodel.MetricNameLabel: "empty_metric"},
@@ -38,7 +37,7 @@ func (node emptyRangeNode) Eval(timestamp time.Time, view *viewAdapter) Matrix {
 	}
 }
 
-func (node emptyRangeNode) EvalBoundaries(timestamp time.Time, view *viewAdapter) Matrix {
+func (node emptyRangeNode) EvalBoundaries(timestamp clientmodel.Timestamp, view *viewAdapter) Matrix {
 	return Matrix{
 		metric.SampleSet{
 			Metric: clientmodel.Metric{clientmodel.MetricNameLabel: "empty_metric"},
@@ -48,7 +47,7 @@ func (node emptyRangeNode) EvalBoundaries(timestamp time.Time, view *viewAdapter
 }
 
 func TestDeltaWithEmptyElementDoesNotCrash(t *testing.T) {
-	now := time.Now()
+	now := clientmodel.Now()
 	vector := deltaImpl(now, nil, []Node{emptyRangeNode{}, &ScalarLiteral{value: 0}}).(Vector)
 	if len(vector) != 0 {
 		t.Fatalf("Expected empty result vector, got: %v", vector)