@Test public void create_snapshot_test_1() throws Exception { StorageInfo storageInfo = new StorageInfo(); storageInfo.setDefaults(); storageInfo.setShouldTransferSnapshots(Boolean.FALSE); storageInfo.setMaxSnapshotDeltas(10); try (TransactionResource tran = Entities.transactionFor(StorageInfo.class)) { Entities.persist(storageInfo); tran.commit(); } VolumeInfo vol = new VolumeInfo("vol-0000001"); vol.setCreateTime(new Date(System.currentTimeMillis() - TimeUnit.MILLISECONDS.convert(1, TimeUnit.HOURS))); vol.setSize(1); vol.setStatus(StorageProperties.Status.available.toString()); try (TransactionResource tran = Entities.transactionFor(VolumeInfo.class)) { Entities.persist(vol); tran.commit(); } SnapshotInfo snap = new SnapshotInfo("snap-0000001"); snap.setIsOrigin(Boolean.TRUE); snap.setProgress("0"); snap.setSizeGb(vol.getSize()); snap.setSnapPointId("snap-point-id-1"); snap.setStartTime(new Date(System.currentTimeMillis() - TimeUnit.MILLISECONDS.convert(20, TimeUnit.MINUTES))); snap.setStatus(StorageProperties.Status.creating.toString()); snap.setVolumeId(vol.getVolumeId()); try (TransactionResource tran = Entities.transactionFor(SnapshotInfo.class)) { Entities.persist(snap); tran.commit(); } final LogicalStorageManager storageManager = context.mock(LogicalStorageManager.class); context.checking(new Expectations() { { oneOf(storageManager).createSnapshot(vol.getVolumeId(), snap.getSnapshotId(), snap.getSnapPointId()); oneOf(storageManager).finishVolume(snap.getSnapshotId()); } }); new SnapshotCreator(vol.getVolumeId(), snap.getSnapshotId(), snap.getSnapPointId(), storageManager, BlockStorageUnitTestSupport.createMockS3SnapshotTransfer(), BlockStorageUnitTestSupport.createMockSnapshotProgressCallback()).run(); context.assertIsSatisfied(); List<SnapshotInfo> snaps; try (TransactionResource tran = Entities.transactionFor(SnapshotInfo.class)) { snaps = Entities.query(new SnapshotInfo()); tran.commit(); } assertTrue("expected to have a result set querying the snapshots table in eucalyptus_storage persistence context", snaps != null); assertTrue("expected 1 SnapshotInfo entity to exist, but found " + snaps.size(), snaps.size() == 1); assertTrue("expected SnapshotInfo entity for " + snap.getSnapshotId() + " but found entity for " + snaps.get(0).getSnapshotId(), snap.getSnapshotId().equals(snaps.get(0).getSnapshotId())); assertTrue("expected SnapshotInfo entity status to be " + StorageProperties.Status.available.toString() + " but found it to be " + snaps.get(0).getStatus(), StorageProperties.Status.available.toString().equals(snaps.get(0).getStatus())); assertTrue("expected SnapshotInfo entity progress to be 100% but found it to be " + snaps.get(0).getProgress(), "100".equals(snaps.get(0).getProgress())); assertTrue("expected SnapshotInfo entity to contain a null reference for snapshotLocation but found " + snaps.get(0).getSnapshotLocation(), null == snaps.get(0).getSnapshotLocation()); assertTrue("expected SnapshotInfo entity to contain null reference for previousSnapshotID but found " + snaps.get(0).getPreviousSnapshotId(), null == snaps.get(0).getPreviousSnapshotId()); }
@Override public void run() { LOG.trace("Starting SnapshotCreator task"); EucaSemaphore semaphore = EucaSemaphoreDirectory.getSolitarySemaphore(this.volumeId); try { Boolean shouldTransferSnapshots = true; String bucket = null; shouldTransferSnapshots = StorageInfo.getStorageInfo().getShouldTransferSnapshots(); if (shouldTransferSnapshots) { if (null == snapshotTransfer) { snapshotTransfer = new S3SnapshotTransfer(this.snapshotId, this.snapshotId); } bucket = snapshotTransfer.prepareForUpload(); if (snapshotTransfer == null || StringUtils.isBlank(bucket)) { throw new EucalyptusCloudException("Failed to initialize snapshot transfer mechanism for uploading " + this.snapshotId); } } try { try { semaphore.acquire(); } catch (InterruptedException ex) { throw new EucalyptusCloudException("Failed to create snapshot " + this.snapshotId + " as the semaphore could not be acquired"); } if (!isSnapshotMarkedFailed(this.snapshotId)) { if (null == progressCallback) { progressCallback = new SnapshotProgressCallback(this.snapshotId); } blockManager.createSnapshot(this.volumeId, this.snapshotId, this.snapPointId); progressCallback.updateBackendProgress(50); } else { throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed by another thread"); } } finally { semaphore.release(); } SnapshotInfo prevSnap = null; SnapshotInfo currSnap = null; Future<String> uploadFuture = null; if (!isSnapshotMarkedFailed(this.snapshotId)) { if (shouldTransferSnapshots) { String snapshotLocation = SnapshotInfo.generateSnapshotLocationURI(SnapshotTransferConfiguration.OSG, bucket, this.snapshotId); try { Integer maxDeltaLimit = StorageInfo.getStorageInfo().getMaxSnapshotDeltas(); maxDeltaLimit = maxDeltaLimit != null ? maxDeltaLimit : 0; if (maxDeltaLimit > 0 && blockManager.supportsIncrementalSnapshots()) { LOG.debug("EBS backend supports incremental snapshots"); int attempts = 0; do { attempts++; prevSnap = fetchPreviousSnapshot(maxDeltaLimit); if (prevSnap != null) { EucaSemaphore prevSnapSemaphore = EucaSemaphoreDirectory.getSolitarySemaphore(prevSnap.getSnapshotId()); try { try { prevSnapSemaphore.acquire(); } catch (InterruptedException ex) { LOG.warn("Cannot update metadata for snapshot " + this.snapshotId + " due to an error acquiring semaphore for a previous snapshot " + prevSnap.getSnapshotId() + ". May retry again later"); continue; } currSnap = updateSnapshotInfo(prevSnap.getSnapshotId(), snapshotLocation); } finally { prevSnapSemaphore.release(); } } else { currSnap = updateSnapshotInfo(snapshotLocation); } } while (currSnap == null && attempts < 10); } else { LOG.debug("Either EBS backend does not support incremental snapshots or the feature is disabled"); currSnap = updateSnapshotInfo(snapshotLocation); } if (currSnap == null) { LOG.warn("Failed to update metadata for snapshot " + this.snapshotId); throw new EucalyptusCloudException("Failed to update metadata for snapshot " + this.snapshotId); } } catch (EucalyptusCloudException e) { throw e; } catch (Exception e) { LOG.warn("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); throw new EucalyptusCloudException("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); } StorageResourceWithCallback srwc = null; StorageResource snapshotResource = null; if (prevSnap != null) { LOG.info("Generate delta between penultimate snapshot " + prevSnap.getSnapshotId() + " and latest snapshot " + this.snapshotId); srwc = blockManager.prepIncrementalSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId, prevSnap.getSnapshotId(), prevSnap.getSnapPointId()); snapshotResource = srwc.getSr(); } else { LOG.info("Upload entire content of snapshot " + this.snapshotId); snapshotResource = blockManager.prepSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId); } if (snapshotResource == null) { LOG.warn("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); throw new EucalyptusCloudException("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); } try { uploadFuture = snapshotTransfer.upload(snapshotResource, progressCallback); } catch (Exception e) { throw new EucalyptusCloudException("Failed to upload snapshot " + this.snapshotId + " to objectstorage", e); } finally { if (srwc != null && srwc.getCallback() != null) { blockManager.executeCallback(srwc.getCallback(), srwc.getSr()); } } } else { LOG.debug("Snapshot uploads are disabled, skipping upload step for " + this.snapshotId); } } else { LOG.warn("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); } try { LOG.debug("Finishing up " + this.snapshotId + " on block storage backend"); blockManager.finishVolume(this.snapshotId); LOG.info("Finished creating " + this.snapshotId + " on block storage backend"); progressCallback.updateBackendProgress(50); } catch (EucalyptusCloudException ex) { LOG.warn("Failed to complete snapshot " + this.snapshotId + " on backend", ex); throw ex; } if (uploadFuture != null) { LOG.debug("Waiting for upload of " + this.snapshotId + " to complete"); if (uploadFuture.get() != null) { LOG.info("Uploaded " + this.snapshotId + " to object storage gateway, etag result - " + uploadFuture.get()); } else { LOG.warn("Failed to upload " + this.snapshotId + " to object storage gateway failed. Check prior logs for exact errors"); throw new EucalyptusCloudException("Failed to upload " + this.snapshotId + " to object storage gateway"); } } if (prevSnap != null) { final int MILLIS_PER_HOUR = 60 * 60 * 1000; final int timeoutMillis = StorageInfo.getStorageInfo().getSnapshotTransferTimeoutInHours() * MILLIS_PER_HOUR; final int pollPeriodMillis = 10000; int waitTimeSoFar = 0; while (waitTimeSoFar < timeoutMillis) { try { if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { break; } } catch (NoSuchElementException nsee) { LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " no longer recorded in the database. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw nsee; } catch (TransactionException te) { LOG.error("General database error. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw te; } try { Thread.sleep(pollPeriodMillis); } catch (InterruptedException e) { throw new RuntimeException("Interrupted while waiting for previous snapshot to complete", e); } waitTimeSoFar += pollPeriodMillis; } if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { if (isSnapshotMarkedAvailable(prevSnap.getSnapshotId())) { markSnapshotAvailable(); } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " finalized into the " + prevSnap.getStatus() + " state. This snapshot delta " + this.snapshotId + " set to 'failed' state because it requires an intact previous snapshot."); } } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " never finalized after " + "waiting " + (timeoutMillis / MILLIS_PER_HOUR) + " hours. " + "This snapshot delta " + this.snapshotId + " set to 'failed' state because " + "it requires an intact previous snapshot."); } } else { markSnapshotAvailable(); LOG.debug("Snapshot " + this.snapshotId + " set to 'available' state"); } } catch (Exception ex) { LOG.error("Failed to create snapshot " + this.snapshotId, ex); try { markSnapshotFailed(); LOG.debug("Snapshot " + this.snapshotId + " set to 'failed' state"); } catch (TransactionException | NoSuchElementException e) { LOG.warn("Cannot update snapshot " + this.snapshotId + " status to 'failed' in DB", e); } } LOG.trace("Finished SnapshotCreator task"); }
SnapshotCreator implements Runnable { @Override public void run() { LOG.trace("Starting SnapshotCreator task"); EucaSemaphore semaphore = EucaSemaphoreDirectory.getSolitarySemaphore(this.volumeId); try { Boolean shouldTransferSnapshots = true; String bucket = null; shouldTransferSnapshots = StorageInfo.getStorageInfo().getShouldTransferSnapshots(); if (shouldTransferSnapshots) { if (null == snapshotTransfer) { snapshotTransfer = new S3SnapshotTransfer(this.snapshotId, this.snapshotId); } bucket = snapshotTransfer.prepareForUpload(); if (snapshotTransfer == null || StringUtils.isBlank(bucket)) { throw new EucalyptusCloudException("Failed to initialize snapshot transfer mechanism for uploading " + this.snapshotId); } } try { try { semaphore.acquire(); } catch (InterruptedException ex) { throw new EucalyptusCloudException("Failed to create snapshot " + this.snapshotId + " as the semaphore could not be acquired"); } if (!isSnapshotMarkedFailed(this.snapshotId)) { if (null == progressCallback) { progressCallback = new SnapshotProgressCallback(this.snapshotId); } blockManager.createSnapshot(this.volumeId, this.snapshotId, this.snapPointId); progressCallback.updateBackendProgress(50); } else { throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed by another thread"); } } finally { semaphore.release(); } SnapshotInfo prevSnap = null; SnapshotInfo currSnap = null; Future<String> uploadFuture = null; if (!isSnapshotMarkedFailed(this.snapshotId)) { if (shouldTransferSnapshots) { String snapshotLocation = SnapshotInfo.generateSnapshotLocationURI(SnapshotTransferConfiguration.OSG, bucket, this.snapshotId); try { Integer maxDeltaLimit = StorageInfo.getStorageInfo().getMaxSnapshotDeltas(); maxDeltaLimit = maxDeltaLimit != null ? maxDeltaLimit : 0; if (maxDeltaLimit > 0 && blockManager.supportsIncrementalSnapshots()) { LOG.debug("EBS backend supports incremental snapshots"); int attempts = 0; do { attempts++; prevSnap = fetchPreviousSnapshot(maxDeltaLimit); if (prevSnap != null) { EucaSemaphore prevSnapSemaphore = EucaSemaphoreDirectory.getSolitarySemaphore(prevSnap.getSnapshotId()); try { try { prevSnapSemaphore.acquire(); } catch (InterruptedException ex) { LOG.warn("Cannot update metadata for snapshot " + this.snapshotId + " due to an error acquiring semaphore for a previous snapshot " + prevSnap.getSnapshotId() + ". May retry again later"); continue; } currSnap = updateSnapshotInfo(prevSnap.getSnapshotId(), snapshotLocation); } finally { prevSnapSemaphore.release(); } } else { currSnap = updateSnapshotInfo(snapshotLocation); } } while (currSnap == null && attempts < 10); } else { LOG.debug("Either EBS backend does not support incremental snapshots or the feature is disabled"); currSnap = updateSnapshotInfo(snapshotLocation); } if (currSnap == null) { LOG.warn("Failed to update metadata for snapshot " + this.snapshotId); throw new EucalyptusCloudException("Failed to update metadata for snapshot " + this.snapshotId); } } catch (EucalyptusCloudException e) { throw e; } catch (Exception e) { LOG.warn("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); throw new EucalyptusCloudException("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); } StorageResourceWithCallback srwc = null; StorageResource snapshotResource = null; if (prevSnap != null) { LOG.info("Generate delta between penultimate snapshot " + prevSnap.getSnapshotId() + " and latest snapshot " + this.snapshotId); srwc = blockManager.prepIncrementalSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId, prevSnap.getSnapshotId(), prevSnap.getSnapPointId()); snapshotResource = srwc.getSr(); } else { LOG.info("Upload entire content of snapshot " + this.snapshotId); snapshotResource = blockManager.prepSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId); } if (snapshotResource == null) { LOG.warn("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); throw new EucalyptusCloudException("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); } try { uploadFuture = snapshotTransfer.upload(snapshotResource, progressCallback); } catch (Exception e) { throw new EucalyptusCloudException("Failed to upload snapshot " + this.snapshotId + " to objectstorage", e); } finally { if (srwc != null && srwc.getCallback() != null) { blockManager.executeCallback(srwc.getCallback(), srwc.getSr()); } } } else { LOG.debug("Snapshot uploads are disabled, skipping upload step for " + this.snapshotId); } } else { LOG.warn("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); } try { LOG.debug("Finishing up " + this.snapshotId + " on block storage backend"); blockManager.finishVolume(this.snapshotId); LOG.info("Finished creating " + this.snapshotId + " on block storage backend"); progressCallback.updateBackendProgress(50); } catch (EucalyptusCloudException ex) { LOG.warn("Failed to complete snapshot " + this.snapshotId + " on backend", ex); throw ex; } if (uploadFuture != null) { LOG.debug("Waiting for upload of " + this.snapshotId + " to complete"); if (uploadFuture.get() != null) { LOG.info("Uploaded " + this.snapshotId + " to object storage gateway, etag result - " + uploadFuture.get()); } else { LOG.warn("Failed to upload " + this.snapshotId + " to object storage gateway failed. Check prior logs for exact errors"); throw new EucalyptusCloudException("Failed to upload " + this.snapshotId + " to object storage gateway"); } } if (prevSnap != null) { final int MILLIS_PER_HOUR = 60 * 60 * 1000; final int timeoutMillis = StorageInfo.getStorageInfo().getSnapshotTransferTimeoutInHours() * MILLIS_PER_HOUR; final int pollPeriodMillis = 10000; int waitTimeSoFar = 0; while (waitTimeSoFar < timeoutMillis) { try { if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { break; } } catch (NoSuchElementException nsee) { LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " no longer recorded in the database. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw nsee; } catch (TransactionException te) { LOG.error("General database error. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw te; } try { Thread.sleep(pollPeriodMillis); } catch (InterruptedException e) { throw new RuntimeException("Interrupted while waiting for previous snapshot to complete", e); } waitTimeSoFar += pollPeriodMillis; } if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { if (isSnapshotMarkedAvailable(prevSnap.getSnapshotId())) { markSnapshotAvailable(); } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " finalized into the " + prevSnap.getStatus() + " state. This snapshot delta " + this.snapshotId + " set to 'failed' state because it requires an intact previous snapshot."); } } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " never finalized after " + "waiting " + (timeoutMillis / MILLIS_PER_HOUR) + " hours. " + "This snapshot delta " + this.snapshotId + " set to 'failed' state because " + "it requires an intact previous snapshot."); } } else { markSnapshotAvailable(); LOG.debug("Snapshot " + this.snapshotId + " set to 'available' state"); } } catch (Exception ex) { LOG.error("Failed to create snapshot " + this.snapshotId, ex); try { markSnapshotFailed(); LOG.debug("Snapshot " + this.snapshotId + " set to 'failed' state"); } catch (TransactionException | NoSuchElementException e) { LOG.warn("Cannot update snapshot " + this.snapshotId + " status to 'failed' in DB", e); } } LOG.trace("Finished SnapshotCreator task"); } }
SnapshotCreator implements Runnable { @Override public void run() { LOG.trace("Starting SnapshotCreator task"); EucaSemaphore semaphore = EucaSemaphoreDirectory.getSolitarySemaphore(this.volumeId); try { Boolean shouldTransferSnapshots = true; String bucket = null; shouldTransferSnapshots = StorageInfo.getStorageInfo().getShouldTransferSnapshots(); if (shouldTransferSnapshots) { if (null == snapshotTransfer) { snapshotTransfer = new S3SnapshotTransfer(this.snapshotId, this.snapshotId); } bucket = snapshotTransfer.prepareForUpload(); if (snapshotTransfer == null || StringUtils.isBlank(bucket)) { throw new EucalyptusCloudException("Failed to initialize snapshot transfer mechanism for uploading " + this.snapshotId); } } try { try { semaphore.acquire(); } catch (InterruptedException ex) { throw new EucalyptusCloudException("Failed to create snapshot " + this.snapshotId + " as the semaphore could not be acquired"); } if (!isSnapshotMarkedFailed(this.snapshotId)) { if (null == progressCallback) { progressCallback = new SnapshotProgressCallback(this.snapshotId); } blockManager.createSnapshot(this.volumeId, this.snapshotId, this.snapPointId); progressCallback.updateBackendProgress(50); } else { throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed by another thread"); } } finally { semaphore.release(); } SnapshotInfo prevSnap = null; SnapshotInfo currSnap = null; Future<String> uploadFuture = null; if (!isSnapshotMarkedFailed(this.snapshotId)) { if (shouldTransferSnapshots) { String snapshotLocation = SnapshotInfo.generateSnapshotLocationURI(SnapshotTransferConfiguration.OSG, bucket, this.snapshotId); try { Integer maxDeltaLimit = StorageInfo.getStorageInfo().getMaxSnapshotDeltas(); maxDeltaLimit = maxDeltaLimit != null ? maxDeltaLimit : 0; if (maxDeltaLimit > 0 && blockManager.supportsIncrementalSnapshots()) { LOG.debug("EBS backend supports incremental snapshots"); int attempts = 0; do { attempts++; prevSnap = fetchPreviousSnapshot(maxDeltaLimit); if (prevSnap != null) { EucaSemaphore prevSnapSemaphore = EucaSemaphoreDirectory.getSolitarySemaphore(prevSnap.getSnapshotId()); try { try { prevSnapSemaphore.acquire(); } catch (InterruptedException ex) { LOG.warn("Cannot update metadata for snapshot " + this.snapshotId + " due to an error acquiring semaphore for a previous snapshot " + prevSnap.getSnapshotId() + ". May retry again later"); continue; } currSnap = updateSnapshotInfo(prevSnap.getSnapshotId(), snapshotLocation); } finally { prevSnapSemaphore.release(); } } else { currSnap = updateSnapshotInfo(snapshotLocation); } } while (currSnap == null && attempts < 10); } else { LOG.debug("Either EBS backend does not support incremental snapshots or the feature is disabled"); currSnap = updateSnapshotInfo(snapshotLocation); } if (currSnap == null) { LOG.warn("Failed to update metadata for snapshot " + this.snapshotId); throw new EucalyptusCloudException("Failed to update metadata for snapshot " + this.snapshotId); } } catch (EucalyptusCloudException e) { throw e; } catch (Exception e) { LOG.warn("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); throw new EucalyptusCloudException("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); } StorageResourceWithCallback srwc = null; StorageResource snapshotResource = null; if (prevSnap != null) { LOG.info("Generate delta between penultimate snapshot " + prevSnap.getSnapshotId() + " and latest snapshot " + this.snapshotId); srwc = blockManager.prepIncrementalSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId, prevSnap.getSnapshotId(), prevSnap.getSnapPointId()); snapshotResource = srwc.getSr(); } else { LOG.info("Upload entire content of snapshot " + this.snapshotId); snapshotResource = blockManager.prepSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId); } if (snapshotResource == null) { LOG.warn("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); throw new EucalyptusCloudException("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); } try { uploadFuture = snapshotTransfer.upload(snapshotResource, progressCallback); } catch (Exception e) { throw new EucalyptusCloudException("Failed to upload snapshot " + this.snapshotId + " to objectstorage", e); } finally { if (srwc != null && srwc.getCallback() != null) { blockManager.executeCallback(srwc.getCallback(), srwc.getSr()); } } } else { LOG.debug("Snapshot uploads are disabled, skipping upload step for " + this.snapshotId); } } else { LOG.warn("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); } try { LOG.debug("Finishing up " + this.snapshotId + " on block storage backend"); blockManager.finishVolume(this.snapshotId); LOG.info("Finished creating " + this.snapshotId + " on block storage backend"); progressCallback.updateBackendProgress(50); } catch (EucalyptusCloudException ex) { LOG.warn("Failed to complete snapshot " + this.snapshotId + " on backend", ex); throw ex; } if (uploadFuture != null) { LOG.debug("Waiting for upload of " + this.snapshotId + " to complete"); if (uploadFuture.get() != null) { LOG.info("Uploaded " + this.snapshotId + " to object storage gateway, etag result - " + uploadFuture.get()); } else { LOG.warn("Failed to upload " + this.snapshotId + " to object storage gateway failed. Check prior logs for exact errors"); throw new EucalyptusCloudException("Failed to upload " + this.snapshotId + " to object storage gateway"); } } if (prevSnap != null) { final int MILLIS_PER_HOUR = 60 * 60 * 1000; final int timeoutMillis = StorageInfo.getStorageInfo().getSnapshotTransferTimeoutInHours() * MILLIS_PER_HOUR; final int pollPeriodMillis = 10000; int waitTimeSoFar = 0; while (waitTimeSoFar < timeoutMillis) { try { if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { break; } } catch (NoSuchElementException nsee) { LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " no longer recorded in the database. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw nsee; } catch (TransactionException te) { LOG.error("General database error. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw te; } try { Thread.sleep(pollPeriodMillis); } catch (InterruptedException e) { throw new RuntimeException("Interrupted while waiting for previous snapshot to complete", e); } waitTimeSoFar += pollPeriodMillis; } if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { if (isSnapshotMarkedAvailable(prevSnap.getSnapshotId())) { markSnapshotAvailable(); } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " finalized into the " + prevSnap.getStatus() + " state. This snapshot delta " + this.snapshotId + " set to 'failed' state because it requires an intact previous snapshot."); } } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " never finalized after " + "waiting " + (timeoutMillis / MILLIS_PER_HOUR) + " hours. " + "This snapshot delta " + this.snapshotId + " set to 'failed' state because " + "it requires an intact previous snapshot."); } } else { markSnapshotAvailable(); LOG.debug("Snapshot " + this.snapshotId + " set to 'available' state"); } } catch (Exception ex) { LOG.error("Failed to create snapshot " + this.snapshotId, ex); try { markSnapshotFailed(); LOG.debug("Snapshot " + this.snapshotId + " set to 'failed' state"); } catch (TransactionException | NoSuchElementException e) { LOG.warn("Cannot update snapshot " + this.snapshotId + " status to 'failed' in DB", e); } } LOG.trace("Finished SnapshotCreator task"); } SnapshotCreator(String volumeId, String snapshotId, String snapPointId, LogicalStorageManager blockManager); protected  SnapshotCreator(String volumeId, String snapshotId, String snapPointId, LogicalStorageManager mockBlockManager,
      SnapshotTransfer mockSnapshotTransfer, SnapshotProgressCallback mockProgressCallback); }
SnapshotCreator implements Runnable { @Override public void run() { LOG.trace("Starting SnapshotCreator task"); EucaSemaphore semaphore = EucaSemaphoreDirectory.getSolitarySemaphore(this.volumeId); try { Boolean shouldTransferSnapshots = true; String bucket = null; shouldTransferSnapshots = StorageInfo.getStorageInfo().getShouldTransferSnapshots(); if (shouldTransferSnapshots) { if (null == snapshotTransfer) { snapshotTransfer = new S3SnapshotTransfer(this.snapshotId, this.snapshotId); } bucket = snapshotTransfer.prepareForUpload(); if (snapshotTransfer == null || StringUtils.isBlank(bucket)) { throw new EucalyptusCloudException("Failed to initialize snapshot transfer mechanism for uploading " + this.snapshotId); } } try { try { semaphore.acquire(); } catch (InterruptedException ex) { throw new EucalyptusCloudException("Failed to create snapshot " + this.snapshotId + " as the semaphore could not be acquired"); } if (!isSnapshotMarkedFailed(this.snapshotId)) { if (null == progressCallback) { progressCallback = new SnapshotProgressCallback(this.snapshotId); } blockManager.createSnapshot(this.volumeId, this.snapshotId, this.snapPointId); progressCallback.updateBackendProgress(50); } else { throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed by another thread"); } } finally { semaphore.release(); } SnapshotInfo prevSnap = null; SnapshotInfo currSnap = null; Future<String> uploadFuture = null; if (!isSnapshotMarkedFailed(this.snapshotId)) { if (shouldTransferSnapshots) { String snapshotLocation = SnapshotInfo.generateSnapshotLocationURI(SnapshotTransferConfiguration.OSG, bucket, this.snapshotId); try { Integer maxDeltaLimit = StorageInfo.getStorageInfo().getMaxSnapshotDeltas(); maxDeltaLimit = maxDeltaLimit != null ? maxDeltaLimit : 0; if (maxDeltaLimit > 0 && blockManager.supportsIncrementalSnapshots()) { LOG.debug("EBS backend supports incremental snapshots"); int attempts = 0; do { attempts++; prevSnap = fetchPreviousSnapshot(maxDeltaLimit); if (prevSnap != null) { EucaSemaphore prevSnapSemaphore = EucaSemaphoreDirectory.getSolitarySemaphore(prevSnap.getSnapshotId()); try { try { prevSnapSemaphore.acquire(); } catch (InterruptedException ex) { LOG.warn("Cannot update metadata for snapshot " + this.snapshotId + " due to an error acquiring semaphore for a previous snapshot " + prevSnap.getSnapshotId() + ". May retry again later"); continue; } currSnap = updateSnapshotInfo(prevSnap.getSnapshotId(), snapshotLocation); } finally { prevSnapSemaphore.release(); } } else { currSnap = updateSnapshotInfo(snapshotLocation); } } while (currSnap == null && attempts < 10); } else { LOG.debug("Either EBS backend does not support incremental snapshots or the feature is disabled"); currSnap = updateSnapshotInfo(snapshotLocation); } if (currSnap == null) { LOG.warn("Failed to update metadata for snapshot " + this.snapshotId); throw new EucalyptusCloudException("Failed to update metadata for snapshot " + this.snapshotId); } } catch (EucalyptusCloudException e) { throw e; } catch (Exception e) { LOG.warn("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); throw new EucalyptusCloudException("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); } StorageResourceWithCallback srwc = null; StorageResource snapshotResource = null; if (prevSnap != null) { LOG.info("Generate delta between penultimate snapshot " + prevSnap.getSnapshotId() + " and latest snapshot " + this.snapshotId); srwc = blockManager.prepIncrementalSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId, prevSnap.getSnapshotId(), prevSnap.getSnapPointId()); snapshotResource = srwc.getSr(); } else { LOG.info("Upload entire content of snapshot " + this.snapshotId); snapshotResource = blockManager.prepSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId); } if (snapshotResource == null) { LOG.warn("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); throw new EucalyptusCloudException("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); } try { uploadFuture = snapshotTransfer.upload(snapshotResource, progressCallback); } catch (Exception e) { throw new EucalyptusCloudException("Failed to upload snapshot " + this.snapshotId + " to objectstorage", e); } finally { if (srwc != null && srwc.getCallback() != null) { blockManager.executeCallback(srwc.getCallback(), srwc.getSr()); } } } else { LOG.debug("Snapshot uploads are disabled, skipping upload step for " + this.snapshotId); } } else { LOG.warn("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); } try { LOG.debug("Finishing up " + this.snapshotId + " on block storage backend"); blockManager.finishVolume(this.snapshotId); LOG.info("Finished creating " + this.snapshotId + " on block storage backend"); progressCallback.updateBackendProgress(50); } catch (EucalyptusCloudException ex) { LOG.warn("Failed to complete snapshot " + this.snapshotId + " on backend", ex); throw ex; } if (uploadFuture != null) { LOG.debug("Waiting for upload of " + this.snapshotId + " to complete"); if (uploadFuture.get() != null) { LOG.info("Uploaded " + this.snapshotId + " to object storage gateway, etag result - " + uploadFuture.get()); } else { LOG.warn("Failed to upload " + this.snapshotId + " to object storage gateway failed. Check prior logs for exact errors"); throw new EucalyptusCloudException("Failed to upload " + this.snapshotId + " to object storage gateway"); } } if (prevSnap != null) { final int MILLIS_PER_HOUR = 60 * 60 * 1000; final int timeoutMillis = StorageInfo.getStorageInfo().getSnapshotTransferTimeoutInHours() * MILLIS_PER_HOUR; final int pollPeriodMillis = 10000; int waitTimeSoFar = 0; while (waitTimeSoFar < timeoutMillis) { try { if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { break; } } catch (NoSuchElementException nsee) { LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " no longer recorded in the database. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw nsee; } catch (TransactionException te) { LOG.error("General database error. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw te; } try { Thread.sleep(pollPeriodMillis); } catch (InterruptedException e) { throw new RuntimeException("Interrupted while waiting for previous snapshot to complete", e); } waitTimeSoFar += pollPeriodMillis; } if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { if (isSnapshotMarkedAvailable(prevSnap.getSnapshotId())) { markSnapshotAvailable(); } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " finalized into the " + prevSnap.getStatus() + " state. This snapshot delta " + this.snapshotId + " set to 'failed' state because it requires an intact previous snapshot."); } } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " never finalized after " + "waiting " + (timeoutMillis / MILLIS_PER_HOUR) + " hours. " + "This snapshot delta " + this.snapshotId + " set to 'failed' state because " + "it requires an intact previous snapshot."); } } else { markSnapshotAvailable(); LOG.debug("Snapshot " + this.snapshotId + " set to 'available' state"); } } catch (Exception ex) { LOG.error("Failed to create snapshot " + this.snapshotId, ex); try { markSnapshotFailed(); LOG.debug("Snapshot " + this.snapshotId + " set to 'failed' state"); } catch (TransactionException | NoSuchElementException e) { LOG.warn("Cannot update snapshot " + this.snapshotId + " status to 'failed' in DB", e); } } LOG.trace("Finished SnapshotCreator task"); } SnapshotCreator(String volumeId, String snapshotId, String snapPointId, LogicalStorageManager blockManager); protected  SnapshotCreator(String volumeId, String snapshotId, String snapPointId, LogicalStorageManager mockBlockManager,
      SnapshotTransfer mockSnapshotTransfer, SnapshotProgressCallback mockProgressCallback); @Override void run(); }
SnapshotCreator implements Runnable { @Override public void run() { LOG.trace("Starting SnapshotCreator task"); EucaSemaphore semaphore = EucaSemaphoreDirectory.getSolitarySemaphore(this.volumeId); try { Boolean shouldTransferSnapshots = true; String bucket = null; shouldTransferSnapshots = StorageInfo.getStorageInfo().getShouldTransferSnapshots(); if (shouldTransferSnapshots) { if (null == snapshotTransfer) { snapshotTransfer = new S3SnapshotTransfer(this.snapshotId, this.snapshotId); } bucket = snapshotTransfer.prepareForUpload(); if (snapshotTransfer == null || StringUtils.isBlank(bucket)) { throw new EucalyptusCloudException("Failed to initialize snapshot transfer mechanism for uploading " + this.snapshotId); } } try { try { semaphore.acquire(); } catch (InterruptedException ex) { throw new EucalyptusCloudException("Failed to create snapshot " + this.snapshotId + " as the semaphore could not be acquired"); } if (!isSnapshotMarkedFailed(this.snapshotId)) { if (null == progressCallback) { progressCallback = new SnapshotProgressCallback(this.snapshotId); } blockManager.createSnapshot(this.volumeId, this.snapshotId, this.snapPointId); progressCallback.updateBackendProgress(50); } else { throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed by another thread"); } } finally { semaphore.release(); } SnapshotInfo prevSnap = null; SnapshotInfo currSnap = null; Future<String> uploadFuture = null; if (!isSnapshotMarkedFailed(this.snapshotId)) { if (shouldTransferSnapshots) { String snapshotLocation = SnapshotInfo.generateSnapshotLocationURI(SnapshotTransferConfiguration.OSG, bucket, this.snapshotId); try { Integer maxDeltaLimit = StorageInfo.getStorageInfo().getMaxSnapshotDeltas(); maxDeltaLimit = maxDeltaLimit != null ? maxDeltaLimit : 0; if (maxDeltaLimit > 0 && blockManager.supportsIncrementalSnapshots()) { LOG.debug("EBS backend supports incremental snapshots"); int attempts = 0; do { attempts++; prevSnap = fetchPreviousSnapshot(maxDeltaLimit); if (prevSnap != null) { EucaSemaphore prevSnapSemaphore = EucaSemaphoreDirectory.getSolitarySemaphore(prevSnap.getSnapshotId()); try { try { prevSnapSemaphore.acquire(); } catch (InterruptedException ex) { LOG.warn("Cannot update metadata for snapshot " + this.snapshotId + " due to an error acquiring semaphore for a previous snapshot " + prevSnap.getSnapshotId() + ". May retry again later"); continue; } currSnap = updateSnapshotInfo(prevSnap.getSnapshotId(), snapshotLocation); } finally { prevSnapSemaphore.release(); } } else { currSnap = updateSnapshotInfo(snapshotLocation); } } while (currSnap == null && attempts < 10); } else { LOG.debug("Either EBS backend does not support incremental snapshots or the feature is disabled"); currSnap = updateSnapshotInfo(snapshotLocation); } if (currSnap == null) { LOG.warn("Failed to update metadata for snapshot " + this.snapshotId); throw new EucalyptusCloudException("Failed to update metadata for snapshot " + this.snapshotId); } } catch (EucalyptusCloudException e) { throw e; } catch (Exception e) { LOG.warn("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); throw new EucalyptusCloudException("Unable to evaluate snapshot location and upload specifics, failing snapshot " + this.snapshotId, e); } StorageResourceWithCallback srwc = null; StorageResource snapshotResource = null; if (prevSnap != null) { LOG.info("Generate delta between penultimate snapshot " + prevSnap.getSnapshotId() + " and latest snapshot " + this.snapshotId); srwc = blockManager.prepIncrementalSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId, prevSnap.getSnapshotId(), prevSnap.getSnapPointId()); snapshotResource = srwc.getSr(); } else { LOG.info("Upload entire content of snapshot " + this.snapshotId); snapshotResource = blockManager.prepSnapshotForUpload(this.volumeId, this.snapshotId, this.snapPointId); } if (snapshotResource == null) { LOG.warn("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); throw new EucalyptusCloudException("Unable to upload snapshot " + this.snapshotId + " due to invalid source"); } try { uploadFuture = snapshotTransfer.upload(snapshotResource, progressCallback); } catch (Exception e) { throw new EucalyptusCloudException("Failed to upload snapshot " + this.snapshotId + " to objectstorage", e); } finally { if (srwc != null && srwc.getCallback() != null) { blockManager.executeCallback(srwc.getCallback(), srwc.getSr()); } } } else { LOG.debug("Snapshot uploads are disabled, skipping upload step for " + this.snapshotId); } } else { LOG.warn("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); throw new EucalyptusCloudException("Snapshot " + this.snapshotId + " marked as failed, aborting upload process"); } try { LOG.debug("Finishing up " + this.snapshotId + " on block storage backend"); blockManager.finishVolume(this.snapshotId); LOG.info("Finished creating " + this.snapshotId + " on block storage backend"); progressCallback.updateBackendProgress(50); } catch (EucalyptusCloudException ex) { LOG.warn("Failed to complete snapshot " + this.snapshotId + " on backend", ex); throw ex; } if (uploadFuture != null) { LOG.debug("Waiting for upload of " + this.snapshotId + " to complete"); if (uploadFuture.get() != null) { LOG.info("Uploaded " + this.snapshotId + " to object storage gateway, etag result - " + uploadFuture.get()); } else { LOG.warn("Failed to upload " + this.snapshotId + " to object storage gateway failed. Check prior logs for exact errors"); throw new EucalyptusCloudException("Failed to upload " + this.snapshotId + " to object storage gateway"); } } if (prevSnap != null) { final int MILLIS_PER_HOUR = 60 * 60 * 1000; final int timeoutMillis = StorageInfo.getStorageInfo().getSnapshotTransferTimeoutInHours() * MILLIS_PER_HOUR; final int pollPeriodMillis = 10000; int waitTimeSoFar = 0; while (waitTimeSoFar < timeoutMillis) { try { if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { break; } } catch (NoSuchElementException nsee) { LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " no longer recorded in the database. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw nsee; } catch (TransactionException te) { LOG.error("General database error. " + "Setting this snapshot delta " + this.snapshotId + " to 'failed' state because it requires an intact previous snapshot."); throw te; } try { Thread.sleep(pollPeriodMillis); } catch (InterruptedException e) { throw new RuntimeException("Interrupted while waiting for previous snapshot to complete", e); } waitTimeSoFar += pollPeriodMillis; } if (isSnapshotMarkedFinalized(prevSnap.getSnapshotId())) { if (isSnapshotMarkedAvailable(prevSnap.getSnapshotId())) { markSnapshotAvailable(); } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " finalized into the " + prevSnap.getStatus() + " state. This snapshot delta " + this.snapshotId + " set to 'failed' state because it requires an intact previous snapshot."); } } else { markSnapshotFailed(); LOG.error("Previous snapshot " + prevSnap.getSnapshotId() + " never finalized after " + "waiting " + (timeoutMillis / MILLIS_PER_HOUR) + " hours. " + "This snapshot delta " + this.snapshotId + " set to 'failed' state because " + "it requires an intact previous snapshot."); } } else { markSnapshotAvailable(); LOG.debug("Snapshot " + this.snapshotId + " set to 'available' state"); } } catch (Exception ex) { LOG.error("Failed to create snapshot " + this.snapshotId, ex); try { markSnapshotFailed(); LOG.debug("Snapshot " + this.snapshotId + " set to 'failed' state"); } catch (TransactionException | NoSuchElementException e) { LOG.warn("Cannot update snapshot " + this.snapshotId + " status to 'failed' in DB", e); } } LOG.trace("Finished SnapshotCreator task"); } SnapshotCreator(String volumeId, String snapshotId, String snapPointId, LogicalStorageManager blockManager); protected  SnapshotCreator(String volumeId, String snapshotId, String snapPointId, LogicalStorageManager mockBlockManager,
      SnapshotTransfer mockSnapshotTransfer, SnapshotProgressCallback mockProgressCallback); @Override void run();  }