Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def setup_logging(loggers_and_levels, logentries_id=None):
log = logging.getLogger('logentries')
log.setLevel(logging.INFO)
logentries_handler = LogentriesHandler(logentries_id)
handler = logging.StreamHandler()
FORMAT = "%(asctime)s:%(levelname)s:%(name)s:%(message)s"
formatter = logging.Formatter(fmt=FORMAT)
handler.setFormatter(formatter)
logentries_handler.setFormatter(formatter)
log.addHandler(handler)
log.addHandler(logentries_handler)
for logger, level in loggers_and_levels:
logger.setLevel(level)
logger.addHandler(handler)
logger.addHandler(logentries_handler)
def __init__(self, token):
self._count = 0.0
self._sum = 0.0
self._lock = Lock()
self.token = token
handler = LogentriesHandler(token)
log.addHandler(handler)
def setup_logging(loggers_and_levels, logentries_id=None):
log = logging.getLogger('logentries')
log.setLevel(logging.INFO)
if logentries_id:
logentries_handler = LogentriesHandler(logentries_id)
handler = logentries_handler
else:
handler = logging.StreamHandler()
FORMAT = "%(asctime)s:%(levelname)s:%(name)s:%(message)s"
formatter = logging.Formatter(fmt=FORMAT)
handler.setFormatter(formatter)
log.addHandler(handler)
for logger, level in loggers_and_levels:
logger.setLevel(level)
logger.addHandler(handler)
def main(gctx=None):
# setup logentries. we forward log messages to it
le_token = "e8549616-0798-4d7e-a2ca-2513ae81fa17"
le_handler = LogentriesHandler(le_token, use_tls=False, verbose=False)
le_level = 20 #logging.INFO
ctx = zmq.Context().instance()
sock = ctx.socket(zmq.PULL)
sock.bind("ipc:///tmp/logmessage")
# and we publish them
pub_sock = messaging.pub_sock('logMessage')
while True:
dat = b''.join(sock.recv_multipart())
dat = dat.decode('utf8')
# print "RECV", repr(dat)
def configure(self):
logger = logging.getLogger('logentries')
handler = LogentriesHandler(self.token)
logger.addHandler(handler)
logger.setLevel(logging.DEBUG)
return logger