bidengine: bid parameters

* skip orders greater than one cpu
* skip orders with max price too low
* price according to memory usage

fixes #291
fixes #295

cmd/common/signal.go

@@ -10,7 +10,7 @@ import (
 func WatchSignals(ctx context.Context, cancel context.CancelFunc) <-chan struct{} {
 	donech := make(chan struct{})
 	sigch := make(chan os.Signal, 1)
-	signal.Notify(sigch, syscall.SIGINT, syscall.SIGHUP)
+	signal.Notify(sigch, syscall.SIGINT, syscall.SIGHUP, syscall.SIGTERM)
 	go func() {
 		defer close(donech)
 		defer signal.Stop(sigch)

glide.yaml

@@ -43,12 +43,10 @@ import:
   - trees/avltree
 - package: github.com/boz/go-lifecycle
 - package: k8s.io/client-go
-  # version: ^8.0.0
   subpackages:
   - '...'
 - package: k8s.io/apiextensions-apiserver
 - package: k8s.io/apimachinery
-  # version: 'kubernetes-1.11.1'
   subpackages:
   - '...'
 - package: github.com/googleapis/gnostic
@@ -64,3 +62,5 @@ import:
 - package: github.com/grpc-ecosystem/grpc-gateway
   version: ^1.4.1
 - package: k8s.io/metrics
+- package: github.com/caarlos0/env
+  version: ^3.3.0

provider/bidengine/config.go

@@ -0,0 +1,8 @@
+package bidengine
+
+type config struct {
+	FulfillmentCPUMax uint `env:"AKASH_PROVIDER_MAX_FULFILLMENT_CPU" envDefault:"1000"`
+
+	FulfillmentMemPriceMin uint `env:"AKASH_PROVIDER_MEM_PRICE_MIN" envDefault:"50"`
+	FulfillmentMemPriceMax uint `env:"AKASH_PROVIDER_MEM_PRICE_MAX" envDefault:"150"`
+}

provider/bidengine/order.go

@@ -10,12 +10,14 @@ import (
 	"github.com/ovrclk/akash/provider/event"
 	"github.com/ovrclk/akash/provider/session"
 	"github.com/ovrclk/akash/types"
+	"github.com/ovrclk/akash/types/unit"
 	"github.com/ovrclk/akash/util/runner"
 	"github.com/tendermint/tmlibs/log"
 )
 
 // order manages bidding and general lifecycle handling of an order.
 type order struct {
+	config      config
 	order       types.OrderID
 	fulfillment *types.Fulfillment
 
@@ -41,6 +43,7 @@ func newOrder(e *service, oid types.OrderID, fulfillment *types.Fulfillment) (*o
 	log := session.Log().With("order", oid)
 
 	order := &order{
+		config:      e.config,
 		order:       oid,
 		fulfillment: fulfillment,
 		session:     session,
@@ -148,13 +151,10 @@ loop:
 
 			group = result.Value().(*types.DeploymentGroup)
 
-			if !matchProviderAttributes(o.session.Provider().Attributes, group.Requirements) {
-				o.log.Debug("unable to fulfill: incompatible attributes")
-				break loop
+			if !o.shouldBid(group) {
+				break
 			}
 
-			// TODO: check if price is too low
-
 			// Begin reserving resources from cluster.
 			clusterch = runner.Do(func() runner.Result {
 				return runner.NewResult(o.cluster.Reserve(o.order, group))
@@ -233,17 +233,65 @@ loop:
 	}
 }
 
-func (o *order) calculatePrice(resources types.ResourceList) uint64 {
-	max := o.resourcesMaxPrice(resources)
-	return uint64(rand.Int63n(int64(max)) + 1)
+func (o *order) shouldBid(group *types.DeploymentGroup) bool {
+
+	// does provider have required attributes?
+	if !matchProviderAttributes(o.session.Provider().Attributes, group.Requirements) {
+		o.log.Debug("unable to fulfill: incompatible attributes")
+		return false
+	}
+
+	// TODO: catch overflow
+	var (
+		cpu   int64
+		mem   int64
+		price int64
+	)
+	for _, rg := range group.GetResources() {
+		cpu += int64(rg.Unit.CPU * rg.Count)
+		mem += int64(rg.Unit.Memory * uint64(rg.Count))
+		price += int64(rg.Price)
+	}
+
+	// requesting too much cpu?
+	if cpu > int64(o.config.FulfillmentCPUMax) || cpu <= 0 {
+		o.log.Info("unable to fulfill: cpu request too high",
+			"cpu-requested", cpu)
+		return false
+	}
+
+	// price max too low?
+	if price*unit.Gi < mem*int64(o.config.FulfillmentMemPriceMin) {
+		o.log.Info("unable to fulfill: price too low",
+			"max-price", price,
+			"min-price", mem*int64(o.config.FulfillmentMemPriceMin)/unit.Gi)
+		return false
+	}
+
+	return true
 }
 
-func (o *order) resourcesMaxPrice(resources types.ResourceList) uint64 {
+func (o *order) calculatePrice(resources types.ResourceList) uint64 {
 	// TODO: catch overflow
-	price := uint64(0)
+	var (
+		mem  int64
+		rmax int64
+	)
+
 	for _, group := range resources.GetResources() {
-		price += group.Price
+		rmax += int64(group.Price)
+		mem += int64(group.Unit.Memory * uint64(group.Count))
+	}
+
+	cmin := uint64(float64(mem) * float64(o.config.FulfillmentMemPriceMin) / float64(unit.Gi))
+	cmax := uint64(float64(mem) * float64(o.config.FulfillmentMemPriceMax) / float64(unit.Gi))
+
+	if cmax > uint64(rmax) {
+		cmax = uint64(rmax)
+	}
+	if cmax == 0 {
+		cmax = 1
 	}
-	o.log.Debug("group max price", "price", price)
-	return price
+
+	return uint64(rand.Int63n(int64(cmax-cmin)) + int64(cmin))
 }

provider/bidengine/service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 
 	lifecycle "github.com/boz/go-lifecycle"
+	"github.com/caarlos0/env"
 	"github.com/ovrclk/akash/provider/cluster"
 	"github.com/ovrclk/akash/provider/event"
 	"github.com/ovrclk/akash/provider/session"
@@ -18,6 +19,11 @@ type Service interface {
 // Service handles bidding on orders.
 func NewService(ctx context.Context, session session.Session, cluster cluster.Cluster, bus event.Bus) (Service, error) {
 
+	config := config{}
+	if err := env.Parse(&config); err != nil {
+		return nil, err
+	}
+
 	sub, err := bus.Subscribe()
 	if err != nil {
 		return nil, err
@@ -34,6 +40,7 @@ func NewService(ctx context.Context, session session.Session, cluster cluster.Cl
 	session.Log().Info("found orders", "count", len(existingOrders))
 
 	s := &service{
+		config:  config,
 		session: session,
 		cluster: cluster,
 		bus:     bus,
@@ -50,6 +57,7 @@ func NewService(ctx context.Context, session session.Session, cluster cluster.Cl
 }
 
 type service struct {
+	config  config
 	session session.Session
 	cluster cluster.Cluster
 

provider/bidengine/service_test.go

@@ -2,6 +2,7 @@ package bidengine_test
 
 import (
 	"context"
+	"strconv"
 	"testing"
 
 	"github.com/ovrclk/akash/provider/bidengine"
@@ -12,6 +13,7 @@ import (
 	"github.com/ovrclk/akash/testutil"
 	txmocks "github.com/ovrclk/akash/txutil/mocks"
 	"github.com/ovrclk/akash/types"
+	"github.com/ovrclk/akash/types/unit"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
@@ -25,7 +27,7 @@ func TestService(t *testing.T) {
 	defer bus.Close()
 
 	deployment := testutil.Deployment(testutil.Address(t), 1)
-	group := testutil.DeploymentGroups(deployment.Address, 2).Items[0]
+	group := makeDeploymentGroup(deployment.Address, 2)
 	order := testutil.Order(deployment.Address, group.Seq, 3)
 	provider := testutil.Provider(testutil.Address(t), 4)
 
@@ -97,7 +99,7 @@ func TestService_Catchup(t *testing.T) {
 	defer bus.Close()
 
 	deployment := testutil.Deployment(testutil.Address(t), 1)
-	group := testutil.DeploymentGroups(deployment.Address, 2).Items[0]
+	group := makeDeploymentGroup(deployment.Address, 2)
 	order := testutil.Order(deployment.Address, group.Seq, 3)
 	provider := testutil.Provider(testutil.Address(t), 4)
 
@@ -156,3 +158,32 @@ func TestService_Catchup(t *testing.T) {
 
 	mock.AssertExpectationsForObjects(t, qclient, txclient, creso, cluster)
 }
+
+func makeDeploymentGroup(daddr []byte, nonce uint64) *types.DeploymentGroup {
+	runit := types.ResourceUnit{
+		CPU:    500,
+		Memory: 256 * unit.Mi,
+		Disk:   5 * unit.Gi,
+	}
+
+	rgroup := types.ResourceGroup{
+		Unit:  runit,
+		Count: 2,
+		Price: 35,
+	}
+
+	pattr := types.ProviderAttribute{
+		Name:  "region",
+		Value: "us-west",
+	}
+
+	return &types.DeploymentGroup{
+		Name: strconv.FormatUint(nonce, 10),
+		DeploymentGroupID: types.DeploymentGroupID{
+			Deployment: daddr,
+			Seq:        nonce,
+		},
+		Resources:    []types.ResourceGroup{rgroup},
+		Requirements: []types.ProviderAttribute{pattr},
+	}
+}