Repository: nchammas/flintrock (language: python)

Issue:
Allow configuration of EC2 spot request duration
When spinning up ec2 spot instances with flintrock for ad-hoc analysis, there can be cases where capacity is not available for a given instance type. Of course, one can wait until AWS releases instances, however often I find that I just cancel the request and spin up a different type of instance. However, as the default timeout of spot requests is 7 days, it may be that after several minutes/hours/days AWS finally release capacity for a cancelled / unneeded request. This then leads the situation where there are "inconsistent" clusters spun up in AWS, and flintrock has not fully configured them. In this case the instances must be manually terminated from the EC2 console. 

Additionally, there can be the scenario where a request is cancelled, a new request for a different instance is then created (with the same name), and both requests end up being fulfilled, which the causes issues with describe/destroy. In this PR, I have added support for using a human readable timedelta (e.g. `'7d'`, `'5m'`) in configuration/cli such that the spot request timeout can be controlled by the user (and it retains the default of 7 days). This would alleviate the problems described above, where I could simply reduce the spot request duration to be much shorter than 7 days.

This PR makes the following changes:
* Adds support for the `ValidUntil` parameter when creating ec2 spot requests
* Retains the default option where requests are valid for 7 days

I tested this PR by running pytest locally and adding new tests for the util function. However, I have not run any of the acceptance tests.
Spot instance tags issue
<!--
    If you are reporting a potential bug, include as many of the following
    as are relevant (and delete this sentence, of course):
-->
* Flintrock version: Master version
* Python version: 3.8.10
* OS: Mac M1

Hi !

It seems that there is an issue regarding `TagSpecifications` in `Make finding cluster nodes more reliable in the face of launch failures`. As we are interested in this new feature, we tried to make a release in our forked version but we got this issue:
```
Parameter validation failed:
Unknown parameter in LaunchSpecification: "TagSpecifications", must be one of: SecurityGroupIds, SecurityGroups, AddressingType, BlockDeviceMappings, EbsOptimized, IamInstanceProfile, ImageId, InstanceType, KernelId, KeyName, Monitoring, NetworkInterfaces, Placement, RamdiskId, SubnetId, UserData
```


If we are not wrong aout this issue, it seems that this issue comes from `TagSpecifications` in `common_launch_specs` that is a parameter of `request_spot_instances` and not a key inside `LaunchSpecification` dict as you can see [here](https://boto3.amazonaws.com/v1/documentation/api/1.21.44/reference/services/ec2.html?highlight=create_instance#EC2.Client.request_spot_instances). Also, we can't put `ResourceType': 'instance'` in `client.request_spot_instances` as explained in the [documentation](https://boto3.amazonaws.com/v1/documentation/api/1.21.44/reference/services/ec2.html?highlight=create_instance#EC2.Client.request_spot_instances):
```
TagSpecifications (list) --
The key-value pair for tagging the Spot Instance request on creation. The value for ResourceType must be spot-instances-request , otherwise the Spot Instance request fails. To tag the Spot Instance request after it has been created, see CreateTags.
```
So, I think, we need to put `ResourceType': 'spot-instances-request'` if we want to tag the spot request instance (not spot instance), if not we can remove `TagSpecifications` from spot request instance. 


Here is a possible solution in `_create_instances`:
1. Remove `TagSpecifications` from `common_launch_specs`
2. Add `TagSpecifications=[{'ResourceType': 'spot-instances-request', 'Tags': tag_specifications}],` in `client.request_spot_instances`  
3. Add `TagSpecifications=[{'ResourceType': 'instance', 'Tags': tag_specifications}],` in `ec2.create_instances`
4. Add tags to spot instances using `create_tags` as last step inside `if spot_price:`:
```python
(ec2.instances
             .filter(
                Filters=[
                    {'Name': 'instance-id', 'Values': [instance.id for instance in cluster_instances]}
                ])
             .create_tags(Tags=tag_specifications))
```

We have done these modifications in [our fork](https://github.com/adotmob/flintrock/commit/1533553efeb7a23aa7c565d13a9225802fc472bc), so if you're okay with that, we can do the same here or if you have maybe a more elegant way to do that it could be interesting. 

I hope it's clear and this helps !

Maxime




Discussion / hints:
Hello @mblack20 and my apologies about taking so long to chime in here! Thank you for contributing to Flintrock. I will take a look at your work and give some feedback this week.
Thanks for the review!

I can understand the reticence to add _yet another_ parameter, however I would argue this is a useful one, as it can help out with issues caused by late fulfilment of clusters, duplicate instances with the same name etc., and in fact in 2+ years of production use of flintrock this has been the only parameter I've been missing.

As for lowering the defaults, that's definitely the easier option, but I'm not sure what a sensible default would be. For example, I'd set it as low as 5 minutes, as if the instance type isn't available immediately I'd rather relaunch with another type. However, I'm not sure this would be suitable for all users and is a big departure from the default of 7 days, hence I opted for moving it to configuration instead.

Happy to hear your thoughts. Thanks!
Looks like we missed adding this new option to the `add-slaves` command.

@mblack20 - No worries if you've moved on, but if you have the time, do you mind adding this new option to the `add-slaves` command, and testing it manually?
Ah sorry I missed that, yes I'll take a look and follow up with a PR
Thanks for the clear report. I appreciate it.

I did not think to test spot instances as I worked on #362. Apologies.

I would welcome a PR to fix this issue. However, I would not want to re-introduce tagging as a separate step, as you proposed in step (4) above. This recreates the original problem of cluster nodes becoming orphaned if something goes wrong between when they are launched and when they are tagged.

We want tagging to happen as part of the same AWS API call that creates the instances, whether they are spot or not.

Is that not possible with spot instances? i.e. Are tags for spot instance _requests_ not automatically also tags for the spot _instances_ themselves?
Hi,

Ok it's more clear now why you don't use `create_tags` anymore. 

But when we tried to launch spot instances with `TagSpecifications` using `request_spot_instances`, the request is tagged but instances doesn't inherit from tags of the request. So it seems, it's not possible to tag spot instances during the same call that creates instances. 
Looks like the boto3 method we use to launch spot instances is [deprecated][1], and the solution is to migrate off that method.

> We strongly discourage using the RequestSpotInstances API because it is a legacy API with no planned investment.

[1]: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/request_spot_instances.html
Oh actually I hadn't seen that ! If we follow AWS recommandation to launch spot instances [Which is the best Spot request method to use?](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/spot-best-practices.html#which-spot-request-method-to-use), it seems that we need to use [create_fleet](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/create_fleet.html#create-fleet). I don't know exactly what is the impact of this change.. But regarding our Tag problem, it seems that `create_fleet` support instance tagging, so we could have a single AWS API call that launches the instances and tag them as described in the [documentation](https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/create_fleet.html#create-fleet): 
```
TagSpecifications (list) –

The key-value pair for tagging the EC2 Fleet request on creation. For more information, see [Tag your resources](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/Using_Tags.html#tag-resources).

If the fleet type is instant, specify a resource type of fleet to tag the fleet or instance to tag the instances at launch.

If the fleet type is maintain or request, specify a resource type of fleet to tag the fleet. You cannot specify a resource type of instance. To tag instances at launch, specify the tags in a [launch template](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-launch-templates.html#create-launch-template).
``` 
I thought `create_instances` is the API we need to use, actually. We are already using it to launch on-demand instances. It just needs different parameters to configure it for spot instances. That way we are using a single launch API across Flintrock. (Note that `RunInstances` from the AWS docs maps to `create_instances` in boto3.)

That was my plan, but I only took a quick look and plan to come back to this later. I think in the future we could migrate from `create_instances` to `create_fleet` if we want to support launching clusters with different master/slave instance types, but that's not required for now.

If you want to take a crack at this, go ahead. Otherwise, I will try to get to this next week.

Fix patch:
diff --git a/CHANGES.md b/CHANGES.md
index 94f0944..31af175 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -8,11 +8,13 @@
 
 * [#348]: Bumped default Spark to 3.2; dropped support for Python 3.6; added CI build for Python 3.10.
 * [#361]: Migrated from AdoptOpenJDK, which is deprecated, to Adoptium OpenJDK.
-* [#362]: Improved Flintrock's ability to cleanup after launch failures.
+* [#362][#366]: Improved Flintrock's ability to cleanup after launch failures.
+* [#366]: Deprecated `--ec2-spot-request-duration`, which is not needed for one-time spot instances launched using the RunInstances API.
 
 [#348]: https://github.com/nchammas/flintrock/pull/348
 [#361]: https://github.com/nchammas/flintrock/pull/361
 [#362]: https://github.com/nchammas/flintrock/pull/362
+[#366]: https://github.com/nchammas/flintrock/pull/366
 
 ## [2.0.0] - 2021-06-10
 
diff --git a/flintrock/config.yaml.template b/flintrock/config.yaml.template
index 7db03c7..fdf8da6 100644
--- a/flintrock/config.yaml.template
+++ b/flintrock/config.yaml.template
@@ -35,7 +35,6 @@ providers:
     # ami: ami-61bbf104  # CentOS 7, us-east-1
     # user: centos
     # spot-price: <price>
-    # spot-request-duration: 7d  # duration a spot request is valid, supports d/h/m/s (e.g. 4d 3h 2m 1s)
     # vpc-id: <id>
     # subnet-id: <id>
     # placement-group: <name>
diff --git a/flintrock/ec2.py b/flintrock/ec2.py
index f7d01ed..ecd1cda 100644
--- a/flintrock/ec2.py
+++ b/flintrock/ec2.py
@@ -27,7 +27,6 @@
 )
 from .ssh import generate_ssh_key_pair
 from .services import SecurityGroupRule
-from .util import duration_to_expiration
 
 logger = logging.getLogger('flintrock.ec2')
 
@@ -275,7 +274,6 @@ def add_slaves(
         identity_file: str,
         num_slaves: int,
         spot_price: float,
-        spot_request_duration: str,
         min_root_ebs_size_gb: int,
         tags: list,
         assume_yes: bool,
@@ -321,7 +319,6 @@ def add_slaves(
                 num_instances=num_slaves,
                 region=self.region,
                 spot_price=spot_price,
-                spot_request_valid_until=duration_to_expiration(spot_request_duration),
                 ami=self.master_instance.image_id,
                 assume_yes=assume_yes,
                 key_name=self.master_instance.key_name,
@@ -704,7 +701,6 @@ def _create_instances(
     num_instances,
     region,
     spot_price,
-    spot_request_valid_until,
     ami,
     assume_yes,
     key_name,
@@ -724,7 +720,6 @@ def _create_instances(
     ec2 = boto3.resource(service_name='ec2', region_name=region)
 
     cluster_instances = []
-    spot_requests = []
     common_launch_specs = {
         'ImageId': ami,
         'KeyName': key_name,
@@ -733,7 +728,6 @@ def _create_instances(
         'Placement': {
             'AvailabilityZone': availability_zone,
             'Tenancy': tenancy,
-            'GroupName': placement_group,
         },
         'SecurityGroupIds': security_group_ids,
         'SubnetId': subnet_id,
@@ -748,80 +742,36 @@ def _create_instances(
         ],
     }
 
+    if spot_price:
+        common_launch_specs.update({
+            'InstanceMarketOptions': {
+                'MarketType': 'spot',
+                'SpotOptions': {
+                    'SpotInstanceType': 'one-time',
+                    'MaxPrice': str(spot_price),
+                    'InstanceInterruptionBehavior': 'terminate',
+                },
+            }
+        })
+    else:
+        common_launch_specs.update({
+            'InstanceInitiatedShutdownBehavior': instance_initiated_shutdown_behavior,
+        })
+        # This can't be part of the previous update because we need a deep merge.
+        common_launch_specs['Placement'].update({
+            'GroupName': placement_group,
+        })
+
     try:
-        if spot_price:
-            user_data = base64.b64encode(user_data.encode('utf-8')).decode()
-            logger.info("Requesting {c} spot instances at a max price of ${p}...".format(
-                c=num_instances, p=spot_price))
-            client = ec2.meta.client
-            spot_requests = client.request_spot_instances(
-                SpotPrice=str(spot_price),
-                InstanceCount=num_instances,
-                ValidUntil=spot_request_valid_until,
-                LaunchSpecification=common_launch_specs,
-            )['SpotInstanceRequests']
-
-            request_ids = [r['SpotInstanceRequestId'] for r in spot_requests]
-            pending_request_ids = request_ids
-
-            while pending_request_ids:
-                logger.info("{grant} of {req} instances granted. Waiting...".format(
-                    grant=num_instances - len(pending_request_ids),
-                    req=num_instances))
-                time.sleep(30)
-                spot_requests = client.describe_spot_instance_requests(
-                    SpotInstanceRequestIds=request_ids)['SpotInstanceRequests']
-
-                failed_requests = [r for r in spot_requests if r['State'] == 'failed']
-                if failed_requests:
-                    failure_reasons = {r['Status']['Code'] for r in failed_requests}
-                    raise Error(
-                        "The spot request failed for the following reason{s}: {reasons}"
-                        .format(
-                            s='' if len(failure_reasons) == 1 else 's',
-                            reasons=', '.join(failure_reasons)))
-
-                pending_request_ids = [
-                    r['SpotInstanceRequestId'] for r in spot_requests
-                    if r['State'] == 'open']
-
-            logger.info("All {c} instances granted.".format(c=num_instances))
-
-            cluster_instances = list(
-                ec2.instances.filter(
-                    Filters=[
-                        {'Name': 'instance-id', 'Values': [r['InstanceId'] for r in spot_requests]}
-                    ]))
-        else:
-            cluster_instances = ec2.create_instances(
-                MinCount=num_instances,
-                MaxCount=num_instances,
-                # Shutdown Behavior is specific to on-demand instances.
-                InstanceInitiatedShutdownBehavior=instance_initiated_shutdown_behavior,
-                **common_launch_specs,
-            )
+        cluster_instances = ec2.create_instances(
+            MinCount=num_instances,
+            MaxCount=num_instances,
+            **common_launch_specs,
+        )
         return cluster_instances
     except (Exception, KeyboardInterrupt) as e:
         if not isinstance(e, KeyboardInterrupt):
             print(e, file=sys.stderr)
-        if spot_requests:
-            request_ids = [r['SpotInstanceRequestId'] for r in spot_requests]
-            if any([r['State'] != 'active' for r in spot_requests]):
-                print("Canceling spot instance requests...", file=sys.stderr)
-                client.cancel_spot_instance_requests(
-                    SpotInstanceRequestIds=request_ids)
-            # Make sure we have the latest information on any launched spot instances.
-            spot_requests = client.describe_spot_instance_requests(
-                SpotInstanceRequestIds=request_ids)['SpotInstanceRequests']
-            instance_ids = [
-                r['InstanceId'] for r in spot_requests
-                if 'InstanceId' in r]
-            if instance_ids:
-                cluster_instances = list(
-                    ec2.instances.filter(
-                        Filters=[
-                            {'Name': 'instance-id', 'Values': instance_ids}
-                        ]))
         raise InterruptedEC2Operation(instances=cluster_instances) from e
 
 
@@ -842,7 +792,6 @@ def launch(
         user,
         security_groups,
         spot_price=None,
-        spot_request_duration=None,
         min_root_ebs_size_gb,
         vpc_id,
         subnet_id,
@@ -916,7 +865,6 @@ def launch(
     common_instance_spec = {
         'region': region,
         'spot_price': spot_price,
-        'spot_request_valid_until': duration_to_expiration(spot_request_duration),
         'ami': ami,
         'assume_yes': assume_yes,
         'key_name': key_name,
diff --git a/flintrock/flintrock.py b/flintrock/flintrock.py
index 38e1188..bd7f353 100644
--- a/flintrock/flintrock.py
+++ b/flintrock/flintrock.py
@@ -39,6 +39,11 @@
 else:
     THIS_DIR = os.path.dirname(os.path.realpath(__file__))
 
+EC2_SPOT_REQUEST_DURATION_DEPRECATION_MESSAGE = (
+    "Deprecation: --ec2-spot-request-duration is deprecated. One-time spot instances do "
+    "not support a request duration. "
+    "For more information see: https://github.com/nchammas/flintrock/pull/366"
+)
 
 logger = logging.getLogger('flintrock.flintrock')
 
@@ -345,8 +350,8 @@ def cli(cli_context, config, provider, debug):
               help="Additional security groups names to assign to the instances. "
                    "You can specify this option multiple times.")
 @click.option('--ec2-spot-price', type=float)
-@click.option('--ec2-spot-request-duration', default='7d',
-              help="Duration a spot request is valid (e.g. 3d 2h 1m).")
+@click.option('--ec2-spot-request-duration',
+              help="(DEPRECATED) Duration a spot request is valid (e.g. 3d 2h 1m).")
 @click.option('--ec2-min-root-ebs-size-gb', type=int, default=30)
 @click.option('--ec2-vpc-id', default='', help="Leave empty for default VPC.")
 @click.option('--ec2-subnet-id', default='')
@@ -414,6 +419,9 @@ def launch(
     """
     Launch a new cluster.
     """
+    if ec2_spot_request_duration:
+        logger.warning(EC2_SPOT_REQUEST_DURATION_DEPRECATION_MESSAGE)
+
     provider = cli_context.obj['provider']
     services = []
 
@@ -511,7 +519,6 @@ def launch(
             user=ec2_user,
             security_groups=ec2_security_groups,
             spot_price=ec2_spot_price,
-            spot_request_duration=ec2_spot_request_duration,
             min_root_ebs_size_gb=ec2_min_root_ebs_size_gb,
             vpc_id=ec2_vpc_id,
             subnet_id=ec2_subnet_id,
@@ -787,8 +794,8 @@ def stop(cli_context, cluster_name, ec2_region, ec2_vpc_id, assume_yes):
               help="Path to SSH .pem file for accessing nodes.")
 @click.option('--ec2-user')
 @click.option('--ec2-spot-price', type=float)
-@click.option('--ec2-spot-request-duration', default='7d',
-              help="Duration a spot request is valid (e.g. 3d 2h 1m).")
+@click.option('--ec2-spot-request-duration',
+              help="(DEPRECATED) Duration a spot request is valid (e.g. 3d 2h 1m).")
 @click.option('--ec2-min-root-ebs-size-gb', type=int, default=30)
 @click.option('--assume-yes/--no-assume-yes', default=False)
 @click.option('--ec2-tag', 'ec2_tags',
@@ -816,6 +823,8 @@ def add_slaves(
     Flintrock will configure new slaves based on information queried
     automatically from the master.
     """
+    if ec2_spot_request_duration:
+        logger.warning(EC2_SPOT_REQUEST_DURATION_DEPRECATION_MESSAGE)
     provider = cli_context.obj['provider']
 
     option_requires(
@@ -842,7 +851,6 @@ def add_slaves(
         provider_options = {
             'min_root_ebs_size_gb': ec2_min_root_ebs_size_gb,
             'spot_price': ec2_spot_price,
-            'spot_request_duration': ec2_spot_request_duration,
             'tags': ec2_tags
         }
     else:
diff --git a/flintrock/util.py b/flintrock/util.py
index 961f143..c625618 100644
--- a/flintrock/util.py
+++ b/flintrock/util.py
@@ -1,7 +1,5 @@
 import os
 import sys
-from datetime import datetime, timedelta, timezone
-from decimal import Decimal
 
 FROZEN = getattr(sys, 'frozen', False)
 
@@ -20,47 +18,6 @@ def get_subprocess_env() -> dict:
     return env
 
 
-def duration_to_timedelta(duration_string):
-    """
-    Convert a time duration string (e.g. 3h 4m 10s) into a timedelta
-    """
-
-    duration_string = duration_string.lower()
-
-    total_seconds = Decimal('0')
-
-    prev_num = []
-    for character in duration_string:
-        if character.isalpha():
-            if prev_num:
-                num = Decimal(''.join(prev_num))
-                if character == 'd':
-                    total_seconds += num * 60 * 60 * 24
-                elif character == 'h':
-                    total_seconds += num * 60 * 60
-                elif character == 'm':
-                    total_seconds += num * 60
-                elif character == 's':
-                    total_seconds += num
-                prev_num = []
-
-        elif character.isnumeric() or character == '.':
-            prev_num.append(character)
-
-    return timedelta(seconds=float(total_seconds))
-
-
-def duration_to_expiration(duration_string):
-    default_duration = timedelta(days=7)
-
-    if not duration_string:
-        expiration = datetime.now(tz=timezone.utc) + default_duration
-    else:
-        expiration = datetime.now(tz=timezone.utc) + duration_to_timedelta(duration_string)
-
-    return expiration
-
-
 def spark_hadoop_build_version(hadoop_version: str) -> str:
     """
     Given a Hadoop version, determine the Hadoop build of Spark to use.


Test patch:
diff --git a/tests/test_util.py b/tests/test_util.py
index 12455e7..d72d9d4 100644
--- a/tests/test_util.py
+++ b/tests/test_util.py
@@ -1,23 +1,4 @@
-from datetime import datetime, timedelta, timezone
-from flintrock.util import (
-    duration_to_timedelta,
-    duration_to_expiration,
-    spark_hadoop_build_version,
-)
-from freezegun import freeze_time
-
-
-def test_duration_to_timedelta():
-    assert duration_to_timedelta('1d') == timedelta(days=1)
-    assert duration_to_timedelta('3d2h1m') == timedelta(days=3, hours=2, minutes=1)
-    assert duration_to_timedelta('4d 2h 1m 5s') == timedelta(days=4, hours=2, minutes=1, seconds=5)
-    assert duration_to_timedelta('36h') == timedelta(hours=36)
-    assert duration_to_timedelta('7d') == timedelta(days=7)
-
-
-@freeze_time("2012-01-14")
-def test_duration_to_expiration():
-    assert duration_to_expiration('5m') == datetime.now(tz=timezone.utc) + timedelta(minutes=5)
+from flintrock.util import spark_hadoop_build_version
 
 
 def test_spark_hadoop_build_version():