(...)
I'm not an email infrastructure specialist, so some tips welcome if there is a way to get better throughput...
***
Did not managed to solve the issue by increasing mail throughput, so if someone have the same kind of issue, only way to avoid pacing down the whole infinite captures management job loop is to "thread" the mail send function so they are sent asynchronously. A single thread managed by a send list (thread safe in Python) is IMO enough, as Python is quite limited by design for thread handling.
In fact, there are 3 lists to handle not only the send requests jobs (not handled directly, done by "send" function that manages the threaded sending), but also sent and fail cases to handle the cleanup asynchronously from the main thread loop.
This also allows to keep the send thread active for a few 10's of seconds after last mail is sent (more would likely lead to a timeout/disconnect from server side, gmail in my case), so the re-connect time (even if this was not a real issue in my measurements) is saved when handling close-in-time bursts of captures.
That's now my python module file for Domoticz (dmtThrMail.py). This may also work for other providers changing smtp server (hardcoded, hereunder):
Code: Select all
# Domoticz python scripts threaded mail module
import sys
import logging
import os
import smtplib
import time
import threading
from email import encoders
from email.mime.application import MIMEApplication
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
# Only gmail supported for now...
# Globals used for thread communication:
global mailSendList
mailSendList = []
global mailSentList
mailSentList = []
global mailFailList
mailFailList = []
# Do not use directly, will be started as a thread by sendThrMail() to
# avoid caller blocking time that can be several tens of seconds with big
# attachments and remain connected for stayConnectedTime seconds after last
# mail was sent, to save reconnect time for close emailing requests.
def send_mail_threaded(sender, passwd, stayConnectedTime, logger):
global mailSendList
global mailSentList
global mailFailList
logger.log(logging.DEBUG, "Connect, stayConnectedTime = %d", stayConnectedTime)
lock = threading.Lock()
# Connect (blocking thread scheduling to prevent timeout/ssl setup issues ?)
#lock.acquire()
try:
logger.log(logging.DEBUG, "(Re-)Connect...")
smtp = smtplib.SMTP_SSL() # timeout=3 ???
#smtp.set_debuglevel(1)
smtp.connect('smtp.gmail.com', 465)
smtp.ehlo()
logger.log(logging.DEBUG, "SMTP init done ! Will login...")
smtp.login(sender, passwd)
logger.log(logging.DEBUG, "Done !")
except: # pylint: disable=bare-except
# On connect error, exit thread immediately
logger.log(logging.INFO, "Connect error : %s.", sys.exc_info()[0])
# Avoid list grow in case of multiple connect issues
if len(mailSendList) > 2:
mailFailList.append(mailSendList.pop(0))
return
#finally:
# lock.release()
# Send mail loop, active until connection release...
tLast = tNow = 0
while tNow-tLast <= stayConnectedTime:
if mailSendList != []:
# Get 1 item from send list
current_mail = mailSendList.pop(0)
logger.log(logging.DEBUG, "Sending {}".format(current_mail))
# Build message
msg = MIMEMultipart()
msg['Subject'] = current_mail['SUBJECT']
msg['To'] = ', '.join(current_mail['RECIPIENTS'])
msg['From'] = sender
if current_mail['BODY']:
msg.attach(MIMEText(current_mail['BODY'], 'plain'))
# Open file attachment, if any
if current_mail['FILENAME']:
try:
# Extract base filename to populate mail header
_path, fileName = os.path.split(current_mail['FILENAME'])
attachment = open(current_mail['FILENAME'], "rb")
payload = MIMEApplication(attachment.read(), _encoder=encoders.encode_base64)
attachment.close()
payload.add_header('Content-Disposition', 'attachment',
filename=fileName)
msg.attach(payload)
except: # pylint: disable=bare-except
logger.log(logging.INFO, "Unable to attach : %s.", current_mail['FILENAME'])
# Send mail, updating sent or fail lists for main thread
composed_msg = msg.as_string()
try:
#start_time = datetime.now()
smtp.sendmail(sender, current_mail['RECIPIENTS'], composed_msg)
#logger.log(logging.DEBUG, "Send time : %ds", (datetime.now()-start_time).seconds)
logger.log(logging.DEBUG, "DONE!!!")
mailSentList.append(current_mail)
except: # pylint: disable=bare-except
logger.log(logging.INFO, "Unable to send mail : %s.", sys.exc_info()[0])
mailFailList.append(current_mail)
tLast = time.time()
else:
time.sleep(0.5)
tNow = time.time()
# Disconnect, blocking thread scheduling till thread end (will auto-release)
lock.acquire()
try:
smtp.quit()
logger.log(logging.DEBUG, "Disconnected !")
except: # pylint: disable=bare-except
logger.log(logging.INFO, "Disconnect error : %s", sys.exc_info()[0])
# Send non blocking emails using a (single) send thread, connection will
# remain for stayConnectedTime seconds after last mail in shared
# send list is over. Use 0 for single notifications, a few 10's of
# seconds for mail loops.
# 3 lists (thread-safe in Python) are used for mail<->base threads communication:
# - Send list: Appended by hereunder function from base thread to provide send jobs to the mail thread
# and poped when send is done (successful or not, see sent/fail lists).
# - Sent list: Appended by mail thread to inform base thread of successful sends, then poped
# by base thread to clear mails/attachments (using clrMails(getSentMailList(), True, logger)).
# - Fail list: Appended by mail thread to inform base thread of failed sends, so attachments may
# be saved for instance (file list returned by clrMails(getFailMailList(), False, logger)).
def sendThrMail(fileAttach, subject, body, recipients, sender, passwd, stayConnectedTime, logger):
global mailSendList
logger.log(logging.INFO, "Send mail (%s attached) to %s..." % (fileAttach, recipients))
# 1st call 'Static' thread object init or when restart is later needed
if (not hasattr(sendThrMail, 'threadMail')) or (not sendThrMail.threadMail.is_alive()):
logger.log(logging.DEBUG, "Mail thread (re-)init")
sendThrMail.threadMail = threading.Thread(name='send_mail', target=send_mail_threaded, args=(sender, passwd, stayConnectedTime, logger,))
threadedMailData = {'RECIPIENTS':recipients, 'SUBJECT':subject, 'BODY':body, 'FILENAME':fileAttach}
mailSendList.append(threadedMailData)
logger.log(logging.DEBUG, "Added %s to thread send list...", threadedMailData)
# (Re-)Start thread if needed
if not sendThrMail.threadMail.is_alive():
logger.log(logging.DEBUG, "(Re-)Start thread...")
sendThrMail.threadMail.start()
# Return thread objet if needed for external sync/join...
return sendThrMail.threadMail
# Utilities for main thread caller
def getSendMailList():
return mailSendList
def getSentMailList():
return mailSentList
def getFailMailList():
return mailFailList
def clrMailList(mailList, rmAttch, logger):
retFiles = []
while (mailList) != []:
popMail = mailList.pop(0)
if popMail['FILENAME'] and os.path.exists(popMail['FILENAME']):
if rmAttch:
try:
logger.log(logging.DEBUG, "Delete %s", popMail['FILENAME'])
os.unlink(popMail['FILENAME'])
except: # pylint: disable=bare-except
logger.log(logging.INFO, "Unable to delete attachment(s) : %s", sys.exc_info()[0])
retFiles.append(popMail['FILENAME'])
else:
retFiles.append(popMail['FILENAME'])
# Return attachment files that were not erased (either because rmAttach was False or could not be done)
return retFiles
At test file for attachments used by the test code may be build using, for instance:
Code: Select all
dd if=/dev/urandom, of=test1M.bin bs=1M count=1Code: Select all
#! /usr/bin/python
# -*- coding: utf-8 -*-
import sys
import logging
import time
from datetime import datetime
from dmtThrMail import sendThrMail, getSendMailList, getSentMailList, getFailMailList, clrMailList
#notifyMail = '[email protected]'
storeMailAcnt = '[email protected]'
storeMailPass = 'DOMOTICZ_MAIL_PASSORWD'
#sendMailList = [storeMailAcnt] + [notifyMail]
sendMailList = [storeMailAcnt]
subjectMail = 'test'
logLevel = 'INFO'
#logLevel = 'DEBUG'
print "Start logger"
logger = logging.getLogger()
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
logger.setLevel(logLevel)
logger.addHandler(handler)
logger.log(logging.INFO, "Started logger at %s (LL=%s)", datetime.now(), logLevel)
logger.log(logging.INFO, "Mail 1")
bodyMail = 'Sent at ' + str(datetime.now())
sendThrMail(None, subjectMail, bodyMail, sendMailList, storeMailAcnt, storeMailPass, 1, logger)
logger.log(logging.INFO, "Mail 2")
bodyMail = 'Sent at ' + str(datetime.now())
sendThrMail(None, subjectMail, bodyMail, sendMailList, storeMailAcnt, storeMailPass, 30, logger)
time.sleep(10)
logger.log(logging.INFO, "Mail 3")
bodyMail = 'Sent at ' + str(datetime.now())
threadMail = sendThrMail('test1M.bin', subjectMail, bodyMail, sendMailList, storeMailAcnt, storeMailPass, 30, logger)
logger.log(logging.INFO, "END")
print getSendMailList()
print getSentMailList()
print getFailMailList()
if threadMail.is_alive():
threadMail.join()
logger.log(logging.INFO, "After JOIN => All send jobs should be over...")
print getSendMailList()
print getSentMailList()
print getFailMailList() # If some mail data still there, need to move attachments to another place? Erase?
logger.log(logging.INFO, "Cleanup to do")
filesNotErased = clrMailList(getSentMailList(), False, logger) # No attachment erase for print
#filesNotErased = clrMailList(getSentMailList(), True, logger) # Auto erase sent attachments version
print filesNotErased
print getSendMailList()