Repository navigation
Fix random failures on functional tests #14331
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from 6 commits
2aac150
bcd65e9
a908fe1
629e881
e64ff0b
30aaacd
cd0fe62
d4e19f7
56076da
688fb58
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,9 +1,6 @@ | ||
| name: Pull Request DataScience | ||
|
|
||
| on: | ||
| push: | ||
| branches: | ||
| - main | ||
| pull_request: | ||
| branches: | ||
| - main | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,3 +1,3 @@ | ||
| { | ||
| "reporterEnabled": "spec,mocha-junit-reporter" | ||
| "reporterEnabled": "./build/ci/scripts/spec_with_pid,mocha-junit-reporter" | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,96 @@ | ||
| 'use strict'; | ||
| /** | ||
| * @module Spec | ||
| */ | ||
| /** | ||
| * Module dependencies. | ||
| */ | ||
|
|
||
| var Base = require('mocha/lib/reporters/base'); | ||
| var constants = require('mocha/lib/runner').constants; | ||
| var EVENT_RUN_BEGIN = constants.EVENT_RUN_BEGIN; | ||
| var EVENT_RUN_END = constants.EVENT_RUN_END; | ||
| var EVENT_SUITE_BEGIN = constants.EVENT_SUITE_BEGIN; | ||
| var EVENT_SUITE_END = constants.EVENT_SUITE_END; | ||
| var EVENT_TEST_FAIL = constants.EVENT_TEST_FAIL; | ||
| var EVENT_TEST_PASS = constants.EVENT_TEST_PASS; | ||
| var EVENT_TEST_PENDING = constants.EVENT_TEST_PENDING; | ||
| var inherits = require('mocha/lib/utils').inherits; | ||
| var color = Base.color; | ||
|
|
||
| /** | ||
| * Expose `Spec`. | ||
| */ | ||
|
|
||
| exports = module.exports = Spec; | ||
|
|
||
| /** | ||
| * Constructs a new `Spec` reporter instance. | ||
| * | ||
| * @public | ||
| * @class | ||
| * @memberof Mocha.reporters | ||
| * @extends Mocha.reporters.Base | ||
| * @param {Runner} runner - Instance triggers reporter actions. | ||
| * @param {Object} [options] - runner options | ||
| */ | ||
| function Spec(runner, options) { | ||
| Base.call(this, runner, options); | ||
|
|
||
| var self = this; | ||
| var indents = 0; | ||
| var n = 0; | ||
|
|
||
| function indent() { | ||
| return Array(indents).join(' '); | ||
| } | ||
|
|
||
| runner.on(EVENT_RUN_BEGIN, function () { | ||
| Base.consoleLog(); | ||
| }); | ||
|
|
||
| runner.on(EVENT_SUITE_BEGIN, function (suite) { | ||
| ++indents; | ||
| Base.consoleLog(color('suite', `${process.pid} %s%s`), indent(), suite.title); | ||
| }); | ||
|
|
||
| runner.on(EVENT_SUITE_END, function () { | ||
| --indents; | ||
| if (indents === 1) { | ||
| Base.consoleLog(); | ||
| } | ||
| }); | ||
|
|
||
| runner.on(EVENT_TEST_PENDING, function (test) { | ||
| var fmt = indent() + color('pending', `${process.pid} - %s`); | ||
| Base.consoleLog(fmt, test.title); | ||
| }); | ||
|
|
||
| runner.on(EVENT_TEST_PASS, function (test) { | ||
| var fmt; | ||
| if (test.speed === 'fast') { | ||
| fmt = indent() + color('checkmark', `${process.pid} ` + Base.symbols.ok) + color('pass', ' %s'); | ||
| Base.consoleLog(fmt, test.title); | ||
| } else { | ||
| fmt = | ||
| indent() + | ||
| color('checkmark', `${process.pid} ` + Base.symbols.ok) + | ||
| color('pass', ' %s') + | ||
| color(test.speed, ' (%dms)'); | ||
| Base.consoleLog(fmt, test.title, test.duration); | ||
| } | ||
| }); | ||
|
|
||
| runner.on(EVENT_TEST_FAIL, function (test) { | ||
| Base.consoleLog(indent() + color('fail', `${process.pid} %d) %s`), ++n, test.title); | ||
| }); | ||
|
|
||
| runner.once(EVENT_RUN_END, self.epilogue.bind(self)); | ||
| } | ||
|
|
||
| /** | ||
| * Inherit from `Base.prototype`. | ||
| */ | ||
| inherits(Spec, Base); | ||
|
|
||
| Spec.description = 'hierarchical & verbose [default]'; |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| Functional test failures related to kernel ports overlapping. |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| import sys | ||
| import argparse | ||
| import os | ||
|
|
||
| os.system("color") | ||
| from pathlib import Path | ||
| import re | ||
|
|
||
| parser = argparse.ArgumentParser(description="Parse a test log into its parts") | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Is this used?
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. logParser? Or the args?
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This file is used (I mentioned it in the PR description) to split a test output file into synchronized parts so you can debug it. |
||
| parser.add_argument("testlog", type=str, nargs=1, help="Log to parse") | ||
| parser.add_argument( | ||
| "--testoutput", action="store_true", help="Show all failures and passes" | ||
| ) | ||
| parser.add_argument( | ||
| "--split", | ||
| action="store_true", | ||
| help="Split into per process files. Each file will have the pid appended", | ||
| ) | ||
| ansi_escape = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])") | ||
| pid_regex = re.compile(r"(\d+).*") | ||
|
|
||
|
|
||
| def printTestOutput(testlog): | ||
| # Find all the lines that don't have a PID in them. These are the test output | ||
| p = Path(testlog[0]) | ||
| with p.open() as f: | ||
| for line in f.readlines(): | ||
| stripped = line.strip() | ||
| if len(stripped) > 2 and stripped[0] == "\x1B" and stripped[1] == "[": | ||
| print(line.rstrip()) # Should be a test line as it has color encoding | ||
|
|
||
|
|
||
| def splitByPid(testlog): | ||
| # Split testlog into prefixed logs based on pid | ||
| baseFile = os.path.splitext(testlog[0])[0] | ||
| p = Path(testlog[0]) | ||
| pids = set() | ||
| logs = {} | ||
| pid = None | ||
| with p.open() as f: | ||
| for line in f.readlines(): | ||
| stripped = ansi_escape.sub("", line.strip()) | ||
| # See if starts with a pid | ||
| if len(stripped) > 0 and stripped[0] <= "9" and stripped[0] >= "0": | ||
| # Pull out the pid | ||
| match = pid_regex.match(stripped) | ||
|
|
||
| # Pids are at least two digits | ||
| if match != None and len(match.group(1)) > 2: | ||
| # Pid is found | ||
| pid = int(match.group(1)) | ||
|
|
||
| # See if we've created a log for this pid or not | ||
| if not pid in pids: | ||
| pids.add(pid) | ||
| logFile = "{}_{}.log".format(baseFile, pid) | ||
| print("Writing to new log: " + logFile) | ||
| logs[pid] = Path(logFile).open(mode="w") | ||
|
|
||
| # Add this line to the log | ||
| if pid != None: | ||
| logs[pid].write(line) | ||
| # Close all of the open logs | ||
| for key in logs: | ||
| logs[key].close() | ||
|
|
||
|
|
||
| def doWork(args): | ||
| if not args.testlog: | ||
| print("Test log should be passed") | ||
| elif args.testoutput: | ||
| printTestOutput(args.testlog) | ||
| elif args.split: | ||
| splitByPid(args.testlog) | ||
| else: | ||
| parser.print_usage() | ||
|
|
||
|
|
||
| def main(): | ||
| doWork(parser.parse_args()) | ||
|
|
||
|
|
||
| if __name__ == "__main__": | ||
| main() | ||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -177,14 +177,20 @@ export abstract class BasePythonDaemon { | |||||
| return Object.keys(options).every((item) => daemonSupportedSpawnOptions.indexOf(item as any) >= 0); | ||||||
| } | ||||||
| protected sendRequestWithoutArgs<R, E, RO>(type: RequestType0<R, E, RO>): Thenable<R> { | ||||||
| return Promise.race([this.connection.sendRequest(type), this.connectionClosedDeferred.promise]); | ||||||
| if (this.proc && this.proc.exitCode === null) { | ||||||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
Suggested change
Else if we have
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The api says it returns a number or null. Never undefined. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yes, just that usage of |
||||||
| return Promise.race([this.connection.sendRequest(type), this.connectionClosedDeferred.promise]); | ||||||
| } | ||||||
| return this.connectionClosedDeferred.promise; | ||||||
| } | ||||||
| protected sendRequest<P, R, E, RO>(type: RequestType<P, R, E, RO>, params?: P): Thenable<R> { | ||||||
| if (!this.isAlive) { | ||||||
| if (!this.isAlive || this.proc.exitCode !== null) { | ||||||
| traceError('Daemon is handling a request after death.'); | ||||||
| } | ||||||
| // Throw an error if the connection has been closed. | ||||||
| return Promise.race([this.connection.sendRequest(type, params), this.connectionClosedDeferred.promise]); | ||||||
| if (this.proc && this.proc.exitCode === null) { | ||||||
| // Throw an error if the connection has been closed. | ||||||
| return Promise.race([this.connection.sendRequest(type, params), this.connectionClosedDeferred.promise]); | ||||||
| } | ||||||
| return this.connectionClosedDeferred.promise; | ||||||
| } | ||||||
| protected throwIfRPCConnectionIsDead() { | ||||||
| if (!this.isAlive) { | ||||||
|
|
||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -2,10 +2,15 @@ | |
| // Licensed under the MIT License. | ||
| 'use strict'; | ||
|
|
||
| import * as fsextra from 'fs-extra'; | ||
| import { inject, injectable } from 'inversify'; | ||
| import * as os from 'os'; | ||
| import * as path from 'path'; | ||
| import * as portfinder from 'portfinder'; | ||
| import { promisify } from 'util'; | ||
| import * as uuid from 'uuid/v4'; | ||
| import { isTestExecution } from '../../common/constants'; | ||
| import { traceInfo } from '../../common/logger'; | ||
| import { IProcessServiceFactory } from '../../common/process/types'; | ||
| import { Resource } from '../../common/types'; | ||
| import { captureTelemetry } from '../../telemetry'; | ||
|
|
@@ -16,20 +21,62 @@ import { KernelDaemonPool } from './kernelDaemonPool'; | |
| import { KernelProcess } from './kernelProcess'; | ||
| import { IKernelConnection, IKernelLauncher, IKernelProcess } from './types'; | ||
|
|
||
| const PortToStartFrom = 9_000; | ||
| const PortFormatString = `kernelLauncherPortStart_{0}.tmp`; | ||
|
|
||
| // Launches and returns a kernel process given a resource or python interpreter. | ||
| // If the given interpreter is undefined, it will try to use the selected interpreter. | ||
| // If the selected interpreter doesn't have a kernel, it will find a kernel on disk and use that. | ||
| @injectable() | ||
| export class KernelLauncher implements IKernelLauncher { | ||
| private static nextFreePortToTryAndUse = PortToStartFrom; | ||
| private static startPortPromise = KernelLauncher.computeStartPort(); | ||
| private static nextFreePortToTryAndUsePromise = KernelLauncher.startPortPromise; | ||
| constructor( | ||
| @inject(IProcessServiceFactory) private processExecutionFactory: IProcessServiceFactory, | ||
| @inject(IDataScienceFileSystem) private readonly fs: IDataScienceFileSystem, | ||
| @inject(KernelDaemonPool) private readonly daemonPool: KernelDaemonPool | ||
| ) {} | ||
|
|
||
| // This function is public so it can be called when a test shuts down | ||
| public static async cleanupStartPort() { | ||
| try { | ||
| // Destroy the file | ||
| const port = await KernelLauncher.startPortPromise; | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Not sure I like this. There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yup, all good, I can see it is |
||
| traceInfo(`Cleaning up port start file : ${port}`); | ||
|
|
||
| const filePath = path.join(os.tmpdir(), PortFormatString.format(port.toString())); | ||
| await fsextra.remove(filePath); | ||
| } catch (exc) { | ||
| // If it fails it doesn't really matter. Just a temp file | ||
| traceInfo(`Kernel port mutex failed to cleanup: `, exc); | ||
| } | ||
| } | ||
|
|
||
| private static async computeStartPort(): Promise<number> { | ||
| if (isTestExecution()) { | ||
| // Since multiple instances of a test may be running, write our best guess to a shared file | ||
| let portStart = 9_000; | ||
| let result = 0; | ||
| while (result === 0 && portStart < 65_000) { | ||
| try { | ||
| // Try creating a file with the port in the name | ||
| const filePath = path.join(os.tmpdir(), PortFormatString.format(portStart.toString())); | ||
| await fsextra.open(filePath, 'wx'); | ||
|
|
||
| // If that works, we have our port | ||
| result = portStart; | ||
| } catch { | ||
| // If that fails, it should mean the file already exists | ||
| portStart += 1_000; | ||
| } | ||
| } | ||
| traceInfo(`Computed port start for KernelLauncher is : ${result}`); | ||
|
|
||
| return result; | ||
| } else { | ||
| return 9_000; | ||
| } | ||
| } | ||
|
|
||
| @captureTelemetry(Telemetry.KernelLauncherPerf) | ||
| public async launch( | ||
| kernelConnectionMetadata: KernelSpecConnectionMetadata | PythonKernelConnectionMetadata, | ||
|
|
@@ -49,18 +96,28 @@ export class KernelLauncher implements IKernelLauncher { | |
| return kernelProcess; | ||
| } | ||
|
|
||
| private async getKernelConnection(): Promise<IKernelConnection> { | ||
| private async getPorts(): Promise<number[]> { | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. This is really minor, but maybe not the exact same name as the portfinder function that we are using? Or change the other one to portfinderGetPorts.
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Sure |
||
| const getPorts = promisify(portfinder.getPorts); | ||
|
|
||
| // Have to wait for static port lookup (it handles case where two VS code instances are running) | ||
| const nextFreePort = await KernelLauncher.nextFreePortToTryAndUsePromise; | ||
| const startPort = await KernelLauncher.startPortPromise; | ||
|
|
||
| // Ports may have been freed, hence start from begining. | ||
| const port = | ||
| KernelLauncher.nextFreePortToTryAndUse > PortToStartFrom + 1_000 | ||
| ? PortToStartFrom | ||
| : KernelLauncher.nextFreePortToTryAndUse; | ||
| const port = nextFreePort > startPort + 1_000 ? startPort : nextFreePort; | ||
|
|
||
| // Then get the next set starting at that point | ||
| const ports = await getPorts(5, { host: '127.0.0.1', port }); | ||
|
|
||
| // We launch restart kernels in the background, its possible other session hasn't started. | ||
| // Ensure we do not use same ports. | ||
| KernelLauncher.nextFreePortToTryAndUse = Math.max(...ports) + 1; | ||
| KernelLauncher.nextFreePortToTryAndUsePromise = Promise.resolve(Math.max(...ports) + 1); | ||
|
|
||
| return ports; | ||
| } | ||
|
|
||
| private async getKernelConnection(): Promise<IKernelConnection> { | ||
| const ports = await this.getPorts(); | ||
| return { | ||
| version: 1, | ||
| key: uuid(), | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Do you want to change this in the PR?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Oh probably not. I'll put it back