Merge pull request #3 from khorshuheng/configurable-load

Configurable load and Feast 0.8 compatibility

Author: pyalex

README.md

@@ -7,6 +7,11 @@ This simple Go service generates load as part of the Feast testing suite. It sit
 ```
 
 ### Usage
+Create a specification file for the load. Refer to the example specification for details.
+```
+LOAD_SPECIFICATION_PATH=example/loadSpec.yml
+```
+
 Start the proxy
 ```
 LOAD_FEAST_SERVING_HOST=feast.serving.example.com LOAD_FEAST_SERVING_PORT=6566 go run main.go
@@ -21,8 +26,8 @@ The following command simply echos the version of Feast Serving. Useful for test
 curl localhost:8080/echo
 ```
 
-This command will send a single GetOnlineFeatures request to the configured Feast serving instance. The `entity_count` parameter is used to set how many entities will be sent (unique users in this case). The higher the number of entities the higher the expected latency.
+This command will send a single or multiple GetOnlineFeatures request(s) depending on the load specification.
 
 ```
-curl localhost:8080/send?entity_count=30
+curl localhost:8080/send
 ```

example/loadSpec.yml

@@ -0,0 +1,49 @@
+# Define the requested entities
+entities:
+  # Entity name
+  - name: "merchant_id"
+    # Type of entity. Supported types: int64, int32, float, double, bool, string
+    type: "int64"
+    # Generate values from file
+    fileSource:
+      # Path to a file that contains the possible values of the entity. The number of
+      # lines must be greater than the entity count for the request.
+      path: "example/restaurant_id.txt"
+  - name: "customer_id"
+    type: "int64"
+    # Applicable only for integer type entities, and if a source file is not defined.
+    # The entity value will be generated randomly based on the min and max values (inclusive).
+    randInt:
+      min: 1
+      max: 100
+
+
+# Each entry defines a request that would be send when the /send endpoint is called.
+# If they are multiple requests, each request will be sent within a goroutine, and the call
+# will return when all requests succesfully receive a response.
+requests:
+  # Entity(s) that corresponds to the features to be retrieved
+  - entities:
+      - "customer_id"
+    # Retrieved features
+    features:
+      - "merchant_orders:lifetime_avg_basket_size"
+      - "merchant_orders:weekday_completed_order"
+      - "merchant_orders:weekend_completed_order"
+    # Number of entities in the request. The entity value are chosen randomly from the source
+    # file defined above.
+    entityCount: 5
+  - entities:
+      - "merchant_id"
+    features:
+      - "customer_orders:cust_total_orders_3months"
+      - "customer_orders:cust_orders_uniq_restaurants_3months"
+    entityCount: 5
+
+  - entities:
+      - "customer_id"
+      - "merchant_id"
+    features:
+      - "merchant_customer_orders:days_since_last_order"
+      - "merchant_customer_orders:int_order_count_90days"
+    entityCount: 5

example/restaurant_id.txt

@@ -0,0 +1,10 @@
+1
+2
+3
+4
+5
+6
+7
+8
+9
+10

generator/generator.go

@@ -0,0 +1,201 @@
+package generator
+
+import (
+	"bufio"
+	"errors"
+	"fmt"
+	feast "github.com/feast-dev/feast/sdk/go"
+	"github.com/feast-dev/feast/sdk/go/protos/feast/types"
+	"math/rand"
+	"os"
+	"strconv"
+	"time"
+)
+
+type LoadSpec struct {
+	EntitySpec   []EntitySpec  `yaml:"entities"`
+	RequestSpecs []RequestSpec `yaml:"requests"`
+}
+
+type EntitySpec struct {
+	Name       string     `yaml:"name"`
+	Type       string     `yaml:"type"`
+	FileSource FileSource `yaml:"fileSource"`
+	RandInt    RandInt    `yaml:"randInt"`
+}
+
+type FileSource struct {
+	Path string `yaml:"path"`
+}
+
+
+type RandInt struct {
+	Min int64 `yaml:"min"`
+	Max int64 `yaml:"max"`
+}
+
+// Generate all possible values for an entity
+type EntityPoolGenerator interface {
+	GenerateEntityValues() ([]*types.Value, error)
+}
+
+// Generate all possible values for an entity from a file source
+type FileSourceEntityValueGenerator struct {
+	entity EntitySpec
+}
+
+func (generator FileSourceEntityValueGenerator) GenerateEntityValues() ([]*types.Value, error) {
+	var entityValues []*types.Value
+	file, err := os.Open(generator.entity.FileSource.Path)
+	if err != nil {
+		return nil, err
+	}
+	scanner := bufio.NewScanner(file)
+	scanner.Split(bufio.ScanLines)
+	for scanner.Scan() {
+		parsedValue, err := generator.parseStrToEntityValue(generator.entity.Type, scanner.Text())
+		if err != nil {
+			return nil, err
+		}
+		entityValues = append(entityValues, parsedValue)
+	}
+	return entityValues, nil
+}
+
+func (generator FileSourceEntityValueGenerator) parseStrToEntityValue(valueType string, valueStr string) (*types.Value, error) {
+	switch valueType {
+	case "string":
+		return feast.StrVal(valueStr), nil
+	case "int32":
+		parsedValue, err := strconv.ParseInt(valueStr, 10, 32)
+		if err != nil {
+			return nil, err
+		}
+		return feast.Int32Val(int32(parsedValue)), nil
+	case "int64":
+		parsedValue, err := strconv.ParseInt(valueStr, 10, 64)
+		if err != nil {
+			return nil, err
+		}
+		return feast.Int64Val(parsedValue), nil
+	case "float":
+		parsedValue, err := strconv.ParseFloat(valueStr, 32)
+		if err != nil {
+			return nil, err
+		}
+		return feast.FloatVal(float32(parsedValue)), nil
+	case "double":
+		parsedValue, err := strconv.ParseFloat(valueStr, 64)
+		if err != nil {
+			return nil, err
+		}
+		return feast.DoubleVal(parsedValue), nil
+	case "bool":
+		parsedValue, err := strconv.ParseBool(valueStr)
+		if err != nil {
+			return nil, err
+		}
+		return feast.BoolVal(parsedValue), nil
+	}
+
+	return nil, errors.New(fmt.Sprintf("Unrecognized value type: %s", valueType))
+}
+
+// Generate all possible values for an entity based on a range of integer
+type RandIntEntityValueGenerator struct {
+	entity EntitySpec
+}
+
+func (generator RandIntEntityValueGenerator) GenerateEntityValues() ([]*types.Value, error) {
+	entityType := generator.entity.Type
+	switch entityType {
+	case "int64":
+		minValue := generator.entity.RandInt.Min
+		maxValue := generator.entity.RandInt.Max
+		poolSize := maxValue - minValue + 1
+		entityValues := make([]*types.Value, poolSize)
+		for i := int64(0); i < poolSize; i++ {
+			entityValues[i] = feast.Int64Val(i + minValue)
+		}
+		return entityValues, nil
+	case "int32":
+		minValue := int32(generator.entity.RandInt.Min)
+		maxValue := int32(generator.entity.RandInt.Max)
+		poolSize := maxValue - minValue + 1
+		entityValues := make([]*types.Value, poolSize)
+		for i := int32(0); i < poolSize; i++ {
+			entityValues[i] = feast.Int32Val(i + minValue)
+		}
+		return entityValues, nil
+	default:
+		return nil, errors.New(fmt.Sprintf("Unsupported entity type: %s", entityType))
+	}
+}
+
+type RequestSpec struct {
+	Entities    []string `yaml:"entities"`
+	Features    []string `yaml:"features"`
+	EntityCount int32    `yaml:"entityCount"`
+}
+
+type RequestGenerator struct {
+	entityToValuePoolMap 	map[string][]*types.Value
+	requests 				[]RequestSpec
+	project					string
+}
+
+func NewRequestGenerator(loadSpec LoadSpec, project string) (RequestGenerator, error) {
+	entityToValuePoolMap := map[string][]*types.Value{}
+	for _, entity := range loadSpec.EntitySpec {
+		var poolGenerator EntityPoolGenerator
+		if entity.FileSource != (FileSource{}) {
+			poolGenerator = FileSourceEntityValueGenerator{entity}
+		} else {
+			poolGenerator = RandIntEntityValueGenerator{entity}
+		}
+		pool, err := poolGenerator.GenerateEntityValues()
+		if err != nil {
+			return RequestGenerator{}, err
+		}
+		entityToValuePoolMap[entity.Name] = pool
+	}
+	return RequestGenerator{
+		entityToValuePoolMap: entityToValuePoolMap,
+		requests: loadSpec.RequestSpecs,
+		project: project,
+	}, nil
+}
+
+
+func (generator *RequestGenerator) GenerateRandomRows(entities []string, entityCount int32) []feast.Row {
+	rows := make([]feast.Row, entityCount)
+	for _, entity := range entities {
+		valuePool := generator.entityToValuePoolMap[entity]
+		rand.Seed(time.Now().UnixNano())
+		rand.Shuffle(len(valuePool), func(i, j int) { valuePool[i], valuePool[j] = valuePool[j], valuePool[i] })
+	}
+
+	for i := int32(0); i < entityCount; i++ {
+		row := feast.Row{}
+		for _, entity := range entities {
+			valuePool := generator.entityToValuePoolMap[entity]
+			row[entity] = valuePool[i]
+		}
+		rows[i]=row
+	}
+
+	return rows
+}
+
+func (generator *RequestGenerator) GenerateRequests() []feast.OnlineFeaturesRequest {
+	var onlineFeatureRequests []feast.OnlineFeaturesRequest
+	for _, request := range generator.requests {
+		entityRows := generator.GenerateRandomRows(request.Entities, request.EntityCount)
+		onlineFeatureRequests = append(onlineFeatureRequests, feast.OnlineFeaturesRequest{
+			Features: request.Features,
+			Entities: entityRows,
+			Project:  generator.project,
+		})
+	}
+	return onlineFeatureRequests
+}

generator/generator_test.go

@@ -0,0 +1,33 @@
+package generator
+
+import (
+	"gopkg.in/yaml.v2"
+	"io/ioutil"
+	"testing"
+)
+
+func TestGenerateRequests(t *testing.T) {
+	yamlSpec, err := ioutil.ReadFile("../example/loadSpec.yml")
+	if err != nil {
+		t.Errorf(err.Error())
+		t.FailNow()
+	}
+	loadSpec := LoadSpec{}
+	err = yaml.Unmarshal(yamlSpec, &loadSpec)
+	if err != nil {
+		t.Errorf(err.Error())
+		t.FailNow()
+	}
+	loadSpec.EntitySpec[0].FileSource.Path = "../example/restaurant_id.txt"
+	requestGenerator, err := NewRequestGenerator(loadSpec, "default")
+	if err != nil {
+		t.Errorf(err.Error())
+		t.FailNow()
+	}
+	requests := requestGenerator.GenerateRequests()
+	if len(requests) != 3 {
+		t.Errorf("Request length not equals to 3")
+		t.FailNow()
+	}
+
+}

go.mod

@@ -1,8 +1,9 @@
 module feast-load-generator
 
-go 1.14
+go 1.15
 
 require (
-	github.com/feast-dev/feast/sdk/go v0.0.0-20200724013123-d07875d4efa1
+	github.com/feast-dev/feast/sdk/go v0.8.2
 	github.com/kelseyhightower/envconfig v1.4.0
+	gopkg.in/yaml.v2 v2.2.2
 )