src/cloud-prune: rework duration conversion, move to cosalib

Rework the duration parsing code to support days, weeks,
months and years. Also use a regexp so we don't need the
space.

In preparation for the container garbage collection code,
which will consume the same policy files, move the
code to the shared cosalib.

Also updates the parse_fcos_version to returns the stream id in
a tuple along the build timestamp.

src/cmd-cloud-prune

@@ -6,11 +6,12 @@
 # Example of policy.yaml
 # rawhide:
 #     # all cloud images
-#     cloud-uploads: 2 years
+#     cloud-uploads: 2y
 #     # artifacts in meta.json's `images` key
-#     images: 2 years
+#     images: 2y
 #     images-keep: [qemu, live-iso]
-#     build: 3 years
+#     build: 3y
+#     containers: 2w
 # The script also updates the builds.json for the respective stream by
 # adding the policy-cleanup key when we set the upload_builds_json flag.
 # It adds the relevant actions completed to that key
@@ -44,7 +45,8 @@ from cosalib.gcp import remove_gcp_image
 from cosalib.aws import deregister_aws_resource
 from cosalib.builds import BUILDFILES
 from cosalib.s3 import s3_copy
-from cosalib.cmdlib import parse_fcos_version_to_timestamp
+from cosalib.cmdlib import parse_fcos_version_to_timestamp_and_stream
+from cosalib.cmdlib import convert_duration_to_days
 
 Build = collections.namedtuple("Build", ["id", "images", "arch", "meta_json"])
 # set metadata caching to 5m
@@ -105,17 +107,17 @@ def main():
     for action in ['cloud-uploads', 'images', 'build']:
         if action not in policy[stream]:
             continue
-        duration = get_period_in_months(policy[stream][action])
-        ref_date = today_date - relativedelta(months=int(duration))
+        duration = convert_duration_to_days(policy[stream][action])
+        ref_date = today_date - relativedelta(days=int(duration))
 
-        print(f"Pruning resources of type {action} older than {duration} months ({ref_date.date()}) on stream {stream}")
+        print(f"Pruning resources of type {action} older than {policy[stream][action]} ({ref_date.date()}) on stream {stream}")
         # Enumerating in reverse to go from the oldest build to the newest one
         for index, build in enumerate(reversed(builds_json_data["builds"])):
             build_id = build["id"]
             if action in build.get("policy-cleanup", []):
                 print(f"Build {build_id} has already had {action} pruning completed")
                 continue
-            build_date = parse_fcos_version_to_timestamp(build_id)
+            (build_date, _) = parse_fcos_version_to_timestamp_and_stream(build_id)
 
             if build_date >= ref_date:
                 break
@@ -172,8 +174,8 @@ def validate_policy(stream, policy):
         actions = policy[stream]
         if 'cloud-uploads' not in actions:
             raise Exception("Pruning for cloud-uploads must be set before we prune the builds")
-        cloud_uploads_duration = get_period_in_months(actions["cloud-uploads"])
-        build_duration = get_period_in_months(actions["build"])
+        cloud_uploads_duration = convert_duration_to_days(actions["cloud-uploads"])
+        build_duration = convert_duration_to_days(actions["build"])
         if cloud_uploads_duration > build_duration:
             raise Exception("Duration of pruning cloud-uploads must be less than or equal to pruning a build")
 
@@ -286,15 +288,5 @@ def delete_gcp_image(build, cloud_config, dry_run):
     return errors
 
 
-def get_period_in_months(duration):
-    val, unit = duration.split(maxsplit=1)
-    if unit in ["years", "year", "y"]:
-        return int(val) * 12
-    elif unit in ["months", "month", "m"]:
-        return int(val)
-    else:
-        raise Exception(f"Duration unit provided is {unit}. Pruning duration is only supported in years and months")
-
-
 if __name__ == "__main__":
     main()

src/cosalib/cmdlib.py

@@ -339,7 +339,7 @@ def get_basearch():
         return get_basearch.saved
 
 
-def parse_fcos_version_to_timestamp(version):
+def parse_fcos_version_to_timestamp_and_stream(version):
     '''
     Parses an FCOS build ID and verifies the versioning is accurate. Then
     it verifies that the parsed timestamp has %Y%m%d format and returns that.
@@ -351,7 +351,34 @@ def parse_fcos_version_to_timestamp(version):
         timestamp = datetime.datetime.strptime(m.group(2), '%Y%m%d')
     except ValueError:
         raise Exception(f"FCOS build {version} has incorrect date format. It should be in (%Y%m%d)")
-    return timestamp
+    return (timestamp, int(m.group(3)))
+
+
+def convert_duration_to_days(duration_arg):
+    """
+    Parses duration strings and convert them into days.
+    The excpected format is Nd/D, nw/W, Nm/M, Ny/Y where N is a positive integer.
+    The return value is the number of days represented, in integer format
+    """
+    match = re.match(r'^([0-9]+)([dDmMyYwW])$', duration_arg)
+
+    if match is None:
+        raise ValueError(f"Incorrect duration '{duration_arg}'. Valid values are in the form of 1d, 2w, 3m, 4y")
+
+    unit = match.group(2)
+    value = int(match.group(1))
+    match unit.lower():
+        case "y":
+            days = value * 365
+        case "m":
+            days = value * 30
+        case "w":
+            days = value * 7
+        case "d":
+            days = value
+        case _:
+            raise ValueError(f"Invalid unit '{match.group(2)}'. Please use y (years), m (months), w (weeks), or d (days).")
+    return days
 
 
 def parse_date_string(date_string):