xos/synchronizers/hpc/hpc_watcher.py

"""
    hpc_watcher.py

    Daemon to watch the health of HPC and RR instances.

    This deamon uses HpcHealthCheck objects in the Data Model to conduct
    periodic tests of HPC and RR nodes. Two types of Health Checks are
    supported:

       kind="dns": checks the request routers to make sure that a DNS
         name is resolveable and returns the right kind of records.

         resource_name should be set to the domain name to lookup.

         result_contains is option and can be used to hold "A", "CNAME", or
            a particular address or hostname that should be contained in the
            query's answer.

       kind="http": checks the hpc nodes to make sure that a URL can be
         retrieved from the node.

         resource_name should be set to the HostName:Url to fetch. For
         example, cdn-stream.htm.fiu.edu:/hft2441/intro.mp4

     In addition to the above, HPC heartbeat probes are conducted, similar to
     the ones that dnsredir conducts.

     The results of health checks are stored in a tag attached to the Instance
     the healthcheck was conducted against. If all healthchecks of a particular
     variety were successful for a instance, then "success" will be stored in
     the tag. Otherwise, the first healthcheck to fail will be stored in the
     tag.

     Ubuntu prereqs:
         apt-get install python-pycurl
         pip install dnslib
"""

import os
import socket
import sys
sys.path.append("/opt/xos")
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "xos.settings")
import django
from django.contrib.contenttypes.models import ContentType
from core.models import *
from services.hpc.models import *
from services.requestrouter.models import *
django.setup()
import time
import pycurl
import traceback
import json
from StringIO import StringIO

from dnslib.dns import DNSRecord,DNSHeader,DNSQuestion,QTYPE
from dnslib.digparser import DigParser

from threading import Thread, Condition

"""
from dnslib import *
q = DNSRecord(q=DNSQuestion("cdn-stream.htm.fiu.edu"))
a_pkt = q.send("150.135.65.10", tcp=False, timeout=10)
a = DNSRecord.parse(a_pkt)

from dnslib import *
q = DNSRecord(q=DNSQuestion("onlab.vicci.org"))
a_pkt = q.send("150.135.65.10", tcp=False, timeout=10)
a = DNSRecord.parse(a_pkt)
"""

class WorkQueue:
    def __init__(self):
        self.job_cv = Condition()
        self.jobs = []
        self.result_cv = Condition()
        self.results = []
        self.outstanding = 0

    def get_job(self):
        self.job_cv.acquire()
        while not self.jobs:
            self.job_cv.wait()
        result = self.jobs.pop()
        self.job_cv.release()
        return result

    def submit_job(self, job):
        self.job_cv.acquire()
        self.jobs.append(job)
        self.job_cv.notify()
        self.job_cv.release()
        self.outstanding = self.outstanding + 1

    def get_result(self):
        self.result_cv.acquire()
        while not self.results:
            self.result_cv.wait()
        result = self.results.pop()
        self.result_cv.release()
        self.outstanding = self.outstanding - 1
        return result

    def submit_result(self, result):
        self.result_cv.acquire()
        self.results.append(result)
        self.result_cv.notify()
        self.result_cv.release()

class DnsResolver(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue
        self.daemon = True
        self.start()

    def run(self):
        while True:
            job = self.queue.get_job()
            self.handle_job(job)
            self.queue.submit_result(job)

    def handle_job(self, job):
        domain = job["domain"]
        server = job["server"]
        port = job["port"]
        result_contains = job.get("result_contains", None)

        try:
            q = DNSRecord(q=DNSQuestion(domain)) #, getattr(QTYPE,"A")))

            a_pkt = q.send(server, port, tcp=False, timeout=10)
            a = DNSRecord.parse(a_pkt)

            found_record = False
            for record in a.rr:
                if (not result_contains):
                    QTYPE_A = getattr(QTYPE,"A")
                    QTYPE_CNAME = getattr(QTYPE, "CNAME")
                    if ((record.rtype==QTYPE_A) or (record.qtype==QTYPE_CNAME)):
                        found_record = True
                else:
                    tmp = QTYPE.get(record.rtype) + str(record.rdata)
                    if (result_contains in tmp):
                        found_record = True

            if not found_record:
                if result_contains:
                    job["status"] =  "%s,No %s records" % (domain, result_contains)
                else:
                    job["status"] =  "%s,No A or CNAME records" % domain

                return

        except Exception, e:
            job["status"] = "%s,Exception: %s" % (domain, str(e))
            return

        job["status"] = "success"

class HpcHeartbeat(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue
        self.daemon = True
        self.start()

    def run(self):
        while True:
            job = self.queue.get_job()
            self.handle_job(job)
            self.queue.submit_result(job)

    def curl_error_message(self, e):
        if e.args[0] == 6:
            return "couldn't resolve host"
        if e.args[0] == 7:
            return "failed to connect"
        return "curl error %d" % e.args[0]

    def handle_job(self, job):
        server = job["server"]
        port = job["port"]

        try:
            buffer = StringIO()
            c = pycurl.Curl()

            c.setopt(c.URL, "http://%s:%s/heartbeat" % (server, port))
            c.setopt(c.WRITEDATA, buffer)
            c.setopt(c.HTTPHEADER, ['host: hpc-heartbeat', 'X-heartbeat: 1'])
            c.setopt(c.TIMEOUT, 10)
            c.setopt(c.CONNECTTIMEOUT, 10)
            c.setopt(c.NOSIGNAL, 1)

            try:
                c.perform()
                response_code = c.getinfo(c.RESPONSE_CODE)
            except Exception, e:
                #traceback.print_exc()
                job["status"] = self.curl_error_message(e)
                return
            finally:
                c.close()

            if response_code != 200:
                job["status"] = "error response %d" % response_code
                return

        except Exception, e:
            job["status"] = "Exception: %s" % str(e)
            return

        job["status"] = "success"

class HpcFetchUrl(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue
        self.daemon = True
        self.start()

    def run(self):
        while True:
            job = self.queue.get_job()
            self.handle_job(job)
            self.queue.submit_result(job)

    def curl_error_message(self, e):
        if e.args[0] == 6:
            return "couldn't resolve host"
        if e.args[0] == 7:
            return "failed to connect"
        return "curl error %d" % e.args[0]

    def handle_job(self, job):
        server = job["server"]
        port = job["port"]
        url = job["url"]
        domain = job["domain"]

        def progress(download_t, download_d, upload_t, upload_d):
            # limit download size to a megabyte
            if (download_d > 1024*1024):
                return 1
            else:
                return 0

        try:
            buffer = StringIO()
            c = pycurl.Curl()

            c.setopt(c.URL, "http://%s:%s/%s" % (server, port, url))
            c.setopt(c.WRITEDATA, buffer)
            c.setopt(c.HTTPHEADER, ['host: ' + domain])
            c.setopt(c.TIMEOUT, 10)
            c.setopt(c.CONNECTTIMEOUT, 10)
            c.setopt(c.NOSIGNAL, 1)
            c.setopt(c.NOPROGRESS, 0)
            c.setopt(c.PROGRESSFUNCTION, progress)

            try:
                try:
                    c.perform()
                except Exception, e:
                    # prevent callback abort from raising exception
                    if (e.args[0] != pycurl.E_ABORTED_BY_CALLBACK):
                        raise
                response_code = c.getinfo(c.RESPONSE_CODE)
                bytes_downloaded = int(c.getinfo(c.SIZE_DOWNLOAD))
                total_time = float(c.getinfo(c.TOTAL_TIME))
            except Exception, e:
                #traceback.print_exc()
                job["status"] = self.curl_error_message(e)
                return
            finally:
                c.close()

            if response_code != 200:
                job["status"] = "error response %s" %  str(response_code)
                return

        except Exception, e:
            #traceback.print_exc()
            job["status"] = "Exception: %s" % str(e)
            return

        job["status"] = "success"
        job["bytes_downloaded"] = bytes_downloaded
        job["total_time"] = total_time

class WatcherWorker(Thread):
    def __init__(self, queue):
        Thread.__init__(self)
        self.queue = queue
        self.daemon = True
        self.start()

    def run(self):
        while True:
            job = self.queue.get_job()
            self.handle_job(job)
            self.queue.submit_result(job)

    def curl_error_message(self, e):
        if e.args[0] == 6:
            return "couldn't resolve host"
        if e.args[0] == 7:
            return "failed to connect"
        return "curl error %d" % e.args[0]

    def handle_job(self, job):
        server = job["server"]
        port = job["port"]

        try:
            buffer = StringIO()
            c = pycurl.Curl()

            c.setopt(c.URL, "http://%s:%s/" % (server, port))
            c.setopt(c.WRITEDATA, buffer)
            c.setopt(c.TIMEOUT, 10)
            c.setopt(c.CONNECTTIMEOUT, 10)
            c.setopt(c.NOSIGNAL, 1)

            try:
                c.perform()
                response_code = c.getinfo(c.RESPONSE_CODE)
            except Exception, e:
                #traceback.print_exc()
                job["status"] = json.dumps( {"status": self.curl_error_message(e)} )
                return
            finally:
                c.close()

            if response_code != 200:
                job["status"] = json.dumps( {"status": "error response %d" % response_code} )
                return

            d = json.loads(buffer.getvalue())
            d["status"] = "success";
            job["status"] = json.dumps(d)

        except Exception, e:
            job["status"] = json.dumps( {"status": "Exception: %s" % str(e)} )
            return

class BaseWatcher(Thread):
    def __init__(self):
        Thread.__init__(self)
        self.daemon = True

    def get_public_ip(self, service, instance):
        network_name = None
        if "hpc" in instance.slice.name:
            network_name = getattr(service, "watcher_hpc_network", None)
        elif "demux" in instance.slice.name:
            network_name = getattr(service, "watcher_dnsdemux_network", None)
        elif "redir" in instance.slice.name:
            network_name = getattr(service, "watcher_dnsredir_network", None)

        if network_name and network_name.lower()=="nat":
            return None

        if (network_name is None) or (network_name=="") or (network_name.lower()=="public"):
            return instance.get_public_ip()

        for ns in instance.ports.all():
            if (ns.ip) and (ns.network.name==network_name):
                return ns.ip

        raise ValueError("Couldn't find network %s" % str(network_name))

    def set_status(self, instance, service, kind, msg, check_error=True):
        #print instance.node.name, kind, msg
        if check_error:
            instance.has_error = (msg!="success")

        instance_type = ContentType.objects.get_for_model(instance)

        t = Tag.objects.filter(service=service, name=kind+".msg", content_type__pk=instance_type.id, object_id=instance.id)
        if t:
            t=t[0]
            if (t.value != msg):
                t.value = msg
                t.save()
        else:
            Tag(service=service, name=kind+".msg", content_object = instance, value=msg).save()

        t = Tag.objects.filter(service=service, name=kind+".time", content_type__pk=instance_type.id, object_id=instance.id)
        if t:
            t=t[0]
            t.value = str(time.time())
            t.save()
        else:
            Tag(service=service, name=kind+".time", content_object = instance, value=str(time.time())).save()

    def get_service_slices(self, service, kind=None):
        try:
            slices = service.slices.all()
        except:
            # buggy data model
            slices = service.service.all()

        if kind:
            return [x for x in slices if (kind in x.name)]
        else:
            return list(slices)

class RRWatcher(BaseWatcher):
    def __init__(self):
        BaseWatcher.__init__(self)

        self.resolver_queue = WorkQueue()
        for i in range(0,10):
            DnsResolver(queue = self.resolver_queue)

    def check_request_routers(self, service, instances):
        for instance in instances:
            instance.has_error = False

            try:
                ip = self.get_public_ip(service, instance)
            except Exception, e:
                self.set_status(instance, service, "watcher.DNS", "exception: %s" % str(e))
                continue
            if not ip:
                try:
                    ip = socket.gethostbyname(instance.node.name)
                except:
                    self.set_status(instance, service, "watcher.DNS", "dns resolution failure")
                    continue

            if not ip:
                self.set_status(instance, service, "watcher.DNS", "no IP address")
                continue

            checks = HpcHealthCheck.objects.filter(kind="dns")
            if not checks:
                self.set_status(instance, service, "watcher.DNS", "no DNS HealthCheck tests configured")

            for check in checks:
                self.resolver_queue.submit_job({"domain": check.resource_name, "server": ip, "port": 53, "instance": instance, "result_contains": check.result_contains})

        while self.resolver_queue.outstanding > 0:
            result = self.resolver_queue.get_result()
            instance = result["instance"]
            if (result["status"]!="success") and (not instance.has_error):
                self.set_status(instance, service, "watcher.DNS", result["status"])

        for instance in instances:
            if not instance.has_error:
                self.set_status(instance, service, "watcher.DNS", "success")

    def run_once(self):
        for hpcService in HpcService.objects.all():
            for slice in self.get_service_slices(hpcService, "dnsdemux"):
                self.check_request_routers(hpcService, slice.instances.all())

        for rrService in RequestRouterService.objects.all():
            for slice in self.get_service_slices(rrService, "dnsdemux"):
                self.check_request_routers(rrService, slice.instances.all())

    def run(self):
        while True:
            self.run_once()
            time.sleep(10)

            django.db.reset_queries()

class HpcProber(BaseWatcher):
    def __init__(self):
        BaseWatcher.__init__(self)

        self.heartbeat_queue = WorkQueue()
        for i in range(0, 10):
            HpcHeartbeat(queue = self.heartbeat_queue)

    def probe_hpc(self, service, instances):
        for instance in instances:
            instance.has_error = False

            self.heartbeat_queue.submit_job({"server": instance.node.name, "port": 8009, "instance": instance})

        while self.heartbeat_queue.outstanding > 0:
            result = self.heartbeat_queue.get_result()
            instance = result["instance"]
            if (result["status"]!="success") and (not instance.has_error):
                self.set_status(instance, service, "watcher.HPC-hb", result["status"])

        for instance in instances:
            if not instance.has_error:
                self.set_status(instance, service, "watcher.HPC-hb", "success")

    def run_once(self):
        for hpcService in HpcService.objects.all():
            for slice in self.get_service_slices(hpcService, "hpc"):
                self.probe_hpc(hpcService, slice.instances.all())

    def run(self):
        while True:
            self.run_once()
            time.sleep(10)

            django.db.reset_queries()

class HpcFetcher(BaseWatcher):
    def __init__(self):
        BaseWatcher.__init__(self)

        self.fetch_queue = WorkQueue()
        for i in range(0, 10):
            HpcFetchUrl(queue = self.fetch_queue)

    def fetch_hpc(self, service, instances):
        for instance in instances:
            instance.has_error = False
            instance.url_status = []

            checks = HpcHealthCheck.objects.filter(kind="http")
            if not checks:
                self.set_status(instance, service, "watcher.HPC-fetch", "no HTTP HealthCheck tests configured")

            for check in checks:
                if (not check.resource_name) or (":" not in check.resource_name):
                    self.set_status(instance, service, "watcher.HPC-fetch", "malformed resource_name: " + str(check.resource_name))
                    break

                (domain, url) = check.resource_name.split(":",1)

                self.fetch_queue.submit_job({"server": instance.node.name, "port": 80, "instance": instance, "domain": domain, "url": url})

        while self.fetch_queue.outstanding > 0:
            result = self.fetch_queue.get_result()
            instance = result["instance"]
            if (result["status"] == "success"):
                instance.url_status.append( (result["domain"] + result["url"], "success", result["bytes_downloaded"], result["total_time"]) )
            if (result["status"]!="success") and (not instance.has_error):
                self.set_status(instance, service, "watcher.HPC-fetch", result["status"])

        for instance in instances:
            self.set_status(instance, service, "watcher.HPC-fetch-urls", json.dumps(instance.url_status), check_error=False)
            if not instance.has_error:
                self.set_status(instance, service, "watcher.HPC-fetch", "success")

    def run_once(self):
        for hpcService in HpcService.objects.all():
            for slice in self.get_service_slices(hpcService, "hpc"):
                try:
                    self.fetch_hpc(hpcService, slice.instances.all())
                except:
                    traceback.print_exc()

    def run(self):
        while True:
            self.run_once()
            time.sleep(10)

            django.db.reset_queries()

class WatcherFetcher(BaseWatcher):
    def __init__(self):
        BaseWatcher.__init__(self)

        self.fetch_queue = WorkQueue()
        for i in range(0, 10):
             WatcherWorker(queue = self.fetch_queue)

    def fetch_watcher(self, service, instances):
        for instance in instances:
            try:
                ip = self.get_public_ip(service, instance)
            except Exception, e:
                self.set_status(instance, service, "watcher.watcher", json.dumps({"status": "exception: %s" % str(e)}) )
                continue
            if not ip:
                try:
                    ip = socket.gethostbyname(instance.node.name)
                except:
                    self.set_status(instance, service, "watcher.watcher", json.dumps({"status": "dns resolution failure"}) )
                    continue

            if not ip:
                self.set_status(instance, service, "watcher.watcher", json.dumps({"status": "no IP address"}) )
                continue

            port = 8015
            if ("redir" in instance.slice.name):
                port = 8016
            elif ("demux" in instance.slice.name):
                port = 8017

            self.fetch_queue.submit_job({"server": ip, "port": port, "instance": instance})

        while self.fetch_queue.outstanding > 0:
            result = self.fetch_queue.get_result()
            instance = result["instance"]
            self.set_status(instance, service, "watcher.watcher", result["status"])

    def run_once(self):
        for hpcService in HpcService.objects.all():
            for slice in self.get_service_slices(hpcService):
                self.fetch_watcher(hpcService, slice.instances.all())

    def run(self):
        while True:
            self.run_once()
            time.sleep(10)

            django.db.reset_queries()


if __name__ == "__main__":
    if "--once" in sys.argv:
        RRWatcher().run_once()
        HpcProber().run_once()
        HpcFetcher().run_once()
        WatcherFetcher().run_once()
    else:
        RRWatcher().start()
        HpcProber().start()
        HpcFetcher().start()
        WatcherFetcher().start()

        print "Running forever..."
        while True:
            time.sleep(60)