Upload bench output as build artifacts (#1478)

* Upload bench output as build artifacts

* Fix tags types

* Pull previous stats from metrics

* Change the default branch for comparison

* Fix formatting

* Fix build errors

* Address review comments

* Dedup some common code

* Add eval for channel info to find branch name

ci/test-bench.sh

@@ -2,6 +2,11 @@
 
 cd "$(dirname "$0")/.."
 
+# shellcheck disable=SC1091
+source ci/upload_ci_artifact.sh
+
+eval "$(ci/channel-info.sh)"
+
 ci/version-check.sh nightly
 export RUST_BACKTRACE=1
 
@@ -12,6 +17,17 @@ _() {
 
 set -o pipefail
 
+UPLOAD_METRICS=""
+TARGET_BRANCH=$BUILDKITE_BRANCH
+if [[ -z $BUILDKITE_BRANCH ]] || ./ci/is-pr.sh; then
+  TARGET_BRANCH=$EDGE_CHANNEL
+else
+  UPLOAD_METRICS="upload"
+fi
+
 BENCH_FILE=bench_output.log
-_ cargo bench --features=unstable --verbose -- -Z unstable-options --format=json | tee $BENCH_FILE
-_ cargo run --release --bin solana-upload-perf -- $BENCH_FILE
+BENCH_ARTIFACT=current_bench_results.log
+_ cargo bench --features=unstable --verbose -- -Z unstable-options --format=json | tee "$BENCH_FILE"
+_ cargo run --release --bin solana-upload-perf -- "$BENCH_FILE" "$TARGET_BRANCH" "$UPLOAD_METRICS" >"$BENCH_ARTIFACT"
+
+upload_ci_artifact "$BENCH_ARTIFACT"

src/bin/upload-perf.rs

@@ -4,11 +4,33 @@ extern crate solana;
 use influx_db_client as influxdb;
 use serde_json::Value;
 use solana::metrics;
+use std::collections::HashMap;
 use std::env;
 use std::fs::File;
 use std::io::{BufRead, BufReader};
 use std::process::Command;
 
+fn get_last_metrics(metric: &str, db: &str, name: &str, branch: &str) -> Result<String, String> {
+    let query = format!(
+        r#"SELECT last("{}") FROM "{}"."autogen"."{}" WHERE "branch"='{}'"#,
+        metric, db, name, branch
+    );
+
+    let response = metrics::query(&query)?;
+
+    match serde_json::from_str(&response) {
+        Result::Ok(v) => {
+            let v: Value = v;
+            let data = &v["results"][0]["series"][0]["values"][0][1];
+            if data.is_null() {
+                return Result::Err("Key not found".to_string());
+            }
+            Result::Ok(data.to_string())
+        }
+        Result::Err(err) => Result::Err(err.to_string()),
+    }
+}
+
 fn main() {
     let args: Vec<String> = env::args().collect();
     // Open the path in read-only mode, returns `io::Result<File>`
@@ -18,33 +40,77 @@ fn main() {
         Ok(file) => file,
     };
 
+    let branch = &args[2];
+    let upload_metrics = args.len() > 2;
+
     let git_output = Command::new("git")
         .args(&["rev-parse", "HEAD"])
         .output()
         .expect("failed to execute git rev-parse");
     let git_commit_hash = String::from_utf8_lossy(&git_output.stdout);
     let trimmed_hash = git_commit_hash.trim().to_string();
-    println!("uploading hash: {}", trimmed_hash);
+
+    let mut last_commit = None;
+    let mut results = HashMap::new();
+
+    let db = env::var("INFLUX_DATABASE").unwrap_or_else(|_| "scratch".to_string());
 
     for line in BufReader::new(file).lines() {
         if let Ok(v) = serde_json::from_str(&line.unwrap()) {
             let v: Value = v;
             if v["type"] == "bench" {
-                println!("{}", v);
-                println!("  {}", v["type"]);
+                let name = v["name"].as_str().unwrap().trim_matches('\"').to_string();
+
+                last_commit = match get_last_metrics(&"commit".to_string(), &db, &name, &branch) {
+                    Result::Ok(v) => Some(v),
+                    Result::Err(_) => None,
+                };
+
                 let median = v["median"].to_string().parse().unwrap();
                 let deviation = v["deviation"].to_string().parse().unwrap();
-                metrics::submit(
-                    influxdb::Point::new(&v["name"].as_str().unwrap().trim_matches('\"'))
-                        .add_field("median", influxdb::Value::Integer(median))
-                        .add_field("deviation", influxdb::Value::Integer(deviation))
-                        .add_field(
-                            "commit",
-                            influxdb::Value::String(git_commit_hash.trim().to_string()),
-                        ).to_owned(),
-                );
+                if upload_metrics {
+                    metrics::submit(
+                        influxdb::Point::new(&v["name"].as_str().unwrap().trim_matches('\"'))
+                            .add_tag("test", influxdb::Value::String("bench".to_string()))
+                            .add_tag("branch", influxdb::Value::String(branch.to_string()))
+                            .add_field("median", influxdb::Value::Integer(median))
+                            .add_field("deviation", influxdb::Value::Integer(deviation))
+                            .add_field(
+                                "commit",
+                                influxdb::Value::String(git_commit_hash.trim().to_string()),
+                            ).to_owned(),
+                    );
+                }
+                let last_median = get_last_metrics(&"median".to_string(), &db, &name, &branch)
+                    .unwrap_or_default();
+                let last_deviation =
+                    get_last_metrics(&"deviation".to_string(), &db, &name, &branch)
+                        .unwrap_or_default();
+
+                results.insert(name, (median, deviation, last_median, last_deviation));
             }
         }
     }
+
+    if let Some(commit) = last_commit {
+        println!(
+            "Comparing current commits: {} against baseline {} on {} branch",
+            trimmed_hash, commit, branch
+        );
+        println!("bench_name, median, last_median, deviation, last_deviation");
+        for (entry, values) in results {
+            println!(
+                "{}, {}, {}, {}, {}",
+                entry, values.0, values.2, values.1, values.3
+            );
+        }
+    } else {
+        println!("No previous results found for {} branch", branch);
+        println!("hash: {}", trimmed_hash);
+        println!("bench_name, median, deviation");
+        for (entry, values) in results {
+            println!("{}, {}, {}", entry, values.0, values.1);
+        }
+    }
     metrics::flush();
 }

src/metrics.rs

@@ -1,5 +1,7 @@
 //! The `metrics` module enables sending measurements to an InfluxDB instance
 
+extern crate reqwest;
+
 use influx_db_client as influxdb;
 use std::env;
 use std::sync::mpsc::{channel, Receiver, RecvTimeoutError, Sender};
@@ -37,11 +39,7 @@ impl InfluxDbMetricsWriter {
     }
 
     fn build_client() -> Option<influxdb::Client> {
-        let host = env::var("INFLUX_HOST")
-            .unwrap_or_else(|_| "https://metrics.solana.com:8086".to_string());
-        let db = env::var("INFLUX_DATABASE").unwrap_or_else(|_| "scratch".to_string());
-        let username = env::var("INFLUX_USERNAME").unwrap_or_else(|_| "scratch_writer".to_string());
-        let password = env::var("INFLUX_PASSWORD").unwrap_or_else(|_| "topsecret".to_string());
+        let (host, db, username, password) = get_env_settings();
 
         debug!("InfluxDB host={} db={} username={}", host, db, username);
         let mut client = influxdb::Client::new_with_option(host, db, None)
@@ -177,6 +175,27 @@ pub fn submit(point: influxdb::Point) {
     agent.submit(point);
 }
 
+fn get_env_settings() -> (String, String, String, String) {
+    let host =
+        env::var("INFLUX_HOST").unwrap_or_else(|_| "https://metrics.solana.com:8086".to_string());
+    let db = env::var("INFLUX_DATABASE").unwrap_or_else(|_| "scratch".to_string());
+    let username = env::var("INFLUX_USERNAME").unwrap_or_else(|_| "scratch_writer".to_string());
+    let password = env::var("INFLUX_PASSWORD").unwrap_or_else(|_| "topsecret".to_string());
+    (host, db, username, password)
+}
+
+pub fn query(q: &str) -> Result<String, String> {
+    let (host, _, username, password) = get_env_settings();
+    let query = format!("{}/query?u={}&p={}&q={}", &host, &username, &password, &q);
+
+    let response = reqwest::get(query.as_str())
+        .map_err(|err| err.to_string())?
+        .text()
+        .map_err(|err| err.to_string())?;
+
+    Result::Ok(response)
+}
+
 /// Blocks until all pending points from previous calls to `submit` have been
 /// transmitted.
 pub fn flush() {