Lustre implementation follow-up (DataDog#20727)

* Handle dict access better

* Use `caplog` fixture to test logging

* Update IP extraction logic

* Add test for malformed IP in param name

* Remove useless import

* Add changelog

* Update lustre/changelog.d/20727.fixed

Co-authored-by: NouemanKHAL <noueman.khalikine@datadoghq.com>

* Apply suggestions from code review

Co-authored-by: NouemanKHAL <noueman.khalikine@datadoghq.com>

* Improve logging as suggested in code review

* Add debug log for non 0 returncode

* Assert at least one filesystem was found in discovery

* Separate log tests

* Improve handle_ip function logic

---------

Co-authored-by: NouemanKHAL <noueman.khalikine@datadoghq.com>

Author: dkirov-dd

lustre/changelog.d/20727.fixed

@@ -0,0 +1 @@
+Improve error handling implementation with enhanced debug logs and refined logic

lustre/datadog_checks/lustre/check.py

@@ -4,6 +4,7 @@
 import re
 import subprocess
 from datetime import datetime, timezone
+from ipaddress import ip_address
 from typing import Any, Dict, List, Set, Tuple, Union
 
 import yaml
@@ -36,20 +37,24 @@ def _get_stat_type(suffix: str, unit: str) -> str:
         return 'gauge'
 
 
-def _handle_ip_in_param(parts: List[str]) -> List[str]:
+def _handle_ip_in_param(parts: List[str]) -> Tuple[List[str], bool]:
     """
     Merge parameter parts corresponding to an IP address.
 
     Example:
         ['some','172','0','0','12@tcp','param']
     =>  ['some','172.0.0.12@tcp', 'param']
     """
-    match_index = 0
-    for i, part in enumerate(parts):
-        if '@' in part:
-            match_index = i
-    new_part = ".".join(parts[match_index - 3 : match_index + 1])
-    return [*parts[: match_index - 3], new_part, *parts[match_index + 1 :]]
+    match_indexes = [i for i in range(len(parts)) if '@' in parts[i]]
+    if len(match_indexes) != 1 or match_indexes[0] < 3:
+        return [], False
+    index = match_indexes[0]
+    new_part = ".".join(parts[index - 3 : index + 1])
+    try:
+        ip_address(new_part.split('@')[0])
+    except ValueError:
+        return [], False
+    return [*parts[: index - 3], new_part, *parts[index + 1 :]], True
 
 
 class LustreCheck(AgentCheck):
@@ -160,6 +165,7 @@ def _update_filesystems(self) -> None:
                     filesystem = match.group(0)
                     filesystems.append(filesystem)
             self.filesystems = list(set(filesystems))  # Remove duplicates
+            assert self.filesystems, f'Nothing matched regex `{filesystem_regex}` in params {lines}'
             self.log.debug('Found filesystem(s): %s', self.filesystems)
         except Exception as e:
             self.log.error('Failed to find filesystems: %s', e)
@@ -194,6 +200,11 @@ def _run_command(self, bin: str, *args: str, sudo: bool = False) -> str:
         try:
             self.log.debug('Running command: %s', cmd)
             output = subprocess.run(cmd, timeout=5, shell=True, capture_output=True, text=True)
+            if not output.returncode == 0 and output.stderr:
+                self.log.debug(
+                    'Command %s exited with returncode %s. Captured stderr: %s', cmd, output.returncode, output.stderr
+                )
+                return ''
             if output.stdout is None:
                 self.log.debug(
                     'Command %s returned no output, check if dd-agent is running\
@@ -218,12 +229,12 @@ def submit_jobstats_metrics(self, filesystems: List[str]) -> None:
             device_name = jobstats_param.split('.')[1]  # For example: lustre-MDT0000
             if not any(device_name.startswith(fs) for fs in filesystems):
                 continue
-            jobstats_metrics = self._get_jobstats_metrics(jobstats_param)['job_stats']
+            jobstats_metrics = self._get_jobstats_metrics(jobstats_param).get('job_stats')
             if jobstats_metrics is None:
                 self.log.debug('No jobstats metrics found for %s', jobstats_param)
                 continue
             for job in jobstats_metrics:
-                job_id = job['job_id']
+                job_id = job.get('job_id', "unknown")
                 tags = self.tags + [f'device_name:{device_name}', f'job_id:{job_id}']
                 for metric_name, metric_values in job.items():
                     if not isinstance(metric_values, dict):
@@ -265,15 +276,19 @@ def _get_jobstats_metrics(self, jobstats_param: str) -> Dict[str, Any]:
         jobstats_output = self._run_command('lctl', 'get_param', '-ny', jobstats_param, sudo=True)
         try:
             return yaml.safe_load(jobstats_output) or {}
-        except KeyError:
-            self.log.debug('No jobstats metrics found for %s', jobstats_param)
+        except Exception as e:
+            self.log.debug('Could not get data for "%s", caught exception: %s', jobstats_param, e)
             return {}
 
     def submit_lnet_stats_metrics(self) -> None:
         '''
         Submit the lnet stats metrics.
         '''
-        lnet_metrics = self._get_lnet_metrics('stats')['statistics']
+        lnet_metrics = self._get_lnet_metrics('stats')
+        if 'statistics' not in lnet_metrics:
+            self.log.debug('Could not find `statistics` property in the output of lnet stats. Output: %s', lnet_metrics)
+            return
+        lnet_metrics = lnet_metrics['statistics']
         for metric in lnet_metrics:
             if metric.endswith('_count') or metric == 'errors':
                 metric_type = 'count'
@@ -285,7 +300,11 @@ def submit_lnet_local_ni_metrics(self) -> None:
         '''
         Submit the lnet local ni metrics.
         '''
-        lnet_local_stats = self._get_lnet_metrics('net')['net']
+        lnet_local_stats = self._get_lnet_metrics('net')
+        if 'net' not in lnet_local_stats:
+            self.log.debug('Could not find `net` property in the output of lnet stats. Output: %s', lnet_local_stats)
+            return
+        lnet_local_stats = lnet_local_stats['net']
         for net in lnet_local_stats:
             net_type = net.get('net type')
             for ni in net.get('local NI(s)', []):
@@ -302,7 +321,11 @@ def submit_lnet_peer_ni_metrics(self) -> None:
         '''
         Submit the lnet peer ni metrics.
         '''
-        lnet_peer_stats = self._get_lnet_metrics('peer')['peer']
+        lnet_peer_stats = self._get_lnet_metrics('peer')
+        if 'peer' not in lnet_peer_stats:
+            self.log.debug('Could not find `peer` property in the output of lnet stats. Output: %s', lnet_peer_stats)
+            return
+        lnet_peer_stats = lnet_peer_stats['peer']
         for peer in lnet_peer_stats:
             nid = peer.get('primary nid')
             for ni in peer.get('peer ni', []):
@@ -347,8 +370,8 @@ def _get_lnet_metrics(self, stats_type: str = 'stats') -> Dict[str, Any]:
         lnet_stats = self._run_command('lnetctl', stats_type, 'show', '-v', self.lnetctl_verbosity, sudo=True)
         try:
             return yaml.safe_load(lnet_stats) or {}
-        except (KeyError, ValueError):
-            self.log.debug('No lnet stats found')
+        except Exception as e:
+            self.log.debug('Could not get lnet %s, caught exception: %s', stats_type, e)
             return {}
 
     def submit_param_data(self, params: Set[LustreParam], filesystems: List[str]) -> None:
@@ -400,32 +423,36 @@ def _submit_stat(self, prefix: str, name: str, value: Dict[str, Any], tags: List
             else:
                 self.log.debug('Unexpected metric value for %s.%s: %s', name, suffix, metric_value)
 
-    def _extract_tags_from_param(self, param_regex: str, param_name: str, wildcards: Tuple[str]) -> List[str]:
+    def _extract_tags_from_param(self, param_regex: str, param_name: str, wildcards: Tuple[str, ...]) -> List[str]:
         '''
         Extract tags from the parameter name based on the regex and wildcard meanings.
         '''
+        if not wildcards:
+            return []
         tags = []
-        if wildcards:
-            regex_parts = param_regex.split('.')
-            param_parts = param_name.split('.')
-            wildcard_number = 0
-            if not len(regex_parts) == len(param_parts):
-                # Edge case: mdt.lustre-MDT0000.exports.172.31.16.218@tcp.stats
-                if any("@" in part for part in param_parts):
-                    # We need to reconstruct the address
-                    param_parts = _handle_ip_in_param(param_parts)
-                else:
-                    self.log.debug('Parameter name %s does not match regex %s', param_name, param_regex)
+        regex_parts = param_regex.split('.')
+        param_parts = param_name.split('.')
+        wildcard_number = 0
+        if not len(regex_parts) == len(param_parts):
+            # Edge case: mdt.lustre-MDT0000.exports.172.31.16.218@tcp.stats
+            if len(regex_parts) + 3 == len(param_parts):
+                # We need to reconstruct the address
+                param_parts, is_valid_ip = _handle_ip_in_param(param_parts)
+                if not is_valid_ip:
+                    self.log.debug("Skipping tags for parameter %s", param_name)
+                    return []
+            else:
+                self.log.debug('Parameter name %s does not match regex %s', param_name, param_regex)
+                return tags
+        for part_number, part in enumerate(regex_parts):
+            if part == '*':
+                if wildcard_number >= len(wildcards):
+                    self.log.debug(
+                        'Found %s wildcards, which exceeds available wildcard tags %s', wildcard_number, wildcards
+                    )
                     return tags
-            for part_number, part in enumerate(regex_parts):
-                if part == '*':
-                    if wildcard_number >= len(wildcards):
-                        self.log.debug(
-                            'Found %s wildcards, which exceeds available wildcard tags %s', wildcard_number, wildcards
-                        )
-                        return tags
-                    tags.append(f'{wildcards[wildcard_number]}:{param_parts[part_number]}')
-                    wildcard_number += 1
+                tags.append(f'{wildcards[wildcard_number]}:{param_parts[part_number]}')
+                wildcard_number += 1
         return tags
 
     def _parse_stats(self, raw_stats: str) -> Dict[str, Dict[str, Union[int, str]]]:

lustre/datadog_checks/lustre/constants.py

@@ -24,8 +24,8 @@
 @dataclass(frozen=True)
 class LustreParam:
     regex: str
-    node_types: Tuple[str]
-    wildcards: Tuple[str] = ()
+    node_types: Tuple[str, ...]
+    wildcards: Tuple[str, ...] = ()
     prefix: str = ''
     fixture: str = ''
 

lustre/tests/test_unit.py

@@ -2,12 +2,14 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
+import logging
+
 import mock
 import pytest
 
 from datadog_checks.dev.utils import get_metadata_metrics
 from datadog_checks.lustre import LustreCheck
-from datadog_checks.lustre.constants import CURATED_PARAMS, DEFAULT_STATS, EXTRA_STATS, JOBSTATS_PARAMS
+from datadog_checks.lustre.constants import CURATED_PARAMS, DEFAULT_STATS, EXTRA_STATS, JOBSTATS_PARAMS, LustreParam
 
 from .metrics import (
     CLIENT_METRICS,
@@ -268,6 +270,10 @@ def test_extract_tags_from_param(mock_lustre_commands):
         )
         assert tags == ['device_name:lustre-MDT0000', 'nid:172.31.16.218@tcp']
 
+        # Test with malformed IP
+        tags = check._extract_tags_from_param('mdt.*.stats', 'mdt.172.malformed.16.218@tcp.stats', ('nid',))
+        assert tags == []
+
         # Test with no wildcards
         tags = check._extract_tags_from_param('mds.MDS.mdt.stats', 'mds.MDS.mdt.stats', ())
         assert tags == []
@@ -473,3 +479,126 @@ def test_run_command_exceptions():
     # Test subprocess exception
     with mock.patch('subprocess.run', side_effect=Exception("Command failed")):
         assert check._run_command('lctl', 'test') == ''
+
+
+def test_node_type_determination_logging(caplog, mock_lustre_commands):
+    """Test logging for node type determination errors."""
+    mapping = {
+        'lctl get_param -ny version': 'all_version.txt',
+        'lctl dl': 'client_dl_yaml.txt',
+    }
+
+    with mock_lustre_commands(mapping):
+        with caplog.at_level(logging.DEBUG):
+            with mock.patch.object(LustreCheck, '_update_devices', side_effect=Exception("Device error")):
+                LustreCheck('lustre', {}, [{}])
+
+        log_text = caplog.text
+        assert 'Failed to determine node type:' in log_text
+
+
+def test_filesystem_discovery_logging(caplog, mock_lustre_commands):
+    """Test logging for filesystem discovery."""
+    filesystem = 'testfilesystem'
+    mapping = {
+        'lctl get_param -ny version': 'all_version.txt',
+        'lctl dl': 'client_dl_yaml.txt',
+        'lctl list_param mdt.*.job_stats': f'mdt.{filesystem}-MDT0000.job_stats',
+    }
+
+    with mock_lustre_commands(mapping):
+        with caplog.at_level(logging.DEBUG):
+            LustreCheck('lustre', {}, [{'node_type': 'mds'}]).update()
+
+        log_text = caplog.text
+        assert 'Finding filesystems...' in log_text
+        assert f'Found filesystem(s): [\'{filesystem}\']' in log_text
+
+    mapping.update(
+        {
+            'lctl list_param mdt.*.job_stats': 'mdt.000.job_stats',
+        }
+    )
+
+    with mock_lustre_commands(mapping):
+        with caplog.at_level(logging.DEBUG):
+            LustreCheck('lustre', {}, [{'node_type': 'mds'}]).update()
+
+        log_text = caplog.text
+        assert 'Finding filesystems...' in log_text
+        assert 'Failed to find filesystems: Nothing matched regex' in log_text
+
+
+def test_lnet_error_logging(caplog, mock_lustre_commands):
+    """Test logging for LNET data retrieval errors."""
+    mapping = {
+        'lctl get_param -ny version': 'all_version.txt',
+        'lctl dl': 'client_dl_yaml.txt',
+        'lnetctl stats show': 'invalid yaml',
+    }
+
+    with mock_lustre_commands(mapping):
+        with caplog.at_level(logging.DEBUG):
+            check = LustreCheck('lustre', {}, [{}])
+            with mock.patch('yaml.safe_load', side_effect=Exception("YAML error")):
+                check._get_lnet_metrics('stats')
+
+        log_text = caplog.text
+        assert 'Could not get lnet stats, caught exception:' in log_text
+
+
+def test_parameter_filtering_logging(caplog, mock_lustre_commands):
+    """Test logging for parameter filtering based on node type and filesystem."""
+    wrong_filesystem_param = 'some.wrongfilesystem-MDT0000.param.mds'
+    mapping = {
+        'lctl get_param -ny version': 'all_version.txt',
+        'lctl dl': 'client_dl_yaml.txt',
+        'lctl list_param some.*.param.mds': wrong_filesystem_param,
+    }
+
+    with mock_lustre_commands(mapping):
+        with caplog.at_level(logging.DEBUG):
+            check = LustreCheck('lustre', {}, [{'node_type': 'mds'}])
+            params = {
+                LustreParam(
+                    regex="some.*.param.client",
+                    node_types=("client",),
+                    wildcards=("device_name",),
+                    prefix="",
+                    fixture="",
+                ),
+                LustreParam(
+                    regex="some.*.param.mds",
+                    node_types=("mds",),
+                    wildcards=("device_name",),
+                    prefix="",
+                    fixture="",
+                ),
+            }
+            check.submit_param_data(params, ['lustre'])
+
+        log_text = caplog.text
+        assert 'Skipping param some.*.param.client for node type mds' in log_text
+        assert f'Skipping param {wrong_filesystem_param} as it did not match any filesystem' in log_text
+
+
+def test_changelog_logging(caplog, mock_lustre_commands):
+    """Test logging for changelog operations."""
+    mapping = {
+        'lctl get_param -ny version': 'all_version.txt',
+        'lctl dl': 'client_dl_yaml.txt',
+        'lfs changelog': 'test_changelog',
+    }
+
+    with mock_lustre_commands(mapping):
+        with caplog.at_level(logging.DEBUG):
+            check = LustreCheck('lustre', {}, [{}])
+
+            # Test changelog cursor error
+            with mock.patch.object(check, 'get_log_cursor', side_effect=Exception("Cursor error")):
+                check._get_changelog('lustre-MDT0000', 100)
+
+        log_text = caplog.text
+        assert 'Could not retrieve log cursor, assuming initialization' in log_text
+        assert 'Fetching changelog from index 0 to 100' in log_text
+        assert 'Collecting changelogs for:' in log_text