Skip to content
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.

Sign up for GitHub

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

log nr of streamed bytes every 30 seconds in follower #885

Merged
merged 2 commits into from
Dec 17, 2024
Merged

log nr of streamed bytes every 30 seconds in follower #885

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
784bdfa
log nr of streamed bytes every 30 seconds in follower
viktorerlingsson Dec 16, 2024
59d569a
update changelog
viktorerlingsson Dec 17, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Queues will no longer be closed if file size is incorrect. Fixes [#669](https://github.com/cloudamqp/lavinmq/issues/669)
- Dont redeclare exchange in java client test [#860](https://github.com/cloudamqp/lavinmq/pull/860)

### Added

- Added some logging for followers [#885](https://github.com/cloudamqp/lavinmq/pull/885)

## [2.0.2] - 2024-11-25

### Fixed
Expand Down
10 changes: 10 additions & 0 deletions src/lavinmq/clustering/client.cr
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ module LavinMQ
@unix_amqp_proxy : Proxy?
@unix_http_proxy : Proxy?
@socket : TCPSocket?
@streamed_bytes = 0_u64

def initialize(@config : Config, @id : Int32, @password : String, proxy = true)
System.maximize_fd_limit
Expand Down Expand Up @@ -209,6 +210,7 @@ module LavinMQ
private def stream_changes(socket, lz4)
acks = Channel(Int64).new(@config.clustering_max_unsynced_actions)
spawn send_ack_loop(acks, socket), name: "Send ack loop"
spawn log_streamed_bytes_loop, name: "Log streamed bytes loop"
loop do
filename_len = lz4.read_bytes Int32, IO::ByteFormat::LittleEndian
next if filename_len.zero?
Expand Down Expand Up @@ -236,6 +238,7 @@ module LavinMQ
@files.delete("#{filename}.tmp").try &.close
end
ack_bytes = len.abs + sizeof(Int64) + filename_len + sizeof(Int32)
@streamed_bytes &+= ack_bytes
acks.send(ack_bytes)
end
ensure
Expand All @@ -256,6 +259,13 @@ module LavinMQ
socket.close rescue nil
end

private def log_streamed_bytes_loop
loop do
sleep 30.seconds
Log.info { "Total streamed bytes: #{@streamed_bytes}" }
end
end

private def authenticate(socket)
socket.write Start
socket.write_bytes @password.bytesize.to_u8, IO::ByteFormat::LittleEndian
Expand Down
Loading