通过SDK执行命令

import com.aliyun.ecs20140526.Client;
import com.aliyun.ecs20140526.models.*;
import com.aliyun.teaopenapi.models.Config;
import com.google.gson.Gson;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;


public class CloudAssistantService {

    /**
     * 从环境变量中获取AccessKeyId、AccessKeySecret
     */
    private static final String ACCESS_KEY_ID = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String ACCESS_KEY_SECRET = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    private static final ScheduledExecutorService SCHEDULER = Executors.newScheduledThreadPool(1);
    private static volatile Client ecsClient;

    private CloudAssistantService() {
    }

    /**
     * 初始化ECS客户端
     *
     * @param regionId 地域ID，用于指定ECS客户端所在的地域
     * @return 返回初始化好的ECS客户端实例
     * <p>
     * 该方法使用双重检查锁定模式来确保线程安全地创建单例ECS客户端。
     * 它首先检查客户端是否已经存在，如果不存在，则在同步块中再次检查并创建新的客户端实例。
     */
    public static Client getEcsClient(String regionId) throws Exception {
        if (ecsClient == null) {
            synchronized (CloudAssistantService.class) {
                if (ecsClient == null) {
                    Config config = new Config().setAccessKeyId(ACCESS_KEY_ID).setAccessKeySecret(ACCESS_KEY_SECRET).setRegionId(regionId);
                    ecsClient = new Client(config);
                }
            }
        }
        return ecsClient;
    }

    public static void main(String[] args_) {
        try {
            // 地域ID
            String regionId = "cn-chengdu";
            getEcsClient(regionId);
            // 执行命令的ECS实例ID
            List<String> instanceIds = Arrays.asList("i-2vcXXXXXXXXXXXXXXXb8", "i-2vcXXXXXXXXXXXXXXXot");
            // 命令内容
            String commandContent = "#!/bin/bash\n cat /proc/meminfo";
            // 命令执行超时时间
            long commandTimeOut = 60;


            // 执行命令
            String invokeId = runCommand(commandContent, regionId, instanceIds, commandTimeOut);
            // 查询命令执行结果
            DescribeInvocationsResponse invocationResult = describeInvocations(regionId, invokeId, commandTimeOut);
            System.out.println("The command execution result:" + new Gson().toJson(invocationResult));
            // 省略记录日志文件操作

        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            SCHEDULER.shutdown();
        }
    }

    /**
     * 在指定实例上执行命令
     *
     * @param commandContent 要执行的命令内容
     * @param regionId       实例所在的地域ID
     * @param instanceIds    要执行命令的ECS实例ID集合
     * @param commandTimeOut 命令执行的超时时间
     * @return 返回命令执行的调用ID
     */
    public static String runCommand(String commandContent, String regionId, List<String> instanceIds, long commandTimeOut) {
        try {
            System.out.println("runCommand start...");
            RunCommandRequest request = new RunCommandRequest();
            request.setRegionId(regionId);
            request.setType(Constants.COMMAND_TYPE.RUN_SHELL_SCRIPT);
            request.setCommandContent(commandContent);
            request.setInstanceId(instanceIds);
            request.setTimeout(commandTimeOut);
            RunCommandResponse runCommandResponse = ecsClient.runCommand(request);
            return runCommandResponse.body.invokeId;
        } catch (Exception e) {
            throw new RuntimeException("runCommand failed", e);
        }
    }

    /**
     * 查询云助手命令执行结果
     *
     * @param regionId       地域ID，用于指定调用结果所属的区域
     * @param invokeId       调用ID，用于唯一标识一次调用
     * @param commandTimeOut 命令执行的超时时间
     */
    public static DescribeInvocationsResponse describeInvocations(String regionId, String invokeId, long commandTimeOut) {
        DescribeInvocationsRequest describeInvocationsRequest = new DescribeInvocationsRequest()
                .setRegionId(regionId)
                .setInvokeId(invokeId);

        long delay = 2;
        // 设置最大重试次数
        int maxRetries = (int) (commandTimeOut / delay); 
        int retryCount = 0;

        try {
            while (retryCount < maxRetries) {
                ScheduledFuture<DescribeInvocationsResponse> future = SCHEDULER.schedule(() ->
                        ecsClient.describeInvocations(describeInvocationsRequest), delay, TimeUnit.SECONDS);
                DescribeInvocationsResponse results = future.get();
                List<DescribeInvocationsResponseBody.DescribeInvocationsResponseBodyInvocationsInvocation> invocationList = results.body.invocations.invocation;
                if (invocationList.isEmpty()) {
                    throw new RuntimeException("The command execution result was not found.");
                }
                DescribeInvocationsResponseBody.DescribeInvocationsResponseBodyInvocationsInvocation invocationResult = results.body.invocations.invocation.get(0);
                String invocationStatus = invocationResult.invocationStatus;
                switch (invocationStatus) {
                    case Constants.INVOCATION_STATUS.PENDING:
                    case Constants.INVOCATION_STATUS.RUNNING:
                    case Constants.INVOCATION_STATUS.STOPPING:
                        retryCount++;
                        continue;
                    default:
                        return results;
                }
            }
            throw new RuntimeException("Max retries exceeded for command execution result.");
        } catch (Exception e) {
            throw new RuntimeException("describeInvocationResults failed", e);
        }
    }


    public static class Constants {
        // 命令类型
        public static final class COMMAND_TYPE {
            // 适用于Linux实例的Shell命令。
            public static final String RUN_SHELL_SCRIPT = "RunShellScript";
            // 适用于Windows实例的Bat命令。
            public static final String RUN_BAT_SCRIPT = "RunBatScript";
            // 适用于Windows实例的PowerShell命令。
            public static final String RUN_POWERSHELL_SCRIPT = "RunPowerShellScript";
        }

        // 云助手命令执行结果
        public static final class INVOCATION_STATUS {
            // 系统正在校验或发送命令。
            public static final String PENDING = "Pending";
            // 命令正在实例上执行。
            public static final String RUNNING = "Running";
            // 正在停止执行的命令。
            public static final String STOPPING = "Stopping";
        }
    }
}

import os
import time
import logging
from alibabacloud_ecs20140526 import models as ecs_20140526_models
from alibabacloud_ecs20140526.client import Client as Ecs20140526Client
from alibabacloud_tea_openapi import models as open_api_models

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

ACCESS_KEY_ID = os.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID")
ACCESS_KEY_SECRET = os.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")

if not ACCESS_KEY_ID or not ACCESS_KEY_SECRET:
    raise EnvironmentError(
        "Missing required environment variables: ALIBABA_CLOUD_ACCESS_KEY_ID and ALIBABA_CLOUD_ACCESS_KEY_SECRET")


def get_ecs_client(region_id):
    config = open_api_models.Config(
        access_key_id=ACCESS_KEY_ID,
        access_key_secret=ACCESS_KEY_SECRET,
        region_id=region_id
    )
    return Ecs20140526Client(config)


def execute_command(client, command_content, region_id, instance_ids, command_timeout, command_type):
    if not instance_ids:
        raise ValueError("Instance IDs list cannot be empty.")

    valid_command_types = ["RunShellScript", "RunBatScript", "RunPowerShellScript"]
    if command_type not in valid_command_types:
        raise ValueError(f"Invalid command type: {command_type}. Valid types are {valid_command_types}.")

    request = ecs_20140526_models.RunCommandRequest()
    request.region_id = region_id
    request.type = command_type
    request.command_content = command_content
    request.instance_ids = instance_ids
    request.timeout = command_timeout

    try:
        run_command_response = client.run_command(request)
        return run_command_response.to_map()['body']['InvokeId']
    except Exception as e:
        logging.error(f"Failed to execute command: {e}")
        raise


def query_invocations(client, region_id, invoke_id):
    request = ecs_20140526_models.DescribeInvocationsRequest()
    request.region_id = region_id
    request.invoke_ids = [invoke_id]

    try:
        describe_invocations_response = client.describe_invocations(request)
        return describe_invocations_response.to_map()['body']
    except Exception as e:
        logging.error(f"Failed to query invocations: {e}")
        raise


def wait_for_command_completion(client, region_id, invoke_id, max_retries, backoff_factor=2):
    retry_count = 0
    while retry_count < max_retries:
        time.sleep(backoff_factor ** retry_count)
        results = query_invocations(client, region_id, invoke_id)
        invocation_list = results.get('Invocations', {}).get('Invocation', [])
        if not invocation_list:
            raise RuntimeError("The command execution result was not found.")

        invocation_result = invocation_list[0]
        invocation_status = invocation_result.get('InvocationStatus')
        logging.info(f"Current invocation status: {invocation_status}")

        if invocation_status == "Finished":
            print("query_invocations result:", results)
            break
        elif invocation_status in ["Failed", "Stopped"]:
            raise RuntimeError(f"Command execution failed with status: {invocation_status}")
        else:
            retry_count += 1
    else:
        raise TimeoutError("Command execution timed out.")


def main():
    # 地域ID
    region_id = "cn-chengdu"
    # 执行命令的ECS实例ID
    instance_ids = ["i-2vcXXXXXXXXXXXXXXXb8", "i-2vcXXXXXXXXXXXXXXXot"]
    # 命令内容
    command_content = "#!/bin/bash\n cat /proc/meminfo"
    # 超时时间
    command_timeout = 60
    # 命令类型，支持RunShellScript、RunBatScript、RunPowerShellScript
    command_type = "RunShellScript"

    client = get_ecs_client(region_id)
    invoke_id = execute_command(client, command_content, region_id, instance_ids, command_timeout, command_type)

    max_retries = max(int(command_timeout // 2), 1)
    wait_for_command_completion(client, region_id, invoke_id, max_retries)


if __name__ == "__main__":
    main()