From 59c4a3b7445a464f99dcd38995a4e687a80be632 Mon Sep 17 00:00:00 2001
From: Aravinda VK <avishwan@redhat.com>
Date: Wed, 12 Oct 2016 23:32:39 +0530
Subject: [PATCH 108/141] geo-rep/eventsapi: Additional Events
Added following events
EVENT_GEOREP_ACTIVE
{
"nodeid": NODEID,
"ts": TIMESTAMP,
"event": "GEOREP_ACTIVE",
"message": {
"master_volume": MASTER_VOLUME_NAME,
"slave_host": SLAVE_HOST,
"slave_volume": SLAVE_VOLUME,
"brick_path": BRICK_PATH
}
}
EVENT_GEOREP_PASSIVE
{
"nodeid": NODEID,
"ts": TIMESTAMP,
"event": "GEOREP_PASSIVE",
"message": {
"master_volume": MASTER_VOLUME_NAME,
"slave_host": SLAVE_HOST,
"slave_volume": SLAVE_VOLUME,
"brick_path": BRICK_PATH
}
}
EVENT_GEOREP_CHECKPOINT_COMPLETED
{
"nodeid": NODEID,
"ts": TIMESTAMP,
"event": "GEOREP_ACTIVE",
"message": {
"master_volume": MASTER_VOLUME_NAME,
"slave_host": SLAVE_HOST,
"slave_volume": SLAVE_VOLUME,
"brick_path": BRICK_PATH,
"checkpoint_time": CHECKPOINT_TIME,
"checkpoint_completion_time": CHECKPOINT_COMPLETION_TIME
}
}
> Reviewed-on: http://review.gluster.org/15630
> Smoke: Gluster Build System <jenkins@build.gluster.org>
> NetBSD-regression: NetBSD Build System <jenkins@build.gluster.org>
> CentOS-regression: Gluster Build System <jenkins@build.gluster.org>
> Reviewed-by: Kotresh HR <khiremat@redhat.com>
> Tested-by: Kotresh HR <khiremat@redhat.com>
BUG: 1380257
Change-Id: I90716175868c59dd65c8d202e73e0ede90347b6a
Signed-off-by: Aravinda VK <avishwan@redhat.com>
Reviewed-on: https://code.engineering.redhat.com/gerrit/87745
Reviewed-by: Atin Mukherjee <amukherj@redhat.com>
---
events/eventskeygen.py | 4 ++
geo-replication/syncdaemon/gsyncd.py | 4 ++
geo-replication/syncdaemon/gsyncdstatus.py | 50 +++++++++++++++++++++++++---
geo-replication/syncdaemon/monitor.py | 5 ++-
geo-replication/syncdaemon/resource.py | 7 +++-
geo-replication/syncdaemon/syncdutils.py | 19 ++++++++++
6 files changed, 81 insertions(+), 8 deletions(-)
diff --git a/events/eventskeygen.py b/events/eventskeygen.py
index 6165b24..1b1c691 100644
--- a/events/eventskeygen.py
+++ b/events/eventskeygen.py
@@ -128,6 +128,10 @@ keys = (
"EVENT_EC_MIN_BRICKS_UP",
#georep async events
"EVENT_GEOREP_FAULTY",
+ "EVENT_GEOREP_CHECKPOINT_COMPLETED",
+ "EVENT_GEOREP_ACTIVE",
+ "EVENT_GEOREP_PASSIVE",
+
#quota async events
"EVENT_QUOTA_CROSSED_SOFT_LIMIT",
diff --git a/geo-replication/syncdaemon/gsyncd.py b/geo-replication/syncdaemon/gsyncd.py
index b459abc..2c80319 100644
--- a/geo-replication/syncdaemon/gsyncd.py
+++ b/geo-replication/syncdaemon/gsyncd.py
@@ -39,6 +39,7 @@ from changelogagent import agent, Changelog
from gsyncdstatus import set_monitor_status, GeorepStatus
from libcxattr import Xattr
import struct
+from syncdutils import get_master_and_slave_data_from_args
ParseError = XET.ParseError if hasattr(XET, 'ParseError') else SyntaxError
@@ -697,8 +698,11 @@ def main_i():
status_get = rconf.get('status_get')
if status_get:
+ master_name, slave_data = get_master_and_slave_data_from_args(args)
for brick in gconf.path:
brick_status = GeorepStatus(gconf.state_file, brick,
+ master_name,
+ slave_data,
getattr(gconf, "pid_file", None))
checkpoint_time = int(getattr(gconf, "checkpoint", "0"))
brick_status.print_status(checkpoint_time=checkpoint_time)
diff --git a/geo-replication/syncdaemon/gsyncdstatus.py b/geo-replication/syncdaemon/gsyncdstatus.py
index f4d50c1..6575fcd 100644
--- a/geo-replication/syncdaemon/gsyncdstatus.py
+++ b/geo-replication/syncdaemon/gsyncdstatus.py
@@ -17,6 +17,8 @@ import json
import time
from datetime import datetime
from errno import EACCES, EAGAIN, ENOENT
+from syncdutils import EVENT_GEOREP_ACTIVE, EVENT_GEOREP_PASSIVE, gf_event
+from syncdutils import EVENT_GEOREP_CHECKPOINT_COMPLETED
DEFAULT_STATUS = "N/A"
MONITOR_STATUS = ("Created", "Started", "Paused", "Stopped")
@@ -114,7 +116,12 @@ def set_monitor_status(status_file, status):
class GeorepStatus(object):
- def __init__(self, monitor_status_file, brick, monitor_pid_file=None):
+ def __init__(self, monitor_status_file, brick, master, slave,
+ monitor_pid_file=None):
+ self.master = master
+ slv_data = slave.split("::")
+ self.slave_host = slv_data[0]
+ self.slave_volume = slv_data[1].split(":")[0] # Remove Slave UUID
self.work_dir = os.path.dirname(monitor_status_file)
self.monitor_status_file = monitor_status_file
self.filename = os.path.join(self.work_dir,
@@ -137,6 +144,10 @@ class GeorepStatus(object):
data = self.default_values
data = mergerfunc(data)
+ # If Data is not changed by merger func
+ if not data:
+ return False
+
with tempfile.NamedTemporaryFile(
'w',
dir=os.path.dirname(self.filename),
@@ -149,6 +160,7 @@ class GeorepStatus(object):
os.O_DIRECTORY)
os.fsync(dirfd)
os.close(dirfd)
+ return True
def reset_on_worker_start(self):
def merger(data):
@@ -163,10 +175,24 @@ class GeorepStatus(object):
def set_field(self, key, value):
def merger(data):
+ # Current data and prev data is same
+ if data[key] == value:
+ return {}
+
data[key] = value
return json.dumps(data)
- self._update(merger)
+ return self._update(merger)
+
+ def trigger_gf_event_checkpoint_completion(self, checkpoint_time,
+ checkpoint_completion_time):
+ gf_event(EVENT_GEOREP_CHECKPOINT_COMPLETED,
+ master_volume=self.master,
+ slave_host=self.slave_host,
+ slave_volume=self.slave_volume,
+ brick_path=self.brick,
+ checkpoint_time=checkpoint_time,
+ checkpoint_completion_time=checkpoint_completion_time)
def set_last_synced(self, value, checkpoint_time):
def merger(data):
@@ -184,9 +210,13 @@ class GeorepStatus(object):
# previously then update the checkpoint completed time
if checkpoint_time > 0 and checkpoint_time <= value[0]:
if data["checkpoint_completed"] == "No":
+ curr_time = int(time.time())
data["checkpoint_time"] = checkpoint_time
- data["checkpoint_completion_time"] = int(time.time())
+ data["checkpoint_completion_time"] = curr_time
data["checkpoint_completed"] = "Yes"
+ self.trigger_gf_event_checkpoint_completion(
+ checkpoint_time, curr_time)
+
return json.dumps(data)
self._update(merger)
@@ -221,10 +251,20 @@ class GeorepStatus(object):
self._update(merger)
def set_active(self):
- self.set_field("worker_status", "Active")
+ if self.set_field("worker_status", "Active"):
+ gf_event(EVENT_GEOREP_ACTIVE,
+ master_volume=self.master,
+ slave_host=self.slave_host,
+ slave_volume=self.slave_volume,
+ brick_path=self.brick)
def set_passive(self):
- self.set_field("worker_status", "Passive")
+ if self.set_field("worker_status", "Passive"):
+ gf_event(EVENT_GEOREP_PASSIVE,
+ master_volume=self.master,
+ slave_host=self.slave_host,
+ slave_volume=self.slave_volume,
+ brick_path=self.brick)
def get_monitor_status(self):
data = ""
diff --git a/geo-replication/syncdaemon/monitor.py b/geo-replication/syncdaemon/monitor.py
index 22cd1cc..7eddd26 100644
--- a/geo-replication/syncdaemon/monitor.py
+++ b/geo-replication/syncdaemon/monitor.py
@@ -212,7 +212,10 @@ class Monitor(object):
"""
if not self.status.get(w[0]['dir'], None):
self.status[w[0]['dir']] = GeorepStatus(gconf.state_file,
- w[0]['dir'])
+ w[0]['dir'],
+ master,
+ "%s::%s" % (slave_host,
+ slave_vol))
set_monitor_status(gconf.state_file, self.ST_STARTED)
self.status[w[0]['dir']].set_worker_status(self.ST_INIT)
diff --git a/geo-replication/syncdaemon/resource.py b/geo-replication/syncdaemon/resource.py
index 3a9aa33..5cf5eb0 100644
--- a/geo-replication/syncdaemon/resource.py
+++ b/geo-replication/syncdaemon/resource.py
@@ -40,7 +40,7 @@ from syncdutils import ChangelogException, ChangelogHistoryNotAvailable
from syncdutils import get_changelog_log_level
from syncdutils import CHANGELOG_AGENT_CLIENT_VERSION
from gsyncdstatus import GeorepStatus
-
+from syncdutils import get_master_and_slave_data_from_args
UrlRX = re.compile('\A(\w+)://([^ *?[]*)\Z')
HostRX = re.compile('[a-zA-Z\d](?:[a-zA-Z\d.-]*[a-zA-Z\d])?', re.I)
@@ -1474,7 +1474,10 @@ class GLUSTER(AbstractUrl, SlaveLocal, SlaveRemote):
changelog_register_failed = False
(inf, ouf, ra, wa) = gconf.rpc_fd.split(',')
changelog_agent = RepceClient(int(inf), int(ouf))
- status = GeorepStatus(gconf.state_file, gconf.local_path)
+ master_name, slave_data = get_master_and_slave_data_from_args(
+ sys.argv)
+ status = GeorepStatus(gconf.state_file, gconf.local_path,
+ master_name, slave_data)
status.reset_on_worker_start()
rv = changelog_agent.version()
if int(rv) != CHANGELOG_AGENT_CLIENT_VERSION:
diff --git a/geo-replication/syncdaemon/syncdutils.py b/geo-replication/syncdaemon/syncdutils.py
index 0bcf5ec..b8e1e6a 100644
--- a/geo-replication/syncdaemon/syncdutils.py
+++ b/geo-replication/syncdaemon/syncdutils.py
@@ -28,10 +28,17 @@ sys.path.insert(1, GLUSTERFS_LIBEXECDIR)
EVENTS_ENABLED = True
try:
from events.eventtypes import GEOREP_FAULTY as EVENT_GEOREP_FAULTY
+ from events.eventtypes import GEOREP_ACTIVE as EVENT_GEOREP_ACTIVE
+ from events.eventtypes import GEOREP_PASSIVE as EVENT_GEOREP_PASSIVE
+ from events.eventtypes import GEOREP_CHECKPOINT_COMPLETED \
+ as EVENT_GEOREP_CHECKPOINT_COMPLETED
except ImportError:
# Events APIs not installed, dummy eventtypes with None
EVENTS_ENABLED = False
EVENT_GEOREP_FAULTY = None
+ EVENT_GEOREP_ACTIVE = None
+ EVENT_GEOREP_PASSIVE = None
+ EVENT_GEOREP_CHECKPOINT_COMPLETED = None
try:
from cPickle import PickleError
@@ -539,3 +546,15 @@ def gf_event(event_type, **kwargs):
if EVENTS_ENABLED:
from events.gf_event import gf_event as gfevent
gfevent(event_type, **kwargs)
+
+
+def get_master_and_slave_data_from_args(args):
+ master_name = None
+ slave_data = None
+ for arg in args:
+ if arg.startswith(":"):
+ master_name = arg.replace(":", "")
+ if "::" in arg:
+ slave_data = arg.replace("ssh://", "")
+
+ return (master_name, slave_data)
--
1.7.1