feat: integrate async in ng

.vscode/launch.json

@@ -2,11 +2,11 @@
     "version": "0.2.0",
     "configurations": [
         {
-            "name": "Debug non-ng example pipeline",
+            "name": "non-ng example pipeline",
             "type": "debugpy",
             "request": "launch",
             "program": "logprep/run_logprep.py",
-            "console": "integratedTerminal",
+            "console": "internalConsole",
             "args": [
                 "run",
                 "examples/exampledata/config/pipeline.yml"
@@ -15,6 +15,21 @@
                 "PROMETHEUS_MULTIPROC_DIR": "tmp/logprep"
             },
             "justMyCode": false
+        },
+        {
+            "name": "ng example pipeline",
+            "type": "debugpy",
+            "request": "launch",
+            "program": "logprep/run_ng.py",
+            "console": "internalConsole",
+            "args": [
+                "run",
+                "examples/exampledata/config/ng_pipeline.yml"
+            ],
+            "env": {
+                "PROMETHEUS_MULTIPROC_DIR": "tmp/logprep"
+            },
+            "justMyCode": false
         }
     ]
 }

benchmark.py

@@ -591,6 +591,7 @@ def benchmark_run(
         binary = "logprep-ng" if ng == 1 else "logprep"
 
         t_startup = time.time()
+
         logprep_proc = popen_cmd([binary, "run", str(pipeline_config)], env=env)
         _current_logprep_proc = logprep_proc
 

examples/exampledata/config/_benchmark_ng_pipeline.yml

@@ -0,0 +1,125 @@
+version: 2
+process_count: 1
+timeout: 5.0
+restart_count: 2
+config_refresh_interval: 5
+error_backlog_size: 15000
+logger:
+  level: DEBUG
+  format: "%(asctime)-15s %(hostname)-5s %(name)-10s %(levelname)-8s: %(message)s"
+  datefmt: "%Y-%m-%d %H:%M:%S"
+  loggers:
+    "py.warnings": {"level": "ERROR"}
+    "Runner": {"level": "INFO"}
+    "Processor": {"level": "ERROR"}
+    "Exporter": {"level": "ERROR"}
+    "uvicorn": {"level": "ERROR"}
+    "uvicorn.access": {"level": "ERROR"}
+    "OpenSearchOutput": {"level": "DEBUG"}
+    "KafkaOutput": {"level": "ERROR"}
+    "Input": {"level": "ERROR"}
+metrics:
+  enabled: true
+  port: 8001
+pipeline:
+  - labelername:
+      type: ng_labeler
+      schema: examples/exampledata/rules/labeler/schema.json
+      include_parent_labels: true
+      rules:
+        - examples/exampledata/rules/labeler/rules
+  - dissector:
+      type: ng_dissector
+      rules:
+        - examples/exampledata/rules/dissector/rules
+  - dropper:
+      type: ng_dropper
+      rules:
+        - examples/exampledata/rules/dropper/rules
+        - filter: "test_dropper"
+          dropper:
+            drop:
+              - drop_me
+          description: "..."
+  - pre_detector:
+      type: ng_pre_detector
+      rules:
+        - examples/exampledata/rules/pre_detector/rules
+      outputs:
+        - opensearch: sre
+      tree_config: examples/exampledata/rules/pre_detector/tree_config.json
+      alert_ip_list_path: examples/exampledata/rules/pre_detector/alert_ips.yml
+  - amides:
+      type: ng_amides
+      rules:
+        - examples/exampledata/rules/amides/rules
+      models_path: examples/exampledata/models/model.zip
+      num_rule_attributions: 10
+      max_cache_entries: 1000000
+      decision_threshold: 0.32
+  - pseudonymizer:
+      type: ng_pseudonymizer
+      pubkey_analyst: examples/exampledata/rules/pseudonymizer/example_analyst_pub.pem
+      pubkey_depseudo: examples/exampledata/rules/pseudonymizer/example_depseudo_pub.pem
+      regex_mapping: examples/exampledata/rules/pseudonymizer/regex_mapping.yml
+      hash_salt: a_secret_tasty_ingredient
+      outputs:
+        - opensearch: pseudonyms
+      rules:
+        - examples/exampledata/rules/pseudonymizer/rules/
+      max_cached_pseudonyms: 1000000
+  - calculator:
+      type: ng_calculator
+      rules:
+        - filter: "test_label: execute"
+          calculator:
+            target_field: "calculation"
+            calc: "1 + 1"
+input:
+  kafka:
+    type: ng_confluentkafka_input
+    topic: consumer
+    kafka_config:
+      bootstrap.servers: 127.0.0.1:9092
+      group.id: cgroup3
+      enable.auto.commit: "true"
+      auto.commit.interval.ms: "10000"
+      enable.auto.offset.store: "false"
+      queued.min.messages: "100000"
+      queued.max.messages.kbytes: "65536"
+      statistics.interval.ms: "60000"
+    preprocessing:
+      version_info_target_field: Logprep_version_info
+      log_arrival_time_target_field: event.ingested
+      hmac:
+        target: <RAW_MSG>
+        key: "thisisasecureandrandomkey"
+        output_field: Full_event
+output:
+  opensearch:
+    type: ng_opensearch_output
+    hosts:
+      - 127.0.0.1:9200
+    default_index: processed
+    default_op_type: create
+    message_backlog_size: 2500
+    timeout: 10000
+    flush_timeout: 60
+    user: admin
+    secret: admin
+    desired_cluster_status: ["green", "yellow"]
+    chunk_size: 25
+error_output:
+  opensearch:
+    type: ng_opensearch_output
+    hosts:
+      - 127.0.0.1:9200
+    default_index: errors
+    default_op_type: create
+    message_backlog_size: 2500
+    timeout: 10000
+    flush_timeout: 60
+    user: admin
+    secret: admin
+    desired_cluster_status: ["green", "yellow"]
+    chunk_size: 25

examples/exampledata/config/_benchmark_non_ng_pipeline.yml

@@ -0,0 +1,125 @@
+version: 2
+process_count: 1
+timeout: 5.0
+restart_count: 2
+config_refresh_interval: 5
+error_backlog_size: 15000
+logger:
+  level: DEBUG
+  format: "%(asctime)-15s %(hostname)-5s %(name)-10s %(levelname)-8s: %(message)s"
+  datefmt: "%Y-%m-%d %H:%M:%S"
+  loggers:
+    "py.warnings": {"level": "ERROR"}
+    "Runner": {"level": "INFO"}
+    "Processor": {"level": "ERROR"}
+    "Exporter": {"level": "ERROR"}
+    "uvicorn": {"level": "ERROR"}
+    "uvicorn.access": {"level": "ERROR"}
+    "OpenSearchOutput": {"level": "DEBUG"}
+    "KafkaOutput": {"level": "ERROR"}
+    "Input": {"level": "ERROR"}
+metrics:
+  enabled: true
+  port: 8001
+pipeline:
+  - labelername:
+      type: labeler
+      schema: examples/exampledata/rules/labeler/schema.json
+      include_parent_labels: true
+      rules:
+        - examples/exampledata/rules/labeler/rules
+  - dissector:
+      type: dissector
+      rules:
+        - examples/exampledata/rules/dissector/rules
+  - dropper:
+      type: dropper
+      rules:
+        - examples/exampledata/rules/dropper/rules
+        - filter: "test_dropper"
+          dropper:
+            drop:
+              - drop_me
+          description: "..."
+  - pre_detector:
+      type: pre_detector
+      rules:
+        - examples/exampledata/rules/pre_detector/rules
+      outputs:
+        - opensearch: sre
+      tree_config: examples/exampledata/rules/pre_detector/tree_config.json
+      alert_ip_list_path: examples/exampledata/rules/pre_detector/alert_ips.yml
+  - amides:
+      type: amides
+      rules:
+        - examples/exampledata/rules/amides/rules
+      models_path: examples/exampledata/models/model.zip
+      num_rule_attributions: 10
+      max_cache_entries: 1000000
+      decision_threshold: 0.32
+  - pseudonymizer:
+      type: pseudonymizer
+      pubkey_analyst: examples/exampledata/rules/pseudonymizer/example_analyst_pub.pem
+      pubkey_depseudo: examples/exampledata/rules/pseudonymizer/example_depseudo_pub.pem
+      regex_mapping: examples/exampledata/rules/pseudonymizer/regex_mapping.yml
+      hash_salt: a_secret_tasty_ingredient
+      outputs:
+        - opensearch: pseudonyms
+      rules:
+        - examples/exampledata/rules/pseudonymizer/rules/
+      max_cached_pseudonyms: 1000000
+  - calculator:
+      type: calculator
+      rules:
+        - filter: "test_label: execute"
+          calculator:
+            target_field: "calculation"
+            calc: "1 + 1"
+input:
+  kafka:
+    type: confluentkafka_input
+    topic: consumer
+    kafka_config:
+      bootstrap.servers: 127.0.0.1:9092
+      group.id: cgroup3
+      enable.auto.commit: "true"
+      auto.commit.interval.ms: "10000"
+      enable.auto.offset.store: "false"
+      queued.min.messages: "100000"
+      queued.max.messages.kbytes: "65536"
+      statistics.interval.ms: "60000"
+    preprocessing:
+      version_info_target_field: Logprep_version_info
+      log_arrival_time_target_field: event.ingested
+      hmac:
+        target: <RAW_MSG>
+        key: "thisisasecureandrandomkey"
+        output_field: Full_event
+output:
+  opensearch:
+    type: opensearch_output
+    hosts:
+      - 127.0.0.1:9200
+    default_index: processed
+    default_op_type: create
+    message_backlog_size: 2500
+    timeout: 10000
+    flush_timeout: 60
+    user: admin
+    secret: admin
+    desired_cluster_status: ["green", "yellow"]
+    chunk_size: 25
+error_output:
+  opensearch:
+    type: opensearch_output
+    hosts:
+      - 127.0.0.1:9200
+    default_index: errors
+    default_op_type: create
+    message_backlog_size: 2500
+    timeout: 10000
+    flush_timeout: 60
+    user: admin
+    secret: admin
+    desired_cluster_status: ["green", "yellow"]
+    chunk_size: 25

examples/exampledata/config/ng_pipeline.yml

@@ -17,6 +17,7 @@ logger:
     "uvicorn.access": {"level": "ERROR"}
     "OpenSearchOutput": {"level": "DEBUG"}
     "KafkaOutput": {"level": "ERROR"}
+    "Input": {"level": "ERROR"}
 metrics:
   enabled: true
   port: 8001

logprep/abc/component.py

@@ -77,7 +77,7 @@ def metric_labels(self) -> dict:
         """Labels for the metrics"""
         return {"component": self._config.type, "name": self.name, "description": "", "type": ""}
 
-    def __init__(self, name: str, configuration: "Config", pipeline_index: int | None = None):
+    def __init__(self, name: str, configuration: Config, pipeline_index: int | None = None):
         self._config = configuration
         self.name = name
         self.pipeline_index = pipeline_index

logprep/abc/processor.py

@@ -151,7 +151,7 @@ def result(self, value: ProcessorResult):
         self._result = value
 
     @property
-    def rules(self) -> list["Rule"]:
+    def rules(self) -> Sequence["Rule"]:
         """Returns all rules
 
         Returns

logprep/connector/http/input.py

@@ -266,7 +266,7 @@ class HttpEndpoint(ABC):
     # pylint: disable=too-many-arguments
     def __init__(
         self,
-        messages: mp.Queue,
+        messages: mp.Queue[dict],
         original_event_field: dict[str, str] | None,
         collect_meta: bool,
         metafield_name: str,
@@ -336,7 +336,7 @@ def put_message(self, event: dict, metadata: dict):
 class JSONHttpEndpoint(HttpEndpoint):
     """:code:`json` endpoint to get json from request"""
 
-    _decoder = msgspec.json.Decoder()
+    _decoder: msgspec.json.Decoder[dict] = msgspec.json.Decoder()
 
     @raise_request_exceptions
     @basic_auth
@@ -360,7 +360,7 @@ async def __call__(self, req, resp, **kwargs):  # pylint: disable=arguments-diff
 class JSONLHttpEndpoint(HttpEndpoint):
     """:code:`jsonl` endpoint to get jsonl from request"""
 
-    _decoder = msgspec.json.Decoder()
+    _decoder: msgspec.json.Decoder[dict] = msgspec.json.Decoder()
 
     @raise_request_exceptions
     @basic_auth
@@ -555,7 +555,7 @@ def __attrs_post_init__(self):
 
     __slots__: list[str] = ["target", "app", "http_server"]
 
-    messages: typing.Optional[Queue] = None
+    messages: Queue[dict] | None = None
 
     _endpoint_registry: Mapping[str, type[HttpEndpoint]] = {
         "json": JSONHttpEndpoint,

logprep/factory.py

@@ -12,7 +12,7 @@ class Factory:
     """Create components for logprep."""
 
     @classmethod
-    def create(cls, configuration: dict) -> Component | None:
+    def create(cls, configuration: dict) -> Component:
         """Create component."""
         if configuration == {} or configuration is None:
             raise InvalidConfigurationError("The component definition is empty.")
@@ -23,16 +23,14 @@ def create(cls, configuration: dict) -> Component | None:
                 f"Found multiple component definitions ({', '.join(configuration.keys())}),"
                 + " but there must be exactly one."
             )
-        for component_name, component_configuration_dict in configuration.items():
-            if configuration == {} or component_configuration_dict is None:
-                raise InvalidConfigurationError(
-                    f'The definition of component "{component_name}" is empty.'
-                )
-            if not isinstance(component_configuration_dict, dict):
-                raise InvalidConfigSpecificationError(component_name)
-            component = Configuration.get_class(component_name, component_configuration_dict)
-            component_configuration = Configuration.create(
-                component_name, component_configuration_dict
+        # we know configuration has exactly one entry
+        [(component_name, component_configuration_dict)] = configuration.items()
+        if component_configuration_dict is None:
+            raise InvalidConfigurationError(
+                f'The definition of component "{component_name}" is empty.'
             )
-            return component(component_name, component_configuration)
-        return None
+        if not isinstance(component_configuration_dict, dict):
+            raise InvalidConfigSpecificationError(component_name)
+        component = Configuration.get_class(component_name, component_configuration_dict)
+        component_configuration = Configuration.create(component_name, component_configuration_dict)
+        return component(component_name, component_configuration)