prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <issue>
  aws - new generic resource action "invoke-stepfunction" enhancement
  Enhancement which behaves just like action invoke-lambda but invokes (aka starts execution of) step function state machines for use cases which need sequencing and conditional logic which is more easily implemented in step functions.  
  
  Doc hint: See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions.html#SFN.Client.start_execution
  
  </issue>
  <code>
  [start of c7n/resources/sfn.py]
  1 # Copyright 2015-2017 Capital One Services, LLC
  2 #
  3 # Licensed under the Apache License, Version 2.0 (the "License");
  4 # you may not use this file except in compliance with the License.
  5 # You may obtain a copy of the License at
  6 #
  7 # http://www.apache.org/licenses/LICENSE-2.0
  8 #
  9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 # See the License for the specific language governing permissions and
  13 # limitations under the License.
  14 from __future__ import absolute_import, division, print_function, unicode_literals
  15 
  16 from c7n.manager import resources
  17 from c7n.query import QueryResourceManager
  18 from c7n.tags import Tag, RemoveTag
  19 
  20 
  21 @resources.register('step-machine')
  22 class StepFunction(QueryResourceManager):
  23     """AWS Step Functions State Machine"""
  24 
  25     class resource_type(object):
  26         service = 'stepfunctions'
  27         enum_spec = ('list_state_machines', 'stateMachines', None)
  28         arn = id = 'stateMachineArn'
  29         name = 'name'
  30         date = 'creationDate'
  31         dimension = None
  32         detail_spec = (
  33             "describe_state_machine", "stateMachineArn",
  34             'stateMachineArn', None)
  35         filter_name = None
  36 
  37 
  38 @StepFunction.action_registry.register('tag')
  39 class TagStepFunction(Tag):
  40     """Action to create tag(s) on a step function
  41 
  42     :example:
  43 
  44     .. code-block:: yaml
  45 
  46             policies:
  47               - name: tag-step-function
  48                 resource: step-machine
  49                 actions:
  50                   - type: tag
  51                     key: target-tag
  52                     value: target-tag-value
  53     """
  54 
  55     permissions = ('stepfunctions:TagResource',)
  56 
  57     def process_resource_set(self, client, resources, tags):
  58 
  59         tags_lower = []
  60 
  61         for tag in tags:
  62             tags_lower.append({k.lower(): v for k, v in tag.items()})
  63 
  64         for r in resources:
  65             client.tag_resource(resourceArn=r['stateMachineArn'], tags=tags_lower)
  66 
  67 
  68 @StepFunction.action_registry.register('remove-tag')
  69 class UnTagStepFunction(RemoveTag):
  70     """Action to create tag(s) on a step function
  71 
  72     :example:
  73 
  74     .. code-block:: yaml
  75 
  76             policies:
  77               - name: step-function-remove-tag
  78                 resource: step-machine
  79                 actions:
  80                   - type: remove-tag
  81                     tags: ["test"]
  82     """
  83 
  84     permissions = ('stepfunctions:UntagResource',)
  85 
  86     def process_resource_set(self, client, resources, tag_keys):
  87 
  88         for r in resources:
  89             client.untag_resource(resourceArn=r['stateMachineArn'], tagKeys=tag_keys)
  90
  [end of c7n/resources/sfn.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: diff --git a/c7n/resources/sfn.py b/c7n/resources/sfn.py
  --- a/c7n/resources/sfn.py
  +++ b/c7n/resources/sfn.py
  @@ -13,9 +13,12 @@
   # limitations under the License.
   from __future__ import absolute_import, division, print_function, unicode_literals
   
  +
  +from c7n.actions import Action
   from c7n.manager import resources
   from c7n.query import QueryResourceManager
   from c7n.tags import Tag, RemoveTag
  +from c7n.utils import type_schema, local_session, dumps, chunks
   
   
   @resources.register('step-machine')
  @@ -35,6 +38,95 @@
           filter_name = None
   
   
  +class InvokeStepFunction(Action):
  +    """Invoke step function on resources.
  +
  +    By default this will invoke a step function for each resource
  +    providing both the `policy` and `resource` as input.
  +
  +    That behavior can be configured setting policy and bulk
  +    boolean flags on the action.
  +
  +    If bulk action parameter is set to true, then the step
  +    function will be invoked in bulk, with a set of resource arns
  +    under the `resources` key.
  +
  +    The size of the batch can be configured via the batch-size
  +    parameter. Note step function state (input, execution, etc)must
  +    fit within 32k, we default to batch size 250.
  +
  +    :example:
  +
  +    .. code-block:: yaml
  +
  +       policies:
  +         - name: invoke-step-function
  +           resource: s3
  +           filters:
  +             - is-log-target
  +             - "tag:IngestSetup": absent
  +           actions:
  +             - type: invoke-sfn
  +               # This will cause the workflow to be invoked
  +               # with many resources arns in a single execution.
  +               # Note this is *not* the default.
  +               bulk: true
  +               batch-size: 10
  +               state-machine: LogIngestSetup
  +    """
  +
  +    schema = type_schema(
  +        'invoke-sfn',
  +        required=['state-machine'],
  +        **{'state-machine': {'type': 'string'},
  +           'batch-size': {'type': 'integer'},
  +           'bulk': {'type': 'boolean'},
  +           'policy': {'type': 'boolean'}})
  +    schema_alias = True
  +    permissions = ('stepfunctions:StartExecution',)
  +
  +    def process(self, resources):
  +        client = local_session(
  +            self.manager.session_factory).client('stepfunctions')
  +        arn = self.data['state-machine']
  +        if not arn.startswith('arn'):
  +            arn = 'arn:aws:states:{}:{}:stateMachine:{}'.format(
  +                self.manager.config.region, self.manager.config.account_id, arn)
  +
  +        params = {'stateMachineArn': arn}
  +        pinput = {}
  +
  +        if self.data.get('policy', True):
  +            pinput['policy'] = dict(self.manager.data)
  +
  +        resource_set = list(zip(self.manager.get_arns(resources), resources))
  +        if self.data.get('bulk', False) is True:
  +            return self.invoke_batch(client, params, pinput, resource_set)
  +
  +        for arn, r in resource_set:
  +            pinput['resource'] = r
  +            params['input'] = dumps(pinput)
  +            r['c7n:execution-arn'] = self.manager.retry(
  +                client.start_execution, **params).get('executionArn')
  +
  +    def invoke_batch(self, client, params, pinput, resource_set):
  +        for batch_rset in chunks(resource_set, self.data.get('batch-size', 250)):
  +            pinput['resources'] = [rarn for rarn, _ in batch_rset]
  +            params['input'] = dumps(pinput)
  +            exec_arn = self.manager.retry(
  +                client.start_execution, **params).get('executionArn')
  +            for _, r in resource_set:
  +                r['c7n:execution-arn'] = exec_arn
  +
  +    @classmethod
  +    def register(cls, registry, key):
  +        for _, r in registry.items():
  +            r.action_registry.register('invoke-sfn', cls)
  +
  +
  +resources.subscribe(resources.EVENT_FINAL, InvokeStepFunction.register)
  +
  +
   @StepFunction.action_registry.register('tag')
   class TagStepFunction(Tag):
       """Action to create tag(s) on a step function
  
verification_info: {"golden_diff": "diff --git a/c7n/resources/sfn.py b/c7n/resources/sfn.py\n--- a/c7n/resources/sfn.py\n+++ b/c7n/resources/sfn.py\n@@ -13,9 +13,12 @@\n # limitations under the License.\n from __future__ import absolute_import, division, print_function, unicode_literals\n \n+\n+from c7n.actions import Action\n from c7n.manager import resources\n from c7n.query import QueryResourceManager\n from c7n.tags import Tag, RemoveTag\n+from c7n.utils import type_schema, local_session, dumps, chunks\n \n \n @resources.register('step-machine')\n@@ -35,6 +38,95 @@\n         filter_name = None\n \n \n+class InvokeStepFunction(Action):\n+    \"\"\"Invoke step function on resources.\n+\n+    By default this will invoke a step function for each resource\n+    providing both the `policy` and `resource` as input.\n+\n+    That behavior can be configured setting policy and bulk\n+    boolean flags on the action.\n+\n+    If bulk action parameter is set to true, then the step\n+    function will be invoked in bulk, with a set of resource arns\n+    under the `resources` key.\n+\n+    The size of the batch can be configured via the batch-size\n+    parameter. Note step function state (input, execution, etc)must\n+    fit within 32k, we default to batch size 250.\n+\n+    :example:\n+\n+    .. code-block:: yaml\n+\n+       policies:\n+         - name: invoke-step-function\n+           resource: s3\n+           filters:\n+             - is-log-target\n+             - \"tag:IngestSetup\": absent\n+           actions:\n+             - type: invoke-sfn\n+               # This will cause the workflow to be invoked\n+               # with many resources arns in a single execution.\n+               # Note this is *not* the default.\n+               bulk: true\n+               batch-size: 10\n+               state-machine: LogIngestSetup\n+    \"\"\"\n+\n+    schema = type_schema(\n+        'invoke-sfn',\n+        required=['state-machine'],\n+        **{'state-machine': {'type': 'string'},\n+           'batch-size': {'type': 'integer'},\n+           'bulk': {'type': 'boolean'},\n+           'policy': {'type': 'boolean'}})\n+    schema_alias = True\n+    permissions = ('stepfunctions:StartExecution',)\n+\n+    def process(self, resources):\n+        client = local_session(\n+            self.manager.session_factory).client('stepfunctions')\n+        arn = self.data['state-machine']\n+        if not arn.startswith('arn'):\n+            arn = 'arn:aws:states:{}:{}:stateMachine:{}'.format(\n+                self.manager.config.region, self.manager.config.account_id, arn)\n+\n+        params = {'stateMachineArn': arn}\n+        pinput = {}\n+\n+        if self.data.get('policy', True):\n+            pinput['policy'] = dict(self.manager.data)\n+\n+        resource_set = list(zip(self.manager.get_arns(resources), resources))\n+        if self.data.get('bulk', False) is True:\n+            return self.invoke_batch(client, params, pinput, resource_set)\n+\n+        for arn, r in resource_set:\n+            pinput['resource'] = r\n+            params['input'] = dumps(pinput)\n+            r['c7n:execution-arn'] = self.manager.retry(\n+                client.start_execution, **params).get('executionArn')\n+\n+    def invoke_batch(self, client, params, pinput, resource_set):\n+        for batch_rset in chunks(resource_set, self.data.get('batch-size', 250)):\n+            pinput['resources'] = [rarn for rarn, _ in batch_rset]\n+            params['input'] = dumps(pinput)\n+            exec_arn = self.manager.retry(\n+                client.start_execution, **params).get('executionArn')\n+            for _, r in resource_set:\n+                r['c7n:execution-arn'] = exec_arn\n+\n+    @classmethod\n+    def register(cls, registry, key):\n+        for _, r in registry.items():\n+            r.action_registry.register('invoke-sfn', cls)\n+\n+\n+resources.subscribe(resources.EVENT_FINAL, InvokeStepFunction.register)\n+\n+\n @StepFunction.action_registry.register('tag')\n class TagStepFunction(Tag):\n     \"\"\"Action to create tag(s) on a step function\n", "issue": "aws - new generic resource action \"invoke-stepfunction\" enhancement\nEnhancement which behaves just like action invoke-lambda but invokes (aka starts execution of) step function state machines for use cases which need sequencing and conditional logic which is more easily implemented in step functions.  \r\n\r\nDoc hint: See https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/stepfunctions.html#SFN.Client.start_execution\n", "before_files": [{"content": "# Copyright 2015-2017 Capital One Services, LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\nfrom __future__ import absolute_import, division, print_function, unicode_literals\n\nfrom c7n.manager import resources\nfrom c7n.query import QueryResourceManager\nfrom c7n.tags import Tag, RemoveTag\n\n\n@resources.register('step-machine')\nclass StepFunction(QueryResourceManager):\n    \"\"\"AWS Step Functions State Machine\"\"\"\n\n    class resource_type(object):\n        service = 'stepfunctions'\n        enum_spec = ('list_state_machines', 'stateMachines', None)\n        arn = id = 'stateMachineArn'\n        name = 'name'\n        date = 'creationDate'\n        dimension = None\n        detail_spec = (\n            \"describe_state_machine\", \"stateMachineArn\",\n            'stateMachineArn', None)\n        filter_name = None\n\n\n@StepFunction.action_registry.register('tag')\nclass TagStepFunction(Tag):\n    \"\"\"Action to create tag(s) on a step function\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: tag-step-function\n                resource: step-machine\n                actions:\n                  - type: tag\n                    key: target-tag\n                    value: target-tag-value\n    \"\"\"\n\n    permissions = ('stepfunctions:TagResource',)\n\n    def process_resource_set(self, client, resources, tags):\n\n        tags_lower = []\n\n        for tag in tags:\n            tags_lower.append({k.lower(): v for k, v in tag.items()})\n\n        for r in resources:\n            client.tag_resource(resourceArn=r['stateMachineArn'], tags=tags_lower)\n\n\n@StepFunction.action_registry.register('remove-tag')\nclass UnTagStepFunction(RemoveTag):\n    \"\"\"Action to create tag(s) on a step function\n\n    :example:\n\n    .. code-block:: yaml\n\n            policies:\n              - name: step-function-remove-tag\n                resource: step-machine\n                actions:\n                  - type: remove-tag\n                    tags: [\"test\"]\n    \"\"\"\n\n    permissions = ('stepfunctions:UntagResource',)\n\n    def process_resource_set(self, client, resources, tag_keys):\n\n        for r in resources:\n            client.untag_resource(resourceArn=r['stateMachineArn'], tagKeys=tag_keys)\n", "path": "c7n/resources/sfn.py"}]}