Skip to content

Commit f11bae3

Browse files
CoderYellowCoderYellow
committed
fix(webflux): name controller arguments explicitly
1 parent 861bc57 commit f11bae3

6 files changed

Lines changed: 156 additions & 9 deletions

File tree

‎autoconfigure/src/main/kotlin/org/openprojectx/hadoop/yarn/log/api/autoconfigure/YarnLogSseController.kt‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -23,12 +23,12 @@ class YarnLogSseController(
2323
produces = [MediaType.TEXT_EVENT_STREAM_VALUE],
2424
)
2525
fun stream(
26-
@PathVariable applicationId: String,
27-
@RequestParam(defaultValue = "true") follow: Boolean,
28-
@RequestParam(required = false) logFiles: Set<String>?,
29-
@RequestParam(required = false) containers: Set<String>?,
30-
@RequestParam(required = false) tailBytes: Long?,
31-
@RequestParam(required = false) pollIntervalMs: Long?,
26+
@PathVariable("applicationId") applicationId: String,
27+
@RequestParam(name = "follow", defaultValue = "true") follow: Boolean,
28+
@RequestParam(name = "logFiles", required = false) logFiles: Set<String>?,
29+
@RequestParam(name = "containers", required = false) containers: Set<String>?,
30+
@RequestParam(name = "tailBytes", required = false) tailBytes: Long?,
31+
@RequestParam(name = "pollIntervalMs", required = false) pollIntervalMs: Long?,
3232
principal: Principal?,
3333
): Flux<ServerSentEvent<YarnLogEvent>> {
3434
val interval = pollIntervalMs?.let(Duration::ofMillis) ?: properties.pollInterval

‎autoconfigure/src/test/kotlin/org/openprojectx/hadoop/yarn/log/api/autoconfigure/YarnLogSseControllerTest.kt‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
package org.openprojectx.hadoop.yarn.log.api.autoconfigure
2+
3+
import com.ninjasquad.springmockk.MockkBean
4+
import io.mockk.every
5+
import io.mockk.slot
6+
import org.openprojectx.hadoop.yarn.log.api.YarnLogEvent
7+
import org.openprojectx.hadoop.yarn.log.api.YarnLogEventType
8+
import org.openprojectx.hadoop.yarn.log.api.YarnLogSource
9+
import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamRequest
10+
import org.openprojectx.hadoop.yarn.log.api.YarnLogStreamService
11+
import org.springframework.beans.factory.annotation.Autowired
12+
import org.springframework.boot.autoconfigure.EnableAutoConfiguration
13+
import org.springframework.boot.webflux.test.autoconfigure.WebFluxTest
14+
import org.springframework.context.annotation.Configuration
15+
import org.springframework.context.annotation.Import
16+
import org.springframework.http.MediaType
17+
import org.springframework.test.context.ContextConfiguration
18+
import org.springframework.test.web.reactive.server.WebTestClient
19+
import reactor.core.publisher.Flux
20+
import kotlin.test.Test
21+
import kotlin.test.assertEquals
22+
import kotlin.test.assertTrue
23+
24+
@WebFluxTest(YarnLogSseController::class)
25+
@Import(YarnLogApiProperties::class)
26+
@ContextConfiguration(
27+
classes = [YarnLogSseControllerTest.TestApplication::class, YarnLogSseController::class],
28+
)
29+
class YarnLogSseControllerTest {
30+
@Autowired
31+
private lateinit var webTestClient: WebTestClient
32+
33+
@MockkBean
34+
private lateinit var service: YarnLogStreamService
35+
36+
@Test
37+
fun `binds named parameters and streams readable SSE events`() {
38+
val requestSlot = slot<YarnLogStreamRequest>()
39+
every { service.stream(capture(requestSlot)) } returns Flux.just(
40+
YarnLogEvent(
41+
type = YarnLogEventType.LOG,
42+
sequence = 1,
43+
applicationId = APPLICATION_ID,
44+
containerId = "container_123_0001_01_000001",
45+
logType = "stdout",
46+
source = YarnLogSource.NODE_MANAGER,
47+
offset = 12,
48+
encoding = "BASE64",
49+
data = "aGVsbG8K",
50+
text = "hello\\n",
51+
),
52+
YarnLogEvent(
53+
type = YarnLogEventType.COMPLETE,
54+
sequence = 2,
55+
applicationId = APPLICATION_ID,
56+
state = "FINISHED",
57+
),
58+
)
59+
60+
val events = webTestClient.get()
61+
.uri { builder ->
62+
builder.path("/api/v1/yarn/applications/$APPLICATION_ID/logs")
63+
.queryParam("follow", true)
64+
.queryParam("logFiles", "stdout,stderr")
65+
.build()
66+
}
67+
.accept(MediaType.TEXT_EVENT_STREAM)
68+
.exchange()
69+
.expectStatus().isOk
70+
.expectHeader().contentTypeCompatibleWith(MediaType.TEXT_EVENT_STREAM)
71+
.returnResult(String::class.java)
72+
.responseBody
73+
.collectList()
74+
.block()!!
75+
76+
assertEquals(APPLICATION_ID, requestSlot.captured.applicationId)
77+
assertEquals(setOf("stdout", "stderr"), requestSlot.captured.logFiles)
78+
assertTrue(requestSlot.captured.follow)
79+
assertTrue(events.any { it.contains("hello\\\\n") })
80+
}
81+
82+
private companion object {
83+
const val APPLICATION_ID = "application_123_0001"
84+
}
85+
86+
@Configuration(proxyBeanMethods = false)
87+
@EnableAutoConfiguration
88+
class TestApplication
89+
}

‎autoconfigure/yarn-log-api-spring-boot-autoconfigure.gradle.kts‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ dependencies {
2626
api("org.springframework.boot:spring-boot-starter")
2727
kapt("org.springframework.boot:spring-boot-configuration-processor")
2828

29-
30-
29+
testImplementation("org.springframework.boot:spring-boot-starter-webflux-test")
30+
testImplementation("com.ninja-squad:springmockk:5.0.1")
31+
testImplementation(kotlin("test"))
3132
}

‎buildSrc/src/main/kotlin/kotlin-jvm.gradle.kts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,9 @@ plugins {
1212
kotlin {
1313
// Use a specific Java version to make it easier to work in different environments.
1414
jvmToolchain(17)
15+
compilerOptions {
16+
javaParameters.set(true)
17+
}
1518
}
1619

1720
dependencies {
@@ -30,4 +33,4 @@ tasks.withType<Test>().configureEach {
3033
TestLogEvent.SKIPPED
3134
)
3235
}
33-
}
36+
}

‎core/build.gradle.kts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,4 +25,5 @@ dependencies {
2525
testImplementation(kotlin("test"))
2626
testImplementation("org.junit.jupiter:junit-jupiter")
2727
testImplementation("io.projectreactor:reactor-test")
28+
testImplementation("org.wiremock:wiremock-standalone:3.13.2")
2829
}

‎core/src/test/kotlin/org/openprojectx/hadoop/yarn/log/api/engine/NodeManagerLogClientWireMockTest.kt‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
package org.openprojectx.hadoop.yarn.log.api.engine
2+
3+
import com.github.tomakehurst.wiremock.client.WireMock.equalTo
4+
import com.github.tomakehurst.wiremock.client.WireMock.get
5+
import com.github.tomakehurst.wiremock.client.WireMock.urlPathEqualTo
6+
import com.github.tomakehurst.wiremock.junit5.WireMockRuntimeInfo
7+
import com.github.tomakehurst.wiremock.junit5.WireMockTest
8+
import org.springframework.web.reactive.function.client.WebClient
9+
import reactor.core.scheduler.Schedulers
10+
import kotlin.test.Test
11+
import kotlin.test.assertContentEquals
12+
import kotlin.test.assertEquals
13+
14+
@WireMockTest
15+
class NodeManagerLogClientWireMockTest {
16+
@Test
17+
fun `requests a tail window and parses the NodeManager response`(wireMock: WireMockRuntimeInfo) {
18+
val logBytes = "tail".encodeToByteArray()
19+
val response = buildString {
20+
appendLine("Container: container_123_0001_01_000001 on node:8042")
21+
appendLine("LogAggregationType: LOCAL")
22+
appendLine("LogType:stdout")
23+
appendLine("LogLength:10")
24+
appendLine("LogContents:")
25+
}.encodeToByteArray() + logBytes + "End of LogType:stdout\n".encodeToByteArray()
26+
wireMock.wireMock.register(
27+
get(urlPathEqualTo("/ws/v1/node/containers/container_123_0001_01_000001/logs/stdout"))
28+
.withQueryParam("size", equalTo("-4"))
29+
.willReturn(com.github.tomakehurst.wiremock.client.WireMock.ok().withBody(response)),
30+
)
31+
val scheduler = Schedulers.newSingle("wiremock-hadoop")
32+
33+
try {
34+
val client = NodeManagerLogClient(
35+
webClient = WebClient.builder().build(),
36+
cookieProvider = HadoopSpnegoCookieProvider(scheduler),
37+
)
38+
39+
val result = client.fetchTail(
40+
nodeHttpAddress = "localhost:${wireMock.httpPort}",
41+
containerId = "container_123_0001_01_000001",
42+
logType = "stdout",
43+
windowBytes = 4,
44+
).block()!!
45+
46+
assertEquals(10, result.fileLength)
47+
assertEquals(6, result.responseStartOffset)
48+
assertContentEquals(logBytes, result.bytes)
49+
} finally {
50+
scheduler.dispose()
51+
}
52+
}
53+
}

0 commit comments

Comments
 (0)