statistics report function added
This commit is contained in:
parent
85310a69a7
commit
ee6c70341b
13 changed files with 902 additions and 721 deletions
|
|
@ -1,303 +1,9 @@
|
|||
# coding: utf-8
|
||||
|
||||
import sys
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import commands
|
||||
import requests
|
||||
from SWSCloudNode.settings import settings
|
||||
from SWSCloudNode.logger import logging
|
||||
from SWSCloudNode import lxc
|
||||
from SWSCloudNode.compute.lxc import LXC
|
||||
from SWSCloudNode.compute.qemu import Qemu
|
||||
from SWSCloudNode.compute.qemu import QemuStats
|
||||
|
||||
|
||||
class Tasks:
|
||||
def __init__(self):
|
||||
self.endpoint = settings.get('server', 'endpoint')
|
||||
self.id = settings.get('server', 'id')
|
||||
self.secret = settings.get('server', 'secret')
|
||||
|
||||
def get_item(self):
|
||||
try:
|
||||
response = requests.get('%s/server_api/task' % self.endpoint, auth=(self.id, self.secret))
|
||||
except Exception as e:
|
||||
logging.error('no connection with %s' % self.endpoint)
|
||||
return None
|
||||
else:
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
logging.error("Unexpected status code: %d" % response.status_code)
|
||||
return None
|
||||
|
||||
|
||||
class Node:
|
||||
def tasks_get(self):
|
||||
try:
|
||||
response = requests.get(
|
||||
'%s/server_api/tasks' % settings.get('server', 'endpoint'),
|
||||
auth=(settings.get('server', 'id'), settings.get('server', 'secret'))
|
||||
)
|
||||
except Exception as e:
|
||||
sys.exit('no connection with %s' % settings.get('server', 'endpoint'))
|
||||
# print e
|
||||
else:
|
||||
return {
|
||||
'status': response.status_code,
|
||||
'results': response.json()
|
||||
}
|
||||
|
||||
def task_status_update(self, task_id, status):
|
||||
response = requests.put(
|
||||
'%s/server_api/tasks/%s' % (
|
||||
settings.get('server', 'endpoint'),
|
||||
task_id
|
||||
),
|
||||
auth=(
|
||||
settings.get('server', 'id'),
|
||||
settings.get('server', 'secret'),
|
||||
),
|
||||
data={
|
||||
"status": status
|
||||
}
|
||||
)
|
||||
return response.json()
|
||||
|
||||
def report_container_stats(self, container_id, statistics):
|
||||
response = requests.post(
|
||||
'%s/server_api/containers/stats/%s' % (
|
||||
settings.get('server', 'endpoint'),
|
||||
container_id
|
||||
),
|
||||
auth=(settings.get('server', 'id'), settings.get('server', 'secret')),
|
||||
data={
|
||||
'status': json.dumps(statistics)
|
||||
}
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
return False
|
||||
|
||||
# TODO: подумать куда переместить
|
||||
def container_config_create(self, container_id, link, ipv4, ipv6):
|
||||
cfg = list()
|
||||
cfg.append("lxc.network.type = veth")
|
||||
cfg.append("lxc.network.flags = up")
|
||||
cfg.append("lxc.network.name = eth0")
|
||||
cfg.append("lxc.network.link = %s" % link)
|
||||
|
||||
if ipv4['ipv4']:
|
||||
# cfg.append('lxc.network.ipv4 = %s/32' % ipv4['ipv4'])
|
||||
cfg.append('lxc.network.ipv4 = %s' % ipv4['ipv4'])
|
||||
cfg.append('lxc.network.ipv4.gateway = %s' % ipv4['ipv4_gateway'])
|
||||
|
||||
if 'ipv6' in ipv6 and 'ipv6_gateway' in ipv6:
|
||||
# cfg.append('lxc.network.ipv6 = %s/64' % ipv6['ipv6'])
|
||||
cfg.append('lxc.network.ipv6 = %s' % ipv6['ipv6'])
|
||||
cfg.append('lxc.network.ipv6.gateway = %s', ipv6['ipv6_gateway'])
|
||||
|
||||
config_file = '/var/lib/gocloud/node/configs/%s.config' % container_id
|
||||
|
||||
cfg_file = open(config_file, 'w')
|
||||
cfg_file.write('\n'.join(cfg))
|
||||
cfg_file.write('\n')
|
||||
cfg_file.close()
|
||||
return True
|
||||
|
||||
def container_authkey_create(self, container_id, auth_key):
|
||||
# create ssh_key.pub
|
||||
authkey_file = '/var/lib/gocloud/node/auth-keys/%s.pub' % container_id
|
||||
ak = open(authkey_file, 'w')
|
||||
ak.write(auth_key)
|
||||
ak.write('\n')
|
||||
ak.close()
|
||||
return True
|
||||
|
||||
|
||||
class Task(Node):
|
||||
def interface2ip(self):
|
||||
# intf = open(self.settings['proxy_interface'], 'r').read().split('\n')[0]
|
||||
interface = "eth0"
|
||||
intf_ip = commands.getoutput("ip address show dev " + interface).split()
|
||||
intf_ip = intf_ip[intf_ip.index('inet') + 1].split('/')[0]
|
||||
return intf_ip
|
||||
|
||||
def container_create(self, task):
|
||||
return True
|
||||
|
||||
def container_destroy(self, task):
|
||||
# check exists container name
|
||||
if lxc.lxc().exists(task['parameters']['hostname']):
|
||||
# todo: if hostname already exists then node callback to server to rename container
|
||||
return False
|
||||
|
||||
lxc.lxc().destroy(task['parameters']['hostname'])
|
||||
|
||||
# TODO: check status
|
||||
|
||||
# delete record from dnsmasq
|
||||
# dnsmasq.Dnsmasq().delete(task['parameters']['hostname'])
|
||||
|
||||
return True
|
||||
|
||||
def container_start(self, task):
|
||||
logging.debug("container_start")
|
||||
print "================ "
|
||||
print task
|
||||
lxc.lxc().start(task['parameters']['hostname'])
|
||||
# TODO: check status
|
||||
return True
|
||||
|
||||
def container_stop(self, task):
|
||||
logging.debug("container_stop")
|
||||
lxc.lxc().stop(task['parameters']['hostname'])
|
||||
# TODO: check status
|
||||
return True
|
||||
|
||||
def container_restart(self, task):
|
||||
logging.debug("container_restart")
|
||||
lxc.lxc().stop(task['parameters']['hostname'])
|
||||
lxc.lxc().start(task['parameters']['hostname'])
|
||||
# TODO: check status
|
||||
return True
|
||||
|
||||
def init(self):
|
||||
task = TCPClient().request(Request().build("task_get", self.request_auth, None))
|
||||
# check exists element 'version'
|
||||
if not "version" in task:
|
||||
logging.error("Response not contain 'version' element")
|
||||
return False
|
||||
|
||||
if task['version'] == "1.0":
|
||||
if task['status'] == 0:
|
||||
if task['method'] == "container_create":
|
||||
# create container
|
||||
result = Task().container_create(task)
|
||||
if not result:
|
||||
return False
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
# todo: hold job if status not 0
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_destroy":
|
||||
# delete hostname from dnsmasq /etc/lxc/dnsmasq.conf
|
||||
result = self.container_destroy(task)
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
# todo: held job if status not 0
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_start":
|
||||
self.container_start(task)
|
||||
# todo: held job if status not 0
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_stop":
|
||||
self.container_stop(task)
|
||||
# TODO: held job if status not 0
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_restart":
|
||||
self.container_restart(task)
|
||||
# todo: held job if status not 0
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_clone":
|
||||
logging.debug("container_clone")
|
||||
# TODO: set clone_IP to DNSMASQ
|
||||
subprocess.call("/usr/bin/lxc-clone -o %(hostname)s -n %(clone_hostname)s" % task['parameters'], shell=True)
|
||||
# TODO: check container status
|
||||
# todo: held job if status not 0
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_mx_add":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
if self.Request_tast_update(task['task_id'])['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_mx_delete":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
if self.Request_tast_update(task['task_id'])['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_web_add":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
nginx.Nginx().vhost_add(task['parameters']['vhost_id'], task['parameters']['vhost'], task['parameters']['container_ip'])
|
||||
nginx.Service().reload()
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
if task['method'] == "service_web_delete":
|
||||
logging.debug("service_web_delete")
|
||||
# todo: held job if status not 0
|
||||
|
||||
print
|
||||
print task
|
||||
|
||||
nginx.Nginx().vhost_delete(task['parameters']['container_ip'], task['parameters']['vhost_id'])
|
||||
nginx.Service().reload()
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_web_update":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
if self.Request_tast_update(task['task_id'])['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_ssh_allow":
|
||||
logging.debug("service_ssh_allow")
|
||||
# todo: held job if status not 0
|
||||
|
||||
values = (self.interface2ip(), task['parameters']['port'], task['parameters']['container_ip'], 22)
|
||||
os.popen("ufw allow %s" % task['parameters']['port'])
|
||||
os.popen("iptables -t nat -I PREROUTING -p tcp -d %s --dport %s -j DNAT --to %s:%s" % values)
|
||||
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_ssh_deny":
|
||||
logging.debug("service_ssh_deny")
|
||||
# todo: held job if status not 0
|
||||
logging.debug("------")
|
||||
logging.debug(task)
|
||||
os.popen("ufw deny %s" % task['parameters']['port'])
|
||||
logging.debug("------")
|
||||
|
||||
task_update_result = TCPClient().request(Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
else:
|
||||
if task['status'] == 4:
|
||||
print "auth fail"
|
||||
else:
|
||||
print "structure version not supported"
|
||||
|
||||
return None
|
||||
from .node import Node, StatisticsReporter
|
||||
from .tasks import Tasks
|
||||
|
|
|
|||
|
|
@ -99,3 +99,9 @@ class Common:
|
|||
"request": json.dumps(request)
|
||||
}
|
||||
return urllib.urlopen(server_url, urllib.urlencode(params)).read()
|
||||
|
||||
@staticmethod
|
||||
def is_root_user():
|
||||
if os.getuid() == 0:
|
||||
return True
|
||||
return False
|
||||
|
|
|
|||
0
SWSCloudNode/compute/__init__.py
Normal file
0
SWSCloudNode/compute/__init__.py
Normal file
1
SWSCloudNode/compute/lxc/__init__.py
Normal file
1
SWSCloudNode/compute/lxc/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
from .lxc import LXC
|
||||
|
|
@ -67,7 +67,7 @@ class _LXCMonitor(threading.Thread):
|
|||
self.join()
|
||||
|
||||
|
||||
class lxc():
|
||||
class LXC(object):
|
||||
def __init__(self):
|
||||
logging.debug("")
|
||||
|
||||
2
SWSCloudNode/compute/qemu/__init__.py
Normal file
2
SWSCloudNode/compute/qemu/__init__.py
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
from .qemu import Qemu
|
||||
from .stats import QemuStats
|
||||
|
|
@ -6,7 +6,7 @@ import subprocess
|
|||
from SWSCloudNode.common import Common
|
||||
|
||||
|
||||
class QEMU:
|
||||
class Qemu:
|
||||
def __init__(self):
|
||||
# qemu+ssh://root@laforge.usersys.redhat.com/system
|
||||
self.conn = libvirt.open("qemu:///system")
|
||||
81
SWSCloudNode/compute/qemu/stats.py
Normal file
81
SWSCloudNode/compute/qemu/stats.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
# Get the network I/O statistics
|
||||
# http://libvirt.org/docs/libvirt-appdev-guide-python/en-US/html/libvirt_application_development_guide_using_python-Guest_Domains-Monitoring-IO_stats.html
|
||||
|
||||
from __future__ import print_function
|
||||
import sys
|
||||
import libvirt
|
||||
from xml.etree import ElementTree
|
||||
|
||||
|
||||
class QemuStats(object):
|
||||
def __init__(self):
|
||||
self.conn = libvirt.open('qemu:///system')
|
||||
|
||||
def __del__(self):
|
||||
self.conn.close()
|
||||
|
||||
def network(self, dom):
|
||||
results = dict(
|
||||
read_bytes=0, read_packets=0, read_errors=0, read_drops=0,
|
||||
write_bytes=0, write_packets=0, write_errors=0, write_drops=0)
|
||||
|
||||
dom = self.conn.lookupByName(dom)
|
||||
# dom = conn.lookupByID(5)
|
||||
if dom == None:
|
||||
return results
|
||||
|
||||
tree = ElementTree.fromstring(dom.XMLDesc())
|
||||
iface = tree.find('devices/interface/target').get('dev')
|
||||
stats = dom.interfaceStats(iface)
|
||||
|
||||
results['read_bytes'] = str(stats[0])
|
||||
results['read_packets'] = str(stats[1])
|
||||
results['read_errors'] = str(stats[2])
|
||||
results['read_drops'] = str(stats[3])
|
||||
results['write_drops'] = str(stats[4])
|
||||
results['write_packets'] = str(stats[5])
|
||||
results['write_errors'] = str(stats[6])
|
||||
results['write_drops'] = str(stats[7])
|
||||
|
||||
return results
|
||||
|
||||
def cpu(self, dom):
|
||||
# http://libvirt.org/docs/libvirt-appdev-guide-python/en-US/html/libvirt_application_development_guide_using_python-Guest_Domains-Monitoring-vCPU.html
|
||||
results = dict(cpu_time=0, system_time=0, user_time=0)
|
||||
if self.conn == None:
|
||||
print('Failed to open connection to qemu:///system', file=sys.stderr)
|
||||
return results
|
||||
|
||||
# dom = conn.lookupByID(5)
|
||||
dom = self.conn.lookupByName(dom)
|
||||
if dom == None:
|
||||
print('Failed to find the domain '+domName, file=sys.stderr)
|
||||
return results
|
||||
|
||||
stats = dom.getCPUStats(True)
|
||||
|
||||
results['cpu_time'] = str(stats[0]['cpu_time'] / 100000)
|
||||
results['system_time'] = str(stats[0]['system_time'] / 100000)
|
||||
results['user_time'] = str(stats[0]['user_time'] / 10000)
|
||||
|
||||
return results
|
||||
|
||||
def memory(self, dom):
|
||||
# http://libvirt.org/docs/libvirt-appdev-guide-python/en-US/html/libvirt_application_development_guide_using_python-Guest_Domains-Monitoring-Memory.html
|
||||
results = dict()
|
||||
if self.conn == None:
|
||||
print('Failed to open connection to qemu:///system', file=sys.stderr)
|
||||
return results
|
||||
|
||||
# dom = conn.lookupByID(5)
|
||||
dom = self.conn.lookupByName(dom)
|
||||
if dom == None:
|
||||
print('Failed to find the domain '+domName, file=sys.stderr)
|
||||
return results
|
||||
|
||||
stats = dom.memoryStats()
|
||||
# print('memory used:')
|
||||
for name in stats:
|
||||
results[name] = stats[name]
|
||||
# print(' '+str(stats[name])+' ('+name+')')
|
||||
return results
|
||||
118
SWSCloudNode/node.py
Normal file
118
SWSCloudNode/node.py
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
# coding: utf-8
|
||||
|
||||
import sys
|
||||
import json
|
||||
import requests
|
||||
from SWSCloudNode.settings import settings
|
||||
|
||||
|
||||
class Node(object):
|
||||
def tasks_get(self):
|
||||
try:
|
||||
response = requests.get(
|
||||
'%s/server_api/tasks' % settings.get('server', 'endpoint'),
|
||||
auth=(settings.get('server', 'id'), settings.get('server', 'secret')))
|
||||
except Exception as e:
|
||||
sys.exit('no connection with %s' % settings.get('server', 'endpoint'))
|
||||
else:
|
||||
return dict(status=response.status_code, results=response.json())
|
||||
|
||||
def task_status_update(self, task_id, status):
|
||||
response = requests.put(
|
||||
'%s/server_api/tasks/%s' % (
|
||||
settings.get('server', 'endpoint'),
|
||||
task_id
|
||||
),
|
||||
auth=(
|
||||
settings.get('server', 'id'),
|
||||
settings.get('server', 'secret'),
|
||||
),
|
||||
data={
|
||||
"status": status
|
||||
}
|
||||
)
|
||||
return response.json()
|
||||
|
||||
def report_container_stats(self, container_id, statistics):
|
||||
response = requests.post(
|
||||
'%s/server_api/containers/stats/%s' % (
|
||||
settings.get('server', 'endpoint'),
|
||||
container_id
|
||||
),
|
||||
auth=(settings.get('server', 'id'), settings.get('server', 'secret')),
|
||||
data={
|
||||
'status': json.dumps(statistics)
|
||||
}
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
return False
|
||||
|
||||
# TODO: подумать куда переместить
|
||||
def container_config_create(self, container_id, link, ipv4, ipv6):
|
||||
"""
|
||||
|
||||
:param container_id:
|
||||
:param link:
|
||||
:param ipv4:
|
||||
:param ipv6:
|
||||
:return:
|
||||
"""
|
||||
cfg = list()
|
||||
cfg.append("lxc.network.type = veth")
|
||||
cfg.append("lxc.network.flags = up")
|
||||
cfg.append("lxc.network.name = eth0")
|
||||
cfg.append("lxc.network.link = %s" % link)
|
||||
|
||||
if ipv4['ipv4']:
|
||||
# cfg.append('lxc.network.ipv4 = %s/32' % ipv4['ipv4'])
|
||||
cfg.append('lxc.network.ipv4 = %s' % ipv4['ipv4'])
|
||||
cfg.append('lxc.network.ipv4.gateway = %s' % ipv4['ipv4_gateway'])
|
||||
|
||||
if 'ipv6' in ipv6 and 'ipv6_gateway' in ipv6:
|
||||
# cfg.append('lxc.network.ipv6 = %s/64' % ipv6['ipv6'])
|
||||
cfg.append('lxc.network.ipv6 = %s' % ipv6['ipv6'])
|
||||
cfg.append('lxc.network.ipv6.gateway = %s', ipv6['ipv6_gateway'])
|
||||
|
||||
config_file = '/var/lib/gocloud/node/configs/%s.config' % container_id
|
||||
|
||||
cfg_file = open(config_file, 'w')
|
||||
cfg_file.write('\n'.join(cfg))
|
||||
cfg_file.write('\n')
|
||||
cfg_file.close()
|
||||
return True
|
||||
|
||||
def container_authkey_create(self, container_id, auth_key):
|
||||
# create ssh_key.pub
|
||||
authkey_file = '/var/lib/gocloud/node/auth-keys/%s.pub' % container_id
|
||||
ak = open(authkey_file, 'w')
|
||||
ak.write(auth_key)
|
||||
ak.write('\n')
|
||||
ak.close()
|
||||
return True
|
||||
|
||||
|
||||
class StatisticsReporter(object):
|
||||
@staticmethod
|
||||
def send_vm_statistics(vm_id, data):
|
||||
response = requests.post(
|
||||
'%s/stats/v1/compute/vms/%s' % (
|
||||
settings.get('statistics', 'endpoint'), vm_id),
|
||||
# TODO: node auth
|
||||
# auth=(settings.get('server', 'id'), settings.get('server', 'secret')),
|
||||
json=data)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def send_containers_statistics(vm_id, data):
|
||||
response = requests.post(
|
||||
'%s/stats/v1/compute/containers/%s' % (
|
||||
settings.get('statistics', 'endpoint'), vm_id),
|
||||
# TODO: node auth
|
||||
# auth=(settings.get('server', 'id'), settings.get('server', 'secret')),
|
||||
json=data)
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
return False
|
||||
243
SWSCloudNode/tasks.py
Normal file
243
SWSCloudNode/tasks.py
Normal file
|
|
@ -0,0 +1,243 @@
|
|||
# coding: utf-8
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
import commands
|
||||
import requests
|
||||
from SWSCloudNode.logger import logging
|
||||
from SWSCloudNode.settings import settings
|
||||
from SWSCloudNode.logger import logging
|
||||
|
||||
|
||||
ALLOWED_TASKS = [
|
||||
'container_delete', 'container_create', 'container_start', 'container_stop', 'container_restart',
|
||||
'vm_create', 'vm_delete', 'vm_start', 'vm_stop', 'vm_restart',
|
||||
]
|
||||
|
||||
|
||||
class Tasks:
|
||||
def __init__(self):
|
||||
self.endpoint = settings.get('server', 'endpoint')
|
||||
self.id = settings.get('server', 'id')
|
||||
self.secret = settings.get('server', 'secret')
|
||||
|
||||
@staticmethod
|
||||
def is_allowed_task(task):
|
||||
if task in ALLOWED_TASKS:
|
||||
return True
|
||||
return False
|
||||
|
||||
def get_item(self):
|
||||
try:
|
||||
response = requests.get('%s/server_api/task' % self.endpoint, auth=(self.id, self.secret))
|
||||
except Exception as e:
|
||||
logging.error('no connection with %s' % self.endpoint)
|
||||
return None
|
||||
else:
|
||||
if response.status_code == 200:
|
||||
return response.json()
|
||||
logging.error("Unexpected status code: %d" % response.status_code)
|
||||
return None
|
||||
|
||||
def interface2ip(self):
|
||||
# intf = open(self.settings['proxy_interface'], 'r').read().split('\n')[0]
|
||||
interface = "eth0"
|
||||
intf_ip = commands.getoutput("ip address show dev " + interface).split()
|
||||
intf_ip = intf_ip[intf_ip.index('inet') + 1].split('/')[0]
|
||||
return intf_ip
|
||||
|
||||
def container_create(self, task):
|
||||
return True
|
||||
|
||||
def container_destroy(self, task):
|
||||
# check exists container name
|
||||
if lxc.lxc().exists(task['parameters']['hostname']):
|
||||
# todo: if hostname already exists then node callback to server to rename container
|
||||
return False
|
||||
|
||||
lxc.lxc().destroy(task['parameters']['hostname'])
|
||||
|
||||
# TODO: check status
|
||||
|
||||
# delete record from dnsmasq
|
||||
# dnsmasq.Dnsmasq().delete(task['parameters']['hostname'])
|
||||
|
||||
return True
|
||||
|
||||
def container_start(self, task):
|
||||
logging.debug("container_start")
|
||||
print "================ "
|
||||
print task
|
||||
lxc.lxc().start(task['parameters']['hostname'])
|
||||
# TODO: check status
|
||||
return True
|
||||
|
||||
|
||||
def container_stop(self, task):
|
||||
logging.debug("container_stop")
|
||||
lxc.lxc().stop(task['parameters']['hostname'])
|
||||
# TODO: check status
|
||||
return True
|
||||
|
||||
|
||||
def container_restart(self, task):
|
||||
logging.debug("container_restart")
|
||||
lxc.lxc().stop(task['parameters']['hostname'])
|
||||
lxc.lxc().start(task['parameters']['hostname'])
|
||||
# TODO: check status
|
||||
return True
|
||||
|
||||
|
||||
def init(self):
|
||||
task = TCPClient().request(Request().build("task_get", self.request_auth, None))
|
||||
# check exists element 'version'
|
||||
if not "version" in task:
|
||||
logging.error("Response not contain 'version' element")
|
||||
return False
|
||||
|
||||
if task['version'] == "1.0":
|
||||
if task['status'] == 0:
|
||||
if task['method'] == "container_create":
|
||||
# create container
|
||||
result = Task().container_create(task)
|
||||
if not result:
|
||||
return False
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
# todo: hold job if status not 0
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_destroy":
|
||||
# delete hostname from dnsmasq /etc/lxc/dnsmasq.conf
|
||||
result = self.container_destroy(task)
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
# todo: held job if status not 0
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_start":
|
||||
self.container_start(task)
|
||||
# todo: held job if status not 0
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_stop":
|
||||
self.container_stop(task)
|
||||
# TODO: held job if status not 0
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_restart":
|
||||
self.container_restart(task)
|
||||
# todo: held job if status not 0
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "container_clone":
|
||||
logging.debug("container_clone")
|
||||
# TODO: set clone_IP to DNSMASQ
|
||||
subprocess.call("/usr/bin/lxc-clone -o %(hostname)s -n %(clone_hostname)s" % task['parameters'],
|
||||
shell=True)
|
||||
# TODO: check container status
|
||||
# todo: held job if status not 0
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_mx_add":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
if self.Request_tast_update(task['task_id'])['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_mx_delete":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
if self.Request_tast_update(task['task_id'])['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_web_add":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
nginx.Nginx().vhost_add(task['parameters']['vhost_id'], task['parameters']['vhost'],
|
||||
task['parameters']['container_ip'])
|
||||
nginx.Service().reload()
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
if task['method'] == "service_web_delete":
|
||||
logging.debug("service_web_delete")
|
||||
# todo: held job if status not 0
|
||||
|
||||
print
|
||||
print task
|
||||
|
||||
nginx.Nginx().vhost_delete(task['parameters']['container_ip'], task['parameters']['vhost_id'])
|
||||
nginx.Service().reload()
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_web_update":
|
||||
logging.debug("create container")
|
||||
# todo: held job if status not 0
|
||||
if self.Request_tast_update(task['task_id'])['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_ssh_allow":
|
||||
logging.debug("service_ssh_allow")
|
||||
# todo: held job if status not 0
|
||||
|
||||
values = (self.interface2ip(), task['parameters']['port'], task['parameters']['container_ip'], 22)
|
||||
os.popen("ufw allow %s" % task['parameters']['port'])
|
||||
os.popen("iptables -t nat -I PREROUTING -p tcp -d %s --dport %s -j DNAT --to %s:%s" % values)
|
||||
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
if task['method'] == "service_ssh_deny":
|
||||
logging.debug("service_ssh_deny")
|
||||
# todo: held job if status not 0
|
||||
logging.debug("------")
|
||||
logging.debug(task)
|
||||
os.popen("ufw deny %s" % task['parameters']['port'])
|
||||
logging.debug("------")
|
||||
|
||||
task_update_result = TCPClient().request(
|
||||
Request().build("task_update", self.request_auth, {"task_id": task['task_id'], "status": 0}))
|
||||
if task_update_result['status'] == 0:
|
||||
return True
|
||||
return False
|
||||
|
||||
else:
|
||||
if task['status'] == 4:
|
||||
print "auth fail"
|
||||
else:
|
||||
print "structure version not supported"
|
||||
|
||||
return None
|
||||
|
|
@ -137,14 +137,19 @@ while True:
|
|||
# Create new virtual machine
|
||||
if task.get('task') == 'vm_create':
|
||||
nodeclient.task_status_update(task['id'], 1)
|
||||
|
||||
vm_id = task['plain']['vm_id']
|
||||
|
||||
# TODO: if container doesn't exists then complete task and report about this fact
|
||||
p = task['plain']
|
||||
try:
|
||||
# автоматически определяем подходящий сетевой интерфейс исходя из имеющегося ipv4
|
||||
interface = Detect().get_suitable_interface(p['ipv4'])
|
||||
# interface = settings.get('node', 'interface')
|
||||
|
||||
# автоматически определяем подходящий сетевой интерфейс исходя из имеющегося ipv4
|
||||
|
||||
interface = Detect().get_suitable_interface(p['ipv4'])
|
||||
if not interface:
|
||||
interface = settings.get('node', 'interface')
|
||||
|
||||
try:
|
||||
qemu.QEMU().create(
|
||||
p['cores'], p['memory'], p['storage'], p['swap'], p['hostname'],
|
||||
p['ipv4'], p['ipv4_gateway'], p['dns1'], p['dns2'], p['password'],
|
||||
|
|
|
|||
|
|
@ -1,17 +1,35 @@
|
|||
#!/usr/bin/env python
|
||||
# coding: utf-8
|
||||
|
||||
from SWSCloudNode import Node
|
||||
from SWSCloudNode import lxc
|
||||
import requests
|
||||
from SWSCloudNode import Node, StatisticsReporter
|
||||
from SWSCloudNode import LXC
|
||||
from SWSCloudNode import Qemu
|
||||
from SWSCloudNode import QemuStats
|
||||
|
||||
cls_lxc = lxc.lxc()
|
||||
cls_lxc = LXC()
|
||||
cls_qemu = Qemu()
|
||||
cls_node = Node()
|
||||
|
||||
containers = cls_lxc.list()
|
||||
|
||||
for container in containers:
|
||||
for container in cls_lxc.list():
|
||||
# print container
|
||||
info = cls_lxc.info(container)
|
||||
info['container_id'] = info['name']
|
||||
print cls_node.report_container_stats(info['container_id'], info)
|
||||
# print info
|
||||
|
||||
|
||||
"""
|
||||
curl localhost:8089/node_stats/v1/compute/vms/04ea5600-89c6-11e6-b1e1-fb8145d56ed7 -X POST --header 'Content-Type: application/json'
|
||||
"""
|
||||
|
||||
vms = cls_qemu.list().get('online')
|
||||
for vm in vms:
|
||||
dom = vms.get(vm).get('hostname')
|
||||
|
||||
data = dict(
|
||||
nework=QemuStats().network(dom),
|
||||
cpu=QemuStats().cpu(dom),
|
||||
memory=QemuStats().memory(dom))
|
||||
|
||||
print StatisticsReporter().send_vm_statistics(dom, data)
|
||||
|
|
|
|||
9
setup.py
9
setup.py
|
|
@ -4,13 +4,13 @@ from setuptools import setup
|
|||
|
||||
setup(
|
||||
name='SWSCloudNode',
|
||||
version='3.1.7',
|
||||
version='3.3.0',
|
||||
author='Vyacheslav Anzhiganov',
|
||||
author_email='vanzhiganov@ya.ru',
|
||||
packages=[
|
||||
'SWSCloudNode',
|
||||
'SWSCloudNode.lxc',
|
||||
'SWSCloudNode.qemu',
|
||||
'SWSCloudNode.compute.lxc',
|
||||
'SWSCloudNode.compute.qemu',
|
||||
],
|
||||
scripts=[
|
||||
'cloud_node_agent.py',
|
||||
|
|
@ -18,6 +18,7 @@ setup(
|
|||
],
|
||||
install_requires=[
|
||||
'requests',
|
||||
'netaddr==0.7.18'
|
||||
'netaddr==0.7.18',
|
||||
'libvirt-python==2.2.0',
|
||||
],
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue