Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.util.custom_logging.fluent_log

"""
Provides a `TraceLogger` implementation that logs to a fluentd collector
"""

import json
import threading
import time

try:
    from fluent.sender import FluentSender
except ImportError:
    FluentSender = None


FLUENT_IMPORT_MESSAGE = "The Python fluent package is required to use this " "feature, please install it"


[docs]class FluentTraceLogger:
[docs] def __init__(self, name, host="localhost", port=24224): assert FluentSender is not None, FLUENT_IMPORT_MESSAGE self.lock = threading.Lock() self.thread_local = threading.local() self.name = name self.sender = FluentSender(self.name, host=host, port=port)
[docs] def context_set(self, key, value): self.lock.acquire() if not hasattr(self.thread_local, "context"): self.thread_local.context = {} self.thread_local.context[key] = value self.lock.release()
[docs] def context_remove(self, key): self.lock.acquire() del self.thread_local.context[key] self.lock.release()
[docs] def log(self, label, event_time=None, **kwargs): self.lock.acquire() if hasattr(self.thread_local, "context"): kwargs.update(self.thread_local.context) self.lock.release() event_time = event_time or time.time() self.sender.emit_with_time(label, int(event_time), json.dumps(kwargs, default=str))