ceph/src/pybind/mgr/dashboard/plugins/motd.py

   1 # -*- coding: utf-8 -*-
   2
   3 import hashlib
   4 import json
   5 from enum import Enum
   6 from typing import Dict, NamedTuple, Optional
   7
   8 from ceph.utils import datetime_now, datetime_to_str, parse_timedelta, str_to_datetime
   9 from mgr_module import CLICommand
  10
  11 from . import PLUGIN_MANAGER as PM
  12 from .plugin import SimplePlugin as SP
  13
  14
  15 class MotdSeverity(Enum):
  16     INFO = 'info'
  17     WARNING = 'warning'
  18     DANGER = 'danger'
  19
  20
  21 class MotdData(NamedTuple):
  22     message: str
  23     md5: str  # The MD5 of the message.
  24     severity: MotdSeverity
  25     expires: str  # The expiration date in ISO 8601. Does not expire if empty.
  26
  27
  28 @PM.add_plugin  # pylint: disable=too-many-ancestors
  29 class Motd(SP):
  30     NAME = 'motd'
  31
  32     OPTIONS = [
  33         SP.Option(
  34             name=NAME,
  35             default='',
  36             type='str',
  37             desc='The message of the day'
  38         )
  39     ]
  40
  41     @PM.add_hook
  42     def register_commands(self):
  43         @CLICommand("dashboard {name} get".format(name=self.NAME))
  44         def _get(_):
  45             stdout: str
  46             value: str = self.get_option(self.NAME)
  47             if not value:
  48                 stdout = 'No message of the day has been set.'
  49             else:
  50                 data = json.loads(value)
  51                 if not data['expires']:
  52                     data['expires'] = "Never"
  53                 stdout = 'Message="{message}", severity="{severity}", ' \
  54                          'expires="{expires}"'.format(**data)
  55             return 0, stdout, ''
  56
  57         @CLICommand("dashboard {name} set".format(name=self.NAME),
  58                     "name=severity,type=CephChoices,strings={} ".format(
  59                         "|".join(s.value for s in MotdSeverity))
  60                     + "name=expires,type=CephString "
  61                     + "name=message,type=CephString")
  62         def _set(_, severity: str, expires: str, message: str):
  63             if expires != '0':
  64                 delta = parse_timedelta(expires)
  65                 if not delta:
  66                     return 1, '', 'Invalid expires format, use "2h", "10d" or "30s"'
  67                 expires = datetime_to_str(datetime_now() + delta)
  68             else:
  69                 expires = ''
  70             value: str = json.dumps({
  71                 'message': message,
  72                 'md5': hashlib.md5(message.encode()).hexdigest(),
  73                 'severity': severity,
  74                 'expires': expires
  75             })
  76             self.set_option(self.NAME, value)
  77             return 0, 'Message of the day has been set.', ''
  78
  79         @CLICommand("dashboard {name} clear".format(name=self.NAME))
  80         def _clear(_):
  81             self.set_option(self.NAME, '')
  82             return 0, 'Message of the day has been cleared.', ''
  83
  84     @PM.add_hook
  85     def get_controllers(self):
  86         from ..controllers import RESTController, UiApiController
  87
  88         @UiApiController('/motd')
  89         class MessageOfTheDay(RESTController):
  90             def list(_) -> Optional[Dict]:  # pylint: disable=no-self-argument
  91                 value: str = self.get_option(self.NAME)
  92                 if not value:
  93                     return None
  94                 data: MotdData = MotdData(**json.loads(value))
  95                 # Check if the MOTD has been expired.
  96                 if data.expires:
  97                     expires = str_to_datetime(data.expires)
  98                     if expires < datetime_now():
  99                         return None
 100                 return data._asdict()
 101
 102         return [MessageOfTheDay]