import json import math import datetime import os from .util.constant import STATUS_SUCCESS, STATUS_FAIL from .util.constant import USER_ADDR from .util.constant import INGESTION_API, META_API, TSG_API, STORAGE_GW_API from .util.constant import INSTANCE_ID_KEY from .util.constant import INSTANCE_ID_PLACEHOLDER from .util.constant import META_ENDPOINT, TSG_ENDPOINT, INGESTION_ENDPOINT, STORAGE_GW_MT_ENDPOINT_PATTERN, STORAGE_GW_ST_ENDPOINT_PATTERN from .util.constant import IS_INTERNAL, IS_MT from .util.retryrequests import RetryRequests from .util.series import Series from .util.timeutil import get_time_offset, str_to_dt, dt_to_str, get_time_list from telemetry import log import pandas as pd REQUEST_TIMEOUT_SECONDS = 120 def get_field_idx(fields, target): for idx, field in enumerate(fields): if field == '__FIELD__.' + target: return idx raise Exception("Not found field {} in {}".format(target, ','.join(fields))) class TSANAClient(object): def __init__(self, username=None, password=None, retrycount=3, retryinterval=1000): self.username = username self.password = password self.retryrequests = RetryRequests(retrycount, retryinterval) self.crt = os.environ['MA_CERT_CRT_PATH'] if 'MA_CERT_CRT_PATH' in os.environ else '' self.key = os.environ['MA_CERT_KEY_PATH'] if 'MA_CERT_KEY_PATH' in os.environ else '' def post(self, api_endpoint, instance_id, api_key, user, path, data): if not api_endpoint.startswith('http'): api_endpoint = "https://" + api_endpoint url = api_endpoint.rstrip('/') + path headers = { "x-api-key": api_key, "x-user": user, "X-Base-Address": api_endpoint, "Instance-Id": instance_id, "X-Action-Source": "Plugin", "Content-Type": "application/json" } if self.username and self.password: auth = (self.username, self.password) else: auth = None if self.crt and self.key: cert = (self.crt, self.key) else: cert = None try: r = self.retryrequests.post(url=url, headers=headers, auth=auth, data=json.dumps(data), timeout=REQUEST_TIMEOUT_SECONDS, cert=cert, verify=False) if r.status_code != 204: try: return r.json() except ValueError as e: return "ValueError: " + str(e) + " Content: " + r.content.decode('UTF-8') except Exception as e: raise Exception('TSANA service api "{}" failed, request:{}, {}'.format(url, json.dumps(data), str(e))) def put(self, api_endpoint, instance_id, api_key, user, path, data): if not api_endpoint.startswith('http'): api_endpoint = "https://" + api_endpoint url = api_endpoint.rstrip('/') + path headers = { "x-api-key": api_key, "x-user": user, "X-Base-Address": api_endpoint, "Instance-Id": instance_id, "X-Action-Source": "Plugin", "Content-Type": "application/json" } if self.username and self.password: auth = (self.username, self.password) else: auth = None if self.crt and self.key: cert = (self.crt, self.key) else: cert = None try: r = self.retryrequests.put(url=url, headers=headers, auth=auth, data=json.dumps(data), timeout=REQUEST_TIMEOUT_SECONDS, cert=cert, verify=False) if r.status_code != 204: try: return r.json() except ValueError: return r.content except Exception as e: raise Exception('TSANA service api "{}" failed, request:{}, {}'.format(url, json.dumps(data), str(e))) def get(self, api_endpoint, instance_id, api_key, user, path): if not api_endpoint.startswith('http'): api_endpoint = "https://" + api_endpoint url = api_endpoint.rstrip('/') + path headers = { "x-api-key": api_key, "x-user": user, "X-Base-Address": api_endpoint, "Instance-Id": instance_id, "X-Action-Source": "Plugin", "Content-Type": "application/json" } if self.username and self.password: auth = (self.username, self.password) else: auth = None if self.crt and self.key: cert = (self.crt, self.key) else: cert = None try: r = self.retryrequests.get(url=url, headers=headers, auth=auth, timeout=REQUEST_TIMEOUT_SECONDS, cert=cert, verify=False) try: return r.json() except ValueError: return r.content except Exception as e: raise Exception('TSANA service api "{}" failed, {}'.format(url, str(e))) ################ GATEWAY API ################ # Query time series from TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # series_sets: Array of series set # start_time: inclusive, the first timestamp to be query # end_time: exclusive # top: top sereis number per series set # Return: # An array of Series object # Series include # series_id: UUID # dim: dimension dict for this series # fields: 1-d string array, ['time', '__VAL__', '__FIELD__.ExpectedValue', '__FIELD__.IsAnomaly', '__FIELD__.PredictionValue', '__FIELD__.PredictionModelScore', '__FIELD__.IsSuppress', '__FIELD__.Period', '__FIELD__.CostPoint', '__FIELD__.Mean', '__FIELD__.STD', '__FIELD__.TrendChangeAnnotate', '__FIELD__.TrendChang...tateIgnore', '__FIELD__.AnomalyAnnotate', ...] # value: 2-d array, [['2020-10-12T17:55:00Z', 1.0, None, None, None, None, None, None, None, None, None, None, None, None, ...]] def get_timeseries_gw(self, parameters, series_sets, start_time, end_time, top=20): if start_time > end_time: raise Exception('start_time should be less than or equal to end_time') instance_id = parameters[INSTANCE_ID_KEY] if IS_INTERNAL: if IS_MT: storage_gw_endpoint = STORAGE_GW_MT_ENDPOINT_PATTERN storage_gw_endpoint = storage_gw_endpoint.replace(INSTANCE_ID_PLACEHOLDER, instance_id) else: storage_gw_endpoint = STORAGE_GW_ST_ENDPOINT_PATTERN meta_endpoint = META_ENDPOINT else: storage_gw_endpoint = parameters['apiEndpointV2'] + STORAGE_GW_API meta_endpoint = parameters['apiEndpointV2'] + META_API end_str = dt_to_str(end_time) start_str = dt_to_str(start_time) multi_series_data = [] total_point_num = 0 loop = 0 # Query each series's tag for data in series_sets: dim = {} if 'dimensionFilter' not in data: data['dimensionFilter'] = data['filters'] for dimkey in data['dimensionFilter']: dim[dimkey] = [data['dimensionFilter'][dimkey]] skip = 0 count = 0 para = dict(metricId=data['metricId'], dimensionFilter=dim, activeSince=start_str) gran_info = (data['metricMeta']['granularityName'], data['metricMeta']['granularityAmount']) data_point_num_per_series = len(get_time_list(start_time, end_time, gran_info)) series_limit_per_call = min(max(100000 // data_point_num_per_series, 1), 1000) while True: # Max data points per call is 100000 ret = self.post(meta_endpoint, instance_id, parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/metrics/' + data['metricId'] + '/series/query?$skip={}&$maxpagesize={}'.format(skip, series_limit_per_call), data=para) if len(ret['value']) == 0: break series_list = [] for s in ret['value']: series = {} series['metricsName'] = s['metricId'] series['begin'] = start_str series['end'] = end_str series['tagSet'] = s['dimension'] series['returnSeriesId'] = True series_list.append(series) if len(series_list) > 0: ret_data = self.post(storage_gw_endpoint, instance_id, parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/api/query_series', data=series_list) sub_multi_series_data = [] for factor in ret_data: if len(factor['values']) <= 0: continue sub_multi_series_data.append(Series(factor['name'], factor['seriesId'], factor['tags'], factor['columns'], factor['values'])) total_point_num += len(factor['values']) log.count("get_data_series_num", 1, endpoint=parameters['apiEndpoint'], group_id=parameters['groupId'], group_name=parameters['groupName'].replace(' ', '_'), instance_id=parameters['instance']['instanceId'], instance_name=parameters['instance']['instanceName'].replace(' ', '_')) log.count("get_data_point_num", len(factor['values']), endpoint=parameters['apiEndpoint'], group_id=parameters['groupId'], group_name=parameters['groupName'].replace(' ', '_'), instance_id=parameters['instance']['instanceId'], instance_name=parameters['instance']['instanceName'].replace(' ', '_')) multi_series_data.extend(sub_multi_series_data) count += len(sub_multi_series_data) if count >= top: log.warning(f"Current series num {count} for series set {json.dumps(data)} has reached limit {top}!") break skip = skip + len(series_list) loop = loop + 1 if loop % 10 == 0: log.info(f"Loop times: {loop}, total series num: {len(multi_series_data)}, total points num {total_point_num}.") # Max data points limit is 4000000, about 400Mb if total_point_num >= 4000000: log.info("Reach total point number limit 4000000.") break if not len(multi_series_data): raise Exception("Series is empty") log.info(f"Get time series completed! Total series num: {len(multi_series_data)}, total points num {total_point_num}.") return multi_series_data ################ META API ################ # To get the meta of a specific datafeed from TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # datafeed_id: an UUID string # Return: # the meta of the specified datafeed, or None if there is something wrong. def get_datafeed_meta(self, parameters, datafeed_id): return self.get(META_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + META_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/datafeeds/' + datafeed_id) # To get the meta of a specific metric from TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # metric_id: an UUID string # Return: # the meta of the specified metric, or None if there is something wrong. def get_metric_meta(self, parameters, metric_id): return self.get(META_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + META_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/metrics/' + metric_id + '/meta') # To get the dimension values of a specific dimension of a metric from TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # metric_id: an UUID string # dimension_name: dimension name for specific metric with metric_id # Return: # the dimension values of a specific dimension of a metric, or None if there is something wrong. def get_dimesion_values(self, parameters, metric_id, dimension_name): dims = self.get(META_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + META_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/metrics/' + metric_id + '/dimensions') if 'dimensions' in dims and dimension_name in dims['dimensions']: return dims['dimensions'][dimension_name] else: return None # Query time series from TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # series_sets: Array of series set # start_time: inclusive, the first timestamp to be query # end_time: exclusive # top: top sereis number per series set # Return: # An array of Series object def get_timeseries(self, parameters, series_sets, start_time, end_time, top=20): if start_time > end_time: raise Exception('start_time should be less than or equal to end_time') end_str = dt_to_str(end_time) start_str = dt_to_str(start_time) multi_series_data = [] total_point_num = 0 loop = 0 # Query each series's tag for data in series_sets: dim = {} if 'dimensionFilter' not in data: data['dimensionFilter'] = data['filters'] for dimkey in data['dimensionFilter']: dim[dimkey] = [data['dimensionFilter'][dimkey]] skip = 0 count = 0 para = dict(metricId=data['metricId'], dimensionFilter=dim, activeSince=start_str) gran_info = (data['metricMeta']['granularityName'], data['metricMeta']['granularityAmount']) data_point_num_per_series = len(get_time_list(start_time, end_time, gran_info)) series_limit_per_call = min(max(100000 // data_point_num_per_series, 1), 1000) while True: # Max data points per call is 100000 ret = self.post(META_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + META_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/metrics/' + data['metricId'] + '/series/query?$skip={}&$maxpagesize={}'.format(skip, series_limit_per_call), data=para) if len(ret['value']) == 0: break series_list = [] for s in ret['value']: s['startTime'] = start_str s['endTime'] = end_str s['returnSeriesId'] = True series_list.append(s) if len(series_list) > 0: ret_data = self.post(META_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + META_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/metrics/series/data', data=dict(value=series_list)) sub_multi_series_data = [] for factor in ret_data['value']: if len(factor['values']) <= 0: continue sub_multi_series_data.append(Series(factor['id']['metricId'], factor['id']['seriesId'], factor['id']['dimension'], factor['fields'], factor['values'])) total_point_num += len(factor['values']) multi_series_data.extend(sub_multi_series_data) count += len(sub_multi_series_data) if count >= top: log.warning(f"Current series num {count} for series set {json.dumps(data)} has reached limit {top}!") break skip = skip + len(series_list) loop = loop + 1 if loop % 10 == 0: log.info(f"Loop times: {loop}, total series num: {len(multi_series_data)}, total points num {total_point_num}.") # Max data points limit is 4000000, about 400Mb if total_point_num >= 4000000: log.info("Reach total point number limit 4000000.") break if not len(multi_series_data): raise Exception("Series is empty") log.info(f"Get time series completed! Total series num: {len(multi_series_data)}, total points num {total_point_num}.") return multi_series_data # Get ranked dimensions # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # metric_id: uuid for metric # dimensions: included dimensions # start_time: inclusive, the first timestamp to be query # top: max count for returned results # skip: offset # Return: # ranked series dimensions def rank_series(self, parameters, metric_id, dimensions, start_time, top=10, skip=0): url = f'/metrics/{metric_id}/rank-series' para = dict(dimensions=dimensions, count=top, startTime=start_time, skip=skip) return self.post(META_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + META_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, url, data=para) ################ INGESTION API ################ # Query data from original data source for datafeed which does not need ingestion # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # datafeed_id: datafeed id for source data # start_time: inclusive, the first timestamp to be query # end_time: exclusive # Return: # LoadSchemaResponse: # SeriesNumber: long number of rows # Schemas:columns, dict which includes # PreviewData: list> which includes # Err: error code def get_source_data(self, parameters, datafeed_id, start_time, end_time): if start_time > end_time: raise Exception('start_time should be less than or equal to end_time') body = { "datafeedId": datafeed_id, "startTime": dt_to_str(start_time), "endTime": dt_to_str(end_time) } ret_data = self.post(INGESTION_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + INGESTION_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/loadSourceData', body) return ret_data # Save a inference result back to TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # metric_id: an UUID string # dimensions: a dict includes dimension name and value # timestamps: an array of timestamps # values: an array of inference result values # fields: an array of field names # field_values: an 2-d array of field values corresponding with fields # push_data_type: 'DatabaseOnly'/'AnomalyDetection', default is 'AnomalyDetection' # Return: # result: STATE_SUCCESS / STATE_FAIL # message: description for the result def save_data_points(self, parameters, metric_id, dimensions, timestamps, values, fields=None, field_values=None, push_data_type='AnomalyDetection'): try: if len(values) <= 0: raise Exception('empty values') body = { "metricId": metric_id, "dimensions": dimensions, "timestamps": timestamps, "values": values, "pushDataType": push_data_type } if fields and len(fields) > 0 and field_values and len(field_values) > 0: body['fields'] = fields body['fieldValues'] = field_values self.post(INGESTION_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + INGESTION_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/pushData', body) return STATUS_SUCCESS, '' except Exception as e: return STATUS_FAIL, str(e) ################ TSG API ################ # To get the detailed info of a specific group from TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # Return: # detailed info of the specified group def get_group_detail(self, parameters): return self.get(TSG_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + TSG_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/timeSeriesGroups/' + parameters['groupId']) # Save a training result back to TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # instance: instance object, which is copied from the inference request, or from the entity # model_id: model id # model_state: model state(TRAINING,READY,FAILED,DELETED) # message: detail message # Return: # result: STATE_SUCCESS / STATE_FAIL # message: description for the result def save_training_result(self, parameters, model_id, model_state:str, message:str): try: body = { 'modelId': model_id, 'state': model_state, 'message': message } self.put(TSG_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + TSG_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/timeSeriesGroups/' + parameters['groupId'] + '/appInstances/' + parameters['instance']['instanceId'] + '/modelKey', body) return STATUS_SUCCESS, '' except Exception as e: return STATUS_FAIL, str(e) # Save a inference result back to TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # instance: instance object, which is copied from the inference request, or from the entity # result: an array of inference result. # Return: # result: STATE_SUCCESS / STATE_FAIL # message: description for the result def save_inference_result(self, parameters, result, batch_size=1000): try: if len(result) <= 0: return STATUS_SUCCESS, '' for batch_index in range(math.ceil(len(result) / batch_size)): body = { 'groupId': parameters['groupId'], 'instanceId': parameters['instance']['instanceId'], 'results': [] } batch_start = batch_index * batch_size for step in range(min(batch_size, len(result) - batch_start)): item = result[batch_start + step] item['timestamp'] = dt_to_str(str_to_dt(item['timestamp'])) body['results'].append({ 'params': parameters['instance']['params'], 'timestamp': item['timestamp'], 'result': item['value'], 'status': item['status'] }) self.post(TSG_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + TSG_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/timeSeriesGroups/' + parameters['groupId'] + '/appInstances/' + parameters['instance']['instanceId'] + '/saveResult', body) return STATUS_SUCCESS, '' except Exception as e: return STATUS_FAIL, str(e) # Save training task status back to TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # instance: instance object, which is copied from the inference request, or from the entity # status: ModelState:Pending, Training, Ready, Deleted, Failed # last_error: last error message # Return: # result: STATE_SUCCESS / STATE_FAIL # message: description for the result def save_training_status(self, task_id, parameters, status, last_error=None): try: context = {} context['groupId'] = parameters['groupId'] context['groupName'] = parameters['groupName'] context['instanceId'] = parameters['instance']['instanceId'] context['instanceName'] = parameters['instance']['instanceName'] context['startTime'] = parameters['startTime'] context['endTime'] = parameters['endTime'] body = { 'taskId': task_id, 'operation': 'Train', 'context': context, 'status': status, 'lastError': str(last_error)[:1000] if last_error is not None else '' } self.post(TSG_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + TSG_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/timeSeriesGroups/' + parameters['groupId'] + '/appInstances/' + parameters['instance']['instanceId'] + '/ops', body) return STATUS_SUCCESS, '' except Exception as e: log.warning(f"Save training status failed. taskId: {task_id}, error: {str(e)}") return STATUS_FAIL, str(e) # Save inference task status back to TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # instance: instance object, which is copied from the inference request, or from the entity # status: InferenceState:Pending, Running, Ready, Failed # last_error: last error message # Return: # result: STATE_SUCCESS / STATE_FAIL # message: description for the result def save_inference_status(self, task_id, parameters, status, last_error=None): try: if not parameters['manually']: return STATUS_SUCCESS, '' context = {} context['groupId'] = parameters['groupId'] context['groupName'] = parameters['groupName'] context['instanceId'] = parameters['instance']['instanceId'] context['instanceName'] = parameters['instance']['instanceName'] context['startTime'] = parameters['startTime'] context['endTime'] = parameters['endTime'] body = { 'taskId': task_id, 'operation': 'Inference', 'context': context, 'status': status, 'lastError': str(last_error)[:1000] if last_error is not None else '' } self.post(TSG_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + TSG_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/timeSeriesGroups/' + parameters['groupId'] + '/appInstances/' + parameters['instance']['instanceId'] + '/ops', body) return STATUS_SUCCESS, '' except Exception as e: log.warning(f"Save inference status failed. taskId: {task_id}, error: {str(e)}") return STATUS_FAIL, str(e) # Get inference result from TSANA # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # instance: instance object, which is copied from the inference request, or from the entity # Return: # result: STATE_SUCCESS / STATE_FAIL # message: description for the result # value: inference result array def get_inference_result(self, parameters): try: ret = self.get(TSG_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + TSG_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, '/timeSeriesGroups/' + parameters['groupId'] + '/appInstances/' + parameters['instance']['instanceId'] + '/history?startTime=' + parameters['startTime'] + '&endTime=' + parameters['endTime']) return STATUS_SUCCESS, '', ret except Exception as e: return STATUS_FAIL, str(e), None # Push alert in general # Parameters: # parameters: a dict object which should includes # apiEndpoint: api endpoint for specific user # apiKey: api key for specific user # groupId: groupId in TSANA, which is copied from inference request, or from the entity # alert_type: alert type # message: alert message # Return: # result: STATE_SUCCESS / STATE_FAIL # message: description for the result def push_alert(self, parameters, alert_type, message): try: url = '/timeSeriesGroups/alert' para = dict(alertType=alert_type, message=message) self.post(TSG_ENDPOINT if IS_INTERNAL else parameters['apiEndpointV2'] + TSG_API, parameters[INSTANCE_ID_KEY], parameters['apiKey'], parameters['groupId'] + USER_ADDR, url, data=para) return STATUS_SUCCESS, '' except Exception as e: return STATUS_FAIL, repr(e)