We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Email server shutdown warning again after extending cluster
Hey, would it be possible to send another hour warning email if a user extends the cluster life?

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `atmo/clusters/models.py`
Content:
```
1 # This Source Code Form is subject to the terms of the Mozilla Public
2 # License, v. 2.0. If a copy of the MPL was not distributed with this
3 # file, you can obtain one at http://mozilla.org/MPL/2.0/.
4 import math
5 from datetime import timedelta
6 
7 from autorepr import autorepr, autostr
8 from django.db import models, transaction
9 from django.utils import timezone
10 
11 from ..models import CreatedByModel, EditedAtModel, URLActionModel
12 from .provisioners import ClusterProvisioner
13 from .queries import ClusterQuerySet, EMRReleaseQuerySet
14 from atmo.stats.models import Metric
15 
16 
17 class EMRRelease(EditedAtModel):
18     version = models.CharField(
19         max_length=50,
20         primary_key=True,
21     )
22     changelog_url = models.TextField(
23         help_text='The URL of the changelog with details about the release.',
24         default='',
25     )
26     help_text = models.TextField(
27         help_text='Optional help text to show for users when creating a cluster.',
28         default='',
29     )
30     is_active = models.BooleanField(
31         help_text='Whether this version should be shown to the user at all.',
32         default=True,
33     )
34     is_experimental = models.BooleanField(
35         help_text='Whether this version should be shown to users as experimental.',
36         default=False,
37     )
38     is_deprecated = models.BooleanField(
39         help_text='Whether this version should be shown to users as deprecated.',
40         default=False,
41     )
42 
43     objects = EMRReleaseQuerySet.as_manager()
44 
45     class Meta:
46         ordering = ['-version']
47         get_latest_by = 'created_at'
48         verbose_name = 'EMR release'
49         verbose_name_plural = 'EMR releases'
50 
51     __str__ = autostr('{self.version}')
52 
53     __repr__ = autorepr(['version', 'is_active', 'is_experimental', 'is_deprecated'])
54 
55 
56 class EMRReleaseModel(models.Model):
57     EMR_RELEASE_HELP = (
58         'Different AWS EMR versions have different versions '
59         'of software like Hadoop, Spark, etc. '
60         'See <a href="'
61         'http://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-whatsnew.html"'
62         '>what\'s new</a> in each.'
63     )
64     emr_release = models.ForeignKey(
65         EMRRelease,
66         verbose_name='EMR release',
67         on_delete=models.PROTECT,
68         related_name='created_%(class)ss',  # e.g. emr_release.created_clusters.all()
69         help_text=EMR_RELEASE_HELP,
70     )
71 
72     class Meta:
73         abstract = True
74 
75 
76 class Cluster(EMRReleaseModel, CreatedByModel, EditedAtModel, URLActionModel):
77     STATUS_STARTING = 'STARTING'
78     STATUS_BOOTSTRAPPING = 'BOOTSTRAPPING'
79     STATUS_RUNNING = 'RUNNING'
80     STATUS_WAITING = 'WAITING'
81     STATUS_TERMINATING = 'TERMINATING'
82     STATUS_TERMINATED = 'TERMINATED'
83     STATUS_TERMINATED_WITH_ERRORS = 'TERMINATED_WITH_ERRORS'
84 
85     ACTIVE_STATUS_LIST = (
86         STATUS_STARTING,
87         STATUS_BOOTSTRAPPING,
88         STATUS_RUNNING,
89         STATUS_WAITING,
90         STATUS_TERMINATING,
91     )
92     READY_STATUS_LIST = [
93         STATUS_RUNNING,
94         STATUS_WAITING,
95     ]
96     TERMINATED_STATUS_LIST = (
97         STATUS_TERMINATED,
98     )
99     FAILED_STATUS_LIST = (
100         STATUS_TERMINATED_WITH_ERRORS,
101     )
102     FINAL_STATUS_LIST = TERMINATED_STATUS_LIST + FAILED_STATUS_LIST
103 
104     STATE_CHANGE_REASON_INTERNAL_ERROR = 'INTERNAL_ERROR'
105     STATE_CHANGE_REASON_VALIDATION_ERROR = 'VALIDATION_ERROR'
106     STATE_CHANGE_REASON_INSTANCE_FAILURE = 'INSTANCE_FAILURE'
107     STATE_CHANGE_REASON_BOOTSTRAP_FAILURE = 'BOOTSTRAP_FAILURE'
108     STATE_CHANGE_REASON_USER_REQUEST = 'USER_REQUEST'
109     STATE_CHANGE_REASON_STEP_FAILURE = 'STEP_FAILURE'
110     STATE_CHANGE_REASON_ALL_STEPS_COMPLETED = 'ALL_STEPS_COMPLETED'
111     FAILED_STATE_CHANGE_REASON_LIST = [
112         STATE_CHANGE_REASON_INTERNAL_ERROR,
113         STATE_CHANGE_REASON_VALIDATION_ERROR,
114         STATE_CHANGE_REASON_INSTANCE_FAILURE,
115         STATE_CHANGE_REASON_BOOTSTRAP_FAILURE,
116         STATE_CHANGE_REASON_STEP_FAILURE,
117     ]
118     REQUESTED_STATE_CHANGE_REASON_LIST = [
119         STATE_CHANGE_REASON_USER_REQUEST,
120     ]
121     COMPLETED_STATE_CHANGE_REASON_LIST = [
122         STATE_CHANGE_REASON_ALL_STEPS_COMPLETED,
123     ]
124     DEFAULT_SIZE = 1
125     DEFAULT_LIFETIME = 8
126 
127     identifier = models.CharField(
128         max_length=100,
129         help_text="Cluster name, used to non-uniqely identify individual clusters."
130     )
131     size = models.IntegerField(
132         help_text="Number of computers used in the cluster."
133     )
134     lifetime = models.PositiveSmallIntegerField(
135         help_text="Lifetime of the cluster after which it's automatically terminated, in hours.",
136         default=DEFAULT_LIFETIME,
137     )
138     lifetime_extension_count = models.PositiveSmallIntegerField(
139         help_text="Number of lifetime extensions.",
140         default=0,
141     )
142     ssh_key = models.ForeignKey(
143         'keys.SSHKey',
144         on_delete=models.SET_NULL,
145         blank=True,
146         null=True,
147         related_name='launched_clusters',  # e.g. ssh_key.launched_clusters.all()
148         help_text="SSH key to use when launching the cluster.",
149     )
150     expires_at = models.DateTimeField(
151         blank=True,
152         null=True,
153         help_text="Date/time that the cluster will expire and automatically be deleted.",
154     )
155     started_at = models.DateTimeField(
156         blank=True,
157         null=True,
158         help_text="Date/time when the cluster was started on AWS EMR.",
159     )
160     ready_at = models.DateTimeField(
161         blank=True,
162         null=True,
163         help_text="Date/time when the cluster was ready to run steps on AWS EMR.",
164     )
165     finished_at = models.DateTimeField(
166         blank=True,
167         null=True,
168         help_text="Date/time when the cluster was terminated or failed on AWS EMR.",
169     )
170     jobflow_id = models.CharField(
171         max_length=50,
172         blank=True,
173         null=True,
174         help_text="AWS cluster/jobflow ID for the cluster, used for cluster management.",
175     )
176     most_recent_status = models.CharField(
177         max_length=50,
178         default='',
179         blank=True,
180         help_text="Most recently retrieved AWS status for the cluster.",
181         db_index=True,
182     )
183     master_address = models.CharField(
184         max_length=255,
185         default='',
186         blank=True,
187         help_text=("Public address of the master node."
188                    "This is only available once the cluster has bootstrapped"),
189     )
190     expiration_mail_sent = models.BooleanField(
191         default=False,
192         help_text="Whether the expiration mail were sent.",
193     )
194 
195     objects = ClusterQuerySet.as_manager()
196 
197     class Meta:
198         permissions = [
199             ('view_cluster', 'Can view cluster'),
200             ('maintain_cluster', 'Can maintain cluster'),
201         ]
202 
203     __str__ = autostr('{self.identifier}')
204 
205     __repr__ = autorepr([
206         'identifier',
207         'most_recent_status',
208         'size',
209         'lifetime',
210         'expires_at',
211         'lifetime_extension_count',
212     ])
213 
214     url_prefix = 'clusters'
215     url_actions = ['detail', 'extend', 'terminate']
216 
217     def get_absolute_url(self):
218         return self.urls.detail
219 
220     @property
221     def is_active(self):
222         """Returns whether the cluster is active or not."""
223         return self.most_recent_status in self.ACTIVE_STATUS_LIST
224 
225     @property
226     def is_terminated(self):
227         """Returns whether the cluster is terminated or not."""
228         return self.most_recent_status in self.TERMINATED_STATUS_LIST
229 
230     @property
231     def is_failed(self):
232         """Returns whether the cluster has failed or not."""
233         return self.most_recent_status in self.FAILED_STATUS_LIST
234 
235     @property
236     def is_terminating(self):
237         """Returns whether the cluster is terminating or not."""
238         return self.most_recent_status == self.STATUS_TERMINATING
239 
240     @property
241     def is_ready(self):
242         """Returns whether the cluster is ready or not."""
243         return self.most_recent_status == self.STATUS_WAITING
244 
245     @property
246     def is_expiring_soon(self):
247         """Returns whether the cluster is expiring in the next hour."""
248         return self.expires_at <= timezone.now() + timedelta(hours=1)
249 
250     @property
251     def provisioner(self):
252         return ClusterProvisioner()
253 
254     @property
255     def info(self):
256         """Returns the provisioning information for the cluster."""
257         return self.provisioner.info(self.jobflow_id)
258 
259     def sync(self, info=None):
260         """Should be called to update latest cluster status in `self.most_recent_status`."""
261         if info is None:
262             info = self.info
263 
264         # Map AWS API fields to Cluster model fields.
265         model_field_map = (
266             ('state', 'most_recent_status'),
267             ('public_dns', 'master_address'),
268             ('creation_datetime', 'started_at'),
269             ('ready_datetime', 'ready_at'),
270             ('end_datetime', 'finished_at'),
271         )
272         save_needed = False
273         date_fields_updated = False
274 
275         # set the various model fields to the value the API returned
276         for api_field, model_field in model_field_map:
277             field_value = info.get(api_field)
278             # Only update the field if the value for a field is not set or it
279             # hasn't changed.
280             if field_value is None or field_value == getattr(self, model_field):
281                 continue
282             setattr(self, model_field, field_value)
283             save_needed = True
284 
285             if model_field in ('started_at', 'ready_at', 'finished_at'):
286                 date_fields_updated = True
287 
288         if save_needed:
289             with transaction.atomic():
290                 self.save()
291 
292         with transaction.atomic():
293             if date_fields_updated:
294 
295                 if self.finished_at:
296                     # When cluster is finished, record normalized instance
297                     # hours.
298                     hours = math.ceil(
299                         (self.finished_at - self.started_at).seconds / 60 / 60
300                     )
301                     normalized_hours = hours * self.size
302                     Metric.record(
303                         'cluster-normalized-instance-hours', normalized_hours,
304                         data={
305                             'identifier': self.identifier,
306                             'size': self.size,
307                             'jobflow_id': self.jobflow_id,
308                         }
309                     )
310 
311                 # When cluster is ready, record a count and time to ready.
312                 if self.ready_at and not self.finished_at:
313                     # A simple count to track number of clusters spun up
314                     # successfully.
315                     Metric.record('cluster-ready', data={
316                         'identifier': self.identifier,
317                         'size': self.size,
318                         'jobflow_id': self.jobflow_id,
319                     })
320                     # Time in seconds it took the cluster to be ready.
321                     time_to_ready = (self.ready_at - self.started_at).seconds
322                     Metric.record(
323                         'cluster-time-to-ready', time_to_ready,
324                         data={
325                             'identifier': self.identifier,
326                             'size': self.size,
327                             'jobflow_id': self.jobflow_id,
328                         }
329                     )
330 
331     def save(self, *args, **kwargs):
332         """Insert the cluster into the database or update it if already
333         present, spawning the cluster if it's not already spawned.
334         """
335         # actually start the cluster
336         if self.jobflow_id is None:
337             self.jobflow_id = self.provisioner.start(
338                 user_username=self.created_by.username,
339                 user_email=self.created_by.email,
340                 identifier=self.identifier,
341                 emr_release=self.emr_release.version,
342                 size=self.size,
343                 public_key=self.ssh_key.key,
344             )
345             # once we've stored the jobflow id we can fetch the status for the first time
346             transaction.on_commit(self.sync)
347 
348             with transaction.atomic():
349                 Metric.record('cluster-emr-version',
350                               data={'version': self.emr_release.version})
351 
352         # set the dates
353         if not self.expires_at:
354             # clusters should expire after the lifetime it's set to
355             self.expires_at = timezone.now() + timedelta(hours=self.lifetime)
356 
357         super().save(*args, **kwargs)
358 
359     def extend(self, hours):
360         """Extend the cluster lifetime by the given number of hours."""
361         self.expires_at = models.F('expires_at') + timedelta(hours=hours)
362         self.lifetime_extension_count = models.F('lifetime_extension_count') + 1
363         self.save()
364 
365         with transaction.atomic():
366             Metric.record('cluster-extension', data={
367                 'identifier': self.identifier,
368                 'size': self.size,
369                 'jobflow_id': self.jobflow_id,
370             })
371 
372     def deactivate(self):
373         """Shutdown the cluster and update its status accordingly"""
374         self.provisioner.stop(self.jobflow_id)
375         self.sync()
376
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/atmo/clusters/models.py b/atmo/clusters/models.py
--- a/atmo/clusters/models.py
+++ b/atmo/clusters/models.py
@@ -360,6 +360,7 @@
         """Extend the cluster lifetime by the given number of hours."""
         self.expires_at = models.F('expires_at') + timedelta(hours=hours)
         self.lifetime_extension_count = models.F('lifetime_extension_count') + 1
+        self.expiration_mail_sent = False
         self.save()
 
         with transaction.atomic():

{"golden_diff": "diff --git a/atmo/clusters/models.py b/atmo/clusters/models.py\n--- a/atmo/clusters/models.py\n+++ b/atmo/clusters/models.py\n@@ -360,6 +360,7 @@\n         \"\"\"Extend the cluster lifetime by the given number of hours.\"\"\"\n         self.expires_at = models.F('expires_at') + timedelta(hours=hours)\n         self.lifetime_extension_count = models.F('lifetime_extension_count') + 1\n+        self.expiration_mail_sent = False\n         self.save()\n \n         with transaction.atomic():\n", "issue": "Email server shutdown warning again after extending cluster\nHey, would it be possible to send another hour warning email if a user extends the cluster life?\n", "before_files": [{"content": "# This Source Code Form is subject to the terms of the Mozilla Public\n# License, v. 2.0. If a copy of the MPL was not distributed with this\n# file, you can obtain one at http://mozilla.org/MPL/2.0/.\nimport math\nfrom datetime import timedelta\n\nfrom autorepr import autorepr, autostr\nfrom django.db import models, transaction\nfrom django.utils import timezone\n\nfrom ..models import CreatedByModel, EditedAtModel, URLActionModel\nfrom .provisioners import ClusterProvisioner\nfrom .queries import ClusterQuerySet, EMRReleaseQuerySet\nfrom atmo.stats.models import Metric\n\n\nclass EMRRelease(EditedAtModel):\n    version = models.CharField(\n        max_length=50,\n        primary_key=True,\n    )\n    changelog_url = models.TextField(\n        help_text='The URL of the changelog with details about the release.',\n        default='',\n    )\n    help_text = models.TextField(\n        help_text='Optional help text to show for users when creating a cluster.',\n        default='',\n    )\n    is_active = models.BooleanField(\n        help_text='Whether this version should be shown to the user at all.',\n        default=True,\n    )\n    is_experimental = models.BooleanField(\n        help_text='Whether this version should be shown to users as experimental.',\n        default=False,\n    )\n    is_deprecated = models.BooleanField(\n        help_text='Whether this version should be shown to users as deprecated.',\n        default=False,\n    )\n\n    objects = EMRReleaseQuerySet.as_manager()\n\n    class Meta:\n        ordering = ['-version']\n        get_latest_by = 'created_at'\n        verbose_name = 'EMR release'\n        verbose_name_plural = 'EMR releases'\n\n    __str__ = autostr('{self.version}')\n\n    __repr__ = autorepr(['version', 'is_active', 'is_experimental', 'is_deprecated'])\n\n\nclass EMRReleaseModel(models.Model):\n    EMR_RELEASE_HELP = (\n        'Different AWS EMR versions have different versions '\n        'of software like Hadoop, Spark, etc. '\n        'See <a href=\"'\n        'http://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-whatsnew.html\"'\n        '>what\\'s new</a> in each.'\n    )\n    emr_release = models.ForeignKey(\n        EMRRelease,\n        verbose_name='EMR release',\n        on_delete=models.PROTECT,\n        related_name='created_%(class)ss',  # e.g. emr_release.created_clusters.all()\n        help_text=EMR_RELEASE_HELP,\n    )\n\n    class Meta:\n        abstract = True\n\n\nclass Cluster(EMRReleaseModel, CreatedByModel, EditedAtModel, URLActionModel):\n    STATUS_STARTING = 'STARTING'\n    STATUS_BOOTSTRAPPING = 'BOOTSTRAPPING'\n    STATUS_RUNNING = 'RUNNING'\n    STATUS_WAITING = 'WAITING'\n    STATUS_TERMINATING = 'TERMINATING'\n    STATUS_TERMINATED = 'TERMINATED'\n    STATUS_TERMINATED_WITH_ERRORS = 'TERMINATED_WITH_ERRORS'\n\n    ACTIVE_STATUS_LIST = (\n        STATUS_STARTING,\n        STATUS_BOOTSTRAPPING,\n        STATUS_RUNNING,\n        STATUS_WAITING,\n        STATUS_TERMINATING,\n    )\n    READY_STATUS_LIST = [\n        STATUS_RUNNING,\n        STATUS_WAITING,\n    ]\n    TERMINATED_STATUS_LIST = (\n        STATUS_TERMINATED,\n    )\n    FAILED_STATUS_LIST = (\n        STATUS_TERMINATED_WITH_ERRORS,\n    )\n    FINAL_STATUS_LIST = TERMINATED_STATUS_LIST + FAILED_STATUS_LIST\n\n    STATE_CHANGE_REASON_INTERNAL_ERROR = 'INTERNAL_ERROR'\n    STATE_CHANGE_REASON_VALIDATION_ERROR = 'VALIDATION_ERROR'\n    STATE_CHANGE_REASON_INSTANCE_FAILURE = 'INSTANCE_FAILURE'\n    STATE_CHANGE_REASON_BOOTSTRAP_FAILURE = 'BOOTSTRAP_FAILURE'\n    STATE_CHANGE_REASON_USER_REQUEST = 'USER_REQUEST'\n    STATE_CHANGE_REASON_STEP_FAILURE = 'STEP_FAILURE'\n    STATE_CHANGE_REASON_ALL_STEPS_COMPLETED = 'ALL_STEPS_COMPLETED'\n    FAILED_STATE_CHANGE_REASON_LIST = [\n        STATE_CHANGE_REASON_INTERNAL_ERROR,\n        STATE_CHANGE_REASON_VALIDATION_ERROR,\n        STATE_CHANGE_REASON_INSTANCE_FAILURE,\n        STATE_CHANGE_REASON_BOOTSTRAP_FAILURE,\n        STATE_CHANGE_REASON_STEP_FAILURE,\n    ]\n    REQUESTED_STATE_CHANGE_REASON_LIST = [\n        STATE_CHANGE_REASON_USER_REQUEST,\n    ]\n    COMPLETED_STATE_CHANGE_REASON_LIST = [\n        STATE_CHANGE_REASON_ALL_STEPS_COMPLETED,\n    ]\n    DEFAULT_SIZE = 1\n    DEFAULT_LIFETIME = 8\n\n    identifier = models.CharField(\n        max_length=100,\n        help_text=\"Cluster name, used to non-uniqely identify individual clusters.\"\n    )\n    size = models.IntegerField(\n        help_text=\"Number of computers used in the cluster.\"\n    )\n    lifetime = models.PositiveSmallIntegerField(\n        help_text=\"Lifetime of the cluster after which it's automatically terminated, in hours.\",\n        default=DEFAULT_LIFETIME,\n    )\n    lifetime_extension_count = models.PositiveSmallIntegerField(\n        help_text=\"Number of lifetime extensions.\",\n        default=0,\n    )\n    ssh_key = models.ForeignKey(\n        'keys.SSHKey',\n        on_delete=models.SET_NULL,\n        blank=True,\n        null=True,\n        related_name='launched_clusters',  # e.g. ssh_key.launched_clusters.all()\n        help_text=\"SSH key to use when launching the cluster.\",\n    )\n    expires_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time that the cluster will expire and automatically be deleted.\",\n    )\n    started_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time when the cluster was started on AWS EMR.\",\n    )\n    ready_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time when the cluster was ready to run steps on AWS EMR.\",\n    )\n    finished_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time when the cluster was terminated or failed on AWS EMR.\",\n    )\n    jobflow_id = models.CharField(\n        max_length=50,\n        blank=True,\n        null=True,\n        help_text=\"AWS cluster/jobflow ID for the cluster, used for cluster management.\",\n    )\n    most_recent_status = models.CharField(\n        max_length=50,\n        default='',\n        blank=True,\n        help_text=\"Most recently retrieved AWS status for the cluster.\",\n        db_index=True,\n    )\n    master_address = models.CharField(\n        max_length=255,\n        default='',\n        blank=True,\n        help_text=(\"Public address of the master node.\"\n                   \"This is only available once the cluster has bootstrapped\"),\n    )\n    expiration_mail_sent = models.BooleanField(\n        default=False,\n        help_text=\"Whether the expiration mail were sent.\",\n    )\n\n    objects = ClusterQuerySet.as_manager()\n\n    class Meta:\n        permissions = [\n            ('view_cluster', 'Can view cluster'),\n            ('maintain_cluster', 'Can maintain cluster'),\n        ]\n\n    __str__ = autostr('{self.identifier}')\n\n    __repr__ = autorepr([\n        'identifier',\n        'most_recent_status',\n        'size',\n        'lifetime',\n        'expires_at',\n        'lifetime_extension_count',\n    ])\n\n    url_prefix = 'clusters'\n    url_actions = ['detail', 'extend', 'terminate']\n\n    def get_absolute_url(self):\n        return self.urls.detail\n\n    @property\n    def is_active(self):\n        \"\"\"Returns whether the cluster is active or not.\"\"\"\n        return self.most_recent_status in self.ACTIVE_STATUS_LIST\n\n    @property\n    def is_terminated(self):\n        \"\"\"Returns whether the cluster is terminated or not.\"\"\"\n        return self.most_recent_status in self.TERMINATED_STATUS_LIST\n\n    @property\n    def is_failed(self):\n        \"\"\"Returns whether the cluster has failed or not.\"\"\"\n        return self.most_recent_status in self.FAILED_STATUS_LIST\n\n    @property\n    def is_terminating(self):\n        \"\"\"Returns whether the cluster is terminating or not.\"\"\"\n        return self.most_recent_status == self.STATUS_TERMINATING\n\n    @property\n    def is_ready(self):\n        \"\"\"Returns whether the cluster is ready or not.\"\"\"\n        return self.most_recent_status == self.STATUS_WAITING\n\n    @property\n    def is_expiring_soon(self):\n        \"\"\"Returns whether the cluster is expiring in the next hour.\"\"\"\n        return self.expires_at <= timezone.now() + timedelta(hours=1)\n\n    @property\n    def provisioner(self):\n        return ClusterProvisioner()\n\n    @property\n    def info(self):\n        \"\"\"Returns the provisioning information for the cluster.\"\"\"\n        return self.provisioner.info(self.jobflow_id)\n\n    def sync(self, info=None):\n        \"\"\"Should be called to update latest cluster status in `self.most_recent_status`.\"\"\"\n        if info is None:\n            info = self.info\n\n        # Map AWS API fields to Cluster model fields.\n        model_field_map = (\n            ('state', 'most_recent_status'),\n            ('public_dns', 'master_address'),\n            ('creation_datetime', 'started_at'),\n            ('ready_datetime', 'ready_at'),\n            ('end_datetime', 'finished_at'),\n        )\n        save_needed = False\n        date_fields_updated = False\n\n        # set the various model fields to the value the API returned\n        for api_field, model_field in model_field_map:\n            field_value = info.get(api_field)\n            # Only update the field if the value for a field is not set or it\n            # hasn't changed.\n            if field_value is None or field_value == getattr(self, model_field):\n                continue\n            setattr(self, model_field, field_value)\n            save_needed = True\n\n            if model_field in ('started_at', 'ready_at', 'finished_at'):\n                date_fields_updated = True\n\n        if save_needed:\n            with transaction.atomic():\n                self.save()\n\n        with transaction.atomic():\n            if date_fields_updated:\n\n                if self.finished_at:\n                    # When cluster is finished, record normalized instance\n                    # hours.\n                    hours = math.ceil(\n                        (self.finished_at - self.started_at).seconds / 60 / 60\n                    )\n                    normalized_hours = hours * self.size\n                    Metric.record(\n                        'cluster-normalized-instance-hours', normalized_hours,\n                        data={\n                            'identifier': self.identifier,\n                            'size': self.size,\n                            'jobflow_id': self.jobflow_id,\n                        }\n                    )\n\n                # When cluster is ready, record a count and time to ready.\n                if self.ready_at and not self.finished_at:\n                    # A simple count to track number of clusters spun up\n                    # successfully.\n                    Metric.record('cluster-ready', data={\n                        'identifier': self.identifier,\n                        'size': self.size,\n                        'jobflow_id': self.jobflow_id,\n                    })\n                    # Time in seconds it took the cluster to be ready.\n                    time_to_ready = (self.ready_at - self.started_at).seconds\n                    Metric.record(\n                        'cluster-time-to-ready', time_to_ready,\n                        data={\n                            'identifier': self.identifier,\n                            'size': self.size,\n                            'jobflow_id': self.jobflow_id,\n                        }\n                    )\n\n    def save(self, *args, **kwargs):\n        \"\"\"Insert the cluster into the database or update it if already\n        present, spawning the cluster if it's not already spawned.\n        \"\"\"\n        # actually start the cluster\n        if self.jobflow_id is None:\n            self.jobflow_id = self.provisioner.start(\n                user_username=self.created_by.username,\n                user_email=self.created_by.email,\n                identifier=self.identifier,\n                emr_release=self.emr_release.version,\n                size=self.size,\n                public_key=self.ssh_key.key,\n            )\n            # once we've stored the jobflow id we can fetch the status for the first time\n            transaction.on_commit(self.sync)\n\n            with transaction.atomic():\n                Metric.record('cluster-emr-version',\n                              data={'version': self.emr_release.version})\n\n        # set the dates\n        if not self.expires_at:\n            # clusters should expire after the lifetime it's set to\n            self.expires_at = timezone.now() + timedelta(hours=self.lifetime)\n\n        super().save(*args, **kwargs)\n\n    def extend(self, hours):\n        \"\"\"Extend the cluster lifetime by the given number of hours.\"\"\"\n        self.expires_at = models.F('expires_at') + timedelta(hours=hours)\n        self.lifetime_extension_count = models.F('lifetime_extension_count') + 1\n        self.save()\n\n        with transaction.atomic():\n            Metric.record('cluster-extension', data={\n                'identifier': self.identifier,\n                'size': self.size,\n                'jobflow_id': self.jobflow_id,\n            })\n\n    def deactivate(self):\n        \"\"\"Shutdown the cluster and update its status accordingly\"\"\"\n        self.provisioner.stop(self.jobflow_id)\n        self.sync()\n", "path": "atmo/clusters/models.py"}], "after_files": [{"content": "# This Source Code Form is subject to the terms of the Mozilla Public\n# License, v. 2.0. If a copy of the MPL was not distributed with this\n# file, you can obtain one at http://mozilla.org/MPL/2.0/.\nimport math\nfrom datetime import timedelta\n\nfrom autorepr import autorepr, autostr\nfrom django.db import models, transaction\nfrom django.utils import timezone\n\nfrom ..models import CreatedByModel, EditedAtModel, URLActionModel\nfrom .provisioners import ClusterProvisioner\nfrom .queries import ClusterQuerySet, EMRReleaseQuerySet\nfrom atmo.stats.models import Metric\n\n\nclass EMRRelease(EditedAtModel):\n    version = models.CharField(\n        max_length=50,\n        primary_key=True,\n    )\n    changelog_url = models.TextField(\n        help_text='The URL of the changelog with details about the release.',\n        default='',\n    )\n    help_text = models.TextField(\n        help_text='Optional help text to show for users when creating a cluster.',\n        default='',\n    )\n    is_active = models.BooleanField(\n        help_text='Whether this version should be shown to the user at all.',\n        default=True,\n    )\n    is_experimental = models.BooleanField(\n        help_text='Whether this version should be shown to users as experimental.',\n        default=False,\n    )\n    is_deprecated = models.BooleanField(\n        help_text='Whether this version should be shown to users as deprecated.',\n        default=False,\n    )\n\n    objects = EMRReleaseQuerySet.as_manager()\n\n    class Meta:\n        ordering = ['-version']\n        get_latest_by = 'created_at'\n        verbose_name = 'EMR release'\n        verbose_name_plural = 'EMR releases'\n\n    __str__ = autostr('{self.version}')\n\n    __repr__ = autorepr(['version', 'is_active', 'is_experimental', 'is_deprecated'])\n\n\nclass EMRReleaseModel(models.Model):\n    EMR_RELEASE_HELP = (\n        'Different AWS EMR versions have different versions '\n        'of software like Hadoop, Spark, etc. '\n        'See <a href=\"'\n        'http://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-whatsnew.html\"'\n        '>what\\'s new</a> in each.'\n    )\n    emr_release = models.ForeignKey(\n        EMRRelease,\n        verbose_name='EMR release',\n        on_delete=models.PROTECT,\n        related_name='created_%(class)ss',  # e.g. emr_release.created_clusters.all()\n        help_text=EMR_RELEASE_HELP,\n    )\n\n    class Meta:\n        abstract = True\n\n\nclass Cluster(EMRReleaseModel, CreatedByModel, EditedAtModel, URLActionModel):\n    STATUS_STARTING = 'STARTING'\n    STATUS_BOOTSTRAPPING = 'BOOTSTRAPPING'\n    STATUS_RUNNING = 'RUNNING'\n    STATUS_WAITING = 'WAITING'\n    STATUS_TERMINATING = 'TERMINATING'\n    STATUS_TERMINATED = 'TERMINATED'\n    STATUS_TERMINATED_WITH_ERRORS = 'TERMINATED_WITH_ERRORS'\n\n    ACTIVE_STATUS_LIST = (\n        STATUS_STARTING,\n        STATUS_BOOTSTRAPPING,\n        STATUS_RUNNING,\n        STATUS_WAITING,\n        STATUS_TERMINATING,\n    )\n    READY_STATUS_LIST = [\n        STATUS_RUNNING,\n        STATUS_WAITING,\n    ]\n    TERMINATED_STATUS_LIST = (\n        STATUS_TERMINATED,\n    )\n    FAILED_STATUS_LIST = (\n        STATUS_TERMINATED_WITH_ERRORS,\n    )\n    FINAL_STATUS_LIST = TERMINATED_STATUS_LIST + FAILED_STATUS_LIST\n\n    STATE_CHANGE_REASON_INTERNAL_ERROR = 'INTERNAL_ERROR'\n    STATE_CHANGE_REASON_VALIDATION_ERROR = 'VALIDATION_ERROR'\n    STATE_CHANGE_REASON_INSTANCE_FAILURE = 'INSTANCE_FAILURE'\n    STATE_CHANGE_REASON_BOOTSTRAP_FAILURE = 'BOOTSTRAP_FAILURE'\n    STATE_CHANGE_REASON_USER_REQUEST = 'USER_REQUEST'\n    STATE_CHANGE_REASON_STEP_FAILURE = 'STEP_FAILURE'\n    STATE_CHANGE_REASON_ALL_STEPS_COMPLETED = 'ALL_STEPS_COMPLETED'\n    FAILED_STATE_CHANGE_REASON_LIST = [\n        STATE_CHANGE_REASON_INTERNAL_ERROR,\n        STATE_CHANGE_REASON_VALIDATION_ERROR,\n        STATE_CHANGE_REASON_INSTANCE_FAILURE,\n        STATE_CHANGE_REASON_BOOTSTRAP_FAILURE,\n        STATE_CHANGE_REASON_STEP_FAILURE,\n    ]\n    REQUESTED_STATE_CHANGE_REASON_LIST = [\n        STATE_CHANGE_REASON_USER_REQUEST,\n    ]\n    COMPLETED_STATE_CHANGE_REASON_LIST = [\n        STATE_CHANGE_REASON_ALL_STEPS_COMPLETED,\n    ]\n    DEFAULT_SIZE = 1\n    DEFAULT_LIFETIME = 8\n\n    identifier = models.CharField(\n        max_length=100,\n        help_text=\"Cluster name, used to non-uniqely identify individual clusters.\"\n    )\n    size = models.IntegerField(\n        help_text=\"Number of computers used in the cluster.\"\n    )\n    lifetime = models.PositiveSmallIntegerField(\n        help_text=\"Lifetime of the cluster after which it's automatically terminated, in hours.\",\n        default=DEFAULT_LIFETIME,\n    )\n    lifetime_extension_count = models.PositiveSmallIntegerField(\n        help_text=\"Number of lifetime extensions.\",\n        default=0,\n    )\n    ssh_key = models.ForeignKey(\n        'keys.SSHKey',\n        on_delete=models.SET_NULL,\n        blank=True,\n        null=True,\n        related_name='launched_clusters',  # e.g. ssh_key.launched_clusters.all()\n        help_text=\"SSH key to use when launching the cluster.\",\n    )\n    expires_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time that the cluster will expire and automatically be deleted.\",\n    )\n    started_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time when the cluster was started on AWS EMR.\",\n    )\n    ready_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time when the cluster was ready to run steps on AWS EMR.\",\n    )\n    finished_at = models.DateTimeField(\n        blank=True,\n        null=True,\n        help_text=\"Date/time when the cluster was terminated or failed on AWS EMR.\",\n    )\n    jobflow_id = models.CharField(\n        max_length=50,\n        blank=True,\n        null=True,\n        help_text=\"AWS cluster/jobflow ID for the cluster, used for cluster management.\",\n    )\n    most_recent_status = models.CharField(\n        max_length=50,\n        default='',\n        blank=True,\n        help_text=\"Most recently retrieved AWS status for the cluster.\",\n        db_index=True,\n    )\n    master_address = models.CharField(\n        max_length=255,\n        default='',\n        blank=True,\n        help_text=(\"Public address of the master node.\"\n                   \"This is only available once the cluster has bootstrapped\"),\n    )\n    expiration_mail_sent = models.BooleanField(\n        default=False,\n        help_text=\"Whether the expiration mail were sent.\",\n    )\n\n    objects = ClusterQuerySet.as_manager()\n\n    class Meta:\n        permissions = [\n            ('view_cluster', 'Can view cluster'),\n            ('maintain_cluster', 'Can maintain cluster'),\n        ]\n\n    __str__ = autostr('{self.identifier}')\n\n    __repr__ = autorepr([\n        'identifier',\n        'most_recent_status',\n        'size',\n        'lifetime',\n        'expires_at',\n        'lifetime_extension_count',\n    ])\n\n    url_prefix = 'clusters'\n    url_actions = ['detail', 'extend', 'terminate']\n\n    def get_absolute_url(self):\n        return self.urls.detail\n\n    @property\n    def is_active(self):\n        \"\"\"Returns whether the cluster is active or not.\"\"\"\n        return self.most_recent_status in self.ACTIVE_STATUS_LIST\n\n    @property\n    def is_terminated(self):\n        \"\"\"Returns whether the cluster is terminated or not.\"\"\"\n        return self.most_recent_status in self.TERMINATED_STATUS_LIST\n\n    @property\n    def is_failed(self):\n        \"\"\"Returns whether the cluster has failed or not.\"\"\"\n        return self.most_recent_status in self.FAILED_STATUS_LIST\n\n    @property\n    def is_terminating(self):\n        \"\"\"Returns whether the cluster is terminating or not.\"\"\"\n        return self.most_recent_status == self.STATUS_TERMINATING\n\n    @property\n    def is_ready(self):\n        \"\"\"Returns whether the cluster is ready or not.\"\"\"\n        return self.most_recent_status == self.STATUS_WAITING\n\n    @property\n    def is_expiring_soon(self):\n        \"\"\"Returns whether the cluster is expiring in the next hour.\"\"\"\n        return self.expires_at <= timezone.now() + timedelta(hours=1)\n\n    @property\n    def provisioner(self):\n        return ClusterProvisioner()\n\n    @property\n    def info(self):\n        \"\"\"Returns the provisioning information for the cluster.\"\"\"\n        return self.provisioner.info(self.jobflow_id)\n\n    def sync(self, info=None):\n        \"\"\"Should be called to update latest cluster status in `self.most_recent_status`.\"\"\"\n        if info is None:\n            info = self.info\n\n        # Map AWS API fields to Cluster model fields.\n        model_field_map = (\n            ('state', 'most_recent_status'),\n            ('public_dns', 'master_address'),\n            ('creation_datetime', 'started_at'),\n            ('ready_datetime', 'ready_at'),\n            ('end_datetime', 'finished_at'),\n        )\n        save_needed = False\n        date_fields_updated = False\n\n        # set the various model fields to the value the API returned\n        for api_field, model_field in model_field_map:\n            field_value = info.get(api_field)\n            # Only update the field if the value for a field is not set or it\n            # hasn't changed.\n            if field_value is None or field_value == getattr(self, model_field):\n                continue\n            setattr(self, model_field, field_value)\n            save_needed = True\n\n            if model_field in ('started_at', 'ready_at', 'finished_at'):\n                date_fields_updated = True\n\n        if save_needed:\n            with transaction.atomic():\n                self.save()\n\n        with transaction.atomic():\n            if date_fields_updated:\n\n                if self.finished_at:\n                    # When cluster is finished, record normalized instance\n                    # hours.\n                    hours = math.ceil(\n                        (self.finished_at - self.started_at).seconds / 60 / 60\n                    )\n                    normalized_hours = hours * self.size\n                    Metric.record(\n                        'cluster-normalized-instance-hours', normalized_hours,\n                        data={\n                            'identifier': self.identifier,\n                            'size': self.size,\n                            'jobflow_id': self.jobflow_id,\n                        }\n                    )\n\n                # When cluster is ready, record a count and time to ready.\n                if self.ready_at and not self.finished_at:\n                    # A simple count to track number of clusters spun up\n                    # successfully.\n                    Metric.record('cluster-ready', data={\n                        'identifier': self.identifier,\n                        'size': self.size,\n                        'jobflow_id': self.jobflow_id,\n                    })\n                    # Time in seconds it took the cluster to be ready.\n                    time_to_ready = (self.ready_at - self.started_at).seconds\n                    Metric.record(\n                        'cluster-time-to-ready', time_to_ready,\n                        data={\n                            'identifier': self.identifier,\n                            'size': self.size,\n                            'jobflow_id': self.jobflow_id,\n                        }\n                    )\n\n    def save(self, *args, **kwargs):\n        \"\"\"Insert the cluster into the database or update it if already\n        present, spawning the cluster if it's not already spawned.\n        \"\"\"\n        # actually start the cluster\n        if self.jobflow_id is None:\n            self.jobflow_id = self.provisioner.start(\n                user_username=self.created_by.username,\n                user_email=self.created_by.email,\n                identifier=self.identifier,\n                emr_release=self.emr_release.version,\n                size=self.size,\n                public_key=self.ssh_key.key,\n            )\n            # once we've stored the jobflow id we can fetch the status for the first time\n            transaction.on_commit(self.sync)\n\n            with transaction.atomic():\n                Metric.record('cluster-emr-version',\n                              data={'version': self.emr_release.version})\n\n        # set the dates\n        if not self.expires_at:\n            # clusters should expire after the lifetime it's set to\n            self.expires_at = timezone.now() + timedelta(hours=self.lifetime)\n\n        super().save(*args, **kwargs)\n\n    def extend(self, hours):\n        \"\"\"Extend the cluster lifetime by the given number of hours.\"\"\"\n        self.expires_at = models.F('expires_at') + timedelta(hours=hours)\n        self.lifetime_extension_count = models.F('lifetime_extension_count') + 1\n        self.expiration_mail_sent = False\n        self.save()\n\n        with transaction.atomic():\n            Metric.record('cluster-extension', data={\n                'identifier': self.identifier,\n                'size': self.size,\n                'jobflow_id': self.jobflow_id,\n            })\n\n    def deactivate(self):\n        \"\"\"Shutdown the cluster and update its status accordingly\"\"\"\n        self.provisioner.stop(self.jobflow_id)\n        self.sync()\n", "path": "atmo/clusters/models.py"}]}