Method record_error_count

redis/observability/metrics.py:241–285

Record error count Args: server_address: Server address server_port: Server port network_peer_address: Network peer address network_peer_port: Network peer port error_type: Error type retry_attempts: Retry atte

(
        self,
        server_address: Optional[str] = None,
        server_port: Optional[int] = None,
        network_peer_address: Optional[str] = None,
        network_peer_port: Optional[int] = None,
        error_type: Optional[Exception] = None,
        retry_attempts: Optional[int] = None,
        is_internal: Optional[bool] = None,
    )

    # Resiliency metric recording methods

    def record_error_count(
        self,
        server_address: Optional[str] = None,
        server_port: Optional[int] = None,
        network_peer_address: Optional[str] = None,
        network_peer_port: Optional[int] = None,
        error_type: Optional[Exception] = None,
        retry_attempts: Optional[int] = None,
        is_internal: Optional[bool] = None,
    ):
        """
        Record error count

        Args:
            server_address: Server address
            server_port: Server port
            network_peer_address: Network peer address
            network_peer_port: Network peer port
            error_type: Error type
            retry_attempts: Retry attempts
            is_internal: Whether the error is internal (e.g., timeout, network error)
        """
        if not hasattr(self, "client_errors"):
            return

        attrs = self.attr_builder.build_base_attributes(
            server_address=server_address,
            server_port=server_port,
        )
        attrs.update(
            self.attr_builder.build_operation_attributes(
                network_peer_address=network_peer_address,
                network_peer_port=network_peer_port,
                retry_attempts=retry_attempts,
            )
        )

        attrs.update(
            self.attr_builder.build_error_attributes(
                error_type=error_type,
                is_internal=is_internal,
            )
        )

        self.client_errors.add(1, attributes=attrs)

    def record_maint_notification_count(
        self,