auth: store cached permission information in a form of interval tree

This commit change the type of cached permission information from the
home made thing to interval tree. It improves computational complexity
of permission checking from O(n) to O(lg n).

auth/range_perm_cache.go

@@ -15,87 +15,11 @@
 package auth
 
 import (
-	"bytes"
-	"sort"
-
 	"github.com/coreos/etcd/auth/authpb"
 	"github.com/coreos/etcd/mvcc/backend"
+	"github.com/coreos/etcd/pkg/adt"
 )
 
-// isSubset returns true if a is a subset of b.
-// If a is a prefix of b, then a is a subset of b.
-// Given intervals [a1,a2) and [b1,b2), is
-// the a interval a subset of b?
-func isSubset(a, b *rangePerm) bool {
-	switch {
-	case len(a.end) == 0 && len(b.end) == 0:
-		// a, b are both keys
-		return bytes.Equal(a.begin, b.begin)
-	case len(b.end) == 0:
-		// b is a key, a is a range
-		return false
-	case len(a.end) == 0:
-		// a is a key, b is a range. need b1 <= a1 and a1 < b2
-		return bytes.Compare(b.begin, a.begin) <= 0 && bytes.Compare(a.begin, b.end) < 0
-	default:
-		// both are ranges. need b1 <= a1 and a2 <= b2
-		return bytes.Compare(b.begin, a.begin) <= 0 && bytes.Compare(a.end, b.end) <= 0
-	}
-}
-
-func isRangeEqual(a, b *rangePerm) bool {
-	return bytes.Equal(a.begin, b.begin) && bytes.Equal(a.end, b.end)
-}
-
-// mergeRangePerms merges adjacent rangePerms.
-func mergeRangePerms(perms []*rangePerm) (rs []*rangePerm) {
-	if len(perms) < 2 {
-		return perms
-	}
-	sort.Sort(RangePermSliceByBegin(perms))
-
-	// initialize with first rangePerm
-	rs = append(rs, perms[0])
-
-	// merge in-place from 2nd
-	for i := 1; i < len(perms); i++ {
-		rp := rs[len(rs)-1]
-
-		// skip duplicate range
-		if isRangeEqual(rp, perms[i]) {
-			continue
-		}
-
-		// merge ["a", "") with ["a", "c")
-		if bytes.Equal(rp.begin, perms[i].begin) && len(rp.end) == 0 && len(perms[i].end) > 0 {
-			rp.end = perms[i].end
-			continue
-		}
-
-		// do not merge ["a", "b") with ["b", "")
-		if bytes.Equal(rp.end, perms[i].begin) && len(perms[i].end) == 0 {
-			rs = append(rs, perms[i])
-			continue
-		}
-
-		// rp.end < perms[i].begin; found beginning of next range
-		if bytes.Compare(rp.end, perms[i].begin) < 0 {
-			rs = append(rs, perms[i])
-			continue
-		}
-
-		rp.end = getMax(rp.end, perms[i].end)
-	}
-	return
-}
-
-func getMax(a, b []byte) []byte {
-	if bytes.Compare(a, b) > 0 {
-		return a
-	}
-	return b
-}
-
 func getMergedPerms(tx backend.BatchTx, userName string) *unifiedRangePermissions {
 	user := getUser(tx, userName)
 	if user == nil {
@@ -103,7 +27,8 @@ func getMergedPerms(tx backend.BatchTx, userName string) *unifiedRangePermission
 		return nil
 	}
 
-	var readPerms, writePerms []*rangePerm
+	readPerms := &adt.IntervalTree{}
+	writePerms := &adt.IntervalTree{}
 
 	for _, roleName := range user.Roles {
 		role := getRole(tx, roleName)
@@ -112,30 +37,36 @@ func getMergedPerms(tx backend.BatchTx, userName string) *unifiedRangePermission
 		}
 
 		for _, perm := range role.KeyPermission {
-			rp := &rangePerm{begin: perm.Key, end: perm.RangeEnd}
+			var ivl adt.Interval
+
+			if len(perm.RangeEnd) != 0 {
+				ivl = adt.NewStringInterval(string(perm.Key), string(perm.RangeEnd))
+			} else {
+				ivl = adt.NewStringPoint(string(perm.Key))
+			}
 
 			switch perm.PermType {
 			case authpb.READWRITE:
-				readPerms = append(readPerms, rp)
-				writePerms = append(writePerms, rp)
+				readPerms.Insert(ivl, struct{}{})
+				writePerms.Insert(ivl, struct{}{})
 
 			case authpb.READ:
-				readPerms = append(readPerms, rp)
+				readPerms.Insert(ivl, struct{}{})
 
 			case authpb.WRITE:
-				writePerms = append(writePerms, rp)
+				writePerms.Insert(ivl, struct{}{})
 			}
 		}
 	}
 
 	return &unifiedRangePermissions{
-		readPerms:  mergeRangePerms(readPerms),
-		writePerms: mergeRangePerms(writePerms),
+		readPerms:  readPerms,
+		writePerms: writePerms,
 	}
 }
 
-func checkKeyPerm(cachedPerms *unifiedRangePermissions, key, rangeEnd []byte, permtyp authpb.Permission_Type) bool {
-	var tocheck []*rangePerm
+func checkKeyInterval(cachedPerms *unifiedRangePermissions, key, rangeEnd string, permtyp authpb.Permission_Type) bool {
+	var tocheck *adt.IntervalTree
 
 	switch permtyp {
 	case authpb.READ:
@@ -146,18 +77,51 @@ func checkKeyPerm(cachedPerms *unifiedRangePermissions, key, rangeEnd []byte, pe
 		plog.Panicf("unknown auth type: %v", permtyp)
 	}
 
-	requiredPerm := &rangePerm{begin: key, end: rangeEnd}
+	ivl := adt.NewStringInterval(key, rangeEnd)
+
+	isContiguous := true
+	var maxEnd, minBegin adt.Comparable
 
-	for _, perm := range tocheck {
-		if isSubset(requiredPerm, perm) {
+	tocheck.Visit(ivl, func(n *adt.IntervalValue) bool {
+		if minBegin == nil {
+			minBegin = n.Ivl.Begin
+			maxEnd = n.Ivl.End
 			return true
 		}
+
+		if maxEnd.Compare(n.Ivl.Begin) < 0 {
+			isContiguous = false
+			return false
+		}
+
+		if n.Ivl.End.Compare(maxEnd) > 0 {
+			maxEnd = n.Ivl.End
+		}
+
+		return true
+	})
+
+	return isContiguous && maxEnd.Compare(ivl.End) >= 0 && minBegin.Compare(ivl.Begin) <= 0
+}
+
+func checkKeyPoint(cachedPerms *unifiedRangePermissions, key string, permtyp authpb.Permission_Type) bool {
+	var tocheck *adt.IntervalTree
+
+	switch permtyp {
+	case authpb.READ:
+		tocheck = cachedPerms.readPerms
+	case authpb.WRITE:
+		tocheck = cachedPerms.writePerms
+	default:
+		plog.Panicf("unknown auth type: %v", permtyp)
 	}
 
-	return false
+	pt := adt.NewStringPoint(key)
+
+	return tocheck.Contains(pt)
 }
 
-func (as *authStore) isRangeOpPermitted(tx backend.BatchTx, userName string, key, rangeEnd []byte, permtyp authpb.Permission_Type) bool {
+func (as *authStore) isRangeOpPermitted(tx backend.BatchTx, userName string, key, rangeEnd string, permtyp authpb.Permission_Type) bool {
 	// assumption: tx is Lock()ed
 	_, ok := as.rangePermCache[userName]
 	if !ok {
@@ -169,7 +133,11 @@ func (as *authStore) isRangeOpPermitted(tx backend.BatchTx, userName string, key
 		as.rangePermCache[userName] = perms
 	}
 
-	return checkKeyPerm(as.rangePermCache[userName], key, rangeEnd, permtyp)
+	if len(rangeEnd) == 0 {
+		return checkKeyPoint(as.rangePermCache[userName], key, permtyp)
+	}
+
+	return checkKeyInterval(as.rangePermCache[userName], key, rangeEnd, permtyp)
 }
 
 func (as *authStore) clearCachedPerm() {
@@ -181,35 +149,6 @@ func (as *authStore) invalidateCachedPerm(userName string) {
 }
 
 type unifiedRangePermissions struct {
-	// readPerms[i] and readPerms[j] (i != j) don't overlap
-	readPerms []*rangePerm
-	// writePerms[i] and writePerms[j] (i != j) don't overlap, too
-	writePerms []*rangePerm
-}
-
-type rangePerm struct {
-	begin, end []byte
-}
-
-type RangePermSliceByBegin []*rangePerm
-
-func (slice RangePermSliceByBegin) Len() int {
-	return len(slice)
-}
-
-func (slice RangePermSliceByBegin) Less(i, j int) bool {
-	switch bytes.Compare(slice[i].begin, slice[j].begin) {
-	case 0: // begin(i) == begin(j)
-		return bytes.Compare(slice[i].end, slice[j].end) == -1
-
-	case -1: // begin(i) < begin(j)
-		return true
-
-	default:
-		return false
-	}
-}
-
-func (slice RangePermSliceByBegin) Swap(i, j int) {
-	slice[i], slice[j] = slice[j], slice[i]
+	readPerms  *adt.IntervalTree
+	writePerms *adt.IntervalTree
 }

auth/range_perm_cache_test.go

@@ -15,155 +15,45 @@
 package auth
 
 import (
-	"bytes"
-	"fmt"
 	"testing"
-)
-
-func isPermsEqual(a, b []*rangePerm) bool {
-	if len(a) != len(b) {
-		return false
-	}
-
-	for i := range a {
-		if len(b) <= i {
-			return false
-		}
-
-		if !bytes.Equal(a[i].begin, b[i].begin) || !bytes.Equal(a[i].end, b[i].end) {
-			return false
-		}
-	}
 
-	return true
-}
+	"github.com/coreos/etcd/auth/authpb"
+	"github.com/coreos/etcd/pkg/adt"
+)
 
-func TestGetMergedPerms(t *testing.T) {
+func TestRangePermission(t *testing.T) {
 	tests := []struct {
-		params []*rangePerm
-		want   []*rangePerm
+		perms []adt.Interval
+		begin string
+		end   string
+		want  bool
 	}{
-		{ // subsets converge
-			[]*rangePerm{{[]byte{2}, []byte{3}}, {[]byte{2}, []byte{5}}, {[]byte{1}, []byte{4}}},
-			[]*rangePerm{{[]byte{1}, []byte{5}}},
-		},
-		{ // subsets converge
-			[]*rangePerm{{[]byte{0}, []byte{3}}, {[]byte{0}, []byte{1}}, {[]byte{2}, []byte{4}}, {[]byte{0}, []byte{2}}},
-			[]*rangePerm{{[]byte{0}, []byte{4}}},
-		},
-		{ // biggest range at the end
-			[]*rangePerm{{[]byte{2}, []byte{3}}, {[]byte{0}, []byte{2}}, {[]byte{1}, []byte{4}}, {[]byte{0}, []byte{5}}},
-			[]*rangePerm{{[]byte{0}, []byte{5}}},
-		},
-		{ // biggest range at the beginning
-			[]*rangePerm{{[]byte{0}, []byte{5}}, {[]byte{2}, []byte{3}}, {[]byte{0}, []byte{2}}, {[]byte{1}, []byte{4}}},
-			[]*rangePerm{{[]byte{0}, []byte{5}}},
-		},
-		{ // no overlapping ranges
-			[]*rangePerm{{[]byte{2}, []byte{3}}, {[]byte{0}, []byte{1}}, {[]byte{4}, []byte{7}}, {[]byte{8}, []byte{15}}},
-			[]*rangePerm{{[]byte{0}, []byte{1}}, {[]byte{2}, []byte{3}}, {[]byte{4}, []byte{7}}, {[]byte{8}, []byte{15}}},
-		},
-		{
-			[]*rangePerm{makePerm("00", "03"), makePerm("18", "19"), makePerm("02", "08"), makePerm("10", "12")},
-			[]*rangePerm{makePerm("00", "08"), makePerm("10", "12"), makePerm("18", "19")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b")},
-			[]*rangePerm{makePerm("a", "b")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("b", "")},
-			[]*rangePerm{makePerm("a", "b"), makePerm("b", "")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("b", "c")},
-			[]*rangePerm{makePerm("a", "c")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "c"), makePerm("b", "d")},
-			[]*rangePerm{makePerm("a", "d")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("b", "c"), makePerm("d", "e")},
-			[]*rangePerm{makePerm("a", "c"), makePerm("d", "e")},
-		},
 		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("c", "d"), makePerm("e", "f")},
-			[]*rangePerm{makePerm("a", "b"), makePerm("c", "d"), makePerm("e", "f")},
+			[]adt.Interval{adt.NewStringInterval("a", "c"), adt.NewStringInterval("x", "z")},
+			"a", "z",
+			false,
 		},
 		{
-			[]*rangePerm{makePerm("e", "f"), makePerm("c", "d"), makePerm("a", "b")},
-			[]*rangePerm{makePerm("a", "b"), makePerm("c", "d"), makePerm("e", "f")},
+			[]adt.Interval{adt.NewStringInterval("a", "f"), adt.NewStringInterval("c", "d"), adt.NewStringInterval("f", "z")},
+			"a", "z",
+			true,
 		},
 		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("c", "d"), makePerm("a", "z")},
-			[]*rangePerm{makePerm("a", "z")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("c", "d"), makePerm("a", "z"), makePerm("1", "9")},
-			[]*rangePerm{makePerm("1", "9"), makePerm("a", "z")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("c", "d"), makePerm("a", "z"), makePerm("1", "a")},
-			[]*rangePerm{makePerm("1", "z")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("a", "z"), makePerm("5", "6"), makePerm("1", "9")},
-			[]*rangePerm{makePerm("1", "9"), makePerm("a", "z")},
-		},
-		{
-			[]*rangePerm{makePerm("a", "b"), makePerm("b", "c"), makePerm("c", "d"), makePerm("b", "f"), makePerm("1", "9")},
-			[]*rangePerm{makePerm("1", "9"), makePerm("a", "f")},
-		},
-		// overlapping
-		{
-			[]*rangePerm{makePerm("a", "f"), makePerm("b", "g")},
-			[]*rangePerm{makePerm("a", "g")},
-		},
-		// keys
-		{
-			[]*rangePerm{makePerm("a", ""), makePerm("b", "")},
-			[]*rangePerm{makePerm("a", ""), makePerm("b", "")},
-		},
-		{
-			[]*rangePerm{makePerm("a", ""), makePerm("a", "c")},
-			[]*rangePerm{makePerm("a", "c")},
-		},
-		{
-			[]*rangePerm{makePerm("a", ""), makePerm("a", "c"), makePerm("b", "")},
-			[]*rangePerm{makePerm("a", "c")},
-		},
-		{
-			[]*rangePerm{makePerm("a", ""), makePerm("b", "c"), makePerm("b", ""), makePerm("c", ""), makePerm("d", "")},
-			[]*rangePerm{makePerm("a", ""), makePerm("b", "c"), makePerm("c", ""), makePerm("d", "")},
-		},
-		// duplicate ranges
-		{
-			[]*rangePerm{makePerm("a", "f"), makePerm("a", "f")},
-			[]*rangePerm{makePerm("a", "f")},
-		},
-		// duplicate keys
-		{
-			[]*rangePerm{makePerm("a", ""), makePerm("a", ""), makePerm("a", "")},
-			[]*rangePerm{makePerm("a", "")},
+			[]adt.Interval{adt.NewStringInterval("a", "d"), adt.NewStringInterval("a", "b"), adt.NewStringInterval("c", "f")},
+			"a", "f",
+			true,
 		},
 	}
 
 	for i, tt := range tests {
-		result := mergeRangePerms(tt.params)
-		if !isPermsEqual(result, tt.want) {
-			t.Errorf("#%d: result=%q, want=%q", i, sprintPerms(result), sprintPerms(tt.want))
+		readPerms := &adt.IntervalTree{}
+		for _, p := range tt.perms {
+			readPerms.Insert(p, struct{}{})
 		}
-	}
-}
 
-func makePerm(a, b string) *rangePerm {
-	return &rangePerm{begin: []byte(a), end: []byte(b)}
-}
-
-func sprintPerms(rs []*rangePerm) (s string) {
-	for _, rp := range rs {
-		s += fmt.Sprintf("[%s %s] ", rp.begin, rp.end)
+		result := checkKeyInterval(&unifiedRangePermissions{readPerms: readPerms}, tt.begin, tt.end, authpb.READ)
+		if result != tt.want {
+			t.Errorf("#%d: result=%t, want=%t", i, result, tt.want)
+		}
 	}
-	return
 }

auth/store.go

@@ -747,7 +747,7 @@ func (as *authStore) isOpPermitted(userName string, revision uint64, key, rangeE
 		return nil
 	}
 
-	if as.isRangeOpPermitted(tx, userName, key, rangeEnd, permTyp) {
+	if as.isRangeOpPermitted(tx, userName, string(key), string(rangeEnd), permTyp) {
 		return nil
 	}