agent/lib/src/runner.dart - mirrors/cocoon - Git at Google

 // Copyright 2016 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 'dart:convert' show utf8;
 import 'dart:io';

 // TODO(dnfield): Move to replacement once we make it more reasnoable to
 // upgrade the agent.
 // ignore: deprecated_member_use
 import 'package:vm_service_client/vm_service_client.dart';

 import 'agent.dart';
 import 'utils.dart';

 /// The default task timeout, if a custom value is not provided.
 ///
 /// This should be the same as `_kDefaultTaskTimeout` defined in https://github.com/flutter/flutter/blob/master/dev/devicelab/lib/framework/framework.dart
 const Duration _kDefaultTaskTimeout = const Duration(minutes: 15);

 /// Extra amount of time we give the devicelab task to finish or timeout on its
 /// own before forcefully quitting it.
 const Duration _kGracePeriod = const Duration(minutes: 1);

 /// Send logs in 10KB chunks.
 const int _kLogChunkSize = 10000;

 /// A result of running a single task.
 ///
 /// In normal circumstances, even when a task fails, the result is parsed from
 /// JSON returned by the task runner process via the VM service. However, if
 /// things are completely out of control and the task runner process is
 /// corrupted a failed result can be instantiated directly using
 /// [TaskResult.failure] constructor.
 class TaskResult {
   /// Parses a task result from JSON.
   TaskResult.parse(Map<String, dynamic> json)
       : succeeded = json['success'] as bool,
         data = json['data'] as Map<String, dynamic>,
         benchmarkScoreKeys = json['benchmarkScoreKeys'] ?? const <String>[],
         reason = json['reason'] as String;

   /// Constructs an unsuccessful result.
   TaskResult.failure(this.reason)
       : this.succeeded = false,
         this.data = const <String, dynamic>{},
         this.benchmarkScoreKeys = const <String>[];

   /// Whether the task succeeded.
   final bool succeeded;

   /// Task-specific JSON data.
   final Map<String, dynamic> data;

   /// Keys in [data] that store scores that will be submitted to Golem.
   ///
   /// Each key is also part of a benchmark's name tracked by Golem.
   /// A benchmark name is computed by combining [Task.name] with a key
   /// separated by a dot. For example, if a task's name is
   /// `"complex_layout__start_up"` and score key is
   /// `"engineEnterTimestampMicros"`, the score will be submitted to Golem under
   /// `"complex_layout__start_up.engineEnterTimestampMicros"`.
   ///
   /// This convention reduces the amount of configuration that needs to be done
   /// to submit benchmark scores to Golem.
   final dynamic benchmarkScoreKeys;

   /// Whether the task failed.
   bool get failed => !succeeded;

   /// Explains the failure reason if [failed].
   final String reason;

   Map<String, dynamic> toJson() {
     return <String, dynamic>{
       'success': succeeded,
       'data': data,
       'benchmarkScoreKeys': benchmarkScoreKeys,
       'reason': reason,
     };
   }
 }

 /// Runs a task in a separate Dart VM and collects the result using the VM
 /// service protocol.
 ///
 /// [taskName] is the name of the task. The corresponding task executable is
 /// expected to be found under `bin/tasks`.
 Future<TaskResult> runTask(Agent agent, CocoonTask task) async {
   String devicelabPath = '${config.flutterDirectory.path}/dev/devicelab';
   String taskExecutable = 'bin/tasks/${task.name}.dart';

   if (!file('$devicelabPath/$taskExecutable').existsSync())
     throw 'Executable Dart file not found: $taskExecutable';

   int vmServicePort = await _findAvailablePort();
   Process runner;
   await inDirectory(devicelabPath, () async {
     runner = await startProcess(
         dartBin,
         <String>[
           '--enable-vm-service=$vmServicePort',
           '--no-pause-isolates-on-exit',
           '--disable-service-auth-codes',
           taskExecutable,
           '--cloud-auth-token=${task.cloudAuthToken}',
         ],
         silent: true);
   });

   bool runnerFinished = false;

   // ignore: unawaited_futures
   runner.exitCode.then((_) {
     runnerFinished = true;
   });

   StringBuffer buffer = StringBuffer();

   Future<Null> sendLog(String message, {bool flush: false}) async {
     buffer.write(toLogString(message));
     logger.info('[task runner] [${task.name}] $message');
     // Send a chunk at a time, or upon request.
     if (flush || buffer.length > _kLogChunkSize) {
       String chunk = buffer.toString();
       buffer = StringBuffer();
       await agent.uploadLogChunk(task.key, chunk);
     }
   }

   await sendLog('Agent ID: ${agent.agentId}', flush: true);

   var stdoutSub =
       runner.stdout.transform(utf8.decoder).listen((String message) async {
     await sendLog(message);
   });
   var stderrSub =
       runner.stderr.transform(utf8.decoder).listen((String message) async {
     await sendLog(message);
   });

   String waitingFor = 'connection';
   try {
     VMIsolateRef isolate = await _connectToRunnerIsolate(vmServicePort);
     waitingFor = 'task completion';

     Duration taskTimeout = task.timeoutInMinutes != 0
         ? Duration(minutes: task.timeoutInMinutes)
         : _kDefaultTaskTimeout;

     Map<String, dynamic> taskResult = await isolate.invokeExtension(
         'ext.cocoonRunTask', <String, String>{
       'timeoutInMinutes': '${taskTimeout.inMinutes}'
     }).timeout(taskTimeout + _kGracePeriod) as Map<String, dynamic>;

     waitingFor = 'task process to exit';
     final Future<dynamic> whenProcessExits = Future.wait<void>([
       runner.exitCode,
       stdoutSub.asFuture(),
       stderrSub.asFuture(),
     ]);
     await whenProcessExits.timeout(const Duration(seconds: 1));
     return TaskResult.parse(taskResult);
   } on TimeoutException catch (timeout) {
     runner.kill(ProcessSignal.sigint);
     return TaskResult.failure(
         'Timeout waiting for $waitingFor: ${timeout.message}');
   } finally {
     await stdoutSub.cancel();
     await stderrSub.cancel();
     await sendLog('Task execution finished', flush: true);
     // Force-quit the task runner process.
     if (!runnerFinished) runner.kill(ProcessSignal.sigkill);
     // Force-quit dangling local processes (such as adb commands).
     await forceQuitRunningProcesses();
   }
 }

 Future<VMIsolateRef> _connectToRunnerIsolate(int vmServicePort) async {
   String url = 'ws://localhost:$vmServicePort/ws';
   DateTime started = DateTime.now();

   // TODO(yjbanov): due to lack of imagination at the moment the handshake with
   //                the task process is very rudimentary and requires this small
   //                delay to let the task process open up the VM service port.
   //                Otherwise we almost always hit the non-ready case first and
   //                wait a whole 1 second, which is annoying.
   await Future<void>.delayed(const Duration(milliseconds: 100));

   while (true) {
     try {
       // Make sure VM server is up by successfully opening and closing a socket.
       await (await WebSocket.connect(url)).close();

       // Look up the isolate.
       VMServiceClient client = VMServiceClient.connect(url);
       VM vm = await client.getVM();
       VMIsolateRef isolate = vm.isolates.single;
       String response =
           await isolate.invokeExtension('ext.cocoonRunnerReady') as String;
       if (response != 'ready') throw 'not ready yet';
       return isolate;
     } catch (error) {
       const Duration connectionTimeout = const Duration(seconds: 10);
       if (DateTime.now().difference(started) > connectionTimeout) {
         throw TimeoutException(
           'Failed to connect to the task runner process',
           connectionTimeout,
         );
       }
       logger.info('VM service not ready yet: $error');
       const Duration pauseBetweenRetries = const Duration(milliseconds: 200);
       logger.info('Will retry in $pauseBetweenRetries.');
       await Future<void>.delayed(pauseBetweenRetries);
     }
   }
 }

 Future<int> _findAvailablePort() async {
   int port = 20000;
   while (true) {
     try {
       ServerSocket socket =
           await ServerSocket.bind(InternetAddress.loopbackIPv4, port);
       await socket.close();
       return port;
     } catch (_) {
       port++;
     }
   }
 }
	// Copyright 2016 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 'dart:convert' show utf8;
	import 'dart:io';

	// TODO(dnfield): Move to replacement once we make it more reasnoable to
	// upgrade the agent.
	// ignore: deprecated_member_use
	import 'package:vm_service_client/vm_service_client.dart';

	import 'agent.dart';
	import 'utils.dart';

	/// The default task timeout, if a custom value is not provided.
	///
	/// This should be the same as `_kDefaultTaskTimeout` defined in https://github.com/flutter/flutter/blob/master/dev/devicelab/lib/framework/framework.dart
	const Duration _kDefaultTaskTimeout = const Duration(minutes: 15);

	/// Extra amount of time we give the devicelab task to finish or timeout on its
	/// own before forcefully quitting it.
	const Duration _kGracePeriod = const Duration(minutes: 1);

	/// Send logs in 10KB chunks.
	const int _kLogChunkSize = 10000;

	/// A result of running a single task.
	///
	/// In normal circumstances, even when a task fails, the result is parsed from
	/// JSON returned by the task runner process via the VM service. However, if
	/// things are completely out of control and the task runner process is
	/// corrupted a failed result can be instantiated directly using
	/// [TaskResult.failure] constructor.
	class TaskResult {
	/// Parses a task result from JSON.
	TaskResult.parse(Map<String, dynamic> json)
	: succeeded = json['success'] as bool,
	data = json['data'] as Map<String, dynamic>,
	benchmarkScoreKeys = json['benchmarkScoreKeys'] ?? const <String>[],
	reason = json['reason'] as String;

	/// Constructs an unsuccessful result.
	TaskResult.failure(this.reason)
	: this.succeeded = false,
	this.data = const <String, dynamic>{},
	this.benchmarkScoreKeys = const <String>[];

	/// Whether the task succeeded.
	final bool succeeded;

	/// Task-specific JSON data.
	final Map<String, dynamic> data;

	/// Keys in [data] that store scores that will be submitted to Golem.
	///
	/// Each key is also part of a benchmark's name tracked by Golem.
	/// A benchmark name is computed by combining [Task.name] with a key
	/// separated by a dot. For example, if a task's name is
	/// `"complex_layout__start_up"` and score key is
	/// `"engineEnterTimestampMicros"`, the score will be submitted to Golem under
	/// `"complex_layout__start_up.engineEnterTimestampMicros"`.
	///
	/// This convention reduces the amount of configuration that needs to be done
	/// to submit benchmark scores to Golem.
	final dynamic benchmarkScoreKeys;

	/// Whether the task failed.
	bool get failed => !succeeded;

	/// Explains the failure reason if [failed].
	final String reason;

	Map<String, dynamic> toJson() {
	return <String, dynamic>{
	'success': succeeded,
	'data': data,
	'benchmarkScoreKeys': benchmarkScoreKeys,
	'reason': reason,
	};
	}
	}

	/// Runs a task in a separate Dart VM and collects the result using the VM
	/// service protocol.
	///
	/// [taskName] is the name of the task. The corresponding task executable is
	/// expected to be found under `bin/tasks`.
	Future<TaskResult> runTask(Agent agent, CocoonTask task) async {
	String devicelabPath = '${config.flutterDirectory.path}/dev/devicelab';
	String taskExecutable = 'bin/tasks/${task.name}.dart';

	if (!file('$devicelabPath/$taskExecutable').existsSync())
	throw 'Executable Dart file not found: $taskExecutable';

	int vmServicePort = await _findAvailablePort();
	Process runner;
	await inDirectory(devicelabPath, () async {
	runner = await startProcess(
	dartBin,
	<String>[
	'--enable-vm-service=$vmServicePort',
	'--no-pause-isolates-on-exit',
	'--disable-service-auth-codes',
	taskExecutable,
	'--cloud-auth-token=${task.cloudAuthToken}',
	],
	silent: true);
	});

	bool runnerFinished = false;

	// ignore: unawaited_futures
	runner.exitCode.then((_) {
	runnerFinished = true;
	});

	StringBuffer buffer = StringBuffer();

	Future<Null> sendLog(String message, {bool flush: false}) async {
	buffer.write(toLogString(message));
	logger.info('[task runner] [${task.name}] $message');
	// Send a chunk at a time, or upon request.
	if (flush \|\| buffer.length > _kLogChunkSize) {
	String chunk = buffer.toString();
	buffer = StringBuffer();
	await agent.uploadLogChunk(task.key, chunk);
	}
	}

	await sendLog('Agent ID: ${agent.agentId}', flush: true);

	var stdoutSub =
	runner.stdout.transform(utf8.decoder).listen((String message) async {
	await sendLog(message);
	});
	var stderrSub =
	runner.stderr.transform(utf8.decoder).listen((String message) async {
	await sendLog(message);
	});

	String waitingFor = 'connection';
	try {
	VMIsolateRef isolate = await _connectToRunnerIsolate(vmServicePort);
	waitingFor = 'task completion';

	Duration taskTimeout = task.timeoutInMinutes != 0
	? Duration(minutes: task.timeoutInMinutes)
	: _kDefaultTaskTimeout;

	Map<String, dynamic> taskResult = await isolate.invokeExtension(
	'ext.cocoonRunTask', <String, String>{
	'timeoutInMinutes': '${taskTimeout.inMinutes}'
	}).timeout(taskTimeout + _kGracePeriod) as Map<String, dynamic>;

	waitingFor = 'task process to exit';
	final Future<dynamic> whenProcessExits = Future.wait<void>([
	runner.exitCode,
	stdoutSub.asFuture(),
	stderrSub.asFuture(),
	]);
	await whenProcessExits.timeout(const Duration(seconds: 1));
	return TaskResult.parse(taskResult);
	} on TimeoutException catch (timeout) {
	runner.kill(ProcessSignal.sigint);
	return TaskResult.failure(
	'Timeout waiting for $waitingFor: ${timeout.message}');
	} finally {
	await stdoutSub.cancel();
	await stderrSub.cancel();
	await sendLog('Task execution finished', flush: true);
	// Force-quit the task runner process.
	if (!runnerFinished) runner.kill(ProcessSignal.sigkill);
	// Force-quit dangling local processes (such as adb commands).
	await forceQuitRunningProcesses();
	}
	}

	Future<VMIsolateRef> _connectToRunnerIsolate(int vmServicePort) async {
	String url = 'ws://localhost:$vmServicePort/ws';
	DateTime started = DateTime.now();

	// TODO(yjbanov): due to lack of imagination at the moment the handshake with
	// the task process is very rudimentary and requires this small
	// delay to let the task process open up the VM service port.
	// Otherwise we almost always hit the non-ready case first and
	// wait a whole 1 second, which is annoying.
	await Future<void>.delayed(const Duration(milliseconds: 100));

	while (true) {
	try {
	// Make sure VM server is up by successfully opening and closing a socket.
	await (await WebSocket.connect(url)).close();

	// Look up the isolate.
	VMServiceClient client = VMServiceClient.connect(url);
	VM vm = await client.getVM();
	VMIsolateRef isolate = vm.isolates.single;
	String response =
	await isolate.invokeExtension('ext.cocoonRunnerReady') as String;
	if (response != 'ready') throw 'not ready yet';
	return isolate;
	} catch (error) {
	const Duration connectionTimeout = const Duration(seconds: 10);
	if (DateTime.now().difference(started) > connectionTimeout) {
	throw TimeoutException(
	'Failed to connect to the task runner process',
	connectionTimeout,
	);
	}
	logger.info('VM service not ready yet: $error');
	const Duration pauseBetweenRetries = const Duration(milliseconds: 200);
	logger.info('Will retry in $pauseBetweenRetries.');
	await Future<void>.delayed(pauseBetweenRetries);
	}
	}
	}

	Future<int> _findAvailablePort() async {
	int port = 20000;
	while (true) {
	try {
	ServerSocket socket =
	await ServerSocket.bind(InternetAddress.loopbackIPv4, port);
	await socket.close();
	return port;
	} catch (_) {
	port++;
	}
	}
	}