diff --git a/CHANGELOG.md b/CHANGELOG.md
index 83a9e2d87..40ba60582 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,10 +6,12 @@ All notable changes to this project will be documented in this file.
 
 ### Added
 
+- status::condition module to compute the cluster resource status ([#571]).
 - Helper function to build RBAC resources ([#572]).
 - Add `ClusterResourceApplyStrategy` to `ClusterResource` ([#573]).
 - Add `ClusterOperation` common struct with `reconcilation_paused` and `stopped` flags ([#573]).
 
+[#571]: https://github.com/stackabletech/operator-rs/pull/571
 [#572]: https://github.com/stackabletech/operator-rs/pull/572
 [#573]: https://github.com/stackabletech/operator-rs/pull/573
 
diff --git a/src/lib.rs b/src/lib.rs
index d7ad9e6af..73b26be83 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -17,6 +17,7 @@ pub mod pod_utils;
 pub mod product_config_utils;
 pub mod product_logging;
 pub mod role_utils;
+pub mod status;
 pub mod utils;
 pub mod validation;
 pub mod yaml;
diff --git a/src/status/condition/daemonset.rs b/src/status/condition/daemonset.rs
new file mode 100644
index 000000000..6bdc39b1b
--- /dev/null
+++ b/src/status/condition/daemonset.rs
@@ -0,0 +1,173 @@
+use crate::status::condition::{
+    ClusterCondition, ClusterConditionSet, ClusterConditionStatus, ClusterConditionType,
+    ConditionBuilder,
+};
+
+use k8s_openapi::api::apps::v1::DaemonSet;
+use kube::ResourceExt;
+use std::cmp;
+
+/// Default implementation to build [`ClusterCondition`]s for
+/// `DaemonSet` resources.
+///
+/// Currently only the `ClusterConditionType::Available` is implemented. This will be extended
+/// to support all `ClusterConditionType`s in the future.
+#[derive(Default)]
+pub struct DaemonSetConditionBuilder {
+    daemon_sets: Vec<DaemonSet>,
+}
+
+impl ConditionBuilder for DaemonSetConditionBuilder {
+    fn build_conditions(&self) -> ClusterConditionSet {
+        vec![self.available()].into()
+    }
+}
+
+impl DaemonSetConditionBuilder {
+    pub fn add(&mut self, ds: DaemonSet) {
+        self.daemon_sets.push(ds);
+    }
+
+    fn available(&self) -> ClusterCondition {
+        let mut available = ClusterConditionStatus::True;
+        let mut unavailable_resources = vec![];
+
+        for ds in &self.daemon_sets {
+            let current_status = Self::daemon_set_available(ds);
+
+            if current_status != ClusterConditionStatus::True {
+                unavailable_resources.push(ds.name_any())
+            }
+
+            available = cmp::max(available, current_status);
+        }
+
+        // We need to sort here to make sure roles and role groups are not changing position
+        // due to the HashMap (random order) logic.
+        unavailable_resources.sort();
+
+        let message = match available {
+            ClusterConditionStatus::True => {
+                "All DaemonSet have the requested amount of ready replicas.".to_string()
+            }
+            ClusterConditionStatus::False => {
+                format!("DaemonSet {unavailable_resources:?} missing ready replicas.")
+            }
+            ClusterConditionStatus::Unknown => "DaemonSet status cannot be determined.".to_string(),
+        };
+
+        ClusterCondition {
+            reason: None,
+            message: Some(message),
+            status: available,
+            type_: ClusterConditionType::Available,
+            last_transition_time: None,
+            last_update_time: None,
+        }
+    }
+
+    /// Returns a condition "Available: True" if the number of ready Pods is greater than zero.
+    /// This is an heuristic that doesn't take into consideration if *all* eligible nodes have
+    /// running Pods.
+    /// Other fields of the daemon set status have been considered and discarded for being even less
+    /// reliable/informative.
+    fn daemon_set_available(ds: &DaemonSet) -> ClusterConditionStatus {
+        let number_ready = ds
+            .status
+            .as_ref()
+            .map(|status| status.number_ready)
+            .unwrap_or_default();
+
+        if number_ready > 0 {
+            ClusterConditionStatus::True
+        } else {
+            ClusterConditionStatus::False
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::status::condition::daemonset::DaemonSetConditionBuilder;
+    use crate::status::condition::{
+        ClusterCondition, ClusterConditionStatus, ClusterConditionType, ConditionBuilder,
+    };
+    use k8s_openapi::api::apps::v1::{DaemonSet, DaemonSetStatus};
+
+    fn build_ds(number_ready: i32) -> DaemonSet {
+        DaemonSet {
+            status: Some(DaemonSetStatus {
+                number_ready,
+                ..DaemonSetStatus::default()
+            }),
+            ..DaemonSet::default()
+        }
+    }
+
+    #[test]
+    fn test_daemon_set_available_true() {
+        let ds = build_ds(1);
+
+        assert_eq!(
+            DaemonSetConditionBuilder::daemon_set_available(&ds),
+            ClusterConditionStatus::True
+        );
+    }
+
+    #[test]
+    fn test_daemon_set_available_false() {
+        let ds = build_ds(0);
+        assert_eq!(
+            DaemonSetConditionBuilder::daemon_set_available(&ds),
+            ClusterConditionStatus::False
+        );
+    }
+
+    #[test]
+    fn test_daemon_set_available_condition_true() {
+        let mut ds_condition_builder = DaemonSetConditionBuilder::default();
+        ds_condition_builder.add(build_ds(1));
+
+        let conditions = ds_condition_builder.build_conditions();
+
+        let got = conditions
+            .conditions
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::True,
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+    }
+
+    #[test]
+    fn test_daemon_set_available_condition_false() {
+        let mut ds_condition_builder = DaemonSetConditionBuilder::default();
+        ds_condition_builder.add(build_ds(0));
+
+        let conditions = ds_condition_builder.build_conditions();
+
+        let got = conditions
+            .conditions
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::False,
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+    }
+}
diff --git a/src/status/condition/mod.rs b/src/status/condition/mod.rs
new file mode 100644
index 000000000..4a7db91ac
--- /dev/null
+++ b/src/status/condition/mod.rs
@@ -0,0 +1,502 @@
+pub mod daemonset;
+pub mod operations;
+pub mod statefulset;
+
+use chrono::Utc;
+use k8s_openapi::apimachinery::pkg::apis::meta::v1::Time;
+use schemars::{self, JsonSchema};
+use serde::{Deserialize, Serialize};
+use strum::EnumCount;
+
+/// A **data structure** that contains a vector of `ClusterCondition`s.
+/// Should usually be implemented on the status of a `CustomResource` or the `CustomResource` itself.
+pub trait HasStatusCondition {
+    fn conditions(&self) -> Vec<ClusterCondition>;
+}
+
+/// A **data structure** that produces a `ClusterConditionSet` containing all required
+/// `ClusterCondition`s.
+pub trait ConditionBuilder {
+    fn build_conditions(&self) -> ClusterConditionSet;
+}
+
+/// Computes the final conditions to be set in the operator status condition field.
+///
+/// # Arguments
+///
+/// * `resource` - A cluster resource or status implementing [`HasStatusCondition`] in order to
+///    retrieve the "current" conditions set in the cluster. This is required to  compute
+///    condition change and set proper update / transition times.
+/// * `condition_builders` - A slice of structs implementing [`ConditionBuilder`]. This can be a
+///    one of the predefined ConditionBuilders like `DaemonSetConditionBuilder` or a custom
+///    implementation for special resources or different behavior.
+///
+/// # Examples
+/// ```
+/// use stackable_operator::status::condition::daemonset::DaemonSetConditionBuilder;
+/// use stackable_operator::status::condition::statefulset::StatefulSetConditionBuilder;
+/// use k8s_openapi::api::apps::v1::{DaemonSet, StatefulSet};
+/// use stackable_operator::status::condition::{ClusterCondition, ConditionBuilder, HasStatusCondition, compute_conditions};
+///
+/// struct ClusterStatus {
+///     conditions: Vec<ClusterCondition>
+/// }
+///
+/// impl HasStatusCondition for ClusterStatus {
+///     fn conditions(&self) -> Vec<ClusterCondition> {
+///         self.conditions.clone()
+///     }
+/// }
+///
+/// let mut daemonset_condition_builder = DaemonSetConditionBuilder::default();
+/// daemonset_condition_builder.add(DaemonSet::default());
+///
+/// let mut statefulset_condition_builder = StatefulSetConditionBuilder::default();
+/// statefulset_condition_builder.add(StatefulSet::default());
+///
+/// let old_status = ClusterStatus {
+///     conditions: vec![]
+/// };
+///
+/// let new_status = ClusterStatus {
+///     conditions: compute_conditions(&old_status,
+///         &[
+///             &daemonset_condition_builder as &dyn ConditionBuilder,
+///             &statefulset_condition_builder as &dyn ConditionBuilder
+///         ]
+///     )
+/// };
+///
+/// ```
+pub fn compute_conditions<T: HasStatusCondition>(
+    resource: &T,
+    condition_builders: &[&dyn ConditionBuilder],
+) -> Vec<ClusterCondition> {
+    let mut new_resource_conditions = ClusterConditionSet::new();
+    // compute current conditions and merge their message if required
+    for cb in condition_builders {
+        let conditions: ClusterConditionSet = cb.build_conditions();
+        new_resource_conditions = new_resource_conditions.merge(conditions, update_message);
+    }
+
+    let old_resource_conditions: ClusterConditionSet = resource.conditions().into();
+    // merge the computed conditions and update e.g. transition timestamps if required
+    old_resource_conditions
+        .merge(new_resource_conditions, update_timestamps)
+        .into()
+}
+
+#[derive(Clone, Debug, Default, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ClusterCondition {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    /// Last time the condition transitioned from one status to another.
+    pub last_transition_time: Option<Time>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    /// The last time this condition was updated.
+    pub last_update_time: Option<Time>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    /// A human readable message indicating details about the transition.
+    pub message: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    /// The reason for the condition's last transition.
+    pub reason: Option<String>,
+    /// Status of the condition, one of True, False, Unknown.
+    pub status: ClusterConditionStatus,
+    /// Type of deployment condition.
+    #[serde(rename = "type")]
+    pub type_: ClusterConditionType,
+}
+
+#[derive(
+    Clone,
+    Copy,
+    Debug,
+    Default,
+    Deserialize,
+    EnumCount,
+    Eq,
+    Hash,
+    JsonSchema,
+    Ord,
+    PartialEq,
+    PartialOrd,
+    Serialize,
+)]
+#[serde(rename_all = "PascalCase")]
+pub enum ClusterConditionType {
+    #[default]
+    /// Available indicates that the binary maintained by the operator (eg: zookeeper for the
+    /// zookeeper-operator), is functional and available in the cluster.
+    Available,
+    /// Degraded indicates that the operand is not functioning completely. An example of a degraded
+    /// state would be if there should be 5 copies of the operand running but only 4 are running.
+    /// It may still be available, but it is degraded.
+    Degraded,
+    /// Progressing indicates that the operator is actively making changes to the binary maintained
+    /// by the operator (eg: zookeeper for the zookeeper-operator).
+    Progressing,
+    /// Paused indicates that the operator is not reconciling the cluster. This may be used for
+    /// debugging or operator updating.
+    ReconciliationPaused,
+    /// Stopped indicates that all the cluster replicas are scaled down to 0. All resources (e.g.
+    /// ConfigMaps, Services etc.) are kept.
+    Stopped,
+}
+
+// These are indices for the `ClusterConditionSet` conditions vector and should not be altered.
+// This may lead to problems when upgrading the operator later on.
+// When introducing a new `ClusterConditionType` it MUST be the direct successor of the last
+// condition index due to the "array" access of the `ClusterConditionSet` conditions vector.
+impl From<ClusterConditionType> for usize {
+    fn from(condition_type: ClusterConditionType) -> Self {
+        match condition_type {
+            ClusterConditionType::Available => 0,
+            ClusterConditionType::Degraded => 1,
+            ClusterConditionType::Progressing => 2,
+            ClusterConditionType::ReconciliationPaused => 3,
+            ClusterConditionType::Stopped => 4,
+        }
+    }
+}
+
+#[derive(
+    Clone, Debug, Default, Deserialize, Eq, JsonSchema, Ord, PartialEq, PartialOrd, Serialize,
+)]
+#[serde(rename_all = "PascalCase")]
+pub enum ClusterConditionStatus {
+    #[default]
+    /// True means a resource is in the condition.
+    True,
+    /// False means a resource is not in the condition.
+    False,
+    /// Unknown means kubernetes cannot decide if a resource is in the condition or not.
+    Unknown,
+}
+
+#[derive(Clone, Default)]
+/// Helper struct to order and merge `ClusterCondition` objects.
+pub struct ClusterConditionSet {
+    conditions: Vec<Option<ClusterCondition>>,
+}
+
+impl ClusterConditionSet {
+    pub fn new() -> Self {
+        ClusterConditionSet {
+            // We use this as a quasi "Set" where each ClusterConditionType has its fixed position
+            // This ensures ordering, and in contrast to e.g. a
+            // BTreeMap<ClusterConditionType, ClusterCondition>, prevents shenanigans like adding a
+            // ClusterCondition (as value) with a different ClusterConditionType than its key.
+            // See "put".
+            conditions: vec![None; ClusterConditionType::COUNT],
+        }
+    }
+
+    /// Adds a [`ClusterCondition`] to its assigned index in the conditions vector.
+    fn put(&mut self, condition: ClusterCondition) {
+        let index: usize = condition.type_.into();
+        self.conditions[index] = Some(condition);
+    }
+
+    /// Merges two [`ClusterConditionSet`]s. The condition_combiner implements the strategy used to
+    /// merge two conditions of the same `type_`.
+    ///
+    /// # Arguments
+    ///
+    /// * `other` - The [`ClusterConditionSet`] to be merged
+    /// * `condition_combiner` - This is either be `update_message` or `update_timestamps`. The
+    /// `update_message` is used to concatenate messages of the same [`ClusterConditionStatus`] and
+    /// the same [`ClusterConditionType`]. The `update_timestamps` is required to merge the old
+    /// cluster status with the new one and update transition timestamps correctly.
+    fn merge(
+        self,
+        other: ClusterConditionSet,
+        condition_combiner: fn(ClusterCondition, ClusterCondition) -> ClusterCondition,
+    ) -> ClusterConditionSet {
+        let mut result = ClusterConditionSet::new();
+
+        // Combine the two condition vectors of old and new `ClusterConditionSet`.
+        for (old_condition, new_condition) in self
+            .conditions
+            .into_iter()
+            .zip(other.conditions.into_iter())
+        {
+            if let Some(condition) = match (old_condition, new_condition) {
+                // If both are set use the `condition_combiner` to update timestamps or concatenate
+                // the message
+                (Some(old), Some(new)) => Some(condition_combiner(old, new)),
+                // No new condition is computed, keep the old one
+                (Some(old), None) => Some(old),
+                // No condition of that type existed, use the new one
+                (None, Some(new)) => Some(new),
+                _ => None,
+            } {
+                result.put(condition);
+            };
+        }
+
+        result
+    }
+}
+
+/// A condition combiner strategy where the timestamps are updated to reflect a
+/// state transition (if needed).
+fn update_timestamps(
+    old_condition: ClusterCondition,
+    new_condition: ClusterCondition,
+) -> ClusterCondition {
+    // sanity check
+    assert_eq!(old_condition.type_, new_condition.type_);
+
+    let now = Time(Utc::now());
+    // No change in status -> update "last_update_time" and keep "last_transition_time"
+    if old_condition.status == new_condition.status {
+        ClusterCondition {
+            last_update_time: Some(now),
+            last_transition_time: old_condition.last_transition_time,
+            ..new_condition
+        }
+    // Change in status -> set new "last_update_time" and "last_transition_time"
+    } else {
+        ClusterCondition {
+            last_update_time: Some(now.clone()),
+            last_transition_time: Some(now),
+            ..new_condition
+        }
+    }
+}
+
+/// A condition combiner strategy with the following properties:
+/// 1. It preserves the condition with the highest status.
+/// 2. It joins the previous messages to the current one if both conditions
+/// have the same status.
+fn update_message(
+    old_condition: ClusterCondition,
+    new_condition: ClusterCondition,
+) -> ClusterCondition {
+    // sanity check
+    assert_eq!(old_condition.type_, new_condition.type_);
+
+    match old_condition.status.cmp(&new_condition.status) {
+        std::cmp::Ordering::Equal => {
+            let message = Some(
+                vec![old_condition.message, new_condition.message]
+                    .into_iter()
+                    .flatten()
+                    .collect::<Vec<String>>()
+                    .join("\n"),
+            );
+
+            ClusterCondition {
+                message,
+                ..new_condition
+            }
+        }
+        std::cmp::Ordering::Less => new_condition,
+        std::cmp::Ordering::Greater => old_condition,
+    }
+}
+
+impl From<ClusterConditionSet> for Vec<ClusterCondition> {
+    fn from(value: ClusterConditionSet) -> Self {
+        value.conditions.into_iter().flatten().collect()
+    }
+}
+
+impl From<Vec<ClusterCondition>> for ClusterConditionSet {
+    fn from(value: Vec<ClusterCondition>) -> Self {
+        let mut result = ClusterConditionSet::new();
+        for c in value {
+            result.put(c);
+        }
+        result
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    struct TestClusterCondition {}
+    impl HasStatusCondition for TestClusterCondition {
+        fn conditions(&self) -> Vec<ClusterCondition> {
+            vec![ClusterCondition {
+                type_: ClusterConditionType::Available,
+                status: ClusterConditionStatus::Unknown,
+                message: Some("TestClusterCondition=Unknown".into()),
+                ..ClusterCondition::default()
+            }]
+        }
+    }
+
+    struct AvailableFalseConditionBuilder1 {}
+    impl ConditionBuilder for AvailableFalseConditionBuilder1 {
+        fn build_conditions(&self) -> ClusterConditionSet {
+            vec![ClusterCondition {
+                type_: ClusterConditionType::Available,
+                status: ClusterConditionStatus::False,
+                message: Some("AvailableFalseConditionBuilder".into()),
+                ..ClusterCondition::default()
+            }]
+            .into()
+        }
+    }
+
+    struct AvailableFalseConditionBuilder2 {}
+    impl ConditionBuilder for AvailableFalseConditionBuilder2 {
+        fn build_conditions(&self) -> ClusterConditionSet {
+            vec![ClusterCondition {
+                type_: ClusterConditionType::Available,
+                status: ClusterConditionStatus::False,
+                message: Some("AvailableFalseConditionBuilder_2".into()),
+                ..ClusterCondition::default()
+            }]
+            .into()
+        }
+    }
+
+    struct AvailableTrueConditionBuilder1 {}
+    impl ConditionBuilder for AvailableTrueConditionBuilder1 {
+        fn build_conditions(&self) -> ClusterConditionSet {
+            vec![ClusterCondition {
+                type_: ClusterConditionType::Available,
+                status: ClusterConditionStatus::True,
+                message: Some("AvailableTrueConditionBuilder_1".into()),
+                ..ClusterCondition::default()
+            }]
+            .into()
+        }
+    }
+
+    struct AvailableTrueConditionBuilder2 {}
+    impl ConditionBuilder for AvailableTrueConditionBuilder2 {
+        fn build_conditions(&self) -> ClusterConditionSet {
+            vec![ClusterCondition {
+                type_: ClusterConditionType::Available,
+                status: ClusterConditionStatus::True,
+                message: Some("AvailableTrueConditionBuilder_2".into()),
+                ..ClusterCondition::default()
+            }]
+            .into()
+        }
+    }
+
+    struct AvailableUnknownConditionBuilder {}
+    impl ConditionBuilder for AvailableUnknownConditionBuilder {
+        fn build_conditions(&self) -> ClusterConditionSet {
+            vec![ClusterCondition {
+                type_: ClusterConditionType::Available,
+                status: ClusterConditionStatus::Unknown,
+                message: Some("AvailableUnknownConditionBuilder".into()),
+                ..ClusterCondition::default()
+            }]
+            .into()
+        }
+    }
+
+    #[test]
+    pub fn test_compute_conditions_with_transition() {
+        let resource = TestClusterCondition {};
+        let condition_builders = &[&AvailableTrueConditionBuilder1 {} as &dyn ConditionBuilder];
+
+        let got = compute_conditions(&resource, condition_builders)
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::True,
+            message: Some("AvailableTrueConditionBuilder_1".into()),
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+        assert_eq!(got.message, expected.message);
+        assert_eq!(got.last_transition_time, got.last_update_time);
+        assert!(got.last_transition_time.is_some());
+    }
+
+    #[test]
+    pub fn test_compute_conditions_message_concatenation() {
+        let resource = TestClusterCondition {};
+        let condition_builders = &[
+            &AvailableTrueConditionBuilder1 {} as &dyn ConditionBuilder,
+            &AvailableTrueConditionBuilder2 {} as &dyn ConditionBuilder,
+        ];
+
+        let got = compute_conditions(&resource, condition_builders)
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::True,
+            message: Some(
+                "AvailableTrueConditionBuilder_1\nAvailableTrueConditionBuilder_2".into(),
+            ),
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+        assert_eq!(got.message, expected.message);
+    }
+
+    #[test]
+    pub fn test_compute_conditions_message_concatenation_with_different_status() {
+        let resource = TestClusterCondition {};
+        let condition_builders = &[
+            &AvailableFalseConditionBuilder1 {} as &dyn ConditionBuilder,
+            &AvailableTrueConditionBuilder1 {} as &dyn ConditionBuilder,
+            &AvailableFalseConditionBuilder2 {} as &dyn ConditionBuilder,
+            &AvailableTrueConditionBuilder2 {} as &dyn ConditionBuilder,
+        ];
+
+        let got = compute_conditions(&resource, condition_builders)
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::False,
+            message: Some(
+                "AvailableFalseConditionBuilder\nAvailableFalseConditionBuilder_2".into(),
+            ),
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+        assert_eq!(got.message, expected.message);
+    }
+
+    #[test]
+    pub fn test_compute_conditions_status_priority() {
+        let resource = TestClusterCondition {};
+        let condition_builders = &[
+            &AvailableUnknownConditionBuilder {} as &dyn ConditionBuilder,
+            &AvailableFalseConditionBuilder1 {} as &dyn ConditionBuilder,
+            &AvailableTrueConditionBuilder1 {} as &dyn ConditionBuilder,
+        ];
+
+        let got = compute_conditions(&resource, condition_builders)
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::Unknown,
+            message: Some("AvailableUnknownConditionBuilder".into()),
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+        assert_eq!(got.message, expected.message);
+    }
+}
diff --git a/src/status/condition/operations.rs b/src/status/condition/operations.rs
new file mode 100644
index 000000000..d0e3c36ba
--- /dev/null
+++ b/src/status/condition/operations.rs
@@ -0,0 +1,150 @@
+use crate::commons::cluster_operation::ClusterOperation;
+use crate::status::condition::{
+    ClusterCondition, ClusterConditionSet, ClusterConditionStatus, ClusterConditionType,
+    ConditionBuilder,
+};
+
+/// Default implementation to build [`ClusterCondition`]s for
+/// the ClusterOperation.
+#[derive(Debug, Clone)]
+pub struct ClusterOperationsConditionBuilder<'a> {
+    cluster_operation: &'a ClusterOperation,
+}
+
+impl<'a> ConditionBuilder for ClusterOperationsConditionBuilder<'a> {
+    fn build_conditions(&self) -> ClusterConditionSet {
+        vec![self.reconciliation_paused(), self.cluster_stopped()].into()
+    }
+}
+
+impl<'a> ClusterOperationsConditionBuilder<'a> {
+    pub fn new(cluster_operation: &'a ClusterOperation) -> Self {
+        Self { cluster_operation }
+    }
+
+    /// Returns the `ReconciliationPaused` cluster condition.
+    fn reconciliation_paused(&self) -> ClusterCondition {
+        let status = if self.cluster_operation.reconciliation_paused {
+            ClusterConditionStatus::True
+        } else {
+            ClusterConditionStatus::False
+        };
+
+        let message = match status {
+            ClusterConditionStatus::True => {
+                "The cluster reconciliation is paused. Only the cluster status is reconciled."
+                    .to_string()
+            }
+            ClusterConditionStatus::False => "The cluster is reconciled normally.".to_string(),
+            ClusterConditionStatus::Unknown => {
+                "The cluster reconciliation status could not be determined.".to_string()
+            }
+        };
+
+        ClusterCondition {
+            reason: None,
+            message: Some(message),
+            status,
+            type_: ClusterConditionType::ReconciliationPaused,
+            last_transition_time: None,
+            last_update_time: None,
+        }
+    }
+
+    /// Returns the `Stopped` cluster condition.
+    fn cluster_stopped(&self) -> ClusterCondition {
+        let status =
+            if self.cluster_operation.stopped && self.cluster_operation.reconciliation_paused {
+                ClusterConditionStatus::Unknown
+            } else if self.cluster_operation.stopped {
+                ClusterConditionStatus::True
+            } else {
+                ClusterConditionStatus::False
+            };
+
+        let message = match status {
+            ClusterConditionStatus::True => {
+                "The cluster is stopped.".to_string()
+            }
+            ClusterConditionStatus::False => {
+                "The cluster is running.".to_string()
+            }
+            ClusterConditionStatus::Unknown => {
+                "The cluster stopped status could not be determined. This might be due to the cluster reconciliation being paused.".to_string()
+            }
+        };
+
+        ClusterCondition {
+            reason: None,
+            message: Some(message),
+            status,
+            type_: ClusterConditionType::Stopped,
+            last_transition_time: None,
+            last_update_time: None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use rstest::*;
+
+    #[rstest]
+    #[case::not_paused_not_stopped(
+        false,
+        false,
+        ClusterConditionStatus::False,
+        ClusterConditionStatus::False
+    )]
+    #[case::paused_not_stopped(
+        true,
+        false,
+        ClusterConditionStatus::True,
+        ClusterConditionStatus::False
+    )]
+    #[case::not_paused_stopped(
+        false,
+        true,
+        ClusterConditionStatus::False,
+        ClusterConditionStatus::True
+    )]
+    #[case::paused_stopped(
+        true,
+        true,
+        ClusterConditionStatus::True,
+        ClusterConditionStatus::Unknown
+    )]
+    fn test_cluster_operation_condition(
+        #[case] reconciliation_paused: bool,
+        #[case] stopped: bool,
+        #[case] expected_paused_status: ClusterConditionStatus,
+        #[case] expected_stopped_status: ClusterConditionStatus,
+    ) {
+        let cluster_operation = ClusterOperation {
+            reconciliation_paused,
+            stopped,
+        };
+
+        let op_condition_builder = ClusterOperationsConditionBuilder::new(&cluster_operation);
+        let conditions = op_condition_builder.build_conditions();
+
+        let got = conditions
+            .conditions
+            .get::<usize>(ClusterConditionType::ReconciliationPaused.into())
+            .cloned()
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(got.status, expected_paused_status);
+
+        let got = conditions
+            .conditions
+            .get::<usize>(ClusterConditionType::Stopped.into())
+            .cloned()
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(got.status, expected_stopped_status);
+    }
+}
diff --git a/src/status/condition/statefulset.rs b/src/status/condition/statefulset.rs
new file mode 100644
index 000000000..6a6d62a9f
--- /dev/null
+++ b/src/status/condition/statefulset.rs
@@ -0,0 +1,189 @@
+use crate::status::condition::{
+    ClusterCondition, ClusterConditionSet, ClusterConditionStatus, ClusterConditionType,
+    ConditionBuilder,
+};
+
+use k8s_openapi::api::apps::v1::StatefulSet;
+use kube::ResourceExt;
+use std::cmp;
+
+/// Default implementation to build [`ClusterCondition`]s for
+/// `StatefulSet` resources.
+///
+/// Currently only the `ClusterConditionType::Available` is implemented. This will be extended
+/// to support all `ClusterConditionType`s in the future.
+#[derive(Default)]
+pub struct StatefulSetConditionBuilder {
+    stateful_sets: Vec<StatefulSet>,
+}
+
+impl ConditionBuilder for StatefulSetConditionBuilder {
+    fn build_conditions(&self) -> ClusterConditionSet {
+        vec![self.available()].into()
+    }
+}
+
+impl StatefulSetConditionBuilder {
+    pub fn add(&mut self, sts: StatefulSet) {
+        self.stateful_sets.push(sts);
+    }
+
+    fn available(&self) -> ClusterCondition {
+        let mut available = ClusterConditionStatus::True;
+        let mut unavailable_resources = vec![];
+        for sts in &self.stateful_sets {
+            let current_status = Self::stateful_set_available(sts);
+
+            if current_status != ClusterConditionStatus::True {
+                unavailable_resources.push(sts.name_any())
+            }
+
+            available = cmp::max(available, current_status);
+        }
+
+        // We need to sort here to make sure roles and role groups are not changing position
+        // due to the HashMap (random order) logic.
+        unavailable_resources.sort();
+
+        let message = match available {
+            ClusterConditionStatus::True => {
+                "All StatefulSet have the requested amount of ready replicas.".to_string()
+            }
+            ClusterConditionStatus::False => {
+                format!("StatefulSet {unavailable_resources:?} missing ready replicas.")
+            }
+            ClusterConditionStatus::Unknown => {
+                "StatefulSet status cannot be determined.".to_string()
+            }
+        };
+
+        ClusterCondition {
+            reason: None,
+            message: Some(message),
+            status: available,
+            type_: ClusterConditionType::Available,
+            last_transition_time: None,
+            last_update_time: None,
+        }
+    }
+
+    /// Returns a condition "Available: True" if the number of requested replicas matches
+    /// the number of available replicas. In addition, there needs to be at least one replica
+    /// available.
+    fn stateful_set_available(sts: &StatefulSet) -> ClusterConditionStatus {
+        let requested_replicas = sts
+            .spec
+            .as_ref()
+            .and_then(|spec| spec.replicas)
+            .unwrap_or_default();
+        let available_replicas = sts
+            .status
+            .as_ref()
+            .and_then(|status| status.available_replicas)
+            .unwrap_or_default();
+
+        if requested_replicas == available_replicas && requested_replicas != 0 {
+            ClusterConditionStatus::True
+        } else {
+            ClusterConditionStatus::False
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use crate::status::condition::statefulset::StatefulSetConditionBuilder;
+    use crate::status::condition::{
+        ClusterCondition, ClusterConditionStatus, ClusterConditionType, ConditionBuilder,
+    };
+    use k8s_openapi::api::apps::v1::{StatefulSet, StatefulSetSpec, StatefulSetStatus};
+
+    fn build_sts(spec_replicas: i32, available_replicas: i32) -> StatefulSet {
+        StatefulSet {
+            spec: Some(StatefulSetSpec {
+                replicas: Some(spec_replicas),
+                ..StatefulSetSpec::default()
+            }),
+            status: Some(StatefulSetStatus {
+                available_replicas: Some(available_replicas),
+                ..StatefulSetStatus::default()
+            }),
+            ..StatefulSet::default()
+        }
+    }
+
+    #[test]
+    fn test_stateful_set_available_true() {
+        let sts = build_sts(3, 3);
+
+        assert_eq!(
+            StatefulSetConditionBuilder::stateful_set_available(&sts),
+            ClusterConditionStatus::True
+        );
+    }
+
+    #[test]
+    fn test_stateful_set_available_false() {
+        let sts = build_sts(3, 2);
+
+        assert_eq!(
+            StatefulSetConditionBuilder::stateful_set_available(&sts),
+            ClusterConditionStatus::False
+        );
+
+        let sts = build_sts(3, 4);
+
+        assert_eq!(
+            StatefulSetConditionBuilder::stateful_set_available(&sts),
+            ClusterConditionStatus::False
+        );
+    }
+
+    #[test]
+    fn test_stateful_set_available_condition_true() {
+        let mut sts_condition_builder = StatefulSetConditionBuilder::default();
+        sts_condition_builder.add(build_sts(3, 3));
+
+        let conditions = sts_condition_builder.build_conditions();
+
+        let got = conditions
+            .conditions
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::True,
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+    }
+
+    #[test]
+    fn test_stateful_set_available_condition_false() {
+        let mut sts_condition_builder = StatefulSetConditionBuilder::default();
+        sts_condition_builder.add(build_sts(3, 2));
+
+        let conditions = sts_condition_builder.build_conditions();
+
+        let got = conditions
+            .conditions
+            .get::<usize>(ClusterConditionType::Available.into())
+            .cloned()
+            .unwrap()
+            .unwrap();
+
+        let expected = ClusterCondition {
+            type_: ClusterConditionType::Available,
+            status: ClusterConditionStatus::False,
+            ..ClusterCondition::default()
+        };
+
+        assert_eq!(got.type_, expected.type_);
+        assert_eq!(got.status, expected.status);
+    }
+}
diff --git a/src/status/mod.rs b/src/status/mod.rs
new file mode 100644
index 000000000..db3947f61
--- /dev/null
+++ b/src/status/mod.rs
@@ -0,0 +1 @@
+pub mod condition;