CBG-5013 move gocb dcp client to separate file #7917
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -40,7 +40,7 @@ type endStreamCallbackFunc func(e endStreamEvent) | |
|
|
||
| var ErrVbUUIDMismatch = errors.New("VbUUID mismatch when failOnRollback set") | ||
|
|
||
| type GoCBDCPClient struct { | ||
| ctx context.Context | ||
| ID string // unique ID for DCPClient - used for DCP stream name, must be unique | ||
| agent *gocbcore.DCPAgent // SDK DCP agent, manages connections and calls back to DCPClient stream observer implementation | ||
|
|
@@ -81,7 +81,7 @@ type DCPClientOptions struct { | |
| CheckpointPrefix string | ||
| } | ||
|
|
||
| func NewDCPClient(ctx context.Context, ID string, callback sgbucket.FeedEventCallbackFunc, options DCPClientOptions, bucket *GocbV2Bucket) (*GoCBDCPClient, error) { | ||
torcolvin marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
|
||
| numVbuckets, err := bucket.GetMaxVbno() | ||
| if err != nil { | ||
|
|
@@ -91,7 +91,7 @@ func NewDCPClient(ctx context.Context, ID string, callback sgbucket.FeedEventCal | |
| return newDCPClientWithForBuckets(ctx, ID, callback, options, bucket, numVbuckets) | ||
| } | ||
|
|
||
| func newDCPClientWithForBuckets(ctx context.Context, ID string, callback sgbucket.FeedEventCallbackFunc, options DCPClientOptions, bucket *GocbV2Bucket, numVbuckets uint16) (*GoCBDCPClient, error) { | ||
|
|
||
| numWorkers := DefaultNumWorkers | ||
| if options.NumWorkers > 0 { | ||
|
|
@@ -106,7 +106,7 @@ func newDCPClientWithForBuckets(ctx context.Context, ID string, callback sgbucke | |
| return nil, fmt.Errorf("callers must specify a checkpoint prefix when persisting metadata") | ||
| } | ||
| } | ||
| client := &GoCBDCPClient{ | ||
| ctx: ctx, | ||
| workers: make([]*DCPWorker, numWorkers), | ||
| numVbuckets: numVbuckets, | ||
|
|
@@ -155,7 +155,7 @@ func newDCPClientWithForBuckets(ctx context.Context, ID string, callback sgbucke | |
| } | ||
|
|
||
| // getCollectionHighSeqNo returns the highSeqNo for a given KV collection ID. | ||
| func (dc *GoCBDCPClient) getCollectionHighSeqNos(collectionID uint32) ([]uint64, error) { | ||
| vbucketSeqnoOptions := gocbcore.GetVbucketSeqnoOptions{} | ||
| if dc.supportsCollections { | ||
| vbucketSeqnoOptions.FilterOptions = &gocbcore.GetVbucketSeqnoFilterOptions{CollectionID: collectionID} | ||
|
|
@@ -213,7 +213,7 @@ func (dc *DCPClient) getCollectionHighSeqNos(collectionID uint32) ([]uint64, err | |
| } | ||
|
|
||
| // getHighSeqNos returns the maximum sequence number for every collection configured by the DCP agent. | ||
| func (dc *GoCBDCPClient) getHighSeqNos() ([]uint64, error) { | ||
| highSeqNos := make([]uint64, dc.numVbuckets) | ||
| // Initialize highSeqNo to the current metadata's StartSeqNo - we don't want to use a value lower than what | ||
| // we've already processed | ||
|
|
@@ -235,7 +235,7 @@ func (dc *DCPClient) getHighSeqNos() ([]uint64, error) { | |
| } | ||
|
|
||
| // configureOneShot sets highSeqnos for a one shot feed. | ||
| func (dc *GoCBDCPClient) configureOneShot() error { | ||
| highSeqNos, err := dc.getHighSeqNos() | ||
| if err != nil { | ||
| return err | ||
|
|
@@ -251,7 +251,7 @@ func (dc *DCPClient) configureOneShot() error { | |
| } | ||
|
|
||
| // Start returns an error and a channel to indicate when the DCPClient is done. If Start returns an error, DCPClient.Close() needs to be called. | ||
| func (dc *GoCBDCPClient) Start() (doneChan chan error, err error) { | ||
| err = dc.initAgent(dc.spec) | ||
| if err != nil { | ||
| return dc.doneChannel, err | ||
|
|
@@ -274,13 +274,13 @@ func (dc *DCPClient) Start() (doneChan chan error, err error) { | |
| } | ||
|
|
||
| // Close is used externally to stop the DCP client. If the client was already closed due to error, returns that error | ||
| func (dc *GoCBDCPClient) Close() error { | ||
| dc.close() | ||
| return dc.getCloseError() | ||
| } | ||
|
|
||
| // GetMetadata returns metadata for all vbuckets | ||
| func (dc *GoCBDCPClient) GetMetadata() []DCPMetadata { | ||
| metadata := make([]DCPMetadata, dc.numVbuckets) | ||
| for i := uint16(0); i < dc.numVbuckets; i++ { | ||
| metadata[i] = dc.metadata.GetMeta(i) | ||
|
|
@@ -290,7 +290,7 @@ func (dc *DCPClient) GetMetadata() []DCPMetadata { | |
|
|
||
| // close is used internally to stop the DCP client. Sends any fatal errors to the client's done channel, and | ||
| // closes that channel. | ||
| func (dc *GoCBDCPClient) close() { | ||
|
|
||
| // set dc.closing to true, avoid re-triggering close if it's already in progress | ||
| if !dc.closing.CompareAndSwap(false, true) { | ||
|
|
@@ -316,7 +316,7 @@ func (dc *DCPClient) close() { | |
| } | ||
|
|
||
| // getAgentConfig returns a gocbcore.DCPAgentConfig for the given BucketSpec | ||
| func (dc *GoCBDCPClient) getAgentConfig(spec BucketSpec) (*gocbcore.DCPAgentConfig, error) { | ||
| connStr, err := spec.GetGoCBConnStringForDCP() | ||
| if err != nil { | ||
| return nil, err | ||
|
|
@@ -360,7 +360,7 @@ func (dc *DCPClient) getAgentConfig(spec BucketSpec) (*gocbcore.DCPAgentConfig, | |
| } | ||
|
|
||
| // initAgent creates a DCP agent and waits for it to be ready | ||
| func (dc *GoCBDCPClient) initAgent(spec BucketSpec) error { | ||
| agentConfig, err := dc.getAgentConfig(spec) | ||
| if err != nil { | ||
| return err | ||
|
|
@@ -405,13 +405,13 @@ func (dc *DCPClient) initAgent(spec BucketSpec) error { | |
| return nil | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) workerForVbno(vbNo uint16) *DCPWorker { | ||
| workerIndex := int(vbNo % uint16(len(dc.workers))) | ||
| return dc.workers[workerIndex] | ||
| } | ||
|
|
||
| // startWorkers initializes the DCP workers to receive stream events from eventFeed | ||
| func (dc *GoCBDCPClient) startWorkers(ctx context.Context) { | ||
|
|
||
| // vbuckets are assigned to workers as vbNo % NumWorkers. Create set of assigned vbuckets | ||
| assignedVbs := make(map[int][]uint16) | ||
|
|
@@ -434,7 +434,7 @@ func (dc *DCPClient) startWorkers(ctx context.Context) { | |
| } | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) openStream(vbID uint16, maxRetries uint32) error { | ||
|
|
||
| var openStreamErr error | ||
| var attempts uint32 | ||
|
|
@@ -488,7 +488,7 @@ func (dc *DCPClient) openStream(vbID uint16, maxRetries uint32) error { | |
| return fmt.Errorf("openStream failed to complete after %d attempts, last error: %w", attempts, openStreamErr) | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) rollback(ctx context.Context, vbID uint16, seqNo gocbcore.SeqNo) { | ||
| if dc.dbStats != nil { | ||
| dc.dbStats.Add("dcp_rollback_count", 1) | ||
| } | ||
|
|
@@ -497,7 +497,7 @@ func (dc *DCPClient) rollback(ctx context.Context, vbID uint16, seqNo gocbcore.S | |
|
|
||
| // openStreamRequest issues the OpenStream request, but doesn't perform any error handling. Callers | ||
| // should generally use openStream() for error and retry handling | ||
| func (dc *GoCBDCPClient) openStreamRequest(vbID uint16) error { | ||
|
|
||
| vbMeta := dc.metadata.GetMeta(vbID) | ||
|
|
||
|
|
@@ -548,7 +548,7 @@ func (dc *DCPClient) openStreamRequest(vbID uint16) error { | |
| // verifyFailoverLog checks for VbUUID changes when failOnRollback is set, and | ||
| // writes the failover log to the client metadata store. If previous VbUUID is zero, it's | ||
| // not considered a rollback - it's not required to initialize vbUUIDs into meta. | ||
| func (dc *GoCBDCPClient) verifyFailoverLog(vbID uint16, f []gocbcore.FailoverEntry) error { | ||
|
|
||
| if dc.failOnRollback { | ||
| previousMeta := dc.metadata.GetMeta(vbID) | ||
|
|
@@ -566,7 +566,7 @@ func (dc *DCPClient) verifyFailoverLog(vbID uint16, f []gocbcore.FailoverEntry) | |
| return nil | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) deactivateVbucket(vbID uint16) { | ||
| dc.activeVbucketLock.Lock() | ||
| delete(dc.activeVbuckets, vbID) | ||
| activeCount := len(dc.activeVbuckets) | ||
|
|
@@ -580,7 +580,7 @@ func (dc *DCPClient) deactivateVbucket(vbID uint16) { | |
| } | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) onStreamEnd(e endStreamEvent) { | ||
| if e.err == nil { | ||
| DebugfCtx(dc.ctx, KeyDCP, "Stream (vb:%d) closed, all items streamed", e.vbID) | ||
| dc.deactivateVbucket(e.vbID) | ||
|
|
@@ -616,12 +616,12 @@ func (dc *DCPClient) onStreamEnd(e endStreamEvent) { | |
| }(e.vbID, retries) | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) fatalError(err error) { | ||
| dc.setCloseError(err) | ||
| dc.close() | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) setCloseError(err error) { | ||
| dc.closeErrorLock.Lock() | ||
| defer dc.closeErrorLock.Unlock() | ||
| // If the DCPClient is already closing, don't update the error. If an initial error triggered the close, | ||
|
|
@@ -635,7 +635,7 @@ func (dc *DCPClient) setCloseError(err error) { | |
| } | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) getCloseError() error { | ||
| dc.closeErrorLock.Lock() | ||
| defer dc.closeErrorLock.Unlock() | ||
| return dc.closeError | ||
|
|
@@ -661,16 +661,18 @@ func getLatestVbUUID(failoverLog []gocbcore.FailoverEntry) (vbUUID gocbcore.VbUU | |
| return entry.VbUUID | ||
| } | ||
|
|
||
| func (dc *GoCBDCPClient) GetMetadataKeyPrefix() string { | ||
| return dc.metadata.GetKeyPrefix() | ||
| } | ||
|
|
||
| // StartWorkersForTest will iterate through dcp workers to start them, to be used for caching testing purposes only. | ||
| func (dc *GoCBDCPClient) StartWorkersForTest(t *testing.T) { | ||
| dc.startWorkers(dc.ctx) | ||
| } | ||
|
|
||
| // NewDCPClientForTest is a test-only function to create a DCP client with a specific number of vbuckets. | ||
| func NewDCPClientForTest(ctx context.Context, t *testing.T, ID string, callback sgbucket.FeedEventCallbackFunc, options DCPClientOptions, bucket *GocbV2Bucket, numVbuckets uint16) (*GoCBDCPClient, error) { | ||
|
There was a problem hiding this comment. The function name 'NewDCPClientForTest' no longer matches the struct type 'GoCBDCPClient'. Consider renaming to 'NewGoCBDCPClientForTest' for consistency. |
||
| return newDCPClientWithForBuckets(ctx, ID, callback, options, bucket, numVbuckets) | ||
| } | ||
|
|
||
| var _ gocbcore.StreamObserver = &GoCBDCPClient{} | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -31,7 +31,7 @@ type dcpDataGen struct { | |||||
| seqAlloc *sequenceAllocator | ||||||
| delays []time.Duration | ||||||
| dbCtx *db.DatabaseContext | ||||||
| client *base.GoCBDCPClient | ||||||
| numChannelsPerDoc int | ||||||
| numTotalChannels int | ||||||
| simRapidUpdate bool | ||||||
|
|
@@ -327,7 +327,7 @@ func (dcp *dcpDataGen) mutateWithDedupe(seqs []uint64, chanCount int, casValue u | |||||
| return encodedVal, chanCount, nil | ||||||
| } | ||||||
|
|
||||||
| func createDCPClient(t *testing.T, ctx context.Context, bucket *base.GocbV2Bucket, callback sgbucket.FeedEventCallbackFunc, dbStats *expvar.Map, numWorkers, numVBuckets int) (*base.GoCBDCPClient, error) { | ||||||
|
There was a problem hiding this comment. The function name 'createDCPClient' no longer matches the struct type 'GoCBDCPClient'. Consider renaming to 'createGoCBDCPClient' for consistency.
Suggested change
Collaborator
Author
There was a problem hiding this comment. I didn't rename this because it is a test tool. |
||||||
| options := base.DCPClientOptions{ | ||||||
| MetadataStoreType: base.DCPMetadataStoreInMemory, | ||||||
| GroupID: "", | ||||||
|
|
||||||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
The struct name has inconsistent casing: 'GoCB' should be 'Gocb' to follow Go naming conventions (GocbDCPClient).
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.
We aren't consistent with
GoCBandGocb. Most things useGoCBexceptGocbV2Bucket.