Source code for octavia.amphorae.backends.agent.api_server.util

# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# 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.


import os
import re
import stat
import subprocess
import typing as tp

import jinja2
from oslo_config import cfg
from oslo_log import log as logging

from octavia.amphorae.backends.utils import ip_advertisement
from octavia.amphorae.backends.utils import network_utils
from octavia.common import constants as consts

CONF = cfg.CONF
LOG = logging.getLogger(__name__)

FRONTEND_BACKEND_PATTERN = re.compile(r'\n(frontend|backend)\s+(\S+)\n')
LISTENER_MODE_PATTERN = re.compile(r'^\s+mode\s+(.*)$', re.MULTILINE)
TLS_CERT_PATTERN = re.compile(r'^\s+bind\s+\S+\s+ssl crt-list\s+(\S*)',
                              re.MULTILINE)
STATS_SOCKET_PATTERN = re.compile(r'stats socket\s+(\S+)')



[docs]
class ParsingError(Exception):
    pass




[docs]
def init_path(lb_id):
    return os.path.join(consts.SYSTEMD_DIR, f'haproxy-{lb_id}.service')




[docs]
def keepalived_lvs_dir():
    return os.path.join(CONF.haproxy_amphora.base_path, 'lvs')




[docs]
def keepalived_lvs_init_path(listener_id):
    return os.path.join(consts.SYSTEMD_DIR,
                        consts.KEEPALIVEDLVS_SYSTEMD %
                        str(listener_id))




[docs]
def keepalived_backend_check_script_dir():
    return os.path.join(CONF.haproxy_amphora.base_path, 'lvs/check/')




[docs]
def keepalived_backend_check_script_path():
    return os.path.join(keepalived_backend_check_script_dir(),
                        'udp_check.sh')




[docs]
def keepalived_lvs_pids_path(listener_id):
    pids_path = {}
    for file_ext in ['pid', 'vrrp.pid', 'check.pid']:
        pids_path[file_ext] = (
            os.path.join(CONF