Update locking components from DR replication changes (#3283)

* Update locking components from DR replication changes

* Fix plugin backend test

* Add a comment about needing the statelock:

helper/consts/replication.go

@@ -3,18 +3,37 @@ package consts
 type ReplicationState uint32
 
 const (
-	ReplicationDisabled ReplicationState = iota
-	ReplicationPrimary
-	ReplicationSecondary
+	_ ReplicationState = iota
+	OldReplicationPrimary
+	OldReplicationSecondary
+	OldReplicationBootstrapping
+
+	ReplicationDisabled           ReplicationState = 0
+	ReplicationPerformancePrimary ReplicationState = 1 << iota
+	ReplicationPerformanceSecondary
+	ReplicationBootstrapping
+	ReplicationDRPrimary
+	ReplicationDRSecondary
 )
 
 func (r ReplicationState) String() string {
 	switch r {
-	case ReplicationSecondary:
-		return "secondary"
-	case ReplicationPrimary:
-		return "primary"
+	case ReplicationPerformanceSecondary:
+		return "perf-secondary"
+	case ReplicationPerformancePrimary:
+		return "perf-primary"
+	case ReplicationBootstrapping:
+		return "bootstrapping"
+	case ReplicationDRPrimary:
+		return "dr-primary"
+	case ReplicationDRSecondary:
+		return "dr-secondary"
 	}
 
 	return "disabled"
 }
+
+func (r ReplicationState) HasState(flag ReplicationState) bool { return r&flag != 0 }
+func (r *ReplicationState) AddState(flag ReplicationState)     { *r |= flag }
+func (r *ReplicationState) ClearState(flag ReplicationState)   { *r &= ^flag }
+func (r *ReplicationState) ToggleState(flag ReplicationState)  { *r ^= flag }

http/sys_rekey.go

@@ -21,7 +21,7 @@ func handleSysRekeyInit(core *vault.Core, recovery bool) http.Handler {
 		}
 
 		repState := core.ReplicationState()
-		if repState == consts.ReplicationSecondary {
+		if repState.HasState(consts.ReplicationPerformanceSecondary) {
 			respondError(w, http.StatusBadRequest,
 				fmt.Errorf("rekeying can only be performed on the primary cluster when replication is activated"))
 			return

logical/plugin/system_test.go

@@ -117,7 +117,7 @@ func TestSystem_replicationState(t *testing.T) {
 	defer client.Close()
 
 	sys := logical.TestSystemView()
-	sys.ReplicationStateVal = consts.ReplicationPrimary
+	sys.ReplicationStateVal = consts.ReplicationPerformancePrimary
 
 	server.RegisterName("Plugin", &SystemViewServer{
 		impl: sys,

vault/core.go

@@ -1891,11 +1891,9 @@ func (c *Core) emitMetrics(stopCh chan struct{}) {
 }
 
 func (c *Core) ReplicationState() consts.ReplicationState {
-	var state consts.ReplicationState
-	c.clusterParamsLock.RLock()
-	state = c.replicationState
-	c.clusterParamsLock.RUnlock()
-	return state
+	c.stateLock.RLock()
+	defer c.stateLock.RUnlock()
+	return c.replicationState
 }
 
 func (c *Core) SealAccess() *SealAccess {

vault/dynamic_system_view.go

@@ -84,13 +84,10 @@ func (d dynamicSystemView) CachingDisabled() bool {
 	return d.core.cachingDisabled || (d.mountEntry != nil && d.mountEntry.Config.ForceNoCache)
 }
 
-// Checks if this is a primary Vault instance.
+// Checks if this is a primary Vault instance. Caller should hold the stateLock
+// in read mode.
 func (d dynamicSystemView) ReplicationState() consts.ReplicationState {
-	var state consts.ReplicationState
-	d.core.clusterParamsLock.RLock()
-	state = d.core.replicationState
-	d.core.clusterParamsLock.RUnlock()
-	return state
+	return d.core.replicationState
 }
 
 // ResponseWrapData wraps the given data in a cubbyhole and returns the

vault/logical_system.go

@@ -1251,12 +1251,10 @@ func (b *SystemBackend) handleMountTable(
 // handleMount is used to mount a new path
 func (b *SystemBackend) handleMount(
 	req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
-	b.Core.clusterParamsLock.RLock()
 	repState := b.Core.replicationState
-	b.Core.clusterParamsLock.RUnlock()
 
 	local := data.Get("local").(bool)
-	if !local && repState == consts.ReplicationSecondary {
+	if !local && repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot add a non-local mount to a replication secondary"), nil
 	}
 
@@ -1381,9 +1379,9 @@ func (b *SystemBackend) handleUnmount(
 	path := data.Get("path").(string)
 	path = sanitizeMountPath(path)
 
-	repState := b.Core.ReplicationState()
+	repState := b.Core.replicationState
 	entry := b.Core.router.MatchingMountEntry(path)
-	if entry != nil && !entry.Local && repState == consts.ReplicationSecondary {
+	if entry != nil && !entry.Local && repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot unmount a non-local mount on a replication secondary"), nil
 	}
 
@@ -1406,9 +1404,7 @@ func (b *SystemBackend) handleUnmount(
 // handleRemount is used to remount a path
 func (b *SystemBackend) handleRemount(
 	req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
-	b.Core.clusterParamsLock.RLock()
 	repState := b.Core.replicationState
-	b.Core.clusterParamsLock.RUnlock()
 
 	// Get the paths
 	fromPath := data.Get("from").(string)
@@ -1423,7 +1419,7 @@ func (b *SystemBackend) handleRemount(
 	toPath = sanitizeMountPath(toPath)
 
 	entry := b.Core.router.MatchingMountEntry(fromPath)
-	if entry != nil && !entry.Local && repState == consts.ReplicationSecondary {
+	if entry != nil && !entry.Local && repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot remount a non-local mount on a replication secondary"), nil
 	}
 
@@ -1519,9 +1515,7 @@ func (b *SystemBackend) handleMountTuneWrite(
 // handleTuneWriteCommon is used to set config settings on a path
 func (b *SystemBackend) handleTuneWriteCommon(
 	path string, data *framework.FieldData) (*logical.Response, error) {
-	b.Core.clusterParamsLock.RLock()
 	repState := b.Core.replicationState
-	b.Core.clusterParamsLock.RUnlock()
 
 	path = sanitizeMountPath(path)
 
@@ -1538,7 +1532,7 @@ func (b *SystemBackend) handleTuneWriteCommon(
 		b.Backend.Logger().Error("sys: tune failed: no mount entry found", "path", path)
 		return handleError(fmt.Errorf("sys: tune of path '%s' failed: no mount entry found", path))
 	}
-	if mountEntry != nil && !mountEntry.Local && repState == consts.ReplicationSecondary {
+	if mountEntry != nil && !mountEntry.Local && repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot tune a non-local mount on a replication secondary"), nil
 	}
 
@@ -1757,12 +1751,10 @@ func (b *SystemBackend) handleAuthTable(
 // handleEnableAuth is used to enable a new credential backend
 func (b *SystemBackend) handleEnableAuth(
 	req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
-	b.Core.clusterParamsLock.RLock()
 	repState := b.Core.replicationState
-	b.Core.clusterParamsLock.RUnlock()
 
 	local := data.Get("local").(bool)
-	if !local && repState == consts.ReplicationSecondary {
+	if !local && repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot add a non-local mount to a replication secondary"), nil
 	}
 
@@ -1833,9 +1825,9 @@ func (b *SystemBackend) handleDisableAuth(
 	path = sanitizeMountPath(path)
 	fullPath := credentialRoutePrefix + path
 
-	repState := b.Core.ReplicationState()
+	repState := b.Core.replicationState
 	entry := b.Core.router.MatchingMountEntry(fullPath)
-	if entry != nil && !entry.Local && repState == consts.ReplicationSecondary {
+	if entry != nil && !entry.Local && repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot unmount a non-local mount on a replication secondary"), nil
 	}
 
@@ -1984,12 +1976,10 @@ func (b *SystemBackend) handleAuditHash(
 // handleEnableAudit is used to enable a new audit backend
 func (b *SystemBackend) handleEnableAudit(
 	req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
-	b.Core.clusterParamsLock.RLock()
 	repState := b.Core.replicationState
-	b.Core.clusterParamsLock.RUnlock()
 
 	local := data.Get("local").(bool)
-	if !local && repState == consts.ReplicationSecondary {
+	if !local && repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot add a non-local mount to a replication secondary"), nil
 	}
 
@@ -2132,10 +2122,8 @@ func (b *SystemBackend) handleKeyStatus(
 // handleRotate is used to trigger a key rotation
 func (b *SystemBackend) handleRotate(
 	req *logical.Request, data *framework.FieldData) (*logical.Response, error) {
-	b.Core.clusterParamsLock.RLock()
 	repState := b.Core.replicationState
-	b.Core.clusterParamsLock.RUnlock()
-	if repState == consts.ReplicationSecondary {
+	if repState.HasState(consts.ReplicationPerformanceSecondary) {
 		return logical.ErrorResponse("cannot rotate on a replication secondary"), nil
 	}
 

vault/mount.go

@@ -544,7 +544,7 @@ func (c *Core) loadMounts() error {
 			// ensure this comes over. If we upgrade first, we simply don't
 			// create the mount, so we won't conflict when we sync. If this is
 			// local (e.g. cubbyhole) we do still add it.
-			if !foundRequired && (c.replicationState != consts.ReplicationSecondary || requiredMount.Local) {
+			if !foundRequired && (c.replicationState.HasState(consts.ReplicationPerformanceSecondary) || requiredMount.Local) {
 				c.mounts.Entries = append(c.mounts.Entries, requiredMount)
 				needPersist = true
 			}

vault/policy_store.go

@@ -145,7 +145,7 @@ func (c *Core) setupPolicyStore() error {
 	sysView := &dynamicSystemView{core: c}
 	c.policyStore = NewPolicyStore(view, sysView)
 
-	if sysView.ReplicationState() == consts.ReplicationSecondary {
+	if c.replicationState.HasState(consts.ReplicationPerformanceSecondary) {
 		// Policies will sync from the primary
 		return nil
 	}