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

    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))