Build and deploy a service to the dev environment

import json import requests # Get environment variables GITHUB_TOKEN = getEnvVar('GITHUB_TOKEN') GITHUB_REPO_OWNER = getEnvVar('GITHUB_REPO_OWNER') # Print input parameters print(f"Input Parameters:\nrepo_name: {repo_name}") # GitHub API URL to list branches url = f"https://api.github.com/repos/{GITHUB_REPO_OWNER}/{repo_name}/branches" # Headers for authentication headers = { 'Authorization': f'Bearer {GITHUB_TOKEN}', 'Accept': 'application/vnd.github.v3+json' } # Make the request to GitHub API response = requests.get(url, headers=headers) # Check if the request was successful if response.status_code == 200: branches = [branch['name'] for branch in response.json()] # print(f"Output:\n{json.dumps(branches, indent=4, default=str)}") else: branches = [] print(f"Failed to fetch branches. Status code: {response.status_code}") branch_ids = [] for branch in branches: if branch.startswith("deployment-branch-"): branch_id = int(branch.split("deployment-branch-")[1]) branch_ids.append(branch_id) latest_branch_id = max(branch_ids) latest_deployment_branch = f"deployment-branch-{latest_branch_id}" print(latest_deployment_branch)

import json import requests source_branch = "main" latest_branch_id = int(latest_deployment_branch.split("deployment-branch-")[1]) new_branch_id = latest_branch_id + 1 new_branch_name = f"deployment-branch-{new_branch_id}" # Get environment variables GITHUB_TOKEN = getEnvVar('GITHUB_TOKEN') GITHUB_REPO_OWNER = getEnvVar('GITHUB_REPO_OWNER') # Print input parameters print(f"Input Parameters:\nrepo_name: {repo_name}\nnew_branch_name: {new_branch_name}\nsource_branch: {source_branch}") # GitHub API URL to get the latest commit of the source branch url = f"https://api.github.com/repos/{GITHUB_REPO_OWNER}/{repo_name}/git/refs/heads/{source_branch}" # Headers for authentication headers = { 'Authorization': f'Bearer {GITHUB_TOKEN}', 'Accept': 'application/vnd.github.v3+json' } # Make the request to GitHub API to get the latest commit SHA of the source branch response = requests.get(url, headers=headers) if response.status_code == 200: source_branch_info = response.json() latest_commit_sha = source_branch_info['object']['sha'] # GitHub API URL to create a new branch url = f"https://api.github.com/repos/{GITHUB_REPO_OWNER}/{repo_name}/git/refs" # Payload to create a new branch payload = { 'ref': f'refs/heads/{new_branch_name}', 'sha': latest_commit_sha } # Make the request to GitHub API to create the new branch response = requests.post(url, headers=headers, json=payload) if response.status_code == 201: new_branch = response.json()['ref'] print(f"Output:\n{json.dumps(new_branch, indent=4, default=str)}") else: new_branch = None print(f"Failed to create new branch. Status code: {response.status_code}") else: new_branch = None print(f"Failed to get the latest commit of the source branch. Status code: {response.status_code}")

import requests import json branch_1 = latest_deployment_branch branch_2 = new_branch_name # Get environment variables GITHUB_TOKEN = getEnvVar('GITHUB_TOKEN') GITHUB_REPO_OWNER = getEnvVar('GITHUB_REPO_OWNER') # Define headers for GitHub API headers = { 'Authorization': f'token {GITHUB_TOKEN}', 'Accept': 'application/vnd.github.v3+json' } # Function to get the creation date of a branch def get_branch_creation_date(repo_owner, repo_name, branch_name): url = f'https://api.github.com/repos/{repo_owner}/{repo_name}/branches/{branch_name}' response = requests.get(url, headers=headers) response.raise_for_status() branch_info = response.json() commit_sha = branch_info['commit']['sha'] url = f'https://api.github.com/repos/{repo_owner}/{repo_name}/commits/{commit_sha}' response = requests.get(url, headers=headers) response.raise_for_status() commit_info = response.json() return commit_info['commit']['committer']['date'] # Get creation dates of the branches branch_1_creation_date = get_branch_creation_date(GITHUB_REPO_OWNER, repo_name, branch_1) branch_2_creation_date = get_branch_creation_date(GITHUB_REPO_OWNER, repo_name, branch_2) print(f'Branch 1 creation date: {branch_1_creation_date}') print(f'Branch 2 creation date: {branch_2_creation_date}') # Function to get commits between two dates def get_commits_between_dates(repo_owner, repo_name, branch_name, start_date, end_date): url = f'https://api.github.com/repos/{repo_owner}/{repo_name}/commits' params = { 'sha': branch_name, 'since': start_date, 'until': end_date } response = requests.get(url, headers=headers, params=params) response.raise_for_status() return response.json() # Get commits between the creation dates of the two branches commits = get_commits_between_dates(GITHUB_REPO_OWNER, repo_name, 'main', branch_1_creation_date, branch_2_creation_date) # Extract required information from commits commit_list = [] for commit in commits: commit_info = { 'commit_id': commit['sha'], 'commit_user': commit['commit']['committer']['name'], 'commit_comments': commit['commit']['message'], 'commit_date': commit['commit']['committer']['date'] } commit_list.append(commit_info) print(json.dumps(commit_list, indent=4, default=str)) # Output the commits commits = commit_list

import json import requests from requests.auth import HTTPBasicAuth ticket_title = f"dev deployment of repo: {repo_name} and branch: {new_branch_name}" # Get environment variables JIRA_USER_NAME = getEnvVar('JIRA_USER_NAME') JIRA_API_KEY = getEnvVar('JIRA_API_KEY') JIRA_BASE_URL = getEnvVar('JIRA_BASE_URL') # Print input parameters print(f"Input Parameters:\nproject_key: {project_key}\nticket_title: {ticket_title}\ncommit_list: {json.dumps(commit_list, indent=4, default=str)}") # Jira API URL to create a new issue url = f"{JIRA_BASE_URL}/rest/api/3/issue" # Headers for authentication headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } # Format the commit log in Atlassian Document Format commit_log_adf = [{ "type": "paragraph", "content": [ { "type": "text", "text": json.dumps(commit, indent=4) } ] } for commit in commit_list] # Payload to create a new issue payload = { "fields": { "project": { "key": project_key }, "summary": ticket_title, "description": { "type": "doc", "version": 1, "content": commit_log_adf }, "issuetype": { "name": "Task" } } } # Make the request to Jira API to create the new issue response = requests.post(url, headers=headers, auth=HTTPBasicAuth(JIRA_USER_NAME, JIRA_API_KEY), json=payload) if response.status_code == 201: issue_key = response.json()['key'] issue_url = f"{JIRA_BASE_URL}/browse/{issue_key}" print(f"Output:\n{json.dumps({'issue_url': issue_url}, indent=4, default=str)}") else: issue_url = None print(f"Failed to create Jira ticket. Status code: {response.status_code}") # Output the issue URL issue_url

import json import requests from requests.auth import HTTPBasicAuth branch_name = new_branch_name # Get environment variables JENKINS_USERNAME = getEnvVar('JENKINS_USERNAME') JENKINS_API_TOKEN = getEnvVar('JENKINS_API_TOKEN') JENKINS_URL = getEnvVar('JENKINS_URL') # Print input parameters print(f"Input Parameters:\njob_name: {job_name}\nbranch_name: {branch_name}") # Jenkins API URL to trigger the job with parameters url = f"{JENKINS_URL}/job/{job_name}/buildWithParameters" # Headers for authentication headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } # Parameters to trigger the job params = { 'Branch': branch_name } # Make the request to Jenkins API to trigger the job response = requests.post(url, headers=headers, auth=HTTPBasicAuth(JENKINS_USERNAME, JENKINS_API_TOKEN), params=params) if response.status_code == 201: print(f"Output:\nJob triggered successfully.") else: print(f"Failed to trigger job. Status code: {response.status_code}")

import requests import json num_results = 10 # Fetch environment variables jenkins_username = getEnvVar('JENKINS_USERNAME') jenkins_api_token = getEnvVar('JENKINS_API_TOKEN') jenkins_url = getEnvVar('JENKINS_URL') # Print input parameters print(f"Job Name: {job_name}") print(f"Number of Results: {num_results}") # Jenkins API URL for job builds api_url = f"{jenkins_url}/job/{job_name}/api/json?tree=builds[number,duration,result]&depth=1&pretty=true" # Make the request to Jenkins API response = requests.get(api_url, auth=(jenkins_username, jenkins_api_token)) # Check if the request was successful if response.status_code != 200: raise Exception(f"Failed to fetch job details: {response.status_code} {response.text}") # Parse the JSON response job_data = response.json() # Extract the last 'num_results' builds builds = job_data.get('builds', [])[:num_results] # Calculate the average duration of successful builds total_duration = 0 successful_builds = 0 for build in builds: if build.get('result') == 'SUCCESS': total_duration += build.get('duration', 0) successful_builds += 1 if successful_builds == 0: average_execution_length = 0 else: average_execution_length = total_duration / successful_builds average_execution_length = int(average_execution_length/1000) # Print the output print(f"Average Execution Length: {average_execution_length}")

import time # Print input parameters print(f"Input Parameters:\n average_execution_length: {average_execution_length} seconds") wait_time_seconds = average_execution_length * 2 print(f"Waiting for {wait_time_seconds} seconds...") time.sleep(wait_time_seconds) # Print output print("Output:\nWait completed.")

import json import requests from requests.auth import HTTPBasicAuth # Get environment variables JENKINS_USERNAME = getEnvVar('JENKINS_USERNAME') JENKINS_API_TOKEN = getEnvVar('JENKINS_API_TOKEN') JENKINS_URL = getEnvVar('JENKINS_URL') # Print input parameters print(f"Input Parameters:\njob_name: {job_name}") # Jenkins API URL to get the last build number url_last_build = f"{JENKINS_URL}/job/{job_name}/lastBuild/api/json" # Headers for authentication headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } # Make the request to Jenkins API to get the last build number response_last_build = requests.get(url_last_build, headers=headers, auth=HTTPBasicAuth(JENKINS_USERNAME, JENKINS_API_TOKEN)) if response_last_build.status_code == 200: last_build_info = response_last_build.json() last_build_number = last_build_info['number'] # Jenkins API URL to get the console log url_console_log = f"{JENKINS_URL}/job/{job_name}/{last_build_number}/logText/progressiveText?start=0" # Make the request to Jenkins API to get the console log response_console_log = requests.get(url_console_log, headers=headers, auth=HTTPBasicAuth(JENKINS_USERNAME, JENKINS_API_TOKEN)) if response_console_log.status_code == 200: console_log = response_console_log.text last_50_lines = '\n'.join(console_log.split('\n')[-50:]) print(f"Output:\n{last_50_lines}") else: print(f"Failed to fetch console log. Status code: {response_console_log.status_code}") else: print(f"Failed to fetch last build number. Status code: {response_last_build.status_code}")

import json import requests from requests.auth import HTTPBasicAuth # Get environment variables JIRA_USER_NAME = getEnvVar('JIRA_USER_NAME') JIRA_API_KEY = getEnvVar('JIRA_API_KEY') JIRA_BASE_URL = getEnvVar('JIRA_BASE_URL') # Print input parameters print(f"Input Parameters:\nissue_url: {issue_url}\nlast_50_lines: {last_50_lines}") # Extract issue key from issue URL issue_key = issue_url.split('/')[-1] # Jira API URL to add a comment to the issue url = f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}/comment" # Headers for authentication headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } # Format the console log in Atlassian Document Format console_log_adf = [{ "type": "paragraph", "content": [ { "type": "text", "text": line } ] } for line in last_50_lines.split('\n')] # Payload to add a comment payload = { "body": { "type": "doc", "version": 1, "content": console_log_adf } } # Make the request to Jira API to add the comment response = requests.post(url, headers=headers, auth=HTTPBasicAuth(JIRA_USER_NAME, JIRA_API_KEY), json=payload) if response.status_code == 201: comment_id = response.json()['id'] print(f"Output:\nComment added successfully with ID: {comment_id}") else: print(f"Failed to add comment. Status code: {response.status_code}")

import json # Print input parameters print(f"Input Parameters:\ncluster_name: {cluster_name}") # Command to update kubeconfig command = f'aws eks update-kubeconfig --name {cluster_name} --region us-east-2' # Execute the command locally host_ip_addr = None update_kubeconfig_output = _exe(host_ip_addr, command) # Print raw output print(f"Raw Output:\n{update_kubeconfig_output}") # Process the output if needed # For now, just print the output as is print(f"Output:\n{update_kubeconfig_output}")

import json # Print input parameters print(f"Input Parameters:\nnamespace: {namespace}") # Command to get pod status command = f'kubectl get pods -n {namespace}' # Execute the command locally host_ip_addr = None pod_status = _exe(host_ip_addr, command) # Print raw output print(f"Raw Output:\n{pod_status}") # Process the output if needed # For now, just print the output as is print(f"Output:\n{pod_status}")

import json import requests from requests.auth import HTTPBasicAuth # Get environment variables JIRA_USER_NAME = getEnvVar('JIRA_USER_NAME') JIRA_API_KEY = getEnvVar('JIRA_API_KEY') JIRA_BASE_URL = getEnvVar('JIRA_BASE_URL') # Print input parameters print(f"Input Parameters:\nissue_url: {issue_url}\npod_status: {pod_status}") # Extract issue key from issue URL issue_key = issue_url.split('/')[-1] # Jira API URL to add a comment to the issue url = f"{JIRA_BASE_URL}/rest/api/3/issue/{issue_key}/comment" # Headers for authentication headers = { 'Content-Type': 'application/json', 'Accept': 'application/json' } # Format the pod status as a code block in Atlassian Document Format pod_status_adf = [{ "type": "codeBlock", "content": [ { "type": "text", "text": pod_status } ] }] # Payload to add a comment payload = { "body": { "type": "doc", "version": 1, "content": pod_status_adf } } # Make the request to Jira API to add the comment response = requests.post(url, headers=headers, auth=HTTPBasicAuth(JIRA_USER_NAME, JIRA_API_KEY), json=payload) if response.status_code == 201: comment_id = response.json()['id'] print(f"Output:\nComment added successfully with ID: {comment_id}") else: print(f"Failed to add comment. Status code: {response.status_code}")

import json from slack_sdk import WebClient from slack_sdk.errors import SlackApiError # Get environment variables SLACK_API_TOKEN = getEnvVar('SLACK_API_TOKEN') # Print input parameters print(f"Input Parameters:\nrepo_name: {repo_name}\nnew_branch_name: {new_branch_name}\nissue_url: {issue_url}") # Initialize Slack client client = WebClient(token=SLACK_API_TOKEN) # Message to be posted message = f"""A new branch `{new_branch_name}` has been created in the repository `{repo_name}`. It has been built and deployed to dev environment. You can track it here: {issue_url}""" try: # Post message to Slack channel response = client.chat_postMessage( channel='#devops', text=message ) print(f"Output:\nMessage posted successfully to Slack channel devops.") except SlackApiError as e: print(f"Failed to post message to Slack channel. Error: {e.response['error']}")