feat: replace Synchronous HTTP Calls with Asynchronous in OPA plugin

[monshri]

📝 This PR addresses the following:

Fix #1931

Previous implementation of OPA plugin had the following issue:
When the OPA plugin is enabled and evaluates policies, each policy check blocks the entire async worker until the HTTP request completes, which caused latency across all concurrent requests, etc.

Fix:

Key improvements:

• Replaced all synchronous HTTP calls with asynchronous equivalents.
• Added retry logic with exponential backoff for transient connection failures.
• Introduced request timeout configuration to prevent indefinite hangs.
• Enhanced logging to capture retry attempts and HTTP status errors.

        for attempt in range(max_retries):
            try:
                response = await client.post(url, json=payload, timeout=seconds)
                response.raise_for_status()
                logger.info(f"OPA POST succeeded on attempt {attempt + 1}")
                return response
            except (httpx.TimeoutException, httpx.RequestError, httpx.ConnectError):
                logger.warning(f"Retry attempt to connect to OPA server {attempt + 1}")
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2**attempt)
            except httpx.HTTPStatusError as e:
                logger.error(f"OPA POST HTTP error (attempt {attempt + 1}, status {e.response.status_code}): 
{e.response.text[:200]}")

2. Added configurable timeout and maximum retry attempts as part of plugin configuration config.yaml

    config:
      # Plugin config dict passed to the plugin constructor
      opa_base_url: "http://127.0.0.1:8181/v1/data/"
      opa_client_retries: 3
      opa_client_timeout: 30s

[terylt]

Is this something that could be done once in the init function of the Plugin? This would save processing.

[araujof]

e.g.,

# Parse once in __init__:
  def __init__(self, config: PluginConfig):
      ...
      # Parse timeout once
      match = re.match(r"(\d+)s", self.opa_config.opa_client_timeout.strip())
      self._timeout_seconds = float(match.group(1)) if match else 30.0

[monshri]

yes that's a good point. I will make that change

[araujof]

Hi Shriti, please look into the recommendations we made.

[araujof]

A new httpx.AsyncClient() is created and destroyed for every OPA policy evaluation. This means:

• Connection establishment overhead on every request
• No connection reuse/pooling
• TCP handshake + TLS negotiation repeated for each policy check
• Under high load with many plugin invocations, this becomes a severe bottleneck

Recommendation:

  # In __init__:
  self._http_client = httpx.AsyncClient(
      timeout=httpx.Timeout(self._timeout_seconds),
      limits=httpx.Limits(max_keepalive_connections=20, max_connections=100),
      http2=True  # Optional: enable HTTP/2 for better performance
  )

  # Add cleanup method:
  async def cleanup(self):
      await self._http_client.aclose()

[araujof]

Maybe set a default (those are max numbers) and allows for configuration, I think.

[monshri]

Done

[araujof]

e.g.,

# Parse once in __init__:
  def __init__(self, config: PluginConfig):
      ...
      # Parse timeout once
      match = re.match(r"(\d+)s", self.opa_config.opa_client_timeout.strip())
      self._timeout_seconds = float(match.group(1)) if match else 30.0

[araujof]

await asyncio.sleep(2**attempt) grows exponentially without limit:

• Attempt 0: 1 second
• Attempt 1: 2 seconds
• Attempt 2: 4 seconds
• Attempt 3: 8 seconds (if max_retries increased)

This could cause excessive delays if max_retries is increased.

Recommendation:

  # Add jitter and cap max delay
  import random
  delay = min(2**attempt, 10) + random.uniform(0, 1)
  await asyncio.sleep(delay)

[monshri]

Sure, makes sense. Will make that change

[monshri]

Done

[araujof]

Nice job on this PR. As noted in the inlined comments, a few suggestions:

1. Connection Pooling: Consider reusing a single httpx.AsyncClient instance instead of creating new ones per request:

  # In __init__:
  self.http_client = httpx.AsyncClient(timeout=httpx.Timeout(30.0))

  # In _evaluate_opa_policy:
  rsp = await self._post_with_retry(client=self.http_client, url=url, payload=payload, ...)

  # Add cleanup method:
  async def cleanup(self):
      await self.http_client.aclose()

This would improve performance by avoiding connection establishment overhead on each OPA request.

2. Regex Compilation: The timeout regex is compiled on every call (line 136):

  match = re.match(r"(\d+)s", self.opa_config.opa_client_timeout.strip())

Could be compiled once in init:

  # In __init__:
  self._timeout_pattern = re.compile(r"(\d+)s")

  # In _post_with_retry:
  match = self._timeout_pattern.match(self.opa_config.opa_client_timeout.strip())

[araujof]

LGTM