Merge "Amphora vertical scaling optimization in Octavia"

Author: Zuul

octavia/amphorae/backends/agent/api_server/amphora_info.py

@@ -17,6 +17,7 @@
 import socket
 import subprocess
 
+from oslo_log import log as logging
 import pyroute2
 import webob
 
@@ -26,6 +27,8 @@
 from octavia.common import constants as consts
 from octavia.common import exceptions
 
+LOG = logging.getLogger(__name__)
+
 
 class AmphoraInfo(object):
     def __init__(self, osutils):
@@ -60,6 +63,9 @@ def compile_amphora_details(self, extend_lvs_driver=None):
         meminfo = self._get_meminfo()
         cpu = self._cpu()
         st = os.statvfs('/')
+        listeners = (
+            sorted(set(haproxy_listener_list + lvs_listener_list))
+            if lvs_listener_list else haproxy_listener_list)
         body = {'hostname': socket.gethostname(),
                 'haproxy_version':
                     self._get_version_of_installed_package('haproxy'),
@@ -68,6 +74,7 @@ def compile_amphora_details(self, extend_lvs_driver=None):
                 'active': True,
                 'haproxy_count':
                     self._count_haproxy_processes(haproxy_listener_list),
+                'cpu_count': os.cpu_count(),
                 'cpu': {
                     'total': cpu['total'],
                     'user': cpu['user'],
@@ -85,11 +92,10 @@ def compile_amphora_details(self, extend_lvs_driver=None):
                     'used': (st.f_blocks - st.f_bfree) * st.f_frsize,
                     'available': st.f_bavail * st.f_frsize},
                 'load': self._load(),
+                'active_tuned_profiles': self._get_active_tuned_profiles(),
                 'topology': consts.TOPOLOGY_SINGLE,
                 'topology_status': consts.TOPOLOGY_STATUS_OK,
-                'listeners': sorted(list(
-                    set(haproxy_listener_list + lvs_listener_list)))
-                if lvs_listener_list else haproxy_listener_list,
+                'listeners': listeners,
                 'packages': {}}
         if extend_body:
             body.update(extend_body)
@@ -188,3 +194,12 @@ def get_interface(self, ip_addr):
                 status=404)
         return webob.Response(json=dict(message='OK', interface=interface),
                               status=200)
+
+    def _get_active_tuned_profiles(self) -> str:
+        """Returns the active TuneD profile(s)"""
+        try:
+            with open("/etc/tuned/active_profile", "r", encoding="utf-8") as f:
+                return f.read(1024).strip()
+        except OSError as ex:
+            LOG.debug("Reading active TuneD profiles failed: %r", ex)
+        return ""

octavia/amphorae/drivers/haproxy/rest_api_driver.py

@@ -219,10 +219,13 @@ def update_amphora_listeners(self, loadbalancer, amphora,
         if has_tcp and not split_config:
             if listeners_to_update:
                 # Generate HaProxy configuration from listener object
+                amp_details = self.clients[amphora.api_version].get_details(
+                    amphora)
                 config = self.jinja_combo.build_config(
                     host_amphora=amphora, listeners=listeners_to_update,
                     tls_certs=certs,
-                    haproxy_versions=haproxy_versions)
+                    haproxy_versions=haproxy_versions,
+                    amp_details=amp_details)
                 self.clients[amphora.api_version].upload_config(
                     amphora, loadbalancer.id, config,
                     timeout_dict=timeout_dict)

octavia/common/jinja/haproxy/combined_listeners/jinja_cfg.py

@@ -14,6 +14,7 @@
 
 import os
 import re
+from typing import Optional
 
 import jinja2
 from octavia_lib.common import constants as lib_consts
@@ -85,11 +86,12 @@ def __init__(self,
         self.connection_logging = connection_logging
 
     def build_config(self, host_amphora, listeners, tls_certs,
-                     haproxy_versions, socket_path=None):
+                     haproxy_versions, amp_details, socket_path=None):
         """Convert a logical configuration to the HAProxy version
 
         :param host_amphora: The Amphora this configuration is hosted on
         :param listener: The listener configuration
+        :param amp_details: Detail information from the amphora
         :param socket_path: The socket path for Haproxy process
         :return: Rendered configuration
         """
@@ -115,7 +117,8 @@ def build_config(self, host_amphora, listeners, tls_certs,
         return self.render_loadbalancer_obj(
             host_amphora, listeners, tls_certs=tls_certs,
             socket_path=socket_path,
-            feature_compatibility=feature_compatibility)
+            feature_compatibility=feature_compatibility,
+            amp_details=amp_details)
 
     def _get_template(self):
         """Returns the specified Jinja configuration template."""
@@ -152,13 +155,15 @@ def _format_log_string(self, load_balancer, protocol):
 
     def render_loadbalancer_obj(self, host_amphora, listeners,
                                 tls_certs=None, socket_path=None,
-                                feature_compatibility=None):
+                                feature_compatibility=None,
+                                amp_details: Optional[dict] = None):
         """Renders a templated configuration from a load balancer object
 
         :param host_amphora: The Amphora this configuration is hosted on
         :param listener: The listener configuration
         :param tls_certs: Dict of the TLS certificates for the listener
         :param socket_path: The socket path for Haproxy process
+        :param amp_details: Detail information from the amphora
         :return: Rendered configuration
         """
         feature_compatibility = feature_compatibility or {}
@@ -167,36 +172,45 @@ def render_loadbalancer_obj(self, host_amphora, listeners,
             listeners[0].load_balancer,
             listeners,
             tls_certs,
-            feature_compatibility,)
+            feature_compatibility)
         if not socket_path:
             socket_path = '%s/%s.sock' % (self.base_amp_path,
                                           listeners[0].load_balancer.id)
         state_file_path = '%s/%s/servers-state' % (
             self.base_amp_path,
             listeners[0].load_balancer.id) if feature_compatibility.get(
             constants.SERVER_STATE_FILE) else ''
-        prometheus_listener = False
-        for listener in listeners:
-            if listener.protocol == lib_consts.PROTOCOL_PROMETHEUS:
-                prometheus_listener = True
-                break
+        prometheus_listener = any(
+            lsnr.protocol == lib_consts.PROTOCOL_PROMETHEUS for lsnr in
+            listeners)
         require_insecure_fork = feature_compatibility.get(
             constants.INSECURE_FORK)
         enable_prometheus = prometheus_listener and feature_compatibility.get(
             lib_consts.PROTOCOL_PROMETHEUS, False)
+
+        jinja_dict = {
+            'loadbalancer': loadbalancer,
+            'stats_sock': socket_path,
+            'log_http': self.log_http,
+            'log_server': self.log_server,
+            'state_file': state_file_path,
+            'administrative_log_facility':
+                CONF.amphora_agent.administrative_log_facility,
+            'user_log_facility':
+                CONF.amphora_agent.user_log_facility,
+            'connection_logging': self.connection_logging,
+            'enable_prometheus': enable_prometheus,
+            'require_insecure_fork': require_insecure_fork,
+        }
+        try:
+            # Enable cpu-pinning only if the amphora TuneD profile is active
+            if "amphora" in amp_details["active_tuned_profiles"].split():
+                jinja_dict["cpu_count"] = int(amp_details["cpu_count"])
+        except (KeyError, TypeError):
+            pass
+
         return self._get_template().render(
-            {'loadbalancer': loadbalancer,
-             'stats_sock': socket_path,
-             'log_http': self.log_http,
-             'log_server': self.log_server,
-             'state_file': state_file_path,
-             'administrative_log_facility':
-                 CONF.amphora_agent.administrative_log_facility,
-             'user_log_facility': CONF.amphora_agent.user_log_facility,
-             'connection_logging': self.connection_logging,
-             'enable_prometheus': enable_prometheus,
-             'require_insecure_fork': require_insecure_fork},
-            constants=constants, lib_consts=lib_consts)
+            jinja_dict, constants=constants, lib_consts=lib_consts)
 
     def _transform_loadbalancer(self, host_amphora, loadbalancer, listeners,
                                 tls_certs, feature_compatibility):

octavia/common/jinja/haproxy/combined_listeners/templates/base.j2

@@ -26,6 +26,10 @@ global
     {% if loadbalancer.global_connection_limit is defined %}
     maxconn {{ loadbalancer.global_connection_limit }}
     {% endif %}
+    {%- if cpu_count is defined and cpu_count > 1 %}
+    nbthread {{ cpu_count - 1 }}
+    cpu-map auto:1/1-{{ cpu_count - 1 }} 1-{{ cpu_count - 1 }}
+    {%- endif %}
     {% set found_ns = namespace(found=false) %}
     {% for listener in loadbalancer.listeners if listener.enabled %}
         {% for pool in listener.pools if pool.enabled %}

octavia/tests/functional/amphorae/backend/agent/api_server/test_server.py

@@ -2964,10 +2964,14 @@ def _test_details(self, distro, mock_subbprocess, mock_hostname,
 
         haproxy_count = random.randrange(0, 100)
         mock_count_haproxy.return_value = haproxy_count
+        tuned_profiles = "virtual-guest optimize-serial-console amphora"
 
-        expected_dict = {'active': True, 'api_version': '1.0',
+        expected_dict = {'active': True,
+                         'active_tuned_profiles': tuned_profiles,
+                         'api_version': '1.0',
                          'cpu': {'soft_irq': cpu_softirq, 'system': cpu_system,
                                  'total': cpu_total, 'user': cpu_user},
+                         'cpu_count': os.cpu_count(),
                          'disk': {'available': disk_available,
                                   'used': disk_used},
                          'haproxy_count': haproxy_count,
@@ -2995,10 +2999,13 @@ def _test_details(self, distro, mock_subbprocess, mock_hostname,
                          'topology_status': consts.TOPOLOGY_STATUS_OK,
                          'lvs_listener_process_count': 0}
 
-        if distro == consts.UBUNTU:
-            rv = self.ubuntu_app.get('/' + api_server.VERSION + '/details')
-        elif distro == consts.CENTOS:
-            rv = self.centos_app.get('/' + api_server.VERSION + '/details')
+        with mock.patch("octavia.amphorae.backends.agent.api_server"
+                        ".amphora_info.open",
+                        mock.mock_open(read_data=tuned_profiles)):
+            if distro == consts.UBUNTU:
+                rv = self.ubuntu_app.get('/' + api_server.VERSION + '/details')
+            elif distro == consts.CENTOS:
+                rv = self.centos_app.get('/' + api_server.VERSION + '/details')
 
         self.assertEqual(200, rv.status_code)
         self.assertEqual(expected_dict,

octavia/tests/unit/amphorae/backends/agent/api_server/test_amphora_info.py

@@ -11,6 +11,7 @@
 #    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 random
 from unittest import mock
 
@@ -151,11 +152,13 @@ def test_compile_amphora_details(self, mhostname, m_count, m_pkg_version,
         original_version = api_server.VERSION
         api_server.VERSION = self.API_VERSION
         expected_dict = {u'active': True,
+                         'active_tuned_profiles': '',
                          u'api_version': self.API_VERSION,
                          u'cpu': {u'soft_irq': u'8336',
                                   u'system': u'52554',
                                   u'total': 7503411,
                                   u'user': u'252551'},
+                         'cpu_count': os.cpu_count(),
                          u'disk': {u'available': 109079126016,
                                    u'used': 25718685696},
                          u'haproxy_count': 5,
@@ -235,11 +238,13 @@ def test_compile_amphora_details_for_ipvs(self, mhostname, m_count,
         original_version = api_server.VERSION
         api_server.VERSION = self.API_VERSION
         expected_dict = {u'active': True,
+                         'active_tuned_profiles': '',
                          u'api_version': self.API_VERSION,
                          u'cpu': {u'soft_irq': u'8336',
                                   u'system': u'52554',
                                   u'total': 7503411,
                                   u'user': u'252551'},
+                         'cpu_count': os.cpu_count(),
                          u'disk': {u'available': 109079126016,
                                    u'used': 25718685696},
                          u'haproxy_count': 5,