-
Notifications
You must be signed in to change notification settings - Fork 311
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.
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
chore: improve rsources service table setup procedure #4165
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change | ||||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
@@ -476,26 +476,40 @@ | |||||||||||||
return time.After(config.GetDuration("Rsources.stats.cleanup.interval", 1, time.Hour)) | ||||||||||||||
} | ||||||||||||||
} | ||||||||||||||
sh.log.Debugf("setting up rsources tables in %s", sh.config.LocalHostname) | ||||||||||||||
if err := setupTables(ctx, sh.localDB, sh.config.LocalHostname, sh.log); err != nil { | ||||||||||||||
|
||||||||||||||
const lockID = 100020001 | ||||||||||||||
|
||||||||||||||
if err := withAdvisoryLock(ctx, sh.localDB, lockID, func(tx *sql.Tx) error { | ||||||||||||||
sh.log.Debugf("setting up rsources tables in %s", sh.config.LocalHostname) | ||||||||||||||
if err := setupTables(ctx, sh.localDB, sh.config.LocalHostname, sh.log); err != nil { | ||||||||||||||
return err | ||||||||||||||
} | ||||||||||||||
if err := migrateFailedKeysTable(ctx, tx); err != nil { | ||||||||||||||
return fmt.Errorf("migrating rsources_failed_keys table: %w", err) | ||||||||||||||
} | ||||||||||||||
sh.log.Debugf("rsources tables setup successfully in %s", sh.config.LocalHostname) | ||||||||||||||
return nil | ||||||||||||||
}); err != nil { | ||||||||||||||
return err | ||||||||||||||
} | ||||||||||||||
if err := migrateFailedKeysTable(ctx, sh.localDB); err != nil { | ||||||||||||||
return fmt.Errorf("failed to migrate rsources_failed_keys table: %w", err) | ||||||||||||||
} | ||||||||||||||
sh.log.Debugf("rsources tables setup successfully in %s", sh.config.LocalHostname) | ||||||||||||||
|
||||||||||||||
if sh.sharedDB != nil { | ||||||||||||||
sh.log.Debugf("setting up rsources tables for shared db %s", sh.config.SharedConn) | ||||||||||||||
if err := setupTables(ctx, sh.sharedDB, "shared", sh.log); err != nil { | ||||||||||||||
return err | ||||||||||||||
} | ||||||||||||||
sh.log.Debugf("rsources tables for shared db %s setup successfully", sh.config.SharedConn) | ||||||||||||||
if err := withAdvisoryLock(ctx, sh.sharedDB, lockID, func(_ *sql.Tx) error { | ||||||||||||||
sh.log.Debugf("setting up rsources tables for shared db %s", sh.config.SharedConn) | ||||||||||||||
if err := setupTables(ctx, sh.sharedDB, "shared", sh.log); err != nil { | ||||||||||||||
return err | ||||||||||||||
} | ||||||||||||||
sh.log.Debugf("rsources tables for shared db %s setup successfully", sh.config.SharedConn) | ||||||||||||||
|
||||||||||||||
sh.log.Debugf("setting up rsources logical replication in %s", sh.config.LocalHostname) | ||||||||||||||
if err := sh.setupLogicalReplication(ctx); err != nil { | ||||||||||||||
return fmt.Errorf("failed to setup rsources logical replication in %s: %w", sh.config.LocalHostname, err) | ||||||||||||||
sh.log.Debugf("setting up rsources logical replication in %s", sh.config.LocalHostname) | ||||||||||||||
if err := sh.setupLogicalReplication(ctx); err != nil { | ||||||||||||||
return fmt.Errorf("logical replication in %q: %w", sh.config.LocalHostname, err) | ||||||||||||||
} | ||||||||||||||
sh.log.Debugf("rsources logical replication setup successfully in %s", sh.config.LocalHostname) | ||||||||||||||
return nil | ||||||||||||||
}); err != nil { | ||||||||||||||
return err | ||||||||||||||
} | ||||||||||||||
sh.log.Debugf("rsources logical replication setup successfully in %s", sh.config.LocalHostname) | ||||||||||||||
} | ||||||||||||||
return nil | ||||||||||||||
Comment on lines
476
to
514
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Consider refactoring the |
||||||||||||||
} | ||||||||||||||
|
@@ -511,15 +525,7 @@ | |||||||||||||
} | ||||||||||||||
|
||||||||||||||
// TODO: Remove this after a few releases | ||||||||||||||
func migrateFailedKeysTable(ctx context.Context, db *sql.DB) error { | ||||||||||||||
tx, err := db.Begin() | ||||||||||||||
if err != nil { | ||||||||||||||
return fmt.Errorf("failed to start transaction: %w", err) | ||||||||||||||
} | ||||||||||||||
defer func() { _ = tx.Rollback() }() | ||||||||||||||
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(100020001)`); err != nil { | ||||||||||||||
return fmt.Errorf("error while acquiring advisory lock for failed keys migration: %w", err) | ||||||||||||||
} | ||||||||||||||
func migrateFailedKeysTable(ctx context.Context, tx *sql.Tx) error { | ||||||||||||||
var previousTableExists bool | ||||||||||||||
row := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT FROM pg_tables WHERE schemaname NOT IN ('pg_catalog','information_schema') AND tablename = 'rsources_failed_keys')`) | ||||||||||||||
if err := row.Scan(&previousTableExists); err != nil { | ||||||||||||||
|
@@ -597,7 +603,7 @@ | |||||||||||||
if _, err := tx.ExecContext(ctx, `drop function if exists ksuid()`); err != nil { | ||||||||||||||
return fmt.Errorf("failed to drop ksuid function: %w", err) | ||||||||||||||
} | ||||||||||||||
return tx.Commit() | ||||||||||||||
return nil | ||||||||||||||
} | ||||||||||||||
|
||||||||||||||
return nil | ||||||||||||||
|
@@ -803,3 +809,18 @@ | |||||||||||||
} | ||||||||||||||
} | ||||||||||||||
} | ||||||||||||||
|
||||||||||||||
func withAdvisoryLock(ctx context.Context, db *sql.DB, lockId int64, f func(tx *sql.Tx) error) error { | ||||||||||||||
tx, err := db.BeginTx(ctx, nil) | ||||||||||||||
if err != nil { | ||||||||||||||
return fmt.Errorf("failed to start transaction: %w", err) | ||||||||||||||
} | ||||||||||||||
defer func() { _ = tx.Rollback() }() | ||||||||||||||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. The deferred rollback in - defer func() { _ = tx.Rollback() }()
+ defer func() {
+ if r := recover(); r != nil || err != nil {
+ _ = tx.Rollback()
+ }
+ }() Committable suggestion
Suggested change
|
||||||||||||||
if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock($1)`, lockId); err != nil { | ||||||||||||||
return fmt.Errorf("acquiring advisory lock: %w", err) | ||||||||||||||
} | ||||||||||||||
if err := f(tx); err != nil { | ||||||||||||||
return err | ||||||||||||||
} | ||||||||||||||
return tx.Commit() | ||||||||||||||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -31,7 +31,7 @@ var _ = Describe("Using sources handler v1", func() { | |
Out: 4, | ||
Failed: 6, | ||
} | ||
BeforeAll(func() { | ||
BeforeEach(func() { | ||
var err error | ||
pool, err = dockertest.NewPool("") | ||
Expect(err).NotTo(HaveOccurred()) | ||
Comment on lines
31
to
37
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
The
The |
||
|
@@ -45,7 +45,7 @@ var _ = Describe("Using sources handler v1", func() { | |
sh = createService(config) | ||
}) | ||
|
||
AfterAll(func() { | ||
AfterEach(func() { | ||
purgeResources(pool, resource.resource) | ||
}) | ||
|
||
|
@@ -337,7 +337,7 @@ var _ = Describe("Using sources handler v1", func() { | |
serviceA, serviceB JobService | ||
) | ||
|
||
BeforeAll(func() { | ||
BeforeEach(func() { | ||
var err error | ||
pool, err = dockertest.NewPool("") | ||
Expect(err).NotTo(HaveOccurred()) | ||
|
@@ -382,7 +382,7 @@ var _ = Describe("Using sources handler v1", func() { | |
serviceB = createService(configB) | ||
}) | ||
|
||
AfterAll(func() { | ||
AfterEach(func() { | ||
purgeResources(pool, pgA.resource, pgB.resource, pgC.resource) | ||
if network != nil { | ||
_ = pool.Client.RemoveNetwork(network.ID) | ||
|
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -199,6 +199,9 @@ func NewJobService(config JobServiceConfig) (JobService, error) { | |
if config.Log == nil { | ||
config.Log = logger.NewLogger().Child("rsources") | ||
} | ||
if config.MaxPoolSize <= 2 { | ||
config.MaxPoolSize = 2 // minimum 2 connections in the pool for proper startup | ||
} | ||
Comment on lines
+202
to
+204
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Ensure that |
||
var ( | ||
localDB, sharedDB *sql.DB | ||
err error | ||
Comment on lines
199
to
207
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Consider checking the error returned by |
||
|
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 test command logic for handling
package
andexclude
variables is correctly implemented. However, consider using$(shell ...)
instead of backticks for command substitution to improve readability and maintainability in the Makefile.