Add support for data_usage event

data/migrations/1_create_msmt_results.sql

@@ -66,8 +66,8 @@ CREATE TABLE `results` (
 
     -- This is a flag used to indicate if the result is done or is currently running.
     `is_done` TINYINT(1) NOT NULL,
-    `data_usage_up` INTEGER NOT NULL,
-    `data_usage_down` INTEGER NOT NULL,
+    `data_usage_up` REAL NOT NULL,
+    `data_usage_down` REAL NOT NULL,
     -- It's probably reasonable to set the maximum length to 260 as this is the
     -- maximum length of file paths on windows.
     `measurement_dir` VARCHAR(260) NOT NULL,

internal/cli/list/list.go

@@ -29,8 +29,8 @@ func init() {
 			msmtSummary := output.MeasurementSummaryData{
 				TotalCount:    0,
 				AnomalyCount:  0,
-				DataUsageUp:   0,
-				DataUsageDown: 0,
+				DataUsageUp:   0.0,
+				DataUsageDown: 0.0,
 				TotalRuntime:  0,
 			}
 			for _, msmt := range measurements {

internal/database/models.go

@@ -91,8 +91,8 @@ type Result struct {
 	Runtime        float64   `db:"runtime"`    // Runtime is expressed in fractional seconds
 	IsViewed       bool      `db:"is_viewed"`
 	IsDone         bool      `db:"is_done"`
-	DataUsageUp    int64     `db:"data_usage_up"`
-	DataUsageDown  int64     `db:"data_usage_down"`
+	DataUsageUp    float64   `db:"data_usage_up"`
+	DataUsageDown  float64   `db:"data_usage_down"`
 	MeasurementDir string    `db:"measurement_dir"`
 }
 

internal/log/handlers/cli/results.go

@@ -113,8 +113,8 @@ func logResultSummary(w io.Writer, f log.Fields) error {
 
 	networks := f.Get("total_networks").(int64)
 	tests := f.Get("total_tests").(int64)
-	dataUp := f.Get("total_data_usage_up").(int64)
-	dataDown := f.Get("total_data_usage_down").(int64)
+	dataUp := f.Get("total_data_usage_up").(float64)
+	dataDown := f.Get("total_data_usage_down").(float64)
 	if tests == 0 {
 		fmt.Fprintf(w, "No results\n")
 		fmt.Fprintf(w, "Try running:\n")
@@ -125,7 +125,7 @@ func logResultSummary(w io.Writer, f log.Fields) error {
 	fmt.Fprintf(w, " │ %s │ %s │ %s │\n",
 		util.RightPad(fmt.Sprintf("%d tests", tests), 12),
 		util.RightPad(fmt.Sprintf("%d nets", networks), 12),
-		util.RightPad(fmt.Sprintf("%d ⬆ %d ⬇", dataUp, dataDown), 12))
+		util.RightPad(fmt.Sprintf("%.0f ⬆ %.0f ⬇", dataUp, dataDown), 12))
 	fmt.Fprintf(w, " └──────────────┴──────────────┴──────────────┘\n")
 
 	return nil

internal/output/output.go

@@ -24,8 +24,8 @@ type MeasurementSummaryData struct {
 	TotalRuntime  float64
 	TotalCount    int64
 	AnomalyCount  int64
-	DataUsageUp   int64
-	DataUsageDown int64
+	DataUsageUp   float64
+	DataUsageDown float64
 }
 
 func MeasurementSummary(msmt MeasurementSummaryData) {
@@ -79,8 +79,8 @@ type ResultItemData struct {
 	NetworkName             string
 	ASN                     uint
 	Done                    bool
-	DataUsageDown           int64
-	DataUsageUp             int64
+	DataUsageDown           float64
+	DataUsageUp             float64
 	Index                   int
 	TotalCount              int
 }
@@ -109,8 +109,8 @@ func ResultItem(result ResultItemData) {
 
 type ResultSummaryData struct {
 	TotalTests         int64
-	TotalDataUsageUp   int64
-	TotalDataUsageDown int64
+	TotalDataUsageUp   float64
+	TotalDataUsageDown float64
 	TotalNetworks      int64
 }
 

nettests/nettests.go

@@ -63,6 +63,12 @@ func (c *Controller) SetInputIdxMap(inputIdxMap map[int64]int64) error {
 	return nil
 }
 
+type StatusEnd struct {
+	DownloadedKB float64 `json:"download_kb"`
+	UploadedKB   float64 `json:"uploaded_kb"`
+	Failure      string  `json:"failure"`
+}
+
 // Init should be called once to initialise the nettest
 func (c *Controller) Init(nt *mk.Nettest) error {
 	log.Debugf("Init: %v", nt)
@@ -273,12 +279,27 @@ func (c *Controller) Init(nt *mk.Nettest) error {
 
 	nt.On("status.end", func(e mk.Event) {
 		log.Debugf("status.end")
+
 		for idx, msmt := range c.msmts {
 			log.Debugf("adding msmt#%d to result", idx)
 			if err := msmt.AddToResult(c.Ctx.DB, c.res); err != nil {
 				log.WithError(err).Error("failed to add to result")
 			}
 		}
+
+		var endMsg StatusEnd
+		err := json.Unmarshal([]byte(e.Value.JSONStr), &endMsg)
+		if err != nil {
+			log.WithError(err).Errorf("failed to extract status.end message %s", e.Value.JSONStr)
+			return
+		}
+
+		if endMsg.Failure != "" {
+			log.Errorf("Failure in status.end: %s", endMsg.Failure)
+		}
+
+		c.res.DataUsageDown += endMsg.DownloadedKB
+		c.res.DataUsageDown += endMsg.UploadedKB
 	})
 
 	log.Debugf("Registered all the handlers")
@@ -303,7 +324,10 @@ func (c *Controller) OnEntry(idx int64, jsonStr string) {
 	log.Debugf("OnEntry")
 
 	var entry Entry
-	json.Unmarshal([]byte(jsonStr), &entry)
+	if err := json.Unmarshal([]byte(jsonStr), &entry); err != nil {
+		log.WithError(err).Error("failed to parse onEntry")
+		return
+	}
 	tk := c.nt.GetTestKeys(entry.TestKeys)
 
 	log.Debugf("Fetching: %s %v", idx, c.msmts[idx])