Add Python logging with same format as Java (#360)

Author: cbornet

langstream-runtime/langstream-runtime-impl/src/main/python/langstream_runtime/__main__.py

@@ -15,13 +15,21 @@
 # limitations under the License.
 #
 
+import logging
 import sys
 
 import yaml
 
 from . import runtime
 
 if __name__ == "__main__":
+    logging.addLevelName(logging.WARNING, "WARN")
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s.%(msecs)03d [%(threadName)s] %(levelname)-5s %(name).36s -- %(message)s",  # noqa: E501
+        datefmt="%H:%M:%S",
+    )
+
     print(sys.argv)
     if len(sys.argv) != 2:
         print("Missing pod configuration file argument")

langstream-runtime/langstream-runtime-impl/src/main/python/langstream_runtime/kafka_connection.py

@@ -36,6 +36,7 @@
 )
 from .topic_connector import TopicConsumer, TopicProducer
 
+LOG = logging.getLogger(__name__)
 STRING_DESERIALIZER = StringDeserializer()
 
 SERIALIZERS = {
@@ -102,7 +103,7 @@ def create_dlq_producer(agent_id, streaming_cluster, configuration):
     dlq_conf = configuration.get("deadLetterTopicProducer")
     if not dlq_conf:
         return None
-    logging.info(
+    LOG.info(
         f"Creating dead-letter topic producer for agent {agent_id} using configuration "
         f"{configuration}"
     )
@@ -155,16 +156,16 @@ def __init__(self, configs):
         self.commit_ever_called = False
 
     def start(self):
-        self.consumer = Consumer(self.configs)
-        logging.info(f"Subscribing consumer to {self.topic}")
+        self.consumer = Consumer(self.configs, logger=LOG)
+        LOG.info(f"Subscribing consumer to {self.topic}")
         self.consumer.subscribe(
             [self.topic], on_assign=self.on_assign, on_revoke=self.on_revoke
         )
 
     def close(self):
         with self.lock:
             if self.consumer:
-                logging.info(
+                LOG.info(
                     f"Closing consumer to {self.topic} with {self.pending_commits} "
                     f"pending commits and {len(self.uncommitted)} uncommitted "
                     f"offsets: {self.uncommitted} "
@@ -188,9 +189,9 @@ def read(self) -> List[KafkaRecord]:
         if message is None:
             return []
         if message.error():
-            logging.error(f"Consumer error: {message.error()}")
+            LOG.error(f"Consumer error: {message.error()}")
             return []
-        logging.debug(
+        LOG.debug(
             f"Received message from Kafka topics {self.consumer.assignment()}:"
             f" {message}"
         )
@@ -223,7 +224,7 @@ def commit(self, records: List[KafkaRecord]):
                     if committed_tp_offset.error:
                         raise KafkaException(committed_tp_offset.error)
                     current_offset = committed_tp_offset.offset
-                    logging.info(
+                    LOG.info(
                         f"Current position on partition {topic_partition} is "
                         f"{current_offset}"
                     )
@@ -269,34 +270,34 @@ def on_commit(self, error, partitions):
         with self.lock:
             self.pending_commits -= 1
         if error:
-            logging.error(f"Error committing offsets on topic {self.topic}: {error}")
+            LOG.error(f"Error committing offsets on topic {self.topic}: {error}")
             if not self.commit_failure:
                 self.commit_failure = KafkaException(error)
         else:
-            logging.debug(f"Offsets committed: {partitions}")
+            LOG.debug(f"Offsets committed: {partitions}")
 
     def on_assign(self, consumer: Consumer, partitions: List[TopicPartition]):
         with self.lock:
-            logging.info(f"Partitions assigned: {partitions}")
+            LOG.info(f"Partitions assigned: {partitions}")
             for partition in partitions:
                 offset = consumer.committed([partition])[0].offset
-                logging.info(f"Last committed offset for {partition} is {offset}")
+                LOG.info(f"Last committed offset for {partition} is {offset}")
                 if offset >= 0:
                     self.committed[partition] = offset
 
     def on_revoke(self, _, partitions: List[TopicPartition]):
         with self.lock:
-            logging.info(f"Partitions revoked: {partitions}")
+            LOG.info(f"Partitions revoked: {partitions}")
             for partition in partitions:
                 if partition in self.committed:
                     offset = self.committed.pop(partition)
-                    logging.info(
+                    LOG.info(
                         f"Current offset {offset} on partition {partition} (revoked)"
                     )
                 if partition in self.uncommitted:
                     offsets = self.uncommitted.pop(partition)
                     if len(offsets) > 0:
-                        logging.warning(
+                        LOG.warning(
                             f"There are uncommitted offsets {offsets} on partition "
                             f"{partition} (revoked), these messages will be "
                             f"re-delivered"
@@ -330,13 +331,13 @@ def __init__(self, configs):
         self.delivery_failure: Optional[Exception] = None
 
     def start(self):
-        self.producer = Producer(self.configs)
+        self.producer = Producer(self.configs, logger=LOG)
 
     def write(self, records: List[Record]):
         for record in records:
             if self.delivery_failure:
                 raise self.delivery_failure
-            logging.info(f"Sending record {record}")
+            LOG.info(f"Sending record {record}")
             headers = []
             if record.headers():
                 for key, value in record.headers():

langstream-runtime/langstream-runtime-impl/src/main/python/langstream_runtime/runtime.py

@@ -44,6 +44,8 @@
     TopicConsumerWithDLQSource,
 )
 
+LOG = logging.getLogger(__name__)
+
 
 def current_time_millis():
     return int(time.time_ns() / 1000_000)
@@ -64,7 +66,7 @@ def __init__(self, configuration):
 
     def handle_errors(self, source_record: Record, error) -> ErrorsProcessingOutcome:
         self.failures += 1
-        logging.info(
+        LOG.info(
             f"Handling error {error} for source record {source_record}, "
             f"errors count {self.failures} (max retries {self.retries})"
         )
@@ -264,7 +266,7 @@ def run_with_server(
 
 
 def run(configuration, agent=None, agent_info: AgentInfo = AgentInfo(), max_loops=-1):
-    logging.info(f"Pod Configuration {configuration}")
+    LOG.info(f"Pod Configuration {configuration}")
 
     if "streamingCluster" not in configuration:
         raise ValueError("streamingCluster cannot be null")
@@ -405,7 +407,7 @@ def run_main_loop(
                                         source,
                                     )
                     except Exception as e:
-                        logging.exception("Error while processing records")
+                        LOG.exception("Error while processing records")
                         # raise the error
                         # this way the consumer will not commit the records
                         raise e
@@ -426,7 +428,7 @@ def run_processor_agent(
     trial_number = 0
     while len(records_to_process) > 0:
         trial_number += 1
-        logging.info(
+        LOG.info(
             f"run processor on {len(records_to_process)} records "
             f"(trial #{trial_number})"
         )
@@ -439,26 +441,26 @@ def run_processor_agent(
             if isinstance(processor_result, Exception):
                 action = errors_handler.handle_errors(source_record, processor_result)
                 if action == ErrorsProcessingOutcome.SKIP:
-                    logging.error(
+                    LOG.error(
                         f"Unrecoverable error {processor_result} while processing the "
                         f"records, skipping"
                     )
                     results_by_record[source_record] = (source_record, processor_result)
                 elif action == ErrorsProcessingOutcome.RETRY:
-                    logging.error(
+                    LOG.error(
                         f"Retryable error {processor_result} while processing the "
                         f"records, retrying"
                     )
                     records_to_process.append(source_record)
                 elif action == ErrorsProcessingOutcome.FAIL:
-                    logging.error(
+                    LOG.error(
                         f"Unrecoverable error {processor_result} while processing some "
                         f"records, failing"
                     )
                     # TODO: replace with custom exception ?
                     source.permanent_failure(source_record, processor_result)
                     if errors_handler.fail_processing_on_permanent_errors():
-                        logging.error("Failing processing on permanent error")
+                        LOG.error("Failing processing on permanent error")
                         raise processor_result
                     # in case the source does not throw an exception we mark the record
                     # as "skipped"
@@ -496,26 +498,26 @@ def write_records_to_the_sink(
             action = errors_handler.handle_errors(source_record, error)
             if action == ErrorsProcessingOutcome.SKIP:
                 # skip (the whole batch)
-                logging.error(
+                LOG.error(
                     f"Unrecoverable error {error} while processing the records, "
                     f"skipping"
                 )
                 source_record_tracker.commit(for_the_sink)
                 return
             elif action == ErrorsProcessingOutcome.RETRY:
                 # retry (the whole batch)
-                logging.error(
+                LOG.error(
                     f"Retryable error {error} while processing the records, retrying"
                 )
             elif action == ErrorsProcessingOutcome.FAIL:
-                logging.error(
+                LOG.error(
                     f"Unrecoverable error {error} while processing some records, "
                     f"failing"
                 )
                 # TODO: replace with custom exception ?
                 source.permanent_failure(source_record, error)
                 if errors_handler.fail_processing_on_permanent_errors():
-                    logging.error("Failing processing on permanent error")
+                    LOG.error("Failing processing on permanent error")
                     raise error
                 # in case the source does not throw an exception we mark the record as
                 # "skipped"
@@ -526,7 +528,7 @@ def write_records_to_the_sink(
 
 class NoopTopicConsumer(TopicConsumer):
     def read(self):
-        logging.info("Sleeping for 1 second, no records...")
+        LOG.info("Sleeping for 1 second, no records...")
         time.sleep(1)
         return []
 

langstream-runtime/langstream-runtime-impl/src/main/python/langstream_runtime/tests/test_runtime.py

@@ -50,7 +50,6 @@ def test_simple_agent(_):
     agent = runtime.init_agent(config)
     agent_info = runtime.AgentInfo()
     runtime.run(config, agent=agent, agent_info=agent_info, max_loops=2)
-    print(json.dumps(agent_info.worker_status(), indent=2))
     assert (
         json.dumps(agent_info.worker_status(), indent=2)
         == """[

langstream-runtime/langstream-runtime-impl/src/main/python/langstream_runtime/topic_connector.py

@@ -26,6 +26,8 @@
     TopicConsumer,
 )
 
+LOG = logging.getLogger(__name__)
+
 
 class TopicConsumerSource(Source):
     def __init__(self, consumer: TopicConsumer):
@@ -38,11 +40,11 @@ def commit(self, records: List[Record]):
         self.consumer.commit(records)
 
     def start(self):
-        logging.info(f"Starting consumer {self.consumer}")
+        LOG.info(f"Starting consumer {self.consumer}")
         self.consumer.start()
 
     def close(self):
-        logging.info(f"Closing consumer {self.consumer}")
+        LOG.info(f"Closing consumer {self.consumer}")
         self.consumer.close()
 
     def agent_info(self) -> Dict[str, Any]:
@@ -66,7 +68,7 @@ def close(self):
         self.dlq_producer.close()
 
     def permanent_failure(self, record: Record, error: Exception):
-        logging.error(f"Sending record to DLQ: {record}")
+        LOG.error(f"Sending record to DLQ: {record}")
         self.dlq_producer.write([record])
 
 
@@ -76,11 +78,11 @@ def __init__(self, producer: TopicProducer):
         self.commit_callback = None
 
     def start(self):
-        logging.info(f"Starting producer {self.producer}")
+        LOG.info(f"Starting producer {self.producer}")
         self.producer.start()
 
     def close(self):
-        logging.info(f"Closing producer {self.producer}")
+        LOG.info(f"Closing producer {self.producer}")
         self.producer.close()
 
     def write(self, records: List[Record]):