Improve role cache efficiency for API key roles (elastic#58156)

This PR ensure that same roles are cached only once even when they are from different API keys.
API key role descriptors and limited role descriptors are now saved in Authentication#metadata
as raw bytes instead of deserialised Map<String, Object>.
Hashes of these bytes are used as keys for API key roles. Only when the required role is not found
in the cache, they will be deserialised to build the RoleDescriptors. The deserialisation is directly
from raw bytes to RoleDescriptors without going through the current detour of
"bytes -> Map -> bytes -> RoleDescriptors".

libs/x-content/src/main/java/org/elasticsearch/common/xcontent/AbstractObjectParser.java

@@ -210,6 +210,12 @@ public void declareLong(BiConsumer<Value, Long> consumer, ParseField field) {
         declareField(consumer, p -> p.longValue(), field, ValueType.LONG);
     }
 
+    public void declareLongOrNull(BiConsumer<Value, Long> consumer, long nullValue, ParseField field) {
+        // Using a method reference here angers some compilers
+        declareField(consumer, p -> p.currentToken() == XContentParser.Token.VALUE_NULL ? nullValue : p.longValue(),
+            field, ValueType.LONG_OR_NULL);
+    }
+
     public void declareInt(BiConsumer<Value, Integer> consumer, ParseField field) {
         // Using a method reference here angers some compilers
         declareField(consumer, p -> p.intValue(), field, ValueType.INT);

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/SecurityContext.java

@@ -10,9 +10,12 @@
 import org.elasticsearch.ElasticsearchSecurityException;
 import org.elasticsearch.Version;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.node.Node;
 import org.elasticsearch.xpack.core.security.authc.Authentication;
 import org.elasticsearch.xpack.core.security.authc.Authentication.AuthenticationType;
@@ -23,14 +26,21 @@
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.Objects;
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import static org.elasticsearch.xpack.core.security.authc.Authentication.VERSION_API_KEY_ROLES_AS_BYTES;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY;
+import static org.elasticsearch.xpack.core.security.authc.AuthenticationField.API_KEY_ROLE_DESCRIPTORS_KEY;
+
 /**
  * A lightweight utility that can find the current user and authentication information for the local thread.
  */
 public class SecurityContext {
+
     private final Logger logger = LogManager.getLogger(SecurityContext.class);
 
     private final ThreadContext threadContext;
@@ -149,8 +159,27 @@ public void executeAfterRewritingAuthentication(Consumer<StoredContext> consumer
         final Authentication authentication = getAuthentication();
         try (ThreadContext.StoredContext ignore = threadContext.stashContext()) {
             setAuthentication(new Authentication(authentication.getUser(), authentication.getAuthenticatedBy(),
-                authentication.getLookedUpBy(), version, authentication.getAuthenticationType(), authentication.getMetadata()));
+                authentication.getLookedUpBy(), version, authentication.getAuthenticationType(),
+                rewriteMetadataForApiKeyRoleDescriptors(version, authentication)));
             consumer.accept(original);
         }
     }
+
+    private Map<String, Object> rewriteMetadataForApiKeyRoleDescriptors(Version streamVersion, Authentication authentication) {
+        Map<String, Object> metadata = authentication.getMetadata();
+        if (authentication.getAuthenticationType() == AuthenticationType.API_KEY
+            && authentication.getVersion().onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)
+            && streamVersion.before(VERSION_API_KEY_ROLES_AS_BYTES)) {
+            metadata = new HashMap<>(metadata);
+            metadata.put(
+                API_KEY_ROLE_DESCRIPTORS_KEY,
+                XContentHelper.convertToMap(
+                    (BytesReference) metadata.get(API_KEY_ROLE_DESCRIPTORS_KEY), false, XContentType.JSON).v2());
+            metadata.put(
+                API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                XContentHelper.convertToMap(
+                    (BytesReference) metadata.get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY), false, XContentType.JSON).v2());
+        }
+        return metadata;
+    }
 }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/Authentication.java

@@ -27,6 +27,8 @@
 // That interface can be removed
 public class Authentication implements ToXContentObject {
 
+    public static final Version VERSION_API_KEY_ROLES_AS_BYTES = Version.V_7_9_0;
+
     private final User user;
     private final RealmRef authenticatedBy;
     private final RealmRef lookedUpBy;

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/AuthenticationField.java

@@ -8,6 +8,8 @@
 public final class AuthenticationField {
 
     public static final String AUTHENTICATION_KEY = "_xpack_security_authentication";
+    public static final String API_KEY_ROLE_DESCRIPTORS_KEY = "_security_api_key_role_descriptors";
+    public static final String API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY = "_security_api_key_limited_by_role_descriptors";
 
     private AuthenticationField() {}
 }