configuration/kubernetes/deployment.yaml

@@ -99,7 +99,7 @@ spec:
         imagePullPolicy: Always
         resources:
           requests:
-            cpu: 25m
+            cpu: 32m
             memory: 32Mi
           limits:
             cpu: 25m

src/pricing/collector.py

@@ -12,13 +12,16 @@ from pricing.exchanges.gemini import geminiPublicTicker, geminiVolAskBid, gemini
 from pricing.exchanges.bitstamp import bitstampVolAskBid, bitstampHighLow, bitstampOpenClose
 from pricing.exchanges.kraken import krakenVolAskBid, krakenHighLow, krakenOpenClose
 
-# from src.utils.databaseConnect import send
+from src.utils.databaseConnect import send
 from src.utils.activemqConnect import activeMQSender
 
-import json, uuid
+import json
 
 from src.utils.jsonLogger import log
 
+btc_usd="/home/price-collector/src/resources/queries/V1_INSERT_NEW_PRICE_RECORD_BTC.graphql"
+xmr_usd="/home/price-collector/src/resources/queries/V1_INSERT_NEW_PRICE_RECORD_XMR.graphql"
+
 def averager(type):
 
     timestamp = datetime.datetime.now() + datetime.timedelta(hours=1)
@@ -74,23 +77,23 @@ def getOpenClose(type):
 
     return open, close
 
-# def sendToGateway(c_type, timestamp, av_price, high, low, vol, o_price, c_price):
+def sendToGateway(c_type, timestamp, av_price, high, low, vol, o_price, c_price):
-#
+
-#     with open(btc_usd, 'r') as file:
+    with open(btc_usd, 'r') as file:
-#         data = file.read()
+        data = file.read()
-#
+
-#     strippedTimestamp = timestamp.replace(minute=0, second=0, microsecond=0)
+    strippedTimestamp = timestamp.replace(minute=0, second=0, microsecond=0)
-#     timestamp = '"'+strippedTimestamp.strftime('%Y-%m-%dT%H:%M:%S')+'"'
+    timestamp = '"'+strippedTimestamp.strftime('%Y-%m-%dT%H:%M:%S')+'"'
-#     type = '"'+c_type+'"'
+    type = '"'+c_type+'"'
-#
+
-#     query = data % (timestamp, type, round(av_price, 2),
+    query = data % (timestamp, type, round(av_price, 2),
-#                                     round(high, 2),
+                                    round(high, 2),
-#                                     round(low, 2),
+                                    round(low, 2),
-#                                     round(o_price, 2),
+                                    round(o_price, 2),
-#                                     round(c_price, 2),
+                                    round(c_price, 2),
-#                                     round(vol, 2))
+                                    round(vol, 2))
-#
+
-#     log("Query sending down to db-gateway -- ({})".format(query), 'INFO')
+    log("Query sending down to db-gateway -- ({})".format(query), 'INFO')
 
     # status, response = send(query, log)
 
@@ -119,12 +122,11 @@ def sentToArtemis(c_type, timestamp, av_price, high, low, vol, o_price, c_price)
 
     messageJson = json.dumps(message, indent = 4)
 
-    syncId = uuid.uuid4()
+    log("Sending message to PricingSave queue", 'INFO')
+    log("Message: {}".format(message), 'INFO')
 
-    log("Sending message to PricingSave queue", 'INFO', syncId)
+    activeMQSender(messageJson)
-    log("Message: {}".format(message), 'INFO', syncId)
 
-    activeMQSender(messageJson, syncId)
 
 def timeFunction():
     global time

src/resources/queries/V1_INSERT_NEW_PRICE_RECORD_BTC.graphql

@@ -0,0 +1 @@
+mutation { createBtc(createdDate: %s, type: %s, average_price: %f, high_price: %f, low_price: %f, open_price: %f, close_price: %f, volume: %f){ id } }

src/utils/activemqConnect.py

@@ -17,20 +17,14 @@ class keys():
     def returnKeys(self):
         return self.addr, self.port, self.amqU, self.amqP
 
-def activeMQSender(message, syncId):
+def activeMQSender(message):
     addr, port, mqUser, mqPass = keys().returnKeys()
 
-    log("Attempting Connection to Artemis...", 'INFO', syncId)
+    log("Attempting Connection to Artemis...", 'INFO')
     con = stomp.Connection([(addr, port)], auto_content_length=False)
     con.connect( mqUser, mqPass, wait=True)
 
-    con.send("PricingSave",
+    con.send("PricingSave", message, content_type="application/json", headers={"Content-Type":"application/json"})
-             message,
-             content_type="application/json",
-             headers={
-                 "Content-Type":"application/json",
-                 "X-CRYPTO-Sync-ID":syncId
-             })
 
     con.disconnect()
 

src/utils/jsonLogger.py

@@ -28,13 +28,13 @@ def setup_logging(log_level='INFO'):
     logHandler.setFormatter(formatter)
     logger.addHandler(logHandler)
 
-def log(message, level, syncId=""):
+def log(message, level):
     logger = logging.getLogger(__name__)
     if level == 'INFO':
-        logger.info(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.info(message)
     elif level == 'WARN':
-        logger.warn(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.warn(message)
     elif level == 'ERR':
-        logger.error(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.error(message)
     elif level == 'DEBUG':
-        logger.debug(message, extra={"X-CRYPTO-Sync-ID" : syncId})
+        logger.debug(message)