Add container tags to test images
[nonrtric.git] / test / cr / app / cr.py
1
2 #  ============LICENSE_START===============================================
3 #  Copyright (C) 2020 Nordix Foundation. All rights reserved.
4 #  ========================================================================
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 #  ============LICENSE_END=================================================
17 #
18
19 from flask import Flask, request, Response
20 from time import sleep
21 import time
22 from datetime import datetime
23 import json
24 import traceback
25 import logging
26 import socket
27 from threading import RLock
28 from hashlib import md5
29
30 # Disable all logging of GET on reading counters and db
31 class AjaxFilter(logging.Filter):
32     def filter(self, record):
33         return ("/counter/" not in record.getMessage()) and ("/db" not in record.getMessage())
34
35 log = logging.getLogger('werkzeug')
36 log.addFilter(AjaxFilter())
37
38 app = Flask(__name__)
39
40 lock = RLock()
41
42 # list of callback messages
43 msg_callbacks={}
44
45 # Server info
46 HOST_IP = "::"
47 HOST_PORT = 2222
48
49 # Metrics vars
50 cntr_msg_callbacks=0
51 cntr_batch_callbacks=0
52 cntr_msg_fetched=0
53 cntr_callbacks={}
54 hosts_set=set()
55
56 # Request and response constants
57 CALLBACK_URL="/callbacks/<string:id>"
58 CALLBACK_MR_URL="/callbacks-mr/<string:id>" #Json list with string encoded items
59 CALLBACK_TEXT_URL="/callbacks-text/<string:id>" # Callback for string of text
60 APP_READ_URL="/get-event/<string:id>"
61 APP_READ_ALL_URL="/get-all-events/<string:id>"
62 DUMP_ALL_URL="/db"
63 NULL_URL="/callbacks-null"  # Url for ignored callback. Callbacks are not checked, counted or stored
64
65 MIME_TEXT="text/plain"
66 MIME_JSON="application/json"
67 CAUGHT_EXCEPTION="Caught exception: "
68 SERVER_ERROR="Server error :"
69 TIME_STAMP="cr-timestamp"
70
71 forced_settings={}
72 forced_settings['delay']=None
73
74
75 # Remote host lookup and print host name
76 def remote_host_logging(request):
77
78     if request.environ.get('HTTP_X_FORWARDED_FOR') is None:
79         host_ip=str(request.environ['REMOTE_ADDR'])
80     else:
81         host_ip=str(request.environ['HTTP_X_FORWARDED_FOR'])
82     prefix='::ffff:'
83     if (host_ip.startswith('::ffff:')):
84         host_ip=host_ip[len(prefix):]
85     try:
86         name, alias, addresslist = socket.gethostbyaddr(host_ip)
87         print("Calling host: "+str(name))
88         hosts_set.add(name)
89     except Exception:
90         print("Calling host not possible to retrieve IP: "+str(host_ip))
91         hosts_set.add(host_ip)
92
93
94 #I'm alive function
95 @app.route('/',
96     methods=['GET'])
97 def index():
98     return 'OK', 200
99
100 ### Callback interface, for control
101
102 # Fetch the oldest callback message for an id
103 # URI and parameter, (GET): /get-event/<id>
104 # response: message + 200 or just 204 or just 500(error)
105 @app.route(APP_READ_URL,
106     methods=['GET'])
107 def receiveresponse(id):
108     global msg_callbacks
109     global cntr_msg_fetched
110
111     with lock:
112         try:
113             if ((id in msg_callbacks.keys()) and (len(msg_callbacks[id]) > 0)):
114                 cntr_msg_fetched+=1
115                 cntr_callbacks[id][1]+=1
116                 msg=msg_callbacks[id][0]
117                 print("Fetching msg for id: "+id+", msg="+str(msg))
118
119                 if (isinstance(msg,dict)):
120                     del msg[TIME_STAMP]
121                     if ("md5" in msg.keys()):
122                         print("EXTRACTED MD5")
123                         msg=msg["md5"]
124                         print("MD5: "+str(msg))
125
126                 del msg_callbacks[id][0]
127                 return json.dumps(msg),200
128             print("No messages for id: "+id)
129         except Exception as e:
130             print(CAUGHT_EXCEPTION+str(e))
131             traceback.print_exc()
132             return "",500
133
134         return "",204
135
136 # Fetch all callback message for an id in an array
137 # URI and parameter, (GET): /get-all-events/<id>
138 # response: message + 200 or just 500(error)
139 @app.route(APP_READ_ALL_URL,
140     methods=['GET'])
141 def receiveresponse_all(id):
142     global msg_callbacks
143     global cntr_msg_fetched
144
145     with lock:
146         try:
147             if ((id in msg_callbacks.keys()) and (len(msg_callbacks[id]) > 0)):
148                 cntr_msg_fetched+=len(msg_callbacks[id])
149                 cntr_callbacks[id][1]+=len(msg_callbacks[id])
150                 msg=msg_callbacks[id]
151                 print("Fetching all msgs for id: "+id+", msg="+str(msg))
152                 for sub_msg in msg:
153                     if (isinstance(sub_msg, dict)):
154                         del sub_msg[TIME_STAMP]
155                 del msg_callbacks[id]
156                 return json.dumps(msg),200
157             print("No messages for id: "+id)
158         except Exception as e:
159             print(CAUGHT_EXCEPTION+str(e))
160             traceback.print_exc()
161             return "",500
162
163         msg=[]
164         return json.dumps(msg),200
165
166 # Receive a callback message
167 # URI and payload, (PUT or POST): /callbacks/<id> <json messages>
168 # response: OK 200 or 500 for other errors
169 @app.route(CALLBACK_URL,
170     methods=['PUT','POST'])
171 def events_write(id):
172     global msg_callbacks
173     global cntr_msg_callbacks
174
175     try:
176         print("Received callback for id: "+id +", content-type="+request.content_type)
177         remote_host_logging(request)
178         print("raw data: str(request.data): "+str(request.data))
179         do_delay()
180         try:
181             if (request.content_type == MIME_JSON):
182                 data = request.data
183                 msg = json.loads(data)
184                 print("Payload(json): "+str(msg))
185             else:
186                 msg={}
187                 print("Payload(content-type="+request.content_type+"). Setting empty json as payload")
188         except Exception as e:
189             msg={}
190             print("(Exception) Payload does not contain any json, setting empty json as payload")
191             traceback.print_exc()
192
193         with lock:
194             cntr_msg_callbacks += 1
195             if (isinstance(msg, dict)):
196                 msg[TIME_STAMP]=str(datetime.now())
197             if (id in msg_callbacks.keys()):
198                 msg_callbacks[id].append(msg)
199             else:
200                 msg_callbacks[id]=[]
201                 msg_callbacks[id].append(msg)
202
203             if (id in cntr_callbacks.keys()):
204                 cntr_callbacks[id][0] += 1
205                 cntr_callbacks[id][2] += 1
206             else:
207                 cntr_callbacks[id]=[]
208                 cntr_callbacks[id].append(1)
209                 cntr_callbacks[id].append(0)
210                 cntr_callbacks[id].append(0)
211
212     except Exception as e:
213         print(CAUGHT_EXCEPTION+str(e))
214         traceback.print_exc()
215         return 'NOTOK',500
216
217     return 'OK',200
218
219
220 # Receive a json callback message with payload formatted according to output from the message router
221 # Array of stringified json objects
222 # URI and payload, (PUT or POST): /callbacks-mr/<id> <json messages>
223 # json is a list of string encoded json items
224 # response: OK 200 or 500 for other errors
225 @app.route(CALLBACK_MR_URL,
226     methods=['PUT','POST'])
227 def events_write_mr(id):
228     global msg_callbacks
229     global cntr_msg_callbacks
230     global cntr_batch_callbacks
231
232     storeas=request.args.get('storeas') #If set, store payload as a md5 hascode and dont log the payload
233                                         #Large payloads will otherwise overload the server
234     try:
235         print("Received callback (mr) for id: "+id +", content-type="+request.content_type)
236         print("raw data: str(request.data): "+str(request.data))
237         if (storeas is None):
238             print("raw data: str(request.data): "+str(request.data))
239         do_delay()
240         list_data=False
241         try:
242             #if (request.content_type == MIME_JSON):
243             if (MIME_JSON in request.content_type):
244                 data = request.data
245                 msg_list = json.loads(data)
246                 if (storeas is None):
247                     print("Payload(json): "+str(msg_list))
248                 list_data=True
249             else:
250                 msg_list=[]
251                 print("Payload(content-type="+request.content_type+"). Setting empty json as payload")
252         except Exception as e:
253             msg_list=[]
254             print("(Exception) Payload does not contain any json, setting empty json as payload")
255             traceback.print_exc()
256
257         with lock:
258             remote_host_logging(request)
259             if (list_data):
260                 cntr_batch_callbacks += 1
261             for msg in msg_list:
262                 if (storeas is None):
263                     msg=json.loads(msg)
264                 else:
265                     #Convert to compact json without ws between parameter and value...
266                     #It seem that ws is added somewhere along to way to this server
267                     msg=json.loads(msg)
268                     msg=json.dumps(msg, separators=(',', ':'))
269
270                     md5msg={}
271                     md5msg["md5"]=md5(msg.encode('utf-8')).hexdigest()
272                     msg=md5msg
273                     print("msg (json converted to md5 hash): "+str(msg["md5"]))
274                 cntr_msg_callbacks += 1
275                 if (isinstance(msg, dict)):
276                     msg[TIME_STAMP]=str(datetime.now())
277                 if (id in msg_callbacks.keys()):
278                     msg_callbacks[id].append(msg)
279                 else:
280                     msg_callbacks[id]=[]
281                     msg_callbacks[id].append(msg)
282
283                 if (id in cntr_callbacks.keys()):
284                     cntr_callbacks[id][0] += 1
285                 else:
286                     cntr_callbacks[id]=[]
287                     cntr_callbacks[id].append(1)
288                     cntr_callbacks[id].append(0)
289                     cntr_callbacks[id].append(0)
290             if (id in msg_callbacks.keys() and list_data):
291                 cntr_callbacks[id][2] += 1
292
293     except Exception as e:
294         print(CAUGHT_EXCEPTION+str(e))
295         traceback.print_exc()
296         return 'NOTOK',500
297
298     return 'OK',200
299
300 # Receive a callback message of a single text message (content type ignored)
301 # or a json array of strings (content type json)
302 # URI and payload, (PUT or POST): /callbacks-text/<id> <text message>
303 # response: OK 200 or 500 for other errors
304 @app.route(CALLBACK_TEXT_URL,
305     methods=['PUT','POST'])
306 def events_write_text(id):
307     global msg_callbacks
308     global cntr_msg_callbacks
309     global cntr_batch_callbacks
310
311     storeas=request.args.get('storeas') #If set, store payload as a md5 hascode and dont log the payload
312                                         #Large payloads will otherwise overload the server
313     try:
314         print("Received callback for id: "+id +", content-type="+request.content_type)
315         remote_host_logging(request)
316         if (storeas is None):
317             print("raw data: str(request.data): "+str(request.data))
318         do_delay()
319
320         try:
321             msg_list=None
322             list_data=False
323             if (MIME_JSON in request.content_type):  #Json array of strings
324                 msg_list=json.loads(request.data)
325                 list_data=True
326             else:
327                 data=request.data.decode("utf-8")    #Assuming string
328                 msg_list=[]
329                 msg_list.append(data)
330             with lock:
331                 cntr_batch_callbacks += 1
332                 for msg in msg_list:
333                     if (storeas == "md5"):
334                         md5msg={}
335                         print("msg: "+str(msg))
336                         print("msg (endcode str): "+str(msg.encode('utf-8')))
337                         md5msg["md5"]=md5(msg.encode('utf-8')).hexdigest()
338                         msg=md5msg
339                         print("msg (data converted to md5 hash): "+str(msg["md5"]))
340
341                     if (isinstance(msg, dict)):
342                         msg[TIME_STAMP]=str(datetime.now())
343
344                     cntr_msg_callbacks += 1
345                     if (id in msg_callbacks.keys()):
346                         msg_callbacks[id].append(msg)
347                     else:
348                         msg_callbacks[id]=[]
349                         msg_callbacks[id].append(msg)
350
351                     if (id in cntr_callbacks.keys()):
352                         cntr_callbacks[id][0] += 1
353                     else:
354                         cntr_callbacks[id]=[]
355                         cntr_callbacks[id].append(1)
356                         cntr_callbacks[id].append(0)
357                         cntr_callbacks[id].append(0)
358                 if (id in cntr_callbacks.keys() and list_data):
359                     cntr_callbacks[id][2] += 1
360         except Exception as e:
361             print(CAUGHT_EXCEPTION+str(e))
362             traceback.print_exc()
363             return 'NOTOK',500
364
365
366     except Exception as e:
367         print(CAUGHT_EXCEPTION+str(e))
368         traceback.print_exc()
369         return 'NOTOK',500
370
371     return 'OK',200
372
373 # Receive a callback message but ignore contents and return 200
374 # URI and payload, (PUT or POST): /callbacks-text/<id> <text message>
375 # response: OK 200
376 @app.route(NULL_URL,
377     methods=['PUT','POST'])
378 def null_url(id):
379     return 'OK',200
380
381 # Dump the whole db of current callbacks
382 # URI and parameter, (GET): /db
383 # response: message + 200
384 @app.route(DUMP_ALL_URL,
385     methods=['GET'])
386 def dump_db():
387     return json.dumps(msg_callbacks),200
388
389 ### Functions for metrics read out ###
390
391 @app.route('/counter/received_callbacks',
392     methods=['GET'])
393 def requests_submitted():
394     req_id = request.args.get('id')
395     if (req_id is None):
396         return Response(str(cntr_msg_callbacks), status=200, mimetype=MIME_TEXT)
397
398     if (req_id in cntr_callbacks.keys()):
399         return Response(str(cntr_callbacks[req_id][0]), status=200, mimetype=MIME_TEXT)
400     else:
401         return Response(str("0"), status=200, mimetype=MIME_TEXT)
402
403 @app.route('/counter/received_callback_batches',
404     methods=['GET'])
405 def batches_submitted():
406     req_id = request.args.get('id')
407     if (req_id is None):
408         return Response(str(cntr_batch_callbacks), status=200, mimetype=MIME_TEXT)
409
410     if (req_id in cntr_callbacks.keys()):
411         return Response(str(cntr_callbacks[req_id][2]), status=200, mimetype=MIME_TEXT)
412     else:
413         return Response(str("0"), status=200, mimetype=MIME_TEXT)
414
415 @app.route('/counter/fetched_callbacks',
416     methods=['GET'])
417 def requests_fetched():
418     req_id = request.args.get('id')
419     if (req_id is None):
420         return Response(str(cntr_msg_fetched), status=200, mimetype=MIME_TEXT)
421
422     if (req_id in cntr_callbacks.keys()):
423         return Response(str(cntr_callbacks[req_id][1]), status=200, mimetype=MIME_TEXT)
424     else:
425         return Response(str("0"), status=200, mimetype=MIME_TEXT)
426
427 @app.route('/counter/current_messages',
428     methods=['GET'])
429 def current_messages():
430     req_id = request.args.get('id')
431     if (req_id is None):
432         return Response(str(cntr_msg_callbacks-cntr_msg_fetched), status=200, mimetype=MIME_TEXT)
433
434     if (req_id in cntr_callbacks.keys()):
435         return Response(str(cntr_callbacks[req_id][0]-cntr_callbacks[req_id][1]), status=200, mimetype=MIME_TEXT)
436     else:
437         return Response(str("0"), status=200, mimetype=MIME_TEXT)
438
439 @app.route('/counter/remote_hosts',
440     methods=['GET'])
441 def remote_hosts():
442     global hosts_set
443
444     hosts=",".join(hosts_set)
445     return Response(str(hosts), status=200, mimetype=MIME_TEXT)
446
447
448 #Set force delay response, in seconds, for all callbacks
449 #/froceesponse?delay=<seconds>
450 @app.route('/forcedelay', methods=['POST'])
451 def forcedelay():
452
453   try:
454     forced_settings['delay']=int(request.args.get('delay'))
455   except Exception:
456     forced_settings['delay']=None
457   return Response("Force delay: " + str(forced_settings['delay']) + " sec set for all callback responses", 200, mimetype=MIME_TEXT)
458
459 # Helper: Delay if delayed response code is set
460 def do_delay():
461   if (forced_settings['delay'] is not None):
462     try:
463       val=int(forced_settings['delay'])
464       if (val < 1):
465           return Response("Force delay too short: " + str(forced_settings['delay']) + " sec", 500, mimetype=MIME_TEXT)
466       print("Delaying "+str(val)+ " sec.")
467       time.sleep(val)
468     except Exception:
469       return Response("Force delay : " + str(forced_settings['delay']) + " sec failed", 500, mimetype=MIME_TEXT)
470 ### Admin ###
471
472 # Reset all messsages and counters
473 @app.route('/reset',
474     methods=['GET', 'POST', 'PUT'])
475 def reset():
476     global msg_callbacks
477     global cntr_msg_fetched
478     global cntr_msg_callbacks
479     global cntr_batch_callbacks
480     global cntr_callbacks
481     global forced_settings
482
483     with lock:
484         msg_callbacks={}
485         cntr_msg_fetched=0
486         cntr_msg_callbacks=0
487         cntr_batch_callbacks=0
488         cntr_callbacks={}
489         forced_settings['delay']=None
490
491         return Response('OK', status=200, mimetype=MIME_TEXT)
492
493 ### Main function ###
494
495 if __name__ == "__main__":
496     app.run(port=HOST_PORT, host=HOST_IP)