Checklists V2

.github/actions/get_aprl/entrypoint.py

@@ -93,6 +93,7 @@ def get_aprl_recos():
     github_file_extension = '.yaml'
     github_branch = 'master'
     retrieved_recos = []
+    timestamp = datetime.date.today().strftime("%B %d, %Y")
     # Get last commit to APRL reco
     if (verbose): print("DEBUG: Scanning GitHub repository {0} for {1} files...".format(github_repo, github_file_extension))
     r = requests.get(f'https://api.github.com/repos/{github_org}/{github_repo}/commits')
@@ -131,7 +132,9 @@ def get_aprl_recos():
                                 item['severity'] = item['recommendationImpact']
                                 item['category'] = item['recommendationControl']
                                 item['guid'] = item['aprlGuid']
-                                item['source'] = file_path
+                                item['sourceFile'] = file_path
+                                item['sourceType'] = 'aprl'
+                                item['timestamp'] = timestamp
                             retrieved_recos += aprl_recos
                             if verbose: print("DEBUG: {0} recommendations found in file {1}".format(len(aprl_recos), file_path))
                     else:

.github/actions/get_service_guides/entrypoint.py

@@ -75,6 +75,11 @@
     args_verbose = (sys.argv[3].lower() == 'true')
 except:
     args_verbose = True
+try:
+    args_overwrite = (sys.argv[4].lower() == 'true')
+except:
+    args_overwrite = False
+
 # These parameters haven't been implemented in the github action
 args_print_json = False
 args_extract_key_phrases_checklist = False
@@ -197,10 +202,11 @@ def short_pillar(pillar):
     else: return pillar
 
 # Function to parse markdown
-def parse_markdown(markdown, service, verbose=False):
+def parse_markdown(markdown, service, source=None, verbose=False):
     recos = []
     waf_pillars = ['cost optimization', 'operational excellence', 'performance efficiency', 'reliability', 'security']
     processing_pillar = ''
+    timestamp = datetime.date.today().strftime("%B %d, %Y")
     if (verbose): print("DEBUG: Processing markdown file...")
     line_count = 0
     for line in markdown.split('\n'):
@@ -211,7 +217,10 @@ def parse_markdown(markdown, service, verbose=False):
             if (verbose): print("DEBUG: Processing pillar '{0}'".format(processing_pillar))
         if (line[0:4] == '> - ') and (processing_pillar != ''):
             reco = line[4:]
-            recos.append({'waf': processing_pillar, 'service': service, 'text': remove_markdown(reco), 'description': '', 'type': 'checklist'})
+            reco_object = {'waf': processing_pillar, 'service': service, 'text': remove_markdown(reco), 'description': '', 'type': 'checklist', 'sourceType': 'wafsg', 'timestamp': timestamp}
+            if source:
+                reco_object['sourceFile'] = source
+            recos.append(reco_object)
         # If line matches a pattern that starts with "|" then comes a text, then "|" and a description and a closing "|"
         if (line[0:1] == '|'):
             line_table_items = line.split('|')
@@ -258,7 +267,7 @@ def get_waf_service_guide_recos():
                         if r.status_code == 200:
                             svcguide = r.text
                             if (args_verbose): print("DEBUG: Parsing service guide '{0}', {1} characters retrieved...".format(file_path, len(svcguide)))
-                            svc_recos = parse_markdown(svcguide, service, verbose=False)
+                            svc_recos = parse_markdown(svcguide, service, source=file_path, verbose=False)
                             if (len(svc_recos) > 0):
                                 retrieved_recos += svc_recos
                                 if args_verbose: print("DEBUG: {0} recommendations found for service '{1}'".format(len(svc_recos), service))
@@ -315,24 +324,33 @@ def get_waf_service_guide_recos():
 # If file exists, try to match the recos in the file by the text field and update the GUIDs
 # If file doesn't exist, generate random GUIDs for each reco
 def update_guids(checklist, filename):
-    # If file exists
+    # If file exists, we can either overwrite it and generate new GUIDs or try to match the recos by text
+    # Note that if matching the recos by GUID, the old recos that do not exactly match the text of the new ones will be lost
     if os.path.isfile(filename):
-        if (args_verbose): print("DEBUG: Retrieving checklist GUIDs from file {0}...".format(filename))
-        existing_checklist = load_json(filename)
-        for reco in checklist['items']:
-            # Find a reco in the existing checklist that matches the text
-            existing_reco = [x for x in existing_checklist['items'] if x['text'] == reco['text']]
-            if len(existing_reco) > 0:
-                # Verify that the existing reco has a GUID
-                if 'guid' in existing_reco[0]:
-                    reco['guid'] = existing_reco[0]['guid']
+        if args_overwrite:
+            if (args_verbose): print("DEBUG: File {0} not found, generating new GUIDs...".format(filename))
+            for reco in checklist['items']:
+                reco['guid'] = str(uuid.uuid4())
+                if 'checklist_match' in reco:
+                    reco['checklist_match_guid'] = str(uuid.uuid4())
+            return checklist
+        else:
+            if (args_verbose): print("DEBUG: Retrieving checklist GUIDs from file {0}...".format(filename))
+            existing_checklist = load_json(filename)
+            for reco in checklist['items']:
+                # Find a reco in the existing checklist that matches the text
+                existing_reco = [x for x in existing_checklist['items'] if x['text'] == reco['text']]
+                if len(existing_reco) > 0:
+                    # Verify that the existing reco has a GUID
+                    if 'guid' in existing_reco[0]:
+                        reco['guid'] = existing_reco[0]['guid']
+                    else:
+                        if (args_verbose): print("DEBUG: reco {0} not found in file {1}, generating new GUID...".format(reco['text'], filename))
+                        reco['guid'] = str(uuid.uuid4())
+                # If no reco was found, generate a new GUID
                 else:
-                    if (args_verbose): print("DEBUG: reco {0} not found in file {1}, generating new GUID...".format(reco['text'], filename))
                     reco['guid'] = str(uuid.uuid4())
-            # If no reco was found, generate a new GUID
-            else:
-                reco['guid'] = str(uuid.uuid4())
-        return checklist
+            return checklist
     # If file doesn't exist, generate GUIDs for each reco
     else:
         if (args_verbose): print("DEBUG: File {0} not found, generating new GUIDs...".format(filename))

.github/actions/get_the_aks_checklist/entrypoint.py

@@ -79,6 +79,8 @@ def get_theaks_recos():
                                     item['waf'] = 'Resiliency'
                                 elif 'operations' in item['category'].lower() or 'management' in item['category'].lower():
                                     item['waf'] = 'Operational Excellence'
+                                item['sourceType'] = 'theakscl'
+                                item['sourceFile'] = file_url
                             retrieved_recos += theaks_recos
                             if verbose: print("DEBUG: {0} recommendations found in file {1}".format(len(theaks_recos), file_path))
                     else:

.github/actions/recov2lint/Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.8-slim-buster
+WORKDIR /app
+COPY requirements.txt requirements.txt
+COPY entrypoint.py entrypoint.py
+RUN pip3 install -r requirements.txt
+ENTRYPOINT ["python3", "/app/entrypoint.py"]

.github/actions/recov2lint/README.md

@@ -0,0 +1,26 @@
+# Retrieve recommendations from Well Architected service guides
+
+This action retrieves the recommendations described in [Well-Architected Service Guides](https://learn.microsoft.com/azure/well-architected/service-guides/?product=popular) and stores it as a new checklist.
+
+## Inputs
+
+## `services`
+
+**Optional** Service(s) whose service guide will be downloaded (leave blank for all service guides). You can specify multiple comma-separated values. Default `""`.
+
+## `output_folder`
+
+**Optional** Folder where the new checklists will be stored. Default `"./checklists-ext"`.
+
+## `verbose`
+
+**Optional** Whether script output is verbose or not. Default `"true"`.
+
+## Example usage
+
+```
+uses: ./.github/actions/get_service_guides
+with:
+  output_file: './checklists'
+  service: 'Azure Kubernetes Service'
+```

.github/actions/recov2lint/action.yml

@@ -0,0 +1,18 @@
+# action.yml
+name: 'Validate PRs for v2 recommendations and checklists.'
+description: 'Verify that no duplicate names exist and that all YAML files conform to the schemas.'
+inputs:
+  folder:
+    description: 'Folder where the recommendations are stored'
+    required: false
+    default: './v2 (string)'
+  verbose:
+    description: 'Verbose output, true/false (string)'
+    required: false
+    default: 'true'
+runs:
+  using: 'docker'
+  image: 'Dockerfile'
+  args:
+    - '${{ inputs.folder }}'
+    - '${{ inputs.verbose }}'

.github/actions/recov2lint/entrypoint.py

@@ -0,0 +1,139 @@
+# This scripts runs checks on the v2 recommendations and checklists
+import jsonschema
+import sys
+import yaml
+import json
+import os
+from pathlib import Path
+from collections import Counter
+
+
+# The script has been modified to be run from a github action with positional parameters
+# 1. Root Folder where the v2 recommendations, checklists and schemas are stored
+# 2. Verbose
+try:
+    root_folder = sys.argv[1]
+except:
+    root_folder = './v2'
+try:
+    verbose = (sys.argv[2].lower() == 'true')
+except:
+    verbose = True
+
+# Print the parameters
+if verbose: print("INFO: Running recov2lint with parameters: root_folder='{0}', verbose={1}".format(root_folder, verbose))
+
+# Constants
+checklist_subfolder = os.path.join(root_folder, 'checklists')
+reco_subfolder = os.path.join(root_folder, 'recos')
+schema_subfolder = os.path.join(root_folder, 'schema')
+reco_schema_file = os.path.join(schema_subfolder, 'recommendation.schema.json')
+checklist_schema_file = os.path.join(schema_subfolder, 'checklist.schema.json')
+
+# Verify that the root folder and the subfolders exist
+if not os.path.exists(root_folder):
+    print(f"ERROR: Root folder '{root_folder}' does not exist.")
+    sys.exit(1)
+if not os.path.exists(checklist_subfolder):
+    print(f"ERROR: Checklist subfolder '{checklist_subfolder}' does not exist.")
+    sys.exit(1)
+if not os.path.exists(reco_subfolder):
+    print(f"ERROR: Reco subfolder '{reco_subfolder}' does not exist.")
+    sys.exit(1)
+if not os.path.exists(schema_subfolder):
+    print(f"ERROR: Schema subfolder '{schema_subfolder}' does not exist.")
+    sys.exit(1)
+if not os.path.exists(reco_schema_file):
+    print(f"ERROR: Reco schema file '{reco_schema_file}' does not exist.")
+    sys.exit(1)
+if not os.path.exists(checklist_schema_file):
+    print(f"ERROR: Checklist schema file '{checklist_schema_file}' does not exist.")
+    sys.exit(1)
+
+# Gets all YAML files in a folder and parses them into a list of objects, adding the filepath for reference
+def get_yml_objects(folder, verbose=False):
+    files = list(Path(folder).rglob( '*.*' ))
+    if verbose: print("DEBUG: Found {0} files in folder {1}".format(len(files), folder))
+    objects = []
+    for file in files:
+        if (file.suffix == '.yaml') or (file.suffix == '.yml'):
+            try:
+                with open(file.resolve()) as f:
+                    object = yaml.safe_load(f)
+            except Exception as e:
+                print("ERROR: Error when loading YAML file {0} - {1}". format(file, str(e)))
+            item = {
+                'filepath': str(file.resolve()),
+                'object': object
+            }
+            objects.append(item)
+    if verbose: print("DEBUG: Loaded {0} objects from folder {1}".format(len(objects), folder))
+    return objects        
+
+# Given a list of objects, compares them with a JSON schema
+def get_invalid_objects(items, schema_file, verbose=False):
+    # Retrieve checklists schema
+    if verbose: print("DEBUG: Loading schema from", schema_file)
+    with open(schema_file, 'r') as stream:
+        try:
+            schema = json.load(stream)
+        except:
+            print("ERROR: Error loading JSON schema from", schema_file)
+            return None
+    # Start validation
+    if verbose: print("DEBUG: Starting validation with schema {0}...".format(schema_file))
+    object_counter = 0
+    finding_counter = 0
+    for item in items:
+        object = item['object']
+        object_counter +=1
+        if 'name' in object:
+            object_name = object['name']
+        else:
+            object_name = 'unnamed'
+        try:
+            jsonschema.validate(object, schema)
+            if verbose: print("DEBUG: Checklist '{0}' in '{1}' validates correctly against the schema.".format(object_name, item['filepath']))
+        except jsonschema.exceptions.ValidationError as e:
+            print("ERROR: Object '{0}' in '{1}' does not validate against the schema.".format(object_name, item['filepath']))
+            print("DEBUG: -", str(e))
+            finding_counter += 1
+        except jsonschema.exceptions.SchemaError as e:
+            print("ERROR: Schema", schema_file, "does not seem to be valid.")
+            if verbose: print("DEBUG: -", str(e))
+            sys.exit(1)
+        except Exception as e:
+            print("ERROR: Unknown error validating checklist '{0}' against the schema {1}: {2}".format(cl['name'], schema_file,str(e)))
+    return finding_counter
+
+
+# Get all recos
+v2recos = get_yml_objects(reco_subfolder)
+# Look for duplicate names
+name_list = [reco['object']['name'] for reco in v2recos if 'name' in reco['object']]
+name_counts = Counter(name_list)
+duplicate_names = [item for item, count in name_counts.items() if count > 1]
+if len(duplicate_names) > 0:
+    print("ERROR: Duplicate reco names found: {0}".format(duplicate_names))
+    sys.exit(1)
+else:
+    print("INFO: No duplicate reco names found in {0} recommendations.".format(len(v2recos)))
+# Validate recos
+reco_errors = get_invalid_objects(v2recos, reco_schema_file, verbose=verbose)
+if reco_errors > 0:
+    print("ERROR: {0} recos did not validate against the schema.".format(reco_errors))
+    sys.exit(1)
+else:
+    print("INFO: {0} recommendations validated from folder {2}, {1} non-compliances found.".format(len(v2recos), reco_errors, reco_subfolder))
+
+# Get all checklists
+v2checklists = get_yml_objects(checklist_subfolder)
+# Validate checklists
+checklist_errors = get_invalid_objects(v2checklists, checklist_schema_file, verbose=verbose)
+if checklist_errors > 0:
+    print("ERROR: {0} checklists did not validate against the schema.".format(checklist_errors))
+    sys.exit(1)
+else:
+    print("INFO: {0} checklists validated from folder {2}, {1} non-compliances found.".format(len(v2checklists), checklist_errors, checklist_subfolder))
+
+

.github/actions/recov2lint/requirements.txt

@@ -0,0 +1,2 @@
+pyyaml
+jsonschema

.github/workflows/autotagv2.yml

@@ -0,0 +1,46 @@
+name: Autotag
+
+env:
+  GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+on:
+  pull_request:
+    branches: [v2]
+    paths:
+      - '**.yml'
+      - '**.yaml'
+
+jobs:
+  autotag:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - id: files
+        uses: masesgroup/retrieve-changed-files@v2
+      - id: alzimpact
+        name: Verify whether the modified files have an impact on the ALZ checklist
+        run: |
+          echo "DEBUG: Running on $SHELL"
+          pip install -r ./scripts/requirements.txt
+          alz_files=$(python3 ./scripts/cl.py list-recos --input-folder ./v2/recos --checklist-file ./v2/checklists/alz.yaml --only-filenames)
+          alz_files_count=$(echo "$alz_files" | wc -l)
+          echo "$alz_files_count reco files found in the ALZ checklist:"
+          echo "$alz_files" | head -2
+          echo "..."
+          echo "$alz_files" | tail -2
+          for input_file in ${{ steps.files.outputs.all }}; do
+            echo "Processing '$input_file'..."
+            if [[ "$alz_files" == *"$input_file"* ]]; then
+              echo "Modification to file '$input_file' detected, which seems to be a reco leveraged by the ALZ checklist"
+              echo "alz_impact=yes" >> $GITHUB_OUTPUT
+            else
+              echo "'$input_file' has no ALZ impact"
+            fi
+          done
+      - name: add ALZ label
+        if: ${{ steps.alzimpact.outputs.alz_impact == 'yes' }}
+        uses: actions-ecosystem/action-add-labels@v1
+        id: addalzlabel
+        with:
+          labels: 'landingzone'
+          github_token: ${{ secrets.WORKFLOW_PAT }}

.github/workflows/linter.yml

@@ -26,7 +26,7 @@ jobs:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
       - name: Check that GUIDs are unique
         id: checklistlint
-        uses: ./.github/actions/get_the_aks_checklist
+        uses: ./.github/actions/review-checklists-lint
         with:
           file_extension: 'en.json'
           key_name: 'guid'

.github/workflows/linterv2.yml

@@ -0,0 +1,21 @@
+---
+name: Lint v2 recommendations and checklists
+on:
+  # push:
+  #  branches-ignore: [main]
+  pull_request:
+    branches: [v2]
+
+jobs:
+  build:
+    name: Lint v2 recommendations and checklists
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout Code
+        uses: actions/checkout@v2
+      - name: Check unique names and schema conformity
+        id: checklistlint
+        uses: ./.github/actions/recov2lint
+        with:
+          folder: './v2'
+          verbose: 'false'