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
9 # http://www.apache.org/licenses/LICENSE-2.0
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=================================================
19 from flask import Flask, request
20 from time import sleep
24 from flask import Flask
25 from flask import Response
27 from threading import RLock, Thread
32 # Disable all logging of GET on reading counters
33 class AjaxFilter(logging.Filter):
34 def filter(self, record):
35 return ("/counter/" not in record.getMessage())
37 log = logging.getLogger('werkzeug')
38 log.addFilter(AjaxFilter())
42 # list of messages to/from Dmaap
53 cntr_msg_requests_submitted=0
54 cntr_msg_requests_fetched=0
55 cntr_msg_responses_submitted=0
56 cntr_msg_responses_fetched=0
58 # Request and response constants
59 ORU_WRITE_URL="/events/unauthenticated.SEC_FAULT_OUTPUT"
60 ORU_READ_URL="/events/unauthenticated.SEC_FAULT_OUTPUT/users/test/"
61 AGENT_WRITE_URL="/events/A1-POLICY-AGENT-WRITE"
62 AGENT_READ_URL="/events/A1-POLICY-AGENT-READ/users/policy-agent"
63 APP_WRITE_URL="/send-request"
64 APP_READ_URL="/receive-response"
65 MIME_TEXT="text/plain"
66 MIME_JSON="application/json"
67 CAUGHT_EXCEPTION="Caught exception: "
68 SERVER_ERROR="Server error :"
72 generic_topics_upload_baseurl=""
75 downloader_thread=None
76 generic_uploader_thread=None
78 # Function to upload PMS messages to dmaap
81 global cntr_msg_requests_fetched
83 print("Starting uploader")
85 headers = {'Content-type': 'application/json', 'Accept': '*/*'}
86 #url="http://"+topic_host+"/events/"+topic_read
90 while (len(msg_requests)>0):
94 print("Sending to dmaap : "+ url)
95 print("Sending to dmaap : "+ msg)
96 resp=requests.post(url, data=msg, headers=headers, timeout=10)
97 if (resp.status_code<199 & resp.status_code > 299):
98 print("Failed, response code: " + str(resp.status_code))
101 print("Dmaap response code: " + str(resp.status_code))
102 print("Dmaap response text: " + str(resp.text))
105 cntr_msg_requests_fetched += 1
106 except Exception as e:
107 print("Failed, exception: "+ str(e))
112 # Function to download PMS messages from dmaap
113 def dmaap_downloader():
115 global cntr_msg_responses_submitted
117 print("Starting uploader")
122 #url="http://"+topic_host+"/events/"+topic_write+"/users/mr-stub?timeout=15000&limit=100"
124 headers = {'Accept': 'application/json'}
125 print("Reading from dmaap: " + url)
126 resp=requests.get(url, headers=headers)
127 if (resp.status_code<199 & resp.status_code > 299):
128 print("Failed, response code: " + resp.status_code)
131 print("Recieved data from dmaap mr")
134 print("Recieved data (raw): " + str(resp.text))
135 if isinstance(data, list):
137 item=json.loads(item)
138 corrid=str(item["correlationId"])
139 status=str(item["status"])
140 msg=str(item["message"])
141 item_str=msg+status[0:3]
143 msg_responses[corrid]=item_str
144 cntr_msg_responses_submitted += 1
146 print("Data from dmaap is not json array: " + str(resp.text))
148 except Exception as e:
149 print("Corrupt data from dmaap mr - dropping " + str(data))
150 print("CAUGHT_EXCEPTION" + str(e) + " "+traceback.format_exc())
152 except Exception as e:
155 # Function to upload generic messages to dmaap
156 def dmaap_generic_uploader():
158 global cntr_msg_requests_fetched
160 print("Starting generic uploader")
162 headers_json = {'Content-type': 'application/json', 'Accept': '*/*'}
163 headers_text = {'Content-type': 'text/plain', 'Accept': '*/*'}
166 if (len(generic_messages)):
167 for topicname in generic_messages.keys(): #topicname contains the path of the topics, eg. "/event/<topic>"
168 topic_queue=generic_messages[topicname]
169 if (len(topic_queue)>0):
170 if (topicname.endswith(".text")):
177 url=generic_topics_upload_baseurl+topicname
178 print("Sending to dmaap : "+ url)
179 print("Sending to dmaap : "+ msg)
180 print("Sending to dmaap : "+ str(headers))
182 resp=requests.post(url, data=msg, headers=headers, timeout=10)
183 if (resp.status_code<199 & resp.status_code > 299):
184 print("Failed, response code: " + str(resp.status_code))
187 print("Dmaap response code: " + str(resp.status_code))
188 print("Dmaap response text: " + str(resp.text))
191 cntr_msg_requests_fetched += 1
192 except Exception as e:
193 print("Failed, exception: "+ str(e))
204 # Helper function to create a Dmaap PMS request message
205 # args : <GET|PUT|DELETE> <correlation-id> <json-string-payload - may be None> <url>
206 # response: json formatted string of a complete Dmaap message
207 def create_message(operation, correlation_id, payload, url):
208 if (payload is None):
210 time_stamp=datetime.datetime.utcnow()
211 msg = '{\"apiVersion\":\"1.0\",\"operation\":\"'+operation+'\",\"correlationId\":\"'+correlation_id+'\",\"originatorId\": \"849e6c6b420\",'
212 msg = msg + '\"payload\":'+payload+',\"requestId\":\"23343221\", \"target\":\"policy-agent\", \"timestamp\":\"'+str(time_stamp)+'\", \"type\":\"request\",\"url\":\"'+url+'\"}'
216 ### MR-stub interface, for MR control
218 # Send a PMS message to MR
219 # URI and parameters (PUT or POST): /send-request?operation=<GET|PUT|POST|DELETE>&url=<url>
220 # response: <correlation-id> (http 200) o4 400 for parameter error or 500 for other errors
221 @app.route(APP_WRITE_URL,
222 methods=['PUT','POST'])
225 global cntr_msg_requests_submitted
227 print("APP_WRITE_URL lock")
229 oper=request.args.get('operation')
231 print(APP_WRITE_URL+" parameter 'operation' missing")
232 return Response('Parameter operation missing in request', status=400, mimetype=MIME_TEXT)
234 url=request.args.get('url')
236 print(APP_WRITE_URL+" parameter 'url' missing")
237 return Response('Parameter url missing in request', status=400, mimetype=MIME_TEXT)
239 if (oper != "GET" and oper != "PUT" and oper != "POST" and oper != "DELETE"):
240 print(APP_WRITE_URL+" parameter 'operation' need to be: DEL|PUT|POST|DELETE")
241 return Response('Parameter operation does not contain DEL|PUT|POST|DELETE in request', status=400, mimetype=MIME_TEXT)
243 print(APP_WRITE_URL+" operation="+oper+" url="+url)
244 correlation_id=str(time.time_ns())
246 if (oper == "PUT") and (request.json is not None):
247 payload=json.dumps(request.json)
249 msg=create_message(oper, correlation_id, payload, url)
251 print(APP_WRITE_URL+" MSG(correlationid = "+correlation_id+"): " + json.dumps(json.loads(msg), indent=2))
252 msg_requests.append(msg)
253 cntr_msg_requests_submitted += 1
254 return Response(correlation_id, status=200, mimetype=MIME_TEXT)
255 except Exception as e:
256 print(APP_WRITE_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
257 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
259 # Receive a PMS message response for MR for the included correlation id
260 # URI and parameter, (GET): /receive-response?correlationid=<correlation-id>
261 # response: <json-array of 1 response> 200 or empty 204 or other errors 500
262 @app.route(APP_READ_URL,
264 def receiveresponse():
266 global cntr_msg_responses_fetched
268 print("APP_READ_URL lock")
270 cid=request.args.get('correlationid')
272 print(APP_READ_URL+" parameter 'correclationid' missing")
273 return Response('Parameter correlationid missing in json', status=500, mimetype=MIME_TEXT)
275 if (cid in msg_responses):
276 answer=msg_responses[cid]
277 del msg_responses[cid]
278 print(APP_READ_URL+" response (correlationid="+cid+"): " + answer)
279 cntr_msg_responses_fetched += 1
280 return Response(answer, status=200, mimetype=MIME_JSON)
282 print(APP_READ_URL+" - no messages (correlationid="+cid+"): ")
283 return Response('', status=204, mimetype=MIME_JSON)
284 except Exception as e:
285 print(APP_READ_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
286 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
288 ### Dmaap interface ###
290 # Read PMS messages stream. URI according to agent configuration.
291 # URI, (GET): /events/A1-POLICY-AGENT-READ/users/policy-agent
292 # response: 200 <json array of request messages>, or 500 for other errors
293 @app.route(AGENT_READ_URL,
297 global cntr_msg_requests_fetched
299 if topic_write or topic_read:
300 return Response('Url not available when running as mrstub frontend', status=404, mimetype=MIME_TEXT)
302 limit=request.args.get('limit')
311 print("Limting number of returned messages to: "+str(limit))
313 timeout=request.args.get('timeout')
314 if (timeout is None):
317 timeout=min(int(timeout),60000)
319 start_time=int(round(time.time() * 1000))
320 current_time=int(round(time.time() * 1000))
322 while(current_time<start_time+int(timeout)):
324 if(len(msg_requests)>0):
328 while(cntr<limit and len(msg_requests)>0):
331 msgs=msgs+msg_requests.pop(0)
332 cntr_msg_requests_fetched += 1
335 print(AGENT_READ_URL+" MSGs: "+json.dumps(json.loads(msgs), indent=2))
336 return Response(msgs, status=200, mimetype=MIME_JSON)
337 except Exception as e:
338 print(AGENT_READ_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
339 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
340 sleep(0.025) # sleep 25 milliseconds
341 current_time=int(round(time.time() * 1000))
343 print("timeout: "+str(timeout)+", start_time: "+str(start_time)+", current_time: "+str(current_time))
344 return Response("[]", status=200, mimetype=MIME_JSON)
346 # Write PMS messages stream. URI according to agent configuration.
347 # URI and payload, (PUT or POST): /events/A1-POLICY-AGENT-WRITE <json array of response messages>
348 # response: OK 200 or 400 for missing json parameters, 500 for other errors
349 @app.route(AGENT_WRITE_URL,
350 methods=['PUT','POST'])
353 global cntr_msg_responses_submitted
355 if topic_write or topic_read:
356 return Response('Url not available when running as mrstub frontend', status=404, mimetype=MIME_TEXT)
359 print("AGENT_WRITE_URL lock")
362 print(AGENT_WRITE_URL+ " json=" + json.dumps(answer, indent=2))
363 if isinstance(answer, dict):
364 #Create a an array if the answer is a dict (single message)
366 answer_list.append(answer)
370 cid=item['correlationId']
372 print(AGENT_WRITE_URL+" parameter 'correlatonid' missing")
373 return Response('Parameter <correlationid> missing in json', status=400, mimetype=MIME_TEXT)
376 print(AGENT_WRITE_URL+" parameter 'msgs' missing")
377 return Response('Parameter >message> missing in json', status=400, mimetype=MIME_TEXT)
378 status=item['status']
380 print(AGENT_WRITE_URL+" parameter 'status' missing")
381 return Response('Parameter <status> missing in json', status=400, mimetype=MIME_TEXT)
382 if isinstance(msg, list) or isinstance(msg, dict):
383 msg_str=json.dumps(msg)+status[0:3]
385 msg_str=msg+status[0:3]
386 msg_responses[cid]=msg_str
387 cntr_msg_responses_submitted += 1
388 print(AGENT_WRITE_URL+ " msg+status (correlationid="+cid+") :" + str(msg_str))
389 except Exception as e:
390 print(AGENT_WRITE_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
391 return Response('{"message": "' + SERVER_ERROR + ' ' + str(e) + '","status":"500"}', status=200, mimetype=MIME_JSON)
393 return Response('{}', status=200, mimetype=MIME_JSON)
395 @app.route(ORU_WRITE_URL,
396 methods=['PUT','POST'])
399 msg=json.dumps(request.json)
400 msg_requests.append(msg)
401 return Response('{}', status=200, mimetype=MIME_JSON)
403 @app.route(ORU_READ_URL,
407 if(len(msg_requests)>0):
408 rsp=msg_requests.pop(0)
411 return Response(json.dumps(res), status=200, mimetype=MIME_JSON)
412 return Response("[]", status=200, mimetype=MIME_JSON)
414 # Generic POST catching all urls starting with /events/<topic>.
415 # Writes the message in a que for that topic
416 @app.route("/events/<path>",
418 def generic_write(path):
419 global generic_messages
420 global cntr_msg_responses_submitted
421 urlkey="/events/"+str(path)
422 write_method=str(request.method)
425 if (urlkey.endswith(".text")):
426 payload=str(request.data.decode('UTF-8'))
427 print(write_method+" on "+urlkey+" text=" + payload)
430 print(write_method+" on "+urlkey+" json=" + json.dumps(payload))
432 if (urlkey in generic_messages.keys()):
433 topicmsgs=generic_messages[urlkey]
435 generic_messages[urlkey]=topicmsgs
437 if isinstance(payload, list):
438 for listitem in payload:
439 topicmsgs.append(listitem)
441 topicmsgs.append(payload)
443 cntr_msg_responses_submitted += 1
444 except Exception as e:
445 print(write_method + "on "+urlkey+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
446 return Response('{"message": "' + SERVER_ERROR + ' ' + str(e) + '","status":"500"}', status=200, mimetype=MIME_JSON)
448 return Response('{}', status=200, mimetype=MIME_JSON)
450 # Generic GET catching all urls starting with /events/. Returns max 4096 json msgs in an array.
451 # Returns only the messages previously written to the same urls
452 @app.route("/events/<path:path>",
454 def generic_read(path):
455 global generic_messages
456 global cntr_msg_requests_fetched
458 if generic_topics_upload_baseurl:
459 return Response('Url not available when running as mrstub frontend', status=404, mimetype=MIME_TEXT)
461 urlpath="/events/"+str(path)
462 urlkey="/events/"+str(path).split("/")[0] #Extract topic
463 print("GET on topic"+urlkey)
464 limit=request.args.get('limit')
473 print("Limting number of returned messages to: "+str(limit))
475 timeout=request.args.get('timeout')
476 if (timeout is None):
479 timeout=min(int(timeout),60000)
481 start_time=int(round(time.time() * 1000))
482 current_time=int(round(time.time() * 1000))
484 if (urlkey in generic_messages.keys()):
485 topicmsgs=generic_messages[urlkey]
487 while(current_time<start_time+int(timeout)):
489 if(len(topicmsgs)>0):
493 while(cntr<limit and len(topicmsgs)>0):
496 msgs=msgs+json.dumps(json.dumps(topicmsgs.pop(0)))
497 cntr_msg_requests_fetched += 1
500 print("GET on "+urlpath+" MSGs: "+msgs)
501 return Response(msgs, status=200, mimetype=MIME_JSON)
502 except Exception as e:
503 print("GET on "+urlpath+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
504 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
505 sleep(0.025) # sleep 25 milliseconds
506 current_time=int(round(time.time() * 1000))
508 print("timeout: "+str(timeout)+", start_time: "+str(start_time)+", current_time: "+str(current_time))
509 return Response("[]", status=200, mimetype=MIME_JSON)
512 ### Functions for metrics read out ###
514 @app.route('/counter/requests_submitted',
516 def requests_submitted():
517 return Response(str(cntr_msg_requests_submitted), status=200, mimetype=MIME_TEXT)
519 @app.route('/counter/requests_fetched',
521 def requests_fetched():
522 return Response(str(cntr_msg_requests_fetched), status=200, mimetype=MIME_TEXT)
524 @app.route('/counter/responses_submitted',
526 def responses_submitted():
527 return Response(str(cntr_msg_responses_submitted), status=200, mimetype=MIME_TEXT)
529 @app.route('/counter/responses_fetched',
531 def responses_fetched():
532 return Response(str(cntr_msg_responses_fetched), status=200, mimetype=MIME_TEXT)
534 @app.route('/counter/current_requests',
536 def current_requests():
537 return Response(str(len(msg_requests)), status=200, mimetype=MIME_TEXT)
539 @app.route('/counter/current_responses',
541 def current_responses():
542 return Response(str(len(msg_responses)), status=200, mimetype=MIME_TEXT)
546 # Reset all messsages and counters
548 methods=['GET', 'POST', 'PUT'])
550 global cntr_msg_requests_submitted
551 global cntr_msg_requests_fetched
552 global cntr_msg_responses_submitted
553 global cntr_msg_responses_fetched
557 cntr_msg_requests_submitted=0
558 cntr_msg_requests_fetched=0
559 cntr_msg_responses_submitted=0
560 cntr_msg_responses_fetched=0
563 return Response('OK', status=200, mimetype=MIME_TEXT)
565 # Get env vars, if present
566 if os.getenv("TOPIC_READ") is not None:
568 print("Env variables:")
569 print("TOPIC_READ:"+os.environ['TOPIC_READ'])
570 print("TOPIC_WRITE:"+os.environ['TOPIC_WRITE'])
572 topic_read=os.environ['TOPIC_READ']
573 topic_write=os.environ['TOPIC_WRITE']
576 if topic_read and downloader_thread is None:
577 downloader_thread=Thread(target=dmaap_downloader)
578 downloader_thread.start()
580 if topic_write and uploader_thread is None:
581 uploader_thread=Thread(target=dmaap_uploader)
582 uploader_thread.start()
584 if os.environ['GENERIC_TOPICS_UPLOAD_BASEURL'] is not None:
585 print("GENERIC_TOPICS_UPLOAD_BASEURL:"+os.environ['GENERIC_TOPICS_UPLOAD_BASEURL'])
586 generic_topics_upload_baseurl=os.environ['GENERIC_TOPICS_UPLOAD_BASEURL']
587 if generic_topics_upload_baseurl and generic_uploader_thread is None:
588 generic_uploader_thread=Thread(target=dmaap_generic_uploader)
589 generic_uploader_thread.start()
591 if os.getenv("TOPIC_READ") is None or os.environ['GENERIC_TOPICS_UPLOAD_BASEURL'] is None:
592 print("No env variables - OK")
594 if __name__ == "__main__":
595 app.run(port=HOST_PORT, host=HOST_IP)