Istio Ingress

#!/usr/bin/env python3
# Copyright 2025 Canonical Ltd.
# See LICENSE file for licensing details.

r"""# Interface Library for istio_ingress_route.

This library wraps relation endpoints for istio_ingress_route. The requirer of this
relation is any charm needing advanced ingress routing (multi-port, multi-protocol).
The provider is the istio-ingress-k8s charm.

## Getting Started

To get started using the library, you just need to fetch the library using `charmcraft`.

```shell
cd some-charm
charmcraft fetch-lib charms.istio_ingress_k8s.v0.istio_ingress_route
```

To use the library from the requirer side:

```yaml
requires:
    ingress:
        interface: istio_ingress_route
```

```python
from charms.istio_ingress_k8s.v0.istio_ingress_route import (
    IstioIngressRouteRequirer,
    IstioIngressRouteConfig,
    Listener,
    HTTPRoute,
    GRPCRoute,
    BackendRef,
    ProtocolType,
    HTTPMethod,
    HTTPRouteMatch,
    HTTPPathMatch,
    PathMatchType,
    GRPCMethodMatch,
    GRPCRouteMatch,
    to_gateway_protocol,  # Helper for charm-side use
)

class MyCharm(CharmBase):
  def __init__(self, *args):
    # ...
    self.ingress = IstioIngressRouteRequirer(
        self,
        relation_name="ingress",
    )

    self.framework.observe(
        self.ingress.on.ready, self._on_ingress_ready
    )

  def _configure_ingress(self):
      # Define listeners - names are auto-generated by the charm
      http_listener = Listener(port=3200, protocol=ProtocolType.HTTP)
      grpc_listener = Listener(port=9096, protocol=ProtocolType.GRPC)

      config = IstioIngressRouteConfig(
          model=self.model.name,  # Requirer's namespace where services live
          listeners=[http_listener, grpc_listener],
          http_routes=[
              HTTPRoute(
                  name="http-route",
                  listener=http_listener,
                  matches=[
                      HTTPRouteMatch(
                          path=HTTPPathMatch(type=HTTPPathMatchType.PathPrefix, value="/api"),
                          method=HTTPMethod.GET
                      )
                  ],
                  backends=[BackendRef(service=self.app.name, port=3200)],
              ),
          ],
          grpc_routes=[
              GRPCRoute(
                  name="grpc-route",
                  listener=grpc_listener,
                  matches=[
                      GRPCRouteMatch(
                          method=GRPCMethodMatch(service="myapp.MyService", method="GetData")
                      )
                  ],
                  backends=[BackendRef(service=self.app.name, port=9096)],
              ),
          ],
      )
      self.ingress.submit_config(config)

  def _on_ingress_ready(self, event):
      # Get the final external URL
      scheme = "https" if self.ingress.tls_enabled else "http"
      url = f"{scheme}://{self.ingress.external_host}"
      # Use this URL for your application configuration
```

To use the library from the provider side (istio-ingress):

```yaml
provides:
    istio-ingress-route:
        interface: istio_ingress_route
```

```python
from charms.istio_ingress_k8s.v0.istio_ingress_route import IstioIngressRouteProvider

class IstioIngressCharm(CharmBase):
  def __init__(self, *args):
    # ...
    self.istio_ingress_route = IstioIngressRouteProvider(
        self,
        external_host=self._external_host,
        tls_enabled=self._is_tls_enabled(),
    )

    self.framework.observe(
        self.istio_ingress_route.on.ready, self._handle_istio_ingress_route_ready
    )

    def _handle_istio_ingress_route_ready(self, event):
        config = self.istio_ingress_route.get_config(event.relation)
        if not config:
            return

        # Transform listeners based on TLS availability
        is_tls_enabled = self._is_tls_enabled()
        for listener in config.listeners:
            gateway_protocol = to_gateway_protocol(listener.protocol, is_tls_enabled)
            # Use gateway_protocol to create Gateway listeners
            # Create HTTPRoutes and GRPCRoutes from config
```
"""

import logging
from abc import ABC
from enum import Enum
from typing import List, Optional

from ops.charm import CharmBase, CharmEvents, RelationEvent
from ops.framework import EventSource, Object, StoredState
from ops.model import Relation
from pydantic import BaseModel, Field, field_validator


# The unique Charmhub library identifier, never change it
LIBID = "3ae88161e5c34ba4b8392c93ef7d12ce"

# Increment this major API version when introducing breaking changes
LIBAPI = 0

# Increment this PATCH version before using `charmcraft publish-lib` or reset
# to 0 if you are raising the major API version
LIBPATCH = 2

log = logging.getLogger(__name__)


# -------------------------------------------------------------------
# Exceptions
# -------------------------------------------------------------------
class IstioIngressRouteException(RuntimeError):
    """Base class for exceptions raised by IstioIngressRoute."""


class UnauthorizedError(IstioIngressRouteException):
    """Raised when the unit needs the leader to perform some action."""


# -------------------------------------------------------------------
# Enums and Helper Functions
# -------------------------------------------------------------------
class ProtocolType(str, Enum):
    """Application-level protocol types.

    Consumers specify the application protocol (HTTP or GRPC).
    The istio-ingress charm automatically applies TLS encryption based on
    certificate availability, upgrading HTTP→HTTPS and GRPC→GRPCS transparently.

    Note: The Gateway API doesn't have GRPC as a distinct protocol type.
    GRPC uses HTTP/2, so it maps to "HTTP" or "HTTPS" in Gateway listeners.
    The difference between HTTP and gRPC traffic is expressed through the
    route type (HTTPRoute vs GRPCRoute).

    Examples:
        >>> # Consumer specifies HTTP
        >>> Listener(name="api", port=8080, protocol=ProtocolType.HTTnamespace=self.model.name,  # Same namespace as ingressP)
        # If TLS available: Gateway serves HTTPS on port 8080
        # If TLS not available: Gateway serves HTTP on port 8080

        >>> # Consumer specifies GRPC
        >>> Listener(name="grpc", port=9090, protocol=ProtocolType.GRPC)
        # If TLS available: Gateway serves HTTPS (HTTP/2 + TLS) on port 9090
        # If TLS not available: Gateway serves HTTP on port 9090
    """

    HTTP = "HTTP"
    GRPC = "GRPC"

    # TODO: Extend support to L4 protocols. See https://github.com/canonical/istio-ingress-k8s-operator/issues/114.
    # TCP = "TCP"
    # UDP = "UDP"


# NOTE: `protocol` arg will be needed if the library needs to support TCP, UDP routes.
def to_gateway_protocol(protocol: ProtocolType, tls_enabled: bool = False) -> str:
    """Map application protocol to Gateway API protocol.

    The Gateway API doesn't have separate HTTP/gRPC protocol types.
    Both use HTTP, with the difference being in the route type (HTTPRoute vs GRPCRoute).

    Args:
        protocol: Application-level protocol (HTTP or GRPC)
        tls_enabled: Whether TLS termination should be applied

    Returns:
        Gateway API protocol string ("HTTP" or "HTTPS")

    Examples:
        >>> to_gateway_protocol(ProtocolType.HTTP, tls_enabled=False)
        'HTTP'
        >>> to_gateway_protocol(ProtocolType.HTTP, tls_enabled=True)
        'HTTPS'
        >>> to_gateway_protocol(ProtocolType.GRPC, tls_enabled=False)
        'HTTP'  # gRPC uses HTTP/2 (h2c when cleartext)
        >>> to_gateway_protocol(ProtocolType.GRPC, tls_enabled=True)
        'HTTPS'  # gRPC uses HTTP/2 over TLS
    """
    # HTTP and GRPC both use HTTP/2
    if tls_enabled:
        return "HTTPS"
    else:
        return "HTTP"


class HTTPMethod(str, Enum):
    """HTTP methods for route matching."""

    GET = "GET"
    POST = "POST"
    PUT = "PUT"
    DELETE = "DELETE"
    PATCH = "PATCH"
    HEAD = "HEAD"
    OPTIONS = "OPTIONS"
    CONNECT = "CONNECT"
    TRACE = "TRACE"


class HTTPPathMatchType(str, Enum):
    """Path match types for HTTP routes."""

    Exact = "Exact"
    PathPrefix = "PathPrefix"
    RegularExpression = "RegularExpression"


# -------------------------------------------------------------------
# Base Models
# -------------------------------------------------------------------
class Listener(BaseModel):
    """Gateway listener configuration.

    Specify the application-level protocol (HTTP or GRPC).
    The istio-ingress charm will automatically upgrade to TLS (HTTPS/GRPCS)
    when certificates are available.

    The listener name is automatically derived from port and protocol by the charm.
    """

    port: int = Field(..., ge=1, le=65535, description="Port number")
    protocol: ProtocolType = Field(..., description="Protocol type")
    # TODO: uncomment the below when support is added for both wildcards and using subdomains
    # hostname: Optional[str] = Field(None, description="Hostname binding for this listener")

    @property
    def name(self) -> str:
        """Get the listener name derived from protocol and port.

        Returns:
            Listener name in format: {protocol}-{port} (e.g., "http-8080", "grpc-9090")
        """
        return f"{self.protocol.value.lower()}-{self.port}"

    @property
    def gateway_protocol(self) -> str:
        """Get the Gateway API protocol (cleartext).

        This maps GRPC -> HTTP, since Gateway API doesn't have a GRPC protocol type.
        Both HTTP and gRPC use HTTP/2; the difference is in the route type.

        Returns:
            Gateway API protocol string without TLS ("HTTP")
        """
        return to_gateway_protocol(self.protocol, tls_enabled=False)


class BackendRef(BaseModel):
    """Reference to a backend service."""

    service: str = Field(..., description="Service name (in same namespace)")
    port: int = Field(..., ge=1, le=65535, description="Service port")
    weight: Optional[int] = Field(None, ge=1, le=100, description="Traffic weight")


# -------------------------------------------------------------------
# Route Match Base Classes
# -------------------------------------------------------------------
class HTTPPathMatch(BaseModel):
    """Path matching configuration for HTTP routes."""

    type: HTTPPathMatchType = Field(
        default=HTTPPathMatchType.PathPrefix,
        description="Type of path match (Exact, PathPrefix, or RegularExpression)"
    )
    value: str = Field(..., description="Path value to match")


class GRPCMethodMatch(BaseModel):
    """gRPC method matching configuration.

    Matches gRPC methods in the format /service/method where:
    - service can be a simple name (e.g., "MyService") or package-qualified (e.g., "package.MyService")
    - method is the RPC method name (optional - if omitted, matches all methods on the service)

    Examples:
        >>> GRPCMethodMatch(service="com.example.UserService")
        # Matches all methods on /com.example.UserService

        >>> GRPCMethodMatch(service="com.example.UserService", method="GetUser")
        # Matches only /com.example.UserService/GetUser

        >>> GRPCMethodMatch(service="UserService", method="CreateUser")
        # Matches /UserService/CreateUser
    """

    service: str = Field(..., description="gRPC service name (e.g., 'package.Service')")
    method: Optional[str] = Field(None, description="gRPC method name (e.g., 'GetUser'). If omitted, matches all methods on the service")


class _RouteMatch(BaseModel, ABC):
    """Base class for route match conditions."""

    headers: Optional[dict] = Field(None, description="Header matches")


class HTTPRouteMatch(_RouteMatch):
    """Match conditions for HTTP routes."""

    path: Optional[HTTPPathMatch] = Field(None, description="Path match configuration")
    method: Optional[HTTPMethod] = Field(None, description="HTTP method")


class GRPCRouteMatch(_RouteMatch):
    """Match conditions for gRPC routes."""

    method: Optional[GRPCMethodMatch] = Field(None, description="gRPC method match configuration")


# -------------------------------------------------------------------
# Route Base Classes
# -------------------------------------------------------------------
class _Route(BaseModel, ABC):
    """Base class for all routes."""

    name: str = Field(..., description="Route name")
    listener: Listener = Field(..., description="Listener this route binds to")
    backends: List[BackendRef] = Field(..., description="Backend services")

    @property
    def protocol(self) -> ProtocolType:
        """Protocol type - overridden in subclasses."""
        raise NotImplementedError


class _L7Route(_Route, ABC):
    """Base class for Layer 7 routes."""

    hostnames: Optional[List[str]] = Field(None, description="Hostnames to match")


# -------------------------------------------------------------------
# Concrete Route Classes
# -------------------------------------------------------------------
class HTTPRoute(_L7Route):
    """HTTP route configuration."""

    matches: Optional[List[HTTPRouteMatch]] = Field(None, description="HTTP match rules")

    @property
    def protocol(self) -> ProtocolType:
        """Protocol type for HTTP routes."""
        return ProtocolType.HTTP


class GRPCRoute(_L7Route):
    """gRPC route configuration."""

    matches: Optional[List[GRPCRouteMatch]] = Field(None, description="gRPC match rules")

    @property
    def protocol(self) -> ProtocolType:
        """Protocol type for gRPC routes."""
        return ProtocolType.GRPC


# -------------------------------------------------------------------
# Main Config
# -------------------------------------------------------------------
class IstioIngressRouteConfig(BaseModel):
    """Complete configuration for istio-ingress-route."""

    model: str = Field(..., description="The model (namespace) where backend services live")
    listeners: List[Listener] = Field(default_factory=list)
    http_routes: List[HTTPRoute] = Field(default_factory=list)
    grpc_routes: List[GRPCRoute] = Field(default_factory=list)


# -------------------------------------------------------------------
# Events
# -------------------------------------------------------------------
class IstioIngressRouteProviderReadyEvent(RelationEvent):
    """Event emitted when istio-ingress is ready to provide ingress for a routed unit."""


class IstioIngressRouteProviderDataRemovedEvent(RelationEvent):
    """Event emitted when a routed ingress relation is removed."""


class IstioIngressRouteRequirerReadyEvent(RelationEvent):
    """Event emitted when a unit requesting ingress has provided all data."""


class IstioIngressRouteRequirerEvents(CharmEvents):
    """Container for IstioIngressRouteRequirer events."""

    ready = EventSource(IstioIngressRouteRequirerReadyEvent)


class IstioIngressRouteProviderEvents(CharmEvents):
    """Container for IstioIngressRouteProvider events."""

    ready = EventSource(IstioIngressRouteProviderReadyEvent)
    data_removed = EventSource(IstioIngressRouteProviderDataRemovedEvent)


# -------------------------------------------------------------------
# Provider
# -------------------------------------------------------------------
class IstioIngressRouteProvider(Object):
    """Implementation of the provider of istio_ingress_route.

    This will be owned by the istio-ingress charm.
    The main idea is that istio-ingress will observe the `ready` event and, upon
    receiving it, will fetch the config from the requirer's application databag,
    apply it (create Gateway listeners and Routes), and update its own app databag
    to let the requirer know that the ingress is ready.
    """

    on = IstioIngressRouteProviderEvents()  # pyright: ignore
    _stored = StoredState()

    def __init__(
        self,
        charm: CharmBase,
        relation_name: str = "istio-ingress-route",
        external_host: str = "",
        *,
        tls_enabled: bool = False,
    ):
        """Constructor for IstioIngressRouteProvider.

        Args:
            charm: The charm that is instantiating the instance.
            relation_name: The name of the relation to bind to
                (defaults to "istio-ingress-route").
            external_host: The external host.
            tls_enabled: Whether TLS is enabled on the gateway.
        """
        super().__init__(charm, relation_name)
        self._stored.set_default(external_host=None, tls_enabled=None)

        self._charm = charm
        self._relation_name = relation_name

        if (
            self._stored.external_host != external_host  # pyright: ignore
            or self._stored.tls_enabled != tls_enabled  # pyright: ignore
        ):
            # If istio-ingress endpoint details changed, update
            self.update_ingress_address(external_host=external_host, tls_enabled=tls_enabled)

        self.framework.observe(
            self._charm.on[relation_name].relation_changed, self._on_relation_changed
        )
        self.framework.observe(
            self._charm.on[relation_name].relation_broken, self._on_relation_broken
        )

    @property
    def external_host(self) -> str:
        """Return the external host set by istio-ingress, if any."""
        self._update_stored()
        return self._stored.external_host or ""  # type: ignore

    @property
    def tls_enabled(self) -> bool:
        """Return whether TLS is enabled on the gateway."""
        self._update_stored()
        return self._stored.tls_enabled or False  # type: ignore

    @property
    def relations(self):
        """The list of Relation instances associated with this endpoint."""
        return list(self._charm.model.relations[self._relation_name])

    def _update_stored(self) -> None:
        """Ensure that the stored data is up-to-date."""
        if not self._charm.unit.is_leader():
            return

        for relation in self._charm.model.relations[self._relation_name]:
            if not relation.app:
                self._stored.external_host = ""
                self._stored.tls_enabled = False
                return
            external_host = relation.data[relation.app].get("external_host", "")
            self._stored.external_host = (
                external_host or self._stored.external_host  # pyright: ignore
            )
            tls_enabled_str = relation.data[relation.app].get("tls_enabled", "False")
            self._stored.tls_enabled = tls_enabled_str == "True"

    def _on_relation_changed(self, event: RelationEvent):
        if self.is_ready(event.relation):
            self.update_ingress_address()
            self.on.ready.emit(relation=event.relation, app=event.relation.app)

    def _on_relation_broken(self, event: RelationEvent):
        self.on.data_removed.emit(relation=event.relation, app=event.relation.app)

    def update_ingress_address(
        self, *, external_host: Optional[str] = None, tls_enabled: Optional[bool] = None
    ):
        """Ensure that requirers know the external host for istio-ingress."""
        if not self._charm.unit.is_leader():
            return

        for relation in self._charm.model.relations[self._relation_name]:
            relation.data[self._charm.app]["external_host"] = external_host or self.external_host
            tls_value = tls_enabled if tls_enabled is not None else self.tls_enabled
            relation.data[self._charm.app]["tls_enabled"] = str(tls_value)

        # We first attempt to write relation data (which may raise) and only then update stored
        # state.
        self._stored.external_host = external_host
        self._stored.tls_enabled = tls_enabled

    def wipe_ingress_data(self, relation: Relation):
        """Clear ingress data from relation.

        This removes the external_host and tls_enabled fields from the provider's
        application databag for the given relation. This is typically used when
        route conflicts are detected or when the ingress should no longer be available.

        Args:
            relation: The relation to clear data from
        """
        if not self._charm.unit.is_leader():
            log.debug("wipe_ingress_data: not leader, skipping")
            return

        try:
            relation.data[self._charm.app].pop("external_host", None)
            relation.data[self._charm.app].pop("tls_enabled", None)
        except Exception as e:
            log.warning(
                f"Error {e} clearing ingress data for relation {relation.name}. "
                "This may be a ghost of a dead relation."
            )

    def is_ready(self, relation: Relation) -> bool:
        """Whether IstioIngressRoute is ready on this relation.

        Returns True when the remote app shared the config; False otherwise.
        """
        if not relation.app or not relation.data[relation.app]:
            return False
        return "config" in relation.data[relation.app]

    def get_config(self, relation: Relation) -> Optional[IstioIngressRouteConfig]:
        """Retrieve the config published by the remote application."""
        if not self.is_ready(relation):
            return None

        config_json = relation.data[relation.app].get("config")
        if not config_json:
            return None

        try:
            return IstioIngressRouteConfig.model_validate_json(config_json)
        except Exception as e:
            log.error(f"Failed to parse config from {relation}: {e}")
            return None


# -------------------------------------------------------------------
# Requirer
# -------------------------------------------------------------------
class IstioIngressRouteRequirer(Object):
    """Handles the requirer side of the istio-ingress-route interface.

    This class provides an API for publishing routing configurations
    to the istio-ingress charm through the `istio-ingress-route` relation.
    """

    on = IstioIngressRouteRequirerEvents()  # pyright: ignore
    _stored = StoredState()

    def __init__(
        self,
        charm: CharmBase,
        relation_name: str = "ingress",
    ):
        """Constructor for IstioIngressRouteRequirer.

        Args:
            charm: The charm that is instantiating the instance.
            relation_name: The name of the relation to bind to (defaults to "ingress").
        """
        super().__init__(charm, relation_name)
        self._stored.set_default(external_host=None, tls_enabled=None)

        self._charm = charm
        self._relation_name = relation_name

        self.framework.observe(
            self._charm.on[relation_name].relation_changed, self._on_relation_changed
        )
        self.framework.observe(
            self._charm.on[relation_name].relation_broken, self._on_relation_broken
        )

    @property
    def external_host(self) -> str:
        """Return the external host set by istio-ingress, if any."""
        self._update_stored()
        return self._stored.external_host or ""  # type: ignore

    @property
    def tls_enabled(self) -> bool:
        """Return whether TLS is enabled on the gateway."""
        self._update_stored()
        return self._stored.tls_enabled or False  # type: ignore

    def _update_stored(self) -> None:
        """Ensure that the stored host is up-to-date."""
        if not self._charm.unit.is_leader():
            return

        for relation in self._charm.model.relations[self._relation_name]:
            if not relation.app:
                self._stored.external_host = ""
                self._stored.tls_enabled = False
                return
            external_host = relation.data[relation.app].get("external_host", "")
            self._stored.external_host = (
                external_host or self._stored.external_host  # pyright: ignore
            )
            tls_enabled_str = relation.data[relation.app].get("tls_enabled", "False")
            self._stored.tls_enabled = tls_enabled_str == "True"

    def _on_relation_changed(self, event: RelationEvent) -> None:
        """Update StoredState with external_host and other information from istio-ingress."""
        self._update_stored()
        if self._charm.unit.is_leader():
            self.on.ready.emit(relation=event.relation, app=event.relation.app)

    def _on_relation_broken(self, event: RelationEvent) -> None:
        """On RelationBroken, clear the stored data if set and emit an event."""
        self._stored.external_host = ""
        self._stored.tls_enabled = False
        if self._charm.unit.is_leader():
            self.on.ready.emit(relation=event.relation, app=event.relation.app)

    def is_ready(self) -> bool:
        """Is the IstioIngressRouteRequirer ready to submit data?"""
        return len(self._charm.model.relations[self._relation_name]) > 0

    def submit_config(self, config: IstioIngressRouteConfig):
        """Submit an ingress configuration to istio-ingress.

        This method publishes routing configuration data to the
        `istio-ingress-route` relation.

        Args:
            config: The IstioIngressRouteConfig to submit.

        Raises:
            UnauthorizedError: If the unit is not the leader.
        """
        if not self._charm.unit.is_leader():
            raise UnauthorizedError()

        relations = self._charm.model.relations[self._relation_name]
        if not relations:
            log.warning(f"No relations found for {self._relation_name}")
            return

        for relation in relations:
            app_databag = relation.data[self._charm.app]
            # Serialize to JSON using Pydantic v2
            app_databag["config"] = config.model_dump_json()