用云函数快速批量处理COS里面的日志

'use strict';
const zlib = require("zlib"),
	readLine = require('readline'),
	COS = require('cos-nodejs-sdk-v5'),
	cos = new COS({
		SecretId: process.env.SECRET_ID,
		SecretKey: process.env.SECRET_KEY,
		KeepAlive: false
	}),
	tencentcloud = require("/opt/node_modules/tencentcloud-sdk-nodejs"), //node14之前的版本的内置SDK不支持cls，需要下载新的sdk用层的方式覆盖进去并用这个方式引用
	ClsClient = tencentcloud.cls.v20201016.Client,
	clientConfig = {
		credential: {
			secretId: process.env.SECRET_ID,
			secretKey: process.env.SECRET_KEY,
		},
		region: process.env.TENCENTCLOUD_REGION,
		profile: {
			httpProfile: {
				endpoint: "cls.tencentcloudapi.com",
			},
		},
	},
	downloadCycle = 60000, //下载周期
	request = require('request');
exports.main_handler = (event, context, callback) => {
	let gunzip = zlib.createGunzip();
	let jsonCount = 0,
		invalidLines = 0;
	let rl = readLine.createInterface(gunzip);
	rl.on('line', function(line) {
		if (/^\s*\{.*\}\s*$/.test(line)) {
			jsonCount++; //收到的行数据似乎是JSON数据
		} else {
			invalidLines++; //收到的一行似乎不是JSON数据
		}
	})
	rl.on('close', () => {
		//因为是demo，这里没有等待所有的可能的并发流都处理完再回调
		//实际使用的时候应该Promise.all或者用异步方式逐个流处理完再回调。
		callback(null, "有效行数: " + jsonCount + " ,无效行数:" + invalidLines)
	})
	if ("Type" in event && event.Type == "Timer") { //定时器触发
		let client = new ClsClient(clientConfig);
		let params = {
			"TopicId": process.env.TopicId,
			"Query": "SCF_Namespace:\"" + process.env.SCF_NAMESPACE + "\"",
			"Count": 10000000, //最大一次下载1000万条。如果一个周期内有可能超过的话需要考虑分包或者该用cos投递
			"From": (Math.floor(Date.now() / downloadCycle - 1) * downloadCycle),
			"To": (Math.floor(Date.now() / downloadCycle) * downloadCycle) //触发点之前的一个完整的时间周期
		};
		//生成日志打包任务，定时器下次触发的时候下载处理
		client.CreateExport(params).then(res => {
			if ("ExportId" in res) {
				console.log("导出任务发起成功，生成的文件等待定时器下次触发的时候处理");
				client.DescribeExports({
					"TopicId": process.env.TopicId
				}).then(res => {
					if ("Exports" in res && res.Exports instanceof Array) {
						for (let i = 0; i < res.Exports.length; i++) {
							let e = res.Exports[i];
							if (e.Status == "Completed") {
								request(e.CosPath).pipe(gunzip);
								client.DeleteExport({"ExportId": e.ExportId});//如果要更保险一点，放到readline的close后面删除更好
							}
						}
					}
				})
			} else {
				console.log("导出任务发起失败，需要告警并走重试逻辑");
			}
		})
	} else if ("Records" in event && event.Records instanceof Array) { //COS触发
		let records = event.Records;
		for (let i = 0; i < records.length; i++) {
			let c = records[i].cos;
			console.log("下载并分析日志文件 " + c.cosObject.key)
			cos.getObject({
				Bucket: c.cosBucket.name + "-" + c.cosBucket.appid,
				Region: c.cosBucket.cosRegion,
				Key: "/" + c.cosObject.key.split("/").slice(3).join("/"),
				Output: gunzip
				//            Output: "/tmp/test.gz"
			}, function(err, data) {
				if (err) console.log("err:" + JSON.stringify(err))
			});
		}
	} else {
		return "未知触发器"
	}
};