teaql_provider_linux/
executor.rs

1use std::collections::HashMap;
2use std::time::SystemTime;
3
4use teaql_core::{EntityDescriptor, Record};
5use teaql_data_service::{
6    DataServiceCapabilities, DataServiceExecutor, DataServiceOperation, ExecutionMetadata,
7    MutationExecutor, MutationRequest, MutationResult, QueryExecutor, QueryRequest, QueryResult,
8    StreamChunk, StreamQueryExecutor, Transaction, TransactionExecutor,
9};
10use teaql_runtime::{InMemoryMetadataStore, MemoryDataService};
11
12use crate::collector::{Collector, ProcessCollector, SystemInfoCollector, ThreadCollector};
13use crate::error::LinuxProviderError;
14
15/// A `DataServiceExecutor` backed by Linux /proc collectors.
16///
17/// Routes queries to the appropriate collector by entity name, collects all records,
18/// then delegates in-memory query processing (filter, sort, project, aggregate) to
19/// `MemoryDataService`.
20pub struct LinuxDataServiceExecutor {
21    collectors: HashMap<String, Box<dyn Collector>>,
22}
23
24impl Default for LinuxDataServiceExecutor {
25    fn default() -> Self {
26        Self::new()
27    }
28}
29
30impl LinuxDataServiceExecutor {
31    /// Create a new executor with the default set of collectors.
32    pub fn new() -> Self {
33        let mut collectors: HashMap<String, Box<dyn Collector>> = HashMap::new();
34        collectors.insert(
35            "SystemInfo".to_owned(),
36            Box::new(SystemInfoCollector),
37        );
38        collectors.insert("Process".to_owned(), Box::new(ProcessCollector));
39        collectors.insert("Thread".to_owned(), Box::new(ThreadCollector));
40        Self { collectors }
41    }
42
43    /// Register an additional collector.
44    pub fn with_collector(mut self, collector: Box<dyn Collector>) -> Self {
45        self.collectors
46            .insert(collector.entity_name().to_owned(), collector);
47        self
48    }
49
50    fn collect_records(&self, entity: &str) -> Result<Vec<Record>, LinuxProviderError> {
51        let collector = self
52            .collectors
53            .get(entity)
54            .ok_or_else(|| LinuxProviderError::UnknownEntity(entity.to_owned()))?;
55        collector.collect_all()
56    }
57}
58
59impl DataServiceExecutor for LinuxDataServiceExecutor {
60    type Error = LinuxProviderError;
61
62    fn capabilities(&self) -> DataServiceCapabilities {
63        DataServiceCapabilities {
64            query: true,
65            mutation: false,
66            transaction: false,
67            schema: false,
68            id_generation: false,
69            batch_mutation: false,
70            returning: false,
71        }
72    }
73}
74
75impl QueryExecutor for LinuxDataServiceExecutor {
76    async fn query(&self, request: QueryRequest) -> Result<QueryResult, Self::Error> {
77        let started_at = SystemTime::now();
78        let entity = &request.query.entity;
79
80        // Collect raw records from the matching collector.
81        let rows = self.collect_records(entity)?;
82
83        // Build a MemoryDataService with the entity registered so fetch_all succeeds.
84        let metadata = InMemoryMetadataStore::new()
85            .with_entity(EntityDescriptor::new(entity));
86        let mut mem = MemoryDataService::new(metadata);
87        mem.seed(entity.clone(), rows);
88
89        let result_rows = mem
90            .fetch_all(&request.query)
91            .map_err(|e| LinuxProviderError::ProcFs(e.to_string()))?;
92
93        let ended_at = SystemTime::now();
94        let count = result_rows.len();
95
96        Ok(QueryResult {
97            rows: result_rows,
98            metadata: ExecutionMetadata {
99                backend: "linux-proc".to_owned(),
100                operation: DataServiceOperation::Query,
101                started_at,
102                ended_at,
103                affected_rows: None,
104                result_count: Some(count),
105                trace_chain: request.trace_chain,
106                comment: request.comment,
107                backend_request_id: None,
108                debug_query: None,
109            },
110        })
111    }
112}
113
114impl MutationExecutor for LinuxDataServiceExecutor {
115    async fn mutate(
116        &self,
117        _request: MutationRequest,
118    ) -> Result<MutationResult, Self::Error> {
119        Err(LinuxProviderError::ProcFs("Linux provider is read-only".to_owned()))
120    }
121}
122
123impl Transaction for LinuxDataServiceExecutor {
124    type Error = LinuxProviderError;
125
126    async fn commit(self) -> Result<(), Self::Error> {
127        Ok(())
128    }
129    async fn rollback(self) -> Result<(), Self::Error> {
130        Ok(())
131    }
132}
133
134impl TransactionExecutor for LinuxDataServiceExecutor {
135    type Tx<'a> = LinuxDataServiceExecutor;
136
137    async fn begin(&self) -> Result<Self::Tx<'_>, Self::Error> {
138        Err(LinuxProviderError::ProcFs("Linux provider does not support transactions".to_owned()))
139    }
140}
141
142impl StreamQueryExecutor for LinuxDataServiceExecutor {
143    async fn query_stream(&self, _request: QueryRequest, _chunk_size: usize) -> Result<Vec<StreamChunk>, Self::Error> {
144        Err(LinuxProviderError::ProcFs("Linux provider does not support streaming".to_owned()))
145    }
146}
teaql_provider_linux/executor.rs

teaql_provider_linux/
executor.rs
