forked from open-lambda/open-lambda
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathserver_legacy.py
More file actions
147 lines (119 loc) · 3.58 KB
/
Copy pathserver_legacy.py
File metadata and controls
147 lines (119 loc) · 3.58 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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
'''
Python Runtime for Docker
Note: SOCK doesn't use this anymore (it uses server.py instead), but
this is still here because we haven't updated docker.go yet.
'''
#pylint: disable=invalid-name,line-too-long,global-statement
import os
import sys
import json
import argparse
import importlib
import 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')
SERVER_PIPE_PATH = os.path.join(HOST_DIR, 'server_pipe')
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, f
if initialized:
return
# assume submitted .py file is /handler/f.py
import f
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(f'bad POST data: "{data}"')
return
self.write(json.dumps(f.f(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():
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(SERVER_PIPE_PATH, 'w', encoding='utf-8') 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]
ret_val = ns.forkenter()
sys.stdout.flush()
if ret_val == 0:
redirect()
# import modules
for mod in mods:
print(f'importing: {mod}')
try:
globals()[mod] = importlib.import_module(mod)
except Exception as err:
print(f'failed to import {mod} with: {err}')
print(f'signal: {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()