运维自动化之nacos，apollo，k8s常用API汇总

　　1. Nacos API
　　本代码包括列出service列表,列出指定服务下实例列表,服务上下线,实例注册,服务信息修改等方法, 更多关于nacosapi的详细信息，请参考https://nacos.io/zh-cn/docs/open-api.html,import requests class NacosAPI(object):     def __init__(self,url,username,password):         self.auth, self.url = {＂username＂: username,＂password＂: password},url         self.response = requests.post(＂%s/nacos/v1/auth/users/login＂ % (self.url), data=self.auth,verify=False).json()         self.params=((＂accessToken＂, self.response[＂accessToken＂]),)     def servicesList(self):         ＂＂＂列出service列表＂＂＂         data=requests.get(＂%s/nacos/v1/ns/catalog/services?pageNo=1&pageSize=100＂ % self.url,params=self.params).text         print(data)     def instanceList(self,service):         ＂＂＂列出指定服务下实例列表＂＂＂         data = requests.get(＂%s/nacos/v1/ns/instance/list?serviceName=%s＂ % (self.url,service), params=self.params).text         print(data)     def changeStatus(self,serviceName,ip,port,enabled):         ＂＂＂实例上下线 enabled:true 上线, false 下线＂＂＂         self.data = {＂serviceName＂: serviceName,＂ip＂: ip,＂port＂:port,＂enabled＂: enabled}         data=requests.put(＂%s/nacos/v1/ns/instance＂ % (self.url), params=self.params, data=self.data , verify=False).text         print(data)     def register(self,serviceName,ip,port):         ＂＂＂实例注册＂＂＂         url=＂%s/nacos/v1/ns/instance＂ % (self.url)         data=requests.post(url,data={＂ip＂:ip,＂port＂:port,＂serviceName＂:serviceName})         print(data.text)     def instanceChange(self):         ＂＂＂修改实例信息＂＂＂         data = {＂serviceName＂: ＂merchant-order＂, ＂ip＂: ＂10.42.4.129＂, ＂port＂: 8080, ＂clusterName＂: ＂DEFAULT＂,＂weight＂: 1, ＂enabled＂: False}         response = requests.put(url=＂%s/nacos/v1/ns/instance＂ % (self.url), data=data)         print(response.text) if __name__==＂__main__＂:     nacos=NacosAPI(＂http://192.168.3.160:8848＂,＂nacos＂,＂NACOS＂)     nacos.register()
　　2.Apollo 前台界面 API
　　因开放平台未支持授权，故抓取了前台界面的api实现批量授权功能，开放平台的管理需要提前授权才能管理import requests class apollo_web():     def __init__(self,potalurl,username,password):         self.potalurl,self.username,self.password=potalurl,username,password         data = {＂username＂: username,＂password＂: password}         response = requests.post(f＂{self.potalurl}/signin＂,data=data,allow_redirects=False)         self.SESSION=dict(response.cookies)[＂SESSION＂]     def Authorize_App(self,Token,app:str):         ＂＂＂授权应用给用户＂＂＂         cookies = {＂NG_TRANSLATE_LANG_KEY＂: ＂zh-CN＂,＂SESSION＂: self.SESSION}         params = ((＂envs＂, [＂DEV＂, ＂SIT＂]),(＂type＂, ＂AppRole＂))         headers = {＂Content-Type＂: ＂application/json＂}         data = ＂{＂appId＂:＂%s＂}＂ % app         response = requests.post(f＂{self.potalurl}/consumers/{Token}/assign-role＂ ,                                  params=params, headers=headers,cookies=cookies, data=data, verify=False)         print(response.text) if __name__==＂__main__＂:     apollo=apollo_web()     apollo.Authorize_App(＂Token＂,＂sms＂)
　　3.Apollo 开放平台 API
　　本代码包含获取配置,修改配置,发布配置,批量发布配置等功能， 在我的日常工作中，批量发布用的很频繁，import  requests,json from loguru import logger class ApolloAPI(object):     def __init__(self,apolloenv):         apollo_info= {＂aliyun＂:{＂url＂:＂http://apollo.ali.cn＂,＂token＂:＂a2814e7476b64fbdb08d9e9ec0b2d2d76e621cac＂}                      ,＂local＂:{＂url＂:＂http://apollo.tenx.cn＂,＂token＂:＂a2814e7476b64fbdb08d9e9ec0b2d2d76e621cac＂}}         self.apollourl=apollo_info[apolloenv].get(＂url＂)         self.apps=[＂zuul＂,＂omc＂,＂oms＂,＂omss＂]         self.headers = {＂Authorization＂: apollo_info[apolloenv].get(＂token＂),＂Content-Type＂: ＂application/json;charset=UTF-8＂}     def GetnamespaceName(self,env,appid,clusterName=＂default＂):         ＂＂＂获取某集群下，指定appid的所有namespace＂＂＂         url = ＂%s/openapi/v1/envs/%s/apps/%s/clusters/%s/namespaces＂ % (self.apollourl,env,appid,clusterName)         data=requests.get(url, headers=self.headers).json()         result=[x[＂namespaceName＂] for x in data ]         return result     def AddApp_Config(self,env,appId,namespaceName,key,value,comment=＂＂,clusterName=＂default＂):         ＂＂＂添加配置＂＂＂         url=＂%s/openapi/v1/envs/%s/apps/%s/clusters/%s/namespaces/%s/items＂ % (self.apollourl,env,appId,clusterName,namespaceName )         data={＂key＂:key,＂value＂:value,＂comment＂:comment,＂dataChangeCreatedBy＂:＂apollo＂}         respose = requests.post(url, json=data, headers=self.headers)         print(＂新增配置 ：环境:%s,appId:%s,namespaceName:%s,key:%s,value:%s,新增配置结果:%s＂ % (env,appId,namespaceName,key,value,respose.text))     def ReleasesApp_OneNamespaceName(self,env,appId,namespaceName,clusterName=＂default＂):         ＂＂＂发布指定appid下指定的namespace＂＂＂         url=＂%s/openapi/v1/envs/%s/apps/%s/clusters/%s/namespaces/%s/releases＂ % (self.apollourl,env,appId,clusterName,namespaceName)         data={＂releaseTitle＂:＂releaseTitle＂,＂releaseComment＂:＂releaseComment＂,＂releasedBy＂:＂apollo＂}         respose=requests.post(url,json=data,headers=self.headers)         logger.info(＂ReleasesApp_OneNamespaceName:respose:%s＂ % (respose.text))         if respose.status_code==200:             return ＂发布成功＂         else:             return ＂发布失败,HTTP状态码:%s＂ % respose.status_code     def ReleasesApp_AllNamespaceName(self,env,appId,clusterName=＂default＂):         ＂＂＂发布指定appid下所有namespace＂＂＂         result=[]         namespaceNames=self.GetnamespaceName(env,appId,clusterName)         for namespaceName in namespaceNames:             logger.info(＂ReleasesApp_AllNamespaceName:app:%s,namespaceName:%s＂ % (appId,namespaceName))             result.append(self.ReleasesApp_OneNamespaceName(env,appId,namespaceName,clusterName))         return result     def ReleasesAllApps(self,env=＂FINDEV＂):         ＂＂＂发布指定环境的所有应用＂＂＂         for app in self.apps:             try:                 result=self.ReleasesApp_AllNamespaceName(env,app)                 logger.info(＂app:%s,%s＂ % (app,result))             except Exception as e:                 logger.info(＂ReleasesAllApps:%s,error:%s＂ % (app,e))     def BeforeReleasesDetailAll(self,env,appId,clusterName=＂default＂):         ＂＂＂获取即将发布后的配置信息＂＂＂         result={}         url=＂%s/openapi/v1/envs/%s/apps/%s/clusters/%s/namespaces＂ % (self.apollourl,env,appId,clusterName)         try:             respose=requests.get(url, headers=self.headers).json()             for x in respose:                 result[x[＂namespaceName＂]]={}                 for i in x[＂items＂]:                     result[x[＂namespaceName＂]][i[＂key＂]]=i[＂value＂]             return result         except Exception as e:             print(＂BeforeReleasesDetailAll:%s:%s＂ % (env,appId))             raise Exception(＂sdsdsdsd＂)     def CurrentReleasesedDetailAll(self,env,appId,clusterName=＂default＂):         ＂＂＂获取发布前的配置信息＂＂＂         result={}         namespaceNames = self.GetnamespaceName(env, appId, clusterName)         for namespaceName in namespaceNames:             url=＂%s/openapi/v1/envs/%s/apps/%s/clusters/%s/namespaces/%s/releases/latest＂ % (self.apollourl,env,appId,clusterName,namespaceName)             respose=requests.get(url, headers=self.headers)             if respose.text:                 resposejson=respose.json()                 result[resposejson[＂namespaceName＂]]=resposejson[＂configurations＂]         return result if __name__==＂__main__＂:     apollo=ApolloAPI(＂ali＂)     apollo.ReleasesAllApps(＂PRE＂)
　　4.K8S API
　　本代码支持文件和文本认证信息，方便根据实际情况开发from kubernetes import client, config from kubernetes.client.rest import ApiException class KubernetesAPI:     def __init__(self,kube_conf):         ＂＂＂文件＂＂＂         config.load_kube_config(kube_conf)     def __init__(self,kube_conf_text):         ＂＂＂文本＂＂＂         yamldata=yaml.load(kube_conf_text, Loader=yaml.Loader)         k8_loader = kube_config.KubeConfigLoader(yamldata)         call_config = type.__call__(Configuration)         k8_loader.load_and_set(call_config)         Configuration.set_default(call_config)     def list_node(self):         ＂＂＂列出node＂＂＂         v1 = client.CoreV1Api()         try:             ret=v1.list_node(watch=False)             result={＂code＂:200,＂data:＂:[{＂uid＂:x.metadata.uid,＂name＂:x.metadata.name,＂creattime＂:x.metadata.creation_timestamp,＂pod_cid_rs＂:x.spec.pod_cid_rs,                            ＂pod_cidr＂:x.spec.pod_cidr,＂unschedulable＂:x.spec.unschedulable} for x in ret.items]}         except ApiException as e:             result = {＂code＂: 500, ＂msg＂: e}         return result     def list_namespace(self):         ＂＂＂列出namespace＂＂＂         v1 = client.CoreV1Api()         try:             ret=v1.list_namespace(watch=False)             result={＂code＂:200,＂data:＂:[{＂uid＂:x.metadata.uid,＂name＂:x.metadata.name,＂creattime＂:x.metadata.creation_timestamp} for x in ret.items]}         except ApiException as e:             result = {＂code＂: 500, ＂msg＂: e}         return result     def list_deploymet(self,namespaces):         ＂＂＂列出指定namespaces的deploymet＂＂＂         v1 = client.AppsV1Api()         try:             ret = v1.list_namespaced_deployment(namespaces,watch=False)             result = {＂code＂: 200, ＂data＂:[{＂uid＂:x.metadata.uid,＂name＂:x.metadata.name,＂namespace＂:x.metadata.namespace,＂replicas＂:x.spec.replicas,                            ＂status＂:x.status.available_replicas,＂creattime＂:x.metadata.creation_timestamp} for x in ret.items]}         except ApiException as e:             result = {＂code＂: 500, ＂msg＂: e}         return result     def list_pods(self,namespace):         v1 = client.CoreV1Api()         try:             ret=v1.list_namespaced_pod(namespace)             result={＂code＂:200,＂data＂:[{＂uid＂:x.metadata.uid,＂name＂:x.metadata.name,＂pod_ip＂:x.status.pod_ip,＂host_ip＂:x.status.host_ip,＂creattime＂:x.metadata.creation_timestamp} for x in ret.items]}         except ApiException as e:             result = {＂code＂: 500, ＂msg＂: e}         return result     def list_pods_deployment(self,namespace,deployment):         ＂＂＂查询deployment的pods＂＂＂         v1 = client.CoreV1Api()         try:             ret=v1.list_namespaced_pod(namespace,label_selector=＂app.kubernetes.io/name=%s＂ % deployment)             result={＂code＂:200,＂data＂:[{＂uid＂:x.metadata.uid,＂name＂:x.metadata.name,＂pod_ip＂:x.status.pod_ip,＂host_ip＂:x.status.host_ip,＂creattime＂:x.metadata.creation_timestamp} for x in ret.items]}         except ApiException as e:             result = {＂code＂: 500, ＂msg＂: e}         return result     def query_pod_stdoutlogs(self,namespaces,pod):         ＂＂＂查看pod stdout日志＂＂＂         v1 = client.CoreV1Api()         try:             result = {＂code＂: 200, ＂data＂:v1.read_namespaced_pod_log(pod,namespaces)}         except ApiException as e:             result = {＂code＂: 500, ＂msg＂: e}         return result     def query_deployment_stdoutlogs(self,namespace,deployment):         ＂＂＂查看deployment下pod的日志＂＂＂         result=[]         querypods=self.list_pods_deployment(namespace,deployment)         if querypods[＂code＂]==200:             for x in querypods[＂data＂]:                 result.append({＂name＂:x[＂name＂],＂logs＂:self.query_pod_stdoutlogs(namespace,x[＂name＂])})         else:             result = {＂code＂: 500, ＂msg＂: ＂查询pod出错:%s＂ % querypods[＂msg＂]}         return result     def remove_pod(self,namespace,podname):         ＂＂＂删除pod＂＂＂         v1 = client.CoreV1Api()         try:             v1.delete_namespaced_pod(podname, namespace)             return {＂code＂: 200, ＂msg＂: ＂删除成功＂}         except ApiException as e:             return {＂code＂: 500, ＂msg＂: str(e)}     def deployment_setvar(self,namespace,deployment,key,value):         ＂＂＂设置deployment变量＂＂＂         v1=client.AppsV1Api()         body={＂spec＂:{＂template＂:{＂spec＂:{＂containers＂:[{＂env＂:[{＂name＂:key,＂value＂:value}],＂name＂:deployment}]}}}}         try:             v1.patch_namespaced_deployment(name=deployment,namespace=namespace,body=body)             return {＂code＂: 200, ＂msg＂:＂设置成功＂}         except ApiException as e:             return {＂code＂:500,＂msg＂:str(e)} if __name__==＂__main__＂:     ss=KubernetesAPI(＂D:Softwareconfig＂)     sss=ss.query_deployment_stdoutlogs(＂dev＂,＂qtz＂)     print(sss)