Gradle, tests runner: don't use separate thread to parse process output

Or --scan will fail (see KT-31891 for more details). To do it in same thread we should parse output on output stream calls. TCServiceMessageOutputStreamHandler will do that. Also, all new lines now preserved, so TCServiceMessagesClient should ignore new lines after messages. New ignoreLineEndingAfterMessage setting will do that. #KT-31891 Fixed
2019-06-09 23:00:39 +03:00
parent 84b87a22e6
commit d0e3ab1beb
6 changed files with 385 additions and 55 deletions
@@ -0,0 +1,119 @@
+/*
+ * Copyright 2010-2019 JetBrains s.r.o. and Kotlin Programming Language contributors.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.gradle.internal.testing
+
+import jetbrains.buildServer.messages.serviceMessages.ServiceMessage
+import jetbrains.buildServer.messages.serviceMessages.ServiceMessageParserCallback
+import org.jetbrains.kotlin.gradle.internal.testing.TCServiceMessageOutputStreamHandler.Companion.MESSAGE_LIMIT_BYTES
+import org.slf4j.Logger
+import java.io.ByteArrayOutputStream
+import java.io.IOException
+import java.io.OutputStream
+
+/**
+ * Provides output stream that handles lines and parsing it for TeamCity server messages.
+ * Calls [client] for each parsed message and regular text.
+ *
+ * TeamCity server messages should ends with new line.
+ * Only messages short than [MESSAGE_LIMIT_BYTES] supported.
+ */
+internal class TCServiceMessageOutputStreamHandler(
+    private val client: ServiceMessageParserCallback,
+    private val onException: () -> Unit,
+    private val logger: Logger,
+    private val messageLimitBytes: Int = MESSAGE_LIMIT_BYTES // for test only
+) : OutputStream() {
+    private var closed: Boolean = false
+    private val buffer = ByteArrayOutputStream()
+    private var overflowInsideMessage: Boolean = false
+
+    @Throws(IOException::class)
+    override fun close() {
+        closed = true
+        flush()
+    }
+
+    override fun write(b: ByteArray, off: Int, len: Int) {
+        if (closed) throw IOException("The stream has been closed.")
+        var i = off
+        var last = off
+        fun bytesToAppend() = i - last
+        val end = off + len
+
+        fun append(len: Int = bytesToAppend()) {
+            buffer.write(b, last, i - last)
+            last += len
+        }
+
+        while (i < end) {
+            val c = b[i++]
+            if (c == '\n'.toByte()) {
+                append()
+                flushLine()
+            } else if (buffer.size() + bytesToAppend() >= messageLimitBytes) {
+                append(messageLimitBytes - buffer.size())
+                overflow()
+            }
+        }
+
+        append()
+    }
+
+    @Throws(IOException::class)
+    override fun write(b: Int) {
+        // helpfully will be inlined at runtime
+        write(byteArrayOf(b.toByte()), 0, 1)
+    }
+
+    private fun flushLine() {
+        overflowInsideMessage = false
+        if (buffer.size() > 0) {
+            val text = buffer.toString("utf-8")
+            parse(text)
+            buffer.reset()
+        }
+    }
+
+    override fun flush() {
+        flushLine()
+    }
+
+    private fun overflow() {
+        val text = buffer.toString("utf-8")
+
+        // support messageLimitBytes inside "##teamcity[...]" (including "##teamcity[]").
+        val i = if (overflowInsideMessage) {
+            if (!text.endsWith("]")) {
+                logger.warn("Cannot process process output: too long teamcity service message (more then 1Mb). Event was lost. See stdout for more details.")
+            }
+            -1
+        } else text.indexOf("##teamcity[")
+        if (i != -1) {
+            client.regularText(text.substring(0, i))
+            buffer.reset()
+            buffer.write(text.substring(i).toByteArray())
+            overflowInsideMessage = true
+        } else {
+            flushLine()
+        }
+    }
+
+    private fun parse(text: String) {
+        try {
+            ServiceMessage.parse(text, client)
+        } catch (e: Exception) {
+            onException()
+            logger.error(
+                "Error while processing test process output message \"$text\"",
+                e
+            )
+        }
+    }
+
+    companion object {
+        private const val MESSAGE_LIMIT_BYTES = 0x100000 // 1Mb
+    }
+}
@@ -24,7 +24,8 @@ data class TCServiceMessagesClientSettings(
    val prependSuiteName: Boolean = false,
    val treatFailedTestOutputAsStacktrace: Boolean = false,
    val stackTraceParser: (String) -> ParsedStackTrace? = { null },
-    val ignoreOutOfRootNodes: Boolean = false
+    val ignoreOutOfRootNodes: Boolean = false,
+    val ignoreLineEndingAfterMessage: Boolean = true
 )

 internal open class TCServiceMessagesClient(
@@ -33,6 +34,7 @@ internal open class TCServiceMessagesClient(
    val log: Logger
 ) : ServiceMessageParserCallback {
    lateinit var rootOperationId: OperationIdentifier
+    var afterMessage = false

    inline fun root(operation: OperationIdentifier, actions: () -> Unit) {
        rootOperationId = operation
@@ -68,20 +70,25 @@ internal open class TCServiceMessagesClient(
            is TestSuiteFinished -> close(message.ts, getSuiteName(message))
            else -> Unit
        }
+
+        afterMessage = true
    }

    protected open fun getSuiteName(message: BaseTestSuiteMessage) = message.suiteName

    override fun regularText(text: String) {
-        log.kotlinDebug { "TCSM stdout captured: $text" }
-        val actualText = text + "\n"
+        val actualText = if (afterMessage && settings.ignoreLineEndingAfterMessage) text.removePrefix("\n") else text
+        if (actualText.isNotEmpty()) {
+            log.kotlinDebug { "TCSM stdout captured: $actualText" }

-        val test = leaf as? TestNode
-        if (test != null) {
-            test.output(StdOut, actualText)
-        } else {
-            printNonTestOutput(actualText)
+            val test = leaf as? TestNode
+            if (test != null) {
+                test.output(StdOut, actualText)
+            } else {
+                printNonTestOutput(actualText)
+            }
        }
+        afterMessage = false
    }

    protected open fun printNonTestOutput(actualText: String) {
@@ -1,18 +1,15 @@
 package org.jetbrains.kotlin.gradle.internal.testing

-import jetbrains.buildServer.messages.serviceMessages.ServiceMessage
 import org.gradle.api.internal.tasks.testing.TestExecuter
 import org.gradle.api.internal.tasks.testing.TestExecutionSpec
 import org.gradle.api.internal.tasks.testing.TestResultProcessor
 import org.gradle.internal.operations.BuildOperationExecutor
+import org.gradle.process.ExecResult
 import org.gradle.process.ProcessForkOptions
 import org.gradle.process.internal.ExecHandle
 import org.gradle.process.internal.ExecHandleFactory
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
-import java.io.PipedInputStream
-import java.io.PipedOutputStream
-import kotlin.concurrent.thread

 open class TCServiceMessagesTestExecutionSpec(
    val forkOptions: ProcessForkOptions,
@@ -39,54 +36,31 @@ class TCServiceMessagesTestExecutor(

    override fun execute(spec: TCServiceMessagesTestExecutionSpec, testResultProcessor: TestResultProcessor) {
        spec.wrapExecute {
-            val stdInPipe = PipedInputStream()
-
            val rootOperation = buildOperationExecutor.currentOperation.parentId

-            outputReaderThread = thread(name = "${spec.forkOptions} output reader") {
-                try {
-                    val client = spec.createClient(testResultProcessor, log)
+            val client = spec.createClient(testResultProcessor, log)

-                    client.root(rootOperation) {
-                        stdInPipe.reader().useLines { lines ->
-                            lines.forEach {
-                                if (shouldStop) {
-                                    client.closeAll()
-                                    return@thread
-                                }
+            try {
+                val exec = execHandleFactory.newExec()
+                spec.forkOptions.copyTo(exec)
+                exec.args = spec.args
+                exec.standardOutput = TCServiceMessageOutputStreamHandler(client, { spec.showSuppressedOutput() }, log)
+                execHandle = exec.build()

-                                try {
-                                    ServiceMessage.parse(it, client)
-                                } catch (e: Exception) {
-                                    spec.showSuppressedOutput()
-                                    log.error(
-                                        "Error while processing test process output message \"$it\"",
-                                        e
-                                    )
-                                }
-                            }
-                        }
-                    }
-                } catch (t: Throwable) {
-                    spec.showSuppressedOutput()
-                    log.error("Error creating TCServiceMessagesClient", t)
+                lateinit var result: ExecResult
+                client.root(rootOperation) {
+                    execHandle!!.start()
+                    result = execHandle!!.waitForFinish()
                }
-            }

-            val exec = execHandleFactory.newExec()
-            spec.forkOptions.copyTo(exec)
-            exec.args = spec.args
-            exec.standardOutput = PipedOutputStream(stdInPipe)
-
-            execHandle = exec.build()
-
-            execHandle!!.start()
-            val result = execHandle!!.waitForFinish()
-            outputReaderThread!!.join()
-
-            if (spec.checkExitCode && result.exitValue != 0) {
+                if (spec.checkExitCode && result.exitValue != 0) {
+                    spec.showSuppressedOutput()
+                    error("$execHandle exited with errors (exit code: ${result.exitValue})")
+                }
+            } catch (e: Throwable) {
+                client.closeAll()
                spec.showSuppressedOutput()
-                error("$execHandle exited with errors (exit code: ${result.exitValue})")
+                throw e
            }
        }
    }
@@ -0,0 +1,167 @@
+/*
+ * Copyright 2010-2019 JetBrains s.r.o. and Kotlin Programming Language contributors.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.gradle.internal.testing
+
+import jetbrains.buildServer.messages.serviceMessages.ServiceMessage
+import jetbrains.buildServer.messages.serviceMessages.ServiceMessageParserCallback
+import org.junit.Test
+import org.slf4j.event.EventRecodingLogger
+import org.slf4j.event.SubstituteLoggingEvent
+import org.slf4j.helpers.SubstituteLogger
+import java.text.ParseException
+import java.util.concurrent.ArrayBlockingQueue
+import kotlin.test.assertEquals
+
+class TCServiceMessageOutputStreamHandlerTest {
+    private val client = Mock()
+    private val logEvents = ArrayBlockingQueue<SubstituteLoggingEvent>(10)
+    private val log = EventRecodingLogger(SubstituteLogger("", logEvents, false), logEvents)
+    private val handler = TCServiceMessageOutputStreamHandler(client, {}, log, 30)
+
+    private val clientCalls get() = client.log.toString()
+    private val logString get() = logEvents.map { it.message }.toString()
+    
+    @Test
+    fun testLines() {
+        handler.write("Test1\n".toByteArray())
+        handler.write("Test2".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `Test1\n`\n" +
+                    "TEXT: `Test2`\n",
+            clientCalls
+        )
+        assertEquals("[]", logString)
+    }
+
+    @Test
+    fun testLines2() {
+        handler.write("Test1\n\r".toByteArray())
+        handler.write("Test2".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `Test1\n`\n" +
+                    "TEXT: `\rTest2`\n",
+            clientCalls
+        )
+        assertEquals("[]", logString)
+    }
+
+    @Test
+    fun testLines3() {
+        handler.write("Test1\n\r\n".toByteArray())
+        handler.write("Test2".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `Test1\n`\n" +
+                    "TEXT: `\r\n`\n" +
+                    "TEXT: `Test2`\n",
+            clientCalls
+        )
+        assertEquals("[]", logString)
+    }
+
+    @Test
+    fun testMessage() {
+        handler.write("xxx##teamcity[testStarted]\n".toByteArray())
+        handler.write("yyy##teamcity[testStarted]".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `xxx`\n" +
+                    "MESSAGE: `##teamcity[testStarted]`\n" +
+                    "TEXT: `\n`\n" + // this will be ignore in org.jetbrains.kotlin.gradle.internal.testing.TCServiceMessagesClient.regularText
+                    "TEXT: `yyy`\n" +
+                    "MESSAGE: `##teamcity[testStarted]`\n",
+            clientCalls
+        )
+        assertEquals("[]", logString)
+    }
+
+    @Test
+    fun testMessageSplit() {
+        handler.write("xxx##teamc".toByteArray())
+        handler.write("ity[testStarted]\n".toByteArray())
+        handler.write("yyy##teamcity[testStarted]".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `xxx`\n" +
+                    "MESSAGE: `##teamcity[testStarted]`\n" +
+                    "TEXT: `\n`\n" + // this will be ignore in org.jetbrains.kotlin.gradle.internal.testing.TCServiceMessagesClient.regularText
+                    "TEXT: `yyy`\n" +
+                    "MESSAGE: `##teamcity[testStarted]`\n",
+            clientCalls
+        )
+        assertEquals("[]", logString)
+    }
+
+    @Test
+    fun testNoOverflowNL() {
+        // "##teamcity[]".length = 12
+        // 30 - 12 = 18
+        //                                     | <- limit here
+        handler.write("012345##teamcity[12345678901234567]\n".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `012345`\n" +
+                    "MESSAGE: `##teamcity[12345678901234567]`\n" +
+                    "TEXT: `\n`\n",
+            clientCalls
+        )
+        assertEquals("[]", logString)
+    }
+
+
+    @Test
+    fun testNoOverflowNoNL() {
+        // "##teamcity[]".length = 12
+        // 30 - 12 = 18
+        //                                     | <- limit here
+        handler.write("012345##teamcity[123456789012345678]\n".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `012345`\n" +
+                    "MESSAGE: `##teamcity[123456789012345678]`\n" +
+                    "TEXT: `\n`\n",
+            clientCalls
+        )
+        assertEquals("[]", logString)
+    }
+
+    @Test
+    fun testOverflow() {
+        // "##teamcity[]".length = 12
+        // 30 - 12 = 18
+        //                                     | <- limit here
+        handler.write("012345##teamcity[1234567890123456789]\n".toByteArray())
+        handler.close()
+        assertEquals(
+            "TEXT: `012345`\n" +
+                    "TEXT: `##teamcity[1234567890123456789`\n" +
+                    "TEXT: `]\n`\n",
+            clientCalls
+        )
+        assertEquals(
+            "[Cannot process process output: too long teamcity service message (more then 1Mb). Event was lost. See stdout for more details.]",
+            logString
+        )
+    }
+
+    class Mock : ServiceMessageParserCallback {
+        val log = StringBuffer()
+
+        override fun parseException(p0: ParseException, p1: String) {
+            log.appendln("EXCEPTION `$p0`, `$p1`")
+        }
+
+        override fun serviceMessage(p0: ServiceMessage) {
+            log.appendln("MESSAGE: `$p0`")
+        }
+
+        override fun regularText(p0: String) {
+            log.appendln("TEXT: `$p0`")
+        }
+    }
+}
@@ -80,8 +80,7 @@ AssertionError: Expected value to be true.
    at mpplib2/build/js_test_node_modules/mpplib2_test.js:59:38
    at Object.fn [as test] (mpplib2/build/tmp/expandedArchives/src/KotlinTestRunner.ts:12:25)
    at Object.test (mpplib2/build/tmp/expandedArchives/src/KotlinTestTeamCityReporter.ts:80:28)
-    at test (mpplib2/build/tmp/expandedArchives/kotlin-test-js-1.3-SNAPSHOT.jar_d60f1e6d0dd94843a03bf98a569bbb73/src/main/kotlin/kotlin/test/TestApi.kt:57:15)
- // root//Test
+    at test (mpplib2/build/tmp/expandedArchives/kotlin-test-js-1.3-SNAPSHOT.jar_d60f1e6d0dd94843a03bf98a569bbb73/src/main/kotlin/kotlin/test/TestApi.kt:57:15) // root//Test
    COMPLETED FAILURE // root//Test
  COMPLETED FAILURE // root/
 COMPLETED FAILURE // root
@@ -0,0 +1,64 @@
+/*
+ * Copyright 2010-2019 JetBrains s.r.o. and Kotlin Programming Language contributors.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the license/LICENSE.txt file.
+ */
+
+package org.jetbrains.kotlin.gradle.internal.testing.tcsmc
+
+import jetbrains.buildServer.messages.serviceMessages.TestFinished
+import jetbrains.buildServer.messages.serviceMessages.TestStarted
+import jetbrains.buildServer.messages.serviceMessages.TestSuiteFinished
+import jetbrains.buildServer.messages.serviceMessages.TestSuiteStarted
+import org.junit.Test
+
+class TestLineEndings : TCServiceMessagesClientTest() {
+    @Test
+    fun test() {
+        assertEvents(
+            """
+STARTED SUITE root // root
+  STARTED SUITE  // root/
+    STARTED TEST displayName: Test, classDisplayName: , className: , name: Test // root//Test
+      StdOut[
+text1
+
+] // root//Test
+      StdOut[
+
+text2
+
+] // root//Test
+      FAILURE Expected <7>, actual <42>
+ // root//Test
+      StdOut[
+text3
+
+] // root//Test
+      StdOut[
+
+text4
+
+] // root//Test
+    COMPLETED FAILURE // root//Test
+  COMPLETED FAILURE // root/
+COMPLETED FAILURE // root
+        """
+        ) {
+            serviceMessage(TestSuiteStarted(""))
+            serviceMessage(TestStarted("Test", false, null))
+            regularText("\n\ntext1\n\n")
+            regularText("\n\ntext2\n\n")
+            serviceMessage(
+                "testFailed",
+                mapOf(
+                    "name" to "Test",
+                    "message" to "Expected <7>, actual <42>"
+                )
+            )
+            regularText("\n\ntext3\n\n")
+            regularText("\n\ntext4\n\n")
+            serviceMessage(TestFinished("Test", 0))
+            serviceMessage(TestSuiteFinished(""))
+        }
+    }
+}