convert expiration indexed in ACLToken table to use `indexerSingle` (#11018)

* move intFromBool to be available for oss

* add expiry indexes

* remove dead code: `TokenExpirationIndex`

* fix remove indexer `TokenExpirationIndex`

* fix rebase issue

agent/consul/state/acl.go

@@ -1,7 +1,6 @@
 package state
 
 import (
-	"encoding/binary"
 	"fmt"
 	"time"
 
@@ -11,54 +10,6 @@ import (
 	pbacl "github.com/hashicorp/consul/proto/pbacl"
 )
 
-type TokenExpirationIndex struct {
-	LocalFilter bool
-}
-
-func (s *TokenExpirationIndex) encodeTime(t time.Time) []byte {
-	val := t.Unix()
-	buf := make([]byte, 8)
-	binary.BigEndian.PutUint64(buf, uint64(val))
-	return buf
-}
-
-func (s *TokenExpirationIndex) FromObject(obj interface{}) (bool, []byte, error) {
-	token, ok := obj.(*structs.ACLToken)
-	if !ok {
-		return false, nil, fmt.Errorf("object is not an ACLToken")
-	}
-	if s.LocalFilter != token.Local {
-		return false, nil, nil
-	}
-	if !token.HasExpirationTime() {
-		return false, nil, nil
-	}
-	if token.ExpirationTime.Unix() < 0 {
-		return false, nil, fmt.Errorf("token expiration time cannot be before the unix epoch: %s", token.ExpirationTime)
-	}
-
-	buf := s.encodeTime(*token.ExpirationTime)
-
-	return true, buf, nil
-}
-
-func (s *TokenExpirationIndex) FromArgs(args ...interface{}) ([]byte, error) {
-	if len(args) != 1 {
-		return nil, fmt.Errorf("must provide only a single argument")
-	}
-	arg, ok := args[0].(time.Time)
-	if !ok {
-		return nil, fmt.Errorf("argument must be a time.Time: %#v", args[0])
-	}
-	if arg.Unix() < 0 {
-		return nil, fmt.Errorf("argument must be a time.Time after the unix epoch: %s", args[0])
-	}
-
-	buf := s.encodeTime(arg)
-
-	return buf, nil
-}
-
 // ACLTokens is used when saving a snapshot
 func (s *Snapshot) ACLTokens() (memdb.ResultIterator, error) {
 	iter, err := s.tx.Get(tableACLTokens, "id")
@@ -850,9 +801,9 @@ func (s *Store) ACLTokenListExpired(local bool, asOf time.Time, max int) (struct
 
 func (s *Store) expiresIndexName(local bool) string {
 	if local {
-		return "expires-local"
+		return indexExpiresLocal
 	}
-	return "expires-global"
+	return indexExpiresGlobal
 }
 
 // ACLTokenDeleteBySecret is used to remove an existing ACL from the state store. If

agent/consul/state/acl_schema.go

@@ -22,6 +22,8 @@ const (
 	indexAuthMethod    = "authmethod"
 	indexLocality      = "locality"
 	indexName          = "name"
+	indexExpiresGlobal = "expires-global"
+	indexExpiresLocal  = "expires-local"
 )
 
 func tokensTableSchema() *memdb.TableSchema {
@@ -84,17 +86,23 @@ func tokensTableSchema() *memdb.TableSchema {
 					writeIndex: writeIndex(indexLocalFromACLToken),
 				},
 			},
-			"expires-global": {
+			indexExpiresGlobal: {
-				Name:         "expires-global",
+				Name:         indexExpiresGlobal,
 				AllowMissing: true,
 				Unique:       false,
-				Indexer:      &TokenExpirationIndex{LocalFilter: false},
+				Indexer: indexerSingle{
+					readIndex:  readIndex(indexFromTimeQuery),
+					writeIndex: writeIndex(indexExpiresGlobalFromACLToken),
 				},
-			"expires-local": {
+			},
-				Name:         "expires-local",
+			indexExpiresLocal: {
+				Name:         indexExpiresLocal,
 				AllowMissing: true,
 				Unique:       false,
-				Indexer:      &TokenExpirationIndex{LocalFilter: true},
+				Indexer: indexerSingle{
+					readIndex:  readIndex(indexFromTimeQuery),
+					writeIndex: writeIndex(indexExpiresLocalFromACLToken),
+				},
 			},
 
 			//DEPRECATED (ACL-Legacy-Compat) - This index is only needed while we support upgrading v1 to v2 acls
@@ -423,3 +431,42 @@ func indexLocalFromACLToken(raw interface{}) ([]byte, error) {
 	b.Bool(p.Local)
 	return b.Bytes(), nil
 }
+
+func indexFromTimeQuery(arg interface{}) ([]byte, error) {
+	p, ok := arg.(*TimeQuery)
+	if !ok {
+		return nil, fmt.Errorf("unexpected type %T for TimeQuery index", arg)
+	}
+
+	var b indexBuilder
+	b.Time(p.Value)
+	return b.Bytes(), nil
+}
+
+func indexExpiresLocalFromACLToken(raw interface{}) ([]byte, error) {
+	return indexExpiresFromACLToken(raw, true)
+}
+
+func indexExpiresGlobalFromACLToken(raw interface{}) ([]byte, error) {
+	return indexExpiresFromACLToken(raw, false)
+}
+
+func indexExpiresFromACLToken(raw interface{}, local bool) ([]byte, error) {
+	p, ok := raw.(*structs.ACLToken)
+	if !ok {
+		return nil, fmt.Errorf("unexpected type %T for structs.ACLToken index", raw)
+	}
+	if p.Local != local {
+		return nil, errMissingValueForIndex
+	}
+	if !p.HasExpirationTime() {
+		return nil, errMissingValueForIndex
+	}
+	if p.ExpirationTime.Unix() < 0 {
+		return nil, fmt.Errorf("token expiration time cannot be before the unix epoch: %s", p.ExpirationTime)
+	}
+
+	var b indexBuilder
+	b.Time(*p.ExpirationTime)
+	return b.Bytes(), nil
+}

agent/consul/state/acl_test.go

@@ -3819,13 +3819,19 @@ func TestTokenPoliciesIndex(t *testing.T) {
 		Name:         "global",
 		AllowMissing: true,
 		Unique:       false,
-		Indexer:      &TokenExpirationIndex{LocalFilter: false},
+		Indexer: indexerSingle{
+			readIndex:  readIndex(indexFromTimeQuery),
+			writeIndex: writeIndex(indexExpiresGlobalFromACLToken),
+		},
 	}
 	localIndex := &memdb.IndexSchema{
 		Name:         "local",
 		AllowMissing: true,
 		Unique:       false,
-		Indexer:      &TokenExpirationIndex{LocalFilter: true},
+		Indexer: indexerSingle{
+			readIndex:  readIndex(indexFromTimeQuery),
+			writeIndex: writeIndex(indexExpiresLocalFromACLToken),
+		},
 	}
 	schema := &memdb.DBSchema{
 		Tables: map[string]*memdb.TableSchema{

agent/consul/state/indexer.go

@@ -5,6 +5,9 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"time"
+
+	"github.com/hashicorp/consul/agent/structs"
 )
 
 // indexerSingle implements both memdb.Indexer and memdb.SingleIndexer. It may
@@ -137,3 +140,27 @@ func (b *indexBuilder) Bytes() []byte {
 func (b *indexBuilder) Bool(v bool) {
 	b.Raw([]byte{intFromBool(v)})
 }
+
+type TimeQuery struct {
+	Value time.Time
+	structs.EnterpriseMeta
+}
+
+// NamespaceOrDefault exists because structs.EnterpriseMeta uses a pointer
+// receiver for this method. Remove once that is fixed.
+func (q TimeQuery) NamespaceOrDefault() string {
+	return q.EnterpriseMeta.NamespaceOrDefault()
+}
+
+// PartitionOrDefault exists because structs.EnterpriseMeta uses a pointer
+// receiver for this method. Remove once that is fixed.
+func (q TimeQuery) PartitionOrDefault() string {
+	return q.EnterpriseMeta.PartitionOrDefault()
+}
+
+func (b *indexBuilder) Time(t time.Time) {
+	val := t.Unix()
+	buf := make([]byte, 8)
+	binary.BigEndian.PutUint64(buf, uint64(val))
+	(*bytes.Buffer)(b).Write(buf)
+}