1#!/usr/bin/env python3

2# This file is part of Checkbox.

3#

4# Copyright 2024 Canonical Ltd.

5# Written by:

6# Stanley Huang <stanley.huang@canonical.com>

7#

8# Checkbox is free software: you can redistribute it and/or modify

9# it under the terms of the GNU General Public License version 3,

10# as published by the Free Software Foundation.

11#

12# Checkbox is distributed in the hope that it will be useful,

13# but WITHOUT ANY WARRANTY; without even the implied warranty of

14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

15# GNU General Public License for more details.

16#

17# You should have received a copy of the GNU General Public License

18# along with Checkbox. If not, see <http://www.gnu.org/licenses/>.

19

20 import re

21 import sys

22 import logging

23 import argparse

24 import shlex

25 import subprocess

26 from pathlib import Path

27 from typing import Dict

28

29

30 class RAIDStats(Dict):

31 device = str

32 mode = str

33

34

35 def get_md_stat(filename: str = "/proc/mdstat") -> list:

36 """

37 Parse the information of RAID devices from /dev/mdstat

38

39 Returns:

40 raid_stats (list): the RAID information

41 """

42 pattern = r"^(md[0-9]+) : active ([a-z0-9]+) [a-z0-9 \[\]]+$"

43 raid_raw_data = ""

44 raid_stats = []

45

46 with Path(filename) as node:

47 raid_raw_data = node.read_text().strip("\n")

48

49 for tmp_data in raid_raw_data.split("\n"):

50 find_result = re.search(pattern, tmp_data)

51 if find_result:

52 raid_node = RAIDStats(

53 device=find_result.groups()[0], mode=find_result.groups()[1]

54 )

55 raid_stats.append(raid_node)

56

57 return raid_stats

58

59

60 def dump_raid_info(nodes: list) -> None:

61 """

62 Display detail information for all of RAID devices

63

64 Args:

65 nodes (list): the name of MD devices

66 """

67 for node in nodes:

68 subprocess.run(shlex.split("mdadm -D /dev/{}".format(node)))

69

70

71 def check_raid_mode_test(modes: str) -> None:

72 """

73 Validate the RAID modes running on the system is expected

74

75 Args:

76 modes (str): the expected RAID modes on the system

77

78 Raises:

79 ValueError: when if RAID modes are not expected

80 """

81 expected_modes = modes.strip().split()

82 cur_raid_stats = get_md_stat()

83

84 active_mode = [stat["mode"] for stat in cur_raid_stats]

85 logging.info("Active RAID mode on the system: %s", active_mode)

86 logging.info("Expected RAID mode: %s", expected_modes)

87

88 dump_raid_info([stat["device"] for stat in cur_raid_stats])

89

90 if sorted(expected_modes) == sorted(active_mode):

91 logging.info("PASS: RAID mode on the system is expected")

92 else:

93 raise ValueError("RAID mode on the system is not expected")

94

95

96 def register_arguments():

97 parser = argparse.ArgumentParser(

98 formatter_class=argparse.RawDescriptionHelpFormatter,

99 description="Check RAID stats and compare mode is expected",

100 )

101 parser.add_argument(

102 "--mode",

103 required=True,

104 type=str,

105 help="The RAID mode is enabeld on the running system",

106 )

107

108 args = parser.parse_args()

109 return args

110

111

112 if __name__ == "__main__":

113

114 root_logger = logging.getLogger()

115 root_logger.setLevel(logging.INFO)

116 logger_format = "%(asctime)s %(levelname)-8s %(message)s"

117 date_format = "%Y-%m-%d %H:%M:%S"

118

119 # Log DEBUG and INFO to stdout, others to stderr

120 stdout_handler = logging.StreamHandler(sys.stdout)

121 stdout_handler.setFormatter(logging.Formatter(logger_format, date_format))

122

123 stderr_handler = logging.StreamHandler(sys.stderr)

124 stderr_handler.setFormatter(logging.Formatter(logger_format, date_format))

125

126 stdout_handler.setLevel(logging.DEBUG)

127 stderr_handler.setLevel(logging.WARNING)

128

129 # Add a filter to the stdout handler to limit log records to

130 # INFO level and below

131 stdout_handler.addFilter(lambda record: record.levelno <= logging.INFO)

132

133 root_logger.addHandler(stderr_handler)

134 root_logger.addHandler(stdout_handler)

135

136 args = register_arguments()

137

138 try:

139 check_raid_mode_test(args.mode)

140 except Exception as err:

141 logging.error(err)