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 :"
74 downloader_thread=None
76 # Function to download messages from dmaap
79 global cntr_msg_requests_fetched
81 print("Starting uploader")
83 headers = {'Content-type': 'application/json', 'Accept': '*/*'}
84 #url="http://"+topic_host+"/events/"+topic_read
88 while (len(msg_requests)>0):
92 print("Sending to dmaap : "+ url)
93 print("Sending to dmaap : "+ msg)
94 resp=requests.post(url, data=msg, headers=headers, timeout=10)
95 if (resp.status_code<199 & resp.status_code > 299):
96 print("Failed, response code: " + str(resp.status_code))
99 print("Dmaap response code: " + str(resp.status_code))
100 print("Dmaap response text: " + str(resp.text))
103 cntr_msg_requests_fetched += 1
104 except Exception as e:
105 print("Failed, exception: "+ str(e))
110 # Function to upload messages to dmaap
111 def dmaap_downloader():
113 global cntr_msg_responses_submitted
115 print("Starting uploader")
120 #url="http://"+topic_host+"/events/"+topic_write+"/users/mr-stub?timeout=15000&limit=100"
122 headers = {'Accept': 'application/json'}
123 print("Reading from dmaap: " + url)
124 resp=requests.get(url, headers=headers)
125 if (resp.status_code<199 & resp.status_code > 299):
126 print("Failed, response code: " + resp.status_code)
129 print("Recieved data from dmaap mr")
132 print("Recieved data (raw): " + str(resp.text))
133 if isinstance(data, list):
135 item=json.loads(item)
136 corrid=str(item["correlationId"])
137 status=str(item["status"])
138 msg=str(item["message"])
139 item_str=msg+status[0:3]
141 msg_responses[corrid]=item_str
142 cntr_msg_responses_submitted += 1
144 print("Data from dmaap is not json array: " + str(resp.text))
146 except Exception as e:
147 print("Corrupt data from dmaap mr - dropping " + str(data))
148 print("CAUGHT_EXCEPTION" + str(e) + " "+traceback.format_exc())
150 except Exception as e:
160 # Helper function to create a Dmaap request message
161 # args : <GET|PUT|DELETE> <correlation-id> <json-string-payload - may be None> <url>
162 # response: json formatted string of a complete Dmaap message
163 def create_message(operation, correlation_id, payload, url):
164 if (payload is None):
166 time_stamp=datetime.datetime.utcnow()
167 msg = '{\"apiVersion\":\"1.0\",\"operation\":\"'+operation+'\",\"correlationId\":\"'+correlation_id+'\",\"originatorId\": \"849e6c6b420\",'
168 msg = msg + '\"payload\":'+payload+',\"requestId\":\"23343221\", \"target\":\"policy-agent\", \"timestamp\":\"'+str(time_stamp)+'\", \"type\":\"request\",\"url\":\"'+url+'\"}'
172 ### MR-stub interface, for MR control
174 # Send a message to MR
175 # URI and parameters (PUT or POST): /send-request?operation=<GET|PUT|POST|DELETE>&url=<url>
176 # response: <correlation-id> (http 200) o4 400 for parameter error or 500 for other errors
177 @app.route(APP_WRITE_URL,
178 methods=['PUT','POST'])
181 global cntr_msg_requests_submitted
183 print("APP_WRITE_URL lock")
185 oper=request.args.get('operation')
187 print(APP_WRITE_URL+" parameter 'operation' missing")
188 return Response('Parameter operation missing in request', status=400, mimetype=MIME_TEXT)
190 url=request.args.get('url')
192 print(APP_WRITE_URL+" parameter 'url' missing")
193 return Response('Parameter url missing in request', status=400, mimetype=MIME_TEXT)
195 if (oper != "GET" and oper != "PUT" and oper != "POST" and oper != "DELETE"):
196 print(APP_WRITE_URL+" parameter 'operation' need to be: DEL|PUT|POST|DELETE")
197 return Response('Parameter operation does not contain DEL|PUT|POST|DELETE in request', status=400, mimetype=MIME_TEXT)
199 print(APP_WRITE_URL+" operation="+oper+" url="+url)
200 correlation_id=str(time.time_ns())
202 if (oper == "PUT") and (request.json is not None):
203 payload=json.dumps(request.json)
205 msg=create_message(oper, correlation_id, payload, url)
207 print(APP_WRITE_URL+" MSG(correlationid = "+correlation_id+"): " + json.dumps(json.loads(msg), indent=2))
208 msg_requests.append(msg)
209 cntr_msg_requests_submitted += 1
210 return Response(correlation_id, status=200, mimetype=MIME_TEXT)
211 except Exception as e:
212 print(APP_WRITE_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
213 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
215 # Receive a message response for MR for the included correlation id
216 # URI and parameter, (GET): /receive-response?correlationid=<correlation-id>
217 # response: <json-array of 1 response> 200 or empty 204 or other errors 500
218 @app.route(APP_READ_URL,
220 def receiveresponse():
222 global cntr_msg_responses_fetched
224 print("APP_READ_URL lock")
226 cid=request.args.get('correlationid')
228 print(APP_READ_URL+" parameter 'correclationid' missing")
229 return Response('Parameter correlationid missing in json', status=500, mimetype=MIME_TEXT)
231 if (cid in msg_responses):
232 answer=msg_responses[cid]
233 del msg_responses[cid]
234 print(APP_READ_URL+" response (correlationid="+cid+"): " + answer)
235 cntr_msg_responses_fetched += 1
236 return Response(answer, status=200, mimetype=MIME_JSON)
238 print(APP_READ_URL+" - no messages (correlationid="+cid+"): ")
239 return Response('', status=204, mimetype=MIME_JSON)
240 except Exception as e:
241 print(APP_READ_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
242 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
244 ### Dmaap interface ###
246 # Read messages stream. URI according to agent configuration.
247 # URI, (GET): /events/A1-POLICY-AGENT-READ/users/policy-agent
248 # response: 200 <json array of request messages>, or 500 for other errors
249 @app.route(AGENT_READ_URL,
253 global cntr_msg_requests_fetched
255 if topic_write or topic_read:
256 return Response('Url not available when running as mrstub frontend', status=404, mimetype=MIME_TEXT)
258 limit=request.args.get('limit')
267 print("Limting number of returned messages to: "+str(limit))
269 timeout=request.args.get('timeout')
270 if (timeout is None):
273 timeout=min(int(timeout),60000)
275 start_time=int(round(time.time() * 1000))
276 current_time=int(round(time.time() * 1000))
278 while(current_time<start_time+int(timeout)):
280 if(len(msg_requests)>0):
284 while(cntr<limit and len(msg_requests)>0):
287 msgs=msgs+msg_requests.pop(0)
288 cntr_msg_requests_fetched += 1
291 print(AGENT_READ_URL+" MSGs: "+json.dumps(json.loads(msgs), indent=2))
292 return Response(msgs, status=200, mimetype=MIME_JSON)
293 except Exception as e:
294 print(AGENT_READ_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
295 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
296 sleep(0.025) # sleep 25 milliseconds
297 current_time=int(round(time.time() * 1000))
299 print("timeout: "+str(timeout)+", start_time: "+str(start_time)+", current_time: "+str(current_time))
300 return Response("[]", status=200, mimetype=MIME_JSON)
302 # Write messages stream. URI according to agent configuration.
303 # URI and payload, (PUT or POST): /events/A1-POLICY-AGENT-WRITE <json array of response messages>
304 # response: OK 200 or 400 for missing json parameters, 500 for other errors
305 @app.route(AGENT_WRITE_URL,
306 methods=['PUT','POST'])
309 global cntr_msg_responses_submitted
311 if topic_write or topic_read:
312 return Response('Url not available when running as mrstub frontend', status=404, mimetype=MIME_TEXT)
315 print("AGENT_WRITE_URL lock")
318 print(AGENT_WRITE_URL+ " json=" + json.dumps(answer, indent=2))
319 if isinstance(answer, dict):
320 #Create a an array if the answer is a dict (single message)
322 answer_list.append(answer)
326 cid=item['correlationId']
328 print(AGENT_WRITE_URL+" parameter 'correlatonid' missing")
329 return Response('Parameter <correlationid> missing in json', status=400, mimetype=MIME_TEXT)
332 print(AGENT_WRITE_URL+" parameter 'msgs' missing")
333 return Response('Parameter >message> missing in json', status=400, mimetype=MIME_TEXT)
334 status=item['status']
336 print(AGENT_WRITE_URL+" parameter 'status' missing")
337 return Response('Parameter <status> missing in json', status=400, mimetype=MIME_TEXT)
338 if isinstance(msg, list) or isinstance(msg, dict):
339 msg_str=json.dumps(msg)+status[0:3]
341 msg_str=msg+status[0:3]
342 msg_responses[cid]=msg_str
343 cntr_msg_responses_submitted += 1
344 print(AGENT_WRITE_URL+ " msg+status (correlationid="+cid+") :" + str(msg_str))
345 except Exception as e:
346 print(AGENT_WRITE_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
347 return Response('{"message": "' + SERVER_ERROR + ' ' + str(e) + '","status":"500"}', status=200, mimetype=MIME_JSON)
349 return Response('{}', status=200, mimetype=MIME_JSON)
351 @app.route(ORU_WRITE_URL,
352 methods=['PUT','POST'])
355 msg=json.dumps(request.json)
356 msg_requests.append(msg)
357 return Response('{}', status=200, mimetype=MIME_JSON)
359 @app.route(ORU_READ_URL,
363 if(len(msg_requests)>0):
364 rsp=msg_requests.pop(0)
367 return Response(json.dumps(res), status=200, mimetype=MIME_JSON)
368 return Response("[]", status=200, mimetype=MIME_JSON)
370 # Generic POST/PUT catching all urls starting with /events/<topic>.
371 # Writes the message in a que for that topic
372 @app.route("/events/<path>",
373 methods=['PUT','POST'])
374 def generic_write(path):
375 global generic_messages
376 global cntr_msg_responses_submitted
377 urlkey="/events/"+str(path)
378 write_method=str(request.method)
382 print(write_method+" on "+urlkey+" json=" + json.dumps(payload))
384 if (urlkey in generic_messages.keys()):
385 topicmsgs=generic_messages[urlkey]
387 generic_messages[urlkey]=topicmsgs
389 if isinstance(payload, list):
390 for listitem in payload:
391 topicmsgs.append(listitem)
393 topicmsgs.append(payload)
395 cntr_msg_responses_submitted += 1
396 except Exception as e:
397 print(write_method + "on "+urlkey+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
398 return Response('{"message": "' + SERVER_ERROR + ' ' + str(e) + '","status":"500"}', status=200, mimetype=MIME_JSON)
400 return Response('{}', status=200, mimetype=MIME_JSON)
402 # Generic GET catching all urls starting with /events/. Returns max 4096 json msgs in an array.
403 # Returns only the messages previously written to the same urls
404 @app.route("/events/<path:path>",
406 def generic_read(path):
407 global generic_messages
408 global cntr_msg_requests_fetched
410 urlpath="/events/"+str(path)
411 urlkey="/events/"+str(path).split("/")[0] #Extract topic
412 print("GET on topic"+urlkey)
413 limit=request.args.get('limit')
422 print("Limting number of returned messages to: "+str(limit))
424 timeout=request.args.get('timeout')
425 if (timeout is None):
428 timeout=min(int(timeout),60000)
430 start_time=int(round(time.time() * 1000))
431 current_time=int(round(time.time() * 1000))
433 if (urlkey in generic_messages.keys()):
434 topicmsgs=generic_messages[urlkey]
436 while(current_time<start_time+int(timeout)):
438 if(len(topicmsgs)>0):
442 while(cntr<limit and len(topicmsgs)>0):
445 msgs=msgs+json.dumps(json.dumps(topicmsgs.pop(0)))
446 cntr_msg_requests_fetched += 1
449 print("GET on "+urlpath+" MSGs: "+msgs)
450 return Response(msgs, status=200, mimetype=MIME_JSON)
451 except Exception as e:
452 print("GET on "+urlpath+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
453 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
454 sleep(0.025) # sleep 25 milliseconds
455 current_time=int(round(time.time() * 1000))
457 print("timeout: "+str(timeout)+", start_time: "+str(start_time)+", current_time: "+str(current_time))
458 return Response("[]", status=200, mimetype=MIME_JSON)
461 ### Functions for metrics read out ###
463 @app.route('/counter/requests_submitted',
465 def requests_submitted():
466 return Response(str(cntr_msg_requests_submitted), status=200, mimetype=MIME_TEXT)
468 @app.route('/counter/requests_fetched',
470 def requests_fetched():
471 return Response(str(cntr_msg_requests_fetched), status=200, mimetype=MIME_TEXT)
473 @app.route('/counter/responses_submitted',
475 def responses_submitted():
476 return Response(str(cntr_msg_responses_submitted), status=200, mimetype=MIME_TEXT)
478 @app.route('/counter/responses_fetched',
480 def responses_fetched():
481 return Response(str(cntr_msg_responses_fetched), status=200, mimetype=MIME_TEXT)
483 @app.route('/counter/current_requests',
485 def current_requests():
486 return Response(str(len(msg_requests)), status=200, mimetype=MIME_TEXT)
488 @app.route('/counter/current_responses',
490 def current_responses():
491 return Response(str(len(msg_responses)), status=200, mimetype=MIME_TEXT)
495 # Reset all messsages and counters
497 methods=['GET', 'POST', 'PUT'])
499 global cntr_msg_requests_submitted
500 global cntr_msg_requests_fetched
501 global cntr_msg_responses_submitted
502 global cntr_msg_responses_fetched
506 cntr_msg_requests_submitted=0
507 cntr_msg_requests_fetched=0
508 cntr_msg_responses_submitted=0
509 cntr_msg_responses_fetched=0
512 return Response('OK', status=200, mimetype=MIME_TEXT)
514 # Get env vars, if present
515 if os.getenv("TOPIC_READ") is not None:
517 print("Env variables:")
518 print("TOPIC_READ:"+os.environ['TOPIC_READ'])
519 print("TOPIC_WRITE:"+os.environ['TOPIC_WRITE'])
521 topic_read=os.environ['TOPIC_READ']
522 topic_write=os.environ['TOPIC_WRITE']
525 if topic_read and downloader_thread is None:
526 downloader_thread=Thread(target=dmaap_downloader)
527 downloader_thread.start()
529 if topic_write and uploader_thread is None:
530 uploader_thread=Thread(target=dmaap_uploader)
531 uploader_thread.start()
534 print("No env variables - OK")
536 if __name__ == "__main__":
537 app.run(port=HOST_PORT, host=HOST_IP)