Rewrite store-query methods that use stream() and intercept/wrap store-query for metrics (#1461)

* Rewrite store-query methods that use stream() and intercept/wrap store-query for metrics

We cannot use the @measure annotation in ArtifactStoreQuery, since that class is instantiated a variable number of
calls for each user request, and instances hold state supporting the fluent API. The StoreDataManager decorator
wraps ArtifactStoreQuery results from the query() method in a query class that measures query calls, delegating the
actual query to the instance that comes from the underlying StoreDataManager delegate.

StoreDataManager stream / getAll methods rely on having in-memory access to ALL of the ArtifactStore instances,
and also on having a small enough instance count to make iterating through all of them feasible. In our production
environment, both of these assumptions are violated. We need a better way to access this information, but to start
I've refactored DefaultArtifactStoreQuery methods that use streams and getAll in the StoreDataManager so they won't
do that.

In the case of retrieving the stores that are in the membership tree of a particular group, I've rewritten that to
avoid recursion.

In addition, I've added a keyStream() method to ArtifactStoreQuery and a streamArtifactStoreKeys() method to
StoreDataManager. This SHOULD be something that we can optimize, ensuring all keys are available in-memory or easy
to deserialize with minimum overhead. We can do a lot of filtering based on the key alone, then additional
filtering can be applied with the matching stores, which would be retrieved directly by key from the filtered
keys in these calls.

* remove duplicate dependencies in 2 poms

* Separate metered-db stuff to break dependency cycle, and fix a few compilation / unit test problems

Author: jdcasey

addons/httprox/common/pom.xml

@@ -80,10 +80,6 @@
       <artifactId>indy-filer-default</artifactId>
       <scope>test</scope>
     </dependency>
-    <dependency>
-      <groupId>org.commonjava.indy</groupId>
-      <artifactId>indy-subsys-jaxrs</artifactId>
-    </dependency>
     <dependency>
       <groupId>org.commonjava.indy</groupId>
       <artifactId>indy-folo-common</artifactId>

api/src/main/java/org/commonjava/indy/data/ArtifactStoreQuery.java

@@ -67,6 +67,12 @@ Stream<T> stream()
     Stream<T> stream( Predicate<ArtifactStore> filter )
             throws IndyDataException;
 
+    Stream<StoreKey> keyStream()
+            throws IndyDataException;
+
+    Stream<StoreKey> keyStream( Predicate<StoreKey> filterPredicate )
+            throws IndyDataException;
+
     List<T> getAll( Predicate<ArtifactStore> filter )
             throws IndyDataException;
 

api/src/main/java/org/commonjava/indy/data/DelegatingArtifactStoreQuery.java

@@ -120,6 +120,20 @@ public Stream<T> stream( final Predicate<ArtifactStore> filter )
         return delegate.stream( filter );
     }
 
+    @Override
+    public Stream<StoreKey> keyStream()
+            throws IndyDataException
+    {
+        return delegate.keyStream();
+    }
+
+    @Override
+    public Stream<StoreKey> keyStream( final Predicate<StoreKey> filterPredicate )
+            throws IndyDataException
+    {
+        return null;
+    }
+
     @Override
     public List<T> getAll( final Predicate<ArtifactStore> filter )
             throws IndyDataException

api/src/main/java/org/commonjava/indy/data/StoreDataManager.java

@@ -129,4 +129,9 @@ void reload()
     boolean isReadonly( ArtifactStore store );
 
     boolean isEmpty();
+
+    /**
+     * Stream of StoreKey instances present in the system.
+     */
+    Stream<StoreKey> streamArtifactStoreKeys();
 }

db/common/src/main/java/org/commonjava/indy/db/common/AbstractStoreDataManager.java

@@ -246,17 +246,20 @@ public abstract void clear( final ChangeSummary summary )
             throws IndyDataException;
 
     @Override
+    @Measure
     public abstract Set<ArtifactStore> getAllArtifactStores()
             throws IndyDataException;
 
     @Override
+    @Measure
     public Stream<ArtifactStore> streamArtifactStores()
             throws IndyDataException
     {
         return getAllArtifactStores().stream();
     }
 
     @Override
+    @Measure
     public abstract Map<StoreKey, ArtifactStore> getArtifactStoresByKey();
 
     @Override

db/common/src/main/java/org/commonjava/indy/db/common/DefaultArtifactStoreQuery.java

@@ -36,11 +36,11 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.HashSet;
+import java.util.LinkedList;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -162,13 +162,15 @@ public List<T> getAll()
     }
 
     @Override
+    @Measure
     public Stream<T> stream()
             throws IndyDataException
     {
         return stream( store -> true );
     }
 
     @Override
+    @Measure
     public Stream<T> stream( Predicate<ArtifactStore> filter )
             throws IndyDataException
     {
@@ -219,6 +221,7 @@ public List<T> getAll( Predicate<ArtifactStore> filter )
     }
 
     @Override
+    @Measure
     public List<T> getAllByDefaultPackageTypes()
             throws IndyDataException
     {
@@ -233,6 +236,7 @@ public List<T> getAllByDefaultPackageTypes()
     }
 
     @Override
+    @Measure
     public T getByName( String name )
             throws IndyDataException
     {
@@ -251,9 +255,7 @@ public boolean containsByName( String name )
     public Set<Group> getGroupsContaining( StoreKey storeKey )
             throws IndyDataException
     {
-        return new DefaultArtifactStoreQuery<>( dataManager, storeKey.getPackageType(), enabled,
-                                                Group.class ).stream(
-                store -> ( (Group) store ).getConstituents().contains( storeKey ) ).collect( Collectors.toSet() );
+        return getAllGroups().stream().filter( g -> g.getConstituents().contains( storeKey ) ).collect( Collectors.toSet() );
     }
 
     @Override
@@ -374,21 +376,15 @@ public List<RemoteRepository> getRemoteRepositoryByUrl( String url )
     public List<ArtifactStore> getOrderedConcreteStoresInGroup( final String groupName )
             throws IndyDataException
     {
-        Map<StoreKey, ArtifactStore> stores = new HashMap<>();
-        stream().forEach( s -> stores.put( s.getKey(), s ) );
-
-        return getGroupOrdering( groupName, stores, false, true );
+        return getGroupOrdering( groupName, false, true );
     }
 
     @Override
     @Measure
     public List<ArtifactStore> getOrderedStoresInGroup( final String groupName )
             throws IndyDataException
     {
-        Map<StoreKey, ArtifactStore> stores = new HashMap<>();
-        stream().forEach( s -> stores.put( s.getKey(), s ) );
-
-        return getGroupOrdering( groupName, stores, true, false );
+        return getGroupOrdering( groupName, true, false );
     }
 
     @Override
@@ -417,8 +413,8 @@ public Set<Group> getGroupsAffectedBy( Collection<StoreKey> keys )
 
         Set<StoreKey> processed = new HashSet<>();
 
-        Set<Group> all = new DefaultArtifactStoreQuery<>( dataManager, toProcess.get( 0 ).getPackageType(), null,
-                                                          Group.class ).stream().collect( Collectors.toSet() );
+        Set<StoreKey> all = new DefaultArtifactStoreQuery<>( dataManager, toProcess.get( 0 ).getPackageType(), null,
+                                                          Group.class ).keyStream().collect( Collectors.toSet() );
 
         while ( !toProcess.isEmpty() )
         {
@@ -433,8 +429,10 @@ public Set<Group> getGroupsAffectedBy( Collection<StoreKey> keys )
             // use this to avoid reprocessing groups we've already encountered.
             processed.add( next );
 
-            for ( ArtifactStore store : all )
+            for ( StoreKey key : all )
             {
+                ArtifactStore store = dataManager.getArtifactStore( key );
+
                 if ( !processed.contains( store.getKey() ) && ( store instanceof Group ) )
                 {
                     Group g = (Group) store;
@@ -452,27 +450,76 @@ public Set<Group> getGroupsAffectedBy( Collection<StoreKey> keys )
         return groups;
     }
 
+    public Stream<StoreKey> keyStream()
+    {
+        return keyStream( null );
+    }
+
+    public Stream<StoreKey> keyStream( Predicate<StoreKey> filterPredicate )
+    {
+        return dataManager.streamArtifactStoreKeys().filter(key -> {
+            if ( packageType != null && !key.getPackageType().equals( packageType ))
+            {
+                return false;
+            }
+
+            if ( types != null && !types.isEmpty() && !types.contains( key.getType() ) )
+            {
+                return false;
+            }
+
+            if ( filterPredicate != null && !filterPredicate.test( key ) )
+            {
+                return false;
+            }
+
+            return true;
+        });
+    }
+
     @Override
+    @Measure
     public List<RemoteRepository> getAllRemoteRepositories()
             throws IndyDataException
     {
-        return new DefaultArtifactStoreQuery<>( dataManager, packageType, enabled,
-                                                RemoteRepository.class ).getAll();
+        return getAllOfType( RemoteRepository.class );
     }
 
     @Override
+    @Measure
     public List<HostedRepository> getAllHostedRepositories()
             throws IndyDataException
     {
-        return new DefaultArtifactStoreQuery<>( dataManager, packageType, enabled,
-                                                HostedRepository.class ).getAll();
+        return getAllOfType( HostedRepository.class );
     }
 
     @Override
+    @Measure
     public List<Group> getAllGroups()
             throws IndyDataException
     {
-        return new DefaultArtifactStoreQuery<>( dataManager, packageType, enabled, Group.class ).getAll();
+        return getAllOfType( Group.class );
+    }
+
+    private <T extends ArtifactStore> List<T> getAllOfType(Class<T> type)
+            throws IndyDataException
+    {
+        List<StoreKey> keys = new DefaultArtifactStoreQuery<>( dataManager, packageType, enabled, type ).keyStream()
+                                                                                                        .collect(
+                                                                                                                Collectors
+                                                                                                                        .toList() );
+
+        List<T> stores = new ArrayList<>();
+        for ( StoreKey key : keys )
+        {
+            T store = (T) dataManager.getArtifactStore( key );
+            if ( store != null && ( enabled == null || store.isDisabled() == !enabled ) )
+            {
+                stores.add( store );
+            }
+        }
+
+        return stores;
     }
 
     @Override
@@ -503,7 +550,7 @@ public DefaultArtifactStoreQuery<T> noPackageType()
         return this;
     }
 
-    private List<ArtifactStore> getGroupOrdering( final String groupName, final Map<StoreKey, ArtifactStore> stores,
+    private List<ArtifactStore> getGroupOrdering( final String groupName,
                                                   final boolean includeGroups, final boolean recurseGroups )
             throws IndyDataException
     {
@@ -512,55 +559,79 @@ private List<ArtifactStore> getGroupOrdering( final String groupName, final Map<
             throw new IndyDataException( "packageType must be set on the query before calling this method!" );
         }
 
-        final Group master = (Group) stores.get( new StoreKey( packageType, StoreType.group, groupName ) );
+        final Group master = (Group) dataManager.getArtifactStore( new StoreKey( packageType, StoreType.group, groupName ) );
         if ( master == null )
         {
             return Collections.emptyList();
         }
 
         final List<ArtifactStore> result = new ArrayList<>();
-        recurseGroup( master, stores, result, new HashSet<>(), includeGroups, recurseGroups );
+        recurseGroup( master, result, new HashSet<>(), includeGroups, recurseGroups );
 
         return result;
     }
 
-    private void recurseGroup( final Group master, final Map<StoreKey, ArtifactStore> stores,
+    private void recurseGroup( final Group master,
                                final List<ArtifactStore> result, final Set<StoreKey> seen, final boolean includeGroups,
                                final boolean recurseGroups )
+            throws IndyDataException
     {
-        if ( master == null || master.isDisabled() && Boolean.TRUE.equals( enabled ) )
-        {
-            return;
-        }
+        AtomicReference<IndyDataException> errorRef = new AtomicReference<>();
+        LinkedList<Group> toCheck = new LinkedList<>();
+        toCheck.add( master );
 
-        List<StoreKey> members = new ArrayList<>( master.getConstituents() );
-        if ( includeGroups )
+        while ( !toCheck.isEmpty() )
         {
-            result.add( master );
-        }
+            Group next = toCheck.removeFirst();
 
-        members.forEach( ( key ) ->
-                         {
-                             if ( !seen.contains( key ) )
+            if ( next == null || next.isDisabled() && Boolean.TRUE.equals( enabled ) )
+            {
+                return;
+            }
+
+            List<StoreKey> members = new ArrayList<>( next.getConstituents() );
+            if ( includeGroups )
+            {
+                result.add( next );
+            }
+
+            // TODO: Need to refactor away from actual recursion.
+            members.forEach( ( key ) ->
                              {
-                                 seen.add( key );
-                                 final StoreType type = key.getType();
-                                 if ( recurseGroups && type == StoreType.group )
-                                 {
-                                     // if we're here, we're definitely recursing groups...
-                                     recurseGroup( (Group) stores.get( key ), stores, result, seen, includeGroups,
-                                                   true );
-                                 }
-                                 else
+                                 if ( !seen.contains( key ) )
                                  {
-                                     final ArtifactStore store = stores.get( key );
-                                     if ( store != null && !( store.isDisabled() && Boolean.TRUE.equals( enabled ) ) )
+                                     seen.add( key );
+                                     final StoreType type = key.getType();
+                                     try
+                                     {
+                                         if ( recurseGroups && type == StoreType.group )
+                                         {
+                                             // if we're here, we're definitely recursing groups...
+                                             Group group = (Group) dataManager.getArtifactStore( key );
+                                             toCheck.addFirst( group );
+                                         }
+                                         else
+                                         {
+                                             final ArtifactStore store = dataManager.getArtifactStore( key );
+                                             if ( store != null && !( store.isDisabled() && Boolean.TRUE.equals( enabled ) ) )
+                                             {
+                                                 result.add( store );
+                                             }
+                                         }
+                                     }
+                                     catch ( IndyDataException e )
                                      {
-                                         result.add( store );
+                                         errorRef.set(e);
                                      }
                                  }
-                             }
-                         } );
+                             } );
+
+            IndyDataException error = errorRef.get();
+            if ( error != null )
+            {
+                throw error;
+            }
+        }
     }
 
 }