rc2

database/api.go

@@ -712,6 +712,7 @@ func (s *Database) ListInstantPeriods(conn net.Conn, req proto.ListInstantPeriod
 	responseWriter, err := transform.NewInstantPeriodsWriter(transform.InstantPeriodsWriterOptions{
 		Dst:            conn,
 		GroupBy:        req.GroupBy,
+		Since:          uint32(since.Unix()),
 		AggregateFuncs: req.AggregateFuncs,
 		FirstHourOfDay: req.FirstHourOfDay,
 	})

examples/requests/requests.go

@@ -246,8 +246,6 @@ GetMetric:
 
 	cumulativeMeasures := GenerateCumulativeMeasures(seriesInDays)
 
-	//pretty.PPrintln("CUMULATIVE MEASURES", cumulativeMeasures)
-
 	err = conn.AppendMeasures(proto.AppendMeasuresReq{
 		MetricID: cumulativeMetricID,
 		Measures: cumulativeMeasures,

transform/aggregate.go

@@ -16,6 +16,7 @@ import (
 type InstantPeriodsWriterOptions struct {
 	Dst            io.Writer
 	GroupBy        diploma.GroupBy
+	Since          uint32
 	AggregateFuncs byte
 	FirstHourOfDay int
 }
@@ -25,6 +26,7 @@ type InstantPeriodsWriter struct {
 	arr            []byte
 	responder      *ChunkedResponder
 	groupBy        diploma.GroupBy
+	since          uint32
 	firstHourOfDay int
 	time2period    func(uint32) time.Time
 	currentPeriod  time.Time
@@ -63,6 +65,7 @@ func NewInstantPeriodsWriter(opt InstantPeriodsWriterOptions) (*InstantPeriodsWr
 		arr:            make([]byte, 12+q*8),
 		responder:      NewChunkedResponder(opt.Dst),
 		groupBy:        opt.GroupBy,
+		since:          opt.Since,
 		firstHourOfDay: opt.FirstHourOfDay,
 	}
 
@@ -179,9 +182,11 @@ func (s *InstantPeriodsWriter) packBlankPeriod() {
 
 func (s *InstantPeriodsWriter) Close() (err error) {
 	if s.entries > 0 {
+		if s.lastTimestamp >= s.since {
 			s.packPeriod(s.lastTimestamp)
 			s.responder.AppendRecord(s.arr)
 		}
+	}
 	return s.responder.Flush()
 }