Merge pull request #134 from m-lab/schema

Add anomolies fields and fix bug.

Author: gfr10598

parser/ndt.go

@@ -28,7 +28,8 @@ 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.
-	MAX_NUM_SNAPSHOTS = 2800
+	MIN_NUM_SNAPSHOTS = 1600 // If fewer than this, then set anomalies.num_snaps
+	MAX_NUM_SNAPSHOTS = 2800 // If more than this, truncate, and set anomolies.num_snaps
 )
 
 //=========================================================================
@@ -327,12 +328,17 @@ func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
 		return
 	}
 
+	valid := true
 	err = snaplog.ValidateSnapshots()
 	if err != nil {
 		log.Printf("ValidateSnapshots failed for %s, when processing: %s\n%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...
@@ -358,6 +364,7 @@ func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
 			return
 		}
 
+		// Delete the constant fields.
 		delete(delta, "TimeStamps")
 		delete(delta, "StartTimeStamp")
 		delete(delta, "StartTimeUsec")
@@ -385,6 +392,12 @@ func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
 		last = &snap
 	}
 
+	if len(deltas) > 0 {
+		// We tag some of the deltas with specific tags, to make them easy
+		// to find.  is_last is the first, but more will be added as we work
+		// out the most useful tags.
+		deltas[len(deltas)-1]["is_last"] = true
+	}
 	final := snaplog.SnapCount() - 1
 	if final > MAX_NUM_SNAPSHOTS {
 		final = MAX_NUM_SNAPSHOTS
@@ -419,6 +432,13 @@ func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
 
 	results["test_id"] = test.fn
 	results["task_filename"] = n.taskFileName
+	if snaplog.SnapCount() > MAX_NUM_SNAPSHOTS || snaplog.SnapCount() < MIN_NUM_SNAPSHOTS {
+		results["anomalies"].(schema.Web100ValueMap)["num_snaps"] = snaplog.SnapCount()
+	}
+	if !valid {
+		results["anomalies"].(schema.Web100ValueMap)["snaplog_error"] = true
+	}
+
 	// This is the timestamp parsed from the filename.
 	lt, err := test.info.Timestamp.MarshalText()
 	if err != nil {
@@ -442,10 +462,11 @@ func (n *NDTParser) getAndInsertValues(test *fileInfoAndData, testType string) {
 		// TODO - metaFile is currently used only to populate the connection spec.
 		// Should we be using it for anything else?
 		n.metaFile.PopulateConnSpec(connSpec)
+	} else {
 		// TODO Add a log once noise is reduced.
 		metrics.WarningCount.WithLabelValues(
 			n.TableName(), testType, "no meta").Inc()
-	} else {
+		results["anomalies"].(schema.Web100ValueMap)["no_meta"] = true
 		// TODO(dev) - use other information to partially populate
 		// the connection spec.
 	}

schema/web100.go

@@ -131,6 +131,7 @@ func NewWeb100FullRecord(version string, logTime int64, connSpec, snapValues map
 		"log_time": 0,
 		// Can this be part of the metadata service?
 		"connection_spec": FullConnectionSpec(),
+		"anomalies":       Web100ValueMap{},
 		"web100_log_entry": map[string]bigquery.Value{
 			"version":         version,
 			"log_time":        logTime,
@@ -210,6 +211,7 @@ func EmptyGeolocation() Web100ValueMap {
 // All undefined fields will be set to null after a BQ insert.
 func NewWeb100MinimalRecord(version string, logTime int64, connSpec, snapValues Web100ValueMap, deltas []Web100ValueMap) Web100ValueMap {
 	return Web100ValueMap{
+		"anomalies": Web100ValueMap{},
 		"web100_log_entry": Web100ValueMap{
 			"version":         version,
 			"log_time":        logTime,