Add Checksum Basic Function to Support Later Checksum Validation for Heartbeats #3930
Conversation
ti-chi-bot
bot
added
release-note
Summary of ChangesHello @hongyunyan, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request lays the foundational groundwork for robust checksum validation within the system. It defines the necessary data structures and communication protocols for tracking and verifying dispatcher set checksums across different nodes. The changes enable the maintainer to manage the expected state of dispatcher sets and react to discrepancies reported via heartbeats, which is crucial for ensuring data consistency and operational stability in a distributed environment. Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces the foundational components for checksum validation in heartbeats, including new protobuf definitions and a nodeSetChecksumManager to manage checksum state on the maintainer side. The code is well-structured and provides a solid base for the feature. I have identified a few areas for improvement, such as potentially dead code, a redundant check, and a minor style issue in the protobuf definition. Additionally, a typo was found in a constant used within the new test files.
| int64 mode = 3; | ||
| uint64 seq = 4; | ||
| DispatcherSetChecksum checksum = 5; | ||
| } |
There was a problem hiding this comment.
The file should end with a newline character. It's a common convention and some tools might expect it.
| if exists { | ||
| if oldCapture == capture { | ||
| log.Warn("dispatcher already exists in expected set, ignore it", | ||
| zap.Stringer("changefeedID", m.changefeedID), | ||
| zap.String("dispatcherID", id.String()), | ||
| zap.String("capture", capture.String()), | ||
| zap.String("mode", common.StringMode(m.mode)), | ||
| ) | ||
| continue | ||
| } | ||
| log.Warn("dispatcher exists in another capture, override expected node", | ||
| zap.Stringer("changefeedID", m.changefeedID), | ||
| zap.String("dispatcherID", id.String()), | ||
| zap.String("oldCapture", oldCapture.String()), | ||
| zap.String("newCapture", capture.String()), | ||
| zap.String("mode", common.StringMode(m.mode)), | ||
| ) | ||
| if oldState, ok := m.state.captures[oldCapture]; ok { | ||
| oldState.checksum.Remove(id) | ||
| } | ||
| } |
There was a problem hiding this comment.
This if exists block appears to be dead code. applyExpectedSet is only called from ResetAndSendFull, which resets m.state just before this call. Consequently, m.state.dispatcherToNode will be empty, and exists will always evaluate to false.
If this logic is intended for future use, please add a comment to clarify. Otherwise, it should be removed to prevent confusion.
| if exists && oldCapture == capture { | ||
| continue | ||
| } | ||
| if exists && oldCapture != "" && oldCapture != capture { |
There was a problem hiding this comment.
The check oldCapture != "" seems redundant. If exists is true, oldCapture should already be a valid, non-empty capture ID. This check can be removed to simplify the code.
| if exists && oldCapture != "" && oldCapture != capture { | |
| if exists && oldCapture != capture { |
|
|
||
| func TestDispatcherSetChecksumResendAndAck(t *testing.T) { | ||
| mgr := newNodeSetChecksumManager( | ||
| common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceNamme), |
There was a problem hiding this comment.
There appears to be a typo in the constant common.DefaultKeyspaceNamme. It should likely be common.DefaultKeyspaceName. Please consider correcting the constant definition in pkg/common/types.go.
|
|
||
| func TestCaptureSetChecksumManagerRemoveNodesCleansState(t *testing.T) { | ||
| mgr := newNodeSetChecksumManager( | ||
| common.NewChangeFeedIDWithName("test", common.DefaultKeyspaceNamme), |
There was a problem hiding this comment.
There appears to be a typo in the constant common.DefaultKeyspaceNamme. It should likely be common.DefaultKeyspaceName. Please consider correcting the constant definition in pkg/common/types.go.
|
/gemini review |
|
/gemini summary |
There was a problem hiding this comment.
Code Review
This pull request introduces a checksum validation mechanism for heartbeats to enhance data integrity, alongside fixing a widespread typo (DefaultKeyspaceNamme to DefaultKeyspaceName). The core of the new functionality lies in the new node_set_checksum_manager.go, which manages checksums for node sets, and the corresponding protobuf definitions. My review identifies a critical issue in how protobuf messages are compared, which could lead to incorrect behavior, and also points out some minor inconsistencies and typos for improvement.
| if ack.ChangefeedID != m.changefeedID.ToPB() { | ||
| return | ||
| } |
There was a problem hiding this comment.
The comparison ack.ChangefeedID != m.changefeedID.ToPB() is incorrect because it compares the memory addresses of the two pointers, not their contents. This will likely always evaluate to true, causing valid acknowledgements to be ignored. You should use proto.Equal to perform a deep comparison of protobuf messages.
You'll also need to add "github.com/gogo/protobuf/proto" to your imports for this change.
| if ack.ChangefeedID != m.changefeedID.ToPB() { | |
| return | |
| } | |
| if !proto.Equal(ack.ChangefeedID, m.changefeedID.ToPB()) { | |
| return | |
| } |
Summary of ChangesThis pull request establishes the foundational components for implementing checksum validation for heartbeats within the system. By defining new protobuf structures and introducing a dedicated Highlights
🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Changelog
Activity
|
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
…igate into 0105-checksum-basic
|
/test all |
|
@hongyunyan: The following test failed, say
Full PR test history. Your PR dashboard. DetailsInstructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here. |
| ) (shouldWarn bool, duration time.Duration) { | ||
| if state == heartbeatpb.ChecksumState_MATCH { | ||
| s.lastObservedState = state | ||
| s.nonMatchSince = time.Time{} |
There was a problem hiding this comment.
Is time.Time{} a standard way to initialize the time.Time ?
How about use the time.Now() at the very first, and then set it to all related fields.
There was a problem hiding this comment.
Yes, and it's more like a reset action
| } | ||
|
|
||
| needResetTimer := s.lastObservedState == heartbeatpb.ChecksumState_MATCH || s.lastObservedState != state | ||
| if needResetTimer || s.nonMatchSince.IsZero() { |
There was a problem hiding this comment.
|| s.nonMatchSince.IsZero() can be combined into the condition check above.
| } | ||
| for _, id := range ids { | ||
| oldNode, exists := m.state.dispatcherToNode[id] | ||
| if exists { |
There was a problem hiding this comment.
if !exists just continue first.
[LGTM Timeline notifier]Timeline:
|
| return | ||
| } | ||
|
|
||
| stateStr := "mismatch" |
There was a problem hiding this comment.
It is better to use a constant here.
|
@3AceShowHand: Your lgtm message is repeated, so it is ignored. DetailsIn response to this: Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. |
|
[APPROVALNOTIFIER] This PR is APPROVED This pull-request has been approved by: 3AceShowHand The full list of commands accepted by this bot can be found here. The pull request process is described here DetailsNeeds approval from an approver in each of these files:
Approvers can indicate their approval by writing |
| ChangefeedID changefeedID = 1; | ||
| uint64 epoch = 2; | ||
| int64 mode = 3; | ||
| uint64 seq = 4; |
There was a problem hiding this comment.
Why do we need epochs and seq? Can't a single seq be sufficient?
| m.mu.Lock() | ||
| nodeState, ok := m.state.nodes[from] | ||
| if !ok { | ||
| m.mu.Unlock() |
There was a problem hiding this comment.
use defer m.mu.Unlock() above
| changefeed: m.changefeedID, | ||
| node: from, | ||
| } | ||
| m.mu.Unlock() |
There was a problem hiding this comment.
use defer m.mu.Unlock() above
| return msgs | ||
| } | ||
|
|
||
| func (m *nodeSetChecksumManager) applyExpectedSet(expected map[node.ID][]common.DispatcherID) { |
There was a problem hiding this comment.
Add a comment to inform that a lock is required before use.
4 participants
What problem does this PR solve?
Issue Number: ref #3848
What is changed and how it works?
This pull request establishes the foundational components for implementing checksum validation for heartbeats within the system. By defining new protobuf structures and introducing a dedicated
nodeSetChecksumManager, it enables the maintainer to track and verify the consistency of dispatcher sets across different nodes. This mechanism is crucial for proactively identifying and addressing discrepancies, thereby enhancing the overall data integrity and operational stability of the distributed environment.Highlights
ChecksumStateenum and messages likeChecksumMeta,DispatcherSetChecksum,DispatcherSetChecksumAckResponse, andDispatcherSetChecksumUpdateRequestto support checksum validation.nodeSetChecksumManagerto manage expected dispatcher set checksums on the maintainer side, including logic for applying incremental changes, flushing updates, handling acknowledgments, and resending messages.IOTypeconstants and handling their encoding/decoding.DefaultKeyspaceNammetoDefaultKeyspaceNameacross various files, improving code consistency.nodeSetChecksumManagerto ensure the reliability of checksum tracking and communication logic.Check List
Tests
Questions
Will it cause performance regression or break compatibility?
Do you need to update user documentation, design documentation or monitoring documentation?
Release note