You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@usergrid.apache.org by gr...@apache.org on 2016/05/02 19:55:11 UTC
[40/50] [abbrv] usergrid git commit: Removing export and import files
along with the other export that uses Jclouds
http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
deleted file mode 100644
index 6c1ffc0..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/ImportServiceImpl.java
+++ /dev/null
@@ -1,1352 +0,0 @@
-/*
- * 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.management.importer;
-
-import com.google.common.base.Preconditions;
-import org.apache.usergrid.batch.JobExecution;
-import org.apache.usergrid.batch.service.SchedulerService;
-import org.apache.usergrid.corepersistence.util.CpNamingUtils;
-import org.apache.usergrid.management.ManagementService;
-import org.apache.usergrid.persistence.*;
-import org.apache.usergrid.persistence.entities.FailedImportEntity;
-import org.apache.usergrid.persistence.entities.FileImport;
-import org.apache.usergrid.persistence.entities.Import;
-import org.apache.usergrid.persistence.entities.JobData;
-import org.apache.usergrid.persistence.exceptions.EntityNotFoundException;
-import org.apache.usergrid.persistence.Query;
-import org.apache.usergrid.persistence.Query.Level;
-import org.apache.usergrid.utils.InflectionUtils;
-import org.codehaus.jackson.JsonFactory;
-import org.codehaus.jackson.JsonParser;
-import org.codehaus.jackson.JsonToken;
-import org.codehaus.jackson.map.ObjectMapper;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import rx.Observable;
-import rx.Subscriber;
-import rx.functions.Action1;
-import rx.functions.Func2;
-import rx.schedulers.Schedulers;
-
-import javax.annotation.Nullable;
-import javax.annotation.PostConstruct;
-import java.io.File;
-import java.util.*;
-
-
-public class ImportServiceImpl implements ImportService {
-
- public static final String IMPORT_ID = "importId";
- public static final String IMPORT_JOB_NAME = "importJob";
- public static final String FILE_IMPORT_ID = "fileImportId";
- public static final String FILE_IMPORT_JOB_NAME = "fileImportJob";
- public static final int HEARTBEAT_COUNT = 50;
-
- public static final String APP_IMPORT_CONNECTION = "imports";
- public static final String IMPORT_FILE_INCLUDES_CONNECTION = "files";
-
- private static final Logger logger = LoggerFactory.getLogger(ImportServiceImpl.class);
-
- int MAX_FILE_IMPORTS = 1000; // max number of file import jobs / import job
-
- protected EntityManagerFactory emf;
-
- private SchedulerService sch;
-
- private ManagementService managementService;
-
- private JsonFactory jsonFactory = new JsonFactory();
-
-
- @PostConstruct
- public void init() {
- }
-
-
- /**
- * This schedules the main import Job.
- *
- * @param config configuration of the job to be scheduled
- * @return it returns the UUID of the scheduled job
- */
- @Override
- public Import schedule(final UUID application, Map<String, Object> config) throws Exception {
-
- Preconditions.checkNotNull(config, "import information cannot be null");
- Preconditions.checkNotNull(application, "application cannot be null");
-
- final EntityManager rootEM;
- try {
- rootEM = emf.getEntityManager(emf.getManagementAppId());
- } catch (Exception e) {
- logger.error("application doesn't exist within the current context", e);
- return null;
- }
-
- Import importEntity = new Import();
- importEntity.setState(Import.State.CREATED);
-
- // create the import entity to store all metadata about the import job
- try {
- importEntity = rootEM.create(importEntity);
- } catch (Exception e) {
- logger.error("Import entity creation failed", e);
- return null;
- }
-
- // update state for import job to created
-
- // set data to be transferred to importInfo
- JobData jobData = new JobData();
- jobData.setProperty("importInfo", config);
- jobData.setProperty(IMPORT_ID, importEntity.getUuid());
-
- long soonestPossible = System.currentTimeMillis() + 250; //sch grace period
-
- // schedule import job
- sch.createJob(IMPORT_JOB_NAME, soonestPossible, jobData);
-
- // update state for import job to created
- importEntity.setState(Import.State.SCHEDULED);
- rootEM.update(importEntity);
-
- final EntityRef appInfo = getApplicationInfoEntity(rootEM, application);
-
- //now link it to the application
- rootEM.createConnection(appInfo, APP_IMPORT_CONNECTION, importEntity);
-
- return importEntity;
- }
-
-
- @Override
- public Results getImports(final UUID applicationId, @Nullable final String ql, @Nullable final String cursor) {
- Preconditions.checkNotNull(applicationId, "applicationId must be specified");
-
- try {
- final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
- final Entity appInfo = getApplicationInfoEntity(rootEm, applicationId);
-
- Query query = Query.fromQLNullSafe(ql);
- query.setCursor(cursor);
-
- //set our entity type
- query.setEntityType(Schema.getDefaultSchema().getEntityType(Import.class));
-
- return rootEm.searchCollection(appInfo, APP_IMPORT_CONNECTION, query);
- } catch (Exception e) {
- throw new RuntimeException("Unable to get import entity", e);
- }
- }
-
-
- @Override
- public Import getImport(final UUID applicationId, final UUID importId) {
- Preconditions.checkNotNull(applicationId, "applicationId must be specified");
- Preconditions.checkNotNull(importId, "importId must be specified");
-
- try {
- final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
- final Entity appInfo = getApplicationInfoEntity(rootEm, applicationId);
- final Import importEntity = rootEm.get(importId, Import.class);
-
- // check if it's on the path
- if (!rootEm.isConnectionMember(appInfo, APP_IMPORT_CONNECTION, importEntity)) {
- return null;
- }
-
- return importEntity;
- } catch (Exception e) {
- throw new RuntimeException("Unable to get import entity", e);
- }
-
- }
-
-
- private Entity getApplicationInfoEntity(final EntityManager rootEm, final UUID applicationId) throws Exception {
- final Entity entity = rootEm.get(new SimpleEntityRef(CpNamingUtils.APPLICATION_INFO, applicationId));
-
- if (entity == null) {
- throw new EntityNotFoundException("Cound not find application with id " + applicationId);
- }
-
- return entity;
- }
-
- @Override
- public Results getFileImports(final UUID applicationId, final UUID importId,
- @Nullable final String ql, @Nullable final String cursor) {
-
- Preconditions.checkNotNull(applicationId, "applicationId must be specified");
- Preconditions.checkNotNull(importId, "importId must be specified");
-
- try {
- final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-
- final Import importEntity = getImport(applicationId, importId);
-
- Query query = Query.fromQLNullSafe(ql);
- query.setCursor(cursor);
- query.setConnectionType(IMPORT_FILE_INCLUDES_CONNECTION);
- query.setResultsLevel(Level.ALL_PROPERTIES);
-
-
- //set our entity type
- query.setEntityType(Schema.getDefaultSchema().getEntityType(FileImport.class));
-
- return rootEm.searchTargetEntities(importEntity, query);
- } catch (Exception e) {
- throw new RuntimeException("Unable to get import entity", e);
- }
-
- }
-
-
- @Override
- public FileImport getFileImport(final UUID applicationId, final UUID importId, final UUID fileImportId) {
- try {
- final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
- final Import importEntity = getImport(applicationId, importId);
-
- if (importEntity == null) {
- throw new EntityNotFoundException("Import not found with id " + importId);
- }
-
- final FileImport fileImport = rootEm.get(importId, FileImport.class);
-
-
- // check if it's on the path
- if (!rootEm.isConnectionMember(importEntity, APP_IMPORT_CONNECTION, fileImport)) {
- return null;
- }
-
- return fileImport;
- } catch (Exception e) {
- throw new RuntimeException("Unable to load file import", e);
- }
- }
-
-
- @Override
- public Results getFailedImportEntities(final UUID applicationId, final UUID importId, final UUID fileImportId,
- @Nullable final String ql, @Nullable final String cursor) {
-
- Preconditions.checkNotNull(applicationId, "applicationId must be specified");
- Preconditions.checkNotNull(importId, "importId must be specified");
- Preconditions.checkNotNull(fileImportId, "fileImportId must be specified");
-
- try {
- final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-
- final FileImport importEntity = getFileImport(applicationId, importId, fileImportId);
-
- Query query = Query.fromQLNullSafe(ql);
- query.setCursor(cursor);
- query.setConnectionType(FileImportTracker.ERRORS_CONNECTION_NAME);
- query.setResultsLevel(Level.ALL_PROPERTIES);
-
-
- //set our entity type
- query.setEntityType(Schema.getDefaultSchema().getEntityType(FailedImportEntity.class));
-
- return rootEm.searchTargetEntities(importEntity, query);
- } catch (Exception e) {
- throw new RuntimeException("Unable to get import entity", e);
- }
- }
-
-
- @Override
- public FailedImportEntity getFailedImportEntity(final UUID applicationId, final UUID importId,
- final UUID fileImportId, final UUID failedImportId) {
- try {
- final EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
-
- final FileImport importEntity = getFileImport(applicationId, importId, fileImportId);
-
- if (importEntity == null) {
- throw new EntityNotFoundException("Import not found with id " + importId);
- }
-
-
- final FailedImportEntity fileImport = rootEm.get(importId, FailedImportEntity.class);
-
-
- // check if it's on the path
- if (!rootEm.isConnectionMember(importEntity, FileImportTracker.ERRORS_CONNECTION_NAME, fileImport)) {
- return null;
- }
-
- return fileImport;
- } catch (Exception e) {
- throw new RuntimeException("Unable to load file import", e);
- }
- }
-
-
- /**
- * This schedules the sub FileImport Job
- *
- * @param file file to be scheduled
- * @return it returns the UUID of the scheduled job
- * @throws Exception
- */
- private JobData createFileTask(Map<String, Object> config, String file, EntityRef importRef) throws Exception {
-
- if (logger.isTraceEnabled()) {
- logger.trace("scheduleFile() for import {}:{} file {}",
- importRef.getType(), importRef.getType(), file);
- }
-
- EntityManager rootEM;
-
- try {
- rootEM = emf.getEntityManager(emf.getManagementAppId());
- } catch (Exception e) {
- logger.error("application doesn't exist within the current context");
- return null;
- }
-
- // create a FileImport entity to store metadata about the fileImport job
- UUID applicationId = (UUID) config.get("applicationId");
- FileImport fileImport = new FileImport(file, applicationId);
- fileImport = rootEM.create(fileImport);
-
- Import importEntity = rootEM.get(importRef, Import.class);
-
- try {
- // create a connection between the main import job and the sub FileImport Job
- rootEM.createConnection(importEntity, IMPORT_FILE_INCLUDES_CONNECTION, fileImport);
-
- if (logger.isTraceEnabled()) {
- logger.trace("Created connection from {}:{} to {}:{}",
- importEntity.getType(), importEntity.getUuid(),
- fileImport.getType(), fileImport.getUuid()
- );
- }
-
- } catch (Exception e) {
- logger.error("Exception creating file task connection", e.getMessage());
- return null;
- }
-
- // mark the File Import Job as created
- fileImport.setState(FileImport.State.CREATED);
- rootEM.update(fileImport);
-
- // set data to be transferred to the FileImport Job
- JobData jobData = new JobData();
- jobData.setProperty("File", file);
- jobData.setProperty(FILE_IMPORT_ID, fileImport.getUuid());
- jobData.addProperties(config);
-
- // update state of the job to Scheduled
- fileImport.setState(FileImport.State.SCHEDULED);
- rootEM.update(fileImport);
-
- return jobData;
- }
-
-
- private int getConnectionCount(final Import importRoot) {
-
- try {
-
- EntityManager rootEM = emf.getEntityManager(emf.getManagementAppId());
- Query query = Query.fromQL("select *");
- query.setEntityType("file_import");
- query.setConnectionType(IMPORT_FILE_INCLUDES_CONNECTION);
- query.setLimit(MAX_FILE_IMPORTS);
-
- // TODO, this won't work with more than 100 files
- Results entities = rootEM.searchTargetEntities(importRoot, query);
- return entities.size();
-
- // see ImportConnectsTest()
-// Results entities = rootEM.getTargetEntities(
-// importRoot, "includes", null, Level.ALL_PROPERTIES );
-// PagingResultsIterator itr = new PagingResultsIterator( entities );
-// int count = 0;
-// while ( itr.hasNext() ) {
-// itr.next();
-// count++;
-// }
-// return count;
- } catch (Exception e) {
- logger.error("application doesn't exist within the current context");
- throw new RuntimeException(e);
- }
- }
-
-
- /**
- * Schedule the file tasks. This must happen in 2 phases. The first is linking the
- * sub files to the master the second is scheduling them to run.
- */
- private JobData scheduleFileTasks(final JobData jobData) {
-
- long soonestPossible = System.currentTimeMillis() + 250; //sch grace period
-
- // schedule file import job
- return sch.createJob(FILE_IMPORT_JOB_NAME, soonestPossible, jobData);
- }
-
-
- /**
- * Query Entity Manager for the state of the Import Entity. This corresponds to the GET /import
- */
- @Override
- public Import.State getState(UUID uuid) throws Exception {
-
- Preconditions.checkNotNull(uuid, "uuid cannot be null");
-
- EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
- //retrieve the import entity.
- Import importUG = rootEm.get(uuid, Import.class);
-
- if (importUG == null) {
- throw new EntityNotFoundException("Could not find entity with uuid " + uuid);
- }
-
- return importUG.getState();
- }
-
-
- /**
- * Query Entity Manager for the error message generated for an import job.
- */
- @Override
- public String getErrorMessage(final UUID uuid) throws Exception {
-
- //get application entity manager
-
- if (uuid == null) {
- logger.error("getErrorMessage(): UUID passed in cannot be null.");
- return "UUID passed in cannot be null";
- }
-
- EntityManager rootEm = emf.getEntityManager(emf.getManagementAppId());
-
- //retrieve the import entity.
- Import importUG = rootEm.get(uuid, Import.class);
-
- if (importUG == null) {
- logger.error("getErrorMessage(): no entity with that uuid was found");
- return "No Such Element found";
- }
- return importUG.getErrorMessage();
- }
-
-
- /**
- * Returns the Import Entity that stores all meta-data for the particular import Job
- *
- * @param jobExecution the import job details
- * @return Import Entity
- */
- @Override
- public Import getImportEntity(final JobExecution jobExecution) throws Exception {
-
- UUID importId = (UUID) jobExecution.getJobData().getProperty(IMPORT_ID);
- EntityManager importManager = emf.getEntityManager(emf.getManagementAppId());
-
- return importManager.get(importId, Import.class);
- }
-
-
- /**
- * Returns the File Import Entity that stores all meta-data for the particular sub File import Job
- *
- * @return File Import Entity
- */
- @Override
- public FileImport getFileImportEntity(final JobExecution jobExecution) throws Exception {
- UUID fileImportId = (UUID) jobExecution.getJobData().getProperty(FILE_IMPORT_ID);
- EntityManager em = emf.getEntityManager(emf.getManagementAppId());
- return em.get(fileImportId, FileImport.class);
- }
-
-
- public EntityManagerFactory getEmf() {
- return emf;
- }
-
-
- public void setEmf(final EntityManagerFactory emf) {
- this.emf = emf;
- }
-
-
- public ManagementService getManagementService() {
-
- return managementService;
- }
-
-
- public void setManagementService(final ManagementService managementService) {
- this.managementService = managementService;
- }
-
-
- public void setSch(final SchedulerService sch) {
- this.sch = sch;
- }
-
-
- /**
- * This method creates sub-jobs for each file i.e. File Import Jobs.
- *
- * @param jobExecution the job created by the scheduler with all the required config data
- */
- @Override
- public void doImport(JobExecution jobExecution) throws Exception {
-
- if (logger.isTraceEnabled()) {
- logger.trace("doImport()");
- }
-
- @SuppressWarnings("unchecked")
- Map<String, Object> config = (Map<String, Object>) jobExecution.getJobData().getProperty("importInfo");
- if (config == null) {
- logger.error("doImport(): Import Information passed through is null");
- return;
- }
-
- @SuppressWarnings("unchecked")
- Map<String, Object> properties = (Map<String, Object>) config.get("properties");
- @SuppressWarnings("unchecked")
- Map<String, Object> storage_info = (Map<String, Object>) properties.get("storage_info");
-
- String bucketName = (String) storage_info.get("bucket_location");
- String accessId = (String) storage_info.get("s3_access_id");
- String secretKey = (String) storage_info.get("s3_key");
-
- // get Import Entity from the management app, update it to show that job has started
-
- final EntityManager rootEM = emf.getEntityManager(emf.getManagementAppId());
- UUID importId = (UUID) jobExecution.getJobData().getProperty(IMPORT_ID);
- Import importEntity = rootEM.get(importId, Import.class);
-
- importEntity.setState(Import.State.STARTED);
- importEntity.setStarted(System.currentTimeMillis());
- importEntity.setErrorMessage(" ");
- rootEM.update(importEntity);
- if (logger.isTraceEnabled()) {
- logger.trace("doImport(): updated state");
- }
-
- // if no S3 importer was passed in then create one
-
- S3Import s3Import;
- Object s3PlaceHolder = jobExecution.getJobData().getProperty("s3Import");
- try {
- if (s3PlaceHolder != null) {
- s3Import = (S3Import) s3PlaceHolder;
- } else {
- s3Import = new S3ImportImpl();
- }
- } catch (Exception e) {
- logger.error("doImport(): Error creating S3Import", e);
- importEntity.setErrorMessage(e.getMessage());
- importEntity.setState(Import.State.FAILED);
- rootEM.update(importEntity);
- return;
- }
-
- // get list of all JSON files in S3 bucket
-
- final List<String> bucketFiles;
- try {
-
- if (config.get("organizationId") == null) {
- logger.error("doImport(): No organization could be found");
- importEntity.setErrorMessage("No organization could be found");
- importEntity.setState(Import.State.FAILED);
- rootEM.update(importEntity);
- return;
-
- } else {
-
- if (config.get("applicationId") == null) {
- throw new UnsupportedOperationException("Import applications not supported");
-
- } else {
- bucketFiles = s3Import.getBucketFileNames(bucketName, ".json", accessId, secretKey);
- }
- }
-
- } catch (OrganizationNotFoundException | ApplicationNotFoundException e) {
- importEntity.setErrorMessage(e.getMessage());
- importEntity.setState(Import.State.FAILED);
- rootEM.update(importEntity);
- return;
- }
-
-
- // schedule a FileImport job for each file found in the bucket
-
- if (bucketFiles.isEmpty()) {
- importEntity.setState(Import.State.FINISHED);
- importEntity.setErrorMessage("No files found in the bucket: " + bucketName);
- rootEM.update(importEntity);
-
- } else {
-
- Map<String, Object> fileMetadata = new HashMap<>();
- ArrayList<Map<String, Object>> value = new ArrayList<>();
- final List<JobData> fileJobs = new ArrayList<>(bucketFiles.size());
-
- // create the Entity Connection and set up metadata for each job
-
- for (String bucketFile : bucketFiles) {
- final JobData jobData = createFileTask(config, bucketFile, importEntity);
- fileJobs.add(jobData);
- }
-
- int retries = 0;
- int maxRetries = 60;
- boolean done = false;
- while (!done && retries++ < maxRetries) {
-
- final int count = getConnectionCount(importEntity);
-
- if (count == fileJobs.size()) {
- if (logger.isTraceEnabled()) {
- logger.trace("Got ALL {} of {} expected connections", count, fileJobs.size());
- }
- done = true;
- } else {
- if (logger.isTraceEnabled()) {
- logger.trace("Got {} of {} expected connections. Waiting...", count, fileJobs.size());
- }
- Thread.sleep(1000);
- }
- }
- if (retries >= maxRetries) {
- throw new RuntimeException("Max retries was reached");
- }
-
- // schedule each job
-
- for (JobData jobData : fileJobs) {
-
- final JobData scheduled = scheduleFileTasks(jobData);
-
- Map<String, Object> fileJobID = new HashMap<>();
- fileJobID.put("FileName", scheduled.getProperty("File"));
- fileJobID.put("JobID", scheduled.getUuid());
- value.add(fileJobID);
- }
-
- fileMetadata.put("files", value);
- importEntity.addProperties(fileMetadata);
- importEntity.setFileCount(fileJobs.size());
- rootEM.update(importEntity);
- }
- }
-
-
- @Override
- public void downloadAndImportFile(JobExecution jobExecution) {
-
- // get values we need
-
- @SuppressWarnings("unchecked")
- Map<String, Object> properties = (Map<String, Object>) jobExecution.getJobData().getProperty("properties");
- if (properties == null) {
- logger.error("downloadAndImportFile(): Import Information passed through is null");
- return;
- }
- @SuppressWarnings("unchecked")
- Map<String, Object> storage_info = (Map<String, Object>) properties.get("storage_info");
-
- String bucketName = (String) storage_info.get("bucket_location");
- String accessId = (String) storage_info.get("s3_access_id");
- String secretKey = (String) storage_info.get("s3_key");
-
- EntityManager rootEM = emf.getEntityManager(emf.getManagementAppId());
-
- // get the file import entity
-
- FileImport fileImport;
- try {
- fileImport = getFileImportEntity(jobExecution);
- } catch (Exception e) {
- logger.error("Error updating fileImport to set state of file import", e);
- return;
- }
-
- // tracker flushes every 100 entities
- final FileImportTracker tracker = new FileImportTracker(emf, fileImport, 100);
-
- String fileName = jobExecution.getJobData().getProperty("File").toString();
- UUID targetAppId = (UUID) jobExecution.getJobData().getProperty("applicationId");
-
- // is job already done?
- if (FileImport.State.FAILED.equals(fileImport.getState())
- || FileImport.State.FINISHED.equals(fileImport.getState())) {
- return;
- }
-
- // update FileImport Entity to indicate that we have started
-
- if (logger.isDebugEnabled()) {
- logger.debug("downloadAndImportFile() for file {} ", fileName);
- }
-
- try {
- rootEM.update(fileImport);
- fileImport.setState(FileImport.State.STARTED);
- rootEM.update(fileImport);
-
- if (rootEM.get(targetAppId) == null) {
- tracker.fatal("Application " + targetAppId + " does not exist");
- return;
- }
-
- } catch (Exception e) {
- tracker.fatal("Application " + targetAppId + " does not exist");
- checkIfComplete(rootEM, fileImport);
- return;
- }
-
- EntityManager targetEm = emf.getEntityManager(targetAppId);
-
- // download file from S3, if no S3 importer was passed in then create one
-
- File downloadedFile = null;
- S3Import s3Import;
- Object s3PlaceHolder = jobExecution.getJobData().getProperty("s3Import");
- try {
- if (s3PlaceHolder != null) {
- s3Import = (S3Import) s3PlaceHolder;
- } else {
- s3Import = new S3ImportImpl();
- }
- } catch (Exception e) {
- tracker.fatal("Error connecting to S3: " + e.getMessage());
- checkIfComplete(rootEM, fileImport);
- return;
- }
-
- try {
- downloadedFile = s3Import.copyFileFromBucket(
- fileName, bucketName, accessId, secretKey);
- } catch (Exception e) {
- tracker.fatal("Error downloading file: " + e.getMessage());
- checkIfComplete(rootEM, fileImport);
- return;
- }
-
- // parse JSON data, create Entities and Connections from import data
-
- try {
- parseEntitiesAndConnectionsFromJson(
- jobExecution, downloadedFile, targetEm, rootEM, fileImport, tracker);
-
- } catch (Exception e) {
- tracker.fatal(e.getMessage());
- }
-
- checkIfComplete(rootEM, fileImport);
- }
-
-
- private Import getImportEntity(final EntityManager rootEm, final FileImport fileImport) {
- try {
- Results importJobResults =
- rootEm.getSourceEntities(fileImport, IMPORT_FILE_INCLUDES_CONNECTION,
- null, Level.ALL_PROPERTIES);
-
- List<Entity> importEntities = importJobResults.getEntities();
- final Import importEntity = (Import) importEntities.get(0).toTypedEntity();
- return importEntity;
- } catch (Exception e) {
- throw new RuntimeException("Unable to import entity");
- }
- }
-
- /**
- * Check if we're the last job on failure
- */
- private void checkIfComplete(final EntityManager rootEM, final FileImport fileImport) {
- int failCount = 0;
- int successCount = 0;
-
- final Import importEntity = getImportEntity(rootEM, fileImport);
-
- try {
-
- // wait for query index to catch up
-
- // TODO: better way to wait for indexes to catch up
- try {
- Thread.sleep(5000);
- } catch (Exception intentionallyIgnored) {
- // intentionally ignored
- }
-
- // get file import entities for this import job
-
- Query query = new Query();
- query.setEntityType(Schema.getDefaultSchema().getEntityType(FileImport.class));
- query.setConnectionType(IMPORT_FILE_INCLUDES_CONNECTION);
- query.setLimit(MAX_FILE_IMPORTS);
-
- Results entities = rootEM.searchTargetEntities(importEntity, query);
- PagingResultsIterator itr = new PagingResultsIterator(entities);
-
- if (!itr.hasNext()) {
- logger.warn("Found no FileImport entities for import {}, unable to check if complete",
- importEntity.getUuid());
- return;
- }
-
- if (logger.isTraceEnabled()) {
- logger.trace("Checking {} file import jobs to see if we are done for file {}",
- new Object[]{entities.size(), fileImport.getFileName()});
- }
-
- // loop through entities, count different types of status
-
- while (itr.hasNext()) {
- FileImport fi = (FileImport) itr.next();
-
- switch (fi.getState()) {
- case FAILED: // failed, but we may not be complete so continue checking
- failCount++;
- break;
- case FINISHED: // finished, we can continue checking
- successCount++;
- continue;
- default: // not something we recognize as complete, short circuit
- if (logger.isDebugEnabled()) {
- logger.debug("not done yet, bail out... {}", fi.getState().toString());
- }
- return;
- }
- }
- } catch (Exception e) {
- failCount++;
- if (importEntity != null) {
- importEntity.setErrorMessage("Error determining status of file import jobs");
- }
- if (logger.isDebugEnabled()) {
- logger.debug("Error determining status of file import jobs", e);
- }
- }
-
- if (logger.isTraceEnabled()) {
- logger.trace("successCount = {} failCount = {}", successCount, failCount);
- }
-
- if (importEntity != null) {
- if (logger.isTraceEnabled()) {
- logger.trace("FINISHED");
- }
-
- if (failCount == 0) {
- importEntity.setState(Import.State.FINISHED);
- } else {
- // we had failures, set it to failed
- importEntity.setState(Import.State.FAILED);
- }
-
- try {
- rootEM.update(importEntity);
- if (logger.isTraceEnabled()) {
- logger.trace("Updated import entity {}:{} with state {}",
- importEntity.getType(), importEntity.getUuid(), importEntity.getState());
- }
- } catch (Exception e) {
- logger.error("Error updating import entity", e);
- }
- }
-
-
- }
-
-
- /**
- * Gets the JSON parser for given file
- *
- * @param collectionFile the file for which JSON parser is required
- */
- private JsonParser getJsonParserForFile(File collectionFile) throws Exception {
- JsonParser jp = jsonFactory.createJsonParser(collectionFile);
- jp.setCodec(new ObjectMapper());
- return jp;
- }
-
-
- /**
- * Imports the entity's connecting references (collections, connections and dictionaries)
- *
- * @param execution The job jobExecution currently running
- * @param file The file to be imported
- * @param em Entity Manager for the application being imported
- * @param rootEm Entity manager for the root applicaition
- * @param fileImport The file import entity
- */
- private void parseEntitiesAndConnectionsFromJson(
- final JobExecution execution,
- final File file,
- final EntityManager em,
- final EntityManager rootEm,
- final FileImport fileImport,
- final FileImportTracker tracker) throws Exception {
-
-
- // tracker flushes every 100 entities
- //final FileImportTracker tracker = new FileImportTracker( emf, fileImport, 100 );
-
- // function to execute for each write event
- final Action1<WriteEvent> doWork = new Action1<WriteEvent>() {
- @Override
- public void call(WriteEvent writeEvent) {
- writeEvent.doWrite(em, fileImport, tracker);
- }
- };
-
- // invokes the heartbeat every HEARTBEAT_COUNT operations
- final Func2<Integer, WriteEvent, Integer> heartbeatReducer = new Func2<Integer, WriteEvent, Integer>() {
- @Override
- public Integer call(final Integer integer, final WriteEvent writeEvent) {
- final int next = integer.intValue() + 1;
- if (next % HEARTBEAT_COUNT == 0) {
- execution.heartbeat();
- }
- return next;
- }
- };
-
-
- // FIRST PASS: import all entities in the file
-
-
- // observable that parses JSON and emits write events
- JsonParser jp = getJsonParserForFile(file);
-
- // TODO: move JSON parser into observable creation so open/close happens within the stream
- // entitiesOnly = true
- final JsonEntityParserObservable jsonObservableEntities =
- new JsonEntityParserObservable(jp, em, rootEm, fileImport, tracker, true);
-
- final Observable<WriteEvent> entityEventObservable = Observable.create(jsonObservableEntities);
-
- // only take while our stats tell us we should continue processing
- // potentially skip the first n if this is a resume operation
- final int entityNumSkip = (int) tracker.getTotalEntityCount();
-
-
- entityEventObservable.takeWhile(writeEvent -> !tracker.shouldStopProcessingEntities()).skip(entityNumSkip)
- .flatMap(writeEvent -> {
- return Observable.just(writeEvent).doOnNext(doWork);
- }, 10).reduce(0, heartbeatReducer).toBlocking().last();
-
-
- jp.close();
-
- if (FileImport.State.FAILED.equals(fileImport.getState())) {
- if (logger.isTraceEnabled()) {
- logger.trace("\n\nFailed to completely write entities, skipping second phase. File: {}\n",
- fileImport.getFileName());
- }
- return;
- }
- if (logger.isTraceEnabled()) {
- logger.trace("\n\nWrote entities. File: {}\n", fileImport.getFileName());
- }
-
-
- // SECOND PASS: import all connections and dictionaries
-
-
- // observable that parses JSON and emits write events
- jp = getJsonParserForFile(file);
-
- // TODO: move JSON parser into observable creation so open/close happens within the stream
- // entitiesOnly = false
- final JsonEntityParserObservable jsonObservableOther =
- new JsonEntityParserObservable(jp, em, rootEm, fileImport, tracker, false);
-
- final Observable<WriteEvent> otherEventObservable = Observable.create(jsonObservableOther);
-
- // only take while our stats tell us we should continue processing
- // potentially skip the first n if this is a resume operation
- final int connectionNumSkip = (int) tracker.getTotalConnectionCount();
-
- // with this code we get asynchronous behavior and testImportWithMultipleFiles will fail
- final int connectionCount = otherEventObservable.takeWhile(
- writeEvent -> !tracker.shouldStopProcessingConnections()).skip(connectionNumSkip).flatMap(entityWrapper -> {
- return Observable.just(entityWrapper).doOnNext(doWork).subscribeOn(Schedulers.io());
-
- }, 10).reduce(0, heartbeatReducer).toBlocking().last();
-
- jp.close();
-
- if (logger.isTraceEnabled()) {
- logger.trace("\n\nparseEntitiesAndConnectionsFromJson(): Wrote others for file {}\n",
- fileImport.getFileName());
- }
-
- if (FileImport.State.FAILED.equals(fileImport.getState())) {
- if (logger.isDebugEnabled()) {
- logger.debug("\n\nparseEntitiesAndConnectionsFromJson(): failed to completely write entities\n");
- }
- return;
- }
-
- // flush the job statistics
- tracker.complete();
-
- if (FileImport.State.FAILED.equals(fileImport.getState())) {
- if (logger.isDebugEnabled()) {
- logger.debug("\n\nFailed to completely write connections and dictionaries. File: {}\n",
- fileImport.getFileName());
- }
- return;
- }
-
- if (logger.isTraceEnabled()) {
- logger.trace("\n\nWrote connections and dictionaries. File: {}\n", fileImport.getFileName());
- }
- }
-
-
- private interface WriteEvent {
- public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker);
- }
-
-
- private final class EntityEvent implements WriteEvent {
- UUID entityUuid;
- String entityType;
- Map<String, Object> properties;
-
- EntityEvent(UUID entityUuid, String entityType, Map<String, Object> properties) {
- this.entityUuid = entityUuid;
- this.entityType = entityType;
- this.properties = properties;
- }
-
-
- // Creates entities
- @Override
- public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker) {
- try {
- if (logger.isTraceEnabled()) {
- logger.trace("Writing imported entity {}:{} into app {}",
- entityType, entityUuid, em.getApplication().getUuid());
- }
-
- em.create(entityUuid, entityType, properties);
-
- tracker.entityWritten();
-
- } catch (Exception e) {
- logger.error("Error writing entity. From file:{}", fileImport.getFileName(), e);
-
- tracker.entityFailed(e.getMessage() + " From file: " + fileImport.getFileName());
- }
- }
- }
-
-
- private final class ConnectionEvent implements WriteEvent {
- EntityRef ownerEntityRef;
- String connectionType;
- EntityRef entityRef;
-
- ConnectionEvent(EntityRef ownerEntityRef, String connectionType, EntityRef entryRef) {
- this.ownerEntityRef = ownerEntityRef;
- this.connectionType = connectionType;
- this.entityRef = entryRef;
- }
-
- // creates connections between entities
- @Override
- public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker tracker) {
-
- try {
- // TODO: do we need to ensure that all Entity events happen first?
- // TODO: what happens if ConnectionEvents happen before all entities are saved?
-
- // Connections are specified as UUIDs with no type
- if (entityRef.getType() == null) {
- entityRef = em.get(ownerEntityRef.getUuid());
- }
-
- if (logger.isTraceEnabled()) {
- logger.trace("Creating connection from {}:{} to {}:{}",
- ownerEntityRef.getType(), ownerEntityRef.getUuid(),
- entityRef.getType(), entityRef.getUuid());
- }
-
- em.createConnection(ownerEntityRef, connectionType, entityRef);
-
- tracker.connectionWritten();
-
- } catch (Exception e) {
- logger.error("Error writing connection. From file: {}", fileImport.getFileName(), e);
-
- tracker.connectionFailed(e.getMessage() + " From file: " + fileImport.getFileName());
- }
- }
- }
-
-
- private final class DictionaryEvent implements WriteEvent {
-
- EntityRef ownerEntityRef;
- String dictionaryName;
- Map<String, Object> dictionary;
-
- DictionaryEvent(EntityRef ownerEntityRef, String dictionaryName, Map<String, Object> dictionary) {
- this.ownerEntityRef = ownerEntityRef;
- this.dictionaryName = dictionaryName;
- this.dictionary = dictionary;
- }
-
- // adds map to the dictionary
- @Override
- public void doWrite(EntityManager em, FileImport fileImport, FileImportTracker stats) {
- try {
-
- if (logger.isTraceEnabled()) {
- logger.trace("Adding map to {}:{} dictionary {}",
- ownerEntityRef.getType(), ownerEntityRef.getType(), dictionaryName);
- }
-
- em.addMapToDictionary(ownerEntityRef, dictionaryName, dictionary);
-
- } catch (Exception e) {
- logger.error("Error writing dictionary. From file: " + fileImport.getFileName(), e);
-
- // TODO add statistics for dictionary writes and failures
- }
- }
- }
-
-
- private final class JsonEntityParserObservable implements Observable.OnSubscribe<WriteEvent> {
- public static final String COLLECTION_OBJECT_NAME = "collections";
- private final JsonParser jp;
- EntityManager em;
- EntityManager rootEm;
- FileImport fileImport;
- FileImportTracker tracker;
- boolean entitiesOnly;
-
-
- JsonEntityParserObservable(
- JsonParser parser,
- EntityManager em,
- EntityManager rootEm,
- FileImport fileImport,
- FileImportTracker tracker,
- boolean entitiesOnly) {
-
- this.jp = parser;
- this.em = em;
- this.rootEm = rootEm;
- this.fileImport = fileImport;
- this.tracker = tracker;
- this.entitiesOnly = entitiesOnly;
- }
-
-
- @Override
- public void call(final Subscriber<? super WriteEvent> subscriber) {
- process(subscriber);
- }
-
-
- private void process(final Subscriber<? super WriteEvent> subscriber) {
-
- try {
-
- // we ignore imported entity type information, entities get the type of the collection
- Stack<JsonToken> objectStartStack = new Stack();
- Stack<String> objectNameStack = new Stack();
- EntityRef lastEntity = null;
-
- String entityType = null;
-
- while (true) {
-
- JsonToken token = jp.nextToken();
-
- // nothing left to do.
- if (token == null) {
- break;
- }
-
- String name = jp.getCurrentName();
-
-
- // start of an object with a field name
-
- if (token.equals(JsonToken.START_OBJECT)) {
-
- objectStartStack.push(token);
-
- // nothing to do
- if (name == null) {
- continue;
- }
-
-
- if ("Metadata".equals(name)) {
-
- Map<String, Object> entityMap = jp.readValueAs(HashMap.class);
-
- UUID uuid = null;
- if (entityMap.get("uuid") != null) {
- uuid = UUID.fromString((String) entityMap.get("uuid"));
- lastEntity = new SimpleEntityRef(entityType, uuid);
- }
-
- if (entitiesOnly) {
- //logger.debug("{}Got entity with uuid {}", indent, lastEntity);
-
- WriteEvent event = new EntityEvent(uuid, entityType, entityMap);
- processWriteEvent(subscriber, event);
- }
- objectStartStack.pop();
- } else if ("connections".equals(name)) {
-
- Map<String, Object> connectionMap = jp.readValueAs(HashMap.class);
-
- for (String type : connectionMap.keySet()) {
- List targets = (List) connectionMap.get(type);
-
- for (Object targetObject : targets) {
- UUID target = UUID.fromString((String) targetObject);
-
- if (!entitiesOnly) {
- //logger.debug("{}Got connection {} to {}",
- //new Object[]{indent, type, target.toString()});
-
- EntityRef entryRef = new SimpleEntityRef(target);
- WriteEvent event = new ConnectionEvent(lastEntity, type, entryRef);
- processWriteEvent(subscriber, event);
- }
- }
- }
-
- objectStartStack.pop();
-
- } else if ("dictionaries".equals(name)) {
-
- Map<String, Object> dictionariesMap = jp.readValueAs(HashMap.class);
- for (String dname : dictionariesMap.keySet()) {
- Map dmap = (Map) dictionariesMap.get(dname);
-
- if (!entitiesOnly) {
- //logger.debug("{}Got dictionary {} size {}",
- //new Object[] {indent, dname, dmap.size() });
-
- WriteEvent event = new DictionaryEvent(lastEntity, dname, dmap);
- processWriteEvent(subscriber, event);
- }
- }
-
- objectStartStack.pop();
-
- } else {
- // push onto object names we don't immediately understand. Used for parent detection
- objectNameStack.push(name);
- }
-
- } else if (token.equals(JsonToken.START_ARRAY)) {
- if (objectNameStack.size() == 1
- && COLLECTION_OBJECT_NAME.equals(objectNameStack.peek())) {
- entityType = InflectionUtils.singularize(name);
- }
-
- } else if (token.equals(JsonToken.END_OBJECT)) {
- objectStartStack.pop();
- }
- }
-
- if (subscriber != null) {
- subscriber.onCompleted();
- }
-
- if (logger.isTraceEnabled()) {
- logger.trace("process(): done parsing JSON");
- }
-
- } catch (Exception e) {
-
- tracker.fatal(e.getMessage());
-
- if (subscriber != null) {
-
- // don't need to blow up here, we handled the problem
- // if we blow up we may prevent in-flight entities from being written
- // subscriber.onError(e);
-
- // but we are done reading entities
- subscriber.onCompleted();
- }
- }
- }
-
- private void processWriteEvent(final Subscriber<? super WriteEvent> subscriber, WriteEvent writeEvent) {
-
- if (subscriber == null) {
-
- // this logic makes it easy to remove Rx for debugging purposes
- // no Rx, just do it
- writeEvent.doWrite(em, fileImport, tracker);
-
- } else {
- subscriber.onNext(writeEvent);
- }
- }
-
- }
-}
-
-
-/**
- * Custom Exception class for Organization Not Found
- */
-class OrganizationNotFoundException extends Exception {
- OrganizationNotFoundException(String s) {
- super(s);
- }
-}
-
-
-/**
- * Custom Exception class for Application Not Found
- */
-class ApplicationNotFoundException extends Exception {
- ApplicationNotFoundException(String s) {
- super(s);
- }
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java
deleted file mode 100644
index 7e09051..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3Import.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * 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.management.importer;
-
-import java.io.File;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-
-
-/**
- * Plug-able S3Import interface.
- */
-public interface S3Import {
-
- List<String> getBucketFileNames(
- String bucketName, String endsWith, String accessId, String secretKey ) throws Exception;
-
- public File copyFileFromBucket(
- String blobFileName, String bucketName, String accessId, String secretKey ) throws Exception;
-}
http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java b/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java
deleted file mode 100644
index 921de64..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/management/importer/S3ImportImpl.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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.management.importer;
-
-import com.google.common.collect.ImmutableSet;
-import com.google.inject.Module;
-import org.apache.commons.io.IOUtils;
-import org.apache.commons.lang.RandomStringUtils;
-import org.jclouds.ContextBuilder;
-import org.jclouds.blobstore.BlobStore;
-import org.jclouds.blobstore.BlobStoreContext;
-import org.jclouds.blobstore.domain.Blob;
-import org.jclouds.blobstore.domain.MutableBlobMetadata;
-import org.jclouds.blobstore.domain.PageSet;
-import org.jclouds.blobstore.domain.StorageMetadata;
-import org.jclouds.blobstore.options.ListContainerOptions;
-import org.jclouds.http.config.JavaUrlHttpCommandExecutorServiceModule;
-import org.jclouds.logging.log4j.config.Log4JLoggingModule;
-import org.jclouds.netty.config.NettyPayloadModule;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.*;
-import java.nio.file.Files;
-import java.util.*;
-
-
-public class S3ImportImpl implements S3Import {
- private static final Logger logger = LoggerFactory.getLogger(S3ImportImpl.class);
-
-
- public File copyFileFromBucket(
- String blobFileName, String bucketName, String accessId, String secretKey ) throws Exception {
-
- // setup to use JCloud BlobStore interface to AWS S3
-
- Properties overrides = new Properties();
- overrides.setProperty("s3" + ".identity", accessId);
- overrides.setProperty("s3" + ".credential", secretKey);
-
- final Iterable<? extends Module> MODULES = ImmutableSet.of(
- new JavaUrlHttpCommandExecutorServiceModule(),
- new Log4JLoggingModule(),
- new NettyPayloadModule());
-
- BlobStoreContext context = ContextBuilder.newBuilder("s3")
- .credentials(accessId, secretKey)
- .modules(MODULES)
- .overrides(overrides)
- .buildView(BlobStoreContext.class);
- BlobStore blobStore = context.getBlobStore();
-
- // get file from configured bucket, copy it to local temp file
-
- Blob blob = blobStore.getBlob(bucketName, blobFileName);
- if ( blob == null) {
- throw new RuntimeException(
- "Blob file name " + blobFileName + " not found in bucket " + bucketName );
- }
-
- FileOutputStream fop = null;
- File tempFile;
- try {
- tempFile = File.createTempFile(bucketName, RandomStringUtils.randomAlphabetic(10));
- tempFile.deleteOnExit();
- fop = new FileOutputStream(tempFile);
- InputStream is = blob.getPayload().openStream();
- IOUtils.copyLarge(is, fop);
- return tempFile;
-
- } finally {
- if ( fop != null ) {
- fop.close();
- }
- }
- }
-
-
- @Override
- public List<String> getBucketFileNames(
- String bucketName, String endsWith, String accessId, String secretKey ) {
-
- // get setup to use JCloud BlobStore interface to AWS S3
-
- Properties overrides = new Properties();
- overrides.setProperty("s3" + ".identity", accessId);
- overrides.setProperty("s3" + ".credential", secretKey);
-
- final Iterable<? extends Module> MODULES = ImmutableSet.of(
- new JavaUrlHttpCommandExecutorServiceModule(),
- new Log4JLoggingModule(),
- new NettyPayloadModule());
-
- BlobStoreContext context = ContextBuilder.newBuilder("s3")
- .credentials(accessId, secretKey)
- .modules(MODULES)
- .overrides(overrides)
- .buildView(BlobStoreContext.class);
- BlobStore blobStore = context.getBlobStore();
-
- // gets all the files in the configured bucket recursively
-
- PageSet<? extends StorageMetadata> pageSets =
- blobStore.list(bucketName, new ListContainerOptions().recursive());
-
- if (logger.isTraceEnabled()) {
- logger.trace(" Found {} files in bucket {}", pageSets.size(), bucketName);
- }
-
- List<String> blobFileNames = new ArrayList<>();
- for ( Object pageSet : pageSets ) {
- String blobFileName = ((MutableBlobMetadata)pageSet).getName();
- if ( blobFileName.endsWith( endsWith )) {
- blobFileNames.add(blobFileName);
- }
- }
-
- return blobFileNames;
- }
-
-
-}
-
http://git-wip-us.apache.org/repos/asf/usergrid/blob/c915a316/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java
----------------------------------------------------------------------
diff --git a/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java b/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java
deleted file mode 100644
index 2929d93..0000000
--- a/stack/services/src/main/java/org/apache/usergrid/services/assets/data/S3BinaryStore.java
+++ /dev/null
@@ -1,346 +0,0 @@
-/*
- * 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.services.assets.data;
-
-
-import com.google.common.collect.ImmutableSet;
-import com.google.common.hash.Hashing;
-import com.google.common.io.Files;
-import com.google.inject.Module;
-import org.apache.commons.codec.binary.Hex;
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.IOUtils;
-import org.apache.usergrid.persistence.Entity;
-import org.apache.usergrid.persistence.EntityManager;
-import org.apache.usergrid.persistence.EntityManagerFactory;
-import org.apache.usergrid.utils.StringUtils;
-import org.jclouds.ContextBuilder;
-import org.jclouds.blobstore.BlobStore;
-import org.jclouds.blobstore.BlobStoreContext;
-import org.jclouds.blobstore.domain.Blob;
-import org.jclouds.blobstore.domain.BlobBuilder;
-import org.jclouds.blobstore.options.GetOptions;
-import org.jclouds.http.config.JavaUrlHttpCommandExecutorServiceModule;
-import org.jclouds.logging.log4j.config.Log4JLoggingModule;
-import org.jclouds.netty.config.NettyPayloadModule;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-
-import java.io.*;
-import java.util.Map;
-import java.util.Properties;
-import java.util.UUID;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-public class S3BinaryStore implements BinaryStore {
-
- private static final Iterable<? extends Module> MODULES = ImmutableSet
- .of( new JavaUrlHttpCommandExecutorServiceModule(), new Log4JLoggingModule(), new NettyPayloadModule() );
-
- private static final Logger logger = LoggerFactory.getLogger( S3BinaryStore.class );
- private static final long FIVE_MB = ( FileUtils.ONE_MB * 5 );
- private static String WORKERS_PROP_NAME = "usergrid.binary.upload-workers";
-
- private BlobStoreContext context;
- private String accessId;
- private String secretKey;
- private String bucketName;
- private ExecutorService executorService;
-
- @Autowired
- private Properties properties;
-
- @Autowired
- private EntityManagerFactory emf;
-
-
- public S3BinaryStore( String accessId, String secretKey, String bucketName ) {
- this.accessId = accessId;
- this.secretKey = secretKey;
- this.bucketName = bucketName;
- }
-
-
- private BlobStoreContext getContext() {
- if ( context == null ) {
- context = ContextBuilder.newBuilder( "aws-s3" ).credentials( accessId, secretKey ).modules( MODULES )
- .buildView( BlobStoreContext.class );
-
- BlobStore blobStore = context.getBlobStore();
- blobStore.createContainerInLocation( null, bucketName );
- }
-
- return context;
- }
-
-
- public void destroy() {
- if ( context != null ) {
- context.close();
- }
- }
-
-
- @Override
- public void write( final UUID appId, final Entity entity, InputStream inputStream ) throws IOException {
-
- // write up to 5mb of data to an byte array
-
- ByteArrayOutputStream baos = new ByteArrayOutputStream();
- long written = IOUtils.copyLarge( inputStream, baos, 0, FIVE_MB );
- byte[] data = baos.toByteArray();
-
- if ( written < FIVE_MB ) { // total smaller than 5mb
-
- final String uploadFileName = AssetUtils.buildAssetKey( appId, entity );
- final String mimeType = AssetMimeHandler.get().getMimeType( entity, data );
-
- final Map<String, Object> fileMetadata = AssetUtils.getFileMetadata( entity );
- fileMetadata.put( AssetUtils.LAST_MODIFIED, System.currentTimeMillis() );
-
- BlobStore blobStore = getContext().getBlobStore();
-
- // need this for JClouds 1.7.x:
-// BlobBuilder.PayloadBlobBuilder bb = blobStore.blobBuilder( uploadFileName )
-// .payload( data ).calculateMD5().contentType( mimeType );
-
- // need this for JClouds 1.8.x:
- BlobBuilder.PayloadBlobBuilder bb = blobStore.blobBuilder(uploadFileName)
- .payload( data )
- .contentMD5( Hashing.md5().newHasher().putBytes( data ).hash() )
- .contentType( mimeType );
-
- fileMetadata.put( AssetUtils.CONTENT_LENGTH, written );
- if ( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ) != null ) {
- bb.contentDisposition( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ).toString() );
- }
- final Blob blob = bb.build();
-
- String md5sum = Hex.encodeHexString( blob.getMetadata().getContentMetadata().getContentMD5() );
- fileMetadata.put( AssetUtils.CHECKSUM, md5sum );
-
- String eTag = blobStore.putBlob( bucketName, blob );
- fileMetadata.put( AssetUtils.E_TAG, eTag );
- }
- else { // bigger than 5mb... dump 5 mb tmp files and upload from them
-
- ExecutorService executors = getExecutorService();
-
- executors.submit( new UploadWorker( appId, entity, inputStream, data, written ) );
- }
- }
-
- private ExecutorService getExecutorService() {
-
- if ( executorService == null ) {
- synchronized (this) {
-
- int workers = 40;
- String workersString = properties.getProperty( WORKERS_PROP_NAME, "40");
-
- if ( StringUtils.isNumeric( workersString ) ) {
- workers = Integer.parseInt( workersString );
- } else if ( !StringUtils.isEmpty( workersString )) {
- logger.error("Ignoring invalid setting for {}", WORKERS_PROP_NAME);
- }
- executorService = Executors.newFixedThreadPool( workers );
- }
- }
-
- return executorService;
- }
-
-
- @Override
- public InputStream read( UUID appId, Entity entity, long offset, long length ) throws IOException {
- BlobStore blobStore = getContext().getBlobStore();
- Blob blob;
- if ( offset == 0 && length == FIVE_MB ) {
- blob = blobStore.getBlob( bucketName, AssetUtils.buildAssetKey( appId, entity ) );
- }
- else {
- GetOptions options = GetOptions.Builder.range( offset, length );
- blob = blobStore.getBlob( bucketName, AssetUtils.buildAssetKey( appId, entity ), options );
- }
- if ( blob == null || blob.getPayload() == null ) {
- return null;
- }
- return blob.getPayload().getInput();
- }
-
-
- @Override
- public InputStream read( UUID appId, Entity entity ) throws IOException {
- return read( appId, entity, 0, FIVE_MB );
- }
-
-
- @Override
- public void delete( UUID appId, Entity entity ) {
- BlobStore blobStore = getContext().getBlobStore();
- blobStore.removeBlob( bucketName, AssetUtils.buildAssetKey( appId, entity ) );
- }
-
- class UploadWorker implements Callable<Void> {
-
- private UUID appId;
- private Entity entity;
- private InputStream inputStream;
- private byte[] data;
- private long written;
-
-
- public UploadWorker( UUID appId, Entity entity, InputStream is, byte[] data, long written ) {
- this.appId = appId;
- this.entity = entity;
- this.inputStream = is;
- this.data = data;
- this.written = written;
- }
-
- @Override
- public Void call() {
-
- if (logger.isTraceEnabled()) {
- logger.trace("Writing temp file for S3 upload");
- }
-
- // determine max size file allowed, default to 50mb
- long maxSizeBytes = 50 * FileUtils.ONE_MB;
- String maxSizeMbString = properties.getProperty( "usergrid.binary.max-size-mb", "50" );
- if (StringUtils.isNumeric( maxSizeMbString )) {
- maxSizeBytes = Long.parseLong( maxSizeMbString ) * FileUtils.ONE_MB;
- }
-
- // always allow files up to 5mb
- if (maxSizeBytes < 5 * FileUtils.ONE_MB ) {
- maxSizeBytes = 5 * FileUtils.ONE_MB;
- }
-
- // write temporary file, slightly larger than our size limit
- OutputStream os = null;
- File tempFile;
- try {
- tempFile = File.createTempFile( entity.getUuid().toString(), "tmp" );
- tempFile.deleteOnExit();
- os = new BufferedOutputStream( new FileOutputStream( tempFile.getAbsolutePath() ) );
- os.write( data );
- written += data.length;
- written += IOUtils.copyLarge( inputStream, os, 0, maxSizeBytes + 1 );
-
- if (logger.isTraceEnabled()) {
- logger.trace("Write temp file {} length {}", tempFile.getName(), written);
- }
-
- } catch ( IOException e ) {
- throw new RuntimeException( "Error creating temp file", e );
-
- } finally {
- if ( os != null ) {
- try {
- os.flush();
- } catch (IOException e) {
- logger.error( "Error flushing data to temporary upload file", e );
- }
- IOUtils.closeQuietly( os );
- }
- }
-
- // if tempFile is too large, delete it, add error to entity file metadata and abort
-
- Map<String, Object> fileMetadata = AssetUtils.getFileMetadata( entity );
-
- if ( tempFile.length() > maxSizeBytes ) {
- if (logger.isDebugEnabled()) {
- logger.debug("File too large. Temp file size (bytes) = {}, Max file size (bytes) = {}",
- tempFile.length(), maxSizeBytes);
- }
- try {
- EntityManager em = emf.getEntityManager( appId );
- fileMetadata.put( "error", "Asset size " + tempFile.length()
- + " is larger than max size of " + maxSizeBytes );
- em.update( entity );
- tempFile.delete();
-
- } catch ( Exception e ) {
- logger.error( "Error updating entity with error message", e);
- }
- return null;
- }
-
- String uploadFileName = AssetUtils.buildAssetKey( appId, entity );
- String mimeType = AssetMimeHandler.get().getMimeType( entity, data );
-
- try { // start the upload
-
- if (logger.isTraceEnabled()) {
- logger.trace("S3 upload thread started");
- }
-
- BlobStore blobStore = getContext().getBlobStore();
-
- // need this for JClouds 1.7.x:
-// BlobBuilder.PayloadBlobBuilder bb = blobStore.blobBuilder( uploadFileName )
-// .payload( tempFile ).calculateMD5().contentType( mimeType );
-
- // need this for JClouds 1.8.x:
- BlobBuilder.PayloadBlobBuilder bb = blobStore.blobBuilder( uploadFileName )
- .payload( tempFile )
- .contentMD5( Files.hash( tempFile, Hashing.md5() ) )
- .contentType( mimeType );
-
- if ( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ) != null ) {
- bb.contentDisposition( fileMetadata.get( AssetUtils.CONTENT_DISPOSITION ).toString() );
- }
- final Blob blob = bb.build();
-
- String md5sum = Hex.encodeHexString( blob.getMetadata().getContentMetadata().getContentMD5() );
- fileMetadata.put( AssetUtils.CHECKSUM, md5sum );
-
- if (logger.isTraceEnabled()) {
- logger.trace("S3 upload starting");
- }
-
- String eTag = blobStore.putBlob( bucketName, blob );
-
- if (logger.isTraceEnabled()) {
- logger.trace("S3 upload complete eTag=" + eTag);
- }
-
- // update entity with eTag
- EntityManager em = emf.getEntityManager( appId );
- fileMetadata.put( AssetUtils.LAST_MODIFIED, System.currentTimeMillis() );
- fileMetadata.put( AssetUtils.CONTENT_LENGTH, written );
- fileMetadata.put( AssetUtils.E_TAG, eTag );
- em.update( entity );
- }
- catch ( Exception e ) {
- logger.error( "error uploading", e );
- }
-
- if ( tempFile != null && tempFile.exists() ) {
- tempFile.delete();
- }
-
- return null;
- }
- }
-}
-