blob: 0ad3df2b34e77a072ccbea63635c1185c0485632 [file] [log] [blame] [edit]
// Copyright (c) 2019, the Dart project authors. Please see the AUTHORS file
// for details. All rights reserved. Use of this source code is governed by a
// BSD-style license that can be found in the LICENSE file.
// Post results from Dart tryjobs and CI builders to Cloud Pub/Sub.
//
// Reads a results.json input file, sends only the changed results from
// that file to the Pub/Sub channel 'results' in the 'dart-ci' project.
// Multiple messages are sent if there are more than 100 changed results,
// so the cloud function only needs to process 100 records within its time
// limit of 60s. Because of this, we never approach the limit of 10 MB
// base64-encoded data bytes per message.
import 'dart:convert';
import 'dart:io';
import 'package:args/args.dart';
import 'package:http/http.dart' as http;
void usage(ArgParser parser, {exitCode = 0}) {
print('''
Usage: post_results_to_pubsub.dart [OPTIONS]
Posts Dart CI results as messages to Google Cloud Pub/Sub
The options are as follows:
${parser.usage}''');
exit(exitCode);
}
const resultsPerMessage = 100;
String getPostUrl(String project) {
return 'https://pubsub.googleapis.com/v1/projects/$project'
'/topics/results:publish';
}
main(List<String> args) async {
final parser = new ArgParser();
parser.addFlag('help', help: 'Show the program usage.', negatable: false);
parser.addOption('auth_token',
abbr: 'a',
help: 'Authorization token with a scope including pubsub publish.');
parser.addOption('result_file',
abbr: 'f', help: 'File containing the results to send');
parser.addOption('id', abbr: 'i', help: 'Buildbucket ID of this build');
parser.addOption('base_revision', help: 'A try build\'s patch base');
parser.addFlag('staging',
abbr: 's', help: 'Publish to the staging system', defaultsTo: false);
final options = parser.parse(args);
if (options['help']) {
usage(parser);
}
if (options['result_file'] == null) {
print('Error: option "result_file" is required.\n');
usage(parser, exitCode: 1);
}
if (options['auth_token'] == null) {
print('Error: option "auth_token" is required.\n');
usage(parser, exitCode: 1);
}
final project = options['staging'] ? "dart-ci-staging" : "dart-ci";
final client = http.Client();
final lines = await File(options['result_file']).readAsLines();
final token = await File(options['auth_token']).readAsString();
final buildbucketID = options['id'];
final baseRevision = options['base_revision'];
if (lines.isEmpty) {
print('No results in input file');
return;
}
// TODO(karlklose): parse and validate data before sending it.
final changedPattern = '"changed":true';
List<String> changedResults =
lines.where((change) => change.contains(changedPattern)).toList();
// We need to send at least one result, to save build metadata to Firestore.
// Send an unchanged result - the cloud function filters these out.
if (changedResults.isEmpty) changedResults = lines.sublist(0, 1);
final chunks = <List<String>>[];
var position = 0;
final lastFullChunkStart = changedResults.length - resultsPerMessage;
while (position <= lastFullChunkStart) {
chunks.add(changedResults.sublist(position, position += resultsPerMessage));
}
if (position < changedResults.length)
chunks.add(changedResults.sublist(position));
// Send pubsub messages.
for (final chunk in chunks) {
// Space messages out to reduce scaling problems
const chunkDelay = Duration(seconds: 2);
if (chunk != chunks.first) {
await Future.delayed(chunkDelay);
}
final message = '[\n${chunk.join(",\n")}\n]';
final base64data = base64Encode(utf8.encode(message.toString()));
final attributes = {
if (chunk == chunks.last) 'num_chunks': chunks.length.toString(),
if (buildbucketID != null) 'buildbucket_id': buildbucketID,
if (baseRevision != null) 'base_revision': baseRevision,
};
final jsonMessage = jsonEncode({
'messages': [
{'attributes': attributes, 'data': base64data}
]
});
final headers = {'Authorization': 'Bearer $token'};
final postUrl = getPostUrl(project);
final response =
await client.post(postUrl, headers: headers, body: jsonMessage);
print('Sent pubsub message containing ${chunk.length} results');
print('Status ${response.statusCode}');
print('Response: ${response.body}');
}
print('Number of Pub/Sub messages sent: ${chunks.length}');
client.close();
}