# Déclarer un cluster virtuel Kafka > Ouvrir un cluster virtuel aux clients Kafka : contrat AsyncAPI, écouteur TLS ou repli du maillage, amonts, clients et chart. Guide - Depuis la version 1.6.0 - Flexway v1.7.1 - https://flexway.io/docs/operations-kafka-virtual-clusters/ > 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/). Le contrat AsyncAPI déclare les sujets servis, `flexway.yaml` l'écouteur et l'amont, les clients pointent sur la passerelle. Ce que voit chaque tenant : [Clients Kafka natifs](https://flexway.io/docs/operations-kafka-clients/). ## Étapes ### Déclarer le cluster au contrat Une opération AsyncAPI marquée `x-flexway-kafka-proxy` est servie aux clients Kafka du cluster qu'elle nomme ; `flexway.yaml` déclare l'écouteur `listeners.kafka` et l'amont. L'exemple complet, ici dans un maillage (un port par courtier) : ```yaml # Kafka clients behind the gateway (1.6.0): producers, consumers and admin tools of any # library reach the virtual cluster "shop" the AsyncAPI contract declares - each tenant inside its own # names, under the same identities as HTTP. Here the fallback inside a mesh, a port per broker; the TLS # listener, brokers named by SNI, is in the guide. version: 1 contracts: asyncapi: ./asyncapi.yaml upstreams: events: protocol: kafka brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093] tls: { server_name: kafka.internal } sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" } timeout: 5s # a connection to a broker, TLS and SASL included auth: issuers: - name: customers discovery: https://login.example.net/.well-known/openid-configuration audience: kafka tenant_claim: org # every topic, group and transaction is the tenant's api_keys: store: "file:./api-keys.yaml" # PLAIN: the key's id as the user, the key as the password listeners: kafka: plaintext: mesh_only: true # clear text only where the mesh encrypts between pods advertised_host: flexway-kafka.flexway.svc.cluster.local clusters: shop: { ports: { from: 19092, to: 19095 }, first_node_id: 0 } # bootstrap 19092, node n at 19093+n ``` ```yaml title="asyncapi.yaml" # examples/kafka-proxy/asyncapi.yaml asyncapi: 3.0.0 info: { title: Shop events, version: "1.0.0" } channels: orders: address: "{tenant}.orders" parameters: tenant: { x-flexway-bind: tenant } messages: order: { payload: { type: object, required: [id], properties: { id: { type: string } } } } operations: produceOrders: action: send channel: { $ref: "#/channels/orders" } x-flexway-upstream: events x-flexway-kafka-proxy: { cluster: shop, transactions: true } security: - $ref: "#/components/securitySchemes/apiKey" - $ref: "#/components/securitySchemes/ordersWrite" consumeOrders: action: receive channel: { $ref: "#/channels/orders" } x-flexway-upstream: events x-flexway-kafka-proxy: { cluster: shop, admin_scope: "orders:admin" } security: - $ref: "#/components/securitySchemes/apiKey" - $ref: "#/components/securitySchemes/ordersRead" components: securitySchemes: apiKey: { type: plain } # SASL PLAIN: an API key bound to the operation ordersWrite: # SASL OAUTHBEARER: a token carrying the scope type: oauth2 flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:write": Produce orders } } } scopes: ["orders:write"] ordersRead: type: oauth2 flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "orders:read": Read orders } } } scopes: ["orders:read"] ``` ### L'écouteur TLS, les courtiers nommés par SNI Un seul port, le nom SNI dit quel courtier est visé - `bootstrap..` pour l'amorçage, `broker-..` pour chacun ; `Metadata` et `FindCoordinator` répondent ces noms. Le certificat couvre `*..` pour chaque cluster virtuel (un joker ne couvre qu'une étiquette) ; tous ces noms résolvent vers la passerelle (un enregistrement DNS joker, ou le `Service` `LoadBalancer` du chart). ```yaml title="flexway.yaml" # flexway.yaml - l'écouteur TLS ; le reste comme l'exemple ci-dessus listeners: kafka: address: ":9093" tls: cert_file: /etc/flexway-kafka-tls/tls.crt # *.shop.kafka.example.com key_file: /etc/flexway-kafka-tls/tls.key client_ca_file: /etc/flexway-kafka-tls/ca.crt # facultatif : les certificats clients (X509) sni_domain: kafka.example.com advertised_port: 9093 # le port que voit le client (celui du LoadBalancer) max_request_bytes: 16MiB # une requête authentifiée ; 64 Kio avant l'authentification max_response_bytes: 64MiB # une réponse du courtier ; un Fetch est tenu sous la moitié max_connections: 4096 auth_timeout: 10s idle_timeout: 10m auth: mtls: clients: - { id: billing, subject: "CN=billing-app", tenant: acme, operations: [consumeOrders] } ``` ### Le repli en clair, un port par courtier Sans TLS, SNI n'existe pas ; chaque cluster virtuel reçoit une plage - `from` pour l'amorçage, `from + 1 + n` pour le nœud `first_node_id + n` -, annoncée sous `advertised_host`. Refusé au démarrage sans `mesh_only: true` : le clair ne traverse qu'un maillage qui chiffre (Linkerd, Istio en mTLS strict). La plage couvre **chaque** identifiant de nœud du cluster, sinon le client reçoit une adresse injoignable et le journal le dit. ### Les amonts Apache Kafka (`apache/kafka`, Confluent Platform, MSK, Strimzi) et Redpanda se déclarent pareil ; la passerelle lit les versions de chaque courtier et n'annonce au client que l'intersection avec ce qu'elle juge : ```yaml upstreams: events: # Apache Kafka : SCRAM sur TLS protocol: kafka brokers: [kafka-0.kafka:9093, kafka-1.kafka:9093, kafka-2.kafka:9093] tls: { ca_file: /etc/flexway-secrets/kafka/ca.crt, server_name: kafka.internal } sasl: { mechanism: SCRAM-SHA-512, username: flexway, password: "file:/etc/flexway-secrets/kafka/password" } events-redpanda: # Redpanda : SASL SCRAM-SHA-256, ou mTLS de la passerelle protocol: kafka brokers: [redpanda-0.redpanda:9093] tls: { ca_file: /etc/flexway-secrets/redpanda/ca.crt, cert_file: /etc/flexway-secrets/redpanda/tls.crt, key_file: /etc/flexway-secrets/redpanda/tls.key } timeout: 5s # une connexion au courtier, TLS et SASL compris ``` ### Les clients Des propriétés Kafka standard, pointées sur la passerelle : ```go // franz-go : une clé d'API en SASL PLAIN sur TLS cl, err := kgo.NewClient( kgo.SeedBrokers("bootstrap.shop.kafka.example.com:9093"), kgo.DialTLSConfig(&tls.Config{MinVersion: tls.VersionTLS12}), kgo.SASL(plain.Auth{User: "acme-orders-producer", Pass: os.Getenv("FLEXWAY_KAFKA_KEY")}.AsMechanism()), kgo.TransactionalID("tx-orders"), // devient acme.~t.tx-orders chez le courtier ) // un jeton : kgo.SASL(oauth.Oauth(func(ctx context.Context) (oauth.Auth, error) { return oauth.Auth{Token: tok}, nil })) ``` ```properties # librdkafka (confluent-kafka-go, Python, .NET) : un jeton OAUTHBEARER de l'émetteur déclaré bootstrap.servers=bootstrap.shop.kafka.example.com:9093 security.protocol=SASL_SSL sasl.mechanisms=OAUTHBEARER sasl.oauthbearer.method=oidc sasl.oauthbearer.client.id=billing sasl.oauthbearer.client.secret=${BILLING_SECRET} sasl.oauthbearer.token.endpoint.url=https://login.example.net/token sasl.oauthbearer.scope=orders:read group.id=billing # devient acme.~g.billing ``` ```properties # Client Java (producteur, consommateur, AdminClient, Kafka Connect) : une clé d'API en PLAIN bootstrap.servers=bootstrap.shop.kafka.example.com:9093 security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="acme-orders-reader" password="${FLEXWAY_KAFKA_KEY}"; # un certificat client à la place : security.protocol=SSL, ssl.keystore.location=..., ssl.truststore.location=... # group.protocol=classic (par défaut) : le protocole de groupe de KIP-848 n'est pas servi ``` ### Le chart : clair dans le maillage seulement Les ports en clair vont dans un `Service` à part, `-flexway-kafka-mesh`, **toujours** `ClusterIP` ; le rendu échoue sans `kafka.networkPolicy.from` (les pods ou namespaces du maillage). ```yaml title="values.yaml" # rendu : le port TLS sur un LoadBalancer, la plage en clair réservée au maillage config: listeners: kafka: address: ":9093" tls: { cert_file: /etc/flexway-kafka-tls/tls.crt, key_file: /etc/flexway-kafka-tls/tls.key } sni_domain: kafka.example.com plaintext: { mesh_only: true, advertised_host: orders-flexway-kafka-mesh.apps.svc.cluster.local, clusters: { shop: { ports: { from: 19092, to: 19095 } } } } kafka: service: { type: LoadBalancer } tls: { clusters: [shop], issuerRef: { name: kafka-ca, kind: ClusterIssuer } } networkPolicy: from: [ { namespaceSelector: { matchLabels: { kubernetes.io/metadata.name: apps } } } ] ``` ```yaml title="values.yaml" # refusé : du clair sans pairs déclarés kafka: networkPolicy: { from: [] } # Error: execution error at (flexway/templates/kafka.yaml): listeners.kafka.plaintext opens ports in # clear: kafka.networkPolicy.enabled and kafka.networkPolicy.from (the mesh's pods or namespaces) are # required - clear text is for the mesh only ``` Le chart rend aussi le `Service` `-flexway-kafka`, sa `NetworkPolicy` et, avec cert-manager, le certificat à joker de chaque cluster (`kafka.tls.clusters`). ## Vérifier 1. `flexway validate` accepte la configuration ; un refus nomme l'extension ou la clé fautive ([Refusé au chargement](https://flexway.io/docs/operations-kafka-reference/#refus%C3%A9-au-chargement)). 2. Un client liste ses sujets sous leur nom court, et seulement ceux que ses opérations lui ouvrent : ```bash kafka-topics --bootstrap-server bootstrap.shop.kafka.example.com:9093 --command-config client.properties --list # orders ``` 3. Le courtier, lui, tient `acme.orders`, et `acme.~g.billing` pour le groupe `billing` ([Les noms des groupes et des transactions](https://flexway.io/docs/operations-kafka-reference/#les-noms-des-groupes-et-des-transactions)). 4. Le `/config` d'administration publie les clusters virtuels sous `kafka`. 5. Un client refusé se lit dans le journal `kafka authentication refused`, avec sa cause et jamais la clé ; les autres cas : [Dépannage](https://flexway.io/docs/operations-kafka-reference/#d%C3%A9pannage). ## À savoir ### Rechargement L'adresse, TLS, `sni_domain`, `advertised_port`, la plage de repli et `max_connections`, `max_total_in_flight_bytes` et `max_tenant_in_flight_bytes` sont lus au démarrage (`listeners.kafka` refusé au rechargement s'ils changent) ; le contrat, les identités, les bornes et les délais s'appliquent à la requête suivante - une connexion garde son identité tant que son justificatif reste celui de la configuration en vigueur ([Sessions : révocation, durée, certificats](https://flexway.io/docs/operations-kafka-reference/#sessions--r%C3%A9vocation-dur%C3%A9e-certificats)), et chaque requête est jugée sur celle-ci. ### ACL du courtier Le principal de la passerelle n'a besoin que des sujets, groupes et identifiants transactionnels des tenants servis ([Une connexion amont par connexion cliente et par courtier](https://flexway.io/docs/operations-kafka-clients/#une-connexion-amont-par-connexion-cliente-et-par-courtier)), sous les noms de [Les noms des groupes et des transactions](https://flexway.io/docs/operations-kafka-reference/#les-noms-des-groupes-et-des-transactions).