metadata: improve error messages when vdi_type is missing

Explicit error message pointing to missing vdi_type on SRMetadata update
Add a distinction for unpacking corrupted empty metadata headers for easier diagnostic

Signed-off-by: Antoine Bartuccio <antoine.bartuccio@vates.tech>

Author: klmp200

drivers/LVHDSR.py

@@ -246,6 +246,10 @@ def updateSRMetadata(self, allocation):
             for vdi in self.session.xenapi.SR.get_VDIs(self.sr_ref):
                 vdi_uuid = self.session.xenapi.VDI.get_uuid(vdi)
 
+                vdi_type = self.session.xenapi.VDI.get_sm_config(vdi).get('vdi_type')
+                if not vdi_type:
+                    raise xs_errors.XenError('MetadataError', opterr=f"Missing `vdi_type` for VDI {vdi_uuid}")
+
                 # Create the VDI entry in the SR metadata
                 vdi_info[vdi_uuid] = \
                 {
@@ -261,7 +265,7 @@ def updateSRMetadata(self, allocation):
                     TYPE_TAG: \
                         self.session.xenapi.VDI.get_type(vdi),
                     VDI_TYPE_TAG: \
-                       self.session.xenapi.VDI.get_sm_config(vdi)['vdi_type'],
+                       vdi_type,
                     READ_ONLY_TAG: \
                         int(self.session.xenapi.VDI.get_read_only(vdi)),
                     METADATA_OF_POOL_TAG: \

drivers/srmetadata.py

@@ -15,6 +15,7 @@
 #
 # Functions to read and write SR metadata
 #
+
 from io import SEEK_SET
 
 import util
@@ -133,11 +134,11 @@ def file_read_wrapper(fd, offset, bytesToRead=METADATA_BLK_SIZE):
             ([fd, offset, bytesToRead], e.errno))
 
 
-def to_utf8(s):
+def to_utf8(s: str) -> bytes:
     return s.encode("utf-8")
 
 
-def from_utf8(bs):
+def from_utf8(bs: bytes) -> str:
     return bs.decode("utf-8")
 
 
@@ -177,13 +178,14 @@ def buildHeader(length, major=metadata.MD_MAJOR, minor=metadata.MD_MINOR):
                    + str(minor))
 
 
-def unpackHeader(header):
-    vals = from_utf8(header).split(HEADER_SEP)
+def unpackHeader(header: bytes):
+    decoded = from_utf8(header)
+    if len(decoded.rstrip('\x00')) == 0:
+        raise xs_errors.XenError('MetadataError', opterr='Empty header')
+    vals = decoded.split(HEADER_SEP)
     if len(vals) != 4 or vals[0] != metadata.HDR_STRING:
-        util.SMlog("Exception unpacking metadata header: "
-                   "Error: Bad header '%s'" % (header))
-        raise xs_errors.XenError('MetadataError', \
-                        opterr='Bad header')
+        util.SMlog(f"Exception unpacking metadata header: Error: Bad header {header!r}")
+        raise xs_errors.XenError('MetadataError', opterr='Bad header')
     return (vals[0], vals[1], vals[2], vals[3])
 
 

tests/test_LVHDSR.py

@@ -276,6 +276,16 @@ def cmd(args):
             self.assertEqual(1, lvm_cache.activate.call_count)
             self.assertEqual(1, lvm_cache.deactivate.call_count)
 
+        # Act (3)
+        # This tests SR metadata updates
+        sr.updateSRMetadata('thick')
+
+        # Test that removing vdi_type on a vdi does crash properly
+        del vdi_data['vdi2_ref']['sm-config']['vdi_type']
+        with self.assertRaises(Exception):
+            # Fail on vdi2_ref
+            sr.updateSRMetadata('thick')
+
     def convert_vdi_to_meta(self, vdi_data):
         metadata = {}
         for item in vdi_data.items():

tests/test_srmetadata.py

@@ -5,6 +5,7 @@
 import uuid
 import unittest
 import unittest.mock as mock
+import xs_errors
 
 from srmetadata import (LVMMetadataHandler, buildHeader, buildXMLSector,
                         getMetadataLength, unpackHeader, updateLengthInHeader,
@@ -25,6 +26,42 @@ def test_unpackHeader(self):
         self.assertEqual(int(major), 1)
         self.assertEqual(int(minor), 2)
 
+    def test_unpackHeader_empty(self):
+        # Given
+        headers = [b"", b"\x00", b"\x00"]
+
+        for header in headers:
+            # When
+            with self.assertRaises(xs_errors.SROSError) as error:
+                unpackHeader(header)
+
+            # Then
+            self.assertEqual(
+                error.exception.args[0],
+                'Error in Metadata volume operation for SR. [opterr=Empty header]'
+            )
+
+    def test_unpackHeader_bad(self):
+        # Given
+        headers = [
+            b"BAD:4096      :1:2" + (b' ' * 493),
+            b"BAD:4096      :1:2",
+            b"XSSM:4096      :1",
+            b"XSSM:4096      ",
+            b"XSSM",
+        ]
+
+        for header in headers:
+            # When
+            with self.assertRaises(xs_errors.SROSError) as error:
+                unpackHeader(header)
+
+            # Then
+            self.assertEqual(
+                error.exception.args[0],
+                'Error in Metadata volume operation for SR. [opterr=Bad header]'
+            )
+
     def test_buildHeader_unpackHeader_roundTrip(self):
         # Given
         orig_length = 12345