Repository: threefoldtech/zos
Pull request #1334 (2021-07-09 09:36:49)

Zos3/capacity reporting


Patch:
diff --git a/client/node.go b/client/node.go
index 86777936..43d456fc 100644
--- a/client/node.go
+++ b/client/node.go
@@ -29,10 +29,10 @@ func (n *NodeClient) DeploymentUpdate(ctx context.Context, dl gridtypes.Deployme
 	return n.bus.Call(ctx, n.nodeTwin, cmd, dl, nil)
 }
 
-func (n *NodeClient) DeploymentGet(ctx context.Context, deploymentID uint32) (dl gridtypes.Deployment, err error) {
+func (n *NodeClient) DeploymentGet(ctx context.Context, contractID uint64) (dl gridtypes.Deployment, err error) {
 	const cmd = "zos.deployment.get"
 	in := args{
-		"deployment_id": deploymentID,
+		"contract_id": contractID,
 	}
 
 	if err = n.bus.Call(ctx, n.nodeTwin, cmd, in, &dl); err != nil {
@@ -42,10 +42,10 @@ func (n *NodeClient) DeploymentGet(ctx context.Context, deploymentID uint32) (dl
 	return dl, nil
 }
 
-func (n *NodeClient) DeploymentDelete(ctx context.Context, deploymentID uint32) error {
+func (n *NodeClient) DeploymentDelete(ctx context.Context, contractID uint64) error {
 	const cmd = "zos.deployment.delete"
 	in := args{
-		"deployment_id": deploymentID,
+		"contract_id": contractID,
 	}
 
 	return n.bus.Call(ctx, n.nodeTwin, cmd, in, nil)
diff --git a/cmds/modules/provisiond/main.go b/cmds/modules/provisiond/main.go
index c2d98e1b..0e336213 100644
--- a/cmds/modules/provisiond/main.go
+++ b/cmds/modules/provisiond/main.go
@@ -257,19 +257,17 @@ func action(cli *cli.Context) error {
 		log.Error().Err(err).Msg("failed to mark module as booted")
 	}
 
-	//TODO: uncomment me
-
-	// reporter, err := NewReported(store, identity, queues)
-	// if err != nil {
-	// 	return errors.Wrap(err, "failed to setup capacity reporter")
-	// }
-	// // also spawn the capacity reporter
-	// go func() {
-	// 	if err := reporter.Run(ctx); err != nil && err != context.Canceled {
-	// 		log.Fatal().Err(err).Msg("capacity reported stopped unexpectedely")
-	// 	}
-	// 	log.Info().Msg("capacity reported stopped")
-	// }()
+	reporter, err := NewReporter(engine, cl, queues)
+	if err != nil {
+		return errors.Wrap(err, "failed to setup capacity reporter")
+	}
+	// also spawn the capacity reporter
+	go func() {
+		if err := reporter.Run(ctx); err != nil && err != context.Canceled {
+			log.Fatal().Err(err).Msg("capacity reported stopped unexpectedely")
+		}
+		log.Info().Msg("capacity reported stopped")
+	}()
 
 	// and start the zbus server in the back ground
 	go func() {
diff --git a/cmds/modules/provisiond/reporter.go b/cmds/modules/provisiond/reporter.go
index 8b6de921..5afedac7 100644
--- a/cmds/modules/provisiond/reporter.go
+++ b/cmds/modules/provisiond/reporter.go
@@ -3,16 +3,20 @@ package provisiond
 import (
 	"bytes"
 	"context"
-	"encoding/hex"
+	"crypto/ed25519"
+	"os"
+	"path/filepath"
 	"time"
 
+	"github.com/centrifuge/go-substrate-rpc-client/v3/types"
 	"github.com/joncrlsn/dque"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
+	"github.com/threefoldtech/zbus"
+	"github.com/threefoldtech/zos/pkg"
 	"github.com/threefoldtech/zos/pkg/environment"
-	"github.com/threefoldtech/zos/pkg/farmer"
 	"github.com/threefoldtech/zos/pkg/gridtypes"
-	"github.com/threefoldtech/zos/pkg/provision/storage"
+	"github.com/threefoldtech/zos/pkg/provision"
 	"github.com/threefoldtech/zos/pkg/stubs"
 	"github.com/threefoldtech/zos/pkg/substrate"
 )
@@ -56,21 +60,30 @@ func (m many) AsError() error {
 	return m
 }
 
+type Consumption struct {
+	substrate.Consumption
+	TwinID uint32
+}
+
+type Report struct {
+	Consumption []Consumption
+}
+
 // Reporter structure
 type Reporter struct {
-	identity  *stubs.IdentityManagerStub
-	storage   *storage.Fs
+	cl        zbus.Client
+	sk        ed25519.PrivateKey
+	engine    provision.Engine
 	queue     *dque.DQue
 	substrate *substrate.Substrate
-	nodeID    string
 }
 
 func reportBuilder() interface{} {
-	return &farmer.Report{}
+	return &Report{}
 }
 
 // NewReporter creates a new capacity reporter
-func NewReporter(store *storage.Fs, identity *stubs.IdentityManagerStub, root string) (*Reporter, error) {
+func NewReporter(engine provision.Engine, cl zbus.Client, root string) (*Reporter, error) {
 	env, err := environment.Get()
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get runtime environment")
@@ -80,54 +93,125 @@ func NewReporter(store *storage.Fs, identity *stubs.IdentityManagerStub, root st
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create substrate client")
 	}
+	const queueName = "consumption"
+	var queue *dque.DQue
+	for i := 0; i < 3; i++ {
+		queue, err = dque.NewOrOpen(queueName, root, 1024, reportBuilder)
+		if err != nil {
+			os.RemoveAll(filepath.Join(root, queueName))
+			continue
+		}
+		break
+	}
 
-	queue, err := dque.NewOrOpen("consumption", root, 1024, reportBuilder)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to setup report persisted queue")
 	}
 
+	identity := stubs.NewIdentityManagerStub(cl)
+	sk := ed25519.PrivateKey(identity.PrivateKey(context.TODO()))
+
 	return &Reporter{
-		storage:   store,
-		identity:  identity,
+		cl:        cl,
+		engine:    engine,
+		sk:        sk,
 		queue:     queue,
 		substrate: sub,
-		nodeID:    identity.NodeID(context.TODO()).Identity(),
 	}, nil
 }
 
-func (r *Reporter) pushOne() error {
+func (r *Reporter) pushOne() ([]Consumption, error) {
 	item, err := r.queue.PeekBlock()
 	if err != nil {
-		return errors.Wrap(err, "failed to peek into capacity queue. #properlyfatal")
+		return nil, errors.Wrap(err, "failed to peek into capacity queue. #properlyfatal")
 	}
 
-	report := item.(*farmer.Report)
+	report := item.(*Report)
 
 	// DEBUG
-	log.Debug().Int64("timestamp", report.Timestamp).Msg("sending capacity report")
-	// TODO: push report to chain
-	panic("not implemented")
-	// if err := r.substrate.NodeReport(r.nodeID, *report); err != nil {
-	// 	return errors.Wrap(err, "failed to publish consumption report")
-	// }
-
-	// only removed if report is reported to
-	// farmer
+	log.Debug().Int("len", len(report.Consumption)).Msgf("sending capacity report")
+
+	consumptions := make([]substrate.Consumption, 0, len(report.Consumption))
+	for _, cmp := range report.Consumption {
+		consumptions = append(consumptions, cmp.Consumption)
+	}
+	if err := r.substrate.Report(r.sk, consumptions); err != nil {
+		return nil, errors.Wrap(err, "failed to publish consumption report")
+	}
+
+	// only removed if report is reported to substrate
 	// remove item from queue
 	_, err = r.queue.Dequeue()
 
-	return err
+	return report.Consumption, err
 }
 
-func (r *Reporter) pusher() {
+func (r *Reporter) pusher(ctx context.Context) {
 	for {
-		if err := r.pushOne(); err != nil {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		// problem is pushOne is a blocker call. so if ctx is canceled
+		// while we are inside pushOne, no way to detect that until the pushOne call
+		// returns
+		reported, err := r.pushOne()
+		if err != nil {
 			log.Error().Err(err).Msg("error while processing capacity report")
-			<-time.After(3 * time.Second)
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(3 * time.Second):
+				continue
+			}
+		}
+
+		log.Debug().Msg("capacity report pushed to chain")
+		if r.queue.Size() == 0 {
+			// we only synchronize once ALL queued reports are pushed.
+			if err := r.synchronize(ctx, reported); err != nil {
+				log.Error().Err(err).Msg("failed to synchronize active contracts")
+			}
 		}
+	}
+}
+
+// synchronize will make sure that the node only runs
+// active contracts.
+func (r *Reporter) synchronize(ctx context.Context, reported []Consumption) error {
+	log.Debug().Msg("synchronize active contracts")
+
+	local := make(map[types.U64]Consumption)
+	for _, report := range reported {
+		local[report.ContractID] = report
+	}
+
+	identity, err := substrate.Identity(r.sk)
+	if err != nil {
+		return errors.Wrap(err, "failed to get node identity")
+	}
+	// the idea here is that we bring ALL active node contracts from chain.
+	// then compare it with what we have atm (the one we just reported)
+	contracts, err := r.substrate.GetNodeContracts(identity.PublicKey, substrate.ContractState{IsCreated: true})
+	if err != nil {
+		return err
+	}
 
-		log.Debug().Msg("capacity report pushed to farmer")
+	for _, contract := range contracts {
+		// is there a consumption report for a contract
+		delete(local, contract.ContractID)
 	}
+	// any LOCAL contract that is not in the map must be decommissioned
+	for _, local := range local {
+		log.Debug().Uint64("contract", uint64(local.ContractID)).Msg("decomission contract because it has been deleted")
+		if err := r.engine.Deprovision(ctx, local.TwinID, uint64(local.ContractID), "contract terminated"); err != nil {
+			log.Error().Err(err).Msgf("failed to decomission contract(%d)", local.ContractID)
+		}
+	}
+
+	return nil
 }
 
 // Run runs the reporter
@@ -136,7 +220,7 @@ func (r *Reporter) Run(ctx context.Context) error {
 	// take into account the following:
 	// every is in seconds.
 
-	go r.pusher()
+	go r.pusher(ctx)
 
 	ticker := time.NewTicker(every * time.Second)
 	defer ticker.Stop()
@@ -154,7 +238,7 @@ func (r *Reporter) Run(ctx context.Context) error {
 			// 5 minute slot.
 			// but if it was stopped before that timestamp, then it will
 			// not be counted.
-			report, err := r.collect(time.Unix(u, 0))
+			report, err := r.collect(ctx, time.Unix(u, 0))
 			if err != nil {
 				log.Error().Err(err).Msg("failed to collect users consumptions")
 				continue
@@ -172,17 +256,20 @@ func (r *Reporter) Run(ctx context.Context) error {
 	}
 }
 
-func (r *Reporter) collect(since time.Time) (rep farmer.Report, err error) {
-	users, err := r.storage.Twins()
+func (r *Reporter) collect(ctx context.Context, since time.Time) (rep Report, err error) {
+	users, err := r.engine.Storage().Twins()
 	if err != nil {
 		return rep, err
 	}
 
-	rep.Timestamp = since.Unix()
-	rep.Consumption = make(map[uint64]farmer.Consumption)
+	// to optimize we get ALL vms metrics in one call.
+	metrics, err := stubs.NewVMModuleStub(r.cl).Metrics(ctx)
+	if err != nil {
+		return Report{}, errors.Wrap(err, "failed to get VMs network metrics")
+	}
 
 	for _, user := range users {
-		cap, err := r.user(since, user)
+		cap, err := r.user(since, user, metrics)
 		if err != nil {
 			log.Error().Err(err).Msg("failed to collect all user capacity")
 			// NOTE: we intentionally not doing a 'continue' or 'return'
@@ -190,81 +277,94 @@ func (r *Reporter) collect(since time.Time) (rep farmer.Report, err error) {
 			// collected some of the user consumption, we still can report that
 		}
 
-		if cap.Capacity.Zero() {
-			continue
-		}
-
-		rep.Consumption[uint64(user)] = cap
+		rep.Consumption = append(rep.Consumption, cap...)
 	}
 
 	return
 }
 
-func (r *Reporter) push(report farmer.Report) error {
-	// create signature
-	var buf bytes.Buffer
-	if err := report.Challenge(&buf); err != nil {
-		return errors.Wrap(err, "failed to create report challenge")
-	}
-
-	signature, err := r.identity.Sign(context.TODO(), buf.Bytes())
-	if err != nil {
-		return errors.Wrap(err, "failed to sign report")
-	}
-
-	report.Signature = hex.EncodeToString(signature)
+func (r *Reporter) push(report Report) error {
 	return r.queue.Enqueue(&report)
 }
 
-func (r *Reporter) user(since time.Time, user uint32) (farmer.Consumption, error) {
+func (r *Reporter) user(since time.Time, user uint32, metrics pkg.MachineMetrics) ([]Consumption, error) {
 	var m many
-	types := gridtypes.Types()
-	consumption := farmer.Consumption{
-		Workloads: make(map[gridtypes.WorkloadType][]gridtypes.WorkloadID),
+
+	var consumptions []Consumption
+	ids, err := r.engine.Storage().ByTwin(user)
+	if err != nil {
+		m = m.append(errors.Wrapf(err, "failed to get reservation for user '%s'", user))
 	}
 
-	for _, typ := range types {
-		consumption.Workloads[typ] = make([]gridtypes.WorkloadID, 0)
-		ids, err := r.storage.ByTwin(user)
+	for _, id := range ids {
+		dl, err := r.engine.Storage().Get(user, id)
 		if err != nil {
-			m = m.append(errors.Wrapf(err, "failed to get reservation for user '%s' type '%s", user, typ))
+			m = m.append(errors.Wrapf(err, "failed to get reservation '%s'", id))
 			continue
 		}
 
-		for _, id := range ids {
-			dl, err := r.storage.Get(user, id)
-			if err != nil {
-				m = m.append(errors.Wrapf(err, "failed to get reservation '%s'", id))
+		consumption := Consumption{
+			TwinID: user,
+			Consumption: substrate.Consumption{
+				ContractID: types.U64(id),
+				Timestamp:  types.U64(since.Unix()),
+			},
+		}
+
+		for i := range dl.Workloads {
+			wl := &dl.Workloads[i]
+
+			if wl.Result.IsNil() {
+				// no results yet
 				continue
 			}
 
-			for i := range dl.Workloads {
-				wl := &dl.Workloads[i]
+			wlID, err := gridtypes.NewWorkloadID(user, id, wl.Name)
+			if err != nil {
+				log.Error().Err(err).Msg("invalid workload id (shouldn't happen here)")
+				continue
+			}
 
-				if wl.Result.IsNil() {
-					// no results yet
+			if r.shouldCount(since, &wl.Result) {
+				cap, err := wl.Capacity()
+				if err != nil {
+					m = m.append(errors.Wrapf(err, "failed to get reservation '%s' capacity", id))
 					continue
 				}
 
-				if r.shouldCount(since, &wl.Result) {
-					cap, err := wl.Capacity()
-					if err != nil {
-						m = m.append(errors.Wrapf(err, "failed to get reservation '%s' capacity", id))
-						continue
-					}
-
-					wlID, _ := gridtypes.NewWorkloadID(user, id, wl.Name)
-					consumption.Workloads[typ] = append(consumption.Workloads[typ], wlID)
-					consumption.Capacity.Add(&cap)
-
-					// special handling for ZMachine types. if they exist
-					// we also need to get network usage.
+				consumption.CRU += types.U64(cap.CRU)
+				consumption.MRU += types.U64(cap.MRU)
+				consumption.HRU += types.U64(cap.HRU)
+				consumption.SRU += types.U64(cap.SRU)
+
+				// special handling for ZMachine types. if they exist
+				// we also need to get network usage.
+				metric, ok := metrics[wlID.String()]
+				if ok {
+					// add metric to consumption
+					consumption.NRU += types.U64(r.computeNU(metric))
 				}
 			}
 		}
+
+		if !consumption.IsEmpty() {
+			consumptions = append(consumptions, consumption)
+		}
 	}
 
-	return consumption, m.AsError()
+	return consumptions, m.AsError()
+}
+
+func (r *Reporter) computeNU(m pkg.MachineMetric) gridtypes.Unit {
+	const (
+		// weights knobs for nu calculations
+		public  float64 = 1.0
+		private float64 = 0.5
+	)
+
+	nu := m.Public.Nu()*public + m.Private.Nu()*private
+
+	return gridtypes.Unit(nu)
 }
 
 func (r *Reporter) shouldCount(since time.Time, result *gridtypes.Result) bool {
@@ -274,11 +374,7 @@ func (r *Reporter) shouldCount(since time.Time, result *gridtypes.Result) bool {
 
 	if result.State == gridtypes.StateDeleted {
 		// if it was stopped before the 'since' .
-		if result.Created.Time().Before(since) {
-			return false
-		}
-		// otherwise it's true
-		return true
+		return !result.Created.Time().Before(since)
 	}
 
 	return false
diff --git a/pkg/farmer/client.go b/pkg/farmer/client.go
deleted file mode 100644
index c391711a..00000000
--- a/pkg/farmer/client.go
+++ /dev/null
@@ -1,85 +0,0 @@
-package farmer
-
-import (
-	"bytes"
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"net/http"
-	"net/url"
-	"path/filepath"
-
-	"github.com/pkg/errors"
-)
-
-const (
-	contentType = "application/json"
-)
-
-var (
-	successCodes = []int{http.StatusOK}
-)
-
-// Client to farmer bot
-type Client struct {
-	url  url.URL
-	base string
-}
-
-// NewClient creates a new instance of client
-func NewClient(u string) (*Client, error) {
-	ul, err := url.Parse(u)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to parse farmer path")
-	}
-
-	path := ul.Path
-	ul.Path = ""
-	return &Client{
-		url:  *ul,
-		base: path,
-	}, nil
-}
-
-func (c *Client) path(rel ...string) string {
-	u := c.url
-	u.Path = filepath.Join(c.base, filepath.Join(rel...))
-	return u.String()
-}
-
-func (c *Client) serialize(o interface{}) (*bytes.Buffer, error) {
-	var bytes bytes.Buffer
-	err := json.NewEncoder(&bytes).Encode(o)
-	return &bytes, err
-}
-
-func (c *Client) response(r *http.Response, o interface{}, codes ...int) error {
-	defer r.Body.Close()
-	if len(codes) == 0 {
-		codes = successCodes
-	}
-
-	in := func(i int, l []int) bool {
-		for _, v := range l {
-			if v == i {
-				return true
-			}
-		}
-		return false
-	}
-
-	if !in(r.StatusCode, codes) {
-		msg, _ := ioutil.ReadAll(r.Body)
-		return fmt.Errorf("invalid response (%s): %s", r.Status, string(msg))
-	}
-
-	defer func() {
-		ioutil.ReadAll(r.Body)
-	}()
-
-	if o != nil {
-		return json.NewDecoder(r.Body).Decode(o)
-	}
-
-	return nil
-}
diff --git a/pkg/farmer/node.go b/pkg/farmer/node.go
deleted file mode 100644
index 611261a7..00000000
--- a/pkg/farmer/node.go
+++ /dev/null
@@ -1,55 +0,0 @@
-package farmer
-
-import (
-	"net/http"
-
-	"github.com/pkg/errors"
-	"github.com/threefoldtech/zos/pkg/gridtypes"
-)
-
-// Location structure
-type Location struct {
-	Latitude  float64 `json:"latitude"`
-	Longitude float64 `json:"longitude"`
-}
-
-// Node structure
-type Node struct {
-	ID       string             `json:"node"`
-	FarmID   uint32             `json:"farm"`
-	Secret   string             `json:"secret"`
-	Location Location           `json:"location"`
-	Capacity gridtypes.Capacity `json:"capacity"`
-}
-
-// NodeRegister register node
-func (c *Client) NodeRegister(node Node) error {
-	url := c.path("nodes")
-	body, err := c.serialize(node)
-	if err != nil {
-		return errors.Wrap(err, "failed to create request body")
-	}
-
-	response, err := http.Post(url, contentType, body)
-	if err != nil {
-		return errors.Wrap(err, "failed to build request")
-	}
-
-	return c.response(response, nil, http.StatusCreated, http.StatusNotModified)
-}
-
-// GatewayRegister registers a node as a gateway
-func (c *Client) GatewayRegister(node Node) error {
-	url := c.path("gateways")
-	body, err := c.serialize(node)
-	if err != nil {
-		return errors.Wrap(err, "failed to create request body")
-	}
-
-	response, err := http.Post(url, contentType, body)
-	if err != nil {
-		return errors.Wrap(err, "failed to build request")
-	}
-
-	return c.response(response, nil, http.StatusCreated, http.StatusNotModified)
-}
diff --git a/pkg/farmer/report.go b/pkg/farmer/report.go
deleted file mode 100644
index 3f4e454a..00000000
--- a/pkg/farmer/report.go
+++ /dev/null
@@ -1,89 +0,0 @@
-package farmer
-
-import (
-	"bytes"
-	"crypto/ed25519"
-	"encoding/hex"
-	"fmt"
-	"io"
-	"net/http"
-
-	"github.com/pkg/errors"
-	"github.com/threefoldtech/zos/pkg/gridtypes"
-)
-
-// Consumption struct
-type Consumption struct {
-	Workloads map[gridtypes.WorkloadType][]gridtypes.WorkloadID `json:"workloads"`
-	Capacity  gridtypes.Capacity                                `json:"capacity"`
-}
-
-// Report is a user report
-type Report struct {
-	Timestamp   int64                  `json:"timestamp"`
-	Signature   string                 `json:"signature"`
-	Consumption map[uint64]Consumption `json:"consumption"`
-}
-
-// Challenge build a report challenge for signing
-func (r *Report) Challenge(w io.Writer) error {
-	var total gridtypes.Capacity
-	for _, consumption := range r.Consumption {
-		total.Add(&consumption.Capacity)
-	}
-
-	if _, err := fmt.Fprintf(w, "%d", r.Timestamp); err != nil {
-		return err
-	}
-
-	if _, err := fmt.Fprintf(w, "%d", total.CRU); err != nil {
-		return err
-	}
-
-	if _, err := fmt.Fprintf(w, "%d", total.MRU); err != nil {
-		return err
-	}
-
-	if _, err := fmt.Fprintf(w, "%d", total.SRU); err != nil {
-		return err
-	}
-
-	if _, err := fmt.Fprintf(w, "%d", total.HRU); err != nil {
-		return err
-	}
-
-	if _, err := fmt.Fprintf(w, "%d", total.IPV4U); err != nil {
-		return err
-	}
-
-	return nil
-}
-
-// Sign the report given sk
-func (r *Report) Sign(sk ed25519.PrivateKey) error {
-	var buf bytes.Buffer
-	if err := r.Challenge(&buf); err != nil {
-		return errors.Wrap(err, "failed to build signing challenge")
-	}
-
-	signature := ed25519.Sign(sk, buf.Bytes())
-	r.Signature = hex.EncodeToString(signature)
-	return nil
-}
-
-// NodeReport send usage reports to farmer
-func (c *Client) NodeReport(nodeID string, report Report) error {
-	url := c.path("nodes", nodeID, "reports")
-
-	body, err := c.serialize(report)
-	if err != nil {
-		return errors.Wrap(err, "failed to create request body")
-	}
-
-	response, err := http.Post(url, contentType, body)
-	if err != nil {
-		return errors.Wrap(err, "failed to build request")
-	}
-
-	return c.response(response, nil, http.StatusCreated, http.StatusOK)
-}
diff --git a/pkg/farmer/substrate.go b/pkg/farmer/substrate.go
deleted file mode 100644
index fa190b56..00000000
--- a/pkg/farmer/substrate.go
+++ /dev/null
@@ -1,54 +0,0 @@
-package farmer
-
-import (
-	"fmt"
-
-	"github.com/pkg/errors"
-	"github.com/threefoldtech/zos/pkg/substrate"
-)
-
-const (
-	//DefaultFarmerTwinPort twin port
-	DefaultFarmerTwinPort = uint16(3000)
-)
-
-// GetFarmTwin gets the IP of a farmer twin given the substrate db url
-// and the farm id
-func GetFarmTwin(url string, id uint32) (*substrate.Twin, error) {
-	sub, err := substrate.NewSubstrate(url)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to connect to substrate")
-	}
-
-	farm, err := sub.GetFarm(uint32(id))
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed to get farm '%d'", id)
-	}
-
-	return sub.GetTwin(uint32(farm.TwinID))
-}
-
-// NewClientFromSubstrate gets a farmer twin client from a substrate url and
-// the farm id
-func NewClientFromSubstrate(url string, id uint32, port ...uint16) (*Client, error) {
-	twin, err := GetFarmTwin(url, id)
-	if err != nil {
-		return nil, errors.Wrap(err, "failed to get farmer twin ip")
-	}
-
-	ip := twin.IPAddress()
-	if len(ip) == 0 {
-		return nil, fmt.Errorf("invalid ip address associated with farmer twin")
-	}
-
-	p := DefaultFarmerTwinPort
-	if len(port) == 1 {
-		p = port[0]
-	} else if len(port) > 1 {
-		return nil, fmt.Errorf("only one port is supported")
-	}
-
-	twinURL := fmt.Sprintf("http://[%s]:%d", ip.String(), p)
-
-	return NewClient(twinURL)
-}
diff --git a/pkg/gridtypes/deployment.go b/pkg/gridtypes/deployment.go
index bf26e7a5..07c353b4 100644
--- a/pkg/gridtypes/deployment.go
+++ b/pkg/gridtypes/deployment.go
@@ -39,7 +39,7 @@ type KeyGetter interface {
 type Deployment struct {
 	Version              int                  `json:"version"`
 	TwinID               uint32               `json:"twin_id"`
-	ContractID           uint32               `json:"contract_id"`
+	ContractID           uint64               `json:"contract_id"`
 	Metadata             string               `json:"metadata"`
 	Description          string               `json:"description"`
 	Expiration           Timestamp            `json:"expiration"`
@@ -163,6 +163,10 @@ func (d *Deployment) Challenge(w io.Writer) error {
 		}
 	}
 
+	if err := d.SignatureRequirement.Challenge(w); err != nil {
+		return err
+	}
+
 	return nil
 }
 
diff --git a/pkg/gridtypes/id.go b/pkg/gridtypes/id.go
index 926e8ec5..800db024 100644
--- a/pkg/gridtypes/id.go
+++ b/pkg/gridtypes/id.go
@@ -45,7 +45,7 @@ func (i WorkloadID) String() string {
 }
 
 // Parts split id into building parts
-func (i WorkloadID) Parts() (twin, deployment uint32, name string, err error) {
+func (i WorkloadID) Parts() (twin uint32, deployment uint64, name string, err error) {
 	_, err = fmt.Sscanf(string(i), "%d-%d-%s", &twin, &deployment, &name)
 	return
 }
@@ -64,7 +64,7 @@ func IsValidName(n Name) error {
 }
 
 // NewWorkloadID creates a new global ID from it's parts
-func NewWorkloadID(twin uint32, deployment uint32, name Name) (WorkloadID, error) {
+func NewWorkloadID(twin uint32, deployment uint64, name Name) (WorkloadID, error) {
 	if err := IsValidName(name); err != nil {
 		return "", err
 	}
diff --git a/pkg/gridtypes/workload.go b/pkg/gridtypes/workload.go
index 1d7e9800..470c1756 100644
--- a/pkg/gridtypes/workload.go
+++ b/pkg/gridtypes/workload.go
@@ -104,8 +104,6 @@ type Workload struct {
 	Type WorkloadType `json:"type"`
 	// Data is the reservation type arguments.
 	Data json.RawMessage `json:"data"`
-	// Date of creation
-	Created Timestamp `json:"created"`
 	// Metadata is custom user metadata
 	Metadata string `json:"metadata"`
 	//Description
diff --git a/pkg/gridtypes/zos/kubernetes.go b/pkg/gridtypes/zos/kubernetes.go
index ea8d5edc..64643845 100644
--- a/pkg/gridtypes/zos/kubernetes.go
+++ b/pkg/gridtypes/zos/kubernetes.go
@@ -1,12 +1,6 @@
 package zos
 
 import (
-	"fmt"
-	"io"
-	"net"
-	"strings"
-
-	"github.com/pkg/errors"
 	"github.com/threefoldtech/zos/pkg/gridtypes"
 )
 
@@ -84,56 +78,3 @@ var vmSize = map[uint8]gridtypes.Capacity{
 		MRU: 1 * gridtypes.Gigabyte,
 	},
 }
-
-// Kubernetes reservation data
-type Kubernetes struct {
-	ZMachine `json:",inline"`
-	// ClusterSecret is the hex encoded encrypted(?) cluster secret.
-	ClusterSecret string `json:"cluster_secret"`
-	// MasterIPs define the URL's for the kubernetes master nodes. If this
-	// list is empty, this node is considered to be a master node.
-	MasterIPs []net.IP `json:"master_ips"`
-
-	DatastoreEndpoint     string `json:"datastore_endpoint"`
-	DisableDefaultIngress bool   `json:"disable_default_ingress"`
-}
-
-// Challenge implementation
-func (k Kubernetes) Challenge(b io.Writer) error {
-	k.ZMachine.Challenge(b)
-
-	if _, err := fmt.Fprintf(b, "%s", k.ClusterSecret); err != nil {
-		return err
-	}
-	for _, ip := range k.MasterIPs {
-		if _, err := fmt.Fprintf(b, "%s", ip.String()); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-// Valid implementation
-func (k Kubernetes) Valid(getter gridtypes.WorkloadGetter) error {
-	err := k.ZMachine.Valid(getter)
-	if err != nil {
-		return err
-	}
-	if strings.ContainsAny(k.ClusterSecret, " \t\r\n\f") {
-		return errors.New("cluster secret shouldn't contain whitespace chars")
-	}
-
-	for _, ip := range k.MasterIPs {
-		if ip.To4() == nil && ip.To16() == nil {
-			return errors.New("invalid master IP")
-		}
-	}
-	return nil
-}
-
-// KubernetesResult result returned by k3s reservation
-type KubernetesResult struct {
-	ID string `json:"id"`
-	IP string `json:"ip"`
-}
diff --git a/pkg/gridtypes/zos/types.go b/pkg/gridtypes/zos/types.go
index e4283cd7..63a70f17 100644
--- a/pkg/gridtypes/zos/types.go
+++ b/pkg/gridtypes/zos/types.go
@@ -7,16 +7,12 @@ import (
 )
 
 const (
-	// ContainerType type
-	ContainerType gridtypes.WorkloadType = "container"
 	// ZMountType type
 	ZMountType gridtypes.WorkloadType = "zmount"
 	// NetworkType type
 	NetworkType gridtypes.WorkloadType = "network"
 	// ZDBType type
 	ZDBType gridtypes.WorkloadType = "zdb"
-	// KubernetesType type
-	KubernetesType gridtypes.WorkloadType = "kubernetes"
 	// ZMachineType type
 	ZMachineType gridtypes.WorkloadType = "zmachine"
 
@@ -28,7 +24,6 @@ func init() {
 	gridtypes.RegisterType(ZMountType, ZMount{})
 	gridtypes.RegisterType(NetworkType, Network{})
 	gridtypes.RegisterType(ZDBType, ZDB{})
-	gridtypes.RegisterType(KubernetesType, Kubernetes{})
 	gridtypes.RegisterType(ZMachineType, ZMachine{})
 	gridtypes.RegisterType(PublicIPType, PublicIP{})
 }
diff --git a/pkg/primitives/statistics.go b/pkg/primitives/statistics.go
index 81b69d04..9ad5968e 100644
--- a/pkg/primitives/statistics.go
+++ b/pkg/primitives/statistics.go
@@ -149,7 +149,7 @@ func (s *Statistics) Provision(ctx context.Context, wl *gridtypes.WorkloadWithID
 	}
 
 	// we add extra overhead for some workload types here
-	if wl.Type == zos.KubernetesType {
+	if wl.Type == zos.ZMachineType {
 		// we add the min of 5% of allocated memory or 1G
 		needed.MRU += gridtypes.Min(needed.MRU*5/100, gridtypes.Gigabyte)
 	} // TODO: other types ?
diff --git a/pkg/primitives/zmount.go b/pkg/primitives/zmount.go
index ba23be25..9d7e31c9 100644
--- a/pkg/primitives/zmount.go
+++ b/pkg/primitives/zmount.go
@@ -38,7 +38,6 @@ func (p *Primitives) zMountProvision(ctx context.Context, wl *gridtypes.Workload
 }
 
 func (p *Primitives) zMountDecommission(ctx context.Context, wl *gridtypes.WorkloadWithID) error {
-	storageClient := stubs.NewStorageModuleStub(p.zbus)
-
-	return storageClient.ReleaseFilesystem(ctx, wl.ID.String())
+	vdisk := stubs.NewVDiskModuleStub(p.zbus)
+	return vdisk.Deallocate(ctx, wl.ID.String())
 }
diff --git a/pkg/provision/api/reservation.go b/pkg/provision/api/reservation.go
index fd31f8fb..293d4718 100644
--- a/pkg/provision/api/reservation.go
+++ b/pkg/provision/api/reservation.go
@@ -59,7 +59,7 @@ func (a *Workloads) createOrUpdate(request *http.Request) (interface{}, mw.Respo
 	return nil, mw.Accepted()
 }
 
-func (a *Workloads) parseIDs(request *http.Request) (twin, id uint32, err error) {
+func (a *Workloads) parseIDs(request *http.Request) (twin uint32, id uint64, err error) {
 	twinVar := mux.Vars(request)["twin"]
 	if len(twinVar) == 0 {
 		return twin, id, fmt.Errorf("invalid twin id format")
@@ -74,12 +74,12 @@ func (a *Workloads) parseIDs(request *http.Request) (twin, id uint32, err error)
 		return twin, id, err
 	}
 
-	idU, err := strconv.ParseUint(idVar, 10, 32)
+	idU, err := strconv.ParseUint(idVar, 10, 64)
 	if err != nil {
 		return twin, id, err
 	}
 
-	return uint32(twinU), uint32(idU), nil
+	return uint32(twinU), idU, nil
 }
 
 func (a *Workloads) delete(request *http.Request) (interface{}, mw.Response) {
diff --git a/pkg/provision/engine.go b/pkg/provision/engine.go
index f931a78c..ab6c80ce 100644
--- a/pkg/provision/engine.go
+++ b/pkg/provision/engine.go
@@ -280,7 +280,7 @@ func (e *NativeEngine) Provision(ctx context.Context, deployment gridtypes.Deplo
 }
 
 // Deprovision workload
-func (e *NativeEngine) Deprovision(ctx context.Context, twin, id uint32, reason string) error {
+func (e *NativeEngine) Deprovision(ctx context.Context, twin uint32, id uint64, reason string) error {
 	deployment, err := e.storage.Get(twin, id)
 	if err != nil {
 		return err
@@ -288,7 +288,7 @@ func (e *NativeEngine) Deprovision(ctx context.Context, twin, id uint32, reason
 
 	log.Debug().
 		Uint32("twin", deployment.TwinID).
-		Uint32("contract", deployment.ContractID).
+		Uint64("contract", deployment.ContractID).
 		Msg("schedule for deprovision")
 
 	job := engineJob{
@@ -348,15 +348,23 @@ func (e *NativeEngine) Run(root context.Context) error {
 		job := obj.(*engineJob)
 		ctx := context.WithValue(root, deploymentKey{}, &job.Target)
 
-		// contract processing
-		ctx, err = e.contract(ctx, &job.Target)
-		if err != nil {
-			job.Target.SetError(err)
-			if err := e.storage.Set(job.Target); err != nil {
-				log.Error().Err(err).Msg("failed to set deployment global error")
+		// contract validation
+		// this should ONLY be done on provosion and update operation
+		if job.Op != opDeprovision {
+			// otherwise, contract validation is needed
+			ctx, err = e.contract(ctx, &job.Target)
+			if err != nil {
+				log.Error().Err(err).Uint64("contract", job.Target.ContractID).Msg("contact validation fails")
+				job.Target.SetError(err)
+				if err := e.storage.Set(job.Target); err != nil {
+					log.Error().Err(err).Msg("failed to set deployment global error")
+				}
+				e.queue.Dequeue()
+
+				continue
 			}
 
-			continue
+			log.Debug().Uint64("contract", job.Target.ContractID).Msg("contact validation pass")
 		}
 
 		switch job.Op {
@@ -386,7 +394,7 @@ func (e *NativeEngine) Run(root context.Context) error {
 			// and update to reflect the current result on those workloads.
 			update, err := job.Source.Upgrade(&job.Target)
 			if err != nil {
-				log.Error().Err(err).Uint32("twin", job.Target.TwinID).Uint32("id", job.Target.ContractID).Msg("failed to get update procedure")
+				log.Error().Err(err).Uint32("twin", job.Target.TwinID).Uint64("id", job.Target.ContractID).Msg("failed to get update procedure")
 				break
 			}
 
@@ -418,6 +426,7 @@ func (e *NativeEngine) contract(ctx context.Context, dl *gridtypes.Deployment) (
 	}
 
 	ctx = context.WithValue(ctx, contractKey{}, contract)
+
 	if contract.Node.String() != e.address {
 		return nil, fmt.Errorf("invalid node address in contract")
 	}
@@ -428,7 +437,7 @@ func (e *NativeEngine) contract(ctx context.Context, dl *gridtypes.Deployment) (
 	}
 
 	if contract.DeploymentHash != hex.EncodeToString(hash) {
-		return nil, errors.Wrap(err, "deployment hash messmatch")
+		return nil, fmt.Errorf("contract hash does not match deployment hash")
 	}
 
 	return ctx, nil
@@ -452,7 +461,7 @@ func (e *NativeEngine) boot(root context.Context) error {
 		for _, id := range ids {
 			dl, err := storage.Get(twin, id)
 			if err != nil {
-				log.Error().Err(err).Uint32("twin", twin).Uint32("id", id).Msg("failed to load deployment")
+				log.Error().Err(err).Uint32("twin", twin).Uint64("id", id).Msg("failed to load deployment")
 				continue
 			}
 			// unfortunately we have to inject this value here
@@ -497,7 +506,7 @@ func (e *NativeEngine) uninstallDeployment(ctx context.Context, getter gridtypes
 			twin, deployment, name, _ := wl.ID.Parts()
 			log := log.With().
 				Uint32("twin", twin).
-				Uint32("deployment", deployment).
+				Uint64("deployment", deployment).
 				Str("name", name).
 				Str("type", wl.Type.String()).
 				Logger()
@@ -528,7 +537,7 @@ func (e *NativeEngine) updateDeployment(ctx context.Context, getter gridtypes.Wo
 			twin, deployment, name, _ := wl.ID.Parts()
 			log := log.With().
 				Uint32("twin", twin).
-				Uint32("deployment", deployment).
+				Uint64("deployment", deployment).
 				Str("name", name).
 				Str("type", wl.Type.String()).
 				Logger()
@@ -584,7 +593,7 @@ func (e *NativeEngine) installDeployment(ctx context.Context, getter gridtypes.W
 			twin, deployment, name, _ := wl.ID.Parts()
 			log := log.With().
 				Uint32("twin", twin).
-				Uint32("deployment", deployment).
+				Uint64("deployment", deployment).
 				Str("name", name).
 				Str("type", wl.Type.String()).
 				Logger()
diff --git a/pkg/provision/interface.go b/pkg/provision/interface.go
index 1bab302a..1994fef3 100644
--- a/pkg/provision/interface.go
+++ b/pkg/provision/interface.go
@@ -19,7 +19,7 @@ type Engine interface {
 	// means that workload has been committed to storage (accepts)
 	// and will be processes later
 	Provision(ctx context.Context, wl gridtypes.Deployment) error
-	Deprovision(ctx context.Context, twin, id uint32, reason string) error
+	Deprovision(ctx context.Context, twin uint32, id uint64, reason string) error
 	Update(ctx context.Context, update gridtypes.Deployment) error
 	Storage() Storage
 	Twins() Twins
@@ -55,9 +55,9 @@ var (
 type Storage interface {
 	Add(wl gridtypes.Deployment) error
 	Set(wl gridtypes.Deployment) error
-	Get(twin, deployment uint32) (gridtypes.Deployment, error)
+	Get(twin uint32, deployment uint64) (gridtypes.Deployment, error)
 	Twins() ([]uint32, error)
-	ByTwin(twin uint32) ([]uint32, error)
+	ByTwin(twin uint32) ([]uint64, error)
 }
 
 // Janitor interface
diff --git a/pkg/provision/mbus/reservation.go b/pkg/provision/mbus/reservation.go
index 01150268..9252120d 100644
--- a/pkg/provision/mbus/reservation.go
+++ b/pkg/provision/mbus/reservation.go
@@ -14,7 +14,7 @@ import (
 )
 
 type idArgs struct {
-	DeploymentID uint32 `json:"deployment_id"`
+	ContractID uint64 `json:"contract_id"`
 }
 
 func (d *Deployments) createOrUpdate(ctx context.Context, payload []byte, update bool) (interface{}, mw.Response) {
@@ -73,7 +73,7 @@ func (d *Deployments) delete(ctx context.Context, payload []byte) (interface{},
 	ctx, cancel := context.WithTimeout(ctx, 3*time.Minute)
 	defer cancel()
 
-	err = d.engine.Deprovision(ctx, rmb.GetTwinID(ctx), args.DeploymentID, "requested by user")
+	err = d.engine.Deprovision(ctx, rmb.GetTwinID(ctx), args.ContractID, "requested by user")
 	if err == context.DeadlineExceeded {
 		return nil, mw.Unavailable(ctx.Err())
 	} else if errors.Is(err, provision.ErrDeploymentNotExists) {
@@ -92,7 +92,7 @@ func (d *Deployments) get(ctx context.Context, payload []byte) (interface{}, mw.
 		return nil, mw.Error(err)
 	}
 
-	deployment, err := d.engine.Storage().Get(rmb.GetTwinID(ctx), args.DeploymentID)
+	deployment, err := d.engine.Storage().Get(rmb.GetTwinID(ctx), args.ContractID)
 	if errors.Is(err, provision.ErrDeploymentNotExists) {
 		return nil, mw.NotFound(fmt.Errorf("workload not found"))
 	} else if err != nil {
diff --git a/pkg/provision/storage/storage.go b/pkg/provision/storage/storage.go
index eb336bac..b7260f13 100644
--- a/pkg/provision/storage/storage.go
+++ b/pkg/provision/storage/storage.go
@@ -147,20 +147,20 @@ func (s *Fs) get(path string) (gridtypes.Deployment, error) {
 }
 
 // Get gets a workload by id
-func (s *Fs) Get(twin, deployment uint32) (gridtypes.Deployment, error) {
+func (s *Fs) Get(twin uint32, deployment uint64) (gridtypes.Deployment, error) {
 	path := s.rooted(filepath.Join(fmt.Sprint(twin), fmt.Sprint(deployment)))
 
 	return s.get(path)
 }
 
 // ByTwin return list of deployment for given twin id
-func (s *Fs) ByTwin(twin uint32) ([]uint32, error) {
+func (s *Fs) ByTwin(twin uint32) ([]uint64, error) {
 	s.m.RLock()
 	defer s.m.RUnlock()
 	return s.byTwin(twin)
 }
 
-func (s *Fs) byTwin(twin uint32) ([]uint32, error) {
+func (s *Fs) byTwin(twin uint32) ([]uint64, error) {
 	base := filepath.Join(s.root, fmt.Sprint(twin))
 
 	entities, err := ioutil.ReadDir(base)
@@ -169,7 +169,7 @@ func (s *Fs) byTwin(twin uint32) ([]uint32, error) {
 	} else if err != nil {
 		return nil, errors.Wrap(err, "failed to list twin directory")
 	}
-	ids := make([]uint32, 0, len(entities))
+	ids := make([]uint64, 0, len(entities))
 	for _, entry := range entities {
 		if entry.IsDir() {
 			continue
@@ -181,7 +181,7 @@ func (s *Fs) byTwin(twin uint32) ([]uint32, error) {
 			continue
 		}
 
-		ids = append(ids, uint32(id))
+		ids = append(ids, uint64(id))
 	}
 
 	return ids, nil
diff --git a/pkg/substrate/account.go b/pkg/substrate/account.go
index 203c797d..fb756bc4 100644
--- a/pkg/substrate/account.go
+++ b/pkg/substrate/account.go
@@ -11,11 +11,16 @@ import (
 	"github.com/cenkalti/backoff"
 	"github.com/centrifuge/go-substrate-rpc-client/v3/signature"
 	"github.com/centrifuge/go-substrate-rpc-client/v3/types"
+	"github.com/jbenet/go-base58"
 	"github.com/pkg/errors"
 	"github.com/vedhavyas/go-subkey"
 	subkeyEd25519 "github.com/vedhavyas/go-subkey/ed25519"
 )
 
+const (
+	network = 42
+)
+
 // AccountID type
 type AccountID types.AccountID
 
@@ -26,13 +31,13 @@ func (a AccountID) PublicKey() ed25519.PublicKey {
 
 // String return string representation of account
 func (a AccountID) String() string {
-	address, _ := subkey.SS58Address(a[:], 42)
+	address, _ := subkey.SS58Address(a[:], network)
 	return address
 }
 
 // MarshalJSON implementation
 func (a AccountID) MarshalJSON() ([]byte, error) {
-	address, err := subkey.SS58Address(a[:], 42)
+	address, err := subkey.SS58Address(a[:], network)
 	if err != nil {
 		return nil, err
 	}
@@ -40,6 +45,20 @@ func (a AccountID) MarshalJSON() ([]byte, error) {
 	return json.Marshal(address)
 }
 
+// FromAddress creates an AccountID from a SS58 address
+func FromAddress(address string) (account AccountID, err error) {
+	bytes := base58.Decode(address)
+	if len(bytes) != 3+len(account) {
+		return account, fmt.Errorf("invalid address length")
+	}
+	if bytes[0] != network {
+		return account, fmt.Errorf("invalid address format")
+	}
+
+	copy(account[:], bytes[1:len(account)+1])
+	return
+}
+
 // keyringPairFromSecret creates KeyPair based on seed/phrase and network
 // Leave network empty for default behavior
 func keyringPairFromSecret(seedOrPhrase string, network uint8) (signature.KeyringPair, error) {
diff --git a/pkg/substrate/contract.go b/pkg/substrate/contract.go
index 0d468fb8..b0e4ad9f 100644
--- a/pkg/substrate/contract.go
+++ b/pkg/substrate/contract.go
@@ -1,6 +1,7 @@
 package substrate
 
 import (
+	"crypto/ed25519"
 	"fmt"
 
 	"github.com/centrifuge/go-substrate-rpc-client/v3/scale"
@@ -10,8 +11,9 @@ import (
 
 // ContractState enum
 type ContractState struct {
-	IsCreated  bool
-	IsDeployed bool
+	IsCreated    bool
+	IsDeleted    bool
+	IsOutOfFunds bool
 }
 
 // Decode implementation for the enum type
@@ -25,7 +27,9 @@ func (r *ContractState) Decode(decoder scale.Decoder) error {
 	case 0:
 		r.IsCreated = true
 	case 1:
-		r.IsDeployed = true
+		r.IsDeleted = true
+	case 2:
+		r.IsOutOfFunds = true
 	default:
 		return fmt.Errorf("unknown CertificateType value")
 	}
@@ -33,6 +37,19 @@ func (r *ContractState) Decode(decoder scale.Decoder) error {
 	return nil
 }
 
+// Encode implementation
+func (r ContractState) Encode(encoder scale.Encoder) (err error) {
+	if r.IsCreated {
+		err = encoder.PushByte(0)
+	} else if r.IsDeleted {
+		err = encoder.PushByte(1)
+	} else if r.IsOutOfFunds {
+		err = encoder.PushByte(2)
+	}
+
+	return
+}
+
 // Contract structure
 type Contract struct {
 	Versioned
@@ -46,6 +63,7 @@ type Contract struct {
 	LastUpdated        types.U64
 	PreviousNUReported types.U64
 	PublicIPs          []PublicIP
+	UnBilled           types.U64
 }
 
 // GetContract we should not have calls to create contract, instead only get
@@ -63,6 +81,28 @@ func (s *Substrate) GetContract(id uint64) (*Contract, error) {
 	return s.getContract(key)
 }
 
+// GetNodeContracts gets all contracts on a node (pk) in given state
+func (s *Substrate) GetNodeContracts(pk []byte, state ContractState) ([]Contract, error) {
+	stateBytes, err := types.EncodeToBytes(state)
+	if err != nil {
+		return nil, err
+	}
+	key, err := types.CreateStorageKey(s.meta, "SmartContractModule", "NodeContracts", pk, stateBytes, nil)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create substrate query key")
+	}
+	var contracts []Contract
+	ok, err := s.cl.RPC.State.GetStorageLatest(key, &contracts)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to lookup contracts")
+	}
+
+	if !ok {
+		return nil, errors.Wrap(ErrNotFound, "contracts not found")
+	}
+
+	return contracts, nil
+}
 func (s *Substrate) getContract(key types.StorageKey) (*Contract, error) {
 	raw, err := s.cl.RPC.State.GetStorageRawLatest(key)
 	if err != nil {
@@ -91,3 +131,34 @@ func (s *Substrate) getContract(key types.StorageKey) (*Contract, error) {
 
 	return &node, nil
 }
+
+// Consumption structure
+type Consumption struct {
+	ContractID types.U64
+	Timestamp  types.U64
+	CRU        types.U64 `json:"cru"`
+	SRU        types.U64 `json:"sru"`
+	HRU        types.U64 `json:"hru"`
+	MRU        types.U64 `json:"mru"`
+	NRU        types.U64 `json:"nru"`
+}
+
+// IsEmpty true if consumption is zero
+func (s *Consumption) IsEmpty() bool {
+	//Unit = gridtypes.Megabyte
+	return s.CRU == 0 && s.SRU == 0 && s.HRU == 0 && s.MRU == 0
+}
+
+// Report send a capacity report to substrate
+func (s *Substrate) Report(sk ed25519.PrivateKey, consumptions []Consumption) error {
+	c, err := types.NewCall(s.meta, "SmartContractModule.add_reports", consumptions)
+	if err != nil {
+		return errors.Wrap(err, "failed to create call")
+	}
+
+	if err := s.call(sk, c); err != nil {
+		return errors.Wrap(err, "failed to create report")
+	}
+
+	return nil
+}
diff --git a/pkg/vm.go b/pkg/vm.go
index 1cd8d786..dc16ffc0 100644
--- a/pkg/vm.go
+++ b/pkg/vm.go
@@ -195,6 +195,24 @@ type NetMetric struct {
 	NetTxBytes   uint64 `json:"net_tx_bytes"`
 }
 
+// Nu calculate network units
+func (n *NetMetric) Nu() float64 {
+	const (
+		// weights knobs for nu calculations
+		bytes   float64 = 1.0
+		packets float64 = 0.2
+		rx      float64 = 1.0
+		tx      float64 = 1.0
+	)
+
+	nu := float64(n.NetRxBytes)*bytes*rx +
+		float64(n.NetTxBytes)*bytes*tx +
+		float64(n.NetRxPackets)*packets*rx +
+		float64(n.NetTxPackets)*packets*tx
+
+	return nu
+}
+
 // MachineMetric is a container for metrics from multiple networks
 // currently only groped as private (wiregaurd + yggdrasil), and public (public Ips)
 type MachineMetric struct {
diff --git a/specs/grid3/readme.md b/specs/grid3/readme.md
index 92230f9f..65e1024a 100644
--- a/specs/grid3/readme.md
+++ b/specs/grid3/readme.md
@@ -77,12 +77,14 @@ Once the contract is terminated, the Ips are returned back to the farmer pool.
 Each farmer object is assigned a Pricing Policy object:
 The pricing policy defines:
 - Currency (TFT, USD, etc…) _we will probably drop that_
+- Unit (Bytes, Kilobyte, Megabyte, Gigabytes)
 - SU
 - CU
 - NU
 
 ## General notes:
 - Each price defines a price for a single UNIT/SECOND. So for example SU is the price of a single Storage Unit per second where a Storage Unit can be 1 Gigabyte
+- NU is an exception because that's the price PER unit (regardless of number of seconds). For example if NU is 10 and Unit ig Gigabyte, so 1 Gigabyte of network traffic costs 10 mill.
 - Price is defined as `mil` of the currency. 1 UNIT = 1,000,000,0 mil
 
 ### Example:
diff --git a/specs/grid3/uml/rmb.wsd b/specs/grid3/uml/rmb.wsd
new file mode 100644
index 00000000..3a0dec4a
--- /dev/null
+++ b/specs/grid3/uml/rmb.wsd
@@ -0,0 +1,19 @@
+@startuml
+actor Client as cl
+participant RMB as rmb
+participant Node as zos
+
+cl --> rmb: Create an RMB message \n(cmd: zos.deployment.deploy, dat: base64(json(deployment)))
+rmb -> zos: Decode payload(dat). \nSo node receive a deployment object
+zos -> zos: Verify hash against contract hash\nVerify signature of the twin
+zos -> rmb: Accepted Or Error
+rmb --> cl: Response from the node
+group If valid
+zos -> zos: Deploy
+end group
+...
+cl --> rmb: Get Deployment (cmd: zos.deployment.get)
+rmb -> zos: Call get method
+zos -> rmb: Get Response
+rmb --> cl: Response
+@enduml


Test patch:
diff --git a/pkg/provision/storage/storage_test.go b/pkg/provision/storage/storage_test.go
index 4fc573ee..de055813 100644
--- a/pkg/provision/storage/storage_test.go
+++ b/pkg/provision/storage/storage_test.go
@@ -43,7 +43,7 @@ func TestStorageAdd(t *testing.T) {
 	require.NoError(err)
 
 	twin := uint32(1)
-	id := uint32(1)
+	id := uint64(1)
 	err = store.Add(gridtypes.Deployment{
 		TwinID:      twin,
 		ContractID:  id,
@@ -74,7 +74,7 @@ func TestStorageSet(t *testing.T) {
 	require.NoError(err)
 
 	twin := uint32(1)
-	id := uint32(1)
+	id := uint64(1)
 	deployment := gridtypes.Deployment{
 		TwinID:      twin,
 		ContractID:  id,
@@ -110,7 +110,7 @@ func TestStorageGet(t *testing.T) {
 	store, err := NewFSStore(root)
 	require.NoError(err)
 	twin := uint32(1)
-	id := uint32(1)
+	id := uint64(1)
 	deployment := gridtypes.Deployment{
 		TwinID:      twin,
 		ContractID:  id,
diff --git a/pkg/substrate/account_test.go b/pkg/substrate/account_test.go
new file mode 100644
index 00000000..362d58ff
--- /dev/null
+++ b/pkg/substrate/account_test.go
@@ -0,0 +1,18 @@
+package substrate
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAddress(t *testing.T) {
+	require := require.New(t)
+
+	address := "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"
+
+	account, err := FromAddress(address)
+	require.NoError(err)
+
+	require.Equal(address, account.String())
+}