[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

@@ -3,10 +3,12 @@
 
 """Cluster-related data structures / model classes."""
 import json
+import math
 from abc import ABC
 from datetime import datetime
+from enum import Enum
 from hashlib import md5
-from typing import Any, Dict, List, Literal, Optional
+from typing import Any, Dict, List, Literal, Optional, Tuple
 
 from charms.opensearch.v0.helper_enums import BaseStrEnum
 from pydantic import BaseModel, Field, root_validator, validator
@@ -153,6 +155,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 +356,246 @@ def promote_failover(self) -> None:
         self.main_app = self.failover_app
         self.main_rel_id = self.failover_rel_id
         self.delete("failover")
+
+
+class ByteUnit(Enum):
+    """As per docs, Java uses the byte format.
+
+    Converts the *B and *iB to the same raw values. For example, can be written as:
+    - 6m: 6 * 1024 * 1024
+    - 6144k: 6144 * 1024
+    - 6291456: 6291456 bytes
+
+    More info: https://dev.java/learn/jvm/tools/core/java/#overview
+    """
+
+    B = 1  # noqa: N815
+    kB = 1024  # noqa: N815
+    mB = 1024 * kB  # noqa: N815
+    gB = 1024 * mB  # noqa: N815
+
+    @staticmethod
+    def get(name: str) -> int:
+        """Convert the value to the required unit."""
+        val = name.lower()
+        if val == "kb" or val == "k":
+            return ByteUnit.kB
+        if val == "mb" or val == "m":
+            return ByteUnit.mB
+        if val == "gb" or val == "g":
+            return ByteUnit.gB
+        return ByteUnit.B
+
+    @staticmethod
+    def previous(val):
+        """Return the previous value of the unit."""
+        if val == ByteUnit.kB:
+            return ByteUnit.B
+        if val == ByteUnit.mB:
+            return ByteUnit.kB
+        if val == ByteUnit.gB:
+            return ByteUnit.mB
+        return ByteUnit.B
+
+    @staticmethod
+    def to_int(value: tuple[str, any]) -> int:
+        """Convert the value to the bytes unit."""
+        if isinstance(value[1], ByteUnit):
+            return value[0] * value[1].value
+
+        unit = ByteUnit.get(value[1]).value
+        return int(value[0]) * unit
+
+    @staticmethod
+    def unit(value: int | float) -> Tuple[float, Any]:
+        """Return the next value of the unit.
+
+        This value must be an integer. If we have a decimal part, then we should round it up.
+        """
+        inter_value = float(value)
+        for u in [ByteUnit.B, ByteUnit.kB, ByteUnit.mB, ByteUnit.gB]:
+            if inter_value < 1024:
+                break
+            inter_value /= 1024
+
+        # Now, we calculate the rounding
+        if u == ByteUnit.B:
+            # We are already in the lowest unit possible, return a rounded value
+            return (int(inter_value), u)
+        # Check if we have a decimal part, if yes, then we multiply the value by 1024
+        dec, _ = math.modf(inter_value)
+        if dec != 0.0:
+            return (int(inter_value * 1024), ByteUnit.previous(u))
+        return (int(inter_value), u)
+
+
+class JavaByteSize:
+    """Java Byte Size tuple representation."""
+
+    def __init__(self, value: str | float | int | None = None, unit: str | ByteUnit | None = None):
+        """Constructor of JavaByteSize.
+
+        Args:
+            value: the value of the size
+            unit: the unit of the size
+        """
+        if not value and not unit:
+            self.value = 0
+            self.unit = ByteUnit.B
+            return
+
+        u = unit
+        if isinstance(unit, str):
+            u = ByteUnit.get(unit)
+        self.value, self.unit = ByteUnit.unit(float(value) * u.value)
+
+    def percent(self, percentage: float) -> int:
+        """Return the percentage of the JavaByteSize."""
+        val = ByteUnit.to_int((self.value, self.unit)) * percentage
+        return JavaByteSize(val, ByteUnit.B)
+
+    def __eq__(self, other: Any) -> bool:
+        """Check if the JavaByteSize is equal to the other value."""
+        if not isinstance(other, JavaByteSize):
+            raise TypeError("Cannot compare JavaByteSize with other types.")
+        return ByteUnit.to_int((self.value, self.unit)) == ByteUnit.to_int(
+            (other.value, other.unit)
+        )
+
+    def __lt__(self, other: Any) -> bool:
+        """Check if the JavaByteSize is less than the other value."""
+        if not isinstance(other, JavaByteSize):
+            raise TypeError("Cannot compare JavaByteSize with other types.")
+        return ByteUnit.to_int((self.value, self.unit)) < ByteUnit.to_int(
+            (other.value, other.unit)
+        )
+
+    def __gt__(self, other: Any) -> bool:
+        """Check if the JavaByteSize is greater than the other value."""
+        if not isinstance(other, JavaByteSize):
+            raise TypeError("Cannot compare JavaByteSize with other types.")
+        return ByteUnit.to_int((self.value, self.unit)) > ByteUnit.to_int(
+            (other.value, other.unit)
+        )
+
+    def __str__(self) -> str:
+        """Return the string representation of the JavaByteSize."""
+        return f"{self.value}{str(self.unit.name.lower())[:1]}"
+
+
+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: JavaByteSize | None = None
+    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=input_str)
+
+    @root_validator
+    def set_options(cls, values):  # noqa: N805
+        """Generate the attributes depending on the input."""
+        heap = JavaByteSize(
+            OpenSearchPerfProfile.meminfo()["MemTotal"][0],
+            OpenSearchPerfProfile.meminfo()["MemTotal"][1],
+        )
+
+        val = values["typ"]
+        if isinstance(val, str):
+            val = PerformanceType(val)
+
+        if val == PerformanceType.PRODUCTION:
+            values["heap_size"] = (
+                heap.percent(0.25)
+                if heap.percent(0.25) > JavaByteSize("1", "g")
+                else JavaByteSize("1", "g")
+            )
+
+        if val == PerformanceType.STAGING:
+            values["heap_size"] = (
+                heap.percent(0.1)
+                if heap.percent(0.1) > JavaByteSize("1", "g")
+                else JavaByteSize("1", "g")
+            )
+
+        if val == PerformanceType.TESTING:
+            values["heap_size"] = JavaByteSize("1", "gB")
+
+        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": "1-all",
+                        },
+                    },
+                },
+            }
+
+            values["charmed_component_templates"] = {
+                "charmed-default-tpl": {
+                    "template": {
+                        "settings": {
+                            "number_of_replicas": "1-all",
+                            "index": {
+                                "codec": "zstd_no_dict",
+                            },
+                        },
+                    },
+                },
+                "charmed-vector-tpl": {
+                    "template": {
+                        "settings": {
+                            "number_of_replicas": "1-all",
+                            "index": {
+                                "codec": "default",
+                            },
+                        },
+                    },
+                },
+                "charmed-ingest-tpl": {
+                    "template": {
+                        "settings": {
+                            "number_of_replicas": "1-all",
+                            "index": {
+                                "codec": "zstd_no_dict",
+                                "flush_threshold_size": (
+                                    str(values["heap_size"].percent(0.25))
+                                    if values["heap_size"].percent(0.25)
+                                    > JavaByteSize("512", "mB")
+                                    else "512m"
+                                ),
+                            },
+                        },
+                    },
+                },
+            }
+
+        return values
+
+    @staticmethod
+    def meminfo() -> dict[str, JavaByteSize]:
+        """Read the /proc/meminfo file and return the values."""
+        with open("/proc/meminfo") as f:
+            meminfo = f.read()
+        return {
+            line.split()[0][:-1]: (
+                int(line.split()[1]),
+                ByteUnit.get(line.split()[2] if len(line.split()) > 2 else "b"),
+            )
+            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
@@ -355,6 +361,13 @@ def cleanup():
                 logger.error("Service previously started but now misses the snap.")
                 return
 
+        # Store the current perf. profile we are applying
+        self.peers_data.put(
+            Scope.UNIT,
+            PERFORMANCE_PROFILE,
+            PerformanceType(self._charm.config.get(PERFORMANCE_PROFILE, "production")),
+        )
+
         # 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 +689,41 @@ def _on_config_changed(self, event: ConfigChangedEvent):  # noqa C901
         if not self.plugin_manager.check_plugin_manager_ready():
             return
 
+        if self.upgrade_in_progress:
+            # Deferring right now is too late anyways
+            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"
-                    )
-                    event.defer()
-                    return
+            restart_requested = self.plugin_manager.run()
+            if (
+                PerformanceType(self.peers_data.get(Scope.UNIT, PERFORMANCE_PROFILE))
+                != self.opensearch.perf_profile
+            ):
+                self.opensearch.perf_profile = OpenSearchPerfProfile.from_str(
+                    self._charm.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)
+
+                # Configure templates if needed
+                self.opensearch.apply_perf_templates_if_neeeded()
+
+                self.peers_data.put(Scope.UNIT, PERFORMANCE_PROFILE, self.opensearch.perf_profile)
+                restart_requested = self.opensearch.is_service_started()
 
+            if restart_requested:
                 self._restart_opensearch_event.emit()
         except (OpenSearchNotFullyReadyError, OpenSearchPluginError) as e:
             if isinstance(e, OpenSearchNotFullyReadyError):