diff --git a/sdk/core/azure-core/HISTORY.md b/sdk/core/azure-core/HISTORY.md
index 93f247b1f1c6..5af465b79aff 100644
--- a/sdk/core/azure-core/HISTORY.md
+++ b/sdk/core/azure-core/HISTORY.md
@@ -10,6 +10,7 @@
 - Tracing: DistributedTracingPolicy now accepts kwargs network_span_namer to change network span name  #7773
 - Tracing: Implementation of AbstractSpan can now use the mixin HttpSpanMixin to get HTTP span update automatically  #7773
 - Tracing: AbstractSpan contract "change_context" introduced  #7773
+- Introduce new policy HttpLoggingPolicy  #7988
 
 ### Bug fixes
 
diff --git a/sdk/core/azure-core/azure/core/pipeline/policies/__init__.py b/sdk/core/azure-core/azure/core/pipeline/policies/__init__.py
index 9e321cee1138..4b724ee41493 100644
--- a/sdk/core/azure-core/azure/core/pipeline/policies/__init__.py
+++ b/sdk/core/azure-core/azure/core/pipeline/policies/__init__.py
@@ -35,7 +35,8 @@
     UserAgentPolicy,
     NetworkTraceLoggingPolicy,
     ContentDecodePolicy,
-    ProxyPolicy
+    ProxyPolicy,
+    HttpLoggingPolicy,
 )
 
 __all__ = [
@@ -52,6 +53,7 @@
     'CustomHookPolicy',
     'DistributedTracingPolicy',
     'RequestHistory',
+    'HttpLoggingPolicy',
 ]
 
 #pylint: disable=unused-import
diff --git a/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py b/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py
index 3df461c0feac..3320f6f094ab 100644
--- a/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py
+++ b/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py
@@ -36,6 +36,7 @@
 import re
 from typing import (Mapping, IO, TypeVar, TYPE_CHECKING, Type, cast, List, Callable, Iterator, # pylint: disable=unused-import
                     Any, Union, Dict, Optional, AnyStr)
+from six.moves import urllib
 
 from azure.core import __version__  as azcore_version
 from azure.core.exceptions import (
@@ -267,6 +268,106 @@ def on_response(self, request, response):
                 _LOGGER.debug("Failed to log response: %s", repr(err))
 
 
+class HttpLoggingPolicy(SansIOHTTPPolicy):
+    """The Pipeline policy that handles logging of HTTP requests and responses.
+    """
+
+    DEFAULT_HEADERS_WHITELIST = set([
+        "x-ms-client-request-id",
+        "x-ms-return-client-request-id",
+        "traceparent",
+        "Accept",
+        "Cache-Control",
+        "Connection",
+        "Content-Length",
+        "Content-Type",
+        "Date",
+        "ETag",
+        "Expires",
+        "If-Match",
+        "If-Modified-Since",
+        "If-None-Match",
+        "If-Unmodified-Since",
+        "Last-Modified",
+        "Pragma",
+        "Request-Id",
+        "Retry-After",
+        "Server",
+        "Transfer-Encoding",
+        "User-Agent"
+    ])
+    REDACTED_PLACEHOLDER = "REDACTED"
+
+    def __init__(self, logger=None, **kwargs):  # pylint: disable=unused-argument
+        self.logger = logger or logging.getLogger(
+            "azure.core.pipeline.policies.http_logging_policy"
+        )
+        self.allowed_query_params = set()
+        self.allowed_header_namers = set(HttpLoggingPolicy.DEFAULT_HEADERS_WHITELIST)
+
+    def _redact_query_param(self, key, value):
+        lower_case_allowed_query_params = [
+            param.lower() for param in self.allowed_query_params
+        ]
+        return value if key.lower() in lower_case_allowed_query_params else HttpLoggingPolicy.REDACTED_PLACEHOLDER
+
+    def _redact_header(self, key, value):
+        lower_case_allowed_header_namers = [
+            header.lower() for header in self.allowed_header_namers
+        ]
+        return value if key.lower() in lower_case_allowed_header_namers else HttpLoggingPolicy.REDACTED_PLACEHOLDER
+
+    def on_request(self, request):
+        # type: (PipelineRequest) -> None
+        """Logs HTTP method, url and headers.
+        :param request: The PipelineRequest object.
+        :type request: ~azure.core.pipeline.PipelineRequest
+        """
+        http_request = request.http_request
+        options = request.context.options
+        # Get logger in my context first (request has been retried)
+        # then read from kwargs (pop if that's the case)
+        # then use my instance logger
+        logger = request.context.setdefault("logger", options.pop("logger", self.logger))
+
+        if not logger.isEnabledFor(logging.INFO):
+            return
+
+        try:
+            parsed_url = list(urllib.parse.urlparse(http_request.url))
+            parsed_qp = urllib.parse.parse_qsl(parsed_url[4], keep_blank_values=True)
+            filtered_qp = [(key, self._redact_query_param(key, value)) for key, value in parsed_qp]
+            # 4 is query
+            parsed_url[4] = "&".join(["=".join(part) for part in filtered_qp])
+            redacted_url = urllib.parse.urlunparse(parsed_url)
+
+            logger.info("Request URL: %r", redacted_url)
+            logger.info("Request method: %r", http_request.method)
+            logger.info("Request headers:")
+            for header, value in http_request.headers.items():
+                value = self._redact_header(header, value)
+                logger.info("    %r: %r", header, value)
+        except Exception as err:  # pylint: disable=broad-except
+            logger.warning("Failed to log request: %s", repr(err))
+
+    def on_response(self, request, response):
+        # type: (PipelineRequest, PipelineResponse) -> None
+        http_response = response.http_response
+        logger = response.context.get("logger")
+
+        if not logger.isEnabledFor(logging.INFO):
+            return
+
+        try:
+            logger.info("Response status: %r", http_response.status_code)
+            logger.info("Response headers:")
+            for res_header, value in http_response.headers.items():
+                value = self._redact_header(res_header, value)
+                logger.info("    %r: %r", res_header, value)
+        except Exception as err:  # pylint: disable=broad-except
+            logger.warning("Failed to log response: %s", repr(err))
+
+
 class ContentDecodePolicy(SansIOHTTPPolicy):
     """Policy for decoding unstreamed response content.
     """
diff --git a/sdk/core/azure-core/tests/test_universal_pipeline.py b/sdk/core/azure-core/tests/test_universal_pipeline.py
index 8ca5d3857131..e0a9079709fc 100644
--- a/sdk/core/azure-core/tests/test_universal_pipeline.py
+++ b/sdk/core/azure-core/tests/test_universal_pipeline.py
@@ -24,6 +24,7 @@
 # THE SOFTWARE.
 #
 #--------------------------------------------------------------------------
+import logging
 try:
     from unittest import mock
 except ImportError:
@@ -49,7 +50,8 @@
 from azure.core.pipeline.policies import (
     NetworkTraceLoggingPolicy,
     ContentDecodePolicy,
-    UserAgentPolicy
+    UserAgentPolicy,
+    HttpLoggingPolicy,
 )
 
 def test_user_agent():
@@ -192,3 +194,179 @@ def body(self):
     raw_deserializer.on_response(None, response)
     result = response.context["deserialized_data"]
     assert result["success"] is True
+
+
+def test_http_logger():
+
+    class MockHandler(logging.Handler):
+        def __init__(self):
+            super(MockHandler, self).__init__()
+            self.messages = []
+        def reset(self):
+            self.messages = []
+        def emit(self, record):
+            self.messages.append(record)
+    mock_handler = MockHandler()
+
+    logger = logging.getLogger("testlogger")
+    logger.addHandler(mock_handler)
+    logger.setLevel(logging.DEBUG)
+
+    policy = HttpLoggingPolicy(logger=logger)
+
+    universal_request = HttpRequest('GET', 'http://127.0.0.1/')
+    http_response = HttpResponse(universal_request, None)
+    http_response.status_code = 202
+    request = PipelineRequest(universal_request, PipelineContext(None))
+
+    # Basics
+
+    policy.on_request(request)
+    response = PipelineResponse(request, http_response, request.context)
+    policy.on_response(request, response)
+
+    assert all(m.levelname == 'INFO' for m in mock_handler.messages)
+    assert len(mock_handler.messages) == 5
+    assert mock_handler.messages[0].message == "Request URL: 'http://127.0.0.1/'"
+    assert mock_handler.messages[1].message == "Request method: 'GET'"
+    assert mock_handler.messages[2].message == 'Request headers:'
+    assert mock_handler.messages[3].message == 'Response status: 202'
+    assert mock_handler.messages[4].message == 'Response headers:'
+
+    mock_handler.reset()
+
+    # Let's make this request a failure, retried twice
+
+    policy.on_request(request)
+    response = PipelineResponse(request, http_response, request.context)
+    policy.on_response(request, response)
+
+    policy.on_request(request)
+    response = PipelineResponse(request, http_response, request.context)
+    policy.on_response(request, response)
+
+    assert all(m.levelname == 'INFO' for m in mock_handler.messages)
+    assert len(mock_handler.messages) == 10
+    assert mock_handler.messages[0].message == "Request URL: 'http://127.0.0.1/'"
+    assert mock_handler.messages[1].message == "Request method: 'GET'"
+    assert mock_handler.messages[2].message == 'Request headers:'
+    assert mock_handler.messages[3].message == 'Response status: 202'
+    assert mock_handler.messages[4].message == 'Response headers:'
+    assert mock_handler.messages[0].message == "Request URL: 'http://127.0.0.1/'"
+    assert mock_handler.messages[1].message == "Request method: 'GET'"
+    assert mock_handler.messages[2].message == 'Request headers:'
+    assert mock_handler.messages[3].message == 'Response status: 202'
+    assert mock_handler.messages[4].message == 'Response headers:'
+
+    mock_handler.reset()
+
+    # Headers and query parameters
+
+    policy.allowed_query_params = ['country']
+
+    universal_request.headers = {
+        "Accept": "Caramel",
+        "Hate": "Chocolat",
+    }
+    http_response.headers = {
+        "Content-Type": "Caramel",
+        "HateToo": "Chocolat",
+    }
+    universal_request.url = "http://127.0.0.1/?country=france&city=aix"
+
+    policy.on_request(request)
+    response = PipelineResponse(request, http_response, request.context)
+    policy.on_response(request, response)
+
+    assert all(m.levelname == 'INFO' for m in mock_handler.messages)
+    assert len(mock_handler.messages) == 9
+    assert mock_handler.messages[0].message == "Request URL: 'http://127.0.0.1/?country=france&city=REDACTED'"
+    assert mock_handler.messages[1].message == "Request method: 'GET'"
+    assert mock_handler.messages[2].message == "Request headers:"
+    # Dict not ordered in Python, exact logging order doesn't matter
+    assert set([
+        mock_handler.messages[3].message,
+        mock_handler.messages[4].message
+    ]) == set([
+        "    'Accept': 'Caramel'",
+        "    'Hate': 'REDACTED'"
+    ])
+    assert mock_handler.messages[5].message == "Response status: 202"
+    assert mock_handler.messages[6].message == "Response headers:"
+    # Dict not ordered in Python, exact logging order doesn't matter
+    assert set([
+        mock_handler.messages[7].message,
+        mock_handler.messages[8].message
+    ]) == set([
+        "    'Content-Type': 'Caramel'",
+        "    'HateToo': 'REDACTED'"
+    ])
+
+    mock_handler.reset()
+
+def test_http_logger_operation_level():
+
+    class MockHandler(logging.Handler):
+        def __init__(self):
+            super(MockHandler, self).__init__()
+            self.messages = []
+        def reset(self):
+            self.messages = []
+        def emit(self, record):
+            self.messages.append(record)
+    mock_handler = MockHandler()
+
+    logger = logging.getLogger("testlogger")
+    logger.addHandler(mock_handler)
+    logger.setLevel(logging.DEBUG)
+
+    policy = HttpLoggingPolicy()
+    kwargs={'logger': logger}
+
+    universal_request = HttpRequest('GET', 'http://127.0.0.1/')
+    http_response = HttpResponse(universal_request, None)
+    http_response.status_code = 202
+    request = PipelineRequest(universal_request, PipelineContext(None, **kwargs))
+
+    # Basics
+
+    policy.on_request(request)
+    response = PipelineResponse(request, http_response, request.context)
+    policy.on_response(request, response)
+
+    assert all(m.levelname == 'INFO' for m in mock_handler.messages)
+    assert len(mock_handler.messages) == 5
+    assert mock_handler.messages[0].message == "Request URL: 'http://127.0.0.1/'"
+    assert mock_handler.messages[1].message == "Request method: 'GET'"
+    assert mock_handler.messages[2].message == 'Request headers:'
+    assert mock_handler.messages[3].message == 'Response status: 202'
+    assert mock_handler.messages[4].message == 'Response headers:'
+
+    mock_handler.reset()
+
+    # Let's make this request a failure, retried twice
+
+    request = PipelineRequest(universal_request, PipelineContext(None, **kwargs))
+
+    policy.on_request(request)
+    response = PipelineResponse(request, http_response, request.context)
+    policy.on_response(request, response)
+
+    policy.on_request(request)
+    response = PipelineResponse(request, http_response, request.context)
+    policy.on_response(request, response)
+
+    assert all(m.levelname == 'INFO' for m in mock_handler.messages)
+    assert len(mock_handler.messages) == 10
+    assert mock_handler.messages[0].message == "Request URL: 'http://127.0.0.1/'"
+    assert mock_handler.messages[1].message == "Request method: 'GET'"
+    assert mock_handler.messages[2].message == 'Request headers:'
+    assert mock_handler.messages[3].message == 'Response status: 202'
+    assert mock_handler.messages[4].message == 'Response headers:'
+    assert mock_handler.messages[0].message == "Request URL: 'http://127.0.0.1/'"
+    assert mock_handler.messages[1].message == "Request method: 'GET'"
+    assert mock_handler.messages[2].message == 'Request headers:'
+    assert mock_handler.messages[3].message == 'Response status: 202'
+    assert mock_handler.messages[4].message == 'Response headers:'
+
+    mock_handler.reset()