update grafana dashboard and change kafka output counts

logprep/connector/confluent_kafka/output.py

@@ -28,15 +28,14 @@
 import json
 from datetime import datetime
 from functools import cached_property, partial
-from logging import Logger
 from socket import getfqdn
 from typing import Optional
 
 from attrs import define, field, validators
 from confluent_kafka import KafkaException, Producer
 
 from logprep.abc.output import CriticalOutputError, FatalOutputError, Output
-from logprep.metrics.metrics import GaugeMetric
+from logprep.metrics.metrics import GaugeMetric, Metric
 from logprep.util.validators import keys_in_validator
 
 DEFAULTS = {
@@ -241,10 +240,10 @@ def store(self, document: dict) -> Optional[bool]:
         configured input
         """
         self.store_custom(document, self._config.topic)
-        self.metrics.number_of_processed_events += 1
         if self.input_connector:
             self.input_connector.batch_finished_callback()
 
+    @Metric.measure_time()
     def store_custom(self, document: dict, target: str) -> None:
         """Write document to Kafka into target topic.
 
@@ -263,6 +262,7 @@ def store_custom(self, document: dict, target: str) -> None:
         try:
             self._producer.produce(target, value=self._encoder.encode(document))
             self._producer.poll(self._config.send_timeout)
+            self.metrics.number_of_processed_events += 1
         except BufferError:
             # block program until buffer is empty
             self._producer.flush(timeout=self._config.flush_timeout)
@@ -271,6 +271,7 @@ def store_custom(self, document: dict, target: str) -> None:
                 self, f"Error storing output document -> {error}", document
             ) from error
 
+    @Metric.measure_time()
     def store_failed(
         self, error_message: str, document_received: dict, document_processed: dict
     ) -> None: