Temporal function which can span across time boundaries #811

nikunjgit · 2018-07-26T19:08:52Z

No description provided.

codecov · 2018-07-26T23:03:49Z

Codecov Report

Merging #811 into master will decrease coverage by 0.1%.
The diff coverage is 66.03%.

@@            Coverage Diff            @@
##           master    #811      +/-   ##
=========================================
- Coverage   78.41%   78.3%   -0.11%     
=========================================
  Files         381     384       +3     
  Lines       32712   33060     +348     
=========================================
+ Hits        25651   25889     +238     
- Misses       5314    5390      +76     
- Partials     1747    1781      +34

Flag	Coverage Δ
#dbnode	`81.42% <ø> (ø)`	⬆️
#m3ninx	`71.99% <ø> (ø)`	⬆️
#query	`66.56% <66.03%> (+0.12%)`	⬆️

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 1dadd52...a51e0cd. Read the comment docs.

benraskin92 · 2018-07-27T13:51:29Z

src/coordinator/block/types.go

 type Bounds struct {
 	Start    time.Time
-	End      time.Time
+	Duration time.Duration
 	StepSize time.Duration
 }

 // TimeForIndex returns the start time for a given index assuming a uniform step size
 func (b Bounds) TimeForIndex(idx int) (time.Time, error) {


Maybe this should return (time.Time, bool) so instead of relying on the error if it's out of bounds, we rely on a true/false

benraskin92 · 2018-07-27T19:15:07Z

src/coordinator/block/types.go

+	start := b.Start.Add(blockDuration * time.Duration(n*multiplier))
+	return Bounds{
+		Start:    start,
+		Duration: blockDuration,


can just do Duration: b.Duration?

b.Duration was being used twice so i just extract it out

benraskin92 · 2018-07-27T19:20:36Z

src/coordinator/block/types_test.go

+	}
+	assert.Equal(t, bounds.Steps(), 0)
+	_, err := bounds.TimeForIndex(0)
+	assert.Error(t, err, "No valid index in this block")


nit: s/No/no

benraskin92 · 2018-07-27T19:20:44Z

src/coordinator/block/types_test.go

+	}
+	assert.Equal(t, bounds.Steps(), 0)
+	_, err = bounds.TimeForIndex(0)
+	assert.Error(t, err, "No valid index in this block")


same as above

benraskin92 · 2018-07-27T19:22:56Z

src/coordinator/block/types_test.go

@@ -0,0 +1,71 @@
+// Copyright (c) 2018 Uber Technologies, Inc.


Want to test Next() and/or Previous()

benraskin92 · 2018-07-27T19:26:23Z

src/coordinator/executor/transform/cache.go

+	defer c.mu.Unlock()
+	_, ok := c.blocks[fromTime(key)]
+	if ok {
+		return errors.New("block already exists")


Does this need to error? If it's already in there, isn't that fine? If you need to, can we just return a bool instead?

a block already there is probably a bug so an error helps us identify it

benraskin92 · 2018-07-27T19:31:15Z

src/coordinator/executor/transform/cache.go

+		if ok {
+			blks[i] = b
+		}
+


nit: newline

benraskin92 · 2018-07-27T19:35:15Z

src/coordinator/functions/fetch.go

+			// Ignore any errors
+			iter, _ := block.StepIter()
+			if iter != nil {
+				fmt.Printf("[fetch node]: meta for the block: %v\n", iter.Meta())


benraskin92 · 2018-07-27T19:51:05Z

src/coordinator/functions/temporal/base.go

+	transformOpts transform.Options
+}
+
+// Process processes a block. The processing steps are as follows:


Nice comment!

benraskin92 · 2018-07-27T19:57:10Z

src/coordinator/functions/temporal/base.go

+// 3. For the blocks after current block, figure out which can be processed right now
+// 4. Process all valid blocks from #3, #4 and mark them as processed
+// 5. Run a sweep face to free up blocks which are no longer needed to be cached
+func (c *baseNode) Process(ID parser.NodeID, b block.Block) error {


Could you add some comments in the code explaining what each section is doing? I think that'd be very helpful.

benraskin92 · 2018-07-31T21:42:50Z

src/coordinator/functions/temporal/base.go

+			}
+
+			builder.AppendValue(i, newVal)
+


nit: remove newline

benraskin92 · 2018-07-31T21:43:55Z

src/coordinator/functions/temporal/base.go

+			values = append(values, s.Values()...)
+		}
+
+		desiredLength := int(aggDuration / bounds.StepSize)


please add some comments in this code

benraskin92 · 2018-07-31T21:45:03Z

src/coordinator/functions/temporal/base.go

+		deps := leftBlks[len(leftBlks)-lStart:]
+		deps = append(deps, rightBlks[:i]...)
+		processRequests = append(processRequests, processRequest{blk: rightBlks[i], deps: deps, bounds: bounds.Next(i + 1)})
+


nit: remove newline

arnikola · 2018-07-31T00:57:35Z

src/coordinator/ts/series.go

@@ -69,7 +69,7 @@ func alignValues(values Values, start, end time.Time, interval time.Duration) (F
 	case Datapoints:
 		return RawPointsToFixedStep(vals, start, end, interval)
 	case FixedResolutionMutableValues:
-		// TODO: Align fixed resolution as well once storages can return those directly
+		// TODO: NearestStart fixed resolution as well once storages can return those directly


nit: Comment mishap?

arnikola · 2018-07-31T00:58:40Z

src/coordinator/block/types.go

 type Bounds struct {
 	Start    time.Time
-	End      time.Time
+	Duration time.Duration


nit: To mirror m3db, might be better to call this BlockSize?

arnikola · 2018-07-31T01:01:10Z

src/coordinator/functions/temporal/base.go

+	}
+
+	rightRangeStart := bounds.Next(maxBlocks).Start
+	queryEndBounds := bounds.Nearest(c.transformOpts.TimeSpec.End.Add(-1 * bounds.StepSize))


nit: End.Sub(bounds.StepSize)

Actually the reason we do it is because Sub works with time whereas Add works with duration.

arnikola · 2018-08-01T08:40:07Z

src/coordinator/api/v1/handler/prometheus/native/common.go

+	endInclusiveVal := r.FormValue(endInclusiveParam)
+	params.IncludeEnd = true
+	if endInclusiveVal != "" {
+		includeEnd, err := strconv.ParseBool(r.FormValue(endInclusiveParam))


nit: use endInclusiveVal instead of getting it from r again

arnikola · 2018-08-01T22:20:54Z

src/coordinator/api/v1/handler/prometheus/native/common.go

+		if err != nil {
+			logging.WithContext(r.Context()).Warn("unable to parse end inclusive flag", zap.Any("error", err))
+		}
+		params.IncludeEnd = includeEnd


nit newline

arnikola · 2018-08-02T11:09:15Z

src/coordinator/executor/transform/cache.go

+// Processed returns all processed block times from the cache
+func (c *TimeCache) Processed() map[time.Time]bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()


Rather than incurring the cost of defer here, unlock manually

arnikola · 2018-08-02T11:18:46Z

src/coordinator/functions/temporal/base.go

+	}
+
+	// Process left side of the range
+	leftBlks, emptyLeftBlocks := c.processLeft(b, bounds, maxBlocks, leftRangeStart)


nit: this would read better if the bool returned by processLeft is true iff required blocks are present, rather than the inverse

simplified some logic now, let me know how it reads

arnikola · 2018-08-02T11:21:26Z

src/coordinator/functions/temporal/base.go

+		leftRangeTimes = append(leftRangeTimes, t)
+	}
+
+	leftBlks := c.cache.MultiGet(leftRangeTimes)


Would it be better to push some logic down to MultiGet, where it shortcircuits out on a missing cache value, then returns the partial list and false, rather than doing additional processing here and in processRight?

yeah sounds good.

arnikola · 2018-08-02T11:24:51Z

src/coordinator/functions/temporal/base.go

+	return rightBlks[:firstNil], firstNil != len(rightBlks)
+}
+
+// processCompletedBlocks processes all blocks for which are dependant blocks are present


nit ... for which all dependent...

arnikola · 2018-08-02T11:28:57Z

src/coordinator/functions/temporal/base.go

+	// Mark all blocks as processed
+	c.cache.MarkProcessed(processedKeys)
+	// Sweep to free blocks from cache with no dependencies
+	c.sweep(c.cache.Processed(), queryStartBounds, queryEndBounds, maxBlocks)


does sweep need to take in cache.Processed() if it's a method on c already?

keeping sweep simple ?

benraskin92 · 2018-08-02T17:07:45Z

src/coordinator/models/tag.go

@@ -214,3 +214,16 @@ func (t Tags) sortKeys() ([]string, int) {
 	sort.Strings(keys)
 	return keys, length
 }
+
+func (t Tags) WithoutName() Tags {


nit: add a comment

no longer part of the diff

arnikola · 2018-08-15T18:33:55Z

src/query/api/v1/handler/prometheus/native/common.go

@@ -45,6 +45,7 @@ const (
 	targetParam = "target"
 	stepParam   = "step"
 	debugParam  = "debug"
+	endInclusiveParam = "end-inclusive"


nit: since including is the default, flip this to excludeEndParam?

arnikola · 2018-08-15T18:55:27Z

src/query/api/v1/handler/prometheus/native/common.go

@@ -150,6 +163,12 @@ func renderResultsJSON(w io.Writer, series []*ts.Series) {
 		vals := s.Values()
 		for i := 0; i < s.Len(); i++ {
 			dp := vals.DatapointAt(i)
+			// Skip points before the query boundary. Ideal place to adjust these would be at the result node but that would make it inefficient
+			// since we would need to create another block just for the sake of restricting the bounds
+			if dp.Timestamp.Before(params.Start) {


Rather than skipping in this loop, could ts.Series keep the Bounds of the block it was generated from, and then we could have something like this on Bounds:

// IndexAtTime returns the first index at or after a given time; // returns 0 if t < start, and -1 if t > end func (b Bounds) IndexAtTime(t time.Time) int { start := b.Start if t.Before(start) { return 0 } if t.After(b.End) { return -1 } return int(math.Ceil(float64(t.Sub(start)) / float64(b.StepSize))) }

and use for i := s.b.IndexAtTime(params.Start) here

So the series bounds are always same as the bounds of the block its generated from. The problem is that we sometimes have to fetch blocks for a longer query duration so that some functions can look back enough. However, when we return, we want to skip the excess points.

I get the motivation, but at the moment it may iterate through a bunch of unnecessary data points; better to calculate the first valid i to start from, rather than trying a bunch and discarding. Not really a big deal with a couple of series, but something like 10,000 series where only the last few datapoints in a 2 hour block are required, we'll be doing a lot of unnecessary looping

hmm I see. I can solve that another way. Essentially, since each series has the same start, I can figure out the actual start for first series and use that for all others.

arnikola · 2018-08-15T18:58:04Z

src/query/block/types.go

 }

 // Steps calculates the number of steps for the bounds
 func (b Bounds) Steps() int {
-	if b.Start.After(b.End) || b.StepSize <= 0 {
+	if b.StepSize <= 0 {


Should the case where StepSize is 0 error out earlier, when we build the bounds in the first place, rather than continuing here which could give weird results?

I think only TimeForIndex doesn't make sense with 0 step size, other things should work. In theory we can use this for things without fixed resolution ? I was thinking about moving the bounds outside the block package and into models.

arnikola · 2018-08-15T19:03:42Z

src/query/block/types.go

+
+// Contains returns whether the time lies between the bounds.
+func (b Bounds) Contains(t time.Time) bool {
+	return !b.Start.After(t) && b.Start.Add(b.Duration).After(t)


nit: might read a little cleaner as

diff := b.Start.Sub(t) return diff >= 0 && diff < b.Duration

arnikola · 2018-08-15T19:08:13Z

src/query/block/types.go

+}
+
+func (b Bounds) nth(n int, forward bool) Bounds {
+	multiplier := 1


nit: might read a little cleaner as

multiplier := time.Duration(n) if !forward { multiplier *= -1 }

arnikola · 2018-08-15T23:20:06Z

src/query/functions/temporal/base.go

+// MarkProcessed is used to mark a block as processed
+func (c *blockCache) markProcessed(keys []time.Time) {
+	c.mu.Lock()
+	defer c.mu.Unlock()


Unlock manually instead

arnikola · 2018-08-15T23:22:21Z

src/query/functions/temporal/base.go

+			}
+		}
+
+		reversed(blks)


Can we use sort.Reverse(blks)?

we don't want the blocks in descending sorted order, this is more about just reversing the array

sort.Reverse just reverses the list, shouldn't do any sorting

don't think that's correct. Check out : https://golang.org/pkg/sort/#Reverse. They have an example there. Reverse needs Len(), Comparator, etc. and sorts the list in reverse order

Fair enough; weird that there's no inbuilt reverse array method

arnikola · 2018-08-15T23:24:09Z

src/query/functions/temporal/base.go

+
+// processCompletedBlocks processes all blocks for which are dependant blocks are present
+func (c *baseNode) processCompletedBlocks(processRequests []processRequest, queryStartBounds, queryEndBounds block.Bounds, maxBlocks int) error {
+	processedKeys := make([]time.Time, len(processRequests))


Should this lock the mutex? Similar for processLeft and processRight

markProcessed and sweep are the only ones mutating the blocks. markProcessed is already within a mutex and i'm not sure if sweep needs it since its fine for sweep to miss a few blocks and they can be sweeped later on. Not sure if processLeft and processRight needs a mutex since the multiGet is already locked.

We might have to revisit this later

arnikola · 2018-08-15T23:26:54Z

src/query/functions/temporal/base.go

+	values := make([]float64, 0, steps)
+
+	seriesMeta := seriesIter.SeriesMeta()
+	resultSeriesMeta := make([]block.SeriesMeta, len(seriesMeta))


Shouldn't resultSeriesMeta be the same as incoming seriesMeta? How do you combine seriesMetas between multiple blocks if a particular series does not exist in a particular block?

in resultSeriesMeta we just need to remove the seriesname (same as Prom), rest is the same. We assume all blocks have the same series. If it doesnt' then that series should be all NaNs for that time range.

Oh does prom drop name here? Weird...

arnikola · 2018-08-15T23:29:05Z

src/query/functions/temporal/base_test.go

+	"github.com/stretchr/testify/require"
+)
+
+type processor struct {


Can we get more tests for cases where we have multiple series per block, also where series are missing from some blocks, etc.?

i don't think series can be missing from some blocks. In that case, they should just be NaNs. I'll add tests for processSingleRequest

Seems inefficient to jam in a bunch of unnecessary NaNs for series that don't exist in blocks just to make processing a little easier; could add up to a lot of unneeded datapoints for sparse series

I think that depends on the iterator but it's probably just gonna return NaNs and not actually store them.

arnikola · 2018-08-16T00:13:51Z

src/query/functions/fetch.go

@@ -56,6 +59,14 @@ func (o FetchOp) OpType() string {
 	return FetchType


Should we make FetchOp and FetchNode private similar to other Op/Nodes?

yeah! Don't want to do it in this diff though

Mind adding a todo jic?

arnikola · 2018-08-20T18:32:48Z

src/query/functions/temporal/base.go

+	for i, j := 0, len(blocks)-1; i < j; i, j = i+1, j-1 {
+		blocks[i], blocks[j] = blocks[j], blocks[i]
+	}
+


nit: newline

arnikola · 2018-08-20T18:32:57Z

src/query/functions/temporal/base.go

+	return blks, nil
+}
+
+func reversed(blocks []block.Block) {


nit: rename to reverse

arnikola · 2018-08-20T18:35:06Z

src/query/functions/temporal/base.go

+			}
+		}
+
+		reversed(blks)


Fair enough; weird that there's no inbuilt reverse array method

arnikola · 2018-08-20T18:37:07Z

src/query/functions/temporal/base.go

+	}
+
+	// Process a single index
+	process := func(i int) (bool, error) {


nit: rename to addIfPresent, maybe add a comment as to what this does. Also, flip to return false if not added, and true if added, since true usually corresponds to the positive case

i made the name clearer and added a comment. I still think its reads better with empty, let me know once you read the new comment/naming

arnikola · 2018-08-20T18:37:22Z

src/query/functions/temporal/base.go

+	defer c.mu.Unlock()
+
+	blks := make([]block.Block, 0, numBlocks)
+	if numBlocks == 0 {


nit: can do this before the mutex lock

arnikola · 2018-08-20T18:51:08Z

src/query/functions/temporal/base.go

+	}
+
+	for seriesIter.Next() {
+		values = values[0:0]


nit: can do values[:0] instead?

arnikola · 2018-08-20T18:51:38Z

src/query/functions/temporal/base.go

+	values := make([]float64, 0, steps)
+
+	seriesMeta := seriesIter.SeriesMeta()
+	resultSeriesMeta := make([]block.SeriesMeta, len(seriesMeta))


Oh does prom drop name here? Weird...

arnikola · 2018-08-20T18:52:00Z

src/query/functions/temporal/base.go

+	}
+
+	bounds := seriesIter.Meta().Bounds
+	steps := int((aggDuration + bounds.Duration) / bounds.StepSize)


Add stepSize==0 sanity check; also can define this closer to where it's needed

should be handled by sanity check on bounds in Process()

arnikola · 2018-08-20T18:52:08Z

src/query/functions/temporal/base.go

+		if err != nil {
+			return err
+		}
+		depIters[i] = iter


nit newline

arnikola · 2018-08-20T18:52:21Z

src/query/functions/temporal/base.go

+}
+
+func (c *baseNode) processSingleRequest(request processRequest) error {
+	aggDuration := c.op.duration


nit: define closer to where it's needed

Fix rebase issues Reduce scopes, add comments Implement count_over_time Fix query ranges to align with prom, add debugging Address comments overtime

arnikola · 2018-08-22T22:02:41Z

src/query/functions/fetch.go

 )

 // FetchType gets the series from storage
 const FetchType = "fetch"

 // FetchOp stores required properties for fetch
+// TODO: Make FetchOp private


arnikola

Approved pending comments on TestSingleProcessRequest

nikunjgit requested review from robskillington, justinjc and benraskin92 July 26, 2018 19:08

benraskin92 reviewed Jul 27, 2018

View reviewed changes

nikunjgit force-pushed the overTime branch from 6dc8f2c to 473cc12 Compare July 27, 2018 19:13

benraskin92 reviewed Jul 27, 2018

View reviewed changes

benraskin92 reviewed Jul 31, 2018

View reviewed changes

src/coordinator/functions/temporal/base.go Outdated

}

builder.AppendValue(i, newVal)

Copy link

Collaborator

benraskin92 Jul 31, 2018

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: remove newline

benraskin92 reviewed Jul 31, 2018

View reviewed changes

arnikola reviewed Aug 2, 2018

View reviewed changes

benraskin92 reviewed Aug 2, 2018

View reviewed changes

nikunjgit force-pushed the overTime branch 2 times, most recently from b64f8ae to d2da696 Compare August 14, 2018 23:18

arnikola reviewed Aug 15, 2018

View reviewed changes

arnikola reviewed Aug 16, 2018

View reviewed changes

nikunjgit force-pushed the overTime branch from a831064 to bc3d3b5 Compare August 16, 2018 23:49

arnikola reviewed Aug 20, 2018

View reviewed changes

nikunjgit added 11 commits August 21, 2018 12:11

Temporal function which can span across time boundaries

283b978

Fix rebase issues Reduce scopes, add comments Implement count_over_time Fix query ranges to align with prom, add debugging Address comments overtime

Fix tests

9099977

Rebase

1a0470a

Fix tests

a8f35e9

Address comments

c74a82d

Address comments

0d1344f

More comments

389db2e

Add more tests

507796e

Comments

7568ebc

Fix test

b1a8fa9

Tests, lint

64976e2

nikunjgit force-pushed the overTime branch from ca59b43 to 64976e2 Compare August 21, 2018 21:20

arnikola reviewed Aug 22, 2018

View reviewed changes

arnikola approved these changes Aug 23, 2018

View reviewed changes

Add test comments

a51e0cd

nikunjgit merged commit eca65a6 into master Aug 23, 2018

arnikola deleted the overTime branch September 6, 2018 17:51

		@@ -0,0 +1,71 @@
		// Copyright (c) 2018 Uber Technologies, Inc.

		@@ -56,6 +59,14 @@ func (o FetchOp) OpType() string {
		return FetchType

Temporal function which can span across time boundaries #811

Temporal function which can span across time boundaries #811

Conversation

nikunjgit commented Jul 26, 2018

codecov bot commented Jul 26, 2018 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

benraskin92 Jul 31, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Jul 26, 2018 •

edited

Loading

benraskin92 Jul 31, 2018 •

edited

Loading