Quotas des clients Kafka

Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la fonctionnalité côté produit : guide produit ; l'appeler : guide développeur.

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). Un client qui ignore le champ - librdkafka ne l'a jamais appliqué (librdkafka#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) : 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 :

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 :

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) ; un quota du courtier sur ce principal freine tous les tenants ensemble, ceux de la passerelle chacun le sien :

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 :

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 :

# 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 :

# 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
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

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.