Migrate the rest of the metrics center library (#73875)

Some notable changes are:
- Add SkiaPerfDestination
- Add LegacyFlutterDestination (for backup options during transitions).
- Add GcsLock

Related issue: https://github.com/flutter/flutter/issues/73872

dev/benchmarks/metrics_center/.gitignore

+secret

dev/benchmarks/metrics_center/LICENSE

+Copyright 2014 The Flutter Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+    * Redistributions of source code must retain the above copyright
+      notice, this list of conditions and the following disclaimer.
+    * Redistributions in binary form must reproduce the above
+      copyright notice, this list of conditions and the following
+      disclaimer in the documentation and/or other materials provided
+      with the distribution.
+    * Neither the name of Google Inc. nor the names of its
+      contributors may be used to endorse or promote products derived
+      from this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
\ No newline at end of file

dev/benchmarks/metrics_center/README.md

+Metrics center is a minimal set of code and services to support multiple perf
+metrics generators (e.g., Cocoon device lab, Cirrus bots, LUCI bots, Firebase
+Test Lab) and destinations (e.g., old Cocoon perf dashboard, Skia perf
+dashboard). The work and maintenance it requires is very close to that of just
+supporting a single generator and destination (e.g., engine bots to Skia perf),
+and the small amount of extra work is designed to make it easy to support more
+generators and destinations in the future.
+
+This is currently under migration. More documentations will be added once the
+migration is done.

dev/benchmarks/metrics_center/lib/metrics_center.dart

+// Copyright 2014 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+export 'src/common.dart';
+export 'src/flutter.dart';
+export 'src/google_benchmark.dart';
+export 'src/skiaperf.dart';

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

@@ -8,6 +8,10 @@ import 'dart:convert';
 import 'package:crypto/crypto.dart';
 import 'package:equatable/equatable.dart';
 
+import 'package:googleapis_auth/auth.dart';
+import 'package:googleapis_auth/auth_io.dart';
+import 'package:http/http.dart';
+
 /// Common format of a metric data point.
 class MetricPoint extends Equatable {
   MetricPoint(
@@ -42,6 +46,23 @@ class MetricPoint extends Equatable {
   List<Object> get props => <Object>[value, tags];
 }
 
+/// Interface to write [MetricPoint].
+abstract class MetricDestination {
+  /// Insert new data points or modify old ones with matching id.
+  Future<void> update(List<MetricPoint> points);
+}
+
+/// Create `AuthClient` in case we only have an access token without the full
+/// credentials json. It's currently the case for Chrmoium LUCI bots.
+AuthClient authClientFromAccessToken(String token, List<String> scopes) {
+  final DateTime anHourLater = DateTime.now().add(const Duration(hours: 1));
+  final AccessToken accessToken =
+      AccessToken('Bearer', token, anHourLater.toUtc());
+  final AccessCredentials accessCredentials =
+      AccessCredentials(accessToken, null, scopes);
+  return authenticatedClient(Client(), accessCredentials);
+}
+
 /// Some common tag keys
 const String kGithubRepoKey = 'gitRepo';
 const String kGitRevisionKey = 'gitRevision';
@@ -52,3 +73,6 @@ const String kSubResultKey = 'subResult';
 /// Known github repo
 const String kFlutterFrameworkRepo = 'flutter/flutter';
 const String kFlutterEngineRepo = 'flutter/engine';
+
+/// The key for the GCP project id in the credentials json.
+const String kProjectId = 'project_id';

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

@@ -3,6 +3,8 @@
 // found in the LICENSE file.
 
 import 'package:metrics_center/src/common.dart';
+import 'package:metrics_center/src/legacy_datastore.dart';
+import 'package:metrics_center/src/legacy_flutter.dart';
 
 /// Convenient class to capture the benchmarks in the Flutter engine repo.
 class FlutterEngineMetricPoint extends MetricPoint {
@@ -20,3 +22,33 @@ class FlutterEngineMetricPoint extends MetricPoint {
           }..addAll(moreTags),
         );
 }
+
+/// All Flutter performance metrics (framework, engine, ...) should be written
+/// to this destination.
+class FlutterDestination extends MetricDestination {
+  // TODO(liyuqian): change the implementation of this class (without changing
+  // its public APIs) to remove `LegacyFlutterDestination` and directly use
+  // `SkiaPerfDestination` once the migration is fully done.
+  FlutterDestination._(this._legacyDestination);
+
+  static Future<FlutterDestination> makeFromCredentialsJson(
+      Map<String, dynamic> json) async {
+    final LegacyFlutterDestination legacyDestination =
+        LegacyFlutterDestination(await datastoreFromCredentialsJson(json));
+    return FlutterDestination._(legacyDestination);
+  }
+
+  static FlutterDestination makeFromAccessToken(
+      String accessToken, String projectId) {
+    final LegacyFlutterDestination legacyDestination = LegacyFlutterDestination(
+        datastoreFromAccessToken(accessToken, projectId));
+    return FlutterDestination._(legacyDestination);
+  }
+
+  @override
+  Future<void> update(List<MetricPoint> points) async {
+    await _legacyDestination.update(points);
+  }
+
+  final LegacyFlutterDestination _legacyDestination;
+}

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

+// Copyright 2014 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'package:googleapis/storage/v1.dart';
+import 'package:googleapis_auth/auth.dart';
+
+/// Global (in terms of earth) mutex using Google Cloud Storage.
+class GcsLock {
+  /// Create a lock with an authenticated client and a GCS bucket name.
+  ///
+  /// The client is used to communicate with Google Cloud Storage APIs.
+  GcsLock(this._client, this._bucketName)
+      : assert(_client != null),
+        assert(_bucketName != null) {
+    _api = StorageApi(_client);
+  }
+
+  /// Create a temporary lock file in GCS, and use it as a mutex mechanism to
+  /// run a piece of code exclusively.
+  ///
+  /// There must be no existing lock file with the same name in order to
+  /// proceed. If multiple [GcsLock]s with the same `bucketName` and
+  /// `lockFileName` try [protectedRun] simultaneously, only one will proceed
+  /// and create the lock file. All others will be blocked.
+  ///
+  /// When [protectedRun] finishes, the lock file is deleted, and other blocked
+  /// [protectedRun] may proceed.
+  ///
+  /// If the lock file is stuck (e.g., `_unlock` is interrupted unexpectedly),
+  /// one may need to manually delete the lock file from GCS to unblock any
+  /// [protectedRun] that may depend on it.
+  Future<void> protectedRun(String lockFileName, Future<void> f()) async {
+    await _lock(lockFileName);
+    try {
+      await f();
+    } catch (e, stacktrace) {
+      print(stacktrace);
+      rethrow;
+    } finally {
+      await _unlock(lockFileName);
+    }
+  }
+
+  Future<void> _lock(String lockFileName) async {
+    final Object object = Object();
+    object.bucket = _bucketName;
+    object.name = lockFileName;
+    final Media content = Media(const Stream<List<int>>.empty(), 0);
+
+    Duration waitPeriod = const Duration(milliseconds: 10);
+    bool locked = false;
+    while (!locked) {
+      try {
+        await _api.objects.insert(object, _bucketName,
+            ifGenerationMatch: '0', uploadMedia: content);
+        locked = true;
+      } on DetailedApiRequestError catch (e) {
+        if (e.status == 412) {
+          // Status 412 means that the lock file already exists. Wait until
+          // that lock file is deleted.
+          await Future<void>.delayed(waitPeriod);
+          waitPeriod *= 2;
+          if (waitPeriod >= _kWarningThreshold) {
+            print(
+              'The lock is waiting for a long time: $waitPeriod. '
+              'If the lock file $lockFileName in bucket $_bucketName '
+              'seems to be stuck (i.e., it was created a long time ago and '
+              'no one seems to be owning it currently), delete it manually '
+              'to unblock this.',
+            );
+          }
+        } else {
+          rethrow;
+        }
+      }
+    }
+  }
+
+  Future<void> _unlock(String lockFileName) async {
+    await _api.objects.delete(_bucketName, lockFileName);
+  }
+
+  StorageApi _api;
+
+  final String _bucketName;
+  final AuthClient _client;
+
+  static const Duration _kWarningThreshold = Duration(seconds: 10);
+}

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

@@ -30,6 +30,6 @@ class GithubHelper {
 
   static final GithubHelper _singleton = GithubHelper._internal();
 
-  final GitHub _github = GitHub();
+  final GitHub _github = GitHub(auth: findAuthenticationFromEnvironment());
   final Map<String, DateTime> _commitDateTimeCache = <String, DateTime>{};
 }

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

+// Copyright 2014 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+// TODO(liyuqian): Remove this file once the migration is fully done and we no
+// longer need to fall back to the datastore.
+
+import 'package:gcloud/db.dart';
+import 'package:googleapis_auth/auth.dart';
+import 'package:googleapis_auth/auth_io.dart';
+
+// The official pub.dev/packages/gcloud documentation uses datastore_impl
+// so we have to ignore implementation_imports here.
+// ignore: implementation_imports
+import 'package:gcloud/src/datastore_impl.dart';
+
+import 'common.dart';
+
+Future<DatastoreDB> datastoreFromCredentialsJson(
+    Map<String, dynamic> json) async {
+  final AutoRefreshingAuthClient client = await clientViaServiceAccount(
+      ServiceAccountCredentials.fromJson(json), DatastoreImpl.SCOPES);
+  return DatastoreDB(DatastoreImpl(client, json[kProjectId] as String));
+}
+
+DatastoreDB datastoreFromAccessToken(String token, String projectId) {
+  final AuthClient client =
+      authClientFromAccessToken(token, DatastoreImpl.SCOPES);
+  return DatastoreDB(DatastoreImpl(client, projectId));
+}

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

+// Copyright 2014 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+// TODO(liyuqian): Remove this legacy file once the migration is fully done.
+// See go/flutter-metrics-center-migration for detailed plans.
+import 'dart:convert';
+import 'dart:math';
+
+import 'package:gcloud/db.dart';
+
+import 'common.dart';
+import 'legacy_datastore.dart';
+
+const String kSourceTimeMicrosName = 'sourceTimeMicros';
+
+// The size of 500 is currently limited by Google datastore. It cannot write
+// more than 500 entities in a single call.
+const int kMaxBatchSize = 500;
+
+/// This model corresponds to the existing data model 'MetricPoint' used in the
+/// flutter-cirrus GCP project.
+///
+/// The originId and sourceTimeMicros fields are no longer used but we are still
+/// providing valid values to them so it's compatible with old code and services
+/// during the migration.
+@Kind(name: 'MetricPoint', idType: IdType.String)
+class LegacyMetricPointModel extends Model<String> {
+  LegacyMetricPointModel({MetricPoint fromMetricPoint}) {
+    if (fromMetricPoint != null) {
+      id = fromMetricPoint.id;
+      value = fromMetricPoint.value;
+      originId = 'legacy-flutter';
+      sourceTimeMicros = null;
+      tags = fromMetricPoint.tags.keys
+          .map((String key) =>
+              jsonEncode(<String, dynamic>{key: fromMetricPoint.tags[key]}))
+          .toList();
+    }
+  }
+
+  @DoubleProperty(required: true, indexed: false)
+  double value;
+
+  @StringListProperty()
+  List<String> tags;
+
+  @StringProperty(required: true)
+  String originId;
+
+  @IntProperty(propertyName: kSourceTimeMicrosName)
+  int sourceTimeMicros;
+}
+
+class LegacyFlutterDestination extends MetricDestination {
+  LegacyFlutterDestination(this._db);
+
+  static Future<LegacyFlutterDestination> makeFromCredentialsJson(
+      Map<String, dynamic> json) async {
+    return LegacyFlutterDestination(await datastoreFromCredentialsJson(json));
+  }
+
+  static LegacyFlutterDestination makeFromAccessToken(
+      String accessToken, String projectId) {
+    return LegacyFlutterDestination(
+        datastoreFromAccessToken(accessToken, projectId));
+  }
+
+  @override
+  Future<void> update(List<MetricPoint> points) async {
+    final List<LegacyMetricPointModel> flutterCenterPoints =
+        points.map((MetricPoint p) => LegacyMetricPointModel(fromMetricPoint: p)).toList();
+
+    for (int start = 0; start < points.length; start += kMaxBatchSize) {
+      final int end = min(start + kMaxBatchSize, points.length);
+      await _db.withTransaction((Transaction tx) async {
+        tx.queueMutations(inserts: flutterCenterPoints.sublist(start, end));
+        await tx.commit();
+      });
+    }
+  }
+
+  final DatastoreDB _db;
+}

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

@@ -6,8 +6,11 @@ import 'dart:convert';
 
 import 'package:gcloud/storage.dart';
 import 'package:googleapis/storage/v1.dart' show DetailedApiRequestError;
+import 'package:googleapis_auth/auth.dart';
+import 'package:googleapis_auth/auth_io.dart';
 
 import 'package:metrics_center/src/common.dart';
+import 'package:metrics_center/src/gcs_lock.dart';
 import 'package:metrics_center/src/github_helper.dart';
 
 // Skia Perf Format is a JSON file that looks like:
@@ -75,7 +78,7 @@ class SkiaPerfPoint extends MetricPoint {
     final String name = p.tags[kNameKey];
 
     if (githubRepo == null || gitHash == null || name == null) {
-      throw '$kGithubRepoKey, $kGitRevisionKey, $kGitRevisionKey must be set in'
+      throw '$kGithubRepoKey, $kGitRevisionKey, $kNameKey must be set in'
           ' the tags of $p.';
     }
 
@@ -201,10 +204,28 @@ class SkiaPerfGcsAdaptor {
   ///
   /// The `objectName` must be a properly formatted string returned by
   /// [computeObjectName].
+  ///
+  /// The read may retry multiple times if transient network errors with code
+  /// 504 happens.
   Future<void> writePoints(
       String objectName, List<SkiaPerfPoint> points) async {
     final String jsonString = jsonEncode(SkiaPerfPoint.toSkiaPerfJson(points));
-    await _gcsBucket.writeBytes(objectName, utf8.encode(jsonString));
+    final List<int> content = utf8.encode(jsonString);
+
+    // Retry multiple times as GCS may return 504 timeout.
+    for (int retry = 0; retry < 5; retry += 1) {
+      try {
+        await _gcsBucket.writeBytes(objectName, content);
+        return;
+      } catch (e) {
+        if (e is DetailedApiRequestError && e.status == 504) {
+          continue;
+        }
+        rethrow;
+      }
+    }
+    // Retry one last time and let the exception go through.
+    await _gcsBucket.writeBytes(objectName, content);
   }
 
   /// Read a list of `SkiaPerfPoint` that have been previously written to the
@@ -290,7 +311,7 @@ class SkiaPerfGcsAdaptor {
   /// 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,
+  static Future<String> computeObjectName(String githubRepo, String revision,
       {GithubHelper githubHelper}) async {
     assert(_githubRepoToGcsName[githubRepo] != null);
     final String topComponent = _githubRepoToGcsName[githubRepo];
@@ -322,6 +343,95 @@ class SkiaPerfGcsAdaptor {
   final Bucket _gcsBucket;
 }
 
+class SkiaPerfDestination extends MetricDestination {
+  SkiaPerfDestination(this._gcs, this._lock);
+
+  static const String kBucketName = 'flutter-skia-perf-prod';
+  static const String kTestBucketName = 'flutter-skia-perf-test';
+
+  /// Create from a full credentials json (of a service account).
+  static Future<SkiaPerfDestination> makeFromGcpCredentials(
+      Map<String, dynamic> credentialsJson,
+      {bool isTesting = false}) async {
+    final AutoRefreshingAuthClient client = await clientViaServiceAccount(
+        ServiceAccountCredentials.fromJson(credentialsJson), Storage.SCOPES);
+    return make(
+      client,
+      credentialsJson[kProjectId] as String,
+      isTesting: isTesting,
+    );
+  }
+
+  /// Create from an access token and its project id.
+  static Future<SkiaPerfDestination> makeFromAccessToken(
+      String token, String projectId,
+      {bool isTesting = false}) async {
+    final AuthClient client = authClientFromAccessToken(token, Storage.SCOPES);
+    return make(client, projectId, isTesting: isTesting);
+  }
+
+  /// Create from an [AuthClient] and a GCP project id.
+  ///
+  /// [AuthClient] can be obtained from functions like `clientViaUserConsent`.
+  static Future<SkiaPerfDestination> make(AuthClient client, String projectId,
+      {bool isTesting = false}) async {
+    final Storage storage = Storage(client, projectId);
+    final String bucketName = isTesting ? kTestBucketName : kBucketName;
+    if (!await storage.bucketExists(bucketName)) {
+      throw 'Bucket $kBucketName does not exist.';
+    }
+    final SkiaPerfGcsAdaptor adaptor =
+        SkiaPerfGcsAdaptor(storage.bucket(bucketName));
+    final GcsLock lock = GcsLock(client, bucketName);
+    return SkiaPerfDestination(adaptor, lock);
+  }
+
+  @override
+  Future<void> update(List<MetricPoint> points) async {
+    // 1st, create a map based on git repo, git revision, and point id. Git repo
+    // and git revision are the top level components of the Skia perf GCS object
+    // name.
+    final Map<String, Map<String, Map<String, SkiaPerfPoint>>> pointMap =
+        <String, Map<String, Map<String, SkiaPerfPoint>>>{};
+    for (final SkiaPerfPoint p
+        in points.map((MetricPoint x) => SkiaPerfPoint.fromPoint(x))) {
+      if (p != null) {
+        pointMap[p.githubRepo] ??= <String, Map<String, SkiaPerfPoint>>{};
+        pointMap[p.githubRepo][p.gitHash] ??= <String, SkiaPerfPoint>{};
+        pointMap[p.githubRepo][p.gitHash][p.id] = p;
+      }
+    }
+
+    // 2nd, read existing points from the gcs object and update with new ones.
+    for (final String repo in pointMap.keys) {
+      for (final String revision in pointMap[repo].keys) {
+        final String objectName =
+            await SkiaPerfGcsAdaptor.computeObjectName(repo, revision);
+        final Map<String, SkiaPerfPoint> newPoints = pointMap[repo][revision];
+        // If too many bots are writing the metrics of a git revision into this
+        // single json file (with name `objectName`), the contention on the lock
+        // might be too high. In that case, break the json file into multiple
+        // json files according to bot names or task names. Skia perf read all
+        // json files in the directory so one can use arbitrary names for those
+        // sharded json file names.
+        _lock.protectedRun('$objectName.lock', () async {
+          final List<SkiaPerfPoint> oldPoints =
+              await _gcs.readPoints(objectName);
+          for (final SkiaPerfPoint p in oldPoints) {
+            if (newPoints[p.id] == null) {
+              newPoints[p.id] = p;
+            }
+          }
+          await _gcs.writePoints(objectName, newPoints.values.toList());
+        });
+      }
+    }
+  }
+
+  final SkiaPerfGcsAdaptor _gcs;
+  final GcsLock _lock;
+}
+
 const String kSkiaPerfGitHashKey = 'gitHash';
 const String kSkiaPerfResultsKey = 'results';
 const String kSkiaPerfValueKey = 'value';

dev/benchmarks/metrics_center/pubspec.yaml

 name: metrics_center
+version: 0.0.4
+description:
+  Support multiple performance metrics sources/formats and destinations.
+homepage:
+  https://github.com/flutter/flutter/tree/master/dev/benchmarks/metrics_center
 
 environment:
   sdk: '>=2.10.0 <3.0.0'
@@ -10,7 +15,6 @@ dependencies:
   googleapis_auth: 0.2.12
   github: 7.0.4
   equatable: 1.2.5
-  mockito: 4.1.1
 
   _discoveryapis_commons: 0.2.0 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   async: 2.5.0-nullsafety.3 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
@@ -36,6 +40,8 @@ dependencies:
 dev_dependencies:
   test: 1.16.0-nullsafety.9
   pedantic: 1.10.0-nullsafety.3
+  mockito: 4.1.1
+  fake_async: 1.2.0-nullsafety.3
 
   _fe_analyzer_shared: 12.0.0 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
   analyzer: 0.40.6 # THIS LINE IS AUTOGENERATED - TO UPDATE USE "flutter update-packages --force-upgrade"
@@ -66,4 +72,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: d5aa
+# PUBSPEC CHECKSUM: 25e6

dev/benchmarks/metrics_center/test/flutter_test.dart

+// Copyright 2014 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'package:metrics_center/src/common.dart';
+import 'package:metrics_center/src/flutter.dart';
+
+import 'common.dart';
+
+void main() {
+  test('FlutterEngineMetricPoint works.', () {
+    const String gitRevision = 'ca799fa8b2254d09664b78ee80c43b434788d112';
+    final FlutterEngineMetricPoint simplePoint = FlutterEngineMetricPoint(
+      'BM_ParagraphLongLayout',
+      287235,
+      gitRevision,
+    );
+    expect(simplePoint.value, equals(287235));
+    expect(simplePoint.tags[kGithubRepoKey], kFlutterEngineRepo);
+    expect(simplePoint.tags[kGitRevisionKey], gitRevision);
+    expect(simplePoint.tags[kNameKey], 'BM_ParagraphLongLayout');
+
+    final FlutterEngineMetricPoint detailedPoint = FlutterEngineMetricPoint(
+      'BM_ParagraphLongLayout',
+      287224,
+      'ca799fa8b2254d09664b78ee80c43b434788d112',
+      moreTags: const <String, String>{
+        'executable': 'txt_benchmarks',
+        'sub_result': 'CPU',
+        kUnitKey: 'ns',
+      },
+    );
+    expect(detailedPoint.value, equals(287224));
+    expect(detailedPoint.tags['executable'], equals('txt_benchmarks'));
+    expect(detailedPoint.tags['sub_result'], equals('CPU'));
+    expect(detailedPoint.tags[kUnitKey], equals('ns'));
+  });
+}
\ No newline at end of file

dev/benchmarks/metrics_center/test/gcs_lock_test.dart

+// Copyright 2014 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:googleapis/storage/v1.dart';
+import 'package:fake_async/fake_async.dart';
+import 'package:gcloud/storage.dart';
+import 'package:googleapis_auth/auth_io.dart';
+import 'package:metrics_center/src/gcs_lock.dart';
+import 'package:metrics_center/src/skiaperf.dart';
+import 'package:mockito/mockito.dart';
+
+import 'common.dart';
+import 'utility.dart';
+
+enum TestPhase {
+  run1,
+  run2,
+}
+
+class MockClient extends Mock implements AuthClient {}
+
+void main() {
+  const Duration kDelayStep = Duration(milliseconds: 10);
+  final Map<String, dynamic> credentialsJson = getTestGcpCredentialsJson();
+
+  test('GcsLock prints warnings for long waits', () {
+    // Capture print to verify error messages.
+    final List<String> prints = <String>[];
+    final ZoneSpecification spec =
+        ZoneSpecification(print: (_, __, ___, String msg) => prints.add(msg));
+
+    Zone.current.fork(specification: spec).run<void>(() {
+      fakeAsync((FakeAsync fakeAsync) {
+        final MockClient mockClient = MockClient();
+        final GcsLock lock = GcsLock(mockClient, 'mockBucket');
+        when(mockClient.send(any)).thenThrow(DetailedApiRequestError(412, ''));
+        final Future<void> runFinished =
+            lock.protectedRun('mock.lock', () async {});
+        fakeAsync.elapse(const Duration(seconds: 10));
+        when(mockClient.send(any)).thenThrow(AssertionError('Stop!'));
+        runFinished.catchError((dynamic e) {
+          final AssertionError error = e as AssertionError;
+          expect(error.message, 'Stop!');
+          print('${error.message}');
+        });
+        fakeAsync.elapse(const Duration(seconds: 20));
+      });
+    });
+
+    const String kExpectedErrorMessage = 'The lock is waiting for a long time: '
+        '0:00:10.240000. If the lock file mock.lock in bucket mockBucket '
+        'seems to be stuck (i.e., it was created a long time ago and no one '
+        'seems to be owning it currently), delete it manually to unblock this.';
+    expect(prints, equals(<String>[kExpectedErrorMessage, 'Stop!']));
+  });
+
+  test('GcsLock integration test: single protectedRun is successful', () async {
+    final AutoRefreshingAuthClient client = await clientViaServiceAccount(
+        ServiceAccountCredentials.fromJson(credentialsJson), Storage.SCOPES);
+    final GcsLock lock = GcsLock(client, SkiaPerfDestination.kTestBucketName);
+    int testValue = 0;
+    await lock.protectedRun('test.lock', () async {
+      testValue = 1;
+    });
+    expect(testValue, 1);
+  }, skip: credentialsJson == null);
+
+  test('GcsLock integration test: protectedRun is exclusive', () async {
+    final AutoRefreshingAuthClient client = await clientViaServiceAccount(
+        ServiceAccountCredentials.fromJson(credentialsJson), Storage.SCOPES);
+    final GcsLock lock1 = GcsLock(client, SkiaPerfDestination.kTestBucketName);
+    final GcsLock lock2 = GcsLock(client, SkiaPerfDestination.kTestBucketName);
+
+    TestPhase phase = TestPhase.run1;
+    final Completer<void> started1 = Completer<void>();
+    final Future<void> finished1 = lock1.protectedRun('test.lock', () async {
+      started1.complete();
+      while (phase == TestPhase.run1) {
+        await Future<void>.delayed(kDelayStep);
+      }
+    });
+
+    await started1.future;
+
+    final Completer<void> started2 = Completer<void>();
+    final Future<void> finished2 = lock2.protectedRun('test.lock', () async {
+      started2.complete();
+    });
+
+    // started2 should not be set even after a long wait because lock1 is
+    // holding the GCS lock file.
+    await Future<void>.delayed(kDelayStep * 10);
+    expect(started2.isCompleted, false);
+
+    // When phase is switched to run2, lock1 should be released soon and
+    // lock2 should soon be able to proceed its protectedRun.
+    phase = TestPhase.run2;
+    await started2.future;
+    await finished1;
+    await finished2;
+  }, skip: credentialsJson == null);
+}

dev/benchmarks/metrics_center/test/google_benchmark_test.dart

@@ -3,7 +3,7 @@
 // found in the LICENSE file.
 
 import 'package:metrics_center/src/common.dart';
-import 'package:metrics_center/google_benchmark.dart';
+import 'package:metrics_center/src/google_benchmark.dart';
 
 import 'common.dart';
 import 'utility.dart';

dev/benchmarks/metrics_center/test/legacy_flutter_test.dart

+// Copyright 2014 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'package:gcloud/src/datastore_impl.dart';
+import 'package:googleapis_auth/auth_io.dart';
+import 'package:metrics_center/src/common.dart';
+import 'package:metrics_center/src/legacy_flutter.dart';
+
+import 'common.dart';
+import 'utility.dart';
+
+const String kTestSourceId = 'test';
+
+void main() {
+  final Map<String, dynamic> credentialsJson = getTestGcpCredentialsJson();
+  test(
+      'LegacyFlutterDestination integration test: '
+      'update does not crash.', () async {
+    final LegacyFlutterDestination dst =
+        await LegacyFlutterDestination.makeFromCredentialsJson(credentialsJson);
+    await dst.update(<MetricPoint>[MetricPoint(1.0, const <String, String>{})]);
+  }, skip: credentialsJson == null);
+
+  test(
+      'LegacyFlutterDestination integration test: '
+      'can update with an access token.', () async {
+    final AutoRefreshingAuthClient client = await clientViaServiceAccount(
+      ServiceAccountCredentials.fromJson(credentialsJson),
+      DatastoreImpl.SCOPES,
+    );
+    final String token = client.credentials.accessToken.data;
+    final LegacyFlutterDestination dst =
+        LegacyFlutterDestination.makeFromAccessToken(
+      token,
+      credentialsJson[kProjectId] as String,
+    );
+    await dst.update(<MetricPoint>[MetricPoint(1.0, const <String, String>{})]);
+  }, skip: credentialsJson == null);
+}

dev/benchmarks/metrics_center/test/skiaperf_test.dart

@@ -9,6 +9,7 @@ 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/gcs_lock.dart';
 import 'package:metrics_center/src/github_helper.dart';
 import 'package:mockito/mockito.dart';
 
@@ -25,11 +26,38 @@ class MockObjectInfo extends Mock implements ObjectInfo {}
 
 class MockGithubHelper extends Mock implements GithubHelper {}
 
+class MockGcsLock implements GcsLock {
+  @override
+  Future<void> protectedRun(
+      String exclusiveObjectName, Future<void> Function() f) async {
+    await f();
+  }
+}
+
+class MockSkiaPerfGcsAdaptor implements SkiaPerfGcsAdaptor {
+  @override
+  Future<List<SkiaPerfPoint>> readPoints(String objectName) async {
+    return _storage[objectName] ?? <SkiaPerfPoint>[];
+  }
+
+  @override
+  Future<void> writePoints(
+      String objectName, List<SkiaPerfPoint> points) async {
+    _storage[objectName] = points.toList();
+  }
+
+  // Map from the object name to the list of SkiaPoint that mocks the GCS.
+  final Map<String, List<SkiaPerfPoint>> _storage =
+      <String, List<SkiaPerfPoint>>{};
+}
+
 Future<void> main() async {
   const double kValue1 = 1.0;
   const double kValue2 = 2.0;
+  const double kValue3 = 3.0;
 
   const String kFrameworkRevision1 = '9011cece2595447eea5dd91adaa241c1c9ef9a33';
+  const String kFrameworkRevision2 = '372fe290e4d4f3f97cbf02a57d235771a9412f10';
   const String kEngineRevision1 = '617938024315e205f26ed72ff0f0647775fa6a71';
   const String kEngineRevision2 = '5858519139c22484aaff1cf5b26bdf7951259344';
   const String kTaskName = 'analyzer_benchmark';
@@ -58,6 +86,17 @@ Future<void> main() async {
     },
   );
 
+  final MetricPoint cocoonPointRev2Metric1 = MetricPoint(
+    kValue3,
+    const <String, String>{
+      kGithubRepoKey: kFlutterFrameworkRepo,
+      kGitRevisionKey: kFrameworkRevision2,
+      kNameKey: kTaskName,
+      kSubResultKey: kMetric1,
+      kUnitKey: 's',
+    },
+  );
+
   final MetricPoint cocoonPointBetaRev1Metric1 = MetricPoint(
     kValue1,
     const <String, String>{
@@ -284,7 +323,7 @@ Future<void> main() async {
             kFlutterFrameworkRepo, kFrameworkRevision1))
         .thenAnswer((_) => Future<DateTime>.value(DateTime(2019, 12, 4, 23)));
     expect(
-      await SkiaPerfGcsAdaptor.comptueObjectName(
+      await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterFrameworkRepo,
         kFrameworkRevision1,
         githubHelper: mockHelper,
@@ -294,7 +333,7 @@ Future<void> main() async {
     when(mockHelper.getCommitDateTime(kFlutterEngineRepo, kEngineRevision1))
         .thenAnswer((_) => Future<DateTime>.value(DateTime(2019, 12, 3, 20)));
     expect(
-      await SkiaPerfGcsAdaptor.comptueObjectName(
+      await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterEngineRepo,
         kEngineRevision1,
         githubHelper: mockHelper,
@@ -304,7 +343,7 @@ Future<void> main() async {
     when(mockHelper.getCommitDateTime(kFlutterEngineRepo, kEngineRevision2))
         .thenAnswer((_) => Future<DateTime>.value(DateTime(2020, 1, 3, 15)));
     expect(
-      await SkiaPerfGcsAdaptor.comptueObjectName(
+      await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterEngineRepo,
         kEngineRevision2,
         githubHelper: mockHelper,
@@ -317,7 +356,7 @@ Future<void> main() async {
     final MockBucket testBucket = MockBucket();
     final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
 
-    final String testObjectName = await SkiaPerfGcsAdaptor.comptueObjectName(
+    final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterFrameworkRepo, kFrameworkRevision1);
 
     final List<SkiaPerfPoint> writePoints = <SkiaPerfPoint>[
@@ -354,7 +393,7 @@ Future<void> main() async {
   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(
+    final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterFrameworkRepo, kFrameworkRevision1);
     when(testBucket.info(testObjectName))
         .thenThrow(Exception('No such object'));
@@ -363,6 +402,7 @@ Future<void> main() async {
 
   // The following is for integration tests.
   Bucket testBucket;
+  GcsLock testLock;
   final Map<String, dynamic> credentialsJson = getTestGcpCredentialsJson();
   if (credentialsJson != null) {
     final ServiceAccountCredentials credentials =
@@ -377,12 +417,13 @@ Future<void> main() async {
 
     assert(await storage.bucketExists(kTestBucketName));
     testBucket = storage.bucket(kTestBucketName);
+    testLock = GcsLock(client, kTestBucketName);
   }
 
   Future<void> skiaPerfGcsAdapterIntegrationTest() async {
     final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
 
-    final String testObjectName = await SkiaPerfGcsAdaptor.comptueObjectName(
+    final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterFrameworkRepo, kFrameworkRevision1);
 
     await skiaPerfGcs.writePoints(testObjectName, <SkiaPerfPoint>[
@@ -411,7 +452,7 @@ Future<void> main() async {
   Future<void> skiaPerfGcsIntegrationTestWithEnginePoints() async {
     final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
 
-    final String testObjectName = await SkiaPerfGcsAdaptor.comptueObjectName(
+    final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterEngineRepo, engineRevision);
 
     await skiaPerfGcs.writePoints(testObjectName, <SkiaPerfPoint>[
@@ -457,26 +498,77 @@ Future<void> main() async {
     skip: testBucket == null,
   );
 
+  // `SkiaPerfGcsAdaptor.computeObjectName` uses `GithubHelper` which requires
+  // network connections. Hence we put them as integration tests instead of unit
+  // tests.
   test(
     'SkiaPerfGcsAdaptor integration test for name computations',
     () async {
       expect(
-        await SkiaPerfGcsAdaptor.comptueObjectName(
+        await SkiaPerfGcsAdaptor.computeObjectName(
             kFlutterFrameworkRepo, kFrameworkRevision1),
         equals(
             'flutter-flutter/2019/12/04/23/$kFrameworkRevision1/values.json'),
       );
       expect(
-        await SkiaPerfGcsAdaptor.comptueObjectName(
+        await SkiaPerfGcsAdaptor.computeObjectName(
             kFlutterEngineRepo, kEngineRevision1),
         equals('flutter-engine/2019/12/03/20/$kEngineRevision1/values.json'),
       );
       expect(
-        await SkiaPerfGcsAdaptor.comptueObjectName(
+        await SkiaPerfGcsAdaptor.computeObjectName(
             kFlutterEngineRepo, kEngineRevision2),
         equals('flutter-engine/2020/01/03/15/$kEngineRevision2/values.json'),
       );
     },
     skip: testBucket == null,
   );
+
+  test('SkiaPerfDestination correctly updates points', () async {
+    final SkiaPerfGcsAdaptor mockGcs = MockSkiaPerfGcsAdaptor();
+    final GcsLock mockLock = MockGcsLock();
+    final SkiaPerfDestination dst = SkiaPerfDestination(mockGcs, mockLock);
+    await dst.update(<MetricPoint>[cocoonPointRev1Metric1]);
+    await dst.update(<MetricPoint>[cocoonPointRev1Metric2]);
+    List<SkiaPerfPoint> points = await mockGcs.readPoints(
+        await SkiaPerfGcsAdaptor.computeObjectName(
+            kFlutterFrameworkRepo, kFrameworkRevision1));
+    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]);
+
+    final MetricPoint updated =
+        MetricPoint(kValue3, cocoonPointRev1Metric1.tags);
+
+    await dst.update(<MetricPoint>[updated, cocoonPointRev2Metric1]);
+
+    points = await mockGcs.readPoints(
+        await SkiaPerfGcsAdaptor.computeObjectName(
+            kFlutterFrameworkRepo, kFrameworkRevision2));
+    expect(points.length, equals(1));
+    expect(points[0].gitHash, equals(kFrameworkRevision2));
+    expect(points[0].value, equals(kValue3));
+
+    points = await mockGcs.readPoints(
+        await SkiaPerfGcsAdaptor.computeObjectName(
+            kFlutterFrameworkRepo, kFrameworkRevision1));
+    expectSetMatch(
+        points.map((SkiaPerfPoint p) => p.value), <double>[kValue2, kValue3]);
+  });
+
+  Future<void> skiaPerfDestinationIntegrationTest() async {
+    final SkiaPerfDestination destination =
+        SkiaPerfDestination(SkiaPerfGcsAdaptor(testBucket), testLock);
+    await destination.update(<MetricPoint>[cocoonPointRev1Metric1]);
+  }
+
+  test(
+    'SkiaPerfDestination integration test',
+    skiaPerfDestinationIntegrationTest,
+    skip: testBucket == null,
+  );
 }