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
30 # Disable all logging of GET on reading counters
31 class AjaxFilter(logging.Filter):
32 def filter(self, record):
33 return ("/counter/" not in record.getMessage())
35 log = logging.getLogger('werkzeug')
36 log.addFilter(AjaxFilter())
40 # list of messages to/from Dmaap
49 cntr_msg_requests_submitted=0
50 cntr_msg_requests_fetched=0
51 cntr_msg_responses_submitted=0
52 cntr_msg_responses_fetched=0
54 # Request and response constants
55 AGENT_WRITE_URL="/events/A1-POLICY-AGENT-WRITE"
56 AGENT_READ_URL="/events/A1-POLICY-AGENT-READ/users/policy-agent"
57 APP_WRITE_URL="/send-request"
58 APP_READ_URL="/receive-response"
59 MIME_TEXT="text/plain"
60 MIME_JSON="application/json"
61 CAUGHT_EXCEPTION="Caught exception: "
62 SERVER_ERROR="Server error :"
71 # Helper function to create a Dmaap request message
72 # args : <GET|PUT|DELETE> <correlation-id> <json-string-payload - may be None> <url>
73 # response: json formatted string of a complete Dmaap message
74 def create_message(operation, correlation_id, payload, url):
77 time_stamp=datetime.datetime.utcnow()
78 msg = '{\"apiVersion\":\"1.0\",\"operation\":\"'+operation+'\",\"correlationId\":\"'+correlation_id+'\",\"originatorId\": \"849e6c6b420\",'
79 msg = msg + '\"payload\":'+payload+',\"requestId\":\"23343221\", \"target\":\"policy-agent\", \"timestamp\":\"'+str(time_stamp)+'\", \"type\":\"request\",\"url\":\"'+url+'\"}'
83 ### MR-stub interface, for MR control
85 # Send a message to MR
86 # URI and parameters (GET): /send-request?operation=<GET|PUT|POST|DELETE>&url=<url>
87 # response: <correlation-id> (http 200) o4 400 for parameter error or 500 for other errors
88 @app.route(APP_WRITE_URL,
89 methods=['PUT','POST'])
92 global cntr_msg_requests_submitted
94 print("APP_WRITE_URL lock")
96 oper=request.args.get('operation')
98 print(APP_WRITE_URL+" parameter 'operation' missing")
99 return Response('Parameter operation missing in request', status=400, mimetype=MIME_TEXT)
101 url=request.args.get('url')
103 print(APP_WRITE_URL+" parameter 'url' missing")
104 return Response('Parameter url missing in request', status=400, mimetype=MIME_TEXT)
106 if (oper != "GET" and oper != "PUT" and oper != "POST" and oper != "DELETE"):
107 print(APP_WRITE_URL+" parameter 'operation' need to be: DEL|PUT|POST|DELETE")
108 return Response('Parameter operation does not contain DEL|PUT|POST|DELETE in request', status=400, mimetype=MIME_TEXT)
110 print(APP_WRITE_URL+" operation="+oper+" url="+url)
111 correlation_id=str(time.time_ns())
113 if (oper == "PUT") and (request.json is not None):
114 payload=json.dumps(request.json)
116 msg=create_message(oper, correlation_id, payload, url)
118 print(APP_WRITE_URL+" MSG(correlationid = "+correlation_id+"): " + json.dumps(json.loads(msg), indent=2))
119 msg_requests.append(msg)
120 cntr_msg_requests_submitted += 1
121 return Response(correlation_id, status=200, mimetype=MIME_TEXT)
122 except Exception as e:
123 print(APP_WRITE_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
124 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
126 # Receive a message response for MR for the included correlation id
127 # URI and parameter, (GET): /receive-response?correlationid=<correlation-id>
128 # response: <json-array of 1 response> 200 or empty 204 or other errors 500
129 @app.route(APP_READ_URL,
131 def receiveresponse():
133 global cntr_msg_responses_fetched
135 print("APP_READ_URL lock")
137 id=request.args.get('correlationid')
139 print(APP_READ_URL+" parameter 'correclationid' missing")
140 return Response('Parameter correlationid missing in json', status=500, mimetype=MIME_TEXT)
142 if (id in msg_responses):
143 answer=msg_responses[id]
144 del msg_responses[id]
145 print(APP_READ_URL+" response (correlationid="+id+"): " + answer)
146 cntr_msg_responses_fetched += 1
147 return Response(answer, status=200, mimetype=MIME_JSON)
149 print(APP_READ_URL+" - no messages (correlationid="+id+"): ")
150 return Response('', status=204, mimetype=MIME_JSON)
151 except Exception as e:
152 print(APP_READ_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
153 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
155 ### Dmaap interface ###
157 # Read messages stream. URI according to agent configuration.
158 # URI, (GET): /events/A1-POLICY-AGENT-READ/users/policy-agent
159 # response: 200 <json array of request messages>, or 500 for other errors
160 @app.route(AGENT_READ_URL,
164 global cntr_msg_requests_fetched
166 limit=request.args.get('limit')
175 print("Limting number of returned messages to: "+str(limit))
177 timeout=request.args.get('timeout')
178 if (timeout is None):
181 timeout=min(int(timeout),60000)
183 startTime=int(round(time.time() * 1000))
184 currentTime=int(round(time.time() * 1000))
186 while(currentTime<startTime+int(timeout)):
188 if(len(msg_requests)>0):
192 while(cntr<limit and len(msg_requests)>0):
195 msgs=msgs+msg_requests.pop(0)
196 cntr_msg_requests_fetched += 1
199 print(AGENT_READ_URL+" MSGs: "+json.dumps(json.loads(msgs), indent=2))
200 return Response(msgs, status=200, mimetype=MIME_JSON)
201 except Exception as e:
202 print(AGENT_READ_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
203 return Response(SERVER_ERROR+" "+str(e), status=500, mimetype=MIME_TEXT)
204 sleep(0.025) # sleep 25 milliseconds
205 currentTime=int(round(time.time() * 1000))
207 print("timeout: "+str(timeout)+", startTime: "+str(startTime)+", currentTime: "+str(currentTime))
208 return Response("[]", status=200, mimetype=MIME_JSON)
210 # Write messages stream. URI according to agent configuration.
211 # URI and payload, (PUT or POST): /events/A1-POLICY-AGENT-WRITE <json array of response messages>
212 # response: OK 200 or 400 for missing json parameters, 500 for other errors
213 @app.route(AGENT_WRITE_URL,
214 methods=['PUT','POST'])
217 global cntr_msg_responses_submitted
219 print("AGENT_WRITE_URL lock")
222 print(AGENT_WRITE_URL+ " json=" + json.dumps(answer, indent=2))
223 if isinstance(answer, dict):
224 #Create a an array if the answer is a dict (single message)
226 answer_list.append(answer)
230 id=item['correlationId']
232 print(AGENT_WRITE_URL+" parameter 'correlatonid' missing")
233 return Response('Parameter <correlationid> missing in json', status=400, mimetype=MIME_TEXT)
236 print(AGENT_WRITE_URL+" parameter 'msgs' missing")
237 return Response('Parameter >message> missing in json', status=400, mimetype=MIME_TEXT)
238 status=item['status']
240 print(AGENT_WRITE_URL+" parameter 'status' missing")
241 return Response('Parameter <status> missing in json', status=400, mimetype=MIME_TEXT)
242 if isinstance(msg, list) or isinstance(msg, dict):
243 msg_str=json.dumps(msg)+status[0:3]
245 msg_str=msg+status[0:3]
246 msg_responses[id]=msg_str
247 cntr_msg_responses_submitted += 1
248 print(AGENT_WRITE_URL+ " msg+status (correlationid="+id+") :" + str(msg_str))
249 except Exception as e:
250 print(AGENT_WRITE_URL+"-"+CAUGHT_EXCEPTION+" "+str(e) + " "+traceback.format_exc())
251 return Response('{"message": "' + SERVER_ERROR + ' ' + str(e) + '","status":"500"}', status=200, mimetype=MIME_JSON)
253 return Response('{}', status=200, mimetype=MIME_JSON)
256 ### Functions for metrics read out ###
258 @app.route('/counter/requests_submitted',
260 def requests_submitted():
261 return Response(str(cntr_msg_requests_submitted), status=200, mimetype=MIME_TEXT)
263 @app.route('/counter/requests_fetched',
265 def requests_fetched():
266 return Response(str(cntr_msg_requests_fetched), status=200, mimetype=MIME_TEXT)
268 @app.route('/counter/responses_submitted',
270 def responses_submitted():
271 return Response(str(cntr_msg_responses_submitted), status=200, mimetype=MIME_TEXT)
273 @app.route('/counter/responses_fetched',
275 def responses_fetched():
276 return Response(str(cntr_msg_responses_fetched), status=200, mimetype=MIME_TEXT)
278 @app.route('/counter/current_requests',
280 def current_requests():
281 return Response(str(len(msg_requests)), status=200, mimetype=MIME_TEXT)
283 @app.route('/counter/current_responses',
285 def current_responses():
286 return Response(str(len(msg_responses)), status=200, mimetype=MIME_TEXT)
290 # Reset all messsages and counters
292 methods=['GET', 'POST', 'PUT'])
294 global cntr_msg_requests_submitted
295 global cntr_msg_requests_fetched
296 global cntr_msg_responses_submitted
297 global cntr_msg_responses_fetched
301 cntr_msg_requests_submitted=0
302 cntr_msg_requests_fetched=0
303 cntr_msg_responses_submitted=0
304 cntr_msg_responses_fetched=0
307 return Response('OK', status=200, mimetype=MIME_TEXT)
309 if __name__ == "__main__":
310 app.run(port=HOST_PORT, host=HOST_IP)