windows 一键扩容

需求背景

公司某业务有个 word 生成 job，可以按照模板或自定义生成 word 和 pdf，由于处理的是 office，所以依赖 windows 环境。 正常业务流量下，一台机器同时部署 3个job节点 即可满足要求，但是当用户需求紧急，需要短时间内生成大量 word 和 pdf 时，当前的部署情况显然无法满足要求。 由于公司业务系统全部部署在阿里云，所以调研了阿里云ECS自动申请以及在windows平台自动部署的技术。

阿里云提供了 open API 可以通过编码的方式来申请 ECS 实例，这里不做过多说明，直接代码体现。

windows下自动部署，采用 python 脚本的方式进行。

open API 自动申请ECS实例

POM 文件

<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-core</artifactId>
    <version>4.4.6</version>
</dependency>
<dependency>
    <groupId>com.aliyun</groupId>
    <artifactId>aliyun-java-sdk-ecs</artifactId>
    <version>4.17.6</version>
</dependency>

ApplyInstanceHelper.java

下面的代码可参考阿里云官网 open api 示例。

import com.alibaba.fastjson.JSON;
import com.aliyuncs.AcsRequest;
import com.aliyuncs.AcsResponse;
import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.ecs.model.v20140526.*;
import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.exceptions.ServerException;
import com.aliyuncs.profile.DefaultProfile;
import com.aliyuncs.profile.IClientProfile;
import com.cosfuture.common.etcd.EtcdUtil;
import com.cosfuture.common.utils.DateUtils;
import com.cosfuture.common.utils.SpringContextUtil;
import com.cosfuture.eiduo.ctb.ops.model.Config;
import com.cosfuture.eiduo.ctb.ops.service.InstanceService;
import com.cosfuture.eiduo.sms.service.SmsService;
import com.google.common.base.Joiner;
import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * 申请实例帮助类.
 *
 * @author leichu 2021-05-13.
 */
public class ApplyInstanceHelper {

	private static final Logger logger = LoggerFactory.getLogger(ApplyInstanceHelper.class);

	private static final String INSTANCE_STATUS_RUNNING = "Running";
	private static final long INSTANCE_STATUS_TOTAL_CHECK_TIME_ELAPSE_MILLISECOND = 60000 * 3;

	private static ApplyInstanceHelper INSTANCE;

	private ApplyInstanceHelper() {

	}

	public static ApplyInstanceHelper getInstance() {
		if (null == INSTANCE) {
			synchronized (ApplyInstanceHelper.class) {
				if (null == INSTANCE) {
					INSTANCE = new ApplyInstanceHelper();
				}
			}
		}
		return INSTANCE;
	}

	public void apply(List<Long> ids, Config config) {
		RunInstancesRequest instancesRequest = runInstancesRequest(config);
		RunInstancesResponse response = callOpenApi(instancesRequest);
		if (response == null) {
			return;
		}
		Map<Long, String> map = Maps.newHashMap();
		List<String> instanceIdSets = response.getInstanceIdSets();
		for (int i = 0; i < ids.size(); i++) {
			map.put(ids.get(i), instanceIdSets.get(i));
		}
		SpringContextUtil.getBean(InstanceService.class).updateRid(map);

		logger.warn("实例创建成功. InstanceIds: {}", instanceIdSets);
		callToDescribeInstances(instanceIdSets);
	}

	public void release(String instanceId) {
		DeleteInstanceRequest request = new DeleteInstanceRequest();
		request.setInstanceId(instanceId);
		request.setForce(true);
		try {
			callOpenApi(request);
			SpringContextUtil.getBean(InstanceService.class).releaseSuccess(Lists.newArrayList(instanceId));
			logger.warn("实例{}已释放！", instanceId);
		} catch (Exception e) {
			logger.error("实例{}释放失败！", instanceId);
		}

	}


	/**
	 * 每3秒中检查一次实例的状态，超时时间设为3分钟。
	 *
	 * @param instanceIds 需要检查的实例ID
	 */
	private void callToDescribeInstances(List<String> instanceIds) {
		Long startTime = System.currentTimeMillis();
		for (; ; ) {
			DateUtils.sleep(3);
			DescribeInstancesRequest describeInstancesRequest = new DescribeInstancesRequest();
			describeInstancesRequest.setRegionId(Config.regionId);
			describeInstancesRequest.setInstanceIds(JSON.toJSONString(instanceIds));
			DescribeInstancesResponse describeInstancesResponse = callOpenApi(describeInstancesRequest);
			Long timeStamp = System.currentTimeMillis();
			if (describeInstancesResponse == null) {
				continue;
			} else {
				for (DescribeInstancesResponse.Instance instance : describeInstancesResponse.getInstances()) {
					logger.info("实例详情: {}", JSON.toJSONString(instance));
					if (INSTANCE_STATUS_RUNNING.equals(instance.getStatus())) {
						List<DescribeInstancesResponse.Instance.NetworkInterface> interfaces = instance.getNetworkInterfaces();
						List<String> ip = interfaces.stream().map(DescribeInstancesResponse.Instance.NetworkInterface::getPrimaryIpAddress).collect(Collectors.toList());
						// 更新实例的状态为 运行中。
						String ipStr = Joiner.on(",").join(ip);
						SpringContextUtil.getBean(InstanceService.class).runSuccess(instance.getInstanceId(), ipStr);
						sendNotify(instance.getInstanceId(), ipStr);
						instanceIds.remove(instance.getInstanceId());
						logger.warn("实例已启动: {}", instance.getInstanceId());
					} else {
						logger.warn("实例正在启动: {}", instance.getInstanceId());
					}
				}
			}
			if (instanceIds.size() == 0) {
				logger.warn("所有实例全部启动完成.");
				return;
			}
			if (timeStamp - startTime > INSTANCE_STATUS_TOTAL_CHECK_TIME_ELAPSE_MILLISECOND) {
				if (instanceIds.size() > 0) {
					logger.warn("实例在{}分钟内启动失败: {}", INSTANCE_STATUS_TOTAL_CHECK_TIME_ELAPSE_MILLISECOND / 60000, JSON.toJSONString(instanceIds));
				} else {
					logger.warn("所有实例均已启动.");
				}
				return;
			}
		}
	}

	/**
	 * 调用OpenAPI的方法，这里进行了错误处理
	 *
	 * @param request AcsRequest, Open API的请求
	 * @param <T>     AcsResponse 请求所对应返回值
	 * @return 返回OpenAPI的调用结果，如果调用失败，则会返回null
	 */
	private <T extends AcsResponse> T callOpenApi(AcsRequest<T> request) {
		IClientProfile profile = DefaultProfile.getProfile(Config.regionId, Config.accessKeyId, Config.accessSecret);
		IAcsClient client = new DefaultAcsClient(profile);
		try {
			T response = client.getAcsResponse(request, false, 0);
			logger.warn("Success. OpenAPI Action: {} call successfully.", request.getActionName());
			return response;
		} catch (ServerException e) {
			logger.error("Fail. Something with your connection with Aliyun go incorrect. ErrorCode: {}", e.getErrCode());
		} catch (ClientException e) {
			logger.error("Fail. Business error. ErrorCode: {}, RequestId: {}", e.getErrCode(), e.getRequestId());
		}
		return null;
	}

	private RunInstancesRequest runInstancesRequest(Config config) {
		RunInstancesRequest runInstancesRequest = new RunInstancesRequest();
		runInstancesRequest.setDryRun(config.dryRun);
		runInstancesRequest.setRegionId(config.regionId);
		runInstancesRequest.setInstanceType(config.getInstanceType());
		runInstancesRequest.setInstanceChargeType(config.instanceChargeType);
		runInstancesRequest.setImageId(config.imageId);
		runInstancesRequest.setSecurityGroupId(config.securityGroupId);
		runInstancesRequest.setPeriod(config.period);
		runInstancesRequest.setPeriodUnit(config.periodUnit);
		runInstancesRequest.setZoneId(config.zoneId);
		runInstancesRequest.setInternetChargeType(config.internetChargeType);
		runInstancesRequest.setVSwitchId(config.vSwitchId);
		runInstancesRequest.setInstanceName(config.getInstanceName());
		runInstancesRequest.setAmount(config.getAmount());
		runInstancesRequest.setInternetMaxBandwidthOut(config.internetMaxBandwidthOut);
		runInstancesRequest.setHostName(config.getHostName());
		runInstancesRequest.setIoOptimized(config.ioOptimized);
		runInstancesRequest.setUniqueSuffix(config.uniqueSuffix);
		runInstancesRequest.setSpotStrategy(config.spotStrategy);
		runInstancesRequest.setSystemDiskSize(config.getSystemDiskSize());
		runInstancesRequest.setSystemDiskCategory(config.systemDiskCategory);
		runInstancesRequest.setPassword(config.getInstancePassword());
		return runInstancesRequest;
	}

	private void sendNotify(String aliId, String ip) {
		try {
			String openIds = EtcdUtil.getV("/webapp/ctb/ops.notify.openid", "");
			List<String> ids = Splitter.on(",").splitToList(openIds);
			String title = "ECS实例启动";
			String content = String.format("实例ID:%s IP:%s", aliId, ip);
			SpringContextUtil.getBean(SmsService.class).sendWeChatMsg(ids, title, content, "");
		} catch (Exception e) {
			logger.error("实例ID:{} IP:{}", aliId, ip, e);
		}
	}

}

自动部署

startPy.bat

将该脚本放入 windows 的开机启动目录下，C:\Users\leichu\AppData\Roaming\Microsoft\Windows\Start Menu\Programs\Startup。

@echo off
c:
cd c:/job/
python download.py
pause

download.py

# encoding=utf-8
import os
import requests
import time

scriptPath = 'https://xxx.aliyuncs.com/script/ctbAutoDeploy.py'
execScriptPath = 'C:/job/ctbAutoDeploy.py'

def main():
    print(">>>>>>>>>>>>>>>>>>> 开始下载脚本 <<<<<<<<<<<<<<<<<<")
    print(f'请求地址：{scriptPath}')
    res = requests.get(scriptPath, stream=True)
    print(f'状态码：{res.status_code}')
    if res.status_code == 200:
        open(execScriptPath, 'wb').write(res.content)
        print(">>>>>>>>>>>>>>>>>>> 脚本下载完成 <<<<<<<<<<<<<<<<<<")
    del res

    print(">>>>>>>>>>>>>>>>>>> 3秒后自动执行部署脚本 <<<<<<<<<<<<<<<<<<")
    time.sleep(3)
    print(">>>>>>>>>>>>>>>>>>> 开始执行部署脚本 <<<<<<<<<<<<<<<<<<")
    os.system(f'python {execScriptPath}')
    print(">>>>>>>>>>>>>>>>>>> 部署脚本执行完成 <<<<<<<<<<<<<<<<<<")

if __name__ == '__main__':
    main()

ctbAutoDeploy.py

# encoding=utf-8
import os
import shutil
import requests
import json

import socket
from pathlib import Path

from xml.dom import minidom

import threading

bizMap = {
    '1': {'name': '错题本', 'alias': 'ctb-gen', 'src': 'ctb-gen-job', 'taskType': 1, 'branch': 'deploy', 'pk': 'ctb-gen-job'},
    '2': {'name': '教师讲义', 'alias': 'jsjy-gen', 'src': 'ctb-gen-job', 'taskType': 3, 'branch': 'deploy', 'pk': 'ctb-gen-job'},
    '3': {'name': '阶段复习', 'alias': 'jdfx-gen', 'src': 'ctb-gen-job', 'taskType': 4, 'branch': 'deploy', 'pk': 'ctb-gen-job'},
    '4': {'name': '原划题平台', 'alias': 'separate-wt', 'src': 'word-transform-job', 'taskType': 1, 'branch': 'deploy', 'pk': 'word-transform-job'},
    '5': {'name': '新资源加工平台', 'alias': 'res-palt-wt', 'src': 'word-transform-job', 'taskType': 1, 'branch': 'res-plat', 'pk': 'res-plat-wt'},
}

def get_host_ip():
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(('8.8.8.8', 80))
        ip = s.getsockname()[0]
    finally:
        s.close()
    return ip

def getBiz():
    url = f'https://www.eiduo.com/ctbOps/getOneUnDeployBiz?ip={get_host_ip()}'
    req = requests.get(url)
    return req.text

def deploySuccess(instanceId):
    url = f'https://www.eiduo.com/ctbOps/deployed?id={instanceId}'
    print(url)
    req = requests.get(url)
    return req.text

def deploy(index, biz, deployDir, pkName):
    newDeployDir = f'{deployDir}_{index}'
    shutil.copytree(f'{deployDir}{pkName}', newDeployDir)

    # shutil.copytree(src, dst)
    # 修改日志文件的标题
    logConfigFile = f'{newDeployDir}/cflogback.xml'
    modifyLogFile(logConfigFile, biz['alias']+f'_{index}')
    # 修改启动脚本的标题
    startScript = f'{newDeployDir}/bin/startup.bat'
    modifyTitle(startScript, biz['alias']+f'_{index}')
    # 启动
    startPro(f'{startScript}')


def main():
    config = getBiz()
    result = json.loads(config)
    print(result)
    if result['data'] == 'error':
        print('部署异常')
        return
    bizCode = result['data']['bizCode']
    biz = bizMap[f'{bizCode}']
    proName = biz['src']
    pkName = biz['pk']
    nodeCnt = result['data']['node']

    deployDir = f"c:/job/{biz['alias']}/"
    if not Path(deployDir).exists():
        os.makedirs(deployDir)

    srcDir = f"c:/job/{biz['alias']}/src/"
    if not Path(srcDir).exists():
        os.makedirs(srcDir)

    # 获取源码
    gitPath = f"http://kitty:123456@gs.mizss.com/cosfuture/{proName}.git"
    os.system(f'cd {srcDir} && git clone {gitPath}')
    branch = biz['branch']
    # 切换到指定的分支
    os.system(f'cd {srcDir}{proName} && git checkout {branch}')
    # if biz['branch'] != 'master':
    #     os.system(f'cd {srcDir}{proName} && git checkout {branch}')

    # 构建 & 打包 & 解压
    os.system(f'cd {srcDir}{proName} && mvn clean package')

    shutil.copyfile(f'{srcDir}{proName}/target/{pkName}-package.zip',
                    f'{deployDir}{pkName}-package.zip')
    os.system(f'Bandizip.exe x -o:{deployDir} {deployDir}{pkName}-package.zip')

    # 错题本修改 taskType 的值
    if biz['src'] == 'ctb-gen-job' and biz['taskType'] != 1:
        xmlPath = f'{deployDir}{pkName}/applicationContext-service.xml'
        modifyXML(xmlPath, biz['taskType'])

    # deploy(0, biz, deployDir, pkName)
    # deploy(1, biz, deployDir, pkName)
    threads = []
    for i in range(0, nodeCnt):
        print(f'节点{i+1}启动中.......')
        t = threading.Thread(target=deploy, args=(i+1, biz, deployDir, pkName))
        threads.append(t)
        t.start()
    #等待所有线程任务结束。
    for t in threads:
        t.join()
    
    print("所有节点部署任务全部完成")

    # 通知部署成功
    res = deploySuccess(result['data']['id'])
    result = json.loads(res)
    if result['data'] == 'success':
        print('部署启动成功')
    else:
        print('部署失败')


def modifyLogFile(logConfigFile, title):
    dom = minidom.parse(logConfigFile)
    root = dom.documentElement
    prop = root.getElementsByTagName('property')
    for pr in prop:
        name = pr.getAttribute('name')
        if name == 'logback.appName':
            pr.setAttribute('value', f'{title}')
            break
    with open(logConfigFile, 'w', encoding='utf-8') as fh:
        dom.writexml(fh, encoding='utf-8')
        print('logConfigFile title 写入 OK!')


def modifyXML(xmlPath, value):
    dom = minidom.parse(xmlPath)
    root = dom.documentElement
    beans = root.getElementsByTagName('bean')
    for bean in beans:
        id = bean.getAttribute('id')
        if id == 'taskType':
            bean.getElementsByTagName(
                'constructor-arg')[0].setAttribute('value', f'{value}')
            break
    with open(xmlPath, 'w', encoding='utf-8') as fh:
        dom.writexml(fh, encoding='utf-8')
        print('taskType 写入 OK!')


def modifyTitle(startScriptPath, title):
    file_data = ""
    with open(startScriptPath, 'r', encoding='utf-8') as fh:
        for line in fh:
            if 'title ' in line:
                line = line.replace(line, f'title {title} \n')
            file_data += line
    with open(startScriptPath, 'w', encoding='utf-8') as fh:
        fh.write(file_data)
    print('startScript title 写入 OK!')


def startPro(startScript):
    print(startScript)
    os.system(f'start {startScript}')

if __name__ == '__main__':
    main()