ISPN-5933 Make RemoteQueryEngine a cache-level component

* a ReflectionMatcher/ProtobufMatcher/CompatibilityReflectionMatcher is no longer instantiated for internal caches as they are not query-enabled
* perform query authorization only once

query/src/main/java/org/infinispan/query/dsl/embedded/impl/BaseEmbeddedQuery.java

@@ -5,8 +5,6 @@
 import org.infinispan.objectfilter.ObjectFilter;
 import org.infinispan.query.dsl.QueryFactory;
 import org.infinispan.query.dsl.impl.BaseQuery;
-import org.infinispan.security.AuthorizationManager;
-import org.infinispan.security.AuthorizationPermission;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -46,14 +44,6 @@ protected BaseEmbeddedQuery(QueryFactory queryFactory, AdvancedCache<?, ?> cache
                                String[] projection, long startOffset, int maxResults) {
       super(queryFactory, jpaQuery, namedParameters, projection, startOffset, maxResults);
       this.cache = cache;
-      ensureAccessPermissions(cache);
-   }
-
-   private void ensureAccessPermissions(AdvancedCache<?, ?> cache) {
-      AuthorizationManager authorizationManager = SecurityActions.getCacheAuthorizationManager(cache);
-      if (authorizationManager != null) {
-         authorizationManager.checkPermission(AuthorizationPermission.BULK_READ);
-      }
    }
 
    @Override

query/src/main/java/org/infinispan/query/dsl/embedded/impl/QueryEngine.java

@@ -29,6 +29,8 @@
 import org.infinispan.query.dsl.impl.BaseQuery;
 import org.infinispan.query.dsl.impl.JPAQueryGenerator;
 import org.infinispan.query.impl.ComponentRegistryUtils;
+import org.infinispan.security.AuthorizationManager;
+import org.infinispan.security.AuthorizationPermission;
 import org.infinispan.util.KeyValuePair;
 
 import java.security.PrivilegedAction;
@@ -47,6 +49,8 @@ public class QueryEngine {
 
    private static final int MAX_EXPANSION_COFACTORS = 16;
 
+   private final AuthorizationManager authorizationManager;
+
    protected final AdvancedCache<?, ?> cache;
 
    /**
@@ -73,9 +77,14 @@ public QueryEngine(AdvancedCache<?, ?> cache, SearchManager searchManager) {
       this.queryCache = ComponentRegistryUtils.getQueryCache(cache);
       this.searchManager = searchManager;
       searchFactory = searchManager != null ? searchManager.unwrap(SearchIntegrator.class) : null;
+      authorizationManager = SecurityActions.getCacheAuthorizationManager(cache);
    }
 
    public BaseQuery buildQuery(QueryFactory queryFactory, String jpqlString, Map<String, Object> namedParameters, long startOffset, int maxResults) {
+      if (authorizationManager != null) {
+         authorizationManager.checkPermission(AuthorizationPermission.BULK_READ);
+      }
+
       checkParameters(namedParameters);
 
       FilterParsingResult<?> parsingResult = parse(jpqlString, namedParameters);

query/src/main/java/org/infinispan/query/impl/LifecycleManager.java

@@ -51,6 +51,7 @@
 import org.infinispan.query.continuous.impl.ContinuousQueryResult;
 import org.infinispan.query.continuous.impl.JPAContinuousQueryCacheEventFilterConverter;
 import org.infinispan.query.spi.ProgrammaticSearchMappingProvider;
+import org.infinispan.registry.InternalCacheRegistry;
 import org.infinispan.registry.impl.ClusterRegistryImpl;
 import org.infinispan.transaction.LockingMode;
 import org.infinispan.util.logging.LogFactory;
@@ -91,21 +92,22 @@ public class LifecycleManager extends AbstractModuleLifecycle {
     */
    @Override
    public void cacheStarting(ComponentRegistry cr, Configuration cfg, String cacheName) {
-      cr.registerComponent(new ReflectionMatcher(null), ReflectionMatcher.class);
-
-      if (cfg.indexing().index().isEnabled()) {
-         log.registeringQueryInterceptor();
-         SearchIntegrator searchFactory = getSearchFactory(cfg.indexing().properties(), cr);
-         createQueryInterceptorIfNeeded(cr, cfg, searchFactory);
-         EmbeddedCacheManager cacheManager = cr.getGlobalComponentRegistry().getComponent(EmbeddedCacheManager.class);
-         addCacheDependencyIfNeeded(cacheName, cacheManager, cfg.indexing().properties());
+      InternalCacheRegistry icr = cr.getGlobalComponentRegistry().getComponent(InternalCacheRegistry.class);
+      if (!icr.isInternalCache(cacheName)) {
+         cr.registerComponent(new ReflectionMatcher(null), ReflectionMatcher.class);
+
+         if (cfg.indexing().index().isEnabled()) {
+            log.registeringQueryInterceptor();
+            SearchIntegrator searchFactory = getSearchFactory(cfg.indexing().properties(), cr);
+            createQueryInterceptorIfNeeded(cr, cfg, searchFactory);
+            EmbeddedCacheManager cacheManager = cr.getGlobalComponentRegistry().getComponent(EmbeddedCacheManager.class);
+            addCacheDependencyIfNeeded(cacheName, cacheManager, cfg.indexing().properties());
+         }
       }
    }
 
    private void addCacheDependencyIfNeeded(String cacheStarting, EmbeddedCacheManager cacheManager, Properties properties) {
-      if (!ClusterRegistryImpl.GLOBAL_REGISTRY_CACHE_NAME.equals(cacheStarting)) {
-         cacheManager.addCacheDependency(cacheStarting, ClusterRegistryImpl.GLOBAL_REGISTRY_CACHE_NAME);
-      }
+      cacheManager.addCacheDependency(cacheStarting, ClusterRegistryImpl.GLOBAL_REGISTRY_CACHE_NAME);
       if (hasInfinispanDirectory(properties) && !DEFAULT_CACHES.contains(cacheStarting)) {
          String metadataCacheName = getMetadataCacheName(properties);
          String lockingCacheName = getLockingCacheName(properties);
@@ -187,11 +189,9 @@ public void cacheStarted(ComponentRegistry cr, String cacheName) {
       registerQueryMBeans(cache.getAdvancedCache(), cr, cacheName);
    }
 
-   private void registerQueryMBeans(AdvancedCache cache,
-         ComponentRegistry cr, String cacheName) {
+   private void registerQueryMBeans(AdvancedCache cache, ComponentRegistry cr, String cacheName) {
       Configuration cfg = cache.getCacheConfiguration();
-      SearchIntegrator sf = getSearchFactory(
-            cfg.indexing().properties(), cr);
+      SearchIntegrator sf = getSearchFactory(cfg.indexing().properties(), cr);
 
       // Resolve MBean server instance
       GlobalConfiguration globalCfg =
@@ -221,8 +221,8 @@ private void registerQueryMBeans(AdvancedCache cache,
             .toManageableComponentMetadata();
       try {
          // TODO: MassIndexer should be some kind of query cache component?
-         DistributedExecutorMassIndexer maxIndexer = new DistributedExecutorMassIndexer(cache, sf);
-         ResourceDMBean mbean = new ResourceDMBean(maxIndexer, massIndexerCompMetadata);
+         DistributedExecutorMassIndexer massIndexer = new DistributedExecutorMassIndexer(cache, sf);
+         ResourceDMBean mbean = new ResourceDMBean(massIndexer, massIndexerCompMetadata);
          ObjectName massIndexerObjName = new ObjectName(jmxDomain + ":"
                + queryGroupName + ",component=" + massIndexerCompMetadata.getJmxObjectName());
          JmxUtil.registerMBean(mbean, massIndexerObjName, mbeanServer);
@@ -237,10 +237,7 @@ private String getQueryGroupName(String cacheManagerName, String cacheName) {
 
    private boolean verifyChainContainsQueryInterceptor(ComponentRegistry cr) {
       InterceptorChain interceptorChain = cr.getComponent(InterceptorChain.class);
-      if (interceptorChain == null) {
-         return false;
-      }
-      return interceptorChain.containsInterceptorType(QueryInterceptor.class, true);
+      return interceptorChain != null && interceptorChain.containsInterceptorType(QueryInterceptor.class, true);
    }
 
    private SearchIntegrator getSearchFactory(Properties indexingProperties, ComponentRegistry cr) {
@@ -250,7 +247,7 @@ private SearchIntegrator getSearchFactory(Properties indexingProperties, Compone
          searchFactory = (SearchIntegrator) component;
       }
       //defend against multiple initialization:
-      if (searchFactory==null) {
+      if (searchFactory == null) {
          GlobalComponentRegistry globalComponentRegistry = cr.getGlobalComponentRegistry();
          EmbeddedCacheManager uninitializedCacheManager = globalComponentRegistry.getComponent(EmbeddedCacheManager.class);
          indexingProperties = addProgrammaticMappings(indexingProperties, cr);
@@ -332,7 +329,7 @@ public void cacheManagerStarting(GlobalComponentRegistry gcr, GlobalConfiguratio
       QueryCache queryCache = new QueryCache();
       gcr.registerComponent(queryCache, QueryCache.class);
 
-      Map<Integer,AdvancedExternalizer<?>> externalizerMap = globalCfg.serialization().advancedExternalizers();
+      Map<Integer, AdvancedExternalizer<?>> externalizerMap = globalCfg.serialization().advancedExternalizers();
       externalizerMap.put(ExternalizerIds.JPA_FILTER_AND_CONVERTER, new JPAFilterAndConverter.JPAFilterAndConverterExternalizer());
       externalizerMap.put(ExternalizerIds.JPA_FILTER_RESULT, new JPAFilterAndConverter.FilterResultExternalizer());
       externalizerMap.put(ExternalizerIds.JPA_CACHE_EVENT_FILTER_CONVERTER, new JPACacheEventFilterConverter.Externalizer());

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/LifecycleManager.java

@@ -1,5 +1,7 @@
 package org.infinispan.query.remote.impl;
 
+import org.infinispan.AdvancedCache;
+import org.infinispan.Cache;
 import org.infinispan.commons.CacheException;
 import org.infinispan.commons.marshall.AdvancedExternalizer;
 import org.infinispan.configuration.cache.Configuration;
@@ -8,6 +10,7 @@
 import org.infinispan.configuration.cache.InterceptorConfiguration;
 import org.infinispan.configuration.cache.InterceptorConfigurationBuilder;
 import org.infinispan.configuration.global.GlobalConfiguration;
+import org.infinispan.context.Flag;
 import org.infinispan.factories.ComponentRegistry;
 import org.infinispan.factories.GlobalComponentRegistry;
 import org.infinispan.factories.components.ComponentMetadataRepo;
@@ -21,6 +24,9 @@
 import org.infinispan.manager.DefaultCacheManager;
 import org.infinispan.manager.EmbeddedCacheManager;
 import org.infinispan.objectfilter.impl.ProtobufMatcher;
+import org.infinispan.protostream.SerializationContext;
+import org.infinispan.query.Search;
+import org.infinispan.query.SearchManager;
 import org.infinispan.query.remote.ProtobufMetadataManager;
 import org.infinispan.query.remote.impl.filter.JPABinaryProtobufFilterAndConverter;
 import org.infinispan.query.remote.impl.filter.JPAContinuousQueryProtobufCacheEventFilterConverter;
@@ -29,6 +35,7 @@
 import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapper;
 import org.infinispan.query.remote.impl.indexing.RemoteValueWrapperInterceptor;
 import org.infinispan.query.remote.impl.logging.Log;
+import org.infinispan.registry.InternalCacheRegistry;
 import org.infinispan.util.logging.LogFactory;
 import org.kohsuke.MetaInfServices;
 
@@ -106,7 +113,8 @@ private void unregisterProtobufMetadataManagerMBean(GlobalComponentRegistry gcr)
     */
    @Override
    public void cacheStarting(ComponentRegistry cr, Configuration cfg, String cacheName) {
-      if (!cacheName.equals(ProtobufMetadataManager.PROTOBUF_METADATA_CACHE_NAME)) {
+      InternalCacheRegistry icr = cr.getGlobalComponentRegistry().getComponent(InternalCacheRegistry.class);
+      if (!icr.isInternalCache(cacheName)) {
          ProtobufMetadataManagerImpl protobufMetadataManager = (ProtobufMetadataManagerImpl) cr.getGlobalComponentRegistry().getComponent(ProtobufMetadataManager.class);
 
          // ensure the protobuf metadata cache is created
@@ -147,7 +155,6 @@ private void createRemoteValueWrapperInterceptor(ComponentRegistry cr, Configura
          }
          if (ic != null) {
             cr.registerComponent(wrapperInterceptor, RemoteValueWrapperInterceptor.class);
-            cr.registerComponent(wrapperInterceptor, wrapperInterceptor.getClass().getName(), true);
          }
          cfg.customInterceptors().interceptors(builder.build().customInterceptors().interceptors());
       }
@@ -156,24 +163,30 @@ private void createRemoteValueWrapperInterceptor(ComponentRegistry cr, Configura
    @Override
    public void cacheStarted(ComponentRegistry cr, String cacheName) {
       Configuration configuration = cr.getComponent(Configuration.class);
-      boolean remoteValueWrappingEnabled = configuration.indexing().index().isEnabled() && !configuration.compatibility().enabled();
-      if (!remoteValueWrappingEnabled) {
-         if (verifyChainContainsRemoteValueWrapperInterceptor(cr)) {
-            throw new IllegalStateException("It was NOT expected to find the RemoteValueWrapperInterceptor registered in the InterceptorChain as indexing was disabled, but it was found");
+      boolean isIndexed = configuration.indexing().index().isEnabled();
+      boolean isCompatMode = configuration.compatibility().enabled();
+      boolean remoteValueWrappingEnabled = isIndexed && !isCompatMode;
+      if (remoteValueWrappingEnabled) {
+         if (!verifyChainContainsRemoteValueWrapperInterceptor(cr)) {
+            throw new IllegalStateException("It was expected to find the RemoteValueWrapperInterceptor registered in the InterceptorChain but it wasn't found");
          }
-         return;
+      } else if (verifyChainContainsRemoteValueWrapperInterceptor(cr)) {
+         throw new IllegalStateException("It was NOT expected to find the RemoteValueWrapperInterceptor registered in the InterceptorChain as indexing was disabled, but it was found");
       }
-      if (!verifyChainContainsRemoteValueWrapperInterceptor(cr)) {
-         throw new IllegalStateException("It was expected to find the RemoteValueWrapperInterceptor registered in the InterceptorChain but it wasn't found");
+
+      InternalCacheRegistry icr = cr.getGlobalComponentRegistry().getComponent(InternalCacheRegistry.class);
+      if (!icr.isInternalCache(cacheName)) {
+         AdvancedCache<?, ?> cache = cr.getComponent(Cache.class).getAdvancedCache().withFlags(Flag.OPERATION_HOTROD);
+         SerializationContext serCtx = ProtobufMetadataManagerImpl.getSerializationContextInternal(cache.getCacheManager());
+         SearchManager searchManager = isIndexed ? Search.getSearchManager(cache) : null;
+         RemoteQueryEngine remoteQueryEngine = new RemoteQueryEngine(cache, searchManager, isCompatMode, serCtx);
+         cr.registerComponent(remoteQueryEngine, RemoteQueryEngine.class);
       }
    }
 
    private boolean verifyChainContainsRemoteValueWrapperInterceptor(ComponentRegistry cr) {
       InterceptorChain interceptorChain = cr.getComponent(InterceptorChain.class);
-      if (interceptorChain == null) {
-         return false;
-      }
-      return interceptorChain.containsInterceptorType(RemoteValueWrapperInterceptor.class, true);
+      return interceptorChain != null && interceptorChain.containsInterceptorType(RemoteValueWrapperInterceptor.class, true);
    }
 
    @Override

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/QueryFacadeImpl.java

@@ -4,12 +4,9 @@
 import org.hibernate.search.engine.impl.nullencoding.NullMarkerCodec;
 import org.infinispan.AdvancedCache;
 import org.infinispan.commons.logging.LogFactory;
-import org.infinispan.configuration.cache.Configuration;
 import org.infinispan.protostream.ProtobufUtil;
 import org.infinispan.protostream.SerializationContext;
 import org.infinispan.protostream.WrappedMessage;
-import org.infinispan.query.Search;
-import org.infinispan.query.SearchManager;
 import org.infinispan.query.dsl.impl.BaseQuery;
 import org.infinispan.query.remote.client.QueryRequest;
 import org.infinispan.query.remote.client.QueryResponse;
@@ -49,19 +46,19 @@ public final class QueryFacadeImpl implements QueryFacade {
 
    @Override
    public byte[] query(AdvancedCache<byte[], byte[]> cache, byte[] query) {
+      RemoteQueryEngine queryEngine = SecurityActions.getCacheComponentRegistry(cache).getComponent(RemoteQueryEngine.class);
+      if (queryEngine == null) {
+         throw log.queryingNotEnabled(cache.getName());
+      }
+
+      SerializationContext serCtx = ProtobufMetadataManagerImpl.getSerializationContextInternal(cache.getCacheManager());
       try {
-         SerializationContext serCtx = ProtobufMetadataManagerImpl.getSerializationContextInternal(cache.getCacheManager());
          QueryRequest request = ProtobufUtil.fromByteArray(serCtx, query, 0, query.length, QueryRequest.class);
 
-         Configuration cacheConfiguration = SecurityActions.getCacheConfiguration(cache);
-         boolean isIndexed = cacheConfiguration.indexing().index().isEnabled();
-         boolean isCompatMode = cacheConfiguration.compatibility().enabled();
-         SearchManager searchManager = isIndexed ? Search.getSearchManager(cache) : null;  // this also checks access permissions
-         RemoteQueryEngine queryEngine = new RemoteQueryEngine(cache, searchManager, isCompatMode, serCtx);
-
          long startOffset = request.getStartOffset() == null ? -1 : request.getStartOffset();
          int maxResults = request.getMaxResults() == null ? -1 : request.getMaxResults();
          Map<String, Object> namedParameters = getNamedParameters(request);
+
          BaseQuery q = queryEngine.buildQuery(null, request.getJpqlString(), namedParameters, startOffset, maxResults);
 
          QueryResponse response = makeResponse(q);

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/SecurityActions.java

@@ -1,11 +1,9 @@
 package org.infinispan.query.remote.impl;
 
 import org.infinispan.AdvancedCache;
-import org.infinispan.configuration.cache.Configuration;
 import org.infinispan.factories.ComponentRegistry;
 import org.infinispan.security.Security;
 import org.infinispan.security.actions.GetCacheComponentRegistryAction;
-import org.infinispan.security.actions.GetCacheConfigurationAction;
 
 import java.security.AccessController;
 import java.security.PrivilegedAction;
@@ -24,10 +22,6 @@ static <T> T doPrivileged(PrivilegedAction<T> action) {
             AccessController.doPrivileged(action) : Security.doPrivileged(action);
    }
 
-   static Configuration getCacheConfiguration(AdvancedCache<?, ?> cache) {
-      return doPrivileged(new GetCacheConfigurationAction(cache));
-   }
-
    static ComponentRegistry getCacheComponentRegistry(AdvancedCache<?, ?> cache) {
       return doPrivileged(new GetCacheComponentRegistryAction(cache));
    }

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/logging/Log.java

@@ -23,4 +23,7 @@ public interface Log extends org.infinispan.util.logging.Log {
 
    @Message(value = "An exception has occurred during query execution", id = 18003)
    CacheException errorExecutingQuery(@Cause Throwable cause);
+
+   @Message(value = "Querying is not enabled on cache %s", id = 18004)
+   CacheException queryingNotEnabled(String cacheName);
 }

server/hotrod/src/main/scala/org/infinispan/server/hotrod/Decoder10.scala

@@ -5,7 +5,6 @@ import org.infinispan.configuration.cache.Configuration
 import org.infinispan.server.core.Operation._
 import HotRodOperation._
 import OperationStatus._
-import org.infinispan.stats.Stats
 import org.infinispan.server.core._
 import collection.mutable
 import collection.immutable
@@ -211,7 +210,7 @@ object Decoder10 extends AbstractVersionedDecoder with ServerConstants with Log
             getKeyMetadata(h, k, cache)
          case QueryRequest =>
             val query = readRangedBytes(buffer)
-            val result = server.getQueryFacades.head.query(cache, query)
+            val result = server.query(cache, query)
             new QueryResponse(h.version, h.messageId, h.cacheName, h.clientIntel,
                h.topologyId, result)
       }

server/hotrod/src/main/scala/org/infinispan/server/hotrod/Decoder2x.scala

@@ -367,7 +367,7 @@ object Decoder2x extends AbstractVersionedDecoder with ServerConstants with Log
             getKeyMetadata(h, k, cache)
          case QueryRequest =>
             val query = readRangedBytes(buffer)
-            val result = server.getQueryFacades.head.query(cache, query)
+            val result = server.query(cache, query)
             new QueryResponse(h.version, h.messageId, h.cacheName, h.clientIntel,
                h.topologyId, result)
          case AddClientListenerRequest =>

server/hotrod/src/main/scala/org/infinispan/server/hotrod/HotRodServer.scala

@@ -1,6 +1,7 @@
 package org.infinispan.server.hotrod
 
 import logging.Log
+import org.infinispan.AdvancedCache
 import org.infinispan.commons.marshall.Marshaller
 import org.infinispan.filter.{ParamKeyValueFilterConverterFactory, KeyValueFilterConverterFactory}
 import org.infinispan.notifications.Listener
@@ -62,7 +63,9 @@ class HotRodServer extends AbstractProtocolServer("HotRod") with Log {
 
    def getMarshaller = marshaller
 
-   def getQueryFacades: Seq[QueryFacade] = queryFacades
+   def query(cache: AdvancedCache[Array[Byte], Array[Byte]], query: Array[Byte]): Array[Byte] = {
+      queryFacades.head.query(cache, query)
+   }
 
    def getClientListenerRegistry: ClientListenerRegistry = clientListenerRegistry