Don't fail to log if we can't redact something (apache#16118)

Rather than dying with an exception, catch it and warn about that,
asking users to report it to us.

Additionally handle the specific case where a file handle/IO object is
logged -- we definitely don't want to iterate over that!

(cherry picked from commit 57bd6fb)
(cherry picked from commit 7603ef6)
(cherry picked from commit 41ae709)
(cherry picked from commit ae0d119)

airflow/utils/log/secrets_masker.py

@@ -16,6 +16,7 @@
 # under the License.
 """Mask sensitive information from logs"""
 import collections
+import io
 import logging
 import re
 from typing import TYPE_CHECKING, Iterable, Optional, Set, TypeVar, Union
@@ -40,6 +41,10 @@
 
     RedactableItem = TypeVar('RedctableItem')
 
+
+log = logging.getLogger(__name__)
+
+
 DEFAULT_SENSITIVE_FIELDS = frozenset(
     {
         'access_token',
@@ -188,24 +193,36 @@ def redact(self, item: "RedactableItem", name: str = None) -> "RedactableItem":
         is redacted.
 
         """
-        if name and should_hide_value_for_key(name):
-            return self._redact_all(item)
-
-        if isinstance(item, dict):
-            return {dict_key: self.redact(subval, dict_key) for dict_key, subval in item.items()}
-        elif isinstance(item, str):
-            if self.replacer:
-                # We can't replace specific values, but the key-based redacting
-                # can still happen, so we can't short-circuit, we need to walk
-                # the strucutre.
-                return self.replacer.sub('***', item)
-            return item
-        elif isinstance(item, (tuple, set)):
-            # Turn set in to tuple!
-            return tuple(self.redact(subval) for subval in item)
-        elif isinstance(item, Iterable):
-            return list(self.redact(subval) for subval in item)
-        else:
+        try:
+            if name and should_hide_value_for_key(name):
+                return self._redact_all(item)
+
+            if isinstance(item, dict):
+                return {dict_key: self.redact(subval, dict_key) for dict_key, subval in item.items()}
+            elif isinstance(item, str):
+                if self.replacer:
+                    # We can't replace specific values, but the key-based redacting
+                    # can still happen, so we can't short-circuit, we need to walk
+                    # the structure.
+                    return self.replacer.sub('***', item)
+                return item
+            elif isinstance(item, (tuple, set)):
+                # Turn set in to tuple!
+                return tuple(self.redact(subval) for subval in item)
+            elif isinstance(item, io.IOBase):
+                return item
+            elif isinstance(item, Iterable):
+                return list(self.redact(subval) for subval in item)
+            else:
+                return item
+        except Exception as e:  # pylint: disable=broad-except
+            log.warning(
+                "Unable to redact %r, please report this via <https://github.com/apache/airflow/issues>. "
+                "Error was: %s: %s",
+                item,
+                type(e).__name__,
+                str(e),
+            )
             return item
 
     # pylint: enable=too-many-return-statements

tests/utils/log/test_secrets_masker.py

@@ -73,6 +73,22 @@ def test_args(self, logger, caplog):
 
         assert caplog.text == "INFO Cannot connect to user:***\n"
 
+    def test_non_redactable(self, logger, caplog):
+        class NonReactable:
+            def __iter__(self):
+                raise RuntimeError("force fail")
+
+            def __repr__(self):
+                return "<NonRedactable>"
+
+        logger.info("Logging %s", NonReactable())
+
+        assert caplog.messages == [
+            "Unable to redact <NonRedactable>, please report this via "
+            + "<https://github.com/apache/airflow/issues>. Error was: RuntimeError: force fail",
+            "Logging <NonRedactable>",
+        ]
+
     def test_extra(self, logger, caplog):
         logger.handlers[0].formatter = ShortExcFormatter("%(levelname)s %(message)s %(conn)s")
         logger.info("Cannot connect", extra={'conn': "user:password"})
@@ -205,6 +221,14 @@ def test_redact(self, patterns, name, value, expected):
 
         assert filt.redact(value, name) == expected
 
+    def test_redact_filehandles(self, caplog):
+        filt = SecretsMasker()
+        with open("/dev/null", "w") as handle:
+            assert filt.redact(handle, None) == handle
+
+        # We shouldn't have logged a warning here
+        assert caplog.messages == []
+
 
 class TestShouldHideValueForKey:
     @pytest.mark.parametrize(