Repository: threefoldtech/zos
Pull request #1619 (2022-03-11 16:00:55)

New reporting
- Track all VMs consumption every 5 min, create a report of the total consumption during a window every hour 
- Report include 
  - Contract ID
  - Report window
  - Report timestap
  - Network consumption only.
- It introduce only a new way to synchronize contracts with the grid, and to handle cancellation events. 

To be done
- Set contract used capacity on creation/update



Patch:
diff --git a/cmds/modules/noded/main.go b/cmds/modules/noded/main.go
index 9ecee25c..07783243 100644
--- a/cmds/modules/noded/main.go
+++ b/cmds/modules/noded/main.go
@@ -227,6 +227,7 @@ func action(cli *cli.Context) error {
 	identityd := stubs.NewIdentityManagerStub(redis)
 	sk := ed25519.PrivateKey(identityd.PrivateKey(ctx))
 	id, err := substrate.NewIdentityFromEd25519Key(sk)
+	log.Info().Str("address", id.Address()).Msg("node address")
 	if err != nil {
 		return err
 	}
diff --git a/cmds/modules/provisiond/events.go b/cmds/modules/provisiond/events.go
new file mode 100644
index 00000000..a1546533
--- /dev/null
+++ b/cmds/modules/provisiond/events.go
@@ -0,0 +1,144 @@
+package provisiond
+
+import (
+	"context"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+	"github.com/threefoldtech/substrate-client"
+	"github.com/threefoldtech/zbus"
+	"github.com/threefoldtech/zos/pkg"
+	"github.com/threefoldtech/zos/pkg/provision"
+	"github.com/threefoldtech/zos/pkg/stubs"
+)
+
+type ContractEventHandler struct {
+	node   uint32
+	pool   substrate.Manager
+	engine provision.Engine
+	cl     zbus.Client
+}
+
+func NewContractEventHandler(node uint32, mgr substrate.Manager, engine provision.Engine, cl zbus.Client) ContractEventHandler {
+	return ContractEventHandler{node: node, pool: mgr, engine: engine, cl: cl}
+}
+
+func (r *ContractEventHandler) current() (map[uint64]uint32, error) {
+	// we need to build a list of all supposedly active contracts on this node
+	storage := r.engine.Storage()
+	twins, err := storage.Twins()
+	if err != nil {
+		return nil, err
+	}
+	active := make(map[uint64]uint32)
+	for _, twin := range twins {
+		deployments, err := storage.ByTwin(twin)
+		if err != nil {
+			return nil, err
+		}
+		for _, id := range deployments {
+			deployment, err := storage.Get(twin, id)
+			if err != nil {
+				return nil, err
+			}
+
+			if len(deployment.Workloads) > 0 {
+				active[deployment.ContractID] = twin
+			}
+		}
+	}
+
+	return active, nil
+}
+
+func (r *ContractEventHandler) sync(ctx context.Context) error {
+	log.Debug().Msg("synchronizing contracts with the chain")
+
+	active, err := r.current()
+	if err != nil {
+		return errors.Wrap(err, "failed to get current active contracts")
+	}
+	sub, err := r.pool.Substrate()
+	if err != nil {
+		return err
+	}
+
+	defer sub.Close()
+	onchain, err := sub.GetNodeContracts(r.node)
+	if err != nil {
+		return errors.Wrap(err, "failed to get active node contracts")
+	}
+
+	for _, contract := range onchain {
+		delete(active, uint64(contract))
+	}
+
+	for contract, twin := range active {
+		// those contracts exits on node but not on chain.
+		// need to be deleted
+		if err := r.engine.Deprovision(ctx, twin, contract, "contract not active on chain"); err != nil {
+			log.Error().Err(err).
+				Uint32("twin", twin).
+				Uint64("contract", contract).
+				Msg("failed to decomission contract")
+		}
+	}
+
+	log.Debug().Msg("synchronization complete")
+	return nil
+}
+
+// Run runs the reporter
+func (r *ContractEventHandler) Run(ctx context.Context) error {
+	// go over all user reservations
+	// take into account the following:
+	// every is in seconds.
+	events := stubs.NewEventsStub(r.cl)
+	stream, err := events.ContractCancelledEvent(ctx)
+	if err != nil {
+		return errors.Wrap(err, "failed to register to node events")
+	}
+
+	if err := r.sync(ctx); err != nil {
+		return errors.Wrap(err, "failed to synchronize active contracts")
+	}
+
+	ticker := time.NewTicker(1 * time.Hour)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+			go func() {
+				if err := r.sync(ctx); err != nil {
+					log.Error().Err(err).Msg("failed to synchronize contracts with the chain")
+				}
+			}()
+		case event := <-stream:
+			if event.Kind == pkg.EventSubscribed {
+				// we run this in a go routine because we don't
+				// want synchronization of contracts on the chain (that can take some time)
+				// to block
+				go func() {
+					if err := r.sync(ctx); err != nil {
+						log.Error().Err(err).Msg("failed to synchronize contracts with the chain")
+					}
+				}()
+				continue
+			}
+
+			log.Debug().Msgf("received a cancel contract event %+v", event)
+
+			// otherwise we know what contract to be deleted
+			if err := r.engine.Deprovision(ctx, event.TwinId, event.Contract, "contract canceled"); err != nil {
+				log.Error().Err(err).
+					Uint32("twin", event.TwinId).
+					Uint64("contract", event.Contract).
+					Msg("failed to decomission contract")
+			}
+		}
+	}
+}
diff --git a/cmds/modules/provisiond/main.go b/cmds/modules/provisiond/main.go
index 097e3a04..3d24f50f 100644
--- a/cmds/modules/provisiond/main.go
+++ b/cmds/modules/provisiond/main.go
@@ -42,7 +42,8 @@ const (
 	statisticsModule = "statistics"
 	gib              = 1024 * 1024 * 1024
 
-	boltStorageDB = "workloads.bolt"
+	boltStorageDB    = "workloads.bolt"
+	metricsStorageDB = "metrics.bolt"
 
 	// deprecated, kept for migration
 	fsStorageDB = "workloads"
@@ -78,8 +79,11 @@ func action(cli *cli.Context) error {
 		rootDir      string = cli.String("root")
 	)
 
-	ctx := context.Background()
-	ctx, _ = utils.WithSignal(ctx)
+	ctx, _ := utils.WithSignal(context.Background())
+
+	utils.OnDone(ctx, func(_ error) {
+		log.Info().Msg("shutting down")
+	})
 
 	// keep checking if limited-cache flag is set
 	if app.CheckFlag(app.LimitedCache) {
@@ -316,16 +320,31 @@ func action(cli *cli.Context) error {
 		log.Error().Err(err).Msg("failed to mark module as booted")
 	}
 
-	reporter, err := NewReporter(engine, node, cl, queues)
+	handler := NewContractEventHandler(node, mgr, engine, cl)
+
+	go func() {
+		if err := handler.Run(ctx); err != nil && err != context.Canceled {
+			log.Fatal().Err(err).Msg("handling contracts events failed")
+		}
+	}()
+
+	reporter, err := NewReporter(filepath.Join(rootDir, metricsStorageDB), 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")
+		for {
+			err := reporter.Run(ctx)
+			if err != context.Canceled {
+				return
+			} else if err != nil {
+				log.Error().Err(err).Msg("capacity reported stopped unexpectedely")
+			}
+
+			<-time.After(10 * time.Second)
 		}
-		log.Info().Msg("capacity reported stopped")
 	}()
 
 	// and start the zbus server in the background
diff --git a/cmds/modules/provisiond/reporter.go b/cmds/modules/provisiond/reporter.go
index bf8f2489..0cc52cb1 100644
--- a/cmds/modules/provisiond/reporter.go
+++ b/cmds/modules/provisiond/reporter.go
@@ -1,12 +1,10 @@
 package provisiond
 
 import (
-	"bytes"
 	"context"
 	"crypto/ed25519"
 	"os"
 	"path/filepath"
-	"strings"
 	"time"
 
 	"github.com/centrifuge/go-substrate-rpc-client/v4/types"
@@ -18,64 +16,25 @@ import (
 	"github.com/threefoldtech/zos/pkg"
 	"github.com/threefoldtech/zos/pkg/environment"
 	"github.com/threefoldtech/zos/pkg/gridtypes"
-	"github.com/threefoldtech/zos/pkg/provision"
+	"github.com/threefoldtech/zos/pkg/rrd"
 	"github.com/threefoldtech/zos/pkg/stubs"
 )
 
 const (
-	every = 60 * 60 // 1 hour
+	every           = 60 * 60 // 1 hour
+	lastReportedKey = ".last-reported-ts"
 )
 
-type many []error
-
-func (m many) Error() string {
-	return m.WithPrefix("")
-}
-
-func (m many) WithPrefix(p string) string {
-	var buf bytes.Buffer
-	for _, err := range m {
-		if buf.Len() > 0 {
-			buf.WriteRune('\n')
-		}
-		if err, ok := err.(many); ok {
-			buf.WriteString(err.WithPrefix(p + " "))
-			continue
-		}
-
-		buf.WriteString(err.Error())
-	}
-
-	return buf.String()
-}
-
-func (m many) append(err error) many {
-	return append(m, err)
-}
-
-func (m many) AsError() error {
-	if len(m) == 0 {
-		return nil
-	}
-
-	return m
-}
-
-type Consumption struct {
-	substrate.Consumption
-	TwinID uint32
-}
-
 type Report struct {
-	Consumption []Consumption
+	Consumption []substrate.Consumption
 }
 
 // Reporter structure
 type Reporter struct {
-	cl        zbus.Client
-	nodeID    uint32
+	cl  zbus.Client
+	rrd rrd.RRD
+
 	identity  substrate.Identity
-	engine    provision.Engine
 	queue     *dque.DQue
 	substrate substrate.Manager
 }
@@ -85,8 +44,10 @@ func reportBuilder() interface{} {
 }
 
 // NewReporter creates a new capacity reporter
-func NewReporter(engine provision.Engine, nodeID uint32, cl zbus.Client, root string) (*Reporter, error) {
-	sub, err := environment.GetSubstrate()
+func NewReporter(metricsPath string, cl zbus.Client, root string) (*Reporter, error) {
+	idMgr := stubs.NewIdentityManagerStub(cl)
+	sk := ed25519.PrivateKey(idMgr.PrivateKey(context.TODO()))
+	id, err := substrate.NewIdentityFromEd25519Key(sk)
 	if err != nil {
 		return nil, err
 	}
@@ -106,48 +67,45 @@ func NewReporter(engine provision.Engine, nodeID uint32, cl zbus.Client, root st
 		return nil, errors.Wrap(err, "failed to setup report persisted queue")
 	}
 
-	idMgr := stubs.NewIdentityManagerStub(cl)
-	sk := ed25519.PrivateKey(idMgr.PrivateKey(context.TODO()))
-	id, err := substrate.NewIdentityFromEd25519Key(sk)
+	sub, err := environment.GetSubstrate()
 	if err != nil {
 		return nil, err
 	}
 
+	rrd, err := rrd.NewRRDBolt(metricsPath, 5*time.Minute, 24*time.Hour)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to create metrics database")
+	}
+
 	return &Reporter{
 		cl:        cl,
-		engine:    engine,
-		nodeID:    nodeID,
+		rrd:       rrd,
 		identity:  id,
 		queue:     queue,
 		substrate: sub,
 	}, nil
 }
 
-func (r *Reporter) pushOne() ([]Consumption, error) {
+func (r *Reporter) pushOne() error {
 	item, err := r.queue.PeekBlock()
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to peek into capacity queue. #properlyfatal")
+		return errors.Wrap(err, "failed to peek into capacity queue. #properlyfatal")
 	}
 
 	report := item.(*Report)
 
 	log.Info().Int("len", len(report.Consumption)).Msgf("sending capacity report")
 
-	consumptions := make([]substrate.Consumption, 0, len(report.Consumption))
-	for _, cmp := range report.Consumption {
-		log.Debug().Uint64("contract", uint64(cmp.ContractID)).Msg("has consumption to report")
-		consumptions = append(consumptions, cmp.Consumption)
-	}
 	sub, err := r.substrate.Substrate()
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to connect to chain")
+		return errors.Wrap(err, "failed to connect to chain")
 	}
 
 	defer sub.Close()
 
-	hash, err := sub.Report(r.identity, consumptions)
+	hash, err := sub.Report(r.identity, report.Consumption)
 	if err != nil {
-		return nil, errors.Wrap(err, "failed to publish consumption report")
+		return errors.Wrap(err, "failed to publish consumption report")
 	}
 
 	log.Info().Str("hash", hash.Hex()).Msg("report block hash")
@@ -156,7 +114,7 @@ func (r *Reporter) pushOne() ([]Consumption, error) {
 	// remove item from queue
 	_, err = r.queue.Dequeue()
 
-	return report.Consumption, err
+	return err
 }
 
 func (r *Reporter) pusher(ctx context.Context) {
@@ -170,7 +128,7 @@ func (r *Reporter) pusher(ctx context.Context) {
 		// 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()
+		err := r.pushOne()
 		if err != nil {
 			log.Error().Err(err).Msg("error while processing capacity report")
 			select {
@@ -180,243 +138,179 @@ func (r *Reporter) pusher(ctx context.Context) {
 				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")
+// getMetrics will collect network consumption every 5 min and store
+// it in the rrd database.
+func (r *Reporter) getMetrics(ctx context.Context) error {
+	log.Debug().Msg("collecting networking metrics")
+	vmd := stubs.NewVMModuleStub(r.cl)
 
-	local := make(map[types.U64]Consumption)
-	for _, report := range reported {
-		local[report.ContractID] = report
+	ctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+	metrics, err := vmd.Metrics(ctx)
+	if err != nil {
+		return err
 	}
-
-	sub, err := r.substrate.Substrate()
+	slot, err := r.rrd.Slot()
 	if err != nil {
-		return errors.Wrap(err, "failed to connect to chain")
+		return errors.Wrap(err, "failed to create rrd slot")
 	}
 
-	defer sub.Close()
+	for vm, consumption := range metrics {
+		nu := r.computeNU(consumption)
+		log.Debug().Str("vm", vm).Uint64("computed", uint64(nu)).Msgf("consumption: %+v", consumption)
+		if err := slot.Counter(vm, float64(nu)); err != nil {
+			return errors.Wrapf(err, "failed to store metrics for '%s'", vm)
+		}
+	}
 
-	// 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 := sub.GetNodeContracts(r.nodeID)
-	if err != nil {
-		return err
+	return nil
+}
+
+func (r *Reporter) metrics(ctx context.Context) {
+	ticker := time.NewTicker(5 * time.Minute)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := r.getMetrics(ctx); err != nil {
+				log.Error().Err(err).Msg("failed to collect network metrics")
+			}
+		}
 	}
+}
 
-	for _, contract := range contracts {
-		// is there a consumption report for a contract
-		delete(local, contract)
+func (r *Reporter) setLastReportTime(ts int64) error {
+	slot, err := r.rrd.Slot()
+	if err != nil {
+		return errors.Wrap(err, "failed to create rrd slot")
 	}
-	// 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)
-		}
+	// we set it in storage anyway in case of reboot
+	if err := slot.Counter(lastReportedKey, float64(ts)); err != nil {
+		return errors.Wrap(err, "failed to set last report time")
 	}
 
 	return nil
 }
 
+func (r *Reporter) getLastReportTime() (int64, bool, error) {
+	stored, ok, err := r.rrd.Last(lastReportedKey)
+	if err != nil {
+		return 0, false, errors.Wrap(err, "failed to get timestamp of last report")
+	}
+	return int64(stored), ok, nil
+}
+
 // Run runs the reporter
 func (r *Reporter) Run(ctx context.Context) error {
 	// go over all user reservations
 	// take into account the following:
 	// every is in seconds.
-	events := stubs.NewEventsStub(r.cl)
-	stream, err := events.ContractCancelledEvent(ctx)
-	if err != nil {
-		return errors.Wrap(err, "failed to register to node events")
-	}
+	ctx, cancel := context.WithCancel(ctx)
+	defer cancel()
 
+	go r.metrics(ctx)
 	go r.pusher(ctx)
 
-	ticker := time.NewTicker(every * time.Second)
-	defer ticker.Stop()
-
 	// we always start by reporting capacity, and then once each
 	// `every` seconds
-report:
 	for {
-		log.Info().Msg("collecting consumption")
-		// align time.
+		log.Debug().Msg("collecting consumption since last report")
 		u := time.Now().Unix()
-		u = (u / every) * every
-		// so any reservation that is deleted but this
-		// happened 'after' this time stamp is still
-		// considered as consumption because it lies in the current
-		// 5 minute slot.
-		// but if it was stopped before that timestamp, then it will
-		// not be counted.
-		report, err := r.collect(ctx, time.Unix(u, 0))
-		if err != nil {
-			log.Error().Err(err).Msg("failed to collect users consumptions")
-			<-time.After(3 * time.Second)
-			continue
-		}
 
-		log.Info().Int("size", len(report.Consumption)).Msg("queue consumption report for reproting")
-		if err := r.push(report); err != nil {
-			log.Error().Err(err).Msg("failed to push capacity report")
+		lastReport, ok, err := r.getLastReportTime()
+		if err != nil {
+			return err
 		}
-
-		log.Info().Msg("consumption report queued")
-		for {
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			case event := <-stream:
-				if event.Kind == pkg.EventSubscribed {
-					// TODO:
-					// possible loss of events. either we synchronize
-					// all contracts now. Or we wait until next timer
-
-					// for now, wait until next report cycle
-					continue
-				}
-				log.Debug().Msgf("received a cancel contract event %+v", event)
-
-				// otherwise we know what contract to be deleted
-				if err := r.engine.Deprovision(ctx, event.TwinId, event.Contract, "contract canceled"); err != nil {
-					log.Error().Err(err).
-						Uint32("twin", event.TwinId).
-						Uint64("contract", event.Contract).
-						Msg("failed to decomission contract")
-				}
-			case <-ticker.C:
-				continue report
+		if !ok {
+			log.Debug().Msg("no previous reports found")
+			// no reports where made. we can't report consumption now
+			// because we have no window. It won't be fair for the user.
+			// (only during update of old nodes that has running workloads)
+			// hence we only set last report to now
+			lastReport = u
+			if err := r.setLastReportTime(u); err != nil {
+				return err
 			}
 		}
-	}
-}
-
-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
-	}
-
-	// to optimize we get ALL vms vmMetrics in one call.
-	vmMetrics, err := stubs.NewVMModuleStub(r.cl).Metrics(ctx)
-	if err != nil {
-		return Report{}, errors.Wrap(err, "failed to get VMs network metrics")
-	}
+		log.Debug().Time("last-report", time.Unix(lastReport, 0)).Msg("time of last report")
+		// compute when we should send next report.
+		delay := (lastReport + every) - u
 
-	gwMetrics, err := stubs.NewGatewayStub(r.cl).Metrics(ctx)
-	if err != nil && !strings.Contains(err.Error(), "metrics not available") {
-		return Report{}, errors.Wrap(err, "failed to get gateway metrics")
-	}
+		if delay < 0 {
+			delay = 0
+		}
+		// wait for delay
+		log.Debug().Int64("duration", delay).Msg("seconds to wait before collecting consumption")
 
-	qsfsMetrics, err := stubs.NewQSFSDStub(r.cl).Metrics(ctx)
-	if err != nil {
-		return Report{}, errors.Wrap(err, "failed to get qsfs metrics")
-	}
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(time.Duration(delay) * time.Second):
+		}
 
-	for _, user := range users {
-		cap, err := r.user(since, user, vmMetrics, gwMetrics, qsfsMetrics)
+		since := time.Unix(lastReport, 0)
+		log.Debug().Time("since", since).Msg("collecting consumption since")
+		ts, err := r.report(ctx, since)
 		if err != nil {
-			log.Error().Err(err).Msg("failed to collect all user capacity")
-			// NOTE: we intentionally not doing a 'continue' or 'return'
-			// here because even if an error is returned we can have partially
-			// collected some of the user consumption, we still can report that
+			return errors.Wrap(err, "failed to create consumption report")
 		}
 
-		rep.Consumption = append(rep.Consumption, cap...)
-	}
-
-	return rep, nil
-}
-
-func (r *Reporter) push(report Report) error {
-	if len(report.Consumption) == 0 {
-		return nil
+		if err := r.setLastReportTime(ts.Unix()); err != nil {
+			return err
+		}
 	}
-	return r.queue.Enqueue(&report)
 }
 
-func (r *Reporter) user(since time.Time, user uint32, vmMetrics pkg.MachineMetrics, gwMetrics pkg.GatewayMetrics, qsfsMetrics pkg.QSFSMetrics) ([]Consumption, error) {
-	var m many
-
-	var consumptions []Consumption
-	ids, err := r.engine.Storage().ByTwin(user)
+func (r *Reporter) report(ctx context.Context, since time.Time) (time.Time, error) {
+	now := time.Now()
+	window := now.Sub(since)
+	values, err := r.rrd.Counters(since)
 	if err != nil {
-		m = m.append(errors.Wrapf(err, "failed to get reservation for user '%s'", user))
+		return now, errors.Wrap(err, "failed to get stored metrics from rrd")
 	}
 
-	for _, id := range ids {
-		dl, err := r.engine.Storage().Get(user, id)
+	reports := make(map[uint64]substrate.Consumption)
+	for key, value := range values {
+		_, deploment, _, err := gridtypes.WorkloadID(key).Parts()
 		if err != nil {
-			m = m.append(errors.Wrapf(err, "failed to get reservation '%d'", id))
+			log.Error().Err(err).Msgf("failed to parse metric key '%s'", key)
 			continue
 		}
 
-		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
+		rep, ok := reports[deploment]
+		if !ok {
+			rep = substrate.Consumption{
+				ContractID: types.U64(deploment),
+				Timestamp:  types.U64(now.Unix()),
+				Window:     types.U64(window / time.Second),
 			}
+		}
 
-			wlID, err := gridtypes.NewWorkloadID(user, id, wl.Name)
-			if err != nil {
-				log.Error().Err(err).Msg("invalid workload id (shouldn't happen here)")
-				continue
-			}
+		rep.NRU += types.U64(value)
+		reports[deploment] = rep
+	}
 
-			if !r.shouldCount(since, &wl.Result) {
-				continue
-			}
+	report := Report{
+		Consumption: make([]substrate.Consumption, 0, len(reports)),
+	}
 
-			cap, err := wl.Capacity()
-			if err != nil {
-				m = m.append(errors.Wrapf(err, "failed to get reservation '%s' capacity", id))
-				continue
-			}
+	for _, v := range reports {
+		report.Consumption = append(report.Consumption, v)
+	}
 
-			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 := vmMetrics[wlID.String()]
-			if ok {
-				// add metric to consumption
-				consumption.NRU += types.U64(r.computeNU(metric))
-			}
-			consumption.NRU += types.U64(qsfsMetrics.Nu(wlID.String()))
-			// special handling for gw types.
-			consumption.NRU += types.U64(gwMetrics.Nu(wlID.String()))
-		}
+	return now, r.push(report)
+}
 
-		if !consumption.IsEmpty() {
-			consumptions = append(consumptions, consumption)
-		}
+func (r *Reporter) push(report Report) error {
+	if len(report.Consumption) == 0 {
+		return nil
 	}
-
-	return consumptions, m.AsError()
+	return r.queue.Enqueue(&report)
 }
 
 func (r *Reporter) computeNU(m pkg.MachineMetric) gridtypes.Unit {
@@ -430,16 +324,3 @@ func (r *Reporter) computeNU(m pkg.MachineMetric) gridtypes.Unit {
 
 	return gridtypes.Unit(nu)
 }
-
-func (r *Reporter) shouldCount(since time.Time, result *gridtypes.Result) bool {
-	if result.State == gridtypes.StateOk {
-		return true
-	}
-
-	if result.State == gridtypes.StateDeleted {
-		// if it was stopped before the 'since' .
-		return !result.Created.Time().Before(since)
-	}
-
-	return false
-}
diff --git a/go.mod b/go.mod
index ec7db38a..f4a6337c 100644
--- a/go.mod
+++ b/go.mod
@@ -62,7 +62,7 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/threefoldtech/0-fs v1.3.1-0.20201203163303-d963de9adea7
 	github.com/threefoldtech/go-rmb v0.1.11-0.20220224131627-825c23c921d3
-	github.com/threefoldtech/substrate-client v0.0.0-20220308140931-04dc2e0dad45
+	github.com/threefoldtech/substrate-client v0.0.0-20220311155549-b19ac3fcada8
 	github.com/threefoldtech/zbus v0.1.5
 	github.com/tinylib/msgp v1.1.5 // indirect
 	github.com/tyler-smith/go-bip39 v1.1.0
diff --git a/go.sum b/go.sum
index 020815ec..7cf2e6c2 100644
--- a/go.sum
+++ b/go.sum
@@ -1070,6 +1070,8 @@ github.com/threefoldtech/substrate-client v0.0.0-20220307082230-d63a6e0ece52 h1:
 github.com/threefoldtech/substrate-client v0.0.0-20220307082230-d63a6e0ece52/go.mod h1:Lpsz7gVNSdbWD8erfVh1AZv2wnProc7g/QHJiYp5lP8=
 github.com/threefoldtech/substrate-client v0.0.0-20220308140931-04dc2e0dad45 h1:2rIq8aNpQqIc8++41l4w0joOUKwI7beh9CY6sJzFBDI=
 github.com/threefoldtech/substrate-client v0.0.0-20220308140931-04dc2e0dad45/go.mod h1:Lpsz7gVNSdbWD8erfVh1AZv2wnProc7g/QHJiYp5lP8=
+github.com/threefoldtech/substrate-client v0.0.0-20220311155549-b19ac3fcada8 h1:uYfTsfBbk4DIsEmM6dKlqQC/a8HpN+aECK61uiddrFQ=
+github.com/threefoldtech/substrate-client v0.0.0-20220311155549-b19ac3fcada8/go.mod h1:Lpsz7gVNSdbWD8erfVh1AZv2wnProc7g/QHJiYp5lP8=
 github.com/threefoldtech/zbus v0.1.5 h1:S9kXbjejoRRnJw1yKHEXFGF2vqL+Drae2n4vpj0pGHo=
 github.com/threefoldtech/zbus v0.1.5/go.mod h1:ZtiRpcqzEBJetVQDsEbw0p48h/AF3O1kf0tvd30I0BU=
 github.com/tinylib/msgp v1.0.2/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE=
diff --git a/pkg/rrd/rrd.go b/pkg/rrd/rrd.go
new file mode 100644
index 00000000..89bd7771
--- /dev/null
+++ b/pkg/rrd/rrd.go
@@ -0,0 +1,224 @@
+package rrd
+
+import (
+	"encoding/binary"
+	"math"
+	"time"
+
+	"github.com/boltdb/bolt"
+	"github.com/pkg/errors"
+)
+
+// RRD is a round robin database of fixed size which is specified on creation
+// RRD stores `counter` values. Which means values that can only go UP.
+// then it's easy to compute the increase of this counter over a given window
+// The database only keep history based on retention.
+type RRD interface {
+	// Slot returns the current window (slot) to store values.
+	Slot() (Slot, error)
+	// Counters, return all stored counters since the given time (since) until now.
+	Counters(since time.Time) (map[string]float64, error)
+	// Last returns the last reported value for a metric given the metric
+	// name
+	Last(key string) (value float64, ok bool, err error)
+}
+
+type Slot interface {
+	// Counter sets (or overrides) the current stored value for this key,
+	// with value
+	Counter(key string, value float64) error
+	// Key return the key of the slot which is the window timestamp
+	Key() uint64
+}
+
+type rrdBolt struct {
+	db        *bolt.DB
+	window    uint64
+	retention uint64
+}
+
+type rrdSlot struct {
+	db  *bolt.DB
+	key uint64
+}
+
+// NewRRDBolt creates a new rrd database that uses bolt as a storage. if window or retention are 0
+// the function will panic. If retnetion is smaller then window the function will panic.
+// retention and window must be multiple of 1 minute.
+func NewRRDBolt(path string, window time.Duration, retention time.Duration) (RRD, error) {
+	return newRRDBolt(path, window, retention)
+}
+
+func newRRDBolt(path string, window time.Duration, retention time.Duration) (*rrdBolt, error) {
+	window = (window / time.Minute) * time.Minute
+	retention = (retention / time.Minute) * time.Minute
+
+	if window == 0 {
+		panic("invalid window, can't be zero")
+	}
+	if retention == 0 {
+		panic("invalid retention, can't be zero")
+	}
+	if retention < window {
+		panic("retention can't be smaller than window")
+	}
+
+	db, err := bolt.Open(path, 0644, bolt.DefaultOptions)
+	if err != nil {
+		return nil, errors.Wrap(err, "failed to open bolt db")
+	}
+
+	return &rrdBolt{
+		db:        db,
+		window:    uint64(window / time.Second),
+		retention: uint64(retention / time.Second),
+	}, nil
+}
+
+func (r *rrdBolt) retain(now uint64) error {
+	retain := now - r.retention
+	return r.db.Update(func(tx *bolt.Tx) error {
+		cur := tx.Cursor()
+
+		for k, _ := cur.First(); k != nil; k, _ = cur.Next() {
+			if len(k) != 8 {
+				continue // uknown key size
+			}
+
+			if lu64(k) <= retain {
+				// delete the bucket
+				// are we sure this is safe to do while iterating ?
+				_ = tx.DeleteBucket(k)
+			}
+		}
+
+		return nil
+	})
+}
+
+func (r *rrdBolt) Slots() ([]uint64, error) {
+	var slots []uint64
+	err := r.db.View(func(tx *bolt.Tx) error {
+		cur := tx.Cursor()
+		for k, _ := cur.First(); k != nil; k, _ = cur.Next() {
+			slots = append(slots, lu64(k))
+		}
+
+		return nil
+	})
+
+	return slots, err
+}
+
+func (r *rrdBolt) Last(key string) (value float64, ok bool, err error) {
+	err = r.db.View(func(tx *bolt.Tx) error {
+		cur := tx.Cursor()
+		for k, _ := cur.Last(); k != nil; k, _ = cur.Prev() {
+			bucket := tx.Bucket(k)
+			bytes := bucket.Get([]byte(key))
+			if bytes != nil {
+				value = lf64(bytes)
+				ok = true
+				break
+			}
+		}
+
+		return nil
+	})
+
+	return
+}
+
+// Counters return increase in counter value since the given
+// start time.
+func (r *rrdBolt) Counters(since time.Time) (map[string]float64, error) {
+	ts := uint64(since.Unix())
+	ts = (ts / r.window) * r.window
+
+	// we start from the previous slot so we check from the last value.
+	ts -= r.window
+	change := make(map[string]float64)
+
+	err := r.db.View(func(tx *bolt.Tx) error {
+		last := make(map[string]float64)
+		first := true
+		cur := tx.Cursor()
+		for k, _ := cur.Seek(u64(ts)); k != nil; k, _ = cur.Next() {
+			bucket := tx.Bucket(k)
+			values := bucket.Cursor()
+			for k, v := values.First(); k != nil; k, v = values.Next() {
+				vf := lf64(v)
+				key := string(k)
+
+				if !first {
+					prev := last[key]
+					if prev > vf {
+						change[key] += vf
+					} else {
+						change[key] += vf - prev
+					}
+				}
+
+				last[key] = vf
+			}
+
+			first = false
+		}
+
+		return nil
+	})
+
+	return change, err
+}
+
+func (r *rrdBolt) slotAt(ts uint64) (*rrdSlot, error) {
+	ts = (ts / r.window) * r.window
+
+	if err := r.retain(ts); err != nil {
+		return nil, errors.Wrap(err, "failed to apply retnetion policy")
+	}
+
+	if err := r.db.Update(func(tx *bolt.Tx) error {
+		_, err := tx.CreateBucketIfNotExists(u64(ts))
+		return errors.Wrap(err, "failed to create slot bucket")
+
+	}); err != nil {
+		return nil, err
+	}
+
+	return &rrdSlot{db: r.db, key: ts}, nil
+}
+
+func (r *rrdBolt) Slot() (Slot, error) {
+	ts := uint64(time.Now().Unix())
+	return r.slotAt(ts)
+}
+
+func lu64(v []byte) uint64 {
+	return binary.BigEndian.Uint64(v)
+}
+
+func u64(u uint64) []byte {
+	var v [8]byte
+	binary.BigEndian.PutUint64(v[:], u)
+	return v[:]
+}
+
+func f64(f float64) []byte {
+	return u64(math.Float64bits(f))
+}
+
+func lf64(v []byte) float64 {
+	return math.Float64frombits(lu64(v))
+}
+
+func (r *rrdSlot) Counter(key string, value float64) error {
+	return r.db.Update(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket(u64(r.key))
+		return bucket.Put([]byte(key), f64(value))
+	})
+}
+
+func (r *rrdSlot) Key() uint64 {
+	return r.key
+}
diff --git a/pkg/vm.go b/pkg/vm.go
index 7e6f7139..a0d8f743 100644
--- a/pkg/vm.go
+++ b/pkg/vm.go
@@ -209,7 +209,7 @@ func (n *NetMetric) Nu() float64 {
 	const (
 		// weights knobs for nu calculations
 		bytes   float64 = 1.0
-		packets float64 = 0.2
+		packets float64 = 0.0
 		rx      float64 = 1.0
 		tx      float64 = 1.0
 	)


Test patch:
diff --git a/pkg/rrd/rrd_test.go b/pkg/rrd/rrd_test.go
new file mode 100644
index 00000000..8bd67e8f
--- /dev/null
+++ b/pkg/rrd/rrd_test.go
@@ -0,0 +1,220 @@
+package rrd
+
+import (
+	"fmt"
+	"math/rand"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestAddSlot(t *testing.T) {
+	require := require.New(t)
+	path := filepath.Join(os.TempDir(), fmt.Sprint(rand.Int63()))
+	defer os.RemoveAll(path)
+
+	window := 1 * time.Minute
+	db, err := NewRRDBolt(path, window, 10*time.Minute)
+	require.NoError(err)
+
+	now := time.Now()
+	slot, err := db.Slot()
+	require.NoError(err)
+
+	w := uint64(window / time.Second) // window in seconds
+	require.Equal((uint64(now.Unix())/w)*w, slot.Key())
+
+	err = slot.Counter("test-1", 120)
+	require.NoError(err)
+}
+
+func TestCountersTwoValues(t *testing.T) {
+	require := require.New(t)
+	path := filepath.Join(os.TempDir(), fmt.Sprint(rand.Int63()))
+	defer os.RemoveAll(path)
+
+	window := 1 * time.Minute
+	db, err := newRRDBolt(path, window, 10*time.Minute)
+	require.NoError(err)
+
+	now := time.Now()
+	slot1, err := db.slotAt(uint64(now.Add(-time.Minute).Unix()))
+	require.NoError(err)
+
+	slotNow, err := db.slotAt(uint64(now.Unix()))
+	require.NoError(err)
+
+	err = slot1.Counter("test-1", 100)
+	require.NoError(err)
+
+	err = slotNow.Counter("test-1", 120)
+	require.NoError(err)
+
+	counters, err := db.Counters(now.Add(-5 * time.Minute))
+	require.NoError(err)
+	require.Len(counters, 1)
+
+	require.EqualValues(20, counters["test-1"])
+}
+
+func TestCountersSeries(t *testing.T) {
+	require := require.New(t)
+	path := filepath.Join(os.TempDir(), fmt.Sprint(rand.Int63()))
+	defer os.RemoveAll(path)
+
+	window := 1 * time.Minute
+	// note retention is only 10 minutes
+	db, err := newRRDBolt(path, window, 10*time.Minute)
+	require.NoError(err)
+
+	over := 20
+	now := time.Now()
+	first := now.Add(-time.Duration(over) * time.Minute)
+
+	// we insert values over the last 20 minutes
+	for i := 0; i <= over; i++ {
+		slot, err := db.slotAt(uint64(first.Add(time.Duration(i) * time.Minute).Unix()))
+		require.NoError(err)
+
+		require.NoError(slot.Counter("test-1", float64(i)))
+	}
+
+	// i get all the values over the last 10 minutes
+	counters, err := db.Counters(now.Add(-10 * time.Minute))
+	require.NoError(err)
+	require.Len(counters, 1)
+
+	// we go up by one for each slot. so query last 10 blocks (including now) should return 9
+	require.EqualValues(9, counters["test-1"])
+}
+
+func TestCountersRandomIncrese(t *testing.T) {
+	require := require.New(t)
+	path := filepath.Join(os.TempDir(), fmt.Sprint(rand.Int63()))
+	defer os.RemoveAll(path)
+
+	window := 1 * time.Minute
+	// note retention is only 10 minutes
+	db, err := newRRDBolt(path, window, 10*time.Minute)
+	require.NoError(err)
+
+	over := 5
+	now := time.Now()
+	//slot, _ := db.slotAt(uint64(now.Add(-5 * time.Microsecond).Unix()))
+
+	first := now.Add(-time.Duration(over) * time.Minute)
+
+	// we insert values over the last 20 minutes
+	var expected int64
+	for i := 0; i <= over; i++ {
+		slot, err := db.slotAt(uint64(first.Add(time.Duration(i) * time.Minute).Unix()))
+		require.NoError(err)
+		v := rand.Int63n(10)
+		if i != 0 {
+			expected += v
+		}
+		require.NoError(slot.Counter("test-1", float64(expected)))
+	}
+
+	// i get all the values over the last 10 minutes
+	counters, err := db.Counters(now.Add(-10 * time.Minute))
+	require.NoError(err)
+	require.Len(counters, 1)
+
+	// we go up by one for each slot. so query last 10 blocks should return 10
+	require.EqualValues(expected, counters["test-1"])
+}
+
+func TestCountersGap(t *testing.T) {
+	require := require.New(t)
+	path := filepath.Join(os.TempDir(), fmt.Sprint(rand.Int63()))
+	defer os.RemoveAll(path)
+
+	window := 1 * time.Minute
+	db, err := newRRDBolt(path, window, 10*time.Minute)
+	require.NoError(err)
+
+	now := time.Now()
+	slot1, err := db.slotAt(uint64(now.Add(3 - time.Minute).Unix()))
+	require.NoError(err)
+
+	slotNow, err := db.slotAt(uint64(now.Unix()))
+	require.NoError(err)
+
+	err = slot1.Counter("test-1", 100)
+	require.NoError(err)
+
+	err = slotNow.Counter("test-1", 120)
+	require.NoError(err)
+
+	counters, err := db.Counters(now.Add(-5 * time.Minute))
+	require.NoError(err)
+	require.Len(counters, 1)
+
+	require.EqualValues(20, counters["test-1"])
+}
+
+func TestCountersRetention(t *testing.T) {
+	require := require.New(t)
+	path := filepath.Join(os.TempDir(), fmt.Sprint(rand.Int63()))
+	defer os.RemoveAll(path)
+
+	window := 1 * time.Minute
+	// note retention is only 10 minutes
+	db, err := newRRDBolt(path, window, 10*time.Minute)
+	require.NoError(err)
+
+	over := 20
+	now := time.Now()
+	first := now.Add(-time.Duration(over) * time.Minute)
+
+	// we insert values over the last 20 minutes
+	for i := 0; i <= over; i++ {
+		slot, err := db.slotAt(uint64(first.Add(time.Duration(i) * time.Minute).Unix()))
+		require.NoError(err)
+
+		require.NoError(slot.Counter("test-1", float64(i)))
+	}
+
+	slots, err := db.Slots()
+	require.NoError(err)
+
+	require.Len(slots, 10)
+	require.EqualValues((now.Add(-9*time.Minute).Unix()/60)*60, slots[0])
+}
+
+func TestCountersLast(t *testing.T) {
+	require := require.New(t)
+	path := filepath.Join(os.TempDir(), fmt.Sprint(rand.Int63()))
+	defer os.RemoveAll(path)
+
+	window := 1 * time.Minute
+	db, err := newRRDBolt(path, window, 10*time.Minute)
+	require.NoError(err)
+
+	_, ok, err := db.Last("test-1")
+	require.NoError(err)
+	require.False(ok)
+
+	now := time.Now()
+	slot1, err := db.slotAt(uint64(now.Add(-5 * time.Minute).Unix()))
+	require.NoError(err)
+
+	slotNow, err := db.slotAt(uint64(now.Add(-2 * time.Minute).Unix()))
+	require.NoError(err)
+
+	err = slot1.Counter("test-1", 100)
+	require.NoError(err)
+
+	err = slotNow.Counter("test-1", 120)
+	require.NoError(err)
+
+	v, ok, err := db.Last("test-1")
+	require.NoError(err)
+	require.True(ok)
+	require.EqualValues(120, v)
+
+}