| // 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.cluster; |
| |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileNotFoundException; |
| import java.io.IOException; |
| import java.net.ConnectException; |
| import java.net.InetSocketAddress; |
| import java.nio.channels.SocketChannel; |
| import java.rmi.RemoteException; |
| import java.sql.Connection; |
| import java.sql.SQLException; |
| import java.sql.SQLRecoverableException; |
| import java.util.ArrayList; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.concurrent.ExecutorService; |
| 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 com.cloud.agent.AgentManager; |
| import com.cloud.agent.AgentManager.OnError; |
| import com.cloud.agent.api.Answer; |
| import com.cloud.agent.api.ChangeAgentAnswer; |
| import com.cloud.agent.api.ChangeAgentCommand; |
| import com.cloud.agent.api.Command; |
| import com.cloud.agent.api.PropagateResourceEventCommand; |
| import com.cloud.agent.api.TransferAgentCommand; |
| import com.cloud.agent.api.ScheduleHostScanTaskCommand; |
| import com.cloud.agent.manager.ClusteredAgentManagerImpl; |
| import com.cloud.agent.manager.Commands; |
| import com.cloud.cluster.agentlb.dao.HostTransferMapDao; |
| import com.cloud.cluster.dao.ManagementServerHostDao; |
| import com.cloud.cluster.dao.ManagementServerHostPeerDao; |
| import com.cloud.configuration.Config; |
| import com.cloud.configuration.dao.ConfigurationDao; |
| import com.cloud.exception.AgentUnavailableException; |
| import com.cloud.exception.OperationTimedoutException; |
| import com.cloud.host.Host; |
| import com.cloud.host.HostVO; |
| import com.cloud.host.Status.Event; |
| import com.cloud.host.dao.HostDao; |
| import com.cloud.resource.ResourceManager; |
| import com.cloud.resource.ResourceState; |
| import com.cloud.serializer.GsonHelper; |
| import com.cloud.utils.DateUtil; |
| import com.cloud.utils.NumbersUtil; |
| import com.cloud.utils.Profiler; |
| import com.cloud.utils.PropertiesUtil; |
| import com.cloud.utils.component.ComponentContext; |
| import com.cloud.utils.component.ComponentLifecycle; |
| import com.cloud.utils.component.ManagerBase; |
| import com.cloud.utils.concurrency.NamedThreadFactory; |
| import com.cloud.utils.db.ConnectionConcierge; |
| import com.cloud.utils.db.DB; |
| import com.cloud.utils.db.SearchCriteria.Op; |
| import com.cloud.utils.db.SearchCriteria2; |
| import com.cloud.utils.db.SearchCriteriaService; |
| import com.cloud.utils.db.Transaction; |
| import com.cloud.utils.events.SubscriptionMgr; |
| import com.cloud.utils.exception.CloudRuntimeException; |
| import com.cloud.utils.exception.ExceptionUtil; |
| import com.cloud.utils.mgmt.JmxUtil; |
| import com.cloud.utils.net.NetUtils; |
| import com.google.gson.Gson; |
| |
| @Local(value = { ClusterManager.class }) |
| public class ClusterManagerImpl extends ManagerBase implements ClusterManager { |
| private static final Logger s_logger = Logger.getLogger(ClusterManagerImpl.class); |
| |
| private static final int EXECUTOR_SHUTDOWN_TIMEOUT = 1000; // 1 second |
| private static final int DEFAULT_OUTGOING_WORKERS = 5; |
| |
| private final List<ClusterManagerListener> _listeners = new ArrayList<ClusterManagerListener>(); |
| private final Map<Long, ManagementServerHostVO> _activePeers = new HashMap<Long, ManagementServerHostVO>(); |
| private int _heartbeatInterval = ClusterManager.DEFAULT_HEARTBEAT_INTERVAL; |
| private int _heartbeatThreshold = ClusterManager.DEFAULT_HEARTBEAT_THRESHOLD; |
| |
| private final Map<String, ClusterService> _clusterPeers; |
| private final Gson _gson; |
| |
| @Inject |
| private AgentManager _agentMgr; |
| @Inject |
| private ClusteredAgentRebalanceService _rebalanceService; |
| @Inject |
| private ResourceManager _resourceMgr; |
| |
| private final ScheduledExecutorService _heartbeatScheduler = Executors.newScheduledThreadPool(1, new NamedThreadFactory("Cluster-Heartbeat")); |
| private final ExecutorService _notificationExecutor = Executors.newFixedThreadPool(1, new NamedThreadFactory("Cluster-Notification")); |
| private final List<ClusterManagerMessage> _notificationMsgs = new ArrayList<ClusterManagerMessage>(); |
| private ConnectionConcierge _heartbeatConnection = null; |
| |
| private final ExecutorService _executor; |
| |
| private ClusterServiceAdapter _currentServiceAdapter; |
| |
| @Inject |
| private List<ClusterServiceAdapter> _serviceAdapters; |
| |
| @Inject private ManagementServerHostDao _mshostDao; |
| @Inject private ManagementServerHostPeerDao _mshostPeerDao; |
| @Inject private HostDao _hostDao; |
| @Inject private HostTransferMapDao _hostTransferDao; |
| @Inject private ConfigurationDao _configDao; |
| |
| // |
| // pay attention to _mshostId and _msid |
| // _mshostId is the primary key of management host table |
| // _msid is the unique persistent identifier that peer name is based upon |
| // |
| private Long _mshostId = null; |
| protected long _msId = ManagementServerNode.getManagementServerId(); |
| protected long _runId = System.currentTimeMillis(); |
| |
| private boolean _peerScanInited = false; |
| |
| private String _clusterNodeIP = "127.0.0.1"; |
| private boolean _agentLBEnabled = false; |
| private double _connectedAgentsThreshold = 0.7; |
| private static boolean _agentLbHappened = false; |
| |
| private final List<ClusterServicePdu> _clusterPduOutgoingQueue = new ArrayList<ClusterServicePdu>(); |
| private final List<ClusterServicePdu> _clusterPduIncomingQueue = new ArrayList<ClusterServicePdu>(); |
| private final Map<Long, ClusterServiceRequestPdu> _outgoingPdusWaitingForAck = new HashMap<Long, ClusterServiceRequestPdu>(); |
| |
| public ClusterManagerImpl() { |
| _clusterPeers = new HashMap<String, ClusterService>(); |
| |
| _gson = GsonHelper.getGson(); |
| |
| // executor to perform remote-calls in another thread context, to avoid potential |
| // recursive remote calls between nodes |
| // |
| _executor = Executors.newCachedThreadPool(new NamedThreadFactory("Cluster-Worker")); |
| setRunLevel(ComponentLifecycle.RUN_LEVEL_FRAMEWORK); |
| } |
| |
| private void registerRequestPdu(ClusterServiceRequestPdu pdu) { |
| synchronized(_outgoingPdusWaitingForAck) { |
| _outgoingPdusWaitingForAck.put(pdu.getSequenceId(), pdu); |
| } |
| } |
| |
| private ClusterServiceRequestPdu popRequestPdu(long ackSequenceId) { |
| synchronized(_outgoingPdusWaitingForAck) { |
| if(_outgoingPdusWaitingForAck.get(ackSequenceId) != null) { |
| ClusterServiceRequestPdu pdu = _outgoingPdusWaitingForAck.get(ackSequenceId); |
| _outgoingPdusWaitingForAck.remove(ackSequenceId); |
| return pdu; |
| } |
| } |
| |
| return null; |
| } |
| |
| private void cancelClusterRequestToPeer(String strPeer) { |
| List<ClusterServiceRequestPdu> candidates = new ArrayList<ClusterServiceRequestPdu>(); |
| synchronized(_outgoingPdusWaitingForAck) { |
| for(Map.Entry<Long, ClusterServiceRequestPdu> entry : _outgoingPdusWaitingForAck.entrySet()) { |
| if(entry.getValue().getDestPeer().equalsIgnoreCase(strPeer)) |
| candidates.add(entry.getValue()); |
| } |
| |
| for(ClusterServiceRequestPdu pdu : candidates) { |
| _outgoingPdusWaitingForAck.remove(pdu.getSequenceId()); |
| } |
| } |
| |
| for(ClusterServiceRequestPdu pdu : candidates) { |
| s_logger.warn("Cancel cluster request PDU to peer: " + strPeer + ", pdu: " + _gson.toJson(pdu)); |
| synchronized(pdu) { |
| pdu.notifyAll(); |
| } |
| } |
| } |
| |
| private void addOutgoingClusterPdu(ClusterServicePdu pdu) { |
| synchronized(_clusterPduOutgoingQueue) { |
| _clusterPduOutgoingQueue.add(pdu); |
| _clusterPduOutgoingQueue.notifyAll(); |
| } |
| } |
| |
| private ClusterServicePdu popOutgoingClusterPdu(long timeoutMs) { |
| synchronized(_clusterPduOutgoingQueue) { |
| try { |
| _clusterPduOutgoingQueue.wait(timeoutMs); |
| } catch (InterruptedException e) { |
| } |
| |
| if(_clusterPduOutgoingQueue.size() > 0) { |
| ClusterServicePdu pdu = _clusterPduOutgoingQueue.get(0); |
| _clusterPduOutgoingQueue.remove(0); |
| return pdu; |
| } |
| } |
| return null; |
| } |
| |
| private void addIncomingClusterPdu(ClusterServicePdu pdu) { |
| synchronized(_clusterPduIncomingQueue) { |
| _clusterPduIncomingQueue.add(pdu); |
| _clusterPduIncomingQueue.notifyAll(); |
| } |
| } |
| |
| private ClusterServicePdu popIncomingClusterPdu(long timeoutMs) { |
| synchronized(_clusterPduIncomingQueue) { |
| try { |
| _clusterPduIncomingQueue.wait(timeoutMs); |
| } catch (InterruptedException e) { |
| } |
| |
| if(_clusterPduIncomingQueue.size() > 0) { |
| ClusterServicePdu pdu = _clusterPduIncomingQueue.get(0); |
| _clusterPduIncomingQueue.remove(0); |
| return pdu; |
| } |
| } |
| return null; |
| } |
| |
| private Runnable getClusterPduSendingTask() { |
| return new Runnable() { |
| @Override |
| public void run() { |
| onSendingClusterPdu(); |
| } |
| }; |
| } |
| |
| private Runnable getClusterPduNotificationTask() { |
| return new Runnable() { |
| @Override |
| public void run() { |
| onNotifyingClusterPdu(); |
| } |
| }; |
| } |
| |
| private void onSendingClusterPdu() { |
| while(true) { |
| try { |
| ClusterServicePdu pdu = popOutgoingClusterPdu(1000); |
| if(pdu == null) |
| continue; |
| |
| ClusterService peerService = null; |
| for(int i = 0; i < 2; i++) { |
| try { |
| peerService = getPeerService(pdu.getDestPeer()); |
| } catch (RemoteException e) { |
| s_logger.error("Unable to get cluster service on peer : " + pdu.getDestPeer()); |
| } |
| |
| if(peerService != null) { |
| try { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Cluster PDU " + getSelfPeerName() + " -> " + pdu.getDestPeer() + ". agent: " + pdu.getAgentId() |
| + ", pdu seq: " + pdu.getSequenceId() + ", pdu ack seq: " + pdu.getAckSequenceId() + ", json: " + pdu.getJsonPackage()); |
| } |
| |
| long startTick = System.currentTimeMillis(); |
| String strResult = peerService.execute(pdu); |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Cluster PDU " + getSelfPeerName() + " -> " + pdu.getDestPeer() + " completed. time: " + |
| (System.currentTimeMillis() - startTick) + "ms. agent: " + pdu.getAgentId() |
| + ", pdu seq: " + pdu.getSequenceId() + ", pdu ack seq: " + pdu.getAckSequenceId() + ", json: " + pdu.getJsonPackage()); |
| } |
| |
| if("true".equals(strResult)) |
| break; |
| |
| } catch (RemoteException e) { |
| invalidatePeerService(pdu.getDestPeer()); |
| if(s_logger.isInfoEnabled()) { |
| s_logger.info("Exception on remote execution, peer: " + pdu.getDestPeer() + ", iteration: " |
| + i + ", exception message :" + e.getMessage()); |
| } |
| } |
| } |
| } |
| } catch(Throwable e) { |
| s_logger.error("Unexcpeted exception: ", e); |
| } |
| } |
| } |
| |
| private void onNotifyingClusterPdu() { |
| while(true) { |
| try { |
| final ClusterServicePdu pdu = popIncomingClusterPdu(1000); |
| if(pdu == null) |
| continue; |
| |
| _executor.execute(new Runnable() { |
| @Override |
| public void run() { |
| if(pdu.getPduType() == ClusterServicePdu.PDU_TYPE_RESPONSE) { |
| ClusterServiceRequestPdu requestPdu = popRequestPdu(pdu.getAckSequenceId()); |
| if(requestPdu != null) { |
| requestPdu.setResponseResult(pdu.getJsonPackage()); |
| synchronized(requestPdu) { |
| requestPdu.notifyAll(); |
| } |
| } else { |
| s_logger.warn("Original request has already been cancelled. pdu: " + _gson.toJson(pdu)); |
| } |
| } else { |
| String result = dispatchClusterServicePdu(pdu); |
| if(result == null) |
| result = ""; |
| |
| if(pdu.getPduType() == ClusterServicePdu.PDU_TYPE_REQUEST) { |
| ClusterServicePdu responsePdu = new ClusterServicePdu(); |
| responsePdu.setPduType(ClusterServicePdu.PDU_TYPE_RESPONSE); |
| responsePdu.setSourcePeer(pdu.getDestPeer()); |
| responsePdu.setDestPeer(pdu.getSourcePeer()); |
| responsePdu.setAckSequenceId(pdu.getSequenceId()); |
| responsePdu.setJsonPackage(result); |
| |
| addOutgoingClusterPdu(responsePdu); |
| } |
| } |
| } |
| }); |
| } catch(Throwable e) { |
| s_logger.error("Unexcpeted exception: ", e); |
| } |
| } |
| } |
| |
| private String handleScheduleHostScanTaskCommand(ScheduleHostScanTaskCommand cmd) { |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Intercepting resource manager command: " + _gson.toJson(cmd)); |
| } |
| |
| try { |
| // schedule a scan task immediately |
| if (_agentMgr instanceof ClusteredAgentManagerImpl) { |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Received notification as part of addHost command to start a host scan task"); |
| } |
| ClusteredAgentManagerImpl clusteredAgentMgr = (ClusteredAgentManagerImpl)_agentMgr; |
| clusteredAgentMgr.scheduleHostScanTask(); |
| } |
| } catch (Exception e) { |
| // Scheduling host scan task in peer MS is a best effort operation during host add, regular host scan |
| // happens at fixed intervals anyways. So handling any exceptions that may be thrown |
| s_logger.warn("Exception happened while trying to schedule host scan task on mgmt server " + getSelfPeerName() + ", ignoring as regular host scan happens at fixed interval anyways", e); |
| return null; |
| } |
| |
| Answer[] answers = new Answer[1]; |
| answers[0] = new Answer(cmd, true, null); |
| return _gson.toJson(answers); |
| } |
| |
| private String dispatchClusterServicePdu(ClusterServicePdu pdu) { |
| |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Dispatch ->" + pdu.getAgentId() + ", json: " + pdu.getJsonPackage()); |
| } |
| |
| Command [] cmds = null; |
| try { |
| cmds = _gson.fromJson(pdu.getJsonPackage(), Command[].class); |
| } catch(Throwable e) { |
| assert(false); |
| s_logger.error("Excection in gson decoding : ", e); |
| } |
| |
| if (cmds.length == 1 && cmds[0] instanceof ChangeAgentCommand) { //intercepted |
| ChangeAgentCommand cmd = (ChangeAgentCommand)cmds[0]; |
| |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Intercepting command for agent change: agent " + cmd.getAgentId() + " event: " + cmd.getEvent()); |
| } |
| boolean result = false; |
| try { |
| result = executeAgentUserRequest(cmd.getAgentId(), cmd.getEvent()); |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Result is " + result); |
| } |
| |
| } catch (AgentUnavailableException e) { |
| s_logger.warn("Agent is unavailable", e); |
| return null; |
| } |
| |
| Answer[] answers = new Answer[1]; |
| answers[0] = new ChangeAgentAnswer(cmd, result); |
| return _gson.toJson(answers); |
| } else if (cmds.length == 1 && cmds[0] instanceof TransferAgentCommand) { |
| TransferAgentCommand cmd = (TransferAgentCommand) cmds[0]; |
| |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Intercepting command for agent rebalancing: agent " + cmd.getAgentId() + " event: " + cmd.getEvent()); |
| } |
| boolean result = false; |
| try { |
| result = rebalanceAgent(cmd.getAgentId(), cmd.getEvent(), cmd.getCurrentOwner(), cmd.getFutureOwner()); |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Result is " + result); |
| } |
| |
| } catch (AgentUnavailableException e) { |
| s_logger.warn("Agent is unavailable", e); |
| return null; |
| } catch (OperationTimedoutException e) { |
| s_logger.warn("Operation timed out", e); |
| return null; |
| } |
| Answer[] answers = new Answer[1]; |
| answers[0] = new Answer(cmd, result, null); |
| return _gson.toJson(answers); |
| } else if (cmds.length == 1 && cmds[0] instanceof PropagateResourceEventCommand ) { |
| PropagateResourceEventCommand cmd = (PropagateResourceEventCommand) cmds[0]; |
| |
| s_logger.debug("Intercepting command to propagate event " + cmd.getEvent().name() + " for host " + cmd.getHostId()); |
| |
| boolean result = false; |
| try { |
| result = executeResourceUserRequest(cmd.getHostId(), cmd.getEvent()); |
| s_logger.debug("Result is " + result); |
| } catch (AgentUnavailableException ex) { |
| s_logger.warn("Agent is unavailable", ex); |
| return null; |
| } |
| |
| Answer[] answers = new Answer[1]; |
| answers[0] = new Answer(cmd, result, null); |
| return _gson.toJson(answers); |
| } else if (cmds.length == 1 && cmds[0] instanceof ScheduleHostScanTaskCommand) { |
| ScheduleHostScanTaskCommand cmd = (ScheduleHostScanTaskCommand) cmds[0]; |
| String response = handleScheduleHostScanTaskCommand(cmd); |
| return response; |
| } |
| |
| try { |
| long startTick = System.currentTimeMillis(); |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Dispatch -> " + pdu.getAgentId() + ", json: " + pdu.getJsonPackage()); |
| } |
| |
| Answer[] answers = sendToAgent(pdu.getAgentId(), cmds, pdu.isStopOnError()); |
| if(answers != null) { |
| String jsonReturn = _gson.toJson(answers); |
| |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Completed dispatching -> " + pdu.getAgentId() + ", json: " + pdu.getJsonPackage() + |
| " in " + (System.currentTimeMillis() - startTick) + " ms, return result: " + jsonReturn); |
| } |
| |
| return jsonReturn; |
| } else { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Completed dispatching -> " + pdu.getAgentId() + ", json: " + pdu.getJsonPackage() + |
| " in " + (System.currentTimeMillis() - startTick) + " ms, return null result"); |
| } |
| } |
| } catch(AgentUnavailableException e) { |
| s_logger.warn("Agent is unavailable", e); |
| } catch (OperationTimedoutException e) { |
| s_logger.warn("Timed Out", e); |
| } |
| |
| return null; |
| } |
| |
| @Override |
| public void OnReceiveClusterServicePdu(ClusterServicePdu pdu) { |
| addIncomingClusterPdu(pdu); |
| } |
| |
| @Override |
| public Answer[] sendToAgent(Long hostId, Command[] cmds, boolean stopOnError) throws AgentUnavailableException, OperationTimedoutException { |
| Commands commands = new Commands(stopOnError ? OnError.Stop : OnError.Continue); |
| for (Command cmd : cmds) { |
| commands.addCommand(cmd); |
| } |
| return _agentMgr.send(hostId, commands); |
| } |
| |
| @Override |
| public boolean executeAgentUserRequest(long agentId, Event event) throws AgentUnavailableException { |
| return _agentMgr.executeUserRequest(agentId, event); |
| } |
| |
| @Override |
| public Boolean propagateAgentEvent(long agentId, Event event) throws AgentUnavailableException { |
| final String msPeer = getPeerName(agentId); |
| if (msPeer == null) { |
| return null; |
| } |
| |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Propagating agent change request event:" + event.toString() + " to agent:" + agentId); |
| } |
| Command[] cmds = new Command[1]; |
| cmds[0] = new ChangeAgentCommand(agentId, event); |
| |
| Answer[] answers = execute(msPeer, agentId, cmds, true); |
| if (answers == null) { |
| throw new AgentUnavailableException(agentId); |
| } |
| |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Result for agent change is " + answers[0].getResult()); |
| } |
| |
| return answers[0].getResult(); |
| } |
| |
| /** |
| * called by DatabaseUpgradeChecker to see if there are other peers running. |
| * |
| * @param notVersion |
| * If version is passed in, the peers CANNOT be running at this version. If version is null, return true if any |
| * peer is running regardless of version. |
| * @return true if there are peers running and false if not. |
| */ |
| public static final boolean arePeersRunning(String notVersion) { |
| return false; // TODO: Leaving this for Kelven to take care of. |
| } |
| |
| @Override |
| public void broadcast(long agentId, Command[] cmds) { |
| Date cutTime = DateUtil.currentGMTTime(); |
| |
| List<ManagementServerHostVO> peers = _mshostDao.getActiveList(new Date(cutTime.getTime() - _heartbeatThreshold)); |
| for (ManagementServerHostVO peer : peers) { |
| String peerName = Long.toString(peer.getMsid()); |
| if (getSelfPeerName().equals(peerName)) { |
| continue; // Skip myself. |
| } |
| try { |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Forwarding " + cmds[0].toString() + " to " + peer.getMsid()); |
| } |
| executeAsync(peerName, agentId, cmds, true); |
| } catch (Exception e) { |
| s_logger.warn("Caught exception while talkign to " + peer.getMsid()); |
| } |
| } |
| } |
| |
| @Override |
| public void executeAsync(String strPeer, long agentId, Command [] cmds, boolean stopOnError) { |
| ClusterServicePdu pdu = new ClusterServicePdu(); |
| pdu.setSourcePeer(getSelfPeerName()); |
| pdu.setDestPeer(strPeer); |
| pdu.setAgentId(agentId); |
| pdu.setJsonPackage(_gson.toJson(cmds, Command[].class)); |
| pdu.setStopOnError(true); |
| addOutgoingClusterPdu(pdu); |
| } |
| |
| @Override |
| public Answer[] execute(String strPeer, long agentId, Command [] cmds, boolean stopOnError) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug(getSelfPeerName() + " -> " + strPeer + "." + agentId + " " + |
| _gson.toJson(cmds, Command[].class)); |
| } |
| |
| ClusterServiceRequestPdu pdu = new ClusterServiceRequestPdu(); |
| pdu.setSourcePeer(getSelfPeerName()); |
| pdu.setDestPeer(strPeer); |
| pdu.setAgentId(agentId); |
| pdu.setJsonPackage(_gson.toJson(cmds, Command[].class)); |
| pdu.setStopOnError(stopOnError); |
| registerRequestPdu(pdu); |
| addOutgoingClusterPdu(pdu); |
| |
| synchronized(pdu) { |
| try { |
| pdu.wait(); |
| } catch (InterruptedException e) { |
| } |
| } |
| |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug(getSelfPeerName() + " -> " + strPeer + "." + agentId + " completed. result: " + |
| pdu.getResponseResult()); |
| } |
| |
| if(pdu.getResponseResult() != null && pdu.getResponseResult().length() > 0) { |
| try { |
| return _gson.fromJson(pdu.getResponseResult(), Answer[].class); |
| } catch(Throwable e) { |
| s_logger.error("Exception on parsing gson package from remote call to " + strPeer); |
| } |
| } |
| |
| return null; |
| } |
| |
| @Override |
| public String getPeerName(long agentHostId) { |
| |
| HostVO host = _hostDao.findById(agentHostId); |
| if(host != null && host.getManagementServerId() != null) { |
| if(getSelfPeerName().equals(Long.toString(host.getManagementServerId()))) { |
| return null; |
| } |
| |
| return Long.toString(host.getManagementServerId()); |
| } |
| return null; |
| } |
| |
| @Override |
| public ManagementServerHostVO getPeer(String mgmtServerId) { |
| return _mshostDao.findByMsid(Long.valueOf(mgmtServerId)); |
| } |
| |
| @Override |
| public String getSelfPeerName() { |
| return Long.toString(_msId); |
| } |
| |
| @Override |
| public String getSelfNodeIP() { |
| return _clusterNodeIP; |
| } |
| |
| @Override |
| public void registerListener(ClusterManagerListener listener) { |
| // Note : we don't check duplicates |
| synchronized (_listeners) { |
| |
| s_logger.info("register cluster listener " + listener.getClass()); |
| |
| _listeners.add(listener); |
| } |
| } |
| |
| @Override |
| public void unregisterListener(ClusterManagerListener listener) { |
| synchronized(_listeners) { |
| s_logger.info("unregister cluster listener " + listener.getClass()); |
| |
| _listeners.remove(listener); |
| } |
| } |
| |
| public void notifyNodeJoined(List<ManagementServerHostVO> nodeList) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Notify management server node join to listeners."); |
| |
| for(ManagementServerHostVO mshost : nodeList) { |
| s_logger.debug("Joining node, IP: " + mshost.getServiceIP() + ", msid: " + mshost.getMsid()); |
| } |
| } |
| |
| synchronized(_listeners) { |
| for(ClusterManagerListener listener : _listeners) { |
| listener.onManagementNodeJoined(nodeList, _mshostId); |
| } |
| } |
| |
| SubscriptionMgr.getInstance().notifySubscribers(ClusterManager.ALERT_SUBJECT, this, |
| new ClusterNodeJoinEventArgs(_mshostId, nodeList)); |
| } |
| |
| public void notifyNodeLeft(List<ManagementServerHostVO> nodeList) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Notify management server node left to listeners."); |
| } |
| |
| for(ManagementServerHostVO mshost : nodeList) { |
| if(s_logger.isDebugEnabled()) |
| s_logger.debug("Leaving node, IP: " + mshost.getServiceIP() + ", msid: " + mshost.getMsid()); |
| cancelClusterRequestToPeer(String.valueOf(mshost.getMsid())); |
| } |
| |
| synchronized(_listeners) { |
| for(ClusterManagerListener listener : _listeners) { |
| listener.onManagementNodeLeft(nodeList, _mshostId); |
| } |
| } |
| |
| SubscriptionMgr.getInstance().notifySubscribers(ClusterManager.ALERT_SUBJECT, this, |
| new ClusterNodeLeftEventArgs(_mshostId, nodeList)); |
| } |
| |
| public void notifyNodeIsolated() { |
| if(s_logger.isDebugEnabled()) |
| s_logger.debug("Notify management server node isolation to listeners"); |
| |
| synchronized(_listeners) { |
| for(ClusterManagerListener listener : _listeners) { |
| listener.onManagementNodeIsolated(); |
| } |
| } |
| } |
| |
| public ClusterService getPeerService(String strPeer) throws RemoteException { |
| synchronized(_clusterPeers) { |
| if(_clusterPeers.containsKey(strPeer)) { |
| return _clusterPeers.get(strPeer); |
| } |
| } |
| |
| ClusterService service = _currentServiceAdapter.getPeerService(strPeer); |
| |
| if(service != null) { |
| synchronized(_clusterPeers) { |
| // re-check the peer map again to deal with the |
| // race conditions |
| if(!_clusterPeers.containsKey(strPeer)) { |
| _clusterPeers.put(strPeer, service); |
| } |
| } |
| } |
| |
| return service; |
| } |
| |
| public void invalidatePeerService(String strPeer) { |
| synchronized(_clusterPeers) { |
| if(_clusterPeers.containsKey(strPeer)) { |
| _clusterPeers.remove(strPeer); |
| } |
| } |
| } |
| |
| private Runnable getHeartbeatTask() { |
| return new Runnable() { |
| @Override |
| public void run() { |
| Transaction txn = Transaction.open("ClusterHeartBeat"); |
| try { |
| Profiler profiler = new Profiler(); |
| Profiler profilerHeartbeatUpdate = new Profiler(); |
| Profiler profilerPeerScan = new Profiler(); |
| Profiler profilerAgentLB = new Profiler(); |
| |
| try { |
| profiler.start(); |
| |
| profilerHeartbeatUpdate.start(); |
| txn.transitToUserManagedConnection(getHeartbeatConnection()); |
| if(s_logger.isTraceEnabled()) { |
| s_logger.trace("Cluster manager heartbeat update, id:" + _mshostId); |
| } |
| |
| _mshostDao.update(_mshostId, getCurrentRunId(), DateUtil.currentGMTTime()); |
| profilerHeartbeatUpdate.stop(); |
| |
| profilerPeerScan.start(); |
| if (s_logger.isTraceEnabled()) { |
| s_logger.trace("Cluster manager peer-scan, id:" + _mshostId); |
| } |
| |
| if (!_peerScanInited) { |
| _peerScanInited = true; |
| initPeerScan(); |
| } |
| |
| peerScan(); |
| profilerPeerScan.stop(); |
| |
| profilerAgentLB.start(); |
| //initiate agent lb task will be scheduled and executed only once, and only when number of agents loaded exceeds _connectedAgentsThreshold |
| if (_agentLBEnabled && !_agentLbHappened) { |
| SearchCriteriaService<HostVO, HostVO> sc = SearchCriteria2.create(HostVO.class); |
| sc.addAnd(sc.getEntity().getManagementServerId(), Op.NNULL); |
| sc.addAnd(sc.getEntity().getType(), Op.EQ, Host.Type.Routing); |
| List<HostVO> allManagedRoutingAgents = sc.list(); |
| |
| sc = SearchCriteria2.create(HostVO.class); |
| sc.addAnd(sc.getEntity().getType(), Op.EQ, Host.Type.Routing); |
| List<HostVO> allAgents = sc.list(); |
| double allHostsCount = allAgents.size(); |
| double managedHostsCount = allManagedRoutingAgents.size(); |
| if (allHostsCount > 0.0) { |
| double load = managedHostsCount/allHostsCount; |
| if (load >= _connectedAgentsThreshold) { |
| s_logger.debug("Scheduling agent rebalancing task as the average agent load " + load + " is more than the threshold " + _connectedAgentsThreshold); |
| _rebalanceService.scheduleRebalanceAgents(); |
| _agentLbHappened = true; |
| } else { |
| s_logger.trace("Not scheduling agent rebalancing task as the averages load " + load + " is less than the threshold " + _connectedAgentsThreshold); |
| } |
| } |
| } |
| profilerAgentLB.stop(); |
| } finally { |
| profiler.stop(); |
| |
| if(profiler.getDuration() >= _heartbeatInterval) { |
| if(s_logger.isDebugEnabled()) |
| s_logger.debug("Management server heartbeat takes too long to finish. profiler: " + profiler.toString() + |
| ", profilerHeartbeatUpdate: " + profilerHeartbeatUpdate.toString() + |
| ", profilerPeerScan: " + profilerPeerScan.toString() + |
| ", profilerAgentLB: " + profilerAgentLB.toString()); |
| } |
| } |
| |
| } catch(CloudRuntimeException e) { |
| s_logger.error("Runtime DB exception ", e.getCause()); |
| |
| if(e.getCause() instanceof ClusterInvalidSessionException) { |
| s_logger.error("Invalid cluster session found, fence it"); |
| queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeIsolated)); |
| } |
| |
| if(isRootCauseConnectionRelated(e.getCause())) { |
| s_logger.error("DB communication problem detected, fence it"); |
| queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeIsolated)); |
| } |
| |
| invalidHeartbeatConnection(); |
| } catch(ActiveFencingException e) { |
| queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeIsolated)); |
| } catch (Throwable e) { |
| s_logger.error("Unexpected exception in cluster heartbeat", e); |
| if(isRootCauseConnectionRelated(e.getCause())) { |
| s_logger.error("DB communication problem detected, fence it"); |
| queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeIsolated)); |
| } |
| |
| invalidHeartbeatConnection(); |
| } finally { |
| txn.transitToAutoManagedConnection(Transaction.CLOUD_DB); |
| txn.close("ClusterHeartBeat"); |
| } |
| } |
| }; |
| } |
| |
| private boolean isRootCauseConnectionRelated(Throwable e) { |
| while (e != null) { |
| if (e instanceof SQLRecoverableException) { |
| return true; |
| } |
| |
| e = e.getCause(); |
| } |
| |
| return false; |
| } |
| |
| private Connection getHeartbeatConnection() throws SQLException { |
| if(_heartbeatConnection == null) { |
| Connection conn = Transaction.getStandaloneConnectionWithException(); |
| _heartbeatConnection = new ConnectionConcierge("ClusterManagerHeartBeat", conn, false); |
| } |
| |
| return _heartbeatConnection.conn(); |
| } |
| |
| private void invalidHeartbeatConnection() { |
| if(_heartbeatConnection != null) { |
| Connection conn = Transaction.getStandaloneConnection(); |
| if (conn != null) { |
| _heartbeatConnection.reset(Transaction.getStandaloneConnection()); |
| } |
| } |
| } |
| |
| private Runnable getNotificationTask() { |
| return new Runnable() { |
| @Override |
| public void run() { |
| while(true) { |
| synchronized(_notificationMsgs) { |
| try { |
| _notificationMsgs.wait(1000); |
| } catch (InterruptedException e) { |
| } |
| } |
| |
| ClusterManagerMessage msg = null; |
| while((msg = getNextNotificationMessage()) != null) { |
| try { |
| switch(msg.getMessageType()) { |
| case nodeAdded: |
| if(msg.getNodes() != null && msg.getNodes().size() > 0) { |
| Profiler profiler = new Profiler(); |
| profiler.start(); |
| |
| notifyNodeJoined(msg.getNodes()); |
| |
| profiler.stop(); |
| if(profiler.getDuration() > 1000) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Notifying management server join event took " + profiler.getDuration() + " ms"); |
| } |
| } else { |
| s_logger.warn("Notifying management server join event took " + profiler.getDuration() + " ms"); |
| } |
| } |
| break; |
| |
| case nodeRemoved: |
| if(msg.getNodes() != null && msg.getNodes().size() > 0) { |
| Profiler profiler = new Profiler(); |
| profiler.start(); |
| |
| notifyNodeLeft(msg.getNodes()); |
| |
| profiler.stop(); |
| if(profiler.getDuration() > 1000) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Notifying management server leave event took " + profiler.getDuration() + " ms"); |
| } |
| } else { |
| s_logger.warn("Notifying management server leave event took " + profiler.getDuration() + " ms"); |
| } |
| } |
| break; |
| |
| case nodeIsolated: |
| notifyNodeIsolated(); |
| break; |
| |
| default : |
| assert(false); |
| break; |
| } |
| |
| } catch (Throwable e) { |
| s_logger.warn("Unexpected exception during cluster notification. ", e); |
| } |
| } |
| |
| try { Thread.sleep(1000); } catch (InterruptedException e) {} |
| } |
| } |
| }; |
| } |
| |
| private void queueNotification(ClusterManagerMessage msg) { |
| synchronized(this._notificationMsgs) { |
| this._notificationMsgs.add(msg); |
| this._notificationMsgs.notifyAll(); |
| } |
| |
| switch(msg.getMessageType()) { |
| case nodeAdded: |
| { |
| List<ManagementServerHostVO> l = msg.getNodes(); |
| if(l != null && l.size() > 0) { |
| for(ManagementServerHostVO mshost: l) { |
| _mshostPeerDao.updatePeerInfo(_mshostId, mshost.getId(), mshost.getRunid(), ManagementServerHost.State.Up); |
| } |
| } |
| } |
| break; |
| |
| case nodeRemoved: |
| { |
| List<ManagementServerHostVO> l = msg.getNodes(); |
| if(l != null && l.size() > 0) { |
| for(ManagementServerHostVO mshost: l) { |
| _mshostPeerDao.updatePeerInfo(_mshostId, mshost.getId(), mshost.getRunid(), ManagementServerHost.State.Down); |
| } |
| } |
| } |
| break; |
| |
| default : |
| break; |
| |
| } |
| } |
| |
| private ClusterManagerMessage getNextNotificationMessage() { |
| synchronized(this._notificationMsgs) { |
| if(this._notificationMsgs.size() > 0) { |
| return this._notificationMsgs.remove(0); |
| } |
| } |
| |
| return null; |
| } |
| |
| private void initPeerScan() { |
| // upon startup, for all inactive management server nodes that we see at startup time, we will send notification also to help upper layer perform |
| // missed cleanup |
| Date cutTime = DateUtil.currentGMTTime(); |
| List<ManagementServerHostVO> inactiveList = _mshostDao.getInactiveList(new Date(cutTime.getTime() - _heartbeatThreshold)); |
| |
| // We don't have foreign key constraints to enforce the mgmt_server_id integrity in host table, when user manually |
| // remove records from mshost table, this will leave orphan mgmt_serve_id reference in host table. |
| List<Long> orphanList = _mshostDao.listOrphanMsids(); |
| if(orphanList.size() > 0) { |
| for(Long orphanMsid : orphanList) { |
| // construct fake ManagementServerHostVO based on orphan MSID |
| s_logger.info("Add orphan management server msid found in host table to initial clustering notification, orphan msid: " + orphanMsid); |
| inactiveList.add(new ManagementServerHostVO(orphanMsid, 0, "orphan", 0, new Date())); |
| } |
| } else { |
| s_logger.info("We are good, no orphan management server msid in host table is found"); |
| } |
| |
| if(inactiveList.size() > 0) { |
| if(s_logger.isInfoEnabled()) { |
| s_logger.info("Found " + inactiveList.size() + " inactive management server node based on timestamp"); |
| for(ManagementServerHostVO host : inactiveList) |
| s_logger.info("management server node msid: " + host.getMsid() + ", name: " + host.getName() + ", service ip: " + host.getServiceIP() + ", version: " + host.getVersion()); |
| } |
| |
| List<ManagementServerHostVO> downHostList = new ArrayList<ManagementServerHostVO>(); |
| for(ManagementServerHostVO host : inactiveList) { |
| if(!pingManagementNode(host)) { |
| s_logger.warn("Management node " + host.getId() + " is detected inactive by timestamp and also not pingable"); |
| downHostList.add(host); |
| } |
| } |
| |
| if(downHostList.size() > 0) |
| this.queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeRemoved, downHostList)); |
| } else { |
| s_logger.info("No inactive management server node found"); |
| } |
| } |
| |
| private void peerScan() throws ActiveFencingException { |
| Date cutTime = DateUtil.currentGMTTime(); |
| |
| Profiler profiler = new Profiler(); |
| profiler.start(); |
| |
| Profiler profilerQueryActiveList = new Profiler(); |
| profilerQueryActiveList.start(); |
| List<ManagementServerHostVO> currentList = _mshostDao.getActiveList(new Date(cutTime.getTime() - _heartbeatThreshold)); |
| profilerQueryActiveList.stop(); |
| |
| Profiler profilerSyncClusterInfo = new Profiler(); |
| profilerSyncClusterInfo.start(); |
| List<ManagementServerHostVO> removedNodeList = new ArrayList<ManagementServerHostVO>(); |
| List<ManagementServerHostVO> invalidatedNodeList = new ArrayList<ManagementServerHostVO>(); |
| |
| if(_mshostId != null) { |
| |
| if(_mshostPeerDao.countStateSeenInPeers(_mshostId, _runId, ManagementServerHost.State.Down) > 0) { |
| String msg = "We have detected that at least one management server peer reports that this management server is down, perform active fencing to avoid split-brain situation"; |
| s_logger.error(msg); |
| throw new ActiveFencingException(msg); |
| } |
| |
| // only if we have already attached to cluster, will we start to check leaving nodes |
| for(Map.Entry<Long, ManagementServerHostVO> entry : _activePeers.entrySet()) { |
| |
| ManagementServerHostVO current = getInListById(entry.getKey(), currentList); |
| if(current == null) { |
| if(entry.getKey().longValue() != _mshostId.longValue()) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Detected management node left, id:" + entry.getKey() + ", nodeIP:" + entry.getValue().getServiceIP()); |
| } |
| removedNodeList.add(entry.getValue()); |
| } |
| } else { |
| if(current.getRunid() == 0) { |
| if(entry.getKey().longValue() != _mshostId.longValue()) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Detected management node left because of invalidated session, id:" + entry.getKey() + ", nodeIP:" + entry.getValue().getServiceIP()); |
| } |
| invalidatedNodeList.add(entry.getValue()); |
| } |
| } else { |
| if(entry.getValue().getRunid() != current.getRunid()) { |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Detected management node left and rejoined quickly, id:" + entry.getKey() + ", nodeIP:" + entry.getValue().getServiceIP()); |
| } |
| |
| entry.getValue().setRunid(current.getRunid()); |
| } |
| } |
| } |
| } |
| } |
| profilerSyncClusterInfo.stop(); |
| |
| Profiler profilerInvalidatedNodeList = new Profiler(); |
| profilerInvalidatedNodeList.start(); |
| // process invalidated node list |
| if(invalidatedNodeList.size() > 0) { |
| for(ManagementServerHostVO mshost : invalidatedNodeList) { |
| _activePeers.remove(mshost.getId()); |
| try { |
| JmxUtil.unregisterMBean("ClusterManager", "Node " + mshost.getId()); |
| } catch(Exception e) { |
| s_logger.warn("Unable to deregiester cluster node from JMX monitoring due to exception " + e.toString()); |
| } |
| } |
| |
| this.queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeRemoved, invalidatedNodeList)); |
| } |
| profilerInvalidatedNodeList.stop(); |
| |
| Profiler profilerRemovedList = new Profiler(); |
| profilerRemovedList.start(); |
| // process removed node list |
| Iterator<ManagementServerHostVO> it = removedNodeList.iterator(); |
| while(it.hasNext()) { |
| ManagementServerHostVO mshost = it.next(); |
| if(!pingManagementNode(mshost)) { |
| s_logger.warn("Management node " + mshost.getId() + " is detected inactive by timestamp and also not pingable"); |
| _activePeers.remove(mshost.getId()); |
| try { |
| JmxUtil.unregisterMBean("ClusterManager", "Node " + mshost.getId()); |
| } catch(Exception e) { |
| s_logger.warn("Unable to deregiester cluster node from JMX monitoring due to exception " + e.toString()); |
| } |
| } else { |
| s_logger.info("Management node " + mshost.getId() + " is detected inactive by timestamp but is pingable"); |
| it.remove(); |
| } |
| } |
| |
| if(removedNodeList.size() > 0) { |
| this.queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeRemoved, removedNodeList)); |
| } |
| profilerRemovedList.stop(); |
| |
| List<ManagementServerHostVO> newNodeList = new ArrayList<ManagementServerHostVO>(); |
| for(ManagementServerHostVO mshost : currentList) { |
| if(!_activePeers.containsKey(mshost.getId())) { |
| _activePeers.put(mshost.getId(), mshost); |
| |
| if(s_logger.isDebugEnabled()) { |
| s_logger.debug("Detected management node joined, id:" + mshost.getId() + ", nodeIP:" + mshost.getServiceIP()); |
| } |
| newNodeList.add(mshost); |
| |
| try { |
| JmxUtil.registerMBean("ClusterManager", "Node " + mshost.getId(), new ClusterManagerMBeanImpl(this, mshost)); |
| } catch(Exception e) { |
| s_logger.warn("Unable to regiester cluster node into JMX monitoring due to exception " + ExceptionUtil.toString(e)); |
| } |
| } |
| } |
| |
| if(newNodeList.size() > 0) { |
| this.queueNotification(new ClusterManagerMessage(ClusterManagerMessage.MessageType.nodeAdded, newNodeList)); |
| } |
| |
| profiler.stop(); |
| |
| if(profiler.getDuration() >= this._heartbeatInterval) { |
| if(s_logger.isDebugEnabled()) |
| s_logger.debug("Peer scan takes too long to finish. profiler: " + profiler.toString() |
| + ", profilerQueryActiveList: " + profilerQueryActiveList.toString() |
| + ", profilerSyncClusterInfo: " + profilerSyncClusterInfo.toString() |
| + ", profilerInvalidatedNodeList: " + profilerInvalidatedNodeList.toString() |
| + ", profilerRemovedList: " + profilerRemovedList.toString()); |
| } |
| } |
| |
| private static ManagementServerHostVO getInListById(Long id, List<ManagementServerHostVO> l) { |
| for(ManagementServerHostVO mshost : l) { |
| if(mshost.getId() == id) { |
| return mshost; |
| } |
| } |
| return null; |
| } |
| |
| @Override @DB |
| public boolean start() { |
| if(s_logger.isInfoEnabled()) { |
| s_logger.info("Starting cluster manager, msid : " + _msId); |
| } |
| |
| Transaction txn = Transaction.currentTxn(); |
| try { |
| txn.start(); |
| |
| final Class<?> c = this.getClass(); |
| String version = c.getPackage().getImplementationVersion(); |
| |
| ManagementServerHostVO mshost = _mshostDao.findByMsid(_msId); |
| if (mshost == null) { |
| mshost = new ManagementServerHostVO(); |
| mshost.setMsid(_msId); |
| mshost.setRunid(this.getCurrentRunId()); |
| mshost.setName(NetUtils.getHostName()); |
| mshost.setVersion(version); |
| mshost.setServiceIP(_clusterNodeIP); |
| mshost.setServicePort(_currentServiceAdapter.getServicePort()); |
| mshost.setLastUpdateTime(DateUtil.currentGMTTime()); |
| mshost.setRemoved(null); |
| mshost.setAlertCount(0); |
| mshost.setState(ManagementServerHost.State.Up); |
| _mshostDao.persist(mshost); |
| |
| if (s_logger.isInfoEnabled()) { |
| s_logger.info("New instance of management server msid " + _msId + " is being started"); |
| } |
| } else { |
| if (s_logger.isInfoEnabled()) { |
| s_logger.info("Management server " + _msId + " is being started"); |
| } |
| |
| _mshostDao.update(mshost.getId(), getCurrentRunId(), NetUtils.getHostName(), version, _clusterNodeIP, _currentServiceAdapter.getServicePort(), DateUtil.currentGMTTime()); |
| } |
| |
| txn.commit(); |
| |
| _mshostId = mshost.getId(); |
| if (s_logger.isInfoEnabled()) { |
| s_logger.info("Management server (host id : " + _mshostId + ") is being started at " + _clusterNodeIP + ":" + _currentServiceAdapter.getServicePort()); |
| } |
| |
| _mshostPeerDao.clearPeerInfo(_mshostId); |
| |
| // use seperate thread for heartbeat updates |
| _heartbeatScheduler.scheduleAtFixedRate(getHeartbeatTask(), _heartbeatInterval, _heartbeatInterval, TimeUnit.MILLISECONDS); |
| _notificationExecutor.submit(getNotificationTask()); |
| |
| } catch (Throwable e) { |
| s_logger.error("Unexpected exception : ", e); |
| txn.rollback(); |
| |
| throw new CloudRuntimeException("Unable to initialize cluster info into database"); |
| } |
| |
| if (s_logger.isInfoEnabled()) { |
| s_logger.info("Cluster manager was started successfully"); |
| } |
| |
| return true; |
| } |
| |
| @Override @DB |
| public boolean stop() { |
| if(_mshostId != null) { |
| ManagementServerHostVO mshost = _mshostDao.findByMsid(_msId); |
| mshost.setState(ManagementServerHost.State.Down); |
| _mshostDao.update(_mshostId, mshost); |
| } |
| |
| _heartbeatScheduler.shutdownNow(); |
| _executor.shutdownNow(); |
| |
| try { |
| _heartbeatScheduler.awaitTermination(EXECUTOR_SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS); |
| _executor.awaitTermination(EXECUTOR_SHUTDOWN_TIMEOUT, TimeUnit.MILLISECONDS); |
| } catch (InterruptedException e) { |
| } |
| |
| if(s_logger.isInfoEnabled()) { |
| s_logger.info("Cluster manager is stopped"); |
| } |
| |
| return true; |
| } |
| |
| @Override |
| public boolean configure(String name, Map<String, Object> params) throws ConfigurationException { |
| if(s_logger.isInfoEnabled()) { |
| s_logger.info("Start configuring cluster manager : " + name); |
| } |
| |
| Map<String, String> configs = _configDao.getConfiguration("management-server", params); |
| |
| String value = configs.get("cluster.heartbeat.interval"); |
| if (value != null) { |
| _heartbeatInterval = NumbersUtil.parseInt(value, ClusterManager.DEFAULT_HEARTBEAT_INTERVAL); |
| } |
| |
| value = configs.get("cluster.heartbeat.threshold"); |
| if (value != null) { |
| _heartbeatThreshold = NumbersUtil.parseInt(value, ClusterManager.DEFAULT_HEARTBEAT_THRESHOLD); |
| } |
| |
| File dbPropsFile = PropertiesUtil.findConfigFile("db.properties"); |
| Properties dbProps = new Properties(); |
| try { |
| dbProps.load(new FileInputStream(dbPropsFile)); |
| } catch (FileNotFoundException e) { |
| throw new ConfigurationException("Unable to find db.properties"); |
| } catch (IOException e) { |
| throw new ConfigurationException("Unable to load db.properties content"); |
| } |
| _clusterNodeIP = dbProps.getProperty("cluster.node.IP"); |
| if (_clusterNodeIP == null) { |
| _clusterNodeIP = "127.0.0.1"; |
| } |
| _clusterNodeIP = _clusterNodeIP.trim(); |
| |
| if(s_logger.isInfoEnabled()) { |
| s_logger.info("Cluster node IP : " + _clusterNodeIP); |
| } |
| |
| if(!NetUtils.isLocalAddress(_clusterNodeIP)) { |
| throw new ConfigurationException("cluster node IP should be valid local address where the server is running, please check your configuration"); |
| } |
| |
| for(int i = 0; i < DEFAULT_OUTGOING_WORKERS; i++) |
| _executor.execute(getClusterPduSendingTask()); |
| |
| // notification task itself in turn works as a task dispatcher |
| _executor.execute(getClusterPduNotificationTask()); |
| |
| if (_serviceAdapters == null) { |
| throw new ConfigurationException("Unable to get cluster service adapters"); |
| } |
| _currentServiceAdapter = _serviceAdapters.get(0); |
| |
| if(_currentServiceAdapter == null) { |
| throw new ConfigurationException("Unable to set current cluster service adapter"); |
| } |
| |
| _agentLBEnabled = Boolean.valueOf(_configDao.getValue(Config.AgentLbEnable.key())); |
| |
| String connectedAgentsThreshold = configs.get("agent.load.threshold"); |
| |
| if (connectedAgentsThreshold != null) { |
| _connectedAgentsThreshold = Double.parseDouble(connectedAgentsThreshold); |
| } |
| |
| this.registerListener(new LockMasterListener(_msId)); |
| |
| checkConflicts(); |
| |
| if(s_logger.isInfoEnabled()) { |
| s_logger.info("Cluster manager is configured."); |
| } |
| return true; |
| } |
| |
| @Override |
| public long getManagementNodeId() { |
| return _msId; |
| } |
| |
| @Override |
| public long getCurrentRunId() { |
| return _runId; |
| } |
| |
| @Override |
| public boolean isManagementNodeAlive(long msid) { |
| ManagementServerHostVO mshost = _mshostDao.findByMsid(msid); |
| if(mshost != null) { |
| if(mshost.getLastUpdateTime().getTime() >= DateUtil.currentGMTTime().getTime() - _heartbeatThreshold) { |
| return true; |
| } |
| } |
| |
| return false; |
| } |
| |
| @Override |
| public boolean pingManagementNode(long msid) { |
| ManagementServerHostVO mshost = _mshostDao.findByMsid(msid); |
| if(mshost == null) { |
| return false; |
| } |
| |
| return pingManagementNode(mshost); |
| } |
| |
| private boolean pingManagementNode(ManagementServerHostVO mshost) { |
| |
| String targetIp = mshost.getServiceIP(); |
| if("127.0.0.1".equals(targetIp) || "0.0.0.0".equals(targetIp)) { |
| s_logger.info("ping management node cluster service can not be performed on self"); |
| return false; |
| } |
| |
| int retry = 10; |
| while (--retry > 0) { |
| SocketChannel sch = null; |
| try { |
| s_logger.info("Trying to connect to " + targetIp); |
| sch = SocketChannel.open(); |
| sch.configureBlocking(true); |
| sch.socket().setSoTimeout(5000); |
| |
| InetSocketAddress addr = new InetSocketAddress(targetIp, mshost.getServicePort()); |
| sch.connect(addr); |
| return true; |
| } catch (IOException e) { |
| if (e instanceof ConnectException) { |
| s_logger.error("Unable to ping management server at " + targetIp + ":" + mshost.getServicePort() + " due to ConnectException", e); |
| return false; |
| } |
| } finally { |
| if (sch != null) { |
| try { |
| sch.close(); |
| } catch (IOException e) { |
| } |
| } |
| } |
| |
| try { |
| Thread.sleep(1000); |
| } catch (InterruptedException ex) { |
| } |
| } |
| |
| s_logger.error("Unable to ping management server at " + targetIp + ":" + mshost.getServicePort() + " after retries"); |
| return false; |
| } |
| |
| |
| @Override |
| public int getHeartbeatThreshold() { |
| return this._heartbeatThreshold; |
| } |
| |
| public int getHeartbeatInterval() { |
| return this._heartbeatInterval; |
| } |
| |
| public void setHeartbeatThreshold(int threshold) { |
| _heartbeatThreshold = threshold; |
| } |
| |
| private void checkConflicts() throws ConfigurationException { |
| Date cutTime = DateUtil.currentGMTTime(); |
| List<ManagementServerHostVO> peers = _mshostDao.getActiveList(new Date(cutTime.getTime() - _heartbeatThreshold)); |
| for(ManagementServerHostVO peer : peers) { |
| String peerIP = peer.getServiceIP().trim(); |
| if(_clusterNodeIP.equals(peerIP)) { |
| if("127.0.0.1".equals(_clusterNodeIP)) { |
| if(pingManagementNode(peer.getMsid())) { |
| String msg = "Detected another management node with localhost IP is already running, please check your cluster configuration"; |
| s_logger.error(msg); |
| throw new ConfigurationException(msg); |
| } else { |
| String msg = "Detected another management node with localhost IP is considered as running in DB, however it is not pingable, we will continue cluster initialization with this management server node"; |
| s_logger.info(msg); |
| } |
| } else { |
| if(pingManagementNode(peer.getMsid())) { |
| String msg = "Detected that another management node with the same IP " + peer.getServiceIP() + " is already running, please check your cluster configuration"; |
| s_logger.error(msg); |
| throw new ConfigurationException(msg); |
| } else { |
| String msg = "Detected that another management node with the same IP " + peer.getServiceIP() |
| + " is considered as running in DB, however it is not pingable, we will continue cluster initialization with this management server node"; |
| s_logger.info(msg); |
| } |
| } |
| } |
| } |
| } |
| |
| @Override |
| public boolean rebalanceAgent(long agentId, Event event, long currentOwnerId, long futureOwnerId) throws AgentUnavailableException, OperationTimedoutException { |
| return _rebalanceService.executeRebalanceRequest(agentId, currentOwnerId, futureOwnerId, event); |
| } |
| |
| @Override |
| public boolean isAgentRebalanceEnabled() { |
| return _agentLBEnabled; |
| } |
| |
| @Override |
| public Boolean propagateResourceEvent(long agentId, ResourceState.Event event) throws AgentUnavailableException { |
| final String msPeer = getPeerName(agentId); |
| if (msPeer == null) { |
| return null; |
| } |
| |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Propagating agent change request event:" + event.toString() + " to agent:" + agentId); |
| } |
| Command[] cmds = new Command[1]; |
| cmds[0] = new PropagateResourceEventCommand(agentId, event); |
| |
| Answer[] answers = execute(msPeer, agentId, cmds, true); |
| if (answers == null) { |
| throw new AgentUnavailableException(agentId); |
| } |
| |
| if (s_logger.isDebugEnabled()) { |
| s_logger.debug("Result for agent change is " + answers[0].getResult()); |
| } |
| |
| return answers[0].getResult(); |
| } |
| |
| @Override |
| public boolean executeResourceUserRequest(long hostId, ResourceState.Event event) throws AgentUnavailableException { |
| return _resourceMgr.executeUserRequest(hostId, event); |
| } |
| } |
