ZMQ pub/sub 在 kubernetes 中连接需要 2 分钟

Question

我有一个 Kubernetes 1.18 集群，使用 weave 作为我的 CNI。我有一个基于 ZMQ 的 pub/sub 应用程序，我经常（并非总是）看到订阅者需要 2 分钟才能收到来自发布者的消息。这似乎是我的 Kubernetes 环境特有的某种套接字超时。

这是我简单的 ZMQ 应用程序示例

#!/bin/env python2
import zmq, sys, time, argparse, logging, datetime, threading
from zmq.utils.monitor import recv_monitor_message

FORMAT = '%(asctime)-15s %(message)s'
logging.basicConfig(format=FORMAT)

if zmq.zmq_version_info() < (4, 0):
    raise RuntimeError("monitoring in libzmq version < 4.0 is not supported")

logging.error("libzmq-%s" % zmq.zmq_version())
if zmq.zmq_version_info() < (4, 0):
    raise RuntimeError("monitoring in libzmq version < 4.0 is not supported")

EVENT_MAP = {}
logging.error("Event names:")
for name in dir(zmq):
    if name.startswith('EVENT_'):
        value = getattr(zmq, name)
        logging.error("%21s : %4i" % (name, value))
        EVENT_MAP[value] = name


def event_monitor(monitor):
    while monitor.poll():
        evt = recv_monitor_message(monitor)
        evt.update({'description': EVENT_MAP[evt['event']]})
        logging.error("Event: {}".format(evt))
        if evt['event'] == zmq.EVENT_MONITOR_STOPPED:
            break
    monitor.close()
    logging.error("event monitor thread done!")

parser = argparse.ArgumentParser("Simple zmq pubsub example")
parser.add_argument("pub_or_sub", help="Either pub or sub")
parser.add_argument("host", help="host address to connect to if sub otherwise the address to bind to")
parser.add_argument("--port", "-p", type=int, help="The port to use", default=4567)
args = parser.parse_args()

context = zmq.Context()

if args.pub_or_sub.lower() == "sub":
    zmq_socket = context.socket(zmq.SUB)
    monitor = zmq_socket.get_monitor_socket()
    t = threading.Thread(target=event_monitor, args=(monitor,))
    t.setDaemon(True)
    t.start()

    zmq_socket.setsockopt(zmq.SUBSCRIBE, "")
    zmq_socket.connect("tcp://{}:{}".format(args.host, args.port))
    while 1:
        if zmq_socket.poll(timeout=1000):
            logging.error("Received: {}".format(zmq_socket.recv()))
        else:
            logging.error("No message available")
elif args.pub_or_sub.lower() == "pub":
    zmq_socket = context.socket(zmq.PUB)
    monitor = zmq_socket.get_monitor_socket()
    t = threading.Thread(target=event_monitor, args=(monitor,))
    t.setDaemon(True)
    t.start()
    zmq_socket.bind("tcp://{}:{}".format(args.host, args.port))
    i = 0
    while 1:
        logging.error("Sending message: {}".format(i))
        zmq_socket.send("Message {} at {}".format(i, datetime.datetime.now()))
        i += 1
        time.sleep(1.0)
else:
    raise RuntimeError("Needs to either be sub or pub nothing else allowed")

以下是我在 Kubernetes 中部署它的方式：

---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: pub-deployment
  labels:
    app: pub
spec:
  replicas: 1
  selector:
    matchLabels:
      app: pub
  template:
    metadata:
      labels:
        app: pub
    spec:
      containers:
      - name: pub
        image: bagoulla/zmq:latest
        command: ["pubsub", "pub", "0.0.0.0"]
---
apiVersion: v1
kind: Service
metadata:
  name: pub
spec:
  selector:
    app: pub
  ports:
    - protocol: TCP
      port: 4567
      targetPort: 4567
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: sub-deployment
  labels:
    app: sub
spec:
  replicas: 1
  selector:
    matchLabels:
      app: sub
  template:
    metadata:
      labels:
        app: sub
    spec:
      containers:
      - name: sub
        image: bagoulla/zmq:latest
        command: ["pubsub", "sub", "pub"]

我希望从订阅者那里看到什么，我确实看到在同一主机上的 Kubernetes 之外运行时（尽管仍在 Docker 中），以下是快速连续重复的直到 pub 容器准备好并路由：

2020-08-16 08:12:09,141 Event: {'endpoint': 'tcp://127.0.0.1:4567', 'event': 2, 'value': 115, 'description': 'EVENT_CONNECT_DELAYED'}
2020-08-16 08:12:09,141 Event: {'endpoint': 'tcp://127.0.0.1:4567', 'event': 128, 'value': 12, 'description': 'EVENT_CLOSED'}
2020-08-16 08:12:09,142 Event: {'endpoint': 'tcp://127.0.0.1:4567', 'event': 4, 'value': 183, 'description': 'EVENT_CONNECT_RETRIED'}
2020-08-16 08:12:09,328 Event: {'endpoint': 'tcp://127.0.0.1:4567', 'event': 2, 'value': 115, 'description': 'EVENT_CONNECT_DELAYED'}

但是我在 Kubernetes 中看到的是：

│ 2020-08-16 05:54:51,724 Event: {'endpoint': 'tcp://pub:4567', 'event': 2, 'value': 115, 'description': 'EVENT_CONNECT_DELAYED'}                                                                                                            │
.... 2 minutes later....
│ 2020-08-16 05:56:59,038 No message available                                                                                                                                                                                               │
│ 2020-08-16 05:56:59,056 Event: {'endpoint': 'tcp://pub:4567', 'event': 128, 'value': 12, 'description': 'EVENT_CLOSED'}                                                                                                                    │
│ 2020-08-16 05:56:59,056 Event: {'endpoint': 'tcp://pub:4567', 'event': 4, 'value': 183, 'description': 'EVENT_CONNECT_RETRIED'}                                                                                                            │
│ 2020-08-16 05:56:59,243 Event: {'endpoint': 'tcp://pub:4567', 'event': 2, 'value': 115, 'description': 'EVENT_CONNECT_DELAYED'}                                                                                                            │
│ 2020-08-16 05:56:59,245 Event: {'endpoint': 'tcp://pub:4567', 'event': 1, 'value': 12, 'description': 'EVENT_CONNECTED'}                                                                                                                   │
│ 2020-08-16 05:56:59,286 Received: Message 127 at 2020-08-16 05:56:59.286036

显然 Kubernetes 中的某些东西阻止了“EVENT_CLOSED”事件及时发生。这可能是什么？

Answer 1

问题是，当服务启动时，它实际上会创建一个 TCP 黑洞，在该黑洞中可以启动 tcp 连接，但永远不会结束连接。用户应该在 TCP 连接上设置超时，以便他们可以重试连接，直到底层部署或 pod 启动并正确路由。对于 ZMQ，这可以通过 ZMQ_CONNECT_TIMEOUT 套接字选项来完成。

ZMQ pub/sub 在 kubernetes 中连接需要 2 分钟

2 minutes for ZMQ pub/sub to connect in kubernetes

zeromq

pyzmq

kubernetes

weave