Change startup procedure

Current startup procedure works on an assumption that we will
deal with asynchronously appearing devices in asynchronous way
(udev rules) and synchronous events in the system (systemd units)
won't interfere. If we would break anything (mounts) we would just
take those units and restart them. This tactic was working as long
as resetting systemd units took reasonable time.

As hackish as it sounds it worked in all systems that the software
has been validated on. Unfortunately it stopped working because
of *.mount units taking MUCH longer time to restart even on
mainstream OSes, so it's time to change.

This change implements open-cas systemd service which will wait
synchronously with systemd bootup process for all required Open CAS
devices to start. If they don't we fail the boot process just as
failing mounts would. We also make sure that this process takes place
before any mounts (aside from root FS and other critical FS's) are
even attempted. Now opencas-mount-utility can be discarded.

To override this behaviour on per-core basis you can specify
lazy_startup=true option in opencas.conf.

Signed-off-by: Jan Musial <jan.musial@intel.com>

utils/60-persistent-storage-cas-load.rules

@@ -3,9 +3,4 @@ SUBSYSTEM!="block", GOTO="cas_loader_end"
 
 RUN+="/lib/opencas/open-cas-loader /dev/$name"
 
-# Work around systemd<->udev interaction, make sure filesystems with labels on
-# cas are mounted properly
-KERNEL!="cas*", GOTO="cas_loader_end"
-IMPORT{builtin}="blkid"
-ENV{ID_FS_USAGE}=="filesystem|other", ENV{ID_FS_LABEL_ENC}=="?*", RUN+="/lib/opencas/open-cas-mount-utility $env{ID_FS_LABEL_ENC}"
 LABEL="cas_loader_end"

utils/Makefile

@@ -9,25 +9,11 @@ UDEV:=$(shell which udevadm)
 SYSTEMCTL := $(shell which systemctl)
 PYTHON3 := $(shell which python3)
 
-ifeq (, $(shell which systemctl))
-define cas_install
-	install -m 755 open-cas-shutdown /etc/init.d/open-cas-shutdown
-	/sbin/chkconfig open-cas-shutdown on; service open-cas-shutdown start
-endef
-else
 ifneq "$(wildcard /usr/lib/systemd/system)" ""
 	SYSTEMD_DIR=/usr/lib/systemd/system
 else
 	SYSTEMD_DIR=/lib/systemd/system
 endif
-define cas_install
-	install -m 644 open-cas-shutdown.service $(SYSTEMD_DIR)/open-cas-shutdown.service
-	install -m 755 -d $(SYSTEMD_DIR)/../system-shutdown
-	install -m 755 open-cas.shutdown $(SYSTEMD_DIR)/../system-shutdown/open-cas.shutdown
-	$(SYSTEMCTL) daemon-reload
-	$(SYSTEMCTL) -q enable open-cas-shutdown
-endef
-endif
 
 # Just a placeholder when running make from parent dir without install/uninstall arg
 all: ;
@@ -42,7 +28,6 @@ else
 	@install -m 644 opencas.py $(CASCTL_DIR)/opencas.py
 	@install -m 755 casctl $(CASCTL_DIR)/casctl
 	@install -m 755 open-cas-loader $(CASCTL_DIR)/open-cas-loader
-	@install -m 755 open-cas-mount-utility $(CASCTL_DIR)/open-cas-mount-utility
 
 	@ln -fs $(CASCTL_DIR)/casctl /sbin/casctl
 
@@ -55,14 +40,19 @@ else
 
 	@install -m 644 casctl.8 /usr/share/man/man8/casctl.8
 
-	$(cas_install)
+	@install -m 644 open-cas-shutdown.service $(SYSTEMD_DIR)/open-cas-shutdown.service
+	@install -m 644 open-cas.service $(SYSTEMD_DIR)/open-cas.service
+	@install -m 755 -d $(SYSTEMD_DIR)/../system-shutdown
+	@install -m 755 open-cas.shutdown $(SYSTEMD_DIR)/../system-shutdown/open-cas.shutdown
+	@$(SYSTEMCTL) daemon-reload
+	@$(SYSTEMCTL) -q enable open-cas-shutdown
+	@$(SYSTEMCTL) -q enable open-cas
 endif
 
 uninstall:
 	@rm $(CASCTL_DIR)/opencas.py
 	@rm $(CASCTL_DIR)/casctl
 	@rm $(CASCTL_DIR)/open-cas-loader
-	@rm $(CASCTL_DIR)/open-cas-mount-utility
 	@rm -rf $(CASCTL_DIR)
 
 	@rm /sbin/casctl
@@ -71,6 +61,15 @@ uninstall:
 
 	@rm /lib/udev/rules.d/60-persistent-storage-cas-load.rules
 	@rm /lib/udev/rules.d/60-persistent-storage-cas.rules
+	@$(UDEV) control --reload-rules
+
+	@$(SYSTEMCTL) -q disable open-cas-shutdown
+	@$(SYSTEMCTL) -q disable open-cas
+	@$(SYSTEMCTL) daemon-reload
+
+	@rm $(SYSTEMD_DIR)/open-cas-shutdown.service
+	@rm $(SYSTEMD_DIR)/open-cas.service
+	@rm $(SYSTEMD_DIR)/../system-shutdown/open-cas.shutdown
 
 
 .PHONY: install uninstall clean distclean

utils/casctl

@@ -100,6 +100,27 @@ def init(force):
 
     exit(exit_code)
 
+
+def settle(timeout, interval):
+    try:
+        not_initialized = opencas.wait_for_startup(timeout, interval)
+    except Exception as e:
+        eprint(e)
+        exit(1)
+
+    if not_initialized:
+        eprint("Open CAS initialization failed. Couldn't set up all required devices")
+        for device in not_initialized:
+            eprint(
+                "Couldn't add device {} as core {} in cache {}".format(
+                    device.device, device.core_id, device.cache_id
+                )
+            )
+        exit(1)
+
+    exit(0)
+
+
 # Stop - detach cores and stop caches
 def stop(flush):
     try:
@@ -107,30 +128,55 @@ def stop(flush):
     except Exception as e:
         eprint(e)
 
+
 # Command line arguments parsing
 
+
 class cas:
     def __init__(self):
-        parser = argparse.ArgumentParser(prog = 'cas')
+        parser = argparse.ArgumentParser(prog="casctl")
-        subparsers = parser.add_subparsers(title = 'actions')
+        subparsers = parser.add_subparsers(title="actions")
 
-        parser_init = subparsers.add_parser('init', help = 'Setup initial configuration')
+        parser_init = subparsers.add_parser("init", help="Setup initial configuration")
-        parser_init.set_defaults(command='init')
+        parser_init.set_defaults(command="init")
-        parser_init.add_argument ('--force', action='store_true', help = 'Force cache start')
+        parser_init.add_argument(
+            "--force", action="store_true", help="Force cache start"
+        )
 
-        parser_start = subparsers.add_parser('start', help = 'Start cache configuration')
+        parser_start = subparsers.add_parser("start", help="Start cache configuration")
-        parser_start.set_defaults(command='start')
+        parser_start.set_defaults(command="start")
 
-        parser_stop = subparsers.add_parser('stop', help = 'Stop cache configuration')
+        parser_settle = subparsers.add_parser(
-        parser_stop.set_defaults(command='stop')
+            "settle", help="Wait for startup of devices"
-        parser_stop.add_argument ('--flush', action='store_true', help = 'Flush data before stopping')
+        )
+        parser_settle.set_defaults(command="settle")
+        parser_settle.add_argument(
+            "--timeout",
+            action="store",
+            help="How long should command wait [s]",
+            default=270,
+            type=int,
+        )
+        parser_settle.add_argument(
+            "--interval",
+            action="store",
+            help="Polling interval [s]",
+            default=5,
+            type=int,
+        )
+
+        parser_stop = subparsers.add_parser("stop", help="Stop cache configuration")
+        parser_stop.set_defaults(command="stop")
+        parser_stop.add_argument(
+            "--flush", action="store_true", help="Flush data before stopping"
+        )
 
         if len(sys.argv[1:]) == 0:
             parser.print_help()
             return
 
         args = parser.parse_args(sys.argv[1:])
-        getattr(self, 'command_' + args.command)(args)
+        getattr(self, "command_" + args.command)(args)
 
     def command_init(self, args):
         init(args.force)
@@ -138,6 +184,9 @@ class cas:
     def command_start(self, args):
         start()
 
+    def command_settle(self, args):
+        settle(args.timeout, args.interval)
+
     def command_stop(self, args):
         stop(args.flush)
 

utils/casctl.8

@@ -23,6 +23,10 @@ Stop all cache instances.
 .B init
 Initial configuration of caches and core devices.
 
+.TP
+.B settle
+Wait for all core devices to be added to respective caches.
+
 .br
 .B CAUTION
 .br
@@ -51,6 +55,17 @@ Flush data before stopping.
 .B --force
 Force cache start even if cache device contains partitions or metadata from previously running cache instances.
 
+.TP
+.SH Options that are valid with settle are:
+
+.TP
+.B --timeout
+How long will command block waiting for devices to start before timing out [s].
+
+.TP
+.B --interval
+How often will command poll for status change [s].
+
 .TP
 .SH Command --help (-h) does not accept any options.
 

utils/open-cas-mount-utility

@@ -1,26 +0,0 @@
-#!/bin/bash
-#
-# Copyright(c) 2012-2019 Intel Corporation
-# SPDX-License-Identifier: BSD-3-Clause-Clear
-#
-
-# Find all mount units, cut to remove list-units decorations
-logger "Open CAS Mount Utility checking for $1 FS label..."
-MOUNT_UNITS=`systemctl --plain list-units | grep \.mount | grep -v '\-\.mount' | awk '{print $1}'`
-
-for unit in $MOUNT_UNITS
-do
-	# Find BindsTo keyword, pry out FS label from the .device unit name
-	label=`systemctl show $unit | grep BindsTo | sed "s/.*label\-\(.*\)\.device/\1/;tx;d;:x"`
-	if [ "$label" == "" ]; then
-		continue
-	fi
-	label_unescaped=$(systemd-escape -u $(systemd-escape -u $label))
-	if [ "$label_unescaped" == "$1" ]; then
-		# If FS label matches restart unit
-		logger "Open CAS Mount Utility restarting $unit..."
-		systemctl restart $unit &> /dev/null
-		exit 0
-	fi
-done
-

utils/open-cas.service

@@ -0,0 +1,22 @@
+#
+# Copyright(c) 2019 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+
+[Unit]
+Description=opencas initialization service
+After=systemd-remount-fs.service
+Before=local-fs-pre.target local-fs.target
+Wants=local-fs-pre.target local-fs.target
+DefaultDependencies=no
+OnFailure=emergency.target
+OnFailureJobMode=isolate
+
+[Service]
+Type=oneshot
+RemainAfterExit=yes
+ExecStart=/sbin/casctl settle --timeout 150 --interval 5
+TimeoutStartSec=3min
+
+[Install]
+RequiredBy=local-fs.target local-fs-pre.target

utils/opencas.conf.5

@@ -36,6 +36,7 @@ Core ID <0-4095>
 .br
 Core device <DEVICE>
 .br
+Extra fields (optional) lazy_startup=<true,false>
 .RE
 .TP
 \fBNOTES\fR

utils/opencas.py

@@ -8,6 +8,7 @@ import csv
 import re
 import os
 import stat
+import time
 
 # Casadm functionality
 
@@ -308,24 +309,35 @@ class cas_config(object):
             return ret
 
     class core_config(object):
-        def __init__(self, cache_id, core_id, path):
+        def __init__(self, cache_id, core_id, path, **params):
             self.cache_id = int(cache_id)
             self.core_id = int(core_id)
             self.device = path
+            self.params = params
 
         @classmethod
         def from_line(cls, line, allow_incomplete=False):
             values = line.split()
-            if len(values) > 3:
+            if len(values) > 4:
-                raise ValueError('Invalid core configuration (too many columns)')
+                raise ValueError("Invalid core configuration (too many columns)")
             elif len(values) < 3:
-                raise ValueError('Invalid core configuration (too few columns)')
+                raise ValueError("Invalid core configuration (too few columns)")
 
             cache_id = int(values[0])
             core_id = int(values[1])
             device = values[2]
 
-            core_config = cls(cache_id, core_id, device)
+            params = dict()
+            if len(values) > 3:
+                for param in values[3].lower().split(","):
+                    param_name, param_value = param.split("=")
+                    if param_name in params:
+                        raise ValueError(
+                            "Invalid core configuration (repeated parameter)"
+                        )
+                    params[param_name] = param_value
+
+            core_config = cls(cache_id, core_id, device, **params)
 
             core_config.validate_config(allow_incomplete)
 
@@ -335,9 +347,24 @@ class cas_config(object):
             self.check_core_id_valid()
             self.check_recursive()
             cas_config.cache_config.check_cache_id_valid(self.cache_id)
+
+            for param_name, param_value in self.params.items():
+                self.validate_parameter(param_name, param_value)
+
             if not allow_incomplete:
                 cas_config.check_block_device(self.device)
 
+        def validate_parameter(self, param_name, param_value):
+            if param_name == "lazy_startup":
+                if param_value.lower() not in ["true", "false"]:
+                    raise ValueError(
+                        "{} is invalid value for '{}' core param".format(
+                            param_value, param_name
+                        )
+                    )
+            else:
+                raise ValueError("'{}' is invalid core param name".format(param_name))
+
         def check_core_id_valid(self):
             if not 0 <= int(self.core_id) <= 4095:
                 raise ValueError('{0} is invalid core id'.format(self.core_id))
@@ -353,7 +380,14 @@ class cas_config(object):
                 raise ValueError('Recursive configuration detected')
 
         def to_line(self):
-            return '{0}\t{1}\t{2}\n'.format(self.cache_id, self.core_id, self.device)
+            ret = "{0}\t{1}\t{2}".format(self.cache_id, self.core_id, self.device)
+            for i, (param, value) in enumerate(self.params.items()):
+                ret += "," if i > 0 else "\t"
+
+                ret += "{0}={1}".format(param, value)
+            ret += "\n"
+
+            return ret
 
     def __init__(self, caches=None, cores=None, version_tag=None):
         self.caches = caches if caches else dict()
@@ -494,6 +528,13 @@ class cas_config(object):
         except:
             raise Exception('Couldn\'t write config file')
 
+    def get_startup_cores(self):
+        return [
+            core
+            for core in self.cores
+            if core.params.get("lazy_startup", "false") == "false"
+        ]
+
 # Config helper functions
 
 
@@ -689,3 +730,69 @@ def stop(flush):
 
     error.raise_nonempty()
 
+
+def get_devices_state():
+    device_list = get_caches_list()
+
+    devices = {"core_pool": [], "caches": {}, "cores": {}}
+
+    core_pool = False
+    prev_cache_id = -1
+
+    for device in device_list:
+        if device["type"] == "core pool":
+            core_pool = True
+            continue
+
+        if device["type"] == "cache":
+            core_pool = False
+            prev_cache_id = int(device["id"])
+            devices["caches"].update(
+                {
+                    int(device["id"]): {
+                        "device": device["disk"],
+                        "status": device["status"],
+                    }
+                }
+            )
+        elif device["type"] == "core":
+            core = {"device": device["disk"], "status": device["status"]}
+            if core_pool:
+                devices["core_pool"].append(core)
+            else:
+                core.update({"cache_id": prev_cache_id})
+                devices["cores"].update(
+                    {(prev_cache_id, int(device["id"])): core}
+                )
+
+    return devices
+
+
+def wait_for_startup(timeout=300, interval=5):
+    try:
+        config = cas_config.from_file(
+            cas_config.default_location, allow_incomplete=True
+        )
+    except Exception as e:
+        raise Exception("Unable to load opencas config. Reason: {0}".format(str(e)))
+
+    stop_time = time.time() + int(timeout)
+
+    not_initialized = None
+    target_core_state = config.get_startup_cores()
+
+    while stop_time > time.time():
+        not_initialized = []
+        runtime_core_state = get_devices_state()["cores"]
+
+        for core in target_core_state:
+            runtime_state = runtime_core_state.get((core.cache_id, core.core_id), None)
+            if not runtime_state or runtime_state["status"] != "Active":
+                not_initialized.append(core)
+
+        if not not_initialized:
+            break
+
+        time.sleep(interval)
+
+    return not_initialized