import yaml from decouple import config import dynatraceAPI import pandas as pd from pagination import Pagionation from key_request_parser import krparser from datetime import datetime, timedelta import datetime import re def getSLO(ENV, DTAPIToken, DTENV): # DTENV = base url # DTAPIToken = sec token dtclient = dynatraceAPI.Dynatrace(DTENV, DTAPIToken) my_params_report = {"pageSize": 25} # gets all slos and filter later api_url_report = "/api/v2/slo" pages = dtclient.returnPageination(api_url_report, my_params_report, "slo") df = pd.DataFrame(pages.elements) df["env"] = ENV return df def previous_week_range(date: datetime): start_date = date + timedelta(-date.weekday(), weeks=-1) end_date = date + timedelta(-date.weekday() - 1) return start_date, end_date def main(): # Get All SLOs reportItem = {} with open("./environment.yaml") as file: env_doc = yaml.safe_load(file) for env, doc in env_doc.items(): token = dict(doc[2]) url = dict(doc[1]) if config(token.get("env-token-name")) != "": print("Gather data, hold on a minute") DTTOKEN = config(token.get("env-token-name")) DTURL = url.get("env-url") # krp = krparser.KRParser(krparser.KROption.VALIDATE_EXISTS | krparser.KROption.VALIDATE_HASDATA ,DTURL, DTTOKEN) slosF = getSLO(env, DTTOKEN, DTURL) slosF = slosF[slosF["name"].str.startswith("TP_")] # parse the metric Expression to get Services and Requests krs = [] krp = krparser.KRParser( krparser.KROption.VALIDATE_EXISTS | krparser.KROption.VALIDATE_HASDATA | krparser.KROption.RESOLVESERVICES, DTURL, DTTOKEN, ) for index, row in slosF.iterrows(): krs.append(krp.parseBySLO(row)) # x = 0 # SLO Name | SERVICE | PROCESS GRUPPE | TAGS # {"sloname": { # "sloname":$sloname$, # "services":[{ # "serviceName": "$servicename$" # }] # }, # "sloname": { # "sloname":$sloname$, # "services":[{ # "serviceName": "$servicename$" # }] for kr in krs: reportItem[kr.metadata["sloName"]] = {} reportItem[kr.metadata["sloName"]]["sloname"] = kr.metadata["sloName"] reportItem[kr.metadata["sloName"]]["services"] = [] reportItem[kr.metadata["sloName"]]["services"].append( {"serviceName": "$servicename$"} ) extract_services_and_requests(kr.metadata["metricExpression"]) print(reportItem) # filtering the SLO dataframe based on those dates and the service name def extract_services_and_requests(metric_expression: str): services = [] requests = [] result = metric_expression.split("entityName.in") regex_pattern = r"~(.*?)~" method_list = ["GET", "POST", "http"] for x in result: matches = re.findall(regex_pattern, x) for match in matches: if match != "SERVICE": # TODO: not working properly. fix needed res = any(ele in match for ele in method_list) if res: requests.extend(matches) else: services.extend(matches) print(f"services: {services}") print("-" * 80) print(f"requests: {requests}") return result if __name__ == "__main__": main()