forked from canonical/operator
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathcharm.py
2050 lines (1576 loc) · 72.3 KB
/
charm.py
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
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# Copyright 2019 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.
"""Base objects for the Charm, events and metadata."""
import dataclasses
import enum
import logging
import pathlib
import warnings
from typing import (
TYPE_CHECKING,
Any,
ClassVar,
Dict,
List,
Literal,
Mapping,
NoReturn,
Optional,
TextIO,
Tuple,
TypedDict,
Union,
cast,
)
from . import model
from ._private import yaml
from .framework import (
EventBase,
EventSource,
Framework,
Handle,
LifecycleEvent,
Object,
ObjectEvents,
)
if TYPE_CHECKING:
from typing_extensions import Required
_Scopes = Literal['global', 'container']
_RelationMetaDict = TypedDict(
'_RelationMetaDict',
{'interface': Required[str], 'limit': int, 'optional': bool, 'scope': _Scopes},
total=False,
)
_MultipleRange = TypedDict('_MultipleRange', {'range': str})
_StorageMetaDict = TypedDict(
'_StorageMetaDict',
{
'type': Required[str],
'description': str,
'shared': bool,
'read-only': bool,
'minimum-size': str,
'location': str,
'multiple-range': str,
'multiple': _MultipleRange,
},
total=False,
)
_ResourceMetaDict = TypedDict(
'_ResourceMetaDict',
{'type': Required[str], 'filename': str, 'description': str},
total=False,
)
_MountDict = TypedDict('_MountDict', {'storage': Required[str], 'location': str}, total=False)
class _ContainerBaseDict(TypedDict):
name: str
channel: str
architectures: List[str]
logger = logging.getLogger(__name__)
class HookEvent(EventBase):
"""Events raised by Juju to progress a charm's lifecycle.
Hooks are callback methods of a charm class (a subclass of
:class:`CharmBase`) that are invoked in response to events raised
by Juju. These callback methods are the means by which a charm
governs the lifecycle of its application.
The :class:`HookEvent` class is the base of a type hierarchy of events
related to the charm's lifecycle.
:class:`HookEvent` subtypes are grouped into the following categories
- Core lifecycle events
- Relation events
- Storage events
- Metric events
"""
class ActionEvent(EventBase):
"""Events raised by Juju when an administrator invokes a Juju Action.
This class is the data type of events triggered when an administrator
invokes a Juju Action. Callbacks bound to these events may be used
for responding to the administrator's Juju Action request.
To read the parameters for the action, see the instance variable
:attr:`~ops.ActionEvent.params`.
To respond with the result of the action, call
:meth:`~ops.ActionEvent.set_results`. To add progress messages that are
visible as the action is progressing use :meth:`~ops.ActionEvent.log`.
"""
id: str = ''
"""The Juju ID of the action invocation."""
params: Dict[str, Any]
"""The parameters passed to the action."""
def __init__(self, handle: 'Handle', id: Optional[str] = None):
super().__init__(handle)
self.id = id # type: ignore (for backwards compatibility)
def defer(self) -> NoReturn:
"""Action events are not deferrable like other events.
This is because an action runs synchronously and the administrator
is waiting for the result.
Raises:
RuntimeError: always.
"""
raise RuntimeError('cannot defer action events')
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to record the action.
Not meant to be called directly by charm code.
"""
self.id = cast(str, snapshot['id'])
# Params are loaded at restore rather than __init__ because
# the model is not available in __init__.
self.params = self.framework.model._backend.action_get()
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
return {'id': self.id}
def set_results(self, results: Dict[str, Any]):
"""Report the result of the action.
Juju eventually only accepts a str:str mapping, so we will attempt
to flatten any more complex data structure like so::
>>> {'a': 'b'} # becomes: 'a'='b'
>>> {'a': {'b': 'c'}} # becomes: 'a.b'='c'
>>> {'a': {'b': 'c', 'd': 'e'}} # becomes: 'a.b'='c', 'a.d' = 'e'
>>> {'a.b': 'c', 'a.d': 'e'} # equivalent to previous
Note that duplicate keys are not allowed, so this is invalid::
>>> {'a': {'b': 'c'}, 'a.b': 'c'}
Note that the resulting keys must start and end with lowercase
alphanumeric, and can only contain lowercase alphanumeric, hyphens
and periods.
Because results are passed to Juju using the command line, the maximum
size is around 100KB. However, actions results are designed to be
small: a few key-value pairs shown in the Juju CLI. If larger content
is needed, store it in a file and use something like ``juju scp``.
If any exceptions occur whilst the action is being handled, juju will
gather any stdout/stderr data (and the return code) and inject them into the
results object. Thus, the results object might contain the following keys,
additionally to those specified by the charm code:
- Stdout
- Stderr
- Stdout-encoding
- Stderr-encoding
- ReturnCode
Args:
results: The result of the action as a Dict
Raises:
ModelError: if a reserved key is used.
ValueError: if ``results`` has a mix of dotted/non-dotted keys that expand out to
result in duplicate keys, for example: :code:`{'a': {'b': 1}, 'a.b': 2}`. Also
raised if a dict is passed with a key that fails to meet the format requirements.
OSError: if extremely large (>100KB) results are provided.
"""
self.framework.model._backend.action_set(results)
def log(self, message: str):
"""Send a message that a user will see while the action is running.
Args:
message: The message for the user.
"""
self.framework.model._backend.action_log(message)
def fail(self, message: str = ''):
"""Report that this action has failed.
Args:
message: Optional message to record why it has failed.
"""
self.framework.model._backend.action_fail(message)
class InstallEvent(HookEvent):
"""Event triggered when a charm is installed.
This event is triggered at the beginning of a charm's
lifecycle. Any associated callback method should be used to
perform one-time setup operations, such as installing prerequisite
software.
"""
class StartEvent(HookEvent):
"""Event triggered immediately after first configuration change.
This event is triggered immediately after the first
:class:`~ops.ConfigChangedEvent`. Callback methods bound to the event should be
used to ensure that the charm's software is in a running state. Note that
the charm's software should be configured so as to persist in this state
through reboots without further intervention on Juju's part.
"""
class StopEvent(HookEvent):
"""Event triggered when a charm is shut down.
This event is triggered when an application's removal is requested
by the client. The event fires immediately before the end of the
unit's destruction sequence. Callback methods bound to this event
should be used to ensure that the charm's software is not running,
and that it will not start again on reboot.
"""
def defer(self) -> NoReturn:
"""Stop events are not deferrable like other events.
This is because the unit is in the process of tearing down, and there
will not be an opportunity for the deferred event to run.
Raises:
RuntimeError: always.
"""
raise RuntimeError('cannot defer stop events')
class RemoveEvent(HookEvent):
"""Event triggered when a unit is about to be terminated.
This event fires prior to Juju removing the charm and terminating its unit.
"""
def defer(self) -> NoReturn:
"""Remove events are not deferrable like other events.
This is because the unit is about to be torn down, and there
will not be an opportunity for the deferred event to run.
Raises:
RuntimeError: always.
"""
raise RuntimeError('cannot defer remove events')
class ConfigChangedEvent(HookEvent):
"""Event triggered when a configuration change occurs.
This event will fire in several situations:
- When the admin reconfigures the charm using the Juju CLI, for example
``juju config mycharm foo=bar``. This event notifies the charm of
its new configuration. (The event itself, however, is not aware of *what*
specifically has changed in the config).
- Right after the unit starts up for the first time.
This event notifies the charm of its initial configuration.
Typically, this event will fire between an :class:`~ops.InstallEvent`
and a :class:~`ops.StartEvent` during the startup sequence
(when a unit is first deployed), but in general it will fire whenever
the unit is (re)started, for example after pod churn on Kubernetes, on unit
rescheduling, on unit upgrade or refresh, and so on.
- As a specific instance of the above point: when networking changes
(if the machine reboots and comes up with a different IP).
- When the app config changes, for example when `juju trust` is run.
Any callback method bound to this event cannot assume that the
software has already been started; it should not start stopped
software, but should (if appropriate) restart running software to
take configuration changes into account.
"""
class UpdateStatusEvent(HookEvent):
"""Event triggered by a status update request from Juju.
This event is periodically triggered by Juju so that it can
provide constant feedback to the administrator about the status of
the application the charm is modeling. Any callback method bound
to this event should determine the "health" of the application and
set the status appropriately.
The interval between :class:`~ops.UpdateStatusEvent` events can
be configured model-wide, e.g. ``juju model-config
update-status-hook-interval=1m``.
"""
class UpgradeCharmEvent(HookEvent):
"""Event triggered by request to upgrade the charm.
This event will be triggered when an administrator executes ``juju
upgrade-charm``. The event fires after Juju has unpacked the upgraded charm
code, and so this event will be handled by the callback method bound to the
event in the new codebase. The associated callback method is invoked
provided there is no existing error state. The callback method should be
used to reconcile current state written by an older version of the charm
into whatever form that is needed by the current charm version.
"""
class PreSeriesUpgradeEvent(HookEvent):
"""Event triggered to prepare a unit for series upgrade.
This event triggers when an administrator executes ``juju upgrade-machine
<machine> prepare``. The event will fire for each unit that is running on the
specified machine. Any callback method bound to this event must prepare the
charm for an upgrade to the series. This may include things like exporting
database content to a version neutral format, or evacuating running
instances to other machines.
It can be assumed that only after all units on a machine have executed the
callback method associated with this event, the administrator will initiate
steps to actually upgrade the series. After the upgrade has been completed,
the :class:`~ops.PostSeriesUpgradeEvent` will fire.
.. jujuremoved:: 4.0
"""
def __init__(self, handle: 'Handle'):
warnings.warn(
'pre-series-upgrade events will not be emitted from Juju 4.0 onwards',
DeprecationWarning,
stacklevel=3,
)
super().__init__(handle)
class PostSeriesUpgradeEvent(HookEvent):
"""Event triggered after a series upgrade.
This event is triggered after the administrator has done a distribution
upgrade (or rolled back and kept the same series). It is called in response
to ``juju upgrade-machine <machine> complete``. Associated charm callback
methods are expected to do whatever steps are necessary to reconfigure their
applications for the new series. This may include things like populating the
upgraded version of a database. Note however charms are expected to check if
the series has actually changed or whether it was rolled back to the
original series.
.. jujuremoved:: 4.0
"""
def __init__(self, handle: 'Handle'):
warnings.warn(
'post-series-upgrade events will not be emitted from Juju 4.0 onwards',
DeprecationWarning,
stacklevel=3,
)
super().__init__(handle)
class LeaderElectedEvent(HookEvent):
"""Event triggered when a new leader has been elected.
Juju will trigger this event when a new leader unit is chosen for
a given application.
This event fires at least once after Juju selects a leader
unit. Callback methods bound to this event may take any action
required for the elected unit to assert leadership. Note that only
the elected leader unit will receive this event.
"""
class LeaderSettingsChangedEvent(HookEvent):
"""Event triggered when leader changes any settings.
.. deprecated:: 2.4.0
This event has been deprecated in favor of using a Peer relation,
and having the leader set a value in the Application data bag for
that peer relation. (See :class:`~ops.RelationChangedEvent`.)
"""
class CollectMetricsEvent(HookEvent):
"""Event triggered by Juju to collect metrics.
Juju fires this event every five minutes for the lifetime of the
unit. Callback methods bound to this event may use the :meth:`add_metrics`
method of this class to send measurements to Juju.
Note that associated callback methods are currently sandboxed in
how they can interact with Juju.
.. jujuremoved:: 4.0
"""
def __init__(self, handle: 'Handle'):
warnings.warn(
'collect-metrics events will not be emitted from Juju 4.0 onwards - '
'consider using the Canonical Observability Stack',
DeprecationWarning,
stacklevel=3,
)
super().__init__(handle)
def add_metrics(
self, metrics: Mapping[str, Union[int, float]], labels: Optional[Mapping[str, str]] = None
):
"""Record metrics that have been gathered by the charm for this unit.
Args:
metrics: Key-value mapping of metrics that have been gathered.
labels: Key-value labels applied to the metrics.
Raises:
ModelError: if invalid keys or values are provided.
"""
self.framework.model._backend.add_metrics(metrics, labels)
class RelationEvent(HookEvent):
"""A base class representing the various relation lifecycle events.
Relation lifecycle events are generated when application units
participate in relations. Units can only participate in relations
after they have been "started", and before they have been
"stopped". Within that time window, the unit may participate in
several different relations at a time, including multiple
relations with the same name.
"""
relation: 'model.Relation'
"""The relation involved in this event."""
app: model.Application
"""The remote application that has triggered this event."""
unit: Optional[model.Unit]
"""The remote unit that has triggered this event.
This will be ``None`` if the relation event was triggered as an
:class:`Application <model.Application>`-level event.
"""
def __init__(
self,
handle: 'Handle',
relation: 'model.Relation',
app: Optional[model.Application] = None,
unit: Optional[model.Unit] = None,
):
super().__init__(handle)
if unit is not None and unit.app != app:
raise RuntimeError(
f'cannot create RelationEvent with application {app} and unit {unit}'
)
self.relation = relation
if app is None:
logger.warning("'app' expected but not received.")
# Do an explicit assignment here so that we can contain the type: ignore.
self.app = None # type: ignore
else:
self.app = app
self.unit = unit
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
snapshot: Dict[str, Any] = {
'relation_name': self.relation.name,
'relation_id': self.relation.id,
}
if self.app:
snapshot['app_name'] = self.app.name
if self.unit:
snapshot['unit_name'] = self.unit.name
return snapshot
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to deserialize the event from disk.
Not meant to be called by charm code.
"""
relation = self.framework.model.get_relation(
snapshot['relation_name'], snapshot['relation_id']
)
if relation is None:
raise ValueError(
'Unable to restore {}: relation {} (id={}) not found.'.format(
self, snapshot['relation_name'], snapshot['relation_id']
)
)
self.relation = relation
app_name = snapshot.get('app_name')
if app_name:
self.app = self.framework.model.get_app(app_name)
else:
logger.warning("'app_name' expected in snapshot but not found.")
self.app = None # type: ignore
unit_name = snapshot.get('unit_name')
if unit_name:
self.unit = self.framework.model.get_unit(unit_name)
else:
self.unit = None
class RelationCreatedEvent(RelationEvent):
"""Event triggered when a new relation is created.
This is triggered when a new relation with another app is added in Juju. This
can occur before units for those applications have started. All existing
relations will trigger `RelationCreatedEvent` before :class:`~ops.StartEvent` is
emitted.
"""
unit: None # pyright: ignore[reportIncompatibleVariableOverride]
"""Always ``None``."""
class RelationJoinedEvent(RelationEvent):
"""Event triggered when a new unit joins a relation.
This event is triggered whenever a new unit of an integrated
application joins the relation. The event fires only when that
remote unit is first observed by the unit. Callback methods bound
to this event may set any local unit data that can be
determined using no more than the name of the joining unit and the
remote ``private-address`` setting, which is always available when
the relation is created and is by convention not deleted.
"""
unit: model.Unit # pyright: ignore[reportIncompatibleVariableOverride]
"""The remote unit that has triggered this event."""
class RelationChangedEvent(RelationEvent):
"""Event triggered when relation data changes.
This event is triggered whenever there is a change to the data bucket for an
integrated application or unit. Look at ``event.relation.data[event.unit/app]``
to see the new information, where ``event`` is the event object passed to
the callback method bound to this event.
This event always fires once, after :class:`~ops.RelationJoinedEvent`, and
will subsequently fire whenever that remote unit changes its data for
the relation. Callback methods bound to this event should be the only ones
that rely on remote relation data. They should not error if the data
is incomplete, since it can be guaranteed that when the remote unit or
application changes its data, the event will fire again.
The data that may be queried, or set, are determined by the relation's
interface.
"""
class RelationDepartedEvent(RelationEvent):
"""Event triggered when a unit leaves a relation.
This is the inverse of the :class:`~ops.RelationJoinedEvent`, representing when a
unit is leaving the relation (the unit is being removed, the app is being
removed, the relation is being removed). For remaining units, this event is
emitted once for each departing unit. For departing units, this event is
emitted once for each remaining unit.
Callback methods bound to this event may be used to remove all
references to the departing remote unit, because there's no
guarantee that it's still part of the system; it's perfectly
probable (although not guaranteed) that the system running that
unit has already shut down.
Once all callback methods bound to this event have been run for such a
relation, the unit agent will fire the :class:`~ops.RelationBrokenEvent`.
"""
unit: model.Unit # pyright: ignore[reportIncompatibleVariableOverride]
"""The remote unit that has triggered this event."""
def __init__(
self,
handle: 'Handle',
relation: 'model.Relation',
app: Optional[model.Application] = None,
unit: Optional[model.Unit] = None,
departing_unit_name: Optional[str] = None,
):
super().__init__(handle, relation, app=app, unit=unit)
self._departing_unit_name = departing_unit_name
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
snapshot = super().snapshot()
if self._departing_unit_name:
snapshot['departing_unit'] = self._departing_unit_name
return snapshot
@property
def departing_unit(self) -> Optional[model.Unit]:
"""The :class:`ops.Unit` that is departing, if any.
Use this method to determine (for example) whether this unit is the
departing one.
"""
# doing this on init would fail because `framework` gets patched in
# post-init
if not self._departing_unit_name:
return None
return self.framework.model.get_unit(self._departing_unit_name)
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to deserialize the event from disk.
Not meant to be called by charm code.
"""
super().restore(snapshot)
self._departing_unit_name = snapshot.get('departing_unit')
class RelationBrokenEvent(RelationEvent):
"""Event triggered when a relation is removed.
If a relation is being removed (``juju remove-relation`` or ``juju
remove-application``), once all the units have been removed, this event will
fire to signal that the relationship has been fully terminated.
The event indicates that the current relation is no longer valid, and that
the charm's software must be configured as though the relation had never
existed. It will only be called after every callback method bound to
:class:`~ops.RelationDepartedEvent` has been run. If a callback method
bound to this event is being executed, it is guaranteed that no remote units
are currently known locally.
"""
unit: None # pyright: ignore[reportIncompatibleVariableOverride]
"""Always ``None``."""
class StorageEvent(HookEvent):
"""Base class representing events to do with storage.
Juju can provide a variety of storage types to a charms. The
charms can define several different types of storage that are
allocated from Juju. Changes in state of storage trigger sub-types
of :class:`StorageEvent`.
"""
storage: 'model.Storage'
"""Storage instance this event refers to."""
def __init__(self, handle: 'Handle', storage: 'model.Storage'):
super().__init__(handle)
self.storage = storage
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
snapshot: Dict[str, Any] = {}
if isinstance(self.storage, model.Storage):
snapshot['storage_name'] = self.storage.name
snapshot['storage_index'] = self.storage.index
snapshot['storage_location'] = str(self.storage.location)
return snapshot
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to deserialize the event from disk.
Not meant to be called by charm code.
"""
storage_name = snapshot.get('storage_name')
storage_index = snapshot.get('storage_index')
storage_location = snapshot.get('storage_location')
if storage_name and storage_index is not None:
storages = self.framework.model.storages[storage_name]
self.storage = next((s for s in storages if s.index == storage_index), None) # type: ignore
if self.storage is None:
raise RuntimeError(
f'failed loading storage (name={storage_name!r}, '
f'index={storage_index!r}) from snapshot'
)
if storage_location is None:
raise RuntimeError(
'failed loading storage location from snapshot.'
f'(name={storage_name!r}, index={storage_index!r}, storage_location=None)'
)
self.storage.location = storage_location
class StorageAttachedEvent(StorageEvent):
"""Event triggered when new storage becomes available.
This event is triggered when new storage is available for the
charm to use.
Callback methods bound to this event allow the charm to run code
when storage has been added. Such methods will be run before the
:class:`~ops.InstallEvent` fires, so that the installation routine may
use the storage. The name prefix of this hook will depend on the
storage key defined in the ``metadata.yaml`` file.
"""
class StorageDetachingEvent(StorageEvent):
"""Event triggered prior to removal of storage.
This event is triggered when storage a charm has been using is
going away.
Callback methods bound to this event allow the charm to run code
before storage is removed. Such methods will be run before storage
is detached, and always before the :class:`~ops.StopEvent` fires, thereby
allowing the charm to gracefully release resources before they are
removed and before the unit terminates. The name prefix of the
hook will depend on the storage key defined in the ``metadata.yaml``
file.
"""
class WorkloadEvent(HookEvent):
"""Base class representing events to do with the workload.
Workload events are generated for all containers that the charm
expects in metadata.
"""
workload: 'model.Container'
"""The workload involved in this event.
Workload currently only can be a :class:`Container <model.Container>`, but
in future may be other types that represent the specific workload type,
for example a machine.
"""
def __init__(self, handle: 'Handle', workload: 'model.Container'):
super().__init__(handle)
self.workload = workload
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
snapshot: Dict[str, Any] = {}
if isinstance(self.workload, model.Container):
snapshot['container_name'] = self.workload.name
return snapshot
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to deserialize the event from disk.
Not meant to be called by charm code.
"""
container_name = snapshot.get('container_name')
if container_name:
self.workload = self.framework.model.unit.get_container(container_name)
else:
self.workload = None # type: ignore
class PebbleReadyEvent(WorkloadEvent):
"""Event triggered when Pebble is ready for a workload.
This event is triggered when the Pebble process for a workload/container
starts up, allowing the charm to configure how services should be launched.
Callback methods bound to this event allow the charm to run code after
a workload has started its Pebble instance and is ready to receive instructions
regarding what services should be started. The name prefix of the hook
will depend on the container key defined in the ``metadata.yaml`` file.
"""
class PebbleNoticeEvent(WorkloadEvent):
"""Base class for Pebble notice events (each notice type is a subclass)."""
notice: model.LazyNotice
"""Provide access to the event notice's details."""
def __init__(
self,
handle: 'Handle',
workload: 'model.Container',
notice_id: str,
notice_type: str,
notice_key: str,
):
super().__init__(handle, workload)
self.notice = model.LazyNotice(workload, notice_id, notice_type, notice_key)
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
d = super().snapshot()
d['notice_id'] = self.notice.id
d['notice_type'] = (
self.notice.type if isinstance(self.notice.type, str) else self.notice.type.value
)
d['notice_key'] = self.notice.key
return d
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to deserialize the event from disk.
Not meant to be called by charm code.
"""
super().restore(snapshot)
notice_id = snapshot.pop('notice_id')
notice_type = snapshot.pop('notice_type')
notice_key = snapshot.pop('notice_key')
self.notice = model.LazyNotice(self.workload, notice_id, notice_type, notice_key)
class PebbleCustomNoticeEvent(PebbleNoticeEvent):
"""Event triggered when a Pebble notice of type "custom" is created or repeats.
.. jujuadded:: 3.4
"""
class PebbleCheckEvent(WorkloadEvent):
"""Base class for Pebble check events."""
info: model.LazyCheckInfo
"""Provide access to the check's current state."""
def __init__(
self,
handle: Handle,
workload: model.Container,
check_name: str,
):
super().__init__(handle, workload)
self.info = model.LazyCheckInfo(workload, check_name)
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
d = super().snapshot()
d['check_name'] = self.info.name
return d
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to deserialize the event from disk.
Not meant to be called by charm code.
"""
check_name = snapshot.pop('check_name')
super().restore(snapshot)
self.info = model.LazyCheckInfo(self.workload, check_name)
class PebbleCheckFailedEvent(PebbleCheckEvent):
"""Event triggered when a Pebble check exceeds the configured failure threshold.
Note that the check may have started passing by the time this event is
emitted (which will mean that a :class:`~ops.PebbleCheckRecoveredEvent` will be
emitted next). If the handler is executing code that should only be done
if the check is currently failing, check the current status with
``event.info.status == ops.pebble.CheckStatus.DOWN``.
.. jujuadded:: 3.6
"""
class PebbleCheckRecoveredEvent(PebbleCheckEvent):
"""Event triggered when a Pebble check recovers.
This event is only triggered when the check has previously reached a failure
state (not simply failed, but failed at least as many times as the
configured threshold).
.. jujuadded:: 3.6
"""
class SecretEvent(HookEvent):
"""Base class for all secret events."""
def __init__(self, handle: 'Handle', id: str, label: Optional[str]):
super().__init__(handle)
self._id = id
self._label = label
@property
def secret(self) -> model.Secret:
"""The secret instance this event refers to.
Note that the secret content is not retrieved from the secret storage
until :meth:`Secret.get_content()` is called.
"""
backend = self.framework.model._backend
return model.Secret(
backend=backend,
id=self._id,
label=self._label,
_secret_set_cache=self.framework.model._cache._secret_set_cache,
)
def snapshot(self) -> Dict[str, Any]:
"""Used by the framework to serialize the event to disk.
Not meant to be called by charm code.
"""
return {'id': self._id, 'label': self._label}
def restore(self, snapshot: Dict[str, Any]):
"""Used by the framework to deserialize the event from disk.
Not meant to be called by charm code.
"""
self._id = cast(str, snapshot['id'])
self._label = cast(Optional[str], snapshot['label'])
class SecretChangedEvent(SecretEvent):
"""Event triggered on the secret observer charm when the secret owner changes its contents.
When the owner of a secret changes the secret's contents, Juju will create
a new secret revision, and all applications or units that are tracking this
secret will be notified via this event that a new revision is available.
Typically, the charm will fetch the new content by calling
:meth:`event.secret.get_content() <ops.Secret.get_content>` with ``refresh=True``
to tell Juju to start tracking the new revision.
.. jujuadded:: 3.0
Charm secrets added in Juju 3.0, user secrets added in Juju 3.3
"""
class SecretRotateEvent(SecretEvent):
"""Event triggered on the secret owner charm when the secret's rotation policy elapses.
This event is fired on the secret owner to inform it that the secret must
be rotated. The event will keep firing until the owner creates a new
revision by calling :meth:`event.secret.set_content() <ops.Secret.set_content>`.
.. jujuadded:: 3.0
"""
def defer(self) -> NoReturn:
"""Secret rotation events are not deferrable (Juju handles re-invocation).
Raises:
RuntimeError: always.
"""