New executors

lib/execution_segment.go

@@ -0,0 +1,207 @@
+/*
+ *
+ * k6 - a next-generation load testing tool
+ * Copyright (C) 2019 Load Impact
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package lib
+
+import (
+	"encoding"
+	"fmt"
+	"math/big"
+	"strings"
+)
+
+// ExecutionSegment represents a (start, end] partition of the total execution
+// work for a specific test. For example, if we want the split the execution of a
+// test in 2 different parts, we can split it in two segments (0, 0.5] and (0,5, 1].
+//
+// We use rational numbers so it's easier to verify the correctness and easier to
+// reason about portions of indivisible things, like VUs. This way, we can easily
+// split a test in thirds (i.e. (0, 1/3], (1/3, 2/3], (2/3, 1]), without fearing
+// that we'll lose a VU along the way...
+//
+// The most important part is that if work is split between multiple k6 instances,
+// each k6 instance can precisely and reproducably calculate its share of the work,
+// just by knowing its own segment. There won't be a need to schedule the
+// execution from a master node, or to even know how many other k6 instances are
+// running!
+type ExecutionSegment struct {
+	// 0 <= from < to <= 1
+	from *big.Rat
+	to   *big.Rat
+
+	// derived, equals to-from, but pre-calculated here for speed
+	length *big.Rat
+}
+
+// Ensure we implement those interfaces
+var _ encoding.TextUnmarshaler = &ExecutionSegment{}
+var _ fmt.Stringer = &ExecutionSegment{}
+
+// Helpful "constants" so we don't initialize them in every function call
+var zeroRat, oneRat = big.NewRat(0, 1), big.NewRat(1, 1) //nolint:gochecknoglobals
+var oneBigInt, twoBigInt = big.NewInt(1), big.NewInt(2)  //nolint:gochecknoglobals
+
+// NewExecutionSegment validates the supplied arguments (basically, that 0 <=
+// from < to <= 1) and either returns an error, or it returns a
+// fully-initialized and usable execution segment.
+func NewExecutionSegment(from, to *big.Rat) (*ExecutionSegment, error) {
+	if from.Cmp(zeroRat) < 0 {
+		return nil, fmt.Errorf("segment start value should be at least 0 but was %s", from.FloatString(2))
+	}
+	if from.Cmp(to) >= 0 {
+		return nil, fmt.Errorf("segment start(%s) should be less than its end(%s)", from.FloatString(2), to.FloatString(2))
+	}
+	if to.Cmp(oneRat) > 0 {
+		return nil, fmt.Errorf("segment end value shouldn't be more than 1 but was %s", to.FloatString(2))
+	}
+	return &ExecutionSegment{
+		from:   from,
+		to:     to,
+		length: new(big.Rat).Sub(to, from),
+	}, nil
+}
+
+// stringToRat is a helper function that tries to convert a string to a rational
+// number while allowing percentage, decimal, and fraction values.
+func stringToRat(s string) (*big.Rat, error) {
+	if strings.HasSuffix(s, "%") {
+		num, ok := new(big.Int).SetString(strings.TrimSuffix(s, "%"), 10)
+		if !ok {
+			return nil, fmt.Errorf("'%s' is not a valid percentage", s)
+		}
+		return new(big.Rat).SetFrac(num, big.NewInt(100)), nil
+	}
+	rat, ok := new(big.Rat).SetString(s)
+	if !ok {
+		return nil, fmt.Errorf("'%s' is not a valid percentage, decimal, fraction or interval value", s)
+	}
+	return rat, nil
+}
+
+// UnmarshalText implements the encoding.TextUnmarshaler interface, so that
+// execution segments can be specified as CLI flags, environment variables, and
+// JSON strings.
+//
+// We are able to parse both single percentage/float/fraction values, and actual
+// (from; to] segments. For the single values, we just treat them as the
+// beginning segment - thus the execution segment can be used as a shortcut for
+// quickly running an arbitrarily scaled-down version of a test.
+//
+// The parsing logic is that values with a colon, i.e. ':', are full segments:
+//  `1/2:3/4`, `0.5:0.75`, `50%:75%`, and even `2/4:75%` should be (1/2, 3/4]
+// And values without a hyphen are the end of a first segment:
+//  `20%`, `0.2`,  and `1/5` should be converted to (0, 1/5]
+// empty values should probably be treated as "1", i.e. the whole execution
+func (es *ExecutionSegment) UnmarshalText(text []byte) (err error) {
+	from := zeroRat
+	toStr := string(text)
+	if strings.ContainsRune(toStr, ':') {
+		fromToStr := strings.SplitN(toStr, ":", 2)
+		toStr = fromToStr[1]
+		if from, err = stringToRat(fromToStr[0]); err != nil {
+			return err
+		}
+	}
+
+	to, err := stringToRat(toStr)
+	if err != nil {
+		return err
+	}
+
+	segment, err := NewExecutionSegment(from, to)
+	if err != nil {
+		return err
+	}
+	*es = *segment
+	return nil
+}
+
+func (es *ExecutionSegment) String() string {
+	if es == nil {
+		return "0:1"
+	}
+	return es.from.RatString() + ":" + es.to.RatString()
+}
+
+// FloatLength is a helper method for getting some more human-readable
+// information about the execution segment.
+func (es *ExecutionSegment) FloatLength() float64 {
+	if es == nil {
+		return 1.0
+	}
+	res, _ := es.length.Float64()
+	return res
+}
+
+//TODO: add a NewFromString() and Split() methods
+
+// helper function for rounding (up) of rational numbers to big.Int values
+func roundUp(rat *big.Rat) *big.Int {
+	quo, rem := new(big.Int).QuoRem(rat.Num(), rat.Denom(), new(big.Int))
+
+	if rem.Mul(rem, twoBigInt).Cmp(rat.Denom()) >= 0 {
+		return quo.Add(quo, oneBigInt)
+	}
+	return quo
+}
+
+// Scale proportionally scales the supplied value, according to the execution
+// segment's position and size of the work.
+func (es *ExecutionSegment) Scale(value int64) int64 {
+	if es == nil { // no execution segment, i.e. 100%
+		return value
+	}
+	// Instead of the first proposal that used remainders and floor:
+	//    floor( (value * from) % 1 + value * length )
+	// We're using an alternative approach with rounding that (hopefully) has
+	// the same properties, but it's simpler and has better precision:
+	//    round( (value * from) - round(value * from) + (value * (to - from)) )?
+	// which reduces to:
+	//    round( (value * to) - round(value * from) )?
+
+	toValue := big.NewRat(value, 1)
+	toValue.Mul(toValue, es.to)
+
+	fromValue := big.NewRat(value, 1)
+	fromValue.Mul(fromValue, es.from)
+
+	toValue.Sub(toValue, new(big.Rat).SetFrac(roundUp(fromValue), oneBigInt))
+
+	return roundUp(toValue).Int64()
+}
+
+// InPlaceScaleRat scales rational numbers in-place - it changes the passed
+// argument (and also returns it, to allow for chaining, like many other big.Rat
+// methods).
+func (es *ExecutionSegment) InPlaceScaleRat(value *big.Rat) *big.Rat {
+	if es == nil { // no execution segment, i.e. 100%
+		return value
+	}
+	return value.Mul(value, es.length)
+}
+
+// CopyScaleRat scales rational numbers without changing them - creates a new
+// bit.Rat object and uses it for the calculation.
+func (es *ExecutionSegment) CopyScaleRat(value *big.Rat) *big.Rat {
+	if es == nil { // no execution segment, i.e. 100%
+		return value
+	}
+	return new(big.Rat).Mul(value, es.length)
+}

lib/scheduler/interfaces.go → lib/execution_segment_test.go

@@ -18,16 +18,6 @@
  *
  */
 
-package scheduler
+package lib
 
-import "time"
-
-// Config is an interface that should be implemented by all scheduler config types
-type Config interface {
-	GetBaseConfig() BaseConfig
-	Validate() []error
-	GetMaxVUs() int64
-	GetMaxDuration() time.Duration // includes max timeouts, to allow us to share VUs between schedulers in the future
-	//TODO: Split(percentages []float64) ([]Config, error)
-	//TODO: String() method that could be used for priting descriptions of the currently running schedulers for the UI?
-}
+//TODO

lib/helpers.go

@@ -0,0 +1,153 @@
+/*
+ *
+ * k6 - a next-generation load testing tool
+ * Copyright (C) 2019 Load Impact
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package lib
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+)
+
+// StrictJSONUnmarshal decodes a JSON in a strict manner, emitting an error if there
+// are unknown fields or unexpected data
+func StrictJSONUnmarshal(data []byte, v interface{}) error {
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.DisallowUnknownFields()
+	dec.UseNumber()
+
+	if err := dec.Decode(&v); err != nil {
+		return err
+	}
+	if dec.More() {
+		//TODO: use a custom error?
+		return fmt.Errorf("unexpected data after the JSON object")
+	}
+	return nil
+}
+
+// GetMaxPlannedVUs returns the maximum number of planned VUs at any stage of
+// the scheduler execution plan.
+func GetMaxPlannedVUs(steps []ExecutionStep) (result uint64) {
+	for _, s := range steps {
+		stepMaxPlannedVUs := s.PlannedVUs
+		if stepMaxPlannedVUs > result {
+			result = stepMaxPlannedVUs
+		}
+	}
+	return result
+}
+
+// GetMaxPossibleVUs returns the maximum number of planned + unplanned (i.e.
+// initialized mid-test) VUs at any stage of the scheduler execution plan.
+// Unplanned VUs are possible in some schedulers, like the arrival-rate ones, as
+// a way to have a low number of pre-allocated VUs, but be able to initialize
+// new ones in the middle of the test, if needed. For example, if the remote
+// system starts responding very slowly and all of the pre-allocated VUs are
+// waiting for it.
+//
+// IMPORTANT 1: Getting planned and unplanned VUs separately for the whole
+// duration of a test can often lead to mistakes. That's why this function is
+// called GetMaxPossibleVUs() and why there is no GetMaxUnplannedVUs() function.
+//
+// As an example, imagine that you have a scheduler with MaxPlannedVUs=20 and
+// MaxUnaplannedVUs=0, followed immediately after by another scheduler with
+// MaxPlannedVUs=10 and MaxUnaplannedVUs=10. If you MaxPlannedVUs for the whole
+// test is 20, and MaxUnaplannedVUs, but since those schedulers won't run
+// concurrently, MaxVUs for the whole test is not 30, rather it's 20, since 20
+// VUs will be sufficient to run the test.
+//
+// IMPORTANT 2: this has one very important exception. The manual execution
+// scheduler doesn't use the MaxUnplannedVUs (i.e. this function will return 0),
+// since their initialization and usage is directly controlled by the user and
+// is effectively bounded only by the resources of the machine k6 is running on.
+func GetMaxPossibleVUs(steps []ExecutionStep) (result uint64) {
+	for _, s := range steps {
+		stepMaxPossibleVUs := s.PlannedVUs + s.MaxUnplannedVUs
+		if stepMaxPossibleVUs > result {
+			result = stepMaxPossibleVUs
+		}
+	}
+	return result
+}
+
+// GetEndOffset returns the time offset of the last step of the execution plan,
+// and whether that step is a final one, i.e. whether the number of planned or
+// unplanned
+func GetEndOffset(steps []ExecutionStep) (lastStepOffset time.Duration, isFinal bool) {
+	if len(steps) == 0 {
+		return 0, true
+	}
+	lastStep := steps[len(steps)-1]
+	return lastStep.TimeOffset, (lastStep.PlannedVUs == 0 && lastStep.MaxUnplannedVUs == 0)
+}
+
+// A helper function for joining error messages into a single string
+func concatErrors(errors []error, separator string) string {
+	errStrings := make([]string, len(errors))
+	for i, e := range errors {
+		errStrings[i] = e.Error()
+	}
+	return strings.Join(errStrings, separator)
+}
+
+// StreamExecutionSteps launches a new goroutine and emits all execution steps
+// at their appropriate time offsets over the returned unbuffered channel. If
+// closeChanWhenDone is specified, it will close the channel after it sends the
+// last step. If it isn't, or if the context is cancelled, the internal
+// goroutine will be stopped, *but the channel will remain open*!
+//
+// As usual, steps in the supplied slice have to be sorted by their TimeOffset
+// values in an ascending order. Of course, multiple events can have the same
+// time offset (incl. 0).
+func StreamExecutionSteps(
+	ctx context.Context, startTime time.Time, steps []ExecutionStep, closeChanWhenDone bool,
+) <-chan ExecutionStep {
+
+	ch := make(chan ExecutionStep)
+	go func() {
+		for _, step := range steps {
+			offsetDiff := step.TimeOffset - time.Since(startTime)
+			if offsetDiff > 0 { // wait until time of event arrives
+				select {
+				case <-ctx.Done():
+					return // exit if context is cancelled
+				case <-time.After(offsetDiff): //TODO: reuse a timer?
+					// do nothing
+				}
+			}
+			select {
+			case <-ctx.Done():
+				return // exit if context is cancelled
+			case ch <- step: // send the step
+			}
+		}
+
+		// Close the channel only if all steps were sent successfully (i.e. the
+		// parent context didn't die) and we were instructed to do so.
+		if closeChanWhenDone {
+			close(ch)
+		}
+	}()
+	return ch
+}