Skip to content

KAFKA-10074: Improve performance of matchingAcls #8769

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ijuma merged 2 commits into from
Jun 1, 2020
+45 −28
Merged

KAFKA-10074: Improve performance of matchingAcls #8769

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bae2ca7
KAFKA-10074: Improve performance of `matchingAcls`
ijuma Jun 1, 2020
100a645
Add clarifying comment.
ijuma Jun 1, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 18 additions & 7 deletions core/src/main/scala/kafka/security/authorizer/AclAuthorizer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ import org.apache.kafka.server.authorizer._
import org.apache.zookeeper.client.ZKClientConfig

import scala.annotation.nowarn
import scala.collection.mutable.ArrayBuffer
import scala.collection.{Seq, mutable}
import scala.jdk.CollectionConverters._
import scala.util.{Failure, Random, Success, Try}
Expand All @@ -63,9 +64,15 @@ object AclAuthorizer {
def exists: Boolean = zkVersion != ZkVersion.UnknownVersion
}

class AclSeqs(classes: Seq[AclEntry]*) {
def find(p: AclEntry => Boolean): Option[AclEntry] = classes.flatMap(_.find(p)).headOption
def isEmpty: Boolean = !classes.exists(_.nonEmpty)
class AclSeqs(seqs: Seq[AclEntry]*) {
def find(p: AclEntry => Boolean): Option[AclEntry] = {

@chia7712 chia7712 Jun 1, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should it need comment to remind reader that this style is for optimization.

@ijuma ijuma Jun 1, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this is obvious, no? find should generally short-circuit and not go through all the items. That's how it works for all collection implementations

I think this kind of comment makes sense in matchingAcls where I added one.

@ijuma ijuma Jun 1, 2020 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@chia7712 I looked at the code again and I guess the intent may not be clear. I added a comment that hopefully clarifies.

// Lazily iterate through the inner `Seq` elements and stop as soon as we find a match
val it = seqs.iterator.flatMap(_.find(p))
if (it.hasNext) Some(it.next)
else None
}

def isEmpty: Boolean = !seqs.exists(_.nonEmpty)
}

val NoAcls = VersionedAcls(Set.empty, ZkVersion.UnknownVersion)
Expand Down Expand Up @@ -344,8 +351,10 @@ class AclAuthorizer extends Authorizer with Logging {
} else false
}

@nowarn("cat=deprecation")
@nowarn("cat=deprecation&cat=optimizer")
private def matchingAcls(resourceType: ResourceType, resourceName: String): AclSeqs = {
// this code is performance sensitive, make sure to run AclAuthorizerBenchmark after any changes

// save aclCache reference to a local val to get a consistent view of the cache during acl updates.
val aclCacheSnapshot = aclCache
val wildcard = aclCacheSnapshot.get(new ResourcePattern(resourceType, ResourcePattern.WILDCARD_RESOURCE, PatternType.LITERAL))
Expand All @@ -356,11 +365,13 @@ class AclAuthorizer extends Authorizer with Logging {
.map(_.acls.toBuffer)
.getOrElse(mutable.Buffer.empty)

val prefixed = aclCacheSnapshot
val prefixed = new ArrayBuffer[AclEntry]
aclCacheSnapshot
.from(new ResourcePattern(resourceType, resourceName, PatternType.PREFIXED))
.to(new ResourcePattern(resourceType, resourceName.take(1), PatternType.PREFIXED))
.flatMap { case (resource, acls) => if (resourceName.startsWith(resource.name)) acls.acls else Seq.empty }
.toBuffer
.foreach { case (resource, acls) =>
if (resourceName.startsWith(resource.name)) prefixed ++= acls.acls
}

new AclSeqs(prefixed, wildcard, literal)
}
Expand Down
48 changes: 27 additions & 21 deletions jmh-benchmarks/src/main/java/org/apache/kafka/jmh/acl/AclAuthorizerBenchmark.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,30 +68,35 @@
@Measurement(iterations = 15)
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)

public class AclAuthorizerBenchmark {
@Param({"5000", "10000", "50000"})
@Param({"10000", "50000", "200000"})
private int resourceCount;
//no. of. rules per resource
@Param({"5", "10", "15"})
@Param({"10", "50"})
private int aclCount;

private int hostPreCount = 1000;
private final int hostPreCount = 1000;
private final String resourceNamePrefix = "foo-bar35_resource-";

private AclAuthorizer aclAuthorizer = new AclAuthorizer();
private KafkaPrincipal principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "test-user");
private final AclAuthorizer aclAuthorizer = new AclAuthorizer();
private final KafkaPrincipal principal = new KafkaPrincipal(KafkaPrincipal.USER_TYPE, "test-user");
private List<Action> actions = new ArrayList<>();
private RequestContext context;

@Setup(Level.Trial)
public void setup() throws Exception {
setFieldValue(aclAuthorizer, AclAuthorizer.class.getDeclaredField("aclCache").getName(),
prepareAclCache());
actions = Collections.singletonList(new Action(AclOperation.WRITE, new ResourcePattern(ResourceType.TOPIC, "resource-1", PatternType.LITERAL),
1, true, true));
context = new RequestContext(new RequestHeader(ApiKeys.PRODUCE, Integer.valueOf(1).shortValue(), "someclient", 1),
"1", InetAddress.getLocalHost(), KafkaPrincipal.ANONYMOUS,
ListenerName.normalised("listener"), SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY);
// By adding `-95` to the resource name prefix, we cause the `TreeMap.from/to` call to return
// most map entries. In such cases, we rely on the filtering based on `String.startsWith`
// to return the matching ACLs. Using a more efficient data structure (e.g. a prefix
// tree) should improve performance significantly).
actions = Collections.singletonList(new Action(AclOperation.WRITE,
new ResourcePattern(ResourceType.TOPIC, resourceNamePrefix + 95, PatternType.LITERAL),
1, true, true));
context = new RequestContext(new RequestHeader(ApiKeys.PRODUCE, Integer.valueOf(1).shortValue(),
"someclient", 1), "1", InetAddress.getLocalHost(), KafkaPrincipal.ANONYMOUS,
ListenerName.normalised("listener"), SecurityProtocol.PLAINTEXT, ClientInformation.EMPTY);
}

private void setFieldValue(Object obj, String fieldName, Object value) throws Exception {
Expand All @@ -103,10 +108,9 @@ private void setFieldValue(Object obj, String fieldName, Object value) throws Ex
private TreeMap<ResourcePattern, VersionedAcls> prepareAclCache() {
Map<ResourcePattern, Set<AclEntry>> aclEntries = new HashMap<>();
for (int resourceId = 0; resourceId < resourceCount; resourceId++) {

ResourcePattern resource = new ResourcePattern(
(resourceId % 10 == 0) ? ResourceType.GROUP : ResourceType.TOPIC,
"resource-" + resourceId,
resourceNamePrefix + resourceId,
(resourceId % 5 == 0) ? PatternType.PREFIXED : PatternType.LITERAL);

Set<AclEntry> entries = aclEntries.computeIfAbsent(resource, k -> new HashSet<>());
Expand All @@ -118,20 +122,22 @@ private TreeMap<ResourcePattern, VersionedAcls> prepareAclCache() {
}
}

ResourcePattern resourceWildcard = new ResourcePattern(ResourceType.TOPIC, ResourcePattern.WILDCARD_RESOURCE, PatternType.LITERAL);
ResourcePattern resourcePrefix = new ResourcePattern(ResourceType.TOPIC, "resource-", PatternType.PREFIXED);

Set<AclEntry> entriesWildcard = aclEntries.computeIfAbsent(resourceWildcard, k -> new HashSet<>());
ResourcePattern resourcePrefix = new ResourcePattern(ResourceType.TOPIC, resourceNamePrefix,
PatternType.PREFIXED);
Set<AclEntry> entriesPrefix = aclEntries.computeIfAbsent(resourcePrefix, k -> new HashSet<>());

for (int hostId = 0; hostId < hostPreCount; hostId++) {
AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId, AclOperation.READ, AclPermissionType.ALLOW);
AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId,
AclOperation.READ, AclPermissionType.ALLOW);
entriesPrefix.add(new AclEntry(ace));
}

ResourcePattern resourceWildcard = new ResourcePattern(ResourceType.TOPIC, ResourcePattern.WILDCARD_RESOURCE,
PatternType.LITERAL);
Set<AclEntry> entriesWildcard = aclEntries.computeIfAbsent(resourceWildcard, k -> new HashSet<>());
// get dynamic entries number for wildcard acl
for (int hostId = 0; hostId < resourceCount / 10; hostId++) {
AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId, AclOperation.READ, AclPermissionType.ALLOW);
AccessControlEntry ace = new AccessControlEntry(principal.toString(), "127.0.0." + hostId,
AclOperation.READ, AclPermissionType.ALLOW);
entriesWildcard.add(new AclEntry(ace));
}

Expand All @@ -155,7 +161,7 @@ public void testAclsIterator() {
}

@Benchmark
public void testAuthorizer() throws Exception {
public void testAuthorizer() {
aclAuthorizer.authorize(context, actions);
}
}