Add scamper queues for batch and daily (#737)

* add scamper queue for batch and daily

* change cloud function.

* fix typo

* add log

* fix unittest

* Create app-scamper.yaml

* Update queue.yaml

* more fix

* Update .travis.yml

* add log

* add log

* fix data type

* Update pt.go

* Update pt_test.go

* more log

* Update .travis.yml

* clean up

Author: yachang

.travis.yml

@@ -207,7 +207,8 @@ deploy:
                $TRAVIS_BRANCH == ss-sandbox-* ||
                $TRAVIS_BRANCH == fast-sandbox-* ||
                $TRAVIS_BRANCH == pt-sandbox-* ||
-               $TRAVIS_BRANCH == disco-sandbox-*
+               $TRAVIS_BRANCH == disco-sandbox-* ||
+               $TRAVIS_BRANCH == scamper-sandbox-*
 
 ## Service: etl-ndt-parser -- AppEngine Flexible Environment.
 - provider: script
@@ -256,6 +257,17 @@ deploy:
     repo: m-lab/etl
     all_branches: true
     condition: $TRAVIS_BRANCH == pt-sandbox-* || $TRAVIS_BRANCH == sandbox-*
+    
+## Service: etl-scamper-parser -- AppEngine Flexible Environment.
+- provider: script
+  script:
+    $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-sandbox
+    SERVICE_ACCOUNT_mlab_sandbox $TRAVIS_BUILD_DIR/cmd/etl_worker app-scamper.yaml
+  skip_cleanup: true
+  on:
+    repo: m-lab/etl
+    all_branches: true
+    condition: $TRAVIS_BRANCH == scamper-sandbox-* || $TRAVIS_BRANCH == sandbox-*
 
 ## Service: etl-disco-parser -- AppEngine Flexible Environment.
 - provider: script
@@ -303,6 +315,8 @@ deploy:
     && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-staging
     SERVICE_ACCOUNT_mlab_staging $TRAVIS_BUILD_DIR/cmd/etl_worker app-traceroute.yaml
     && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-staging
+    SERVICE_ACCOUNT_mlab_staging $TRAVIS_BUILD_DIR/cmd/etl_worker app-scamper.yaml
+    && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-staging
     SERVICE_ACCOUNT_mlab_staging $TRAVIS_BUILD_DIR/cmd/etl_worker app-sidestream.yaml
     && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-staging
     SERVICE_ACCOUNT_mlab_staging $TRAVIS_BUILD_DIR/cmd/etl_worker fast-sidestream.yaml
@@ -369,6 +383,8 @@ deploy:
     && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-oti
     SERVICE_ACCOUNT_mlab_oti $TRAVIS_BUILD_DIR/cmd/etl_worker app-traceroute.yaml
     && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-oti
+    SERVICE_ACCOUNT_mlab_oti $TRAVIS_BUILD_DIR/cmd/etl_worker app-scamper.yaml
+    && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-oti
     SERVICE_ACCOUNT_mlab_oti $TRAVIS_BUILD_DIR/cmd/etl_worker app-sidestream.yaml
     && $TRAVIS_BUILD_DIR/travis/deploy_app.sh mlab-oti
     SERVICE_ACCOUNT_mlab_oti $TRAVIS_BUILD_DIR/cmd/etl_worker app-disco.yaml

appengine/queue.yaml

@@ -696,6 +696,49 @@ queue:
     max_backoff_seconds: 20
     max_doublings: 0
 
+# Scamper JSON barch queue.
+- name: etl-scamper-batch-0
+  target: etl-batch-parser
+  rate: 1/s
+  bucket_size: 10
+  # max concurrent is limited, because we will share the pipeline with other experiment types
+  max_concurrent_requests: 20
+  retry_parameters:
+    task_age_limit: 12h
+    # min and max are same, so that queue rate only diminishes as tasks are drained
+    # it is set rather low, because traceroute days have relatively few files compared to NDT
+    # and we want the rate limited by the rate/concurrent requests, not by the backoff
+    min_backoff_seconds: 20
+    max_backoff_seconds: 20
+- name: etl-scamper-batch-1
+  target: etl-batch-parser
+  rate: 1/s
+  bucket_size: 10
+  max_concurrent_requests: 20
+  retry_parameters:
+    task_age_limit: 12h
+    min_backoff_seconds: 20
+    max_backoff_seconds: 20
+- name: etl-scamper-batch-2
+  target: etl-batch-parser
+  rate: 1/s
+  bucket_size: 10
+  max_concurrent_requests: 20
+  retry_parameters:
+    task_age_limit: 12h
+    min_backoff_seconds: 20
+    max_backoff_seconds: 20
+- name: etl-scamper-batch-3
+  target: etl-batch-parser
+  rate: 1/s
+  bucket_size: 10
+  max_concurrent_requests: 20
+  retry_parameters:
+    task_age_limit: 12h
+    min_backoff_seconds: 20
+    max_backoff_seconds: 20
+
+
 - name: etl-traceroute-queue
   target: etl-traceroute-parser
   # Average rate at which to release tasks to the service.  Default is 5/sec
@@ -707,6 +750,17 @@ queue:
   # Maximum number of concurrent requests.
   max_concurrent_requests: 5
 
+- name: etl-scamper-queue
+  target: etl-scamper-parser
+  # Average rate at which to release tasks to the service.  Default is 5/sec
+  # This is actually the rate at which tokens are added to the bucket.
+  rate: 1/s
+  # Number of tokens that can accumulate in the bucket.  Default is 5.  This should
+  # have very little impact for our environment.
+  bucket_size: 10
+  # Maximum number of concurrent requests.
+  max_concurrent_requests: 5
+
 - name: etl-sidestream-queue
   target: etl-sidestream-parser
   # Average rate at which to release tasks to the service.  Default is 5/sec

cmd/etl_worker/app-scamper.yaml

@@ -0,0 +1,47 @@
+# Use github release to trigger deployment.
+
+runtime: custom
+env: flex
+service: etl-scamper-parser
+
+# Resource and scaling options. For more background, see:
+#   https://cloud.google.com/appengine/docs/flexible/go/configuring-your-app-with-app-yaml
+
+# TODO(dev): adjust CPU and memory based on actual requirements.
+resources:
+  cpu: 2
+  # Instances support between [(cpu * 0.9) - 0.4, (cpu * 6.5) - 0.4]
+  # Actual memory available is exposed via GAE_MEMORY_MB environment variable.
+  # even though the parser has extremely limited RAM requirements, the
+  # minimum RAM assignable to 2 CPUs is around 1.5GB
+  # We see some OOMs at 3 GB (in etl-batch-parser), so let's try 6
+  memory_gb: 6
+
+automatic_scaling:
+  # We expect fairly steady load, so a modest minimum will rarely cost us anything.
+  min_num_instances: 2
+  max_num_instances: 20
+  cool_down_period_sec: 3000
+  # We don't care much about latency, so a high utilization is desireable.
+  cpu_utilization:
+    target_utilization: 0.6
+
+# Note: add a public port for GCE auto discovery by prometheus.
+# TODO(dev): are any values redundant or irrelevant?
+network:
+  instance_tag: etl-parser
+  name: default
+  # Forward port 9090 on the GCE instance address to the same port in the
+  # container address. Only forward TCP traffic.
+  # Note: the default AppEngine container port 8080 cannot be forwarded.
+  forwarded_ports:
+    - 9090/tcp
+
+env_variables:
+  # These should be substituted in the travis deployment script.
+  RELEASE_TAG: ${TRAVIS_TAG}
+  COMMIT_HASH: ${TRAVIS_COMMIT}
+
+  MAX_WORKERS: 5
+  BIGQUERY_PROJECT: '${INJECTED_PROJECT}'  # Overrides GCLOUD_PROJECT
+  # BIGQUERY_DATASET: 'base_tables' # Overrided computed dataset.

etl/globals.go

@@ -233,6 +233,7 @@ var (
 		"paris-traceroute": PT,
 		"switch":           SW,
 		"tcpinfo":          TCPINFO,
+		"traceroute":       PT,
 	}
 
 	// DataTypeToTable maps from data type to BigQuery table name.

etl/globals_test.go

@@ -92,6 +92,14 @@ func TestValidateTestPath(t *testing.T) {
 				"archive-mlab-oti", "", "paris-traceroute", "2019/06/11", "20190611", "000002", "", "mlab2", "bom01", "paris-traceroute", "0000", "", ".tgz",
 			},
 		},
+		{
+			name:     "scamper-tgz",
+			path:     `gs://archive-mlab-oti/ndt/traceroute/2019/06/20/20190620T224809.435046Z-traceroute-mlab1-den06-ndt.tgz`,
+			wantType: etl.PT,
+			want: &etl.DataPath{
+				"archive-mlab-oti", "ndt", "traceroute", "2019/06/20", "20190620", "224809", "traceroute", "mlab1", "den06", "ndt", "", "", ".tgz",
+			},
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

functions/enqueue.js

@@ -15,8 +15,9 @@ exports.queueForFile = function (filename) {
     experiment_to_task_queue = {
         "switch": "etl-disco-queue",
         "ndt": "etl-ndt-queue",
+        "ndt/traceroute": "etl-scamper-queue",
         "sidestream": "etl-sidestream-queue",
-        "paris-traceroute": "etl-paris-traceroute-queue"
+        "paris-traceroute": "etl-traceroute-queue"
     };
     // TODO - fix this.
     for (key in experiment_to_task_queue) {
@@ -98,4 +99,4 @@ exports.createProdTaskOnFileNotification = function (event, callback) {
 
 exports.createSandboxTaskOnEmbargoFileNotification = exports.createSandboxTaskOnFileNotification
 exports.createStagingTaskOnEmbargoFileNotification = exports.createStagingTaskOnFileNotification
-exports.createProdTaskOnEmbargoFileNotification = exports.createProdTaskOnFileNotification
+exports.createProdTaskOnEmbargoFileNotification = exports.createProdTaskOnFileNotification

parser/pt.go

@@ -163,6 +163,7 @@ func ParseJSON(testName string, rawContent []byte, tableName string, taskFilenam
 			err = json.Unmarshal([]byte(output), &scamperResult)
 			if err != nil {
 				// fail and return here.
+				log.Printf("extra jasonnet processing failed for %s", testName)
 				return schema.PTTest{}, err
 			}
 		}
@@ -498,7 +499,7 @@ func (pt *PTParser) NumBufferedTests() int {
 
 // IsParsable returns the canonical test type and whether to parse data.
 func (pt *PTParser) IsParsable(testName string, data []byte) (string, bool) {
-	if strings.HasSuffix(testName, ".paris") {
+	if strings.HasSuffix(testName, ".paris") || strings.HasSuffix(testName, ".jsonl") {
 		return "paris", true
 	}
 	return "unknown", false
@@ -517,7 +518,7 @@ func (pt *PTParser) ParseAndInsert(meta map[string]bigquery.Value, testName stri
 	}
 
 	// Process the json output of Scamper binary.
-	if strings.Contains(pt.taskFileName, "jsonl") {
+	if strings.Contains(testName, "jsonl") {
 		ptTest, err := ParseJSON(testName, rawContent, pt.TableName(), pt.taskFileName)
 		if err == nil {
 			err := pt.AddRow(&ptTest)
@@ -527,6 +528,9 @@ func (pt *PTParser) ParseAndInsert(meta map[string]bigquery.Value, testName stri
 				pt.PutAsync(pt.TakeRows())
 				pt.AddRow(&ptTest)
 			}
+		} else {
+			// Modify metrics
+			log.Printf("JSON parsing failed with error %v for %s", err, testName)
 		}
 		return nil
 	}