建站
咨詢
售后
建站咨詢
新聞中心
這里有您想知道的互聯(lián)網(wǎng)營(yíng)銷解決方案
通過(guò)AnsibleAPI獲取主機(jī)信息-創(chuàng)新互聯(lián)
在我上一篇文章中，編寫了 Ansible Ad-Hoc & Playbook 的 API，
這一篇將使用 Ansible API 獲取主機(jī)詳細(xì)信息，CMDB系統(tǒng)中最基本的功能
話不多說(shuō)，還是直接上代碼
# - encoding:utf-8 -
import os
import sys
\# 因?yàn)槲沂窃?django 中使用，所以我加上了下面這段代碼。自已測(cè)試時(shí)不用加
import django
Search_Path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.append(Search_Path)
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'AutoOPPlatform.settings')
django.setup()
import ansibleAPI
from collections import namedtuple
DEVICES = namedtuple('mount', ['device', 'point', "total_size", "free_size"])
class Resource(object):
def __init__(self, hostname=None, sshport=22, username=None, password=None, **kwargs):
if hostname:
self.hostname = hostname
self.sshport = sshport
self.username = username
self.password = password
self.group = None
if kwargs:
self.hostname = None
self.group = kwargs
self.info = self._get_info()
def _get_info(self):
if self.hostname:
execute = ansibleAPI.AnsibleAPI([dict(hostname=self.hostname, port=self.sshport,
username=self.username, password=self.password)])
execute.PrivateAdHoc('setup')
msg = execute.adhoc_result()
for k in msg.keys():
if k == 'success' and msg['success']:
return dict(success=msg['success'])
elif k == 'unreachable' and msg['unreachable']:
return dict(error=msg['unreachable'])
elif k == 'failed' and msg['failed']:
return dict(error=msg['failed'])
if self.group:
execute = ansibleAPI.AnsibleAPI(self.group)
execute.PrivateAdHoc('setup')
msg = execute.adhoc_result()
return msg
def get_info(self):
return self.info
def get_failed(self):
return self.info['failed']
def get_unreachable(self):
return self.info['unreachable']
def get_success(self):
success = []
for k in self.info['success']:
success.append(k)
return success
def get_hostname(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_hostname']
def ip(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_default_ipv4']['address']
def netmask(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_default_ipv4']['netmask']
def gateway(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_default_ipv4']['gateway']
def broadcast(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_default_ipv4']['broadcast']
def mac(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_default_ipv4']['macaddress']
def memory_cards(self):
if self.hostname:
execute = ansibleAPI.AnsibleAPI([dict(hostname=self.hostname, port=self.sshport,
username=self.username, password=self.password)])
execute.PrivateAdHoc('shell', 'dmidecode -t memory|grep Size|grep MB|wc -l')
msg = execute.adhoc_result()
for k in msg.keys():
if k == 'success' and msg['success']:
return msg['success'][self.hostname]['stdout']
elif k == 'unreachable' and msg['unreachable']:
return 0
elif k == 'failed' and msg['failed']:
return 0
if self.group:
execute = ansibleAPI.AnsibleAPI(self.group)
execute.PrivateAdHoc('shell', 'dmidecode -t memory|grep Size|grep MB|wc -l')
msg = execute.adhoc_result()
return msg
def memory_total(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_memtotal_mb']
def memory_free(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_memfree_mb']
def memory_used(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_memory_mb']['real']['used']
def swap_total(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_swaptotal_mb']
def swap_free(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_swapfree_mb']
def swap_used(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_memory_mb']['swap']['used']
def memory(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_memory_mb']
def cpu_brand_model(self):
take_repetition = set(self.info['success'][self.hostname]['ansible_facts']['ansible_processor'])
take_repetition.remove('GenuineIntel')
if len(take_repetition) == 1:
return list(take_repetition)[0]
elif len(take_repetition) == 0:
return None
else:
cpu = ""
for n in take_repetition:
cpu = str(cpu) + str(n) + "\n"
return cpu
def cpu_number(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_processor_count']
def cpu_core(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_processor_cores']
def cpu_thread(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_processor_vcpus']
def kernel(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_kernel']
def device(self):
device_size = dict()
for dev, info in self.info['success'][self.hostname]['ansible_facts']['ansible_devices'].iteritems():
if dev != 'sr0':
device_size.setdefault(dev, info['size'])
return device_size
def device_number(self):
return len(self.device())
def device_total_size(self):
return reduce(lambda x, y: x + y, self.device().values())
def partitions(self):
partitions = dict()
for dev, info in self.info['success'][self.hostname]['ansible_facts']['ansible_devices'].iteritems():
if dev != 'sr0':
partitions.setdefault(dev, dict())
for partition, v in info['partitions'].iteritems():
partitions[dev].setdefault(partition, v['size'])
return partitions
def mounts(self):
mount_list = []
for dev in self.info['success'][self.hostname]['ansible_facts']['ansible_mounts']:
mount = DEVICES(device=dev['device'], point=dev['mount'],
total_size=dev['size_total'], free_size=dev['size_available'])
mount_list.append(mount)
return mount_list
def system(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_system']
def architecture(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_machine']
def distribution(self):
return "%s %s" % (self.info['success'][self.hostname]['ansible_facts']['ansible_distribution'],
self.info['success'][self.hostname]['ansible_facts']['ansible_distribution_version'])
def uuid(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_product_uuid']
def vendor(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_system_vendor']
def product_name(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_product_name']
def serial_number(self):
return self.info['success'][self.hostname]['ansible_facts']['ansible_product_serial']
if __name__ == '__main__':
host = Resource('10.10.181.132', sshport=20003, username='root', password=123456)
print host.memory_cards()
#hosts = [{'hostname': '10.0.2.6'}, {'hostname': '10.0.2.7'}, {'hostname': '10.0.2.8'}]
#group = Resource(group1=dict(hosts=hosts, vars={}))
#print group.get_failed()
#print group.get_unreachable()
#print group.get_success()另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無(wú)理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國(guó)服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡(jiǎn)單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢(shì)，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場(chǎng)景需求。
分享題目：通過(guò)AnsibleAPI獲取主機(jī)信息-創(chuàng)新互聯(lián)
轉(zhuǎn)載注明：http://www.ef60e0e.cn/article/pecod.html
