远程访问运行在 kubernetes 中的 Kafka

Posted 2023-03-10

【中文标题】远程访问运行在 kubernetes 中的 Kafka

【英文标题】：Remotely accessing Kafka running inside kubernetes

【发布时间】：2020-01-25 09:04:39

【问题描述】：

我有一个单节点 Kafka 代理在单节点 kubernetes 环境的 pod 内运行。我正在将此图像用于 kafka：https://hub.docker.com/r/wurstmeister/kafka

kafka 版本 = 1.1.0

Kubernetes 集群在服务器上的虚拟机内运行。 VM 在活动接口 ens32 上具有以下 IP - 192.168.3.102

Kafka.yaml

apiVersion: extensions/v1beta1
kind: Deployment
metadata:
  namespace: casb-deployment
  name: kafkaservice
spec:
  replicas: 1
  template:
    metadata:
      labels:
        app: kafkaservice
    spec:
      hostname: kafkaservice
      containers:
      - name: kafkaservice
        imagePullPolicy: IfNotPresent
        image: wurstmeister/kafka:1.1.0
        env:
         - name: KAFKA_BROKER_ID
           value: "1"
#         - name: KAFKA_ADVERTISED_HOST_NAME
#           value: "kafkaservice"
         - name: KAFKA_LISTENER_SECURITY_PROTOCOL_MAP
           value: "INTERNAL_PLAINTEXT:PLAINTEXT,EXTERNAL_PLAINTEXT:PLAINTEXT"
         - name: KAFKA_LISTENERS
           value: "INTERNAL_PLAINTEXT://0.0.0.0:9092,EXTERNAL_PLAINTEXT://0.0.0.0:9093"
         - name: KAFKA_ADVERTISED_LISTENERS
           value: "INTERNAL_PLAINTEXT://kafkaservice:9092,EXTERNAL_PLAINTEXT://192.168.3.102:9093"
         - name: KAFKA_INTER_BROKER_LISTENER_NAME
           value: "INTERNAL_PLAINTEXT"
         - name: KAFKA_CREATE_TOPICS
           value: "topic-1:100:1,topic-2:1:1"  
         - name: KAFKA_ZOOKEEPER_CONNECT
           value: "zookeeper:2181"
        ports: 
        - name: port9092
          containerPort: 9092
        - name: port9093
          containerPort: 9093
        volumeMounts:
        - mountPath: /kafka/kafka-logs-kafkaservice
          name: kafka-volume
      volumes:
      - name: kafka-volume
        hostPath:
          path: /home/volume/kafka-logs

---

apiVersion: v1
kind: Service
metadata:
  namespace: casb-deployment
  name: kafkaservice
  labels:
    app: kafkaservice
spec:
  selector:
    app: kafkaservice
  ports:
  - name: port9092
    port: 9092
    targetPort: 9092
    protocol: TCP

---

apiVersion: v1
kind: Service
metadata:
  namespace: casb-deployment
  name: kafkaservice-external
  labels:
    app: kafkaservice-external
spec:
  selector:
    app: kafkaservice
  ports:
  - name: port9093
    port: 9093
    protocol: TCP
    nodePort: 30035
  type: NodePort

我能够从我的本地机器ping 192.168.3.102 ping 虚拟机，即 kubernetes 节点，并且我正在使用 nodePort 来公开服务。

我也可以通过 telnet telnet 192.168.3.102 30035 得到：

Trying 192.168.3.102...
Connected to 192.168.3.102.
Escape character is '^]'.

我尝试在本地机器上运行 kafka 控制台消费者和生产者：

消费者： bin/kafka-console-consumer.sh --bootstrap-server 192.168.3.102:30035 --topic topic-1 --from-beginning

输出：

[2019-09-25 12:30:40,716] WARN [Consumer clientId=consumer-1, groupId=console-consumer-20551] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

制片人：

bin/kafka-console-producer.sh --broker-list 192.168.3.102:30035 --topic topic-1

输出：

[2019-09-25 12:32:07,958] WARN [Producer clientId=console-producer] Connection to node 1 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

网络统计：

netstat -tulpn | grep 30035
tcp6       0      0 :::30035                :::*                    LISTEN      113545/kube-proxy

我尝试运行基于 python 的消费者，即kafka-python==1.4.2，它给了我以下日志：

[2019-09-25T12:15:39+0500] INFO kafka.client Bootstrapping cluster metadata from [('192.168.3.102', 30035, <AddressFamily.AF_INET: 2>)]
[2019-09-25T12:15:39+0500] INFO kafka.conn <BrokerConnection node_id=bootstrap host=192.168.3.102:30035 <connecting> [IPv4 ('192.168.3.102', 30035)]>: connecting to 192.168.3.102:30035 [('192.168.3.102', 30035) IPv4]
[2019-09-25T12:15:39+0500] INFO kafka.conn <BrokerConnection node_id=bootstrap host=192.168.3.102:30035 <connecting> [IPv4 ('192.168.3.102', 30035)]>: Connection complete.
[2019-09-25T12:15:39+0500] INFO kafka.client Bootstrap succeeded: found 1 brokers and 26 topics.
[2019-09-25T12:15:39+0500] INFO kafka.conn <BrokerConnection node_id=bootstrap host=192.168.3.102:30035 <connected> [IPv4 ('192.168.3.102', 30035)]>: Closing connection. 
[2019-09-25T12:15:39+0500] INFO kafka.conn <BrokerConnection node_id=1 host=192.168.3.102:9093 <connecting> [IPv4 ('192.168.3.102', 9093)]>: connecting to 192.168.3.102:9093 [('192.168.3.102', 9093) IPv4]
[2019-09-25T12:15:39+0500] INFO kafka.conn Probing node 1 broker version
[2019-09-25T12:15:39+0500] ERROR kafka.conn Connect attempt to <BrokerConnection node_id=1 host=192.168.3.102:9093 <connecting> [IPv4 ('192.168.3.102', 9093)]> returned error 111. Disconnecting.
[2019-09-25T12:15:39+0500] INFO kafka.conn <BrokerConnection node_id=1 host=192.168.3.102:9093 <connecting> [IPv4 ('192.168.3.102', 9093)]>: Closing connection. ConnectionError: 111 ECONNREFUSED
[2019-09-25T12:15:40+0500] INFO kafka.conn <BrokerConnection node_id=1 host=192.168.3.102:9093 <connecting> [IPv4 ('192.168.3.102', 9093)]>: connecting to 192.168.3.102:9093 [('192.168.3.102', 9093) IPv4]
[2019-09-25T12:15:40+0500] ERROR kafka.conn Connect attempt to <BrokerConnection node_id=1 host=192.168.3.102:9093 <connecting> [IPv4 ('192.168.3.102', 9093)]> returned error 111. Disconnecting.
[2019-09-25T12:15:40+0500] INFO kafka.conn <BrokerConnection node_id=1 host=192.168.3.102:9093 <connecting> [IPv4 ('192.168.3.102', 9093)]>: Closing connection. ConnectionError: 111 ECONNREFUSED
[2019-09-25T12:15:40+0500] INFO Activity URL collector Exception in activity url collector: NoBrokersAvailable

从日志看来，连接似乎已经建立，即

<connecting> [IPv4 ('192.168.3.102', 30035)]>: Connection complete.
Bootstrap succeeded: found 1 brokers and 26 topics.

然后它就断开了。

请帮助我弄清楚我缺少什么以及如何解决这个问题。谢谢。

【问题讨论】：

尝试kafkacat -L -b 192.168.3.102:30035正确执行kafka连接可以为元数据建立

【参考方案1】：

我还建议在 Kubernetes 上为 Kafka 使用 Strimzi。对于外部访问，这篇文章救了我https://developers.redhat.com/blog/2019/06/11/accessing-apache-kafka-in-strimzi-part-4-load-balancers/。 我的配置如下：

apiVersion: kafka.strimzi.io/v1beta1
kind: Kafka
metadata:
  name: my-cluster
spec:
  kafka:
    version: 2.4.0
    replicas: 1
    listeners:
      plain: 
      tls: 
      external:
        type: loadbalancer
        tls: false
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      log.message.format.version: "2.4"
    storage:
      type: ephemeral
  zookeeper:
    replicas: 1
    storage:
      type: ephemeral
  entityOperator:
    topicOperator: 
    userOperator: 

这用于检索 IP

kubectl get service my-cluster-kafka-external-bootstrap -o=jsonpath='.status.loadBalancer.ingress[0].ip"\n"'

【参考方案2】：

AWS/DigitalOcean/GCE/Azure Kubernetes 上 Kafka/Zookeeper 的简单配置，可外部访问：

https://github.com/StanislavKo/k8s_digitalocean_kafka

您可以通过常规二进制协议从 AWS/DO/GCE 外部连接到 Kafka。连接是 PLAINTEXT 或 SASL_PLAINTEXT（用户名/密码）。

Kafka 集群是 StatefulSet，因此您可以轻松扩展集群。

【讨论】：

请谨慎链接到您自己在不同网站上的内容，您不想成为spammer。您应该在此处包含大部分内容，并且仅将链接用作参考。

【参考方案3】：

您正在引导至端口 30035，并获得初始连接，然后为后续连接返回通告的端口 9093，而不是 30035。

您需要 NodePort 和广告中的一个相同，或者至少两者都是可外部路由的。如果你的代码在你的主机上运行，​​你还需要在你的虚拟机上进行端口转发

注意：存在用于在 Kubernetes 中设置 Kafka 的 Confluent 或 Strimzi Helm Charts