Add to_hex_string UDAF
[com/gs-lite.git] / bin / gshub3.py
1 #!/usr/bin/python3.6
2
3 # ------------------------------------------------
4 #   Copyright 2014 AT&T Intellectual Property
5 #   Licensed under the Apache License, Version 2.0 (the "License");
6 #   you may not use this file except in compliance with the License.
7 #   You may obtain a copy of the License at
8 #
9 #     http://www.apache.org/licenses/LICENSE-2.0
10 #
11 #   Unless required by applicable law or agreed to in writing, software
12 #   distributed under the License is distributed on an "AS IS" BASIS,
13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 #   See the License for the specific language governing permissions and
15 #   limitations under the License.
16 # -------------------------------------------
17
18 # Implementation of GSHUB REST service
19 # for announcement and discovery of gs instances, sources and sinks
20
21 from http.server import BaseHTTPRequestHandler, HTTPServer
22 from socketserver import ThreadingMixIn
23 import threading
24 import getopt
25 import sys
26 import re
27 import cgi
28 import socket
29 import json
30
31 # lis of URLS for all the REST calls we will serve
32 DISCOVER_INSTANCE_URL = '/v1/discover-instance'
33 DISCOVER_INITINSTANCE_URL = '/v1/discover-initialized-instance'
34 DISCOVER_SOURCE_URL =  '/v1/discover-source'
35 DISCOVER_SINK_URL =  '/v1/discover-sink'
36 DISCOVER_STARTPROCESSING_URL = '/v1/discover-start-processing'
37 ANNOUNCE_INSTANCE_URL =  '/v1/announce-instance'
38 ANNOUNCE_INITINSTANCE_URL =  '/v1/announce-initialized-instance'
39 ANNOUNCE_SOURCE_URL =  '/v1/announce-source'
40 ANNOUNCE_SINK_URL =  '/v1/announce-sink'
41 ANNOUNCE_STARTPROCESSING_URL = '/v1/announce-start-processing'
42 ANNOUNCE_STREAM_SUBSCRIPTION = '/v1/announce-stream-subscription'
43 ANNOUNCE_FTA_INSTANCE = '/v1/announce-fta-instance'
44 ANNOUNCE_METRICS = '/v1/log-metrics'
45
46 # gs instance endpoints
47 gs_instances = {}
48
49 # initialized gs instances
50 gs_init_instances = {}
51
52 # instances for which processing started
53 gs_startprocessing_instances = {}
54
55 # source endpoints
56 sources = {}
57
58 # sink endpoints
59 sinks = {}
60
61
62 # exctract endpoint information from json data
63 def extract_endpoint(data) :
64         name = ''
65         ip = ''
66         port = 0
67                 
68         try :
69                 doc = json.loads(str(data, 'utf-8'))    
70         except :
71                 print ('Invalid json message ' + str(data, 'utf-8'))
72                 return []
73         
74         for key in doc.keys() :
75                 if key == 'name' :
76                         name = doc[key]
77                 elif key == 'ip' :
78                         ip = doc[key]
79                         # validate ip address           
80                         try :
81                                 socket.inet_pton(socket.AF_INET, ip)
82                         except :
83                                 print ('Invalid IPV4 address ' + ip)
84                                 ip = ''
85                 elif key == 'port' :
86                         # validate port number
87                         try :
88                                 port = int(doc[key])
89                         except :
90                                 print ('Invalid port number ' + doc[key])
91                                 port = 0
92         
93         if name == '' or ip == '' or port == 0 :
94                 print ('Name, ip or port is missing from json message ' + str(doc))
95                 return []
96         
97
98         return [name, ip, port]
99
100
101 # extract instance name from json data
102 def extract_instance_name(data) :
103         name = ''
104         
105         try :
106                 doc = json.loads(str(data, 'utf-8'))    
107         except :
108                 print ('Invalid json message ' + str(data, "utf-8"))
109                 return ''
110         
111         for key in doc.keys() :
112                 if key == 'name' :
113                         name = doc[key]
114         
115         if name == '' :
116                 print ('Name field is missing in json message ' + str(doc))
117         elif (name in gs_instances) == False:
118                 print ('Attempt to announce the initialization or start of processing for unknown instance ' + name)
119                 name = ''
120                 
121         return name
122         
123 # handler for HTTP requests. We will override do_PORT and do_GET of BaseHTTPRequestHandler
124 class HTTPRequestHandler(BaseHTTPRequestHandler):
125
126         def do_POST(self):
127                 if re.search(ANNOUNCE_INSTANCE_URL, self.path) != None:         
128                         if self.headers.get_content_type() == 'application/json' :
129                                 # Find content length
130                                 content_len = 0
131                                 for i in range(len(self.headers.keys())):
132                                         if self.headers.keys()[i] == 'Content-Length' :
133                                                 content_len = int (self.headers.values()[i])
134                                                 break
135                                 if content_len != 0 :
136                                         # extract endpoint information
137                                         endpoint = extract_endpoint(self.rfile.read(content_len))
138                                         if endpoint == [] :
139                                                 self.send_response(400)
140                                         else :
141                                                 self.send_response(200)
142                                                 gs_instances[endpoint[0]] = [endpoint[1], endpoint[2]]
143                                 else :
144                                         self.send_response(400)
145
146                         else:
147                                 self.send_response(400)
148                         self.end_headers()              
149                         
150                 elif re.search(ANNOUNCE_INITINSTANCE_URL, self.path) != None:
151                         if self.headers.get_content_type() == 'application/json' :
152                                 # Find content length
153                                 content_len = 0
154                                 for i in range(len(self.headers.keys())):
155                                         if self.headers.keys()[i] == 'Content-Length' :
156                                                 content_len = int (self.headers.values()[i])
157                                                 break
158                                 if content_len != 0 :
159                                         # extract name of initialized gs instance                               
160                                         name = extract_instance_name(self.rfile.read(content_len))
161                                         if name == '' :
162                                                 self.send_response(400)
163                                         else :
164                                                 self.send_response(200)
165                                                 gs_init_instances[name] = 1
166                                 else :
167                                         self.send_response(400)
168
169                         else:
170                                 self.send_response(400)
171                         self.end_headers()      
172                         
173                 elif re.search(ANNOUNCE_SOURCE_URL, self.path) != None:
174                         if self.headers.get_content_type() == 'application/json' :
175                                 # Find content length
176                                 content_len = 0
177                                 for i in range(len(self.headers.keys())):
178                                         if self.headers.keys()[i] == 'Content-Length' :
179                                                 content_len = int (self.headers.values()[i])
180                                                 break
181                                 if content_len != 0 :
182                                         # extract endpoint information                          
183                                         endpoint = extract_endpoint(self.rfile.read(content_len))
184                                         if endpoint == [] :
185                                                 self.send_response(400)
186                                         else :
187                                                 self.send_response(200)
188                                                 sources[endpoint[0]] = [endpoint[1], endpoint[2]]
189                                 else :
190                                         self.send_response(400)
191
192                         else:
193                                 self.send_response(400)
194                         self.end_headers()
195                         
196                 elif re.search(ANNOUNCE_SINK_URL, self.path) != None:
197                         if self.headers.get_content_type() == 'application/json' :
198                                 # Find content length
199                                 content_len = 0
200                                 for i in range(len(self.headers.keys())):
201                                         if self.headers.keys()[i] == 'Content-Length' :
202                                                 content_len = int (self.headers.values()[i])
203                                                 break
204                                 if content_len != 0 :
205                                         # extract endpoint information                          
206                                         endpoint = extract_endpoint(self.rfile.read(content_len))
207                                         if endpoint == [] :
208                                                 self.send_response(400)
209                                         else :
210                                                 self.send_response(200)
211                                                 sinks[endpoint[0]] = [endpoint[1], endpoint[2]]
212                                 else :
213                                         self.send_response(400)
214
215                         else:
216                                 self.send_response(400)
217                         self.end_headers()      
218                         
219                 elif re.search(ANNOUNCE_STARTPROCESSING_URL, self.path) != None:
220                         if self.headers.get_content_type() == 'application/json' :
221                                 # Find content length
222                                 content_len = 0
223                                 for i in range(len(self.headers.keys())):
224                                         if self.headers.keys()[i] == 'Content-Length' :
225                                                 content_len = int (self.headers.values()[i])
226                                                 break
227                                 if content_len != 0 :
228                                         # extract name of initialized gs instance                               
229                                         name = extract_instance_name(self.rfile.read(content_len))
230                                         if name == '' :
231                                                 self.send_response(400)
232                                         else :
233                                                 self.send_response(200)
234                                                 gs_startprocessing_instances[name] = 1
235                                 else :
236                                         self.send_response(400)
237
238                         else:
239                                 self.send_response(400)
240                         self.end_headers()
241                         
242                 # we do not do any processing for ANNOUNCE_STREAM_SUBSCRIPTION, ANNOUNCE_FTA_INSTANCE and ANNOUNCE_METRICS in gshub simulator           
243                 elif (re.search(ANNOUNCE_STREAM_SUBSCRIPTION, self.path) != None) or (re.search(ANNOUNCE_FTA_INSTANCE, self.path) != None) or (re.search(ANNOUNCE_METRICS, self.path) != None):
244                         if self.headers.get_content_type() == 'application/json' :
245                                 # Find content length
246                                 content_len = 0
247                                 for i in range(len(self.headers.keys())):
248                                         if self.headers.keys()[i] == 'Content-Length' :
249                                                 content_len = int (self.headers.values()[i])
250                                                 break
251                                 if content_len != 0 :                           
252                                         self.send_response(200)
253                                 else :
254                                         self.send_response(400)
255
256                         else:
257                                 self.send_response(400)
258                         self.end_headers()      
259                         
260                 else:
261                         self.send_response(404)
262                         self.end_headers()
263                 return
264
265         def do_GET(self):
266                 if re.search(DISCOVER_INSTANCE_URL + '/*', self.path) != None:
267                         instance = self.path.split('/')[-1]
268                         # check if this instance is registered
269                         if instance in gs_instances :
270                                 self.send_response(200)
271                                 self.send_header('Content-Type', 'application/json')
272                                 self.end_headers()
273                                 self.wfile.write(bytes("{\"ip\" : \"" + gs_instances[instance][0] + "\", \"port\": " + str(gs_instances[instance][1]) + "}", "utf-8"))
274                         else:
275                                 self.send_response(400)
276                                 self.end_headers()
277
278
279                 elif re.search(DISCOVER_INITINSTANCE_URL + '/*', self.path) != None:
280                         instance = self.path.split('/')[-1]
281                         # check if this instance is initialized
282                         if instance in gs_init_instances :
283                                 self.send_response(200)
284                                 self.send_header('Content-Type', 'application/json')
285                                 self.end_headers()
286                                 self.wfile.write(bytes("{\"ip\" : \"" + gs_instances[instance][0] + "\", \"port\": " + str(gs_instances[instance][1]) + "}", "utf-8"))
287                         else:
288                                 self.send_response(400)
289                                 self.end_headers()
290
291                 elif re.search(DISCOVER_SOURCE_URL + '/*', self.path) != None:
292                         source = self.path.split('/')[-1]
293                         # check if it is a registered source
294                         if source in sources :
295                                 self.send_response(200)
296                                 self.send_header('Content-Type', 'application/json')
297                                 self.end_headers()
298                                 self.wfile.write(bytes("{\"ip\" : \"" + sources[source][0] + "\", \"port\": " + str(sources[source][1]) + "}", "utf-8"))
299                         else:
300                                 self.send_response(400)
301                                 self.end_headers()
302                                 
303                 elif re.search(DISCOVER_SINK_URL + '/*', self.path) != None:
304                         sink = self.path.split('/')[-1]
305                         # check if it is a registered sink
306                         if sink in sinks :
307                                 self.send_response(200)
308                                 self.send_header('Content-Type', 'application/json')
309                                 self.end_headers()
310                                 self.wfile.write(bytes("{\"ip\" : \"" + sinks[sink][0] + "\", \"port\": " + str(sinks[sink][1]) + "}", "utf-8"))
311                         else:
312                                 self.send_response(400)
313                                 self.end_headers()
314                                 
315                 elif re.search(DISCOVER_STARTPROCESSING_URL + '/*', self.path) != None:
316                         instance = self.path.split('/')[-1]
317                         # check if this instance is initialized
318                         if instance in gs_startprocessing_instances :
319                                 self.send_response(200)
320                                 self.send_header('Content-Type', 'application/json')
321                                 self.end_headers()
322                                 self.wfile.write(bytes("{}", "utf-8"))
323                         else:
324                                 self.send_response(400)
325                                 self.end_headers()                              
326                 else:
327                         self.send_response(404)
328                         self.end_headers()
329                 return
330
331
332 # we will use standard python threaded HTTP server
333 class ThreadedHTTPServer(ThreadingMixIn, HTTPServer):
334         allow_reuse_address = True
335
336         def shutdown(self):
337                 self.socket.close()
338                 HTTPServer.shutdown(self)
339
340 class SimpleHttpServer:
341         def __init__(self, ip, port):
342                 self.server = ThreadedHTTPServer((ip,port), HTTPRequestHandler)
343
344         def start(self):
345                 self.server_thread = threading.Thread(target=self.server.serve_forever)
346                 self.server_thread.daemon = True
347                 self.server_thread.start()
348                 
349         def waitForThread(self):
350                 self.server_thread.join()
351
352         def stop(self):
353                 self.server.shutdown()
354                 self.waitForThread()
355
356
357 # print usage instructions
358 def usage():
359         print ('./gshub.py [-p port]')
360         
361         
362
363 def main():
364         # process command-line arguments
365         try:
366                 opts, args = getopt.getopt(sys.argv[1:], "hp:v", ["help", "port="])
367         except getopt.GetoptError as err:
368                 # print help information and exit:
369                 print(str(err)) 
370                 usage()
371                 sys.exit(2)
372
373         port = 0
374         for o, a in opts:
375                 if o in ("-h", "--help"):
376                         usage()
377                         sys.exit(0)
378                 elif o in ("-p", "--port"):
379                         port = int(a)
380                 else:
381                         print ('Unknown command-line option ' + o)
382
383         # start HTTP server to serve REST calls
384         server = SimpleHttpServer('127.0.0.1', port)
385
386         # record HTTP server address in gshub.log
387         f = open('gshub.log', 'w')
388         f.write('127.0.0.1:' + str(server.server.server_port) + '\n')
389         f.close()
390                 
391         print ('GSHUB Running on port ' + str(server.server.server_port) + ' ...')
392         server.start()
393         server.waitForThread()
394
395
396 if __name__ == "__main__":
397         main()
398