fix(loki-stream): reconnect when websocket is unexpectedly closed#1287
Conversation
|
Warning Review limit reached
More reviews will be available in 51 minutes and 30 seconds. Learn how PR review limits work. Your organization has used up its prepaid credits, and credit purchases are no longer available. Enable the review add-on in the billing tab to keep reviews running — you're only billed for reviews past your plan's rate limits ($0.25/file). ⌛ How to resolve this issue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based credits. 🚦 How do rate limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan refill rate. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, the refill rate gradually slows as usage increases. The highest same-day bursts are limited more strictly. Please see our Fair Usage Limits Policy for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
📝 WalkthroughWalkthrough
ChangesWebSocket ライブラリ移行と Stream イテレータ再実装
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~20 minutes 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Preview (prod backend + PR dashboard) → https://1287.ns-preview.trapti.tech/ |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/infrastructure/log/loki/streamer.go`:
- Around line 134-137: The backoffCount variable is incremented on read errors
(around line 156) but never reset, causing the streaming loop to exit
permanently after 5 disconnections even if those are normal reconnections. To
fix this, reset backoffCount to 0 whenever a message is successfully received
and processed (after handling MessageText), so that only consecutive errors
count toward the maximum backoff limit. Additionally, add a sleep duration
between retry attempts before the next iteration of the for loop checking
backoffCount, implementing actual exponential backoff behavior rather than
immediate retries that would hammer the server.
- Around line 173-181: The unconditional blocking send `ch <- l` in the loop
will cause a goroutine leak when context is canceled. When the channel buffer
(capacity 100) fills up and ctx.Done() is triggered, the send blocks forever and
prevents `defer close(ch)` from executing. Replace the blocking send with a
select statement that includes both the channel send case and a `<-ctx.Done()`
case, allowing the goroutine to exit gracefully and close the channel when the
context is canceled.
- Line 151: The conn.Read call in the streamer.go file uses the default 32KiB
read limit from coder/websocket, which can cause connection failures when Loki
tail batches exceed this size. Before calling conn.Read(ctx), increase the read
limit using conn.SetReadLimit(...) with an appropriate higher value to
accommodate larger Loki batch messages and prevent frequent reconnections due to
read errors.
- Around line 144-158: The WebSocket connection created by websocket.Dial in the
streamer.go file is not being explicitly closed, causing resource leaks. The
conn object needs to be closed in all code paths: when context.Canceled causes
an early return on line 153, and before breaking from the loop on line 157 to
avoid accumulating multiple unclosed connections during reconnect attempts. Use
a defer statement immediately after successfully dialing the WebSocket endpoint,
or ensure conn.Close() or conn.CloseNow() is called before each return and
before the break statement that triggers reconnection attempts. The
coder/websocket v1.8.x library requires explicit closure, so every connection
established by websocket.Dial must be properly cleaned up.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 907543c7-85f0-4e9e-aa31-d50e34a5e0aa
⛔ Files ignored due to path filters (1)
go.sumis excluded by!**/*.sum
📒 Files selected for processing (2)
go.modpkg/infrastructure/log/loki/streamer.go
| continue // fail-safe | ||
|
|
||
| for { | ||
| typ, b, err := conn.Read(ctx) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor
🧩 Analysis chain
🌐 Web query:
What is the default read message size limit in github.com/coder/websocket Conn, and how is it changed?
💡 Result:
The default read message size limit in the github.com/coder/websocket library is 32,768 bytes (32 KB) [1][2][3]. You can change or disable this limit using the SetReadLimit(n int64) method on the connection object [1][3]. Setting this value to -1 will completely disable the read message size limit [1][2][3]. When a message exceeds the configured limit, the connection will be closed with a StatusMessageTooBig error [1][3]. Starting from recent versions, you can also detect this condition by checking if the returned error wraps ErrMessageTooBig [4][1].
Citations:
- 1: coder/websocket@7d7c644
- 2: message reader has a default read limit of 32768, net.Conn wrapper doesn't limit size of messages coder/websocket#382
- 3: https://github.com/nhooyr/websocket-old/blob/master/read.go
- 4: refactor: add
ErrMessageTooBigsentinel error for limited reads coder/websocket#535
🏁 Script executed:
find . -name "streamer.go" -type f | head -20Repository: traPtitech/NeoShowcase
Length of output: 252
🏁 Script executed:
cat -n ./pkg/infrastructure/log/loki/streamer.go | head -160 | tail -30Repository: traPtitech/NeoShowcase
Length of output: 1119
🏁 Script executed:
cat -n ./pkg/infrastructure/log/loki/streamer.go | sed -n '144,200p'Repository: traPtitech/NeoShowcase
Length of output: 1641
🏁 Script executed:
grep -n "SetReadLimit\|websocket.Dial" ./pkg/infrastructure/log/loki/streamer.goRepository: traPtitech/NeoShowcase
Length of output: 243
🏁 Script executed:
head -30 ./pkg/infrastructure/log/loki/streamer.goRepository: traPtitech/NeoShowcase
Length of output: 742
conn.Read のデフォルト読み取り上限（32768 バイト）に注意してください。
coder/websocket の Conn はデフォルトで 32KiB の読み取り上限を持ち、これを超えるメッセージは読み取りエラーになりコネクションが切断されます。Loki の tail バッチが大きい場合に頻繁な再接続を誘発する恐れがあります。必要に応じて conn.SetReadLimit(...) で上限を引き上げてください。
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@pkg/infrastructure/log/loki/streamer.go` at line 151, The conn.Read call in
the streamer.go file uses the default 32KiB read limit from coder/websocket,
which can cause connection failures when Loki tail batches exceed this size.
Before calling conn.Read(ctx), increase the read limit using
conn.SetReadLimit(...) with an appropriate higher value to accommodate larger
Loki batch messages and prevent frequent reconnections due to read errors.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
pkg/infrastructure/log/loki/streamer.go (1)
142-158: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueレシーバ
lをループ変数lでシャドウしており可読性・保守性を損ねます。
for l, err := range logSeqのループ変数lがlokiStreamerレシーバlを隠蔽しています。現状はループ内でレシーバのメソッドを呼ばないため動作しますが、将来l.xxx()を追加した際に意図せぬ参照ミスを招きます。ログ要素はlog等へ改名することを推奨します。♻️ 改名例
- for l, err := range logSeq { + for log, err := range logSeq { if errors.Is(err, context.Canceled) { return } else if err != nil { slog.WarnContext(ctx, "unexpected error occurred while reading log stream", "error", err) break // retry } - if l.Time.After(lastSeenTime) { - lastSeenTime = l.Time - } + if log.Time.After(lastSeenTime) { + lastSeenTime = log.Time + } select { - case ch <- l: + case ch <- log: case <-ctx.Done(): return } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@pkg/infrastructure/log/loki/streamer.go` around lines 142 - 158, The loop variable `l` in the `for l, err := range logSeq` statement shadows the receiver `l` of the method, creating variable shadowing that reduces code maintainability and could cause reference errors if receiver methods are added in the future. Rename the loop variable from `l` to a more descriptive name such as `log` and update all references to this variable within the loop block (including `l.Time` which should become `log.Time` and the send operation `ch <- l` which should become `ch <- log`).
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@pkg/infrastructure/log/loki/streamer.go`:
- Around line 150-158: In the log streaming logic within the `readStream`
method, logs at or before the `lastSeenTime` are being re-fetched from Loki on
reconnection because the current implementation lacks a filter to skip
already-sent logs. The code currently only updates `lastSeenTime` when a log's
timestamp is after `lastSeenTime`, but does not skip logs with timestamps before
or equal to `lastSeenTime`. Add a condition that checks if
`l.Time.Before(lastSeenTime)` or if the times are equal, and skip sending those
logs (use `continue` to move to the next iteration) before the `select`
statement that sends logs to the channel. This ensures that previously delivered
logs are not duplicated during reconnections triggered by `tail_max_duration`.
---
Nitpick comments:
In `@pkg/infrastructure/log/loki/streamer.go`:
- Around line 142-158: The loop variable `l` in the `for l, err := range logSeq`
statement shadows the receiver `l` of the method, creating variable shadowing
that reduces code maintainability and could cause reference errors if receiver
methods are added in the future. Rename the loop variable from `l` to a more
descriptive name such as `log` and update all references to this variable within
the loop block (including `l.Time` which should become `log.Time` and the send
operation `ch <- l` which should become `ch <- log`).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: b35bc372-fa38-4b1d-b135-94c873012c23
📒 Files selected for processing (1)
pkg/infrastructure/log/loki/streamer.go
pirosiki197
force-pushed
the
fix/loki-stream
branch
from
4d122ff to
ed89b38
Compare
1 participant
なぜやるか
loki側の
tail_max_durationを過ぎても自動で再接続するようにしたいやったこと
再接続するようにした
ついでにwebsocketのライブラリもメンテナンスされているものに変えた
やらなかったこと
資料
Summary by CodeRabbit
Release Notes
Chores（保守）
Refactor（リファクター）