<fix>[cloudbus]: support send confirm

1. send in CloudBus means submit a message send task.
function return do not means sending completion.
so there is no way to set a limition in sender.
add a FutureCompletion to confirm message send.
provide a way to limited send.

2. fix mn heartbeat thread interrupt handle

3. refactor duplicate code for getChainInfo, fix synchronized target.

This patch is cherry-picked from ZSTAC-71213
(commit: af58834)

Resolves: ZSV-11510

Change-Id: I686361627566737a7775706263796b736967666c

Author: Zhang Wenhao

conf/errorCodes/sys.xml

@@ -105,5 +105,10 @@
         <id>1090</id>
         <description>Multiple reasons</description>
     </code>
+
+    <code>
+        <id>2111</id>
+        <description>CloudBus message sending related error</description>
+    </code>
 </error>
 

core/src/main/java/org/zstack/core/Platform.java

@@ -50,6 +50,7 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Modifier;
+import java.net.Inet4Address;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
 import java.net.SocketException;
@@ -817,9 +818,14 @@ private static String getManagementServerIpInternal() {
             for (NetworkInterface iface : Collections.list(nets)) {
                 String name = iface.getName();
                 if (defaultLine.contains(name)) {
-                    InetAddress ia = iface.getInetAddresses().nextElement();
-                    ip = ia.getHostAddress();
-                    break;
+                    for (InetAddress ia : Collections.list(iface.getInetAddresses())) {
+                        ip = ia.getHostAddress();
+                        if (ia instanceof Inet4Address) {
+                            // we prefer IPv4 address
+                            ip = ia.getHostAddress();
+                            break;
+                        }
+                    }
                 }
             }
         } catch (SocketException e) {

core/src/main/java/org/zstack/core/cloudbus/CloudBus.java

@@ -3,6 +3,7 @@
 import org.springframework.http.HttpEntity;
 import org.zstack.header.Component;
 import org.zstack.header.Service;
+import org.zstack.header.core.FutureCompletion;
 import org.zstack.header.errorcode.ErrorCode;
 import org.zstack.header.exception.CloudConfigureFailException;
 import org.zstack.header.message.*;
@@ -12,14 +13,18 @@
 import java.util.function.Consumer;
 
 public interface CloudBus extends Component {
-    void send(Message msg);
-    
+    /**
+     * submit a message sending task into the queue.
+     * @return {@link FutureCompletion} which can be used to check whether the message was successfully sent or not.
+     */
+    FutureCompletion send(Message msg);
+
     <T extends Message> void send(List<T> msgs);
 
     @Deprecated
     void send(APIMessage msg, Consumer<APIEvent> consumer);
 
-    void send(NeedReplyMessage msg, CloudBusCallBack callback);
+    FutureCompletion send(NeedReplyMessage msg, CloudBusCallBack callback);
 
     @Deprecated
     void send(List<? extends NeedReplyMessage> msgs, CloudBusListCallBack callBack);

core/src/main/java/org/zstack/core/cloudbus/CloudBusImpl2.java

@@ -19,6 +19,7 @@
 import org.zstack.core.timeout.ApiTimeoutManager;
 import org.zstack.header.Constants;
 import org.zstack.header.Service;
+import org.zstack.header.core.FutureCompletion;
 import org.zstack.header.log.NoLogging;
 import org.zstack.header.apimediator.APIIsReadyToGoMsg;
 import org.zstack.header.apimediator.APIIsReadyToGoReply;
@@ -106,6 +107,11 @@ public class CloudBusImpl2 implements CloudBus, CloudBusIN, ManagementNodeChange
     private final String AMQP_PROPERTY_HEADER__COMPRESSED = "compressed";
 
     private String SERVICE_ID = makeLocalServiceId("cloudbus");
+    public static final FutureCompletion SEND_CONFIRMED = new FutureCompletion(null);
+
+    static {
+        SEND_CONFIRMED.success();
+    }
 
     public void setDEFAULT_MESSAGE_TIMEOUT(long timeout) {
         this.DEFAULT_MESSAGE_TIMEOUT = timeout;
@@ -1323,8 +1329,9 @@ private void send(Message msg, Boolean noNeedReply) {
     }
 
     @Override
-    public void send(Message msg) {
+    public FutureCompletion send(Message msg) {
         send(msg, true);
+        return SEND_CONFIRMED;
     }
 
     @Override
@@ -1354,7 +1361,7 @@ private void evaluateMessageTimeout(NeedReplyMessage msg) {
     }
 
     @Override
-    public void send(final NeedReplyMessage msg, final CloudBusCallBack callback) {
+    public FutureCompletion send(final NeedReplyMessage msg, final CloudBusCallBack callback) {
         evaluateMessageTimeout(msg);
 
         Envelope e = new Envelope() {
@@ -1405,6 +1412,7 @@ List<Message> getRequests() {
         envelopes.put(msg.getId(), e);
 
         send(msg, false);
+        return SEND_CONFIRMED;
     }
 
     private MessageReply createTimeoutReply(NeedReplyMessage m) {

core/src/main/java/org/zstack/core/cloudbus/CloudBusImpl3.java

@@ -49,6 +49,7 @@
 
 import static org.zstack.core.Platform.*;
 import static org.zstack.core.cloudbus.CloudBusGlobalProperty.SYNC_CALL_TIMEOUT;
+import static org.zstack.header.errorcode.SysErrors.CLOUD_BUS_SEND_ERROR;
 import static org.zstack.utils.BeanUtils.getProperty;
 import static org.zstack.utils.BeanUtils.setProperty;
 
@@ -101,6 +102,7 @@ public class CloudBusImpl3 implements CloudBus, CloudBusIN {
     private final static TimeoutRestTemplate http = RESTFacade.createRestTemplate(CoreGlobalProperty.REST_FACADE_READ_TIMEOUT, CoreGlobalProperty.REST_FACADE_CONNECT_TIMEOUT);
 
     public static final String HTTP_BASE_URL = "/cloudbus";
+    public static final FutureCompletion SEND_CONFIRMED = new FutureCompletion(null);
 
     {
         if (CloudBusGlobalProperty.MESSAGE_LOG != null) {
@@ -114,6 +116,8 @@ public class CloudBusImpl3 implements CloudBus, CloudBusIN {
             CloudBusGlobalProperty.HTTP_CONTEXT_PATH = "";
             CloudBusGlobalProperty.HTTP_PORT = 8989;
         }
+
+        SEND_CONFIRMED.success();
     }
 
     public static String getManagementNodeUUIDFromServiceID(String serviceID) {
@@ -125,6 +129,12 @@ public static String getManagementNodeUUIDFromServiceID(String serviceID) {
         return ss[0];
     }
 
+    private FutureCompletion sendFail(ErrorCode errorCode) {
+        FutureCompletion c = new FutureCompletion(null);
+        c.fail(errorCode);
+        return c;
+    }
+
     private abstract class Envelope {
         long startTime;
 
@@ -250,8 +260,8 @@ public void deActiveService(String id) {
     }
 
     @Override
-    public void send(Message msg) {
-        send(msg, true);
+    public FutureCompletion send(Message msg) {
+        return send(msg, true);
     }
 
     @Override
@@ -291,11 +301,11 @@ private MessageReply createTimeoutReply(NeedReplyMessage m) {
     }
 
     @Override
-    public void send(NeedReplyMessage msg, CloudBusCallBack callback) {
+    public FutureCompletion send(NeedReplyMessage msg, CloudBusCallBack callback) {
         evaluateMessageTimeout(msg);
         if (msg.getTimeout() <= 1) {
             callback.run(createTimeoutReply(msg));
-            return;
+            return SEND_CONFIRMED;
         }
 
         Envelope e = new Envelope() {
@@ -346,7 +356,7 @@ public void timeout() {
 
         envelopes.put(msg.getId(), e);
         msgExts.forEach(m -> m.afterAddEnvelopes(msg.getId()));
-        send(msg, false);
+        return send(msg, false);
     }
 
     @Override
@@ -454,7 +464,7 @@ public void reply(Message request, MessageReply reply) {
                 callReplyPreSendingExtensions(reply, (NeedReplyMessage) request);
             } catch (Exception e) {
                 logger.error("failed to call pre-sending reply extension:", e);
-                reply.setError(operr(e.getMessage()));
+                reply.setError(err(CLOUD_BUS_SEND_ERROR, e.getMessage()));
             }
         }
 
@@ -526,29 +536,36 @@ public MessageSender(Message msg) {
             localSend = !CloudBusGlobalProperty.HTTP_ALWAYS && managementNodeId.equals(Platform.getManagementServerId());
         }
 
-        void send() {
+        FutureCompletion send() {
             try {
-                doSend();
+                return doSend();
             } catch (Throwable th) {
-                replyErrorIfNeeded(operr(th.getMessage()));
+                ErrorCode err = err(CLOUD_BUS_SEND_ERROR, th.getMessage());
+                replyErrorIfNeeded(err);
+
+                FutureCompletion c = new FutureCompletion(null);
+                c.fail(err);
+                return c;
             }
         }
 
-        private void doSend() {
+        private FutureCompletion doSend() {
             if (msg instanceof Event) {
                 eventSend();
-                return;
+                return SEND_CONFIRMED;
             }
 
             if (localSend) {
                 localSend();
+                return SEND_CONFIRMED;
             } else {
-                httpSend();
+                return httpSend();
             }
         }
 
-        private void httpSendInQueue(String ip) {
-            thdf.chainSubmit(new ChainTask(null) {
+        private FutureCompletion httpSendInQueue(String ip) {
+            FutureCompletion sendCompletion = new FutureCompletion(null);
+            thdf.chainSubmit(new ChainTask(sendCompletion) {
                 @Override
                 public String getSyncSignature() {
                     return "http-send-in-queue";
@@ -557,6 +574,7 @@ public String getSyncSignature() {
                 @Override
                 public void run(SyncTaskChain chain) {
                     httpSend(ip);
+                    sendCompletion.success();
                     chain.next();
                 }
 
@@ -570,25 +588,29 @@ public String getName() {
                     return getSyncSignature();
                 }
             });
+            return sendCompletion;
         }
 
-        private void httpSend() {
+        private FutureCompletion httpSend() {
             buildSchema(msg);
+            String ip;
             try {
-                String ip = destMaker.getNodeInfo(managementNodeId).getNodeIP();
-                httpSendInQueue(ip);
+                ip = destMaker.getNodeInfo(managementNodeId).getNodeIP();
             } catch (ManagementNodeNotFoundException e) {
-                if (msg instanceof MessageReply) {
-                    if (!deadMessageManager.handleManagementNodeNotFoundError(managementNodeId, msg, () -> {
-                        String ip = destMaker.getNodeInfo(managementNodeId).getNodeIP();
-                        httpSendInQueue(ip);
-                    })) {
-                        throw e;
-                    }
+                boolean errorHandled = msg instanceof MessageReply &&
+                        deadMessageManager.handleManagementNodeNotFoundError(managementNodeId, msg, () -> {
+                            String otherIp = destMaker.getNodeInfo(managementNodeId).getNodeIP();
+                            logger.warn(String.format("resend the message[id:%s] to node[ip:%s]", msg.getId(), otherIp));
+                            httpSendInQueue(otherIp);
+                        });
+                if (errorHandled) {
+                    return SEND_CONFIRMED;
                 } else {
                     throw e;
                 }
             }
+
+            return httpSendInQueue(ip);
         }
 
         private void httpSend(String ip) {
@@ -612,12 +634,12 @@ protected ResponseEntity<String> call() {
                 }.run();
 
                 if (!rsp.getStatusCode().is2xxSuccessful()) {
-                    replyErrorIfNeeded(operr("HTTP ERROR, status code: %s, body: %s", rsp.getStatusCode(), rsp.getBody()));
+                    replyErrorIfNeeded(err(CLOUD_BUS_SEND_ERROR, "HTTP ERROR, status code: %s, body: %s", rsp.getStatusCode(), rsp.getBody()));
                 }
             } catch (OperationFailureException e) {
                 replyErrorIfNeeded(e.getErrorCode());
             } catch (Throwable e) {
-                replyErrorIfNeeded(operr(e.getMessage()));
+                replyErrorIfNeeded(err(CLOUD_BUS_SEND_ERROR, e.getMessage()));
             }
         }
 
@@ -1194,7 +1216,7 @@ private void evalThreadContextToMessage(Message msg) {
         }
     }
 
-    private void doSendAndCallExtensions(Message msg) {
+    private FutureCompletion doSendAndCallExtensions(Message msg) {
         // for unit test finding invocation chain
         MessageCommandRecorder.record(msg.getClass());
 
@@ -1209,20 +1231,20 @@ private void doSendAndCallExtensions(Message msg) {
             interceptor.beforeSendMessage(msg);
         }
 
-        doSend(msg);
+        return doSend(msg);
     }
 
-    private void doSend(Message msg) {
+    private FutureCompletion doSend(Message msg) {
         evalThreadContextToMessage(msg);
 
         if (logger.isTraceEnabled() && islogMessage(msg)) {
             logger.trace(String.format("[msg send]: %s", dumpMessage(msg)));
         }
 
-        new MessageSender(msg).send();
+        return new MessageSender(msg).send();
     }
 
-    private void send(Message msg, Boolean noNeedReply) {
+    private FutureCompletion send(Message msg, Boolean noNeedReply) {
         if (msg.getServiceId() == null) {
             throw new IllegalArgumentException(String.format("service id cannot be null: %s", msg.getClass().getName()));
         }
@@ -1238,7 +1260,7 @@ private void send(Message msg, Boolean noNeedReply) {
             msg.putHeaderEntry(NO_NEED_REPLY_MSG, noNeedReply.toString());
         }
 
-        doSendAndCallExtensions(msg);
+        return doSendAndCallExtensions(msg);
     }
 
     private void restoreFromSchema(Message msg, Map raw) throws ClassNotFoundException {