Dev mac experiments

.gitignore

@@ -29,3 +29,4 @@ examples/k8s/charts
 target
 wheelhouse
 requirements.*
+.DS_Store

examples/exampledata/config/http_pipeline.yml

@@ -3,6 +3,7 @@ process_count: 4
 config_refresh_interval: 5
 profile_pipelines: false
 restart_count: 3
+error_backlog_size: 150
 logger:
   level: INFO
   loggers:
@@ -29,7 +30,7 @@ metrics:
 input:
   httpinput:
     type: http_input
-    message_backlog_size: 1500000
+    message_backlog_size: 150
     collect_meta: true
     metafield_name: "@metadata"
     uvicorn_config:

examples/exampledata/config/pipeline.yml

@@ -3,9 +3,9 @@ process_count: 2
 timeout: 0.1
 restart_count: 2
 config_refresh_interval: 5
-error_backlog_size: 1500000
+error_backlog_size: 150
 logger:
-  level: INFO
+  level: DEBUG
   format: "%(asctime)-15s %(hostname)-5s %(name)-10s %(levelname)-8s: %(message)s"
   datefmt: "%Y-%m-%d %H:%M:%S"
   loggers:

logprep/connector/http/input.py

@@ -479,7 +479,7 @@ def _get_asgi_app(endpoints_config: dict) -> falcon.asgi.App:
 
     def _get_event(self, timeout: float) -> Tuple:
         """Returns the first message from the queue"""
-        self.metrics.message_backlog_size += self.messages.qsize()
+        #self.metrics.message_backlog_size += self.messages.qsize()
         try:
             message = self.messages.get(timeout=timeout)
             raw_message = str(message).encode("utf8")

logprep/framework/pipeline.py

@@ -336,7 +336,7 @@ def _drain_input_queues(self) -> None:
         if not hasattr(self._input, "messages"):
             return
         if isinstance(self._input.messages, multiprocessing.queues.Queue):
-            while self._input.messages.qsize():
+            while not self._input.messages.empty():
                 self.process_pipeline()
 
     def stop(self) -> None:

logprep/framework/pipeline_manager.py

@@ -52,7 +52,7 @@ def throttle(self, batch_size=1):
 
     def put(self, obj, block=True, timeout=None, batch_size=1):
         """Put an obj into the queue."""
-        self.throttle(batch_size)
+        #self.throttle(batch_size)
         super().put(obj, block=block, timeout=timeout)
 
 

logprep/runner.py

@@ -174,8 +174,8 @@ def _iterate(self):
                 self.exit_code = EXITCODES.PIPELINE_ERROR.value
                 self._logger.error("Restart count exceeded. Exiting.")
                 sys.exit(self.exit_code)
-            if self._manager.error_queue is not None:
-                self.metrics.number_of_events_in_error_queue += self._manager.error_queue.qsize()
+            #if self._manager.error_queue is not None:
+            #    self.metrics.number_of_events_in_error_queue += self._manager.error_queue.qsize()
             self._manager.restart_failed_pipeline()
 
     def reload_configuration(self):

logprep/util/logging.py

@@ -4,6 +4,12 @@
 import multiprocessing as mp
 from logging.handlers import QueueListener
 from socket import gethostname
+import platform
+if platform.system() != "Linux":
+    from multiprocessing import set_start_method
+    set_start_method("fork")
+
+
 
 logqueue = mp.Queue(-1)