import gzip from ..config import config from ..utils import o11y, utils import csv import time from ..utils.objstorage.s3 import S3,get_files from ..utils.objstorage.gcs import GCS,get_files as gcs_get_files import datetime class SlowLog: """ SlowLog class for the SDK. """ def __init__(self, cfg: config.Config): self.api = o11y.O11yAPI(cfg) self.cfg = cfg def getSlowLog(self, clusterId: str, isServerless: bool, startTimestemp: int, endTimestemp: int): """ Get slow log data from o11y. """ # we need to get the cluster info first cluster = self.api.getO11yClusterInfo(clusterId,isServerless) if isServerless: return self.getSlowLogObjStorage(cluster, startTimestemp, endTimestemp) else: return self.getDedicatedSlowLog(cluster, startTimestemp, endTimestemp) def getDedicatedSlowLog(self, cluster: o11y.ClusterInfo, startTimestemp: int, endTimestemp: int): bucketName = utils.getO11yBucketName(self.cfg.env, cluster.region) if cluster.vendor =='gcp': s3 = GCS(bucketName,"gcp-autonomous-prod") else: s3 = S3(bucketName) need_download = [] commList=self.getCommomPrefix(startTimestemp, endTimestemp) for commperfex in commList: prefix = utils.getO11yObjPrefix(cluster.tenantID, cluster.projectID, cluster.applicationID,commperfex) #get obj list objList = s3.get_file_list(prefix) for obj in objList: timestemp = self.getTimestemp(obj) if timestemp >= startTimestemp and timestemp <= endTimestemp: need_download.append(obj) print(need_download) #download obj logdata = '' if cluster.vendor =='gcp': downloaded=gcs_get_files(5,"gcp-autonomous-prod",bucketName,need_download) for content in downloaded: logdata += content logdata += '\n' else: downloaded=get_files(5,bucketName,need_download) for content in downloaded: content = gzip.decompress(content).decode("utf-8") logdata += content logdata += '\n' return logdata def getSlowLogObjStorage(self, cluster: o11y.ClusterInfo, startTimestemp: int, endTimestemp: int): bucketName = utils.getO11yBucketName(self.cfg.env, cluster.region) if cluster.vendor =='gcp': s3 = GCS(bucketName,"gcp-autonomous-prod") else: s3 = S3(bucketName) #build prefix need_download = [] commList=self.getCommomPrefix(startTimestemp, endTimestemp) for commperfex in commList: prefix = utils.getO11yServerlessObjPerfix(cluster.tenantID, cluster.projectID, cluster.applicationID,commperfex) print(prefix) #get obj list objList = s3.get_file_list(prefix) for obj in objList: timestemp = self.getTimestemp(obj) if timestemp >= startTimestemp and timestemp <= endTimestemp: need_download.append(obj) print(need_download) #download obj logdata = '' if cluster.vendor =='gcp': downloaded=gcs_get_files(5,"gcp-autonomous-prod",bucketName,need_download) for content in downloaded: logdata += content logdata += '\n' else: downloaded=get_files(5,bucketName,need_download) for content in downloaded: content = gzip.decompress(content).decode("utf-8") logdata += content logdata += '\n' return logdata def getCommomPrefix(self,startTimestemp: int, endTimestemp: int): start = str(startTimestemp) end = str(endTimestemp) tmp =[] for (fist, second) in zip(start, end): if fist != second: break tmp.append(fist) if len(tmp) == 1: # if comm prefix length less than 2, we need to splite the time range firstend=(startTimestemp//100000000+1)*100000000-1 # get the first range end secondend=firstend+1 # get the second range start return self.getCommomPrefix(startTimestemp, firstend)+self.getCommomPrefix(secondend, endTimestemp) return [''.join(tmp)] def getTimestemp(self, objName: str): #example 0/prod-devtier/serverless/1372813089187041280/1372813089206301327/1379661944642684098/logs/slowlog/1676466135-a054c164-5d13-4604-a3cb-ad612a102d33.log.gz #get the last part tmp = objName.split('/')[-1] #get the first part tmp = tmp.split('-')[0] return int(tmp) def parseContent(self, content:str): import re lines = content.split('\n') time_regex = re.compile(r'^# Time: (\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{9}Z)') result = [] current_record = {} for line in lines: time_match = time_regex.match(line) if time_match: if current_record: current_record['value']= '\n'.join(current_record['value']) result.append(current_record) current_record = {} # 将时间字符串截断为6位微秒，然后转换为 datetime 对象 time_str = time_match.group(1) truncated_time_str = time_str[:-4] + 'Z' time_format = '%Y-%m-%dT%H:%M:%S.%fZ' dt = datetime.datetime.strptime(truncated_time_str, time_format) current_record['timestamp'] = int(dt.timestamp()) if 'value' not in current_record: current_record['value'] = [] current_record['value'].append(line) else: current_record['value'].append(line) # 添加最后一条记录 if current_record: result.append(current_record) return result def getSlowLogLoki(self, cluster: o11y.ClusterInfo, startTimestemp: int, endTimestemp: int): # get ds path dsPath = utils.getO11yLokiPath(cluster.vendor, cluster.region) # get slow log data # get metrics data part by part logData = dict() # build slow log rules rules = config.RulesConfig([config.Rule("slowlog", '{{cluster_env="{}", namespace="tidb{}", container="slowlog"}} |= ``'.format( self.cfg.env.name, cluster.applicationID))]) # query metrics data every 5 min startTimeRange = startTimestemp endTimeRange = startTimestemp while endTimeRange < endTimestemp: endTimeRange = endTimeRange+5*60 if endTimeRange + \ 5*60 < endTimestemp else endTimestemp print('get slow log data from {} to {}\n'.format( startTimeRange, endTimeRange)) for rule in rules.rules: partLogData = self.api.getO11yLoki( dsPath, rule.expr, startTimeRange, endTimeRange) for log in partLogData: if logData.get(log["stream"]["instance"]) is None: logData[log["stream"]["instance"]] = [] for value in log['values']: logData[log["stream"]["instance"]].append({ # convert nano to second "timestemp": int(value[0])//1000000000, "value": value[1] }) startTimeRange = endTimeRange # sleep 0.5s~1.5s to avoid too many requests time.sleep(0.5+int(time.time()) % 1) return logData def writeToCSV(self, logData, filePath: str): """ Write slow log data to csv file. """ with open(filePath, 'w') as f: csvwriter = csv.writer(f, delimiter=',',) # write header csvwriter.writerow(["instance", "timestemp", "value"]) # write data for instance in logData: for value in logData[instance]: csvwriter.writerow( [instance, value["timestemp"], value["value"]]) def writeToLog(self, logData, filePath: str): """ Write slow log data to log file. """ with open(filePath, 'w') as f: # write data f.write(logData)