more test fix

Author: dwoz

salt/channel/client.py

@@ -197,8 +197,10 @@ async def _send_with_retry(self, load, tries, timeout):
                 )
                 break
             except Exception as exc:  # pylint: disable=broad-except
-                log.trace("Failed to send msg %r", exc)
-                if _try >= tries:
+                log.trace(
+                    "Failed to send msg attempt=%s tries=%s exc=%r", _try, tries, exc
+                )
+                if tries is not None and tries > 0 and _try >= tries:
                     raise
                 else:
                     _try += 1
@@ -513,6 +515,12 @@ async def connect(self):
                 publish_port = self.auth.creds["publish_port"]
             # TODO: The zeromq transport does not use connect_callback and
             # disconnect_callback.
+            log.debug(
+                "AsyncPubChannel connecting to %s:%s (authenticated=%s)",
+                self.opts.get("master_ip", "127.0.0.1"),
+                publish_port,
+                self.auth.authenticated,
+            )
             await self.transport.connect(
                 publish_port, self.connect_callback, self.disconnect_callback
             )
@@ -598,6 +606,17 @@ async def connect_callback(self, result):
         try:
             # Force re-auth on reconnect since the master
             # may have been restarted
+            try:
+                self.token = self.auth.gen_token(b"salt")
+            except Exception:  # pylint: disable=broad-except
+                log.exception(
+                    "Failed to generate authentication token for publish channel"
+                )
+            log.debug(
+                "AsyncPubChannel connect_callback (reconnected=%s, authenticated=%s)",
+                self._reconnected,
+                self.auth.authenticated,
+            )
             await self.send_id(self.token, self._reconnected)
             self.connected = True
             self.event.fire_event({"master": self.opts["master"]}, "__master_connected")
@@ -643,6 +662,15 @@ def disconnect_callback(self):
         if self._closing:
             return
         self.connected = False
+        if getattr(self, "auth", None) is not None:
+            try:
+                self.auth.invalidate()
+            except Exception:  # pylint: disable=broad-except
+                log.exception("Failed to invalidate authentication after disconnect")
+        log.debug(
+            "AsyncPubChannel disconnect_callback invoked (reconnected=%s)",
+            self._reconnected,
+        )
         self.event.fire_event({"master": self.opts["master"]}, "__master_disconnected")
 
     def _verify_master_signature(self, payload):
@@ -672,6 +700,10 @@ async def _decode_payload(self, payload):
             try:
                 payload["load"] = self.auth.crypticle.loads(payload["load"])
             except salt.crypt.AuthenticationError:
+                log.debug(
+                    "AsyncPubChannel payload decrypt failed, attempting re-auth (authenticated=%s)",
+                    self.auth.authenticated,
+                )
                 reauth = True
             if reauth:
                 try:

salt/crypt.py

@@ -821,6 +821,14 @@ def authenticate(self, callback=None):
         This function will de-dupe all calls here and return a *single* future
         for the sign-in-- whis way callers can all assume there aren't others
         """
+        log.debug(
+            "%s authenticate() invoked (future_exists=%s, future_done=%s)",
+            self,
+            hasattr(self, "_authenticate_future"),
+            getattr(
+                getattr(self, "_authenticate_future", None), "done", lambda: False
+            )(),
+        )
         # if an auth is in flight-- and not done-- just pass that back as the future to wait on
         if (
             hasattr(self, "_authenticate_future")
@@ -856,20 +864,58 @@ async def _authenticate(self):
         :rtype: Crypticle
         :returns: A crypticle used for encryption operations
         """
+        log.debug("%s starting _authenticate()", self)
         acceptance_wait_time = self.opts["acceptance_wait_time"]
         acceptance_wait_time_max = self.opts["acceptance_wait_time_max"]
         if not acceptance_wait_time_max:
             acceptance_wait_time_max = acceptance_wait_time
         creds = None
+        log.debug(
+            "%s auth config timeout=%s tries=%s safemode=%s",
+            self,
+            self.opts.get("auth_timeout"),
+            self.opts.get("auth_tries"),
+            self.opts.get("auth_safemode"),
+        )
 
         with salt.channel.client.AsyncReqChannel.factory(
             self.opts, crypt="clear", io_loop=self._loop_raw
         ) as channel:
             error = None
+            attempt = 0
+            max_attempts = self.opts.get("auth_tries", 0) or 0
             while True:
+                attempt += 1
                 try:
-                    creds = await self.sign_in(channel=channel)
+                    creds = await self.sign_in(channel=channel, tries=1)
+                    log.debug("%s sign_in returned %r", self, creds)
                 except SaltClientError as exc:
+                    message = str(exc)
+                    if any(
+                        pattern in message
+                        for pattern in (
+                            "Attempt to authenticate with the salt master failed",
+                            "-|RETRY|-",
+                        )
+                    ):
+                        log.warning(
+                            "%s sign_in transient failure: %s (will retry)",
+                            self,
+                            message,
+                        )
+                        if acceptance_wait_time:
+                            await asyncio.sleep(acceptance_wait_time)
+                        if acceptance_wait_time < acceptance_wait_time_max:
+                            acceptance_wait_time += acceptance_wait_time
+                            log.debug(
+                                "Authentication wait time is %s",
+                                acceptance_wait_time,
+                            )
+                        if max_attempts > 0 and attempt >= max_attempts:
+                            error = exc
+                            break
+                        continue
+                    log.warning("%s sign_in raised fatal error: %s", self, exc)
                     error = exc
                     break
                 if creds == "retry":
@@ -902,6 +948,13 @@ async def _authenticate(self):
                         log.debug(
                             "Authentication wait time is %s", acceptance_wait_time
                         )
+                    if max_attempts > 0 and attempt >= max_attempts:
+                        error = SaltClientError(
+                            "Authentication retries exhausted for {}".format(
+                                self.opts["id"]
+                            )
+                        )
+                        break
                     continue
                 elif creds == "bad enc algo":
                     log.error(
@@ -929,6 +982,11 @@ async def _authenticate(self):
                     )
                 self._authenticate_future.set_exception(error)
             else:
+                log.debug(
+                    "%s received credentials (publish_port=%s)",
+                    self,
+                    creds.get("publish_port"),
+                )
                 key = self.__key(self.opts)
                 new_aes, changed_aes, changed_session = False, False, False
                 if key not in AsyncAuth.creds_map:
@@ -947,6 +1005,13 @@ async def _authenticate(self):
                     self._crypticle = Crypticle(self.opts, creds["aes"])
                     self._session_crypticle = Crypticle(self.opts, creds["session"])
 
+                log.debug(
+                    "%s authenticated with master %s (publish_port=%s)",
+                    self,
+                    self.opts.get("master"),
+                    creds.get("publish_port"),
+                )
+
                 self._authenticate_future.set_result(
                     True
                 )  # mark the sign-in as complete

salt/metaproxy/deltaproxy.py

@@ -228,6 +228,25 @@ async def _sync_all_task():
         salt.engines.start_engines, self.opts, self.process_manager, proxy=self.proxy
     )
 
+    # At this point the control proxy finished its own initialization, allow
+    # requests targeting the control proxy to flow even if sub proxies are
+    # still spinning up (which can take a while when managing dozens of ids).
+    setup_future = getattr(self, "_setup_future", None)
+    self.ready = True
+    if self.opts.get("return_retry_timer", 0) < 20:
+        new_retry = 20
+        self.opts["return_retry_timer"] = new_retry
+        current_max = self.opts.get("return_retry_timer_max", new_retry)
+        if current_max < new_retry:
+            current_max = new_retry
+        self.opts["return_retry_timer_max"] = max(current_max, new_retry + 10)
+        log.debug(
+            "Control proxy %s adjusted return retry timers to %s-%s seconds",
+            self.opts["id"],
+            self.opts["return_retry_timer"],
+            self.opts["return_retry_timer_max"],
+        )
+
     proxy_init_func_name = f"{fq_proxyname}.init"
     proxy_shutdown_func_name = f"{fq_proxyname}.shutdown"
     if (
@@ -425,16 +444,22 @@ async def _sync_grains_task():
     if self.opts["proxy"].get("parallel_startup"):
         log.warning("Initiating parallel startup for proxies")
         waitfor = []
+        max_concurrency = max(1, self.opts["proxy"].get("startup_concurrency", 4))
+        semaphore = asyncio.Semaphore(max_concurrency)
+
         for _id in self.opts["proxy"].get("ids", []):
-            waitfor.append(
-                subproxy_post_master_init(
-                    _id,
-                    uid,
-                    self.opts,
-                    self.proxy,
-                    self.utils,
-                )
-            )
+
+            async def _initialise_subproxy(sub_id, *, _sem=semaphore):
+                async with _sem:
+                    return await subproxy_post_master_init(
+                        sub_id,
+                        uid,
+                        self.opts,
+                        self.proxy,
+                        self.utils,
+                    )
+
+            waitfor.append(_initialise_subproxy(_id))
 
         try:
             results = await asyncio.gather(*waitfor)
@@ -504,28 +529,41 @@ async def subproxy_post_master_init(minion_id, uid, opts, main_proxy, main_utils
     proxy_grains = {}
     proxy_pillar = {}
 
-    proxyopts = opts.copy()
-    proxyopts["id"] = minion_id
+    loop = asyncio.get_running_loop()
 
-    proxyopts = salt.config.proxy_config(
-        opts["conf_file"], defaults=proxyopts, minion_id=minion_id
-    )
-    proxyopts.update({"id": minion_id, "proxyid": minion_id, "subproxy": True})
+    def _load_proxyopts_and_grains():
+        proxyopts_local = opts.copy()
+        proxyopts_local["id"] = minion_id
 
-    proxy_context = {"proxy_id": minion_id}
+        proxyopts_local = salt.config.proxy_config(
+            opts["conf_file"], defaults=proxyopts_local, minion_id=minion_id
+        )
+        proxyopts_local.update(
+            {"id": minion_id, "proxyid": minion_id, "subproxy": True}
+        )
 
-    # We need grains first to be able to load pillar, which is where we keep the proxy
-    # configurations
-    proxy_grains = salt.loader.grains(
-        proxyopts, proxy=main_proxy, context=proxy_context
+        proxy_context_local = {"proxy_id": minion_id}
+
+        proxy_grains_local = salt.loader.grains(
+            proxyopts_local, proxy=main_proxy, context=proxy_context_local
+        )
+        return proxyopts_local, proxy_context_local, proxy_grains_local
+
+    proxyopts, proxy_context, proxy_grains = await loop.run_in_executor(
+        None, _load_proxyopts_and_grains
     )
-    proxy_pillar = await salt.pillar.get_async_pillar(
-        proxyopts,
-        proxy_grains,
-        minion_id,
-        saltenv=proxyopts["saltenv"],
-        pillarenv=proxyopts.get("pillarenv"),
-    ).compile_pillar()
+
+    def _compile_proxy_pillar():
+        pillar = salt.pillar.get_pillar(
+            proxyopts,
+            proxy_grains,
+            minion_id,
+            saltenv=proxyopts["saltenv"],
+            pillarenv=proxyopts.get("pillarenv"),
+        )
+        return pillar.compile_pillar()
+
+    proxy_pillar = await loop.run_in_executor(None, _compile_proxy_pillar)
 
     proxyopts["proxy"] = proxy_pillar.get("proxy", {})
     if not proxyopts["proxy"]:
@@ -534,8 +572,6 @@ async def subproxy_post_master_init(minion_id, uid, opts, main_proxy, main_utils
         )
         return {"proxy_minion": None, "proxy_opts": {}}
 
-    loop = asyncio.get_running_loop()
-
     def _finish_subproxy_setup():
         proxyopts["proxy"].pop("ids", None)
 
@@ -1049,12 +1085,18 @@ async def handle_payload(self, payload):
     Verify the publication and then pass
     the payload along to _handle_decoded_payload.
     """
+    log.debug(
+        "Control proxy %s received payload enc=%s keys=%s",
+        self.opts["id"],
+        payload.get("enc") if payload else None,
+        sorted(payload["load"].keys()) if payload and "load" in payload else None,
+    )
     setup_future = getattr(self, "_setup_future", None)
     if setup_future is not None and not setup_future.done():
-        log.warning("Control proxy waiting for setup to finish before handling payload")
+        log.debug("Control proxy waiting for setup to finish before handling payload")
         await asyncio.shield(setup_future)
     else:
-        log.warning("Control proxy setup already complete; handling payload")
+        log.debug("Control proxy setup already complete; handling payload")
     if payload is not None and payload["enc"] == "aes":
         # First handle payload for the "control" proxy
         if self._target_load(payload["load"]):