[DPE-4115] Performance Profile Support

config.yaml

@@ -36,3 +36,13 @@ options:
     default: true
     type: boolean
     description: Enable opensearch-knn
+
+  profile:
+    type: string
+    default: "production"
+    description: |
+      Profile representing the scope of deployment, and used to tune resource allocation.
+      Allowed values are: "production", "staging" or "testing"
+      Production will tune opensearch for maximum performance while default will tune for
+      minimal running performance.
+      Performance tuning is described on: https://opensearch.org/docs/latest/tuning-your-cluster/performance/

lib/charms/opensearch/v0/constants_charm.py

@@ -118,3 +118,5 @@
 
 # User-face Backup ID format
 OPENSEARCH_BACKUP_ID_FORMAT = "%Y-%m-%dT%H:%M:%SZ"
+
+PERFORMANCE_PROFILE = "profile"

lib/charms/opensearch/v0/helper_conf_setter.py

@@ -272,14 +272,13 @@ def replace(
             output_file: Target file for the result config, by default same as config_file
         """
         path = f"{self.base_path}{config_file}"
-
         if not exists(path):
             raise FileNotFoundError(f"{path} not found.")
 
         with open(path, "r+") as f:
             data = f.read()
 
-            if regex and old_val and re.compile(old_val).match(data):
+            if regex and old_val and re.compile(old_val, re.MULTILINE).findall(data):
                 data = re.sub(r"{}".format(old_val), f"{new_val}", data)
             elif old_val and old_val in data:
                 data = data.replace(old_val, new_val)

lib/charms/opensearch/v0/models.py

@@ -23,6 +23,10 @@
 LIBPATCH = 1
 
 
+MIN_HEAP_SIZE = 1024 * 1024  # 1GB in KB
+MAX_HEAP_SIZE = 32 * MIN_HEAP_SIZE  # 32GB in KB
+
+
 class Model(ABC, BaseModel):
     """Base model class."""
 
@@ -153,6 +157,14 @@ class DeploymentType(BaseStrEnum):
     OTHER = "other"
 
 
+class PerformanceType(BaseStrEnum):
+    """Performance types available."""
+
+    PRODUCTION = "production"
+    STAGING = "staging"
+    TESTING = "testing"
+
+
 class StartMode(BaseStrEnum):
     """Mode of start of units in this deployment."""
 
@@ -346,3 +358,76 @@ def promote_failover(self) -> None:
         self.main_app = self.failover_app
         self.main_rel_id = self.failover_rel_id
         self.delete("failover")
+
+
+class OpenSearchPerfProfile(Model):
+    """Generates an immutable description of the performance profile."""
+
+    class Config:
+        """Pydantic config for this model."""
+
+        arbitrary_types_allowed = True
+
+    typ: PerformanceType
+    heap_size_in_kb: int = MIN_HEAP_SIZE
+    opensearch_yml: Dict[str, str] = {}
+    charmed_index_template: Dict[str, str] = {}
+    charmed_component_templates: Dict[str, str] = {}
+
+    @classmethod
+    def from_str(cls, input_str: str):
+        """Create a new instance of this class from a stringified json/dict repr."""
+        return cls(typ=PerformanceType(input_str))
+
+    @root_validator
+    def set_options(cls, values):  # noqa: N805
+        """Generate the attributes depending on the input."""
+        val = values["typ"]
+        if isinstance(val, str):
+            val = PerformanceType(val)
+
+        if val == PerformanceType.PRODUCTION:
+            values["heap_size_in_kb"] = min(
+                int(0.50 * OpenSearchPerfProfile.meminfo()["MemTotal"]),
+                MAX_HEAP_SIZE,
+            )
+
+        if val == PerformanceType.STAGING:
+            values["heap_size_in_kb"] = min(
+                int(0.25 * OpenSearchPerfProfile.meminfo()["MemTotal"]),
+                MAX_HEAP_SIZE,
+            )
+
+        if val == PerformanceType.TESTING:
+            values["heap_size_in_kb"] = MIN_HEAP_SIZE
+
+        if val != PerformanceType.TESTING:
+            values["opensearch_yml"] = {"indices.memory.index_buffer_size": "25%"}
+
+            values["charmed_index_template"] = {
+                "charmed-index-tpl": {
+                    "index_patterns": ["*"],
+                    "template": {
+                        "settings": {
+                            "number_of_replicas": "2",
+                        },
+                    },
+                },
+            }
+
+        return values
+
+    @staticmethod
+    def meminfo() -> dict[str, float]:
+        """Read the /proc/meminfo file and return the values.
+
+        According to the kernel source code, the values are always in kB:
+            https://github.com/torvalds/linux/blob/
+                2a130b7e1fcdd83633c4aa70998c314d7c38b476/fs/proc/meminfo.c#L31
+        """
+        with open("/proc/meminfo") as f:
+            meminfo = f.read()
+
+        return {
+            line.split()[0][:-1]: float(line.split()[1]) for line in meminfo.split("\n") if line
+        }

lib/charms/opensearch/v0/opensearch_base_charm.py

@@ -11,6 +11,7 @@
 
 from charms.grafana_agent.v0.cos_agent import COSAgentProvider
 from charms.opensearch.v0.constants_charm import (
+    PERFORMANCE_PROFILE,
     AdminUser,
     AdminUserInitProgress,
     AdminUserNotConfigured,
@@ -49,7 +50,12 @@
     generate_hashed_password,
     generate_password,
 )
-from charms.opensearch.v0.models import DeploymentDescription, DeploymentType
+from charms.opensearch.v0.models import (
+    DeploymentDescription,
+    DeploymentType,
+    OpenSearchPerfProfile,
+    PerformanceType,
+)
 from charms.opensearch.v0.opensearch_backups import backup
 from charms.opensearch.v0.opensearch_config import OpenSearchConfig
 from charms.opensearch.v0.opensearch_distro import OpenSearchDistribution
@@ -166,12 +172,21 @@ def __init__(self, handle, *, ignore_lock=False):
         super().__init__(handle, ignore_lock=ignore_lock)
 
 
+class _ApplyProfileTemplatesOpenSearch(EventBase):
+    """Attempt to acquire lock & restart OpenSearch.
+
+    The main reason to have a separate event, is to be able to wait for the cluster.
+    It defers otherwise and only defers the execution of this particular task.
+    """
+
+
 class OpenSearchBaseCharm(CharmBase, abc.ABC):
     """Base class for OpenSearch charms."""
 
     _start_opensearch_event = EventSource(_StartOpenSearch)
     _restart_opensearch_event = EventSource(_RestartOpenSearch)
     _upgrade_opensearch_event = EventSource(_UpgradeOpenSearch)
+    _apply_profile_templates_event = EventSource(_ApplyProfileTemplatesOpenSearch)
 
     def __init__(self, *args, distro: Type[OpenSearchDistribution] = None):
         super().__init__(*args)
@@ -208,6 +223,10 @@ def __init__(self, *args, distro: Type[OpenSearchDistribution] = None):
         self.framework.observe(self._restart_opensearch_event, self._restart_opensearch)
         self.framework.observe(self._upgrade_opensearch_event, self._upgrade_opensearch)
 
+        self.framework.observe(
+            self._apply_profile_templates_event, self._on_apply_profile_templates
+        )
+
         self.framework.observe(self.on.leader_elected, self._on_leader_elected)
         self.framework.observe(self.on.start, self._on_start)
         self.framework.observe(self.on.update_status, self._on_update_status)
@@ -250,6 +269,24 @@ def __init__(self, *args, distro: Type[OpenSearchDistribution] = None):
         # in the deferred event queue
         self._is_peer_rel_changed_deferred = False
 
+    def _on_apply_profile_templates(self, event: EventBase):
+        """Apply the profile templates.
+
+        The main reason to have a separate event, is to be able to wait for the cluster. It
+        defers otherwise and only defers the execution of this particular task.
+        """
+        if not self.opensearch_peer_cm.deployment_desc():
+            logger.info("Applying profile templates but cluster not ready yet.")
+            event.defer()
+            return
+
+        if self.opensearch_peer_cm.deployment_desc().typ != DeploymentType.MAIN_ORCHESTRATOR:
+            logger.info("Applying profile templates but not the main orchestrator.")
+            return
+
+        # Configure templates if needed
+        self.opensearch.apply_perf_templates_if_needed()
+
     @property
     @abc.abstractmethod
     def _upgrade(self) -> typing.Optional[upgrade.Upgrade]:
@@ -355,6 +392,15 @@ def cleanup():
                 logger.error("Service previously started but now misses the snap.")
                 return
 
+        # Store the current perf. profile we are applying
+        if self.unit.is_leader():
+            self.peers_data.put(
+                Scope.APP,
+                PERFORMANCE_PROFILE,
+                PerformanceType(self.config.get(PERFORMANCE_PROFILE, PerformanceType.PRODUCTION)),
+            )
+        self._apply_profile_templates_event.emit()
+
         # apply the directives computed and emitted by the peer cluster manager
         if not self._apply_peer_cm_directives_and_check_if_can_start():
             event.defer()
@@ -676,22 +722,49 @@ def _on_config_changed(self, event: ConfigChangedEvent):  # noqa C901
         if not self.plugin_manager.check_plugin_manager_ready():
             return
 
+        if self.upgrade_in_progress:
+            # The following changes in _on_config_changed are not supported during an upgrade
+            # Therefore, we leave now
+            logger.warning(
+                "Changing config during an upgrade is not supported. The charm may be in a broken, "
+                "unrecoverable state"
+            )
+            event.defer()
+            return
+
         try:
             if not self.plugin_manager.check_plugin_manager_ready():
                 raise OpenSearchNotFullyReadyError()
 
             if self.unit.is_leader():
                 self.status.set(MaintenanceStatus(PluginConfigCheck), app=True)
 
-            if self.plugin_manager.run():
-                if self.upgrade_in_progress:
-                    logger.warning(
-                        "Changing config during an upgrade is not supported. The charm may be in a broken, "
-                        "unrecoverable state"
+            restart_requested = self.plugin_manager.run()
+            if (
+                self.peers_data.get(Scope.APP, PERFORMANCE_PROFILE)
+                and PerformanceType(self.peers_data.get(Scope.APP, PERFORMANCE_PROFILE))
+                != self.opensearch.perf_profile.typ
+            ):
+                self.opensearch.perf_profile = OpenSearchPerfProfile.from_str(
+                    self.config.get(PERFORMANCE_PROFILE)
+                )
+                # If we have a running service, and our profile changed
+                # then we need a restart to apply the new profile
+                self.opensearch_config.apply_performance_profile(self.opensearch.perf_profile)
+                self._apply_profile_templates_event.emit()
+                if self.unit.is_leader():
+                    self.peers_data.put(
+                        Scope.APP,
+                        PERFORMANCE_PROFILE,
+                        PerformanceType(
+                            self.config.get(PERFORMANCE_PROFILE, PerformanceType.PRODUCTION)
+                        ),
                     )
-                    event.defer()
-                    return
+                restart_requested = True
 
+            # Finally, we only need to restart in this case if we have already requested a restart
+            # and the service is actually running
+            if restart_requested and self.opensearch.is_service_started():
                 self._restart_opensearch_event.emit()
         except (OpenSearchNotFullyReadyError, OpenSearchPluginError) as e:
             if isinstance(e, OpenSearchNotFullyReadyError):

lib/charms/opensearch/v0/opensearch_config.py

@@ -8,7 +8,7 @@
 
 from charms.opensearch.v0.constants_tls import CertType
 from charms.opensearch.v0.helper_security import normalized_tls_subject
-from charms.opensearch.v0.models import App
+from charms.opensearch.v0.models import App, OpenSearchPerfProfile
 from charms.opensearch.v0.opensearch_distro import OpenSearchDistribution
 
 # The unique Charmhub library identifier, never change it
@@ -69,6 +69,25 @@ def set_client_auth(self):
             "-Djdk.tls.client.protocols=TLSv1.2",
         )
 
+    def apply_performance_profile(self, profile: OpenSearchPerfProfile):
+        """Apply the performance profile to the opensearch config."""
+        self._opensearch.config.replace(
+            self.JVM_OPTIONS,
+            "-Xms[0-9]+[kmgKMG]",
+            f"-Xms{str(profile.heap_size_in_kb)}k",
+            regex=True,
+        )
+
+        self._opensearch.config.replace(
+            self.JVM_OPTIONS,
+            "-Xmx[0-9]+[kmgKMG]",
+            f"-Xmx{str(profile.heap_size_in_kb)}k",
+            regex=True,
+        )
+
+        for key, val in profile.opensearch_yml.items():
+            self._opensearch.config.put(self.CONFIG_YML, key, val)
+
     def set_admin_tls_conf(self, secrets: Dict[str, any]):
         """Configures the admin certificate."""
         self._opensearch.config.put(

lib/charms/opensearch/v0/opensearch_distro.py

@@ -18,7 +18,7 @@
 
 import requests
 import urllib3.exceptions
-from charms.opensearch.v0.constants_charm import GeneratedRoles
+from charms.opensearch.v0.constants_charm import PERFORMANCE_PROFILE, GeneratedRoles
 from charms.opensearch.v0.helper_charm import (
     format_unit_name,
     mask_sensitive_information,
@@ -27,7 +27,12 @@
 from charms.opensearch.v0.helper_conf_setter import YamlConfigSetter
 from charms.opensearch.v0.helper_http import error_http_retry_log
 from charms.opensearch.v0.helper_networking import get_host_ip, is_reachable
-from charms.opensearch.v0.models import App, StartMode
+from charms.opensearch.v0.models import (
+    App,
+    OpenSearchPerfProfile,
+    PerformanceType,
+    StartMode,
+)
 from charms.opensearch.v0.opensearch_exceptions import (
     OpenSearchCmdError,
     OpenSearchError,
@@ -98,6 +103,9 @@ def __init__(self, charm, peer_relation_name: str):
         self.config = YamlConfigSetter(base_path=self.paths.conf)
         self._charm = charm
         self._peer_relation_name = peer_relation_name
+        self.perf_profile = OpenSearchPerfProfile.from_str(
+            self._charm.config.get(PERFORMANCE_PROFILE, "production")
+        )
 
     def install(self):
         """Install the package."""
@@ -398,6 +406,32 @@ def _build_paths(self) -> Paths:
         """Build the Paths object."""
         pass
 
+    def apply_perf_templates_if_needed(self):  # noqa: C901
+        """Apply performance templates if needed."""
+        if self.perf_profile.typ == PerformanceType.TESTING:
+            # We try to remove the index and components' templates
+            for endpoint in [
+                "/_index_template/charmed-index-tpl",
+            ]:
+                try:
+                    self.request("DELETE", endpoint)
+                except OpenSearchHttpError:
+                    pass
+            # Nothing to do anymore
+            return
+
+        for idx, template in self.perf_profile.charmed_index_template.items():
+            try:
+                self.request("POST", f"/_index_template/{idx}", template)
+            except OpenSearchHttpError as e:
+                logger.error(f"Failed to apply index template: {e}")
+
+        for idx, template in self.perf_profile.charmed_component_templates.items():
+            try:
+                self.request("POST", f"/_component_template/{idx}", template)
+            except OpenSearchHttpError as e:
+                logger.error(f"Failed to apply component template: {e}")
+
     def _set_env_variables(self):
         """Set the necessary environment variables."""
         os.environ["OPENSEARCH_HOME"] = self.paths.home