fix merge issues

Author: FrenchGithubUser

synapse/federation/federation_server.py

@@ -668,10 +668,10 @@ async def on_pdu_request(
     async def on_query_request(
         self, query_type: str, args: dict[str, str]
     ) -> tuple[int, dict[str, Any]]:
-        received_queries_counter.labels(
-            type=query_type,
-            **{SERVER_NAME_LABEL: self.server_name},
-        ).inc()
+        received_queries_counter.add(
+            1,
+            {"type": query_type, SERVER_NAME_LABEL: self.server_name},
+        )
         resp = await self.registry.on_query(query_type, args)
         return 200, resp
 

synapse/replication/http/_base.py

@@ -25,6 +25,8 @@
 from inspect import signature
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, ClassVar
 
+from prometheus_client import Gauge
+
 from twisted.internet.error import ConnectError, DNSLookupError
 from twisted.web.server import Request
 
@@ -47,9 +49,10 @@
 
 logger = logging.getLogger(__name__)
 
-_pending_outgoing_requests = meter.create_up_down_counter(
+_pending_outgoing_requests = Gauge(
     "synapse_pending_outgoing_replication_requests",
-    description="Number of active outgoing replication requests, by replication method name",
+    "Number of active outgoing replication requests, by replication method name",
+    labelnames=["name", SERVER_NAME_LABEL],
 )
 
 _outgoing_request_counter = meter.create_counter(
@@ -209,9 +212,9 @@ def make_client(cls, hs: "HomeServer") -> Callable:
 
         instance_map = hs.config.worker.instance_map
 
-        outgoing_gauge = meter.create_up_down_counter(
-            "synapse_pending_outgoing_replication_requests",
-            description="Number of active outgoing replication requests, by replication method name",
+        outgoing_gauge = _pending_outgoing_requests.labels(
+            name=cls.NAME,
+            **{SERVER_NAME_LABEL: server_name},
         )
 
         replication_secret = None
@@ -228,143 +231,65 @@ async def send_request(
             streams = hs.get_replication_command_handler().get_streams_to_replicate()
             replication = hs.get_replication_data_handler()
 
-            outgoing_gauge.add(
-                1,
-                {"name": cls.NAME, SERVER_NAME_LABEL: server_name},
-            )
-            if instance_name == local_instance_name:
-                raise Exception("Trying to send HTTP request to self")
-            if instance_name not in instance_map:
-                raise Exception(
-                    "Instance %r not in 'instance_map' config" % (instance_name,)
-                )
-
-            data = await cls._serialize_payload(**kwargs)
-
-            if cls.METHOD != "GET" and cls.WAIT_FOR_STREAMS:
-                # Include the current stream positions that we write to. We
-                # don't do this for GETs as they don't have a body, and we
-                # generally assume that a GET won't rely on data we have
-                # written.
-                if _STREAM_POSITION_KEY in data:
+            with outgoing_gauge.track_inprogress():
+                if instance_name == local_instance_name:
+                    raise Exception("Trying to send HTTP request to self")
+                if instance_name not in instance_map:
                     raise Exception(
-                        "data to send contains %r key", _STREAM_POSITION_KEY
+                        "Instance %r not in 'instance_map' config" % (instance_name,)
                     )
 
-                data[_STREAM_POSITION_KEY] = {
-                    "streams": {
-                        stream.NAME: stream.minimal_local_current_token()
-                        for stream in streams
-                    },
-                    "instance_name": local_instance_name,
-                }
-
-            url_args = [
-                urllib.parse.quote(kwargs[name], safe="") for name in cls.PATH_ARGS
-            ]
-
-            if cls.CACHE:
-                txn_id = random_string(10)
-                url_args.append(txn_id)
-
-            if cls.METHOD == "POST":
-                request_func: Callable[..., Awaitable[Any]] = client.post_json_get_json
-            elif cls.METHOD == "PUT":
-                request_func = client.put_json
-            elif cls.METHOD == "GET":
-                request_func = client.get_json
-            else:
-                # We have already asserted in the constructor that a
-                # compatible was picked, but lets be paranoid.
-                raise Exception(
-                    "Unknown METHOD on %s replication endpoint" % (cls.NAME,)
-                )
-
-            # Hard code a special scheme to show this only used for replication. The
-            # instance_name will be passed into the ReplicationEndpointFactory to
-            # determine connection details from the instance_map.
-            uri = "synapse-replication://%s/_synapse/replication/%s/%s" % (
-                instance_name,
-                cls.NAME,
-                "/".join(url_args),
-            )
+                data = await cls._serialize_payload(**kwargs)
 
-            headers: Dict[bytes, List[bytes]] = {}
-            # Add an authorization header, if configured.
-            if replication_secret:
-                headers[b"Authorization"] = [b"Bearer " + replication_secret]
-            opentracing.inject_header_dict(headers, check_destination=False)
-
-            try:
-                # Keep track of attempts made so we can bail if we don't manage to
-                # connect to the target after N tries.
-                attempts = 0
-                # We keep retrying the same request for timeouts. This is so that we
-                # have a good idea that the request has either succeeded or failed
-                # on the master, and so whether we should clean up or not.
-                while True:
-                    try:
-                        result = await request_func(uri, data, headers=headers)
-                        break
-                    except RequestTimedOutError:
-                        if not cls.RETRY_ON_TIMEOUT:
-                            raise
-
-                        logger.warning("%s request timed out; retrying", cls.NAME)
-
-                        # If we timed out we probably don't need to worry about backing
-                        # off too much, but lets just wait a little anyway.
-                        await clock.sleep(1)
-                    except (ConnectError, DNSLookupError) as e:
-                        if not cls.RETRY_ON_CONNECT_ERROR:
-                            raise
-                        if attempts > cls.RETRY_ON_CONNECT_ERROR_ATTEMPTS:
-                            raise
-
-                        delay = 2**attempts
-                        logger.warning(
-                            "%s request connection failed; retrying in %ds: %r",
-                            cls.NAME,
-                            delay,
-                            e,
+                if cls.METHOD != "GET" and cls.WAIT_FOR_STREAMS:
+                    # Include the current stream positions that we write to. We
+                    # don't do this for GETs as they don't have a body, and we
+                    # generally assume that a GET won't rely on data we have
+                    # written.
+                    if _STREAM_POSITION_KEY in data:
+                        raise Exception(
+                            "data to send contains %r key", _STREAM_POSITION_KEY
                         )
 
-                        await clock.sleep(delay)
-                        attempts += 1
-            except HttpResponseException as e:
-                # We convert to SynapseError as we know that it was a SynapseError
-                # on the main process that we should send to the client. (And
-                # importantly, not stack traces everywhere)
-                _outgoing_request_counter.add(
-                    1,
-                    {
-                        "name": cls.NAME,
-                        "code": e.code,
-                        SERVER_NAME_LABEL: server_name,
-                    },
-                )
-                raise e.to_synapse_error()
-            except Exception as e:
-                _outgoing_request_counter.add(
-                    1,
-                    {
-                        "name": cls.NAME,
-                        "code": "ERR",
-                        SERVER_NAME_LABEL: server_name,
-                    },
+                    data[_STREAM_POSITION_KEY] = {
+                        "streams": {
+                            stream.NAME: stream.minimal_local_current_token()
+                            for stream in streams
+                        },
+                        "instance_name": local_instance_name,
+                    }
+
+                url_args = [
+                    urllib.parse.quote(kwargs[name], safe="") for name in cls.PATH_ARGS
+                ]
+
+                if cls.CACHE:
+                    txn_id = random_string(10)
+                    url_args.append(txn_id)
+
+                if cls.METHOD == "POST":
+                    request_func: Callable[..., Awaitable[Any]] = (
+                        client.post_json_get_json
+                    )
+                elif cls.METHOD == "PUT":
+                    request_func = client.put_json
+                elif cls.METHOD == "GET":
+                    request_func = client.get_json
+                else:
+                    # We have already asserted in the constructor that a
+                    # compatible was picked, but lets be paranoid.
+                    raise Exception(
+                        "Unknown METHOD on %s replication endpoint" % (cls.NAME,)
+                    )
+
+                # Hard code a special scheme to show this only used for replication. The
+                # instance_name will be passed into the ReplicationEndpointFactory to
+                # determine connection details from the instance_map.
+                uri = "synapse-replication://%s/_synapse/replication/%s/%s" % (
+                    instance_name,
+                    cls.NAME,
+                    "/".join(url_args),
                 )
-                raise SynapseError(
-                    502, f"Failed to talk to {instance_name} process"
-                ) from e
-
-            _outgoing_request_counter.add(
-                1,
-                {
-                    "name": cls.NAME,
-                    "code": 200,
-                    SERVER_NAME_LABEL: server_name,
-                },
-            )
 
                 headers: dict[bytes, list[bytes]] = {}
                 # Add an authorization header, if configured.
@@ -412,27 +337,32 @@ async def send_request(
                     # We convert to SynapseError as we know that it was a SynapseError
                     # on the main process that we should send to the client. (And
                     # importantly, not stack traces everywhere)
-                    _outgoing_request_counter.labels(
-                        name=cls.NAME,
-                        code=e.code,
-                        **{SERVER_NAME_LABEL: server_name},
-                    ).inc()
+                    _outgoing_request_counter.add(
+                        1,
+                        {
+                            "name": cls.NAME,
+                            "code": e.code,
+                            SERVER_NAME_LABEL: server_name,
+                        },
+                    )
                     raise e.to_synapse_error()
                 except Exception as e:
-                    _outgoing_request_counter.labels(
-                        name=cls.NAME,
-                        code="ERR",
-                        **{SERVER_NAME_LABEL: server_name},
-                    ).inc()
+                    _outgoing_request_counter.add(
+                        1,
+                        {
+                            "name": cls.NAME,
+                            "code": "ERR",
+                            SERVER_NAME_LABEL: server_name,
+                        },
+                    )
                     raise SynapseError(
                         502, f"Failed to talk to {instance_name} process"
                     ) from e
 
-                _outgoing_request_counter.labels(
-                    name=cls.NAME,
-                    code=200,
-                    **{SERVER_NAME_LABEL: server_name},
-                ).inc()
+                _outgoing_request_counter.add(
+                    1,
+                    {"name": cls.NAME, "code": 200, SERVER_NAME_LABEL: server_name},
+                )
 
                 # Wait on any streams that the remote may have written to.
                 for stream_name, position in result.pop(
@@ -444,7 +374,7 @@ async def send_request(
                         position=position,
                     )
 
-            return result
+                return result
 
         return send_request
 

synapse/util/ratelimitutils.py

@@ -28,7 +28,6 @@
 from typing import (
     Any,
     Callable,
-    ContextManager,
     Iterator,
     Mapping,
     MutableSet,