feat(observability): 리포트 다운로드·SSE 스트림·JWT 인증 구현 #29 #72
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.controller | ||
|
|
||
| import hs.kr.entrydsm.observability.adapterin.web.sse.SseBroadcaster | ||
| import hs.kr.entrydsm.observability.adapterin.web.sse.SseConnectionLimiter | ||
| import hs.kr.entrydsm.observability.domain.enum.ErrorCode | ||
| import hs.kr.entrydsm.observability.domain.exception.MonitorDomainException | ||
| import jakarta.servlet.http.HttpServletRequest | ||
| import java.util.concurrent.TimeUnit | ||
| import org.springframework.http.MediaType | ||
| import org.springframework.web.bind.annotation.GetMapping | ||
| import org.springframework.web.bind.annotation.RestController | ||
| import org.springframework.web.servlet.mvc.method.annotation.SseEmitter | ||
|
|
||
| @RestController | ||
| class MonitorStreamController( | ||
| private val sseBroadcaster: SseBroadcaster, | ||
| private val connectionLimiter: SseConnectionLimiter, | ||
| ) { | ||
| @GetMapping("/api/monitor/v11/stream", produces = [MediaType.TEXT_EVENT_STREAM_VALUE]) | ||
| fun stream(httpRequest: HttpServletRequest): SseEmitter { | ||
| val clientKey = clientIp(httpRequest) | ||
| if (!connectionLimiter.tryAcquire(clientKey)) { | ||
| throw MonitorDomainException(ErrorCode.TOO_MANY_CONNECTIONS) | ||
| } | ||
| val emitter = SseEmitter(TimeUnit.MINUTES.toMillis(EMITTER_TIMEOUT_MINUTES)) | ||
| emitter.onCompletion { connectionLimiter.release(clientKey) } | ||
| emitter.onTimeout { connectionLimiter.release(clientKey) } | ||
| emitter.onError { connectionLimiter.release(clientKey) } | ||
| sseBroadcaster.register(emitter) | ||
| return emitter | ||
| } | ||
|
|
||
| private fun clientIp(request: HttpServletRequest): String = | ||
| request.getHeader("X-Forwarded-For") | ||
| ?.substringBefore(",") | ||
| ?.trim() | ||
| ?.takeIf { it.isNotBlank() } | ||
| ?: request.remoteAddr | ||
|
There was a problem hiding this comment. 🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
# X-Forwarded-For를 제거·재설정하는 신뢰된 프록시 설정과 Spring forwarded-header 설정을 확인한다.
fd -HI -t f . | rg -i '(ingress|gateway|nginx|traefik|apache|caddy|values.*ya?ml|.*\.conf$)' |
while IFS= read -r file; do
rg -n -C 3 'X-Forwarded-For|ForwardedHeaderFilter|forward-headers|remoteIpHeader|proxy_set_header' "$file" || true
doneRepository: EntryDSM/entrydsm-platform Length of output: 164 🏁 Script executed: #!/bin/bash
set -euo pipefail
files=(
systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/controller/MonitorStreamController.kt
systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/sse/SseConnectionLimiter.kt
)
for file in "${files[@]}"; do
echo "===== $file ====="
wc -l "$file"
cat -n "$file"
done
echo "===== limiter usages ====="
rg -n -C 4 'SseConnectionLimiter|tryAcquire|release|clientIp|X-Forwarded-For|ForwardedHeaderFilter|server\.forward-headers|forward-headers-strategy|remoteAddr' . \
-g '!build' -g '!dist' -g '!node_modules' || true
echo "===== proxy and application configuration files ====="
git ls-files | rg -i '(^|/)(application[^/]*\.(yml|yaml|properties|conf)|.*(ingress|gateway|nginx|traefik|apache|caddy|values).*|.*\.(conf|yaml|yml))$' |
while IFS= read -r file; do
matches=$(rg -n -C 3 'X-Forwarded-For|ForwardedHeaderFilter|forward-headers|remoteIpHeader|proxy_set_header|forward-headers-strategy|server\.forward-headers' "$file" || true)
if [ -n "$matches" ]; then
echo "===== $file ====="
printf '%s\n' "$matches"
fi
doneRepository: EntryDSM/entrydsm-platform Length of output: 33135 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "===== observability adapter tests ====="
git ls-files 'systems/observability/observability-adapter-in/src/test/**' || true
rg -n -C 3 'MonitorStreamController|SseConnectionLimiter|X-Forwarded-For|clientIp' \
systems/observability/observability-adapter-in/src/test \
systems/observability/observability-adapter-in/src/main \
2>/dev/null || true
echo "===== all forwarded-header references ====="
rg -n -C 2 'X-Forwarded-For|ForwardedHeaderFilter|forward-headers-strategy|server\.forward-headers|proxy_set_header' . \
-g '!build' -g '!dist' -g '!node_modules' || true
echo "===== deterministic limiter behavior model ====="
python3 - <<'PY'
MAX_CONNECTIONS_PER_KEY = 3
counts = {}
def try_acquire(key):
count = counts.setdefault(key, 0)
count += 1
counts[key] = count
if count <= MAX_CONNECTIONS_PER_KEY:
return True
counts[key] -= 1
return False
def release(key):
if key in counts:
counts[key] -= 1
for i in range(10_000):
key = f"attacker-{i}"
assert try_acquire(key)
release(key)
print(f"distinct_keys_after_release={len(counts)}")
print(f"nonzero_keys_after_release={sum(v != 0 for v in counts.values())}")
print(f"sample_count={counts['attacker-0']}")
PYRepository: EntryDSM/entrydsm-platform Length of output: 14774 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "===== SseBroadcaster lifecycle ====="
fd -HI -t f 'SseBroadcaster.kt' systems/observability/observability-adapter-in |
while IFS= read -r file; do
cat -n "$file"
done
echo "===== atomic-removal interleaving model ====="
python3 - <<'PY'
# Model a common proposed fix:
# release decrements, then removes the map entry when the value reaches zero.
# tryAcquire obtains the AtomicInteger reference first, then increments it.
map_entry = {"k": 1}
entry_reference = map_entry
# Existing connection releases after the new request obtained the reference.
map_entry.pop("k")
entry_reference["value"] = 1 # conceptual AtomicInteger held by the new request
print("map_contains_new_acquisition=", "k" in map_entry)
print("new_acquisition_count_is_tracked=", "k" in map_entry and map_entry["k"] is entry_reference)
echo = None
PYRepository: EntryDSM/entrydsm-platform Length of output: 3748 신뢰된 프록시가 정규화한 IP만 제한 키로 사용하십시오. 현재 구현은 요청 경로와 관계없이
📍 Affects 2 files
🤖 Prompt for AI Agents |
||
|
|
||
| companion object { | ||
| private const val EMITTER_TIMEOUT_MINUTES = 30L | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| @@ -0,0 +1,27 @@ | ||||||||||||||||||||||||||||||||||
| package hs.kr.entrydsm.observability.adapterin.web.controller | ||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||
| import hs.kr.entrydsm.observability.adapterin.web.dto.common.ApiResponse | ||||||||||||||||||||||||||||||||||
| import hs.kr.entrydsm.observability.adapterin.web.dto.common.toResponse | ||||||||||||||||||||||||||||||||||
| import hs.kr.entrydsm.observability.adapterin.web.dto.response.ReportGeneratedResponse | ||||||||||||||||||||||||||||||||||
| import hs.kr.entrydsm.observability.application.port.`in`.GenerateReportUseCase | ||||||||||||||||||||||||||||||||||
| import hs.kr.entrydsm.observability.domain.enum.ErrorCode | ||||||||||||||||||||||||||||||||||
| import hs.kr.entrydsm.observability.domain.enum.ReportFormat | ||||||||||||||||||||||||||||||||||
| import hs.kr.entrydsm.observability.domain.exception.MonitorDomainException | ||||||||||||||||||||||||||||||||||
| import org.springframework.web.bind.annotation.GetMapping | ||||||||||||||||||||||||||||||||||
| import org.springframework.web.bind.annotation.RequestParam | ||||||||||||||||||||||||||||||||||
| import org.springframework.web.bind.annotation.RestController | ||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||
| @RestController | ||||||||||||||||||||||||||||||||||
| class ReportController( | ||||||||||||||||||||||||||||||||||
| private val generateReportUseCase: GenerateReportUseCase, | ||||||||||||||||||||||||||||||||||
| ) { | ||||||||||||||||||||||||||||||||||
| @GetMapping("/api/monitor/v11/reports") | ||||||||||||||||||||||||||||||||||
| fun generate( | ||||||||||||||||||||||||||||||||||
| @RequestParam(defaultValue = "xlsx") format: String, | ||||||||||||||||||||||||||||||||||
| ): ApiResponse<ReportGeneratedResponse> { | ||||||||||||||||||||||||||||||||||
| val parsedFormat = runCatching { ReportFormat.valueOf(format.trim().uppercase()) } | ||||||||||||||||||||||||||||||||||
| .getOrElse { throw MonitorDomainException(ErrorCode.INVALID_FORMAT) } | ||||||||||||||||||||||||||||||||||
| val result = generateReportUseCase.generate(parsedFormat) | ||||||||||||||||||||||||||||||||||
| return ApiResponse(data = result.toResponse()) | ||||||||||||||||||||||||||||||||||
|
There was a problem hiding this comment. 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win 변경된 두 웹 흐름에 MVC 테스트를 추가하십시오. 현재 추가된
As per coding guidelines, " 📍 Affects 2 files
🤖 Prompt for AI AgentsSources: Coding guidelines, Path instructions 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win 보고서 생성을 Line 18의 요청은 파일을 저장하고 Redis 다운로드 토큰을 생성합니다. 수정 예시-import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.PostMapping
...
- `@GetMapping`("/api/monitor/v11/reports")
+ `@PostMapping`("/api/monitor/v11/reports")📝 Committable suggestion
Suggested change
🤖 Prompt for AI Agents |
||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.controller | ||
|
|
||
| import hs.kr.entrydsm.observability.application.port.out.ReportObjectStoragePort | ||
| import org.springframework.http.HttpHeaders | ||
| import org.springframework.http.MediaType | ||
| import org.springframework.http.ResponseEntity | ||
| import org.springframework.web.bind.annotation.GetMapping | ||
| import org.springframework.web.bind.annotation.RequestParam | ||
| import org.springframework.web.bind.annotation.RestController | ||
|
|
||
| /** 리포트 다운로드 URL이 가리키는 실제 파일 서빙. S3 presigned URL의 로컬 대체 구현. */ | ||
| @RestController | ||
| class ReportDownloadController( | ||
| private val reportObjectStoragePort: ReportObjectStoragePort, | ||
| ) { | ||
| @GetMapping("/api/monitor/v11/reports/download") | ||
| fun download(@RequestParam token: String): ResponseEntity<ByteArray> { | ||
| val downloaded = reportObjectStoragePort.resolve(token) ?: return ResponseEntity.notFound().build() | ||
| return ResponseEntity.ok() | ||
| .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename=\"${downloaded.fileName}\"") | ||
| .contentType(MediaType.APPLICATION_OCTET_STREAM) | ||
| .body(downloaded.bytes) | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.dto.response | ||
|
|
||
| import java.time.Instant | ||
|
|
||
| data class LiveLogEventResponse( | ||
| val kind: String, | ||
| val level: String, | ||
| val source: String, | ||
| val message: String, | ||
| val pageUrl: String, | ||
| val occurredAt: Instant, | ||
| ) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.dto.response | ||
|
|
||
| import java.time.Instant | ||
|
|
||
| data class ReportGeneratedResponse( | ||
| val status: String, | ||
| val downloadUrl: String, | ||
| val fileName: String, | ||
| val sizeBytes: Long, | ||
| val expiresAt: Instant, | ||
| ) |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.security | ||
|
|
||
| import hs.kr.entrydsm.observability.domain.enum.ErrorCode | ||
| import hs.kr.entrydsm.observability.domain.exception.MonitorDomainException | ||
| import io.jsonwebtoken.JwtException | ||
| import io.jsonwebtoken.Jwts | ||
| import io.jsonwebtoken.security.Keys | ||
| import jakarta.servlet.http.HttpServletRequest | ||
| import jakarta.servlet.http.HttpServletResponse | ||
| import java.nio.charset.StandardCharsets | ||
| import org.springframework.stereotype.Component | ||
| import org.springframework.web.servlet.HandlerInterceptor | ||
|
|
||
| /** | ||
| * ponytail: 이 저장소엔 아직 Spring Security/JWT 발급 체계가 병합되어 있지 않아, | ||
| * 전체 Security 스택 대신 Bearer 토큰 서명·발급자·role 클레임만 확인하는 경량 인터셉터로 둔다. | ||
| * identity의 정식 인증이 병합되면 같은 시크릿을 공유하도록 설정만 맞추면 된다. | ||
| */ | ||
| @Component | ||
| class JwtAuthInterceptor( | ||
| private val properties: JwtAuthProperties, | ||
| ) : HandlerInterceptor { | ||
| private val key by lazy { Keys.hmacShaKeyFor(properties.secret.toByteArray(StandardCharsets.UTF_8)) } | ||
|
|
||
| override fun preHandle(request: HttpServletRequest, response: HttpServletResponse, handler: Any): Boolean { | ||
| val token = bearerToken(request) ?: throw MonitorDomainException(ErrorCode.UNAUTHORIZED) | ||
| val claims = try { | ||
| Jwts.parser() | ||
| .verifyWith(key) | ||
| .requireIssuer(properties.issuer) | ||
| .build() | ||
| .parseSignedClaims(token) | ||
| .payload | ||
| } catch (exception: JwtException) { | ||
| throw MonitorDomainException(ErrorCode.UNAUTHORIZED) | ||
| } catch (exception: IllegalArgumentException) { | ||
| throw MonitorDomainException(ErrorCode.UNAUTHORIZED) | ||
| } | ||
| if (claims["role"] as? String != ADMIN_ROLE) { | ||
| throw MonitorDomainException(ErrorCode.FORBIDDEN) | ||
| } | ||
| return true | ||
| } | ||
|
|
||
| private fun bearerToken(request: HttpServletRequest): String? { | ||
| val header = request.getHeader("Authorization") ?: return null | ||
| if (!header.startsWith(BEARER_PREFIX)) return null | ||
| return header.removePrefix(BEARER_PREFIX).trim().takeIf { it.isNotEmpty() } | ||
| } | ||
|
|
||
| companion object { | ||
| private const val BEARER_PREFIX = "Bearer " | ||
| private const val ADMIN_ROLE = "ADMIN" | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.security | ||
|
|
||
| import org.springframework.boot.context.properties.ConfigurationProperties | ||
| import org.springframework.stereotype.Component | ||
|
|
||
| /** | ||
| * identity의 auth.jwt.secret/issuer와 설정 키 이름을 통일해, 정식 인증이 병합되면 | ||
| * 같은 시크릿을 가리키도록 맞추기만 하면 되게 한다. | ||
| */ | ||
| @Component | ||
| @ConfigurationProperties(prefix = "auth.jwt") | ||
| class JwtAuthProperties { | ||
| lateinit var secret: String | ||
| lateinit var issuer: String | ||
|
Comment on lines
+18
to
+21
There was a problem hiding this comment. 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
ast-grep outline systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/security/JwtAuthProperties.kt --items all
rg -n -C 3 'ConfigurationPropertiesScan|EnableConfigurationProperties|auth\.jwt|auth:|jwt:|starter-validation|jakarta\.validation' .Repository: EntryDSM/entrydsm-platform Length of output: 17569 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '--- JwtAuthProperties.kt ---'
cat -n systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/security/JwtAuthProperties.kt
echo '--- JwtAuthInterceptor usages and implementation ---'
rg -n -C 5 'class JwtAuthInterceptor|JwtAuthProperties|\.secret|\.issuer' systems/observability
echo '--- configuration registration ---'
rg -n -C 4 'ComponentScan|ConfigurationProperties|JwtAuthInterceptor|WebMvcConfigurer|addInterceptors' systems/observability
echo '--- related tests ---'
rg -n -C 5 'JwtAuthProperties|JwtAuthInterceptor|auth\.jwt|JWT_SECRET|JWT_ISSUER' systems/observability --glob '*Test.kt' --glob '*test.yaml' --glob '*test.yml' --glob 'application*.yaml' --glob 'application*.yml'Repository: EntryDSM/entrydsm-platform Length of output: 34168 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path("systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/security/JwtAuthProperties.kt")
text = p.read_text()
required = ["lateinit var secret: String", "lateinit var issuer: String"]
print("properties_file_exists:", p.exists())
for item in required:
print(f"{item!r}:", item in text)
for path in Path("systems/observability").rglob("*.kt"):
t = path.read_text()
if "JwtAuthProperties" in t or "JwtAuthInterceptor" in t:
print(f"reference: {path}")
PYRepository: EntryDSM/entrydsm-platform Length of output: 955 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '--- build and test configuration for observability modules ---'
rg -n -C 5 'spring-boot-starter-validation|spring-boot-starter-test|testImplementation|application-test|`@SpringBootTest`|`@WebMvcTest`' systems/observability kotlin.MODULE.bazel WORKSPACE* MODULE.bazel 2>/dev/null || true
echo '--- all JWT property declarations and configuration files ---'
rg -n -C 3 'auth:[[:space:]]*$|jwt:[[:space:]]*$|secret:|issuer:' systems/observability --glob '*.yaml' --glob '*.yml' --glob '*.kt'
echo '--- interceptor source candidates ---'
fd -i 'JwtAuthInterceptor|JwtAuthProperties' systems/observabilityRepository: EntryDSM/entrydsm-platform Length of output: 4652 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '--- observability bootstrap entry point and dependencies ---'
fd -i 'application|main|BUILD|MODULE' systems/observability
rg -n -C 5 'SpringBootApplication|observability-adapter-in|resources|application.yaml' systems/observability --glob '*.kt' --glob 'BUILD*' --glob '*.bzl' --glob '*.yaml' --glob '*.yml'
echo '--- interceptor complete relevant flow ---'
cat -n systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/security/JwtAuthInterceptor.kt | sed -n '19,45p'
echo '--- bootstrap application configuration ---'
cat -n systems/observability/observability-bootstrap/src/main/resources/application.yaml | sed -n '20,32p'Repository: EntryDSM/entrydsm-platform Length of output: 8693 🌐 Web query:
💡 Result: In Spring Boot, using Kotlin's lateinit properties with Citations:
기본 JWT 시크릿으로 애플리케이션이 시작되지 않게 하세요.
🧰 Tools🪛 detekt (1.23.8)[warning] 13-13: Usages of lateinit should be avoided. (detekt.potential-bugs.LateinitUsage) [warning] 14-14: Usages of lateinit should be avoided. (detekt.potential-bugs.LateinitUsage) 🤖 Prompt for AI AgentsSource: Linters/SAST tools |
||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.security | ||
|
|
||
| import org.springframework.context.annotation.Configuration | ||
| import org.springframework.web.servlet.config.annotation.InterceptorRegistry | ||
| import org.springframework.web.servlet.config.annotation.WebMvcConfigurer | ||
|
|
||
| @Configuration | ||
| class WebMvcConfig( | ||
| private val jwtAuthInterceptor: JwtAuthInterceptor, | ||
| ) : WebMvcConfigurer { | ||
| override fun addInterceptors(registry: InterceptorRegistry) { | ||
| registry.addInterceptor(jwtAuthInterceptor) | ||
| .addPathPatterns("/api/monitor/v11/**") | ||
| .excludePathPatterns( | ||
| "/api/monitor/v11/collect/session", | ||
| "/api/monitor/v11/collect/client-log", | ||
| ) | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.sse | ||
|
|
||
| import hs.kr.entrydsm.observability.adapterin.web.dto.common.toResponse | ||
| import hs.kr.entrydsm.observability.application.port.`in`.GetDashboardSnapshotUseCase | ||
| import java.util.concurrent.CopyOnWriteArrayList | ||
| import java.util.concurrent.atomic.AtomicLong | ||
| import org.springframework.scheduling.annotation.Scheduled | ||
| import org.springframework.stereotype.Component | ||
| import org.springframework.web.servlet.mvc.method.annotation.SseEmitter | ||
|
|
||
| /** | ||
| * ponytail: 커넥션 목록을 인메모리로만 들고 있어 다중 인스턴스에서는 인스턴스별로 브로드캐스트가 갈린다. | ||
| * 수평 확장이 필요해지면 Redis Pub/Sub로 교체한다. | ||
| */ | ||
| @Component | ||
| class SseBroadcaster( | ||
| private val getDashboardSnapshotUseCase: GetDashboardSnapshotUseCase, | ||
| ) { | ||
| private val emitters = CopyOnWriteArrayList<SseEmitter>() | ||
| private val eventIdSeq = AtomicLong(System.currentTimeMillis()) | ||
|
|
||
| fun register(emitter: SseEmitter) { | ||
| emitters.add(emitter) | ||
| emitter.onCompletion { emitters.remove(emitter) } | ||
| emitter.onTimeout { emitters.remove(emitter) } | ||
| emitter.onError { emitters.remove(emitter) } | ||
| sendInitialSnapshot(emitter) | ||
| } | ||
|
|
||
| fun publishLog(payload: Any) { | ||
| broadcast("log", payload) | ||
| } | ||
|
|
||
| @Scheduled(fixedRate = 5000) | ||
| fun broadcastFrequent() { | ||
| if (emitters.isEmpty()) return | ||
| val snapshot = getDashboardSnapshotUseCase.getSnapshot(null) | ||
| broadcast("traffic", snapshot.traffic.toResponse()) | ||
| broadcast("api", snapshot.api.toResponse()) | ||
| broadcast("business", snapshot.business.toResponse()) | ||
| broadcast("service", snapshot.services.toResponse()) | ||
| } | ||
|
|
||
| @Scheduled(fixedRate = 300000) | ||
| fun broadcastResource() { | ||
| if (emitters.isEmpty()) return | ||
| broadcast("resource", getDashboardSnapshotUseCase.getSnapshot(null).resource.toResponse()) | ||
| } | ||
|
|
||
| @Scheduled(fixedRate = 15000) | ||
| fun ping() { | ||
| emitters.forEach { emitter -> | ||
| runCatching { emitter.send(SseEmitter.event().comment("ping")) } | ||
| .onFailure { emitters.remove(emitter) } | ||
| } | ||
| } | ||
|
|
||
| private fun sendInitialSnapshot(emitter: SseEmitter) { | ||
| runCatching { | ||
| emitter.send(SseEmitter.event().reconnectTime(RECONNECT_TIME_MS)) | ||
| val snapshot = getDashboardSnapshotUseCase.getSnapshot(null) | ||
| emitter.send(SseEmitter.event().id(nextId()).name("snapshot").data(snapshot.toResponse())) | ||
| }.onFailure { emitters.remove(emitter) } | ||
| } | ||
|
|
||
| private fun broadcast(eventName: String, data: Any) { | ||
| emitters.forEach { emitter -> | ||
| runCatching { emitter.send(SseEmitter.event().id(nextId()).name(eventName).data(data)) } | ||
| .onFailure { emitters.remove(emitter) } | ||
|
Comment on lines
+68
to
+71
There was a problem hiding this comment. 🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- target outline ---'
ast-grep outline systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/sse/SseBroadcaster.kt || true
printf '%s\n' '--- target source ---'
cat -n systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/sse/SseBroadcaster.kt
printf '%s\n' '--- relevant symbols and scheduler configuration ---'
rg -n -S \
'SseBroadcaster|broadcastFrequent|broadcastResource|ping|Scheduled|TaskScheduler|ThreadPoolTaskScheduler|SseEmitter|\.send\(' \
systems gradle* build.gradle* settings.gradle* 2>/dev/null | head -n 300
printf '%s\n' '--- related tests ---'
rg -l -S 'SseBroadcaster|SseEmitter|broadcastFrequent|broadcastResource' . --glob '*Test.kt' --glob '*Tests.kt' --glob '*test*' | sortRepository: EntryDSM/entrydsm-platform Length of output: 12866 🌐 Web query:
💡 Result: SseEmitter writes to the HTTP response are blocking operations [1]. When using SseEmitter, these blocking writes are executed on a separate thread, which is managed by a configured AsyncTaskExecutor [1]. It is important to distinguish between the AsyncTaskExecutor used for SseEmitter and the TaskScheduler typically used for Citations: 🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- scheduling and executor configuration ---'
rg -n -S \
'`@EnableScheduling`|`@EnableAsync`|SchedulingConfigurer|configureTasks|TaskScheduler|TaskExecutor|AsyncTaskExecutor|ThreadPoolTaskScheduler|ThreadPoolTaskExecutor|spring\.task\.scheduling|scheduling' \
. --glob '!**/build/**' --glob '!**/node_modules/**' | head -n 400
printf '%s\n' '--- broadcaster call sites ---'
rg -n -S 'sseBroadcaster\.(register|publishLog)|SseLiveLogPublisher|MonitorStreamController' \
systems --glob '*.kt' | head -n 300
printf '%s\n' '--- stream and publisher implementations ---'
for f in $(fd -t f -e kt . systems/observability/observability-adapter-in/src/main/kotlin | \
xargs rg -l -S 'class MonitorStreamController|class SseLiveLogPublisher'); do
echo "--- $f"
cat -n "$f"
done
printf '%s\n' '--- test files in the observability adapter-in subsystem ---'
git ls-files 'systems/observability/observability-adapter-in' | \
rg '(^|/)(test|tests)(/|$)|Test\.kt$|Tests\.kt$' || trueRepository: EntryDSM/entrydsm-platform Length of output: 5359 🏁 Script executed: #!/bin/bash
set -eu
python3 - <<'PY'
from pathlib import Path
import re
target = Path("systems/observability/observability-adapter-in/src/main/kotlin/hs/kr/entrydsm/observability/adapterin/web/sse/SseBroadcaster.kt")
text = target.read_text()
checks = {
"scheduled methods call broadcast": bool(re.search(r'`@Scheduled`[\s\S]*?fun broadcastFrequent[\s\S]*?broadcast\(', text)),
"scheduled methods call emitter.send": bool(re.search(r'`@Scheduled`[\s\S]*?fun ping[\s\S]*?emitter\.send\(', text)),
"broadcast iterates emitters": bool(re.search(r'emitters\.forEach\s*\{\s*emitter\s*->[\s\S]*?emitter\.send\(', text)),
"send failures remove emitter": bool(re.search(r'\.onFailure\s*\{\s*emitters\.remove\(emitter\)', text)),
}
for name, result in checks.items():
print(f"{name}: {result}")
for m in re.finditer(r'(`@Scheduled`[^\n]*\n\s*fun\s+\w+\(\)[\s\S]*?)(?=\n\s*`@Scheduled`|\n\s*private fun|\n\s*companion object)', text):
block = m.group(1)
method = re.search(r'fun\s+(\w+)', block).group(1)
sends = len(re.findall(r'\.send\(', block))
broadcasts = len(re.findall(r'\bbroadcast\(', block))
print(f"{method}: send_calls={sends}, broadcast_calls={broadcasts}")
PYRepository: EntryDSM/entrydsm-platform Length of output: 454 🏁 Script executed: #!/bin/bash
set -eu
printf '%s\n' '--- application bootstrap ---'
cat -n systems/observability/observability-bootstrap/src/main/kotlin/hs/kr/entrydsm/ExampleApplication.kt
printf '%s\n' '--- Spring dependency versions ---'
rg -n -S \
'spring-boot|spring-framework|springBoot|org\.springframework' \
gradle.properties settings.gradle* build.gradle* systems --glob '*.gradle' --glob '*.gradle.kts' --glob '*.toml' --glob '*.properties' | head -n 250
printf '%s\n' '--- all scheduled methods ---'
rg -n -B 3 -A 12 '`@Scheduled`' . --glob '*.kt' --glob '*.java' --glob '!**/build/**'Repository: EntryDSM/entrydsm-platform Length of output: 6422 🌐 Web query:
💡 Result: By default, Spring's Citations: 🌐 Web query:
💡 Result: In Spring MVC, while ResponseBodyEmitter and its subclass SseEmitter support asynchronous streaming, individual writes to the response remain inherently blocking [1]. When you return an SseEmitter from a controller method, Spring MVC executes the writing process on a separate thread provided by the configured AsyncTaskExecutor [1]. This architecture ensures that the main servlet container thread is not blocked, allowing it to return to the thread pool while the SSE stream continues to process data [1]. Key points regarding the interaction between these components include: 1. Blocking Nature of Writes: Although the overall programming model is asynchronous, the actual I/O operations for each write to the response are blocking [1]. This is a fundamental difference between Spring MVC and Spring WebFlux; WebFlux utilizes non-blocking I/O and does not require a dedicated thread per write [1]. 2. Role of AsyncTaskExecutor: The AsyncTaskExecutor is responsible for handling the blocking writes when streaming, as well as executing tasks associated with asynchronous request processing (such as Callable or DeferredResult) [1]. 3. Performance Considerations: Spring explicitly notes that the default AsyncTaskExecutor is often not suitable for production environments under heavy load [1]. It is recommended to configure a custom TaskExecutor (e.g., using a ThreadPoolTaskExecutor) with appropriately tuned parameters to prevent thread exhaustion or performance degradation during streaming operations [1]. By offloading these blocking writes to a separate thread pool managed by the AsyncTaskExecutor, Spring MVC maintains scalability even while using blocking I/O for streaming responses [1]. Citations: 느린 SSE 구독자의 전송을 분리하십시오.
전용 bounded executor와 연결별 bounded outbound queue를 사용하십시오. 큐가 가득 차면 해당 emitter를 종료하십시오. 스케줄러 및 API 요청 스레드에서 직접 🤖 Prompt for AI Agents |
||
| } | ||
| } | ||
|
|
||
| private fun nextId(): String = eventIdSeq.incrementAndGet().toString() | ||
|
|
||
| companion object { | ||
| private const val RECONNECT_TIME_MS = 5000L | ||
| } | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.sse | ||
|
|
||
| import java.util.concurrent.ConcurrentHashMap | ||
| import java.util.concurrent.atomic.AtomicInteger | ||
| import org.springframework.stereotype.Component | ||
|
|
||
| /** | ||
| * ponytail: 단일 인스턴스 가정의 인메모리 카운터. 인증 붙기 전이라 계정 대신 IP 기준으로 제한한다. | ||
| * JWT 인증이 연결되면 계정 단위로 교체한다. | ||
| */ | ||
| @Component | ||
| class SseConnectionLimiter { | ||
| private val counts = ConcurrentHashMap<String, AtomicInteger>() | ||
|
|
||
| fun tryAcquire(key: String): Boolean { | ||
| val count = counts.computeIfAbsent(key) { AtomicInteger(0) } | ||
| if (count.incrementAndGet() <= MAX_CONNECTIONS_PER_KEY) return true | ||
| count.decrementAndGet() | ||
| return false | ||
| } | ||
|
|
||
| fun release(key: String) { | ||
| counts[key]?.decrementAndGet() | ||
| } | ||
|
|
||
| companion object { | ||
| private const val MAX_CONNECTIONS_PER_KEY = 3 | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,24 @@ | ||
| package hs.kr.entrydsm.observability.adapterin.web.sse | ||
|
|
||
| import hs.kr.entrydsm.observability.adapterin.web.dto.response.LiveLogEventResponse | ||
| import hs.kr.entrydsm.observability.application.port.out.ClientLogInput | ||
| import hs.kr.entrydsm.observability.application.port.out.LiveLogPublisherPort | ||
| import org.springframework.stereotype.Component | ||
|
|
||
| @Component | ||
| class SseLiveLogPublisher( | ||
| private val sseBroadcaster: SseBroadcaster, | ||
| ) : LiveLogPublisherPort { | ||
| override fun publishClientLog(input: ClientLogInput) { | ||
| sseBroadcaster.publishLog( | ||
| LiveLogEventResponse( | ||
| kind = "CLIENT", | ||
| level = input.level.name, | ||
| source = input.source.name, | ||
| message = input.message, | ||
| pageUrl = input.pageUrl, | ||
| occurredAt = input.occurredAt, | ||
| ), | ||
| ) | ||
| } | ||
| } |
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.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
Repository: EntryDSM/entrydsm-platform
Length of output: 21189
🏁 Script executed:
Repository: EntryDSM/entrydsm-platform
Length of output: 30115
🌐 Web query:
Spring Framework ResponseBodyEmitter SseEmitter onCompletion onTimeout onError callback lifecycle timeout error completion💡 Result:
In the Spring Framework, ResponseBodyEmitter and its subclass SseEmitter provide mechanisms for asynchronous request processing and streaming, where lifecycle events are managed via specific callback methods [1][2][3]. Lifecycle Callbacks The following methods allow applications to register hooks for specific lifecycle events: - onCompletion(Runnable callback): Registers a callback invoked when the async request completes for any reason, including successful completion, timeout, or network error [1][3]. This is primarily used to detect that the emitter instance is no longer usable [1]. - onTimeout(Runnable callback): Registers a callback invoked when the async request times out [1][3]. - onError(Consumer callback): Registers a callback invoked when an error occurs during async request processing [1][3]. These callbacks are executed from a container thread [1][3]. Timeout and Completion Behavior - Request Timeout: You can specify a custom timeout value (in milliseconds) via the constructor when creating a ResponseBodyEmitter or SseEmitter [1][2]. If not set, the framework uses the default configured in the MVC Java Config or the underlying servlet container [2][4]. - Completion: Applications can call complete or completeWithError(Throwable) to signal the end of request processing [1][5]. - Important Implementation Notes: - When an IOException occurs (e.g., the client disconnects), the application should not manually call complete or completeWithError. Instead, the servlet container initiates an AsyncListener notification, and Spring MVC automatically handles the cleanup [6]. - Recent Spring Framework versions (e.g., 6.2.x) have seen internal adjustments regarding when these flags are set and when callbacks are triggered to address issues with state management and callback invocation consistency [7][8][9]. Users should be aware that attempting to trigger completion within a callback (like onTimeout) may behave differently depending on the specific patch version of the framework [9]. For robust streaming, it is recommended to send periodic heartbeats (e.g., empty comments) because the Servlet API does not provide immediate notification when a remote client disappears, and writes will fail if the client has disconnected [6].
Citations:
SseEmitter.onCompletion()behavior between Spring 6.2.3 and 6.2.5 spring-projects/spring-framework#34762🏁 Script executed:
Repository: EntryDSM/entrydsm-platform
Length of output: 324
🏁 Script executed:
Repository: EntryDSM/entrydsm-platform
Length of output: 324
연결 슬롯을 한 번만 해제하십시오.
ResponseBodyEmitter는 timeout 또는 error 뒤에도onCompletion을 호출할 수 있습니다. 현재release가 중복 실행되어 카운터가 음수가 되고, IP별 연결 제한 3개를 초과할 수 있습니다.onCompletion에서만 해제하거나AtomicBoolean으로 중복 실행을 방지하십시오. timeout과 error 종료 시 카운터가 정확히 한 번 감소하는 테스트도 추가하십시오.🤖 Prompt for AI Agents