File: classdef.py

package info (click to toggle)
elasticsearch-curator 8.0.21-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 2,716 kB
  • sloc: python: 17,838; makefile: 159; sh: 156
file content (245 lines) | stat: -rw-r--r-- 9,732 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
"""Other Classes"""

import logging
from es_client.exceptions import FailedValidation
from es_client.helpers.schemacheck import password_filter
from es_client.helpers.utils import get_yaml
from curator import IndexList, SnapshotList
from curator.actions import CLASS_MAP
from curator.exceptions import ConfigurationError
from curator.helpers.testers import validate_actions

# Let me tell you the story of the nearly wasted afternoon and the research that went
# into this seemingly simple work-around. Actually, no. It's even more wasted time
# writing that story here. Suffice to say that I couldn't use the CLASS_MAP with class
# objects to directly map them to class instances. The Wrapper class and the
# ActionDef.instantiate method do all of the work for me, allowing me to easily and
# cleanly pass *args and **kwargs to the individual action classes of CLASS_MAP.


class Wrapper:
    """Wrapper Class"""

    def __init__(self, cls):
        """Instantiate with passed Class (not instance or object)

        :param cls: A class (not an instance of the class)
        """
        #: The class itself (not an instance of it), passed from ``cls``
        self.class_object = cls
        #: An instance of :py:attr:`class_object`
        self.class_instance = None

    def set_instance(self, *args, **kwargs):
        """Set up :py:attr:`class_instance` from :py:attr:`class_object`"""
        self.class_instance = self.class_object(*args, **kwargs)

    def get_instance(self, *args, **kwargs):
        """Return the instance with ``*args`` and ``**kwargs``"""
        self.set_instance(*args, **kwargs)
        return self.class_instance


class ActionsFile:
    """Class to parse and verify entire actions file

    Individual actions are :py:class:`~.curator.classdef.ActionDef` objects
    """

    def __init__(self, action_file):
        self.logger = logging.getLogger(__name__)
        #: The full, validated configuration from ``action_file``.
        self.fullconfig = self.get_validated(action_file)
        self.logger.debug('Action Configuration: %s', password_filter(self.fullconfig))
        #: A dict of all actions in the provided configuration. Each original key name
        #: is preserved and the value is now an
        #: :py:class:`~.curator.classdef.ActionDef`, rather than a dict.
        self.actions = None
        self.set_actions(self.fullconfig['actions'])

    def get_validated(self, action_file):
        """
        :param action_file: The path to a valid YAML action configuration file
        :type action_file: str

        :returns: The result from passing ``action_file`` to
            :py:func:`~.curator.helpers.testers.validate_actions`
        """
        try:
            return validate_actions(get_yaml(action_file))
        except (FailedValidation, UnboundLocalError) as err:
            self.logger.critical('Configuration Error: %s', err)
            raise ConfigurationError from err

    def parse_actions(self, all_actions):
        """Parse the individual actions found in ``all_actions['actions']``

        :param all_actions: All actions, each its own dictionary behind a numeric key.
            Making the keys numeric guarantees that if they are sorted, they will
            always be executed in order.

        :type all_actions: dict

        :returns:
        :rtype: list of :py:class:`~.curator.classdef.ActionDef`
        """
        acts = {}
        for idx in all_actions.keys():
            acts[idx] = ActionDef(all_actions[idx])
        return acts

    def set_actions(self, all_actions):
        """Set the actions via :py:meth:`~.curator.classdef.ActionsFile.parse_actions`

        :param all_actions: All actions, each its own dictionary behind a numeric key.
            Making the keys numeric guarantees that if they are sorted, they will
            always be executed in order.
        :type all_actions: dict
        :rtype: None
        """
        self.actions = self.parse_actions(all_actions)


# In this case, I just don't care that pylint thinks I'm overdoing it with attributes
# pylint: disable=too-many-instance-attributes
class ActionDef:
    """Individual Action Definition Class

    Instances of this class represent an individual action from an action file.
    """

    def __init__(self, action_dict):
        #: The whole action dictionary
        self.action_dict = action_dict
        #: The action name
        self.action = None
        #: The action's class (Alias, Allocation, etc.)
        self.action_cls = None
        #: Only when action is alias will this be a :py:class:`~.curator.IndexList`
        self.alias_adds = None
        #: Only when action is alias will this be a :py:class:`~.curator.IndexList`
        self.alias_removes = None
        #: The list class, either :py:class:`~.curator.IndexList` or
        #: :py:class:`~.curator.SnapshotList`. Default is
        #: :py:class:`~.curator.IndexList`
        self.list_obj = Wrapper(IndexList)
        #: The action ``description``
        self.description = None
        #: The action ``options`` :py:class:`dict`
        self.options = {}
        #: The action ``filters`` :py:class:`list`
        self.filters = None
        #: The action option ``disable_action``
        self.disabled = None
        #: The action option ``continue_if_exception``
        self.cif = None
        #: The action option ``timeout_override``
        self.timeout_override = None
        #: The action option ``ignore_empty_list``
        self.iel = None
        #: The action option ``allow_ilm_indices``
        self.allow_ilm = None
        self.set_root_attrs()
        self.set_option_attrs()
        self.log_the_options()
        self.get_action_class()

    def instantiate(self, attribute, *args, **kwargs):
        """
        Convert ``attribute`` from being a :py:class:`~.curator.classdef.Wrapper` of a
        Class to an instantiated object of that Class.

        This is madness or genius. You decide. This entire method plus the
        :py:class:`~.curator.classdef.Wrapper` class came about because I couldn't
        cleanly instantiate a class variable into a class object. It works, and that's
        good enough for me.

        :param attribute: The `name` of an attribute that references a Wrapper class
            instance
        :type attribute: str
        """
        try:
            wrapper = getattr(self, attribute)
        except AttributeError as exc:
            raise AttributeError(
                f'Bad Attribute: {attribute}. Exception: {exc}'
            ) from exc
        setattr(self, attribute, self.get_obj_instance(wrapper, *args, **kwargs))

    def get_obj_instance(self, wrapper, *args, **kwargs):
        """Get the class instance wrapper identified by ``wrapper``
        Pass all other args and kwargs to the
        :py:meth:`~.curator.classdef.Wrapper.get_instance` method.

        :returns: An instance of the class that :py:class:`~.curator.classdef.Wrapper`
            is wrapping
        """
        if not isinstance(wrapper, Wrapper):
            raise ConfigurationError(
                f'{__name__} was passed wrapper which was of type {type(wrapper)}'
            )
        return wrapper.get_instance(*args, **kwargs)

    def set_alias_extras(self):
        """Populate the :py:attr:`alias_adds` and :py:attr:`alias_removes` attributes"""
        self.alias_adds = Wrapper(IndexList)
        self.alias_removes = Wrapper(IndexList)

    def get_action_class(self):
        """Get the action class from :py:const:`~.curator.actions.CLASS_MAP`

        Do extra setup when action is ``alias``

        Set :py:attr:`list_obj` to :py:class:`~.curator.SnapshotList` when
        :py:attr:`~.curator.classdef.ActionDef.action` is ``delete_snapshots`` or
        ``restore``
        """

        self.action_cls = Wrapper(CLASS_MAP[self.action])
        if self.action == 'alias':
            self.set_alias_extras()
        if self.action in ['delete_snapshots', 'restore']:
            self.list_obj = Wrapper(SnapshotList)

    def set_option_attrs(self):
        """
        Iteratively get the keys and values from
        :py:attr:`~.curator.classdef.ActionDef.options` and set the attributes
        """
        attmap = {
            'disable_action': 'disabled',
            'continue_if_exception': 'cif',
            'ignore_empty_list': 'iel',
            'allow_ilm_indices': 'allow_ilm',
            'timeout_override': 'timeout_override',
        }
        for key in self.action_dict['options']:
            if key in attmap:
                setattr(self, attmap[key], self.action_dict['options'][key])
            else:
                self.options[key] = self.action_dict['options'][key]

    def set_root_attrs(self):
        """
        Iteratively get the keys and values from
        :py:attr:`~.curator.classdef.ActionDef.action_dict` and set the attributes
        """
        for key, value in self.action_dict.items():
            # Gonna grab options in get_option_attrs()
            if key == 'options':
                continue
            if value is not None:
                setattr(self, key, value)

    def log_the_options(self):
        """Log options at initialization time"""
        logger = logging.getLogger('curator.cli.ActionDef')
        msg = (
            f'For action {self.action}: disable_action={self.disabled}'
            f'continue_if_exception={self.cif}, '
            f'timeout_override={self.timeout_override}, '
            f'ignore_empty_list={self.iel}, allow_ilm_indices={self.allow_ilm}'
        )
        logger.debug(msg)
        if self.allow_ilm:
            logger.warning('Permitting operation on indices with an ILM policy')