Skip to content

Commit

Permalink
Merge pull request #24 from GhostofGoes/fix-groundtruth
Browse files Browse the repository at this point in the history
Handler fixes
  • Loading branch information
GhostofGoes authored Aug 14, 2024
2 parents 9b8cbca + b974454 commit 32cc1b8
Show file tree
Hide file tree
Showing 2 changed files with 10 additions and 10 deletions.
18 changes: 9 additions & 9 deletions src/pybennu/pybennu/executables/groundtruth_monitor.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
#!/usr/bin/env python3

""" A monitor that ingests messages published by a provider into elastic
This script will run on an ELK box as a daemon. It subscribes to a provider
Expand All @@ -18,16 +20,14 @@
"""

import os
import json
import argparse
import signal
import sys
import threading
from datetime import datetime
from configparser import ConfigParser, NoOptionError

from elasticsearch import Elasticsearch
from elasticsearch import helpers
from elasticsearch import Elasticsearch, helpers

from pybennu.distributed.subscriber import Subscriber
from pybennu.providers.utils.daemon import Daemon
Expand Down Expand Up @@ -71,7 +71,7 @@ def __init__(self, args):
E.Endpoint_str_set(self.publish_endpoint, 'udp://239.0.0.1:40000')
elastic_ip = '127.0.0.1'
self._filter = '/'

if 'power-groundtruth-monitor' in config.sections():
try:
E.Endpoint_str_set(self.publish_endpoint, config.get('power-groundtruth-monitor',
Expand All @@ -84,7 +84,7 @@ def __init__(self, args):
except NoOptionError:
print("WARNING: elasticsearch-ip not found in the configuration file."
" Using default: {}".format(elastic_ip))
try:
try:
self._filter = config.get('power-groundtruth-monitor', 'filter').strip()
except NoOptionError:
print("WARNING: filter not found in the configuration file."
Expand All @@ -102,7 +102,7 @@ def __init__(self, args):
except NoOptionError:
print("WARNING: elasticsearch-ip not found in the configuration file."
" Using default: {}".format(elastic_ip))
try:
try:
self._filter = config.get('groundtruth-monitor', 'filter').strip()
except NoOptionError:
print("WARNING: filter not found in the configuration file."
Expand Down Expand Up @@ -201,7 +201,7 @@ def __parse_filter(self):
self.fields.append('')
self.devices.append('')
self.providers.append('')

def __filter_message(self, message_dict):
in_filter = False
for i in range(len(self.providers)):
Expand All @@ -216,7 +216,7 @@ def push_updates_to_elastic(self):
messages = list(self.new_data)
self.new_data = []
index = 'bennu-' + datetime.now().strftime('%Y.%m.%d')
actions = [{'_index': index, 'type': 'doc', '_source': message} for message in messages]
actions = [{'_index': index, 'type': 'doc', '_source': message} for message in messages]
try:
res = helpers.bulk(self.__es, actions, request_timeout=30)
except:
Expand Down Expand Up @@ -256,7 +256,7 @@ def main():
args = parser.parse_args()

handle = GroundTruthMonitor(args)
handle.handler = handle._subscription_handler
handle.subscription_handler = handle._subscription_handler
if args.operation == 'start':
if args.daemonize:
handle.start()
Expand Down
2 changes: 1 addition & 1 deletion src/pybennu/pybennu/executables/pybennu_probe.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ def main():
endpoint = E.new_Endpoint()
E.Endpoint_str_set(endpoint, args.endpoint)
probe = client.Client(endpoint)
probe.handler = handler
probe.reply_handler = handler

msg = ""
if args.command == 'query':
Expand Down

0 comments on commit 32cc1b8

Please sign in to comment.