用云函数操作时序数据库CTSDB

'use strict';
const request = require('request'),
db_username=process.env.DB_USERNAME,
db_password = process.env.DB_PWD,
db_ip = process.env.DB_IP,
db_port = process.env.DB_PORT,
db_url = "http://"+db_username+":"+db_password+"@"+db_ip+":"+db_port,
moveCount = parseInt(process.env.MOVE_COUNT),
metricMap = {
	"tags":
	{
		"yakID":"string",
		"pastureID":"string",
		"MAC":"string"
	},
	"time":
	{
		"name":"timestamp",
		"format":"epoch_millis"
	},
	"fields":
	{
		"range1":"short",
		"range2":"short",
		"range3":"short",
		"range4":"short",
		"range5":"short"
	},
	"options":
	{
		"expire_day":100,
		"refresh_interval":"10s",
		"number_of_shards":24
	}
};
let lastRecord;
exports.main_handler = async (event, context, callback) => {
	var metricName = "yakMove";
	var url = db_url+"/_metric/"+metricName,res;
    var log = [];
	//获取CTSDB基本信息
	res = await req("get",{url:db_url,headers:{"content-type": "application/json"}})
	console.log(JSON.stringify(res))
    log.push("兼容ElasticSearch版本："+res.version.number);
    log.push("lucene版本"+res.version.lucene_version);

	/** */
	if((await req("get",{url:url,headers:{"content-type": "application/json"}})).status==200){
		log.push("metric已经存在");
		//删除metric
		res = await req("delete",{
			url:url,
			headers:{"content-type": "application/json"} 
		})
		log.push("删除metric"+metricName+"操作结果"+(res.message?res.message:res.reason))
	}
/**/
	if((await req("get",{url:url,headers:{"content-type": "application/json"}})).status!=200){
		log.push("metri不存在");
		//创建metric
		res = await req("put",{
			url:url,
			headers:{"content-type": "application/json"} ,
			body:JSON.stringify(metricMap)
		})
		log.push("新增metric"+metricName+"操作结果"+(res.message?res.message:res.reason))
	}else{
		log.push("metric "+metricName+" 已经存在，测试插入"+moveCount+"条数据")
	}

	url = db_url+"/"+metricName+"/doc/_bulk"
	//生成随机数据
	var data = genBulkData(moveCount);
	var d=new Date()
	//插入数据到metric中
	res = await req("post",{
		url:url,
		headers:{"content-type": "application/json"} ,
		body:data
	});
	console.log("操作结果"+JSON.stringify(res))
	if(res.error){
		console.log("出错了！");
		log.push("出现错误:"+JSON.stringify(res.error)+" 导致批量失败")		
	}else{
		if(res.errors){
			for(var i=0;i<res.items.length;i++){
				if(res.items[i].index.error){
					log.push("出现错误:"+JSON.stringify(res.items[i].index.error))
				}
			}			
		}else{
			console.log("全部写入成功");
		}
		log.push("插入"+moveCount+"条记录用时"+(new Date()-d)+"毫秒，数据库用时"+res.took+"毫秒，平均每条"+res.took/moveCount+"毫秒");
	}
	console.log("\n"+log.join("\n"))
	callback(null,log.join(" "))
	/**/
};
function req(method,reqbody){
	return new Promise(resolve => {
		request[method](reqbody,(error, response, data)=>{
			resolve(JSON.parse(data));
		});
	})
}
var counter=0
function getRandomMove(){
	
	return {
		"yakID":Math.round(Math.random()*100000000),
		"pastureID":Math.round(Math.random()*100000000),
		"MAC":"00 00 00 00 00 00",
		"timestamp":new Date().getTime(),
		"range1":(counter++ == 1)?(99999999999999999999999999999):Math.round(Math.random()*1000),
		"range2":Math.round(Math.random()*2000),
		"range3":Math.round(Math.random()*3000),
		"range4":Math.round(Math.random()*2000),
		"range5":Math.round(Math.random()*1000)		
	}
}
function genBulkData(n){
	var result = []
	for(var i=0;i<n;i++){
		result.push('{"index":{}}');
		result.push(JSON.stringify(getRandomMove()))
	}
	lastRecord = result[n-1];
	console.log("last inserted record:"+JSON.stringify(lastRecord))
	return result.join("\n")+"\n"
}