torc 0.20.7

Workflow management system
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266

//! Shared data models for report commands.
//!
//! These structs define the JSON output format for `torc reports` commands
//! and are used by both the producers (in `commands/reports.rs`) and consumers
//! (in `commands/recover.rs` and elsewhere).

use serde::{Deserialize, Serialize};

/// Output of `torc reports check-resource-utilization`
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceUtilizationReport {
    pub workflow_id: i64,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub run_id: Option<i64>,
    pub total_results: usize,
    pub over_utilization_count: usize,
    pub violations: Vec<ResourceViolation>,
    /// Number of jobs with resource violations (only present when `--include-failed` is used)
    #[serde(default, skip_serializing_if = "is_zero")]
    pub resource_violations_count: usize,
    /// Jobs that exceeded resource allocations (only present when `--include-failed` is used)
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub resource_violations: Vec<ResourceViolationInfo>,
}

fn is_zero(n: &usize) -> bool {
    *n == 0
}

/// A resource utilization violation (job exceeded its specified resources)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceViolation {
    pub job_id: i64,
    pub job_name: String,
    pub resource_type: String,
    pub specified: String,
    pub peak_used: String,
    pub over_utilization: String,
}

/// Information about a job that exceeded resource allocation.
///
/// This includes jobs that failed during execution and completed jobs
/// that exceeded their configured memory, CPU, or runtime limits.
/// Used for proactive resource optimization and recovery diagnostics.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResourceViolationInfo {
    pub job_id: i64,
    pub job_name: String,
    pub return_code: i64,
    pub exec_time_minutes: f64,
    pub configured_memory: String,
    pub configured_runtime: String,
    pub configured_cpus: i64,

    /// Peak memory usage in bytes (if available from resource monitoring)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub peak_memory_bytes: Option<i64>,

    /// Human-readable peak memory (e.g., "1.5 GB")
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub peak_memory_formatted: Option<String>,

    /// Whether this job violated memory limits
    #[serde(default, skip_serializing_if = "is_false")]
    pub memory_violation: bool,

    /// Reason for OOM detection (e.g., "memory_exceeded", "sigkill_137")
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub oom_reason: Option<String>,

    /// How much memory was over-utilized (e.g., "+25.3%")
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub memory_over_utilization: Option<String>,

    /// Whether this job likely failed due to timeout
    #[serde(default, skip_serializing_if = "is_false")]
    pub likely_timeout: bool,

    /// Reason for timeout detection (e.g., "sigxcpu_152")
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub timeout_reason: Option<String>,

    /// Runtime utilization percentage (e.g., "95.2%")
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub runtime_utilization: Option<String>,

    /// Whether this job exceeded its CPU allocation
    #[serde(default, skip_serializing_if = "is_false")]
    pub likely_cpu_violation: bool,

    /// Peak CPU percentage used (e.g., 501.4%)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub peak_cpu_percent: Option<f64>,

    /// Whether this job exceeded its runtime allocation
    #[serde(default, skip_serializing_if = "is_false")]
    pub likely_runtime_violation: bool,
}

fn is_false(b: &bool) -> bool {
    !*b
}

/// Output of `torc reports results`
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ResultsReport {
    pub workflow_id: i64,
    pub workflow_name: String,
    pub workflow_user: String,
    pub all_runs: bool,
    pub total_results: usize,
    /// Job result records
    ///
    /// Note: This field is named `results` in JSON output (not `jobs`).
    pub results: Vec<JobResultRecord>,
}

/// A single job result record with log file paths
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JobResultRecord {
    pub job_id: i64,
    pub job_name: String,
    pub status: String,
    pub run_id: i64,
    pub return_code: i64,
    pub completion_time: String,
    pub exec_time_minutes: f64,
    pub compute_node_id: i64,

    /// Path to job stdout log
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub job_stdout: Option<String>,

    /// Path to job stderr log
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub job_stderr: Option<String>,

    /// Type of compute node ("local" or "slurm")
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub compute_node_type: Option<String>,

    /// Path to job runner log file
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub job_runner_log: Option<String>,

    /// Slurm job ID (only for slurm jobs)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub slurm_job_id: Option<String>,

    /// Path to Slurm stdout log (only for slurm jobs)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub slurm_stdout: Option<String>,

    /// Path to Slurm stderr log (only for slurm jobs)
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub slurm_stderr: Option<String>,
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_resource_utilization_report_serialization() {
        let report = ResourceUtilizationReport {
            workflow_id: 1,
            run_id: Some(2),
            total_results: 10,
            over_utilization_count: 2,
            violations: vec![ResourceViolation {
                job_id: 1,
                job_name: "test_job".to_string(),
                resource_type: "Memory".to_string(),
                specified: "1.0 GB".to_string(),
                peak_used: "1.5 GB".to_string(),
                over_utilization: "+50.0%".to_string(),
            }],
            resource_violations_count: 1,
            resource_violations: vec![ResourceViolationInfo {
                job_id: 2,
                job_name: "failed_job".to_string(),
                return_code: 137i64,
                exec_time_minutes: 5.5,
                configured_memory: "2g".to_string(),
                configured_runtime: "PT1H".to_string(),
                configured_cpus: 4i64,
                peak_memory_bytes: Some(3_000_000_000),
                peak_memory_formatted: Some("2.8 GB".to_string()),
                memory_violation: true,
                oom_reason: Some("sigkill_137".to_string()),
                memory_over_utilization: Some("+40.0%".to_string()),
                likely_timeout: false,
                timeout_reason: None,
                runtime_utilization: Some("9.2%".to_string()),
                likely_cpu_violation: false,
                peak_cpu_percent: None,
                likely_runtime_violation: false,
            }],
        };

        let json = serde_json::to_string(&report).unwrap();
        let parsed: ResourceUtilizationReport = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.workflow_id, 1);
        assert_eq!(parsed.resource_violations.len(), 1);
        assert!(parsed.resource_violations[0].memory_violation);
    }

    #[test]
    fn test_results_report_serialization() {
        let report = ResultsReport {
            workflow_id: 1,
            workflow_name: "test_workflow".to_string(),
            workflow_user: "testuser".to_string(),
            all_runs: false,
            total_results: 1,
            results: vec![JobResultRecord {
                job_id: 1,
                job_name: "job1".to_string(),
                status: "Completed".to_string(),
                run_id: 1,
                return_code: 0i64,
                completion_time: "2024-01-01T00:00:00Z".to_string(),
                exec_time_minutes: 10.5,
                compute_node_id: 1,
                job_stdout: Some("/path/to/stdout".to_string()),
                job_stderr: Some("/path/to/stderr".to_string()),
                compute_node_type: Some("slurm".to_string()),
                job_runner_log: Some("/path/to/runner.log".to_string()),
                slurm_job_id: Some("12345".to_string()),
                slurm_stdout: Some("/path/to/slurm.out".to_string()),
                slurm_stderr: Some("/path/to/slurm.err".to_string()),
            }],
        };

        let json = serde_json::to_string(&report).unwrap();
        let parsed: ResultsReport = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.workflow_id, 1);
        assert_eq!(parsed.results.len(), 1);
        assert_eq!(parsed.results[0].slurm_job_id, Some("12345".to_string()));
    }

    #[test]
    fn test_resource_violation_info_optional_fields() {
        // Test that optional fields can be omitted
        // Note: return_code and configured_cpus are i64 in the struct
        let json = r#"{
            "job_id": 1,
            "job_name": "test",
            "return_code": 1,
            "exec_time_minutes": 5.0,
            "configured_memory": "1g",
            "configured_runtime": "PT1H",
            "configured_cpus": 2
        }"#;

        let parsed: ResourceViolationInfo = serde_json::from_str(json).unwrap();
        assert_eq!(parsed.job_id, 1);
        assert!(!parsed.memory_violation);
        assert!(!parsed.likely_timeout);
        assert!(!parsed.likely_runtime_violation);
        assert!(parsed.peak_memory_bytes.is_none());
    }
}