player_plugin_loader/
benchmark.rs

1use super::*;
2use std::collections::BTreeMap;
3
4use player_plugin::{MAX_PLUGIN_DIAGNOSTICS, MAX_PLUGIN_MEASUREMENTS, PluginReference};
5
6pub struct BenchmarkSinkPluginSession {
7    sinks: Vec<Arc<dyn BenchmarkSink>>,
8    references: Vec<PluginReference>,
9}
10
11impl std::fmt::Debug for BenchmarkSinkPluginSession {
12    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
13        f.debug_struct("BenchmarkSinkPluginSession")
14            .field("sink_count", &self.sinks.len())
15            .field("references", &self.references)
16            .finish()
17    }
18}
19
20impl BenchmarkSinkPluginSession {
21    pub fn from_registry(
22        registry: &PluginRegistry,
23        references: impl IntoIterator<Item = PluginReference>,
24    ) -> Result<Self, PluginSelectionError> {
25        let mut sinks = Vec::new();
26        let mut resolved_references = Vec::new();
27        for reference in references {
28            let resolved = registry.resolve_benchmark_sink(&reference)?;
29            resolved_references.push(resolved.reference().clone());
30            sinks.push(resolved.capability());
31        }
32        Ok(Self {
33            sinks,
34            references: resolved_references,
35        })
36    }
37
38    pub fn is_empty(&self) -> bool {
39        self.sinks.is_empty()
40    }
41
42    pub fn references(&self) -> &[PluginReference] {
43        &self.references
44    }
45
46    pub fn on_event_batch_json(
47        &self,
48        batch_json: &str,
49    ) -> Result<BenchmarkSinkReport, BenchmarkSinkError> {
50        let batch = serde_json::from_str::<BenchmarkEventBatch>(batch_json).map_err(|error| {
51            BenchmarkSinkError::PayloadCodec(format!(
52                "decode benchmark event batch payload failed: {error}"
53            ))
54        })?;
55        Ok(self.on_event_batch(&batch))
56    }
57
58    pub fn on_event_batch_report_json(
59        &self,
60        batch_json: &str,
61    ) -> Result<String, BenchmarkSinkError> {
62        serde_json::to_string(&self.on_event_batch_json(batch_json)?).map_err(|error| {
63            BenchmarkSinkError::PayloadCodec(format!(
64                "encode benchmark sink status failed: {error}"
65            ))
66        })
67    }
68
69    pub fn on_event_batch(&self, batch: &BenchmarkEventBatch) -> BenchmarkSinkReport {
70        let mut report = BenchmarkSinkReport::default();
71        if let Err(error) = batch.validate() {
72            report.dropped_events = batch.events.len() as u64;
73            report
74                .diagnostics
75                .push(sink_error_diagnostic("host", &error));
76            return report;
77        }
78        for sink in &self.sinks {
79            match sink.on_event_batch(batch) {
80                Ok(status) => {
81                    report.accepted_events += status.accepted_events;
82                }
83                Err(error) => {
84                    report.dropped_events += batch.events.len() as u64;
85                    push_diagnostic(&mut report, sink_error_diagnostic(sink.name(), &error));
86                }
87            }
88        }
89        report
90    }
91
92    pub fn flush(&self) -> BenchmarkSinkReport {
93        let mut report = BenchmarkSinkReport::default();
94        for sink in &self.sinks {
95            match sink.flush() {
96                Ok(sink_report) => {
97                    merge_report(&mut report, sink_report);
98                }
99                Err(error) => {
100                    push_diagnostic(&mut report, sink_error_diagnostic(sink.name(), &error));
101                }
102            }
103        }
104        report
105    }
106
107    pub fn flush_json(&self) -> Result<String, BenchmarkSinkError> {
108        serde_json::to_string(&self.flush()).map_err(|error| {
109            BenchmarkSinkError::PayloadCodec(format!(
110                "encode benchmark sink report failed: {error}"
111            ))
112        })
113    }
114}
115
116fn merge_report(report: &mut BenchmarkSinkReport, incoming: BenchmarkSinkReport) {
117    report.accepted_events = report
118        .accepted_events
119        .saturating_add(incoming.accepted_events);
120    report.dropped_events = report
121        .dropped_events
122        .saturating_add(incoming.dropped_events);
123    extend_bounded(
124        &mut report.measurements,
125        incoming.measurements,
126        MAX_PLUGIN_MEASUREMENTS,
127    );
128    extend_bounded(
129        &mut report.threshold_violations,
130        incoming.threshold_violations,
131        MAX_PLUGIN_MEASUREMENTS,
132    );
133    for diagnostic in incoming.diagnostics {
134        push_diagnostic(report, diagnostic);
135    }
136}
137
138fn extend_bounded<T>(target: &mut Vec<T>, incoming: Vec<T>, limit: usize) {
139    let remaining = limit.saturating_sub(target.len());
140    target.extend(incoming.into_iter().take(remaining));
141}
142
143fn push_diagnostic(report: &mut BenchmarkSinkReport, diagnostic: PluginDiagnostic) {
144    if report.diagnostics.len() < MAX_PLUGIN_DIAGNOSTICS {
145        report.diagnostics.push(diagnostic);
146    }
147}
148
149fn sink_error_diagnostic(name: &str, error: &BenchmarkSinkError) -> PluginDiagnostic {
150    let code = match error {
151        BenchmarkSinkError::PayloadCodec(_) => "benchmark.payload_codec",
152        BenchmarkSinkError::AbiViolation(_) => "benchmark.abi_violation",
153        BenchmarkSinkError::SinkFailed(_) => "benchmark.sink_failed",
154        BenchmarkSinkError::ProtocolViolation(_) => "benchmark.protocol_violation",
155    };
156    PluginDiagnostic {
157        code: code.to_owned(),
158        severity: PluginDiagnosticSeverity::Error,
159        message: "benchmark sink operation failed".to_owned(),
160        attributes: BTreeMap::from([("sink".to_owned(), name.to_owned())]),
161    }
162}
player_plugin_loader/benchmark.rs

player_plugin_loader/
benchmark.rs
