2分钟让ZMQ pub / sub连接kubernetes

如何解决2分钟让ZMQ pub / sub连接kubernetes

我有一个使用weave作为我的CNI的Kubernetes 1.18集群。我有一个基于ZMQ的发布/订阅应用程序，经常（并非总是）看到订阅者需要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,))
    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,'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','description': 'EVENT_CONNECT_DELAYED'}

但是我在Kubernetes中看到的却是：

│ 2020-08-16 05:54:51,724 Event: {'endpoint': 'tcp://pub:4567','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','description': 'EVENT_CLOSED'}                                                                                                                    │
│ 2020-08-16 05:56:59,'description': 'EVENT_CONNECT_RETRIED'}                                                                                                            │
│ 2020-08-16 05:56:59,243 Event: {'endpoint': 'tcp://pub:4567','description': 'EVENT_CONNECT_DELAYED'}                                                                                                            │
│ 2020-08-16 05:56:59,245 Event: {'endpoint': 'tcp://pub:4567','event': 1,'description': 'EVENT_CONNECTED'}                                                                                                                   │
│ 2020-08-16 05:56:59,286 Received: Message 127 at 2020-08-16 05:56:59.286036  

显然，Kubernetes内的某些情况阻止了“ EVENT_CLOSED”事件在及时的庄园中发生。这可能是什么？

解决方法

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