| #!/usr/bin/env python |
| |
| import mesos |
| import os |
| import sys |
| import time |
| import httplib |
| import Queue |
| import threading |
| |
| from optparse import OptionParser |
| from subprocess import * |
| from socket import gethostname |
| |
| MIN_SERVERS = 1 |
| START_THRESHOLD = 25 |
| KILL_THRESHOLD = 5 |
| HAPROXY_EXE = "/root/haproxy-1.3.20/haproxy" #EC2 |
| #HAPROXY_EXE = "/scratch/haproxy-1.3.25/haproxy" #Rcluster |
| |
| class ApacheWebFWScheduler(mesos.Scheduler): |
| def __init__(self): |
| mesos.Scheduler.__init__(self) |
| self.lock = threading.RLock() |
| self.id = 0 |
| self.haproxy = -1 |
| self.reconfigs = 0 |
| self.servers = {} |
| self.overloaded = False |
| |
| def registered(self, driver, fid): |
| print "Mesos haproxy+apache scheduler registered as framework #%s" % fid |
| self.driver = driver |
| |
| def getFrameworkName(self, driver): |
| return "haproxy+apache" |
| |
| def getExecutorInfo(self, driver): |
| execPath = os.path.join(os.getcwd(), "startapache.sh") |
| return mesos.ExecutorInfo(execPath, "") |
| |
| def reconfigure(self): |
| print "reconfiguring haproxy" |
| name = "/tmp/haproxy.conf.%d" % self.reconfigs |
| with open(name, 'w') as config: |
| with open('haproxy.config.template', 'r') as template: |
| for line in template: |
| config.write(line) |
| for id, host in self.servers.iteritems(): |
| config.write(" ") |
| config.write("server %d %s:80 check\n" % (id, host)) |
| |
| cmd = [] |
| if self.haproxy != -1: |
| cmd = [HAPROXY_EXE, |
| "-f", |
| name, |
| "-sf", |
| str(self.haproxy.pid)] |
| else: |
| cmd = [HAPROXY_EXE, |
| "-f", |
| name] |
| |
| self.haproxy = Popen(cmd, shell = False) |
| self.reconfigs += 1 |
| |
| def resourceOffer(self, driver, oid, slave_offers): |
| print "Got resource offer %s with %s slots." % (oid, len(slave_offers)) |
| self.lock.acquire() |
| tasks = [] |
| for offer in slave_offers: |
| if offer.host in self.servers.values(): |
| print "Rejecting slot on host " + offer.host + " because we've launched a server on that machine already." |
| #print "self.servers currently looks like: " + str(self.servers) |
| elif not self.overloaded and len(self.servers) > 0: |
| print "Rejecting slot because we've launched enough tasks." |
| elif int(offer.params['mem']) < 1024: |
| print "Rejecting offer because it doesn't contain enough memory (it has " + offer.params['mem'] + " and we need 1024mb." |
| elif int(offer.params['cpus']) < 1: |
| print "Rejecting offer because it doesn't contain enough CPUs." |
| else: |
| print "Offer is for " + offer.params['cpus'] + " CPUS and " + offer.params["mem"] + " MB on host " + offer.host |
| params = {"cpus": "1", "mem": "1024"} |
| td = mesos.TaskDescription(self.id, offer.slaveId, "server %s" % self.id, params, "") |
| print "Accepting task, id=" + str(self.id) + ", params: " + params['cpus'] + " CPUS, and " + params['mem'] + " MB, on node " + offer.host |
| tasks.append(td) |
| self.servers[self.id] = offer.host |
| self.id += 1 |
| self.overloaded = False |
| driver.replyToOffer(oid, tasks, {"timeout":"1"}) |
| #driver.replyToOffer(oid, tasks, {}) |
| print "done with resourceOffer()" |
| self.lock.release() |
| |
| def statusUpdate(self, driver, status): |
| print "received status update from taskID " + str(status.taskId) + ", with state: " + str(status.state) |
| reconfigured = False |
| self.lock.acquire() |
| if status.taskId in self.servers.keys(): |
| if status.state == mesos.TASK_STARTING: |
| print "Task " + str(status.taskId) + " reported that it is STARTING." |
| del self.servers[status.taskId] |
| self.reconfigure() |
| reconfigured = True |
| if status.state == mesos.TASK_RUNNING: |
| print "Task " + str(status.taskId) + " reported that it is RUNNING, reconfiguring haproxy to include it in webfarm now." |
| self.reconfigure() |
| reconfigured = True |
| if status.state == mesos.TASK_FINISHED: |
| del self.servers[status.taskId] |
| print "Task " + str(status.taskId) + " reported FINISHED (state " + status.state + ")." |
| self.reconfigure() |
| reconfigured = True |
| if status.state == mesos.TASK_FAILED: |
| print "Task " + str(status.taskId) + " reported that it FAILED!" |
| del self.servers[status.taskId] |
| self.reconfigure() |
| reconfigured = True |
| if status.state == mesos.TASK_KILLED: |
| print "Task " + str(status.taskId) + " reported that it was KILLED!" |
| del self.servers[status.taskId] |
| self.reconfigure() |
| reconfigured = True |
| if status.state == mesos.TASK_LOST: |
| print "Task " + str(status.taskId) + " reported was LOST!" |
| del self.servers[status.taskId] |
| self.reconfigure() |
| reconfigured = True |
| self.lock.release() |
| if reconfigured: |
| None#driver.reviveOffers() |
| print "done in statusupdate" |
| |
| def scaleUp(self): |
| print "SCALING UP" |
| self.lock.acquire() |
| self.overloaded = True |
| self.lock.release() |
| |
| def scaleDown(self, id): |
| print "SCALING DOWN (removing server %d)" % id |
| kill = False |
| self.lock.acquire() |
| if self.overloaded: |
| self.overloaded = False |
| else: |
| kill = True |
| self.lock.release() |
| if kill: |
| self.driver.killTask(id) |
| |
| |
| def monitor(sched): |
| print "in MONITOR()" |
| while True: |
| time.sleep(1) |
| print "done sleeping" |
| try: |
| conn = httplib.HTTPConnection("r3.millennium.berkeley.edu:9001") |
| print "done creating connection" |
| conn.request("GET", "/stats;csv") |
| print "done with request()" |
| res = conn.getresponse() |
| print "testing response status" |
| if (res.status != 200): |
| print "response != 200" |
| continue |
| else: |
| print "got some stats" |
| data = res.read() |
| lines = data.split('\n')[2:-2] |
| |
| data = data.split('\n') |
| data = data[1].split(',') |
| |
| if int(data[33]) >= START_THRESHOLD: |
| sched.scaleUp() |
| elif int(data[4]) <= KILL_THRESHOLD: |
| minload, minid = (sys.maxint, 0) |
| for l in lines: |
| cols = l.split(',') |
| id = int(cols[1]) |
| load = int(cols[4]) |
| if load < minload: |
| minload = load |
| minid = id |
| |
| if len(lines) > MIN_SERVERS and minload == 0: |
| sched.scaleDown(minid) |
| |
| conn.close() |
| except Exception, e: |
| print "exception in monitor()" |
| continue |
| print "done in MONITOR()" |
| |
| if __name__ == "__main__": |
| parser = OptionParser(usage = "Usage: %prog mesos_master") |
| |
| (options,args) = parser.parse_args() |
| if len(args) < 1: |
| print >> sys.stderr, "At least one parameter required." |
| print >> sys.stderr, "Use --help to show usage." |
| exit(2) |
| |
| print "sched = ApacheWebFWScheduler()" |
| sched = ApacheWebFWScheduler() |
| |
| print "Connecting to mesos master %s" % args[0] |
| driver = mesos.MesosSchedulerDriver(sched, sys.argv[1]) |
| |
| threading.Thread(target = monitor, args=[sched]).start() |
| |
| driver.run() |
| |
| print "Scheduler finished!" |
