Use compact formatting for tracing if running through cargo

Cargo.lock

@@ -1425,7 +1425,6 @@ name = "lldap-controller"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "async-trait",
  "chrono",
  "cynic",
  "futures",

Cargo.toml

@@ -30,7 +30,6 @@ tracing = "0.1.41"
 thiserror = "2.0.12"
 chrono = "0.4.40"
 passwords = "3.1.16"
-async-trait = "0.1.88"
 reqwest = { version = "0.12.14", default-features = false, features = [
   "json",
   "rustls-tls",

queries/src/lib.rs

@@ -86,6 +86,30 @@ pub struct GetGroups {
     pub groups: Vec<Group>,
 }
 
+#[derive(cynic::QueryVariables, Debug)]
+pub struct CreateGroupVariables<'a> {
+    pub name: &'a str,
+}
+
+#[derive(cynic::QueryFragment, Debug)]
+#[cynic(graphql_type = "Mutation", variables = "CreateGroupVariables")]
+pub struct CreateGroup {
+    #[arguments(name: $name)]
+    pub create_group: Group,
+}
+
+#[derive(cynic::QueryVariables, Debug)]
+pub struct DeleteGroupVariables {
+    pub id: i32,
+}
+
+#[derive(cynic::QueryFragment, Debug)]
+#[cynic(graphql_type = "Mutation", variables = "DeleteGroupVariables")]
+pub struct DeleteGroup {
+    #[arguments(groupId: $id)]
+    pub delete_group: Success,
+}
+
 #[cfg(test)]
 mod tests {
     use cynic::{MutationBuilder, QueryBuilder};
@@ -139,4 +163,18 @@ mod tests {
 
         insta::assert_snapshot!(operation.query);
     }
+
+    #[test]
+    fn create_group_gql_output() {
+        let operation = CreateGroup::build(CreateGroupVariables { name: "group" });
+
+        insta::assert_snapshot!(operation.query);
+    }
+
+    #[test]
+    fn delete_group_gql_output() {
+        let operation = DeleteGroup::build(DeleteGroupVariables { id: 0 });
+
+        insta::assert_snapshot!(operation.query);
+    }
 }

queries/src/snapshots/queries__tests__create_group_gql_output.snap

@@ -0,0 +1,10 @@
+---
+source: queries/src/lib.rs
+expression: operation.query
+---
+mutation CreateGroup($name: String!) {
+  createGroup(name: $name) {
+    id
+    displayName
+  }
+}

queries/src/snapshots/queries__tests__delete_group_gql_output.snap

@@ -0,0 +1,9 @@
+---
+source: queries/src/lib.rs
+expression: operation.query
+---
+mutation DeleteGroup($id: Int!) {
+  deleteGroup(groupId: $id) {
+    ok
+  }
+}

src/bin/crdgen.rs

@@ -2,7 +2,8 @@ use kube::CustomResourceExt;
 
 fn main() {
     print!(
-        "{}",
-        serde_yaml::to_string(&lldap_controller::resources::ServiceUser::crd()).unwrap()
+        "{}---\n{}",
+        serde_yaml::to_string(&lldap_controller::resources::ServiceUser::crd()).unwrap(),
+        serde_yaml::to_string(&lldap_controller::resources::Group::crd()).unwrap()
     )
 }

src/context.rs

@@ -1,10 +1,10 @@
-use async_trait::async_trait;
 use k8s_openapi::api::core::v1::Secret;
 use kube::runtime::events::{Event, EventType, Recorder, Reporter};
 use kube::{Resource, ResourceExt};
 
 use crate::lldap::LldapConfig;
 
+#[derive(Clone)]
 pub struct Context {
     pub client: kube::Client,
     pub lldap_config: LldapConfig,
@@ -26,7 +26,7 @@ impl Context {
     }
 }
 
-#[async_trait]
+#[allow(async_fn_in_trait)]
 pub trait ControllerEvents {
     type Error;
 
@@ -38,16 +38,23 @@ pub trait ControllerEvents {
     where
         T: Resource<DynamicType = ()> + Sync;
 
+    async fn group_created<T>(&self, obj: &T, name: &str) -> Result<(), Self::Error>
+    where
+        T: Resource<DynamicType = ()> + Sync;
+
     async fn user_deleted<T>(&self, obj: &T, username: &str) -> Result<(), Self::Error>
     where
         T: Resource<DynamicType = ()> + Sync;
 
+    async fn group_deleted<T>(&self, obj: &T, name: &str) -> Result<(), Self::Error>
+    where
+        T: Resource<DynamicType = ()> + Sync;
+
     async fn user_not_found<T>(&self, obj: &T, username: &str) -> Result<(), Self::Error>
     where
         T: Resource<DynamicType = ()> + Sync;
 }
 
-#[async_trait]
 impl ControllerEvents for Recorder {
     type Error = kube::Error;
 
@@ -85,6 +92,23 @@ impl ControllerEvents for Recorder {
         .await
     }
 
+    async fn group_created<T>(&self, obj: &T, name: &str) -> Result<(), Self::Error>
+    where
+        T: Resource<DynamicType = ()> + Sync,
+    {
+        self.publish(
+            &Event {
+                type_: EventType::Normal,
+                reason: "GroupCreated".into(),
+                note: Some(format!("Created group '{name}'")),
+                action: "GroupCreated".into(),
+                secondary: None,
+            },
+            &obj.object_ref(&()),
+        )
+        .await
+    }
+
     async fn user_deleted<T>(&self, obj: &T, username: &str) -> Result<(), Self::Error>
     where
         T: Resource<DynamicType = ()> + Sync,
@@ -102,6 +126,23 @@ impl ControllerEvents for Recorder {
         .await
     }
 
+    async fn group_deleted<T>(&self, obj: &T, name: &str) -> Result<(), Self::Error>
+    where
+        T: Resource<DynamicType = ()> + Sync,
+    {
+        self.publish(
+            &Event {
+                type_: EventType::Normal,
+                reason: "GroupDeleted".into(),
+                note: Some(format!("Deleted group '{name}'")),
+                action: "GroupDeleted".into(),
+                secondary: None,
+            },
+            &obj.object_ref(&()),
+        )
+        .await
+    }
+
     async fn user_not_found<T>(&self, obj: &T, username: &str) -> Result<(), Self::Error>
     where
         T: Resource<DynamicType = ()> + Sync,

src/lldap.rs

@@ -8,9 +8,10 @@ use lldap_auth::opaque::AuthenticationError;
 use lldap_auth::registration::ServerRegistrationStartResponse;
 use lldap_auth::{opaque, registration};
 use queries::{
-    AddUserToGroup, AddUserToGroupVariables, CreateUser, CreateUserVariables, DeleteUser,
-    DeleteUserVariables, GetGroups, GetUser, GetUserVariables, Group, RemoveUserFromGroup,
-    RemoveUserFromGroupVariables, User,
+    AddUserToGroup, AddUserToGroupVariables, CreateGroup, CreateGroupVariables, CreateUser,
+    CreateUserVariables, DeleteGroup, DeleteGroupVariables, DeleteUser, DeleteUserVariables,
+    GetGroups, GetUser, GetUserVariables, Group, RemoveUserFromGroup, RemoveUserFromGroupVariables,
+    User,
 };
 use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue};
 use tracing::{debug, trace};
@@ -41,6 +42,7 @@ fn check_graphql_errors<T>(response: GraphQlResponse<T>) -> Result<T> {
         .expect("Data should be valid if there are no error"))
 }
 
+#[derive(Clone)]
 pub struct LldapConfig {
     username: String,
     password: String,
@@ -150,6 +152,32 @@ impl LldapClient {
         Ok(check_graphql_errors(response)?.groups)
     }
 
+    pub async fn create_group(&self, name: &str) -> Result<Group> {
+        let operation = CreateGroup::build(CreateGroupVariables { name });
+
+        let response = self
+            .client
+            .post(format!("{}/api/graphql", self.url))
+            .run_graphql(operation)
+            .await?;
+
+        Ok(check_graphql_errors(response)?.create_group)
+    }
+
+    pub async fn delete_group(&self, id: i32) -> Result<()> {
+        let operation = DeleteGroup::build(DeleteGroupVariables { id });
+
+        let response = self
+            .client
+            .post(format!("{}/api/graphql", self.url))
+            .run_graphql(operation)
+            .await?;
+
+        check_graphql_errors(response)?;
+
+        Ok(())
+    }
+
     pub async fn add_user_to_group(&self, username: &str, group: i32) -> Result<()> {
         let operation = AddUserToGroup::build(AddUserToGroupVariables { username, group });
 

src/main.rs

@@ -3,30 +3,47 @@ use std::time::Duration;
 
 use futures::StreamExt;
 use k8s_openapi::api::core::v1::Secret;
-use kube::runtime::Controller;
-use kube::runtime::controller::Action;
-use kube::{Api, Client as KubeClient};
+use kube::runtime::controller::{self, Action};
+use kube::runtime::reflector::ObjectRef;
+use kube::runtime::{Controller, watcher};
+use kube::{Api, Client as KubeClient, Resource};
 use lldap_controller::context::Context;
 use lldap_controller::lldap::LldapConfig;
-use lldap_controller::resources::{self, ServiceUser, reconcile};
+use lldap_controller::resources::{self, Error, Group, ServiceUser, reconcile};
 use tracing::{debug, info, warn};
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::util::SubscriberInitExt;
 use tracing_subscriber::{EnvFilter, Registry};
 
-fn error_policy(_obj: Arc<ServiceUser>, err: &resources::Error, _ctx: Arc<Context>) -> Action {
+fn error_policy<T>(_obj: Arc<T>, err: &resources::Error, _ctx: Arc<Context>) -> Action {
     warn!("error: {}", err);
     Action::requeue(Duration::from_secs(5))
 }
 
+async fn log_status<T>(
+    res: Result<(ObjectRef<T>, Action), controller::Error<Error, watcher::Error>>,
+) where
+    T: Resource,
+{
+    match res {
+        Ok(obj) => debug!("reconciled {:?}", obj.0.name),
+        Err(err) => warn!("reconcile failed: {}", err),
+    }
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let logger = tracing_subscriber::fmt::layer().json();
     let env_filter = EnvFilter::try_from_default_env()
         .or_else(|_| EnvFilter::try_new("info"))
         .expect("Fallback should be valid");
 
-    Registry::default().with(logger).with(env_filter).init();
+    if std::env::var("CARGO").is_ok() {
+        let logger = tracing_subscriber::fmt::layer().compact();
+        Registry::default().with(logger).with(env_filter).init();
+    } else {
+        let logger = tracing_subscriber::fmt::layer().json();
+        Registry::default().with(logger).with(env_filter).init();
+    }
 
     info!("Starting controller");
 
@@ -41,17 +58,20 @@ async fn main() -> anyhow::Result<()> {
     let service_users = Api::<ServiceUser>::all(client.clone());
     let secrets = Api::<Secret>::all(client.clone());
 
-    Controller::new(service_users.clone(), Default::default())
+    let service_user_controller = Controller::new(service_users, Default::default())
         .owns(secrets, Default::default())
+        .shutdown_on_signal()
+        .run(reconcile, error_policy, Arc::new(data.clone()))
+        .for_each(log_status);
+
+    let groups = Api::<Group>::all(client.clone());
+
+    let group_controller = Controller::new(groups, Default::default())
         .shutdown_on_signal()
         .run(reconcile, error_policy, Arc::new(data))
-        .for_each(|res| async move {
-            match res {
-                Ok(obj) => debug!("reconciled {:?}", obj.0.name),
-                Err(err) => warn!("reconcile failed: {}", err),
-            }
-        })
-        .await;
+        .for_each(log_status);
+
+    tokio::join!(service_user_controller, group_controller);
 
     Ok(())
 }

src/resources/group.rs

@@ -0,0 +1,76 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use kube::CustomResource;
+use kube::runtime::controller::Action;
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use tracing::{debug, trace};
+
+use super::{Error, Reconcile, Result};
+use crate::context::{Context, ControllerEvents};
+
+#[derive(CustomResource, Deserialize, Serialize, Clone, Debug, JsonSchema)]
+#[kube(kind = "Group", group = "lldap.huizinga.dev", version = "v1")]
+#[kube(
+    shortname = "lg",
+    doc = "Custom resource for managing Groups inside of LLDAP"
+)]
+#[serde(rename_all = "camelCase")]
+pub struct GroupSpec {}
+
+impl Reconcile for Group {
+    async fn reconcile(self: Arc<Self>, ctx: Arc<Context>) -> Result<Action> {
+        let name = self
+            .metadata
+            .name
+            .clone()
+            .ok_or(Error::MissingObjectKey(".metadata.name"))?;
+
+        debug!(name, "Apply");
+
+        let lldap_client = ctx.lldap_config.build_client().await?;
+
+        trace!(name, "Get existing groups");
+        let groups = lldap_client.get_groups().await?;
+
+        if !groups.iter().any(|group| group.display_name == name) {
+            trace!("Group does not exist yet");
+
+            lldap_client.create_group(&name).await?;
+
+            ctx.recorder.group_created(self.as_ref(), &name).await?;
+        } else {
+            trace!("Group already exists");
+        }
+
+        Ok(Action::requeue(Duration::from_secs(3600)))
+    }
+
+    async fn cleanup(self: Arc<Self>, ctx: Arc<Context>) -> Result<Action> {
+        let name = self
+            .metadata
+            .name
+            .clone()
+            .ok_or(Error::MissingObjectKey(".metadata.name"))?;
+
+        debug!(name, "Cleanup");
+
+        let lldap_client = ctx.lldap_config.build_client().await?;
+
+        trace!(name, "Get existing groups");
+        let groups = lldap_client.get_groups().await?;
+
+        if let Some(group) = groups.iter().find(|group| group.display_name == name) {
+            trace!(name, "Deleting group");
+
+            lldap_client.delete_group(group.id).await?;
+
+            ctx.recorder.group_deleted(self.as_ref(), &name).await?;
+        } else {
+            trace!(name, "Group does not exist")
+        }
+
+        Ok(Action::await_change())
+    }
+}

src/resources/mod.rs

@@ -1,10 +1,9 @@
+mod group;
 mod service_user;
 
 use core::fmt;
 use std::sync::Arc;
 
-use async_trait::async_trait;
-use k8s_openapi::NamespaceResourceScope;
 use kube::runtime::controller::Action;
 use kube::runtime::finalizer;
 use kube::{Api, Resource, ResourceExt};
@@ -12,6 +11,7 @@ use serde::Serialize;
 use serde::de::DeserializeOwned;
 use tracing::{debug, instrument};
 
+pub use self::group::Group;
 pub use self::service_user::ServiceUser;
 use crate::context::Context;
 use crate::lldap;
@@ -38,7 +38,6 @@ impl From<finalizer::Error<Self>> for Error {
 
 type Result<T, E = Error> = std::result::Result<T, E>;
 
-#[async_trait]
 trait Reconcile {
     async fn reconcile(self: Arc<Self>, ctx: Arc<Context>) -> Result<Action>;
 
@@ -48,19 +47,12 @@ trait Reconcile {
 #[instrument(skip(obj, ctx))]
 pub async fn reconcile<T>(obj: Arc<T>, ctx: Arc<Context>) -> Result<Action>
 where
-    T: Resource<Scope = NamespaceResourceScope>
-        + ResourceExt
-        + Clone
-        + Serialize
-        + DeserializeOwned
-        + fmt::Debug
-        + Reconcile,
+    T: Resource + ResourceExt + Clone + Serialize + DeserializeOwned + fmt::Debug + Reconcile,
     <T as Resource>::DynamicType: Default,
 {
     debug!(name = obj.name_any(), "Reconcile");
 
-    let namespace = obj.namespace().expect("Resource is namespace scoped");
-    let service_users = Api::<T>::namespaced(ctx.client.clone(), &namespace);
+    let service_users = Api::<T>::all(ctx.client.clone());
 
     Ok(
         finalizer(&service_users, &ctx.controller_name, obj, |event| async {

src/resources/service_user.rs

@@ -3,7 +3,6 @@ use std::str::from_utf8;
 use std::sync::Arc;
 use std::time::Duration;
 
-use async_trait::async_trait;
 use chrono::{DateTime, Utc};
 use k8s_openapi::api::core::v1::Secret;
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::OwnerReference;
@@ -76,7 +75,6 @@ fn format_username(name: &str, namespace: &str) -> String {
     format!("{name}.{namespace}")
 }
 
-#[async_trait]
 impl Reconcile for ServiceUser {
     async fn reconcile(self: Arc<Self>, ctx: Arc<Context>) -> Result<Action> {
         let name = self

yaml/group.yaml

@@ -0,0 +1,5 @@
+apiVersion: lldap.huizinga.dev/v1
+kind: Group
+metadata:
+  name: test-group
+spec: {}

yaml/service-user.yaml

@@ -1,6 +1,5 @@
 apiVersion: lldap.huizinga.dev/v1
 kind: ServiceUser
 metadata:
-  name: authelia
-spec:
-  passwordManager: false
+  name: test-user
+spec: {}