EOX GitLab Instance

Commit 4ce3ccf1 authored by Lubomir Doležal's avatar Lubomir Doležal
Browse files

update ingestor to push via vsq

parent 258d0f96
Pipeline #17476 failed with stages
in 2 minutes and 12 seconds
......@@ -53,6 +53,6 @@ ENV INSTANCE_ID="view-server_ingestor" \
INGESTOR_SUCCESS_DIR="/var/ingestor/success" \
INGESTOR_FAIL_DIR="/var/ingestor/fail"
CMD ["gunicorn3", "-c", "config.py", "app"]
CMD ["gunicorn", "-c", "config.py", "app"]
LABEL version="1.5.2"
......@@ -32,7 +32,8 @@ import logging.config
import json
from flask import Flask, request, Response
import redis
from redis import Redis
from vsq.queue import Queue
from ingestor.browse_report import parse_browse_report
from ingestor.util import converter, save_endpoint_report, browse_name
......@@ -69,36 +70,39 @@ logging.config.dictConfig({
}
})
client = redis.Redis(
queue_name = os.environ['REDIS_PREPROCESS_MD_QUEUE_KEY']
queue = Queue(queue_name, Redis(
host=os.environ['REDIS_HOST'],
port=int(os.environ.get('REDIS_PORT', '6379')),
charset="utf-8",
decode_responses=True,
charset='utf-8',
decode_responses=True)
)
queue_name = os.environ['REDIS_PREPROCESS_MD_QUEUE_KEY']
@application.route('/', methods=['POST'])
def ingest():
try:
request.get_data()
if isinstance(request.data, (bytes, bytearray)):
data_from_request = request.data.decode('utf-8')
elif isinstance(request.data, str):
data_from_request = request.data
else:
logger.warn("Something wierd received, neither byte or string.")
browse_report = parse_browse_report(request.data)
logger.debug(browse_report)
if isinstance(browse_report, str):
queue_content = browse_report
filename = queue_content
filename = browse_report
else:
queue_content = json.dumps(
browse_report, default=converter
)
filename = browse_name(browse_report)
task = queue.put({
'type': 'message',
'payload': browse_report,
})
logger.debug("%s %s %s" % (task.id, task.status, task.created))
client.lpush(queue_name, queue_content)
save_endpoint_report(filename, request.data, True)
save_endpoint_report(filename, data_from_request, True)
return Response(status=202)
except Exception as e:
save_endpoint_report(filename, request.data, False)
save_endpoint_report(filename, data_from_request, False)
return Response(str(e), status=400)
......@@ -34,7 +34,8 @@ import functools
import operator
import pyinotify
import redis
from redis import Redis
from vsq.queue import Queue
from ingestor.browse_report import parse_browse_report
from ingestor.util import converter, save_mount_report
......@@ -76,14 +77,12 @@ inotify_modes = os.environ.get('INOTIFY_MASKS', 'IN_MOVED_TO,IN_CLOSE_WRITE').sp
# bitwise or on all elements of configured inotify modes to watch for
inotify_mask = functools.reduce(operator.or_, [getattr(pyinotify, mode, 0) for mode in inotify_modes], 0)
client = redis.Redis(
queue = Queue(queue_name, Redis(
host=os.environ['REDIS_HOST'],
port=int(os.environ.get('REDIS_PORT', '6379')),
charset='utf-8',
decode_responses=True,
decode_responses=True)
)
watchmanager = pyinotify.WatchManager()
......@@ -95,15 +94,11 @@ class EventHandler(pyinotify.ProcessEvent):
with open(event.pathname) as f:
browse_report = parse_browse_report(f)
logger.debug(browse_report)
if isinstance(browse_report, str):
queue_content = browse_report
else:
queue_content = json.dumps(
browse_report, default=converter
)
client.lpush(queue_name, queue_content)
task = queue.put({
'type': 'message',
'payload': browse_report,
})
logger.debug("%s %s %s" % (task.id, task.status, task.created))
save_mount_report(event.pathname, True)
except Exception as e:
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment