Merge pull request #3 from prashantramangupta/master

raamb · web-flow · commit 75c79b5a8971 · 2019-05-07T16:49:53.000+05:30
migration of daemon reports data from redshift to RDS.
diff --git a/build.sh b/build.sh
diff --git a/constant.py b/constant.py
@@ -0,0 +1,20 @@
+NETWORKS = {
+    999: {
+        'db': {'DB_HOST': '',
+               'DB_USER': '',
+               'DB_PASSWORD': '',
+               'DB_NAME': '',
+               'DB_PORT': 3306
+               }
+    }
+}
+SLACK_HOOK = {
+    'hostname' : '',
+    'port': 443,
+    'path': '',
+    'method': 'POST',
+    'headers': {
+        'Content-Type': 'application/json'
+    }
+}
+METRICS_NETWORK_ID  = 999
diff --git a/lambda_handler.py b/lambda_handler.py
diff --git a/platform-usage-authorizer/authorizer.py b/platform-usage-authorizer/authorizer.py
@@ -0,0 +1,14 @@
+from repository import Repository
+from constant import METRICS_NETWORK_ID
+
+class Token:
+    def __init__(self):
+        self.repo = Repository(METRICS_NETWORK_ID)
+
+    def validate_token(self, daemon_id, token):
+        print("validate_token::daemon_id: ", daemon_id);
+        qry = "SELECT * FROM daemon_token WHERE daemon_id = %s and token = %s "
+        res = self.repo.execute(qry, [daemon_id, token])
+        if len(res) > 0:
+            return {'validated': True}
+        return {'validated': False}
diff --git a/platform-usage-authorizer/build.sh b/platform-usage-authorizer/build.sh
@@ -0,0 +1,2 @@
+rm daemon_authorizer.zip
+zip -r daemon_authorizer.zip * -x \*venv\*
diff --git a/platform-usage-authorizer/lambda_handler.py b/platform-usage-authorizer/lambda_handler.py
@@ -0,0 +1,67 @@
+import json
+import os
+
+from authorizer import Token
+from utils import Utils
+
+obj_util = Utils()
+
+def request_handler(event, context):
+    if 'path' not in event:
+        return get_response("400", {"status": "failed", "error": "Bad Request"})
+    try:
+        data = None
+        payload_dict = None
+        path = event['path'].lower()
+        if "/event" == path:
+            payload_dict = event['headers']
+            print("Processing [" + str(path) + "] with body [" + str(payload_dict) + "]")
+            token_instance = Token()
+            data = token_instance.validate_token(daemon_id=payload_dict['x-daemonid'],
+                                                 token=payload_dict['x-token'])
+        else:
+            return get_response(500, "Invalid URL path.")
+        return get_lambda_authorizer_response_format(event=event, allow=data.get('validated', False))
+    except Exception as e:
+        err_msg = {"status": "failed",
+                   "error": repr(e),
+                   "api": event['path'],
+                   "payload": payload_dict,
+                   "type": "authorize"}
+        obj_util.report_slack(1, str(err_msg))
+        return get_response(500, err_msg)
+
+
+# Generate response JSON that API gateway expects from the lambda function
+def get_response(status_code, message):
+    return {
+        'statusCode': status_code,
+        'body': json.dumps(message),
+        'headers': {
+            'Content-Type': 'application/json',
+            "X-Requested-With": '*',
+            "Access-Control-Allow-Headers": 'Access-Control-Allow-Origin, Content-Type,X-Amz-Date,Authorization,X-Api-Key,x-requested-with',
+            "Access-Control-Allow-Origin": '*',
+            "Access-Control-Allow-Methods": 'GET,OPTIONS,POST'
+        }
+    }
+
+
+def get_lambda_authorizer_response_format(event, allow):
+    response = {
+        "principalId": os.environ['principalId'],
+        "policyDocument": {
+            "Version": '2012-10-17',
+            "Statement": [
+                {
+                    "Action": 'execute-api:Invoke',
+                    "Resource": event['methodArn']
+                }
+            ]
+        }
+    }
+    if allow:
+        response["policyDocument"]["Statement"][0]["Effect"] = 'Allow'
+    else:
+        response["policyDocument"]["Statement"][0]["Effect"] = 'Deny'
+    return response
diff --git a/platform-usage-authorizer/requirement.txt b/platform-usage-authorizer/requirement.txt
@@ -0,0 +1,2 @@
+pymysql==0.9.2
+requests==2.20.1
diff --git a/platform-usage-process/build.sh b/platform-usage-process/build.sh
@@ -0,0 +1,2 @@
+rm platform_usage_process.zip
+zip -r platform_usage_process.zip * -x \*venv\*
diff --git a/platform-usage-process/lambda_handler.py b/platform-usage-process/lambda_handler.py
@@ -0,0 +1,76 @@
+import json
+
+from metrics import Metrics
+from register import Token
+from utils import Utils
+
+obj_util = Utils()
+
+
+def request_handler(event, context):
+    if 'path' not in event:
+        return get_response("400", {"status": "failed",
+                                    "error": "Bad Request"})
+    try:
+        data = None
+        payload_dict = None
+        path = event['path'].lower()
+        payload_dict = payload_check(payload=event['body'], path=path)
+        if "/register" == path:
+            token_instance = Token()
+            data = token_instance.process_token(daemon_id=payload_dict['daemonId'])
+        elif "/event" == path:
+            obj_metrics = Metrics()
+            if payload_dict['type'] == 'request':
+                obj_metrics.handle_request_type(payload_dict)
+            elif payload_dict['type'] == 'response':
+                obj_metrics.handle_response_type(payload_dict)
+            data = {}
+        else:
+            return get_response(500, "Invalid URL path.")
+
+        if data is None:
+            response = get_response("400", {"status": "failed",
+                                            "error": "Bad Request",
+                                            "api": event['path'],
+                                            "payload": payload_dict})
+        else:
+            if data.get('error', '') == '':
+                response = get_response("200", {"status": "success", "data": data})
+            else:
+                error = data['error']
+                data.pop('error')
+                response = get_response("200", {"status": "failed", "data": data, "error": error})
+
+        return response
+    except Exception as e:
+        err_msg = {"status": "failed",
+                   "error": repr(e),
+                   "api": event['path'],
+                   "payload": payload_dict,
+                   "type": "process"}
+        obj_util.report_slack(1, str(err_msg))
+        return get_response(500, err_msg)
+
+
+def payload_check(payload, path):
+    payload_dict = None
+    if payload is not None and len(payload) > 0:
+        payload_dict = json.loads(payload)
+    print("Processing [" + str(path) + "] with body [" + str(payload) + "]")
+    return payload_dict
+
+
+# Generate response JSON that API gateway expects from the lambda function
+def get_response(status_code, message):
+    return {
+        'statusCode': status_code,
+        'body': json.dumps(message),
+        'headers': {
+            'Content-Type': 'application/json',
+            "X-Requested-With": '*',
+            "Access-Control-Allow-Headers": 'Access-Control-Allow-Origin, Content-Type,X-Amz-Date,Authorization,X-Api-Key,x-requested-with',
+            "Access-Control-Allow-Origin": '*',
+            "Access-Control-Allow-Methods": 'GET,OPTIONS,POST'
+        }
+    }
diff --git a/platform-usage-process/metrics.py b/platform-usage-process/metrics.py
@@ -0,0 +1,42 @@
+from datetime import datetime as dt
+
+from constant import METRICS_NETWORK_ID
+from repository import Repository
+
+
+class Metrics:
+    def __init__(self):
+        self.repo = Repository(METRICS_NETWORK_ID)
+
+    def handle_request_type(self, params):
+        try:
+            insrt_dm_rq_sts = "INSERT INTO daemon_request_stats (ethereum_json_rpc_endpoint, group_id, input_data_size, " \
+                              "organization_id, registry_address_key, request_id, request_received_time, service_id, " \
+                              "service_method, row_created, row_updated) " \
+                              "VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
+            dm_req_params = (params['ethereum_json_rpc_endpoint'], params['group_id'], params['input_data_size'],
+                             params['organization_id'], params['registry_address_key'], params['request_id'],
+                             params['request_received_time'][:19], params['service_id'], params['service_method'],
+                             dt.utcnow(),
+                             dt.utcnow())
+            self.repo.execute(insrt_dm_rq_sts, dm_req_params)
+        except Exception as e:
+            print(repr(e))
+            raise e
+
+    def handle_response_type(self, params):
+        try:
+            insrt_dm_rs_sts = "INSERT INTO daemon_response_stats (error_message, ethereum_json_rpc_endpoint, group_id, " \
+                              "organization_id, registry_address_key, request_id, request_received_time, response_code, " \
+                              "response_sent_time, response_time, service_id, service_method, row_created, row_updated) " \
+                              "VALUES(%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"
+            dm_rs_params = (params['error_message'], params['ethereum_json_rpc_endpoint'], params['group_id'],
+                            params['organization_id'], params['registry_address_key'], params['request_id'],
+                            params['request_received_time'][:19], params['response_code'],
+                            params['response_sent_time'][:19],
+                            params['response_time'], params['service_id'], params['service_method'], dt.utcnow(),
+                            dt.utcnow())
+            self.repo.execute(insrt_dm_rs_sts, dm_rs_params)
+        except Exception as e:
+            print(repr(e))
+            raise e
diff --git a/platform-usage-process/register.py b/platform-usage-process/register.py
@@ -3,13 +3,13 @@
 import time
 from datetime import datetime as dt
 
+from constant import METRICS_NETWORK_ID
 from repository import Repository
 
 
 class Token:
-    def __init__(self, net_id):
-        self.net_id = net_id
-        self.repo = Repository(net_id)
+    def __init__(self):
+        self.repo = Repository(METRICS_NETWORK_ID)
 
     def process_token(self, daemon_id):
         print("process_token::daemon_id: ", daemon_id);
@@ -33,15 +33,6 @@ def process_token(self, daemon_id):
             result["token"] = token
         return result
 
-    def validate_token(self, daemon_id, token):
-        print("validate_token::daemon_id: ", daemon_id);
-        qry = "SELECT * FROM daemon_token WHERE daemon_id = %s and token = %s "
-        res = self.repo.execute(qry, [daemon_id, token])
-        if len(res) > 0:
-            return {'validated': True}
-        return {'validated': False}
-
-
     def daemon_id_exist(self, daemon_id):
         print("daemon_id_exist::daemon_id: ", daemon_id);
         qry = "SELECT * FROM daemon_token WHERE daemon_id = %s"
diff --git a/platform-usage-process/requirement.txt b/platform-usage-process/requirement.txt
@@ -0,0 +1,2 @@
+pymysql==0.9.2
+requests==2.20.1
diff --git a/requirement.txt b/requirement.txt
diff --git a/utils.py b/utils.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+rm daemon_authorizer.zip`
	`2`	`+zip -r daemon_authorizer.zip * -x \venv\`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,2 @@`
	`1`	`+rm platform_usage_process.zip`
	`2`	`+zip -r platform_usage_process.zip * -x \venv\`