Merge pull request #2 from lomik/rollup_until

Rollup until

Author: lomik

helper/rollup/rollup.go

@@ -135,6 +135,18 @@ func (r *Rollup) Match(metric string) *Pattern {
 	return r.Default
 }
 
+func (r *Rollup) Step(metric string, from int32) int32 {
+	pattern := r.Match(metric)
+	now := int32(time.Now().Unix())
+
+	for i := range pattern.Retention {
+		if i == len(pattern.Retention)-1 || from > now-pattern.Retention[i+1].Age {
+			return pattern.Retention[i].Precision
+		}
+	}
+	return pattern.Retention[len(pattern.Retention)-1].Precision
+}
+
 func doMetricPrecision(points []point.Point, precision int32, aggr func([]point.Point) float64) []point.Point {
 	l := len(points)
 	var i, n int

helper/rollup/rollup_test.go

@@ -1,7 +1,9 @@
 package rollup
 
 import (
+	"fmt"
 	"testing"
+	"time"
 
 	"github.com/lomik/graphite-clickhouse/helper/point"
 )
@@ -73,3 +75,63 @@ func TestMetricPrecision(t *testing.T) {
 		point.AssertListEq(t, test[1], result)
 	}
 }
+
+func TestMetricStep(t *testing.T) {
+	config := `
+<graphite_rollup>
+ 	<pattern>
+ 		<regexp>^metric\.</regexp>
+ 		<function>any</function>
+ 		<retention>
+ 			<age>0</age>
+ 			<precision>1</precision>
+ 		</retention>
+ 		<retention>
+ 			<age>3600</age>
+ 			<precision>10</precision>
+ 		</retention>
+ 	</pattern>
+ 	<default>
+ 		<function>max</function>
+ 		<retention>
+ 			<age>0</age>
+ 			<precision>60</precision>
+ 		</retention>
+ 		<retention>
+ 			<age>3600</age>
+ 			<precision>300</precision>
+ 		</retention>
+ 		<retention>
+ 			<age>86400</age>
+ 			<precision>3600</precision>
+ 		</retention>
+ 	</default>
+</graphite_rollup>
+`
+	r, err := ParseXML([]byte(config))
+	if err != nil {
+		t.Fatal(err)
+	}
+	now := int32(time.Now().Unix())
+
+	tests := []struct {
+		name         string
+		from         int32
+		expectedStep int32
+	}{
+		{"metric.foo.first-retention", now - 500, 1},
+		{"metric.foo.second-retention", now - 3600, 10},
+		{"foo.bar.default-first-retention", now - 500, 60},
+		{"foo.bar.default-second-retention", now - 3700, 300},
+		{"foo.bar.default-last-retention", now - 87000, 3600},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("metric=%v (from=now-%v)", test.name, now-test.from), func(t *testing.T) {
+			step := r.Step(test.name, test.from)
+			if step != test.expectedStep {
+				t.Fatalf("metric=%v (from=now-%v), expected step=%v, actual step=%v", test.name, now-test.from, test.expectedStep, step)
+			}
+		})
+	}
+}

render/handler.go

@@ -69,16 +69,9 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	dateWhere := fmt.Sprintf(
-		"(Date >='%s' AND Date <= '%s' AND Time >= %d AND Time <= %d)",
-		time.Unix(fromTimestamp, 0).Format("2006-01-02"),
-		time.Unix(untilTimestamp, 0).Format("2006-01-02"),
-		fromTimestamp,
-		untilTimestamp,
-	)
-
 	var pathWhere string
 
+	maxStep := int32(0)
 	if find.HasWildcard(target) {
 		// Search in small index table first
 		treeWhere := find.MakeWhere(target, true)
@@ -105,6 +98,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 			if p == "" {
 				continue
 			}
+			step := h.config.Rollup.Step(p, int32(fromTimestamp))
+			if step > maxStep {
+				maxStep = step
+			}
 
 			if !first {
 				listBuf.Write([]byte{','})
@@ -131,8 +128,18 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		// pathWhere = makeWhere(target, false)
 	} else {
 		pathWhere = fmt.Sprintf("Path = '%s'", target)
+		maxStep = h.config.Rollup.Step(target, int32(fromTimestamp))
 	}
 
+	until := untilTimestamp - untilTimestamp % int64(maxStep) + int64(maxStep) - 1
+	dateWhere := fmt.Sprintf(
+		"(Date >='%s' AND Date <= '%s' AND Time >= %d AND Time <= %d)",
+		time.Unix(fromTimestamp, 0).Format("2006-01-02"),
+		time.Unix(untilTimestamp, 0).Format("2006-01-02"),
+		fromTimestamp,
+		until,
+	)
+
 	// @TODO: change format to RowBinary
 	query := fmt.Sprintf(
 		`
@@ -325,7 +332,7 @@ func (h *Handler) ReplyProtobuf(w http.ResponseWriter, r *http.Request, points [
 
 		var index int32
 		// skip points before start
-		for index = 0; points[index].Time < start; index++ {
+		for index = 0; index < int32(len(points)) && points[index].Time < start; index++ {
 		}
 
 		for i := int32(0); i < count; i++ {