Repository: threefoldtech/zos
Pull request #1678 (2022-05-05 11:57:21)

Rrd Calculations
This PR does the following:
- It tries to be safe and avoid over charging the user by
  - Always track the last reported value. 
  - Increase in consumption (increments) are calculated on reporting a reading not when calculating the consumption over a period.
  - Those 2 steps allows us to safely compute the consumption over a period even with gaps.
- Add GW metrics to reporting


Patch:
diff --git a/.github/workflows/grid-deploy.yaml b/.github/workflows/grid-deploy.yaml
new file mode 100644
index 00000000..285a4a7f
--- /dev/null
+++ b/.github/workflows/grid-deploy.yaml
@@ -0,0 +1,30 @@
+name: Deploy
+
+on:
+  workflow_dispatch:
+    inputs:
+      grid:
+        description: "grid to deploy"
+        required: true
+        default: "testing"
+        type: choice
+        options:
+          - testing
+          - production
+      version:
+        description: "version to release"
+        required: true
+        type: string
+jobs:
+  link-release:
+    name: linking
+    runs-on: ubuntu-latest
+    steps:
+      - name: Symlink flist (development)
+        uses: threefoldtech/publish-flist@master
+        with:
+          token: ${{ secrets.HUB_JWT }}
+          action: crosslink
+          user: tf-zos
+          name: zos:${{ github.event.inputs.grid }}-3:latest.flist
+          target: tf-autobuilder/zos:${{ github.event.inputs.version }}.flist
diff --git a/.github/workflows/publish-pre-release.yaml b/.github/workflows/publish-pre-release.yaml
index 09ca1205..b3d35aad 100644
--- a/.github/workflows/publish-pre-release.yaml
+++ b/.github/workflows/publish-pre-release.yaml
@@ -58,23 +58,3 @@ jobs:
           user: tf-autobuilder
           name: zos-${{ github.ref }}.flist
           target: zos:${{ github.ref }}.flist
-
-      - name: Symlink flist (development)
-        if: success()
-        uses: threefoldtech/publish-flist@master
-        with:
-          token: ${{ secrets.HUB_JWT }}
-          action: crosslink
-          user: tf-zos
-          name: zos:development-3:latest.flist
-          target: tf-autobuilder/zos:${{ github.ref }}.flist
-
-      - name: Symlink flist (testing)
-        if: success()
-        uses: threefoldtech/publish-flist@master
-        with:
-          token: ${{ secrets.HUB_JWT }}
-          action: crosslink
-          user: tf-zos
-          name: zos:testing-3:latest.flist
-          target: tf-autobuilder/zos:${{ github.ref }}.flist
diff --git a/.github/workflows/publish-release.yaml b/.github/workflows/publish-release.yaml
index e00f6c37..c7ef7903 100644
--- a/.github/workflows/publish-release.yaml
+++ b/.github/workflows/publish-release.yaml
@@ -59,33 +59,3 @@ jobs:
           user: tf-autobuilder
           name: zos-${{ github.ref }}.flist
           target: zos:${{ github.ref }}.flist
-
-      - name: Symlink flist (development)
-        if: success()
-        uses: threefoldtech/publish-flist@master
-        with:
-          token: ${{ secrets.HUB_JWT }}
-          action: crosslink
-          user: tf-zos
-          name: zos:development-3:latest.flist
-          target: tf-autobuilder/zos:${{ github.ref }}.flist
-
-      - name: Symlink flist (testing)
-        if: success()
-        uses: threefoldtech/publish-flist@master
-        with:
-          token: ${{ secrets.HUB_JWT }}
-          action: crosslink
-          user: tf-zos
-          name: zos:testing-3:latest.flist
-          target: tf-autobuilder/zos:${{ github.ref }}.flist
-
-      - name: Symlink flist (production)
-        if: success()
-        uses: threefoldtech/publish-flist@master
-        with:
-          token: ${{ secrets.HUB_JWT }}
-          action: crosslink
-          user: tf-zos
-          name: zos:production-3:latest.flist
-          target: tf-autobuilder/zos:${{ github.ref }}.flist
diff --git a/cmds/modules/provisiond/main.go b/cmds/modules/provisiond/main.go
index c9e43955..6a73419d 100644
--- a/cmds/modules/provisiond/main.go
+++ b/cmds/modules/provisiond/main.go
@@ -42,8 +42,11 @@ const (
 	statisticsModule = "statistics"
 	gib              = 1024 * 1024 * 1024
 
-	boltStorageDB    = "workloads.bolt"
-	metricsStorageDB = "metrics.bolt"
+	boltStorageDB = "workloads.bolt"
+	// old style rrd, make sure we clean it up
+	metricsStorageDBOld = "metrics.bolt"
+	// new style db after rrd implementation change
+	metricsStorageDB = "metrics-diff.bolt"
 
 	// deprecated, kept for migration
 	fsStorageDB = "workloads"
@@ -350,6 +353,9 @@ func action(cli *cli.Context) error {
 		}
 	}()
 
+	// clean up old rrd db that uses previous style reporting
+	_ = os.Remove(filepath.Join(rootDir, metricsStorageDBOld))
+
 	reporter, err := NewReporter(filepath.Join(rootDir, metricsStorageDB), cl, queues)
 	if err != nil {
 		return errors.Wrap(err, "failed to setup capacity reporter")
diff --git a/cmds/modules/provisiond/reporter.go b/cmds/modules/provisiond/reporter.go
index a4bc62db..a88aac53 100644
--- a/cmds/modules/provisiond/reporter.go
+++ b/cmds/modules/provisiond/reporter.go
@@ -141,9 +141,8 @@ func (r *Reporter) pusher(ctx context.Context) {
 	}
 }
 
-// getMetrics will collect network consumption every 5 min and store
-// it in the rrd database.
-func (r *Reporter) getMetrics(ctx context.Context) error {
+// getVmMetrics will collect network consumption for vms and store it in the given slot
+func (r *Reporter) getVmMetrics(ctx context.Context, slot rrd.Slot) error {
 	log.Debug().Msg("collecting networking metrics")
 	vmd := stubs.NewVMModuleStub(r.cl)
 
@@ -153,10 +152,6 @@ func (r *Reporter) getMetrics(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	slot, err := r.rrd.Slot()
-	if err != nil {
-		return errors.Wrap(err, "failed to create rrd slot")
-	}
 
 	for vm, consumption := range metrics {
 		nu := r.computeNU(consumption)
@@ -169,6 +164,58 @@ func (r *Reporter) getMetrics(ctx context.Context) error {
 	return nil
 }
 
+// getVmMetrics will collect network consumption every 5 min and store
+// it in the rrd database.
+func (r *Reporter) getGwMetrics(ctx context.Context, slot rrd.Slot) error {
+	log.Debug().Msg("collecting networking metrics")
+	gw := stubs.NewGatewayStub(r.cl)
+
+	ctx, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	defer cancel()
+	metrics, err := gw.Metrics(ctx)
+	if err != nil {
+		return err
+	}
+	sent := metrics.Sent
+	recv := metrics.Received
+
+	sums := make(map[string]float64)
+	for wl, v := range sent {
+		sums[wl] = v + recv[wl]
+		delete(recv, wl)
+	}
+
+	for wl, v := range recv {
+		sums[wl] = v
+	}
+
+	for wl, nu := range sums {
+		log.Debug().Str("gw", wl).Uint64("computed", uint64(nu)).Msg("consumption")
+		if err := slot.Counter(wl, float64(nu)); err != nil {
+			return errors.Wrapf(err, "failed to store metrics for '%s'", wl)
+		}
+	}
+
+	return nil
+}
+
+func (r *Reporter) getMetrics(ctx context.Context) error {
+	slot, err := r.rrd.Slot()
+	if err != nil {
+		return err
+	}
+
+	if err := r.getVmMetrics(ctx, slot); err != nil {
+		log.Error().Err(err).Msg("failed to get vm network consumption")
+	}
+
+	if err := r.getGwMetrics(ctx, slot); err != nil {
+		log.Error().Err(err).Msg("failed to get gateway network consumption")
+	}
+
+	return nil
+}
+
 func (r *Reporter) metrics(ctx context.Context) {
 	ticker := time.NewTicker(5 * time.Minute)
 	defer ticker.Stop()
@@ -295,9 +342,7 @@ func (r *Reporter) report(ctx context.Context, since time.Time) (time.Time, erro
 		reports[deploment] = rep
 	}
 
-	report := Report{
-		Consumption: make([]substrate.NruConsumption, 0, len(reports)),
-	}
+	var report Report
 
 	for _, v := range reports {
 		if v.NRU == 0 {
diff --git a/pkg/rrd/rrd.go b/pkg/rrd/rrd.go
index f9d3f4f7..7a9f3b97 100644
--- a/pkg/rrd/rrd.go
+++ b/pkg/rrd/rrd.go
@@ -2,6 +2,8 @@ package rrd
 
 import (
 	"encoding/binary"
+	"fmt"
+	"io"
 	"math"
 	"time"
 
@@ -23,6 +25,10 @@ type RRD interface {
 	Last(key string) (value float64, ok bool, err error)
 }
 
+type Printer interface {
+	Print(w io.Writer) error
+}
+
 type Slot interface {
 	// Counter sets (or overrides) the current stored value for this key,
 	// with value
@@ -31,6 +37,11 @@ type Slot interface {
 	Key() uint64
 }
 
+const (
+	lastBucket = ".last"
+	keyLen     = 8 // float64 size
+)
+
 type rrdBolt struct {
 	db        *bolt.DB
 	window    uint64
@@ -75,14 +86,52 @@ func newRRDBolt(path string, window time.Duration, retention time.Duration) (*rr
 	}, nil
 }
 
+func (r *rrdBolt) printBucket(bucket *bolt.Bucket, out io.Writer) error {
+	cur := bucket.Cursor()
+	for k, v := cur.First(); k != nil; k, v = cur.Next() {
+		if _, err := fmt.Fprintf(out, "\t%s: %f\n", k, lf64(v)); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (r *rrdBolt) Print(out io.Writer) error {
+	return r.db.View(func(tx *bolt.Tx) error {
+		last := tx.Bucket([]byte(lastBucket))
+		if _, err := fmt.Fprintln(out, lastBucket); err != nil {
+			return err
+		}
+		if err := r.printBucket(last, out); err != nil {
+			return err
+		}
+
+		cur := tx.Cursor()
+		for k, _ := cur.First(); k != nil; k, _ = cur.Next() {
+			if len(k) != keyLen {
+				continue
+			}
+			n := lu64(k)
+			fmt.Fprintf(out, "%s (%d)\n", time.Unix(int64(n), 0).String(), n)
+			bucket := tx.Bucket(k)
+			if err := r.printBucket(bucket, out); err != nil {
+				return err
+			}
+		}
+
+		return 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 len(k) != keyLen {
+				continue // unknown key size
 			}
 
 			if lu64(k) <= retain {
@@ -101,6 +150,9 @@ func (r *rrdBolt) Slots() ([]uint64, error) {
 	err := r.db.View(func(tx *bolt.Tx) error {
 		cur := tx.Cursor()
 		for k, _ := cur.First(); k != nil; k, _ = cur.Next() {
+			if len(k) != keyLen {
+				continue
+			}
 			slots = append(slots, lu64(k))
 		}
 
@@ -112,17 +164,7 @@ func (r *rrdBolt) Slots() ([]uint64, error) {
 
 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
-			}
-		}
-
+		value, ok = getLast(tx, key)
 		return nil
 	})
 
@@ -140,29 +182,24 @@ func (r *rrdBolt) Counters(since time.Time) (map[string]float64, error) {
 	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() {
+			if len(k) != keyLen {
+				// this check can be replaced by
+				// bytes.Equal(k, []byte(lastBucket))
+				// to make sure this is a slot bucket
+				// but it's faster to check the length
+				continue
+			}
+
 			bucket := tx.Bucket(k)
 			values := bucket.Cursor()
 			for k, v := values.First(); k != nil; k, v = values.Next() {
-				vf := lf64(v)
+				diff := lf64(v)
 				key := string(k)
 
-				if !first {
-					prev := last[key]
-					if prev > vf {
-						change[key] += vf
-					} else {
-						change[key] += vf - prev
-					}
-				}
-
-				last[key] = vf
+				change[key] += diff
 			}
-
-			first = false
 		}
 
 		return nil
@@ -194,6 +231,38 @@ func (r *rrdBolt) Slot() (Slot, error) {
 	return r.slotAt(ts)
 }
 
+func (r *rrdSlot) Counter(key string, value float64) error {
+	return r.db.Update(func(tx *bolt.Tx) error {
+		last, ok := getLast(tx, key)
+		if err := setLast(tx, key, value); err != nil {
+			return err
+		}
+		// there was no last value, we just
+		// return
+		if !ok {
+			return nil
+		}
+		diff := 0.0
+		if value > last {
+			diff = value - last
+		} else {
+			// this is either an overflow
+			// or counter has been reset (node was restarted hence)
+			// metrics are counting from 0 again.
+			// hence it's safer to assume diff is just the value
+			// reported
+			diff = value
+		}
+
+		bucket := tx.Bucket(u64(r.key))
+		return bucket.Put([]byte(key), f64(diff))
+	})
+}
+
+func (r *rrdSlot) Key() uint64 {
+	return r.key
+}
+
 func lu64(v []byte) uint64 {
 	return binary.BigEndian.Uint64(v)
 }
@@ -212,13 +281,26 @@ 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 getLast(tx *bolt.Tx, key string) (value float64, ok bool) {
+	bucket := tx.Bucket([]byte(lastBucket))
+	if bucket == nil {
+		return
+	}
+
+	bytes := bucket.Get([]byte(key))
+	if bytes != nil {
+		value = lf64(bytes)
+		ok = true
+	}
+
+	return
 }
 
-func (r *rrdSlot) Key() uint64 {
-	return r.key
+func setLast(tx *bolt.Tx, key string, value float64) error {
+	bucket, err := tx.CreateBucketIfNotExists([]byte(lastBucket))
+	if err != nil {
+		return err
+	}
+
+	return bucket.Put([]byte(key), f64(value))
 }


Test patch:
diff --git a/pkg/rrd/rrd_test.go b/pkg/rrd/rrd_test.go
index 31175629..3094ce91 100644
--- a/pkg/rrd/rrd_test.go
+++ b/pkg/rrd/rrd_test.go
@@ -86,9 +86,7 @@ func TestCountersSeries(t *testing.T) {
 	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"])
+	require.EqualValues(10, counters["test-1"])
 }
 
 func TestCountersRandomIncrese(t *testing.T) {
@@ -230,20 +228,25 @@ func TestCountersMultipleReports(t *testing.T) {
 	now := time.Now()
 
 	lastReportTime := now.Unix()
+	slot, err := db.slotAt(uint64(now.Add(-5 * time.Minute).Unix()))
+	require.NoError(slot.Counter("test-0", 0))
+	require.NoError(err)
+
 	total := 0.0
 	for i := 0; i <= 24; i++ {
 		slot, err := db.slotAt(uint64(now.Add(time.Duration(i) * 5 * time.Minute).Unix()))
 		require.NoError(err)
-		err = slot.Counter("test-0", float64(i))
-		require.NoError(err)
 		if i%6 == 0 && i != 0 {
 			counters, err := db.Counters(time.Unix(lastReportTime, 0))
 			require.NoError(err)
 			require.Len(counters, 1)
-			lastReportTime = now.Add(time.Duration(i) * 5 * time.Minute).Unix()
+			lastReportTime = int64(slot.Key()) // now.Add(time.Duration(i) * 5 * time.Minute).Unix()
 			require.EqualValues(6, counters["test-0"])
 			total += counters["test-0"]
 		}
+		err = slot.Counter("test-0", float64(i)+1)
+		require.NoError(err)
+
 	}
 
 	require.EqualValues(24, total)