# -*- coding: utf-8 -*- # tencentcloud-sdk-python from qcloud_cos import CosConfig, CosS3Client, CosClientError, CosServiceError from tencentcloud.common import credential from tencentcloud.common.profile.client_profile import ClientProfile from tencentcloud.common.profile.http_profile import HttpProfile from tencentcloud.common.exception.tencent_cloud_sdk_exception import TencentCloudSDKException from tencentcloud.tem.v20210701 import tem_client, models import json import os ########## # 公共参数 # ########## # 密钥参数，替换为用户的secret_id和secret_key # 例: secret_id = "xxx" secret_id = "" # 例: secret_key = "xxx" secret_key = "" # 大地域 big_region = "China" ############### # 应用与环境信息 # ############### # 应用名 # 例: application_name = "xxx" application_name = "" # 环境名 # 例: environment_name = "xxx" environment_name = "" # 编程语言, 可选值 JAVA / PYTHON / GO / NODE / OTHER # 例: coding_language = "JAVA" coding_language = "JAVA" # 部署方式, 可选值 JAR(通过 jar 包部署) / WAR (通过 war 包部署) / IMAGE (通过镜像部署) deploy_mode = "JAR" ########## # 规格信息 # ########## # cpu 规格, 单位核 cpu_spec = 1 # 内存规格, 单位Gi memory_spec = 1 # 初始实例数 (仅在首次部署时生效) init_pod_num = 1 ############################ # IMAGE 部署时, 可配置以下参数 # ############################ # 镜像仓库类型, 可选值 0 (个人版TCR) / 1 (企业版TCR) / 2 (公共镜像) repo_type = 0 # 企业版TCR实例ID(仅在镜像仓库为企业版TCR是生效) tcr_instance_id = "" # 镜像仓库地址 # 例: ccr.ccs.tencentyun.com/namespace/repo:tag image_repo = "" ################################ # JAR / WAR 部署时, 可配置以下参数 # ################################ # pkg_path 代表 JAR / WAR 部署 # 例: /xxx/demo.jar pkg_path = "" # JDK版本, 可选值 OPEN:8 / OPEN:11 / KONA:8 / KONA:11 jdk_version = "KONA:11" # 操作系统, 当 jdk_version 为 OPEN:8 / OPEN:11 时，可选值为 ALPINE / CENTOS, 当 jdk_version 为 KONA:8 / KONA:11 时，可选值为 ALPINE / TENCENTOS os_flavour = "ALPINE" # 版本号 deploy_version = "v1" ########## # 部署策略 # ########## # 总分批数 total_batch_count = 1 # 分批策略, 可选值 0 (全自动) / 1 (全手动) / 2 (小批量验证) deploy_strategy_type = 0 # 小批量验证(Beta批次)实例数 beta_batch_num = 1 # 每批暂停间隔, 单位秒 batch_interval = 60 # 最小可用实例数 min_available = 1 # 是否强制发布 force_deploy = False ########################################### # 其他部署参数 (选填, 如不填将以上次部署结果为准) # ########################################### # jvm 参数 (仅当 coding_language 为 JAVA 时生效) # 例: jvm_opts = "-Xms512m -Xmx1g" jvm_opts = None # 环境变量 # 例: # env_conf = [ # { # "Key": "自定义环境变量键", # "Value": "自定义环境变量值", # "Type": "default" # }, # { # "Key": "引用明文配置键", # "Value": "引用明文配置值", # "Type": "referenced", # "Config": "明文配置名称" # }, # { # "Key": "引用密文配置键", # "Value": "引用密文配置值", # "Type": "referenced", # "Secret": "密文配置名称" # } # ] env_conf = None # 应用启停 # 启动后执行的脚本 # 例: post_start = "/bin/sh\n-c\necho \"hello world\"" post_start = None # 停止前执行的脚本 # 例: pre_stop = "/bin/sh\n-c\necho \"hello world\"" pre_stop = None # 健康检查 # 就绪探针配置 # 例: # readiness = { # "Type": "HttpGet", # "Protocol": "HTTP", # "Path": "/health", # "Port": 8075, # "InitialDelaySeconds": 5, # "TimeoutSeconds": 3, # "PeriodSeconds": 30 # } readiness = None # 存活探针配置 # 例: # liveness = { # "Type": "HttpGet", # "Protocol": "HTTP", # "Path": "/health", # "Port": 8075, # "InitialDelaySeconds": 5, # "TimeoutSeconds": 3, # "PeriodSeconds": 30 # } liveness = None # 启动探针配置 # 例: # startupProbe = { # "Type": "HttpGet", # "Protocol": "HTTP", # "Path": "/health", # "Port": 8075, # "InitialDelaySeconds": 5, # "TimeoutSeconds": 3, # "PeriodSeconds": 30 # } startupProbe = None # 安全组 # 例: security_group_ids = ["sg-xxxxxxxx"] security_group_ids = None # 持久化存储 # 数据卷配置 # 例: # storage_confs = [ # { # "StorageVolName": "存储卷名称", # "StorageVolPath": "存储卷路径", # "StorageVolIp": "存储卷IP" # } # ] storage_confs = None # 数据卷挂载配置 # 例: # storage_mount_confs = [ # { # "VolumeName": "数据卷名", # "MountPath": "数据卷绑定路径", # } # ] storage_mount_confs = None # 监控配置 # 自动apm采集(skywalking), 可选值 1(开启) / 0 (关闭) # 例: enable_tracing = 1 enable_tracing = None # 配置文件 # 挂载配置信息 # 例: # setting_confs = [ # { # "ConfigDataName": "明文配置名称", # "MountedPath": "挂载路径" # }, # { # "SecretDataName": "密文配置名称", # "MountedPath": "挂载路径" # } # ] setting_confs = None # 访问配置 # 例: # service = { # "ServicePortMappingList": [ # { # "Type": "CLUSTER", # "ServiceName": "k8s-service", # "PortMappingItemList": [ # { # "Port": 8080, # "TargetPort": 8077, # "Protocol": "TCP" # } # ] # }, # { # "Type": "EXTERNAL", # "ServiceName": "public-pvc", # "PortMappingItemList": [ # { # "Port": 8080, # "TargetPort": 8080, # "Protocol": "TCP" # } # ] # }, # { # "Type": "VPC", # "ServiceName": "private-pvc", # "SubnetId": "subnet-plyc619s", # "PortMappingItemList": [ # { # "Port": 8080, # "TargetPort": 8080, # "Protocol": "TCP" # } # ] # } # ], # "EnableRegistryNextDeploy": 0, # "FlushAll": True # } service = None def init_client(): try: endpoint = "tem.tencentcloudapi.com" region = "" if big_region == "China": region = "ap-nanjing" elif big_region == "Asia": region == "ap-tokyo" cred = credential.Credential(secret_id, secret_key) http_profile = HttpProfile() http_profile.endpoint = endpoint client_profile = ClientProfile() client_profile.httpProfile = http_profile return tem_client.TemClient(cred, region, client_profile) except TencentCloudSDKException as err: print(err) raise err def describe_environment_id_and_deploy_region(): try: req = models.DescribeEnvironmentsRequest() params = { "Filters": [ { "Name": "EnvironmentName", "Value": [environment_name] } ], "Limit": -1, "Offset": -1 } req.from_json_string(json.dumps(params)) resp = client.DescribeEnvironments(req) print("DescribeEnvironments: " + resp.to_json_string()) environments = resp.Result.Records if len(environments) > 0: environment = next(filter(lambda env: env.EnvironmentName == environment_name, environments)) return environment.EnvironmentId, environment.Region else: raise Exception("environment not found") except TencentCloudSDKException as err: print(err) raise err def describe_application_id_and_if_first_time_deploy(): try: req = models.DescribeApplicationsRequest() params = { "Filters": [ { "Name": "ApplicationName", "Value": [application_name] } ], "Limit": -1, "Offset": -1 } req.from_json_string(json.dumps(params)) resp = client.DescribeApplications(req) print("DescribeApplications: " + resp.to_json_string()) applications = resp.Result.Records if len(applications) > 0: application = next(filter(lambda app: app.ApplicationName == application_name, applications)) return application.ApplicationId, application.ActiveVersions[0].Status == "un_deploy" or len(list(filter(lambda active_version: active_version.EnvironmentId == environment_id, application.ActiveVersions))) == 0 else: req = models.CreateApplicationRequest() params = { "ApplicationName": application_name, "Description": "", "RepoType": 3, "CodingLanguage": coding_language } req.from_json_string(json.dumps(params)) resp = client.CreateApplication(req) print("CreateApplication: " + resp.to_json_string()) return resp.Result, True except TencentCloudSDKException as err: print(err) raise err def push_to_cos(): file_name = os.path.basename(pkg_path) try: req = models.CreateCosTokenRequest() params = { "ApplicationId": application_id, "PkgName": file_name, "OptType": 1 } req.from_json_string(json.dumps(params)) resp = client.CreateCosToken(req) print("CreateCosToken: " + resp.to_json_string()) except TencentCloudSDKException as err: print(err) raise err cos_secret_id = resp.Result.TmpSecretId cos_secret_key = resp.Result.TmpSecretKey cos_session_token = resp.Result.SessionToken cos_full_path = resp.Result.FullPath cos_bucket = resp.Result.Bucket cos_region = resp.Result.Region scheme = 'https' config = CosConfig( Region=cos_region, SecretId=cos_secret_id, SecretKey=cos_secret_key, Token=cos_session_token, Scheme=scheme ) cos_client = CosS3Client(config) for i in range(0, 10): try: response = cos_client.upload_file( Bucket=cos_bucket, LocalFilePath=pkg_path, Key=cos_full_path ) print("PushToCos: " + json.dumps(response)) break except CosClientError or CosServiceError as err: print(err) raise err return cos_full_path def deploy_application(): params = extract_common_deploy_param() if deploy_mode == "IMAGE": return deploy_application_image(params) else: return deploy_application_package(params) def extract_common_deploy_param(): common_deploy_param = { "InitPodNum": init_pod_num } application_info = {} if not first_time_deploy: try: # DescribeApplicationInfo params = { "ApplicationId": application_id, "EnvironmentId": environment_id } req = models.DescribeApplicationInfoRequest() req.from_json_string(json.dumps(params)) resp = client.DescribeApplicationInfo(req) print("DescribeApplicationInfo: " + resp.to_json_string()) application_info = json.loads(resp.to_json_string())["Result"] # DescribeDeployApplicationDetail req = models.DescribeDeployApplicationDetailRequest() req.from_json_string(json.dumps(params)) resp = client.DescribeDeployApplicationDetail(req) print("DescribeDeployApplicationDetail: " + resp.to_json_string()) application_detail = json.loads(resp.to_json_string())["Result"] common_deploy_param.update({ "InitPodNum": application_info["InitPodNum"], "DeployStrategyConf": { "TotalBatchCount": total_batch_count, "BetaBatchNum": beta_batch_num, "DeployStrategyType": deploy_strategy_type, "BatchInterval": batch_interval, "MinAvailable": min_available, "Force": force_deploy if application_detail.get("Status", "") == "Finish" else True } }) except TencentCloudSDKException as err: print(err) raise err common_deploy_param.update({ "ApplicationId": application_id, "EnvironmentId": environment_id, "DeployMode": deploy_mode, "ConfEdited": True, "CpuSpec": application_info.get("CpuSpec", None) if cpu_spec is None else cpu_spec, "MemorySpec": application_info.get("MemorySpec", None) if memory_spec is None else memory_spec, "JvmOpts": application_info.get("JvmOpts", None) if jvm_opts is None else jvm_opts, "EnvConf": application_info.get("EnvConf", None) if env_conf is None else env_conf, "PostStart": application_info.get("PostStart", None) if post_start is None else post_start, "PreStop": application_info.get("PreStop", None) if pre_stop is None else pre_stop, "Readiness": application_info.get("Readiness", None) if readiness is None else readiness, "Liveness": application_info.get("Liveness", None) if liveness is None else liveness, "StartupProbe": application_info.get("StartupProbe", None) if startupProbe is None else startupProbe, "SecurityGroupIds": application_info.get("SecurityGroupIds", None) if security_group_ids is None else security_group_ids, "StorageConfs": application_info.get("StorageConfs", None) if storage_confs is None else storage_confs, "StorageMountConfs": application_info.get("StorageMountConfs", None) if storage_mount_confs is None else storage_mount_confs, "EnableTracing": application_info.get("EnableTracing", None) if enable_tracing is None else enable_tracing, "SettingConfs": application_info.get("SettingConfs", None) if setting_confs is None else setting_confs, "Service": application_info.get("Service", None) if service is None else service }) return common_deploy_param def deploy_application_package(params): pkg_name = push_to_cos() try: req = models.DeployApplicationRequest() params.update({ "JdkVersion": jdk_version, "PkgName": pkg_name, "SpeedUp": deploy_mode == "JAR" and jdk_version.split(":")[1] == "11", "OsFlavour": os_flavour, "DeployVersion": deploy_version, }) req.from_json_string(json.dumps(params)) resp = client.DeployApplication(req) print("DeployApplication: " + resp.to_json_string()) return resp.Result except TencentCloudSDKException as err: print(err) raise err def deploy_application_image(params): try: req = models.DeployApplicationRequest() image_repo_server = image_repo.split("/")[0] image_tag = image_repo.split(":")[-1] img_repo = image_repo[len(image_repo_server) + 1: len(image_repo) - len(image_tag) - 1] params.update({ "RepoServer": image_repo_server, "RepoType": repo_type, "ImgRepo": img_repo, "DeployVersion": image_tag, "TcrInstanceId": tcr_instance_id }) req.from_json_string(json.dumps(params)) resp = client.DeployApplication(req) print("DeployApplication: " + resp.to_json_string()) return resp.Result except TencentCloudSDKException as err: print(err) raise err if __name__ == "__main__": client = init_client() environment_id, deploy_region = describe_environment_id_and_deploy_region() application_id, first_time_deploy = describe_application_id_and_if_first_time_deploy() deploy_application()