pyrox.dev
1#!/usr/bin/env python
2import uuid
3import time
4import subprocess
5import os
6
7import sys
8
9from mesos.interface import Scheduler
10from mesos.native import MesosSchedulerDriver
11from mesos.interface import mesos_pb2
12
13def log(msg):
14 process = subprocess.Popen("systemd-cat", stdin=subprocess.PIPE)
15 (out,err) = process.communicate(msg)
16
17class NixosTestScheduler(Scheduler):
18 def __init__(self):
19 self.master_ip = sys.argv[1]
20 self.download_uri = sys.argv[2]
21
22 def resourceOffers(self, driver, offers):
23 log("XXX got resource offer")
24
25 offer = offers[0]
26 task = self.new_task(offer)
27 uri = task.command.uris.add()
28 uri.value = self.download_uri
29 task.command.value = "cat test.result"
30 driver.launchTasks(offer.id, [task])
31
32 def statusUpdate(self, driver, update):
33 log("XXX status update")
34 if update.state == mesos_pb2.TASK_FAILED:
35 log("XXX test task failed with message: " + update.message)
36 driver.stop()
37 sys.exit(1)
38 elif update.state == mesos_pb2.TASK_FINISHED:
39 driver.stop()
40 sys.exit(0)
41
42 def new_task(self, offer):
43 task = mesos_pb2.TaskInfo()
44 id = uuid.uuid4()
45 task.task_id.value = str(id)
46 task.slave_id.value = offer.slave_id.value
47 task.name = "task {}".format(str(id))
48
49 cpus = task.resources.add()
50 cpus.name = "cpus"
51 cpus.type = mesos_pb2.Value.SCALAR
52 cpus.scalar.value = 0.1
53
54 mem = task.resources.add()
55 mem.name = "mem"
56 mem.type = mesos_pb2.Value.SCALAR
57 mem.scalar.value = 32
58
59 return task
60
61if __name__ == '__main__':
62 log("XXX framework started")
63
64 framework = mesos_pb2.FrameworkInfo()
65 framework.user = "root"
66 framework.name = "nixos-test-framework"
67 driver = MesosSchedulerDriver(
68 NixosTestScheduler(),
69 framework,
70 sys.argv[1] + ":5050"
71 )
72 driver.run()