Merge pull request #3708 from abhishek-sa1/pub/k8s_telemetry

Rename ldms store theme to slurm-cluster, update ldms port configuration, adding ome kafka topic creation

Author: abhishek-sa1

common/library/module_utils/input_validation/schema/telemetry_config.json

@@ -20,18 +20,24 @@
     },
     "ldms_agg_port": {
       "type": "integer",
-      "const": 6001,
-      "description": "LDMS aggregator starting port (FIXED - cannot be changed)"
+      "minimum": 6001,
+      "maximum": 6100,
+      "default": 6001,
+      "description": "LDMS Aggregator port on service k8s cluster. Valid range: 6001-6100. Default: 6001"
     },
     "ldms_store_port": {
       "type": "integer",
-      "const": 6001,
-      "description": "LDMS store daemon port (FIXED - cannot be changed)"
+      "minimum": 6001,
+      "maximum": 6100,
+      "default": 6001,
+      "description": "LDMS store daemon port on service k8s cluster. Valid range: 6001-6100. Can be the same as ldms_agg_port (isolated by pod). Default: 6001"
     },
     "ldms_sampler_port": {
       "type": "integer",
-      "const": 10001,
-      "description": "LDMS sampler port on compute nodes (FIXED - cannot be changed)"
+      "minimum": 10001,
+      "maximum": 10100,
+      "default": 10001,
+      "description": "LDMS sampler port on compute nodes. Valid range: 10001-10100. Default: 10001"
     },
     "ldms_sampler_configurations": {
     "type": "array",
@@ -49,8 +55,9 @@
         },
         "activation_parameters": {
           "type": "string",
-          "description": "Activation parameters as a string (e.g., 'interval=1000000 offset=0')",
-          "pattern": "^(?=.*\\binterval=[1-9][0-9]*\\b)(?:.*\\boffset=[0-9]+\\b)?$"
+          "description": "Activation parameters as a string (e.g., 'interval=1000000 offset=0'). Format: 'interval=<microseconds>' with optional 'offset=<microseconds>' separated by space.",
+          "pattern": "^interval=[1-9][0-9]*(?:\\s+offset=[0-9]+)?$",
+          "errorMessage": "Must be in format 'interval=<non-zero-number>' or 'interval=<non-zero-number> offset=<number>'. Example: 'interval=1000000' or 'interval=1000000 offset=0'"
         }
       },
       "required": ["plugin_name", "activation_parameters"],
@@ -114,8 +121,43 @@
           "type": "integer"
         },
         "topic_partitions": {
-          "type": "integer",
-          "minimum": 1
+          "type": "array",
+          "minItems": 1,
+          "maxItems": 3,
+          "items": {
+            "type": "object",
+            "properties": {
+              "name": {
+                "type": "string",
+                "enum": ["idrac", "ldms", "ome"],
+                "description": "CONSTANT: Fixed topic names that cannot be changed. Only 'idrac', 'ldms', and 'ome' are allowed.",
+                "errorMessage": {
+                  "enum": "Invalid topic name. Only 'idrac', 'ldms', and 'ome' are allowed as Kafka topic names. Custom topic names are not supported."
+                }
+              },
+              "partitions": {
+                "type": "integer",
+                "minimum": 1,
+                "maximum": 100,
+                "description": "Number of partitions for the topic (1-100). This is the only configurable parameter."
+              }
+            },
+            "required": ["name", "partitions"],
+            "additionalProperties": false,
+            "errorMessage": {
+              "required": {
+                "name": "Topic 'name' is required and must be one of: 'idrac', 'ldms', 'ome'",
+                "partitions": "Topic 'partitions' is required and must be between 1-100"
+              }
+            }
+          },
+          "uniqueItems": true,
+          "description": "IMPORTANT: At least one Kafka topic must be defined. Topic names 'idrac', 'ldms', and 'ome' are CONSTANTS. 'idrac' is required if idrac_telemetry_support is true and kafka is in idrac_telemetry_collection_type. 'ldms' is required if LDMS software is configured in software_config.json (automatic detection). 'ome' is optional. Only partition counts can be changed.",
+          "errorMessage": {
+            "minItems": "At least 1 Kafka topic must be defined. Configure based on enabled features.",
+            "maxItems": "Maximum 3 topics allowed: 'idrac', 'ldms', and 'ome'",
+            "uniqueItems": "Each topic (idrac, ldms, ome) must appear only once"
+          }
         }
       },
       "required": [

common/library/module_utils/input_validation/validation_flows/common_validation.py

@@ -1137,6 +1137,124 @@ def validate_telemetry_config(
             en_us_validation_msg.TELEMETRY_SERVICE_CLUSTER_ENTRY_MISSING_ROLES_CONFIG_MSG
             )    
         )
+    
+    # Determine LDMS support from software_config.json
+    # software_config.json is in the same directory as telemetry_config.yml
+    ldms_support_from_software_config = False
+    input_dir = os.path.dirname(input_file_path)
+    software_config_file_path = os.path.join(input_dir, "software_config.json")
+    
+    logger.info(f"Checking for LDMS software in: {software_config_file_path}")
+    
+    if os.path.exists(software_config_file_path):
+        try:
+            with open(software_config_file_path, 'r') as f:
+                software_config = json.load(f)
+                softwares = software_config.get("softwares", [])
+                ldms_support_from_software_config = any(
+                    software.get("name") == "ldms" for software in softwares
+                )
+                logger.info(f"LDMS software detected in software_config.json: {ldms_support_from_software_config}")
+                if ldms_support_from_software_config:
+                    logger.info("LDMS software found - 'ldms' topic will be required in kafka_configurations.topic_partitions")
+        except (json.JSONDecodeError, IOError) as e:
+            logger.warn(f"Could not load software_config.json: {e}")
+    else:
+        logger.info(f"software_config.json not found at: {software_config_file_path}")
+    
+    # Validate topic_partitions configuration
+    kafka_config = data.get("kafka_configurations", {})
+    topic_partitions = kafka_config.get("topic_partitions", [])
+    idrac_telemetry_collection_type = data.get("idrac_telemetry_collection_type", "")
+    
+    # Check if LDMS software is configured but kafka_configurations is missing entirely
+    if ldms_support_from_software_config and not kafka_config:
+        errors.append(create_error_msg(
+            "kafka_configurations",
+            "not defined",
+            "LDMS software is configured in software_config.json, but kafka_configurations section is missing in telemetry_config.yml. "
+            "Please define kafka_configurations with at least the 'ldms' topic in topic_partitions."
+        ))
+    
+    # Check if LDMS software is configured but no topics are defined
+    if ldms_support_from_software_config and kafka_config and not topic_partitions:
+        errors.append(create_error_msg(
+            "kafka_configurations.topic_partitions",
+            "not defined",
+            "LDMS software is configured in software_config.json, but kafka_configurations.topic_partitions is not defined. "
+            "Please define at least the 'ldms' topic in topic_partitions."
+        ))
+    
+    if topic_partitions:
+        # Ensure at least one topic is defined
+        if len(topic_partitions) < 1:
+            errors.append(create_error_msg(
+                "kafka_configurations.topic_partitions",
+                "is empty",
+                "At least one Kafka topic must be defined"
+            ))
+        
+        # Collect topic names and validate each one
+        topic_names = []
+        allowed_topics = {"idrac", "ldms", "ome"}
+        
+        for idx, topic in enumerate(topic_partitions):
+            if "name" not in topic:
+                errors.append(create_error_msg(
+                    f"kafka_configurations.topic_partitions[{idx}]",
+                    "missing 'name' field",
+                    "Each topic must have a 'name' field"
+                ))
+                continue
+            
+            topic_name = topic.get("name")
+            topic_names.append(topic_name)
+            
+            # Validate each topic name individually
+            if topic_name not in allowed_topics:
+                errors.append(create_error_msg(
+                    f"kafka_configurations.topic_partitions[{idx}].name",
+                    topic_name,
+                    f"Invalid topic name '{topic_name}'. Only 'idrac', 'ldms', and 'ome' are allowed as Kafka topic names. Custom topic names are not supported."
+                ))
+        
+        present_topics = set(topic_names)
+        
+        # Debug logging
+        logger.info(f"Telemetry validation - Present topics: {present_topics}")
+        logger.info(f"Telemetry validation - Allowed topics: {allowed_topics}")
+        
+        # Validate required topics based on feature flags
+        # If iDRAC telemetry is enabled with Kafka, idrac topic is required
+        if idrac_telemetry_support and 'kafka' in idrac_telemetry_collection_type.split(','):
+            if 'idrac' not in present_topics:
+                errors.append(create_error_msg(
+                    "kafka_configurations.topic_partitions",
+                    "missing 'idrac' topic",
+                    "idrac topic is required when idrac_telemetry_support is true and 'kafka' is in idrac_telemetry_collection_type"
+                ))
+        
+        # If LDMS software is configured in software_config.json, ldms topic is required
+        logger.info(f"Checking LDMS topic requirement - ldms_support_from_software_config: {ldms_support_from_software_config}")
+        if ldms_support_from_software_config and 'ldms' not in present_topics:
+            logger.error(f"LDMS topic validation FAILED - 'ldms' topic is missing from present_topics: {present_topics}")
+            errors.append(create_error_msg(
+                "kafka_configurations.topic_partitions",
+                "missing 'ldms' topic",
+                "ldms topic is required when LDMS software is configured in software_config.json"
+            ))
+        elif ldms_support_from_software_config:
+            logger.info(f"LDMS topic validation PASSED - 'ldms' found in present_topics: {present_topics}")
+        
+        # Check for duplicate topic names
+        if len(topic_names) != len(set(topic_names)):
+            duplicates = [name for name in topic_names if topic_names.count(name) > 1]
+            errors.append(create_error_msg(
+                "kafka_configurations.topic_partitions",
+                f"duplicate topics: {', '.join(set(duplicates))}",
+                "Each topic must be defined only once"
+            ))
+    
     return errors
 
 def validate_additional_software(

discovery/discovery.yml

@@ -36,8 +36,7 @@
     - name: Set dynamic run tags including 'provision'
       when: not config_file_status | default(false) | bool
       ansible.builtin.set_fact:
-        omnia_run_tags: "{{ ((ansible_run_tags | default([]))
-         + ['provision', 'slurm', 'slurm_custom', 'security', 'service_k8s', 'csi_driver_powerscale']) | unique }}" # noqa: yaml[line-length]
+        omnia_run_tags: "{{ (ansible_run_tags | default([]) + ['provision']) + ['slurm'] + ['slurm_custom'] + ['security'] + ['service_k8s'] + ['csi_driver_powerscale'] + ['telemetry'] | unique }}" # noqa: yaml[line-length]
         cacheable: true
 
 - name: Invoke validate_config.yml to perform L1 and L2 validations

discovery/roles/configure_ochami/templates/ldms/ldms_conf.sh.j2

@@ -14,12 +14,22 @@ else
     echo "Environment file /opt/ovis-ldms/etc/profile.d/set-ovis-variables.sh not found. Continuing..." | tee -a "$LOG_FILE"
 fi
 
-# --- Verify environment file existence ---
-echo "Checking for LDMS sampler environment file..." | tee -a "$LOG_FILE"
+# --- Copy LDMS sampler environment file from NFS share ---
+echo "Copying LDMS sampler environment file from NFS share..." | tee -a "$LOG_FILE"
+if [ -f {{ client_mount_path }}/ldms/samplers/ldmsd.sampler.env ]; then
+    sudo cp {{ client_mount_path }}/ldms/samplers/ldmsd.sampler.env /opt/ovis-ldms/etc/ldms/ldmsd.sampler.env
+    echo "✓ Copied ldmsd.sampler.env" | tee -a "$LOG_FILE"
+else
+    echo "Warning: ldmsd.sampler.env not found in NFS share" | tee -a "$LOG_FILE"
+fi
+
+# --- Source environment file to get port ---
 if [ -f /opt/ovis-ldms/etc/ldms/ldmsd.sampler.env ]; then
-    echo "Found ldmsd.sampler.env" | tee -a "$LOG_FILE"
+    source /opt/ovis-ldms/etc/ldms/ldmsd.sampler.env
+    echo "✓ Sourced ldmsd.sampler.env (Port: $LDMSD_PORT)" | tee -a "$LOG_FILE"
 else
-    echo "Warning: ldmsd.sampler.env not found" | tee -a "$LOG_FILE"
+    LDMSD_PORT=10001
+    echo "Warning: Using default port $LDMSD_PORT" | tee -a "$LOG_FILE"
 fi
 
 # --- Configure and enable LDMS service ---
@@ -36,13 +46,13 @@ fi
 
 # Configure firewall safely
 
-echo "Configuring firewall for LDMS port 10001..." | tee -a "$LOG_FILE"
-sudo firewall-cmd --permanent --add-port=10001/tcp
+echo "Configuring firewall for LDMS port $LDMSD_PORT..." | tee -a "$LOG_FILE"
+sudo firewall-cmd --permanent --add-port=$LDMSD_PORT/tcp
 sudo firewall-cmd --reload
 
 # --- Verify LDMS connection and metrics ---
 echo "Verifying LDMS connection and metrics..." | tee -a "$LOG_FILE"
-/opt/ovis-ldms/sbin/ldms_ls -a ovis -A conf=/opt/ovis-ldms/etc/ldms/ldmsauth.conf -p 10001 -h localhost | tee -a "$LOG_FILE"
-/opt/ovis-ldms/sbin/ldms_ls -l -a ovis -A conf=/opt/ovis-ldms/etc/ldms/ldmsauth.conf -p 10001 -h localhost > /tmp/metrics
+/opt/ovis-ldms/sbin/ldms_ls -a ovis -A conf=/opt/ovis-ldms/etc/ldms/ldmsauth.conf -p $LDMSD_PORT -h localhost | tee -a "$LOG_FILE"
+/opt/ovis-ldms/sbin/ldms_ls -l -a ovis -A conf=/opt/ovis-ldms/etc/ldms/ldmsauth.conf -p $LDMSD_PORT -h localhost > /tmp/metrics
 
 echo "===== LDMS setup completed =====" | tee -a "$LOG_FILE"

discovery/roles/telemetry/files/nersc-ldms-aggr/README.md

@@ -61,7 +61,7 @@ For csm: make_host_map.csm.py:
 3. Write to out_dir/host_map.json
 
 For dell: make_host_map.dell.py:
-1. Copy host_map.r7525.json to out_dir/host_map.json
+1. Copy host_map.slurm-cluster.json to out_dir/host_map.json
 
 Create ldms config and prepare chart (nersc_ldms_make_ldms_config.py)
 1. Create `ldmsd` config and environment variable files for each `ldmsd` to distribute the producers across daemons and enable daemons to find each other.
@@ -256,11 +256,11 @@ helm install -n telemetry nersc-ldms-aggr nersc-ldms-aggr --values values.yaml
 
 # Let it startup
 kubectl  -n telemetry top pods --containers |grep ldms
-nersc-ldms-aggr-0                           r7525-0                    2m           16Mi            
+nersc-ldms-aggr-0                           slurm-cluster-0            2m           16Mi            
 nersc-ldms-exporter-0                       exporter                   1m           27Mi            
 nersc-ldms-exporter-1                       exporter                   1m           24Mi            
 nersc-ldms-exporter-2                       exporter                   1m           24Mi            
-nersc-ldms-store-r7525-0                    store                      2m           10Mi            
+nersc-ldms-store-slurm-cluster-0            store                      2m           10Mi            
 nersc-ldms-stream-0                         stream                     1m           13Mi  
 
 ```

discovery/roles/telemetry/files/nersc-ldms-aggr/host_map.r7525.json renamed to discovery/roles/telemetry/files/nersc-ldms-aggr/host_map.slurm-cluster.json

@@ -1,4 +1,4 @@
 #!/bin/bash
 
 if [ ! -d "$
-cp host_map.r7525.json out_dir 
+cp host_map.slurm-cluster.json out_dir 

discovery/roles/telemetry/files/nersc-ldms-aggr/make_host_map.bash

@@ -47,7 +47,7 @@ def main(self):
         os.makedirs(self.out_dir, exist_ok=True)
 
         # PLACE HOLDER: just copy the example file for now
-        shutil.copy("host_map.r7525.json", self.out_dir)
+        shutil.copy("host_map.slurm-cluster.json", self.out_dir)
 
 def main():
     parser = argparse.ArgumentParser()

discovery/roles/telemetry/files/nersc-ldms-aggr/make_host_map.dell.py

@@ -166,6 +166,7 @@ def make_agg_configs(self):  # pylint: disable=too-many-locals
     def make_store_configs(self):  # pylint: disable=too-many-locals
         """Generate store configuration files."""
         logging.info("Make Store Configs")
+        
         for ldmsd_name, ldmsd_conf in self.config['node_types'].items():
             # grab auth data
             auth_type = ldmsd_conf.get('auth_type')
@@ -537,7 +538,7 @@ def make_config_store(self, ldmsd_name, ldmsd_agg_name, ldmsd_agg_port,  # pylin
         cfg.append("prdcr_start_regex regex=.*")
         cfg.extend([
             "# Store in kafka - port 9092 (plaintext, no TLS, no auth)",
-            "# NOTE: store_avro_kafka plugin cannot configure TLS/SSL",
+            "# NOTE: store_avro_kafka plugin using plaintext for anonymous access",
             "# Using plaintext listener on port 9092 (internal cluster only)",
             "load name=store_avro_kafka",
             "config name=store_avro_kafka encoding=json topic=ldms",