# Quotas des clients Kafka

> Freiner producteurs et consommateurs Kafka par principal et par tenant, sans jamais couper, et lire les signaux de l'écouteur.

Guide - Depuis la version 1.6.0 - Flexway v1.7.1 - https://flexway.io/docs/operations-kafka-quotas/

> Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la fonctionnalité côté produit : [guide produit](https://flexway.io/docs/product-overview/) ; l'appeler : [guide développeur](https://flexway.io/docs/developer-overview/).

Les quotas de l'écouteur Kafka se déclarent sous `listeners.kafka.quotas` ; ses signaux disent ensuite qui est freiné, refusé ou fermé, et pourquoi.

## Étapes

### Quotas : l'étranglement de KIP-219

Octets produits, octets lus et requêtes par seconde, pour chaque **principal vérifié** et pour chaque
**tenant** (`listeners.kafka.quotas`). Jamais une coupure : au-delà, la réponse
part **tout de suite** avec `throttle_time_ms`, puis la passerelle **ne lit plus** la connexion du client
pendant ce délai ([KIP-219](https://cwiki.apache.org/confluence/display/KAFKA/KIP-219+-+Improve+quota+communication)). Un client qui ignore le
champ - librdkafka ne l'a jamais appliqué ([librdkafka#5592](https://github.com/confluentinc/librdkafka/issues/5592)) -
est freiné quand même : ce qu'il envoie attend dans la socket. Sans `quotas`, rien n'est étranglé.

| Clé | Pourquoi | Défaut |
| --- | --- | --- |
| `client.producer_byte_rate` | Octets produits par seconde (requêtes `Produce`, en-têtes compris) par un principal : le sujet du jeton, la clé d'API ou le certificat - **jamais** le `client.id`, qu'un client choisit librement | aucune borne |
| `client.consumer_byte_rate` | Octets lus par seconde (réponses `Fetch`) par un principal | aucune borne |
| `client.request_rate` | Requêtes par seconde d'un principal, toutes API, une fois authentifié | aucune borne |
| `tenant.producer_byte_rate`, `tenant.consumer_byte_rate`, `tenant.request_rate` | Les mêmes pour tous les principaux d'un tenant ensemble : un tenant n'achète pas plus de débit en multipliant les clés | aucune borne |
| `window` | La rafale admise d'un coup : un seau tient `débit x window` (les fenêtres de quota de Kafka font environ une seconde) | `1s` (de `100ms` à `10s`) |
| `max_throttle` | Plafond de `throttle_time_ms` - et du silence qui suit : sous le `request.timeout.ms` des clients (30 s par défaut chez Kafka), qu'un étranglement ne doit jamais faire expirer | `10s` (de `100ms` à `30s` ; défaut provisoire, à fixer par les mesures de performance) |
| `quota` | Le nom d'une politique `quota` ([Poser des quotas](https://flexway.io/docs/operations-quotas/)) : le budget **jour ou mois** de requêtes `Produce` du tenant, compté sous ce nom comme en HTTP ; épuisé, le `Produce` est refusé `POLICY_VIOLATION` (44, non réessayable) et rien n'est écrit | aucun |

Toutes ces clés sont rechargeables : le débit en vigueur s'applique à la requête suivante, ce qu'un principal a dépensé est gardé. Avec `server.counters`,
les seaux sont partagés entre répliques comme ceux des limites HTTP : chaque réplique décide seule et se
réconcilie toutes les `sync` (un débit tenu à une fenêtre `sync` près ; écart pas encore mesuré). Sans,
chaque réplique compte pour elle.

| Cas | Ce qui se passe |
| --- | --- |
| Client qui respecte KIP-219 (Java, franz-go) | Réponse **immédiate** avec `throttle_time_ms` ; il attend ce délai avant sa requête suivante |
| Client qui ignore le champ (librdkafka : confluent-kafka-go, Python, .NET) | Même réponse ; il renvoie aussitôt, mais sa requête suivante n'est servie qu'à l'échéance : le débit tient sans sa coopération |
| `Fetch` d'un principal ou d'un tenant déjà en dette | Réponse **vide** immédiate avec `throttle_time_ms`, rien n'est demandé au courtier |
| Courtier qui étrangle aussi la passerelle (ses propres quotas sur le principal de la passerelle) | `throttle_time_ms` = le **plus grand** des deux délais ; le silence du courtier est respecté : rien ne lui est envoyé sur cette connexion avant son terme |
| Quota jour ou mois du tenant épuisé | Refus `POLICY_VIOLATION` sur `Produce` (cause `quota`), pas un étranglement ; les lectures (`Fetch`) ne sont **pas encore** comptées au budget : leur code de refus doit d'abord être vérifié contre les principales bibliothèques clientes (un consommateur Java lève une exception sur un code inattendu) |

### Par tenant

Un tenant ne prend pas toute la bande passante des courtiers, quel que soit son nombre de clés :

```yaml
listeners:
  kafka:
    quotas:
      tenant: { producer_byte_rate: 10MiB, consumer_byte_rate: 40MiB, request_rate: 2000 }
```

### Par client

Chaque principal (jeton, clé, certificat) a son seau, sous le seau de son tenant :

```yaml
listeners:
  kafka:
    quotas:
      client: { producer_byte_rate: 1MiB, consumer_byte_rate: 4MiB, request_rate: 200 }
      tenant: { producer_byte_rate: 10MiB, consumer_byte_rate: 40MiB }
      window: 2s           # des rafales de deux secondes de débit
      max_throttle: 5s     # des clients au request.timeout.ms court
server:
  counters: { store: "${COUNTERS_URL}" }   # le débit tenu sur toutes les répliques ensemble
```

### Avec les quotas du courtier

Rien à déclarer : la passerelle relaie le plus grand des deux délais. Le
courtier voit le principal de la passerelle (`upstreams.<nom>.sasl.username`) et l'identifiant client du
tenant (`acme.app-1`, [Les noms des groupes et des transactions](https://flexway.io/docs/operations-kafka-reference/#les-noms-des-groupes-et-des-transactions)) ; un quota du courtier sur
ce principal freine **tous** les tenants ensemble, ceux de la passerelle chacun le sien :

```bash
kafka-configs --bootstrap-server kafka-0.kafka:9093 --command-config admin.properties \
  --alter --entity-type users --entity-name flexway --add-config producer_byte_rate=104857600
```

### Budget jour ou mois

Le même forfait qu'en HTTP, compté sur les `Produce` :

```yaml
listeners:
  kafka:
    quotas: { quota: plans }
server:
  counters: { store: "${COUNTERS_URL}" }   # exigé par toute politique quota
policies:
  definitions:
    plans: { type: quota, config: { limits: { day: 100000, month: 2000000 } } }
```

## Vérifier

### Un client librdkafka

Rien à changer côté client ; ce qu'il voit au-delà du quota :

```properties
# confluent-kafka-go, Python, .NET : des propriétés ordinaires
bootstrap.servers=bootstrap.shop.kafka.example.com:9093
security.protocol=SASL_SSL
sasl.mechanisms=PLAIN
sasl.username=acme-orders-producer
sasl.password=${FLEXWAY_KAFKA_KEY}
# au-delà du quota : le producteur avance au débit déclaré ; sa statistique "throttle" (statistics.interval.ms)
# montre throttle_time_ms ; request.timeout.ms (30 s) reste au-dessus de max_throttle
```

### Lire les signaux

Noms Prometheus :

```promql
# part des requêtes étranglées, par cluster virtuel
sum by (cluster) (rate(flexway_kafka_requests_total{decision="throttled"}[5m]))
  / sum by (cluster) (rate(flexway_kafka_requests_total[5m]))
# refus par cause (hors quota jour ou mois)
sum by (cause) (rate(flexway_kafka_requests_total{decision="refused", cause!="quota"}[5m]))
# connexions refusées par borne ; fermetures normales à part
sum by (bound) (rate(flexway_kafka_connections_refused_total[5m]))
sum by (cause) (rate(flexway_kafka_connections_closed_total[5m]))
# attente p95 de la part d'un tenant
histogram_quantile(0.95, sum by (le, tenant) (rate(flexway_kafka_tenant_budget_wait_seconds_bucket[5m])))
```

## Signaux : métriques, traces, audit

Les métriques `flexway.kafka.*` partent en OTLP avec les autres (même chemin, `service.instance.id` par
réplique) ; chaque dimension est **fermée** : clusters du contrat, noms de la table des API, jeux ci-dessous,
et `tenant` sous la règle de toutes les dimensions de tenant - 200 tenants gardent leur série, les
suivants partagent `_other`. **Jamais** un sujet, une clé, un certificat, un `client.id`, un sujet Kafka
physique ni une adresse.

| Métrique | Dimensions | Ce qu'elle dit |
| --- | --- | --- |
| `flexway.kafka.connections` (jauge) | `cluster`, `tenant`, `mechanism` (`oauthbearer`, `api_key`, `mtls`, `none` avant l'authentification ; `_none` et `_anonymous` avant le SNI) | Connexions ouvertes : sur une réplique, l'occupation de `max_connections` - qui ne refuse rien, une connexion de plus attend dans la file du noyau |
| `flexway.kafka.requests` | `cluster`, `tenant`, `api`, `decision` (`allowed`, `throttled`, `refused`), `cause` (`acl`, `admin_api`, `quota`, `unknown_topic`, `authentication`, `kms_unavailable`, `unsupported`, `key_destroyed` ; `none`) | Requêtes jugées ; `refused` dès qu'une partie est refusée par la passerelle |
| `flexway.kafka.bytes` | `cluster`, `tenant`, `direction` (`produce`, `fetch`) | Les octets que comptent les quotas |
| `flexway.kafka.throttle` (histogramme, s) | `cluster`, `tenant`, `quota` (`producer_byte_rate`, `consumer_byte_rate`, `request_rate`, `broker`), `scope` (`client`, `tenant`, `broker`) | `throttle_time_ms` de chaque réponse étranglée |
| `flexway.kafka.connections.refused` | `bound` (`max_unauthenticated_connections`, `max_connections_per_ip`, `max_connections_per_tenant`, `max_connection_rate_per_tenant`), `tenant` pour les deux bornes du tenant seulement | Connexions refusées par une borne ; jamais avant l'authentification un tenant, jamais une adresse ; le refus par `max_connections_per_tenant` n'est **pas** compté aussi en `authentication` |
| `flexway.kafka.connections.closed` | `cause` (`auth_timeout` sans tenant, `idle_timeout`, `max_session_lifetime`), `tenant` | Fermetures **normales**, jamais mêlées aux refus ; `max_session_lifetime` : session arrivée à son terme (la borne, ou `exp`, ou `NotAfter`) sans ré-authentification |
| `flexway.kafka.tenant_budget` | `cluster`, `tenant`, `outcome` (`waited`, `closed`) | Réponses qui ont attendu la part du tenant (`max_tenant_in_flight_bytes`, mode `wait`) ; connexions fermées à elle (mode `close`) |
| `flexway.kafka.tenant_budget.wait` (histogramme, s) | `cluster`, `tenant` | Combien de temps une réponse a attendu cette part |
| `flexway.kafka.encryption` | `cluster`, `tenant`, `op` (`encrypt`, `decrypt`), `outcome` (`ok`, `kms_unavailable`, `key_destroyed`) | [Chiffrement par champ](https://flexway.io/docs/operations-kafka-encryption/) |
| `flexway.kafka.upstream.connections` | `cluster`, `broker` (identifiant de nœud, ou `bootstrap`), `outcome` (`ok`, `failed`) | Connexions de la passerelle à chaque courtier : la santé par courtier |

| Clé | Pourquoi | Défaut |
| --- | --- | --- |
| `observability.kafka.budget_wait_buckets` | Bornes de `flexway.kafka.tenant_budget.wait`, croissantes, 20 au plus : à resserrer quand les attentes normales d'un parc tiennent sous la milliseconde. **Non rechargeable** (les bornes d'un histogramme se fixent à sa création) | `1ms`, `5ms`, `10ms`, `50ms`, `100ms`, `500ms`, `1s`, `5s`, `10s`, `30s` (le délai d'écriture) |
| `observability.kafka.closures` | Compter les fermetures normales : `false` sur un parc de clients éphémères où elles ne disent rien. Rechargeable | `true` |

```yaml
observability:
  kafka:
    budget_wait_buckets: [100us, 1ms, 10ms, 100ms, 1s, 10s, 30s]
    closures: false
```

### Traces

Un span par requête **authentifiée** échantillonnée (`kafka <API>`, attributs `kafka.cluster`,
`tenant`, `auth.mechanism`, `kafka.decision`, `kafka.cause`, `kafka.throttle_time_ms`) - jamais un sujet ni
un nom de sujet Kafka.

### Audit

Chaque requête refusée et chaque requête
d'**administration** (permise ou refusée), dans la même chaîne que HTTP - `host` = cluster virtuel,
`method` = `KAFKA`, `operation` = l'API, `code` = la cause, le sujet haché ; les requêtes de données
permises sont comptées, pas journalisées une à une.