[event] add optimizer event stream publisher - #289
Conversation
![qoderai[bot]](https://avatars.githubusercontent.com/in/1396158?s=60&v=4){kind=small}
There was a problem hiding this comment.
Overall this is a clean, well-tested addition. The sink/publisher split is a good seam, the framing protocol is simple and bounded, and the tests cover the handshake, backpressure, and drop paths thoroughly.
The main things I'd like to see addressed before merging:
- SIGPIPE risk.
tcp_event_sink.ccuses::write()on TCP sockets; a peer disconnect can raiseSIGPIPEand terminate the worker/IO thread unless the process ignores it globally. Using::send(..., MSG_NOSIGNAL)is the safer fix. - Binding address. The event port binds to
INADDR_ANY. For an internal diagnostic channel, defaulting to loopback or makingbind_addressconfigurable is safer. - Lock contention in
TcpEventSink::Send(). The mutex is held while framing and writing to every consumer. Framing outside the lock and copying ready FDs before writing would reduce contention with the IO loop.
Minor nits are noted inline (queue-size cap, an avoidable copy, etc.).
🤖 Generated by Qoder
| TcpEventSink::WriteResult TcpEventSink::WriteFrame(int fd, const std::string &wire) { | ||
| std::size_t sent = 0; | ||
| while (sent < wire.size()) { | ||
| const ssize_t n = ::write(fd, wire.data() + sent, wire.size() - sent); |
There was a problem hiding this comment.
Writing to a TCP socket with ::write() can raise SIGPIPE if the peer has closed its half of the connection. Since the process does not appear to ignore SIGPIPE globally, a consumer disconnecting mid-send could terminate the publisher worker or IO thread. Use ::send(fd, ..., MSG_NOSIGNAL) here, or ensure SIGPIPE is ignored process-wide.
🤖 Generated by Qoder
|
|
||
| sockaddr_in addr{}; | ||
| addr.sin_family = AF_INET; | ||
| addr.sin_addr.s_addr = htonl(INADDR_ANY); |
There was a problem hiding this comment.
Binding the optimizer event stream to INADDR_ANY exposes this port on all network interfaces. Because this is a diagnostic/replay channel, consider defaulting to INADDR_LOOPBACK or adding a bind_address config option so it is not reachable from untrusted networks by default.
🤖 Generated by Qoder
There was a problem hiding this comment.
key is an int64_t, so the range-for currently copies each value. Use const auto &key to avoid the unnecessary copy.
🤖 Generated by Qoder
| return true; | ||
| }; | ||
|
|
||
| if (!read_positive_size("queue_size", &parsed.queue_size, std::numeric_limits<std::size_t>::max()) || |
There was a problem hiding this comment.
Capping queue_size at size_t::max() means a malformed config can request an impossibly large queue and cause bad_alloc during startup. Consider bounding this to a sane operational maximum (e.g., a few million).
🤖 Generated by Qoder
| bool delivered = false; | ||
| std::vector<int> closable; | ||
| { | ||
| std::lock_guard<std::mutex> lock(connections_mutex_); |
There was a problem hiding this comment.
Send() holds connections_mutex_ while it frames the payload and writes to every ready socket. This serializes the publisher worker with the IO loop and prevents it from processing handshakes/disconnects while a large frame is being pushed. Consider framing the payload outside the lock, copying the ready FDs, and writing without holding the mutex.
🤖 Generated by Qoder
baaf630 to
b65d0d6
Compare
b65d0d6 to
0858baa
Compare
|
@codex review |
There was a problem hiding this comment.
The redesign from a dedicated TCP sink to a gRPC server-streaming service on the existing meta port cleanly addresses the SIGPIPE, bind-address, and Send() lock-contention issues raised in the earlier review. The publisher/sink split and per-subscriber bounded queues are easy to reason about, and the test coverage is good.
A few things worth considering before merging:
SubscriptionEventSink::Send()incrementsdropped_per subscriber that fails to enqueue, so a single event can inflate the drop counter by up tomax_subscriberswhileSend()still returnstrue. Make the metric semantics explicit, or count a drop only when no subscriber accepted the event.OptimizerEventServiceGRpc::SubscribeEvents()can returnRESOURCE_EXHAUSTEDifStop()races between thestopped()check andSubscribe(). In that caseUNAVAILABLEwould be more accurate.GetConfiguration()returns topology without a leader check. If an optimizer connects to a standby, it may receive stale or empty configuration and no events. Consider enforcing the same leader-only behavior as other meta requests, or documenting that optimizers must target the leader.- A malformed
event_publishers_configsstring disables both the log and optimizer publishers. If only the optimizer section is invalid, the log publisher is still taken down; consider isolating the failure. - The new
OptimizerEventStreamServicestreams cache-read events and full topology on the existing insecure gRPC port. This is consistent with the current Meta/Admin services, but confirm that network segmentation is sufficient for this new data channel. - One prior nit still remains:
OptimizerEventPublisher::Convert()copies each block key by value in the range-for loop; usingconst auto &keyavoids the copy.
🤖 Generated by Qoder
🤖 Generated by Qoder
| if (subscription->Enqueue(event)) { | ||
| delivered = true; | ||
| } else { | ||
| dropped_.fetch_add(1); |
There was a problem hiding this comment.
dropped_ is incremented once for every subscriber whose queue is full, and again when there are no subscribers at all. That means a single event can inflate dropped_ by up to max_subscribers, and Send() can return true while dropped_ grows. If this counter is meant to be compared against forwarded_, the per-subscriber semantics are surprising. Consider documenting it explicitly or counting a drop only when no subscriber accepted the event.
🤖 Generated by Qoder
| return grpc::Status(grpc::StatusCode::UNAVAILABLE, "optimizer event publisher is unavailable"); | ||
| } | ||
| auto subscription = sink_->Subscribe(request->consumer_id()); | ||
| if (!subscription) { |
There was a problem hiding this comment.
sink_->stopped() and sink_->Subscribe() are not atomic. If Stop() is called between the two, Subscribe() returns nullptr and the RPC reports RESOURCE_EXHAUSTED even though the sink is unavailable. Returning UNAVAILABLE when stopped_ is true would be more accurate; consider checking sink_->stopped() again after a failed subscribe.
🤖 Generated by Qoder
| return grpc::Status::OK; | ||
| } | ||
|
|
||
| RequestContext request_context(request->trace_id()); |
There was a problem hiding this comment.
GetConfiguration returns the KVCM topology without checking whether this node is the leader. If an optimizer connects to a standby, it may receive stale or empty configuration and no events. Consider either enforcing a leader check (consistent with other meta requests) or documenting that optimizers must target the leader.
🤖 Generated by Qoder
| void Server::RegisterEventPublishers(const std::shared_ptr<EventManager> &event_manager) { | ||
| const auto &event_publishers_configs = config_.event_publishers_configs(); | ||
| EventPublishersConfig publishers_config; | ||
| if (!event_publishers_configs.empty() && !publishers_config.FromJsonString(event_publishers_configs)) { |
There was a problem hiding this comment.
A malformed event_publishers_configs string disables both the log and optimizer publishers. If only the optimizer config is invalid, the log publisher is still taken down. Consider falling back to the log publisher when the optimizer section is the only invalid part, or at least logging which publisher is affected.
🤖 Generated by Qoder
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: dffe0bd623
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| } | ||
| } | ||
|
|
||
| status->set_code(proto::optimizer::OK); |
There was a problem hiding this comment.
During startup, and permanently on a follower, RegistryManager::ListInstanceGroup can legitimately return an empty or partially recovered in-memory registry, yet this method reports OK. Server::Start exposes the RPC before starting leader election, while registry recovery happens later in OnBecomeLeader; RegistryManager already exposes IsRecoverComplete() and the protocol defines SERVICE_NOT_READY/SERVER_NOT_LEADER. An optimizer that bootstraps in this window can accept an empty snapshot and then discard streamed events for unknown instances, so return a non-OK application status until the node is the recovered leader.
Useful? React with 👍 / 👎.
| // Single threaded on purpose: the replay requires per-instance events to | ||
| // arrive in non-decreasing timestamp order, and one worker draining one | ||
| // queue preserves that ordering. |
There was a problem hiding this comment.
With concurrent successful cache reads for the same instance, one serving thread can call SetEventTriggerTime() first but reach BasicEnqueue() after another thread, so FIFO queue order can contain decreasing timestamps. A single worker preserves only enqueue order, not the claimed timestamp order, and the replay/TTL logic assumes non-decreasing timestamps; this can reorder accesses or evaluate expirations against time moving backward. Serialize or clamp ordering per instance before sending events.
Useful? React with 👍 / 👎.
| if (subscription->Enqueue(event)) { | ||
| delivered = true; | ||
| } else { | ||
| dropped_.fetch_add(1); |
There was a problem hiding this comment.
When a subscriber is slower than subscriber_queue_size, this path silently drops the event and only increments an in-process counter that is neither included in the stream nor registered as a metric; it is otherwise logged only when the sink stops. The optimizer therefore continues producing capacity curves from an incomplete trace without knowing that samples were lost. Surface per-subscriber loss through stream metadata, sequence gaps, or operational metrics so affected results can be rejected.
Useful? React with 👍 / 👎.
| if (!sink_ || sink_->stopped()) { | ||
| return grpc::Status(grpc::StatusCode::UNAVAILABLE, "optimizer event publisher is unavailable"); |
There was a problem hiding this comment.
A follower accepts SubscribeEvents whenever the process-local sink is running, and an existing stream is not closed when OnNoLongerLeader disables leader-only cache reads. Consequently, after failover the optimizer can remain attached to the old leader on a healthy but permanently idle stream and never reconnect to the new event producer; the same occurs if it initially contacts a follower. Reject subscriptions unless the node is the recovered leader and close active subscriptions on demotion so clients receive an error and reconnect.
Useful? React with 👍 / 👎.
| "//kv_cache_manager/common:logger", | ||
| "//kv_cache_manager/event/optimizer_stream:event_sink", | ||
| "//kv_cache_manager/event/spec_events", | ||
| "//kv_cache_manager/protocol/protobuf:service_cc_proto", |
There was a problem hiding this comment.
This target introduces a direct event → protocol module dependency, but neither the root dependency diagram nor docs/design/module_architecture.md was updated; the latter still depicts event as depending only on common. Update both architecture diagrams and the event module description so future dependency checks account for this new edge.
AGENTS.md reference: AGENTS.md:L38-L38
Useful? React with 👍 / 👎.
1 participant
Summary
OptimizerEventStreamService.SubscribeEventsas a server-streaming gRPC API on KVCM's existing Meta RPC portlogandoptimizerpublisher configuration, keeping log publishing enabled by default and optimizer publishing opt-inConfiguration
Enabling the optimizer publisher does not add a listening port. Optimizer connects to KVCM's existing
kvcm.service.rpc_portand callsSubscribeEvents.Testing
VcnsHf3fsAllocatorTestbuild is blocked by a missinghiredis.hdependency outside this change