Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use collect_metric_data method in localstack

Best Python code snippet using localstack_python

mysql_metric_collect.py

Source:mysql_metric_collect.py

...168def get_hour():169    return time.strftime("%Y-%m-%d %H:00:00", time.localtime())170def get_day():171    return time.strftime("%Y-%m-%d 00:00:00", time.localtime())172def collect_metric_data(host,port):173	mysql_metric_list={'datadir':'','log_slave_updates':0,'binlog_format':'',"read_only":0,'version':'',"tx_isolation":'','data_size':0,'log_size':0,'sql_slave_skip_counter':'','max_connections':0,'wait_timeout':0,'interactive_timeout':0,'gtid_mode':0,'db_role':0}174	print host175	conn=get_conn(user=admin_user,passwd=admin_passwd,host=host,port=port)176	conn=conn['value']177	sql='show global variables ;'178	sql_rt=exec_sql(conn,sql)179	rt={}180	com_rt={}181	if sql_rt['status']!=0:182		return sql_rt183	for i in sql_rt['result']:184		if i['Variable_name'] in mysql_metric_list.keys():185	        	mysql_metric_list[i['Variable_name']]=i['Value']		186	mysql_metric_list['version']=mysql_metric_list['version'].split('-')[0]187        data_cmd='du -s -b %s'% mysql_metric_list['datadir']188        data_rt=sys_cmd(host,data_cmd)189	if data_rt['status']==0:190		mysql_metric_list['data_size']=data_rt['result'].split()[0]191	basedir=mysql_metric_list['datadir'].split('mysqldata')[0]192	log_dir='%smysqllog'%basedir193        log_cmd='du -s -b %s'% log_dir194        log_rt=sys_cmd(host,log_cmd)195	if log_rt['status']==0:196		mysql_metric_list['log_size']=log_rt['result'].split()[0]197	sql='show slave status;'198	sql_rt=exec_sql(conn,sql)199	rt={}200	if sql_rt['status']==0:201		if len(sql_rt['result'])>0:202			sql_rt=sql_rt['result'][0]203			io_running=sql_rt['Slave_IO_Running']204			sql_running=sql_rt['Slave_SQL_Running']205			if io_running=='Yes' or sql_running=='Yes':206				mysql_metric_list['db_role']=1207	rt={}208	rt['status']=0209	rt['result']=mysql_metric_list210	return rt211			212#print rt213def main():214    try:215        all_port=get_all_port()216        all_port=all_port['result']217	#all_port=[{'hostname':"db4v.infra.bjac.pdtv.it","port":3306}]218        for i in all_port:219            hostname=i['hostname']220            port=i['port']221            per_com_rt=collect_metric_data(hostname,port)222            if per_com_rt['status']==-1:223                continue224            temp_ops=per_com_rt['result']225	    per_com_rt=temp_ops226            cur_time=get_day()227            sql="replace into cmdb.mysql_metric_day (port,hostname,day_time,data_size,read_only,db_role,version,tx_isolation,skip_counter,max_connection,wait_timeout,interactive_timeout,gtid_mode,log_size,binlog_format,log_slave_updates) values (%d,'%s','%s',%d,'%s',%d,'%s','%s','%s',%d,%d,%d,'%s',%d,'%s','%s')"%(int(port),hostname,cur_time,int(per_com_rt['data_size']),per_com_rt['read_only'],int(per_com_rt['db_role']),per_com_rt['version'],per_com_rt['tx_isolation'],per_com_rt['sql_slave_skip_counter'],int(per_com_rt['max_connections']),int(per_com_rt['wait_timeout']),int(per_com_rt['interactive_timeout']),per_com_rt['gtid_mode'],int(per_com_rt['log_size']),per_com_rt['binlog_format'],per_com_rt['log_slave_updates'])228            conn=get_conn(user=admin_user,passwd=admin_passwd,host=admin_host,port=admin_port)229            conn=conn['value']230            conn.autocommit(1) 231            rt_sql=exec_sql(conn,sql)232    except Exception,e:233        print traceback.print_exc() ...

mongodb_ops_collect.py

Source:mongodb_ops_collect.py

...169def get_hour():170    return time.strftime("%Y-%m-%d %H:00:00", time.localtime())171def get_day():172    return time.strftime("%Y-%m-%d 00:00:00", time.localtime())173def collect_metric_data(host,port):174	rt={}175	mongodb_metric_list={}176	rep_name='repl_%d'%int(port)177	client=pymongo.MongoClient(host,port,replicaSet=rep_name)178	db=client['admin']179	db.authenticate('admin','pandatv#>!$999')180	m_rt=db.command(pymongo.son_manipulator.SON([('serverStatus', 1)]))181	m_rt=m_rt['opcounters']182	rt['status']=0183	rt['result']=m_rt184	return rt185			186#print rt187def main():188    try:189        all_port=get_all_port()190        all_port=all_port['result']191	#all_port=[{'hostname':"db4v.infra.bjac.pdtv.it","port":3306}]192        for i in all_port:193            hostname=i['hostname']194            port=i['port']195            per_com_rt=collect_metric_data(hostname,port)196	    print per_com_rt197            if per_com_rt['status']==-1:198                continue199            temp_ops=per_com_rt['result']200	    per_com_rt=temp_ops201            cur_time=get_hour()202	    sql="replace into cmdb.mongodb_hour_ops(hostname,port,day_time,total_command,`select_ops`,`insert_ops`,`update_ops`,`delete_ops`) values('%s',%d,'%s',%d,%d,%d,%d,%d)"%(hostname,int(port),cur_time,per_com_rt['command'],per_com_rt['query'],per_com_rt['insert'],per_com_rt['update'],per_com_rt['delete'])203	    print sql204            conn=get_conn(user=admin_user,passwd=admin_passwd,host=admin_host,port=admin_port)205            conn=conn['value']206            conn.autocommit(1) 207            rt_sql=exec_sql(conn,sql)208    except Exception,e:209        print traceback.print_exc() ...

example.py

Source:example.py

1from fbprophet import Prophet2from fbprophet.plot import add_changepoints_to_plot3import pandas as pd4import matplotlib.pyplot as plt5cap = 0.15 6floor = 0.07df = pd.read_csv("./collect_metric_data/prophet_data.csv")8#df = pd.read_csv("./data/example_wp_log_R.csv")9print(df.head())10df['cap'] = cap11df['floor'] = floor 12m = Prophet() # Defalut growth='linear'13m.fit(df) 14#m = Prophet(growth = 'logistic') 15#m.fit(df)16future = m.make_future_dataframe(periods=25, freq='H')17print(future.tail())18future['cap'] = cap19future['floor'] = floor 20# Trend Chagepoints21#m = Prophet(changepoint_prior_scale = 0.5) 22#forecast = m.fit(df).predict(future)23# 24forecast = m.predict(future) 25#print(forecast.tail())26print(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(10))27fig1 = m.plot(forecast)28fig2 = m.plot_components(forecast)29a = add_changepoints_to_plot(fig1.gca(), m, forecast) ...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.