core: add ACL role state schema and functionality.

This commit includes the new state schema for ACL roles along with
state interaction functions for CRUD actions.

The change also includes snapshot persist and restore
functionality and the addition of FSM messages for Raft updates
which will come via RPC endpoints.

nomad/fsm.go

@@ -58,6 +58,7 @@ const (
 	SecureVariablesSnapshot              SnapshotType = 22
 	SecureVariablesQuotaSnapshot         SnapshotType = 23
 	RootKeyMetaSnapshot                  SnapshotType = 24
+	ACLRoleSnapshot                      SnapshotType = 25
 
 	// Namespace appliers were moved from enterprise and therefore start at 64
 	NamespaceSnapshot SnapshotType = 64
@@ -325,6 +326,10 @@ func (n *nomadFSM) Apply(log *raft.Log) interface{} {
 		return n.applyRootKeyMetaUpsert(msgType, buf[1:], log.Index)
 	case structs.RootKeyMetaDeleteRequestType:
 		return n.applyRootKeyMetaDelete(msgType, buf[1:], log.Index)
+	case structs.ACLRolesUpsertRequestType:
+		return n.applyACLRolesUpsert(msgType, buf[1:], log.Index)
+	case structs.ACLRolesDeleteByIDRequestType:
+		return n.applyACLRolesDeleteByID(msgType, buf[1:], log.Index)
 	}
 
 	// Check enterprise only message types.
@@ -1750,6 +1755,20 @@ func (n *nomadFSM) restoreImpl(old io.ReadCloser, filter *FSMFilter) error {
 			if err := restore.RootKeyMetaRestore(keyMeta); err != nil {
 				return err
 			}
+		case ACLRoleSnapshot:
+
+			// Create a new ACLRole object, so we can decode the message into
+			// it.
+			aclRole := new(structs.ACLRole)
+
+			if err := dec.Decode(aclRole); err != nil {
+				return err
+			}
+
+			// Perform the restoration.
+			if err := restore.ACLRoleRestore(aclRole); err != nil {
+				return err
+			}
 
 		default:
 			// Check if this is an enterprise only object being restored
@@ -2010,6 +2029,36 @@ func (n *nomadFSM) applyDeleteServiceRegistrationByNodeID(msgType structs.Messag
 	return nil
 }
 
+func (n *nomadFSM) applyACLRolesUpsert(msgType structs.MessageType, buf []byte, index uint64) interface{} {
+	defer metrics.MeasureSince([]string{"nomad", "fsm", "apply_acl_role_upsert"}, time.Now())
+	var req structs.ACLRolesUpsertRequest
+	if err := structs.Decode(buf, &req); err != nil {
+		panic(fmt.Errorf("failed to decode request: %v", err))
+	}
+
+	if err := n.state.UpsertACLRoles(msgType, index, req.ACLRoles); err != nil {
+		n.logger.Error("UpsertACLRoles failed", "error", err)
+		return err
+	}
+
+	return nil
+}
+
+func (n *nomadFSM) applyACLRolesDeleteByID(msgType structs.MessageType, buf []byte, index uint64) interface{} {
+	defer metrics.MeasureSince([]string{"nomad", "fsm", "apply_acl_role_delete_by_id"}, time.Now())
+	var req structs.ACLRolesDeleteByIDRequest
+	if err := structs.Decode(buf, &req); err != nil {
+		panic(fmt.Errorf("failed to decode request: %v", err))
+	}
+
+	if err := n.state.DeleteACLRolesByID(msgType, index, req.ACLRoleIDs); err != nil {
+		n.logger.Error("DeleteACLRolesByID failed", "error", err)
+		return err
+	}
+
+	return nil
+}
+
 type FSMFilter struct {
 	evaluator *bexpr.Evaluator
 }
@@ -2218,6 +2267,10 @@ func (s *nomadSnapshot) Persist(sink raft.SnapshotSink) error {
 		sink.Cancel()
 		return err
 	}
+	if err := s.persistACLRoles(sink, encoder); err != nil {
+		sink.Cancel()
+		return err
+	}
 	return nil
 }
 
@@ -2845,6 +2898,33 @@ func (s *nomadSnapshot) persistRootKeyMeta(sink raft.SnapshotSink,
 	return nil
 }
 
+func (s *nomadSnapshot) persistACLRoles(sink raft.SnapshotSink,
+	encoder *codec.Encoder) error {
+
+	// Get all the ACL roles.
+	ws := memdb.NewWatchSet()
+	aclRolesIter, err := s.snap.GetACLRoles(ws)
+	if err != nil {
+		return err
+	}
+
+	for {
+		// Get the next item.
+		for raw := aclRolesIter.Next(); raw != nil; raw = aclRolesIter.Next() {
+
+			// Prepare the request struct.
+			role := raw.(*structs.ACLRole)
+
+			// Write out an ACL role snapshot.
+			sink.Write([]byte{byte(ACLRoleSnapshot)})
+			if err := encoder.Encode(role); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
 // Release is a no-op, as we just need to GC the pointer
 // to the state store snapshot. There is nothing to explicitly
 // cleanup.

nomad/fsm_test.go

@@ -2893,6 +2893,43 @@ func TestFSM_SnapshotRestore_ServiceRegistrations(t *testing.T) {
 	require.ElementsMatch(t, restoredRegs, serviceRegs)
 }
 
+func TestFSM_SnapshotRestore_ACLRoles(t *testing.T) {
+	ci.Parallel(t)
+
+	// Create our initial FSM which will be snapshotted.
+	fsm := testFSM(t)
+	testState := fsm.State()
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, testState.UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Generate and upsert some ACL roles.
+	aclRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+	require.NoError(t, testState.UpsertACLRoles(structs.MsgTypeTestSetup, 10, aclRoles))
+
+	// Perform a snapshot restore.
+	restoredFSM := testSnapshotRestore(t, fsm)
+	restoredState := restoredFSM.State()
+
+	// List the ACL roles from restored state and ensure everything is as
+	// expected.
+	iter, err := restoredState.GetACLRoles(memdb.NewWatchSet())
+	require.NoError(t, err)
+
+	var restoredACLRoles []*structs.ACLRole
+
+	for raw := iter.Next(); raw != nil; raw = iter.Next() {
+		restoredACLRoles = append(restoredACLRoles, raw.(*structs.ACLRole))
+	}
+	require.ElementsMatch(t, restoredACLRoles, aclRoles)
+}
+
 func TestFSM_ReconcileSummaries(t *testing.T) {
 	ci.Parallel(t)
 	// Add some state
@@ -3413,6 +3450,73 @@ func TestFSM_SnapshotRestore_SecureVariables(t *testing.T) {
 	require.ElementsMatch(t, restoredSVs, svs)
 }
 
+func TestFSM_ApplyACLRolesUpsert(t *testing.T) {
+	ci.Parallel(t)
+	fsm := testFSM(t)
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, fsm.State().UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Generate the upsert request and apply the change.
+	req := structs.ACLRolesUpsertRequest{
+		ACLRoles: []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()},
+	}
+	buf, err := structs.Encode(structs.ACLRolesUpsertRequestType, req)
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(makeLog(buf)))
+
+	// Read out both ACL roles and perform an equality check using the hash.
+	ws := memdb.NewWatchSet()
+	out, err := fsm.State().GetACLRoleByName(ws, req.ACLRoles[0].Name)
+	require.NoError(t, err)
+	require.Equal(t, req.ACLRoles[0].Hash, out.Hash)
+
+	out, err = fsm.State().GetACLRoleByName(ws, req.ACLRoles[1].Name)
+	require.NoError(t, err)
+	require.Equal(t, req.ACLRoles[1].Hash, out.Hash)
+}
+
+func TestFSM_ApplyACLRolesDeleteByID(t *testing.T) {
+	ci.Parallel(t)
+	fsm := testFSM(t)
+
+	// Create the policies our ACL roles wants to link to.
+	policy1 := mock.ACLPolicy()
+	policy1.Name = "mocked-test-policy-1"
+	policy2 := mock.ACLPolicy()
+	policy2.Name = "mocked-test-policy-2"
+
+	require.NoError(t, fsm.State().UpsertACLPolicies(
+		structs.MsgTypeTestSetup, 10, []*structs.ACLPolicy{policy1, policy2}))
+
+	// Generate and upsert two ACL roles.
+	aclRoles := []*structs.ACLRole{mock.ACLRole(), mock.ACLRole()}
+	require.NoError(t, fsm.State().UpsertACLRoles(structs.MsgTypeTestSetup, 10, aclRoles))
+
+	// Build and apply our message.
+	req := structs.ACLRolesDeleteByIDRequest{ACLRoleIDs: []string{aclRoles[0].ID, aclRoles[1].ID}}
+	buf, err := structs.Encode(structs.ACLRolesDeleteByIDRequestType, req)
+	require.NoError(t, err)
+	require.Nil(t, fsm.Apply(makeLog(buf)))
+
+	// List all ACL roles within state to ensure both have been removed.
+	ws := memdb.NewWatchSet()
+	iter, err := fsm.State().GetACLRoles(ws)
+	require.NoError(t, err)
+
+	var count int
+	for raw := iter.Next(); raw != nil; raw = iter.Next() {
+		count++
+	}
+	require.Equal(t, 0, count)
+}
+
 func TestFSM_ACLEvents(t *testing.T) {
 	ci.Parallel(t)
 

nomad/mock/mock.go

@@ -2438,3 +2438,19 @@ func mockSecureVariableMetadata() structs.SecureVariableMetadata {
 	}
 	return out
 }
+
+func ACLRole() *structs.ACLRole {
+	role := structs.ACLRole{
+		ID:          uuid.Generate(),
+		Name:        fmt.Sprintf("acl-role-%s", uuid.Short()),
+		Description: "mocked-test-acl-role",
+		Policies: []*structs.ACLRolePolicyLink{
+			{Name: "mocked-test-policy-1"},
+			{Name: "mocked-test-policy-2"},
+		},
+		CreateIndex: 10,
+		ModifyIndex: 10,
+	}
+	role.SetHash()
+	return &role
+}

nomad/state/schema.go

@@ -17,6 +17,7 @@ const (
 	TableSecureVariables       = "secure_variables"
 	TableSecureVariablesQuotas = "secure_variables_quota"
 	TableRootKeyMeta           = "secure_variables_root_key_meta"
+	TableACLRoles              = "acl_roles"
 )
 
 const (
@@ -29,6 +30,7 @@ const (
 	indexExpiresLocal  = "expires-local"
 	indexKeyID         = "key_id"
 	indexPath          = "path"
+	indexName          = "name"
 )
 
 var (
@@ -80,6 +82,7 @@ func init() {
 		secureVariablesTableSchema,
 		secureVariablesQuotasTableSchema,
 		secureVariablesRootKeyMetaSchema,
+		aclRolesTableSchema,
 	}...)
 }
 
@@ -1390,3 +1393,27 @@ func secureVariablesRootKeyMetaSchema() *memdb.TableSchema {
 		},
 	}
 }
+
+func aclRolesTableSchema() *memdb.TableSchema {
+	return &memdb.TableSchema{
+		Name: TableACLRoles,
+		Indexes: map[string]*memdb.IndexSchema{
+			indexID: {
+				Name:         indexID,
+				AllowMissing: false,
+				Unique:       true,
+				Indexer: &memdb.StringFieldIndex{
+					Field: "ID",
+				},
+			},
+			indexName: {
+				Name:         indexName,
+				AllowMissing: false,
+				Unique:       true,
+				Indexer: &memdb.StringFieldIndex{
+					Field: "Name",
+				},
+			},
+		},
+	}
+}