krabka_admin_ui/
admin.rs

1use std::collections::BTreeMap;
2
3use krabka_client_admin::{
4    AclEntry, AclEntryFilter, AdminClient, AdminError, BrokerResult, ConfigResource,
5    CreateAclOutcome, CreatePartitionsOutcome, CreateTopicOutcome, DeleteAclFilterOutcome,
6    DeleteTopicOutcome, DescribeClusterOptions, KafkaError, LogDirInfo, ScramUserOutcome,
7    TopicMetadata, TopicPartitionReplica, UserQuotaConfig, UserScramCredential,
8    UserScramCredentials, groups::ListGroupsOptions,
9};
10
11use crate::dto::{
12    AclRow, GroupRow, KafkaErrorDto, LogDirRow, QuotaRow, ResourceOutcome, TopicRow, UserRow,
13};
14
15pub struct AdminFacade {
16    client: AdminClient,
17}
18
19impl AdminFacade {
20    #[must_use]
21    pub const fn new(client: AdminClient) -> Self {
22        Self { client }
23    }
24
25    #[cfg_attr(test, mutants::skip)]
26    pub fn client_mut(&mut self) -> &mut AdminClient {
27        &mut self.client
28    }
29
30    #[cfg_attr(test, mutants::skip)]
31    /// # Errors
32    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
33    pub async fn topics(&mut self) -> Result<Vec<TopicRow>, AdminError> {
34        let metadata = self.client.metadata(&[]).await?;
35
36        Ok(topic_rows(metadata))
37    }
38
39    #[cfg_attr(test, mutants::skip)]
40    /// # Errors
41    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
42    pub async fn groups(&mut self) -> Result<Vec<GroupRow>, AdminError> {
43        let groups = self
44            .client
45            .list_groups(&ListGroupsOptions::default())
46            .await?
47            .valid
48            .into_iter()
49            .map(|listing| listing.group_id)
50            .collect();
51
52        Ok(group_rows(groups))
53    }
54
55    #[cfg_attr(test, mutants::skip)]
56    /// # Errors
57    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
58    pub async fn log_dirs(&mut self) -> Result<Vec<LogDirRow>, AdminError> {
59        let brokers = self
60            .client
61            .describe_cluster(DescribeClusterOptions::default())
62            .await?
63            .nodes
64            .into_iter()
65            .map(|node| node.id)
66            .collect::<Vec<_>>();
67        let log_dirs = self.client.describe_log_dirs(&brokers, None).await;
68
69        Ok(log_dir_rows(log_dirs))
70    }
71
72    #[cfg_attr(test, mutants::skip)]
73    /// # Errors
74    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
75    pub async fn acls(&mut self) -> Result<Vec<AclRow>, AdminError> {
76        Ok(acl_rows(self.acl_entries().await?))
77    }
78
79    #[cfg_attr(test, mutants::skip)]
80    /// Every ACL the broker reports, unformatted.
81    ///
82    /// Login derives the operator's capabilities from these entries, so it
83    /// needs the typed entry rather than the display row.
84    ///
85    /// # Errors
86    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
87    pub async fn acl_entries(&mut self) -> Result<Vec<AclEntry>, AdminError> {
88        self.client.describe_acls(&AclEntryFilter::default()).await
89    }
90
91    #[cfg_attr(test, mutants::skip)]
92    /// # Errors
93    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
94    pub async fn quotas_for_user(&mut self, username: &str) -> Result<Vec<QuotaRow>, AdminError> {
95        let quotas = self.client.describe_user_quotas(username).await?;
96
97        Ok(quota_rows(username, quotas))
98    }
99
100    #[cfg_attr(test, mutants::skip)]
101    /// # Errors
102    /// Returns an error when the request is invalid, authentication or session validation fails, or the broker admin operation reports a failure.
103    pub async fn users(&mut self) -> Result<Vec<UserRow>, AdminError> {
104        let users = self.client.describe_user_scram_credentials(None).await?;
105
106        Ok(user_rows(users))
107    }
108}
109
110#[must_use]
111pub fn topic_rows(metadata: TopicMetadata) -> Vec<TopicRow> {
112    metadata
113        .topics
114        .into_iter()
115        .map(|topic| TopicRow {
116            name: topic.name,
117            topic_id: topic.topic_id.map(|id| id.to_string()),
118            partition_count: topic.partition_count,
119            replication_factor: topic.replication_factor,
120            error: topic.error.as_ref().map(KafkaErrorDto::from),
121        })
122        .collect()
123}
124
125#[must_use]
126pub fn group_rows(group_ids: Vec<String>) -> Vec<GroupRow> {
127    group_ids
128        .into_iter()
129        .map(|group_id| GroupRow { group_id })
130        .collect()
131}
132
133#[must_use]
134pub fn acl_rows(acls: Vec<AclEntry>) -> Vec<AclRow> {
135    acls.into_iter()
136        .map(|acl| AclRow {
137            resource: format!("{:?}:{}", acl.resource_type, acl.resource_name),
138            pattern_type: format!("{:?}", acl.pattern_type),
139            principal: acl.principal,
140            host: acl.host,
141            operation: format!("{:?}", acl.operation),
142            permission: format!("{:?}", acl.permission_type),
143        })
144        .collect()
145}
146
147#[must_use]
148pub fn quota_rows(username: &str, quotas: UserQuotaConfig) -> Vec<QuotaRow> {
149    quotas
150        .into_iter()
151        .map(|(quota_type, value)| QuotaRow {
152            entity: username.to_string(),
153            quota_type,
154            value: format_quota_value(value),
155        })
156        .collect()
157}
158
159#[must_use]
160pub fn user_rows(users: Vec<UserScramCredentials>) -> Vec<UserRow> {
161    users
162        .into_iter()
163        .map(|user| UserRow {
164            username: user.username,
165            principal: scram_credential_summary(&user.credentials),
166        })
167        .collect()
168}
169
170fn scram_credential_summary(credentials: &[UserScramCredential]) -> String {
171    credentials
172        .iter()
173        .map(|credential| credential.mechanism.as_str())
174        .collect::<Vec<_>>()
175        .join(", ")
176}
177
178#[must_use]
179pub fn resource_outcome_rows<T>(outcomes: Vec<T>) -> Vec<ResourceOutcome>
180where
181    T: IntoResourceOutcomeDto,
182{
183    outcomes
184        .into_iter()
185        .map(IntoResourceOutcomeDto::into_resource_outcome)
186        .collect()
187}
188
189#[must_use]
190pub fn quota_mutation_outcome(
191    username: &str,
192    quota_type: &str,
193    error: Option<KafkaError>,
194) -> ResourceOutcome {
195    kafka_error_outcome(format!("{username}:{quota_type}"), error)
196}
197
198/// One row for each partition of each log dir of each broker. A broker that
199/// failed, and a log dir that failed with no partitions, each give one row
200/// that carries the error.
201#[must_use]
202pub fn log_dir_rows(log_dirs: BTreeMap<i32, BrokerResult<Vec<LogDirInfo>>>) -> Vec<LogDirRow> {
203    log_dirs
204        .into_iter()
205        .flat_map(|(broker_id, result)| match result {
206            Ok(dirs) => dirs
207                .into_iter()
208                .flat_map(|dir| log_dir_partition_rows(broker_id, dir))
209                .collect::<Vec<_>>(),
210            Err(error) => vec![LogDirRow {
211                broker_id,
212                log_dir: String::new(),
213                topic: String::new(),
214                partition: -1,
215                partition_size: 0,
216                offset_lag: 0,
217                is_future_key: false,
218                error: Some(KafkaErrorDto::from(&error)),
219            }],
220        })
221        .collect()
222}
223
224fn log_dir_partition_rows(broker_id: i32, log_dir: LogDirInfo) -> Vec<LogDirRow> {
225    let log_dir_name = log_dir.log_dir;
226    let log_dir_error = log_dir.error.as_ref().map(KafkaErrorDto::from);
227
228    if log_dir.topics.is_empty() && log_dir_error.is_some() {
229        return vec![LogDirRow {
230            broker_id,
231            log_dir: log_dir_name,
232            topic: String::new(),
233            partition: -1,
234            partition_size: 0,
235            offset_lag: 0,
236            is_future_key: false,
237            error: log_dir_error,
238        }];
239    }
240
241    log_dir
242        .topics
243        .into_iter()
244        .flat_map(move |topic| {
245            let log_dir_name = log_dir_name.clone();
246            let log_dir_error = log_dir_error.clone();
247
248            topic
249                .partitions
250                .into_iter()
251                .map(move |partition| LogDirRow {
252                    broker_id,
253                    log_dir: log_dir_name.clone(),
254                    topic: topic.name.clone(),
255                    partition: partition.partition_index,
256                    partition_size: partition.partition_size,
257                    offset_lag: partition.offset_lag,
258                    is_future_key: partition.is_future_key,
259                    error: log_dir_error.clone(),
260                })
261        })
262        .collect()
263}
264
265fn format_quota_value(value: f64) -> String {
266    if value.fract() == 0.0 {
267        return format!("{value:.0}");
268    }
269
270    value.to_string()
271}
272
273fn kafka_error_outcome(resource: String, error: Option<KafkaError>) -> ResourceOutcome {
274    ResourceOutcome {
275        resource,
276        error: error.map(|error| KafkaErrorDto {
277            code: error.code,
278            name: error.name.to_string(),
279            message: error.message,
280        }),
281    }
282}
283
284pub trait IntoResourceOutcomeDto {
285    fn into_resource_outcome(self) -> ResourceOutcome;
286}
287
288impl IntoResourceOutcomeDto for CreateTopicOutcome {
289    fn into_resource_outcome(self) -> ResourceOutcome {
290        kafka_error_outcome(self.name, self.error)
291    }
292}
293
294impl IntoResourceOutcomeDto for DeleteTopicOutcome {
295    fn into_resource_outcome(self) -> ResourceOutcome {
296        kafka_error_outcome(self.name, self.error)
297    }
298}
299
300impl IntoResourceOutcomeDto for CreatePartitionsOutcome {
301    fn into_resource_outcome(self) -> ResourceOutcome {
302        kafka_error_outcome(self.name, self.error)
303    }
304}
305
306impl IntoResourceOutcomeDto for (ConfigResource, Result<(), KafkaError>) {
307    fn into_resource_outcome(self) -> ResourceOutcome {
308        kafka_error_outcome(self.0.name, self.1.err())
309    }
310}
311
312impl IntoResourceOutcomeDto for CreateAclOutcome {
313    fn into_resource_outcome(self) -> ResourceOutcome {
314        kafka_error_outcome("acl".to_string(), self.error)
315    }
316}
317
318impl IntoResourceOutcomeDto for DeleteAclFilterOutcome {
319    fn into_resource_outcome(self) -> ResourceOutcome {
320        kafka_error_outcome("acl-filter".to_string(), self.error)
321    }
322}
323
324impl IntoResourceOutcomeDto for ScramUserOutcome {
325    fn into_resource_outcome(self) -> ResourceOutcome {
326        kafka_error_outcome(self.username, self.error)
327    }
328}
329
330impl IntoResourceOutcomeDto for (TopicPartitionReplica, BrokerResult<()>) {
331    fn into_resource_outcome(self) -> ResourceOutcome {
332        kafka_error_outcome(
333            format!("{}-{}", self.0.topic, self.0.partition),
334            self.1.err(),
335        )
336    }
337}
338
339#[must_use]
340pub fn singleton_quota_config(quota_type: String, value: f64) -> UserQuotaConfig {
341    BTreeMap::from([(quota_type, value)])
342}
krabka_admin_ui/admin.rs

krabka_admin_ui/
admin.rs
