Damans227 commented on code in PR #13345: URL: https://github.com/apache/cloudstack/pull/13345#discussion_r3624822556
########## agent/src/main/java/com/cloud/agent/ServerAttache.java: ########## @@ -0,0 +1,489 @@ +// 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.agent; + +import com.cloud.agent.api.Answer; +import com.cloud.agent.api.Command; +import com.cloud.agent.transport.Request; +import com.cloud.agent.transport.Response; +import com.cloud.exception.AgentUnavailableException; +import com.cloud.exception.CloudException; +import com.cloud.exception.OperationTimedoutException; +import com.cloud.utils.concurrency.NamedThreadFactory; +import com.cloud.utils.nio.Link; +import org.apache.cloudstack.managed.context.ManagedContextRunnable; +import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.ThreadContext; + +import java.io.IOException; +import java.nio.channels.ClosedChannelException; +import java.security.SecureRandom; +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.Map; +import java.util.Optional; +import java.util.Random; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * ServerAttache provides basic server communication commands to be implemented. + * + * @author mprokopchuk Review Comment: Nit: leftover personal `@author mprokopchuk` javadoc tag (recurs in several of the other new files here too: SynchronousListener.java, AgentConnectStatusAnswer.java, AgentConnectStatusCommand.java, ThreadContextUtil.java, ConfigKeyUtil.java, BackoffFactory.java, ExponentialWithJitterBackoff.java). Probably copied over from wherever this was originally authored, worth stripping. ########## agent/src/main/java/com/cloud/agent/HostConnectProcess.java: ########## @@ -0,0 +1,355 @@ +// 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.agent; + +import com.cloud.agent.api.AgentConnectStatusAnswer; +import com.cloud.agent.api.AgentConnectStatusCommand; +import com.cloud.agent.api.Answer; +import com.cloud.agent.api.Command; +import com.cloud.agent.api.StartupAnswer; +import com.cloud.agent.api.StartupCommand; +import com.cloud.agent.properties.AgentProperties; +import com.cloud.agent.properties.AgentPropertiesFileHandler; +import com.cloud.agent.transport.Request; +import com.cloud.exception.CloudException; +import com.cloud.exception.OperationTimedoutException; +import com.cloud.host.Status; +import com.cloud.resource.ResourceStatusUpdater; +import com.cloud.resource.ServerResource; +import com.cloud.utils.concurrency.NamedThreadFactory; +import com.cloud.utils.nio.Link; +import org.apache.cloudstack.threadcontext.ThreadContextUtil; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.ThreadContext; + +import java.io.IOException; +import java.nio.channels.ClosedChannelException; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Predicate; + +public class HostConnectProcess { + private static final Logger logger = LogManager.getLogger(HostConnectProcess.class); + + public static final int DEFAULT_ASYNC_COMMAND_TIMEOUT_SEC = + AgentPropertiesFileHandler.getPropertyValue(AgentProperties.ASYNC_COMMAND_TIMEOUT_SEC); + + public static final int DEFAULT_ASYNC_STARTUP_COMMAND_TIMEOUT_SEC = + AgentPropertiesFileHandler.getPropertyValue(AgentProperties.ASYNC_STARTUP_COMMAND_TIMEOUT_SEC); + + static final long HOST_STATUS_CHECK_INITIAL_DELAY_SEC = 10; + private long hostStatusCheckDelaySec = AgentPropertiesFileHandler.getPropertyValue(AgentProperties.AGENT_HOST_STATUS_CHECK_DELAY_SEC); Review Comment: Nit: this class mixes underscore-prefixed fields (`_link`, `_futureRef`) with camelCase fields like this one in the same class. ########## agent/src/main/java/com/cloud/agent/ServerAttache.java: ########## @@ -0,0 +1,489 @@ +// 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.agent; + +import com.cloud.agent.api.Answer; +import com.cloud.agent.api.Command; +import com.cloud.agent.transport.Request; +import com.cloud.agent.transport.Response; +import com.cloud.exception.AgentUnavailableException; +import com.cloud.exception.CloudException; +import com.cloud.exception.OperationTimedoutException; +import com.cloud.utils.concurrency.NamedThreadFactory; +import com.cloud.utils.nio.Link; +import org.apache.cloudstack.managed.context.ManagedContextRunnable; +import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.ThreadContext; + +import java.io.IOException; +import java.nio.channels.ClosedChannelException; +import java.security.SecureRandom; +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.Map; +import java.util.Optional; +import java.util.Random; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * ServerAttache provides basic server communication commands to be implemented. + * + * @author mprokopchuk + */ +public class ServerAttache { + private static final Logger logger = LogManager.getLogger(ServerAttache.class); + + private static final ScheduledExecutorService s_listenerExecutor = Executors.newScheduledThreadPool(10, + new NamedThreadFactory("ListenerTimer")); + + protected static Comparator<Request> s_reqComparator = (o1, o2) -> { + long seq1 = o1.getSequence(); + long seq2 = o2.getSequence(); + if (seq1 < seq2) { + return -1; + } else if (seq1 > seq2) { + return 1; + } else { + return 0; + } + }; + + protected static Comparator<Object> s_seqComparator = (o1, o2) -> { + long seq1 = ((Request) o1).getSequence(); + long seq2 = (Long) o2; + if (seq1 < seq2) { + return -1; + } else if (seq1 > seq2) { + return 1; + } else { + return 0; + } + }; + + private static final Random s_rand = new SecureRandom(); + protected String _name; + private Link _link; + protected ConcurrentHashMap<Long, ServerListener> _waitForList; + protected ConcurrentHashMap<Long, java.util.concurrent.ScheduledFuture<?>> _alarmFutures; + protected LinkedList<Request> _requests; + protected Long _currentSequence; + protected long _nextSequence; + + protected ServerAttache(Link link) { + _name = link.getIpAddress(); + _link = link; + _waitForList = new ConcurrentHashMap<>(); + _alarmFutures = new ConcurrentHashMap<>(); + _requests = new LinkedList<>(); + _nextSequence = Long.valueOf(s_rand.nextInt(Short.MAX_VALUE)) << 48; + } + + @Override + public String toString() { + return String.format("ServerAttache %s", ReflectionToStringBuilderUtils.reflectOnlySelectedFields(this, + "_name")); + } + + public synchronized long getNextSequence() { + return ++_nextSequence; + } + + protected synchronized void addRequest(Request req) { + int index = findRequest(req); + assert (index < 0) : "How can we get index again? " + index + ":" + req.toString(); + _requests.add(-index - 1, req); + } + + protected void cancel(Request req) { + cancel(req.getSequence()); + } + + protected synchronized void cancel(long seq) { + logger.debug(log(seq, "Cancelling.")); + + ServerListener listener = _waitForList.remove(seq); + if (listener != null) { + listener.processDisconnect(); + } + int index = findRequest(seq); + if (index >= 0) { + _requests.remove(index); + } + } + + protected synchronized int findRequest(Request req) { + return Collections.binarySearch(_requests, req, s_reqComparator); + } + + protected synchronized int findRequest(long seq) { + return Collections.binarySearch(_requests, seq, s_seqComparator); + } + + protected String log(long seq, String msg) { + return "Seq " + _name + "-" + seq + ": " + msg; + } + + protected void registerListener(long seq, ServerListener listener) { + if (logger.isTraceEnabled()) { + logger.trace(log(seq, "Registering listener")); + } + if (listener.getTimeout() != -1) { + java.util.concurrent.ScheduledFuture<?> alarmFuture = + s_listenerExecutor.schedule(new Alarm(seq), listener.getTimeout(), TimeUnit.SECONDS); + _alarmFutures.put(seq, alarmFuture); + } + _waitForList.put(seq, listener); + } + + protected ServerListener unregisterListener(long sequence) { + if (logger.isTraceEnabled()) { + logger.trace(log(sequence, "Unregistering listener")); + } + java.util.concurrent.ScheduledFuture<?> alarmFuture = _alarmFutures.remove(sequence); + if (alarmFuture != null) { + alarmFuture.cancel(false); + } + return _waitForList.remove(sequence); + } + + protected ServerListener getListener(long sequence) { + return _waitForList.get(sequence); + } + + public String getName() { + return _name; + } + + public int getQueueSize() { + return _requests.size(); + } + + public boolean processAnswers(long seq, Response resp) { + resp.logD("Processing: ", true); + boolean processed = false; + Answer[] answers = resp.getAnswers(); + try { + ServerListener monitor = getListener(seq); + if (monitor == null) { + if (answers[0] != null && answers[0].getResult()) { + processed = true; + } + logger.debug(log(seq, "Unable to find listener.")); + } else { + processed = monitor.processAnswers(seq, answers); + logger.trace(log(seq, (processed ? "" : " did not ") + " processed ")); Review Comment: Nit: this produces "did not processed" (double space, missing verb) when not processed. ########## agent/src/main/java/com/cloud/agent/AgentShell.java: ########## @@ -568,6 +582,7 @@ public static void main(String[] args) { shell.init(args); shell.start(); } catch (ConfigurationException e) { + LOGGER.fatal(e.getMessage(), e); Review Comment: Nit: this now logs the same message twice, `LOGGER.fatal` here followed by `System.out.println` right below with the same `e.getMessage()`. ########## agent/src/main/java/com/cloud/agent/HostConnectProcess.java: ########## @@ -0,0 +1,355 @@ +// 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.agent; + +import com.cloud.agent.api.AgentConnectStatusAnswer; +import com.cloud.agent.api.AgentConnectStatusCommand; +import com.cloud.agent.api.Answer; +import com.cloud.agent.api.Command; +import com.cloud.agent.api.StartupAnswer; +import com.cloud.agent.api.StartupCommand; +import com.cloud.agent.properties.AgentProperties; +import com.cloud.agent.properties.AgentPropertiesFileHandler; +import com.cloud.agent.transport.Request; +import com.cloud.exception.CloudException; +import com.cloud.exception.OperationTimedoutException; +import com.cloud.host.Status; +import com.cloud.resource.ResourceStatusUpdater; +import com.cloud.resource.ServerResource; +import com.cloud.utils.concurrency.NamedThreadFactory; +import com.cloud.utils.nio.Link; +import org.apache.cloudstack.threadcontext.ThreadContextUtil; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.ThreadContext; + +import java.io.IOException; +import java.nio.channels.ClosedChannelException; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Predicate; + +public class HostConnectProcess { + private static final Logger logger = LogManager.getLogger(HostConnectProcess.class); + + public static final int DEFAULT_ASYNC_COMMAND_TIMEOUT_SEC = + AgentPropertiesFileHandler.getPropertyValue(AgentProperties.ASYNC_COMMAND_TIMEOUT_SEC); + + public static final int DEFAULT_ASYNC_STARTUP_COMMAND_TIMEOUT_SEC = + AgentPropertiesFileHandler.getPropertyValue(AgentProperties.ASYNC_STARTUP_COMMAND_TIMEOUT_SEC); + + static final long HOST_STATUS_CHECK_INITIAL_DELAY_SEC = 10; + private long hostStatusCheckDelaySec = AgentPropertiesFileHandler.getPropertyValue(AgentProperties.AGENT_HOST_STATUS_CHECK_DELAY_SEC); + private final AtomicReference<ScheduledFuture<?>> hostStatusFutureRef = new AtomicReference<>(); + private final Agent agent; + private ScheduledExecutorService hostStatusExecutor; + + public HostConnectProcess(Agent agent) { + this.agent = agent; + initExecutors(); + } + + private void initExecutors() { + stop(); + var threadFactory = new NamedThreadFactory("Agent-" + HostStatusTask.class.getSimpleName()); + hostStatusExecutor = Executors.newScheduledThreadPool(1, threadFactory); + } + + /** + * Stops the whole connect process and cancels all scheduled asynchronous tasks. + * Returns {@link Boolean#TRUE} if {@link HostConnectProcess} was waiting for {@link StartupAnswer}. + */ + public boolean stop() { + logger.debug("Stopping connect process. The process is active: {}", isInProgress()); + stopHostStatusExecutor(); + logger.debug("Stopped executor"); + Optional<? extends ScheduledFuture<?>> hostStatusOpt = Optional.ofNullable(hostStatusFutureRef.getAndSet(null)) + .filter(Predicate.not(ScheduledFuture::isCancelled)); + + hostStatusOpt.ifPresent(future -> future.cancel(true)); + logger.debug("Cancelled future"); + + return hostStatusOpt.isPresent(); + } + + private void stopHostStatusExecutor() { + if (hostStatusExecutor != null) { + hostStatusExecutor.shutdownNow(); + hostStatusExecutor = null; + } + } + + public void scheduleConnectProcess(Link link, boolean connectionTransfer) { + logger.debug("Scheduling connect process for {}", link); + initExecutors(); + + var task = new HostStatusTask(link, connectionTransfer, agent, hostStatusFutureRef); + var future = hostStatusExecutor.scheduleWithFixedDelay(ThreadContextUtil.wrapThreadContext(task), + HOST_STATUS_CHECK_INITIAL_DELAY_SEC, + hostStatusCheckDelaySec, TimeUnit.SECONDS); + hostStatusFutureRef.set(future); + } + + /** + * Returns {@link Boolean#TRUE} if {@link HostStatusTask} created and scheduled. + * That means there is already {@link Status#Connecting} process is running. + */ + public boolean isInProgress() { + return Optional.ofNullable(hostStatusFutureRef.get()) + .filter(Predicate.not(ScheduledFuture::isCancelled)).isPresent(); + } + + public void updateHostStatusCheckDelay(int newDelaySec) { + logger.info("Updating host status check delay from {} to {} seconds", hostStatusCheckDelaySec, newDelaySec); + this.hostStatusCheckDelaySec = newDelaySec; + } + + /** + * Task wait for the Host to be available to connect to submit {@link StartupCommand}. Review Comment: Nit: javadoc typo, "Task wait for the Host" should be "waits". ########## server/src/main/java/org/apache/cloudstack/agent/lb/IndirectAgentLBServiceImpl.java: ########## @@ -446,6 +446,7 @@ protected boolean migrateNonRoutingHostAgentsInZone(String fromMsUuid, long from break; } + // FIXME: it is fire and forget, Management Server will never know if task failed Review Comment: Nit: new `FIXME` comment, might be worth a tracked follow-up instead of an inline note. ########## engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java: ########## @@ -1253,15 +1635,16 @@ public Answer easySend(final Long hostId, final Command cmd) { } return answer; - + // FIXME: There are a lot of result != null checks in callers. Should this method trow exception? Review Comment: Nit: leftover `FIXME` comment carried over in the moved code. ########## core/src/main/java/org/apache/cloudstack/threadcontext/ThreadContextUtil.java: ########## @@ -0,0 +1,59 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +package org.apache.cloudstack.threadcontext; + +import org.apache.logging.log4j.ThreadContext; + +import java.util.HashMap; +import java.util.Map; + +/** + * Utility class, helps to propagate {@link ThreadContext} values from parent to child threads. + * + * @author mprokopchuk + */ +public class ThreadContextUtil { + /** + * Wrap {@link Runnable} to propagate {@link ThreadContext} values. + * + * @param delegate + * @return + */ + public static Runnable wrapThreadContext(Runnable delegate) { + @SuppressWarnings("unchecked") Review Comment: Nit: `@SuppressWarnings("unchecked")` here (and again below) looks unnecessary, no actual unchecked cast on this line since `ThreadContext.getContext()` is already generically typed. Also the javadoc above has empty `@param`/`@return` tags with no description. ########## framework/db/src/main/java/com/cloud/utils/db/GlobalLock.java: ########## @@ -45,127 +43,248 @@ * </p> */ public class GlobalLock { - protected Logger logger = LogManager.getLogger(getClass()); + protected final static Logger logger = LogManager.getLogger(GlobalLock.class); private String name; - private int lockCount = 0; - private Thread ownerThread = null; - - private int referenceCount = 0; - private long holdingStartTick = 0; - - private static Map<String, GlobalLock> s_lockMap = new HashMap<String, GlobalLock>(); + /** + * DB lock count. + * Increments on {@link GlobalLock#lock(int)} and decrements on {@link GlobalLock#unlock()}. + * Upon {@link GlobalLock#unlock()}, if {@link GlobalLock#lockCount} is less than 1, then lock removed from DB + */ + private int lockCount; + + /** + * Internal (in-memory) lock count. + * Increments on {@link GlobalLock#addRef()} and indirectly on {@link GlobalLock#getInternLock(String)} and + * decrements on {@link GlobalLock#releaseRef()}, {@link GlobalLock#unlock()} and on {@link GlobalLock#lock(int)} + * if DB lock is unsuccessful + */ + private int referenceCount; + + /** + * Thread that owns lock. If lock called from different thread, it will be waiting for the owner to unlock it + * within requested timeout. If owner thread call {@link GlobalLock#lock(int)} again, then + * {@link GlobalLock#lockCount} will be incremented. + * If {@link GlobalLock#unlock()} called by owner thread, or DB lock will be unsuccessful, then owner thread will be + * nullified. + */ + private Thread ownerThread; + + /** + * Variable to hold lock duration in milliseconds. Used for information only. + */ + private long holdingStartTick; + + /** + * Holds all created locks. + */ + private static Map<String, GlobalLock> s_lockMap = new HashMap<>(); + + /** + * Create lock. + * + * @param name lock name + */ private GlobalLock(String name) { this.name = name; } + /** + * Increment reference count to lock. + * + * @return reference count + */ public int addRef() { synchronized (this) { referenceCount++; return referenceCount; } } + /** + * Decrement reference count to lock. + * + * @return reference count + */ public int releaseRef() { - int refCount; - boolean needToRemove = false; synchronized (this) { + if (logger.isDebugEnabled()) { + logger.debug("Releasing reference for internal lock {}, reference count: {}, lock count: {}", + name, referenceCount, lockCount); + } referenceCount--; - refCount = referenceCount; - - if (referenceCount < 0) - logger.warn("Unmatched Global lock " + name + " reference usage detected, check your code!"); - if (referenceCount == 0) + if (referenceCount < 0) { + logger.warn("Unmatched internal lock {} reference usage detected (reference count: {}, " + + "lock count: {}), check your code!", name, referenceCount, lockCount); + } else if (referenceCount < 1) { needToRemove = true; + } } - if (needToRemove) + if (needToRemove) { + if (logger.isDebugEnabled()) { + logger.debug("Need to release internal lock {}", name); + } releaseInternLock(name); + } + if (logger.isDebugEnabled()) { + logger.debug("Released reference for lock {}, reference count: {}", name, referenceCount); + } + return referenceCount; + } - return refCount; + public static boolean isLockAvailable(String name) { + if (logger.isDebugEnabled()) { + logger.debug("Checking lock availability for {}", name); + } + boolean result = false; + try { + result = DbUtil.isFreeLock(name); + } finally { + if (logger.isDebugEnabled()) { + logger.debug("Result of checking lock availability for {}: {}", name, result); + } + } + return result; } + /** + * Registers internal lock (in memory) object. Does not create any lock in DB yet. + * + * @param name lock name + * @return lock object + */ public static GlobalLock getInternLock(String name) { synchronized (s_lockMap) { + GlobalLock lock; if (s_lockMap.containsKey(name)) { - GlobalLock lock = s_lockMap.get(name); - lock.addRef(); - return lock; + lock = s_lockMap.get(name); + if (logger.isDebugEnabled()) { + logger.debug("Internal lock {} already exists with reference count {} and lock count {}", + name, lock.referenceCount, lock.lockCount); + } } else { - GlobalLock lock = new GlobalLock(name); - lock.addRef(); + lock = new GlobalLock(name); + if (logger.isDebugEnabled()) { + logger.debug("Internal lock {} does not exist, adding", name); + } s_lockMap.put(name, lock); - return lock; } + lock.addRef(); + if (logger.isDebugEnabled()) { + logger.debug("Added reference to internal lock {}, reference count {}, lock count {}", + name, lock.referenceCount, lock.lockCount); + } + return lock; } } + /** + * Unregister internal lock (in memory) object. Does not remove any lock from DB. + * + * @param name lock name + */ private void releaseInternLock(String name) { synchronized (s_lockMap) { GlobalLock lock = s_lockMap.get(name); if (lock != null) { - if (lock.referenceCount == 0) + if (lock.referenceCount == 0) { + if (logger.isDebugEnabled()) { + logger.debug("Released internal lock {}", name); + } s_lockMap.remove(name); + } else { + if (logger.isDebugEnabled()) { + logger.debug("Not releasing internal lock {} as it has references count: {}, lock count: {}", + name, lock.referenceCount, lock.lockCount); + } + } } else { - logger.warn("Releasing " + name + ", but it is already released."); + logger.warn("Internal lock {} already released", name); } } } + /** + * Acquire or join existing DB lock. + * + * @param timeoutSeconds time in seconds during which lock needs to be obtained (it is not the lock duration) + * @return true if lock successfully obtained + */ public boolean lock(int timeoutSeconds) { int remainingMilliSeconds = timeoutSeconds * 1000; Profiler profiler = new Profiler(); boolean interrupted = false; try { while (true) { synchronized (this) { - if (ownerThread != null && ownerThread == Thread.currentThread()) { - logger.warn("Global lock re-entrance detected"); - + if (ownerThread == Thread.currentThread()) { + logger.warn("Global lock {} re-entrance detected, owner thread: {}, reference count: {}, " + + "lock count: {}", name, getThreadName(ownerThread), referenceCount, lockCount); + // if it is re-entrance, then we may have more lock counts than needed? lockCount++; - if (logger.isTraceEnabled()) - logger.trace("lock " + name + " is acquired, lock count :" + lockCount); + if (logger.isDebugEnabled()) { + logger.debug("Global lock {} joined, reference count: {}, lock count: {}", + name, referenceCount, lockCount); + } return true; - } - - if (ownerThread != null) { + } else if (ownerThread != null) { profiler.start(); try { - wait((timeoutSeconds) * 1000L); + logger.debug("Waiting {} seconds to acquire global lock {}", timeoutSeconds, name); + wait(timeoutSeconds * 1000L); } catch (InterruptedException e) { interrupted = true; } profiler.stop(); remainingMilliSeconds -= profiler.getDurationInMillis(); - if (remainingMilliSeconds < 0) + if (remainingMilliSeconds < 0) { + logger.warn("Timeout of {} seconds to acquire global lock {} has been reached, " + + "owner thread {}, reference count: {}, lock count: {}", timeoutSeconds, name, getThreadName(ownerThread), referenceCount, lockCount); return false; + } continue; } else { // take ownership temporarily to prevent others enter into stage of acquiring DB lock ownerThread = Thread.currentThread(); + // XXX: do we need it here (???) Review Comment: Nit: leftover casual "XXX: do we need it here (???)" comment carried over in the moved code. ########## agent/src/main/java/com/cloud/agent/ServerAttache.java: ########## @@ -0,0 +1,489 @@ +// 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.agent; + +import com.cloud.agent.api.Answer; +import com.cloud.agent.api.Command; +import com.cloud.agent.transport.Request; +import com.cloud.agent.transport.Response; +import com.cloud.exception.AgentUnavailableException; +import com.cloud.exception.CloudException; +import com.cloud.exception.OperationTimedoutException; +import com.cloud.utils.concurrency.NamedThreadFactory; +import com.cloud.utils.nio.Link; +import org.apache.cloudstack.managed.context.ManagedContextRunnable; +import org.apache.cloudstack.utils.reflectiontostringbuilderutils.ReflectionToStringBuilderUtils; +import org.apache.commons.lang3.ArrayUtils; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; +import org.apache.logging.log4j.ThreadContext; + +import java.io.IOException; +import java.nio.channels.ClosedChannelException; +import java.security.SecureRandom; +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.LinkedList; +import java.util.Map; +import java.util.Optional; +import java.util.Random; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * ServerAttache provides basic server communication commands to be implemented. + * + * @author mprokopchuk + */ +public class ServerAttache { + private static final Logger logger = LogManager.getLogger(ServerAttache.class); + + private static final ScheduledExecutorService s_listenerExecutor = Executors.newScheduledThreadPool(10, + new NamedThreadFactory("ListenerTimer")); + + protected static Comparator<Request> s_reqComparator = (o1, o2) -> { + long seq1 = o1.getSequence(); + long seq2 = o2.getSequence(); + if (seq1 < seq2) { + return -1; + } else if (seq1 > seq2) { + return 1; + } else { + return 0; + } + }; + + protected static Comparator<Object> s_seqComparator = (o1, o2) -> { + long seq1 = ((Request) o1).getSequence(); + long seq2 = (Long) o2; + if (seq1 < seq2) { + return -1; + } else if (seq1 > seq2) { + return 1; + } else { + return 0; + } + }; + + private static final Random s_rand = new SecureRandom(); + protected String _name; + private Link _link; + protected ConcurrentHashMap<Long, ServerListener> _waitForList; + protected ConcurrentHashMap<Long, java.util.concurrent.ScheduledFuture<?>> _alarmFutures; + protected LinkedList<Request> _requests; + protected Long _currentSequence; + protected long _nextSequence; + + protected ServerAttache(Link link) { + _name = link.getIpAddress(); + _link = link; + _waitForList = new ConcurrentHashMap<>(); + _alarmFutures = new ConcurrentHashMap<>(); + _requests = new LinkedList<>(); + _nextSequence = Long.valueOf(s_rand.nextInt(Short.MAX_VALUE)) << 48; + } + + @Override + public String toString() { + return String.format("ServerAttache %s", ReflectionToStringBuilderUtils.reflectOnlySelectedFields(this, + "_name")); + } + + public synchronized long getNextSequence() { + return ++_nextSequence; + } + + protected synchronized void addRequest(Request req) { + int index = findRequest(req); + assert (index < 0) : "How can we get index again? " + index + ":" + req.toString(); + _requests.add(-index - 1, req); + } + + protected void cancel(Request req) { + cancel(req.getSequence()); + } + + protected synchronized void cancel(long seq) { + logger.debug(log(seq, "Cancelling.")); + + ServerListener listener = _waitForList.remove(seq); + if (listener != null) { + listener.processDisconnect(); + } + int index = findRequest(seq); + if (index >= 0) { + _requests.remove(index); + } + } + + protected synchronized int findRequest(Request req) { + return Collections.binarySearch(_requests, req, s_reqComparator); + } + + protected synchronized int findRequest(long seq) { + return Collections.binarySearch(_requests, seq, s_seqComparator); + } + + protected String log(long seq, String msg) { + return "Seq " + _name + "-" + seq + ": " + msg; + } + + protected void registerListener(long seq, ServerListener listener) { + if (logger.isTraceEnabled()) { + logger.trace(log(seq, "Registering listener")); + } + if (listener.getTimeout() != -1) { + java.util.concurrent.ScheduledFuture<?> alarmFuture = + s_listenerExecutor.schedule(new Alarm(seq), listener.getTimeout(), TimeUnit.SECONDS); + _alarmFutures.put(seq, alarmFuture); + } + _waitForList.put(seq, listener); + } + + protected ServerListener unregisterListener(long sequence) { + if (logger.isTraceEnabled()) { + logger.trace(log(sequence, "Unregistering listener")); + } + java.util.concurrent.ScheduledFuture<?> alarmFuture = _alarmFutures.remove(sequence); + if (alarmFuture != null) { + alarmFuture.cancel(false); + } + return _waitForList.remove(sequence); + } + + protected ServerListener getListener(long sequence) { + return _waitForList.get(sequence); + } + + public String getName() { + return _name; + } + + public int getQueueSize() { + return _requests.size(); + } + + public boolean processAnswers(long seq, Response resp) { + resp.logD("Processing: ", true); + boolean processed = false; + Answer[] answers = resp.getAnswers(); + try { + ServerListener monitor = getListener(seq); + if (monitor == null) { + if (answers[0] != null && answers[0].getResult()) { + processed = true; + } + logger.debug(log(seq, "Unable to find listener.")); + } else { + processed = monitor.processAnswers(seq, answers); + logger.trace(log(seq, (processed ? "" : " did not ") + " processed ")); + if (!monitor.isRecurring()) { + unregisterListener(seq); + } + } + } finally { + // we should always trigger next command execution, even in failure cases - otherwise in exception case + // all the remaining will be stuck in the sync queue forever + if (resp.executeInSequence()) { + sendNext(seq); + } + } + return processed; + } + + protected void cancelAllCommands() { + for (Iterator<Map.Entry<Long, ServerListener>> it = _waitForList.entrySet().iterator(); it.hasNext(); ) { + Map.Entry<Long, ServerListener> entry = it.next(); + it.remove(); + + long seq = entry.getKey(); + java.util.concurrent.ScheduledFuture<?> alarmFuture = _alarmFutures.remove(seq); + if (alarmFuture != null) { + alarmFuture.cancel(false); + } + ServerListener monitor = entry.getValue(); + logger.debug(log(seq, "Sending disconnect to " + monitor.getClass())); + monitor.processDisconnect(); + } + } + + public void cleanup() { + cancelAllCommands(); + _requests.clear(); + } + + @Override + public boolean equals(Object obj) { + // Return false straight away. + if (obj == null) { + return false; + } + // No need to handle a ClassCastException. If the classes are different, then equals can return false + // straight ahead. + if (this.getClass() != obj.getClass()) { + return false; + } + ServerAttache that = (ServerAttache) obj; + return _name.equals(that.getName()); + } + + public void send(Request req, ServerListener listener) throws CloudException { + long seq = req.getSequence(); + if (listener != null) { + registerListener(seq, listener); + } + + synchronized (this) { + try { + if (isClosed()) { + throw new CloudException("The link to the server " + _name + " has been closed", new ClosedChannelException()); + } + + if (req.executeInSequence() && _currentSequence != null) { + req.logD("Waiting for Seq " + _currentSequence + " Scheduling: ", true); + addRequest(req); + return; + } + + // If we got to here either we're not supposed to set the _currentSequence or it is null already. + req.logD("Sending ", true); + send(req); + + if (req.executeInSequence() && _currentSequence == null) { + _currentSequence = seq; + logger.trace(log(seq, " is current sequence")); + } + } catch (CloudException e) { + logger.info(log(seq, "Unable to send due to " + e.getMessage()), e); + cancel(seq); + throw e; + } catch (Exception e) { + logger.warn(log(seq, "Unable to send due to " + e.getMessage()), e); + cancel(seq); + throw new CloudException("Problem due to other exception " + e.getMessage(), e); + } + } + } + + public Answer[] send(Request req, int wait) throws CloudException { + if (req.getSequence() <= 0) { + req.setSequence(getNextSequence()); + } + SynchronousListener sl = new SynchronousListener(); + sl.setTimeout(wait + 5); + send(req, sl); + + long seq = req.getSequence(); + try { + for (int i = 0; i < 2; i++) { + Answer[] answers = null; + try { + answers = sl.waitFor(wait); + } catch (InterruptedException e) { + logger.debug(log(seq, "Interrupted")); + } + if (answers != null) { + new Response(req, answers).logD("Received: ", false); + return answers; + } + + answers = sl.getAnswers(); // Try it again. + if (answers != null) { + new Response(req, answers).logD("Received after timeout: ", true); + return answers; + } + + Long current = _currentSequence; + if (current != null && seq != current) { + if (logger.isDebugEnabled()) { + logger.debug(log(seq, "Waited too long.")); + } + throw new OperationTimedoutException(req.getCommands(), -1, seq, wait, false); + } + logger.debug(log(seq, "Waiting some more time because this is the current command")); + } + throw new OperationTimedoutException(req.getCommands(), -1, seq, wait * 2, true); + } catch (OperationTimedoutException e) { + logger.warn(log(seq, "Timed out on " + req)); + cancel(seq); + Long current = _currentSequence; + if (req.executeInSequence() && (current != null && current == seq)) { + sendNext(seq); + } + throw e; + } catch (Exception e) { + logger.warn(log(seq, "Exception while waiting for answer"), e); + cancel(seq); + Long current = _currentSequence; + if (req.executeInSequence() && (current != null && current == seq)) { + sendNext(seq); + } + throw new OperationTimedoutException(req.getCommands(), -1, seq, wait, false); + } finally { + unregisterListener(seq); + } + } + + protected synchronized void sendNext(long seq) { + _currentSequence = null; + if (_requests.isEmpty()) { + if (logger.isDebugEnabled()) { + logger.debug(log(seq, "No more commands found")); + } + return; + } + + Request req = _requests.pop(); + if (logger.isDebugEnabled()) { + logger.debug(log(req.getSequence(), "Sending now. is current sequence.")); + } + try { + send(req); + } catch (CloudException e) { + logger.debug(log(req.getSequence(), "Unable to send the next sequence")); + cancel(req.getSequence()); + } + _currentSequence = req.getSequence(); + } + + public void process(Answer[] answers) { + //do nothing + } + + /** + * sends the request asynchronously. + * + * @param req + * @throws AgentUnavailableException + */ + public synchronized void send(final Request req) throws CloudException { + try { + _link.send(req.toBytes()); + } catch (ClosedChannelException e) { + throw new CloudException("Channel is closed", e); + } + } + + /** + * Process disconnect. + */ + public void disconnect() { + synchronized (this) { + logger.debug("Processing Disconnect."); + if (_link != null) { + _link.close(); + _link.terminated(); + } + _link = null; + } + cancelAllCommands(); + _requests.clear(); + } + + /** + * Is the agent closed for more commands? + * + * @return true if unable to reach agent or false if reachable. + */ + protected boolean isClosed() { + return _link == null; + } + + public Link getLink() { + return _link; + } + + public <T> T send(Long agentId, Command[] commands, Class<T> answerType, int asyncCommandTimeoutSec) throws IOException { + String logId = Optional.ofNullable(ThreadContext.get("logcontextid")) + .filter(String.class::isInstance) + .map(String.class::cast) + .orElse(null); + if (logId != null) { + for (Command command : commands) { + if (command.getContextParam("logid") == null) { + command.setContextParam("logid", logId); + } + } + } + Link link = getLink(); + String commandName = commands[0].getClass().getSimpleName(); + + Long id = Optional.ofNullable(agentId).orElse(-1L); + Request request = new Request(id, -1, commands, true, false); + Answer[] answers; + try { + answers = send(request, asyncCommandTimeoutSec); + } catch (OperationTimedoutException e) { + String msg = String.format("Failed to retrieve answer for command %s to %s", commandName, link); + logger.error(msg, e); + throw new RuntimeException(msg, e); + } catch (CloudException e) { + String msg = String.format("Failed to send command %s to %s", commandName, link); + logger.error(msg, e); + Throwable cause = e.getCause(); + if (cause instanceof ClosedChannelException) { + throw (ClosedChannelException) cause; + } + throw new RuntimeException(msg, e); + } + if (ArrayUtils.isEmpty(answers)) { + String msg = String.format("Received empty %s response from %s", commandName, link); + logger.warn(msg); + throw new IllegalArgumentException(msg); + } + Answer answer = answers[0]; + String details = answer.getDetails(); + if (!answer.getResult()) { + String msg = String.format("Received unsuccessful %s response status from %s: %s", commandName, link, + details); + logger.warn(msg); + throw new IllegalArgumentException(msg); + } + String answerName = answer.getClass().getSimpleName(); + if (!answerType.isInstance(answer)) { + String msg = String.format("Received unexpected %s response type %s from %s: %s", commandName, + answerName, link, details); + logger.warn(msg); + throw new IllegalArgumentException(msg); + } + return answerType.cast(answer); + } + + protected class Alarm extends ManagedContextRunnable { + long _seq; + + public Alarm(long seq) { + _seq = seq; + } + + @Override + protected void runInContext() { + try { + ServerListener listener = unregisterListener(_seq); + if (listener != null) { + cancel(_seq); + listener.processTimeout(_seq); + } + } catch (Exception e) { + ServerAttache.logger.warn("Exception ", e); Review Comment: Nit: vague log message ("Exception " with a trailing space), no context on what actually failed. ########## engine/orchestration/src/main/java/com/cloud/agent/manager/AgentManagerImpl.java: ########## @@ -762,40 +985,124 @@ public void notifyMonitorsOfNewlyAddedHost(long hostId) { protected AgentAttache notifyMonitorsOfConnection(final AgentAttache attache, final StartupCommand[] cmds, final boolean forRebalance) throws ConnectionException { final long hostId = attache.getId(); final HostVO host = _hostDao.findById(hostId); - for (final Pair<Integer, Listener> monitor : _hostMonitors) { - logger.debug("Sending Connect to listener: {}, for rebalance: {}", monitor.second().getClass().getSimpleName(), forRebalance); - for (StartupCommand cmd : cmds) { - try { - logger.debug("process connection to issue: {} for host: {}, forRebalance: {}", ReflectionToStringBuilderUtils.reflectOnlySelectedFields(cmd, "id", "type", "msHostList", "connectionTransferred"), hostId, forRebalance); - monitor.second().processConnect(host, cmd, forRebalance); - } catch (final ConnectionException ce) { - if (ce.isSetupError()) { - logger.warn("Monitor {} says there is an error in the connect process for {} due to {}", monitor.second().getClass().getSimpleName(), hostId, ce.getMessage()); - handleDisconnectWithoutInvestigation(attache, Event.AgentDisconnected, true, true); - throw ce; - } else { - logger.info("Monitor {} says not to continue the connect process for {} due to {}", monitor.second().getClass().getSimpleName(), hostId, ce.getMessage()); - handleDisconnectWithoutInvestigation(attache, Event.ShutdownRequested, true, true); - return attache; - } - } catch (final HypervisorVersionChangedException hvce) { - handleDisconnectWithoutInvestigation(attache, Event.ShutdownRequested, true, true); - throw new CloudRuntimeException("Unable to connect " + attache.getId(), hvce); - } catch (final Exception e) { - logger.error("Monitor {} says there is an error in the connect process for {} due to {}", monitor.second().getClass().getSimpleName(), hostId, e.getMessage(), e); - handleDisconnectWithoutInvestigation(attache, Event.AgentDisconnected, true, true); - throw new CloudRuntimeException("Unable to connect " + attache.getId(), e); - } + Optional<HostVO> hostOpt = Optional.ofNullable(host); + String hostName = hostOpt.map(HostVO::getName).orElse(null); + String hostUuid = hostOpt.map(HostVO::getUuid).orElse(null); + boolean processSuccessful; + Profiler processProfiler = new Profiler(); + processProfiler.start(); + logger.debug("Process connect for host: {} ({}) for rebalance: {}", hostName, hostUuid, forRebalance); + try { + processSuccessful = performProcessConnect(attache, cmds, forRebalance, hostName, hostUuid, host); + } finally { + processProfiler.stop(); + long processDurationSec = processProfiler.getDurationInMillis() / 1000; + logger.debug("Finished process connect for host: {} ({}) for rebalance: {} duration, s: {}", + hostName, hostUuid, forRebalance, processDurationSec); + } + if (processSuccessful) { + sendReadyCommand(attache, host, hostId); + + agentStatusTransitTo(host, Event.Ready, _nodeId); + attache.ready(); + } + return attache; + } + + /** + * @return true if commands successfully completed + */ + private boolean performProcessConnect(AgentAttache attache, StartupCommand[] cmds, boolean forRebalance, String hostName, String hostUuid, HostVO host) throws ConnectionException { + int monitorsSize = _hostMonitors.size(); + boolean processSuccessful = true; + for (int monitorIndex = 0; monitorIndex < monitorsSize && processSuccessful; monitorIndex++) { + Pair<Integer, Listener> monitor = _hostMonitors.get(monitorIndex); + Listener listener = monitor.second(); + String monitorClassName = listener.getClass().getSimpleName(); + Profiler monitorProfiler = new Profiler(); + logger.debug("Process connect for host: {} ({}) listener {} ({} of {}) for rebalance: {}", + hostName, hostUuid, monitorClassName, monitorIndex, monitorsSize, forRebalance); + try { + monitorProfiler.start(); + processSuccessful = performProcessConnect(attache, cmds, forRebalance, hostName, hostUuid, monitorClassName, listener, host); + } finally { + monitorProfiler.stop(); + long monitorDurationSec = monitorProfiler.getDurationInMillis() / 1000; + logger.debug("Finished process connect for host: {} ({}) listener: {} ({} of {}) for rebalance: {} duration, s: {}", + hostName, hostUuid, monitorClassName, monitorIndex, monitorsSize, forRebalance, monitorDurationSec); + } + } + return processSuccessful; + } + + /** + * @return true if commands successfully completed + */ + private boolean performProcessConnect(AgentAttache attache, StartupCommand[] cmds, boolean forRebalance, String hostName, String hostUuid, String monitorClassName, Listener listener, HostVO host) throws ConnectionException { + boolean processSuccessful = true; + int commandsSize = cmds.length; + for (int commandIndex = 0; commandIndex < commandsSize && processSuccessful; commandIndex++) { + StartupCommand command = cmds[commandIndex]; + String commandClassName = command.getClass().getSimpleName(); + boolean connTransferred = command.isConnectionTransferred(); + long processStart = System.currentTimeMillis(); + try { + logger.debug("Process command: {} ({} of {}) in connect for host: {} ({}) listener: {} for rebalance: {} connection transferred: {}", + commandClassName, commandIndex, commandsSize, hostName, hostUuid, monitorClassName, forRebalance, connTransferred); + listener.processConnect(host, command, forRebalance); + } catch (Exception e) { + handleConnectionException(attache, e, processStart, host, monitorClassName); + processSuccessful = false; + } finally { + long commandDurationSec = (System.currentTimeMillis() - processStart) / 1000; + logger.debug("Finished processing command: {} ({} of {}) in connect for host: {} ({}) listener: {} for rebalance: {} connection transferred: {} duration, s: {}", + commandClassName, commandIndex, commandsSize, hostName, hostUuid, monitorClassName, forRebalance, connTransferred, commandDurationSec); + } + } + return processSuccessful; + } + + private void handleConnectionException(AgentAttache attache, Exception e, long processStart, HostVO host, String monitorClassName) throws ConnectionException { + long processDurationMs = System.currentTimeMillis() - processStart; + StringBuilder summaryBuilder = getSummaryMsgBuilder("Failed to connect Host", + EventTypes.EVENT_HOST_RECONNECT, host.getUuid(), host.getName(), null, + e.getMessage(), monitorClassName, processDurationMs); + logger.fatal(summaryBuilder.toString(), e); + + // add alert to the Host here + if (e instanceof ConnectionException) { + ConnectionException ce = (ConnectionException) e; + // XXX: in case of Storage Pool issue we are ending up here Review Comment: Nit: leftover `XXX` comment carried over in the moved code. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
