src/charm.py

#! /usr/bin/env python3

# Copyright 2021 Canonical Ltd
#
# 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 dataclasses
import logging
import pathlib

from ops_openstack.plugins.classes import CinderStoragePluginCharm
from ops_openstack.core import charm_class, get_charm_class
from ops.main import main
from ops.model import BlockedStatus

logger = logging.getLogger(__name__)


@dataclasses.dataclass(frozen=True)
class StorPoolConfItems:
    sp_api_http_host: str
    sp_api_http_port: str
    sp_auth_token: str

    @classmethod
    def from_config(cls, data) -> "StorPoolConfItems":
        args = {
            field.name: str(data[field.name.replace("_", "-")]) for field in dataclasses.fields(cls)
        }
        return cls(**args)

    def to_ini_key_value_pairs(self) -> str:
        return "".join(
            f"{name.upper()}={value}\n" for name, value in dataclasses.asdict(self).items()
        )


class CinderCharmBase(CinderStoragePluginCharm):

    PACKAGES = ["charm-cinder-storpool-deps", "cinder-common"]
    MANDATORY_CONFIG = [
        "protocol",
        "storpool-template",
        "sp-api-http-host",
        "sp-api-http-port",
        "sp-auth-token",
    ]
    # Overriden from the parent. May be set depending on the charm's properties
    stateless = True
    active_active = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def _check_for_config_errors(self, config):
        missing = []
        for mandatory in self.MANDATORY_CONFIG:
            if mandatory not in config:
                missing.append(mandatory)

        if missing:
            return f"Mandatory options are missing: {', '.join(missing)}"

        if config["protocol"] not in ["block", "iscsi"]:
            return (
                f"""Invalid 'protocol' option provided: '{config["protocol"]}';"""
                "valid are 'block' and 'iscsi'"
            )

        if config["protocol"] == "block":
            return "'protocol' value 'block' not yet supported"

        if not (0 < config["sp-api-http-port"] < 65536):
            return (
                f"""'sp-api-http-port' ('{config["sp-api-http-port"]}')"""
                "is not a valid port (0-65535)"
            )

    def on_config(self, event):
        config = dict(self.framework.model.config)
        conf_error = self._check_for_config_errors(config)
        if conf_error is not None:
            logger.error(conf_error)
            self.unit.status = BlockedStatus(conf_error)
            self._stored.is_started = False

            return

        self.create_storpool_conf(StorPoolConfItems.from_config(config))

        super().on_config(event)

        self._stored.is_started = True

    def cinder_configuration(self, config):
        conf_error = self._check_for_config_errors(config)
        if conf_error is not None:
            logger.error(conf_error)
            self._stored.is_started = False

            return []

        # Return the configuration to be set by the principal.
        backend_name = config.get("volume-backend-name", self.framework.model.app.name)
        volume_driver = "cinder.volume.drivers.storpool.StorPoolDriver"

        options = [
            ("volume_driver", volume_driver),
            ("volume_backend_name", backend_name),
            ("storpool_template", config["storpool-template"]),
            ("sp_api_http_host", config["sp-api-http-host"]),
            ("sp_api_http_port", config["sp-api-http-port"]),
            ("sp_auth_token", config["sp-auth-token"]),
        ]

        if config.get("use-multipath"):
            options.extend(
                [
                    ("use_multipath_for_image_xfer", True),
                    ("enforce_multipath_for_image_xfer", True),
                ]
            )

        self.create_storpool_conf(StorPoolConfItems.from_config(config))

        self._stored.is_started = True

        return options

    @staticmethod
    def create_storpool_conf(sp_conf_items: StorPoolConfItems):
        pathlib.Path("/etc/storpool.conf").write_text(
            "# Do not edit; this file is generated by the cinder-storpool charm.\n"
            + sp_conf_items.to_ini_key_value_pairs(),
            encoding="UTF-8",
        )


@charm_class
class CinderStorPoolCharm(CinderCharmBase):
    release = "yoga"


if __name__ == "__main__":
    # main(get_charm_class_for_release())
    # main(CinderStorPoolCharm)
    main(get_charm_class(release="yoga"))