[designate] Health probe for Designate components

openstack/designate/bin/designate-producer-start

@@ -2,9 +2,15 @@
 
 set -ex
 
+#
+# to make agents named after the node
+#
+echo "[DEFAULT]" > /etc/designate/hostname.conf
+echo "host = $NODE_NAME" >> /etc/designate/hostname.conf
+
 #
 # other configs
 #
 cp /designate-etc/* /etc/designate/
 
-exec designate-producer --config-file /etc/designate/designate.conf --config-file /etc/designate/secrets.conf --log-config-append /etc/designate/logging.conf
+exec designate-producer --config-file /etc/designate/designate.conf  --config-file /etc/designate/hostname.conf --config-file /etc/designate/secrets.conf --log-config-append /etc/designate/logging.conf

openstack/designate/templates/bin/_health-probe.py

@@ -0,0 +1,241 @@
+#!/usr/bin/env python
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Health probe script for OpenStack service that uses RPC/unix domain socket for
+communication. Check's the RPC tcp socket status on the process and send
+message to service through rpc call method and expects a reply.
+Use oslo's ping method that is designed just for such simple purpose.
+
+Script returns failure to Kubernetes only when
+  a. TCP socket for the RPC communication are not established.
+  b. service is not reachable or
+  c. service times out sending a reply.
+
+sys.stderr.write() writes to pod's events on failures.
+
+Usage example for Designate Central:
+# python health-probe.py --config-file /etc/designate/secrets.conf \
+#  --config-file /etc/designate/hostname.conf \
+#  --service-queue-name central \
+#  --liveness-probe
+
+"""
+
+import json
+import os
+import psutil
+import signal
+import sys
+
+from oslo_config import cfg
+from oslo_context import context
+from oslo_log import log
+import oslo_messaging
+
+rpc_timeout = int(os.getenv('RPC_PROBE_TIMEOUT', '60'))
+rpc_retries = int(os.getenv('RPC_PROBE_RETRIES', '2'))
+
+tcp_established = "ESTABLISHED"
+
+
+def check_service_status(transport):
+    """Verify service status. Return success if service consumes message"""
+    try:
+        service_queue_name = cfg.CONF.service_queue_name
+        service_hostname = cfg.CONF.host
+        target = oslo_messaging.Target(
+            topic=service_queue_name,
+            server=(service_queue_name + "." + service_hostname),
+            namespace='baseapi',
+            version="1.1")
+        if hasattr(oslo_messaging, 'get_rpc_client'):
+            client = oslo_messaging.get_rpc_client(transport, target,
+                                                   timeout=rpc_timeout,
+                                                   retry=rpc_retries)
+        else:
+            client = oslo_messaging.RPCClient(transport, target,
+                                              timeout=rpc_timeout,
+                                              retry=rpc_retries)
+        client.call(context.RequestContext(),
+                    'ping',
+                    arg=None)
+    except oslo_messaging.exceptions.MessageDeliveryFailure:
+        # Log to pod events
+        sys.stderr.write("Health probe unable to reach message bus")
+        sys.exit(0)  # return success
+    except oslo_messaging.rpc.client.RemoteError as re:
+        message = getattr(re, "message", str(re))
+        if ("Endpoint does not support RPC method" in message) or \
+                ("Endpoint does not support RPC version" in message):
+            sys.exit(0)  # Call reached the service
+        else:
+            sys.stderr.write("Health probe unable to reach service")
+            sys.exit(1)  # return failure
+    except oslo_messaging.exceptions.MessagingTimeout:
+        sys.stderr.write("Health probe timed out. Agent is down or response "
+                         "timed out")
+        sys.exit(1)  # return failure
+    except Exception as ex:
+        message = getattr(ex, "message", str(ex))
+        sys.stderr.write("Health probe caught exception sending message to "
+                         "service: %s" % message)
+        sys.exit(0)
+    except:
+        sys.stderr.write("Health probe caught exception sending message to"
+                         " service")
+        sys.exit(0)
+
+
+def tcp_socket_status(process, ports):
+    """Check the tcp socket status on a process"""
+    for p in psutil.process_iter():
+        sys.stderr.write("Checking process %s\n" % p)
+        try:
+            with p.oneshot():
+                if process in " ".join(p.cmdline()):
+                    pcon = p.connections()
+                    sys.stderr.write("Process %s found with pid %s\n" % (process, p.pid))
+                    for con in pcon:
+                        try:
+                            rport = con.raddr[1]
+                            status = con.status
+                        except IndexError:
+                            continue
+                        if rport in ports and status == tcp_established:
+                            return 1
+        except psutil.Error:
+            continue
+    return 0
+
+
+def configured_port_in_conf():
+    """Get the rabbitmq/Database port configured in config file"""
+
+    rabbit_ports = frozenset([5672])
+    database_ports = frozenset([3306])
+
+    return rabbit_ports, database_ports
+
+
+def test_tcp_socket(service):
+    """Check tcp socket to rabbitmq/db is in Established state"""
+    dict_services = {
+        "api": "designate-api",
+        "producer": "designate-producer",
+        "worker": "designate-worker",
+        "mdns": "designate-mdns",
+        "central": "designate-central",
+    }
+    r_ports, d_ports = configured_port_in_conf()
+
+    if service in dict_services:
+        proc = dict_services[service]
+        transport = oslo_messaging.TransportURL.parse(cfg.CONF)
+        if r_ports and tcp_socket_status(proc, r_ports) == 0:
+            sys.stderr.write("RabbitMQ socket not established for service "
+                             "%s with transport %s" % (proc, transport))
+            # Do not kill the pod if RabbitMQ is not reachable/down
+            if not cfg.CONF.liveness_probe:
+                sys.exit(1)
+
+        # let's do the db check
+        if service not in ["api", "producer"]:
+            if d_ports and tcp_socket_status(proc, d_ports) == 0:
+                sys.stderr.write("Database socket not established for service "
+                                 "%s with transport %s" % (proc, transport))
+                # Do not kill the pod if database is not reachable/down
+                # there could be no socket as well as typically connections
+                # get closed after an idle timeout
+                # Just log it to pod events
+                if not cfg.CONF.liveness_probe:
+                    sys.exit(1)
+
+
+def test_rpc_liveness():
+    """Test if service can consume message from queue"""
+    oslo_messaging.set_transport_defaults(control_exchange='designate')
+
+    rabbit_group = cfg.OptGroup(name='oslo_messaging_rabbit',
+                                title='RabbitMQ options')
+    cfg.CONF.register_group(rabbit_group)
+    cfg.CONF.register_cli_opt(cfg.StrOpt('service-queue-name'))
+    cfg.CONF.register_cli_opt(cfg.BoolOpt('liveness-probe', default=False,
+                                          required=False))
+    cfg.CONF.register_cli_opt(cfg.StrOpt('host'))
+    cfg.CONF(sys.argv[1:])
+
+    log.logging.basicConfig(level=log.INFO)
+
+    try:
+        transport = oslo_messaging.get_rpc_transport(cfg.CONF)
+    except Exception as ex:
+        message = getattr(ex, "message", str(ex))
+        sys.stderr.write("Message bus driver load error: %s" % message)
+        sys.exit(0)  # return success
+
+    if not cfg.CONF.transport_url or \
+            not cfg.CONF.service_queue_name:
+        sys.stderr.write("Both message bus URL and service's queue name are "
+                         "required for health probe to work")
+        sys.exit(0)  # return success
+
+    try:
+        cfg.CONF.set_override('rabbit_max_retries', 2,
+                              group=rabbit_group)  # 3 attempts
+    except cfg.NoSuchOptError as ex:
+        cfg.CONF.register_opt(cfg.IntOpt('rabbit_max_retries', default=2),
+                              group=rabbit_group)
+
+    service = cfg.CONF.service_queue_name
+    test_tcp_socket(service)
+
+    check_service_status(transport)
+
+def check_pid_running(pid):
+    if psutil.pid_exists(int(pid)):
+       return True
+    else:
+       return False
+
+if __name__ == "__main__":
+
+    if "liveness-probe" in ','.join(sys.argv):
+        pidfile = "/tmp/liveness.pid"  #nosec
+    else:
+        pidfile = "/tmp/readiness.pid"  #nosec
+    data = {}
+    if os.path.isfile(pidfile):
+        with open(pidfile,'r') as f:
+            file_content = f.read().strip()
+            if file_content:
+                data = json.loads(file_content)
+
+    if 'pid' in data and check_pid_running(data['pid']):
+        if 'exit_count' in data and data['exit_count'] > 1:
+            # Third time in, kill the previous process
+            os.kill(int(data['pid']), signal.SIGTERM)
+        else:
+            data['exit_count'] = data.get('exit_count', 0) + 1
+            with open(pidfile, 'w') as f:
+                json.dump(data, f)
+            sys.exit(0)
+    data['pid'] = os.getpid()
+    data['exit_count'] = 0
+    with open(pidfile, 'w') as f:
+        json.dump(data, f)
+
+    test_rpc_liveness()
+
+    sys.exit(0)  # return success

openstack/designate/templates/etc/_designate.conf.tpl

@@ -53,6 +53,8 @@ quota_api_export_size = {{ .Values.quota_api_export_size | default 1000 }}
 rpc_response_timeout = {{ .Values.rpc_response_timeout | default .Values.global.rpc_response_timeout | default 300 }}
 rpc_workers = {{ .Values.rpc_workers | default .Values.global.rpc_workers | default 1 }}
 
+rpc_ping_enabled = true
+
 wsgi_default_pool_size = {{ .Values.wsgi_default_pool_size | default .Values.global.wsgi_default_pool_size | default 100 }}
 min_pool_size = {{ .Values.min_pool_size | default .Values.global.min_pool_size | default 10 }}
 max_pool_size = {{ .Values.max_pool_size | default .Values.global.max_pool_size | default 100 }}
@@ -66,7 +68,7 @@ heartbeat_in_pthread = false
 rabbit_interval_max = 3
 rabbit_retry_backoff = 1
 kombu_reconnect_delay = 0.1
-heartbeat_timeout_threshold = 15
+heartbeat_timeout_threshold = 30
 heartbeat_rate = 3
 
 [oslo_messaging_notifications]