1#!/usr/bin/env python3 

2# 

3# Copyright (C) 2020 Vates SAS - ronan.abhamon@vates.fr 

4# 

5# This program is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# (at your option) any later version. 

9# This program is distributed in the hope that it will be useful, 

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

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

12# GNU General Public License for more details. 

13# 

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

15# along with this program. If not, see <https://www.gnu.org/licenses/>. 

16# 

17 

18 

19from linstorvolumemanager import \ 

20 get_controller_uri, LinstorVolumeManager, LinstorVolumeManagerError 

21import linstor 

22import re 

23import util 

24 

25 

26class LinstorJournalerError(Exception): 

27 pass 

28 

29# ============================================================================== 

30 

31 

32class LinstorJournaler: 

33 """ 

34 Simple journaler that uses LINSTOR properties for persistent "storage". 

35 A journal is a id-value pair, and there can be only one journal for a 

36 given id. An identifier is juste a transaction name. 

37 """ 

38 

39 REG_TYPE = re.compile('^([^/]+)$') 

40 REG_TRANSACTION = re.compile('^[^/]+/([^/]+)$') 

41 

42 """ 

43 Types of transaction in the journal. 

44 """ 

45 CLONE = 'clone' 

46 INFLATE = 'inflate' 

47 ZERO = 'zero' 

48 

49 @staticmethod 

50 def default_logger(*args): 

51 print(args) 

52 

53 def __init__( 

54 self, 

55 group_name, 

56 uri=None, 

57 native_client=None, 

58 logger=default_logger.__func__ 

59 ): 

60 self._namespace = '{}journal/'.format( 

61 LinstorVolumeManager._build_sr_namespace() 

62 ) 

63 self._logger = logger 

64 self._journal = self._create_journal_instance( 

65 group_name, self._namespace, uri=uri, native_client=native_client 

66 ) 

67 

68 def create(self, type, identifier, value): 

69 # TODO: Maybe rename to 'add' in the future (in Citrix code too). 

70 

71 key = self._get_key(type, identifier) 

72 

73 # 1. Ensure transaction doesn't exist. 

74 current_value = self.get(type, identifier) 

75 if current_value is not None: 

76 raise LinstorJournalerError( 

77 'Journal transaction already exists for \'{}:{}\': {}' 

78 .format(type, identifier, current_value) 

79 ) 

80 

81 # 2. Write! 

82 try: 

83 self._reset_namespace() 

84 self._logger( 

85 'Create journal transaction \'{}:{}\''.format(type, identifier) 

86 ) 

87 self._journal[key] = str(value) 

88 except Exception as e: 

89 try: 

90 self._journal.pop(key, 'empty') 

91 except Exception as e2: 

92 self._logger( 

93 'Failed to clean up failed journal write: {} (Ignored)' 

94 .format(e2) 

95 ) 

96 

97 raise LinstorJournalerError( 

98 'Failed to write to journal: {}'.format(e) 

99 ) 

100 

101 def remove(self, type, identifier): 

102 key = self._get_key(type, identifier) 

103 try: 

104 self._reset_namespace() 

105 self._logger( 

106 'Destroy journal transaction \'{}:{}\'' 

107 .format(type, identifier) 

108 ) 

109 self._journal.pop(key) 

110 except Exception as e: 

111 raise LinstorJournalerError( 

112 'Failed to remove transaction \'{}:{}\': {}' 

113 .format(type, identifier, e) 

114 ) 

115 

116 def get(self, type, identifier): 

117 self._reset_namespace() 

118 return self._journal.get(self._get_key(type, identifier)) 

119 

120 def get_all(self, type): 

121 entries = {} 

122 

123 self._journal.namespace = self._namespace + '{}/'.format(type) 

124 for (key, value) in self._journal.items(): 

125 res = self.REG_TYPE.match(key) 

126 if res: 

127 identifier = res.groups()[0] 

128 entries[identifier] = value 

129 return entries 

130 

131 # Added to compatibility with Citrix API. 

132 def getAll(self, type): 

133 return self.get_all(type) 

134 

135 def has_entries(self, identifier): 

136 self._reset_namespace() 

137 for (key, value) in self._journal.items(): 

138 res = self.REG_TRANSACTION.match(key) 

139 if res: 

140 current_identifier = res.groups()[0] 

141 if current_identifier == identifier: 

142 return True 

143 return False 

144 

145 # Added to compatibility with Citrix API. 

146 def hasJournals(self, identifier): 

147 return self.has_entries(identifier) 

148 

149 def _reset_namespace(self): 

150 self._journal.namespace = self._namespace 

151 

152 @classmethod 

153 def _create_journal_instance(cls, group_name, namespace, *, uri=None, native_client=None): 

154 if not uri and not native_client: 

155 raise LinstorVolumeManagerError( 

156 'Either a URI to the LINSTOR controller or a LINSTOR client must be provided' 

157 ) 

158 

159 if native_client: 

160 return linstor.KV( 

161 LinstorVolumeManager.build_group_name(group_name), 

162 existing_client=native_client, 

163 namespace=namespace 

164 ) 

165 

166 def connect(uri): 

167 if not uri: 

168 uri = get_controller_uri() 

169 if not uri: 

170 raise LinstorVolumeManagerError( 

171 'Unable to find controller uri...' 

172 ) 

173 return linstor.KV( 

174 LinstorVolumeManager.build_group_name(group_name), 

175 uri=uri, 

176 namespace=namespace 

177 ) 

178 

179 try: 

180 return connect(uri) 

181 except (linstor.errors.LinstorNetworkError, LinstorVolumeManagerError): 

182 pass 

183 

184 return util.retry( 

185 lambda: connect(None), 

186 maxretry=10, 

187 exceptions=[ 

188 linstor.errors.LinstorNetworkError, LinstorVolumeManagerError 

189 ] 

190 ) 

191 

192 @staticmethod 

193 def _get_key(type, identifier): 

194 return '{}/{}'.format(type, identifier)