Add Data Bucket Stream to consistency checks.

bt5/erp5_neo_administration/ExtensionTemplateItem/portal_components/extension.erp5.NEOAdministration.py

@@ -6,6 +6,48 @@ from ZODB import DB
 from neo.client.Storage import Storage
 import hashlib
 
+def DataBucketStream_getChecksumListFromNEONodeListForKey(self, \
+                            node_list, \
+                            ca_file, \
+                            cert_file, \
+                            key_file, \
+                            key, \
+                            threshold):
+  """
+  Directly connect to NEO backends and check checksums of this Data Bucket Stream for this key.
+  """
+  checksum_list = []
+
+  # get directly checksum as we have access to data stream over self
+  data = self.getBucket(key)
+  data = data[:threshold]
+  checksum = hashlib.sha256(data).hexdigest()
+  checksum_list.append(checksum)
+
+  for node in node_list:
+    kw = {'master_nodes': node[0],
+          'name':         node[1],
+          'ca':           ca_file,
+          'cert':         cert_file,
+          'key':          key_file}
+
+    # make a direct connection
+    stor = Storage(**kw)
+    db = DB(stor)
+    conn = db.open()
+    root = conn.root()
+    data_stream_id = self.getId()
+    data_stream = root['Application'].erp5.data_stream_module[data_stream_id]
+    data = data_stream.getBucket(key)
+    data = data[:threshold]
+    conn.close()
+    db.close()
+
+    checksum = hashlib.sha256(data).hexdigest()
+    checksum_list.append(checksum)
+
+  return checksum_list
+
 def DataStream_getChecksumListFromNEONodeListForStartStopOffset(self, \
                             node_list, \
                             ca_file, \

bt5/erp5_neo_administration/SkinTemplateItem/portal_skins/erp5_neo_administration/DataBucketStream_getChecksumListFromNEONodeListForKey.xml

+<?xml version="1.0"?>
+<ZopeData>
+  <record id="1" aka="AAAAAAAAAAE=">
+    <pickle>
+      <global name="ExternalMethod" module="Products.ExternalMethod.ExternalMethod"/>
+    </pickle>
+    <pickle>
+      <dictionary>
+        <item>
+            <key> <string>_function</string> </key>
+            <value> <string>DataBucketStream_getChecksumListFromNEONodeListForKey</string> </value>
+        </item>
+        <item>
+            <key> <string>_module</string> </key>
+            <value> <string>NEOAdministration</string> </value>
+        </item>
+        <item>
+            <key> <string>id</string> </key>
+            <value> <string>DataBucketStream_getChecksumListFromNEONodeListForKey</string> </value>
+        </item>
+        <item>
+            <key> <string>title</string> </key>
+            <value> <string></string> </value>
+        </item>
+      </dictionary>
+    </pickle>
+  </record>
+</ZopeData>

bt5/erp5_neo_administration/SkinTemplateItem/portal_skins/erp5_neo_administration/DataStream_checkIfNEOCloneBackupIsConsistent.py

 """
-  Check replication of Data Stream's content amongst NEO clones.
+  Check replication of Data Stream's or Data Bucket Stream's content amongst NEO clones.
 """
 import random
 
-# sometimes we have HUGE streams so we need to check some randomly generated
-# portion of a Data Stream
-max_stop_offset = context.getSize()
-start_offset = int(random.random() * max_stop_offset)
-stop_offset = start_offset + threshold
-if stop_offset>max_stop_offset:
-  stop_offset = max_stop_offset
+portal_type = context.getPortalType()
+if portal_type == 'Data Stream':
+  # sometimes we have HUGE streams so we need to check some randomly generated
+  # portion of a Data Stream
+  max_stop_offset = context.getSize()
+  start_offset = int(random.random() * max_stop_offset)
+  stop_offset = start_offset + threshold
+  if stop_offset > max_stop_offset:
+    stop_offset = max_stop_offset
+  checksum_list = context.DataStream_getChecksumListFromNEONodeListForStartStopOffset(
+                            neo_node_list,
+                            neo_cert_list[0],
+                            neo_cert_list[1],
+                            neo_cert_list[2],
+                            start_offset,
+                            stop_offset)
+elif portal_type == 'Data Bucket Stream':
+  # choose a random bucket key as raw data source
+  key_list = context.getKeyList()
+  # choose random key
+  key = random.choice(key_list)
+  checksum_list = context.DataBucketStream_getChecksumListFromNEONodeListForKey(
+                            neo_node_list,
+                            neo_cert_list[0],
+                            neo_cert_list[1],
+                            neo_cert_list[2],
+                            key,
+                            threshold)
 
-checksum_list = context.DataStream_getChecksumListFromNEONodeListForStartStopOffset(
-                          neo_node_list,
-                          neo_cert_list[0],
-                          neo_cert_list[1],
-                          neo_cert_list[2],
-                          start_offset,
-                          stop_offset)
 if len(set(checksum_list)) > 1:
   # one of checksums didn't match
   print "PROBLEM:", context.getRelativeUrl(), checksum_list

bt5/erp5_neo_administration/SkinTemplateItem/portal_skins/erp5_neo_administration/ERP5Site_checkNEOCloneBackupConsistency.py

@@ -15,13 +15,14 @@ active_process.setTitle("NEO_Clone_check")
 # which Data Streams to check ...
 data_stream_list = context.ERP5Site_getDataStreamListToCheck()
 
-# we can only check if we have some data inside thus filter out
-data_stream_list = [x for x in data_stream_list if x.getSize() > 0]
-
 for data_stream in data_stream_list:
-  tag = '%s_consistency_check' %data_stream.getPath()
-  data_stream.activate(tag = tag,
-                       active_process = active_process.getPath()).DataStream_checkIfNEOCloneBackupIsConsistent(
+  portal_type = data_stream.getPortalType()
+  if ((portal_type == "Data Stream" and data_stream.getSize() > 0) or
+     (portal_type == "Data Bucket Stream" and len(data_stream.getKeyList()) > 0)):
+    # Data Stream or Data Bucket Stream needs to have data ...
+    tag = '%s_consistency_check' %data_stream.getPath()
+    data_stream.activate(tag = tag,
+                         active_process = active_process.getPath()).DataStream_checkIfNEOCloneBackupIsConsistent(
                                                                     neo_node_list,
                                                                     neo_cert_list,
                                                                     threshold)