diff --git a/python_modules/dagster/dagster/_core/definitions/instigation_logger.py b/python_modules/dagster/dagster/_core/definitions/instigation_logger.py
index 0d4914ba48ef7..6a03c20755324 100644
--- a/python_modules/dagster/dagster/_core/definitions/instigation_logger.py
+++ b/python_modules/dagster/dagster/_core/definitions/instigation_logger.py
@@ -7,8 +7,6 @@
 from contextlib import ExitStack
 from typing import IO, TYPE_CHECKING, Any, Optional
 
-from dagster_shared import seven
-
 from dagster._core.log_manager import LOG_RECORD_METADATA_ATTR
 from dagster._core.storage.compute_log_manager import ComputeIOType, ComputeLogManager
 from dagster._core.utils import coerce_valid_log_level
@@ -73,10 +71,12 @@ def emit(self, record: logging.LogRecord):
             record_dict["exc_info"] = "".join(traceback.format_exception(*exc_info))
 
         try:
-            self._write_stream.write(seven.json.dumps(record_dict) + "\n")
+            self._write_stream.write(json.dumps(record_dict, default=str, sort_keys=True) + "\n")
         except Exception:
             sys.stderr.write(
-                f"Exception writing to logger event stream: {serializable_error_info_from_exc_info(sys.exc_info())}\n"
+                f"Exception writing to logger event stream: {serializable_error_info_from_exc_info(sys.exc_info())}."
+                f" The originating log call was made at: {record.pathname}:{record.funcName} (line {record.lineno})."
+                "\n"
             )
 
 
@@ -179,7 +179,7 @@ def get_instigation_log_records(
             continue
 
         try:
-            records.append(seven.json.loads(line))
+            records.append(json.loads(line, strict=False))
         except json.JSONDecodeError:
             continue
     return records
diff --git a/python_modules/dagster/dagster_tests/storage_tests/test_instigation_logger.py b/python_modules/dagster/dagster_tests/storage_tests/test_instigation_logger.py
index 4fe1f34227d53..2cea99922ef69 100644
--- a/python_modules/dagster/dagster_tests/storage_tests/test_instigation_logger.py
+++ b/python_modules/dagster/dagster_tests/storage_tests/test_instigation_logger.py
@@ -2,7 +2,10 @@
 from unittest import mock
 
 import dagster as dg
-from dagster._core.definitions.instigation_logger import InstigationLogger
+from dagster._core.definitions.instigation_logger import (
+    InstigationLogger,
+    get_instigation_log_records,
+)
 from dagster._core.storage.noop_compute_log_manager import NoOpComputeLogManager
 
 
@@ -74,3 +77,43 @@ def test_instigation_logger_log_failure(capsys):
             captured.err.count("Exception closing logger write stream: Exception: OOPS ON EXIT")
             == 1
         )
+
+
+def test_instigation_logger_logs_weird_record_ok(capsys):
+    class NonJsonSerializable:
+        """A class that json.dumps doesn't know what to do with (by default)."""
+
+        def __str__(self):
+            return "non-jsonable-object"
+
+    log_key = ["test-repo", "test-instigator", "123"]
+
+    with dg.instance_for_test() as instance:
+        with InstigationLogger(
+            log_key=log_key,
+            instance=instance,
+            repository_name="test-repo",
+            instigator_name="test-instigator",
+        ) as logger:
+            logger.info("log without extras")
+            logger.info("log with extras", extra={"non_json": NonJsonSerializable()})
+
+            assert logger.has_captured_logs()
+
+        captured = capsys.readouterr()
+
+        assert "Exception initializing logger write stream" not in captured.err
+        assert "Exception writing to logger event stream" not in captured.err
+        assert "Exception closing logger write stream" not in captured.err
+        assert "NonJsonSerializable is not JSON serializable" not in captured.err
+
+        records = get_instigation_log_records(instance, log_key)
+
+        assert len(records) >= 2
+        assert "test-repo - test-instigator - log with extras" in {
+            record.get("msg") for record in records if record.get("msg")
+        }
+        matching_records = [record for record in records if record.get("non_json")]
+        assert matching_records and matching_records[0]["non_json"] == "non-jsonable-object"
+
+        instance.compute_log_manager.delete_logs(log_key=log_key)