package org.openslx.bwlp.sat.fileserv;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.sql.SQLException;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.net.ssl.SSLContext;
import org.apache.log4j.Logger;
import org.openslx.bwlp.sat.database.mappers.DbImage;
import org.openslx.bwlp.sat.util.Configuration;
import org.openslx.bwlp.sat.util.Constants;
import org.openslx.bwlp.sat.util.FileSystem;
import org.openslx.bwlp.sat.util.Formatter;
import org.openslx.bwlp.thrift.iface.ImageDetailsRead;
import org.openslx.bwlp.thrift.iface.ImagePublishData;
import org.openslx.bwlp.thrift.iface.ImageVersionWrite;
import org.openslx.bwlp.thrift.iface.TransferInformation;
import org.openslx.bwlp.thrift.iface.TransferState;
import org.openslx.bwlp.thrift.iface.TransferStatus;
import org.openslx.bwlp.thrift.iface.UserInfo;
import org.openslx.filetransfer.Downloader;
import org.openslx.filetransfer.util.FileChunk;
import org.openslx.filetransfer.util.IncomingTransferBase;
import org.openslx.util.ThriftUtil;
import org.openslx.util.vm.DiskImage;
import org.openslx.util.vm.DiskImage.UnknownImageFormatException;
public class IncomingDataTransfer extends IncomingTransferBase {
private static final Logger LOGGER = Logger.getLogger(IncomingDataTransfer.class);
private static final long MIN_FREE_SPACE_BYTES = FileChunk.CHUNK_SIZE * (2 + Constants.MAX_UPLOADS);
/**
* User owning this uploaded file.
*/
private final UserInfo owner;
/**
* Base image this upload is a new version for.
*/
private final ImageDetailsRead image;
/**
* Flags to set for this new image version. Optional field.
*/
private ImageVersionWrite versionSettings = null;
/**
* Description of this VM - binary dump of e.g. the *.vmx file (VMware)
*/
private final byte[] machineDescription;
/**
* Indicated whether the version information was written to db already.
* Disallow setVersionData in that case.
*/
private final AtomicBoolean versionWrittenToDb = new AtomicBoolean();
/**
* Set if this is a download from the master server
*/
private final TransferInformation masterTransferInfo;
public IncomingDataTransfer(String uploadId, UserInfo owner, ImageDetailsRead image,
File destinationFile, long fileSize, List<byte[]> sha1Sums, byte[] machineDescription)
throws FileNotFoundException {
super(uploadId, destinationFile, fileSize, sha1Sums);
this.owner = owner;
this.image = image;
this.machineDescription = machineDescription;
this.masterTransferInfo = null;
}
public IncomingDataTransfer(ImagePublishData publishData, File tmpFile, TransferInformation transferInfo)
throws FileNotFoundException {
super(UUID.randomUUID().toString(), tmpFile, publishData.fileSize,
ThriftUtil.unwrapByteBufferList(transferInfo.blockHashes));
ImageDetailsRead idr = new ImageDetailsRead();
idr.setCreateTime(publishData.createTime);
idr.setDescription(publishData.description);
idr.setImageBaseId(publishData.imageBaseId);
idr.setImageName(publishData.imageName);
idr.setIsTemplate(publishData.isTemplate);
idr.setLatestVersionId(publishData.imageVersionId);
idr.setOsId(publishData.osId);
idr.setOwnerId(publishData.user.userId);
idr.setTags(publishData.tags);
idr.setUpdaterId(publishData.user.userId);
idr.setUpdateTime(publishData.createTime);
idr.setVirtId(publishData.virtId);
this.owner = publishData.user;
this.image = idr;
this.machineDescription = ThriftUtil.unwrapByteBuffer(transferInfo.machineDescription);
this.masterTransferInfo = transferInfo;
this.versionSettings = new ImageVersionWrite(false);
}
/**
* Called periodically if this is a transfer from the master server, so we
* can make sure the transfer is running.
*/
public void heartBeat(ExecutorService pool) {
if (masterTransferInfo == null)
return;
synchronized (this) {
if (getActiveConnectionCount() >= 1)
return;
Downloader downloader = null;
if (masterTransferInfo.plainPort != 0) {
try {
downloader = new Downloader(Configuration.getMasterServerAddress(),
masterTransferInfo.plainPort, Constants.TRANSFER_TIMEOUT, null,
masterTransferInfo.token);
} catch (Exception e1) {
LOGGER.debug("Plain connect failed", e1);
downloader = null;
}
}
if (downloader == null && masterTransferInfo.sslPort != 0) {
try {
downloader = new Downloader(Configuration.getMasterServerAddress(),
masterTransferInfo.sslPort, Constants.TRANSFER_TIMEOUT, SSLContext.getDefault(), // TODO: Use the TLSv1.2 one once the master is ready
masterTransferInfo.token);
} catch (Exception e2) {
LOGGER.debug("SSL connect failed", e2);
downloader = null;
}
}
if (downloader == null) {
LOGGER.warn("Could not connect to master server for downloading " + image.imageName);
return;
}
addConnection(downloader, pool);
}
}
/**
* Set meta data for this image version.
*
* @param user
*
* @param data
*/
public boolean setVersionData(UserInfo user, ImageVersionWrite data) {
synchronized (versionWrittenToDb) {
if (versionWrittenToDb.get()) {
return false;
}
if (!user.userId.equals(owner.userId)) {
return false;
}
versionSettings = data;
return true;
}
}
/**
* Called when the upload finished.
*/
@Override
protected synchronized boolean finishIncomingTransfer() {
if (getState() != TransferState.FINISHED)
return false;
potentialFinishTime.set(System.currentTimeMillis());
// If owner is not set, this was a repair-transfer, which downloads directly to the existing target file.
// Nothing more to do in that case.
if (isRepairUpload())
return true;
LOGGER.info("Finalizing uploaded image " + image.imageName);
// Ready to go. First step: Rename temp file to something usable
String ext = "img";
try {
ext = new DiskImage(getTmpFileName()).format.extension;
} catch (IOException | UnknownImageFormatException e1) {
}
File destination = new File(getTmpFileName().getParent(), Formatter.vmName(owner, image.imageName,
ext));
// Sanity check: destination should be a sub directory of the vmStorePath
String relPath = FileSystem.getRelativePath(destination, Configuration.getVmStoreBasePath());
if (relPath == null) {
LOGGER.warn(destination.getAbsolutePath() + " is not a subdir of "
+ Configuration.getVmStoreBasePath().getAbsolutePath());
cancel();
return false;
}
if (relPath.length() > 200) {
LOGGER.error("Generated file name is >200 chars. DB will not like it");
}
// Execute rename
boolean ret = false;
Exception renameException = null;
try {
ret = getTmpFileName().renameTo(destination);
} catch (Exception e) {
ret = false;
renameException = e;
}
if (!ret) {
// Rename failed :-(
LOGGER.warn(
"Could not rename '" + getTmpFileName().getAbsolutePath() + "' to '"
+ destination.getAbsolutePath() + "'", renameException);
cancel();
return false;
}
// Now insert meta data into DB
try {
synchronized (versionWrittenToDb) {
DbImage.createImageVersion(image.imageBaseId, getId(), owner, getFileSize(), relPath,
versionSettings, getChunks(), machineDescription);
versionWrittenToDb.set(true);
}
} catch (SQLException e) {
LOGGER.error("Error finishing upload: Inserting version to DB failed", e);
// Also delete uploaded file, as there is no reference to it
FileSystem.deleteAsync(destination);
cancel();
return false;
}
return true;
}
@Override
public synchronized void cancel() {
super.cancel();
if (!isRepairUpload() && getTmpFileName().exists()) {
FileSystem.deleteAsync(getTmpFileName());
}
}
public boolean isRepairUpload() {
return owner == null;
}
/**
* Get user owning this upload. Can be null in special cases.
*
* @return instance of UserInfo for the according user.
*/
public UserInfo getOwner() {
return this.owner;
}
public synchronized TransferStatus getStatus() {
return new TransferStatus(getChunks().getStatusArray(), getState());
}
@Override
public boolean isActive() {
return getState() == TransferState.IDLE || getState() == TransferState.WORKING;
}
@Override
protected void finalize() {
try {
super.finalize();
} catch (Throwable t) {
}
try {
cancel();
} catch (Throwable t) {
}
}
@Override
protected boolean hasEnoughFreeSpace() {
return FileSystem.getAvailableStorageBytes() > MIN_FREE_SPACE_BYTES;
}
@Override
public TransferInformation getTransferInfo() {
return new TransferInformation(getId(), FileServer.instance().getPlainPort(), FileServer.instance()
.getSslPort());
}
@Override
public String getRelativePath() {
return FileSystem.getRelativePath(getTmpFileName(), Configuration.getVmStoreBasePath());
}
@Override
protected void chunkStatusChanged(FileChunk chunk) {
// TODO Update in DB in case this is a repair upload
}
}
