It's not enough to check for the kernel module in the case of cluster

mirrors, we must also check that the log daemon (cmirrord) is running.
The log module can be auto-loaded, but the daemon cannot be
"auto-started".  Failing to check for the daemon produces cryptic
messages that customers have a hard time deciphering.  (The system
messages do report that the log daemon is not running, but people
don't seem to find this message easily.)

Here are examples of what is printed when the module is available,
but the log daemon has not been started.

[root@bp-01 LVM2]# lvcreate -m1 -l1 -n lv vg
  Shared cluster mirrors are not available.

[root@bp-01 LVM2]# lvcreate -m1 -l1 -n lv vg -v
    Setting logging type to disk
    Finding volume group "vg"
    Archiving volume group "vg" metadata (seqno 3).
    Creating logical volume lv
    Executing: /sbin/modprobe dm-log-userspace
    Cluster mirror log daemon is not running
  Shared cluster mirrors are not available.
    Creating volume group backup "/etc/lvm/backup/vg" (seqno 4).

WHATS_NEW

@@ -1,5 +1,6 @@
 Version 2.02.71 -
 ===============================
+  Check if cluster log daemon is running before allowing cmirror create.
   Check if LV with specified name already exists when splitting a mirror.
   Fix suspend/resume logic for LVs resulting from splitting a mirror.
   Switch cmirrord and clvmd to use dm_create_lockfile.

lib/metadata/mirror.c

@@ -80,6 +80,36 @@ int lv_is_mirrored(const struct logical_volume *lv)
 	return 0;
 }
 
+/*
+ * cluster_mirror_is_available
+ *
+ * Check if the proper kernel module and log daemon are running.
+ * Caller should check for 'vg_is_clustered(lv->vg)' before making
+ * this call.
+ *
+ * Returns: 1 if available, 0 otherwise
+ */
+static int cluster_mirror_is_available(struct logical_volume *lv)
+{
+       unsigned attr = 0;
+       struct cmd_context *cmd = lv->vg->cmd;
+       const struct segment_type *segtype;
+
+       if (!(segtype = get_segtype_from_string(cmd, "mirror")))
+               return_0;
+
+       if (!segtype->ops->target_present)
+               return_0;
+
+       if (!segtype->ops->target_present(lv->vg->cmd, NULL, &attr))
+               return_0;
+
+       if (!(attr & MIRROR_LOG_CLUSTERED))
+               return 0;
+
+       return 1;
+}
+
 /*
  * Returns the number of mirrors of the LV
  */
@@ -1940,6 +1970,12 @@ int lv_add_mirrors(struct cmd_context *cmd, struct logical_volume *lv,
 		return 0;
 	}
 
+	if (vg_is_clustered(lv->vg) &&  !(lv->status & ACTIVATE_EXCL) &&
+	    !cluster_mirror_is_available(lv)) {
+		log_error("Shared cluster mirrors are not available.");
+		return 0;
+	}
+
 	/* For corelog mirror, activation code depends on
 	 * the global mirror_in_sync status. As we are adding
 	 * a new mirror, it should be set as 'out-of-sync'

lib/mirror/mirrored.c

@@ -519,6 +519,19 @@ static int _mirrored_target_present(struct cmd_context *cmd,
 					_mirror_attributes |= MIRROR_LOG_CLUSTERED;
 			} else if (module_present(cmd, "log-userspace"))
 				_mirror_attributes |= MIRROR_LOG_CLUSTERED;
+
+                        if (!(_mirror_attributes & MIRROR_LOG_CLUSTERED))
+                                log_verbose("Cluster mirror log module is not available");
+
+                        /*
+                         * The cluster mirror log daemon must be running,
+                         * otherwise, the kernel module will fail to make
+                         * contact.
+                         */
+                        if (!dm_daemon_is_running(CMIRRORD_PIDFILE)) {
+                                log_verbose("Cluster mirror log daemon is not running");
+                                _mirror_attributes &= ~MIRROR_LOG_CLUSTERED;
+                        }
 		}
 		*attributes = _mirror_attributes;
 	}

libdm/libdevmapper.h

@@ -988,6 +988,13 @@ int dm_asprintf(char **buf, const char *format, ...)
  */
 int dm_create_lockfile(const char* lockfile);
 
+/*
+ * Query whether a daemon is running based on its lockfile
+ *
+ * Returns: 1 if running, 0 if not
+ */
+int dm_daemon_is_running(const char* lockfile);
+
 /*********************
  * regular expressions
  *********************/

libdm/libdm-file.c

@@ -166,3 +166,29 @@ fail:
 
 	return 0;
 }
+
+int dm_daemon_is_running(const char* lockfile)
+{
+       int fd;
+       struct flock lock;
+
+       if((fd = open(lockfile, O_RDONLY)) < 0)
+               return 0;
+
+       lock.l_type = F_WRLCK;
+       lock.l_start = 0;
+       lock.l_whence = SEEK_SET;
+       lock.l_len = 0;
+       if (fcntl(fd, F_GETLK, &lock) < 0) {
+               log_error("Cannot check lock status of lockfile [%s], error was [%s]",
+                         lockfile, strerror(errno));
+               if (close(fd))
+                       stack;
+               return 0;
+       }
+
+       if (close(fd))
+               stack;
+
+       return (lock.l_type == F_UNLCK) ? 0 : 1;
+}