flexway
FRENTalk to an expert

Declare a Kafka virtual cluster

Open a virtual cluster to Kafka clients: AsyncAPI contract, TLS listener or mesh fallback, upstreams, clients and chart.

GuideOperationsSince version 1.6.0

Markdown

On this page

Operations guide: configure, deploy and observe the gateway. What the feature does, product side: product guide; calling it: developer guide.

The AsyncAPI contract declares the topics served, flexway.yaml the listener and the upstream, the clients point at the gateway. What each tenant sees: Native Kafka clients.

Steps

Declare the cluster in the contract

An AsyncAPI operation marked x-flexway-kafka-proxy is served to the Kafka clients of the cluster it names; flexway.yaml declares the listeners.kafka listener and the upstream. The full example, here inside a mesh (a port per broker):

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

The TLS listener, brokers named by SNI

One port, the SNI name says which broker is meant - bootstrap.<cluster>.<sni_domain> to bootstrap, broker-<id>.<cluster>.<sni_domain> for each; Metadata and FindCoordinator answer these names. The certificate covers *.<cluster>.<sni_domain> for each virtual cluster (a wildcard covers one label); every such name resolves to the gateway (a wildcard DNS record, or the chart's LoadBalancer Service).

flexway.yaml

# flexway.yaml - the TLS listener; the rest as in the example above
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    # optional: client certificates (X509)
    sni_domain: kafka.example.com
    advertised_port: 9093                # the port the client sees (the LoadBalancer's)
    max_request_bytes: 16MiB             # an authenticated request; 64 KiB before authentication
    max_response_bytes: 64MiB            # a broker's answer; a Fetch is held below half of it
    max_connections: 4096
    auth_timeout: 10s
    idle_timeout: 10m
auth:
  mtls:
    clients:
      - { id: billing, subject: "CN=billing-app", tenant: acme, operations: [consumeOrders] }

The plaintext fallback, a port per broker

Without TLS there is no SNI; each virtual cluster gets a range - from to bootstrap, from + 1 + n for node first_node_id + n -, announced under advertised_host. Refused at start without mesh_only: true: clear text only crosses a mesh that encrypts (Linkerd, Istio in strict mTLS). The range covers every node id of the cluster; otherwise the client gets an unreachable address and the log says so.

Upstreams

Apache Kafka (apache/kafka, Confluent Platform, MSK, Strimzi) and Redpanda are declared alike; the gateway reads each broker's versions and announces to the client only their intersection with what it judges:

upstreams:
  events:                                # Apache Kafka: SCRAM over 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, or the gateway's mTLS
    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                          # a connection to a broker, TLS and SASL included

Clients

Standard Kafka properties, pointed at the gateway:

// franz-go: an API key over SASL PLAIN on 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"), // becomes acme.~t.tx-orders at the broker
)
// a token: kgo.SASL(oauth.Oauth(func(ctx context.Context) (oauth.Auth, error) { return oauth.Auth{Token: tok}, nil }))
# librdkafka (confluent-kafka-go, Python, .NET): an OAUTHBEARER token of the declared issuer
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                         # becomes acme.~g.billing
# Java client (producer, consumer, AdminClient, Kafka Connect): an API key over 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}";
# a client certificate instead: security.protocol=SSL, ssl.keystore.location=..., ssl.truststore.location=...
# group.protocol=classic (the default): KIP-848's group protocol is not served

The chart: clear text inside the mesh only

The plaintext ports go into a Service of their own, <release>-flexway-kafka-mesh, always ClusterIP; the render fails without kafka.networkPolicy.from (the mesh's pods or namespaces).

values.yaml

# renders: the TLS port on a LoadBalancer, the plaintext range kept for the mesh
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

# refused: clear text with no declared peers
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

The chart also renders the <release>-flexway-kafka Service, its NetworkPolicy and, with cert-manager, each cluster's wildcard certificate (kafka.tls.clusters).

Verify

  1. flexway validate accepts the configuration; a refusal names the faulty extension or key (Refused at load).

  2. A client lists its topics under their short names, and only those its operations open:

    kafka-topics --bootstrap-server bootstrap.shop.kafka.example.com:9093 --command-config client.properties --list
# orders

  3. The broker holds acme.orders, and acme.~g.billing for the group billing (Names of groups and transactions).

  4. The admin /config publishes the virtual clusters under kafka.

  5. A refused client shows in the log kafka authentication refused, with its cause and never the key; the other cases: Troubleshooting.

Good to know

Reload

The address, TLS, sni_domain, advertised_port, the fallback range and max_connections, max_total_in_flight_bytes and max_tenant_in_flight_bytes are read at start (listeners.kafka is refused on reload if they change); the contract, identities, bounds and timeouts apply to the next request - a connection keeps its identity as long as its credential stays the configuration's in force (Sessions: revocation, lifetime, certificates), and each request is judged on it.

Broker ACLs

The gateway's principal only needs the topics, groups and transactional ids of the tenants served (One upstream connection per client connection and broker), under the names of Names of groups and transactions.

Next steps