Merge pull request #302 from m-lab/no-delta

Suppress delta generation

Author: gfr10598

cmd/etl_worker/app-ndt.yaml

@@ -41,4 +41,5 @@ env_variables:
   BIGQUERY_PROJECT: 'mlab-sandbox'
   BIGQUERY_DATASET: 'mlab_sandbox'
   ANNOTATE_IP: 'true'
+  NDT_OMIT_DELTAS: 'true'
   # TODO add custom service-account, instead of using default credentials.

parser/ndt.go

@@ -5,7 +5,9 @@ import (
 	"errors"
 	"fmt"
 	"log"
+	"os"
 	"regexp"
+	"strconv"
 	"strings"
 	"time"
 
@@ -18,6 +20,11 @@ import (
 	"github.com/m-lab/etl/web100"
 )
 
+var (
+	// NDTOmitDeltas flag indicates if deltas should be suppressed.
+	NDTOmitDeltas, _ = strconv.ParseBool(os.Getenv("NDT_OMIT_DELTAS"))
+)
+
 const (
 	// Some snaplogs are very large, and we don't want to parse the entire
 	// snaplog, when there is no value.  However, although the nominal test
@@ -29,7 +36,6 @@ const (
 	// TODO - in future, we should probably detect when the connection state changes
 	// from established, as there is little reason to parse snapshots beyond that
 	// point.
-
 	minNumSnapshots = 1600 // If fewer than this, then set anomalies.num_snaps
 	maxNumSnapshots = 2800 // If more than this, truncate, and set anomolies.num_snaps
 )
@@ -341,59 +347,29 @@ func (n *NDTParser) processTest(test *fileInfoAndData, testType string) {
 	n.getAndInsertValues(test, testType)
 }
 
-func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
-	// Extract the values from the last snapshot.
-	metrics.WorkerState.WithLabelValues("parse").Inc()
-	defer metrics.WorkerState.WithLabelValues("parse").Dec()
-
-	if !strings.HasSuffix(test.fn, ".gz") {
-		metrics.WarningCount.WithLabelValues(
-			n.TableName(), testType, "uncompressed file").Inc()
-	}
-
-	snaplog, err := web100.NewSnapLog(test.data)
-	if err != nil {
-		metrics.ErrorCount.WithLabelValues(
-			n.TableName(), testType, "snaplog failure").Inc()
-		log.Printf("Unable to parse snaplog for %s, when processing: %s\n%s\n",
-			test.fn, n.taskFileName, err)
-		return
-	}
-
-	valid := true
-	err = snaplog.ValidateSnapshots()
-	if err != nil {
-		log.Printf("ValidateSnapshots failed for %s, when processing: %s (%s)\n",
-			test.fn, n.taskFileName, err)
-		metrics.WarningCount.WithLabelValues(
-			n.TableName(), testType, "validate failed").Inc()
-		// If ValidateSnapshots returns error, it generally means that there
-		// is a problem with the last snapshot, typically a truncated file.
-		// In most cases, there are still many valid snapshots.
-		valid = false
-	}
-
-	// HACK - just to see how expensive the Values() call is...
-	// parse ALL the snapshots.
-	last := &web100.Snapshot{}
-	var deltas []schema.Web100ValueMap
+func (n *NDTParser) getDeltas(snaplog *web100.SnapLog, testType string) ([]schema.Web100ValueMap, int) {
+	deltas := []schema.Web100ValueMap{}
 	deltaFieldCount := 0
+	if NDTOmitDeltas {
+		return deltas, deltaFieldCount
+	}
 	snapshotCount := 0
+	last := &web100.Snapshot{}
 	for count := 0; count < snaplog.SnapCount() && count < maxNumSnapshots; count++ {
 		snap, err := snaplog.Snapshot(count)
 		if err != nil {
 			// TODO - refine label and maybe write a log?
 			metrics.TestCount.WithLabelValues(
 				n.TableName(), testType, "snapshot failure").Inc()
-			return
+			return nil, 0
 		}
 		// Proper sizing avoids evacuate, saving about 20%, excluding BQ code.
 		delta := schema.EmptySnap10()
 		snap.SnapshotDeltas(last, delta)
 		if err != nil {
 			metrics.ErrorCount.WithLabelValues(
 				n.TableName(), testType, "snapValues failure").Inc()
-			return
+			return nil, 0
 		}
 
 		// Delete the constant fields.
@@ -430,6 +406,49 @@ func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
 		// out the most useful tags.
 		deltas[len(deltas)-1]["is_last"] = true
 	}
+
+	return deltas, deltaFieldCount
+}
+
+func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
+	// Extract the values from the last snapshot.
+	metrics.WorkerState.WithLabelValues("parse").Inc()
+	defer metrics.WorkerState.WithLabelValues("parse").Dec()
+
+	if !strings.HasSuffix(test.fn, ".gz") {
+		metrics.WarningCount.WithLabelValues(
+			n.TableName(), testType, "uncompressed file").Inc()
+	}
+
+	snaplog, err := web100.NewSnapLog(test.data)
+	if err != nil {
+		metrics.ErrorCount.WithLabelValues(
+			n.TableName(), testType, "snaplog failure").Inc()
+		log.Printf("Unable to parse snaplog for %s, when processing: %s\n%s\n",
+			test.fn, n.taskFileName, err)
+		return
+	}
+
+	valid := true
+	err = snaplog.ValidateSnapshots()
+	if err != nil {
+		log.Printf("ValidateSnapshots failed for %s, when processing: %s (%s)\n",
+			test.fn, n.taskFileName, err)
+		metrics.WarningCount.WithLabelValues(
+			n.TableName(), testType, "validate failed").Inc()
+		// If ValidateSnapshots returns error, it generally means that there
+		// is a problem with the last snapshot, typically a truncated file.
+		// In most cases, there are still many valid snapshots.
+		valid = false
+	}
+
+	var deltas []schema.Web100ValueMap
+	deltaFieldCount := 0
+	deltas, deltaFieldCount = n.getDeltas(snaplog, testType)
+	if deltas == nil {
+		// There was some kind of major failure parsing snapshots.
+		return
+	}
 	final := snaplog.SnapCount() - 1
 	if final > maxNumSnapshots {
 		final = maxNumSnapshots