You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudstack.apache.org by ko...@apache.org on 2015/01/14 06:09:28 UTC
git commit: updated refs/heads/volume-upload to 627f5a6
Repository: cloudstack
Updated Branches:
refs/heads/volume-upload c2cf2503c -> 627f5a62d
volume upload: management server polling and upload status from agent
MS polling logic to query status for volumes that are uploaded
Project: http://git-wip-us.apache.org/repos/asf/cloudstack/repo
Commit: http://git-wip-us.apache.org/repos/asf/cloudstack/commit/627f5a62
Tree: http://git-wip-us.apache.org/repos/asf/cloudstack/tree/627f5a62
Diff: http://git-wip-us.apache.org/repos/asf/cloudstack/diff/627f5a62
Branch: refs/heads/volume-upload
Commit: 627f5a62dc71d7cfda53fa2b6bc4852a312aee02
Parents: c2cf250
Author: Koushik Das <ko...@apache.org>
Authored: Wed Jan 14 10:19:12 2015 +0530
Committer: Koushik Das <ko...@apache.org>
Committed: Wed Jan 14 10:19:12 2015 +0530
----------------------------------------------------------------------
api/src/com/cloud/storage/Volume.java | 19 +-
.../storage/command/UploadStatusAnswer.java | 53 ++++
.../storage/command/UploadStatusCommand.java | 44 ++++
.../datastore/db/VolumeDataStoreDao.java | 3 +
.../image/db/VolumeDataStoreDaoImpl.java | 25 +-
.../cloud/agent/manager/MockStorageManager.java | 5 +
.../agent/manager/MockStorageManagerImpl.java | 9 +-
.../agent/manager/SimulatorManagerImpl.java | 3 +
.../spring-server-core-managers-context.xml | 1 +
.../cloud/storage/ImageStoreUploadMonitor.java | 27 ++
.../storage/ImageStoreUploadMonitorImpl.java | 250 +++++++++++++++++++
.../resource/NfsSecondaryStorageResource.java | 10 +
12 files changed, 442 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/api/src/com/cloud/storage/Volume.java
----------------------------------------------------------------------
diff --git a/api/src/com/cloud/storage/Volume.java b/api/src/com/cloud/storage/Volume.java
index 89d08f7..50ff210 100755
--- a/api/src/com/cloud/storage/Volume.java
+++ b/api/src/com/cloud/storage/Volume.java
@@ -45,9 +45,12 @@ public interface Volume extends ControlledEntity, Identity, InternalIdentity, Ba
Destroy("The volume is destroyed, and can't be recovered."),
Destroying("The volume is destroying, and can't be recovered."),
UploadOp("The volume upload operation is in progress or in short the volume is on secondary storage"),
- Uploading("volume is uploading"),
- Copying("volume is copying from image store to primary, in case it's an uploaded volume"),
- Uploaded("volume is uploaded");
+ Copying("Volume is copying from image store to primary, in case it's an uploaded volume"),
+ Uploaded("Volume is uploaded"),
+ NotUploaded("The volume entry is just created in DB, not yet uploaded"),
+ UploadInProgress("Volume upload is in progress"),
+ UploadError("Volume upload encountered some error"),
+ UploadAbandoned("Volume upload is abandoned since the upload was never initiated within a specificed time");
String _description;
@@ -100,7 +103,12 @@ public interface Volume extends ControlledEntity, Identity, InternalIdentity, Ba
s_fsm.addTransition(new StateMachine2.Transition<State, Event>(Uploaded, Event.DestroyRequested, Destroy, null));
s_fsm.addTransition(new StateMachine2.Transition<State, Event>(Expunged, Event.ExpungingRequested, Expunged, null));
s_fsm.addTransition(new StateMachine2.Transition<State, Event>(Expunged, Event.OperationSucceeded, Expunged, null));
- s_fsm.addTransition(new StateMachine2.Transition<State, Event>(Expunged, Event.OperationFailed, Expunged,null));
+ s_fsm.addTransition(new StateMachine2.Transition<State, Event>(Expunged, Event.OperationFailed, Expunged, null));
+ s_fsm.addTransition(new StateMachine2.Transition<State, Event>(NotUploaded, Event.OperationTimeout, UploadAbandoned, null));
+ s_fsm.addTransition(new StateMachine2.Transition<State, Event>(NotUploaded, Event.UploadRequested, UploadInProgress, null));
+ s_fsm.addTransition(new StateMachine2.Transition<State, Event>(UploadInProgress, Event.OperationSucceeded, Uploaded, null));
+ s_fsm.addTransition(new StateMachine2.Transition<State, Event>(UploadInProgress, Event.OperationFailed, UploadError, null));
+ s_fsm.addTransition(new StateMachine2.Transition<State, Event>(UploadInProgress, Event.OperationTimeout, UploadError, null));
}
}
@@ -120,7 +128,8 @@ public interface Volume extends ControlledEntity, Identity, InternalIdentity, Ba
SnapshotRequested,
DestroyRequested,
ExpungingRequested,
- ResizeRequested;
+ ResizeRequested,
+ OperationTimeout;
}
/**
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/core/src/org/apache/cloudstack/storage/command/UploadStatusAnswer.java
----------------------------------------------------------------------
diff --git a/core/src/org/apache/cloudstack/storage/command/UploadStatusAnswer.java b/core/src/org/apache/cloudstack/storage/command/UploadStatusAnswer.java
new file mode 100644
index 0000000..887f3ab
--- /dev/null
+++ b/core/src/org/apache/cloudstack/storage/command/UploadStatusAnswer.java
@@ -0,0 +1,53 @@
+//
+// 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.cloudstack.storage.command;
+
+import com.cloud.agent.api.Answer;
+
+public class UploadStatusAnswer extends Answer {
+ public static enum UploadStatus {
+ UNKNOWN, IN_PROGRESS, COMPLETED, ERROR
+ }
+
+ private UploadStatus status;
+
+ protected UploadStatusAnswer() {
+ }
+
+ public UploadStatusAnswer(UploadStatusCommand cmd, UploadStatus status, String msg) {
+ super(cmd, false, msg);
+ this.status = UploadStatus.ERROR;
+ }
+
+ public UploadStatusAnswer(UploadStatusCommand cmd, Exception e) {
+ super(cmd, false, e.getMessage());
+ this.status = UploadStatus.ERROR;
+ }
+
+ public UploadStatusAnswer(UploadStatusCommand cmd, UploadStatus status) {
+ super(cmd, true, null);
+ this.status = status;
+ }
+
+ public UploadStatus getStatus() {
+ return status;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/core/src/org/apache/cloudstack/storage/command/UploadStatusCommand.java
----------------------------------------------------------------------
diff --git a/core/src/org/apache/cloudstack/storage/command/UploadStatusCommand.java b/core/src/org/apache/cloudstack/storage/command/UploadStatusCommand.java
new file mode 100644
index 0000000..957854e
--- /dev/null
+++ b/core/src/org/apache/cloudstack/storage/command/UploadStatusCommand.java
@@ -0,0 +1,44 @@
+//
+// 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.cloudstack.storage.command;
+
+import com.cloud.agent.api.Command;
+
+public class UploadStatusCommand extends Command {
+
+ private long entityId;
+
+ protected UploadStatusCommand() {
+ }
+
+ public UploadStatusCommand(long entityId) {
+ this.entityId = entityId;
+ }
+
+ public long getEntityId() {
+ return entityId;
+ }
+
+ @Override
+ public boolean executeInSequence() {
+ return false;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java
----------------------------------------------------------------------
diff --git a/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java b/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java
index a056152..e65880d 100644
--- a/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java
+++ b/engine/schema/src/org/apache/cloudstack/storage/datastore/db/VolumeDataStoreDao.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.cloudstack.engine.subsystem.api.storage.DataObjectInStore;
import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreStateMachine;
+import com.cloud.storage.Volume;
import com.cloud.utils.db.GenericDao;
import com.cloud.utils.fsm.StateDao;
@@ -48,4 +49,6 @@ public interface VolumeDataStoreDao extends GenericDao<VolumeDataStoreVO, Long>,
void expireDnldUrlsForZone(Long dcId);
List<VolumeDataStoreVO> listUploadedVolumesByStoreId(long id);
+
+ List<VolumeDataStoreVO> listByVolumeState(Volume.State... states);
}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java
----------------------------------------------------------------------
diff --git a/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java b/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java
index b71eb2c..117e622 100644
--- a/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java
+++ b/engine/storage/src/org/apache/cloudstack/storage/image/db/VolumeDataStoreDaoImpl.java
@@ -27,7 +27,6 @@ import javax.naming.ConfigurationException;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;
-
import org.apache.cloudstack.engine.subsystem.api.storage.DataObjectInStore;
import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
@@ -36,7 +35,11 @@ import org.apache.cloudstack.engine.subsystem.api.storage.ObjectInDataStoreState
import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
+import com.cloud.storage.Volume;
+import com.cloud.storage.VolumeVO;
+import com.cloud.storage.dao.VolumeDao;
import com.cloud.utils.db.GenericDaoBase;
+import com.cloud.utils.db.JoinBuilder.JoinType;
import com.cloud.utils.db.SearchBuilder;
import com.cloud.utils.db.SearchCriteria;
import com.cloud.utils.db.SearchCriteria.Op;
@@ -53,11 +56,15 @@ public class VolumeDataStoreDaoImpl extends GenericDaoBase<VolumeDataStoreVO, Lo
private SearchBuilder<VolumeDataStoreVO> storeVolumeSearch;
private SearchBuilder<VolumeDataStoreVO> downloadVolumeSearch;
private SearchBuilder<VolumeDataStoreVO> uploadVolumeSearch;
+ private SearchBuilder<VolumeVO> volumeOnlySearch;
+ private SearchBuilder<VolumeDataStoreVO> uploadVolumeStateSearch;
private static final String EXPIRE_DOWNLOAD_URLS_FOR_ZONE = "update volume_store_ref set download_url_created=? where store_id in (select id from image_store where data_center_id=?)";
@Inject
DataStoreManager storeMgr;
+ @Inject
+ VolumeDao volumeDao;
@Override
public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
@@ -102,6 +109,13 @@ public class VolumeDataStoreDaoImpl extends GenericDaoBase<VolumeDataStoreVO, Lo
uploadVolumeSearch.and("destroyed", uploadVolumeSearch.entity().getDestroyed(), SearchCriteria.Op.EQ);
uploadVolumeSearch.done();
+ volumeOnlySearch = volumeDao.createSearchBuilder();
+ volumeOnlySearch.and("state", volumeOnlySearch.entity().getState(), Op.IN);
+ uploadVolumeStateSearch = createSearchBuilder();
+ uploadVolumeStateSearch.join("volumeOnlySearch", volumeOnlySearch, volumeOnlySearch.entity().getId(), uploadVolumeStateSearch.entity().getVolumeId(), JoinType.LEFT);
+ uploadVolumeStateSearch.and("destroyed", uploadVolumeStateSearch.entity().getDestroyed(), SearchCriteria.Op.EQ);
+ uploadVolumeStateSearch.done();
+
return true;
}
@@ -304,4 +318,13 @@ public class VolumeDataStoreDaoImpl extends GenericDaoBase<VolumeDataStoreVO, Lo
}
}
+
+ @Override
+ public List<VolumeDataStoreVO> listByVolumeState(Volume.State... states) {
+ SearchCriteria<VolumeDataStoreVO> sc = uploadVolumeStateSearch.create();
+ sc.setJoinParameters("volumeOnlySearch", "state", (Object[])states);
+ sc.setParameters("destroyed", false);
+ return listIncludingRemovedBy(sc);
+ }
+
}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManager.java
----------------------------------------------------------------------
diff --git a/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManager.java b/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManager.java
index 71b5e4d..6f5a494 100644
--- a/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManager.java
+++ b/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManager.java
@@ -19,6 +19,8 @@ package com.cloud.agent.manager;
import org.apache.cloudstack.storage.command.DeleteCommand;
import org.apache.cloudstack.storage.command.DownloadCommand;
import org.apache.cloudstack.storage.command.DownloadProgressCommand;
+import org.apache.cloudstack.storage.command.UploadStatusAnswer;
+import org.apache.cloudstack.storage.command.UploadStatusCommand;
import com.cloud.agent.api.Answer;
import com.cloud.agent.api.AttachIsoCommand;
@@ -106,4 +108,7 @@ public interface MockStorageManager extends Manager {
StoragePoolInfo getLocalStorage(String hostGuid, Long storageSize);
CopyVolumeAnswer CopyVolume(CopyVolumeCommand cmd);
+
+ public UploadStatusAnswer getUploadStatus(UploadStatusCommand cmd);
+
}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManagerImpl.java
----------------------------------------------------------------------
diff --git a/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManagerImpl.java b/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManagerImpl.java
index 205e485..306826d 100644
--- a/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManagerImpl.java
+++ b/plugins/hypervisors/simulator/src/com/cloud/agent/manager/MockStorageManagerImpl.java
@@ -33,10 +33,12 @@ import javax.naming.ConfigurationException;
import org.apache.log4j.Logger;
import org.springframework.stereotype.Component;
-
import org.apache.cloudstack.storage.command.DeleteCommand;
import org.apache.cloudstack.storage.command.DownloadCommand;
import org.apache.cloudstack.storage.command.DownloadProgressCommand;
+import org.apache.cloudstack.storage.command.UploadStatusAnswer;
+import org.apache.cloudstack.storage.command.UploadStatusAnswer.UploadStatus;
+import org.apache.cloudstack.storage.command.UploadStatusCommand;
import com.cloud.agent.api.Answer;
import com.cloud.agent.api.AttachIsoCommand;
@@ -1255,4 +1257,9 @@ public class MockStorageManagerImpl extends ManagerBase implements MockStorageMa
return new CopyVolumeAnswer(cmd, true, null, primaryStorage.getMountPoint(), vol.getPath());
}
}
+
+ @Override
+ public UploadStatusAnswer getUploadStatus(UploadStatusCommand cmd) {
+ return new UploadStatusAnswer(cmd, UploadStatus.UNKNOWN);
+ }
}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/plugins/hypervisors/simulator/src/com/cloud/agent/manager/SimulatorManagerImpl.java
----------------------------------------------------------------------
diff --git a/plugins/hypervisors/simulator/src/com/cloud/agent/manager/SimulatorManagerImpl.java b/plugins/hypervisors/simulator/src/com/cloud/agent/manager/SimulatorManagerImpl.java
index 0ea6855..27be6f8 100644
--- a/plugins/hypervisors/simulator/src/com/cloud/agent/manager/SimulatorManagerImpl.java
+++ b/plugins/hypervisors/simulator/src/com/cloud/agent/manager/SimulatorManagerImpl.java
@@ -35,6 +35,7 @@ import org.apache.cloudstack.storage.command.DeleteCommand;
import org.apache.cloudstack.storage.command.DownloadCommand;
import org.apache.cloudstack.storage.command.DownloadProgressCommand;
import org.apache.cloudstack.storage.command.StorageSubSystemCommand;
+import org.apache.cloudstack.storage.command.UploadStatusCommand;
import com.cloud.agent.api.Answer;
import com.cloud.agent.api.AttachIsoCommand;
@@ -371,6 +372,8 @@ public class SimulatorManagerImpl extends ManagerBase implements SimulatorManage
answer = _mockStorageMgr.ComputeChecksum((ComputeChecksumCommand)cmd);
} else if (cmd instanceof CreatePrivateTemplateFromVolumeCommand) {
answer = _mockStorageMgr.CreatePrivateTemplateFromVolume((CreatePrivateTemplateFromVolumeCommand)cmd);
+ } else if (cmd instanceof UploadStatusCommand) {
+ answer = _mockStorageMgr.getUploadStatus((UploadStatusCommand)cmd);
} else if (cmd instanceof MaintainCommand) {
answer = _mockAgentMgr.maintain((MaintainCommand)cmd);
} else if (cmd instanceof GetVmStatsCommand) {
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/server/resources/META-INF/cloudstack/core/spring-server-core-managers-context.xml
----------------------------------------------------------------------
diff --git a/server/resources/META-INF/cloudstack/core/spring-server-core-managers-context.xml b/server/resources/META-INF/cloudstack/core/spring-server-core-managers-context.xml
index 6d8f32e..cef60a1 100644
--- a/server/resources/META-INF/cloudstack/core/spring-server-core-managers-context.xml
+++ b/server/resources/META-INF/cloudstack/core/spring-server-core-managers-context.xml
@@ -264,4 +264,5 @@
<property name="gslbServiceProviders" value="#{gslbServiceProvidersRegistry.registered}" />
</bean>
<bean id="certServiceImpl" class="org.apache.cloudstack.network.lb.CertServiceImpl" />
+ <bean id="imageStoreUploadMonitorImpl" class="com.cloud.storage.ImageStoreUploadMonitorImpl" />
</beans>
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/server/src/com/cloud/storage/ImageStoreUploadMonitor.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/storage/ImageStoreUploadMonitor.java b/server/src/com/cloud/storage/ImageStoreUploadMonitor.java
new file mode 100755
index 0000000..c093125
--- /dev/null
+++ b/server/src/com/cloud/storage/ImageStoreUploadMonitor.java
@@ -0,0 +1,27 @@
+// 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 com.cloud.storage;
+
+import com.cloud.utils.component.Manager;
+
+/**
+ * Monitor upload progress of all entities.
+ *
+ */
+public interface ImageStoreUploadMonitor extends Manager {
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/server/src/com/cloud/storage/ImageStoreUploadMonitorImpl.java
----------------------------------------------------------------------
diff --git a/server/src/com/cloud/storage/ImageStoreUploadMonitorImpl.java b/server/src/com/cloud/storage/ImageStoreUploadMonitorImpl.java
new file mode 100755
index 0000000..d7deca3
--- /dev/null
+++ b/server/src/com/cloud/storage/ImageStoreUploadMonitorImpl.java
@@ -0,0 +1,250 @@
+// 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 com.cloud.storage;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import javax.ejb.Local;
+import javax.inject.Inject;
+import javax.naming.ConfigurationException;
+
+import org.apache.log4j.Logger;
+import org.springframework.stereotype.Component;
+import org.apache.cloudstack.engine.subsystem.api.storage.DataStore;
+import org.apache.cloudstack.engine.subsystem.api.storage.DataStoreManager;
+import org.apache.cloudstack.engine.subsystem.api.storage.EndPoint;
+import org.apache.cloudstack.engine.subsystem.api.storage.EndPointSelector;
+import org.apache.cloudstack.framework.config.dao.ConfigurationDao;
+import org.apache.cloudstack.managed.context.ManagedContextRunnable;
+import org.apache.cloudstack.storage.command.UploadStatusAnswer;
+import org.apache.cloudstack.storage.command.UploadStatusCommand;
+import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreDao;
+import org.apache.cloudstack.storage.datastore.db.VolumeDataStoreVO;
+import org.apache.cloudstack.utils.identity.ManagementServerNode;
+
+import com.cloud.agent.Listener;
+import com.cloud.agent.api.AgentControlAnswer;
+import com.cloud.agent.api.AgentControlCommand;
+import com.cloud.agent.api.Answer;
+import com.cloud.agent.api.Command;
+import com.cloud.agent.api.StartupCommand;
+import com.cloud.exception.ConnectionException;
+import com.cloud.host.Host;
+import com.cloud.host.Status;
+import com.cloud.host.dao.HostDao;
+import com.cloud.storage.Volume.Event;
+import com.cloud.storage.dao.VolumeDao;
+import com.cloud.utils.component.ManagerBase;
+import com.cloud.utils.concurrency.NamedThreadFactory;
+import com.cloud.utils.db.Transaction;
+import com.cloud.utils.db.TransactionCallbackNoReturn;
+import com.cloud.utils.db.TransactionStatus;
+import com.cloud.utils.fsm.NoTransitionException;
+import com.cloud.utils.fsm.StateMachine2;
+
+/**
+ * Monitors the progress of upload.
+ */
+@Component
+@Local(value = {ImageStoreUploadMonitor.class})
+public class ImageStoreUploadMonitorImpl extends ManagerBase implements ImageStoreUploadMonitor, Listener {
+
+ static final Logger s_logger = Logger.getLogger(ImageStoreUploadMonitorImpl.class);
+
+ @Inject
+ private ConfigurationDao _configDao;
+ @Inject
+ private VolumeDao _volumeDao;
+ @Inject
+ private VolumeDataStoreDao _volumeDataStoreDao;
+ @Inject
+ private HostDao _hostDao;
+ @Inject
+ private EndPointSelector _epSelector;
+ @Inject
+ private DataStoreManager storeMgr;
+
+ private long _nodeId;
+ private ScheduledExecutorService _executor = null;
+ private int _monitoringInterval;
+ private long _uploadOperationTimeout;
+
+ @Override
+ public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
+ _executor = Executors.newScheduledThreadPool(1, new NamedThreadFactory("Upload-Monitor"));
+ _monitoringInterval = 60;
+ _uploadOperationTimeout = 10 * 60 * 1000; // 10 minutes
+ _nodeId = ManagementServerNode.getManagementServerId();
+ return true;
+ }
+
+ @Override
+ public boolean start() {
+ _executor.scheduleWithFixedDelay(new UploadStatusCheck(), _monitoringInterval, _monitoringInterval, TimeUnit.SECONDS);
+ return true;
+ }
+
+ @Override
+ public boolean stop() {
+ _executor.shutdownNow();
+ return true;
+ }
+
+ @Override
+ public boolean processAnswers(long agentId, long seq, Answer[] answers) {
+ return false;
+ }
+
+ @Override
+ public boolean processCommands(long agentId, long seq, Command[] commands) {
+ return false;
+ }
+
+ @Override
+ public AgentControlAnswer processControlCommand(long agentId, AgentControlCommand cmd) {
+ return null;
+ }
+
+ @Override
+ public boolean processDisconnect(long agentId, Status state) {
+ return false;
+ }
+
+ @Override
+ public boolean isRecurring() {
+ return false;
+ }
+
+ @Override
+ public int getTimeout() {
+ return 0;
+ }
+
+ @Override
+ public boolean processTimeout(long agentId, long seq) {
+ return false;
+ }
+
+ @Override
+ public void processConnect(Host host, StartupCommand cmd, boolean forRebalance) throws ConnectionException {
+ }
+
+ protected class UploadStatusCheck extends ManagedContextRunnable {
+
+ public UploadStatusCheck() {
+ }
+
+ @Override
+ protected void runInContext() {
+ // 1. Select all entries with download_state = Not_Downloaded or Download_In_Progress
+ // 2. Get corresponding volume
+ // 3. Get EP using _epSelector
+ // 4. Check if SSVM is owned by this MS
+ // 5. If owned by MS then send command to appropriate SSVM
+ // 6. In listener check for the answer and update DB accordingly
+ List<VolumeDataStoreVO> volumeDataStores = _volumeDataStoreDao.listByVolumeState(Volume.State.NotUploaded, Volume.State.UploadInProgress);
+ for (VolumeDataStoreVO volumeDataStore : volumeDataStores) {
+ DataStore dataStore = storeMgr.getDataStore(volumeDataStore.getDataStoreId(), DataStoreRole.Image);
+ EndPoint ep = _epSelector.select(dataStore, volumeDataStore.getExtractUrl());
+ if (ep == null) {
+ s_logger.warn("There is no secondary storage VM for image store " + dataStore.getName());
+ continue;
+ }
+ VolumeVO volume = _volumeDao.findById(volumeDataStore.getVolumeId());
+ if (volume == null) {
+ s_logger.warn("Volume with id " + volumeDataStore.getVolumeId() + " not found");
+ continue;
+ }
+ Host host = _hostDao.findById(ep.getId());
+ if (host != null && host.getManagementServerId() != null && _nodeId == host.getManagementServerId().longValue()) {
+ UploadStatusCommand cmd = new UploadStatusCommand(volume.getId());
+ Answer answer = ep.sendMessage(cmd);
+ if (answer == null || !(answer instanceof UploadStatusAnswer)) {
+ s_logger.warn("No or invalid answer corresponding to UploadStatusCommand for volume " + volumeDataStore.getVolumeId());
+ continue;
+ }
+ handleStatusResponse((UploadStatusAnswer)answer, volume, volumeDataStore);
+ }
+ }
+ }
+
+ private void handleStatusResponse(final UploadStatusAnswer answer, final VolumeVO volume, final VolumeDataStoreVO volumeDataStore) {
+ final StateMachine2<Volume.State, Event, Volume> stateMachine = Volume.State.getStateMachine();
+ Transaction.execute(new TransactionCallbackNoReturn() {
+ @Override
+ public void doInTransactionWithoutResult(TransactionStatus status) {
+ VolumeVO tmpVolume = _volumeDao.findById(volume.getId());
+ VolumeDataStoreVO tmpVolumeDataStore = _volumeDataStoreDao.findById(volumeDataStore.getId());
+ try {
+ switch (answer.getStatus()) {
+ case COMPLETED:
+ tmpVolumeDataStore.setDownloadState(VMTemplateStorageResourceAssoc.Status.UPLOADED);
+ stateMachine.transitTo(tmpVolume, Event.OperationSucceeded, null, _volumeDao);
+ if (s_logger.isDebugEnabled()) {
+ s_logger.debug("Volume " + tmpVolume.getUuid() + " uploaded successfully");
+ }
+ break;
+ case IN_PROGRESS:
+ tmpVolumeDataStore.setDownloadState(VMTemplateStorageResourceAssoc.Status.UPLOAD_IN_PROGRESS);
+ // check for timeout
+ if (tmpVolumeDataStore.getDownloadState() == VMTemplateStorageResourceAssoc.Status.NOT_UPLOADED ||
+ tmpVolumeDataStore.getDownloadState() == VMTemplateStorageResourceAssoc.Status.UPLOAD_IN_PROGRESS) {
+ if (System.currentTimeMillis() - tmpVolumeDataStore.getCreated().getTime() > _uploadOperationTimeout) {
+ tmpVolumeDataStore.setDownloadState(VMTemplateStorageResourceAssoc.Status.ABANDONED);
+ stateMachine.transitTo(tmpVolume, Event.OperationTimeout, null, _volumeDao);
+ if (s_logger.isDebugEnabled()) {
+ s_logger.debug("Volume " + tmpVolume.getUuid() + " failed to upload due to operation timed out");
+ }
+ }
+ }
+ break;
+ case ERROR:
+ tmpVolumeDataStore.setDownloadState(VMTemplateStorageResourceAssoc.Status.UPLOAD_ERROR);
+ stateMachine.transitTo(tmpVolume, Event.OperationFailed, null, _volumeDao);
+ if (s_logger.isDebugEnabled()) {
+ s_logger.debug("Volume " + tmpVolume.getUuid() + " failed to upload. Error details: " + answer.getDetails());
+ }
+ break;
+ case UNKNOWN:
+ // check for timeout
+ if (tmpVolumeDataStore.getDownloadState() == VMTemplateStorageResourceAssoc.Status.NOT_UPLOADED ||
+ tmpVolumeDataStore.getDownloadState() == VMTemplateStorageResourceAssoc.Status.UPLOAD_IN_PROGRESS) {
+ if (System.currentTimeMillis() - tmpVolumeDataStore.getCreated().getTime() > _uploadOperationTimeout) {
+ tmpVolumeDataStore.setDownloadState(VMTemplateStorageResourceAssoc.Status.ABANDONED);
+ stateMachine.transitTo(tmpVolume, Event.OperationTimeout, null, _volumeDao);
+ if (s_logger.isDebugEnabled()) {
+ s_logger.debug("Volume " + tmpVolume.getUuid() + " failed to upload due to operation timed out");
+ }
+ }
+ }
+ break;
+ }
+ _volumeDataStoreDao.update(tmpVolumeDataStore.getId(), tmpVolumeDataStore);
+ } catch (NoTransitionException e) {
+ s_logger.error("Unexpected error " + e.getMessage());
+ }
+ }
+ });
+ }
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/cloudstack/blob/627f5a62/services/secondary-storage/server/src/org/apache/cloudstack/storage/resource/NfsSecondaryStorageResource.java
----------------------------------------------------------------------
diff --git a/services/secondary-storage/server/src/org/apache/cloudstack/storage/resource/NfsSecondaryStorageResource.java b/services/secondary-storage/server/src/org/apache/cloudstack/storage/resource/NfsSecondaryStorageResource.java
index 3e6ae3f..457f1a2 100755
--- a/services/secondary-storage/server/src/org/apache/cloudstack/storage/resource/NfsSecondaryStorageResource.java
+++ b/services/secondary-storage/server/src/org/apache/cloudstack/storage/resource/NfsSecondaryStorageResource.java
@@ -50,6 +50,7 @@ import com.cloud.utils.nio.HandlerFactory;
import com.cloud.utils.nio.Link;
import com.cloud.utils.nio.NioServer;
import com.cloud.utils.nio.Task;
+
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
@@ -70,6 +71,9 @@ import org.apache.cloudstack.storage.command.CopyCommand;
import org.apache.cloudstack.storage.command.DeleteCommand;
import org.apache.cloudstack.storage.command.DownloadCommand;
import org.apache.cloudstack.storage.command.DownloadProgressCommand;
+import org.apache.cloudstack.storage.command.UploadStatusAnswer;
+import org.apache.cloudstack.storage.command.UploadStatusAnswer.UploadStatus;
+import org.apache.cloudstack.storage.command.UploadStatusCommand;
import org.apache.cloudstack.storage.template.DownloadManager;
import org.apache.cloudstack.storage.template.DownloadManagerImpl;
import org.apache.cloudstack.storage.template.DownloadManagerImpl.ZfsPathParser;
@@ -238,6 +242,8 @@ public class NfsSecondaryStorageResource extends ServerResourceBase implements S
return execute((CopyCommand)cmd);
} else if (cmd instanceof DeleteCommand) {
return execute((DeleteCommand)cmd);
+ } else if (cmd instanceof UploadStatusCommand) {
+ return execute((UploadStatusCommand)cmd);
} else {
return Answer.createUnsupportedCommandAnswer(cmd);
}
@@ -1616,6 +1622,10 @@ public class NfsSecondaryStorageResource extends ServerResourceBase implements S
return new Answer(cmd, success, result);
}
+ private UploadStatusAnswer execute(UploadStatusCommand cmd) {
+ return new UploadStatusAnswer(cmd, UploadStatus.COMPLETED);
+ }
+
protected GetStorageStatsAnswer execute(final GetStorageStatsCommand cmd) {
DataStoreTO store = cmd.getStore();
if (store instanceof S3TO || store instanceof SwiftTO) {