dashboard/app: scale batching

3 files changed, 65 insertions, 5 deletions

diff --git a/tools/syz-bq.sh b/tools/syz-bq.sh
index 3319f4064..7bf33bbca 100755
--- a/tools/syz-bq.sh
+++ b/tools/syz-bq.sh
@@ -67,6 +67,21 @@ CREATE TABLE IF NOT EXISTS
 gcloud spanner databases ddl update coverage --instance=syzbot --project=syzkaller \
  --ddl="$create_table"
 
+echo "making sure spanner table 'merge_history' exists"
+create_table=$( echo -n '
+CREATE TABLE IF NOT EXISTS
+  merge_history (
+    "namespace" text,
+    "repo" text,
+    "commit" text,
+    "duration" bigint,
+    "dateto" date,
+    "totalrows" bigint,
+  PRIMARY KEY
+    (duration, dateto, commit) );')
+gcloud spanner databases ddl update coverage --instance=syzbot --project=syzkaller \
+ --ddl="$create_table"
+
 echo "Workdir: $workdir"
 base_dir="${workdir}repos/linux_kernels"
 if [ ! -d $base_dir ]; then
@@ -89,10 +104,29 @@ then
 fi
 echo The latest commit as of $to_date is $base_commit.
 
-# rm -rf $base_dir
-# echo Temp dir $base_dir deleted.
-
 from_date=$(date -d "$to_date - $duration days" +%Y-%m-%d)
+# every partition covers 1 day
+query=$(cat <<-END
+SELECT
+  sum(total_rows) as total_rows,
+FROM
+  syzkaller.syzbot_coverage.INFORMATION_SCHEMA.PARTITIONS
+WHERE
+  table_name = '${namespace}' AND
+  PARSE_DATE('%Y%m%d', partition_id) >= '${from_date}' AND
+  PARSE_DATE('%Y%m%d', partition_id) <= '${to_date}';
+END
+)
+
+total_rows=$(bq query --format=csv --use_legacy_sql=false "$query" | tail -n +2)
+if (( total_rows <= 0 ))
+then
+  echo error: no source rows in bigquery available
+  exit
+else
+  echo $total_rows rows are available for processing
+fi
+
 sessionID=$(cat /proc/sys/kernel/random/uuid)
 gsURI=$(echo gs://syzbot-temp/bq-exports/${sessionID}/*.csv.gz)
 echo fetching data from bigquery
@@ -129,7 +163,8 @@ go run ./tools/syz-covermerger/ -workdir $workdir \
   -save-to-spanner true \
   -namespace $namespace \
   -duration $duration \
-  -date-to $to_date
+  -date-to $to_date \
+  -total-rows $total_rows
 
 echo Cleanup
 rm -rf $sessionDir
diff --git a/tools/syz-covermerger/db.go b/tools/syz-covermerger/db.go
index 996e7a6e0..7ba72702d 100644
--- a/tools/syz-covermerger/db.go
+++ b/tools/syz-covermerger/db.go
@@ -24,8 +24,17 @@ type DBRecord struct {
 	Covered      int64
 }
 
+type DBHistoryRecord struct {
+	Namespace string
+	Repo      string
+	Commit    string
+	Duration  int64
+	DateTo    civil.Date
+	TotalRows int64
+}
+
 func saveToSpanner(ctx context.Context, projectID string, coverage map[string]*Coverage,
-	template *DBRecord) {
+	template *DBRecord, totalRows int64) {
 	client, err := spanner.NewClient(ctx, "projects/"+projectID+"/instances/syzbot/databases/coverage")
 	if err != nil {
 		panic(fmt.Sprintf("spanner.NewClient() failed: %s", err.Error()))
@@ -58,6 +67,20 @@ func saveToSpanner(ctx context.Context, projectID string, coverage map[string]*C
 			mutations = nil
 		}
 	}
+
+	var historyInsert *spanner.Mutation
+	if historyInsert, err = spanner.InsertOrUpdateStruct("merge_history", &DBHistoryRecord{
+		Namespace: template.Namespace,
+		Repo:      template.Repo,
+		Commit:    template.Commit,
+		Duration:  template.Duration,
+		DateTo:    template.DateTo,
+		TotalRows: totalRows,
+	}); err != nil {
+		panic(fmt.Sprintf("failed to spanner.InsertStruct(): %s", err.Error()))
+	}
+	mutations = append(mutations, historyInsert)
+
 	if _, err = client.Apply(ctx, mutations); err != nil {
 		panic(fmt.Sprintf("failed to spanner.Apply(inserts): %s", err.Error()))
 	}
diff --git a/tools/syz-covermerger/syz_covermerger.go b/tools/syz-covermerger/syz_covermerger.go
index 063e8c85b..15aab4abb 100644
--- a/tools/syz-covermerger/syz_covermerger.go
+++ b/tools/syz-covermerger/syz_covermerger.go
@@ -28,6 +28,7 @@ var (
 	flagDateTo        = flag.String("date-to", "", "[optional] used to mark DB records")
 	flagSaveToSpanner = flag.String("save-to-spanner", "", "[optional] save aggregation to spanner")
 	flagProjectID     = flag.String("project-id", "syzkaller", "[optional] target spanner db project")
+	flagTotalRows     = flag.Int64("total-rows", 0, "[optional] source size, is used for version contol")
 )
 
 func main() {
@@ -65,6 +66,7 @@ func main() {
 				Duration:  *flagDuration,
 				DateTo:    dateTo,
 			},
+			*flagTotalRows,
 		)
 	}
 }