flexway
FRENParler à un expert

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.

GuideExploitationDepuis la version 1.6.0

Markdown

Sur cette page

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.

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.

É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) :

# 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

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.<cluster>.<sni_domain> pour l'amorçage, broker-<id>.<cluster>.<sni_domain> pour chacun ; Metadata et FindCoordinator répondent ces noms. Le certificat couvre *.<cluster>.<sni_domain> 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).

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 :

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 :

// 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 }))
# 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
# 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, <release>-flexway-kafka-mesh, toujours ClusterIP ; le rendu échoue sans kafka.networkPolicy.from (les pods ou namespaces du maillage).

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 } } } ]

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 <release>-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).

  2. Un client liste ses sujets sous leur nom court, et seulement ceux que ses opérations lui ouvrent :

    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).

  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.

À 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), 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), sous les noms de Les noms des groupes et des transactions.

Étapes suivantes