Deploying and Managing Streams for Apache Kafka Proxy on OpenShift

A KafkaProxy resource represents an instance of the Streams for Apache Kafka Proxy. Conceptually, it is the top-level resource that links together KafkaProxyIngress, VirtualKafkaCluster, KafkaService, and KafkaProtocolFilter resources to form a complete working proxy.

KafkaProxy resources are referenced by KafkaProxyIngress and VirtualKafkaCluster resources to define how the proxy is exposed and what it proxies.

kind: KafkaProxy
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: simple
spec: {} 

kind: KafkaProxy
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: simple
spec: {} 

A KafkaProxyIngress resource defines the networking configuration that allows Kafka clients to connect to a VirtualKafkaCluster.

It is uniquely associated with a single KafkaProxy instance, but it is not uniquely associated with a VirtualKafkaCluster and can be used by multiple VirtualKafkaCluster instances.

The KafkaProxyIngress resource supports the following ingress types to configure networking access to the virtual cluster:

The clusterIP ingress types support both TCP (plain) and TLS connections. The loadBalancer type exclusively supports TLS.

When using TLS, you specify a TLS server certificate in the ingress configuration of the VirtualKafkaCluster resource.

When using loadBalancer, changes to your DNS may be required.

The following table summarizes the supported ingress types.

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: cluster-ip
spec:
  proxyRef: 
    name: simple
  clusterIP: 
    protocol: TCP 

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: cluster-ip
spec:
  proxyRef: 

    name: simple
  clusterIP: 

    protocol: TCP 

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: cluster-ip
spec:
  proxyRef:
    name: simple
  clusterIP:
    protocol: TLS 

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: cluster-ip
spec:
  proxyRef:
    name: simple
  clusterIP:
    protocol: TLS 

$(virtualClusterName).kafkaproxy.example.com

$(virtualClusterName).kafkaproxy.example.com

broker-$(nodeId).$(virtualClusterName).kafkaproxy.example.com

broker-$(nodeId).$(virtualClusterName).kafkaproxy.example.com

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: load-balancer
spec:
  proxyRef: 
    name: simple
  loadBalancer: 
    bootstrapAddress: "$(virtualClusterName).kafkaproxy.example.com" 
    advertisedBrokerAddressPattern: "broker-$(nodeId).$(virtualClusterName).kafkaproxy.example.com" 

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  namespace: my-proxy
  name: load-balancer
spec:
  proxyRef: 

    name: simple
  loadBalancer: 

    bootstrapAddress: "$(virtualClusterName).kafkaproxy.example.com" 

    advertisedBrokerAddressPattern: "broker-$(nodeId).$(virtualClusterName).kafkaproxy.example.com" 

A proxied Kafka cluster is configured in a KafkaService resource, which specifies how the proxy connects to the cluster. The Kafka cluster may or may not be running in the same OpenShift cluster as the proxy: Network connectivity is all that’s required.

This example shows a KafkaService defining how to connect to a Kafka cluster at kafka.example.com.

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092 
  nodeIdRanges: 
    - name: brokers 
      start: 0 
      end: 5 
  # ...

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092 

  nodeIdRanges: 

    - name: brokers 

      start: 0 

      end: 5 

  # ...

A VirtualKafkaCluster resource defines a logical Kafka cluster that is accessible to clients over the network.

The virtual cluster references the following resources, which must be in the same namespace:

This example shows a VirtualKafkaCluster, exposing it to Kafka clients running on the same OpenShift cluster. It uses plain TCP (as opposed to TLS) as the transport protocol.

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: my-cluster
  namespace: my-proxy
spec:
  proxyRef: 
    name: simple
  targetKafkaServiceRef: 
    name: my-cluster
  ingresses:
    - ingressRef: 
        name: cluster-ip

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: my-cluster
  namespace: my-proxy
spec:
  proxyRef: 

    name: simple
  targetKafkaServiceRef: 

    name: my-cluster
  ingresses:
    - ingressRef: 

        name: cluster-ip

This example shows a VirtualKafkaCluster, exposing it to Kafka clients running both on and off-cluster, both using TLS. Because TLS is used, the ingress configuration must reference a TLS server certificate.

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: my-cluster
  namespace: my-proxy
spec:
  proxyRef:
    name: simple
  targetKafkaServiceRef:
    name: my-cluster
  ingresses:
    - ingressRef:
        name: cluster-ip
        certificateRef:
          name: 'cluster-ip-server-cert' 
          kind: Secret
    - ingressRef:
        name: load-balancer
        certificateRef:
          name: 'external-server-cert' 
          kind: Secret

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: my-cluster
  namespace: my-proxy
spec:
  proxyRef:
    name: simple
  targetKafkaServiceRef:
    name: my-cluster
  ingresses:
    - ingressRef:
        name: cluster-ip
        certificateRef:
          name: 'cluster-ip-server-cert' 

          kind: Secret
    - ingressRef:
        name: load-balancer
        certificateRef:
          name: 'external-server-cert' 

          kind: Secret

oc create secret tls <secret-name> --namespace <namespace> --cert=<path/to/cert/file> --key=<path/to/key/file>

oc create secret tls <secret-name> --namespace <namespace> --cert=<path/to/cert/file> --key=<path/to/key/file>

oc create secret tls <secret-name> --namespace <namespace> --cert=<path/to/cert/file> --key=<path/to/key/file>

oc create secret tls <secret-name> --namespace <namespace> --cert=<path/to/cert/file> --key=<path/to/key/file>

When you create a VirtualKafkaCluster, the operator checks whether the following exist:

The result is reported in status.conditions with a ResolvedRefs condition accordingly.

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  # ...
  generation: 12
spec:
  # ...
status:
  observedGeneration: 12 
  conditions:
    - type: ResolvedRefs 
      status: True 
      observedGeneration: 12

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  # ...
  generation: 12
spec:
  # ...
status:
  observedGeneration: 12 

  conditions:
    - type: ResolvedRefs 

      status: True 

      observedGeneration: 12

A status value of False means that one or more of the referenced resources is missing. In this case, the condition includes reason and message properties with more details.

When a VirtualKafkaCluster has a valid spec, the operator attempts to configure the proxy instance accordingly. This might not be possible. For example, the spec may be valid but incompatible with other virtual clusters running in the same proxy instance.

The operator sets a condition type of Accepted in status.conditions to indicate whether or not a virtual cluster has been successfully configured within a proxy instance.

This example shows a VirtualKafkaCluster, exposing it to Kafka clients running on the same OpenShift cluster. It uses TLS as the transport protocol so that communication between Kafka clients and the proxy is encrypted.

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: my-cluster
  namespace: my-proxy
spec:
  proxyRef: 
    name: simple
  targetKafkaServiceRef: 
    name: my-cluster
  ingresses:
    - ingressRef: 
        name: cluster-ip
      tls: 
        certificateRef:
          name: server-certificate
          kind: Secret

kind: VirtualKafkaCluster
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: my-cluster
  namespace: my-proxy
spec:
  proxyRef: 

    name: simple
  targetKafkaServiceRef: 

    name: my-cluster
  ingresses:
    - ingressRef: 

        name: cluster-ip
      tls: 

        certificateRef:
          name: server-certificate
          kind: Secret

Within a VirtualKafkaCluster, an ingress’s tls property configures TLS for that ingress. The tls.certificateRef specifies the Secret resource holding the TLS server certificate that the proxy uses for clients connecting through this ingress. The referenced KafkaProxyIngress also needs to be configured for TLS.

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: cluster-ip
  namespace: my-proxy
spec:
  proxyRef: 
    name: simple
  clusterIP: 
    protocol: TLS 

kind: KafkaProxyIngress
apiVersion: kroxylicious.io/v1alpha1
metadata:
  name: cluster-ip
  namespace: my-proxy
spec:
  proxyRef: 

    name: simple
  clusterIP: 

    protocol: TLS 

You can configure a virtual cluster ingress to request or require Kafka clients to authenticate to the proxy using TLS. This configuration is known as mutual TLS (mTLS), because both the client and the proxy authenticate each other using TLS.

kind: VirtualKafkaCluster
metadata:
  # ...
spec:
  # ...
  ingresses:
    - ingressRef:
        name: cluster-ip
      tls:
        certificateRef:
          # ...
        trustAnchorRef: 
          kind: ConfigMap 
          name: trusted-cas 
          key: trusted-cas.pem 
        tlsClientAuthentication: REQUIRED 

kind: VirtualKafkaCluster
metadata:
  # ...
spec:
  # ...
  ingresses:
    - ingressRef:
        name: cluster-ip
      tls:
        certificateRef:
          # ...
        trustAnchorRef: 

          kind: ConfigMap 

          name: trusted-cas 

          key: trusted-cas.pem 

        tlsClientAuthentication: REQUIRED 

Some older versions of TLS (and SSL before it) are now considered insecure. These versions remain enabled by default in order to maximize interoperability between TLS clients and servers that only support older versions.

If the Kafka cluster than you want to connect to supports newer TLS versions, you can disable the proxy’s support for older, insecure versions. For example, if the Kafka cluster supports TLSv1.1, TLSv1.2 and TLSv1.3 you might choose to enable only TLSv1.3 support. This would reduce the susceptibility to a TLS downgrade attack.

You can restrict which TLS protocol versions the proxy supports for client-to-proxy connections by configuring the protocols property.

kind: VirtualKafkaCluster
metadata:
  # ...
spec:
  # ...
  ingresses:
    - ingressRef:
        name: cluster-ip
      tls:
        certificateRef:
          # ...
        protocols: 
          allow: 
            - TLSv1.3

kind: VirtualKafkaCluster
metadata:
  # ...
spec:
  # ...
  ingresses:
    - ingressRef:
        name: cluster-ip
      tls:
        certificateRef:
          # ...
        protocols: 

          allow: 

            - TLSv1.3

Alternatively, you can use deny to specify protocol versions to exclude.

The names of the TLS protocol versions supported depend on the JVM in the proxy container image.

A cipher suite is a set of cryptographic algorithms that together provide the security guarantees offered by TLS. During TLS negotiation, a server and client agree on a common cipher suite that they both support.

Some older cipher suites are now considered insecure, but may be enabled on the Kafka cluster to allow older clients to connect.

The cipher suites enabled by default in the proxy depend on the JVM used in the proxy image and the TLS protocol version that is negotiated.

To prevent TLS downgrade attacks, you can disable cipher suites known to be insecure or no longer recommended. However, the proxy and the cluster must support at least one cipher suite in common.

You can restrict which TLS cipher suites the proxy uses when negotiating client-to-proxy connections by configuring the cipherSuites property.

kind: VirtualKafkaCluster
metadata:
  # ...
spec:
  # ...
  ingresses:
    - ingressRef:
        name: cluster-ip
      tls:
        certificateRef:
          # ...
        cipherSuites: 
          allow: 
           - TLS_AES_128_GCM_SHA256
           - TLS_AES_256_GCM_SHA384

kind: VirtualKafkaCluster
metadata:
  # ...
spec:
  # ...
  ingresses:
    - ingressRef:
        name: cluster-ip
      tls:
        certificateRef:
          # ...
        cipherSuites: 

          allow: 

           - TLS_AES_128_GCM_SHA256
           - TLS_AES_256_GCM_SHA384

Alternatively, you can use deny to specify cipher suites to exclude. The names of the cipher suites supported depend on the JVM in the proxy container image.

By default, the proxy uses the platform’s default trust store when connecting to the proxied cluster over TLS. This works if the cluster’s TLS certificates are signed by a well-known public Certificate Authority (CA), but fails if they’re signed by a private CA instead.

This example configures a KafkaService to trust TLS certificates signed by any Certificate Authority (CA) listed in the trusted-cas.pem entry of the ConfigMap named trusted-cas.

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    trustAnchorRef: 
      kind: ConfigMap 
      name: trusted-cas 
      key: trusted-cas.pem 
    # ...

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    trustAnchorRef: 

      kind: ConfigMap 

      name: trusted-cas 

      key: trusted-cas.pem 

    # ...

Some Kafka clusters require mutual TLS (mTLS) authentication. You can configure the proxy to present a TLS client certificate using the KafkaService resource.

The TLS client certificate you provide must have been issued by a Certificate Authority (CA) that’s trusted by the proxied cluster.

This example configures a KafkaService to use a TLS client certificate stored in a Secret named tls-cert-for-kafka.example.com.

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    trustAnchorRef:
      kind: ConfigMap
      name: trusted-cas
      key: trusted-cas.pem
    certificateRef: 
      kind: Secret 
      name: tls-cert-for-kafka.example.com 
    # ...

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    trustAnchorRef:
      kind: ConfigMap
      name: trusted-cas
      key: trusted-cas.pem
    certificateRef: 

      kind: Secret 

      name: tls-cert-for-kafka.example.com 

    # ...

Some older versions of TLS (and SSL before it) are now considered insecure. These versions remain enabled by default in order to maximize interoperability between TLS clients and servers that only support older versions.

If the Kafka cluster than you want to connect to supports newer TLS versions, you can disable the proxy’s support for older, insecure versions. For example, if the Kafka cluster supports TLSv1.1, TLSv1.2 and TLSv1.3 you might choose to enable only TLSv1.3 support. This would reduce the susceptibility to a TLS downgrade attack.

This example configures a KafkaService to allow only TLS v1.3 when connecting to kafka.example.com.

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    # ...
    protocols: 
      allow: 
        - TLSv1.3

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    # ...
    protocols: 

      allow: 

        - TLSv1.3

The protocols property also supports deny, if you prefer to list the versions to exclude instead. The names of the TLS protocol versions supported depend on the JVM in the proxy container image.

A cipher suite is a set of cryptographic algorithms that together provide the security guarantees offered by TLS. During TLS negotiation, a server and client agree on a common cipher suite that they both support.

Some older cipher suites are now considered insecure, but may be enabled on the Kafka cluster to allow older clients to connect.

The cipher suites enabled by default in the proxy depend on the JVM used in the proxy image and the TLS protocol version that is negotiated.

To prevent TLS downgrade attacks, you can disable cipher suites known to be insecure or no longer recommended. However, the proxy and the cluster must support at least one cipher suite in common.

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    # ...
    cipherSuites: 
      allow: 
       - TLS_AES_128_GCM_SHA256
       - TLS_AES_256_GCM_SHA384

kind: KafkaService
metadata:
  # ...
spec:
  bootstrapServers: kafka.example.com:9092
  tls:
    # ...
    cipherSuites: 

      allow: 

       - TLS_AES_128_GCM_SHA256
       - TLS_AES_256_GCM_SHA384

The cipherSuites property also supports deny, if you prefer to list the cipher suites to exclude instead. The names of the cipher suites supported depend on the JVM in the proxy container image.

Filter resources can be configured to handle security-sensitive values like passwords or keys by referencing OpenShift Secret and ConfigMap resources.

kind: KafkaProtocolFilter
metadata:
  # ...
spec:
  type: RecordEncryption 
  configTemplate: 
    kms: VaultKmsService
    kmsConfig:
      vaultTransitEngineUrl: http://vault.vault.svc.cluster.local:8200/v1/transit
      vaultToken:
        password: ${secret:vault:token} 
    selector: TemplateKekSelector
    selectorConfig:
      template: "$(topicName)" 

kind: KafkaProtocolFilter
metadata:
  # ...
spec:
  type: RecordEncryption 

  configTemplate: 

    kms: VaultKmsService
    kmsConfig:
      vaultTransitEngineUrl: http://vault.vault.svc.cluster.local:8200/v1/transit
      vaultToken:
        password: ${secret:vault:token} 

    selector: TemplateKekSelector
    selectorConfig:
      template: "$(topicName)" 

Connection metrics count the TCP connections made from the client to the proxy (kroxylicious_client_to_proxy_request_total) and from the proxy to the broker (kroxylicious_proxy_to_server_connections_total). These metrics count connection attempts, so the connection count is incremented even if the connection attempt ultimately fails.

In addition to the count metrics, there are active connection gauge metrics that track the current number of open connections, and error metrics.

Connection and connection error metrics include the following labels: virtual_cluster (the virtual cluster’s name) and node_id (the broker’s node ID). When the client connects to the boostrap endpoint of the virtual cluster, a node ID value of bootstrap is recorded.

The kroxylicious_client_to_proxy_errors_total metric also counts connection errors that occur before a virtual cluster has been identified. For these specific errors, the virtual_cluster and node_id labels are set to an empty string ("").

Message metrics count and record the sizes of the Kafka protocol requests and responses that flow through the proxy.

Use these metrics to help understand:

The size recorded is the encoded size of the protocol message. It includes the 4 byte message size.

Filters can alter the flow of messages through the proxy or the content of the message. This is apparent through the metrics.

Figure 7.1. Downstream and upstream message metrics in the proxy

View larger image

Message metrics include the following labels: virtual_cluster (the virtual cluster’s name), node_id (the broker’s node ID), api_key (the message type), api_version, and decoded (a flag indicating if the message was decoded by the proxy).

When the client connects to the boostrap endpoint of the virtual cluster, metrics are recorded with a node ID value of bootstrap.

This procedure describes how to ingest metrics from the Streams for Apache Kafka Proxy Operator into Prometheus.

This procedure describes how to ingest metrics from the Streams for Apache Kafka Proxy into Prometheus.

This procedure describes how to override the logging level of the Streams for Apache Kafka Proxy.

This procedure describes how to override the logging level of the Streams for Apache Kafka Proxy Operator. It applies when the operator was installed by OLM.

This procedure describes how to override the logging level of the Streams for Apache Kafka Proxy Operator. It applies when the operator was installed from the YAML bundle.