from datetime import datetime import pytz import hashlib import hmac import json import textwrap from dateutil.tz import tzutc import requests from yarl import URL from config import configurations # for API from flask import Flask, request, jsonify #import logging # == Configurations cfg = configurations[1] # Manager's endpoint MANAGER_SERVICE_URL = URL(cfg['MANAGER_ENDPOINT']) # Access key of a user (use superadmin's key to query every sessions) ACCESS_KEY = cfg['ADMIN_ACCESS_KEY'] # Secret key of a user SECRET_KEY = cfg['ADMIN_SECRET_KEY'] # API version. Refer to https://github.com/lablup/backend.ai-manager/blob/05f21a09bc6eff8678747f1199c48ed221be974a/src/ai/backend/manager/server.py#L82. API_VERSION = cfg['API_VERSION'] # Hash algorithm to generate Backend.AI request signature HASH_TYPE = cfg['HASH_TYPE'] def generate_signature( *, method, version, endpoint, date, rel_url, content_type, access_key, secret_key, hash_type, ): hash_type = hash_type hostname = endpoint._val.netloc body_hash = hashlib.new(hash_type, b'').hexdigest() sign_str = '{}\n{}\n{}\nhost:{}\ncontent-type:{}\nx-backendai-version:{}\n{}'.format( method.upper(), rel_url, date.isoformat(), hostname, content_type.lower(), version, body_hash ) sign_bytes = sign_str.encode() sign_key = hmac.new(secret_key.encode(), date.strftime('%Y%m%d').encode(), hash_type).digest() sign_key = hmac.new(sign_key, hostname.encode(), hash_type).digest() signature = hmac.new(sign_key, sign_bytes, hash_type).hexdigest() headers = { 'Authorization': 'BackendAI signMethod=HMAC-{}, credential={}:{}'.format( hash_type.upper(), access_key, signature ), } return headers, signature def query_agent_list(): # == Generate authorization header. # HTTP method to use: POST, GET, HEAD, etc. method = 'POST' # Request date time. Should be UTC. date = datetime.now(tzutc()) # Request url. To fetch compute session list, a client have to use # a GraphQL request. Manager serves every GraphQl requests by a single # URL, `/admin/graphql`. rel_url = '/admin/graphql' # Content type of the request. content_type = 'application/json' hdrs, _ = generate_signature( method=method, version=API_VERSION, endpoint=MANAGER_SERVICE_URL, date=date, rel_url=str(rel_url), content_type=content_type, access_key=ACCESS_KEY, secret_key=SECRET_KEY, hash_type=HASH_TYPE, ) # == Build request headers headers = { # Client can set User-Agent string, which will be displayed in # Manager's log. This is optional, but good to know to categorize # the origin of the request. 'User-Agent': 'Backed.AI Session Query Test', # Following three headers are mandatory. 'Content-Type': content_type, 'X-BackendAI-Version': API_VERSION, 'Date': date.isoformat(), } headers.update(hdrs) # == Prepare session query (using GraphQL) # Fields to fetch. Refer following link to see possible fields. # https://github.com/lablup/backend.ai-manager/blob/main/src/ai/backend/manager/models/kernel.py#L687 fields = [ 'id', 'occupied_slots', 'available_slots', 'live_stat', ] # Query definition for compute_session_list can be found at # https://github.com/lablup/backend.ai-manager/blob/05f21a09bc6eff8678747f1199c48ed221be974a/src/ai/backend/manager/models/gql.py#L403 query = ''' query($limit:Int!, $offset:Int!) { agent_list(limit:$limit, offset:$offset) { items { $fields } total_count } } ''' query = query.replace('$fields', ' '.join(fields)) query = textwrap.dedent(query).strip() variables = { 'limit': 50, 'offset': 0, #'sess_id': '91edc3fd-8c23-45e1-93b6-d15a2a0cd888', } data = { 'query': query, 'variables': variables, } # Send requests and return response after parsing into JSON. r = requests.post(MANAGER_SERVICE_URL / rel_url[1:], headers=headers, data=json.dumps(data)) result = r.json() return result def extract_utilization(): #agents's type is dict agents=query_agent_list() result_dict={} item = 1 for agent in agents['agent_list']['items']: if agent['id'] == 'i-gs-gpu-1080ti-01': break ag=agent['id'] #print(ag) #print(type(ag)) #live's type is dict live = json.loads(agent['live_stat']) #6400% 중에 pct %. cpu=live['node']['cpu_util']['pct'] cpu=float(cpu) #100%기준으로 표시하며 소수점 둘째 자리까지만 사용 cpu=round(int(cpu)/64, 2) mem=live['node']['mem']['pct'] #800% 중에 current %. cuda=live['node']['cuda_util']['current'] cuda=float(cuda) #100%기준으로 표시하며 소수점 둘째 자리까지만 사용 cuda=round(int(cuda)/8, 2) disk =live['node']['disk']['pct'] result_dict['item'+str(item)] ={ 'id':ag, 'cpu':cpu, 'cuda':cuda, 'mem':mem, 'disk':disk} #result_dict[agent['id']] ={'cpu':cpu, 'cuda':cuda, 'mem':mem, 'disk':disk} item+=1 #print("cpu_util : ",cpu/64,"%") #print("memory_util : ", mem,"%") #print("gpu_util : ", cuda/8,"%") #print("disk_util : ", disk, "%") #current_time = str(datetime.now()) kst = pytz.timezone('Asia/Seoul') now = datetime.now(kst) # ISO 8601 형식으로 출력 current_time=now.isoformat() # adding time, root of the result_dict result_dict['time']=current_time result_dict ={"results":result_dict} # json 결과값을 Content-type: application/json으로 반환하도록 jsonify 사용 result_json=jsonify(result_dict) return result_json # == API Logger == #logging.basicConfig(filename = "../logs/monitoringAPI.log", level = logging.DEBUG) # == restAPI == app = Flask(__name__) @app.route('/api/getMonitoring', methods=['GET']) def myfunction_api(): result = extract_utilization() return result if __name__ == "__main__": app.run(host='0.0.0.0', port=31000, debug=True) # api url # http://10.231.238.231:31000/api/getMonitoring