| /* |
| * Copyright 2014 The Apache Software Foundation. |
| * |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.usergrid.corepersistence; |
| |
| |
| import com.google.common.base.Optional; |
| import com.google.common.cache.CacheBuilder; |
| import com.google.common.cache.CacheLoader; |
| import com.google.common.cache.LoadingCache; |
| import com.google.common.util.concurrent.UncheckedExecutionException; |
| import com.google.inject.Injector; |
| import com.google.inject.Key; |
| import com.google.inject.TypeLiteral; |
| import org.apache.commons.lang.StringUtils; |
| import org.apache.usergrid.corepersistence.asyncevents.AsyncEventService; |
| import org.apache.usergrid.corepersistence.index.CollectionSettingsFactory; |
| import org.apache.usergrid.corepersistence.index.ReIndexRequestBuilder; |
| import org.apache.usergrid.corepersistence.index.ReIndexService; |
| import org.apache.usergrid.corepersistence.service.CollectionService; |
| import org.apache.usergrid.corepersistence.service.ConnectionService; |
| import org.apache.usergrid.corepersistence.util.CpNamingUtils; |
| import org.apache.usergrid.exception.ConflictException; |
| import org.apache.usergrid.locking.LockManager; |
| import org.apache.usergrid.mq.QueueManagerFactory; |
| import org.apache.usergrid.persistence.*; |
| import org.apache.usergrid.persistence.actorsystem.ActorSystemFig; |
| import org.apache.usergrid.persistence.actorsystem.ActorSystemManager; |
| import org.apache.usergrid.persistence.cassandra.CassandraService; |
| import org.apache.usergrid.persistence.cassandra.CounterUtils; |
| import org.apache.usergrid.persistence.cassandra.Setup; |
| import org.apache.usergrid.persistence.collection.EntityCollectionManager; |
| import org.apache.usergrid.persistence.collection.exception.CollectionRuntimeException; |
| import org.apache.usergrid.persistence.collection.serialization.impl.migration.EntityIdScope; |
| import org.apache.usergrid.persistence.collection.uniquevalues.UniqueValuesService; |
| import org.apache.usergrid.persistence.core.metrics.MetricsFactory; |
| import org.apache.usergrid.persistence.core.migration.data.MigrationDataProvider; |
| import org.apache.usergrid.persistence.core.scope.ApplicationScope; |
| import org.apache.usergrid.persistence.core.scope.ApplicationScopeImpl; |
| import org.apache.usergrid.persistence.core.util.Health; |
| import org.apache.usergrid.persistence.entities.Application; |
| import org.apache.usergrid.persistence.exceptions.ApplicationAlreadyExistsException; |
| import org.apache.usergrid.persistence.exceptions.DuplicateUniquePropertyExistsException; |
| import org.apache.usergrid.persistence.exceptions.EntityNotFoundException; |
| import org.apache.usergrid.persistence.graph.*; |
| import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType; |
| import org.apache.usergrid.persistence.index.EntityIndex; |
| import org.apache.usergrid.persistence.model.entity.Id; |
| import org.apache.usergrid.persistence.model.entity.SimpleId; |
| import org.apache.usergrid.persistence.model.util.UUIDGenerator; |
| import org.apache.usergrid.persistence.qakka.App; |
| import org.apache.usergrid.persistence.qakka.distributed.DistributedQueueService; |
| import org.apache.usergrid.persistence.qakka.distributed.impl.QueueActorRouterProducer; |
| import org.apache.usergrid.persistence.qakka.distributed.impl.QueueSenderRouterProducer; |
| import org.apache.usergrid.persistence.qakka.distributed.impl.QueueWriterRouterProducer; |
| import org.apache.usergrid.utils.UUIDUtils; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.springframework.beans.BeansException; |
| import org.springframework.context.ApplicationContext; |
| import org.springframework.context.ApplicationContextAware; |
| import rx.Observable; |
| |
| import java.util.*; |
| |
| import static java.lang.String.CASE_INSENSITIVE_ORDER; |
| import static org.apache.usergrid.persistence.Schema.PROPERTY_NAME; |
| import static org.apache.usergrid.persistence.Schema.TYPE_APPLICATION; |
| |
| |
| /** |
| * Implement good-old Usergrid EntityManagerFactory with the new-fangled Core Persistence API. |
| * This is where we keep track of applications and system properties. |
| */ |
| public class CpEntityManagerFactory implements EntityManagerFactory, ApplicationContextAware { |
| |
| private static final Logger logger = LoggerFactory.getLogger( CpEntityManagerFactory.class ); |
| |
| private final EntityManagerFig entityManagerFig; |
| private final ActorSystemFig actorSystemFig; |
| |
| private ApplicationContext applicationContext; |
| |
| private Setup setup = null; |
| |
| EntityManager managementAppEntityManager = null; |
| |
| // cache of already instantiated entity managers |
| private final String ENTITY_MANAGER_CACHE_SIZE = "entity.manager.cache.size"; |
| private final LoadingCache<UUID, EntityManager> entityManagers; |
| |
| private final ApplicationIdCache applicationIdCache; |
| |
| Application managementApp = null; |
| |
| private ManagerCache managerCache; |
| |
| private CassandraService cassandraService; |
| private CounterUtils counterUtils; |
| private Injector injector; |
| private final ReIndexService reIndexService; |
| private final MetricsFactory metricsFactory; |
| private final AsyncEventService indexService; |
| private final CollectionService collectionService; |
| private final ConnectionService connectionService; |
| private final GraphManagerFactory graphManagerFactory; |
| private final CollectionSettingsFactory collectionSettingsFactory; |
| private ActorSystemManager actorSystemManager; |
| private final LockManager lockManager; |
| |
| private final QueueManagerFactory queueManagerFactory; |
| |
| public static final String MANAGEMENT_APP_INIT_MAXRETRIES= "management.app.init.max-retries"; |
| public static final String MANAGEMENT_APP_INIT_INTERVAL = "management.app.init.interval"; |
| |
| |
| public CpEntityManagerFactory( |
| final CassandraService cassandraService, final CounterUtils counterUtils, final Injector injector ) { |
| |
| this.cassandraService = cassandraService; |
| this.counterUtils = counterUtils; |
| this.injector = injector; |
| |
| this.reIndexService = injector.getInstance(ReIndexService.class); |
| this.entityManagerFig = injector.getInstance(EntityManagerFig.class); |
| this.actorSystemFig = injector.getInstance( ActorSystemFig.class ); |
| this.managerCache = injector.getInstance( ManagerCache.class ); |
| this.metricsFactory = injector.getInstance( MetricsFactory.class ); |
| this.indexService = injector.getInstance( AsyncEventService.class ); |
| this.graphManagerFactory = injector.getInstance( GraphManagerFactory.class ); |
| this.collectionService = injector.getInstance( CollectionService.class ); |
| this.connectionService = injector.getInstance( ConnectionService.class ); |
| this.collectionSettingsFactory = injector.getInstance( CollectionSettingsFactory.class ); |
| |
| Properties properties = cassandraService.getProperties(); |
| this.entityManagers = createEntityManagerCache( properties ); |
| |
| logger.info("EntityManagerFactoring starting..."); |
| |
| if ( actorSystemFig.getEnabled() ) { |
| try { |
| logger.info("Akka cluster starting..."); |
| |
| // TODO: fix this kludge |
| injector.getInstance( App.class ); |
| this.actorSystemManager = injector.getInstance( ActorSystemManager.class ); |
| |
| actorSystemManager.registerRouterProducer( injector.getInstance( UniqueValuesService.class ) ); |
| actorSystemManager.registerRouterProducer( injector.getInstance( QueueActorRouterProducer.class ) ); |
| actorSystemManager.registerRouterProducer( injector.getInstance( QueueWriterRouterProducer.class ) ); |
| actorSystemManager.registerRouterProducer( injector.getInstance( QueueSenderRouterProducer.class ) ); |
| actorSystemManager.start(); |
| actorSystemManager.waitForClientActor(); |
| |
| DistributedQueueService distributedQueueService = |
| injector.getInstance( DistributedQueueService.class ); |
| |
| distributedQueueService.init(); |
| |
| } catch (Throwable t) { |
| logger.error("Error starting Akka", t); |
| throw t; |
| } |
| } |
| this.lockManager = injector.getInstance( LockManager.class ); |
| this.queueManagerFactory = injector.getInstance( QueueManagerFactory.class ); |
| |
| |
| // this line always needs to be last due to the temporary cicular dependency until spring is removed |
| this.applicationIdCache = injector.getInstance(ApplicationIdCacheFactory.class).getInstance( |
| getManagementEntityManager() ); |
| |
| checkManagementApp( properties ); |
| } |
| |
| |
| private LoadingCache<UUID, EntityManager> createEntityManagerCache(Properties properties) { |
| |
| int entityManagerCacheSize = 100; |
| try { |
| entityManagerCacheSize = Integer.parseInt( properties.getProperty( ENTITY_MANAGER_CACHE_SIZE, "100" )); |
| } catch ( Exception e ) { |
| logger.error("Error parsing " + ENTITY_MANAGER_CACHE_SIZE + ". Will use " + entityManagerCacheSize, e ); |
| } |
| |
| return CacheBuilder.newBuilder() |
| .maximumSize(entityManagerCacheSize) |
| .build(new CacheLoader<UUID, EntityManager>() { |
| |
| public EntityManager load( UUID appId ) { // no checked exception |
| |
| // create new entity manager and pre-fetch its application |
| EntityManager entityManager = _getEntityManager( appId ); |
| Application app = null; |
| Throwable throwable = null; |
| try { |
| app = entityManager.getApplication(); |
| } catch (Throwable t) { |
| throwable = t; |
| } |
| |
| // the management app is a special case |
| if ( CpNamingUtils.MANAGEMENT_APPLICATION_ID.equals( appId ) ) { |
| |
| if ( app != null ) { |
| // we successfully fetched up the management app, cache it for a rainy day |
| managementAppEntityManager = entityManager; |
| |
| } else if ( managementAppEntityManager != null ) { |
| // failed to fetch management app, use cached one |
| entityManager = managementAppEntityManager; |
| logger.error("Failed to fetch management app"); |
| } |
| } |
| |
| // missing keyspace means we have not done bootstrap yet |
| final boolean isBootstrapping; |
| if ( throwable instanceof CollectionRuntimeException ) { |
| CollectionRuntimeException cre = (CollectionRuntimeException) throwable; |
| isBootstrapping = cre.isBootstrapping(); |
| } else { |
| isBootstrapping = false; |
| } |
| |
| // work around for https://issues.apache.org/jira/browse/USERGRID-1291 |
| // throw exception so that we do not cache |
| // TODO: determine how application name can intermittently be null |
| if ( app != null && app.getName() == null ) { |
| throw new RuntimeException( "Name is null for application " + appId, throwable ); |
| } |
| |
| if ( app == null && !isBootstrapping ) { |
| throw new RuntimeException( "Error getting application " + appId, throwable ); |
| |
| } // else keyspace is missing because setup/bootstrap not done yet |
| |
| return entityManager; |
| } |
| }); |
| } |
| |
| |
| private void checkManagementApp(Properties properties) { |
| |
| int maxRetries = 100; |
| try { |
| maxRetries = Integer.parseInt( properties.getProperty( MANAGEMENT_APP_INIT_MAXRETRIES, "100" )); |
| |
| } catch ( Exception e ) { |
| logger.error("Error parsing " + MANAGEMENT_APP_INIT_MAXRETRIES + ". Will use " + maxRetries, e ); |
| } |
| |
| int interval = 1000; |
| try { |
| interval = Integer.parseInt( properties.getProperty( MANAGEMENT_APP_INIT_INTERVAL, "1000" )); |
| |
| } catch ( Exception e ) { |
| logger.error("Error parsing " + MANAGEMENT_APP_INIT_INTERVAL + ". Will use " + maxRetries, e ); |
| } |
| |
| // hold up construction until we can access the management app |
| int retries = 0; |
| boolean managementAppFound = false; |
| boolean bootstrapping = false; |
| Set<Class> seenBefore = new HashSet<>(10); |
| while ( !managementAppFound && retries++ < maxRetries ) { |
| try { |
| // bypass entity manager cache and get managementApp |
| managementApp = _getEntityManager( getManagementAppId() ).getApplication(); |
| managementAppFound = true; |
| |
| } catch ( Throwable t ) { |
| |
| if ( t instanceof CollectionRuntimeException ) { |
| CollectionRuntimeException cre = (CollectionRuntimeException)t; |
| if ( cre.isBootstrapping() ) { |
| // we're bootstrapping, ignore this and continue |
| bootstrapping = true; |
| break; |
| } |
| } |
| Throwable cause = t; |
| |
| // there was an error, be as informative as possible |
| StringBuilder sb = new StringBuilder(); |
| sb.append(retries).append(": Error ("); |
| |
| if ( t instanceof UncheckedExecutionException ) { |
| UncheckedExecutionException uee = (UncheckedExecutionException)t; |
| if ( uee.getCause() instanceof RuntimeException ) { |
| cause = uee.getCause().getCause(); |
| sb.append(cause.getClass().getSimpleName()).append(") ") |
| .append(uee.getCause().getMessage()); |
| } else { |
| cause = uee.getCause(); |
| sb.append(cause.getClass().getSimpleName()).append(") ").append(t.getMessage()); |
| } |
| } else { |
| sb.append(t.getCause().getClass().getSimpleName()).append(") ").append(t.getMessage()); |
| } |
| |
| String msg = sb.toString(); |
| if ( !seenBefore.contains( cause.getClass() ) ) { |
| logger.error( msg, t); |
| } else { |
| logger.error(msg); |
| } |
| seenBefore.add( cause.getClass() ); |
| |
| try { Thread.sleep( interval ); } catch (InterruptedException ignored) {} |
| } |
| } |
| |
| if ( !managementAppFound && !bootstrapping ) { |
| // exception here will prevent WAR from being deployed |
| throw new RuntimeException( "Unable to get management app after " + retries + " retries" ); |
| } |
| } |
| |
| |
| public CounterUtils getCounterUtils() { |
| return counterUtils; |
| } |
| |
| |
| public CassandraService getCassandraService() { |
| return cassandraService; |
| } |
| |
| |
| private void initMgmtAppInternal() { |
| |
| EntityManager em = getEntityManager(getManagementAppId()); |
| indexService.queueInitializeApplicationIndex(CpNamingUtils.getApplicationScope(getManagementAppId())); |
| |
| try { |
| if ( em.getApplication() == null ) { |
| logger.info("Creating management application"); |
| Map mgmtAppProps = new HashMap<String, Object>(); |
| mgmtAppProps.put(PROPERTY_NAME, CassandraService.MANAGEMENT_APPLICATION); |
| em.create( getManagementAppId(), TYPE_APPLICATION, mgmtAppProps); |
| em.getApplication(); |
| } |
| |
| } catch (Exception ex) { |
| throw new RuntimeException("Fatal error creating management application", ex); |
| } |
| } |
| |
| |
| private Observable<EntityIdScope> getAllEntitiesObservable(){ |
| return injector.getInstance( Key.get(new TypeLiteral< MigrationDataProvider<EntityIdScope>>(){})).getData(); |
| } |
| |
| |
| @Override |
| public EntityManager getEntityManager(UUID applicationId) { |
| try { |
| return entityManagers.get( applicationId ); |
| } |
| catch ( Throwable t ) { |
| logger.error("Error getting entity manager", t); |
| } |
| return _getEntityManager(applicationId); |
| } |
| |
| |
| private EntityManager _getEntityManager( UUID applicationId ) { |
| |
| EntityManager em = new CpEntityManager( |
| cassandraService, |
| counterUtils, |
| indexService, |
| managerCache, |
| metricsFactory, |
| actorSystemFig, |
| entityManagerFig, |
| graphManagerFactory, |
| collectionService, |
| connectionService, |
| collectionSettingsFactory, |
| applicationId, |
| queueManagerFactory); |
| |
| return em; |
| } |
| |
| @Override |
| public Entity createApplicationV2(String organizationName, String name) throws Exception { |
| return createApplicationV2( organizationName, name, null, null, false); |
| } |
| |
| |
| @Override |
| public Entity createApplicationV2( |
| String orgName, String name, UUID applicationId, Map<String, Object> properties, boolean forMigration) throws Exception { |
| |
| String appName = buildAppName( orgName, name ); |
| |
| final UUID appId = applicationIdCache.getApplicationId( appName ); |
| |
| if ( appId != null ) { |
| throw new ApplicationAlreadyExistsException( name ); |
| } |
| |
| applicationId = applicationId==null ? UUIDGenerator.newTimeUUID() : applicationId; |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("New application orgName {} orgAppName {} id {} ", |
| orgName, name, applicationId.toString()); |
| } |
| |
| return initializeApplicationV2( orgName, applicationId, appName, properties, forMigration); |
| } |
| |
| |
| |
| private String buildAppName( String organizationName, String name ) { |
| return StringUtils.lowerCase(name.contains("/") ? name : organizationName + "/" + name); |
| } |
| |
| |
| /** |
| * @return UUID of newly created Entity of type application_info |
| */ |
| @Override |
| public Entity initializeApplicationV2(String organizationName, final UUID applicationId, String name, |
| Map<String, Object> properties, boolean forMigration) throws Exception { |
| |
| // Ensure the management application is initialized |
| initMgmtAppInternal(); |
| |
| // Get entity managers by bypassing the entity manager cache because it expects apps to already exist |
| final EntityManager managementEm = _getEntityManager( getManagementAppId() ); |
| EntityManager appEm = _getEntityManager(applicationId); |
| |
| final String appName = buildAppName(organizationName, name); |
| |
| // check for pre-existing application |
| |
| if ( lookupApplication( appName ) != null ) { |
| throw new ApplicationAlreadyExistsException( appName ); |
| } |
| |
| // Initialize the index for this new application |
| appEm.initializeIndex(); |
| indexService.queueInitializeApplicationIndex(CpNamingUtils.getApplicationScope(applicationId)); |
| if ( properties == null ) { |
| properties = new TreeMap<>( CASE_INSENSITIVE_ORDER); |
| } |
| properties.put( PROPERTY_NAME, appName ); |
| appEm.create(applicationId, TYPE_APPLICATION, properties); |
| |
| // only reset roles if this application isn't being migrated (meaning dictionary and role data already exists) |
| if(!forMigration){ |
| appEm.resetRoles(); |
| } |
| |
| |
| |
| // create application info entity in the management app |
| |
| Map<String, Object> appInfoMap = new HashMap<String, Object>() {{ |
| put( PROPERTY_NAME, appName ); |
| put( "org", organizationName ); |
| }}; |
| |
| Entity appInfo; |
| try { |
| appInfo = managementEm.create(new SimpleId(applicationId,CpNamingUtils.APPLICATION_INFO), appInfoMap); |
| } catch (DuplicateUniquePropertyExistsException e) { |
| throw new ApplicationAlreadyExistsException(appName); |
| } |
| |
| // evict app Id from cache |
| applicationIdCache.evictAppId(appName); |
| |
| logger.info("Initialized application {}, uuid {}", appName, appInfo.getUuid().toString()); |
| return appInfo; |
| } |
| |
| |
| |
| /** |
| * Delete Application. |
| * |
| * <p>The Application Entity is be moved to a Deleted_Applications collection and the |
| * Application index will be removed. |
| * |
| * <p>TODO: add scheduled task that can completely delete all deleted application data.</p> |
| * |
| * @param applicationId UUID of Application to be deleted. |
| */ |
| @Override |
| public void deleteApplication(UUID applicationId) throws Exception { |
| |
| // find application_info for application to delete |
| |
| migrateAppInfo(applicationId, CpNamingUtils.APPLICATION_INFO, CpNamingUtils.DELETED_APPLICATION_INFOS, CpNamingUtils.DELETED_APPLICATION_INFO).toBlocking() |
| .lastOrDefault( null ); |
| } |
| |
| //TODO: return status for restore |
| @Override |
| public Entity restoreApplication(UUID applicationId) throws Exception { |
| |
| // get the deleted_application_info for the deleted app |
| return (Entity) migrateAppInfo( applicationId, CpNamingUtils.DELETED_APPLICATION_INFO, |
| CpNamingUtils.APPLICATION_INFOS , CpNamingUtils.APPLICATION_INFO ).lastOrDefault( null ) |
| .map( appInfo -> { |
| |
| //start the index rebuild |
| final ReIndexRequestBuilder builder = reIndexService.getBuilder().withApplicationId( applicationId ); |
| reIndexService.rebuildIndex( builder ); |
| |
| //load the entity |
| final EntityManager managementEm = getEntityManager( getManagementAppId() ); |
| try { |
| return managementEm.get( new SimpleEntityRef( CpNamingUtils.APPLICATION_INFO, applicationId ) ); |
| } |
| catch ( Exception e ) { |
| logger.error( "Failed to get entity", e ); |
| throw new RuntimeException( e ); |
| } |
| } ) |
| .toBlocking().lastOrDefault(null); |
| |
| } |
| |
| // @Override |
| |
| |
| /** |
| * Migrate the application from one type to another. Used in delete and restore |
| * @param applicationUUID The applicationUUID |
| * @param deleteTypeName The type to use on the delete |
| * @param createCollectionName The name of the collection to write the entity into |
| * @param createTypeName The type to use on the create |
| * @return |
| * @throws Exception |
| */ |
| private Observable migrateAppInfo(final UUID applicationUUID, final String deleteTypeName, final String createCollectionName, final String createTypeName ) throws Exception { |
| |
| final ApplicationScope managementAppScope = CpNamingUtils.getApplicationScope(CpNamingUtils.MANAGEMENT_APPLICATION_ID); |
| final EntityManager managementEm = getEntityManager(CpNamingUtils.MANAGEMENT_APPLICATION_ID); |
| |
| //the application id we will be removing |
| final Id deleteApplicationId = new SimpleId(applicationUUID, deleteTypeName ); |
| |
| //the application id we'll be creating |
| final Id createApplicationId = new SimpleId( applicationUUID, createTypeName ); |
| |
| //the application scope of the deleted app to clean it's index |
| final ApplicationScope deleteApplicationScope = new ApplicationScopeImpl(deleteApplicationId); |
| |
| Entity oldAppEntity = managementEm.get(new SimpleEntityRef( deleteTypeName, applicationUUID)); |
| |
| if(oldAppEntity == null){ |
| throw new EntityNotFoundException( String.format("Could not find application with UUID '%s'", applicationUUID) ); |
| } |
| |
| |
| // ensure that there is not already a deleted app with the same name |
| |
| final EntityRef alias = managementEm.getAlias( createCollectionName, oldAppEntity.getName() ); |
| if ( alias != null ) { |
| throw new ConflictException( "Cannot delete app with same name as already deleted app" ); |
| } |
| // make a copy of the app to delete application_info entity |
| // and put it in a deleted_application_info collection |
| |
| final Entity newAppEntity = |
| managementEm.create( new SimpleId( applicationUUID, createTypeName ), oldAppEntity.getProperties() ); |
| |
| // copy its connections too |
| |
| final Set<String> connectionTypes = managementEm.getConnectionTypes( oldAppEntity ); |
| Observable copyConnections = Observable.from( connectionTypes ).doOnNext( connType -> { |
| try { |
| final Results connResults = |
| managementEm.getTargetEntities( oldAppEntity, connType, null, Query.Level.ALL_PROPERTIES ); |
| connResults.getEntities().forEach( entity -> { |
| try { |
| managementEm.createConnection( newAppEntity, connType, entity ); |
| } |
| catch ( Exception e ) { |
| throw new RuntimeException( e ); |
| } |
| } ); |
| } |
| catch ( Exception e ) { |
| throw new RuntimeException( e ); |
| } |
| } ); |
| |
| final Id managementAppId = CpNamingUtils.getManagementApplicationId(); |
| final EntityIndex aei = getManagementIndex(); |
| final GraphManager managementGraphManager = managerCache.getGraphManager(managementAppScope); |
| final Edge createEdge = CpNamingUtils.createCollectionEdge(managementAppId, createCollectionName, createApplicationId); |
| |
| final Observable createNodeGraph = managementGraphManager.writeEdge(createEdge); |
| |
| final Observable deleteAppFromIndex = aei.deleteApplication(); |
| |
| return Observable |
| .merge( copyConnections, createNodeGraph, deleteAppFromIndex ) |
| .doOnCompleted( () -> { |
| try { |
| if ( oldAppEntity != null ) { |
| managementEm.delete( oldAppEntity ); |
| applicationIdCache.evictAppId( oldAppEntity.getName() ); |
| } |
| EntityIndex ei = getManagementIndex(); |
| ei.refreshAsync().toBlocking().last(); |
| } catch (Exception e) { |
| throw new RuntimeException(e); |
| } |
| |
| } ); |
| } |
| |
| |
| @Override |
| public UUID importApplication( |
| String organization, UUID applicationId, |
| String name, Map<String, Object> properties) throws Exception { |
| |
| throw new UnsupportedOperationException("Not supported yet."); |
| } |
| |
| |
| public UUID lookupApplication(String orgAppName ) throws Exception { |
| return applicationIdCache.getApplicationId(orgAppName); |
| } |
| |
| |
| @Override |
| public Map<String, UUID> getApplications() throws Exception { |
| return getApplications( CpNamingUtils.getEdgeTypeFromCollectionName( CpNamingUtils.APPLICATION_INFOS ) ); |
| } |
| |
| |
| @Override |
| public Map<String, UUID> getDeletedApplications() throws Exception { |
| return getApplications( CpNamingUtils.getEdgeTypeFromCollectionName( CpNamingUtils.DELETED_APPLICATION_INFOS ) ); |
| } |
| |
| |
| private Map<String, UUID> getApplications(final String edgeType) throws Exception { |
| |
| ApplicationScope appScope = |
| CpNamingUtils.getApplicationScope(CpNamingUtils.MANAGEMENT_APPLICATION_ID); |
| GraphManager gm = managerCache.getGraphManager(appScope); |
| |
| EntityManager managementEM = getEntityManager(CpNamingUtils.MANAGEMENT_APPLICATION_ID); |
| Application managementApp = managementEM.getApplication(); |
| if( managementApp == null ) { |
| throw new RuntimeException("Management App " |
| + CpNamingUtils.MANAGEMENT_APPLICATION_ID + " should never be null"); |
| } |
| Id managementId = new SimpleId( managementApp.getUuid(), managementApp.getType() ); |
| |
| |
| if (logger.isDebugEnabled()) { |
| logger.debug("getApplications(): Loading edges of edgeType {} from {}:{}", |
| edgeType, managementId.getType(), managementId.getUuid()); |
| } |
| |
| Observable<MarkedEdge> edges = gm.loadEdgesFromSource( |
| new SimpleSearchByEdgeType( managementId, edgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, |
| Optional.<Edge>absent() ) ); |
| |
| final EntityCollectionManager ecm = managerCache.getEntityCollectionManager( appScope ); |
| |
| //buffer our edges and batch fetch the app infos for faster I/O |
| return edges.map( edge -> { |
| return edge.getTargetNode(); |
| } ).buffer( 100 ).flatMap( entityIds -> { |
| return ecm.load( entityIds ); |
| } ) |
| .flatMap( entitySet -> Observable.from( entitySet.getEntities() ) ) |
| //collect all the app infos into a single map for return |
| .collect( () -> new HashMap<String, UUID>(), ( appMap, entity ) -> { |
| |
| if ( !entity.getEntity().isPresent() ) { |
| return; |
| } |
| |
| final org.apache.usergrid.persistence.model.entity.Entity entityData = |
| entity.getEntity().get(); |
| |
| final UUID applicationId = entity.getId().getUuid(); |
| final String applicationName = ( String ) entityData.getField( PROPERTY_NAME ).getValue(); |
| |
| appMap.put( applicationName , applicationId ); |
| } ).toBlocking().last(); |
| } |
| |
| |
| @Override |
| public void setup() throws Exception { |
| getSetup().initSchema(false); |
| lockManager.setup(); |
| } |
| |
| |
| @Override |
| public void bootstrap() throws Exception { |
| |
| // Always make sure the database schema is initialized |
| getSetup().initSchema(false); |
| |
| // Roll the new 2.x Migration classes to the latest version supported |
| getSetup().runDataMigration(); |
| |
| // Make sure the management application is created |
| initMgmtAppInternal(); |
| |
| // Ensure management app is initialized |
| getSetup().initMgmtApp(); |
| |
| } |
| |
| |
| @Override |
| public Map<String, String> getServiceProperties() { |
| |
| Map<String, String> props = new HashMap<String,String>(); |
| |
| EntityManager em = getEntityManager(getManagementAppId()); |
| Query q = Query.fromQL("select *"); |
| Results results = null; |
| try { |
| results = em.searchCollection( em.getApplicationRef(), "propertymaps", q); |
| |
| } catch (Exception ex) { |
| logger.error("Error getting system properties", ex); |
| } |
| |
| if ( results == null || results.isEmpty() ) { |
| return props; |
| } |
| |
| org.apache.usergrid.persistence.Entity e = results.getEntity(); |
| for ( String key : e.getProperties().keySet() ) { |
| props.put( key, props.get(key).toString() ); |
| } |
| return props; |
| } |
| |
| |
| @Override |
| public boolean updateServiceProperties(Map<String, String> properties) { |
| |
| EntityManager em = getEntityManager(getManagementAppId()); |
| Query q = Query.fromQL("select *"); |
| Results results = null; |
| try { |
| results = em.searchCollection( em.getApplicationRef(), "propertymaps", q); |
| |
| } catch (Exception ex) { |
| logger.error("Error getting system properties", ex); |
| return false; |
| } |
| |
| org.apache.usergrid.persistence.Entity propsEntity = null; |
| |
| if ( !results.isEmpty() ) { |
| propsEntity = results.getEntity(); |
| |
| } else { |
| propsEntity = EntityFactory.newEntity( UUIDUtils.newTimeUUID(), "propertymap"); |
| } |
| |
| // intentionally going only one-level deep into fields and treating all |
| // values as strings because that is all we need for service properties |
| for ( String key : properties.keySet() ) { |
| propsEntity.setProperty(key, properties.get(key).toString()); |
| } |
| |
| try { |
| em.update( propsEntity ); |
| |
| } catch (Exception ex) { |
| logger.error("Error updating service properties", ex); |
| return false; |
| } |
| |
| return true; |
| } |
| |
| |
| @Override |
| public boolean setServiceProperty(final String name, final String value) { |
| return updateServiceProperties(new HashMap<String, String>() {{ |
| put(name, value); |
| }}); |
| } |
| |
| |
| @Override |
| public boolean deleteServiceProperty(String name) { |
| |
| EntityManager em = getEntityManager(getManagementAppId()); |
| |
| |
| Query q = Query.fromQL( "select *"); |
| Results results = null; |
| try { |
| results = em.searchCollection( em.getApplicationRef(), "propertymaps", q); |
| |
| } catch (Exception ex) { |
| logger.error("Error getting service property for delete of property: {}", name, ex); |
| return false; |
| } |
| |
| org.apache.usergrid.persistence.Entity propsEntity = null; |
| |
| if ( !results.isEmpty() ) { |
| propsEntity = results.getEntity(); |
| |
| } else { |
| propsEntity = EntityFactory.newEntity( UUIDUtils.newTimeUUID(), "propertymap"); |
| } |
| |
| try { |
| ((AbstractEntity)propsEntity).clearDataset( name ); |
| em.update( propsEntity ); |
| |
| } catch (Exception ex) { |
| logger.error("Error deleting service property orgAppName: {}", name, ex); |
| return false; |
| } |
| |
| return true; |
| } |
| |
| public ApplicationContext getApplicationContext() { |
| return applicationContext; |
| } |
| |
| @Override |
| public void setApplicationContext( ApplicationContext applicationContext ) throws BeansException { |
| this.applicationContext = applicationContext; |
| // try { |
| // setup(); |
| // } catch (Exception ex) { |
| // logger.error("Error setting up EMF", ex); |
| // } |
| } |
| |
| |
| @Override |
| public long performEntityCount() { |
| //TODO, this really needs to be a task that writes this data somewhere since this will get |
| //progressively slower as the system expands |
| return (Long) getAllEntitiesObservable().countLong().toBlocking().last(); |
| } |
| |
| |
| |
| @Override |
| public UUID getManagementAppId() { |
| return CpNamingUtils.MANAGEMENT_APPLICATION_ID; |
| } |
| |
| @Override |
| public EntityManager getManagementEntityManager() { |
| return getEntityManager(CpNamingUtils.MANAGEMENT_APPLICATION_ID); |
| } |
| |
| |
| /** |
| * Gets the setup. |
| * @return Setup helper |
| */ |
| public Setup getSetup() { |
| if ( setup == null ) { |
| setup = new CpSetup( this, cassandraService, injector ); |
| } |
| return setup; |
| } |
| |
| |
| /** |
| * TODO, these 3 methods are super janky. During refactoring we should clean this model up |
| */ |
| public EntityIndex.IndexRefreshCommandInfo refreshIndex(UUID applicationId) { |
| return getEntityManager(applicationId).refreshIndex(); |
| } |
| |
| |
| |
| private EntityIndex getManagementIndex() { |
| |
| return |
| managerCache.getEntityIndex( // management app |
| CpNamingUtils.getApplicationScope(getManagementAppId())); |
| } |
| |
| |
| |
| |
| @Override |
| public void flushEntityManagerCaches() { |
| |
| managerCache.invalidate(); |
| |
| applicationIdCache.evictAll(); |
| |
| Map<UUID, EntityManager> entityManagersMap = entityManagers.asMap(); |
| for ( UUID appUuid : entityManagersMap.keySet() ) { |
| EntityManager em = entityManagersMap.get(appUuid); |
| em.flushManagerCaches(); |
| } |
| } |
| |
| |
| @Override |
| public Health getEntityStoreHealth() { |
| |
| // could use any collection scope here, does not matter |
| EntityCollectionManager ecm = managerCache.getEntityCollectionManager( |
| new ApplicationScopeImpl( new SimpleId( CpNamingUtils.MANAGEMENT_APPLICATION_ID, "application" ) ) ); |
| |
| return ecm.getHealth(); |
| } |
| |
| |
| |
| @Override |
| public Health getIndexHealth() { |
| |
| return getManagementIndex().getIndexHealth(); |
| } |
| |
| @Override |
| public void initializeManagementIndex(){ |
| getManagementIndex().initialize(); |
| } |
| } |
