diff --git a/sentry_sdk/_experimental_logger.py b/sentry_sdk/_experimental_logger.py
index 1f3cd5e443..d28ff69483 100644
--- a/sentry_sdk/_experimental_logger.py
+++ b/sentry_sdk/_experimental_logger.py
@@ -1,5 +1,6 @@
 # NOTE: this is the logger sentry exposes to users, not some generic logger.
 import functools
+import time
 from typing import Any
 
 from sentry_sdk import get_client, get_current_scope
@@ -9,7 +10,27 @@ def _capture_log(severity_text, severity_number, template, **kwargs):
     # type: (str, int, str, **Any) -> None
     client = get_client()
     scope = get_current_scope()
-    client.capture_log(scope, severity_text, severity_number, template, **kwargs)
+
+    attrs = {
+        "sentry.message.template": template,
+    }  # type: dict[str, str | bool | float | int]
+    if "attributes" in kwargs:
+        attrs.update(kwargs.pop("attributes"))
+    for k, v in kwargs.items():
+        attrs[f"sentry.message.parameters.{k}"] = v
+
+    # noinspection PyProtectedMember
+    client._capture_experimental_log(
+        scope,
+        {
+            "severity_text": severity_text,
+            "severity_number": severity_number,
+            "attributes": attrs,
+            "body": template.format(**kwargs),
+            "time_unix_nano": time.time_ns(),
+            "trace_id": None,
+        },
+    )
 
 
 trace = functools.partial(_capture_log, "trace", 1)
diff --git a/sentry_sdk/client.py b/sentry_sdk/client.py
index 0f97394561..df6764a508 100644
--- a/sentry_sdk/client.py
+++ b/sentry_sdk/client.py
@@ -1,6 +1,5 @@
 import json
 import os
-import time
 import uuid
 import random
 import socket
@@ -210,8 +209,8 @@ def capture_event(self, *args, **kwargs):
         # type: (*Any, **Any) -> Optional[str]
         return None
 
-    def capture_log(self, scope, severity_text, severity_number, template, **kwargs):
-        # type: (Scope, str, int, str, **Any) -> None
+    def _capture_experimental_log(self, scope, log):
+        # type: (Scope, Log) -> None
         pass
 
     def capture_session(self, *args, **kwargs):
@@ -863,47 +862,36 @@ def capture_event(
 
         return return_value
 
-    def capture_log(self, scope, severity_text, severity_number, template, **kwargs):
-        # type: (Scope, str, int, str, **Any) -> None
+    def _capture_experimental_log(self, current_scope, log):
+        # type: (Scope, Log) -> None
         logs_enabled = self.options["_experiments"].get("enable_sentry_logs", False)
         if not logs_enabled:
             return
+        isolation_scope = current_scope.get_isolation_scope()
 
         headers = {
             "sent_at": format_timestamp(datetime.now(timezone.utc)),
         }  # type: dict[str, object]
 
-        attrs = {
-            "sentry.message.template": template,
-        }  # type: dict[str, str | bool | float | int]
-
-        kwargs_attributes = kwargs.get("attributes")
-        if kwargs_attributes is not None:
-            attrs.update(kwargs_attributes)
-
         environment = self.options.get("environment")
-        if environment is not None:
-            attrs["sentry.environment"] = environment
+        if environment is not None and "sentry.environment" not in log["attributes"]:
+            log["attributes"]["sentry.environment"] = environment
 
         release = self.options.get("release")
-        if release is not None:
-            attrs["sentry.release"] = release
+        if release is not None and "sentry.release" not in log["attributes"]:
+            log["attributes"]["sentry.release"] = release
 
-        span = scope.span
-        if span is not None:
-            attrs["sentry.trace.parent_span_id"] = span.span_id
+        span = current_scope.span
+        if span is not None and "sentry.trace.parent_span_id" not in log["attributes"]:
+            log["attributes"]["sentry.trace.parent_span_id"] = span.span_id
 
-        for k, v in kwargs.items():
-            attrs[f"sentry.message.parameters.{k}"] = v
-
-        log = {
-            "severity_text": severity_text,
-            "severity_number": severity_number,
-            "body": template.format(**kwargs),
-            "attributes": attrs,
-            "time_unix_nano": time.time_ns(),
-            "trace_id": None,
-        }  # type: Log
+        if log.get("trace_id") is None:
+            transaction = current_scope.transaction
+            propagation_context = isolation_scope.get_active_propagation_context()
+            if transaction is not None:
+                log["trace_id"] = transaction.trace_id
+            elif propagation_context is not None:
+                log["trace_id"] = propagation_context.trace_id
 
         # If debug is enabled, log the log to the console
         debug = self.options.get("debug", False)
@@ -917,15 +905,10 @@ def capture_log(self, scope, severity_text, severity_number, template, **kwargs)
                 "fatal": logging.CRITICAL,
             }
             logger.log(
-                severity_text_to_logging_level.get(severity_text, logging.DEBUG),
+                severity_text_to_logging_level.get(log["severity_text"], logging.DEBUG),
                 f'[Sentry Logs] {log["body"]}',
             )
 
-        propagation_context = scope.get_active_propagation_context()
-        if propagation_context is not None:
-            headers["trace_id"] = propagation_context.trace_id
-            log["trace_id"] = propagation_context.trace_id
-
         envelope = Envelope(headers=headers)
 
         before_emit_log = self.options["_experiments"].get("before_emit_log")
diff --git a/sentry_sdk/consts.py b/sentry_sdk/consts.py
index f9317242cd..e4f156256a 100644
--- a/sentry_sdk/consts.py
+++ b/sentry_sdk/consts.py
@@ -78,6 +78,7 @@ class CompressionAlgo(Enum):
                 Callable[[str, MetricValue, MeasurementUnit, MetricTags], bool]
             ],
             "metric_code_locations": Optional[bool],
+            "enable_sentry_logs": Optional[bool],
         },
         total=False,
     )
diff --git a/sentry_sdk/integrations/logging.py b/sentry_sdk/integrations/logging.py
index 3777381b83..2114f4867a 100644
--- a/sentry_sdk/integrations/logging.py
+++ b/sentry_sdk/integrations/logging.py
@@ -1,8 +1,10 @@
+import json
 import logging
 from datetime import datetime, timezone
 from fnmatch import fnmatch
 
 import sentry_sdk
+from sentry_sdk.client import BaseClient
 from sentry_sdk.utils import (
     to_string,
     event_from_exception,
@@ -11,7 +13,7 @@
 )
 from sentry_sdk.integrations import Integration
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Tuple
 
 if TYPE_CHECKING:
     from collections.abc import MutableMapping
@@ -61,14 +63,23 @@ def ignore_logger(
 class LoggingIntegration(Integration):
     identifier = "logging"
 
-    def __init__(self, level=DEFAULT_LEVEL, event_level=DEFAULT_EVENT_LEVEL):
-        # type: (Optional[int], Optional[int]) -> None
+    def __init__(
+        self,
+        level=DEFAULT_LEVEL,
+        event_level=DEFAULT_EVENT_LEVEL,
+        sentry_logs_level=DEFAULT_LEVEL,
+    ):
+        # type: (Optional[int], Optional[int], Optional[int]) -> None
         self._handler = None
         self._breadcrumb_handler = None
+        self._sentry_logs_handler = None
 
         if level is not None:
             self._breadcrumb_handler = BreadcrumbHandler(level=level)
 
+        if sentry_logs_level is not None:
+            self._sentry_logs_handler = SentryLogsHandler(level=sentry_logs_level)
+
         if event_level is not None:
             self._handler = EventHandler(level=event_level)
 
@@ -83,6 +94,12 @@ def _handle_record(self, record):
         ):
             self._breadcrumb_handler.handle(record)
 
+        if (
+            self._sentry_logs_handler is not None
+            and record.levelno >= self._sentry_logs_handler.level
+        ):
+            self._sentry_logs_handler.handle(record)
+
     @staticmethod
     def setup_once():
         # type: () -> None
@@ -296,3 +313,90 @@ def _breadcrumb_from_record(self, record):
             "timestamp": datetime.fromtimestamp(record.created, timezone.utc),
             "data": self._extra_from_record(record),
         }
+
+
+def _python_level_to_otel(record_level):
+    # type: (int) -> Tuple[int, str]
+    for py_level, otel_severity_number, otel_severity_text in [
+        (50, 21, "fatal"),
+        (40, 17, "error"),
+        (30, 13, "warn"),
+        (20, 9, "info"),
+        (10, 5, "debug"),
+        (5, 1, "trace"),
+    ]:
+        if record_level >= py_level:
+            return otel_severity_number, otel_severity_text
+    return 0, "default"
+
+
+class SentryLogsHandler(_BaseHandler):
+    """
+    A logging handler that records Sentry logs for each Python log record.
+
+    Note that you do not have to use this class if the logging integration is enabled, which it is by default.
+    """
+
+    def emit(self, record):
+        # type: (LogRecord) -> Any
+        with capture_internal_exceptions():
+            self.format(record)
+            if not self._can_record(record):
+                return
+
+            client = sentry_sdk.get_client()
+            if not client.is_active():
+                return
+
+            if not client.options["_experiments"].get("enable_sentry_logs", False):
+                return
+
+            SentryLogsHandler._capture_log_from_record(client, record)
+
+    @staticmethod
+    def _capture_log_from_record(client, record):
+        # type: (BaseClient, LogRecord) -> None
+        scope = sentry_sdk.get_current_scope()
+        otel_severity_number, otel_severity_text = _python_level_to_otel(record.levelno)
+        attrs = {
+            "sentry.message.template": (
+                record.msg if isinstance(record.msg, str) else json.dumps(record.msg)
+            ),
+        }  # type: dict[str, str | bool | float | int]
+        if record.args is not None:
+            if isinstance(record.args, tuple):
+                for i, arg in enumerate(record.args):
+                    attrs[f"sentry.message.parameters.{i}"] = (
+                        arg if isinstance(arg, str) else json.dumps(arg)
+                    )
+        if record.lineno:
+            attrs["code.line.number"] = record.lineno
+        if record.pathname:
+            attrs["code.file.path"] = record.pathname
+        if record.funcName:
+            attrs["code.function.name"] = record.funcName
+
+        if record.thread:
+            attrs["thread.id"] = record.thread
+        if record.threadName:
+            attrs["thread.name"] = record.threadName
+
+        if record.process:
+            attrs["process.pid"] = record.process
+        if record.processName:
+            attrs["process.executable.name"] = record.processName
+        if record.name:
+            attrs["logger.name"] = record.name
+
+        # noinspection PyProtectedMember
+        client._capture_experimental_log(
+            scope,
+            {
+                "severity_text": otel_severity_text,
+                "severity_number": otel_severity_number,
+                "body": record.message,
+                "attributes": attrs,
+                "time_unix_nano": int(record.created * 1e9),
+                "trace_id": None,
+            },
+        )
diff --git a/tests/test_logs.py b/tests/test_logs.py
index 173a4028d6..9527fb9807 100644
--- a/tests/test_logs.py
+++ b/tests/test_logs.py
@@ -1,19 +1,28 @@
+import logging
 import sys
+from typing import List, Any
 from unittest import mock
 import pytest
 
 import sentry_sdk
 from sentry_sdk import _experimental_logger as sentry_logger
-
+from sentry_sdk.integrations.logging import LoggingIntegration
 
 minimum_python_37 = pytest.mark.skipif(
     sys.version_info < (3, 7), reason="Asyncio tests need Python >= 3.7"
 )
 
 
+def otel_attributes_to_dict(otel_attrs: List[Any]):
+    return {item["key"]: item["value"] for item in otel_attrs}
+
+
 @minimum_python_37
 def test_logs_disabled_by_default(sentry_init, capture_envelopes):
     sentry_init()
+
+    python_logger = logging.Logger("some-logger")
+
     envelopes = capture_envelopes()
 
     sentry_logger.trace("This is a 'trace' log.")
@@ -22,6 +31,7 @@ def test_logs_disabled_by_default(sentry_init, capture_envelopes):
     sentry_logger.warn("This is a 'warn' log...")
     sentry_logger.error("This is a 'error' log...")
     sentry_logger.fatal("This is a 'fatal' log...")
+    python_logger.warning("sad")
 
     assert len(envelopes) == 0
 
@@ -64,14 +74,14 @@ def test_logs_basics(sentry_init, capture_envelopes):
 @minimum_python_37
 def test_logs_before_emit_log(sentry_init, capture_envelopes):
     def _before_log(record, hint):
-        assert list(record.keys()) == [
+        assert set(record.keys()) == {
             "severity_text",
             "severity_number",
             "body",
             "attributes",
             "time_unix_nano",
             "trace_id",
-        ]
+        }
 
         if record["severity_text"] in ["fatal", "error"]:
             return None
@@ -123,34 +133,14 @@ def test_logs_attributes(sentry_init, capture_envelopes):
     log_item = envelopes[0].items[0].payload.json
     assert log_item["body"]["stringValue"] == "The recorded value was 'some value'"
 
-    assert log_item["attributes"][1] == {
-        "key": "attr_int",
-        "value": {"intValue": "1"},
-    }  # TODO: this is strange.
-    assert log_item["attributes"][2] == {
-        "key": "attr_float",
-        "value": {"doubleValue": 2.0},
-    }
-    assert log_item["attributes"][3] == {
-        "key": "attr_bool",
-        "value": {"boolValue": True},
-    }
-    assert log_item["attributes"][4] == {
-        "key": "attr_string",
-        "value": {"stringValue": "string attribute"},
-    }
-    assert log_item["attributes"][5] == {
-        "key": "sentry.environment",
-        "value": {"stringValue": "production"},
-    }
-    assert log_item["attributes"][6] == {
-        "key": "sentry.release",
-        "value": {"stringValue": mock.ANY},
-    }
-    assert log_item["attributes"][7] == {
-        "key": "sentry.message.parameters.my_var",
-        "value": {"stringValue": "some value"},
-    }
+    attrs = otel_attributes_to_dict(log_item["attributes"])
+    assert attrs["attr_int"] == {"intValue": "1"}
+    assert attrs["attr_float"] == {"doubleValue": 2.0}
+    assert attrs["attr_bool"] == {"boolValue": True}
+    assert attrs["attr_string"] == {"stringValue": "string attribute"}
+    assert attrs["sentry.environment"] == {"stringValue": "production"}
+    assert attrs["sentry.release"] == {"stringValue": mock.ANY}
+    assert attrs["sentry.message.parameters.my_var"] == {"stringValue": "some value"}
 
 
 @minimum_python_37
@@ -172,37 +162,33 @@ def test_logs_message_params(sentry_init, capture_envelopes):
         envelopes[0].items[0].payload.json["body"]["stringValue"]
         == "The recorded value was '1'"
     )
-    assert envelopes[0].items[0].payload.json["attributes"][-1] == {
-        "key": "sentry.message.parameters.int_var",
-        "value": {"intValue": "1"},
-    }  # TODO: this is strange.
+    assert otel_attributes_to_dict(envelopes[0].items[0].payload.json["attributes"])[
+        "sentry.message.parameters.int_var"
+    ] == {"intValue": "1"}
 
     assert (
         envelopes[1].items[0].payload.json["body"]["stringValue"]
         == "The recorded value was '2.0'"
     )
-    assert envelopes[1].items[0].payload.json["attributes"][-1] == {
-        "key": "sentry.message.parameters.float_var",
-        "value": {"doubleValue": 2.0},
-    }
+    assert otel_attributes_to_dict(envelopes[1].items[0].payload.json["attributes"])[
+        "sentry.message.parameters.float_var"
+    ] == {"doubleValue": 2.0}
 
     assert (
         envelopes[2].items[0].payload.json["body"]["stringValue"]
         == "The recorded value was 'False'"
     )
-    assert envelopes[2].items[0].payload.json["attributes"][-1] == {
-        "key": "sentry.message.parameters.bool_var",
-        "value": {"boolValue": False},
-    }
+    assert otel_attributes_to_dict(envelopes[2].items[0].payload.json["attributes"])[
+        "sentry.message.parameters.bool_var"
+    ] == {"boolValue": False}
 
     assert (
         envelopes[3].items[0].payload.json["body"]["stringValue"]
         == "The recorded value was 'some string value'"
     )
-    assert envelopes[3].items[0].payload.json["attributes"][-1] == {
-        "key": "sentry.message.parameters.string_var",
-        "value": {"stringValue": "some string value"},
-    }
+    assert otel_attributes_to_dict(envelopes[3].items[0].payload.json["attributes"])[
+        "sentry.message.parameters.string_var"
+    ] == {"stringValue": "some string value"}
 
 
 @minimum_python_37
@@ -235,8 +221,63 @@ def test_logs_tied_to_spans(sentry_init, capture_envelopes):
         with sentry_sdk.start_span(description="test-span") as span:
             sentry_logger.warn("This is a log tied to a span")
 
+    attrs = otel_attributes_to_dict(envelopes[0].items[0].payload.json["attributes"])
+    assert attrs["sentry.trace.parent_span_id"] == {"stringValue": span.span_id}
+
+
+@minimum_python_37
+def test_logger_integration_warning(sentry_init, capture_envelopes):
+    """
+    The python logger module should create 'warn' sentry logs if the flag is on.
+    """
+    sentry_init(_experiments={"enable_sentry_logs": True})
+    envelopes = capture_envelopes()
+
+    python_logger = logging.Logger("test-logger")
+    python_logger.warning("this is %s a template %s", "1", "2")
+
     log_entry = envelopes[0].items[0].payload.json
-    assert log_entry["attributes"][-1] == {
-        "key": "sentry.trace.parent_span_id",
-        "value": {"stringValue": span.span_id},
+    attrs = otel_attributes_to_dict(log_entry["attributes"])
+    assert attrs["sentry.message.template"] == {
+        "stringValue": "this is %s a template %s"
     }
+    assert "code.file.path" in attrs
+    assert "code.line.number" in attrs
+    assert attrs["logger.name"] == {"stringValue": "test-logger"}
+    assert attrs["sentry.environment"] == {"stringValue": "production"}
+    assert attrs["sentry.message.parameters.0"] == {"stringValue": "1"}
+    assert attrs["sentry.message.parameters.1"]
+    assert log_entry["severityNumber"] == 13
+    assert log_entry["severityText"] == "warn"
+
+
+@minimum_python_37
+def test_logger_integration_debug(sentry_init, capture_envelopes):
+    """
+    The python logger module should not create 'debug' sentry logs if the flag is on by default
+    """
+    sentry_init(_experiments={"enable_sentry_logs": True})
+    envelopes = capture_envelopes()
+
+    python_logger = logging.Logger("test-logger")
+    python_logger.debug("this is %s a template %s", "1", "2")
+
+    assert len(envelopes) == 0
+
+
+@minimum_python_37
+def test_no_log_infinite_loop(sentry_init, capture_envelopes):
+    """
+    If 'debug' mode is true, and you set a low log level in the logging integration, there should be no infinite loops.
+    """
+    sentry_init(
+        _experiments={"enable_sentry_logs": True},
+        integrations=[LoggingIntegration(sentry_logs_level=logging.DEBUG)],
+        debug=True,
+    )
+    envelopes = capture_envelopes()
+
+    python_logger = logging.Logger("test-logger")
+    python_logger.debug("this is %s a template %s", "1", "2")
+
+    assert len(envelopes) == 1