summaryrefslogtreecommitdiffstats
path: root/events
diff options
context:
space:
mode:
Diffstat (limited to 'events')
-rw-r--r--events/eventskeygen.py4
-rw-r--r--events/src/eventsapiconf.py.in8
-rw-r--r--events/src/eventsconfig.json3
-rw-r--r--events/src/gf_event.py20
-rw-r--r--events/src/glustereventsd.py95
-rw-r--r--events/src/peer_eventsapi.py4
-rw-r--r--events/src/utils.py28
7 files changed, 76 insertions, 86 deletions
diff --git a/events/eventskeygen.py b/events/eventskeygen.py
index 30b518dc18a..9d8a97c3030 100644
--- a/events/eventskeygen.py
+++ b/events/eventskeygen.py
@@ -102,7 +102,9 @@ ERRORS = (
"EVENT_ERROR_INVALID_INPUTS",
"EVENT_ERROR_SOCKET",
"EVENT_ERROR_CONNECT",
- "EVENT_ERROR_SEND"
+ "EVENT_ERROR_SEND",
+ "EVENT_ERROR_RESOLVE",
+ "EVENT_ERROR_MSG_FORMAT",
)
if gen_header_type == "C_HEADER":
diff --git a/events/src/eventsapiconf.py.in b/events/src/eventsapiconf.py.in
index 03dd0e8d5d7..fad96ca2cc7 100644
--- a/events/src/eventsapiconf.py.in
+++ b/events/src/eventsapiconf.py.in
@@ -9,7 +9,7 @@
# cases as published by the Free Software Foundation.
#
-SERVER_ADDRESS = "@localstatedir@/run/gluster/events.sock"
+SERVER_ADDRESS = "0.0.0.0"
DEFAULT_CONFIG_FILE = "@SYSCONF_DIR@/glusterfs/eventsconfig.json"
CUSTOM_CONFIG_FILE_TO_SYNC = "/events/config.json"
CUSTOM_CONFIG_FILE = "@GLUSTERD_WORKDIR@" + CUSTOM_CONFIG_FILE_TO_SYNC
@@ -17,7 +17,9 @@ WEBHOOKS_FILE_TO_SYNC = "/events/webhooks.json"
WEBHOOKS_FILE = "@GLUSTERD_WORKDIR@" + WEBHOOKS_FILE_TO_SYNC
LOG_FILE = "@localstatedir@/log/glusterfs/events.log"
EVENTSD = "glustereventsd"
-CONFIG_KEYS = ["log_level"]
+CONFIG_KEYS = ["log_level", "port"]
BOOL_CONFIGS = []
-RESTART_CONFIGS = []
+INT_CONFIGS = ["port"]
+RESTART_CONFIGS = ["port"]
EVENTS_ENABLED = @EVENTS_ENABLED@
+UUID_FILE = "@GLUSTERD_WORKDIR@/glusterd.info"
diff --git a/events/src/eventsconfig.json b/events/src/eventsconfig.json
index ce2c775f0bd..45730f9bb83 100644
--- a/events/src/eventsconfig.json
+++ b/events/src/eventsconfig.json
@@ -1,3 +1,4 @@
{
- "log_level": "INFO"
+ "log_level": "INFO",
+ "port": 24009
}
diff --git a/events/src/gf_event.py b/events/src/gf_event.py
index 20dfc8a4f17..f9ece6adc28 100644
--- a/events/src/gf_event.py
+++ b/events/src/gf_event.py
@@ -16,7 +16,7 @@ import time
from eventsapiconf import SERVER_ADDRESS, EVENTS_ENABLED
from eventtypes import all_events
-from utils import logger, setup_logger
+from utils import logger, setup_logger, get_config
# Run this when this lib loads
setup_logger()
@@ -31,10 +31,9 @@ def gf_event(event_type, **kwargs):
return
try:
- client = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
- client.connect(SERVER_ADDRESS)
+ client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
except socket.error as e:
- logger.error("Unable to connect to events.sock: {0}".format(e))
+ logger.error("Unable to connect to events Server: {0}".format(e))
return
# Convert key value args into KEY1=VALUE1;KEY2=VALUE2;..
@@ -45,7 +44,18 @@ def gf_event(event_type, **kwargs):
# <TIMESTAMP> <EVENT_TYPE> <MSG>
msg = "{0} {1} {2}".format(int(time.time()), event_type, msg.strip(";"))
+ port = get_config("port")
+ if port is None:
+ logger.error("Unable to get eventsd port details")
+ return
+
try:
- client.sendall(msg)
+ sent = client.sendto(msg, (SERVER_ADDRESS, port))
+ assert sent == len(msg)
except socket.error as e:
logger.error("Unable to Send message: {0}".format(e))
+ except AssertionError:
+ logger.error("Unable to send message. Sent: {0}, Actual: {1}".format(
+ sent, len(msg)))
+ finally:
+ client.close()
diff --git a/events/src/glustereventsd.py b/events/src/glustereventsd.py
index 3fa57686a8b..91a0743ff22 100644
--- a/events/src/glustereventsd.py
+++ b/events/src/glustereventsd.py
@@ -11,12 +11,10 @@
#
from __future__ import print_function
-import asyncore
-import socket
-import os
-from multiprocessing import Process, Queue
import sys
import signal
+import SocketServer
+import socket
from eventtypes import all_events
import handlers
@@ -24,26 +22,19 @@ import utils
from eventsapiconf import SERVER_ADDRESS
from utils import logger
-# Global Queue, EventsHandler will add items to the queue
-# and process_event will gets each item and handles it
-events_queue = Queue()
-events_server_pid = None
+class GlusterEventsRequestHandler(SocketServer.BaseRequestHandler):
-def process_event():
- """
- Seperate process which handles all the incoming events from Gluster
- processes.
- """
- while True:
- data = events_queue.get()
- logger.debug("EVENT: {0}".format(repr(data)))
+ def handle(self):
+ data = self.request[0].strip()
+ logger.debug("EVENT: {0} from {1}".format(repr(data),
+ self.client_address[0]))
try:
# Event Format <TIMESTAMP> <TYPE> <DETAIL>
ts, key, value = data.split(" ", 2)
except ValueError:
logger.warn("Invalid Event Format {0}".format(data))
- continue
+ return
data_dict = {}
try:
@@ -51,7 +42,7 @@ def process_event():
data_dict = dict(x.split('=') for x in value.split(';'))
except ValueError:
logger.warn("Unable to parse Event {0}".format(data))
- continue
+ return
try:
# Event Type to Function Map, Recieved event data will be in
@@ -75,68 +66,28 @@ def process_event():
handlers.generic_handler(ts, int(key), data_dict)
-def process_event_wrapper():
- try:
- process_event()
- except KeyboardInterrupt:
- return
-
-
-class GlusterEventsHandler(asyncore.dispatcher_with_send):
-
- def handle_read(self):
- data = self.recv(8192)
- if data:
- events_queue.put(data)
- self.send(data)
-
-
-class GlusterEventsServer(asyncore.dispatcher):
-
- def __init__(self):
- global events_server_pid
- asyncore.dispatcher.__init__(self)
- # Start the Events listener process which listens to
- # the global queue
- p = Process(target=process_event_wrapper)
- p.start()
- events_server_pid = p.pid
-
- # Create UNIX Domain Socket, bind to path
- self.create_socket(socket.AF_UNIX, socket.SOCK_STREAM)
- self.bind(SERVER_ADDRESS)
- self.listen(5)
-
- def handle_accept(self):
- pair = self.accept()
- if pair is not None:
- sock, addr = pair
- GlusterEventsHandler(sock)
-
-
def signal_handler_sigusr2(sig, frame):
- if events_server_pid is not None:
- os.kill(events_server_pid, signal.SIGUSR2)
utils.load_all()
def init_event_server():
utils.setup_logger()
-
- # Delete Socket file if Exists
- try:
- os.unlink(SERVER_ADDRESS)
- except OSError:
- if os.path.exists(SERVER_ADDRESS):
- print ("Failed to cleanup socket file {0}".format(SERVER_ADDRESS),
- file=sys.stderr)
- sys.exit(1)
-
utils.load_all()
- # Start the Eventing Server, UNIX DOMAIN SOCKET Server
- GlusterEventsServer()
- asyncore.loop()
+ port = utils.get_config("port")
+ if port is None:
+ sys.stderr.write("Unable to get Port details from Config\n")
+ sys.exit(1)
+
+ # Start the Eventing Server, UDP Server
+ try:
+ server = SocketServer.ThreadingUDPServer(
+ (SERVER_ADDRESS, port),
+ GlusterEventsRequestHandler)
+ except socket.error as e:
+ sys.stderr.write("Failed to start Eventsd: {0}\n".format(e))
+ sys.exit(1)
+ server.serve_forever()
def main():
diff --git a/events/src/peer_eventsapi.py b/events/src/peer_eventsapi.py
index 7887d77351c..f4447784f90 100644
--- a/events/src/peer_eventsapi.py
+++ b/events/src/peer_eventsapi.py
@@ -31,6 +31,7 @@ from events.eventsapiconf import (WEBHOOKS_FILE_TO_SYNC,
EVENTSD,
CONFIG_KEYS,
BOOL_CONFIGS,
+ INT_CONFIGS,
RESTART_CONFIGS)
@@ -462,6 +463,9 @@ class ConfigSetCmd(Cmd):
if args.name in BOOL_CONFIGS:
v = boolify(args.value)
+ if args.name in INT_CONFIGS:
+ v = int(args.value)
+
new_data[args.name] = v
file_content_overwrite(CUSTOM_CONFIG_FILE, new_data)
diff --git a/events/src/utils.py b/events/src/utils.py
index 772221a1e25..386e8f28449 100644
--- a/events/src/utils.py
+++ b/events/src/utils.py
@@ -17,11 +17,10 @@ import requests
from eventsapiconf import (LOG_FILE,
WEBHOOKS_FILE,
DEFAULT_CONFIG_FILE,
- CUSTOM_CONFIG_FILE)
+ CUSTOM_CONFIG_FILE,
+ UUID_FILE)
import eventtypes
-from gluster.cliutils import get_node_uuid
-
# Webhooks list
_webhooks = {}
@@ -32,6 +31,23 @@ _config = {}
# Init Logger instance
logger = logging.getLogger(__name__)
+NodeID = None
+
+
+def get_node_uuid():
+ val = None
+ with open(UUID_FILE) as f:
+ for line in f:
+ if line.startswith("UUID="):
+ val = line.strip().split("=")[-1]
+ break
+ return val
+
+
+def get_config(key):
+ if not _config:
+ load_config()
+ return _config.get(key, None)
def get_event_type_name(idx):
@@ -109,8 +125,12 @@ def load_all():
def publish(ts, event_key, data):
+ global NodeID
+ if NodeID is None:
+ NodeID = get_node_uuid()
+
message = {
- "nodeid": get_node_uuid(),
+ "nodeid": NodeID,
"ts": int(ts),
"event": get_event_type_name(event_key),
"message": data