Add SkiaPerfGcsAdaptor and its tests (#70674)

dev/benchmarks/metrics_center/lib/src/skiaperf.dart

@@ -2,7 +2,13 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+import 'dart:convert';
+
+import 'package:gcloud/storage.dart';
+import 'package:googleapis/storage/v1.dart' show DetailedApiRequestError;
+
 import 'package:metrics_center/src/common.dart';
+import 'package:metrics_center/src/github_helper.dart';
 
 // Skia Perf Format is a JSON file that looks like:
 
@@ -181,6 +187,141 @@ class SkiaPerfPoint extends MetricPoint {
   final Map<String, String> _options;
 }
 
+/// Handle writing and updates of Skia perf GCS buckets.
+class SkiaPerfGcsAdaptor {
+  /// Construct the adaptor given the associated GCS bucket where the data is
+  /// read from and written to.
+  SkiaPerfGcsAdaptor(this._gcsBucket) : assert(_gcsBucket != null);
+
+  /// Used by Skia to differentiate json file format versions.
+  static const int version = 1;
+
+  /// Write a list of SkiaPerfPoint into a GCS file with name `objectName` in
+  /// the proper json format that's understandable by Skia perf services.
+  ///
+  /// The `objectName` must be a properly formatted string returned by
+  /// [computeObjectName].
+  Future<void> writePoints(
+      String objectName, List<SkiaPerfPoint> points) async {
+    final String jsonString = jsonEncode(SkiaPerfPoint.toSkiaPerfJson(points));
+    await _gcsBucket.writeBytes(objectName, utf8.encode(jsonString));
+  }
+
+  /// Read a list of `SkiaPerfPoint` that have been previously written to the
+  /// GCS file with name `objectName`.
+  ///
+  /// The Github repo and revision of those points will be inferred from the
+  /// `objectName`.
+  ///
+  /// Return an empty list if the object does not exist in the GCS bucket.
+  ///
+  /// The read may retry multiple times if transient network errors with code
+  /// 504 happens.
+  Future<List<SkiaPerfPoint>> readPoints(String objectName) async {
+    // Retry multiple times as GCS may return 504 timeout.
+    for (int retry = 0; retry < 5; retry += 1) {
+      try {
+        return await _readPointsWithoutRetry(objectName);
+      } catch (e) {
+        if (e is DetailedApiRequestError && e.status == 504) {
+          continue;
+        }
+        rethrow;
+      }
+    }
+    // Retry one last time and let the exception go through.
+    return await _readPointsWithoutRetry(objectName);
+  }
+
+  Future<List<SkiaPerfPoint>> _readPointsWithoutRetry(String objectName) async {
+    ObjectInfo info;
+
+    try {
+      info = await _gcsBucket.info(objectName);
+    } catch (e) {
+      if (e.toString().contains('No such object')) {
+        return <SkiaPerfPoint>[];
+      } else {
+        rethrow;
+      }
+    }
+
+    final Stream<List<int>> stream = _gcsBucket.read(objectName);
+    final Stream<int> byteStream = stream.expand((List<int> x) => x);
+    final Map<String, dynamic> decodedJson =
+        jsonDecode(utf8.decode(await byteStream.toList()))
+            as Map<String, dynamic>;
+
+    final List<SkiaPerfPoint> points = <SkiaPerfPoint>[];
+
+    final String firstGcsNameComponent = objectName.split('/')[0];
+    _populateGcsNameToGithubRepoMapIfNeeded();
+    final String githubRepo = _gcsNameToGithubRepo[firstGcsNameComponent];
+    assert(githubRepo != null);
+
+    final String gitHash = decodedJson[kSkiaPerfGitHashKey] as String;
+    final Map<String, dynamic> results =
+        decodedJson[kSkiaPerfResultsKey] as Map<String, dynamic>;
+    for (final String name in results.keys) {
+      final Map<String, dynamic> subResultMap =
+          results[name][kSkiaPerfDefaultConfig] as Map<String, dynamic>;
+      for (final String subResult
+          in subResultMap.keys.where((String s) => s != kSkiaPerfOptionsKey)) {
+        points.add(SkiaPerfPoint._(
+          githubRepo,
+          gitHash,
+          name,
+          subResult,
+          subResultMap[subResult] as double,
+          (subResultMap[kSkiaPerfOptionsKey] as Map<String, dynamic>)
+              .cast<String, String>(),
+          info.downloadLink.toString(),
+        ));
+      }
+    }
+    return points;
+  }
+
+  /// Compute the GCS file name that's used to store metrics for a given commit
+  /// (git revision).
+  ///
+  /// Skia perf needs all directory names to be well formatted. The final name
+  /// of the json file (currently `values.json`) can be arbitrary, and multiple
+  /// json files can be put in that leaf directory. We intend to use multiple
+  /// json files in the future to scale up the system if too many writes are
+  /// competing for the same json file.
+  static Future<String> comptueObjectName(String githubRepo, String revision,
+      {GithubHelper githubHelper}) async {
+    assert(_githubRepoToGcsName[githubRepo] != null);
+    final String topComponent = _githubRepoToGcsName[githubRepo];
+    final DateTime t = await (githubHelper ?? GithubHelper())
+        .getCommitDateTime(githubRepo, revision);
+    final String month = t.month.toString().padLeft(2, '0');
+    final String day = t.day.toString().padLeft(2, '0');
+    final String hour = t.hour.toString().padLeft(2, '0');
+    final String dateComponents = '${t.year}/$month/$day/$hour';
+    return '$topComponent/$dateComponents/$revision/values.json';
+  }
+
+  static final Map<String, String> _githubRepoToGcsName = <String, String>{
+    kFlutterFrameworkRepo: 'flutter-flutter',
+    kFlutterEngineRepo: 'flutter-engine',
+  };
+  static final Map<String, String> _gcsNameToGithubRepo = <String, String>{};
+
+  static void _populateGcsNameToGithubRepoMapIfNeeded() {
+    if (_gcsNameToGithubRepo.isEmpty) {
+      for (final String repo in _githubRepoToGcsName.keys) {
+        final String gcsName = _githubRepoToGcsName[repo];
+        assert(_gcsNameToGithubRepo[gcsName] == null);
+        _gcsNameToGithubRepo[gcsName] = repo;
+      }
+    }
+  }
+
+  final Bucket _gcsBucket;
+}
+
 const String kSkiaPerfGitHashKey = 'gitHash';
 const String kSkiaPerfResultsKey = 'results';
 const String kSkiaPerfValueKey = 'value';

dev/benchmarks/metrics_center/pubspec.yaml

@@ -7,7 +7,6 @@ dependencies:
   args: 1.6.0
   crypto: 2.1.5
   gcloud: 0.7.3
-  googleapis: 0.56.1
   googleapis_auth: 0.2.12
   github: 7.0.3
   equatable: 1.2.5
@@ -66,4 +65,4 @@ dev_dependencies:
   webkit_inspection_protocol: 0.7.4 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   yaml: 2.2.1 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
 
-# PUBSPEC CHECKSUM: 734c
+# PUBSPEC CHECKSUM: 6a9a

dev/benchmarks/metrics_center/test/skiaperf_test.dart

@@ -6,17 +6,32 @@
 
 import 'dart:convert';
 
+import 'package:gcloud/storage.dart';
+import 'package:googleapis/storage/v1.dart' show DetailedApiRequestError;
+import 'package:googleapis_auth/auth_io.dart';
+import 'package:metrics_center/src/github_helper.dart';
+import 'package:mockito/mockito.dart';
+
 import 'package:metrics_center/src/common.dart';
 import 'package:metrics_center/src/flutter.dart';
 import 'package:metrics_center/src/skiaperf.dart';
 
 import 'common.dart';
+import 'utility.dart';
+
+class MockBucket extends Mock implements Bucket {}
+
+class MockObjectInfo extends Mock implements ObjectInfo {}
+
+class MockGithubHelper extends Mock implements GithubHelper {}
 
-void main() {
+Future<void> main() async {
   const double kValue1 = 1.0;
   const double kValue2 = 2.0;
 
   const String kFrameworkRevision1 = '9011cece2595447eea5dd91adaa241c1c9ef9a33';
+  const String kEngineRevision1 = '617938024315e205f26ed72ff0f0647775fa6a71';
+  const String kEngineRevision2 = '5858519139c22484aaff1cf5b26bdf7951259344';
   const String kTaskName = 'analyzer_benchmark';
   const String kMetric1 = 'flutter_repo_batch_maximum';
   const String kMetric2 = 'flutter_repo_watch_maximum';
@@ -262,4 +277,206 @@ void main() {
       throwsA(anything),
     );
   });
+
+  test('SkiaPerfGcsAdaptor computes name correctly', () async {
+    final MockGithubHelper mockHelper = MockGithubHelper();
+    when(mockHelper.getCommitDateTime(
+            kFlutterFrameworkRepo, kFrameworkRevision1))
+        .thenAnswer((_) => Future<DateTime>.value(DateTime(2019, 12, 4, 23)));
+    expect(
+      await SkiaPerfGcsAdaptor.comptueObjectName(
+        kFlutterFrameworkRepo,
+        kFrameworkRevision1,
+        githubHelper: mockHelper,
+      ),
+      equals('flutter-flutter/2019/12/04/23/$kFrameworkRevision1/values.json'),
+    );
+    when(mockHelper.getCommitDateTime(kFlutterEngineRepo, kEngineRevision1))
+        .thenAnswer((_) => Future<DateTime>.value(DateTime(2019, 12, 3, 20)));
+    expect(
+      await SkiaPerfGcsAdaptor.comptueObjectName(
+        kFlutterEngineRepo,
+        kEngineRevision1,
+        githubHelper: mockHelper,
+      ),
+      equals('flutter-engine/2019/12/03/20/$kEngineRevision1/values.json'),
+    );
+    when(mockHelper.getCommitDateTime(kFlutterEngineRepo, kEngineRevision2))
+        .thenAnswer((_) => Future<DateTime>.value(DateTime(2020, 1, 3, 15)));
+    expect(
+      await SkiaPerfGcsAdaptor.comptueObjectName(
+        kFlutterEngineRepo,
+        kEngineRevision2,
+        githubHelper: mockHelper,
+      ),
+      equals('flutter-engine/2020/01/03/15/$kEngineRevision2/values.json'),
+    );
+  });
+
+  test('Successfully read mock GCS that fails 1st time with 504', () async {
+    final MockBucket testBucket = MockBucket();
+    final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
+
+    final String testObjectName = await SkiaPerfGcsAdaptor.comptueObjectName(
+        kFlutterFrameworkRepo, kFrameworkRevision1);
+
+    final List<SkiaPerfPoint> writePoints = <SkiaPerfPoint>[
+      SkiaPerfPoint.fromPoint(cocoonPointRev1Metric1),
+    ];
+    final String skiaPerfJson =
+        jsonEncode(SkiaPerfPoint.toSkiaPerfJson(writePoints));
+    await skiaPerfGcs.writePoints(testObjectName, writePoints);
+    verify(testBucket.writeBytes(testObjectName, utf8.encode(skiaPerfJson)));
+
+    // Emulate the first network request to fail with 504.
+    when(testBucket.info(testObjectName))
+        .thenThrow(DetailedApiRequestError(504, 'Test Failure'));
+
+    final MockObjectInfo mockObjectInfo = MockObjectInfo();
+    when(mockObjectInfo.downloadLink)
+        .thenReturn(Uri.https('test.com', 'mock.json'));
+    when(testBucket.info(testObjectName))
+        .thenAnswer((_) => Future<ObjectInfo>.value(mockObjectInfo));
+    when(testBucket.read(testObjectName))
+        .thenAnswer((_) => Stream<List<int>>.value(utf8.encode(skiaPerfJson)));
+
+    final List<SkiaPerfPoint> readPoints =
+        await skiaPerfGcs.readPoints(testObjectName);
+    expect(readPoints.length, equals(1));
+    expect(readPoints[0].testName, kTaskName);
+    expect(readPoints[0].subResult, kMetric1);
+    expect(readPoints[0].value, kValue1);
+    expect(readPoints[0].githubRepo, kFlutterFrameworkRepo);
+    expect(readPoints[0].gitHash, kFrameworkRevision1);
+    expect(readPoints[0].jsonUrl, 'https://test.com/mock.json');
+  });
+
+  test('Return empty list if the GCS file does not exist', () async {
+    final MockBucket testBucket = MockBucket();
+    final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
+    final String testObjectName = await SkiaPerfGcsAdaptor.comptueObjectName(
+        kFlutterFrameworkRepo, kFrameworkRevision1);
+    when(testBucket.info(testObjectName))
+        .thenThrow(Exception('No such object'));
+    expect((await skiaPerfGcs.readPoints(testObjectName)).length, 0);
+  });
+
+  // The following is for integration tests.
+  Bucket testBucket;
+  final Map<String, dynamic> credentialsJson = getTestGcpCredentialsJson();
+  if (credentialsJson != null) {
+    final ServiceAccountCredentials credentials =
+        ServiceAccountCredentials.fromJson(credentialsJson);
+
+    final AutoRefreshingAuthClient client =
+        await clientViaServiceAccount(credentials, Storage.SCOPES);
+    final Storage storage =
+        Storage(client, credentialsJson['project_id'] as String);
+
+    const String kTestBucketName = 'flutter-skia-perf-test';
+
+    assert(await storage.bucketExists(kTestBucketName));
+    testBucket = storage.bucket(kTestBucketName);
+  }
+
+  Future<void> skiaPerfGcsAdapterIntegrationTest() async {
+    final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
+
+    final String testObjectName = await SkiaPerfGcsAdaptor.comptueObjectName(
+        kFlutterFrameworkRepo, kFrameworkRevision1);
+
+    await skiaPerfGcs.writePoints(testObjectName, <SkiaPerfPoint>[
+      SkiaPerfPoint.fromPoint(cocoonPointRev1Metric1),
+      SkiaPerfPoint.fromPoint(cocoonPointRev1Metric2),
+    ]);
+
+    final List<SkiaPerfPoint> points =
+        await skiaPerfGcs.readPoints(testObjectName);
+    expect(points.length, equals(2));
+    expectSetMatch(
+        points.map((SkiaPerfPoint p) => p.testName), <String>[kTaskName]);
+    expectSetMatch(points.map((SkiaPerfPoint p) => p.subResult),
+        <String>[kMetric1, kMetric2]);
+    expectSetMatch(
+        points.map((SkiaPerfPoint p) => p.value), <double>[kValue1, kValue2]);
+    expectSetMatch(points.map((SkiaPerfPoint p) => p.githubRepo),
+        <String>[kFlutterFrameworkRepo]);
+    expectSetMatch(points.map((SkiaPerfPoint p) => p.gitHash),
+        <String>[kFrameworkRevision1]);
+    for (int i = 0; i < 2; i += 1) {
+      expect(points[0].jsonUrl, startsWith('https://'));
+    }
+  }
+
+  Future<void> skiaPerfGcsIntegrationTestWithEnginePoints() async {
+    final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
+
+    final String testObjectName = await SkiaPerfGcsAdaptor.comptueObjectName(
+        kFlutterEngineRepo, engineRevision);
+
+    await skiaPerfGcs.writePoints(testObjectName, <SkiaPerfPoint>[
+      SkiaPerfPoint.fromPoint(enginePoint1),
+      SkiaPerfPoint.fromPoint(enginePoint2),
+    ]);
+
+    final List<SkiaPerfPoint> points =
+        await skiaPerfGcs.readPoints(testObjectName);
+    expect(points.length, equals(2));
+    expectSetMatch(
+      points.map((SkiaPerfPoint p) => p.testName),
+      <String>[engineMetricName, engineMetricName],
+    );
+    expectSetMatch(
+      points.map((SkiaPerfPoint p) => p.value),
+      <double>[engineValue1, engineValue2],
+    );
+    expectSetMatch(
+      points.map((SkiaPerfPoint p) => p.githubRepo),
+      <String>[kFlutterEngineRepo],
+    );
+    expectSetMatch(
+        points.map((SkiaPerfPoint p) => p.gitHash), <String>[engineRevision]);
+    for (int i = 0; i < 2; i += 1) {
+      expect(points[0].jsonUrl, startsWith('https://'));
+    }
+  }
+
+  // To run the following integration tests, there must be a valid Google Cloud
+  // Project service account credentials in secret/test_gcp_credentials.json so
+  // `testBucket` won't be null. Currently, these integration tests are skipped
+  // in the CI, and only verified locally.
+  test(
+    'SkiaPerfGcsAdaptor passes integration test with Google Cloud Storage',
+    skiaPerfGcsAdapterIntegrationTest,
+    skip: testBucket == null,
+  );
+
+  test(
+    'SkiaPerfGcsAdaptor integration test with engine points',
+    skiaPerfGcsIntegrationTestWithEnginePoints,
+    skip: testBucket == null,
+  );
+
+  test(
+    'SkiaPerfGcsAdaptor integration test for name computations',
+    () async {
+      expect(
+        await SkiaPerfGcsAdaptor.comptueObjectName(
+            kFlutterFrameworkRepo, kFrameworkRevision1),
+        equals(
+            'flutter-flutter/2019/12/04/23/$kFrameworkRevision1/values.json'),
+      );
+      expect(
+        await SkiaPerfGcsAdaptor.comptueObjectName(
+            kFlutterEngineRepo, kEngineRevision1),
+        equals('flutter-engine/2019/12/03/20/$kEngineRevision1/values.json'),
+      );
+      expect(
+        await SkiaPerfGcsAdaptor.comptueObjectName(
+            kFlutterEngineRepo, kEngineRevision2),
+        equals('flutter-engine/2020/01/03/15/$kEngineRevision2/values.json'),
+      );
+    },
+    skip: testBucket == null,
+  );
 }

dev/benchmarks/metrics_center/test/utility.dart

@@ -2,9 +2,22 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+import 'dart:convert';
+import 'dart:io';
+
 import 'common.dart';
 
 // This will be used in many of our unit tests.
 void expectSetMatch<T>(Iterable<T> actual, Iterable<T> expected) {
   expect(Set<T>.from(actual), equals(Set<T>.from(expected)));
 }
+
+// May return null if the credentials file doesn't exist.
+Map<String, dynamic> getTestGcpCredentialsJson() {
+  final File f = File('secret/test_gcp_credentials.json');
+  if (!f.existsSync()) {
+    return null;
+  }
+  return jsonDecode(File('secret/test_gcp_credentials.json').readAsStringSync())
+      as Map<String, dynamic>;
+}