Repository: usergrid Updated Branches: refs/heads/master ea1ba360d -> 8b63aae7d
Initial commit of export application API. Project: http://git-wip-us.apache.org/repos/asf/usergrid/repo Commit: http://git-wip-us.apache.org/repos/asf/usergrid/commit/82e7ec57 Tree: http://git-wip-us.apache.org/repos/asf/usergrid/tree/82e7ec57 Diff: http://git-wip-us.apache.org/repos/asf/usergrid/diff/82e7ec57 Branch: refs/heads/master Commit: 82e7ec57bef24c8b75d3f3479952522fdd916bfa Parents: ea1ba36 Author: Michael Russo <russomich...@google.com> Authored: Mon Mar 27 00:01:38 2017 -0700 Committer: Michael Russo <russomich...@google.com> Committed: Mon Mar 27 00:01:38 2017 -0700 ---------------------------------------------------------------------- .../usergrid/corepersistence/CoreModule.java | 4 + .../export/ExportRequestBuilder.java | 47 ++++ .../export/ExportRequestBuilderImpl.java | 65 +++++ .../corepersistence/export/ExportService.java | 49 ++++ .../export/ExportServiceImpl.java | 235 +++++++++++++++++++ .../util/ObjectJsonSerializer.java | 17 ++ .../rest/applications/ApplicationResource.java | 48 ++++ 7 files changed, 465 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/usergrid/blob/82e7ec57/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java index ef4bb04..af297f2 100644 --- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java +++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/CoreModule.java @@ -20,6 +20,8 @@ import com.google.inject.*; import com.google.inject.assistedinject.FactoryModuleBuilder; import com.google.inject.multibindings.Multibinder; import org.apache.usergrid.corepersistence.asyncevents.*; +import org.apache.usergrid.corepersistence.export.ExportService; +import org.apache.usergrid.corepersistence.export.ExportServiceImpl; import org.apache.usergrid.corepersistence.index.*; import org.apache.usergrid.corepersistence.migration.CoreMigration; import org.apache.usergrid.corepersistence.migration.CoreMigrationPlugin; @@ -137,6 +139,8 @@ public class CoreModule extends AbstractModule { bind( ReIndexService.class ).to( ReIndexServiceImpl.class ); + bind( ExportService.class ).to( ExportServiceImpl.class ); + install( new FactoryModuleBuilder().implement( AggregationService.class, AggregationServiceImpl.class ) .build( AggregationServiceFactory.class ) ); http://git-wip-us.apache.org/repos/asf/usergrid/blob/82e7ec57/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilder.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilder.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilder.java new file mode 100644 index 0000000..71c64f6 --- /dev/null +++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilder.java @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.export; + + +import com.google.common.base.Optional; +import org.apache.usergrid.persistence.core.scope.ApplicationScope; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; + + +/** + * A builder interface to build our re-index request + */ +public interface ExportRequestBuilder { + + /** + * Set the application id + */ + ExportRequestBuilder withApplicationId(final UUID applicationId); + + + /** + * Get the application scope + * @return + */ + Optional<ApplicationScope> getApplicationScope(); + +} http://git-wip-us.apache.org/repos/asf/usergrid/blob/82e7ec57/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilderImpl.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilderImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilderImpl.java new file mode 100644 index 0000000..73fcec4 --- /dev/null +++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportRequestBuilderImpl.java @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.export; + + +import com.google.common.base.Optional; +import org.apache.usergrid.corepersistence.util.CpNamingUtils; +import org.apache.usergrid.persistence.core.scope.ApplicationScope; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; + + +/** + * Index service request builder + */ +public class ExportRequestBuilderImpl implements ExportRequestBuilder { + + private Optional<UUID> withApplicationId = Optional.absent(); + private Optional<String> withCollectionName = Optional.absent(); + private Optional<String> cursor = Optional.absent(); + private Optional<Long> updateTimestamp = Optional.absent(); + private Optional<Integer> delayTimer = Optional.absent(); + private Optional<TimeUnit> timeUnitOptional = Optional.absent(); + + + /*** + * + * @param applicationId The application id + * @return + */ + @Override + public ExportRequestBuilder withApplicationId(final UUID applicationId ) { + this.withApplicationId = Optional.fromNullable( applicationId ); + return this; + } + + @Override + public Optional<ApplicationScope> getApplicationScope() { + + if ( this.withApplicationId.isPresent() ) { + return Optional.of( CpNamingUtils.getApplicationScope( withApplicationId.get() ) ); + } + + return Optional.absent(); + } + +} http://git-wip-us.apache.org/repos/asf/usergrid/blob/82e7ec57/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportService.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportService.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportService.java new file mode 100644 index 0000000..7615448 --- /dev/null +++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportService.java @@ -0,0 +1,49 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.export; + + +import java.io.IOException; +import java.io.OutputStream; + +/** + * An interface for exporting all entities within an application + */ +public interface ExportService { + + + /** + * Perform an application export + * + * @param exportRequestBuilder The builder to build the request + */ + void export(final ExportRequestBuilder exportRequestBuilder, OutputStream stream) throws IOException; + + + /** + * Generate a build for the index + */ + ExportRequestBuilder getBuilder(); + + + enum Status{ + STARTED, INPROGRESS, COMPLETE, UNKNOWN; + } +} http://git-wip-us.apache.org/repos/asf/usergrid/blob/82e7ec57/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportServiceImpl.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportServiceImpl.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportServiceImpl.java new file mode 100644 index 0000000..ebbcc58 --- /dev/null +++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/export/ExportServiceImpl.java @@ -0,0 +1,235 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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.export; + + +import com.google.common.base.Optional; +import com.google.inject.Inject; +import com.google.inject.Singleton; +import org.apache.usergrid.corepersistence.ManagerCache; +import org.apache.usergrid.corepersistence.index.*; +import org.apache.usergrid.corepersistence.rx.impl.AllEntityIdsObservable; +import org.apache.usergrid.corepersistence.util.CpEntityMapUtils; +import org.apache.usergrid.corepersistence.util.CpNamingUtils; +import org.apache.usergrid.corepersistence.util.ObjectJsonSerializer; +import org.apache.usergrid.persistence.collection.EntityCollectionManager; +import org.apache.usergrid.persistence.collection.EntityCollectionManagerFactory; +import org.apache.usergrid.persistence.core.scope.ApplicationScope; +import org.apache.usergrid.persistence.graph.GraphManager; +import org.apache.usergrid.persistence.graph.SearchByEdgeType; +import org.apache.usergrid.persistence.graph.impl.SimpleSearchByEdgeType; +import org.apache.usergrid.persistence.map.MapManager; +import org.apache.usergrid.persistence.map.MapManagerFactory; +import org.apache.usergrid.persistence.map.MapScope; +import org.apache.usergrid.persistence.map.impl.MapScopeImpl; +import org.apache.usergrid.persistence.model.entity.Entity; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import rx.Observable; +import rx.schedulers.Schedulers; + + +import java.io.*; +import java.util.Map; +import java.util.zip.ZipEntry; +import java.util.zip.ZipOutputStream; + + +@Singleton +public class ExportServiceImpl implements ExportService { + + private static final Logger logger = LoggerFactory.getLogger( ReIndexServiceImpl.class ); + + private static final MapScope RESUME_MAP_SCOPE = + new MapScopeImpl( CpNamingUtils.getManagementApplicationId(), "export-status" ); + + + private static final String MAP_COUNT_KEY = "count"; + private static final String MAP_STATUS_KEY = "status"; + private static final String MAP_UPDATED_KEY = "lastUpdated"; + + + private final AllEntityIdsObservable allEntityIdsObservable; + private final MapManager mapManager; + private final MapManagerFactory mapManagerFactory; + private final CollectionSettingsFactory collectionSettingsFactory; + private final EntityCollectionManagerFactory entityCollectionManagerFactory; + private final ManagerCache managerCache; + + ObjectJsonSerializer jsonSerializer = ObjectJsonSerializer.INSTANCE; + + + + @Inject + public ExportServiceImpl(final AllEntityIdsObservable allEntityIdsObservable, + final ManagerCache managerCache, + final MapManagerFactory mapManagerFactory, + final CollectionSettingsFactory collectionSettingsFactory, + final EntityCollectionManagerFactory entityCollectionManagerFactory) { + this.allEntityIdsObservable = allEntityIdsObservable; + this.collectionSettingsFactory = collectionSettingsFactory; + this.entityCollectionManagerFactory = entityCollectionManagerFactory; + this.mapManagerFactory = mapManagerFactory; + this.mapManager = mapManagerFactory.createMapManager( RESUME_MAP_SCOPE ); + this.managerCache = managerCache; + } + + + @Override + public void export(final ExportRequestBuilder exportRequestBuilder, OutputStream stream) throws IOException { + + final ZipOutputStream zipOutputStream = new ZipOutputStream(stream); + + //final AtomicInteger count = new AtomicInteger(); + final ApplicationScope appScope = exportRequestBuilder.getApplicationScope().get(); + final Observable<ApplicationScope> applicationScopes = Observable.just(appScope); + + + //final String jobId = StringUtils.sanitizeUUID( UUIDGenerator.newTimeUUID() ); + final EntityCollectionManager ecm = entityCollectionManagerFactory.createCollectionManager(appScope); + + final String rootPath = appScope.getApplication().getUuid().toString(); + + GraphManager gm = managerCache.getGraphManager( appScope ); + + allEntityIdsObservable.getEdgesToEntities( applicationScopes, Optional.absent(), Optional.absent() ) + .doOnNext( edgeScope -> { + + try { + + // load the entity and convert to a normal map + Entity entity = ecm.load(edgeScope.getEdge().getTargetNode()).toBlocking().lastOrDefault(null); + Map entityMap = CpEntityMapUtils.toMap(entity); + + if (entity != null) { + final String filenameWithPath = rootPath + "/" + + edgeScope.getEdge().getSourceNode().getUuid().toString() + "_" + + edgeScope.getEdge().getType() + "_" + + edgeScope.getEdge().getTargetNode().getUuid().toString() + ".json"; + + logger.debug("adding zip entry: {}", filenameWithPath); + zipOutputStream.putNextEntry(new ZipEntry(filenameWithPath)); + + + logger.debug("writing and flushing entity to zip stream: {}", jsonSerializer.toString(entityMap)); + zipOutputStream.write(jsonSerializer.toString(entityMap).getBytes()); + zipOutputStream.closeEntry(); + zipOutputStream.flush(); + + } else { + logger.warn("{} did not have corresponding entity, not writing", edgeScope.toString()); + } + + } catch (IOException e) { + logger.warn("Unable to create entry in zip export for edge {}", edgeScope); + } + + //writeStateMeta( jobId, Status.INPROGRESS, count.addAndGet(1), System.currentTimeMillis() ); + }) + .flatMap( edgeScope -> { + + // find all connection types for the each entity emitted from the app + return gm.getEdgeTypesFromSource(CpNamingUtils.createConnectionTypeSearch(edgeScope.getEdge().getTargetNode())) + .flatMap(emittedEdgeType -> { + + logger.debug("loading edges of type {} from node {}", emittedEdgeType, edgeScope.getEdge().getTargetNode()); + return gm.loadEdgesFromSource(new SimpleSearchByEdgeType( edgeScope.getEdge().getTargetNode(), + emittedEdgeType, Long.MAX_VALUE, SearchByEdgeType.Order.DESCENDING, Optional.absent() )); + + }).doOnNext( markedEdge -> { + + if (!markedEdge.isDeleted()){ + + // todo, probably don't need to load the target node itself since it would be loaded in normal collection walking + Entity entity = ecm.load(markedEdge.getTargetNode()).toBlocking().lastOrDefault(null); + + Map entityMap = CpEntityMapUtils.toMap(entity); + + try { + final String filenameWithPath = rootPath + "/" + + markedEdge.getSourceNode().getUuid().toString() + "_" + + markedEdge.getType() + "_" + + markedEdge.getTargetNode().getUuid().toString() + ".json"; + + logger.debug("adding zip entry: {}", filenameWithPath); + zipOutputStream.putNextEntry(new ZipEntry(filenameWithPath)); + + logger.debug("writing and flushing entity to zip stream: {}", jsonSerializer.toString(entityMap)); + zipOutputStream.write(jsonSerializer.toString(entityMap).getBytes()); + zipOutputStream.closeEntry(); + zipOutputStream.flush(); + + } catch (IOException e) { + logger.warn("Unable to create entry in zip export for edge {}", markedEdge.toString()); + } + } + + }); + + }) + .doOnCompleted(() -> { + + //writeStateMeta( jobId, Status.COMPLETE, count.get(), System.currentTimeMillis() ); + try { + logger.debug("closing zip stream"); + zipOutputStream.close(); + + } catch (IOException e) { + logger.error( "unable to close zip stream"); + } + + }) + .subscribeOn( Schedulers.io() ).toBlocking().lastOrDefault(null); + } + + + @Override + public ExportRequestBuilder getBuilder() { + return new ExportRequestBuilderImpl(); + } + + + + + /** + * Write our state meta data into cassandra so everyone can see it + * @param jobId + * @param status + * @param processedCount + * @param lastUpdated + */ + private void writeStateMeta( final String jobId, final Status status, final long processedCount, + final long lastUpdated ) { + + if(logger.isDebugEnabled()) { + logger.debug( "Flushing state for jobId {}, status {}, processedCount {}, lastUpdated {}", + jobId, status, processedCount, lastUpdated); + } + + mapManager.putString( jobId + MAP_STATUS_KEY, status.name() ); + mapManager.putLong( jobId + MAP_COUNT_KEY, processedCount ); + mapManager.putLong( jobId + MAP_UPDATED_KEY, lastUpdated ); + } + + +} + + http://git-wip-us.apache.org/repos/asf/usergrid/blob/82e7ec57/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/ObjectJsonSerializer.java ---------------------------------------------------------------------- diff --git a/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/ObjectJsonSerializer.java b/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/ObjectJsonSerializer.java index 4e5873a..b704afa 100644 --- a/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/ObjectJsonSerializer.java +++ b/stack/core/src/main/java/org/apache/usergrid/corepersistence/util/ObjectJsonSerializer.java @@ -77,6 +77,23 @@ public final class ObjectJsonSerializer { return stringValue; } + public <T> String toString( final T toSerialize ) { + + Preconditions.checkNotNull( toSerialize, "toSerialize must not be null" ); + final String stringValue; + //mark this version as empty + + //Convert to internal entity map + try { + stringValue = MAPPER.writeValueAsString( toSerialize ); + } + catch ( JsonProcessingException jpe ) { + throw new RuntimeException( "Unable to serialize entity", jpe ); + } + + return stringValue; + } + public <T extends Serializable> T fromString( final String value, final Class<T> toSerialize ) { http://git-wip-us.apache.org/repos/asf/usergrid/blob/82e7ec57/stack/rest/src/main/java/org/apache/usergrid/rest/applications/ApplicationResource.java ---------------------------------------------------------------------- diff --git a/stack/rest/src/main/java/org/apache/usergrid/rest/applications/ApplicationResource.java b/stack/rest/src/main/java/org/apache/usergrid/rest/applications/ApplicationResource.java index 9836f1c..920287b 100644 --- a/stack/rest/src/main/java/org/apache/usergrid/rest/applications/ApplicationResource.java +++ b/stack/rest/src/main/java/org/apache/usergrid/rest/applications/ApplicationResource.java @@ -25,6 +25,11 @@ import org.apache.amber.oauth2.common.message.OAuthResponse; import org.apache.amber.oauth2.common.message.types.GrantType; import org.apache.shiro.authz.UnauthorizedException; import org.apache.shiro.codec.Base64; +import org.apache.usergrid.corepersistence.export.ExportRequestBuilder; +import org.apache.usergrid.corepersistence.export.ExportRequestBuilderImpl; +import org.apache.usergrid.corepersistence.export.ExportService; +import org.apache.usergrid.corepersistence.index.ReIndexRequestBuilder; +import org.apache.usergrid.corepersistence.index.ReIndexService; import org.apache.usergrid.management.ApplicationInfo; import org.apache.usergrid.management.exceptions.DisabledAdminUserException; import org.apache.usergrid.management.exceptions.DisabledAppUserException; @@ -37,6 +42,7 @@ import org.apache.usergrid.persistence.entities.Application; import org.apache.usergrid.persistence.entities.User; import org.apache.usergrid.persistence.exceptions.EntityNotFoundException; import org.apache.usergrid.persistence.index.query.Identifier; +import org.apache.usergrid.persistence.index.utils.UUIDUtils; import org.apache.usergrid.rest.AbstractContextResource; import org.apache.usergrid.rest.ApiResponse; import org.apache.usergrid.rest.applications.assets.AssetsResource; @@ -58,6 +64,8 @@ import org.springframework.stereotype.Component; import javax.ws.rs.*; import javax.ws.rs.core.*; +import java.io.IOException; +import java.io.OutputStream; import java.io.UnsupportedEncodingException; import java.net.URLEncoder; import java.util.Map; @@ -676,4 +684,44 @@ public class ApplicationResource extends CollectionResource { } + private ExportService getExportService() { + return injector.getInstance( ExportService.class ); + } + + + @GET + @Path("export") + @RequireApplicationAccess + @Produces({"application/zip"}) + public Response getExport( @Context UriInfo ui, + @QueryParam("callback") @DefaultValue("callback") String callback ) + throws Exception { + + if (logger.isTraceEnabled()) { + logger.trace("ApplicationResource.getExport"); + } + + if ( !isApplicationAdmin( Identifier.fromUUID( applicationId ) ) ) { + throw new UnauthorizedException(); + } + + + final ExportRequestBuilder request = new ExportRequestBuilderImpl().withApplicationId( applicationId ); + StreamingOutput stream = new StreamingOutput() { + @Override + public void write(OutputStream outputStream) throws IOException, WebApplicationException { + getExportService().export(request,outputStream); + } + }; + return Response + .ok(stream) + .header("Content-Disposition", "attachment; filename=\"usergrid_export-"+System.currentTimeMillis()+".zip\"") + .build(); + } + + + + + + }
