Scale gcs metrics center write with task name

packages/metrics_center/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 1.0.1
+
+- `update` now requires taskName to scale metric writes
+
 # 1.0.0
 
 - Null safety support

packages/metrics_center/lib/src/common.dart

@@ -49,7 +49,8 @@ class MetricPoint extends Equatable {
 /// Interface to write [MetricPoint].
 abstract class MetricDestination {
   /// Insert new data points or modify old ones with matching id.
-  Future<void> update(List<MetricPoint> points, DateTime commitTime);
+  Future<void> update(
+      List<MetricPoint> points, DateTime commitTime, String taskName);
 }
 
 /// Create `AuthClient` in case we only have an access token without the full

packages/metrics_center/lib/src/flutter.dart

@@ -52,8 +52,9 @@ class FlutterDestination extends MetricDestination {
   }
 
   @override
-  Future<void> update(List<MetricPoint> points, DateTime commitTime) async {
-    await _skiaPerfDestination.update(points, commitTime);
+  Future<void> update(
+      List<MetricPoint> points, DateTime commitTime, String taskName) async {
+    await _skiaPerfDestination.update(points, commitTime, taskName);
   }
 
   final SkiaPerfDestination _skiaPerfDestination;

packages/metrics_center/lib/src/skiaperf.dart

@@ -312,12 +312,12 @@ class SkiaPerfGcsAdaptor {
   /// (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> computeObjectName(
-      String githubRepo, String? revision, DateTime commitTime) async {
+  /// of the json file can be arbitrary, and multiple json files can be put
+  /// in that leaf directory. We are using multiple json files divided by test
+  /// names to scale up the system to avoid too many writes competing for
+  /// the same json file.
+  static Future<String> computeObjectName(String githubRepo, String? revision,
+      DateTime commitTime, String taskName) async {
     assert(_githubRepoToGcsName[githubRepo] != null);
     final String? topComponent = _githubRepoToGcsName[githubRepo];
     // [commitTime] is not guranteed to be UTC. Ensure it is so all results
@@ -327,7 +327,7 @@ class SkiaPerfGcsAdaptor {
     final String day = commitUtcTime.day.toString().padLeft(2, '0');
     final String hour = commitUtcTime.hour.toString().padLeft(2, '0');
     final String dateComponents = '${commitUtcTime.year}/$month/$day/$hour';
-    return '$topComponent/$dateComponents/$revision/values.json';
+    return '$topComponent/$dateComponents/$revision/${taskName}_values.json';
   }
 
   static final Map<String, String> _githubRepoToGcsName = <String, String>{
@@ -392,7 +392,8 @@ class SkiaPerfDestination extends MetricDestination {
   }
 
   @override
-  Future<void> update(List<MetricPoint> points, DateTime commitTime) async {
+  Future<void> update(
+      List<MetricPoint> points, DateTime commitTime, String taskName) 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.
@@ -411,14 +412,13 @@ class SkiaPerfDestination extends MetricDestination {
     for (final String repo in pointMap.keys) {
       for (final String? revision in pointMap[repo]!.keys) {
         final String objectName = await SkiaPerfGcsAdaptor.computeObjectName(
-            repo, revision, commitTime);
+            repo, revision, commitTime, taskName);
         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.
+        // Too many bots writing the metrics of a git revision into a single json
+        // file will cause high contention on the lock. We use multiple
+        // json files according to 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);

packages/metrics_center/pubspec.yaml

@@ -1,5 +1,5 @@
 name: metrics_center
-version: 1.0.0
+version: 1.0.1
 description:
   Support multiple performance metrics sources/formats and destinations.
 repository: https://github.com/flutter/packages/tree/master/packages/metrics_center

packages/metrics_center/test/flutter_test.dart

@@ -45,6 +45,6 @@ void main() {
         await FlutterDestination.makeFromCredentialsJson(credentialsJson!,
             isTesting: true);
     dst.update(<FlutterEngineMetricPoint>[simplePoint],
-        DateTime.fromMillisecondsSinceEpoch(123));
+        DateTime.fromMillisecondsSinceEpoch(123), 'test');
   }, skip: credentialsJson == null);
 }

packages/metrics_center/test/skiaperf_test.dart

@@ -317,24 +317,28 @@ Future<void> main() async {
         kFlutterFrameworkRepo,
         kFrameworkRevision1,
         DateTime.utc(2019, 12, 04, 23),
+        'test',
       ),
-      equals('flutter-flutter/2019/12/04/23/$kFrameworkRevision1/values.json'),
+      equals(
+          'flutter-flutter/2019/12/04/23/$kFrameworkRevision1/test_values.json'),
     );
     expect(
       await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterEngineRepo,
         kEngineRevision1,
         DateTime.utc(2019, 12, 03, 20),
+        'test',
       ),
-      equals('flutter-engine/2019/12/03/20/$kEngineRevision1/values.json'),
+      equals('flutter-engine/2019/12/03/20/$kEngineRevision1/test_values.json'),
     );
     expect(
       await SkiaPerfGcsAdaptor.computeObjectName(
         kFlutterEngineRepo,
         kEngineRevision2,
         DateTime.utc(2020, 01, 03, 15),
+        'test',
       ),
-      equals('flutter-engine/2020/01/03/15/$kEngineRevision2/values.json'),
+      equals('flutter-engine/2020/01/03/15/$kEngineRevision2/test_values.json'),
     );
   });
 
@@ -343,9 +347,11 @@ Future<void> main() async {
     final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
 
     final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
-        kFlutterFrameworkRepo,
-        kFrameworkRevision1,
-        DateTime.fromMillisecondsSinceEpoch(123));
+      kFlutterFrameworkRepo,
+      kFrameworkRevision1,
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
 
     final List<SkiaPerfPoint> writePoints = <SkiaPerfPoint>[
       SkiaPerfPoint.fromPoint(cocoonPointRev1Metric1),
@@ -384,9 +390,11 @@ Future<void> main() async {
     final MockBucket testBucket = MockBucket();
     final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket);
     final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
-        kFlutterFrameworkRepo,
-        kFrameworkRevision1,
-        DateTime.fromMillisecondsSinceEpoch(123));
+      kFlutterFrameworkRepo,
+      kFrameworkRevision1,
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
     when(testBucket.info(testObjectName))
         .thenThrow(Exception('No such object'));
     expect((await skiaPerfGcs.readPoints(testObjectName)).length, 0);
@@ -416,9 +424,11 @@ Future<void> main() async {
     final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket!);
 
     final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
-        kFlutterFrameworkRepo,
-        kFrameworkRevision1,
-        DateTime.fromMillisecondsSinceEpoch(123));
+      kFlutterFrameworkRepo,
+      kFrameworkRevision1,
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
 
     await skiaPerfGcs.writePoints(testObjectName, <SkiaPerfPoint>[
       SkiaPerfPoint.fromPoint(cocoonPointRev1Metric1),
@@ -447,9 +457,11 @@ Future<void> main() async {
     final SkiaPerfGcsAdaptor skiaPerfGcs = SkiaPerfGcsAdaptor(testBucket!);
 
     final String testObjectName = await SkiaPerfGcsAdaptor.computeObjectName(
-        kFlutterEngineRepo,
-        engineRevision,
-        DateTime.fromMillisecondsSinceEpoch(123));
+      kFlutterEngineRepo,
+      engineRevision,
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
 
     await skiaPerfGcs.writePoints(testObjectName, <SkiaPerfPoint>[
       SkiaPerfPoint.fromPoint(enginePoint1),
@@ -505,25 +517,30 @@ Future<void> main() async {
           kFlutterFrameworkRepo,
           kFrameworkRevision1,
           DateTime.utc(2019, 12, 04, 23),
+          'test',
         ),
         equals(
-            'flutter-flutter/2019/12/04/23/$kFrameworkRevision1/values.json'),
+            'flutter-flutter/2019/12/04/23/$kFrameworkRevision1/test_values.json'),
       );
       expect(
         await SkiaPerfGcsAdaptor.computeObjectName(
           kFlutterEngineRepo,
           kEngineRevision1,
           DateTime.utc(2019, 12, 03, 20),
+          'test',
         ),
-        equals('flutter-engine/2019/12/03/20/$kEngineRevision1/values.json'),
+        equals(
+            'flutter-engine/2019/12/03/20/$kEngineRevision1/test_values.json'),
       );
       expect(
         await SkiaPerfGcsAdaptor.computeObjectName(
           kFlutterEngineRepo,
           kEngineRevision2,
           DateTime.utc(2020, 01, 03, 15),
+          'test',
         ),
-        equals('flutter-engine/2020/01/03/15/$kEngineRevision2/values.json'),
+        equals(
+            'flutter-engine/2020/01/03/15/$kEngineRevision2/test_values.json'),
       );
     },
     skip: testBucket == null,
@@ -533,13 +550,23 @@ Future<void> main() async {
     final SkiaPerfGcsAdaptor mockGcs = MockSkiaPerfGcsAdaptor();
     final GcsLock mockLock = MockGcsLock();
     final SkiaPerfDestination dst = SkiaPerfDestination(mockGcs, mockLock);
-    await dst.update(<MetricPoint>[cocoonPointRev1Metric1],
-        DateTime.fromMillisecondsSinceEpoch(123));
-    await dst.update(<MetricPoint>[cocoonPointRev1Metric2],
-        DateTime.fromMillisecondsSinceEpoch(123));
-    List<SkiaPerfPoint> points = await mockGcs.readPoints(
-        await SkiaPerfGcsAdaptor.computeObjectName(kFlutterFrameworkRepo,
-            kFrameworkRevision1, DateTime.fromMillisecondsSinceEpoch(123)));
+    await dst.update(
+      <MetricPoint>[cocoonPointRev1Metric1],
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
+    await dst.update(
+      <MetricPoint>[cocoonPointRev1Metric2],
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
+    List<SkiaPerfPoint> points =
+        await mockGcs.readPoints(await SkiaPerfGcsAdaptor.computeObjectName(
+      kFlutterFrameworkRepo,
+      kFrameworkRevision1,
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    ));
     expect(points.length, equals(2));
     expectSetMatch(
         points.map((SkiaPerfPoint p) => p.testName), <String>[kTaskName]);
@@ -551,28 +578,42 @@ Future<void> main() async {
     final MetricPoint updated =
         MetricPoint(kValue3, cocoonPointRev1Metric1.tags);
 
-    await dst.update(<MetricPoint>[updated, cocoonPointRev2Metric1],
-        DateTime.fromMillisecondsSinceEpoch(123));
+    await dst.update(
+      <MetricPoint>[updated, cocoonPointRev2Metric1],
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
 
-    points = await mockGcs.readPoints(
-        await SkiaPerfGcsAdaptor.computeObjectName(kFlutterFrameworkRepo,
-            kFrameworkRevision2, DateTime.fromMillisecondsSinceEpoch(123)));
+    points =
+        await mockGcs.readPoints(await SkiaPerfGcsAdaptor.computeObjectName(
+      kFlutterFrameworkRepo,
+      kFrameworkRevision2,
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    ));
     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, DateTime.fromMillisecondsSinceEpoch(123)));
+    points =
+        await mockGcs.readPoints(await SkiaPerfGcsAdaptor.computeObjectName(
+      kFlutterFrameworkRepo,
+      kFrameworkRevision1,
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    ));
     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],
-        DateTime.fromMillisecondsSinceEpoch(123));
+    await destination.update(
+      <MetricPoint>[cocoonPointRev1Metric1],
+      DateTime.fromMillisecondsSinceEpoch(123),
+      'test',
+    );
   }
 
   test(