Add new field Counter to switch schema. (#951)

* Add new field Counter to switch schema.

* Changes Counter to a uint64, since that is what DISCOv2 outputs.

* Change Value to a float64 and Counter to int64.

* Removes all JSON type hints from switch schema struct fields.

* Removes all code/logic related to Cloud Functions.

* Adds comment on why struct field types differ from DISCOv2.

* Adds update-schema back to Travis. Adds auto detect of SwitchStats for update-schema.

* Removes partField definition from CreateOrUpdateSwitchStats.

* Replaces receiver var s with row.

* Removes an outdated comment. Removes cf-prod case for update-schema deployment.

* Adds cases for SwitchStats in switch block.

* Changes int64 'timestamp's to time.Time type.

* Manually changes type of Sample.Timestamp for BigQuery.

* Adds a comment about why we subtitute an int64 for a TIMESTAMP.

* Moves substitutions before search for docs for schema.

Author: nkinkade

.travis.yml

@@ -9,11 +9,6 @@
 #  * test and build the go code
 #  * on success, deploy the result when the origin branch matches a supported
 #    deployment target.
-#
-# NOTE: Cloud functions only support primitive IAM roles: Owner, Editor, Viewer.
-# See: https://cloud.google.com/functions/docs/concepts/iam
-# TODO(soltesz): Add deployment automation when fine-grained permissions are
-# possible.
 
 dist: bionic
 language: go
@@ -25,12 +20,6 @@ env:
 
 before_install:
 - sudo apt-get install -y jq  # Dependency for sync_tables_with_schema.sh.
-# Install javascript libraries
-- pushd $TRAVIS_BUILD_DIR/functions
-- npm install --verbose
-- pushd embargo
-- npm install --verbose
-- popd; popd
 
 # Coverage tools
 - go get github.com/mattn/goveralls
@@ -84,13 +73,6 @@ script:
 - diff /tmp/current-bindata.go schema/bindata.go || (
   echo "Files do not match; run 'update go-bindata, go generate ./schema' and commit changes" && false )
 
-# Run all javascript tests.
-- pushd $TRAVIS_BUILD_DIR/functions
-- npm test
-- pushd embargo
-- npm test
-- popd; popd
-
 # To start the Go tests, run all the non-integration tests.
 # Currently skipping storage tests, because they depend on GCS, and there is
 # no emulator.
@@ -185,20 +167,6 @@ deploy:
     all_branches: true
     condition: $TRAVIS_BRANCH == qp-sandbox-*
 
-## Service: cloud function -- AppEngine Flexible Environment.
-- provider: script
-  script:
-    $TRAVIS_BUILD_DIR/travis/activate_service_account.sh SERVICE_ACCOUNT_mlab_sandbox
-    && cd $TRAVIS_BUILD_DIR/functions
-    && gcloud config set project mlab-sandbox
-    && gcloud functions deploy createSandboxTaskOnFileNotification --stage-bucket=functions-mlab-sandbox --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=archive-mlab-sandbox
-    && gcloud functions deploy createSandboxTaskOnEmbargoFileNotification --stage-bucket=functions-mlab-sandbox --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=embargo-mlab-sandbox
-  skip_cleanup: true
-  on:
-    repo: m-lab/etl
-    all_branches: true
-    condition: $TRAVIS_BRANCH == cf-sandbox-* || $TRAVIS_BRANCH == sandbox-*
-
 ## Task Queues
 - provider: script
   script:
@@ -294,11 +262,6 @@ deploy:
        $TRAVIS_BUILD_DIR/travis/kubectl.sh mlab-staging data-processing ./apply-cluster.sh
     && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-staging
        SERVICE_ACCOUNT_mlab_staging $TRAVIS_BUILD_DIR/cmd/etl_worker app-batch.yaml
-    && cd $TRAVIS_BUILD_DIR/functions
-    && gcloud functions deploy createStagingTaskOnFileNotification --project=mlab-staging --stage-bucket=functions-mlab-staging --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=archive-mlab-staging
-    && gcloud functions deploy createStagingTaskOnEmbargoFileNotification --project=mlab-staging --stage-bucket=functions-mlab-staging --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=embargo-mlab-staging
-    && cd $TRAVIS_BUILD_DIR/functions/embargo
-    && gcloud functions deploy embargoOnFileNotificationStaging --project=mlab-staging --stage-bucket=functions-mlab-staging --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=scraper-mlab-staging
 #    && $TRAVIS_BUILD_DIR/etl-schema/schema/sync_tables_with_schema.sh mlab-staging batch nodryrun
 #    && $TRAVIS_BUILD_DIR/etl-schema/schema/sync_tables_with_schema.sh mlab-staging base_tables nodryrun
   skip_cleanup: true
@@ -313,24 +276,6 @@ deploy:
 #  Triggers when *ANY* branch is tagged with one of these tags'
 ######################################################################
 
-## Service: cloud function -- AppEngine Flexible Environment.
-# TODO - update-schema is run on many triggers, often multiple times.  Should clean it up.
-- provider: script
-  script:
-    $TRAVIS_BUILD_DIR/travis/activate_service_account.sh SERVICE_ACCOUNT_mlab_oti
-    && $TRAVIS_BUILD_DIR/travis/run_with_application_credentials.sh mlab-oti
-       SERVICE_ACCOUNT_mlab_oti $TRAVIS_BUILD_DIR/cmd/update-schema update-schema
-    && cd $TRAVIS_BUILD_DIR/functions
-    && gcloud functions deploy createProdTaskOnFileNotification --project=mlab-oti --stage-bucket=functions-mlab-oti --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=archive-mlab-oti
-    && gcloud functions deploy createProdTaskOnEmbargoFileNotification --project=mlab-oti --stage-bucket=functions-mlab-oti --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=embargo-mlab-oti
-    && cd $TRAVIS_BUILD_DIR/functions/embargo
-    && gcloud functions deploy embargoOnFileNotificationOti --project=mlab-oti --stage-bucket=functions-mlab-oti --trigger-event=providers/cloud.storage/eventTypes/object.change --trigger-resource=scraper-mlab-oti
-  skip_cleanup: true
-  on:
-    repo: m-lab/etl
-    all_branches: true
-    condition: $TRAVIS_BRANCH == cf-prod-* || $TRAVIS_BRANCH == ndt-prod-* || $TRAVIS_BRANCH == prod-*
-
 ###################### UNIVERSAL PARSER ###############################
 # Deploys Universal K8S Parser.
 # Triggers when *ANY* branch is tagged with u-prod-* OR prod-*
@@ -400,3 +345,17 @@ deploy:
     repo: m-lab/etl
     all_branches: true
     condition: $TRAVIS_TAG == ndt-prod-* || $TRAVIS_TAG == prod-*
+
+
+##################### UPDATE SCHEMAS ###############################
+# Updates or creates schemas in BigQuery
+- provider: script
+  script:
+    $TRAVIS_BUILD_DIR/travis/activate_service_account.sh SERVICE_ACCOUNT_mlab_oti
+    && $TRAVIS_BUILD_DIR/travis/run_with_application_credentials.sh mlab-oti
+       SERVICE_ACCOUNT_mlab_oti $TRAVIS_BUILD_DIR/cmd/update-schema update-schema
+  skip_cleanup: true
+  on:
+    repo: m-lab/etl
+    all_branches: true
+    condition: $TRAVIS_BRANCH == ndt-prod-* || $TRAVIS_BRANCH == prod-*

cmd/update-schema/update.go

@@ -66,6 +66,13 @@ func CreateOrUpdateAnnotationRow(project string, dataset string, table string) e
 	return CreateOrUpdate(schema, project, dataset, table, "Date")
 }
 
+func CreateOrUpdateSwitchStats(project string, dataset string, table string) error {
+	row := schema.SwitchStats{}
+	schema, err := row.Schema()
+	rtx.Must(err, "SwitchStats.Schema")
+	return CreateOrUpdate(schema, project, dataset, table, "")
+}
+
 // CreateOrUpdate will update or create a table from the given schema.
 func CreateOrUpdate(schema bigquery.Schema, project, dataset, table, partField string) error {
 	name := project + "." + dataset + "." + table
@@ -141,6 +148,12 @@ func updateLegacyTables(project string) int {
 	if err := CreateOrUpdateNDT5ResultRow(project, "batch", "ndt5"); err != nil {
 		errCount++
 	}
+	if err := CreateOrUpdateSwitchStats(project, "base_tables", "switch"); err != nil {
+		errCount++
+	}
+	if err := CreateOrUpdateSwitchStats(project, "batch", "switch"); err != nil {
+		errCount++
+	}
 	return errCount
 }
 
@@ -216,6 +229,14 @@ func main() {
 			errCount++
 		}
 
+	case "switch":
+		if err := CreateOrUpdateSwitchStats(*project, "base_tables", "switch"); err != nil {
+			errCount++
+		}
+		if err := CreateOrUpdateSwitchStats(*project, "batch", "switch"); err != nil {
+			errCount++
+		}
+
 	default:
 		log.Fatal("invalid updateType: ", *updateType)
 	}

schema/switch_schema.go

@@ -1,28 +1,71 @@
 package schema
 
-import "time"
+import (
+	"cloud.google.com/go/bigquery"
+
+	"github.com/m-lab/go/cloud/bqx"
+
+	"time"
+)
 
 // Sample is an individual measurement taken by DISCO.
+// NOTE: the types of the fields in this struct differ from the types used
+// natively by the structs in DISCOv2. In DiSCOv2 Value is a uint64, but must
+// be a float here because DISCOv1 outputs floats. float64 should be able to
+// accommodate both types of input values safely. For Counter, DISCOv2 uses a
+// uint64, but BigQuery does not support the notion of unsigned integers, so we
+// use int64 here, which should be safe, too.
 type Sample struct {
-	Timestamp int64   `json:"timestamp,int64" bigquery:"timestamp"`
-	Value     float32 `json:"value,float32" bigquery:"value"`
+	Timestamp int64   `json:"timestamp" bigquery:"timestamp"`
+	Value     float64 `json:"value" bigquery:"value"`
+	Counter   int64   `json:"counter" bigquery:"counter"`
 }
 
 // SwitchStats represents a row of data taken from the raw DISCO export file.
 type SwitchStats struct {
-	TaskFilename  string    `json:"task_filename,string" bigquery:"task_filename"`
-	TestID        string    `json:"test_id,string" bigquery:"test_id"`
+	TaskFilename  string    `json:"task_filename" bigquery:"task_filename"`
+	TestID        string    `json:"test_id" bigquery:"test_id"`
 	ParseTime     time.Time `json:"parse_time" bigquery:"parse_time"`
-	ParserVersion string    `json:"parser_version,string" bigquery:"parser_version"`
-	LogTime       int64     `json:"log_time,int64" bigquery:"log_time"`
+	ParserVersion string    `json:"parser_version" bigquery:"parser_version"`
+	LogTime       time.Time `json:"log_time" bigquery:"log_time"`
 	Sample        []Sample  `json:"sample" bigquery:"sample"`
 	Metric        string    `json:"metric" bigquery:"metric"`
 	Hostname      string    `json:"hostname" bigquery:"hostname"`
 	Experiment    string    `json:"experiment" bigquery:"experiment"`
 }
 
 // Size estimates the number of bytes in the SwitchStats object.
-func (s *SwitchStats) Size() int {
-	return (len(s.TaskFilename) + len(s.TestID) + 8 +
-		12*len(s.Sample) + len(s.Metric) + len(s.Hostname) + len(s.Experiment))
+func (row *SwitchStats) Size() int {
+	return (len(row.TaskFilename) + len(row.TestID) + 8 +
+		12*len(row.Sample) + len(row.Metric) + len(row.Hostname) + len(row.Experiment))
+}
+
+// Schema returns the BigQuery schema for SwitchStats.
+func (row *SwitchStats) Schema() (bigquery.Schema, error) {
+	sch, err := bigquery.InferSchema(row)
+	if err != nil {
+		return bigquery.Schema{}, err
+	}
+
+	// The raw data from DISCO stores the timestamp of a sample as an integer (a
+	// UNIX timestamp), but BigQuery represent the value as type TIMESTAMP.
+	// TODO: DISCO should probably store the value as time.Time to avoid the
+	// need for this.
+	subs := map[string]bigquery.FieldSchema{
+		"timestamp": bigquery.FieldSchema{
+			Name:        "timestamp",
+			Description: "",
+			Repeated:    false,
+			Required:    false,
+			Type:        "TIMESTAMP"},
+	}
+	c := bqx.Customize(sch, subs)
+
+	docs := FindSchemaDocsFor(row)
+	for _, doc := range docs {
+		bqx.UpdateSchemaDescription(c, doc)
+	}
+	rr := bqx.RemoveRequired(c)
+
+	return rr, err
 }