installers/charm/pol/src/pod_spec.py

   1 #!/usr/bin/env python3
   2 # Copyright 2020 Canonical Ltd.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: legal@canonical.com
  18 #
  19 # To get in touch with the maintainers, please contact:
  20 # osm-charmers@lists.launchpad.net
  21 ##
  22
  23 import logging
  24 from pydantic import BaseModel, constr, PositiveInt
  25 from typing import Any, Dict, List
  26
  27 logger = logging.getLogger(__name__)
  28
  29
  30 class ConfigData(BaseModel):
  31     """Configuration data model."""
  32
  33     log_level: constr(regex=r"^(INFO|DEBUG)$")
  34
  35
  36 class RelationData(BaseModel):
  37     """Relation data model."""
  38
  39     message_host: constr(min_length=1)
  40     message_port: PositiveInt
  41     database_uri: constr(regex=r"^(mongodb://)")
  42
  43
  44 def _make_pod_ports(port: int) -> List[Dict[str, Any]]:
  45     """Generate pod ports details.
  46
  47     Args:
  48         port (int): port to expose.
  49
  50     Returns:
  51         List[Dict[str, Any]]: pod port details.
  52     """
  53     return [{"name": "pol", "containerPort": port, "protocol": "TCP"}]
  54
  55
  56 def _make_pod_envconfig(
  57     config: Dict[str, Any], relation_state: Dict[str, Any]
  58 ) -> Dict[str, Any]:
  59     """Generate pod environment configuration.
  60
  61     Args:
  62         config (Dict[str, Any]): configuration information.
  63         relation_state (Dict[str, Any]): relation state information.
  64
  65     Returns:
  66         Dict[str, Any]: pod environment configuration.
  67     """
  68     envconfig = {
  69         # General configuration
  70         "ALLOW_ANONYMOUS_LOGIN": "yes",
  71         "OSMPOL_GLOBAL_LOGLEVEL": config["log_level"],
  72         # Kafka configuration
  73         "OSMPOL_MESSAGE_HOST": relation_state["message_host"],
  74         "OSMPOL_MESSAGE_DRIVER": "kafka",
  75         "OSMPOL_MESSAGE_PORT": relation_state["message_port"],
  76         # Database configuration
  77         "OSMPOL_DATABASE_DRIVER": "mongo",
  78         "OSMPOL_DATABASE_URI": relation_state["database_uri"],
  79     }
  80
  81     return envconfig
  82
  83
  84 def _make_startup_probe() -> Dict[str, Any]:
  85     """Generate startup probe.
  86
  87     Returns:
  88         Dict[str, Any]: startup probe.
  89     """
  90     return {
  91         "exec": {"command": ["/usr/bin/pgrep", "python3"]},
  92         "initialDelaySeconds": 60,
  93         "timeoutSeconds": 5,
  94     }
  95
  96
  97 def _make_readiness_probe() -> Dict[str, Any]:
  98     """Generate readiness probe.
  99
 100     Returns:
 101         Dict[str, Any]: readiness probe.
 102     """
 103     return {
 104         "exec": {
 105             "command": ["sh", "-c", "osm-pol-healthcheck || exit 1"],
 106         },
 107         "periodSeconds": 10,
 108         "timeoutSeconds": 5,
 109         "successThreshold": 1,
 110         "failureThreshold": 3,
 111     }
 112
 113
 114 def _make_liveness_probe() -> Dict[str, Any]:
 115     """Generate liveness probe.
 116
 117     Returns:
 118         Dict[str, Any]: liveness probe.
 119     """
 120     return {
 121         "exec": {
 122             "command": ["sh", "-c", "osm-pol-healthcheck || exit 1"],
 123         },
 124         "initialDelaySeconds": 45,
 125         "periodSeconds": 10,
 126         "timeoutSeconds": 5,
 127         "successThreshold": 1,
 128         "failureThreshold": 3,
 129     }
 130
 131
 132 def make_pod_spec(
 133     image_info: Dict[str, str],
 134     config: Dict[str, Any],
 135     relation_state: Dict[str, Any],
 136     app_name: str = "pol",
 137     port: int = 80,
 138 ) -> Dict[str, Any]:
 139     """Generate the pod spec information.
 140
 141     Args:
 142         image_info (Dict[str, str]): Object provided by
 143                                      OCIImageResource("image").fetch().
 144         config (Dict[str, Any]): Configuration information.
 145         relation_state (Dict[str, Any]): Relation state information.
 146         app_name (str, optional): Application name. Defaults to "pol".
 147         port (int, optional): Port for the container. Defaults to 80.
 148
 149     Returns:
 150         Dict[str, Any]: Pod spec dictionary for the charm.
 151     """
 152     if not image_info:
 153         return None
 154
 155     ConfigData(**(config))
 156     RelationData(**(relation_state))
 157
 158     ports = _make_pod_ports(port)
 159     env_config = _make_pod_envconfig(config, relation_state)
 160
 161     return {
 162         "version": 3,
 163         "containers": [
 164             {
 165                 "name": app_name,
 166                 "imageDetails": image_info,
 167                 "imagePullPolicy": "Always",
 168                 "ports": ports,
 169                 "envConfig": env_config,
 170             }
 171         ],
 172         "kubernetesResources": {
 173             "ingressResources": [],
 174         },
 175     }