feat: introduce scylla dedup #4922

cisse21 · 2024-07-23T12:53:14Z

Description

Introduces integration with scyllaDB for dedup
Has multiple modes but defaulted to badger for backwards compatibility
Has mirror modes which do dual writes but read from either badger or scylla

Linear Ticket

Fixes PIPE-1354

Security

The code changed/added as part of this pull request won't create any security issues with how the software is being used.

coderabbitai · 2024-07-23T12:53:20Z

Important

Review skipped

Auto reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Walkthrough

The recent changes enhance the deduplication service across multiple components by modifying method signatures for better error handling and data management. Key functions now accept structured data types, such as maps instead of slices, allowing for more efficient key-value pair handling. Additionally, new structs and methods were introduced to improve the integration between different database services, promoting a robust architecture for maintaining data consistency.

Changes

Files	Change Summary
`mocks/services/dedup/mock_dedup.go`, `services/dedup/dedup.go`	`Commit` function signature changed from `[]string` to `map[string]types.KeyValue` for better data handling.
`processor/manager.go`, `processor/processor.go`	Enhanced error handling in `Start` method and modified `Setup` to return an error.
`processor/processor_test.go`	Updated test cases to align with the new `Commit` method signature and improved error handling.
`processor/worker.go`	Changed `dedupKeys` in `storeMessage` from `map[string]struct{}` to `map[string]dedupTypes.KeyValue`.
`services/dedup/badger/badger.go`, `services/dedup/scylla/scylla.go`	Introduced `Dedup` and `ScyllaDB` structs with methods for handling database interactions.
`services/dedup/mirrorBadger/mirrorBadger.go`, `services/dedup/mirrorScylla/mirrorScylla.go`	Created new structs for mirroring services with methods for resource management and data transactions.
`services/dedup/types/types.go`	Added `WorkspaceId` field to `KeyValue` struct for enhanced data representation.

Sequence Diagram(s)

sequenceDiagram
    participant C as Client
    participant S as Service
    participant DB as Database
    
    C->>S: Call Commit(keys)
    S->>DB: Store key-value pairs
    DB-->>S: Acknowledge success
    S-->>C: Return success

sequenceDiagram
    participant C as Client
    participant S as Service
    participant M as Manager
    
    C->>M: Start process
    M->>S: Call Setup()
    alt Success
        S-->>M: Setup successful
        M-->>C: Process started
    else Failure
        S-->>M: Error
        M-->>C: Return error
    end

Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

Share

Tips

Chat

There are 3 ways to chat with CodeRabbit:

Review comments: Directly reply to a review comment made by CodeRabbit. Example:
- I pushed a fix in commit <commit_id>.
- Generate unit testing code for this file.
- Open a follow-up GitHub issue for this discussion.
Files and specific lines of code (under the "Files changed" tab): Tag @coderabbitai in a new review comment at the desired location with your query. Examples:
- @coderabbitai generate unit testing code for this file.
- @coderabbitai modularize this function.
PR comments: Tag @coderabbitai in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples:
- @coderabbitai generate interesting stats about this repository and render them as a table.
- @coderabbitai show all the console.log statements in this repository.
- @coderabbitai read src/utils.ts and generate unit testing code.
- @coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.
- @coderabbitai help me debug CodeRabbit configuration file.

Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments.

CodeRabbit Commands (invoked as PR comments)

@coderabbitai pause to pause the reviews on a PR.
@coderabbitai resume to resume the paused reviews.
@coderabbitai review to trigger an incremental review. This is useful when automatic reviews are disabled for the repository.
@coderabbitai full review to do a full review from scratch and review all the files again.
@coderabbitai summary to regenerate the summary of the PR.
@coderabbitai resolve resolve all the CodeRabbit review comments.
@coderabbitai configuration to show the current CodeRabbit configuration for the repository.
@coderabbitai help to get help.

Additionally, you can add @coderabbitai ignore anywhere in the PR description to prevent this PR from being reviewed.

CodeRabbit Configuration File (`.coderabbit.yaml`)

You can programmatically configure CodeRabbit by adding a .coderabbit.yaml file to the root of your repository.
Please see the configuration documentation for more information.
If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: # yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json

Documentation and Community

Visit our Documentation for detailed information on how to use CodeRabbit.
Join our Discord Community to get help, request features, and share feedback.
Follow us on X/Twitter for updates and announcements.

codecov · 2024-07-29T07:05:15Z

Codecov Report

Attention: Patch coverage is 86.59794% with 26 lines in your changes missing coverage. Please review.

Project coverage is 74.40%. Comparing base (77b75fb) to head (4edc80d).
Report is 7 commits behind head on master.

Files	Patch %	Lines
services/dedup/badger/badger.go	85.18%	4 Missing and 4 partials ⚠️
services/dedup/scylla/scylla.go	89.74%	4 Missing and 4 partials ⚠️
processor/manager.go	33.33%	1 Missing and 1 partial ⚠️
processor/processor.go	71.42%	1 Missing and 1 partial ⚠️
services/dedup/dedup.go	87.50%	1 Missing and 1 partial ⚠️
services/dedup/mirrorBadger/mirrorBadger.go	88.88%	1 Missing and 1 partial ⚠️
services/dedup/mirrorScylla/mirrorScylla.go	88.88%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #4922      +/-   ##
==========================================
+ Coverage   74.37%   74.40%   +0.02%     
==========================================
  Files         428      431       +3     
  Lines       49909    50053     +144     
==========================================
+ Hits        37119    37241     +122     
- Misses      10340    10352      +12     
- Partials     2450     2460      +10

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

services/dedup/scylla/scylla.go

services/dedup/badger/badger_test.go

services/dedup/dedup_test.go

services/dedup/mirrorBadger/mirrorBadger.go

services/dedup/mirrorScylla/mirrorScylla.go

mihir20

not very sure on mirror scylla and mirror badger implementations.

Co-authored-by: Mihir Gandhi <[email protected]>

mihir20 · 2024-08-12T06:29:14Z

services/dedup/mirrorBadger/mirrorBadger.go

+}
+
+func (mb *MirrorBadger) Get(kv types.KeyValue) (bool, int64, error) {
+	_, _, _ = mb.scylla.Get(kv)


What if we handle the error here? In worst case scylla will get populated and badger will not, anyways we are throwing a panic in processor in such cases so everything will be retried.

It should be OK given that scylla with this configuration is not essential. Still, we want to know if mirroring isn't working, right? We could start with logging these. Wdyt?

lvrach · 2024-08-12T12:59:19Z

services/dedup/types/types.go

-	Value int64
+	Key         string
+	Value       int64
+	WorkspaceId string


Suggested change

WorkspaceId string

WorkspaceID string

lvrach · 2024-08-12T13:00:35Z

services/dedup/scylla/scylla_test.go

+		require.Nil(t, err)
+		require.False(t, found)
+	})
+	t.Run("Same messageID should be deduped for same workspace from cache", func(t *testing.T) {
+		key1 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		key2 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}
+		found, _, err := scylla.Get(key1)
+		require.Nil(t, err)
+		require.True(t, found)
+		found, _, err = scylla.Get(key2)
+		require.Nil(t, err)


[nit-pick] for consistency and improved semantics

Suggested change

require.Nil(t, err)

require.False(t, found)

})

t.Run("Same messageID should be deduped for same workspace from cache", func(t *testing.T) {

key1 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}

key2 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}

found, _, err := scylla.Get(key1)

require.Nil(t, err)

require.True(t, found)

found, _, err = scylla.Get(key2)

require.Nil(t, err)

require.NoError(t, err)

require.False(t, found)

})

t.Run("Same messageID should be deduped for same workspace from cache", func(t *testing.T) {

key1 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}

key2 := types.KeyValue{Key: "b", Value: 1, WorkspaceId: "test"}

found, _, err := scylla.Get(key1)

require.NoError(t, err)

require.True(t, found)

found, _, err = scylla.Get(key2)

require.NoError(t, err)

lvrach · 2024-08-12T13:02:39Z

services/dedup/scylla/scylla.go

+				})
+			}
+			if err := d.scylla.ExecuteBatch(scyllaBatch); err != nil {
+				return fmt.Errorf("error committing keys: %v", err)


[minor]

Suggested change

return fmt.Errorf("error committing keys: %v", err)

return fmt.Errorf("committing keys: %v", err)

lvrach · 2024-08-12T13:08:08Z

services/dedup/scylla/scylla.go

+		d.createTableMap[kv.WorkspaceId] = &sync.Once{}
+		once = d.createTableMap[kv.WorkspaceId]
+	}
+	d.createTableMu.Unlock()
+	once.Do(func() {
+		query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text PRIMARY KEY,size bigint) WITH bloom_filter_fp_chance = 0.005;", d.keyspace, kv.WorkspaceId)
+		err = d.scylla.Query(query).Exec()
+	})
+	if err != nil {


You can use sync.OnceValue to store and reuse the error.

In the existing code, in case of an error, only the first goroutine will return the error.

Suggested change

d.createTableMap[kv.WorkspaceId] = &sync.Once{}

once = d.createTableMap[kv.WorkspaceId]

}

d.createTableMu.Unlock()

once.Do(func() {

query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text PRIMARY KEY,size bigint) WITH bloom_filter_fp_chance = 0.005;", d.keyspace, kv.WorkspaceId)

err = d.scylla.Query(query).Exec()

})

if err != nil {

d.createTableMap[kv.WorkspaceId] = &sync.OnceValue[error]{}

once = d.createTableMap[kv.WorkspaceId]

}

d.createTableMu.Unlock()

err := once.Do(func() error {

query := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s.%s (id text PRIMARY KEY,size bigint) WITH bloom_filter_fp_chance = 0.005;", d.keyspace, kv.WorkspaceId)

return d.scylla.Query(query).Exec()

})

if err != nil {

Also, you might want to move this logic to a separate unexported method, and call it even from Commit. It depends on how robust you want to make your API, normally Get for a workspace should always happen before a commit.

fracasula · 2024-08-12T14:34:29Z

services/dedup/badger/badger.go

-	log := logger.NewLogger().Child("dedup")
+func NewBadgerDB(conf *config.Config, stats stats.Stats, path string) *Dedup {
+	dedupWindow := conf.GetReloadableDurationVar(3600, time.Second, "Dedup.dedupWindow", "Dedup.dedupWindowInS")
+	log := logger.NewLogger().Child("Dedup")
 	badgerOpts := badger.
 		DefaultOptions(path).
 		WithCompression(options.None).


[question] Out of curiosity... did you test this with compression at all? If yes, what did you observe?

This was an already existing code and I didn't make any changes in the existing functionality

fracasula · 2024-08-12T14:35:03Z

services/dedup/badger/badger.go

-	log := logger.NewLogger().Child("dedup")
+func NewBadgerDB(conf *config.Config, stats stats.Stats, path string) *Dedup {
+	dedupWindow := conf.GetReloadableDurationVar(3600, time.Second, "Dedup.dedupWindow", "Dedup.dedupWindowInS")
+	log := logger.NewLogger().Child("Dedup")
 	badgerOpts := badger.
 		DefaultOptions(path).
 		WithCompression(options.None).
 		WithIndexCacheSize(16 << 20). // 16mb
 		WithNumGoroutines(1).


[question] Could increasing this number help?

fracasula · 2024-08-12T14:38:34Z

services/dedup/badger/badger.go

-		return 0, false, err
-	}
-	err = d.badgerDB.View(func(txn *badger.Txn) error {
+	err := d.badgerDB.View(func(txn *badger.Txn) error {
 		item, err := txn.Get([]byte(key))
 		if err != nil {
 			return err


[question on line 83] Probably not related to this PR but it seems weird so I'll ask.

Why do we do this?

payloadSize, _ = strconv.ParseInt(string(itemValue), 10, 64) found = true

Wouldn't it make more sense to do this instead?

payloadSize, err = strconv.ParseInt(string(itemValue), 10, 64) if err == nil { found = true }

I'm saying this, because if we can get the value from the database but it cannot be parsed then we wrote something invalid. We might as well say we couldn't find it? 🤔 WDYT?

I think this was done because we care more whether the key was present or not and not the value for the key

fracasula · 2024-08-12T14:55:04Z

services/dedup/mirrorBadger/mirrorBadger_test.go

+	logger.Reset()
+	misc.Init()
+
+	dbPath := os.TempDir() + "/dedup_test"


Suggested change

dbPath := os.TempDir() + "/dedup_test"

dbPath := t.TempDir() + "/dedup_test"

It checks for existence as well so you could remove the _ = os.RemoveAll(dbPath) that you have on line 27.
It should also do the cleanup for you if I'm not mistaken:

c.tempDir, c.tempDirErr = os.MkdirTemp("", pattern) if c.tempDirErr == nil { c.Cleanup(func() { if err := removeAll(c.tempDir); err != nil { c.Errorf("TempDir RemoveAll cleanup: %v", err) } }) }

fracasula · 2024-08-12T14:59:33Z

services/dedup/mirrorScylla/mirrorScylla_test.go

+	dbPath := os.TempDir() + "/dedup_test"
+	defer func() { _ = os.RemoveAll(dbPath) }()
+	_ = os.RemoveAll(dbPath)


Same comment as above. Could use t.TempDir().

fracasula

Approved but I left a few questions and a suggestion for your tests 👍

cisse21 added 6 commits July 22, 2024 15:08

chore: refactor dedup package

c8f3c55

Merge branch 'master' into chore.refactorDedup

5348742

Merge branch 'master' into chore.refactorDedup

847391f

chore: merge master

b76ea10

feat: introduce scylla dedup

20d740d

chore: review comments

efb7c8c

cisse21 changed the title ~~Feat.introduce scylla dedup~~ feat: introduce scylla dedup Jul 23, 2024

Merge branch 'chore.refactorDedup' into feat.introduceScyllaDedup

7bb2dd9

cisse21 force-pushed the feat.introduceScyllaDedup branch from 229a938 to 7bb2dd9 Compare July 23, 2024 13:36

chore: review comments

35a0edf

cisse21 force-pushed the chore.refactorDedup branch from efb7c8c to 35a0edf Compare July 23, 2024 14:37

cisse21 force-pushed the feat.introduceScyllaDedup branch from 21ba90a to e31686d Compare July 24, 2024 05:27

Merge branch 'chore.refactorDedup' into feat.introduceScyllaDedup

e31db3d

cisse21 force-pushed the feat.introduceScyllaDedup branch 4 times, most recently from 8acf78e to d15f936 Compare July 24, 2024 09:44

chore: refactor dedup package

8e393ae

cisse21 force-pushed the feat.introduceScyllaDedup branch from d15f936 to 8e393ae Compare July 24, 2024 11:54

Base automatically changed from chore.refactorDedup to master July 25, 2024 12:53

chore: merge master

a2f3d27

cisse21 force-pushed the feat.introduceScyllaDedup branch from 31cbcff to a2f3d27 Compare July 26, 2024 06:01

Merge branch 'master' into feat.introduceScyllaDedup

e33dcca

cisse21 force-pushed the feat.introduceScyllaDedup branch 4 times, most recently from ff5fd37 to 0a698ae Compare July 29, 2024 06:31

rudderlabs deleted a comment from coderabbitai bot Jul 29, 2024

ktgowtham requested review from fracasula and lvrach August 6, 2024 11:05

$fracasula$

fracasula reviewed Aug 7, 2024

View reviewed changes

services/dedup/scylla/scylla.go Outdated Show resolved Hide resolved

cisse21 force-pushed the feat.introduceScyllaDedup branch 2 times, most recently from 1a20fb3 to 2a083d2 Compare August 8, 2024 11:33

chore: review comments

69f170e

cisse21 force-pushed the feat.introduceScyllaDedup branch from 2a083d2 to 69f170e Compare August 8, 2024 11:41

mihir20 reviewed Aug 9, 2024

View reviewed changes

Apply suggestions from code review

4edc80d

Co-authored-by: Mihir Gandhi <[email protected]>

cisse21 force-pushed the feat.introduceScyllaDedup branch from 0c3f0d7 to 4edc80d Compare August 9, 2024 09:16

cisse21 requested review from mihir20 and fracasula August 12, 2024 06:13

mihir20 reviewed Aug 12, 2024

View reviewed changes

chore: add more tests

d44779f

lvrach approved these changes Aug 12, 2024

View reviewed changes

$fracasula$

fracasula reviewed Aug 12, 2024

View reviewed changes

$fracasula$

fracasula approved these changes Aug 12, 2024

View reviewed changes

cisse21 added 2 commits August 13, 2024 12:10

Merge branch 'master' into feat.introduceScyllaDedup

7fc3426

chore: review comments

1eba0cb

cisse21 merged commit 31a033d into master Aug 13, 2024
51 checks passed

cisse21 deleted the feat.introduceScyllaDedup branch August 13, 2024 07:10

This was referenced Aug 16, 2024

chore: release 1.32.0 #5003

Closed

chore: prerelease 1.32.0-rc.1 #5006

Merged

chore: release 1.32.0 #5007

Merged

chore: prerelease 1.32.0-rc.2 #5008

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: introduce scylla dedup #4922

feat: introduce scylla dedup #4922

cisse21 commented Jul 23, 2024 •

edited

Loading

coderabbitai bot commented Jul 23, 2024 •

edited

Loading

Review skipped

Chat

CodeRabbit Commands (invoked as PR comments)

CodeRabbit Configuration File (`.coderabbit.yaml`)

Documentation and Community

codecov bot commented Jul 29, 2024 •

edited

Loading

mihir20 left a comment •

edited

Loading

mihir20 Aug 12, 2024

$@fracasula$ fracasula Aug 12, 2024

lvrach Aug 12, 2024

lvrach Aug 12, 2024

lvrach Aug 12, 2024

lvrach Aug 12, 2024

lvrach Aug 12, 2024

$@fracasula$ fracasula Aug 12, 2024

cisse21 Aug 13, 2024

$@fracasula$ fracasula Aug 12, 2024

$@fracasula$ fracasula Aug 12, 2024

cisse21 Aug 13, 2024

$@fracasula$ fracasula Aug 12, 2024

$@fracasula$ fracasula Aug 12, 2024

$@fracasula$ fracasula Aug 12, 2024

$@fracasula$ fracasula left a comment

	return fmt.Errorf("error committing keys: %v", err)
	return fmt.Errorf("committing keys: %v", err)

	dbPath := os.TempDir() + "/dedup_test"
	dbPath := t.TempDir() + "/dedup_test"

feat: introduce scylla dedup #4922

feat: introduce scylla dedup #4922

Conversation

cisse21 commented Jul 23, 2024 • edited Loading

Description

Linear Ticket

Security

coderabbitai bot commented Jul 23, 2024 • edited Loading

Review skipped

Walkthrough

Changes

Sequence Diagram(s)

Chat

CodeRabbit Commands (invoked as PR comments)

CodeRabbit Configuration File (.coderabbit.yaml)

Documentation and Community

codecov bot commented Jul 29, 2024 • edited Loading

Codecov Report

mihir20 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fracasula left a comment

Choose a reason for hiding this comment

cisse21 commented Jul 23, 2024 •

edited

Loading

coderabbitai bot commented Jul 23, 2024 •

edited

Loading

CodeRabbit Configuration File (`.coderabbit.yaml`)

codecov bot commented Jul 29, 2024 •

edited

Loading

mihir20 left a comment •

edited

Loading

$@fracasula$ fracasula left a comment