chore: cherry-pick upstream tagging fixes onto July release branch

.github/workflows/stacklet-release.yml

@@ -26,15 +26,16 @@ jobs:
           persist-credentials: false
 
       # This repo only ever ships prereleases stamped with a +HASH local
-      # version segment (…aN+HASH). A version without both is a misconfiguration
+      # version segment (…aN+HASH, or …aN.postM+HASH for a hotfix on a
+      # release branch). A version without both is a misconfiguration
       # — fail loudly before building or publishing anything.
       - name: Assert prerelease version
         shell: bash
         run: |
           set -euo pipefail
           version=$(grep -m1 '^version' pyproject.toml | cut -d'"' -f2)
-          if [[ ! "$version" =~ a[0-9]+[+][0-9a-zA-Z.]+$ ]]; then
-            echo "::error::cloud-custodian-prerelease must ship a stamped prerelease version (…aN+HASH); got '$version'"
+          if [[ ! "$version" =~ a[0-9]+(\.post[0-9]+)?[+][0-9a-zA-Z.]+$ ]]; then
+            echo "::error::cloud-custodian-prerelease must ship a stamped prerelease version (…aN+HASH or …aN.postM+HASH); got '$version'"
             exit 1
           fi
           echo "Releasing prerelease version $version"

c7n/lookup.py

@@ -6,28 +6,52 @@
 import copy
 
 
+class Skip:
+    """Sentinel returned by Lookup.resolve_value to omit a value entirely."""
+
+    __slots__ = ()
+
+    def __repr__(self):
+        return 'Lookup.SKIP'
+
+
 class Lookup:
     RESOURCE_SOURCE = 'resource'
 
+    SKIP = Skip()
+
+    # A lookup needs a key, a fallback, or both. With a key it resolves against
+    # the resource and falls back to 'default-value' when the path yields
+    # nothing. With no key it is just the fallback -- which callers that pass a
+    # name to resolve_value treat as a conditional default, writing it only
+    # when that name is not already present on the resource.
     schema = {
         'type': 'object',
-        'oneOf': [
-            {
-                'properties': {
-                    'type': {'type': 'string', 'enum': [RESOURCE_SOURCE]},
-                    'key': {'type': 'string'}
-                },
-                'additionalProperties': False,
-                'required': ['type', 'key']
-            }
+        'properties': {
+            'type': {'type': 'string', 'enum': [RESOURCE_SOURCE]},
+            'key': {'type': 'string'},
+            'default-value': {},
+        },
+        'additionalProperties': False,
+        'required': ['type'],
+        'anyOf': [
+            {'required': ['key']},
+            {'required': ['default-value']},
         ]
     }
 
     @staticmethod
-    def lookup_type(schema):
+    def lookup_type(schema, default_schema=None):
+        """Schema for a value that may be given directly or looked up.
+
+        ``default_schema`` types the lookup's fallback, and defaults to
+        ``schema``. Pass it when the fallback is narrower than the value --
+        tag values, for instance, accept unquoted YAML scalars for backwards
+        compatibility, but a fallback is always written as a string.
+        """
         lookup_schema = copy.deepcopy(Lookup.schema)
-        for lookup in lookup_schema['oneOf']:
-            lookup['properties']['default-value'] = schema
+        lookup_schema['properties']['default-value'] = (
+            schema if default_schema is None else default_schema)
 
         return {
             'oneOf': [
@@ -36,6 +60,30 @@ def lookup_type(schema):
             ]
         }
 
+    @staticmethod
+    def resolve_value(spec, resource, name=None, current_names=()):
+        """Resolve a value spec against a resource.
+
+        Handles both lookup forms and plain scalars:
+
+        - a scalar is returned unchanged
+        - ``{type: resource, key: <jmespath>}`` resolves against the resource,
+          falling back to ``default-value`` when the path yields nothing
+        - ``{type: resource, default-value: <x>}`` (no key) is a conditional
+          default: it resolves to ``x`` only when ``name`` is absent from
+          ``current_names``, and to ``Lookup.SKIP`` otherwise.
+        """
+        if not Lookup.is_lookup(spec):
+            return spec
+        if 'key' not in spec and name in current_names:
+            return Lookup.SKIP
+        return Lookup.extract(spec, resource)
+
+    @staticmethod
+    def has_lookups(spec_map):
+        """True if any value in the mapping is a lookup rather than a scalar."""
+        return any(Lookup.is_lookup(v) for v in spec_map.values())
+
     @staticmethod
     def extract(source, data=None):
         if Lookup.is_lookup(source):
@@ -54,10 +102,11 @@ def get_value(source, data=None):
 
     @staticmethod
     def get_value_from_resource(source, resource):
-        value = jmespath_search(source['key'], resource)
+        if 'key' in source:
+            value = jmespath_search(source['key'], resource)
+            if value is not None:
+                return value
 
-        if value is not None:
-            return value
         if 'default-value' not in source:
             raise Exception('Lookup for key, {}, returned None'.format(source['key']))
         else:

c7n/resources/asg.py

@@ -20,9 +20,11 @@
 from c7n.manager import resources
 from c7n import query
 from c7n.resources.securityhub import PostFinding
-from c7n.tags import TagActionFilter, DEFAULT_TAG, TagCountFilter, TagTrim, TagDelayedAction
+from c7n.tags import (
+    TagActionFilter, DEFAULT_TAG, TagCountFilter, TagTrim, TagDelayedAction,
+    TagValueResolver, has_dynamic_tag_values, tag_value_schema)
 from c7n.utils import (
-    FormatDate, local_session, type_schema, chunks, get_retry, select_keys)
+    local_session, type_schema, chunks, get_retry, select_keys)
 
 from .ec2 import deserialize_user_data
 
@@ -1104,7 +1106,7 @@ def process_resource_set(self, client, asgs, tags):
 
 @ASG.action_registry.register('tag')
 @ASG.action_registry.register('mark')
-class Tag(Action):
+class Tag(TagValueResolver, Action):
     """Action to add a tag to an ASG
 
     The *propagate* parameter can be used to specify that the tag being added
@@ -1125,13 +1127,32 @@ class Tag(Action):
                     key: OwnerName
                     value: OwnerName
                     propagate: true
+
+    Tag values may also be looked up per-asg from the asg's attributes:
+
+    .. code-block:: yaml
+
+            policies:
+              - name: asg-tag-owner
+                resource: asg
+                actions:
+                  - type: tag
+                    propagate: true
+                    tags:
+                      Owner:
+                        type: resource
+                        key: "Tags[?Key=='Team'].Value | [0]"  # from the Team tag
+                        default-value: unknown
+                      CostCenter:
+                        type: resource
+                        default-value: unassigned  # set only if CostCenter absent
     """
 
     schema = type_schema(
         'tag',
         key={'type': 'string'},
-        value={'type': 'string'},
-        tags={'type': 'object'},
+        value=tag_value_schema(),
+        tags={'type': 'object', 'additionalProperties': tag_value_schema()},
         # Backwards compatibility
         tag={'type': 'string'},
         msg={'type': 'string'},
@@ -1141,34 +1162,57 @@ class Tag(Action):
     permissions = ('autoscaling:CreateOrUpdateTags',)
     batch_size = 1
 
-    def get_tag_set(self):
-        tags = []
+    def get_tag_spec_map(self):
+        spec_map = {}
+        tags = self.data.get('tags', {})
         key = self.data.get('key', self.data.get('tag', DEFAULT_TAG))
-        value = self.data.get(
-            'value', self.data.get(
-                'msg', 'AutoScaleGroup does not meet policy guidelines'))
+        value = self.data.get('value', self.data.get('msg'))
+        # The mark default only applies to the single-tag form. A policy that
+        # spells out a tags mapping gets that mapping and nothing else --
+        # otherwise every asg picks up a marker tag it never asked for, and a
+        # conditional default can never resolve to an empty payload.
+        if value is None and not tags:
+            value = 'AutoScaleGroup does not meet policy guidelines'
         if key and value:
-            tags.append({'Key': key, 'Value': value})
+            spec_map[key] = value
+
+        spec_map.update(tags)
 
-        for k, v in self.data.get('tags', {}).items():
-            tags.append({'Key': k, 'Value': v})
+        return spec_map
 
-        return tags
+    def resolve_tag_sets(self, asgs, spec_map):
+        """Yield (asg_set, tags) pairs to submit.
+
+        Static values are shared by every asg, so they go out in batches.
+        Dynamic values are resolved per asg and then regrouped, so asgs that
+        land on the same payload still batch together, and an asg whose tags
+        all resolve to skip drops out entirely.
+        """
+        if not has_dynamic_tag_values(spec_map):
+            tags = [{'Key': k, 'Value': v} for k, v in spec_map.items()]
+            self.interpolate_values(tags)
+            for asg_set in chunks(asgs, self.batch_size):
+                yield asg_set, tags
+            return
+
+        for asg_set, resolved in self.resolve_and_group(asgs, spec_map):
+            tags = [{'Key': k, 'Value': v} for k, v in resolved.items()]
+            for chunk in chunks(asg_set, self.batch_size):
+                yield chunk, tags
 
     def process(self, asgs):
-        tags = self.get_tag_set()
+        spec_map = self.get_tag_spec_map()
         error = None
 
-        self.interpolate_values(tags)
-
         client = self.get_client()
         with self.executor_factory(max_workers=2) as w:
             futures = {}
-            for asg_set in chunks(asgs, self.batch_size):
+            for asg_set, tags in self.resolve_tag_sets(asgs, spec_map):
                 futures[w.submit(
-                    self.process_resource_set, client, asg_set, tags)] = asg_set
+                    self.process_resource_set, client, asg_set, tags)] = (
+                        asg_set, tags)
             for f in as_completed(futures):
-                asg_set = futures[f]
+                asg_set, tags = futures[f]
                 if f.exception():
                     self.log.exception(
                         "Exception tagging tag:%s error:%s asg:%s" % (
@@ -1194,14 +1238,6 @@ def process_resource_set(self, client, asgs, tags):
                 a.setdefault('Tags', []).append(atags)
         self.manager.retry(client.create_or_update_tags, Tags=tag_params)
 
-    def interpolate_values(self, tags):
-        params = {
-            'account_id': self.manager.config.account_id,
-            'now': FormatDate.utcnow(),
-            'region': self.manager.config.region}
-        for t in tags:
-            t['Value'] = t['Value'].format(**params)
-
     def get_client(self):
         return local_session(self.manager.session_factory).client('autoscaling')
 

c7n/resources/athena.py

@@ -2,7 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 from c7n.actions import Action
 from c7n.manager import resources
-from c7n import query
+from c7n import query, tags
 from c7n.utils import type_schema, local_session
 
 from .aws import shape_validate
@@ -63,9 +63,7 @@ def process(self, resources):
 
 @resources.register("athena-data-catalog")
 class AthenaDataCatalog(query.QueryResourceManager):
-    source_mapping = {
-        "describe": query.DescribeWithResourceTags,
-    }
+    source_mapping = {"describe": query.DescribeWithResourceTags}
 
     class resource_type(query.TypeInfo):
         service = "athena"
@@ -78,6 +76,44 @@ class resource_type(query.TypeInfo):
         permissions_augment = ("athena:ListTagsForResource",)
 
 
+class SkipAwsManagedCatalog:
+    """Mixin for AthenaDataCatalog actions that cannot operate on the
+    AWS-managed AwsDataCatalog entry (e.g. tag/untag via Resource Groups
+    Tagging API).  Filters it out before delegating to the real action and
+    emits a warning so operators know why the entry was skipped."""
+
+    def process(self, resources):
+        excluded = [r for r in resources if r.get("CatalogName") == "AwsDataCatalog"]
+        if excluded:
+            self.manager.log.warning(
+                "Skipping %d AWS-managed AwsDataCatalog resource(s) "
+                "which cannot be tagged via Resource Groups Tagging API",
+                len(excluded),
+            )
+        resources = [r for r in resources if r.get("CatalogName") != "AwsDataCatalog"]
+        if not resources:
+            return
+        return super().process(resources)
+
+
+@AthenaDataCatalog.action_registry.register("mark")
+@AthenaDataCatalog.action_registry.register("tag")
+class DataCatalogTag(SkipAwsManagedCatalog, tags.UniversalTag):
+    pass
+
+
+@AthenaDataCatalog.action_registry.register("unmark")
+@AthenaDataCatalog.action_registry.register("untag")
+@AthenaDataCatalog.action_registry.register("remove-tag")
+class DataCatalogUntag(SkipAwsManagedCatalog, tags.UniversalUntag):
+    pass
+
+
+@AthenaDataCatalog.action_registry.register("mark-for-op")
+class DataCatalogMarkForOp(SkipAwsManagedCatalog, tags.UniversalTagDelayedAction):
+    pass
+
+
 @resources.register("athena-capacity-reservation")
 class AthenaCapacityReservation(query.QueryResourceManager):
     source_mapping = {