status-im
/
consul
mirror of https://github.com/status-im/consul.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #8282 from hashicorp/dnephin/unmethod-store-funcs-oss 

state: convert methods that don't use their receiver to functions
This commit is contained in:
Daniel Nephin 2020-07-16 16:04:19 -04:00 committed by GitHub
parent 061276007e ecccb30690
commit 4e16e99850
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 514 additions and 515 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

271
agent/consul/state/acl.go
View File

@ -230,7 +230,7 @@ func (s *Snapshot) ACLTokens() (memdb.ResultIterator, error) {
// ACLToken is used when restoring from a snapshot. For general inserts, use ACL.
func (s *Restore) ACLToken(token *structs.ACLToken) error {
return s.store.aclTokenInsert(s.tx, token)
return aclTokenInsert(s.tx, token)
}
// ACLPolicies is used when saving a snapshot
@ -243,7 +243,7 @@ func (s *Snapshot) ACLPolicies() (memdb.ResultIterator, error) {
}
func (s *Restore) ACLPolicy(policy *structs.ACLPolicy) error {
return s.store.aclPolicyInsert(s.tx, policy)
return aclPolicyInsert(s.tx, policy)
}
// ACLRoles is used when saving a snapshot
@ -256,7 +256,7 @@ func (s *Snapshot) ACLRoles() (memdb.ResultIterator, error) {
}
func (s *Restore) ACLRole(role *structs.ACLRole) error {
return s.store.aclRoleInsert(s.tx, role)
return aclRoleInsert(s.tx, role)
}
// ACLBindingRules is used when saving a snapshot
@ -269,7 +269,7 @@ func (s *Snapshot) ACLBindingRules() (memdb.ResultIterator, error) {
}
func (s *Restore) ACLBindingRule(rule *structs.ACLBindingRule) error {
return s.store.aclBindingRuleInsert(s.tx, rule)
return aclBindingRuleInsert(s.tx, rule)
}
// ACLAuthMethods is used when saving a snapshot
@ -282,7 +282,7 @@ func (s *Snapshot) ACLAuthMethods() (memdb.ResultIterator, error) {
}
func (s *Restore) ACLAuthMethod(method *structs.ACLAuthMethod) error {
return s.store.aclAuthMethodInsert(s.tx, method)
return aclAuthMethodInsert(s.tx, method)
}
// ACLBootstrap is used to perform a one-time ACL bootstrap operation on a
@ -304,7 +304,7 @@ func (s *Store) ACLBootstrap(idx, resetIndex uint64, token *structs.ACLToken, le
}
}
if err := s.aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
if err := aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
return fmt.Errorf("failed inserting bootstrap token: %v", err)
}
if err := tx.Insert("index", &IndexEntry{"acl-token-bootstrap", idx}); err != nil {
@ -339,7 +339,7 @@ func (s *Store) CanBootstrapACLToken() (bool, uint64, error) {
// to update the name. Unlike the older functions to operate specifically on role or policy links
// this function does not itself handle the case where the id cannot be found. Instead the
// getName function should handle that and return an error if necessary
func (s *Store) resolveACLLinks(tx *txn, links []agentpb.ACLLink, getName func(*txn, string) (string, error)) (int, error) {
func resolveACLLinks(tx *txn, links []agentpb.ACLLink, getName func(*txn, string) (string, error)) (int, error) {
var numValid int
for linkIndex, link := range links {
if link.ID != "" {
@ -365,7 +365,7 @@ func (s *Store) resolveACLLinks(tx *txn, links []agentpb.ACLLink, getName func(*
// associated with the ID of the link. Ideally this will be a no-op if the names are already correct
// however if a linked resource was renamed it might be stale. This function will treat the incoming
// links with copy-on-write semantics and its output will indicate whether any modifications were made.
func (s *Store) fixupACLLinks(tx *txn, original []agentpb.ACLLink, getName func(*txn, string) (string, error)) ([]agentpb.ACLLink, bool, error) {
func fixupACLLinks(tx *txn, original []agentpb.ACLLink, getName func(*txn, string) (string, error)) ([]agentpb.ACLLink, bool, error) {
owned := false
links := original
@ -405,11 +405,11 @@ func (s *Store) fixupACLLinks(tx *txn, original []agentpb.ACLLink, getName func(
return links, owned, nil
}
func (s *Store) resolveTokenPolicyLinks(tx *txn, token *structs.ACLToken, allowMissing bool) (int, error) {
func resolveTokenPolicyLinks(tx *txn, token *structs.ACLToken, allowMissing bool) (int, error) {
var numValid int
for linkIndex, link := range token.Policies {
if link.ID != "" {
policy, err := s.getPolicyWithTxn(tx, nil, link.ID, s.aclPolicyGetByID, &token.EnterpriseMeta)
policy, err := getPolicyWithTxn(tx, nil, link.ID, aclPolicyGetByID, &token.EnterpriseMeta)
if err != nil {
return 0, err
@ -433,7 +433,7 @@ func (s *Store) resolveTokenPolicyLinks(tx *txn, token *structs.ACLToken, allowM
// stale when a linked policy was deleted or renamed. This will correct them and generate a newly allocated
// token only when fixes are needed. If the policy links are still accurate then we just return the original
// token.
func (s *Store) fixupTokenPolicyLinks(tx *txn, original *structs.ACLToken) (*structs.ACLToken, error) {
func fixupTokenPolicyLinks(tx *txn, original *structs.ACLToken) (*structs.ACLToken, error) {
owned := false
token := original
@ -449,7 +449,7 @@ func (s *Store) fixupTokenPolicyLinks(tx *txn, original *structs.ACLToken) (*str
return nil, fmt.Errorf("Detected corrupted token within the state store - missing policy link ID")
}
policy, err := s.getPolicyWithTxn(tx, nil, link.ID, s.aclPolicyGetByID, &token.EnterpriseMeta)
policy, err := getPolicyWithTxn(tx, nil, link.ID, aclPolicyGetByID, &token.EnterpriseMeta)
if err != nil {
return nil, err
@ -479,11 +479,11 @@ func (s *Store) fixupTokenPolicyLinks(tx *txn, original *structs.ACLToken) (*str
return token, nil
}
func (s *Store) resolveTokenRoleLinks(tx *txn, token *structs.ACLToken, allowMissing bool) (int, error) {
func resolveTokenRoleLinks(tx *txn, token *structs.ACLToken, allowMissing bool) (int, error) {
var numValid int
for linkIndex, link := range token.Roles {
if link.ID != "" {
role, err := s.getRoleWithTxn(tx, nil, link.ID, s.aclRoleGetByID, &token.EnterpriseMeta)
role, err := getRoleWithTxn(tx, nil, link.ID, aclRoleGetByID, &token.EnterpriseMeta)
if err != nil {
return 0, err
@ -507,7 +507,7 @@ func (s *Store) resolveTokenRoleLinks(tx *txn, token *structs.ACLToken, allowMis
// stale when a linked role was deleted or renamed. This will correct them and generate a newly allocated
// token only when fixes are needed. If the role links are still accurate then we just return the original
// token.
func (s *Store) fixupTokenRoleLinks(tx *txn, original *structs.ACLToken) (*structs.ACLToken, error) {
func fixupTokenRoleLinks(tx *txn, original *structs.ACLToken) (*structs.ACLToken, error) {
owned := false
token := original
@ -523,7 +523,7 @@ func (s *Store) fixupTokenRoleLinks(tx *txn, original *structs.ACLToken) (*struc
return nil, fmt.Errorf("Detected corrupted token within the state store - missing role link ID")
}
role, err := s.getRoleWithTxn(tx, nil, link.ID, s.aclRoleGetByID, &original.EnterpriseMeta)
role, err := getRoleWithTxn(tx, nil, link.ID, aclRoleGetByID, &original.EnterpriseMeta)
if err != nil {
return nil, err
@ -553,10 +553,10 @@ func (s *Store) fixupTokenRoleLinks(tx *txn, original *structs.ACLToken) (*struc
return token, nil
}
func (s *Store) resolveRolePolicyLinks(tx *txn, role *structs.ACLRole, allowMissing bool) error {
func resolveRolePolicyLinks(tx *txn, role *structs.ACLRole, allowMissing bool) error {
for linkIndex, link := range role.Policies {
if link.ID != "" {
policy, err := s.getPolicyWithTxn(tx, nil, link.ID, s.aclPolicyGetByID, &role.EnterpriseMeta)
policy, err := getPolicyWithTxn(tx, nil, link.ID, aclPolicyGetByID, &role.EnterpriseMeta)
if err != nil {
return err
@ -579,7 +579,7 @@ func (s *Store) resolveRolePolicyLinks(tx *txn, role *structs.ACLRole, allowMiss
// stale when a linked policy was deleted or renamed. This will correct them and generate a newly allocated
// role only when fixes are needed. If the policy links are still accurate then we just return the original
// role.
func (s *Store) fixupRolePolicyLinks(tx *txn, original *structs.ACLRole) (*structs.ACLRole, error) {
func fixupRolePolicyLinks(tx *txn, original *structs.ACLRole) (*structs.ACLRole, error) {
owned := false
role := original
@ -595,7 +595,7 @@ func (s *Store) fixupRolePolicyLinks(tx *txn, original *structs.ACLRole) (*struc
return nil, fmt.Errorf("Detected corrupted role within the state store - missing policy link ID")
}
policy, err := s.getPolicyWithTxn(tx, nil, link.ID, s.aclPolicyGetByID, &original.EnterpriseMeta)
policy, err := getPolicyWithTxn(tx, nil, link.ID, aclPolicyGetByID, &original.EnterpriseMeta)
if err != nil {
return nil, err
@ -631,7 +631,7 @@ func (s *Store) ACLTokenSet(idx uint64, token *structs.ACLToken, legacy bool) er
defer tx.Abort()
// Call set on the ACL
if err := s.aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
if err := aclTokenSetTxn(tx, idx, token, false, false, false, legacy); err != nil {
return err
}
@ -643,7 +643,7 @@ func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, cas, allo
defer tx.Abort()
for _, token := range tokens {
if err := s.aclTokenSetTxn(tx, idx, token, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, false); err != nil {
if err := aclTokenSetTxn(tx, idx, token, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, false); err != nil {
return err
}
}
@ -653,7 +653,7 @@ func (s *Store) ACLTokenBatchSet(idx uint64, tokens structs.ACLTokens, cas, allo
// aclTokenSetTxn is the inner method used to insert an ACL token with the
// proper indexes into the state store.
func (s *Store) aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, legacy bool) error {
func aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas, allowMissingPolicyAndRoleIDs, prohibitUnprivileged, legacy bool) error {
// Check that the ID is set
if token.SecretID == "" {
return ErrMissingACLTokenSecret
@ -676,7 +676,7 @@ func (s *Store) aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas
// Check for an existing ACL
// DEPRECATED (ACL-Legacy-Compat) - transition to using accessor index instead of secret once v1 compat is removed
_, existing, err := s.aclTokenGetFromIndex(tx, token.SecretID, "id", nil)
_, existing, err := aclTokenGetFromIndex(tx, token.SecretID, "id", nil)
if err != nil {
return fmt.Errorf("failed token lookup: %s", err)
}
@ -710,22 +710,22 @@ func (s *Store) aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas
token.AccessorID = original.AccessorID
}
if err := s.aclTokenUpsertValidateEnterprise(tx, token, original); err != nil {
if err := aclTokenUpsertValidateEnterprise(tx, token, original); err != nil {
return err
}
var numValidPolicies int
if numValidPolicies, err = s.resolveTokenPolicyLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
if numValidPolicies, err = resolveTokenPolicyLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
return err
}
var numValidRoles int
if numValidRoles, err = s.resolveTokenRoleLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
if numValidRoles, err = resolveTokenRoleLinks(tx, token, allowMissingPolicyAndRoleIDs); err != nil {
return err
}
if token.AuthMethod != "" {
method, err := s.getAuthMethodWithTxn(tx, nil, token.AuthMethod, token.ACLAuthMethodEnterpriseMeta.ToEnterpriseMeta())
method, err := getAuthMethodWithTxn(tx, nil, token.AuthMethod, token.ACLAuthMethodEnterpriseMeta.ToEnterpriseMeta())
if err != nil {
return err
} else if method == nil {
@ -774,7 +774,7 @@ func (s *Store) aclTokenSetTxn(tx *txn, idx uint64, token *structs.ACLToken, cas
// ensure that a hash is set
token.SetHash(false)
return s.aclTokenInsert(tx, token)
return aclTokenInsert(tx, token)
}
// ACLTokenGetBySecret is used to look up an existing ACL token by its SecretID.
@ -792,12 +792,12 @@ func (s *Store) aclTokenGet(ws memdb.WatchSet, value, index string, entMeta *str
tx := s.db.Txn(false)
defer tx.Abort()
token, err := s.aclTokenGetTxn(tx, ws, value, index, entMeta)
token, err := aclTokenGetTxn(tx, ws, value, index, entMeta)
if err != nil {
return 0, nil, err
}
idx := s.aclTokenMaxIndex(tx, token, entMeta)
idx := aclTokenMaxIndex(tx, token, entMeta)
return idx, token, nil
}
@ -807,7 +807,7 @@ func (s *Store) ACLTokenBatchGet(ws memdb.WatchSet, accessors []string) (uint64,
tokens := make(structs.ACLTokens, 0)
for _, accessor := range accessors {
token, err := s.aclTokenGetTxn(tx, ws, accessor, "accessor", nil)
token, err := aclTokenGetTxn(tx, ws, accessor, "accessor", nil)
if err != nil {
return 0, nil, fmt.Errorf("failed acl token lookup: %v", err)
}
@ -823,8 +823,8 @@ func (s *Store) ACLTokenBatchGet(ws memdb.WatchSet, accessors []string) (uint64,
return idx, tokens, nil
}
func (s *Store) aclTokenGetTxn(tx *txn, ws memdb.WatchSet, value, index string, entMeta *structs.EnterpriseMeta) (*structs.ACLToken, error) {
watchCh, rawToken, err := s.aclTokenGetFromIndex(tx, value, index, entMeta)
func aclTokenGetTxn(tx *txn, ws memdb.WatchSet, value, index string, entMeta *structs.EnterpriseMeta) (*structs.ACLToken, error) {
watchCh, rawToken, err := aclTokenGetFromIndex(tx, value, index, entMeta)
if err != nil {
return nil, fmt.Errorf("failed acl token lookup: %v", err)
}
@ -832,11 +832,11 @@ func (s *Store) aclTokenGetTxn(tx *txn, ws memdb.WatchSet, value, index string,
if rawToken != nil {
token := rawToken.(*structs.ACLToken)
token, err := s.fixupTokenPolicyLinks(tx, token)
token, err := fixupTokenPolicyLinks(tx, token)
if err != nil {
return nil, err
}
token, err = s.fixupTokenRoleLinks(tx, token)
token, err = fixupTokenRoleLinks(tx, token)
if err != nil {
return nil, err
}
@ -861,11 +861,11 @@ func (s *Store) ACLTokenList(ws memdb.WatchSet, local, global bool, policy, role
needLocalityFilter := false
if policy == "" && role == "" && methodName == "" {
if global == local {
iter, err = s.aclTokenListAll(tx, entMeta)
iter, err = aclTokenListAll(tx, entMeta)
} else if global {
iter, err = s.aclTokenListGlobal(tx, entMeta)
iter, err = aclTokenListGlobal(tx, entMeta)
} else {
iter, err = s.aclTokenListLocal(tx, entMeta)
iter, err = aclTokenListLocal(tx, entMeta)
}
} else if policy != "" && role == "" && methodName == "" {
@ -877,7 +877,7 @@ func (s *Store) ACLTokenList(ws memdb.WatchSet, local, global bool, policy, role
needLocalityFilter = true
} else if policy == "" && role == "" && methodName != "" {
iter, err = s.aclTokenListByAuthMethod(tx, methodName, methodMeta, entMeta)
iter, err = aclTokenListByAuthMethod(tx, methodName, methodMeta, entMeta)
needLocalityFilter = true
} else {
@ -910,11 +910,11 @@ func (s *Store) ACLTokenList(ws memdb.WatchSet, local, global bool, policy, role
var result structs.ACLTokens
for raw := iter.Next(); raw != nil; raw = iter.Next() {
token := raw.(*structs.ACLToken)
token, err := s.fixupTokenPolicyLinks(tx, token)
token, err := fixupTokenPolicyLinks(tx, token)
if err != nil {
return 0, nil, err
}
token, err = s.fixupTokenRoleLinks(tx, token)
token, err = fixupTokenRoleLinks(tx, token)
if err != nil {
return 0, nil, err
}
@ -922,8 +922,7 @@ func (s *Store) ACLTokenList(ws memdb.WatchSet, local, global bool, policy, role
}
// Get the table index.
idx := s.aclTokenMaxIndex(tx, nil, entMeta)
idx := aclTokenMaxIndex(tx, nil, entMeta)
return idx, result, nil
}
@ -1022,7 +1021,7 @@ func (s *Store) ACLTokenBatchDelete(idx uint64, tokenIDs []string) error {
defer tx.Abort()
for _, tokenID := range tokenIDs {
if err := s.aclTokenDeleteTxn(tx, idx, tokenID, "accessor", nil); err != nil {
if err := aclTokenDeleteTxn(tx, idx, tokenID, "accessor", nil); err != nil {
return err
}
}
@ -1034,16 +1033,16 @@ func (s *Store) aclTokenDelete(idx uint64, value, index string, entMeta *structs
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclTokenDeleteTxn(tx, idx, value, index, entMeta); err != nil {
if err := aclTokenDeleteTxn(tx, idx, value, index, entMeta); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclTokenDeleteTxn(tx *txn, idx uint64, value, index string, entMeta *structs.EnterpriseMeta) error {
func aclTokenDeleteTxn(tx *txn, idx uint64, value, index string, entMeta *structs.EnterpriseMeta) error {
// Look up the existing token
_, token, err := s.aclTokenGetFromIndex(tx, value, index, entMeta)
_, token, err := aclTokenGetFromIndex(tx, value, index, entMeta)
if err != nil {
return fmt.Errorf("failed acl token lookup: %v", err)
}
@ -1056,12 +1055,12 @@ func (s *Store) aclTokenDeleteTxn(tx *txn, idx uint64, value, index string, entM
return fmt.Errorf("Deletion of the builtin anonymous token is not permitted")
}
return s.aclTokenDeleteWithToken(tx, token.(*structs.ACLToken), idx)
return aclTokenDeleteWithToken(tx, token.(*structs.ACLToken), idx)
}
func (s *Store) aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, methodMeta *structs.EnterpriseMeta) error {
func aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, methodMeta *structs.EnterpriseMeta) error {
// collect all the tokens linked with the given auth method.
iter, err := s.aclTokenListByAuthMethod(tx, methodName, methodMeta, structs.WildcardEnterpriseMeta())
iter, err := aclTokenListByAuthMethod(tx, methodName, methodMeta, structs.WildcardEnterpriseMeta())
if err != nil {
return fmt.Errorf("failed acl token lookup: %v", err)
}
@ -1075,7 +1074,7 @@ func (s *Store) aclTokenDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodNam
if len(tokens) > 0 {
// delete them all
for _, token := range tokens {
if err := s.aclTokenDeleteWithToken(tx, token, idx); err != nil {
if err := aclTokenDeleteWithToken(tx, token, idx); err != nil {
return err
}
}
@ -1089,7 +1088,7 @@ func (s *Store) ACLPolicyBatchSet(idx uint64, policies structs.ACLPolicies) erro
defer tx.Abort()
for _, policy := range policies {
if err := s.aclPolicySetTxn(tx, idx, policy); err != nil {
if err := aclPolicySetTxn(tx, idx, policy); err != nil {
return err
}
}
@ -1101,14 +1100,14 @@ func (s *Store) ACLPolicySet(idx uint64, policy *structs.ACLPolicy) error {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclPolicySetTxn(tx, idx, policy); err != nil {
if err := aclPolicySetTxn(tx, idx, policy); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclPolicySetTxn(tx *txn, idx uint64, policy *structs.ACLPolicy) error {
func aclPolicySetTxn(tx *txn, idx uint64, policy *structs.ACLPolicy) error {
// Check that the ID is set
if policy.ID == "" {
return ErrMissingACLPolicyID
@ -1119,7 +1118,7 @@ func (s *Store) aclPolicySetTxn(tx *txn, idx uint64, policy *structs.ACLPolicy)
}
var existing *structs.ACLPolicy
_, existingRaw, err := s.aclPolicyGetByID(tx, policy.ID, nil)
_, existingRaw, err := aclPolicyGetByID(tx, policy.ID, nil)
if err != nil {
return err
}
@ -1146,7 +1145,7 @@ func (s *Store) aclPolicySetTxn(tx *txn, idx uint64, policy *structs.ACLPolicy)
}
// ensure the name is unique (cannot conflict with another policy with a different ID)
_, nameMatch, err := s.aclPolicyGetByName(tx, policy.Name, &policy.EnterpriseMeta)
_, nameMatch, err := aclPolicyGetByName(tx, policy.Name, &policy.EnterpriseMeta)
if err != nil {
return err
}
@ -1154,7 +1153,7 @@ func (s *Store) aclPolicySetTxn(tx *txn, idx uint64, policy *structs.ACLPolicy)
return fmt.Errorf("A policy with name %q already exists", policy.Name)
}
if err := s.aclPolicyUpsertValidateEnterprise(tx, policy, existing); err != nil {
if err := aclPolicyUpsertValidateEnterprise(tx, policy, existing); err != nil {
return err
}
@ -1168,15 +1167,15 @@ func (s *Store) aclPolicySetTxn(tx *txn, idx uint64, policy *structs.ACLPolicy)
}
// Insert the ACL
return s.aclPolicyInsert(tx, policy)
return aclPolicyInsert(tx, policy)
}
func (s *Store) ACLPolicyGetByID(ws memdb.WatchSet, id string, entMeta *structs.EnterpriseMeta) (uint64, *structs.ACLPolicy, error) {
return s.aclPolicyGet(ws, id, s.aclPolicyGetByID, entMeta)
return s.aclPolicyGet(ws, id, aclPolicyGetByID, entMeta)
}
func (s *Store) ACLPolicyGetByName(ws memdb.WatchSet, name string, entMeta *structs.EnterpriseMeta) (uint64, *structs.ACLPolicy, error) {
return s.aclPolicyGet(ws, name, s.aclPolicyGetByName, entMeta)
return s.aclPolicyGet(ws, name, aclPolicyGetByName, entMeta)
}
func (s *Store) ACLPolicyBatchGet(ws memdb.WatchSet, ids []string) (uint64, structs.ACLPolicies, error) {
@ -1185,7 +1184,7 @@ func (s *Store) ACLPolicyBatchGet(ws memdb.WatchSet, ids []string) (uint64, stru
policies := make(structs.ACLPolicies, 0)
for _, pid := range ids {
policy, err := s.getPolicyWithTxn(tx, ws, pid, s.aclPolicyGetByID, nil)
policy, err := getPolicyWithTxn(tx, ws, pid, aclPolicyGetByID, nil)
if err != nil {
return 0, nil, err
}
@ -1204,7 +1203,7 @@ func (s *Store) ACLPolicyBatchGet(ws memdb.WatchSet, ids []string) (uint64, stru
type aclPolicyGetFn func(*txn, string, *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error)
func (s *Store) getPolicyWithTxn(tx *txn, ws memdb.WatchSet, value string, fn aclPolicyGetFn, entMeta *structs.EnterpriseMeta) (*structs.ACLPolicy, error) {
func getPolicyWithTxn(tx *txn, ws memdb.WatchSet, value string, fn aclPolicyGetFn, entMeta *structs.EnterpriseMeta) (*structs.ACLPolicy, error) {
watchCh, policy, err := fn(tx, value, entMeta)
if err != nil {
return nil, fmt.Errorf("failed acl policy lookup: %v", err)
@ -1222,12 +1221,12 @@ func (s *Store) aclPolicyGet(ws memdb.WatchSet, value string, fn aclPolicyGetFn,
tx := s.db.Txn(false)
defer tx.Abort()
policy, err := s.getPolicyWithTxn(tx, ws, value, fn, entMeta)
policy, err := getPolicyWithTxn(tx, ws, value, fn, entMeta)
if err != nil {
return 0, nil, err
}
idx := s.aclPolicyMaxIndex(tx, policy, entMeta)
idx := aclPolicyMaxIndex(tx, policy, entMeta)
return idx, policy, nil
}
@ -1236,7 +1235,7 @@ func (s *Store) ACLPolicyList(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta
tx := s.db.Txn(false)
defer tx.Abort()
iter, err := s.aclPolicyList(tx, entMeta)
iter, err := aclPolicyList(tx, entMeta)
if err != nil {
return 0, nil, fmt.Errorf("failed acl policy lookup: %v", err)
}
@ -1248,17 +1247,17 @@ func (s *Store) ACLPolicyList(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta
}
// Get the table index.
idx := s.aclPolicyMaxIndex(tx, nil, entMeta)
idx := aclPolicyMaxIndex(tx, nil, entMeta)
return idx, result, nil
}
func (s *Store) ACLPolicyDeleteByID(idx uint64, id string, entMeta *structs.EnterpriseMeta) error {
return s.aclPolicyDelete(idx, id, s.aclPolicyGetByID, entMeta)
return s.aclPolicyDelete(idx, id, aclPolicyGetByID, entMeta)
}
func (s *Store) ACLPolicyDeleteByName(idx uint64, name string, entMeta *structs.EnterpriseMeta) error {
return s.aclPolicyDelete(idx, name, s.aclPolicyGetByName, entMeta)
return s.aclPolicyDelete(idx, name, aclPolicyGetByName, entMeta)
}
func (s *Store) ACLPolicyBatchDelete(idx uint64, policyIDs []string) error {
@ -1266,7 +1265,7 @@ func (s *Store) ACLPolicyBatchDelete(idx uint64, policyIDs []string) error {
defer tx.Abort()
for _, policyID := range policyIDs {
if err := s.aclPolicyDeleteTxn(tx, idx, policyID, s.aclPolicyGetByID, nil); err != nil {
if err := aclPolicyDeleteTxn(tx, idx, policyID, aclPolicyGetByID, nil); err != nil {
return err
}
}
@ -1277,14 +1276,14 @@ func (s *Store) aclPolicyDelete(idx uint64, value string, fn aclPolicyGetFn, ent
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclPolicyDeleteTxn(tx, idx, value, fn, entMeta); err != nil {
if err := aclPolicyDeleteTxn(tx, idx, value, fn, entMeta); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclPolicyDeleteTxn(tx *txn, idx uint64, value string, fn aclPolicyGetFn, entMeta *structs.EnterpriseMeta) error {
func aclPolicyDeleteTxn(tx *txn, idx uint64, value string, fn aclPolicyGetFn, entMeta *structs.EnterpriseMeta) error {
// Look up the existing token
_, rawPolicy, err := fn(tx, value, entMeta)
if err != nil {
@ -1301,7 +1300,7 @@ func (s *Store) aclPolicyDeleteTxn(tx *txn, idx uint64, value string, fn aclPoli
return fmt.Errorf("Deletion of the builtin global-management policy is not permitted")
}
return s.aclPolicyDeleteWithPolicy(tx, policy, idx)
return aclPolicyDeleteWithPolicy(tx, policy, idx)
}
func (s *Store) ACLRoleBatchSet(idx uint64, roles structs.ACLRoles, allowMissingPolicyIDs bool) error {
@ -1309,7 +1308,7 @@ func (s *Store) ACLRoleBatchSet(idx uint64, roles structs.ACLRoles, allowMissing
defer tx.Abort()
for _, role := range roles {
if err := s.aclRoleSetTxn(tx, idx, role, allowMissingPolicyIDs); err != nil {
if err := aclRoleSetTxn(tx, idx, role, allowMissingPolicyIDs); err != nil {
return err
}
}
@ -1321,14 +1320,14 @@ func (s *Store) ACLRoleSet(idx uint64, role *structs.ACLRole) error {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclRoleSetTxn(tx, idx, role, false); err != nil {
if err := aclRoleSetTxn(tx, idx, role, false); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclRoleSetTxn(tx *txn, idx uint64, role *structs.ACLRole, allowMissing bool) error {
func aclRoleSetTxn(tx *txn, idx uint64, role *structs.ACLRole, allowMissing bool) error {
// Check that the ID is set
if role.ID == "" {
return ErrMissingACLRoleID
@ -1338,7 +1337,7 @@ func (s *Store) aclRoleSetTxn(tx *txn, idx uint64, role *structs.ACLRole, allowM
return ErrMissingACLRoleName
}
_, existingRaw, err := s.aclRoleGetByID(tx, role.ID, nil)
_, existingRaw, err := aclRoleGetByID(tx, role.ID, nil)
if err != nil {
return fmt.Errorf("failed acl role lookup: %v", err)
}
@ -1349,7 +1348,7 @@ func (s *Store) aclRoleSetTxn(tx *txn, idx uint64, role *structs.ACLRole, allowM
}
// ensure the name is unique (cannot conflict with another role with a different ID)
_, nameMatch, err := s.aclRoleGetByName(tx, role.Name, &role.EnterpriseMeta)
_, nameMatch, err := aclRoleGetByName(tx, role.Name, &role.EnterpriseMeta)
if err != nil {
return fmt.Errorf("failed acl role lookup: %v", err)
}
@ -1357,7 +1356,7 @@ func (s *Store) aclRoleSetTxn(tx *txn, idx uint64, role *structs.ACLRole, allowM
return fmt.Errorf("A role with name %q already exists", role.Name)
}
if err := s.resolveRolePolicyLinks(tx, role, allowMissing); err != nil {
if err := resolveRolePolicyLinks(tx, role, allowMissing); err != nil {
return err
}
@ -1376,7 +1375,7 @@ func (s *Store) aclRoleSetTxn(tx *txn, idx uint64, role *structs.ACLRole, allowM
}
}
if err := s.aclRoleUpsertValidateEnterprise(tx, role, existing); err != nil {
if err := aclRoleUpsertValidateEnterprise(tx, role, existing); err != nil {
return err
}
@ -1389,17 +1388,17 @@ func (s *Store) aclRoleSetTxn(tx *txn, idx uint64, role *structs.ACLRole, allowM
role.ModifyIndex = idx
}
return s.aclRoleInsert(tx, role)
return aclRoleInsert(tx, role)
}
type aclRoleGetFn func(*txn, string, *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error)
func (s *Store) ACLRoleGetByID(ws memdb.WatchSet, id string, entMeta *structs.EnterpriseMeta) (uint64, *structs.ACLRole, error) {
return s.aclRoleGet(ws, id, s.aclRoleGetByID, entMeta)
return s.aclRoleGet(ws, id, aclRoleGetByID, entMeta)
}
func (s *Store) ACLRoleGetByName(ws memdb.WatchSet, name string, entMeta *structs.EnterpriseMeta) (uint64, *structs.ACLRole, error) {
return s.aclRoleGet(ws, name, s.aclRoleGetByName, entMeta)
return s.aclRoleGet(ws, name, aclRoleGetByName, entMeta)
}
func (s *Store) ACLRoleBatchGet(ws memdb.WatchSet, ids []string) (uint64, structs.ACLRoles, error) {
@ -1408,7 +1407,7 @@ func (s *Store) ACLRoleBatchGet(ws memdb.WatchSet, ids []string) (uint64, struct
roles := make(structs.ACLRoles, 0, len(ids))
for _, rid := range ids {
role, err := s.getRoleWithTxn(tx, ws, rid, s.aclRoleGetByID, nil)
role, err := getRoleWithTxn(tx, ws, rid, aclRoleGetByID, nil)
if err != nil {
return 0, nil, err
}
@ -1423,7 +1422,7 @@ func (s *Store) ACLRoleBatchGet(ws memdb.WatchSet, ids []string) (uint64, struct
return idx, roles, nil
}
func (s *Store) getRoleWithTxn(tx *txn, ws memdb.WatchSet, value string, fn aclRoleGetFn, entMeta *structs.EnterpriseMeta) (*structs.ACLRole, error) {
func getRoleWithTxn(tx *txn, ws memdb.WatchSet, value string, fn aclRoleGetFn, entMeta *structs.EnterpriseMeta) (*structs.ACLRole, error) {
watchCh, rawRole, err := fn(tx, value, entMeta)
if err != nil {
return nil, fmt.Errorf("failed acl role lookup: %v", err)
@ -1432,7 +1431,7 @@ func (s *Store) getRoleWithTxn(tx *txn, ws memdb.WatchSet, value string, fn aclR
if rawRole != nil {
role := rawRole.(*structs.ACLRole)
role, err := s.fixupRolePolicyLinks(tx, role)
role, err := fixupRolePolicyLinks(tx, role)
if err != nil {
return nil, err
}
@ -1446,12 +1445,12 @@ func (s *Store) aclRoleGet(ws memdb.WatchSet, value string, fn aclRoleGetFn, ent
tx := s.db.Txn(false)
defer tx.Abort()
role, err := s.getRoleWithTxn(tx, ws, value, fn, entMeta)
role, err := getRoleWithTxn(tx, ws, value, fn, entMeta)
if err != nil {
return 0, nil, err
}
idx := s.aclRoleMaxIndex(tx, role, entMeta)
idx := aclRoleMaxIndex(tx, role, entMeta)
return idx, role, nil
}
@ -1466,7 +1465,7 @@ func (s *Store) ACLRoleList(ws memdb.WatchSet, policy string, entMeta *structs.E
if policy != "" {
iter, err = aclRoleListByPolicy(tx, policy, entMeta)
} else {
iter, err = s.aclRoleList(tx, entMeta)
iter, err = aclRoleList(tx, entMeta)
}
if err != nil {
@ -1477,7 +1476,7 @@ func (s *Store) ACLRoleList(ws memdb.WatchSet, policy string, entMeta *structs.E
var result structs.ACLRoles
for raw := iter.Next(); raw != nil; raw = iter.Next() {
role := raw.(*structs.ACLRole)
role, err := s.fixupRolePolicyLinks(tx, role)
role, err := fixupRolePolicyLinks(tx, role)
if err != nil {
return 0, nil, err
}
@ -1485,17 +1484,17 @@ func (s *Store) ACLRoleList(ws memdb.WatchSet, policy string, entMeta *structs.E
}
// Get the table index.
idx := s.aclRoleMaxIndex(tx, nil, entMeta)
idx := aclRoleMaxIndex(tx, nil, entMeta)
return idx, result, nil
}
func (s *Store) ACLRoleDeleteByID(idx uint64, id string, entMeta *structs.EnterpriseMeta) error {
return s.aclRoleDelete(idx, id, s.aclRoleGetByID, entMeta)
return s.aclRoleDelete(idx, id, aclRoleGetByID, entMeta)
}
func (s *Store) ACLRoleDeleteByName(idx uint64, name string, entMeta *structs.EnterpriseMeta) error {
return s.aclRoleDelete(idx, name, s.aclRoleGetByName, entMeta)
return s.aclRoleDelete(idx, name, aclRoleGetByName, entMeta)
}
func (s *Store) ACLRoleBatchDelete(idx uint64, roleIDs []string) error {
@ -1503,7 +1502,7 @@ func (s *Store) ACLRoleBatchDelete(idx uint64, roleIDs []string) error {
defer tx.Abort()
for _, roleID := range roleIDs {
if err := s.aclRoleDeleteTxn(tx, idx, roleID, s.aclRoleGetByID, nil); err != nil {
if err := aclRoleDeleteTxn(tx, idx, roleID, aclRoleGetByID, nil); err != nil {
return err
}
}
@ -1514,14 +1513,14 @@ func (s *Store) aclRoleDelete(idx uint64, value string, fn aclRoleGetFn, entMeta
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclRoleDeleteTxn(tx, idx, value, fn, entMeta); err != nil {
if err := aclRoleDeleteTxn(tx, idx, value, fn, entMeta); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclRoleDeleteTxn(tx *txn, idx uint64, value string, fn aclRoleGetFn, entMeta *structs.EnterpriseMeta) error {
func aclRoleDeleteTxn(tx *txn, idx uint64, value string, fn aclRoleGetFn, entMeta *structs.EnterpriseMeta) error {
// Look up the existing role
_, rawRole, err := fn(tx, value, entMeta)
if err != nil {
@ -1534,7 +1533,7 @@ func (s *Store) aclRoleDeleteTxn(tx *txn, idx uint64, value string, fn aclRoleGe
role := rawRole.(*structs.ACLRole)
return s.aclRoleDeleteWithRole(tx, role, idx)
return aclRoleDeleteWithRole(tx, role, idx)
}
func (s *Store) ACLBindingRuleBatchSet(idx uint64, rules structs.ACLBindingRules) error {
@ -1542,7 +1541,7 @@ func (s *Store) ACLBindingRuleBatchSet(idx uint64, rules structs.ACLBindingRules
defer tx.Abort()
for _, rule := range rules {
if err := s.aclBindingRuleSetTxn(tx, idx, rule); err != nil {
if err := aclBindingRuleSetTxn(tx, idx, rule); err != nil {
return err
}
}
@ -1554,13 +1553,13 @@ func (s *Store) ACLBindingRuleSet(idx uint64, rule *structs.ACLBindingRule) erro
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclBindingRuleSetTxn(tx, idx, rule); err != nil {
if err := aclBindingRuleSetTxn(tx, idx, rule); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclBindingRuleSetTxn(tx *txn, idx uint64, rule *structs.ACLBindingRule) error {
func aclBindingRuleSetTxn(tx *txn, idx uint64, rule *structs.ACLBindingRule) error {
// Check that the ID and AuthMethod are set
if rule.ID == "" {
return ErrMissingACLBindingRuleID
@ -1569,7 +1568,7 @@ func (s *Store) aclBindingRuleSetTxn(tx *txn, idx uint64, rule *structs.ACLBindi
}
var existing *structs.ACLBindingRule
_, existingRaw, err := s.aclBindingRuleGetByID(tx, rule.ID, nil)
_, existingRaw, err := aclBindingRuleGetByID(tx, rule.ID, nil)
if err != nil {
return fmt.Errorf("failed acl binding rule lookup: %v", err)
}
@ -1584,17 +1583,17 @@ func (s *Store) aclBindingRuleSetTxn(tx *txn, idx uint64, rule *structs.ACLBindi
rule.ModifyIndex = idx
}
if err := s.aclBindingRuleUpsertValidateEnterprise(tx, rule, existing); err != nil {
if err := aclBindingRuleUpsertValidateEnterprise(tx, rule, existing); err != nil {
return err
}
if _, method, err := s.aclAuthMethodGetByName(tx, rule.AuthMethod, &rule.EnterpriseMeta); err != nil {
if _, method, err := aclAuthMethodGetByName(tx, rule.AuthMethod, &rule.EnterpriseMeta); err != nil {
return fmt.Errorf("failed acl auth method lookup: %v", err)
} else if method == nil {
return fmt.Errorf("failed inserting acl binding rule: auth method not found")
}
return s.aclBindingRuleInsert(tx, rule)
return aclBindingRuleInsert(tx, rule)
}
func (s *Store) ACLBindingRuleGetByID(ws memdb.WatchSet, id string, entMeta *structs.EnterpriseMeta) (uint64, *structs.ACLBindingRule, error) {
@ -1605,7 +1604,7 @@ func (s *Store) aclBindingRuleGet(ws memdb.WatchSet, value string, entMeta *stru
tx := s.db.Txn(false)
defer tx.Abort()
watchCh, rawRule, err := s.aclBindingRuleGetByID(tx, value, entMeta)
watchCh, rawRule, err := aclBindingRuleGetByID(tx, value, entMeta)
if err != nil {
return 0, nil, fmt.Errorf("failed acl binding rule lookup: %v", err)
}
@ -1616,7 +1615,7 @@ func (s *Store) aclBindingRuleGet(ws memdb.WatchSet, value string, entMeta *stru
rule = rawRule.(*structs.ACLBindingRule)
}
idx := s.aclBindingRuleMaxIndex(tx, rule, entMeta)
idx := aclBindingRuleMaxIndex(tx, rule, entMeta)
return idx, rule, nil
}
@ -1630,9 +1629,9 @@ func (s *Store) ACLBindingRuleList(ws memdb.WatchSet, methodName string, entMeta
err error
)
if methodName != "" {
iter, err = s.aclBindingRuleListByAuthMethod(tx, methodName, entMeta)
iter, err = aclBindingRuleListByAuthMethod(tx, methodName, entMeta)
} else {
iter, err = s.aclBindingRuleList(tx, entMeta)
iter, err = aclBindingRuleList(tx, entMeta)
}
if err != nil {
return 0, nil, fmt.Errorf("failed acl binding rule lookup: %v", err)
@ -1646,7 +1645,7 @@ func (s *Store) ACLBindingRuleList(ws memdb.WatchSet, methodName string, entMeta
}
// Get the table index.
idx := s.aclBindingRuleMaxIndex(tx, nil, entMeta)
idx := aclBindingRuleMaxIndex(tx, nil, entMeta)
return idx, result, nil
}
@ -1660,7 +1659,7 @@ func (s *Store) ACLBindingRuleBatchDelete(idx uint64, bindingRuleIDs []string) e
defer tx.Abort()
for _, bindingRuleID := range bindingRuleIDs {
s.aclBindingRuleDeleteTxn(tx, idx, bindingRuleID, nil)
aclBindingRuleDeleteTxn(tx, idx, bindingRuleID, nil)
}
return tx.Commit()
}
@ -1669,16 +1668,16 @@ func (s *Store) aclBindingRuleDelete(idx uint64, id string, entMeta *structs.Ent
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclBindingRuleDeleteTxn(tx, idx, id, entMeta); err != nil {
if err := aclBindingRuleDeleteTxn(tx, idx, id, entMeta); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclBindingRuleDeleteTxn(tx *txn, idx uint64, id string, entMeta *structs.EnterpriseMeta) error {
func aclBindingRuleDeleteTxn(tx *txn, idx uint64, id string, entMeta *structs.EnterpriseMeta) error {
// Look up the existing binding rule
_, rawRule, err := s.aclBindingRuleGetByID(tx, id, entMeta)
_, rawRule, err := aclBindingRuleGetByID(tx, id, entMeta)
if err != nil {
return fmt.Errorf("failed acl binding rule lookup: %v", err)
}
@ -1689,15 +1688,15 @@ func (s *Store) aclBindingRuleDeleteTxn(tx *txn, idx uint64, id string, entMeta
rule := rawRule.(*structs.ACLBindingRule)
if err := s.aclBindingRuleDeleteWithRule(tx, rule, idx); err != nil {
if err := aclBindingRuleDeleteWithRule(tx, rule, idx); err != nil {
return fmt.Errorf("failed deleting acl binding rule: %v", err)
}
return nil
}
func (s *Store) aclBindingRuleDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, entMeta *structs.EnterpriseMeta) error {
func aclBindingRuleDeleteAllForAuthMethodTxn(tx *txn, idx uint64, methodName string, entMeta *structs.EnterpriseMeta) error {
// collect them all
iter, err := s.aclBindingRuleListByAuthMethod(tx, methodName, entMeta)
iter, err := aclBindingRuleListByAuthMethod(tx, methodName, entMeta)
if err != nil {
return fmt.Errorf("failed acl binding rule lookup: %v", err)
}
@ -1711,7 +1710,7 @@ func (s *Store) aclBindingRuleDeleteAllForAuthMethodTxn(tx *txn, idx uint64, met
if len(rules) > 0 {
// delete them all
for _, rule := range rules {
if err := s.aclBindingRuleDeleteWithRule(tx, rule, idx); err != nil {
if err := aclBindingRuleDeleteWithRule(tx, rule, idx); err != nil {
return err
}
}
@ -1727,7 +1726,7 @@ func (s *Store) ACLAuthMethodBatchSet(idx uint64, methods structs.ACLAuthMethods
for _, method := range methods {
// this is only used when doing batch insertions for upgrades and replication. Therefore
// we take whatever those said.
if err := s.aclAuthMethodSetTxn(tx, idx, method); err != nil {
if err := aclAuthMethodSetTxn(tx, idx, method); err != nil {
return err
}
}
@ -1738,14 +1737,14 @@ func (s *Store) ACLAuthMethodSet(idx uint64, method *structs.ACLAuthMethod) erro
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclAuthMethodSetTxn(tx, idx, method); err != nil {
if err := aclAuthMethodSetTxn(tx, idx, method); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclAuthMethodSetTxn(tx *txn, idx uint64, method *structs.ACLAuthMethod) error {
func aclAuthMethodSetTxn(tx *txn, idx uint64, method *structs.ACLAuthMethod) error {
// Check that the Name and Type are set
if method.Name == "" {
return ErrMissingACLAuthMethodName
@ -1754,12 +1753,12 @@ func (s *Store) aclAuthMethodSetTxn(tx *txn, idx uint64, method *structs.ACLAuth
}
var existing *structs.ACLAuthMethod
_, existingRaw, err := s.aclAuthMethodGetByName(tx, method.Name, &method.EnterpriseMeta)
_, existingRaw, err := aclAuthMethodGetByName(tx, method.Name, &method.EnterpriseMeta)
if err != nil {
return fmt.Errorf("failed acl auth method lookup: %v", err)
}
if err := s.aclAuthMethodUpsertValidateEnterprise(tx, method, existing); err != nil {
if err := aclAuthMethodUpsertValidateEnterprise(tx, method, existing); err != nil {
return err
}
@ -1773,7 +1772,7 @@ func (s *Store) aclAuthMethodSetTxn(tx *txn, idx uint64, method *structs.ACLAuth
method.ModifyIndex = idx
}
return s.aclAuthMethodInsert(tx, method)
return aclAuthMethodInsert(tx, method)
}
func (s *Store) ACLAuthMethodGetByName(ws memdb.WatchSet, name string, entMeta *structs.EnterpriseMeta) (uint64, *structs.ACLAuthMethod, error) {
@ -1784,18 +1783,18 @@ func (s *Store) aclAuthMethodGet(ws memdb.WatchSet, name string, entMeta *struct
tx := s.db.Txn(false)
defer tx.Abort()
method, err := s.getAuthMethodWithTxn(tx, ws, name, entMeta)
method, err := getAuthMethodWithTxn(tx, ws, name, entMeta)
if err != nil {
return 0, nil, err
}
idx := s.aclAuthMethodMaxIndex(tx, method, entMeta)
idx := aclAuthMethodMaxIndex(tx, method, entMeta)
return idx, method, nil
}
func (s *Store) getAuthMethodWithTxn(tx *txn, ws memdb.WatchSet, name string, entMeta *structs.EnterpriseMeta) (*structs.ACLAuthMethod, error) {
watchCh, rawMethod, err := s.aclAuthMethodGetByName(tx, name, entMeta)
func getAuthMethodWithTxn(tx *txn, ws memdb.WatchSet, name string, entMeta *structs.EnterpriseMeta) (*structs.ACLAuthMethod, error) {
watchCh, rawMethod, err := aclAuthMethodGetByName(tx, name, entMeta)
if err != nil {
return nil, fmt.Errorf("failed acl auth method lookup: %v", err)
}
@ -1812,7 +1811,7 @@ func (s *Store) ACLAuthMethodList(ws memdb.WatchSet, entMeta *structs.Enterprise
tx := s.db.Txn(false)
defer tx.Abort()
iter, err := s.aclAuthMethodList(tx, entMeta)
iter, err := aclAuthMethodList(tx, entMeta)
if err != nil {
return 0, nil, fmt.Errorf("failed acl auth method lookup: %v", err)
}
@ -1825,7 +1824,7 @@ func (s *Store) ACLAuthMethodList(ws memdb.WatchSet, entMeta *structs.Enterprise
}
// Get the table index.
idx := s.aclAuthMethodMaxIndex(tx, nil, entMeta)
idx := aclAuthMethodMaxIndex(tx, nil, entMeta)
return idx, result, nil
}
@ -1843,7 +1842,7 @@ func (s *Store) ACLAuthMethodBatchDelete(idx uint64, names []string, entMeta *st
// deleted. However we never actually batch these deletions as auth methods are not replicated
// Therefore this is fine but if we ever change that precondition then this will be wrong (unless
// we ensure all deletions in a batch should have the same enterprise meta)
s.aclAuthMethodDeleteTxn(tx, idx, name, entMeta)
aclAuthMethodDeleteTxn(tx, idx, name, entMeta)
}
return tx.Commit()
@ -1853,16 +1852,16 @@ func (s *Store) aclAuthMethodDelete(idx uint64, name string, entMeta *structs.En
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.aclAuthMethodDeleteTxn(tx, idx, name, entMeta); err != nil {
if err := aclAuthMethodDeleteTxn(tx, idx, name, entMeta); err != nil {
return err
}
return tx.Commit()
}
func (s *Store) aclAuthMethodDeleteTxn(tx *txn, idx uint64, name string, entMeta *structs.EnterpriseMeta) error {
func aclAuthMethodDeleteTxn(tx *txn, idx uint64, name string, entMeta *structs.EnterpriseMeta) error {
// Look up the existing method
_, rawMethod, err := s.aclAuthMethodGetByName(tx, name, entMeta)
_, rawMethod, err := aclAuthMethodGetByName(tx, name, entMeta)
if err != nil {
return fmt.Errorf("failed acl auth method lookup: %v", err)
}
@ -1873,13 +1872,13 @@ func (s *Store) aclAuthMethodDeleteTxn(tx *txn, idx uint64, name string, entMeta
method := rawMethod.(*structs.ACLAuthMethod)
if err := s.aclBindingRuleDeleteAllForAuthMethodTxn(tx, idx, method.Name, entMeta); err != nil {
if err := aclBindingRuleDeleteAllForAuthMethodTxn(tx, idx, method.Name, entMeta); err != nil {
return err
}
if err := s.aclTokenDeleteAllForAuthMethodTxn(tx, idx, method.Name, entMeta); err != nil {
if err := aclTokenDeleteAllForAuthMethodTxn(tx, idx, method.Name, entMeta); err != nil {
return err
}
return s.aclAuthMethodDeleteWithMethod(tx, method, idx)
return aclAuthMethodDeleteWithMethod(tx, method, idx)
}

56
agent/consul/state/acl_events_test.go
View File

@ -13,39 +13,39 @@ import (
func TestACLChangeUnsubscribeEvent(t *testing.T) {
cases := []struct {
Name string
Setup func(s *Store, tx *txn) error
Mutate func(s *Store, tx *txn) error
Setup func(tx *txn) error
Mutate func(tx *txn) error
expected stream.Event
}{
{
Name: "token create",
Mutate: func(s *Store, tx *txn) error {
return s.aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
Mutate: func(tx *txn) error {
return aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
},
expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
},
{
Name: "token update",
Setup: func(s *Store, tx *txn) error {
return s.aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
Setup: func(tx *txn) error {
return aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
},
Mutate: func(s *Store, tx *txn) error {
Mutate: func(tx *txn) error {
// Add a policy to the token (never mind it doesn't exist for now) we
// allow it in the set command below.
token := newACLToken(1)
token.Policies = []structs.ACLTokenPolicyLink{{ID: "33333333-1111-1111-1111-111111111111"}}
return s.aclTokenSetTxn(tx, tx.Index, token, false, true, false, false)
return aclTokenSetTxn(tx, tx.Index, token, false, true, false, false)
},
expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
},
{
Name: "token delete",
Setup: func(s *Store, tx *txn) error {
return s.aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
Setup: func(tx *txn) error {
return aclTokenSetTxn(tx, tx.Index, newACLToken(1), false, false, false, false)
},
Mutate: func(s *Store, tx *txn) error {
Mutate: func(tx *txn) error {
token := newACLToken(1)
return s.aclTokenDeleteTxn(tx, tx.Index, token.AccessorID, "id", nil)
return aclTokenDeleteTxn(tx, tx.Index, token.AccessorID, "id", nil)
},
expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
},
@ -58,19 +58,19 @@ func TestACLChangeUnsubscribeEvent(t *testing.T) {
{
Name: "policy update",
Setup: newACLPolicyWithSingleToken,
Mutate: func(s *Store, tx *txn) error {
Mutate: func(tx *txn) error {
policy := newACLPolicy(1)
policy.Rules = `operator = "write"`
return s.aclPolicySetTxn(tx, tx.Index, policy)
return aclPolicySetTxn(tx, tx.Index, policy)
},
expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
},
{
Name: "policy delete",
Setup: newACLPolicyWithSingleToken,
Mutate: func(s *Store, tx *txn) error {
Mutate: func(tx *txn) error {
policy := newACLPolicy(1)
return s.aclPolicyDeleteTxn(tx, tx.Index, policy.ID, s.aclPolicyGetByID, nil)
return aclPolicyDeleteTxn(tx, tx.Index, policy.ID, aclPolicyGetByID, nil)
},
expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
},
@ -83,23 +83,23 @@ func TestACLChangeUnsubscribeEvent(t *testing.T) {
{
Name: "role update",
Setup: newACLRoleWithSingleToken,
Mutate: func(s *Store, tx *txn) error {
Mutate: func(tx *txn) error {
role := newACLRole(1, newACLRolePolicyLink(1))
policy2 := newACLPolicy(2)
role.Policies = append(role.Policies, structs.ACLRolePolicyLink{
ID: policy2.ID,
Name: policy2.Name,
})
return s.aclRoleSetTxn(tx, tx.Index, role, true)
return aclRoleSetTxn(tx, tx.Index, role, true)
},
expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
},
{
Name: "role delete",
Setup: newACLRoleWithSingleToken,
Mutate: func(s *Store, tx *txn) error {
Mutate: func(tx *txn) error {
role := newACLRole(1, newACLRolePolicyLink(1))
return s.aclRoleDeleteTxn(tx, tx.Index, role.ID, s.aclRoleGetByID, nil)
return aclRoleDeleteTxn(tx, tx.Index, role.ID, aclRoleGetByID, nil)
},
expected: stream.NewCloseSubscriptionEvent(newSecretIDs(1)),
},
@ -114,7 +114,7 @@ func TestACLChangeUnsubscribeEvent(t *testing.T) {
// Bypass the publish mechanism for this test or we get into odd
// recursive stuff...
setupTx := s.db.WriteTxn(10)
require.NoError(t, tc.Setup(s, setupTx))
require.NoError(t, tc.Setup(setupTx))
// Commit the underlying transaction without using wrapped Commit so we
// avoid the whole event publishing system for setup here. It _should_
// work but it makes debugging test hard as it will call the function
@ -123,7 +123,7 @@ func TestACLChangeUnsubscribeEvent(t *testing.T) {
}
tx := s.db.WriteTxn(100)
require.NoError(t, tc.Mutate(s, tx))
require.NoError(t, tc.Mutate(tx))
// Note we call the func under test directly rather than publishChanges so
// we can test this in isolation.
@ -137,24 +137,24 @@ func TestACLChangeUnsubscribeEvent(t *testing.T) {
}
}
func newACLRoleWithSingleToken(s *Store, tx *txn) error {
func newACLRoleWithSingleToken(tx *txn) error {
role := newACLRole(1, newACLRolePolicyLink(1))
if err := s.aclRoleSetTxn(tx, tx.Index, role, true); err != nil {
if err := aclRoleSetTxn(tx, tx.Index, role, true); err != nil {
return err
}
token := newACLToken(1)
token.Roles = append(token.Roles, structs.ACLTokenRoleLink{ID: role.ID})
return s.aclTokenSetTxn(tx, tx.Index, token, false, false, false, false)
return aclTokenSetTxn(tx, tx.Index, token, false, false, false, false)
}
func newACLPolicyWithSingleToken(s *Store, tx *txn) error {
func newACLPolicyWithSingleToken(tx *txn) error {
policy := newACLPolicy(1)
if err := s.aclPolicySetTxn(tx, tx.Index, policy); err != nil {
if err := aclPolicySetTxn(tx, tx.Index, policy); err != nil {
return err
}
token := newACLToken(1)
token.Policies = append(token.Policies, structs.ACLTokenPolicyLink{ID: policy.ID})
return s.aclTokenSetTxn(tx, tx.Index, token, false, false, false, false)
return aclTokenSetTxn(tx, tx.Index, token, false, false, false, false)
}
func newSecretIDs(ids ...int) []string {

72
agent/consul/state/acl_oss.go
View File

@ -206,7 +206,7 @@ func authMethodsTableSchema() *memdb.TableSchema {
///// ACL Policy Functions /////
///////////////////////////////////////////////////////////////////////////////
func (s *Store) aclPolicyInsert(tx *txn, policy *structs.ACLPolicy) error {
func aclPolicyInsert(tx *txn, policy *structs.ACLPolicy) error {
if err := tx.Insert("acl-policies", policy); err != nil {
return fmt.Errorf("failed inserting acl policy: %v", err)
}
@ -218,19 +218,19 @@ func (s *Store) aclPolicyInsert(tx *txn, policy *structs.ACLPolicy) error {
return nil
}
func (s *Store) aclPolicyGetByID(tx *txn, id string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func aclPolicyGetByID(tx *txn, id string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("acl-policies", "id", id)
}
func (s *Store) aclPolicyGetByName(tx *txn, name string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func aclPolicyGetByName(tx *txn, name string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("acl-policies", "name", name)
}
func (s *Store) aclPolicyList(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclPolicyList(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-policies", "id")
}
func (s *Store) aclPolicyDeleteWithPolicy(tx *txn, policy *structs.ACLPolicy, idx uint64) error {
func aclPolicyDeleteWithPolicy(tx *txn, policy *structs.ACLPolicy, idx uint64) error {
// remove the policy
if err := tx.Delete("acl-policies", policy); err != nil {
return fmt.Errorf("failed deleting acl policy: %v", err)
@ -243,11 +243,11 @@ func (s *Store) aclPolicyDeleteWithPolicy(tx *txn, policy *structs.ACLPolicy, id
return nil
}
func (s *Store) aclPolicyMaxIndex(tx *txn, _ *structs.ACLPolicy, _ *structs.EnterpriseMeta) uint64 {
func aclPolicyMaxIndex(tx *txn, _ *structs.ACLPolicy, _ *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "acl-policies")
}
func (s *Store) aclPolicyUpsertValidateEnterprise(*txn, *structs.ACLPolicy, *structs.ACLPolicy) error {
func aclPolicyUpsertValidateEnterprise(*txn, *structs.ACLPolicy, *structs.ACLPolicy) error {
return nil
}
@ -259,7 +259,7 @@ func (s *Store) ACLPolicyUpsertValidateEnterprise(*structs.ACLPolicy, *structs.A
///// ACL Token Functions /////
///////////////////////////////////////////////////////////////////////////////
func (s *Store) aclTokenInsert(tx *txn, token *structs.ACLToken) error {
func aclTokenInsert(tx *txn, token *structs.ACLToken) error {
// insert the token into memdb
if err := tx.Insert("acl-tokens", token); err != nil {
return fmt.Errorf("failed inserting acl token: %v", err)
@ -273,19 +273,19 @@ func (s *Store) aclTokenInsert(tx *txn, token *structs.ACLToken) error {
return nil
}
func (s *Store) aclTokenGetFromIndex(tx *txn, id string, index string, entMeta *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func aclTokenGetFromIndex(tx *txn, id string, index string, entMeta *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("acl-tokens", index, id)
}
func (s *Store) aclTokenListAll(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclTokenListAll(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-tokens", "id")
}
func (s *Store) aclTokenListLocal(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclTokenListLocal(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-tokens", "local", true)
}
func (s *Store) aclTokenListGlobal(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclTokenListGlobal(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-tokens", "local", false)
}
@ -297,11 +297,11 @@ func aclTokenListByRole(tx ReadTxn, role string, _ *structs.EnterpriseMeta) (mem
return tx.Get("acl-tokens", "roles", role)
}
func (s *Store) aclTokenListByAuthMethod(tx *txn, authMethod string, _, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclTokenListByAuthMethod(tx *txn, authMethod string, _, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-tokens", "authmethod", authMethod)
}
func (s *Store) aclTokenDeleteWithToken(tx *txn, token *structs.ACLToken, idx uint64) error {
func aclTokenDeleteWithToken(tx *txn, token *structs.ACLToken, idx uint64) error {
// remove the token
if err := tx.Delete("acl-tokens", token); err != nil {
return fmt.Errorf("failed deleting acl token: %v", err)
@ -314,11 +314,11 @@ func (s *Store) aclTokenDeleteWithToken(tx *txn, token *structs.ACLToken, idx ui
return nil
}
func (s *Store) aclTokenMaxIndex(tx *txn, _ *structs.ACLToken, entMeta *structs.EnterpriseMeta) uint64 {
func aclTokenMaxIndex(tx *txn, _ *structs.ACLToken, entMeta *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "acl-tokens")
}
func (s *Store) aclTokenUpsertValidateEnterprise(tx *txn, token *structs.ACLToken, existing *structs.ACLToken) error {
func aclTokenUpsertValidateEnterprise(tx *txn, token *structs.ACLToken, existing *structs.ACLToken) error {
return nil
}
@ -330,7 +330,7 @@ func (s *Store) ACLTokenUpsertValidateEnterprise(token *structs.ACLToken, existi
///// ACL Role Functions /////
///////////////////////////////////////////////////////////////////////////////
func (s *Store) aclRoleInsert(tx *txn, role *structs.ACLRole) error {
func aclRoleInsert(tx *txn, role *structs.ACLRole) error {
// insert the role into memdb
if err := tx.Insert("acl-roles", role); err != nil {
return fmt.Errorf("failed inserting acl role: %v", err)
@ -343,15 +343,15 @@ func (s *Store) aclRoleInsert(tx *txn, role *structs.ACLRole) error {
return nil
}
func (s *Store) aclRoleGetByID(tx *txn, id string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func aclRoleGetByID(tx *txn, id string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("acl-roles", "id", id)
}
func (s *Store) aclRoleGetByName(tx *txn, name string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func aclRoleGetByName(tx *txn, name string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("acl-roles", "name", name)
}
func (s *Store) aclRoleList(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclRoleList(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-roles", "id")
}
@ -359,7 +359,7 @@ func aclRoleListByPolicy(tx ReadTxn, policy string, _ *structs.EnterpriseMeta) (
return tx.Get("acl-roles", "policies", policy)
}
func (s *Store) aclRoleDeleteWithRole(tx *txn, role *structs.ACLRole, idx uint64) error {
func aclRoleDeleteWithRole(tx *txn, role *structs.ACLRole, idx uint64) error {
// remove the role
if err := tx.Delete("acl-roles", role); err != nil {
return fmt.Errorf("failed deleting acl role: %v", err)
@ -372,11 +372,11 @@ func (s *Store) aclRoleDeleteWithRole(tx *txn, role *structs.ACLRole, idx uint64
return nil
}
func (s *Store) aclRoleMaxIndex(tx *txn, _ *structs.ACLRole, _ *structs.EnterpriseMeta) uint64 {
func aclRoleMaxIndex(tx *txn, _ *structs.ACLRole, _ *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "acl-roles")
}
func (s *Store) aclRoleUpsertValidateEnterprise(tx *txn, role *structs.ACLRole, existing *structs.ACLRole) error {
func aclRoleUpsertValidateEnterprise(tx *txn, role *structs.ACLRole, existing *structs.ACLRole) error {
return nil
}
@ -388,7 +388,7 @@ func (s *Store) ACLRoleUpsertValidateEnterprise(role *structs.ACLRole, existing
///// ACL Binding Rule Functions /////
///////////////////////////////////////////////////////////////////////////////
func (s *Store) aclBindingRuleInsert(tx *txn, rule *structs.ACLBindingRule) error {
func aclBindingRuleInsert(tx *txn, rule *structs.ACLBindingRule) error {
// insert the role into memdb
if err := tx.Insert("acl-binding-rules", rule); err != nil {
return fmt.Errorf("failed inserting acl role: %v", err)
@ -402,19 +402,19 @@ func (s *Store) aclBindingRuleInsert(tx *txn, rule *structs.ACLBindingRule) erro
return nil
}
func (s *Store) aclBindingRuleGetByID(tx *txn, id string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func aclBindingRuleGetByID(tx *txn, id string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("acl-binding-rules", "id", id)
}
func (s *Store) aclBindingRuleList(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclBindingRuleList(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-binding-rules", "id")
}
func (s *Store) aclBindingRuleListByAuthMethod(tx *txn, method string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclBindingRuleListByAuthMethod(tx *txn, method string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-binding-rules", "authmethod", method)
}
func (s *Store) aclBindingRuleDeleteWithRule(tx *txn, rule *structs.ACLBindingRule, idx uint64) error {
func aclBindingRuleDeleteWithRule(tx *txn, rule *structs.ACLBindingRule, idx uint64) error {
// remove the rule
if err := tx.Delete("acl-binding-rules", rule); err != nil {
return fmt.Errorf("failed deleting acl binding rule: %v", err)
@ -427,11 +427,11 @@ func (s *Store) aclBindingRuleDeleteWithRule(tx *txn, rule *structs.ACLBindingRu
return nil
}
func (s *Store) aclBindingRuleMaxIndex(tx *txn, _ *structs.ACLBindingRule, entMeta *structs.EnterpriseMeta) uint64 {
func aclBindingRuleMaxIndex(tx *txn, _ *structs.ACLBindingRule, entMeta *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "acl-binding-rules")
}
func (s *Store) aclBindingRuleUpsertValidateEnterprise(tx *txn, rule *structs.ACLBindingRule, existing *structs.ACLBindingRule) error {
func aclBindingRuleUpsertValidateEnterprise(tx *txn, rule *structs.ACLBindingRule, existing *structs.ACLBindingRule) error {
return nil
}
@ -443,7 +443,7 @@ func (s *Store) ACLBindingRuleUpsertValidateEnterprise(rule *structs.ACLBindingR
///// ACL Auth Method Functions /////
///////////////////////////////////////////////////////////////////////////////
func (s *Store) aclAuthMethodInsert(tx *txn, method *structs.ACLAuthMethod) error {
func aclAuthMethodInsert(tx *txn, method *structs.ACLAuthMethod) error {
// insert the role into memdb
if err := tx.Insert("acl-auth-methods", method); err != nil {
return fmt.Errorf("failed inserting acl role: %v", err)
@ -457,15 +457,15 @@ func (s *Store) aclAuthMethodInsert(tx *txn, method *structs.ACLAuthMethod) erro
return nil
}
func (s *Store) aclAuthMethodGetByName(tx *txn, method string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func aclAuthMethodGetByName(tx *txn, method string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("acl-auth-methods", "id", method)
}
func (s *Store) aclAuthMethodList(tx *txn, entMeta *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func aclAuthMethodList(tx *txn, entMeta *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("acl-auth-methods", "id")
}
func (s *Store) aclAuthMethodDeleteWithMethod(tx *txn, method *structs.ACLAuthMethod, idx uint64) error {
func aclAuthMethodDeleteWithMethod(tx *txn, method *structs.ACLAuthMethod, idx uint64) error {
// remove the method
if err := tx.Delete("acl-auth-methods", method); err != nil {
return fmt.Errorf("failed deleting acl auth method: %v", err)
@ -478,11 +478,11 @@ func (s *Store) aclAuthMethodDeleteWithMethod(tx *txn, method *structs.ACLAuthMe
return nil
}
func (s *Store) aclAuthMethodMaxIndex(tx *txn, _ *structs.ACLAuthMethod, entMeta *structs.EnterpriseMeta) uint64 {
func aclAuthMethodMaxIndex(tx *txn, _ *structs.ACLAuthMethod, entMeta *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "acl-auth-methods")
}
func (s *Store) aclAuthMethodUpsertValidateEnterprise(tx *txn, method *structs.ACLAuthMethod, existing *structs.ACLAuthMethod) error {
func aclAuthMethodUpsertValidateEnterprise(tx *txn, method *structs.ACLAuthMethod, existing *structs.ACLAuthMethod) error {
return nil
}

14
agent/consul/state/acl_test.go
View File

@ -4105,7 +4105,7 @@ func TestStateStore_resolveACLLinks(t *testing.T) {
},
}
_, err := s.resolveACLLinks(tx, links, func(*txn, string) (string, error) {
_, err := resolveACLLinks(tx, links, func(*txn, string) (string, error) {
err := fmt.Errorf("Should not be attempting to resolve an empty id")
require.Fail(t, err.Error())
return "", err
@ -4131,7 +4131,7 @@ func TestStateStore_resolveACLLinks(t *testing.T) {
},
}
numValid, err := s.resolveACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
numValid, err := resolveACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
switch linkID {
case "e81887b4-836b-4053-a1fa-7e8305902be9":
return "foo", nil
@ -4161,7 +4161,7 @@ func TestStateStore_resolveACLLinks(t *testing.T) {
},
}
numValid, err := s.resolveACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
numValid, err := resolveACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
require.Equal(t, "b985e082-25d3-45a9-9dd8-fd1a41b83b0d", linkID)
return "", nil
})
@ -4201,7 +4201,7 @@ func TestStateStore_fixupACLLinks(t *testing.T) {
tx := s.db.Txn(false)
defer tx.Abort()
newLinks, cloned, err := s.fixupACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
newLinks, cloned, err := fixupACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
switch linkID {
case "40b57f86-97ea-40e4-a99a-c399cc81f4dd":
return "foo", nil
@ -4228,7 +4228,7 @@ func TestStateStore_fixupACLLinks(t *testing.T) {
tx := s.db.Txn(false)
defer tx.Abort()
newLinks, cloned, err := s.fixupACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
newLinks, cloned, err := fixupACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
switch linkID {
case "40b57f86-97ea-40e4-a99a-c399cc81f4dd":
return "foo", nil
@ -4260,7 +4260,7 @@ func TestStateStore_fixupACLLinks(t *testing.T) {
tx := s.db.Txn(false)
defer tx.Abort()
newLinks, cloned, err := s.fixupACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
newLinks, cloned, err := fixupACLLinks(tx, links, func(_ *txn, linkID string) (string, error) {
switch linkID {
case "40b57f86-97ea-40e4-a99a-c399cc81f4dd":
return "foo", nil
@ -4287,7 +4287,7 @@ func TestStateStore_fixupACLLinks(t *testing.T) {
tx := s.db.Txn(false)
defer tx.Abort()
_, _, err := s.fixupACLLinks(tx, links, func(*txn, string) (string, error) {
_, _, err := fixupACLLinks(tx, links, func(*txn, string) (string, error) {
return "", fmt.Errorf("Resolver Error")
})

298
agent/consul/state/catalog.go
View File

File diff suppressed because it is too large Load Diff

60
agent/consul/state/catalog_oss.go
View File

@ -168,7 +168,7 @@ func serviceKindIndexName(kind structs.ServiceKind, _ *structs.EnterpriseMeta) s
}
}
func (s *Store) catalogUpdateServicesIndexes(tx *txn, idx uint64, _ *structs.EnterpriseMeta) error {
func catalogUpdateServicesIndexes(tx *txn, idx uint64, _ *structs.EnterpriseMeta) error {
// overall services index
if err := indexUpdateMaxTxn(tx, idx, "services"); err != nil {
return fmt.Errorf("failed updating index: %s", err)
@ -177,7 +177,7 @@ func (s *Store) catalogUpdateServicesIndexes(tx *txn, idx uint64, _ *structs.Ent
return nil
}
func (s *Store) catalogUpdateServiceKindIndexes(tx *txn, kind structs.ServiceKind, idx uint64, _ *structs.EnterpriseMeta) error {
func catalogUpdateServiceKindIndexes(tx *txn, kind structs.ServiceKind, idx uint64, _ *structs.EnterpriseMeta) error {
// service-kind index
if err := indexUpdateMaxTxn(tx, idx, serviceKindIndexName(kind, nil)); err != nil {
return fmt.Errorf("failed updating index: %s", err)
@ -186,7 +186,7 @@ func (s *Store) catalogUpdateServiceKindIndexes(tx *txn, kind structs.ServiceKin
return nil
}
func (s *Store) catalogUpdateServiceIndexes(tx *txn, serviceName string, idx uint64, _ *structs.EnterpriseMeta) error {
func catalogUpdateServiceIndexes(tx *txn, serviceName string, idx uint64, _ *structs.EnterpriseMeta) error {
// per-service index
if err := indexUpdateMaxTxn(tx, idx, serviceIndexName(serviceName, nil)); err != nil {
return fmt.Errorf("failed updating index: %s", err)
@ -195,81 +195,81 @@ func (s *Store) catalogUpdateServiceIndexes(tx *txn, serviceName string, idx uin
return nil
}
func (s *Store) catalogUpdateServiceExtinctionIndex(tx *txn, idx uint64, _ *structs.EnterpriseMeta) error {
func catalogUpdateServiceExtinctionIndex(tx *txn, idx uint64, _ *structs.EnterpriseMeta) error {
if err := tx.Insert("index", &IndexEntry{serviceLastExtinctionIndexName, idx}); err != nil {
return fmt.Errorf("failed updating missing service extinction index: %s", err)
}
return nil
}
func (s *Store) catalogInsertService(tx *txn, svc *structs.ServiceNode) error {
func catalogInsertService(tx *txn, svc *structs.ServiceNode) error {
// Insert the service and update the index
if err := tx.Insert("services", svc); err != nil {
return fmt.Errorf("failed inserting service: %s", err)
}
if err := s.catalogUpdateServicesIndexes(tx, svc.ModifyIndex, &svc.EnterpriseMeta); err != nil {
if err := catalogUpdateServicesIndexes(tx, svc.ModifyIndex, &svc.EnterpriseMeta); err != nil {
return err
}
if err := s.catalogUpdateServiceIndexes(tx, svc.ServiceName, svc.ModifyIndex, &svc.EnterpriseMeta); err != nil {
if err := catalogUpdateServiceIndexes(tx, svc.ServiceName, svc.ModifyIndex, &svc.EnterpriseMeta); err != nil {
return err
}
if err := s.catalogUpdateServiceKindIndexes(tx, svc.ServiceKind, svc.ModifyIndex, &svc.EnterpriseMeta); err != nil {
if err := catalogUpdateServiceKindIndexes(tx, svc.ServiceKind, svc.ModifyIndex, &svc.EnterpriseMeta); err != nil {
return err
}
return nil
}
func (s *Store) catalogServicesMaxIndex(tx *txn, _ *structs.EnterpriseMeta) uint64 {
func catalogServicesMaxIndex(tx *txn, _ *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "services")
}
func (s *Store) catalogServiceMaxIndex(tx *txn, serviceName string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
func catalogServiceMaxIndex(tx *txn, serviceName string, _ *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch("index", "id", serviceIndexName(serviceName, nil))
}
func (s *Store) catalogServiceKindMaxIndex(tx *txn, ws memdb.WatchSet, kind structs.ServiceKind, entMeta *structs.EnterpriseMeta) uint64 {
func catalogServiceKindMaxIndex(tx *txn, ws memdb.WatchSet, kind structs.ServiceKind, entMeta *structs.EnterpriseMeta) uint64 {
return maxIndexWatchTxn(tx, ws, serviceKindIndexName(kind, nil))
}
func (s *Store) catalogServiceList(tx *txn, _ *structs.EnterpriseMeta, _ bool) (memdb.ResultIterator, error) {
func catalogServiceList(tx *txn, _ *structs.EnterpriseMeta, _ bool) (memdb.ResultIterator, error) {
return tx.Get("services", "id")
}
func (s *Store) catalogServiceListByKind(tx *txn, kind structs.ServiceKind, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogServiceListByKind(tx *txn, kind structs.ServiceKind, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("services", "kind", string(kind))
}
func (s *Store) catalogServiceListByNode(tx *txn, node string, _ *structs.EnterpriseMeta, _ bool) (memdb.ResultIterator, error) {
func catalogServiceListByNode(tx *txn, node string, _ *structs.EnterpriseMeta, _ bool) (memdb.ResultIterator, error) {
return tx.Get("services", "node", node)
}
func (s *Store) catalogServiceNodeList(tx *txn, name string, index string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogServiceNodeList(tx *txn, name string, index string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("services", index, name)
}
func (s *Store) catalogServiceLastExtinctionIndex(tx *txn, _ *structs.EnterpriseMeta) (interface{}, error) {
func catalogServiceLastExtinctionIndex(tx *txn, _ *structs.EnterpriseMeta) (interface{}, error) {
return tx.First("index", "id", serviceLastExtinctionIndexName)
}
func (s *Store) catalogMaxIndex(tx *txn, _ *structs.EnterpriseMeta, checks bool) uint64 {
func catalogMaxIndex(tx *txn, _ *structs.EnterpriseMeta, checks bool) uint64 {
if checks {
return maxIndexTxn(tx, "nodes", "services", "checks")
}
return maxIndexTxn(tx, "nodes", "services")
}
func (s *Store) catalogMaxIndexWatch(tx *txn, ws memdb.WatchSet, _ *structs.EnterpriseMeta, checks bool) uint64 {
func catalogMaxIndexWatch(tx *txn, ws memdb.WatchSet, _ *structs.EnterpriseMeta, checks bool) uint64 {
if checks {
return maxIndexWatchTxn(tx, ws, "nodes", "services", "checks")
}
return maxIndexWatchTxn(tx, ws, "nodes", "services")
}
func (s *Store) catalogUpdateCheckIndexes(tx *txn, idx uint64, _ *structs.EnterpriseMeta) error {
func catalogUpdateCheckIndexes(tx *txn, idx uint64, _ *structs.EnterpriseMeta) error {
// update the universal index entry
if err := tx.Insert("index", &IndexEntry{"checks", idx}); err != nil {
return fmt.Errorf("failed updating index: %s", err)
@ -277,53 +277,53 @@ func (s *Store) catalogUpdateCheckIndexes(tx *txn, idx uint64, _ *structs.Enterp
return nil
}
func (s *Store) catalogChecksMaxIndex(tx *txn, _ *structs.EnterpriseMeta) uint64 {
func catalogChecksMaxIndex(tx *txn, _ *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "checks")
}
func (s *Store) catalogListChecksByNode(tx *txn, node string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogListChecksByNode(tx *txn, node string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("checks", "node", node)
}
func (s *Store) catalogListChecksByService(tx *txn, service string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogListChecksByService(tx *txn, service string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("checks", "service", service)
}
func (s *Store) catalogListChecksInState(tx *txn, state string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogListChecksInState(tx *txn, state string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
// simpler than normal due to the use of the CompoundMultiIndex
return tx.Get("checks", "status", state)
}
func (s *Store) catalogListChecks(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogListChecks(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("checks", "id")
}
func (s *Store) catalogListNodeChecks(tx *txn, node string) (memdb.ResultIterator, error) {
func catalogListNodeChecks(tx *txn, node string) (memdb.ResultIterator, error) {
return tx.Get("checks", "node_service_check", node, false)
}
func (s *Store) catalogListServiceChecks(tx *txn, node string, service string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogListServiceChecks(tx *txn, node string, service string, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("checks", "node_service", node, service)
}
func (s *Store) catalogInsertCheck(tx *txn, chk *structs.HealthCheck, idx uint64) error {
func catalogInsertCheck(tx *txn, chk *structs.HealthCheck, idx uint64) error {
// Insert the check
if err := tx.Insert("checks", chk); err != nil {
return fmt.Errorf("failed inserting check: %s", err)
}
if err := s.catalogUpdateCheckIndexes(tx, idx, &chk.EnterpriseMeta); err != nil {
if err := catalogUpdateCheckIndexes(tx, idx, &chk.EnterpriseMeta); err != nil {
return err
}
return nil
}
func (s *Store) catalogChecksForNodeService(tx *txn, node string, service string, entMeta *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func catalogChecksForNodeService(tx *txn, node string, service string, entMeta *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
return tx.Get("checks", "node_service", node, service)
}
func (s *Store) validateRegisterRequestTxn(tx *txn, args *structs.RegisterRequest) (*structs.EnterpriseMeta, error) {
func validateRegisterRequestTxn(tx *txn, args *structs.RegisterRequest) (*structs.EnterpriseMeta, error) {
return nil, nil
}

18
agent/consul/state/catalog_test.go
View File

@ -113,18 +113,18 @@ func TestStateStore_ensureNoNodeWithSimilarNameTxn(t *testing.T) {
Address: "2.3.4.5",
}
// Lets conflict with node1 (has an ID)
if err := s.ensureNoNodeWithSimilarNameTxn(tx, node, false); err == nil {
if err := ensureNoNodeWithSimilarNameTxn(tx, node, false); err == nil {
t.Fatalf("Should return an error since another name with similar name exists")
}
if err := s.ensureNoNodeWithSimilarNameTxn(tx, node, true); err == nil {
if err := ensureNoNodeWithSimilarNameTxn(tx, node, true); err == nil {
t.Fatalf("Should return an error since another name with similar name exists")
}
// Lets conflict with node without ID
node.Node = "NoDe2"
if err := s.ensureNoNodeWithSimilarNameTxn(tx, node, false); err == nil {
if err := ensureNoNodeWithSimilarNameTxn(tx, node, false); err == nil {
t.Fatalf("Should return an error since another name with similar name exists")
}
if err := s.ensureNoNodeWithSimilarNameTxn(tx, node, true); err != nil {
if err := ensureNoNodeWithSimilarNameTxn(tx, node, true); err != nil {
t.Fatalf("Should not clash with another similar node name without ID, err:=%q", err)
}
@ -134,7 +134,7 @@ func TestStateStore_ensureNoNodeWithSimilarNameTxn(t *testing.T) {
Node: "node1",
Address: "2.3.4.5",
}
if err := s.ensureNoNodeWithSimilarNameTxn(tx, newNode, false); err == nil {
if err := ensureNoNodeWithSimilarNameTxn(tx, newNode, false); err == nil {
t.Fatalf("Should return an error since the previous node is still healthy")
}
s.ensureCheckTxn(tx, 5, &structs.HealthCheck{
@ -142,7 +142,7 @@ func TestStateStore_ensureNoNodeWithSimilarNameTxn(t *testing.T) {
CheckID: structs.SerfCheckID,
Status: api.HealthCritical,
})
if err := s.ensureNoNodeWithSimilarNameTxn(tx, newNode, false); err != nil {
if err := ensureNoNodeWithSimilarNameTxn(tx, newNode, false); err != nil {
t.Fatal(err)
}
}
@ -4386,7 +4386,7 @@ func TestStateStore_ensureServiceCASTxn(t *testing.T) {
// attempt to update with a 0 index
tx := s.db.WriteTxnRestore()
err := s.ensureServiceCASTxn(tx, 3, "node1", &ns)
err := ensureServiceCASTxn(tx, 3, "node1", &ns)
require.Equal(t, err, errCASCompareFailed)
require.NoError(t, tx.Commit())
@ -4401,7 +4401,7 @@ func TestStateStore_ensureServiceCASTxn(t *testing.T) {
ns.ModifyIndex = 99
// attempt to update with a non-matching index
tx = s.db.WriteTxnRestore()
err = s.ensureServiceCASTxn(tx, 4, "node1", &ns)
err = ensureServiceCASTxn(tx, 4, "node1", &ns)
require.Equal(t, err, errCASCompareFailed)
require.NoError(t, tx.Commit())
@ -4416,7 +4416,7 @@ func TestStateStore_ensureServiceCASTxn(t *testing.T) {
ns.ModifyIndex = 2
// update with the matching modify index
tx = s.db.WriteTxnRestore()
err = s.ensureServiceCASTxn(tx, 7, "node1", &ns)
err = ensureServiceCASTxn(tx, 7, "node1", &ns)
require.NoError(t, err)
require.NoError(t, tx.Commit())

62
agent/consul/state/config_entry.go
View File

@ -96,22 +96,22 @@ func (s *Snapshot) ConfigEntries() ([]structs.ConfigEntry, error) {
// ConfigEntry is used when restoring from a snapshot.
func (s *Restore) ConfigEntry(c structs.ConfigEntry) error {
return s.store.insertConfigEntryWithTxn(s.tx, c.GetRaftIndex().ModifyIndex, c)
return insertConfigEntryWithTxn(s.tx, c.GetRaftIndex().ModifyIndex, c)
}
// ConfigEntry is called to get a given config entry.
func (s *Store) ConfigEntry(ws memdb.WatchSet, kind, name string, entMeta *structs.EnterpriseMeta) (uint64, structs.ConfigEntry, error) {
tx := s.db.Txn(false)
defer tx.Abort()
return s.configEntryTxn(tx, ws, kind, name, entMeta)
return configEntryTxn(tx, ws, kind, name, entMeta)
}
func (s *Store) configEntryTxn(tx *txn, ws memdb.WatchSet, kind, name string, entMeta *structs.EnterpriseMeta) (uint64, structs.ConfigEntry, error) {
func configEntryTxn(tx *txn, ws memdb.WatchSet, kind, name string, entMeta *structs.EnterpriseMeta) (uint64, structs.ConfigEntry, error) {
// Get the index
idx := maxIndexTxn(tx, configTableName)
// Get the existing config entry.
watchCh, existing, err := s.firstWatchConfigEntryWithTxn(tx, kind, name, entMeta)
watchCh, existing, err := firstWatchConfigEntryWithTxn(tx, kind, name, entMeta)
if err != nil {
return 0, nil, fmt.Errorf("failed config entry lookup: %s", err)
}
@ -138,10 +138,10 @@ func (s *Store) ConfigEntries(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta
func (s *Store) ConfigEntriesByKind(ws memdb.WatchSet, kind string, entMeta *structs.EnterpriseMeta) (uint64, []structs.ConfigEntry, error) {
tx := s.db.Txn(false)
defer tx.Abort()
return s.configEntriesByKindTxn(tx, ws, kind, entMeta)
return configEntriesByKindTxn(tx, ws, kind, entMeta)
}
func (s *Store) configEntriesByKindTxn(tx *txn, ws memdb.WatchSet, kind string, entMeta *structs.EnterpriseMeta) (uint64, []structs.ConfigEntry, error) {
func configEntriesByKindTxn(tx *txn, ws memdb.WatchSet, kind string, entMeta *structs.EnterpriseMeta) (uint64, []structs.ConfigEntry, error) {
// Get the index
idx := maxIndexTxn(tx, configTableName)
@ -180,7 +180,7 @@ func (s *Store) EnsureConfigEntry(idx uint64, conf structs.ConfigEntry, entMeta
// ensureConfigEntryTxn upserts a config entry inside of a transaction.
func (s *Store) ensureConfigEntryTxn(tx *txn, idx uint64, conf structs.ConfigEntry, entMeta *structs.EnterpriseMeta) error {
// Check for existing configuration.
existing, err := s.firstConfigEntryWithTxn(tx, conf.GetKind(), conf.GetName(), entMeta)
existing, err := firstConfigEntryWithTxn(tx, conf.GetKind(), conf.GetName(), entMeta)
if err != nil {
return fmt.Errorf("failed configuration lookup: %s", err)
}
@ -200,11 +200,11 @@ func (s *Store) ensureConfigEntryTxn(tx *txn, idx uint64, conf structs.ConfigEnt
return err // Err is already sufficiently decorated.
}
if err := s.validateConfigEntryEnterprise(tx, conf); err != nil {
if err := validateConfigEntryEnterprise(tx, conf); err != nil {
return err
}
return s.insertConfigEntryWithTxn(tx, idx, conf)
return insertConfigEntryWithTxn(tx, idx, conf)
}
// EnsureConfigEntryCAS is called to do a check-and-set upsert of a given config entry.
@ -213,7 +213,7 @@ func (s *Store) EnsureConfigEntryCAS(idx, cidx uint64, conf structs.ConfigEntry,
defer tx.Abort()
// Check for existing configuration.
existing, err := s.firstConfigEntryWithTxn(tx, conf.GetKind(), conf.GetName(), entMeta)
existing, err := firstConfigEntryWithTxn(tx, conf.GetKind(), conf.GetName(), entMeta)
if err != nil {
return false, fmt.Errorf("failed configuration lookup: %s", err)
}
@ -247,7 +247,7 @@ func (s *Store) DeleteConfigEntry(idx uint64, kind, name string, entMeta *struct
defer tx.Abort()
// Try to retrieve the existing config entry.
existing, err := s.firstConfigEntryWithTxn(tx, kind, name, entMeta)
existing, err := firstConfigEntryWithTxn(tx, kind, name, entMeta)
if err != nil {
return fmt.Errorf("failed config entry lookup: %s", err)
}
@ -282,14 +282,14 @@ func (s *Store) DeleteConfigEntry(idx uint64, kind, name string, entMeta *struct
return tx.Commit()
}
func (s *Store) insertConfigEntryWithTxn(tx *txn, idx uint64, conf structs.ConfigEntry) error {
func insertConfigEntryWithTxn(tx *txn, idx uint64, conf structs.ConfigEntry) error {
if conf == nil {
return fmt.Errorf("cannot insert nil config entry")
}
// If the config entry is for a terminating or ingress gateway we update the memdb table
// that associates gateways <-> services.
if conf.GetKind() == structs.TerminatingGateway || conf.GetKind() == structs.IngressGateway {
err := s.updateGatewayServices(tx, idx, conf, conf.GetEnterpriseMeta())
err := updateGatewayServices(tx, idx, conf, conf.GetEnterpriseMeta())
if err != nil {
return fmt.Errorf("failed to associate services to gateway: %v", err)
}
@ -333,16 +333,16 @@ func (s *Store) validateProposedConfigEntryInGraph(
case structs.ServiceSplitter:
case structs.ServiceResolver:
case structs.IngressGateway:
err := s.checkGatewayClash(tx, name, structs.IngressGateway, structs.TerminatingGateway, entMeta)
err := checkGatewayClash(tx, name, structs.IngressGateway, structs.TerminatingGateway, entMeta)
if err != nil {
return err
}
err = s.validateProposedIngressProtocolsInServiceGraph(tx, next, entMeta)
err = validateProposedIngressProtocolsInServiceGraph(tx, next, entMeta)
if err != nil {
return err
}
case structs.TerminatingGateway:
err := s.checkGatewayClash(tx, name, structs.TerminatingGateway, structs.IngressGateway, entMeta)
err := checkGatewayClash(tx, name, structs.TerminatingGateway, structs.IngressGateway, entMeta)
if err != nil {
return err
}
@ -353,12 +353,12 @@ func (s *Store) validateProposedConfigEntryInGraph(
return s.validateProposedConfigEntryInServiceGraph(tx, kind, name, next, validateAllChains, entMeta)
}
func (s *Store) checkGatewayClash(
func checkGatewayClash(
tx *txn,
name, selfKind, otherKind string,
entMeta *structs.EnterpriseMeta,
) error {
_, entry, err := s.configEntryTxn(tx, nil, otherKind, name, entMeta)
_, entry, err := configEntryTxn(tx, nil, otherKind, name, entMeta)
if err != nil {
return err
}
@ -393,7 +393,7 @@ func (s *Store) validateProposedConfigEntryInServiceGraph(
// somehow omit the ones that have a default protocol configured.
for _, kind := range serviceGraphKinds {
_, entries, err := s.configEntriesByKindTxn(tx, nil, kind, structs.WildcardEnterpriseMeta())
_, entries, err := configEntriesByKindTxn(tx, nil, kind, structs.WildcardEnterpriseMeta())
if err != nil {
return err
}
@ -688,7 +688,7 @@ func (s *Store) getProxyConfigEntryTxn(
overrides map[structs.ConfigEntryKindName]structs.ConfigEntry,
entMeta *structs.EnterpriseMeta,
) (uint64, *structs.ProxyConfigEntry, error) {
idx, entry, err := s.configEntryWithOverridesTxn(tx, ws, structs.ProxyDefaults, name, overrides, entMeta)
idx, entry, err := configEntryWithOverridesTxn(tx, ws, structs.ProxyDefaults, name, overrides, entMeta)
if err != nil {
return 0, nil, err
} else if entry == nil {
@ -713,7 +713,7 @@ func (s *Store) getServiceConfigEntryTxn(
overrides map[structs.ConfigEntryKindName]structs.ConfigEntry,
entMeta *structs.EnterpriseMeta,
) (uint64, *structs.ServiceConfigEntry, error) {
idx, entry, err := s.configEntryWithOverridesTxn(tx, ws, structs.ServiceDefaults, serviceName, overrides, entMeta)
idx, entry, err := configEntryWithOverridesTxn(tx, ws, structs.ServiceDefaults, serviceName, overrides, entMeta)
if err != nil {
return 0, nil, err
} else if entry == nil {
@ -738,7 +738,7 @@ func (s *Store) getRouterConfigEntryTxn(
overrides map[structs.ConfigEntryKindName]structs.ConfigEntry,
entMeta *structs.EnterpriseMeta,
) (uint64, *structs.ServiceRouterConfigEntry, error) {
idx, entry, err := s.configEntryWithOverridesTxn(tx, ws, structs.ServiceRouter, serviceName, overrides, entMeta)
idx, entry, err := configEntryWithOverridesTxn(tx, ws, structs.ServiceRouter, serviceName, overrides, entMeta)
if err != nil {
return 0, nil, err
} else if entry == nil {
@ -763,7 +763,7 @@ func (s *Store) getSplitterConfigEntryTxn(
overrides map[structs.ConfigEntryKindName]structs.ConfigEntry,
entMeta *structs.EnterpriseMeta,
) (uint64, *structs.ServiceSplitterConfigEntry, error) {
idx, entry, err := s.configEntryWithOverridesTxn(tx, ws, structs.ServiceSplitter, serviceName, overrides, entMeta)
idx, entry, err := configEntryWithOverridesTxn(tx, ws, structs.ServiceSplitter, serviceName, overrides, entMeta)
if err != nil {
return 0, nil, err
} else if entry == nil {
@ -788,7 +788,7 @@ func (s *Store) getResolverConfigEntryTxn(
overrides map[structs.ConfigEntryKindName]structs.ConfigEntry,
entMeta *structs.EnterpriseMeta,
) (uint64, *structs.ServiceResolverConfigEntry, error) {
idx, entry, err := s.configEntryWithOverridesTxn(tx, ws, structs.ServiceResolver, serviceName, overrides, entMeta)
idx, entry, err := configEntryWithOverridesTxn(tx, ws, structs.ServiceResolver, serviceName, overrides, entMeta)
if err != nil {
return 0, nil, err
} else if entry == nil {
@ -802,7 +802,7 @@ func (s *Store) getResolverConfigEntryTxn(
return idx, resolver, nil
}
func (s *Store) configEntryWithOverridesTxn(
func configEntryWithOverridesTxn(
tx *txn,
ws memdb.WatchSet,
kind string,
@ -819,10 +819,10 @@ func (s *Store) configEntryWithOverridesTxn(
}
}
return s.configEntryTxn(tx, ws, kind, name, entMeta)
return configEntryTxn(tx, ws, kind, name, entMeta)
}
func (s *Store) validateProposedIngressProtocolsInServiceGraph(
func validateProposedIngressProtocolsInServiceGraph(
tx *txn,
next structs.ConfigEntry,
entMeta *structs.EnterpriseMeta,
@ -837,7 +837,7 @@ func (s *Store) validateProposedIngressProtocolsInServiceGraph(
}
validationFn := func(svc structs.ServiceName, expectedProto string) error {
_, svcProto, err := s.protocolForService(tx, nil, svc)
_, svcProto, err := protocolForService(tx, nil, svc)
if err != nil {
return err
}
@ -866,18 +866,18 @@ func (s *Store) validateProposedIngressProtocolsInServiceGraph(
// protocolForService returns the service graph protocol associated to the
// provided service, checking all relevant config entries.
func (s *Store) protocolForService(
func protocolForService(
tx *txn,
ws memdb.WatchSet,
svc structs.ServiceName,
) (uint64, string, error) {
// Get the global proxy defaults (for default protocol)
maxIdx, proxyConfig, err := s.configEntryTxn(tx, ws, structs.ProxyDefaults, structs.ProxyConfigGlobal, structs.DefaultEnterpriseMeta())
maxIdx, proxyConfig, err := configEntryTxn(tx, ws, structs.ProxyDefaults, structs.ProxyConfigGlobal, structs.DefaultEnterpriseMeta())
if err != nil {
return 0, "", err
}
idx, serviceDefaults, err := s.configEntryTxn(tx, ws, structs.ServiceDefaults, svc.Name, &svc.EnterpriseMeta)
idx, serviceDefaults, err := configEntryTxn(tx, ws, structs.ServiceDefaults, svc.Name, &svc.EnterpriseMeta)
if err != nil {
return 0, "", err
}

6
agent/consul/state/config_entry_oss.go
View File

@ -49,17 +49,17 @@ func configTableSchema() *memdb.TableSchema {
}
}
func (s *Store) firstConfigEntryWithTxn(tx *txn,
func firstConfigEntryWithTxn(tx *txn,
kind, name string, entMeta *structs.EnterpriseMeta) (interface{}, error) {
return tx.First(configTableName, "id", kind, name)
}
func (s *Store) firstWatchConfigEntryWithTxn(tx *txn,
func firstWatchConfigEntryWithTxn(tx *txn,
kind, name string, entMeta *structs.EnterpriseMeta) (<-chan struct{}, interface{}, error) {
return tx.FirstWatch(configTableName, "id", kind, name)
}
func (s *Store) validateConfigEntryEnterprise(tx *txn, conf structs.ConfigEntry) error {
func validateConfigEntryEnterprise(tx *txn, conf structs.ConfigEntry) error {
return nil
}

12
agent/consul/state/connect_ca.go
View File

@ -113,10 +113,10 @@ func (s *Store) CAConfig(ws memdb.WatchSet) (uint64, *structs.CAConfiguration, e
tx := s.db.Txn(false)
defer tx.Abort()
return s.caConfigTxn(tx, ws)
return caConfigTxn(tx, ws)
}
func (s *Store) caConfigTxn(tx *txn, ws memdb.WatchSet) (uint64, *structs.CAConfiguration, error) {
func caConfigTxn(tx *txn, ws memdb.WatchSet) (uint64, *structs.CAConfiguration, error) {
// Get the CA config
ch, c, err := tx.FirstWatch(caConfigTableName, "id")
if err != nil {
@ -233,10 +233,10 @@ func (s *Store) CARoots(ws memdb.WatchSet) (uint64, structs.CARoots, error) {
tx := s.db.Txn(false)
defer tx.Abort()
return s.caRootsTxn(tx, ws)
return caRootsTxn(tx, ws)
}
func (s *Store) caRootsTxn(tx *txn, ws memdb.WatchSet) (uint64, structs.CARoots, error) {
func caRootsTxn(tx *txn, ws memdb.WatchSet) (uint64, structs.CARoots, error) {
// Get the index
idx := maxIndexTxn(tx, caRootTableName)
@ -459,12 +459,12 @@ func (s *Store) CARootsAndConfig(ws memdb.WatchSet) (uint64, structs.CARoots, *s
tx := s.db.Txn(false)
defer tx.Abort()
confIdx, config, err := s.caConfigTxn(tx, ws)
confIdx, config, err := caConfigTxn(tx, ws)
if err != nil {
return 0, nil, nil, fmt.Errorf("failed CA config lookup: %v", err)
}
rootsIdx, roots, err := s.caRootsTxn(tx, ws)
rootsIdx, roots, err := caRootsTxn(tx, ws)
if err != nil {
return 0, nil, nil, fmt.Errorf("failed CA roots lookup: %v", err)
}

20
agent/consul/state/federation_state.go
View File

@ -63,7 +63,7 @@ func (s *Store) FederationStateBatchSet(idx uint64, configs structs.FederationSt
defer tx.Abort()
for _, config := range configs {
if err := s.federationStateSetTxn(tx, idx, config); err != nil {
if err := federationStateSetTxn(tx, idx, config); err != nil {
return err
}
}
@ -76,7 +76,7 @@ func (s *Store) FederationStateSet(idx uint64, config *structs.FederationState)
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.federationStateSetTxn(tx, idx, config); err != nil {
if err := federationStateSetTxn(tx, idx, config); err != nil {
return err
}
@ -84,7 +84,7 @@ func (s *Store) FederationStateSet(idx uint64, config *structs.FederationState)
}
// federationStateSetTxn upserts a federation state inside of a transaction.
func (s *Store) federationStateSetTxn(tx *txn, idx uint64, config *structs.FederationState) error {
func federationStateSetTxn(tx *txn, idx uint64, config *structs.FederationState) error {
if config.Datacenter == "" {
return fmt.Errorf("missing datacenter on federation state")
}
@ -131,10 +131,10 @@ func (s *Store) federationStateSetTxn(tx *txn, idx uint64, config *structs.Feder
func (s *Store) FederationStateGet(ws memdb.WatchSet, datacenter string) (uint64, *structs.FederationState, error) {
tx := s.db.Txn(false)
defer tx.Abort()
return s.federationStateGetTxn(tx, ws, datacenter)
return federationStateGetTxn(tx, ws, datacenter)
}
func (s *Store) federationStateGetTxn(tx *txn, ws memdb.WatchSet, datacenter string) (uint64, *structs.FederationState, error) {
func federationStateGetTxn(tx *txn, ws memdb.WatchSet, datacenter string) (uint64, *structs.FederationState, error) {
// Get the index
idx := maxIndexTxn(tx, federationStateTableName)
@ -161,10 +161,10 @@ func (s *Store) federationStateGetTxn(tx *txn, ws memdb.WatchSet, datacenter str
func (s *Store) FederationStateList(ws memdb.WatchSet) (uint64, []*structs.FederationState, error) {
tx := s.db.Txn(false)
defer tx.Abort()
return s.federationStateListTxn(tx, ws)
return federationStateListTxn(tx, ws)
}
func (s *Store) federationStateListTxn(tx *txn, ws memdb.WatchSet) (uint64, []*structs.FederationState, error) {
func federationStateListTxn(tx *txn, ws memdb.WatchSet) (uint64, []*structs.FederationState, error) {
// Get the index
idx := maxIndexTxn(tx, federationStateTableName)
@ -185,7 +185,7 @@ func (s *Store) FederationStateDelete(idx uint64, datacenter string) error {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.federationStateDeleteTxn(tx, idx, datacenter); err != nil {
if err := federationStateDeleteTxn(tx, idx, datacenter); err != nil {
return err
}
@ -197,7 +197,7 @@ func (s *Store) FederationStateBatchDelete(idx uint64, datacenters []string) err
defer tx.Abort()
for _, datacenter := range datacenters {
if err := s.federationStateDeleteTxn(tx, idx, datacenter); err != nil {
if err := federationStateDeleteTxn(tx, idx, datacenter); err != nil {
return err
}
}
@ -205,7 +205,7 @@ func (s *Store) FederationStateBatchDelete(idx uint64, datacenters []string) err
return tx.Commit()
}
func (s *Store) federationStateDeleteTxn(tx *txn, idx uint64, datacenter string) error {
func federationStateDeleteTxn(tx *txn, idx uint64, datacenter string) error {
// Try to retrieve the existing federation state.
existing, err := tx.First(federationStateTableName, "id", datacenter)
if err != nil {

10
agent/consul/state/intention.go
View File

@ -136,7 +136,7 @@ func (s *Store) Intentions(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta) (
idx = 1
}
iter, err := s.intentionListTxn(tx, entMeta)
iter, err := intentionListTxn(tx, entMeta)
if err != nil {
return 0, nil, fmt.Errorf("failed intention lookup: %s", err)
}
@ -160,7 +160,7 @@ func (s *Store) IntentionSet(idx uint64, ixn *structs.Intention) error {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.intentionSetTxn(tx, idx, ixn); err != nil {
if err := intentionSetTxn(tx, idx, ixn); err != nil {
return err
}
@ -169,7 +169,7 @@ func (s *Store) IntentionSet(idx uint64, ixn *structs.Intention) error {
// intentionSetTxn is the inner method used to insert an intention with
// the proper indexes into the state store.
func (s *Store) intentionSetTxn(tx *txn, idx uint64, ixn *structs.Intention) error {
func intentionSetTxn(tx *txn, idx uint64, ixn *structs.Intention) error {
// ID is required
if ixn.ID == "" {
return ErrMissingIntentionID
@ -287,7 +287,7 @@ func (s *Store) IntentionDelete(idx uint64, id string) error {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.intentionDeleteTxn(tx, idx, id); err != nil {
if err := intentionDeleteTxn(tx, idx, id); err != nil {
return fmt.Errorf("failed intention delete: %s", err)
}
@ -296,7 +296,7 @@ func (s *Store) IntentionDelete(idx uint64, id string) error {
// intentionDeleteTxn is the inner method used to delete a intention
// with the proper indexes into the state store.
func (s *Store) intentionDeleteTxn(tx *txn, idx uint64, queryID string) error {
func intentionDeleteTxn(tx *txn, idx uint64, queryID string) error {
// Pull the query.
wrapped, err := tx.First(intentionsTableName, "id", queryID)
if err != nil {

2
agent/consul/state/intention_oss.go
View File

@ -7,7 +7,7 @@ import (
memdb "github.com/hashicorp/go-memdb"
)
func (s *Store) intentionListTxn(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
func intentionListTxn(tx *txn, _ *structs.EnterpriseMeta) (memdb.ResultIterator, error) {
// Get all intentions
return tx.Get(intentionsTableName, "id")
}

38
agent/consul/state/kvs.go
View File

@ -69,7 +69,7 @@ func (s *Snapshot) Tombstones() (memdb.ResultIterator, error) {
// KVS is used when restoring from a snapshot. Use KVSSet for general inserts.
func (s *Restore) KVS(entry *structs.DirEntry) error {
if err := s.store.insertKVTxn(s.tx, entry, true); err != nil {
if err := insertKVTxn(s.tx, entry, true); err != nil {
return fmt.Errorf("failed inserting kvs entry: %s", err)
}
@ -105,7 +105,7 @@ func (s *Store) KVSSet(idx uint64, entry *structs.DirEntry) error {
defer tx.Abort()
// Perform the actual set.
if err := s.kvsSetTxn(tx, idx, entry, false); err != nil {
if err := kvsSetTxn(tx, idx, entry, false); err != nil {
return err
}
@ -117,7 +117,7 @@ func (s *Store) KVSSet(idx uint64, entry *structs.DirEntry) error {
// If updateSession is true, then the incoming entry will set the new
// session (should be validated before calling this). Otherwise, we will keep
// whatever the existing session is.
func (s *Store) kvsSetTxn(tx *txn, idx uint64, entry *structs.DirEntry, updateSession bool) error {
func kvsSetTxn(tx *txn, idx uint64, entry *structs.DirEntry, updateSession bool) error {
// Retrieve an existing KV pair
existingNode, err := firstWithTxn(tx, "kvs", "id", entry.Key, &entry.EnterpriseMeta)
if err != nil {
@ -153,7 +153,7 @@ func (s *Store) kvsSetTxn(tx *txn, idx uint64, entry *structs.DirEntry, updateSe
entry.ModifyIndex = idx
// Store the kv pair in the state store and update the index.
if err := s.insertKVTxn(tx, entry, false); err != nil {
if err := insertKVTxn(tx, entry, false); err != nil {
return fmt.Errorf("failed inserting kvs entry: %s", err)
}
@ -165,12 +165,12 @@ func (s *Store) KVSGet(ws memdb.WatchSet, key string, entMeta *structs.Enterpris
tx := s.db.Txn(false)
defer tx.Abort()
return s.kvsGetTxn(tx, ws, key, entMeta)
return kvsGetTxn(tx, ws, key, entMeta)
}
// kvsGetTxn is the inner method that gets a KVS entry inside an existing
// transaction.
func (s *Store) kvsGetTxn(tx *txn,
func kvsGetTxn(tx *txn,
ws memdb.WatchSet, key string, entMeta *structs.EnterpriseMeta) (uint64, *structs.DirEntry, error) {
// Get the table index.
@ -209,7 +209,7 @@ func (s *Store) kvsListTxn(tx *txn,
// Get the table indexes.
idx := kvsMaxIndex(tx, entMeta)
lindex, entries, err := s.kvsListEntriesTxn(tx, ws, prefix, entMeta)
lindex, entries, err := kvsListEntriesTxn(tx, ws, prefix, entMeta)
if err != nil {
return 0, nil, fmt.Errorf("failed kvs lookup: %s", err)
}
@ -267,7 +267,7 @@ func (s *Store) kvsDeleteTxn(tx *txn, idx uint64, key string, entMeta *structs.E
return fmt.Errorf("failed adding to graveyard: %s", err)
}
return s.kvsDeleteWithEntry(tx, entry.(*structs.DirEntry), idx)
return kvsDeleteWithEntry(tx, entry.(*structs.DirEntry), idx)
}
// KVSDeleteCAS is used to try doing a KV delete operation with a given
@ -319,7 +319,7 @@ func (s *Store) KVSSetCAS(idx uint64, entry *structs.DirEntry) (bool, error) {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
set, err := s.kvsSetCASTxn(tx, idx, entry)
set, err := kvsSetCASTxn(tx, idx, entry)
if !set || err != nil {
return false, err
}
@ -330,7 +330,7 @@ func (s *Store) KVSSetCAS(idx uint64, entry *structs.DirEntry) (bool, error) {
// kvsSetCASTxn is the inner method used to do a CAS inside an existing
// transaction.
func (s *Store) kvsSetCASTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool, error) {
func kvsSetCASTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool, error) {
// Retrieve the existing entry.
existing, err := firstWithTxn(tx, "kvs", "id", entry.Key, &entry.EnterpriseMeta)
if err != nil {
@ -351,7 +351,7 @@ func (s *Store) kvsSetCASTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool
}
// If we made it this far, we should perform the set.
if err := s.kvsSetTxn(tx, idx, entry, false); err != nil {
if err := kvsSetTxn(tx, idx, entry, false); err != nil {
return false, err
}
return true, nil
@ -383,7 +383,7 @@ func (s *Store) KVSLock(idx uint64, entry *structs.DirEntry) (bool, error) {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
locked, err := s.kvsLockTxn(tx, idx, entry)
locked, err := kvsLockTxn(tx, idx, entry)
if !locked || err != nil {
return false, err
}
@ -394,7 +394,7 @@ func (s *Store) KVSLock(idx uint64, entry *structs.DirEntry) (bool, error) {
// kvsLockTxn is the inner method that does a lock inside an existing
// transaction.
func (s *Store) kvsLockTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool, error) {
func kvsLockTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool, error) {
// Verify that a session is present.
if entry.Session == "" {
return false, fmt.Errorf("missing session")
@ -437,7 +437,7 @@ func (s *Store) kvsLockTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool,
entry.ModifyIndex = idx
// If we made it this far, we should perform the set.
if err := s.kvsSetTxn(tx, idx, entry, true); err != nil {
if err := kvsSetTxn(tx, idx, entry, true); err != nil {
return false, err
}
return true, nil
@ -449,7 +449,7 @@ func (s *Store) KVSUnlock(idx uint64, entry *structs.DirEntry) (bool, error) {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
unlocked, err := s.kvsUnlockTxn(tx, idx, entry)
unlocked, err := kvsUnlockTxn(tx, idx, entry)
if !unlocked || err != nil {
return false, err
}
@ -460,7 +460,7 @@ func (s *Store) KVSUnlock(idx uint64, entry *structs.DirEntry) (bool, error) {
// kvsUnlockTxn is the inner method that does an unlock inside an existing
// transaction.
func (s *Store) kvsUnlockTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool, error) {
func kvsUnlockTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool, error) {
// Verify that a session is present.
if entry.Session == "" {
return false, fmt.Errorf("missing session")
@ -490,7 +490,7 @@ func (s *Store) kvsUnlockTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool
entry.ModifyIndex = idx
// If we made it this far, we should perform the set.
if err := s.kvsSetTxn(tx, idx, entry, true); err != nil {
if err := kvsSetTxn(tx, idx, entry, true); err != nil {
return false, err
}
return true, nil
@ -498,7 +498,7 @@ func (s *Store) kvsUnlockTxn(tx *txn, idx uint64, entry *structs.DirEntry) (bool
// kvsCheckSessionTxn checks to see if the given session matches the current
// entry for a key.
func (s *Store) kvsCheckSessionTxn(tx *txn,
func kvsCheckSessionTxn(tx *txn,
key string, session string, entMeta *structs.EnterpriseMeta) (*structs.DirEntry, error) {
entry, err := firstWithTxn(tx, "kvs", "id", key, entMeta)
@ -519,7 +519,7 @@ func (s *Store) kvsCheckSessionTxn(tx *txn,
// kvsCheckIndexTxn checks to see if the given modify index matches the current
// entry for a key.
func (s *Store) kvsCheckIndexTxn(tx *txn,
func kvsCheckIndexTxn(tx *txn,
key string, cidx uint64, entMeta *structs.EnterpriseMeta) (*structs.DirEntry, error) {
entry, err := firstWithTxn(tx, "kvs", "id", key, entMeta)

6
agent/consul/state/kvs_oss.go
View File

@ -16,7 +16,7 @@ func kvsIndexer() *memdb.StringFieldIndex {
}
}
func (s *Store) insertKVTxn(tx *txn, entry *structs.DirEntry, updateMax bool) error {
func insertKVTxn(tx *txn, entry *structs.DirEntry, updateMax bool) error {
if err := tx.Insert("kvs", entry); err != nil {
return err
}
@ -33,7 +33,7 @@ func (s *Store) insertKVTxn(tx *txn, entry *structs.DirEntry, updateMax bool) er
return nil
}
func (s *Store) kvsListEntriesTxn(tx *txn, ws memdb.WatchSet, prefix string, entMeta *structs.EnterpriseMeta) (uint64, structs.DirEntries, error) {
func kvsListEntriesTxn(tx *txn, ws memdb.WatchSet, prefix string, entMeta *structs.EnterpriseMeta) (uint64, structs.DirEntries, error) {
var ents structs.DirEntries
var lindex uint64
@ -81,7 +81,7 @@ func kvsMaxIndex(tx *txn, entMeta *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "kvs", "tombstones")
}
func (s *Store) kvsDeleteWithEntry(tx *txn, entry *structs.DirEntry, idx uint64) error {
func kvsDeleteWithEntry(tx *txn, entry *structs.DirEntry, idx uint64) error {
// Delete the entry and update the index.
if err := tx.Delete("kvs", entry); err != nil {
return fmt.Errorf("failed deleting kvs entry: %s", err)

8
agent/consul/state/prepared_query.go
View File

@ -133,7 +133,7 @@ func (s *Store) PreparedQuerySet(idx uint64, query *structs.PreparedQuery) error
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.preparedQuerySetTxn(tx, idx, query); err != nil {
if err := preparedQuerySetTxn(tx, idx, query); err != nil {
return err
}
@ -142,7 +142,7 @@ func (s *Store) PreparedQuerySet(idx uint64, query *structs.PreparedQuery) error
// preparedQuerySetTxn is the inner method used to insert a prepared query with
// the proper indexes into the state store.
func (s *Store) preparedQuerySetTxn(tx *txn, idx uint64, query *structs.PreparedQuery) error {
func preparedQuerySetTxn(tx *txn, idx uint64, query *structs.PreparedQuery) error {
// Check that the ID is set.
if query.ID == "" {
return ErrMissingQueryID
@ -249,7 +249,7 @@ func (s *Store) PreparedQueryDelete(idx uint64, queryID string) error {
tx := s.db.WriteTxn(idx)
defer tx.Abort()
if err := s.preparedQueryDeleteTxn(tx, idx, queryID); err != nil {
if err := preparedQueryDeleteTxn(tx, idx, queryID); err != nil {
return fmt.Errorf("failed prepared query delete: %s", err)
}
@ -258,7 +258,7 @@ func (s *Store) PreparedQueryDelete(idx uint64, queryID string) error {
// preparedQueryDeleteTxn is the inner method used to delete a prepared query
// with the proper indexes into the state store.
func (s *Store) preparedQueryDeleteTxn(tx *txn, idx uint64, queryID string) error {
func preparedQueryDeleteTxn(tx *txn, idx uint64, queryID string) error {
// Pull the query.
wrapped, err := tx.First("prepared-queries", "id", queryID)
if err != nil {

24
agent/consul/state/session.go
View File

@ -146,7 +146,7 @@ func (s *Snapshot) Sessions() (memdb.ResultIterator, error) {
// Session is used when restoring from a snapshot. For general inserts, use
// SessionCreate.
func (s *Restore) Session(sess *structs.Session) error {
if err := s.store.insertSessionTxn(s.tx, sess, sess.ModifyIndex, true); err != nil {
if err := insertSessionTxn(s.tx, sess, sess.ModifyIndex, true); err != nil {
return fmt.Errorf("failed inserting session: %s", err)
}
@ -166,7 +166,7 @@ func (s *Store) SessionCreate(idx uint64, sess *structs.Session) error {
// future.
// Call the session creation
if err := s.sessionCreateTxn(tx, idx, sess); err != nil {
if err := sessionCreateTxn(tx, idx, sess); err != nil {
return err
}
@ -176,7 +176,7 @@ func (s *Store) SessionCreate(idx uint64, sess *structs.Session) error {
// sessionCreateTxn is the inner method used for creating session entries in
// an open transaction. Any health checks registered with the session will be
// checked for failing status. Returns any error encountered.
func (s *Store) sessionCreateTxn(tx *txn, idx uint64, sess *structs.Session) error {
func sessionCreateTxn(tx *txn, idx uint64, sess *structs.Session) error {
// Check that we have a session ID
if sess.ID == "" {
return ErrMissingSessionID
@ -208,12 +208,12 @@ func (s *Store) sessionCreateTxn(tx *txn, idx uint64, sess *structs.Session) err
}
// Verify that all session checks exist
if err := s.validateSessionChecksTxn(tx, sess); err != nil {
if err := validateSessionChecksTxn(tx, sess); err != nil {
return err
}
// Insert the session
if err := s.insertSessionTxn(tx, sess, idx, false); err != nil {
if err := insertSessionTxn(tx, sess, idx, false); err != nil {
return fmt.Errorf("failed inserting session: %s", err)
}
@ -228,7 +228,7 @@ func (s *Store) SessionGet(ws memdb.WatchSet,
defer tx.Abort()
// Get the table index.
idx := s.sessionMaxIndex(tx, entMeta)
idx := sessionMaxIndex(tx, entMeta)
// Look up the session by its ID
watchCh, session, err := firstWatchWithTxn(tx, "sessions", "id", sessionID, entMeta)
@ -249,7 +249,7 @@ func (s *Store) SessionList(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta)
defer tx.Abort()
// Get the table index.
idx := s.sessionMaxIndex(tx, entMeta)
idx := sessionMaxIndex(tx, entMeta)
// Query all of the active sessions.
sessions, err := getWithTxn(tx, "sessions", "id_prefix", "", entMeta)
@ -274,10 +274,10 @@ func (s *Store) NodeSessions(ws memdb.WatchSet, nodeID string, entMeta *structs.
defer tx.Abort()
// Get the table index.
idx := s.sessionMaxIndex(tx, entMeta)
idx := sessionMaxIndex(tx, entMeta)
// Get all of the sessions which belong to the node
result, err := s.nodeSessionsTxn(tx, ws, nodeID, entMeta)
result, err := nodeSessionsTxn(tx, ws, nodeID, entMeta)
if err != nil {
return 0, nil, err
}
@ -313,7 +313,7 @@ func (s *Store) deleteSessionTxn(tx *txn, idx uint64, sessionID string, entMeta
// Delete the session and write the new index.
session := sess.(*structs.Session)
if err := s.sessionDeleteWithSession(tx, session, idx); err != nil {
if err := sessionDeleteWithSession(tx, session, idx); err != nil {
return fmt.Errorf("failed deleting session: %v", err)
}
@ -346,7 +346,7 @@ func (s *Store) deleteSessionTxn(tx *txn, idx uint64, sessionID string, entMeta
// respects the transaction we are in.
e := obj.(*structs.DirEntry).Clone()
e.Session = ""
if err := s.kvsSetTxn(tx, idx, e, true); err != nil {
if err := kvsSetTxn(tx, idx, e, true); err != nil {
return fmt.Errorf("failed kvs update: %s", err)
}
@ -403,7 +403,7 @@ func (s *Store) deleteSessionTxn(tx *txn, idx uint64, sessionID string, entMeta
// Do the delete in a separate loop so we don't trash the iterator.
for _, id := range ids {
if err := s.preparedQueryDeleteTxn(tx, idx, id); err != nil {
if err := preparedQueryDeleteTxn(tx, idx, id); err != nil {
return fmt.Errorf("failed prepared query delete: %s", err)
}
}

14
agent/consul/state/session_oss.go
View File

@ -35,7 +35,7 @@ func nodeChecksIndexer() *memdb.CompoundIndex {
}
}
func (s *Store) sessionDeleteWithSession(tx *txn, session *structs.Session, idx uint64) error {
func sessionDeleteWithSession(tx *txn, session *structs.Session, idx uint64) error {
if err := tx.Delete("sessions", session); err != nil {
return fmt.Errorf("failed deleting session: %s", err)
}
@ -48,7 +48,7 @@ func (s *Store) sessionDeleteWithSession(tx *txn, session *structs.Session, idx
return nil
}
func (s *Store) insertSessionTxn(tx *txn, session *structs.Session, idx uint64, updateMax bool) error {
func insertSessionTxn(tx *txn, session *structs.Session, idx uint64, updateMax bool) error {
if err := tx.Insert("sessions", session); err != nil {
return err
}
@ -80,11 +80,11 @@ func (s *Store) insertSessionTxn(tx *txn, session *structs.Session, idx uint64,
return nil
}
func (s *Store) allNodeSessionsTxn(tx *txn, node string) (structs.Sessions, error) {
return s.nodeSessionsTxn(tx, nil, node, nil)
func allNodeSessionsTxn(tx *txn, node string) (structs.Sessions, error) {
return nodeSessionsTxn(tx, nil, node, nil)
}
func (s *Store) nodeSessionsTxn(tx *txn,
func nodeSessionsTxn(tx *txn,
ws memdb.WatchSet, node string, entMeta *structs.EnterpriseMeta) (structs.Sessions, error) {
sessions, err := tx.Get("sessions", "node", node)
@ -100,11 +100,11 @@ func (s *Store) nodeSessionsTxn(tx *txn,
return result, nil
}
func (s *Store) sessionMaxIndex(tx *txn, entMeta *structs.EnterpriseMeta) uint64 {
func sessionMaxIndex(tx *txn, entMeta *structs.EnterpriseMeta) uint64 {
return maxIndexTxn(tx, "sessions")
}
func (s *Store) validateSessionChecksTxn(tx *txn, session *structs.Session) error {
func validateSessionChecksTxn(tx *txn, session *structs.Session) error {
// Go over the session checks and ensure they exist.
for _, checkID := range session.CheckIDs() {
check, err := tx.First("checks", "id", session.Node, string(checkID))

38
agent/consul/state/txn.go
View File

@ -15,7 +15,7 @@ func (s *Store) txnKVS(tx *txn, idx uint64, op *structs.TxnKVOp) (structs.TxnRes
switch op.Verb {
case api.KVSet:
entry = &op.DirEnt
err = s.kvsSetTxn(tx, idx, entry, false)
err = kvsSetTxn(tx, idx, entry, false)
case api.KVDelete:
err = s.kvsDeleteTxn(tx, idx, op.DirEnt.Key, &op.DirEnt.EnterpriseMeta)
@ -33,7 +33,7 @@ func (s *Store) txnKVS(tx *txn, idx uint64, op *structs.TxnKVOp) (structs.TxnRes
case api.KVCAS:
var ok bool
entry = &op.DirEnt
ok, err = s.kvsSetCASTxn(tx, idx, entry)
ok, err = kvsSetCASTxn(tx, idx, entry)
if !ok && err == nil {
err = fmt.Errorf("failed to set key %q, index is stale", op.DirEnt.Key)
}
@ -41,7 +41,7 @@ func (s *Store) txnKVS(tx *txn, idx uint64, op *structs.TxnKVOp) (structs.TxnRes
case api.KVLock:
var ok bool
entry = &op.DirEnt
ok, err = s.kvsLockTxn(tx, idx, entry)
ok, err = kvsLockTxn(tx, idx, entry)
if !ok && err == nil {
err = fmt.Errorf("failed to lock key %q, lock is already held", op.DirEnt.Key)
}
@ -49,13 +49,13 @@ func (s *Store) txnKVS(tx *txn, idx uint64, op *structs.TxnKVOp) (structs.TxnRes
case api.KVUnlock:
var ok bool
entry = &op.DirEnt
ok, err = s.kvsUnlockTxn(tx, idx, entry)
ok, err = kvsUnlockTxn(tx, idx, entry)
if !ok && err == nil {
err = fmt.Errorf("failed to unlock key %q, lock isn't held, or is held by another session", op.DirEnt.Key)
}
case api.KVGet:
_, entry, err = s.kvsGetTxn(tx, nil, op.DirEnt.Key, &op.DirEnt.EnterpriseMeta)
_, entry, err = kvsGetTxn(tx, nil, op.DirEnt.Key, &op.DirEnt.EnterpriseMeta)
if entry == nil && err == nil {
err = fmt.Errorf("key %q doesn't exist", op.DirEnt.Key)
}
@ -73,13 +73,13 @@ func (s *Store) txnKVS(tx *txn, idx uint64, op *structs.TxnKVOp) (structs.TxnRes
}
case api.KVCheckSession:
entry, err = s.kvsCheckSessionTxn(tx, op.DirEnt.Key, op.DirEnt.Session, &op.DirEnt.EnterpriseMeta)
entry, err = kvsCheckSessionTxn(tx, op.DirEnt.Key, op.DirEnt.Session, &op.DirEnt.EnterpriseMeta)
case api.KVCheckIndex:
entry, err = s.kvsCheckIndexTxn(tx, op.DirEnt.Key, op.DirEnt.ModifyIndex, &op.DirEnt.EnterpriseMeta)
entry, err = kvsCheckIndexTxn(tx, op.DirEnt.Key, op.DirEnt.ModifyIndex, &op.DirEnt.EnterpriseMeta)
case api.KVCheckNotExists:
_, entry, err = s.kvsGetTxn(tx, nil, op.DirEnt.Key, &op.DirEnt.EnterpriseMeta)
_, entry, err = kvsGetTxn(tx, nil, op.DirEnt.Key, &op.DirEnt.EnterpriseMeta)
if entry != nil && err == nil {
err = fmt.Errorf("key %q exists", op.DirEnt.Key)
}
@ -115,7 +115,7 @@ func (s *Store) txnSession(tx *txn, idx uint64, op *structs.TxnSessionOp) error
switch op.Verb {
case api.SessionDelete:
err = s.sessionDeleteWithSession(tx, &op.Session, idx)
err = sessionDeleteWithSession(tx, &op.Session, idx)
default:
err = fmt.Errorf("unknown Session verb %q", op.Verb)
}
@ -130,9 +130,9 @@ func (s *Store) txnSession(tx *txn, idx uint64, op *structs.TxnSessionOp) error
func (s *Store) txnIntention(tx *txn, idx uint64, op *structs.TxnIntentionOp) error {
switch op.Op {
case structs.IntentionOpCreate, structs.IntentionOpUpdate:
return s.intentionSetTxn(tx, idx, op.Intention)
return intentionSetTxn(tx, idx, op.Intention)
case structs.IntentionOpDelete:
return s.intentionDeleteTxn(tx, idx, op.Intention.ID)
return intentionDeleteTxn(tx, idx, op.Intention.ID)
default:
return fmt.Errorf("unknown Intention op %q", op.Op)
}
@ -211,7 +211,7 @@ func (s *Store) txnNode(tx *txn, idx uint64, op *structs.TxnNodeOp) (structs.Txn
func (s *Store) txnService(tx *txn, idx uint64, op *structs.TxnServiceOp) (structs.TxnResults, error) {
switch op.Verb {
case api.ServiceGet:
entry, err := s.getNodeServiceTxn(tx, op.Node, op.Service.ID, &op.Service.EnterpriseMeta)
entry, err := getNodeServiceTxn(tx, op.Node, op.Service.ID, &op.Service.EnterpriseMeta)
switch {
case err != nil:
return nil, err
@ -222,14 +222,14 @@ func (s *Store) txnService(tx *txn, idx uint64, op *structs.TxnServiceOp) (struc
}
case api.ServiceSet:
if err := s.ensureServiceTxn(tx, idx, op.Node, &op.Service); err != nil {
if err := ensureServiceTxn(tx, idx, op.Node, &op.Service); err != nil {
return nil, err
}
entry, err := s.getNodeServiceTxn(tx, op.Node, op.Service.ID, &op.Service.EnterpriseMeta)
entry, err := getNodeServiceTxn(tx, op.Node, op.Service.ID, &op.Service.EnterpriseMeta)
return newTxnResultFromNodeServiceEntry(entry), err
case api.ServiceCAS:
err := s.ensureServiceCASTxn(tx, idx, op.Node, &op.Service)
err := ensureServiceCASTxn(tx, idx, op.Node, &op.Service)
switch {
case err == errCASCompareFailed:
err := fmt.Errorf("failed to set service %q on node %q, index is stale", op.Service.ID, op.Node)
@ -238,7 +238,7 @@ func (s *Store) txnService(tx *txn, idx uint64, op *structs.TxnServiceOp) (struc
return nil, err
}
entry, err := s.getNodeServiceTxn(tx, op.Node, op.Service.ID, &op.Service.EnterpriseMeta)
entry, err := getNodeServiceTxn(tx, op.Node, op.Service.ID, &op.Service.EnterpriseMeta)
return newTxnResultFromNodeServiceEntry(entry), err
case api.ServiceDelete:
@ -276,7 +276,7 @@ func (s *Store) txnCheck(tx *txn, idx uint64, op *structs.TxnCheckOp) (structs.T
switch op.Verb {
case api.CheckGet:
_, entry, err = s.getNodeCheckTxn(tx, op.Check.Node, op.Check.CheckID, &op.Check.EnterpriseMeta)
_, entry, err = getNodeCheckTxn(tx, op.Check.Node, op.Check.CheckID, &op.Check.EnterpriseMeta)
if entry == nil && err == nil {
err = fmt.Errorf("check %q on node %q doesn't exist", op.Check.CheckID, op.Check.Node)
}
@ -284,7 +284,7 @@ func (s *Store) txnCheck(tx *txn, idx uint64, op *structs.TxnCheckOp) (structs.T
case api.CheckSet:
err = s.ensureCheckTxn(tx, idx, &op.Check)
if err == nil {
_, entry, err = s.getNodeCheckTxn(tx, op.Check.Node, op.Check.CheckID, &op.Check.EnterpriseMeta)
_, entry, err = getNodeCheckTxn(tx, op.Check.Node, op.Check.CheckID, &op.Check.EnterpriseMeta)
}
case api.CheckCAS:
@ -295,7 +295,7 @@ func (s *Store) txnCheck(tx *txn, idx uint64, op *structs.TxnCheckOp) (structs.T
err = fmt.Errorf("failed to set check %q on node %q, index is stale", entry.CheckID, entry.Node)
break
}
_, entry, err = s.getNodeCheckTxn(tx, op.Check.Node, op.Check.CheckID, &op.Check.EnterpriseMeta)
_, entry, err = getNodeCheckTxn(tx, op.Check.Node, op.Check.CheckID, &op.Check.EnterpriseMeta)
case api.CheckDelete:
err = s.deleteCheckTxn(tx, idx, op.Check.Node, op.Check.CheckID, &op.Check.EnterpriseMeta)