diff --git a/sentry_sdk/_log_batcher.py b/sentry_sdk/_log_batcher.py
index b139b56da2..6fa501ab7b 100644
--- a/sentry_sdk/_log_batcher.py
+++ b/sentry_sdk/_log_batcher.py
@@ -6,7 +6,27 @@
 if TYPE_CHECKING:
     from typing import Any
 
-    from sentry_sdk._types import Log
+    from sentry_sdk._types import Log, SerializedAttributeValue
+
+
+def _sanitize_surrogates(s: str) -> str:
+    try:
+        s.encode("utf-8")
+        return s
+    except UnicodeEncodeError:
+        return s.encode("utf-16", "surrogatepass").decode("utf-16", "replace")
+
+
+def _serialize_and_sanitize_attribute(val: "Any") -> "SerializedAttributeValue":
+    serialized = serialize_attribute(val)
+    s_val = serialized["value"]
+    if isinstance(s_val, str):
+        serialized["value"] = _sanitize_surrogates(s_val)
+    elif isinstance(s_val, list):
+        serialized["value"] = [
+            _sanitize_surrogates(x) if isinstance(x, str) else x for x in s_val
+        ]
+    return serialized
 
 
 class LogBatcher(Batcher["Log"]):
@@ -28,9 +48,10 @@ def _to_transport_format(item: "Log") -> "Any":
         res = {
             "timestamp": int(item["time_unix_nano"]) / 1.0e9,
             "level": str(item["severity_text"]),
-            "body": str(item["body"]),
+            "body": _sanitize_surrogates(str(item["body"])),
             "attributes": {
-                k: serialize_attribute(v) for (k, v) in item["attributes"].items()
+                _sanitize_surrogates(str(k)): _serialize_and_sanitize_attribute(v)
+                for (k, v) in item["attributes"].items()
             },
         }
 
diff --git a/tests/test_logs.py b/tests/test_logs.py
index 90b5236f2a..4743f3d65d 100644
--- a/tests/test_logs.py
+++ b/tests/test_logs.py
@@ -1010,3 +1010,56 @@ def _flush(self):
     batcher._flush_loop()
 
     assert calls == [1]
+
+
+@minimum_python_37
+def test_logs_lone_surrogates(sentry_init, capture_items):
+    import logging
+
+    from sentry_sdk._log_batcher import LogBatcher
+    from sentry_sdk.utils import json_dumps
+
+    sentry_init(enable_logs=True)
+    items = capture_items("log")
+
+    # String with a lone surrogate, e.g. from os.fsdecode on non-UTF-8 bytes
+    name_with_surrogate = b"report-\xff.csv".decode("utf-8", "surrogateescape")
+    attr_key_with_surrogate = b"key-\xff".decode("utf-8", "surrogateescape")
+
+    sentry_sdk.logger.info("opened {name}", name=name_with_surrogate)
+    logging.getLogger("app").warning("could not read %s", name_with_surrogate)
+
+    get_client().flush()
+
+    assert len(items) == 2
+
+    for item in items:
+        encoded = json_dumps(item.payload)
+        assert b"\\udc" not in encoded
+        assert b"\\ufffd" in encoded or "\ufffd".encode() in encoded
+
+    # Direct unit test on LogBatcher._to_transport_format
+    raw_log = {
+        "severity_text": "info",
+        "severity_number": 9,
+        "time_unix_nano": 1700000000000000000,
+        "body": f"file: {name_with_surrogate}",
+        "attributes": {
+            attr_key_with_surrogate: name_with_surrogate,
+            "paths": [name_with_surrogate, "normal.csv"],
+        },
+    }
+
+    res = LogBatcher._to_transport_format(raw_log)
+    assert "\ufffd" in res["body"]
+    assert "\udcff" not in res["body"]
+    assert "key-\ufffd" in res["attributes"]
+    assert res["attributes"]["key-\ufffd"]["value"] == "report-\ufffd.csv"
+    assert res["attributes"]["paths"]["value"] == [
+        "report-\ufffd.csv",
+        "normal.csv",
+    ]
+
+    encoded_direct = json_dumps(res)
+    assert b"\\udc" not in encoded_direct
+    assert b"\\ufffd" in encoded_direct