You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@stratos.apache.org by ni...@apache.org on 2014/10/10 16:20:50 UTC
[1/5] git commit: Changes to member context removal logic.
Repository: stratos
Updated Branches:
refs/heads/container-autoscaling ff3722268 -> f37ece08b
Changes to member context removal logic.
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/e57843ce
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/e57843ce
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/e57843ce
Branch: refs/heads/container-autoscaling
Commit: e57843ced3d1fa1da0f6ad51e315427a596a7324
Parents: 0fe9648
Author: Nirmal Fernando <ni...@gmail.com>
Authored: Fri Oct 10 19:48:30 2014 +0530
Committer: Nirmal Fernando <ni...@gmail.com>
Committed: Fri Oct 10 19:50:27 2014 +0530
----------------------------------------------------------------------
.../runtime/FasterLookUpDataHolder.java | 69 +++++++++++++++-----
1 file changed, 51 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/e57843ce/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/runtime/FasterLookUpDataHolder.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/runtime/FasterLookUpDataHolder.java b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/runtime/FasterLookUpDataHolder.java
index a8c47ce..2a83908 100644
--- a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/runtime/FasterLookUpDataHolder.java
+++ b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/runtime/FasterLookUpDataHolder.java
@@ -32,6 +32,7 @@ import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledFuture;
/**
* This object holds all runtime data and provides faster access. This is a Singleton class.
@@ -60,6 +61,12 @@ public class FasterLookUpDataHolder implements Serializable{
private Map<String, MemberContext> memberIdToContext = new ConcurrentHashMap<String, MemberContext>();
/**
+ * Key - member id
+ * Value - ScheduledFuture task
+ */
+ private transient Map<String, ScheduledFuture<?>> memberIdToScheduledTask = new ConcurrentHashMap<String, ScheduledFuture<?>>();
+
+ /**
* Key - Kubernetes cluster id
* Value - {@link KubernetesClusterContext}
*/
@@ -278,42 +285,60 @@ public class FasterLookUpDataHolder implements Serializable{
}
}
- public void removeMemberContextsOfCluster(String clusterId) {
+ public void addScheduledFutureJob(String memberId, ScheduledFuture<?> job) {
+ memberIdToScheduledTask.put(memberId, job);
+ }
+
+ public List<MemberContext> removeMemberContextsOfCluster(String clusterId) {
List<MemberContext> ctxts = clusterIdToMemberContext.remove(clusterId);
if(ctxts == null) {
- return;
+ return new ArrayList<MemberContext>();
}
for (MemberContext memberContext : ctxts) {
String memberId = memberContext.getMemberId();
memberIdToContext.remove(memberId);
+ stopTask(memberIdToScheduledTask.remove(memberId));
}
if(log.isDebugEnabled()) {
log.debug("Removed Member Context from the information model. "+ctxt);
}
+ return ctxts;
}
- public void removeMemberContext(String memberId, String clusterId) {
- memberIdToContext.remove(memberId);
+ public MemberContext removeMemberContext(String memberId, String clusterId) {
+ MemberContext returnedCtxt = memberIdToContext.remove(memberId);
List<MemberContext> ctxts = clusterIdToMemberContext.get(clusterId);
- if(ctxts == null) {
- return;
+
+ if (ctxts != null) {
+
+ List<MemberContext> newCtxts = new ArrayList<MemberContext>(ctxts);
+
+ for (Iterator<MemberContext> iterator = newCtxts.iterator(); iterator.hasNext();) {
+ MemberContext memberContext = (MemberContext) iterator.next();
+ if(memberId.equals(memberContext.getMemberId())) {
+ if(log.isDebugEnabled()) {
+
+ log.debug("MemberContext [id]: "+memberId+" removed from information model.");
+ }
+ iterator.remove();
+ }
+ }
+
+ clusterIdToMemberContext.put(clusterId, newCtxts);
}
- List<MemberContext> newCtxts = new ArrayList<MemberContext>(ctxts);
+ stopTask(memberIdToScheduledTask.remove(memberId));
- for (Iterator<MemberContext> iterator = newCtxts.iterator(); iterator.hasNext();) {
- MemberContext memberContext = (MemberContext) iterator.next();
- if(memberId.equals(memberContext.getMemberId())) {
- if(log.isDebugEnabled()) {
-
- log.debug("MemberContext [id]: "+memberId+" removed from information model.");
- }
- iterator.remove();
- }
- }
+ return returnedCtxt;
- clusterIdToMemberContext.put(clusterId, newCtxts);
+ }
+
+ private void stopTask(ScheduledFuture<?> task) {
+ if (task != null) {
+
+ task.cancel(true);
+ }
}
public MemberContext getMemberContextOfMemberId(String memberId) {
@@ -408,5 +433,13 @@ public class FasterLookUpDataHolder implements Serializable{
Map<String, KubernetesClusterContext> kubClusterIdToKubClusterContext) {
this.kubClusterIdToKubClusterContext = kubClusterIdToKubClusterContext;
}
+
+ public Map<String, ScheduledFuture<?>> getMemberIdToScheduledTask() {
+ return memberIdToScheduledTask;
+ }
+
+ public void setMemberIdToScheduledTask(Map<String, ScheduledFuture<?>> memberIdToScheduledTask) {
+ this.memberIdToScheduledTask = memberIdToScheduledTask;
+ }
}
\ No newline at end of file
[4/5] git commit: PodActivationWatcher is responsible for checking
Pod's state and setting IPs and sending member spawned event.
Posted by ni...@apache.org.
PodActivationWatcher is responsible for checking Pod's state and setting IPs and sending member spawned event.
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/ff2a02fa
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/ff2a02fa
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/ff2a02fa
Branch: refs/heads/container-autoscaling
Commit: ff2a02fa1e1428ccb129774618c258aa83da3e6d
Parents: 78ee28b
Author: Nirmal Fernando <ni...@gmail.com>
Authored: Fri Oct 10 19:44:39 2014 +0530
Committer: Nirmal Fernando <ni...@gmail.com>
Committed: Fri Oct 10 19:50:27 2014 +0530
----------------------------------------------------------------------
.../controller/util/PodActivationWatcher.java | 74 ++++++++++++++++++++
1 file changed, 74 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/ff2a02fa/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/util/PodActivationWatcher.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/util/PodActivationWatcher.java b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/util/PodActivationWatcher.java
new file mode 100644
index 0000000..13d1ac6
--- /dev/null
+++ b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/util/PodActivationWatcher.java
@@ -0,0 +1,74 @@
+/*
+ * 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.stratos.cloud.controller.util;
+
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.stratos.cloud.controller.pojo.MemberContext;
+import org.apache.stratos.cloud.controller.registry.RegistryManager;
+import org.apache.stratos.cloud.controller.runtime.FasterLookUpDataHolder;
+import org.apache.stratos.cloud.controller.topology.TopologyBuilder;
+import org.apache.stratos.kubernetes.client.KubernetesApiClient;
+import org.apache.stratos.kubernetes.client.model.Pod;
+
+/**
+ * Checks whether a container is active and update the {@link FasterLookUpDataHolder}.
+ */
+public class PodActivationWatcher implements Runnable {
+
+ private static final Log LOG = LogFactory
+ .getLog(PodActivationWatcher.class);
+ private String podId;
+ private MemberContext ctxt;
+ private KubernetesApiClient kubApi;
+
+ public PodActivationWatcher(String podId, MemberContext ctxt, KubernetesApiClient kubApi) {
+ this.podId = podId;
+ this.ctxt = ctxt;
+ this.kubApi = kubApi;
+ }
+
+ @Override
+ public void run() {
+ try {
+ Pod pod = kubApi.getPod(podId);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("PodActivationWatcher running : "+pod.getCurrentState().getStatus());
+ }
+ if ("Running".equals(pod.getCurrentState().getStatus())) {
+ String hostIP = pod.getCurrentState().getHost();
+ ctxt.setPublicIpAddress(hostIP);
+ ctxt.setPrivateIpAddress(hostIP);
+ FasterLookUpDataHolder.getInstance().addMemberContext(ctxt);
+ // trigger topology
+ TopologyBuilder.handleMemberSpawned(ctxt.getCartridgeType(), ctxt.getClusterId(),
+ null, hostIP, hostIP, ctxt);
+
+ RegistryManager.getInstance().persist(FasterLookUpDataHolder.getInstance());
+
+ }
+
+ } catch (Exception e) {
+ LOG.error("Container Activation Watcher Failed.. ", e);
+ }
+
+ }
+
+}
[5/5] git commit: Return member contexts from termination related
APIs and allocate IPs in a scheduled executor task.
Posted by ni...@apache.org.
Return member contexts from termination related APIs and allocate IPs in a scheduled executor task.
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/f37ece08
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/f37ece08
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/f37ece08
Branch: refs/heads/container-autoscaling
Commit: f37ece08b47a1182cde73d1a59934d1372c9867c
Parents: e57843c
Author: Nirmal Fernando <ni...@gmail.com>
Authored: Fri Oct 10 19:50:15 2014 +0530
Committer: Nirmal Fernando <ni...@gmail.com>
Committed: Fri Oct 10 19:50:28 2014 +0530
----------------------------------------------------------------------
.../impl/CloudControllerServiceImpl.java | 378 +++++++++----------
.../interfaces/CloudControllerService.java | 8 +-
.../jcloud/ComputeServiceBuilderUtil.java | 1 -
3 files changed, 190 insertions(+), 197 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/f37ece08/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/impl/CloudControllerServiceImpl.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/impl/CloudControllerServiceImpl.java b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/impl/CloudControllerServiceImpl.java
index be8702b..25e19ce 100644
--- a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/impl/CloudControllerServiceImpl.java
+++ b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/impl/CloudControllerServiceImpl.java
@@ -25,6 +25,7 @@ import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.stratos.cloud.controller.concurrent.PartitionValidatorCallable;
+import org.apache.stratos.cloud.controller.concurrent.ScheduledThreadExecutor;
import org.apache.stratos.cloud.controller.concurrent.ThreadExecutor;
import org.apache.stratos.cloud.controller.deployment.partition.Partition;
import org.apache.stratos.cloud.controller.exception.*;
@@ -42,6 +43,7 @@ import org.apache.stratos.cloud.controller.topology.TopologyBuilder;
import org.apache.stratos.cloud.controller.topology.TopologyManager;
import org.apache.stratos.cloud.controller.util.CloudControllerConstants;
import org.apache.stratos.cloud.controller.util.CloudControllerUtil;
+import org.apache.stratos.cloud.controller.util.PodActivationWatcher;
import org.apache.stratos.cloud.controller.validate.interfaces.PartitionValidator;
import org.apache.stratos.common.constants.StratosConstants;
import org.apache.stratos.kubernetes.client.KubernetesApiClient;
@@ -74,7 +76,7 @@ import java.util.concurrent.Future;
*/
public class CloudControllerServiceImpl implements CloudControllerService {
- private static final Log log = LogFactory
+ private static final Log LOG = LogFactory
.getLog(CloudControllerServiceImpl.class);
private FasterLookUpDataHolder dataHolder = FasterLookUpDataHolder
.getInstance();
@@ -103,20 +105,20 @@ public class CloudControllerServiceImpl implements CloudControllerService {
currentData.setCartridges(serializedObj.getCartridges());
currentData.setKubClusterIdToKubClusterContext(serializedObj.getKubClusterIdToKubClusterContext());
- if(log.isDebugEnabled()) {
+ if(LOG.isDebugEnabled()) {
- log.debug("Cloud Controller Data is retrieved from registry.");
+ LOG.debug("Cloud Controller Data is retrieved from registry.");
}
} else {
- if(log.isDebugEnabled()) {
+ if(LOG.isDebugEnabled()) {
- log.debug("Cloud Controller Data cannot be found in registry.");
+ LOG.debug("Cloud Controller Data cannot be found in registry.");
}
}
} catch (Exception e) {
String msg = "Unable to acquire data from Registry. Hence, any historical data will not get reflected.";
- log.warn(msg, e);
+ LOG.warn(msg, e);
}
}
@@ -124,26 +126,23 @@ public class CloudControllerServiceImpl implements CloudControllerService {
public void deployCartridgeDefinition(CartridgeConfig cartridgeConfig) throws InvalidCartridgeDefinitionException,
InvalidIaasProviderException {
- if (cartridgeConfig == null) {
- String msg = "Invalid Cartridge Definition: Definition is null.";
- log.error(msg);
- throw new IllegalArgumentException(msg);
-
- }
+
+ handleNullObject(cartridgeConfig, "Invalid Cartridge Definition: Definition is null.");
- if(log.isDebugEnabled()){
- log.debug("Cartridge definition: " + cartridgeConfig.toString());
+ if(LOG.isDebugEnabled()){
+ LOG.debug("Cartridge definition: " + cartridgeConfig.toString());
}
Cartridge cartridge = null;
try {
+ // cartridge can never be null
cartridge = CloudControllerUtil.toCartridge(cartridgeConfig);
} catch (Exception e) {
String msg =
"Invalid Cartridge Definition: Cartridge Type: " +
cartridgeConfig.getType()+
". Cause: Cannot instantiate a Cartridge Instance with the given Config. "+e.getMessage();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new InvalidCartridgeDefinitionException(msg, e);
}
@@ -154,7 +153,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg = "Invalid Cartridge Definition: Cartridge Type: "
+ cartridgeConfig.getType()
+ ". Cause: Iaases of this Cartridge is null or empty.";
- log.error(msg);
+ LOG.error(msg);
throw new InvalidCartridgeDefinitionException(msg);
}
@@ -187,7 +186,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
TopologyBuilder.handleServiceCreated(cartridgeList);
// transaction ends
- log.info("Successfully deployed the Cartridge definition: " + cartridgeType);
+ LOG.info("Successfully deployed the Cartridge definition: " + cartridgeType);
}
private void populateNewCartridge(Cartridge cartridge,
@@ -203,8 +202,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String partitionId = entry.getKey();
IaasProvider oldIaasProvider = entry.getValue();
if (newIaasProviders.contains(oldIaasProvider)) {
- if (log.isDebugEnabled()) {
- log.debug("Copying a partition from the Cartridge that is undeployed, to the new Cartridge. "
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Copying a partition from the Cartridge that is undeployed, to the new Cartridge. "
+ "[partition id] : "+partitionId+" [cartridge type] "+cartridge.getType() );
}
cartridge.addIaasProvider(partitionId, newIaasProviders.get(newIaasProviders.indexOf(oldIaasProvider)));
@@ -221,8 +220,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// invalidate partition validation cache
dataHolder.removeFromCartridgeTypeToPartitionIds(cartridgeType);
- if (log.isDebugEnabled()) {
- log.debug("Partition cache invalidated for cartridge "+cartridgeType);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition cache invalidated for cartridge "+cartridgeType);
}
persist();
@@ -232,14 +231,14 @@ public class CloudControllerServiceImpl implements CloudControllerService {
cartridgeList.add(cartridge);
TopologyBuilder.handleServiceRemoved(cartridgeList);
- if(log.isInfoEnabled()) {
- log.info("Successfully undeployed the Cartridge definition: " + cartridgeType);
+ if(LOG.isInfoEnabled()) {
+ LOG.info("Successfully undeployed the Cartridge definition: " + cartridgeType);
}
return;
}
}
String msg = "Cartridge [type] "+cartridgeType+" is not a deployed Cartridge type.";
- log.error(msg);
+ LOG.error(msg);
throw new InvalidCartridgeTypeException(msg);
}
@@ -247,41 +246,28 @@ public class CloudControllerServiceImpl implements CloudControllerService {
public MemberContext startInstance(MemberContext memberContext) throws
UnregisteredCartridgeException, InvalidIaasProviderException {
- if(log.isDebugEnabled()) {
- log.debug("CloudControllerServiceImpl:startInstance");
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("CloudControllerServiceImpl:startInstance");
}
- if (memberContext == null) {
- String msg = "Instance start-up failed. Member is null.";
- log.error(msg);
- throw new IllegalArgumentException(msg);
- }
+ handleNullObject(memberContext, "Instance start-up failed. Member is null.");
String clusterId = memberContext.getClusterId();
Partition partition = memberContext.getPartition();
- if(log.isDebugEnabled()) {
- log.debug("Received an instance spawn request : " + memberContext.toString());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Received an instance spawn request : " + memberContext);
}
Template template = null;
- if (partition == null) {
- String msg =
- "Instance start-up failed. Specified Partition is null. " +
- memberContext.toString();
- log.error(msg);
- throw new IllegalArgumentException(msg);
- }
+ handleNullObject(partition, "Instance start-up failed. Specified Partition is null. " +
+ memberContext);
String partitionId = partition.getId();
ClusterContext ctxt = dataHolder.getClusterContext(clusterId);
- if (ctxt == null) {
- String msg = "Instance start-up failed. Invalid cluster id. " + memberContext.toString();
- log.error(msg);
- throw new IllegalArgumentException(msg);
- }
+ handleNullObject(ctxt, "Instance start-up failed. Invalid cluster id. " + memberContext);
String cartridgeType = ctxt.getCartridgeType();
@@ -291,7 +277,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Instance start-up failed. No matching Cartridge found [type] "+cartridgeType +". "+
memberContext.toString();
- log.error(msg);
+ LOG.error(msg);
throw new UnregisteredCartridgeException(msg);
}
@@ -300,8 +286,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
IaasProvider iaasProvider = cartridge.getIaasProviderOfPartition(partitionId);
if (iaasProvider == null) {
- if (log.isDebugEnabled()) {
- log.debug("IaasToPartitionMap "+cartridge.hashCode()
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("IaasToPartitionMap "+cartridge.hashCode()
+ " for cartridge "+cartridgeType+ " and for partition: "+partitionId);
}
String msg = "Instance start-up failed. "
@@ -314,7 +300,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
+ cartridge.getPartitionToIaasProvider().keySet()
.toString() + ". " + memberContext.toString()
+ ". ";
- log.fatal(msg);
+ LOG.fatal(msg);
throw new InvalidIaasProviderException(msg);
}
String type = iaasProvider.getType();
@@ -339,20 +325,20 @@ public class CloudControllerServiceImpl implements CloudControllerService {
Iaas iaas = iaasProvider.getIaas();
- if (log.isDebugEnabled()) {
- log.debug("Payload: " + payload.toString());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Payload: " + payload.toString());
}
if (iaas == null) {
- if(log.isDebugEnabled()) {
- log.debug("Iaas is null of Iaas Provider: "+type+". Trying to build IaaS...");
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Iaas is null of Iaas Provider: "+type+". Trying to build IaaS...");
}
try {
iaas = CloudControllerUtil.getIaas(iaasProvider);
} catch (InvalidIaasProviderException e) {
String msg ="Instance start up failed. "+memberContext.toString()+
"Unable to build Iaas of this IaasProvider [Provider] : " + type+". Cause: "+e.getMessage();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new InvalidIaasProviderException(msg, e);
}
@@ -383,24 +369,24 @@ public class CloudControllerServiceImpl implements CloudControllerService {
"Failed to start an instance. " +
memberContext.toString() +
". Reason : Jclouds Template is null for iaas provider [type]: "+iaasProvider.getType();
- log.error(msg);
+ LOG.error(msg);
throw new InvalidIaasProviderException(msg);
}
//Start instance start up in a new thread
ThreadExecutor exec = ThreadExecutor.getInstance();
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller is starting the instance start up thread.");
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller is starting the instance start up thread.");
}
exec.execute(new JcloudsInstanceCreator(memberContext, iaasProvider, cartridgeType));
- log.info("Instance is successfully starting up. "+memberContext.toString());
+ LOG.info("Instance is successfully starting up. "+memberContext.toString());
return memberContext;
} catch (Exception e) {
String msg = "Failed to start an instance. " + memberContext.toString()+" Cause: "+e.getMessage();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new IllegalStateException(msg, e);
}
@@ -414,8 +400,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String snapshotId = volume.getSnapshotId();
if(StringUtils.isNotEmpty(volume.getVolumeId())){
// volumeID is specified, so not creating additional volumes
- if(log.isDebugEnabled()){
- log.debug("Volume creation is skipping since a volume ID is specified. [Volume ID]" + volume.getVolumeId());
+ if(LOG.isDebugEnabled()){
+ LOG.debug("Volume creation is skipping since a volume ID is specified. [Volume ID]" + volume.getVolumeId());
}
volume.setId(volume.getVolumeId());
}else{
@@ -431,8 +417,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
StringBuilder persistencePayload = new StringBuilder();
if(isPersistenceMappingAvailable(ctx)){
for(Volume volume : ctx.getVolumes()){
- if(log.isDebugEnabled()){
- log.debug("Adding persistence mapping " + volume.toString());
+ if(LOG.isDebugEnabled()){
+ LOG.debug("Adding persistence mapping " + volume.toString());
}
if(persistencePayload.length() != 0) {
persistencePayload.append("|");
@@ -445,8 +431,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
persistencePayload.append(volume.getMappingPath());
}
}
- if(log.isDebugEnabled()){
- log.debug("Persistence payload is" + persistencePayload.toString());
+ if(LOG.isDebugEnabled()){
+ LOG.debug("Persistence payload is" + persistencePayload.toString());
}
return persistencePayload;
}
@@ -470,7 +456,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
} catch (RegistryException e) {
String msg = "Failed to persist the Cloud Controller data in registry. Further, transaction roll back also failed.";
- log.fatal(msg);
+ LOG.fatal(msg);
throw new CloudControllerException(msg, e);
}
}
@@ -486,7 +472,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if(memberId == null) {
String msg = "Termination failed. Null member id.";
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -494,7 +480,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if(ctxt == null) {
String msg = "Termination failed. Invalid Member Id: "+memberId;
- log.error(msg);
+ LOG.error(msg);
throw new InvalidMemberException(msg);
}
@@ -524,7 +510,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// these will never be null, since we do not add null values for these.
Cartridge cartridge = dataHolder.getCartridge(cartridgeType);
- log.info("Starting to terminate an instance with member id : " + memberId +
+ LOG.info("Starting to terminate an instance with member id : " + memberId +
" in partition id: " + partitionId + " of cluster id: " + clusterId +
" and of cartridge type: " + cartridgeType);
@@ -533,7 +519,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
"Termination of Member Id: " + memberId + " failed. " +
"Cannot find a matching Cartridge for type: " +
cartridgeType;
- log.error(msg);
+ LOG.error(msg);
throw new InvalidCartridgeTypeException(msg);
}
@@ -543,7 +529,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Termination failed. Cannot find a node id for Member Id: " +
memberId;
- log.error(msg);
+ LOG.error(msg);
throw new InvalidMemberException(msg);
}
@@ -558,7 +544,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
} catch (Exception e) {
String msg =
"Instance termination failed. "+ctxt.toString();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new CloudControllerException(msg, e);
}
@@ -600,23 +586,23 @@ public class CloudControllerServiceImpl implements CloudControllerService {
.getComputeService();
Template template = iaasProvider.getTemplate();
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller is delegating request to start an instance for "
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller is delegating request to start an instance for "
+ memberContext + " to Jclouds layer.");
}
// create and start a node
Set<? extends NodeMetadata> nodes = computeService
.createNodesInGroup(group, 1, template);
node = nodes.iterator().next();
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller received a response for the request to start "
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller received a response for the request to start "
+ memberContext + " from Jclouds layer.");
}
if (node == null) {
String msg = "Null response received for instance start-up request to Jclouds.\n"
+ memberContext.toString();
- log.error(msg);
+ LOG.error(msg);
throw new IllegalStateException(msg);
}
@@ -625,13 +611,13 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (nodeId == null) {
String msg = "Node id of the starting instance is null.\n"
+ memberContext.toString();
- log.fatal(msg);
+ LOG.fatal(msg);
throw new IllegalStateException(msg);
}
memberContext.setNodeId(nodeId);
- if (log.isDebugEnabled()) {
- log.debug("Node id was set. " + memberContext.toString());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Node id was set. " + memberContext.toString());
}
// attach volumes
@@ -649,7 +635,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
} catch (Exception e) {
// continue without throwing an exception, since
// there is an instance already running
- log.error("Attaching Volume to Instance [ "
+ LOG.error("Attaching Volume to Instance [ "
+ instanceId + " ] failed!", e);
}
}
@@ -658,13 +644,13 @@ public class CloudControllerServiceImpl implements CloudControllerService {
} catch (Exception e) {
String msg = "Failed to start an instance. " + memberContext.toString()+" Cause: "+e.getMessage();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new IllegalStateException(msg, e);
}
try{
- if (log.isDebugEnabled()) {
- log.debug("IP allocation process started for "+memberContext);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("IP allocation process started for "+memberContext);
}
String autoAssignIpProp =
iaasProvider.getProperty(CloudControllerConstants.AUTO_ASSIGN_IP_PROPERTY);
@@ -682,8 +668,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// check if floating ip is well defined in cartridge definition
if (pre_defined_ip != null) {
if (isValidIpAddress(pre_defined_ip)) {
- if(log.isDebugEnabled()) {
- log.debug("CloudControllerServiceImpl:IpAllocator:pre_defined_ip: invoking associatePredefinedAddress" + pre_defined_ip);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("CloudControllerServiceImpl:IpAllocator:pre_defined_ip: invoking associatePredefinedAddress" + pre_defined_ip);
}
ip = iaas.associatePredefinedAddress(node, pre_defined_ip);
@@ -692,7 +678,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg = "Error occurred while allocating predefined floating ip address: " + pre_defined_ip +
" / allocated ip:" + ip +
" - terminating node:" + memberContext.toString();
- log.error(msg);
+ LOG.error(msg);
// terminate instance
terminate(iaasProvider,
node.getId(), memberContext);
@@ -701,7 +687,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
} else {
String msg = "Invalid floating ip address configured: " + pre_defined_ip +
" - terminating node:" + memberContext.toString();
- log.error(msg);
+ LOG.error(msg);
// terminate instance
terminate(iaasProvider,
node.getId(), memberContext);
@@ -709,8 +695,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
}
} else {
- if(log.isDebugEnabled()) {
- log.debug("CloudControllerServiceImpl:IpAllocator:no (valid) predefined floating ip configured, "
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("CloudControllerServiceImpl:IpAllocator:no (valid) predefined floating ip configured, "
+ "selecting available one from pool");
}
// allocate an IP address - manual IP assigning mode
@@ -718,7 +704,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (ip != null) {
memberContext.setAllocatedIpAddress(ip);
- log.info("Allocated an ip address: "
+ LOG.info("Allocated an ip address: "
+ memberContext.toString());
}
}
@@ -735,7 +721,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
ip = node.getPublicAddresses().iterator().next();
publicIp = ip;
memberContext.setPublicIpAddress(ip);
- log.info("Retrieving Public IP Address : " + memberContext.toString());
+ LOG.info("Retrieving Public IP Address : " + memberContext.toString());
}
// private IP
@@ -743,7 +729,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
node.getPrivateAddresses().iterator().hasNext()) {
ip = node.getPrivateAddresses().iterator().next();
memberContext.setPrivateIpAddress(ip);
- log.info("Retrieving Private IP Address. " + memberContext.toString());
+ LOG.info("Retrieving Private IP Address. " + memberContext.toString());
}
dataHolder.addMemberContext(memberContext);
@@ -767,17 +753,17 @@ public class CloudControllerServiceImpl implements CloudControllerService {
cartridgeType,
MemberStatus.Created.toString(),
node);
- if (log.isDebugEnabled()) {
- log.debug("Node details: " + node.toString());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Node details: " + node.toString());
}
- if (log.isDebugEnabled()) {
- log.debug("IP allocation process ended for "+memberContext);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("IP allocation process ended for "+memberContext);
}
} catch (Exception e) {
String msg = "Error occurred while allocating an ip address. " + memberContext.toString();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new CloudControllerException(msg, e);
}
@@ -793,12 +779,12 @@ public class CloudControllerServiceImpl implements CloudControllerService {
@Override
public void terminateAllInstances(String clusterId) throws InvalidClusterException {
- log.info("Starting to terminate all instances of cluster : "
+ LOG.info("Starting to terminate all instances of cluster : "
+ clusterId);
if(clusterId == null) {
String msg = "Instance termination failed. Cluster id is null.";
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -806,7 +792,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if(ctxts == null) {
String msg = "Instance termination failed. No members found for cluster id: "+clusterId;
- log.warn(msg);
+ LOG.warn(msg);
return;
}
@@ -836,7 +822,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Instance termination failed. " +ctxt.toString() +
". Cause: Unable to build Iaas of this " + iaasProvider.toString();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new CloudControllerException(msg, e);
}
@@ -853,7 +839,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
iaas.releaseAddress(ctxt.getAllocatedIpAddress());
}
- log.info("Member is terminated: "+ctxt.toString());
+ LOG.info("Member is terminated: "+ctxt.toString());
return iaasProvider;
}
@@ -870,8 +856,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
Iaas iaas = iaasProvider.getIaas();
iaas.detachVolume(ctxt.getInstanceId(), volumeId);
} catch (ResourceNotFoundException ignore) {
- if(log.isDebugEnabled()) {
- log.debug(ignore);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(ignore);
}
}
}
@@ -914,7 +900,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if(cartridgeType == null || clusterId == null || payload == null || hostName == null) {
String msg = "Null Argument/s detected: Cartridge type: "+cartridgeType+", " +
"Cluster Id: "+clusterId+", Payload: "+payload+", Host name: "+hostName;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -923,7 +909,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg = "Registration of cluster: "+clusterId+
" failed. - Unregistered Cartridge type: " + cartridgeType;
- log.error(msg);
+ LOG.error(msg);
throw new UnregisteredCartridgeException(msg);
}
@@ -940,7 +926,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
persist();
- log.info("Successfully registered: "+registrant);
+ LOG.info("Successfully registered: "+registrant);
return true;
}
@@ -1050,7 +1036,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg = "Cannot find a Cartridge having a type of "
+ cartridgeType + ". Hence unable to find information.";
- log.error(msg);
+ LOG.error(msg);
throw new UnregisteredCartridgeException(msg);
}
@@ -1062,7 +1048,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (ctxt == null) {
String msg = "Instance start-up failed. Invalid cluster id. " + clusterId;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1073,7 +1059,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (cartridge == null) {
String msg =
"Instance start-up failed. No matching Cartridge found [type] "+cartridgeType +". ";
- log.error(msg);
+ LOG.error(msg);
throw new UnregisteredClusterException(msg);
}
@@ -1091,7 +1077,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
ClusterContext ctxt = dataHolder.getClusterContext(clusterId_);
if(ctxt == null) {
String msg = "Unregistration of service cluster failed. Cluster not found: " + clusterId_;
- log.error(msg);
+ LOG.error(msg);
}
Collection<Member> members = TopologyManager.getTopology().
getService(ctxt.getCartridgeType()).getCluster(clusterId_).getMembers();
@@ -1118,7 +1104,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
terminateInstance(member.getMemberId());
} catch (Exception e) {
// we are not gonna stop the execution due to errors.
- log.warn("Instance termination failed of member [id] " + member.getMemberId(), e);
+ LOG.warn("Instance termination failed of member [id] " + member.getMemberId(), e);
}
}
}
@@ -1129,7 +1115,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
ClusterContext ctxt = dataHolder.getClusterContext(clusterId_);
if(ctxt == null) {
String msg = "Unregistration of service cluster failed. Cluster not found: " + clusterId_;
- log.error(msg);
+ LOG.error(msg);
}
Collection<Member> members = TopologyManager.getTopology().
getService(ctxt.getCartridgeType()).getCluster(clusterId_).getMembers();
@@ -1141,7 +1127,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
CloudControllerUtil.sleep(1000);
}
- log.info("Unregistration of service cluster: " + clusterId_);
+ LOG.info("Unregistration of service cluster: " + clusterId_);
deleteVolumes(ctxt);
onClusterRemoval(clusterId_);
}
@@ -1164,8 +1150,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
volume.setId(null);
}
} catch(Exception ignore) {
- if(log.isErrorEnabled()) {
- log.error("Error while deleting volume [id] "+ volume.getId(), ignore);
+ if(LOG.isErrorEnabled()) {
+ LOG.error("Error while deleting volume [id] "+ volume.getId(), ignore);
}
}
}
@@ -1190,7 +1176,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
terminateAllContainers(clusterId);
} catch (InvalidClusterException e) {
String msg = "Docker instance termination fails for cluster: "+clusterId;
- log.error(msg, e);
+ LOG.error(msg, e);
throw new UnregisteredClusterException(msg, e);
}
// send cluster removal notifications and update the state
@@ -1209,8 +1195,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// cache hit for this cartridge
// get list of partitions
validatedPartitions = validatedCache.get(cartridgeType);
- if (log.isDebugEnabled()) {
- log.debug("Partition validation cache hit for cartridge type: "+cartridgeType);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition validation cache hit for cartridge type: "+cartridgeType);
}
}
@@ -1218,15 +1204,15 @@ public class CloudControllerServiceImpl implements CloudControllerService {
Map<String, IaasProvider> partitionToIaasProviders =
new ConcurrentHashMap<String, IaasProvider>();
- if (log.isDebugEnabled()) {
- log.debug("Deployment policy validation started for cartridge type: "+cartridgeType);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Deployment policy validation started for cartridge type: "+cartridgeType);
}
Cartridge cartridge = dataHolder.getCartridge(cartridgeType);
if (cartridge == null) {
String msg = "Invalid Cartridge Type: " + cartridgeType;
- log.error(msg);
+ LOG.error(msg);
throw new InvalidCartridgeTypeException(msg);
}
@@ -1260,11 +1246,11 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// add to cache
this.dataHolder.addToCartridgeTypeToPartitionIdMap(cartridgeType, partitionId);
- if (log.isDebugEnabled()) {
- log.debug("Partition "+partitionId+" added to the cache against cartridge type: "+cartridgeType);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Partition "+partitionId+" added to the cache against cartridge type: "+cartridgeType);
}
} catch (Exception e) {
- log.error(e.getMessage(), e);
+ LOG.error(e.getMessage(), e);
throw new InvalidPartitionException(e.getMessage(), e);
}
}
@@ -1275,7 +1261,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// persist data
persist();
- log.info("All partitions "+CloudControllerUtil.getPartitionIds(partitions)+
+ LOG.info("All partitions "+CloudControllerUtil.getPartitionIds(partitions)+
" were validated successfully, against the Cartridge: "+cartridgeType);
return true;
@@ -1299,7 +1285,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Invalid Partition - " + partition.toString()+". Cause: Iaas Provider " +
"is null for Partition Provider: "+provider;
- log.error(msg);
+ LOG.error(msg);
throw new InvalidPartitionException(msg);
}
@@ -1313,7 +1299,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Invalid Partition - " + partition.toString() +
". Cause: Unable to build Iaas of this IaasProvider [Provider] : " + provider+". "+e.getMessage();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new InvalidPartitionException(msg, e);
}
@@ -1336,26 +1322,26 @@ public class CloudControllerServiceImpl implements CloudControllerService {
public MemberContext[] startContainers(ContainerClusterContext containerClusterContext)
throws UnregisteredCartridgeException {
- if(log.isDebugEnabled()) {
- log.debug("CloudControllerServiceImpl:startContainers");
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("CloudControllerServiceImpl:startContainers");
}
if (containerClusterContext == null) {
String msg = "Instance start-up failed. ContainerClusterContext is null.";
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
String clusterId = containerClusterContext.getClusterId();
- if(log.isDebugEnabled()) {
- log.debug("Received an instance spawn request : " + containerClusterContext.toString());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Received an instance spawn request : " + containerClusterContext.toString());
}
ClusterContext ctxt = dataHolder.getClusterContext(clusterId);
if (ctxt == null) {
String msg = "Instance start-up failed. Invalid cluster id. " + containerClusterContext.toString();
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1367,7 +1353,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Instance start-up failed. No matching Cartridge found [type] "+cartridgeType +". "+
containerClusterContext.toString();
- log.error(msg);
+ LOG.error(msg);
throw new UnregisteredCartridgeException(msg);
}
@@ -1385,15 +1371,15 @@ public class CloudControllerServiceImpl implements CloudControllerService {
ContainerClusterContextToReplicationController controllerFunction = new ContainerClusterContextToReplicationController();
ReplicationController controller = controllerFunction.apply(containerClusterContext);
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller is delegating request to start a replication controller "+controller+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller is delegating request to start a replication controller "+controller+
" for "+ containerClusterContext + " to Kubernetes layer.");
}
kubApi.createReplicationController(controller);
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller successfully started the controller "
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller successfully started the controller "
+ controller + " via Kubernetes layer.");
}
@@ -1401,8 +1387,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
ContainerClusterContextToKubernetesService serviceFunction = new ContainerClusterContextToKubernetesService();
Service service = serviceFunction.apply(containerClusterContext);
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller is delegating request to start a service "+service+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller is delegating request to start a service "+service+
" for "+ containerClusterContext + " to Kubernetes layer.");
}
@@ -1412,8 +1398,8 @@ public class CloudControllerServiceImpl implements CloudControllerService {
ctxt.addProperty(StratosConstants.ALLOCATED_SERVICE_HOST_PORT, service.getPort());
dataHolder.addClusterContext(ctxt);
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller successfully started the service "
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller successfully started the service "
+ controller + " via Kubernetes layer.");
}
@@ -1427,9 +1413,9 @@ public class CloudControllerServiceImpl implements CloudControllerService {
for (int i = 0; i < expectedCount ; i++) {
newlyCreatedPods = kubApi.getSelectedPods(new Label[]{l});
- if (log.isDebugEnabled()) {
+ if (LOG.isDebugEnabled()) {
- log.debug("Pods Count: "+newlyCreatedPods.length+" for cluster: "+clusterId);
+ LOG.debug("Pods Count: "+newlyCreatedPods.length+" for cluster: "+clusterId);
}
if(newlyCreatedPods.length == expectedCount) {
break;
@@ -1437,9 +1423,9 @@ public class CloudControllerServiceImpl implements CloudControllerService {
Thread.sleep(5000);
}
- if (log.isDebugEnabled()) {
+ if (LOG.isDebugEnabled()) {
- log.debug(String.format("Pods created : %s for cluster : %s",newlyCreatedPods.length, clusterId));
+ LOG.debug(String.format("Pods created : %s for cluster : %s",newlyCreatedPods.length, clusterId));
}
List<MemberContext> memberContexts = new ArrayList<MemberContext>();
@@ -1455,15 +1441,12 @@ public class CloudControllerServiceImpl implements CloudControllerService {
.getProperties(), StratosConstants.ALLOCATED_SERVICE_HOST_PORT,
String.valueOf(service.getPort())));
- dataHolder.addMemberContext(context);
-
- // trigger topology
- // update the topology with the newly spawned member
- TopologyBuilder.handleMemberSpawned(cartridgeType, clusterId, null,
- pod.getCurrentState().getPodIP(), pod.getCurrentState().getHostIP(), context);
- // publish data
- // TODO
-// CartridgeInstanceDataPublisher.publish(context.getMemberId(), null, null, context.getClusterId(), cartridgeType, MemberStatus.Created.toString(), node);
+ // wait till Pod status turns to running and send member spawned.
+ ScheduledThreadExecutor exec = ScheduledThreadExecutor.getInstance();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller is starting the instance start up thread.");
+ }
+ dataHolder.addScheduledFutureJob(context.getMemberId(), exec.schedule(new PodActivationWatcher(pod.getId(), context, kubApi), 5000));
memberContexts.add(context);
}
@@ -1471,13 +1454,13 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// persist in registry
persist();
- log.info("Kubernetes entities are successfully starting up. "+memberContexts);
+ LOG.info("Kubernetes entities are successfully starting up. "+memberContexts);
return memberContexts.toArray(new MemberContext[0]);
} catch (Exception e) {
String msg = "Failed to start an instance. " + containerClusterContext.toString()+" Cause: "+e.getMessage();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new IllegalStateException(msg, e);
}
}
@@ -1489,7 +1472,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (propVal == null) {
String msg = "Instance start-up failed. Cannot find '"+
StratosConstants.KUBERNETES_MIN_REPLICAS+"' in " + ctxt;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1503,7 +1486,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (propVal == null) {
String msg = "Instance start-up failed. Cannot find '"+
StratosConstants.KUBERNETES_MIN_REPLICAS+"' in " + ctxt;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1534,14 +1517,14 @@ public class CloudControllerServiceImpl implements CloudControllerService {
}
@Override
- public void terminateAllContainers(String clusterId)
+ public MemberContext[] terminateAllContainers(String clusterId)
throws InvalidClusterException {
ClusterContext ctxt = dataHolder.getClusterContext(clusterId);
if (ctxt == null) {
String msg = "Kubernetes units temrination failed. Invalid cluster id. "+clusterId;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1551,7 +1534,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (kubernetesClusterId == null) {
String msg = "Kubernetes units termination failed. Cannot find '"+
StratosConstants.KUBERNETES_CLUSTER_ID+"'. " + ctxt;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1560,7 +1543,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
if (kubClusterContext == null) {
String msg = "Kubernetes units termination failed. Cannot find a matching Kubernetes Cluster for cluster id: "
+kubernetesClusterId;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1570,7 +1553,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
kubApi.deleteService(CloudControllerUtil.getCompatibleId(clusterId));
} catch (KubernetesClientException e) {
// we're not going to throw this error, but proceed with other deletions
- log.error("Failed to delete Kubernetes service with id: "+clusterId, e);
+ LOG.error("Failed to delete Kubernetes service with id: "+clusterId, e);
}
// set replicas=0 for the replication controller
@@ -1578,7 +1561,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
kubApi.updateReplicationController(clusterId, 0);
} catch (KubernetesClientException e) {
// we're not going to throw this error, but proceed with other deletions
- log.error("Failed to update Kubernetes Controller with id: "+clusterId, e);
+ LOG.error("Failed to update Kubernetes Controller with id: "+clusterId, e);
}
// delete pods forcefully
@@ -1595,12 +1578,12 @@ public class CloudControllerServiceImpl implements CloudControllerService {
kubApi.deletePod(pod.getId());
} catch (KubernetesClientException ignore) {
// we can't do nothing here
- log.warn(String.format("Failed to delete Pod [%s] forcefully!", pod.getId()));
+ LOG.warn(String.format("Failed to delete Pod [%s] forcefully!", pod.getId()));
}
}
} catch (KubernetesClientException e) {
// we're not going to throw this error, but proceed with other deletions
- log.error("Failed to delete pods forcefully for cluster: "+clusterId, e);
+ LOG.error("Failed to delete pods forcefully for cluster: "+clusterId, e);
}
// delete the replication controller.
@@ -1608,7 +1591,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
kubApi.deleteReplicationController(clusterId);
} catch (KubernetesClientException e) {
String msg = "Failed to delete Kubernetes Controller with id: "+clusterId;
- log.error(msg, e);
+ LOG.error(msg, e);
throw new InvalidClusterException(msg, e);
}
@@ -1619,29 +1602,35 @@ public class CloudControllerServiceImpl implements CloudControllerService {
kubClusterContext.deallocateHostPort(Integer
.parseInt(allocatedPort));
} else {
- log.warn("Host port dealloacation failed due to a missing property: "
+ LOG.warn("Host port dealloacation failed due to a missing property: "
+ StratosConstants.ALLOCATED_SERVICE_HOST_PORT);
}
- dataHolder.removeMemberContextsOfCluster(clusterId);
+ List<MemberContext> removedMembers = dataHolder.removeMemberContextsOfCluster(clusterId);
+
+ for (MemberContext memberContext : removedMembers) {
+ logTermination(memberContext);
+ }
// persist
persist();
+
+ return removedMembers.toArray(new MemberContext[0]);
}
@Override
public MemberContext[] updateContainers(String clusterId, int replicas)
throws UnregisteredCartridgeException {
- if(log.isDebugEnabled()) {
- log.debug("CloudControllerServiceImpl:updateContainers for cluster : "+clusterId);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("CloudControllerServiceImpl:updateContainers for cluster : "+clusterId);
}
ClusterContext ctxt = dataHolder.getClusterContext(clusterId);
if (ctxt == null) {
String msg = "Instance start-up failed. Invalid cluster id. " + clusterId;
- log.error(msg);
+ LOG.error(msg);
throw new IllegalArgumentException(msg);
}
@@ -1653,7 +1642,7 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Instance start-up failed. No matching Cartridge found [type] "+cartridgeType
+". [cluster id] "+ clusterId;
- log.error(msg);
+ LOG.error(msg);
throw new UnregisteredCartridgeException(msg);
}
@@ -1666,22 +1655,22 @@ public class CloudControllerServiceImpl implements CloudControllerService {
String msg =
"Instance start-up failed. No matching Kubernetes Context Found for [id] "+kubernetesClusterId
+". [cluster id] "+ clusterId;
- log.error(msg);
+ LOG.error(msg);
throw new UnregisteredCartridgeException(msg);
}
KubernetesApiClient kubApi = kubClusterContext.getKubApi();
// update the replication controller - cluster id = replication controller id
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller is delegating request to update a replication controller "+clusterId+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller is delegating request to update a replication controller "+clusterId+
" to Kubernetes layer.");
}
kubApi.updateReplicationController(clusterId, replicas);
- if (log.isDebugEnabled()) {
- log.debug("Cloud Controller successfully updated the controller "
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Cloud Controller successfully updated the controller "
+ clusterId + " via Kubernetes layer.");
}
@@ -1695,9 +1684,9 @@ public class CloudControllerServiceImpl implements CloudControllerService {
for (int i = 0; i < replicas ; i++) {
allPods = kubApi.getSelectedPods(new Label[]{l});
- if (log.isDebugEnabled()) {
+ if (LOG.isDebugEnabled()) {
- log.debug("Pods Count: "+allPods.length+" for cluster: "+clusterId);
+ LOG.debug("Pods Count: "+allPods.length+" for cluster: "+clusterId);
}
if(allPods.length == replicas) {
break;
@@ -1705,9 +1694,9 @@ public class CloudControllerServiceImpl implements CloudControllerService {
Thread.sleep(5000);
}
- if (log.isDebugEnabled()) {
+ if (LOG.isDebugEnabled()) {
- log.debug(String.format("Pods created : %s for cluster : %s",allPods.length, clusterId));
+ LOG.debug(String.format("Pods created : %s for cluster : %s",allPods.length, clusterId));
}
List<MemberContext> memberContexts = new ArrayList<MemberContext>();
@@ -1744,19 +1733,19 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// persist in registry
persist();
- log.info("Kubernetes entities are successfully starting up. "+memberContexts);
+ LOG.info("Kubernetes entities are successfully starting up. "+memberContexts);
return memberContexts.toArray(new MemberContext[0]);
} catch (Exception e) {
String msg = "Failed to update containers belong to cluster " + clusterId+". Cause: "+e.getMessage();
- log.error(msg, e);
+ LOG.error(msg, e);
throw new IllegalStateException(msg, e);
}
}
@Override
- public void terminateContainer(String memberId) throws MemberTerminationFailedException {
+ public MemberContext terminateContainer(String memberId) throws MemberTerminationFailedException {
handleNullObject(memberId, "Failed to terminate member. Invalid Member id. [Member id] " + memberId);
@@ -1789,19 +1778,22 @@ public class CloudControllerServiceImpl implements CloudControllerService {
// member id = pod id
kubApi.deletePod(memberId);
- dataHolder.removeMemberContext(memberId, clusterId);
- // persist
- persist();
+ MemberContext removedMemberContext = dataHolder.removeMemberContext(memberId, clusterId);
+
+ logTermination(removedMemberContext);
+
+ return removedMemberContext;
+
} catch (KubernetesClientException e) {
String msg = String.format("Failed to terminate member [Member id] %s", memberId);
- log.error(msg, e);
+ LOG.error(msg, e);
throw new MemberTerminationFailedException(msg, e);
}
}
private void handleNullObject(Object obj, String errorMsg) {
if (obj == null) {
- log.error(errorMsg);
+ LOG.error(errorMsg);
throw new IllegalArgumentException(errorMsg);
}
}
http://git-wip-us.apache.org/repos/asf/stratos/blob/f37ece08/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/interfaces/CloudControllerService.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/interfaces/CloudControllerService.java b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/interfaces/CloudControllerService.java
index 398467c..d81eab7 100644
--- a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/interfaces/CloudControllerService.java
+++ b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/interfaces/CloudControllerService.java
@@ -133,22 +133,24 @@ public interface CloudControllerService {
/**
* Terminate all containers of the given cluster.
* @param clusterId id of the subjected cluster.
+ * @return terminated {@link MemberContext}s
* @throws InvalidClusterException
*/
- void terminateAllContainers(String clusterId) throws InvalidClusterException;
+ MemberContext[] terminateAllContainers(String clusterId) throws InvalidClusterException;
/**
* Terminate a given member/Kubernetes Pod.
* @param memberId member/Pod id to be terminated.
+ * @return terminated {@link MemberContext}
* @throws MemberTerminationFailedException
*/
- void terminateContainer(String memberId) throws MemberTerminationFailedException;
+ MemberContext terminateContainer(String memberId) throws MemberTerminationFailedException;
/**
* Update the Kubernetes controller created for the given cluster with the specified number of replicas.
* @param clusterId id of the subjected cluster.
* @param replicas total number of replicas to be set to the controller.
- * @return newly created Members if any.
+ * @return newly created Members if any / terminated {@link MemberContext} in scale down scenario.
* @throws InvalidClusterException
*/
MemberContext[] updateContainers(String clusterId, int replicas) throws UnregisteredCartridgeException;
http://git-wip-us.apache.org/repos/asf/stratos/blob/f37ece08/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/jcloud/ComputeServiceBuilderUtil.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/jcloud/ComputeServiceBuilderUtil.java b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/jcloud/ComputeServiceBuilderUtil.java
index ac30f01..d8bda9e 100644
--- a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/jcloud/ComputeServiceBuilderUtil.java
+++ b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/jcloud/ComputeServiceBuilderUtil.java
@@ -47,7 +47,6 @@ public class ComputeServiceBuilderUtil {
private static final Log log = LogFactory.getLog(ComputeServiceBuilderUtil.class);
public static byte[] getUserData(String payloadFileName) {
- // String userData = null;
byte[] bytes = null;
try {
File file = new File(payloadFileName);
[3/5] git commit: RestClient now using a ConnectionManager pool.
Posted by ni...@apache.org.
RestClient now using a ConnectionManager pool.
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/78ee28b2
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/78ee28b2
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/78ee28b2
Branch: refs/heads/container-autoscaling
Commit: 78ee28b2c2f44df6ab8e9db668f80a8c1861ecbf
Parents: ff37222
Author: Nirmal Fernando <ni...@gmail.com>
Authored: Fri Oct 10 19:42:49 2014 +0530
Committer: Nirmal Fernando <ni...@gmail.com>
Committed: Fri Oct 10 19:50:27 2014 +0530
----------------------------------------------------------------------
.../kubernetes/client/rest/RestClient.java | 47 +++++++++++++-------
1 file changed, 30 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/78ee28b2/components/org.apache.stratos.kubernetes.client/src/main/java/org/apache/stratos/kubernetes/client/rest/RestClient.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.kubernetes.client/src/main/java/org/apache/stratos/kubernetes/client/rest/RestClient.java b/components/org.apache.stratos.kubernetes.client/src/main/java/org/apache/stratos/kubernetes/client/rest/RestClient.java
index 061c6d6..7d55c70 100644
--- a/components/org.apache.stratos.kubernetes.client/src/main/java/org/apache/stratos/kubernetes/client/rest/RestClient.java
+++ b/components/org.apache.stratos.kubernetes.client/src/main/java/org/apache/stratos/kubernetes/client/rest/RestClient.java
@@ -18,23 +18,30 @@
*/
package org.apache.stratos.kubernetes.client.rest;
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.net.SocketException;
import java.net.URI;
-import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpDelete;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpRequestBase;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.impl.conn.PoolingClientConnectionManager;
public class RestClient {
private DefaultHttpClient httpClient;
+
+ public RestClient() {
+ PoolingClientConnectionManager cm = new PoolingClientConnectionManager();
+ // Increase max total connection to 200
+ cm.setMaxTotal(200);
+ // Increase default max connection per route to 50
+ cm.setDefaultMaxPerRoute(50);
+
+ httpClient = new DefaultHttpClient(cm);
+ }
/**
* Handle http post request. Return String
@@ -54,9 +61,9 @@ public class RestClient {
* if any errors occur when executing the request
*/
public KubernetesResponse doPost(URI resourcePath, String jsonParamString) throws Exception{
+ HttpPost postRequest = null;
try {
- httpClient = new DefaultHttpClient();
- HttpPost postRequest = new HttpPost(resourcePath);
+ postRequest = new HttpPost(resourcePath);
StringEntity input = new StringEntity(jsonParamString);
input.setContentType("application/json");
@@ -66,7 +73,7 @@ public class RestClient {
return response;
} finally {
- httpClient.getConnectionManager().shutdown();
+ releaseConnection(postRequest);
}
}
@@ -86,22 +93,22 @@ public class RestClient {
* if any errors occur when executing the request
*/
public KubernetesResponse doGet(URI resourcePath) throws Exception{
+ HttpGet getRequest = null;
try {
- httpClient = new DefaultHttpClient();
- HttpGet getRequest = new HttpGet(resourcePath);
+ getRequest = new HttpGet(resourcePath);
getRequest.addHeader("Content-Type", "application/json");
KubernetesResponse response = httpClient.execute(getRequest, new KubernetesResponseHandler());
return response;
} finally {
- httpClient.getConnectionManager().shutdown();
+ releaseConnection(getRequest);
}
}
public KubernetesResponse doDelete(URI resourcePath) throws Exception {
+ HttpDelete httpDelete = null;
try {
- httpClient = new DefaultHttpClient();
- HttpDelete httpDelete = new HttpDelete(resourcePath);
+ httpDelete = new HttpDelete(resourcePath);
httpDelete.addHeader("Content-Type", "application/json");
KubernetesResponse response = httpClient.execute(httpDelete, new KubernetesResponseHandler());
@@ -109,15 +116,15 @@ public class RestClient {
return response;
} finally {
- httpClient.getConnectionManager().shutdown();
+ releaseConnection(httpDelete);
}
}
public KubernetesResponse doPut(URI resourcePath, String jsonParamString) throws Exception {
+ HttpPut putRequest = null;
try {
- httpClient = new DefaultHttpClient();
- HttpPut putRequest = new HttpPut(resourcePath);
+ putRequest = new HttpPut(resourcePath);
StringEntity input = new StringEntity(jsonParamString);
input.setContentType("application/json");
@@ -128,8 +135,14 @@ public class RestClient {
return response;
} finally {
- httpClient.getConnectionManager().shutdown();
+ releaseConnection(putRequest);
}
}
+
+ private void releaseConnection(HttpRequestBase request) {
+ if (request != null) {
+ request.releaseConnection();
+ }
+ }
}
\ No newline at end of file
[2/5] git commit: Scheduled executor service pool to schedule tasks
in CC.
Posted by ni...@apache.org.
Scheduled executor service pool to schedule tasks in CC.
Project: http://git-wip-us.apache.org/repos/asf/stratos/repo
Commit: http://git-wip-us.apache.org/repos/asf/stratos/commit/0fe96488
Tree: http://git-wip-us.apache.org/repos/asf/stratos/tree/0fe96488
Diff: http://git-wip-us.apache.org/repos/asf/stratos/diff/0fe96488
Branch: refs/heads/container-autoscaling
Commit: 0fe96488a6ebf1e5ba3e0304811ac07341513680
Parents: ff2a02f
Author: Nirmal Fernando <ni...@gmail.com>
Authored: Fri Oct 10 19:47:37 2014 +0530
Committer: Nirmal Fernando <ni...@gmail.com>
Committed: Fri Oct 10 19:50:27 2014 +0530
----------------------------------------------------------------------
.../concurrent/ScheduledThreadExecutor.java | 78 ++++++++++++++++++++
1 file changed, 78 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/stratos/blob/0fe96488/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/concurrent/ScheduledThreadExecutor.java
----------------------------------------------------------------------
diff --git a/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/concurrent/ScheduledThreadExecutor.java b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/concurrent/ScheduledThreadExecutor.java
new file mode 100644
index 0000000..56f2a2c
--- /dev/null
+++ b/components/org.apache.stratos.cloud.controller/src/main/java/org/apache/stratos/cloud/controller/concurrent/ScheduledThreadExecutor.java
@@ -0,0 +1,78 @@
+/*
+ * 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.stratos.cloud.controller.concurrent;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This class can be used to create a scheduled executor pool, and hand over new jobs to the pool.
+ */
+public class ScheduledThreadExecutor {
+ private ScheduledExecutorService executor;
+
+ private static class Holder {
+ private static final ScheduledThreadExecutor INSTANCE = new ScheduledThreadExecutor();
+ }
+
+ public static ScheduledThreadExecutor getInstance() {
+ return Holder.INSTANCE;
+ }
+
+ private ScheduledThreadExecutor() {
+ executor = Executors.newScheduledThreadPool(50);
+ }
+
+ public ScheduledFuture<?> schedule(Runnable job, int interval){
+ return executor.scheduleAtFixedRate(job, 0, interval, TimeUnit.MILLISECONDS);
+ }
+
+ public List<ScheduledFuture<?>> scheduleAll(Runnable[] jobs, int interval){
+ List<ScheduledFuture<?>> list = new ArrayList<ScheduledFuture<?>>();
+ for (Runnable job : jobs) {
+
+ list.add(this.schedule(job, interval));
+ }
+ return list;
+ }
+
+ public void shutdown() {
+ executor.shutdown(); // Disable new tasks from being submitted
+ try {
+ // Wait a while for existing tasks to terminate
+ if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
+ executor.shutdownNow(); // Cancel currently executing tasks
+ // Wait a while for tasks to respond to being cancelled
+ if (!executor.awaitTermination(60, TimeUnit.SECONDS))
+ System.err.println("Pool did not terminate");
+ }
+ } catch (InterruptedException ie) {
+ // (Re-)Cancel if current thread also interrupted
+ executor.shutdownNow();
+ // Preserve interrupt status
+ Thread.currentThread().interrupt();
+ }
+ }
+
+
+}