opensource/teleport
3
0
Fork 0
mirror of https://github.com/gravitational/teleport.git synced 2026-09-24 16:17:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix redshift auto user deadlocking (#43335)

Browse Source
This commit is contained in:
Gavin Frazar
2024-07-02 01:04:41 +00:00
committed by GitHub
parent 0426187806
commit f2c2e42f19
5 changed files with 122 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
e2e/aws/redshift_test.go
+18 -3
View File
@@ -89,6 +89,15 @@ func testRedshiftCluster(t *testing.T) {
waitForDatabases(t, cluster.Process, redshiftDBName)
db, err := cluster.Process.GetAuthServer().GetDatabase(ctx, redshiftDBName)
require.NoError(t, err)
// make sure the cloud tag is imported as a label and sets the admin
_ = mustGetDBAdmin(t, db)
// but then ignore the tag and use a randomized admin name with a randomized
// schema in its search_path to prevent tests from interfering with
// eachother.
labels := db.GetStaticLabels()
labels[types.DatabaseAdminLabel] = "test_admin_" + randASCII(t, 6)
cluster.Process.GetAuthServer().UpdateDatabase(ctx, db)
require.NoError(t, err)
adminUser := mustGetDBAdmin(t, db)
conn := connectAsRedshiftClusterAdmin(t, ctx, db.GetAWS().Redshift.ClusterID)
@@ -96,9 +105,14 @@ func testRedshiftCluster(t *testing.T) {
// create a new schema with tables that can only be accessed if the
// auto roles are granted by Teleport automatically.
testSchema := "test_" + randASCII(t, 4)
testSchema := "test_" + randASCII(t, 8)
_, err = conn.Exec(ctx, fmt.Sprintf("CREATE SCHEMA %q", testSchema))
require.NoError(t, err)
// now the admin will install its procedures in the test schema.
_, err = conn.Exec(ctx, fmt.Sprintf(`ALTER USER %q SET SEARCH_PATH = %q`, adminUser.Name, testSchema))
require.NoError(t, err)
_, err = conn.Exec(ctx, fmt.Sprintf("GRANT USAGE,CREATE ON SCHEMA %q TO %q", testSchema, adminUser.Name))
require.NoError(t, err)
t.Cleanup(func() {
// users/roles can only be dropped after we drop the schema+table.
// So, rather than juggling the order of drops, just attempt to drop
@@ -106,10 +120,11 @@ func testRedshiftCluster(t *testing.T) {
// actually created successfully.
for _, stmt := range []string{
fmt.Sprintf("DROP SCHEMA %q CASCADE", testSchema),
fmt.Sprintf("DROP ROLE %q", autoRole1),
fmt.Sprintf("DROP ROLE %q", autoRole2),
fmt.Sprintf("DROP USER IF EXISTS %q", autoUserKeep),
fmt.Sprintf("DROP USER IF EXISTS %q", autoUserDrop),
fmt.Sprintf("DROP ROLE %q", autoRole1),
fmt.Sprintf("DROP ROLE %q", autoRole2),
fmt.Sprintf("DROP USER IF EXISTS %q", adminUser.Name),
} {
_, err := conn.Exec(ctx, stmt)
assert.NoError(t, err, "test cleanup failed, stmt=%q", stmt)
lib/srv/db/common/autousers.go
+3
View File
@@ -77,6 +77,7 @@ func (a *UserProvisioner) Activate(ctx context.Context, sessionCtx *Session) (fu
retryCtx, cancel := context.WithTimeout(ctx, defaults.DatabaseConnectTimeout)
defer cancel()
a.Log.WithField("user", sessionCtx.DatabaseUser).Debug("Activating database user")
lease, err := services.AcquireSemaphoreWithRetry(retryCtx, a.makeAcquireSemaphoreConfig(sessionCtx))
if err != nil {
return nil, trace.Wrap(err)
@@ -119,6 +120,7 @@ func (a *UserProvisioner) Teardown(ctx context.Context, sessionCtx *Session) err
func (a *UserProvisioner) deactivate(ctx context.Context, sessionCtx *Session) error {
// Observe.
defer methodCallMetrics("UserProvisioner:Deactivate", teleport.ComponentDatabase, sessionCtx.Database)()
a.Log.WithField("user", sessionCtx.DatabaseUser).Debug("Deactivating database user")
retryCtx, cancel := context.WithTimeout(ctx, defaults.DatabaseConnectTimeout)
defer cancel()
@@ -147,6 +149,7 @@ func (a *UserProvisioner) deactivate(ctx context.Context, sessionCtx *Session) e
func (a *UserProvisioner) delete(ctx context.Context, sessionCtx *Session) error {
// Observe.
defer methodCallMetrics("UserProvisioner:Delete", teleport.ComponentDatabase, sessionCtx.Database)()
a.Log.WithField("user", sessionCtx.DatabaseUser).Debug("Deleting database user")
retryCtx, cancel := context.WithTimeout(ctx, defaults.DatabaseConnectTimeout)
defer cancel()
lib/srv/db/postgres/engine.go
+1 -1
View File
@@ -139,7 +139,7 @@ func (e *Engine) HandleConnection(ctx context.Context, sessionCtx *common.Sessio
defer func() {
err := e.GetUserProvisioner(e).Teardown(ctx, sessionCtx)
if err != nil {
e.Log.WithError(err).Error("Failed to teardown auto user.")
e.Log.WithError(err).WithField("user", sessionCtx.DatabaseUser).Error("Failed to teardown auto user.")
}
}()
// This is where we connect to the actual Postgres database.
lib/srv/db/postgres/sql/redshift-deactivate-user.sql
+6 -2
View File
@@ -11,11 +11,15 @@ BEGIN
IF EXISTS (SELECT user_name FROM stv_sessions WHERE user_name = CONCAT('IAM:', username)) THEN
RAISE EXCEPTION 'TP000: User has active connections';
ELSE
-- Disable ability to login for the user.
-- We do this before revoking roles so that the pg_shadow table
-- (oid 1260) lock is acquired before the pg_role (oid 4775) and
-- pg_identity (oid 4771) table locks, so that the locks are acquired in
-- the same order in the activate/deactivate/delete procedures.
EXECUTE 'ALTER USER ' || QUOTE_IDENT(username) || ' WITH CONNECTION LIMIT 0';
-- Revoke all role memberships except teleport-auto-user.
FOR rec IN select role_name FROM svv_user_grants WHERE user_name = username AND admin_option = false AND role_name != 'teleport-auto-user' LOOP
EXECUTE 'REVOKE ROLE ' || QUOTE_IDENT(rec.role_name) || ' FROM ' || QUOTE_IDENT(username);
END LOOP;
-- Disable ability to login for the user.
EXECUTE 'ALTER USER ' || QUOTE_IDENT(username) || ' WITH CONNECTION LIMIT 0';
END IF;
END;$$;
lib/srv/db/postgres/users.go
+94 -18
View File
@@ -25,13 +25,18 @@ import (
"errors"
"fmt"
"strings"
"time"
"github.com/gravitational/trace"
"github.com/jackc/pgconn"
"github.com/jackc/pgerrcode"
"github.com/jackc/pgx/v4"
"github.com/lib/pq"
"github.com/sirupsen/logrus"
"github.com/gravitational/teleport/api/types"
apiawsutils "github.com/gravitational/teleport/api/utils/aws"
"github.com/gravitational/teleport/api/utils/retryutils"
"github.com/gravitational/teleport/lib/srv/db/common"
"github.com/gravitational/teleport/lib/srv/db/common/databaseobjectimportrule"
"github.com/gravitational/teleport/lib/srv/db/common/permissions"
@@ -70,7 +75,10 @@ func (e *Engine) ActivateUser(ctx context.Context, sessionCtx *common.Session) e
// We could call this once when the database is being initialized but
// doing it here has a nice "self-healing" property in case the Teleport
// bookkeeping group or stored procedures get deleted or changed offband.
err = e.initAutoUsers(ctx, sessionCtx, conn)
logger := e.Log.WithField("user", sessionCtx.DatabaseUser)
err = withRetry(ctx, logger, func() error {
return trace.Wrap(e.initAutoUsers(ctx, sessionCtx, conn))
})
if err != nil {
return trace.Wrap(err)
}
@@ -80,10 +88,13 @@ func (e *Engine) ActivateUser(ctx context.Context, sessionCtx *common.Session) e
return trace.Wrap(err)
}
e.Log.WithField("user", sessionCtx.DatabaseUser).WithField("roles", roles).Info("Activating PostgreSQL user")
_, err = conn.Exec(ctx, activateQuery, sessionCtx.DatabaseUser, roles)
logger.WithField("roles", roles).Info("Activating PostgreSQL user")
err = withRetry(ctx, logger, func() error {
_, err = conn.Exec(ctx, activateQuery, sessionCtx.DatabaseUser, roles)
return trace.Wrap(err)
})
if err != nil {
e.Log.WithError(err).Debug("Call teleport_activate_user failed.")
logger.WithError(err).Debug("Call teleport_activate_user failed.")
errOut := convertActivateError(sessionCtx, err)
e.Audit.OnDatabaseUserCreate(ctx, sessionCtx, errOut)
return trace.Wrap(errOut)
@@ -99,7 +110,7 @@ func (e *Engine) ActivateUser(ctx context.Context, sessionCtx *common.Session) e
err = e.applyPermissions(ctx, sessionCtx)
if err != nil {
e.Log.WithError(err).Warn("Failed to apply permissions.")
logger.WithError(err).Warn("Failed to apply permissions.")
return trace.Wrap(err)
}
return nil
@@ -302,9 +313,12 @@ func (e *Engine) DeactivateUser(ctx context.Context, sessionCtx *common.Session)
}
defer conn.Close(ctx)
e.Log.WithField("user", sessionCtx.DatabaseUser).Info("Deactivating PostgreSQL user.")
_, err = conn.Exec(ctx, deactivateQuery, sessionCtx.DatabaseUser)
logger := e.Log.WithField("user", sessionCtx.DatabaseUser)
logger.Info("Deactivating PostgreSQL user.")
err = withRetry(ctx, logger, func() error {
_, err = conn.Exec(ctx, deactivateQuery, sessionCtx.DatabaseUser)
return trace.Wrap(err)
})
if err != nil {
e.Audit.OnDatabaseUserDeactivate(ctx, sessionCtx, false, err)
return trace.NewAggregate(errRemove, trace.Wrap(err))
@@ -329,15 +343,18 @@ func (e *Engine) DeleteUser(ctx context.Context, sessionCtx *common.Session) err
}
defer conn.Close(ctx)
e.Log.WithField("user", sessionCtx.DatabaseUser).Info("Deleting PostgreSQL user.")
logger := e.Log.WithField("user", sessionCtx.DatabaseUser)
logger.Info("Deleting PostgreSQL user.")
var state string
switch {
case sessionCtx.Database.IsRedshift():
err = e.deleteUserRedshift(ctx, sessionCtx, conn, &state)
default:
err = conn.QueryRow(ctx, deleteQuery, sessionCtx.DatabaseUser).Scan(&state)
}
err = withRetry(ctx, logger, func() error {
switch {
case sessionCtx.Database.IsRedshift():
return trace.Wrap(e.deleteUserRedshift(ctx, sessionCtx, conn, &state))
default:
return trace.Wrap(conn.QueryRow(ctx, deleteQuery, sessionCtx.DatabaseUser).Scan(&state))
}
})
if err != nil {
return trace.NewAggregate(errRemove, trace.Wrap(err))
}
@@ -345,12 +362,12 @@ func (e *Engine) DeleteUser(ctx context.Context, sessionCtx *common.Session) err
deleted := true
switch state {
case common.SQLStateUserDropped:
e.Log.WithField("user", sessionCtx.DatabaseUser).Debug("User deleted successfully.")
logger.Debug("User deleted successfully.")
case common.SQLStateUserDeactivated:
deleted = false
e.Log.WithField("user", sessionCtx.DatabaseUser).Info("Unable to delete user, it was disabled instead.")
logger.Info("Unable to delete user, it was disabled instead.")
default:
e.Log.WithField("user", sessionCtx.DatabaseUser).Warn("Unable to determine user deletion state.")
logger.Warn("Unable to determine user deletion state.")
}
e.Audit.OnDatabaseUserDeactivate(ctx, sessionCtx, deleted, nil)
@@ -532,3 +549,62 @@ var (
removePermissionsProcName: removePermissionsProc,
}
)
// withRetry is a helper for auto user operations that runs a given func a
// finite number of times until it returns nil error or the given context is
// done.
func withRetry(ctx context.Context, log logrus.FieldLogger, f func() error) error {
linear, err := retryutils.NewLinear(retryutils.LinearConfig{
// arbitrarily copied settings from retry logic in lib/backend/pgbk.
First: 0,
Step: 100 * time.Millisecond,
Max: 750 * time.Millisecond,
Jitter: retryutils.NewHalfJitter(),
})
if err != nil {
return trace.Wrap(err)
}
// retry a finite number of times before giving up.
for i := 0; i < 10; i++ {
err := f()
if err == nil {
return nil
}
if isRetryable(err) {
log.WithError(err).Debug("User operation failed, retrying")
} else {
return trace.Wrap(err)
}
linear.Inc()
select {
case <-linear.After():
case <-ctx.Done():
return trace.Wrap(ctx.Err())
}
}
return trace.Wrap(err, "too many retries")
}
// isRetryable returns true if an error can be retried.
func isRetryable(err error) bool {
var pgErr *pgconn.PgError
err = trace.Unwrap(err)
if errors.As(err, &pgErr) {
// https://www.postgresql.org/docs/current/mvcc-serialization-failure-handling.html
switch pgErr.Code {
case pgerrcode.DeadlockDetected, pgerrcode.SerializationFailure,
pgerrcode.UniqueViolation, pgerrcode.ExclusionViolation:
return true
}
}
// Redshift reports this with a vague SQLSTATE XX000, which is the internal
// error code, but this is a serialization error that rolls back the
// transaction, so it should be retried.
if strings.Contains(err.Error(), "conflict with concurrent transaction") {
return true
}
return pgconn.SafeToRetry(err)
}