forked from open-lambda/open-lambda
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver.py
More file actions
132 lines (108 loc) · 3.32 KB
/
Copy pathserver.py
File metadata and controls
132 lines (108 loc) · 3.32 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
import os, sys, json, argparse, importlib, traceback
import tornado.ioloop
import tornado.web
import tornado.httpserver
import tornado.netutil
HOST_DIR = '/host'
PKGS_DIR = '/packages'
HANDLER_DIR = '/handler'
sys.path.append(PKGS_DIR)
sys.path.append(HANDLER_DIR)
FS_PATH = os.path.join(HOST_DIR, 'fs.sock')
SOCK_PATH = os.path.join(HOST_DIR, 'ol.sock')
STDOUT_PATH = os.path.join(HOST_DIR, 'stdout')
STDERR_PATH = os.path.join(HOST_DIR, 'stderr')
PROCESSES_DEFAULT = 10
initialized = False
parser = argparse.ArgumentParser(description='Listen and serve cache requests or lambda invocations.')
parser.add_argument('--cache', action='store_true', default=False, help='Begin as a cache entry.')
# run after forking into sandbox
def init():
global initialized, lambda_func
if initialized:
return
# assume submitted .py file is /handler/lambda_func.py
import lambda_func
initialized = True
class SockFileHandler(tornado.web.RequestHandler):
def post(self):
try:
data = self.request.body
try :
event = json.loads(data)
except:
self.set_status(400)
self.write('bad POST data: "%s"'%str(data))
return
self.write(json.dumps(lambda_func.handler(event)))
except Exception:
self.set_status(500) # internal error
self.write(traceback.format_exc())
tornado_app = tornado.web.Application([
(r".*", SockFileHandler),
])
# listen on sock file with Tornado
def lambda_server():
global HOST_PIPE
init()
server = tornado.httpserver.HTTPServer(tornado_app)
socket = tornado.netutil.bind_unix_socket(SOCK_PATH)
server.add_socket(socket)
# notify worker server that we are ready through stdout
# flush is necessary, and don't put it after tornado start; won't work
with open('/host/server_pipe', 'w') as pipe:
pipe.write('ready')
tornado.ioloop.IOLoop.instance().start()
server.start(PROCESSES_DEFAULT)
# listen for fds to forkenter
def cache_loop():
import ns
signal = "cache"
r = -1
count = 0
# only child meant to serve ever escapes the loop
while r != 0 or signal == "cache":
if r == 0:
print('RESET')
flush()
ns.reset()
print('LISTENING')
flush()
data = ns.fdlisten(FS_PATH).split()
flush()
mods = data[:-1]
signal = data[-1]
r = ns.forkenter()
sys.stdout.flush()
if r == 0:
redirect()
# import modules
for mod in mods:
print('importing: %s' % mod)
try:
globals()[mod] = importlib.import_module(mod)
except Exception as e:
print('failed to import %s with: %s' % (mod, e))
print('signal: %s' % signal)
flush()
print('')
flush()
count += 1
print('SERVING HANDLERS')
flush()
lambda_server()
def flush():
sys.stdout.flush()
sys.stderr.flush()
def redirect():
sys.stdout.close()
sys.stderr.close()
sys.stdout = open(STDOUT_PATH, 'w')
sys.stderr = open(STDERR_PATH, 'w')
if __name__ == '__main__':
args = parser.parse_args()
redirect()
if args.cache:
cache_loop()
else:
lambda_server()