simon987/chan_feed
1
0
Fork 0
mirror of https://github.com/simon987/chan_feed.git synced 2025-04-20 02:26:44 +00:00
Code Issues Projects Releases Wiki Activity

buffer monitoring

Browse Source
This commit is contained in:
simon 2019-12-19 09:19:51 -05:00
parent a60b646e35
commit 7a360de2ab
2 changed files with 10 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
monitoring.py
View File

@ -1,26 +0,0 @@
import traceback
from influxdb import InfluxDBClient
from util import logger
client = InfluxDBClient("localhost", 8086, "root", "root", "chan_feed")
def init():
db_exists = False
for db in client.get_list_database():
if db["name"] == "chan_feed":
db_exists = True
break
if not db_exists:
client.create_database("chan_feed")
def log(event):
try:
client.write_points(event)
except Exception as e:
logger.debug(traceback.format_exc())
logger.error(str(e))

17
run.py
View File

@ -9,8 +9,8 @@ from queue import Queue
from threading import Thread from threading import Thread
import pika import pika
from hexlib.monitoring import Monitoring
import monitoring
from chan.chan import CHANS from chan.chan import CHANS
from post_process import post_process from post_process import post_process
from util import logger, Web from util import logger, Web
@ -19,9 +19,14 @@ MONITORING = True
BYPASS_RPS = False BYPASS_RPS = False
DBNAME = "chan_feed"
if MONITORING:
influxdb = Monitoring(DBNAME, logger=logger, batch_size=100, flush_on_exit=True)
class ChanScanner: class ChanScanner:
def __init__(self, helper, proxy): def __init__(self, helper, proxy):
self.web = Web(monitoring if MONITORING else None, rps=helper.rps, get_method=helper.get_method, proxy=proxy) self.web = Web(influxdb if MONITORING else None, rps=helper.rps, get_method=helper.get_method, proxy=proxy)
self.helper = helper self.helper = helper
self.state = ChanState() self.state = ChanState()
@ -142,7 +147,7 @@ class ChanState:
def publish_worker(queue: Queue, helper, p): def publish_worker(queue: Queue, helper, p):
channel = connect() channel = connect()
web = Web(monitoring if MONITORING else None, rps=helper.rps, get_method=helper.get_method, proxy=p) web = Web(influxdb if MONITORING else None, rps=helper.rps, get_method=helper.get_method, proxy=p)
while True: while True:
try: try:
@ -170,7 +175,7 @@ def publish(item, board, helper, channel, web):
if MONITORING: if MONITORING:
distance = datetime.utcnow() - datetime.utcfromtimestamp(helper.item_mtime(item)) distance = datetime.utcnow() - datetime.utcfromtimestamp(helper.item_mtime(item))
monitoring.log([{ influxdb.log([{
"measurement": chan, "measurement": chan,
"time": str(datetime.utcnow()), "time": str(datetime.utcnow()),
"tags": { "tags": {
@ -182,7 +187,7 @@ def publish(item, board, helper, channel, web):
}]) }])
break break
except Exception as e: except Exception as e:
logger.debug(traceback.format_exc()) # logger.debug(traceback.format_exc())
logger.error(str(e)) logger.error(str(e))
time.sleep(0.5) time.sleep(0.5)
channel = connect() channel = connect()
@ -213,8 +218,6 @@ if __name__ == "__main__":
if BYPASS_RPS: if BYPASS_RPS:
chan_helper.rps = 10 chan_helper.rps = 10
if MONITORING:
monitoring.init()
state = ChanState() state = ChanState()
publish_q = Queue() publish_q = Queue()