diff --git a/src/builder/kvp.rs b/src/builder/kvp.rs
new file mode 100644
index 000000000..301bbc4f0
--- /dev/null
+++ b/src/builder/kvp.rs
@@ -0,0 +1,73 @@
+use std::collections::BTreeMap;
+
+use crate::types::{Annotation, KeyValuePairExt, KeyValuePairParseError, Label};
+
+pub type AnnotationListBuilder = KeyValuePairBuilder<Annotation>;
+pub type LabelListBuilder = KeyValuePairBuilder<Label>;
+
+pub struct KeyValuePairBuilder<P: KeyValuePairExt> {
+    prefix: Option<String>,
+    kvps: Vec<P>,
+}
+
+impl<P: KeyValuePairExt> KeyValuePairBuilder<P> {
+    /// Creates a new key/value pair list builder with items of type `P`. The
+    /// optional `prefix` will be attached to each added pair via the
+    /// [`KeyValuePairBuilder::add`] method. This builder makes it easier to
+    /// build a list of key/value pairs which all share a **common** key prefix,
+    /// like `app.kubernetes.io/<name>=<value>`.
+    pub fn new<T>(prefix: Option<T>) -> Self
+    where
+        T: Into<String>,
+    {
+        Self {
+            prefix: prefix.map(Into::into),
+            kvps: Vec::new(),
+        }
+    }
+
+    /// Tries to add a new key/value pair to this builder. This method ensures
+    /// that the optional key prefix is valid and in addition also validates
+    /// the key name. Both values must not exceed the maximum length of 253
+    /// characters for the prefix and 63 characters for the name. Also, both
+    /// need to only contain allowed characters.
+    pub fn add<T>(&mut self, name: T, value: T) -> Result<&mut Self, KeyValuePairParseError>
+    where
+        T: Into<String>,
+    {
+        let kvp = P::new(self.prefix.clone(), name.into(), value.into())?;
+        self.kvps.push(kvp);
+        Ok(self)
+    }
+
+    /// Builds a [`BTreeMap<String, P>`] out of the added pairs. This mthods is
+    /// useful when we need a map of items with the underlying type `P` still
+    /// preserved. The Kubernetes API uses [`BTreeMap<String, String>`] to
+    /// handle lists of labels and annotations. The build such a map, use the
+    /// `build_raw` method instead.
+    pub fn build(self) -> BTreeMap<String, P> {
+        self.kvps.into_iter().map(|kvp| (kvp.key(), kvp)).collect()
+    }
+
+    /// Builds a [`BTreeMap<String, String>`] out of the added pairs. This
+    /// method is useful when the returned value is directly passed to
+    /// the Kubernetes API or data structures.
+    pub fn build_raw(self) -> BTreeMap<String, String> {
+        self.kvps
+            .into_iter()
+            .map(|kvp| (kvp.key(), kvp.value().clone()))
+            .collect()
+    }
+}
+
+pub trait KeyValuePairMapExt {
+    fn into_raw(self) -> BTreeMap<String, String>;
+}
+
+impl<P: KeyValuePairExt> KeyValuePairMapExt for BTreeMap<String, P> {
+    fn into_raw(self) -> BTreeMap<String, String> {
+        self.into_iter()
+            .map(|(name, kvp)| (name, kvp.value().clone()))
+            .collect()
+    }
+}
diff --git a/src/builder/meta.rs b/src/builder/meta.rs
index dcee94da2..5024f5aeb 100644
--- a/src/builder/meta.rs
+++ b/src/builder/meta.rs
@@ -1,10 +1,16 @@
-use crate::error::{Error, OperatorResult};
-use crate::labels::{self, ObjectLabels};
+use std::collections::BTreeMap;
+
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, OwnerReference};
 use kube::{Resource, ResourceExt};
-use std::collections::BTreeMap;
 use tracing::warn;
 
+use crate::{
+    builder::kvp::KeyValuePairMapExt,
+    error::{Error, OperatorResult},
+    labels::{self, ObjectLabels},
+    types::{Annotation, KeyValuePairExt, Label, LabelParseError},
+};
+
 /// A builder to build [`ObjectMeta`] objects.
 ///
 /// Of special interest is the [`Self::ownerreference_from_resource()`] function.
@@ -13,12 +19,12 @@ use tracing::warn;
 /// It is strongly recommended to always call [`Self::with_recommended_labels()`]!
 #[derive(Clone, Default)]
 pub struct ObjectMetaBuilder {
-    name: Option<String>,
+    annotations: BTreeMap<String, Annotation>,
+    owner_reference: Option<OwnerReference>,
+    labels: BTreeMap<String, Label>,
     generate_name: Option<String>,
     namespace: Option<String>,
-    ownerreference: Option<OwnerReference>,
-    labels: Option<BTreeMap<String, String>>,
-    annotations: Option<BTreeMap<String, String>>,
+    name: Option<String>,
 }
 
 impl ObjectMetaBuilder {
@@ -27,60 +33,60 @@ impl ObjectMetaBuilder {
     }
 
     /// This sets the name and namespace from a given resource
-    pub fn name_and_namespace<T: Resource>(&mut self, resource: &T) -> &mut Self {
+    pub fn name_and_namespace<T: Resource>(mut self, resource: &T) -> Self {
         self.name = Some(resource.name_any());
         self.namespace = resource.namespace();
         self
     }
 
-    pub fn name_opt(&mut self, name: impl Into<Option<String>>) -> &mut Self {
+    pub fn name_opt(mut self, name: impl Into<Option<String>>) -> Self {
         self.name = name.into();
         self
     }
 
-    pub fn name(&mut self, name: impl Into<String>) -> &mut Self {
+    pub fn name(mut self, name: impl Into<String>) -> Self {
         self.name = Some(name.into());
         self
     }
 
-    pub fn generate_name(&mut self, generate_name: impl Into<String>) -> &mut Self {
+    pub fn generate_name(mut self, generate_name: impl Into<String>) -> Self {
         self.generate_name = Some(generate_name.into());
         self
     }
 
-    pub fn generate_name_opt(&mut self, generate_name: impl Into<Option<String>>) -> &mut Self {
+    pub fn generate_name_opt(mut self, generate_name: impl Into<Option<String>>) -> Self {
         self.generate_name = generate_name.into();
         self
     }
 
-    pub fn namespace_opt(&mut self, namespace: impl Into<Option<String>>) -> &mut Self {
+    pub fn namespace_opt(mut self, namespace: impl Into<Option<String>>) -> Self {
         self.namespace = namespace.into();
         self
     }
 
-    pub fn namespace(&mut self, namespace: impl Into<String>) -> &mut Self {
+    pub fn namespace(mut self, namespace: impl Into<String>) -> Self {
         self.namespace = Some(namespace.into());
         self
     }
 
-    pub fn ownerreference(&mut self, ownerreference: OwnerReference) -> &mut Self {
-        self.ownerreference = Some(ownerreference);
+    pub fn ownerreference(mut self, ownerreference: OwnerReference) -> Self {
+        self.owner_reference = Some(ownerreference);
         self
     }
 
-    pub fn ownerreference_opt(&mut self, ownerreference: Option<OwnerReference>) -> &mut Self {
-        self.ownerreference = ownerreference;
+    pub fn ownerreference_opt(mut self, ownerreference: Option<OwnerReference>) -> Self {
+        self.owner_reference = ownerreference;
         self
     }
 
     /// This can be used to set the `OwnerReference` to the provided resource.
     pub fn ownerreference_from_resource<T: Resource<DynamicType = ()>>(
-        &mut self,
+        mut self,
         resource: &T,
         block_owner_deletion: Option<bool>,
         controller: Option<bool>,
-    ) -> OperatorResult<&mut Self> {
-        self.ownerreference = Some(
+    ) -> OperatorResult<Self> {
+        self.owner_reference = Some(
             OwnerReferenceBuilder::new()
                 .initialize_from_resource(resource)
                 .block_owner_deletion_opt(block_owner_deletion)
@@ -92,55 +98,41 @@ impl ObjectMetaBuilder {
 
     /// This adds a single annotation to the existing annotations.
     /// It'll override an annotation with the same key.
-    pub fn with_annotation(
-        &mut self,
-        annotation_key: impl Into<String>,
-        annotation_value: impl Into<String>,
-    ) -> &mut Self {
-        self.annotations
-            .get_or_insert_with(BTreeMap::new)
-            .insert(annotation_key.into(), annotation_value.into());
+    pub fn with_annotation(mut self, annotation: Annotation) -> Self {
+        self.annotations.insert(annotation.key(), annotation);
         self
     }
 
     /// This adds multiple annotations to the existing annotations.
     /// Any existing annotation with a key that is contained in `annotations` will be overwritten
-    pub fn with_annotations(&mut self, annotations: BTreeMap<String, String>) -> &mut Self {
-        self.annotations
-            .get_or_insert_with(BTreeMap::new)
-            .extend(annotations);
+    pub fn with_annotations(mut self, annotations: BTreeMap<String, Annotation>) -> Self {
+        self.annotations.extend(annotations);
         self
     }
 
     /// This will replace all existing annotations
-    pub fn annotations(&mut self, annotations: BTreeMap<String, String>) -> &mut Self {
-        self.annotations = Some(annotations);
+    pub fn annotations(mut self, annotations: BTreeMap<String, Annotation>) -> Self {
+        self.annotations = annotations;
         self
     }
 
     /// This adds a single label to the existing labels.
     /// It'll override a label with the same key.
-    pub fn with_label(
-        &mut self,
-        label_key: impl Into<String>,
-        label_value: impl Into<String>,
-    ) -> &mut Self {
-        self.labels
-            .get_or_insert_with(BTreeMap::new)
-            .insert(label_key.into(), label_value.into());
+    pub fn with_label(mut self, label: Label) -> Self {
+        self.labels.insert(label.key(), label);
         self
     }
 
     /// This adds multiple labels to the existing labels.
     /// Any existing label with a key that is contained in `labels` will be overwritten
-    pub fn with_labels(&mut self, labels: BTreeMap<String, String>) -> &mut Self {
-        self.labels.get_or_insert_with(BTreeMap::new).extend(labels);
+    pub fn with_labels(mut self, labels: BTreeMap<String, Label>) -> Self {
+        self.labels.extend(labels);
         self
     }
 
     /// This will replace all existing labels
-    pub fn labels(&mut self, labels: BTreeMap<String, String>) -> &mut Self {
-        self.labels = Some(labels);
+    pub fn labels(mut self, labels: BTreeMap<String, Label>) -> Self {
+        self.labels = labels;
         self
     }
 
@@ -149,17 +141,15 @@ impl ObjectMetaBuilder {
     /// The only reasons it is not _required_ is to make testing easier and to allow for more
     /// flexibility if needed.
     pub fn with_recommended_labels<T: Resource>(
-        &mut self,
+        mut self,
         object_labels: ObjectLabels<T>,
-    ) -> &mut Self {
-        let recommended_labels = labels::get_recommended_labels(object_labels);
-        self.labels
-            .get_or_insert_with(BTreeMap::new)
-            .extend(recommended_labels);
-        self
+    ) -> Result<Self, LabelParseError> {
+        let recommended_labels = labels::get_recommended_labels(object_labels)?;
+        self.labels.extend(recommended_labels);
+        Ok(self)
     }
 
-    pub fn build(&self) -> ObjectMeta {
+    pub fn build(self) -> ObjectMeta {
         // if 'generate_name' and 'name' are set, Kubernetes will prioritize the 'name' field and
         // 'generate_name' has no impact.
         if let (Some(name), Some(generate_name)) = (&self.name, &self.generate_name) {
@@ -170,16 +160,19 @@ impl ObjectMetaBuilder {
             );
         }
 
+        let annotations = (!self.annotations.is_empty()).then_some(self.annotations.into_raw());
+        let labels = (!self.labels.is_empty()).then_some(self.labels.into_raw());
+
         ObjectMeta {
             generate_name: self.generate_name.clone(),
             name: self.name.clone(),
             namespace: self.namespace.clone(),
             owner_references: self
-                .ownerreference
+                .owner_reference
                 .as_ref()
                 .map(|ownerreference| vec![ownerreference.clone()]),
-            labels: self.labels.clone(),
-            annotations: self.annotations.clone(),
+            labels,
+            annotations,
             ..ObjectMeta::default()
         }
     }
@@ -305,7 +298,7 @@ impl OwnerReferenceBuilder {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::builder::meta::ObjectMetaBuilder;
+    use crate::{builder::meta::ObjectMetaBuilder, types::Annotation};
     use k8s_openapi::api::core::v1::Pod;
 
     #[test]
@@ -329,7 +322,8 @@ mod tests {
                 role: "role",
                 role_group: "rolegroup",
             })
-            .with_annotation("foo", "bar")
+            .unwrap()
+            .with_annotation(Annotation::new(None, "foo", "bar").unwrap())
             .build();
 
         assert_eq!(meta.generate_name, Some("generate_foo".to_string()));
diff --git a/src/builder/mod.rs b/src/builder/mod.rs
index 40d72fe7e..a72ff12a9 100644
--- a/src/builder/mod.rs
+++ b/src/builder/mod.rs
@@ -5,6 +5,7 @@
 //!
 pub mod configmap;
 pub mod event;
+pub mod kvp;
 pub mod meta;
 pub mod pod;
 
diff --git a/src/builder/pod/container.rs b/src/builder/pod/container.rs
index f7d7a1d94..f666ae590 100644
--- a/src/builder/pod/container.rs
+++ b/src/builder/pod/container.rs
@@ -1,8 +1,9 @@
+use std::fmt;
+
 use k8s_openapi::api::core::v1::{
     ConfigMapKeySelector, Container, ContainerPort, EnvVar, EnvVarSource, ObjectFieldSelector,
     Probe, ResourceRequirements, SecretKeySelector, SecurityContext, VolumeMount,
 };
-use std::fmt;
 
 use crate::{
     commons::product_image_selection::ResolvedProductImage, error::Error,
diff --git a/src/builder/pod/mod.rs b/src/builder/pod/mod.rs
index d54d6b110..3ccce91cc 100644
--- a/src/builder/pod/mod.rs
+++ b/src/builder/pod/mod.rs
@@ -5,26 +5,31 @@ pub mod volume;
 
 use std::collections::BTreeMap;
 
-use crate::builder::meta::ObjectMetaBuilder;
-use crate::commons::affinity::StackableAffinity;
-use crate::commons::product_image_selection::ResolvedProductImage;
-use crate::commons::resources::{
-    ComputeResource, ResourceRequirementsExt, ResourceRequirementsType, LIMIT_REQUEST_RATIO_CPU,
-    LIMIT_REQUEST_RATIO_MEMORY,
-};
-use crate::error::{Error, OperatorResult};
-
-use super::{ListenerOperatorVolumeSourceBuilder, ListenerReference, VolumeBuilder};
 use k8s_openapi::{
     api::core::v1::{
         Affinity, Container, LocalObjectReference, NodeAffinity, Pod, PodAffinity, PodAntiAffinity,
-        PodCondition, PodSecurityContext, PodSpec, PodStatus, PodTemplateSpec,
-        ResourceRequirements, Toleration, Volume,
+        PodCondition, PodSecurityContext, PodSpec, PodStatus, PodTemplateSpec, Toleration, Volume,
     },
     apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::ObjectMeta},
 };
 use tracing::warn;
 
+use crate::{
+    builder::{
+        meta::ObjectMetaBuilder, resources::ResourceRequirementsBuilder,
+        ListenerOperatorVolumeSourceBuilder, ListenerReference, VolumeBuilder,
+    },
+    commons::{
+        affinity::StackableAffinity,
+        product_image_selection::ResolvedProductImage,
+        resources::{
+            ComputeResource, ResourceRequirementsExt, ResourceRequirementsType,
+            LIMIT_REQUEST_RATIO_CPU, LIMIT_REQUEST_RATIO_MEMORY,
+        },
+    },
+    error::{Error, OperatorResult},
+};
+
 /// A builder to build [`Pod`] or [`PodTemplateSpec`] objects.
 #[derive(Clone, Default)]
 pub struct PodBuilder {
@@ -68,10 +73,10 @@ impl PodBuilder {
 
     pub fn metadata_builder<F>(&mut self, f: F) -> &mut Self
     where
-        F: Fn(&mut ObjectMetaBuilder) -> &mut ObjectMetaBuilder,
+        F: Fn(ObjectMetaBuilder) -> ObjectMetaBuilder,
     {
-        let mut builder = ObjectMetaBuilder::new();
-        let builder = f(&mut builder);
+        let builder = ObjectMetaBuilder::new();
+        let builder = f(builder);
         self.metadata = Some(builder.build());
         self
     }
@@ -187,22 +192,16 @@ impl PodBuilder {
         // instead be set inside the container builder. Having container types
         // should greatly simplify setting the default resource requirements.
         // This method should instead be "as dumb" as it can be and should
-        // simply add the provided container to the internal vector. The problem
-        // with the solution down below is that wie side-step the common
-        // interface provided by the `with_resource`, `with_cpu` and
-        // `with_memory` methods of the builder.
-
-        if container.resources.is_none() {
-            let limits = Some(BTreeMap::from([
-                ("cpu".to_string(), Quantity("10m".to_string())),
-                ("memory".to_string(), Quantity("128Mi".to_string())),
-            ]));
-            container.resources = Some(ResourceRequirements {
-                limits: limits.clone(),
-                requests: limits,
-                ..ResourceRequirements::default()
-            });
-        }
+        // simply add the provided container to the internal vector.
+
+        container.resources = container.resources.or(Some(
+            ResourceRequirementsBuilder::new()
+                .with_cpu_request("10m")
+                .with_cpu_limit("10m")
+                .with_memory_request("128Mi")
+                .with_memory_limit("128Mi")
+                .build(),
+        ));
 
         self.init_containers
             .get_or_insert_with(Vec::new)
diff --git a/src/builder/pod/resources.rs b/src/builder/pod/resources.rs
index d8dccf267..486749018 100644
--- a/src/builder/pod/resources.rs
+++ b/src/builder/pod/resources.rs
@@ -6,12 +6,10 @@ use k8s_openapi::{
 use tracing::warn;
 
 use crate::{
-    commons::resources::ResourceRequirementsType, cpu::CpuQuantity, error::OperatorResult,
-    memory::MemoryQuantity,
+    commons::resources::ResourceRequirementsType, constants::resources::RESOURCE_DENYLIST,
+    cpu::CpuQuantity, error::OperatorResult, memory::MemoryQuantity,
 };
 
-const RESOURCE_DENYLIST: &[&str] = &["cpu", "memory"];
-
 #[derive(Debug, Default)]
 pub struct ResourceRequirementsBuilder<CR, CL, MR, ML> {
     other: BTreeMap<String, BTreeMap<ResourceRequirementsType, Quantity>>,
@@ -230,17 +228,8 @@ impl ResourceRequirementsBuilder<Quantity, Quantity, Quantity, Quantity> {
         }
 
         // Only add limits/requests when there is actually stuff to add
-        let limits = if limits.is_empty() {
-            None
-        } else {
-            Some(limits)
-        };
-
-        let requests = if requests.is_empty() {
-            None
-        } else {
-            Some(requests)
-        };
+        let requests = (!requests.is_empty()).then_some(requests);
+        let limits = (!limits.is_empty()).then_some(limits);
 
         ResourceRequirements {
             limits,
diff --git a/src/builder/pod/volume.rs b/src/builder/pod/volume.rs
index d565ddc33..1b6703fa0 100644
--- a/src/builder/pod/volume.rs
+++ b/src/builder/pod/volume.rs
@@ -1,18 +1,17 @@
-use k8s_openapi::api::core::v1::{
-    EphemeralVolumeSource, PersistentVolumeClaimSpec, PersistentVolumeClaimTemplate,
-    ResourceRequirements, VolumeMount,
-};
 use k8s_openapi::{
     api::core::v1::{
         CSIVolumeSource, ConfigMapVolumeSource, DownwardAPIVolumeSource, EmptyDirVolumeSource,
-        HostPathVolumeSource, PersistentVolumeClaimVolumeSource, ProjectedVolumeSource,
-        SecretVolumeSource, Volume,
+        EphemeralVolumeSource, HostPathVolumeSource, PersistentVolumeClaimSpec,
+        PersistentVolumeClaimTemplate, PersistentVolumeClaimVolumeSource, ProjectedVolumeSource,
+        ResourceRequirements, SecretVolumeSource, Volume, VolumeMount,
     },
     apimachinery::pkg::api::resource::Quantity,
 };
-use std::collections::BTreeMap;
 
-use crate::builder::ObjectMetaBuilder;
+use crate::{
+    builder::{kvp::AnnotationListBuilder, ObjectMetaBuilder},
+    types::{Annotation, AnnotationParseError, KeyValuePairExt},
+};
 
 /// A builder to build [`Volume`] objects.
 /// May only contain one `volume_source` at a time.
@@ -306,10 +305,9 @@ impl SecretOperatorVolumeSourceBuilder {
     }
 
     pub fn build(&self) -> EphemeralVolumeSource {
-        let mut attrs = BTreeMap::from([(
-            "secrets.stackable.tech/class".to_string(),
-            self.secret_class.clone(),
-        )]);
+        // TODO (Techassi): Get rid of all unwraps
+        let mut attrs = AnnotationListBuilder::new(Some("secrets.stackable.tech"));
+        attrs.add("class", &self.secret_class).unwrap();
 
         if !self.scopes.is_empty() {
             let mut scopes = String::new();
@@ -326,26 +324,25 @@ impl SecretOperatorVolumeSourceBuilder {
                     }
                 }
             }
-            attrs.insert("secrets.stackable.tech/scope".to_string(), scopes);
+            attrs.add("scope", &scopes).unwrap();
         }
 
         if let Some(format) = &self.format {
-            attrs.insert(
-                "secrets.stackable.tech/format".to_string(),
-                format.as_ref().to_string(),
-            );
+            attrs.add("format", format.as_ref()).unwrap();
         }
 
         if !self.kerberos_service_names.is_empty() {
-            attrs.insert(
-                "secrets.stackable.tech/kerberos.service.names".to_string(),
-                self.kerberos_service_names.join(","),
-            );
+            attrs
+                .add(
+                    "kerberos.service.names",
+                    &self.kerberos_service_names.join(","),
+                )
+                .unwrap();
         }
 
         EphemeralVolumeSource {
             volume_claim_template: Some(PersistentVolumeClaimTemplate {
-                metadata: Some(ObjectMetaBuilder::new().annotations(attrs).build()),
+                metadata: Some(ObjectMetaBuilder::new().annotations(attrs.build()).build()),
                 spec: PersistentVolumeClaimSpec {
                     storage_class_name: Some("secrets.stackable.tech".to_string()),
                     resources: Some(ResourceRequirements {
@@ -390,16 +387,14 @@ pub enum ListenerReference {
 
 impl ListenerReference {
     /// Return the key and value for a Kubernetes object annotation
-    fn to_annotation(&self) -> (String, String) {
+    fn to_annotation(&self) -> Result<Annotation, AnnotationParseError> {
         match self {
-            ListenerReference::ListenerClass(value) => (
-                "listeners.stackable.tech/listener-class".into(),
-                value.into(),
-            ),
-            ListenerReference::ListenerName(value) => (
-                "listeners.stackable.tech/listener-name".into(),
-                value.into(),
-            ),
+            ListenerReference::ListenerClass(value) => {
+                Annotation::new(Some("listeners.stackable.tech"), "listener-class", value)
+            }
+            ListenerReference::ListenerName(value) => {
+                Annotation::new(Some("listeners.stackable.tech"), "listener-name", value)
+            }
         }
     }
 }
@@ -445,11 +440,13 @@ impl ListenerOperatorVolumeSourceBuilder {
 
     /// Build an [`EphemeralVolumeSource`] from the builder
     pub fn build(&self) -> EphemeralVolumeSource {
+        let annotation = self.listener_reference.to_annotation().unwrap();
+
         EphemeralVolumeSource {
             volume_claim_template: Some(PersistentVolumeClaimTemplate {
                 metadata: Some(
                     ObjectMetaBuilder::new()
-                        .annotations([self.listener_reference.to_annotation()].into())
+                        .annotations([(annotation.key(), annotation)].into())
                         .build(),
                 ),
                 spec: PersistentVolumeClaimSpec {
diff --git a/src/cli.rs b/src/cli.rs
index d5c203576..8a70ea438 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -116,8 +116,6 @@ use std::{
     path::{Path, PathBuf},
 };
 
-pub const AUTHOR: &str = "Stackable GmbH - info@stackable.de";
-
 /// Framework-standardized commands
 ///
 /// If you need operator-specific commands then you can flatten [`Command`] into your own command enum. For example:
diff --git a/src/cluster_resources.rs b/src/cluster_resources.rs
index 422e88123..272d0ecae 100644
--- a/src/cluster_resources.rs
+++ b/src/cluster_resources.rs
@@ -1,6 +1,15 @@
 //! A structure containing the cluster resources.
+use std::{
+    collections::{BTreeMap, HashSet},
+    fmt::Debug,
+};
+
+use serde::{de::DeserializeOwned, Serialize};
+use strum::Display;
+use tracing::{debug, info, warn};
 
 use crate::{
+    builder::kvp::LabelListBuilder,
     client::{Client, GetApi},
     commons::{
         cluster_operation::ClusterOperation,
@@ -9,6 +18,11 @@ use crate::{
             LIMIT_REQUEST_RATIO_CPU, LIMIT_REQUEST_RATIO_MEMORY,
         },
     },
+    constants::labels::{
+        LABEL_KEY_APP_INSTANCE, LABEL_KEY_APP_MANAGED_BY, LABEL_KEY_APP_NAME,
+        LABEL_KEY_NAME_APP_INSTANCE, LABEL_KEY_NAME_APP_MANAGED_BY, LABEL_KEY_NAME_APP_NAME,
+        LABEL_KEY_PREFIX_APP_KUBERNETES,
+    },
     error::{Error, OperatorResult},
     k8s_openapi::{
         api::{
@@ -23,20 +37,11 @@ use crate::{
         apimachinery::pkg::apis::meta::v1::{LabelSelector, LabelSelectorRequirement},
         NamespaceResourceScope,
     },
-    kube::{Resource, ResourceExt},
-    labels::{APP_INSTANCE_LABEL, APP_MANAGED_BY_LABEL, APP_NAME_LABEL},
+    kube::{core::ErrorResponse, Resource, ResourceExt},
+    types::{Label, LabelParseError},
     utils::format_full_controller_name,
 };
 
-use kube::core::ErrorResponse;
-use serde::{de::DeserializeOwned, Serialize};
-use std::{
-    collections::{BTreeMap, HashSet},
-    fmt::Debug,
-};
-use strum::Display;
-use tracing::{debug, info, warn};
-
 #[cfg(doc)]
 use crate::k8s_openapi::api::{
     apps::v1::Deployment,
@@ -402,17 +407,14 @@ impl ClusterResources {
 
     /// Return required labels for cluster resources to be uniquely identified for clean up.
     // TODO: This is a (quick-fix) helper method but should be replaced by better label handling
-    pub fn get_required_labels(&self) -> BTreeMap<String, String> {
-        vec![
-            (
-                APP_INSTANCE_LABEL.to_string(),
-                self.app_instance.to_string(),
-            ),
-            (APP_MANAGED_BY_LABEL.to_string(), self.manager.to_string()),
-            (APP_NAME_LABEL.to_string(), self.app_name.to_string()),
-        ]
-        .into_iter()
-        .collect()
+    pub fn get_required_labels(&self) -> Result<BTreeMap<String, Label>, LabelParseError> {
+        let mut labels = LabelListBuilder::new(Some(LABEL_KEY_PREFIX_APP_KUBERNETES));
+
+        labels.add(LABEL_KEY_NAME_APP_INSTANCE, &self.app_instance)?;
+        labels.add(LABEL_KEY_NAME_APP_MANAGED_BY, &self.manager)?;
+        labels.add(LABEL_KEY_NAME_APP_NAME, &self.app_name)?;
+
+        Ok(labels.build())
     }
 
     /// Adds a resource to the cluster resources.
@@ -441,7 +443,11 @@ impl ClusterResources {
     ) -> OperatorResult<T> {
         Self::check_labels(
             resource.labels(),
-            &[APP_INSTANCE_LABEL, APP_MANAGED_BY_LABEL, APP_NAME_LABEL],
+            &[
+                LABEL_KEY_APP_INSTANCE,
+                LABEL_KEY_APP_MANAGED_BY,
+                LABEL_KEY_APP_NAME,
+            ],
             &[&self.app_instance, &self.manager, &self.app_name],
         )?;
 
@@ -661,17 +667,17 @@ impl ClusterResources {
         let label_selector = LabelSelector {
             match_expressions: Some(vec![
                 LabelSelectorRequirement {
-                    key: APP_INSTANCE_LABEL.into(),
+                    key: LABEL_KEY_APP_INSTANCE.into(),
                     operator: "In".into(),
                     values: Some(vec![self.app_instance.to_owned()]),
                 },
                 LabelSelectorRequirement {
-                    key: APP_NAME_LABEL.into(),
+                    key: LABEL_KEY_APP_NAME.into(),
                     operator: "In".into(),
                     values: Some(vec![self.app_name.to_owned()]),
                 },
                 LabelSelectorRequirement {
-                    key: APP_MANAGED_BY_LABEL.into(),
+                    key: LABEL_KEY_APP_MANAGED_BY.into(),
                     operator: "In".into(),
                     values: Some(vec![self.manager.to_owned()]),
                 },
diff --git a/src/commons/affinity.rs b/src/commons/affinity.rs
index f5ca97be3..8fdc3d0dd 100644
--- a/src/commons/affinity.rs
+++ b/src/commons/affinity.rs
@@ -12,11 +12,23 @@ use serde::{Deserialize, Serialize};
 use stackable_operator_derive::Fragment;
 
 use crate::{
+    builder::kvp::LabelListBuilder,
     config::merge::{Atomic, Merge},
-    labels::{APP_COMPONENT_LABEL, APP_INSTANCE_LABEL, APP_NAME_LABEL},
+    constants::{
+        affinity::TOPOLOGY_KEY_HOSTNAME,
+        labels::{
+            LABEL_KEY_NAME_APP_COMPONENT, LABEL_KEY_NAME_APP_INSTANCE, LABEL_KEY_NAME_APP_NAME,
+            LABEL_KEY_PREFIX_APP_KUBERNETES,
+        },
+    },
+    types::KeyValuePairParseError,
 };
 
-pub const TOPOLOGY_KEY_HOSTNAME: &str = "kubernetes.io/hostname";
+#[derive(Debug, thiserror::Error)]
+pub enum StackableAffinityError {
+    #[error("key/value pair error")]
+    KeyValuePairError(#[from] KeyValuePairParseError),
+}
 
 #[derive(Clone, Debug, Default, Deserialize, Fragment, JsonSchema, PartialEq, Serialize)]
 #[fragment(path_overrides(fragment = "crate::config::fragment"))]
@@ -132,24 +144,25 @@ pub fn affinity_between_role_pods(
     cluster_name: &str,
     role: &str,
     weight: i32,
-) -> WeightedPodAffinityTerm {
-    WeightedPodAffinityTerm {
+) -> Result<WeightedPodAffinityTerm, StackableAffinityError> {
+    let mut builder = LabelListBuilder::new(Some(LABEL_KEY_PREFIX_APP_KUBERNETES));
+    builder.add(LABEL_KEY_NAME_APP_NAME, app_name)?;
+    builder.add(LABEL_KEY_NAME_APP_INSTANCE, cluster_name)?;
+    builder.add(LABEL_KEY_NAME_APP_COMPONENT, role)?;
+    let match_labels = builder.build_raw();
+
+    Ok(WeightedPodAffinityTerm {
         pod_affinity_term: PodAffinityTerm {
             label_selector: Some(LabelSelector {
                 match_expressions: None,
-                match_labels: Some(BTreeMap::from([
-                    (APP_NAME_LABEL.to_string(), app_name.to_string()),
-                    (APP_INSTANCE_LABEL.to_string(), cluster_name.to_string()),
-                    (APP_COMPONENT_LABEL.to_string(), role.to_string()),
-                    // We don't include the role-group label here, as the affinity should be between all rolegroups of the given role
-                ])),
+                match_labels: Some(match_labels),
             }),
             namespace_selector: None,
             namespaces: None,
             topology_key: TOPOLOGY_KEY_HOSTNAME.to_string(),
         },
         weight,
-    }
+    })
 }
 
 /// Creates a `WeightedPodAffinityTerm`, which expresses a affinity towards all Pods of the given product (`app_name`) instance (`cluster_name`).
@@ -159,22 +172,24 @@ pub fn affinity_between_cluster_pods(
     app_name: &str,
     cluster_name: &str,
     weight: i32,
-) -> WeightedPodAffinityTerm {
-    WeightedPodAffinityTerm {
+) -> Result<WeightedPodAffinityTerm, StackableAffinityError> {
+    let mut builder = LabelListBuilder::new(Some(LABEL_KEY_PREFIX_APP_KUBERNETES));
+    builder.add(LABEL_KEY_NAME_APP_NAME, app_name)?;
+    builder.add(LABEL_KEY_NAME_APP_INSTANCE, cluster_name)?;
+    let match_labels = builder.build_raw();
+
+    Ok(WeightedPodAffinityTerm {
         pod_affinity_term: PodAffinityTerm {
             label_selector: Some(LabelSelector {
                 match_expressions: None,
-                match_labels: Some(BTreeMap::from([
-                    (APP_NAME_LABEL.to_string(), app_name.to_string()),
-                    (APP_INSTANCE_LABEL.to_string(), cluster_name.to_string()),
-                ])),
+                match_labels: Some(match_labels),
             }),
             namespace_selector: None,
             namespaces: None,
             topology_key: TOPOLOGY_KEY_HOSTNAME.to_string(),
         },
         weight,
-    }
+    })
 }
 
 #[cfg(test)]
@@ -191,11 +206,14 @@ mod tests {
 
     #[test]
     fn test_affinity_merge_new_attributes() {
+        let affinity_between_role_pods =
+            affinity_between_role_pods("kafka", "simple-kafka", "broker", 70).unwrap();
+
         let default_affinity = StackableAffinityFragment {
             pod_affinity: None,
             pod_anti_affinity: Some(PodAntiAffinity {
                 preferred_during_scheduling_ignored_during_execution: Some(vec![
-                    affinity_between_role_pods("kafka", "simple-kafka", "broker", 70),
+                    affinity_between_role_pods,
                 ]),
                 required_during_scheduling_ignored_during_execution: None,
             }),
@@ -304,11 +322,14 @@ mod tests {
 
     #[test]
     fn test_affinity_merge_overwrite_existing_attribute() {
+        let affinity_between_role_pods =
+            affinity_between_role_pods("kafka", "simple-kafka", "broker", 70).unwrap();
+
         let default_affinity = StackableAffinityFragment {
             pod_affinity: None,
             pod_anti_affinity: Some(PodAntiAffinity {
                 preferred_during_scheduling_ignored_during_execution: Some(vec![
-                    affinity_between_role_pods("kafka", "simple-kafka", "broker", 70),
+                    affinity_between_role_pods,
                 ]),
                 required_during_scheduling_ignored_during_execution: None,
             }),
@@ -374,7 +395,7 @@ mod tests {
         let cluster_name = "simple-kafka";
         let role = "broker";
 
-        let anti_affinity = affinity_between_role_pods(app_name, cluster_name, role, 70);
+        let anti_affinity = affinity_between_role_pods(app_name, cluster_name, role, 70).unwrap();
         assert_eq!(
             anti_affinity,
             WeightedPodAffinityTerm {
@@ -407,7 +428,7 @@ mod tests {
         let app_name = "kafka";
         let cluster_name = "simple-kafka";
 
-        let anti_affinity = affinity_between_cluster_pods(app_name, cluster_name, 20);
+        let anti_affinity = affinity_between_cluster_pods(app_name, cluster_name, 20).unwrap();
         assert_eq!(
             anti_affinity,
             WeightedPodAffinityTerm {
diff --git a/src/commons/rbac.rs b/src/commons/rbac.rs
index b0f51f8a1..ed0c67ee4 100644
--- a/src/commons/rbac.rs
+++ b/src/commons/rbac.rs
@@ -1,17 +1,24 @@
-use crate::builder::ObjectMetaBuilder;
-use crate::error::OperatorResult;
-use crate::k8s_openapi::api::core::v1::ServiceAccount;
-use crate::k8s_openapi::api::rbac::v1::{RoleBinding, RoleRef, Subject};
-use kube::{Resource, ResourceExt};
 use std::collections::BTreeMap;
 
+use kube::{Resource, ResourceExt};
+
+use crate::{
+    builder::ObjectMetaBuilder,
+    error::OperatorResult,
+    k8s_openapi::api::{
+        core::v1::ServiceAccount,
+        rbac::v1::{RoleBinding, RoleRef, Subject},
+    },
+    types::Label,
+};
+
 /// Build RBAC objects for the product workloads.
 /// The `rbac_prefix` is meant to be the product name, for example: zookeeper, airflow, etc.
 /// and it is a assumed that a ClusterRole named `{rbac_prefix}-clusterrole` exists.
 pub fn build_rbac_resources<T: Clone + Resource<DynamicType = ()>>(
     resource: &T,
     rbac_prefix: &str,
-    labels: BTreeMap<String, String>,
+    labels: BTreeMap<String, Label>,
 ) -> OperatorResult<(ServiceAccount, RoleBinding)> {
     let sa_name = service_account_name(rbac_prefix);
     let service_account = ServiceAccount {
diff --git a/src/constants/affinity.rs b/src/constants/affinity.rs
new file mode 100644
index 000000000..e90224999
--- /dev/null
+++ b/src/constants/affinity.rs
@@ -0,0 +1 @@
+pub const TOPOLOGY_KEY_HOSTNAME: &str = "kubernetes.io/hostname";
diff --git a/src/constants/cli.rs b/src/constants/cli.rs
new file mode 100644
index 000000000..e40acc90c
--- /dev/null
+++ b/src/constants/cli.rs
@@ -0,0 +1 @@
+pub const AUTHOR: &str = "Stackable GmbH - info@stackable.de";
diff --git a/src/constants/labels.rs b/src/constants/labels.rs
new file mode 100644
index 000000000..7c6e06c92
--- /dev/null
+++ b/src/constants/labels.rs
@@ -0,0 +1,50 @@
+/// The label key prefix used for Kubernetes apps
+pub const LABEL_KEY_PREFIX_APP_KUBERNETES: &str = "app.kubernetes.io";
+
+/// The label key name identifying the tool used to manage the operation of an
+/// application, e.g. "helm"
+pub const LABEL_KEY_NAME_APP_MANAGED_BY: &str = "managed-by";
+
+pub const LABEL_KEY_NAME_APP_ROLE_GROUP: &str = "role-group";
+
+/// The label key name identifying the application component within the
+/// architecture, e.g. "database"
+pub const LABEL_KEY_NAME_APP_COMPONENT: &str = "component";
+
+/// The label key name identifying the application instance, e.g. "mysql-abcxzy"
+pub const LABEL_KEY_NAME_APP_INSTANCE: &str = "instance";
+
+/// The label key name identifying the application version, e.g. a semantic
+/// version, revision hash, etc, like "5.7.21"
+pub const LABEL_KEY_NAME_APP_VERSION: &str = "version";
+
+/// The label key name identifying the higher level application this app is part
+/// of, e.g. "wordpress".
+pub const LABEL_KEY_NAME_APP_PART_OF: &str = "part-of";
+
+/// The label key name identifying the application name e.g. "mysql"
+pub const LABEL_KEY_NAME_APP_NAME: &str = "name";
+
+// COMBINATIONS //
+
+/// The label key identifying the tool used to manage the operation of an
+/// application: `app.kubernetes.io/managed-by`
+pub const LABEL_KEY_APP_MANAGED_BY: &str = const_format::concatcp!(
+    LABEL_KEY_PREFIX_APP_KUBERNETES,
+    "/",
+    LABEL_KEY_NAME_APP_MANAGED_BY
+);
+
+/// The label key identifying the application instance: `app.kubernetes.io/instance`
+pub const LABEL_KEY_APP_INSTANCE: &str = const_format::concatcp!(
+    LABEL_KEY_PREFIX_APP_KUBERNETES,
+    "/",
+    LABEL_KEY_NAME_APP_INSTANCE
+);
+
+/// The label key indentifying the application name: `app.kubernetes.io/name`
+pub const LABEL_KEY_APP_NAME: &str = const_format::concatcp!(
+    LABEL_KEY_PREFIX_APP_KUBERNETES,
+    "/",
+    LABEL_KEY_NAME_APP_NAME
+);
diff --git a/src/constants/mod.rs b/src/constants/mod.rs
new file mode 100644
index 000000000..3e0716fe7
--- /dev/null
+++ b/src/constants/mod.rs
@@ -0,0 +1,5 @@
+pub mod affinity;
+pub mod cli;
+pub mod labels;
+pub mod resources;
+pub mod validation;
diff --git a/src/constants/resources.rs b/src/constants/resources.rs
new file mode 100644
index 000000000..2690e8f46
--- /dev/null
+++ b/src/constants/resources.rs
@@ -0,0 +1 @@
+pub(crate) const RESOURCE_DENYLIST: &[&str] = &["cpu", "memory"];
diff --git a/src/constants/validation.rs b/src/constants/validation.rs
new file mode 100644
index 000000000..49fc8f541
--- /dev/null
+++ b/src/constants/validation.rs
@@ -0,0 +1,18 @@
+use const_format::concatcp;
+
+pub(crate) const RFC_1123_LABEL_FMT: &str = "[a-z0-9]([-a-z0-9]*[a-z0-9])?";
+pub(crate) const RFC_1123_SUBDOMAIN_FMT: &str =
+    concatcp!(RFC_1123_LABEL_FMT, "(\\.", RFC_1123_LABEL_FMT, ")*");
+pub(crate) const RFC_1123_SUBDOMAIN_ERROR_MSG: &str = "a lowercase RFC 1123 subdomain must consist of lower case alphanumeric characters, '-' or '.', and must start and end with an alphanumeric character";
+pub(crate) const RFC_1123_LABEL_ERROR_MSG: &str = "a lowercase RFC 1123 label must consist of lower case alphanumeric characters, '-' or '.', and must start and end with an alphanumeric character";
+
+// This is a subdomain's max length in DNS (RFC 1123)
+pub const RFC_1123_SUBDOMAIN_MAX_LENGTH: usize = 253;
+// Minimal length reuquired by RFC 1123 is 63. Up to 255 allowed, unsupported by k8s.
+pub const RFC_1123_LABEL_MAX_LENGTH: usize = 63;
+
+pub(crate) const RFC_1035_LABEL_FMT: &str = "[a-z]([-a-z0-9]*[a-z0-9])?";
+pub(crate) const RFC_1035_LABEL_ERR_MSG: &str = "a DNS-1035 label must consist of lower case alphanumeric characters or '-', start with an alphabetic character, and end with an alphanumeric character";
+
+// This is a label's max length in DNS (RFC 1035)
+pub const RFC_1035_LABEL_MAX_LENGTH: usize = RFC_1123_LABEL_MAX_LENGTH;
diff --git a/src/labels.rs b/src/labels.rs
index 17d75c6ca..a81f2fdbb 100644
--- a/src/labels.rs
+++ b/src/labels.rs
@@ -1,5 +1,14 @@
-use crate::utils::format_full_controller_name;
-use const_format::concatcp;
+use crate::{
+    builder::kvp::LabelListBuilder,
+    constants::labels::{
+        LABEL_KEY_NAME_APP_COMPONENT, LABEL_KEY_NAME_APP_INSTANCE, LABEL_KEY_NAME_APP_MANAGED_BY,
+        LABEL_KEY_NAME_APP_NAME, LABEL_KEY_NAME_APP_ROLE_GROUP, LABEL_KEY_NAME_APP_VERSION,
+        LABEL_KEY_PREFIX_APP_KUBERNETES,
+    },
+    types::{Label, LabelParseError},
+    utils::format_full_controller_name,
+};
+
 use kube::api::{Resource, ResourceExt};
 use std::collections::BTreeMap;
 
@@ -8,22 +17,6 @@ use crate::builder::ObjectMetaBuilder;
 #[cfg(doc)]
 use crate::commons::product_image_selection::ResolvedProductImage;
 
-const APP_KUBERNETES_LABEL_BASE: &str = "app.kubernetes.io/";
-
-/// The name of the application e.g. "mysql"
-pub const APP_NAME_LABEL: &str = concatcp!(APP_KUBERNETES_LABEL_BASE, "name");
-/// A unique name identifying the instance of an application e.g. "mysql-abcxzy"
-pub const APP_INSTANCE_LABEL: &str = concatcp!(APP_KUBERNETES_LABEL_BASE, "instance");
-/// The current version of the application (e.g., a semantic version, revision hash, etc.) e.g."5.7.21"
-pub const APP_VERSION_LABEL: &str = concatcp!(APP_KUBERNETES_LABEL_BASE, "version");
-/// The component within the architecture e.g. database
-pub const APP_COMPONENT_LABEL: &str = concatcp!(APP_KUBERNETES_LABEL_BASE, "component");
-/// The name of a higher level application this one is part of e.g. "wordpress"
-pub const APP_PART_OF_LABEL: &str = concatcp!(APP_KUBERNETES_LABEL_BASE, "part-of");
-/// The tool being used to manage the operation of an application e.g. helm
-pub const APP_MANAGED_BY_LABEL: &str = concatcp!(APP_KUBERNETES_LABEL_BASE, "managed-by");
-pub const APP_ROLE_GROUP_LABEL: &str = concatcp!(APP_KUBERNETES_LABEL_BASE, "role-group");
-
 /// Recommended labels to set on objects created by Stackable operators
 ///
 /// See [`get_recommended_labels`] and [`ObjectMetaBuilder::with_recommended_labels`].
@@ -65,56 +58,23 @@ pub fn get_recommended_labels<T>(
         role,
         role_group,
     }: ObjectLabels<T>,
-) -> BTreeMap<String, String>
+) -> Result<BTreeMap<String, Label>, LabelParseError>
 where
     T: Resource,
 {
-    let mut labels = role_group_selector_labels(owner, app_name, role, role_group);
+    let mut labels = LabelListBuilder::new(Some(LABEL_KEY_PREFIX_APP_KUBERNETES));
+    labels.add(
+        LABEL_KEY_NAME_APP_MANAGED_BY,
+        &format_full_controller_name(operator_name, controller_name),
+    )?;
+    labels.add(LABEL_KEY_NAME_APP_INSTANCE, &owner.name_any())?;
+    labels.add(LABEL_KEY_NAME_APP_ROLE_GROUP, role_group)?;
+    labels.add(LABEL_KEY_NAME_APP_VERSION, app_version)?;
+    labels.add(LABEL_KEY_NAME_APP_COMPONENT, role)?;
+    labels.add(LABEL_KEY_NAME_APP_NAME, app_name)?;
 
     // TODO: Add operator version label
     // TODO: part-of is empty for now, decide on how this can be used in a proper fashion
-    labels.insert(APP_VERSION_LABEL.to_string(), app_version.to_string());
-    labels.insert(
-        APP_MANAGED_BY_LABEL.to_string(),
-        format_full_controller_name(operator_name, controller_name),
-    );
-
-    labels
-}
-
-/// The labels required to match against objects of a certain role, assuming that those objects
-/// are defined using [`get_recommended_labels`]
-pub fn role_group_selector_labels<T: Resource>(
-    owner: &T,
-    app_name: &str,
-    role: &str,
-    role_group: &str,
-) -> BTreeMap<String, String> {
-    let mut labels = role_selector_labels(owner, app_name, role);
-    labels.insert(APP_ROLE_GROUP_LABEL.to_string(), role_group.to_string());
-    labels
-}
-
-/// The labels required to match against objects of a certain role group, assuming that those objects
-/// are defined using [`get_recommended_labels`]
-pub fn role_selector_labels<T: Resource>(
-    owner: &T,
-    app_name: &str,
-    role: &str,
-) -> BTreeMap<String, String> {
-    let mut labels = build_common_labels_for_all_managed_resources(app_name, &owner.name_any());
-    labels.insert(APP_COMPONENT_LABEL.to_string(), role.to_string());
-    labels
-}
 
-/// The APP_NAME_LABEL (Spark, Kafka, ZooKeeper...) and APP_INSTANCES_LABEL (simple, test ...) are
-/// required to identify resources that belong to a certain owner object (such as a particular `ZookeeperCluster`).
-pub fn build_common_labels_for_all_managed_resources(
-    app_name: &str,
-    owner_name: &str,
-) -> BTreeMap<String, String> {
-    let mut labels = BTreeMap::new();
-    labels.insert(APP_NAME_LABEL.to_string(), app_name.to_string());
-    labels.insert(APP_INSTANCE_LABEL.to_string(), owner_name.to_string());
-    labels
+    Ok(labels.build())
 }
diff --git a/src/lib.rs b/src/lib.rs
index 73b26be83..b3255307c 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -4,6 +4,7 @@ pub mod client;
 pub mod cluster_resources;
 pub mod commons;
 pub mod config;
+pub mod constants;
 pub mod cpu;
 pub mod crd;
 pub mod error;
@@ -18,6 +19,7 @@ pub mod product_config_utils;
 pub mod product_logging;
 pub mod role_utils;
 pub mod status;
+pub mod types;
 pub mod utils;
 pub mod validation;
 pub mod yaml;
diff --git a/src/types/annotation.rs b/src/types/annotation.rs
new file mode 100644
index 000000000..855b97d73
--- /dev/null
+++ b/src/types/annotation.rs
@@ -0,0 +1,85 @@
+use std::{fmt::Display, str::FromStr};
+
+use crate::types::{KeyParseError, KeyValuePair, KeyValuePairExt, KeyValuePairParseError};
+
+pub type AnnotationParseError = KeyValuePairParseError;
+pub type AnnotationKeyParseError = KeyParseError;
+
+/// Annotations are used to attach arbitrary non-identifying metadata to objects,
+/// like pods. They are modeled after the following [specs][1]. It is highly
+/// recommended to use [`Annotation::new`] to create a new label. This method
+/// ensures that no maximum length restrictions are violated. [`Annotation`]
+/// also implements [`FromStr`], which allows parsing a annotation from a
+/// string.
+///
+/// Additionally, [`Annotation`] implements [`Display`], which formats the
+/// label using the following format: `(<prefix>/)<name>=<value>`.
+///
+/// ### Examples
+///
+/// ```
+/// use stackable_operator::types::{Annotation, KeyValuePairExt};
+/// use std::str::FromStr;
+///
+/// let label = Annotation::new(Some("stackable.tech"), "node", "1");
+/// let label = Annotation::from_str("stackable.tech/node=1").unwrap();
+/// let label = Annotation::try_from("stackable.tech/node=1").unwrap();
+/// ```
+///
+/// [1]: https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/
+#[derive(Debug, Clone)]
+pub struct Annotation(KeyValuePair);
+
+impl KeyValuePairExt for Annotation {
+    /// Creates a new label (key/value pair). The key consists of an optional
+    /// `prefix` and a name.
+    ///
+    /// ```
+    /// use stackable_operator::types::{Annotation, KeyValuePairExt};
+    ///
+    /// // stackable.tech/node=1
+    /// let annotattion = Annotation::new(Some("stackable.tech"), "node", "1");
+    /// ```
+    fn new<T>(prefix: Option<T>, name: T, value: T) -> Result<Self, AnnotationParseError>
+    where
+        T: Into<String>,
+    {
+        let kvp = KeyValuePair::new(prefix, name, value)?;
+        Ok(Self(kvp))
+    }
+
+    /// Returns the annotation key as a formatted [`String`]. If the key contains
+    /// a prefix, the key has a format like `<prefix>/<name>`. If not, the key
+    /// only consists of a name.
+    fn key(&self) -> String {
+        self.0.key()
+    }
+
+    /// Returns the annotation value as a [`String`].
+    fn value(&self) -> &String {
+        self.0.value()
+    }
+}
+
+impl FromStr for Annotation {
+    type Err = AnnotationParseError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        let kvp = KeyValuePair::from_str(input)?;
+        Ok(Self(kvp))
+    }
+}
+
+impl TryFrom<&str> for Annotation {
+    type Error = AnnotationParseError;
+
+    fn try_from(input: &str) -> Result<Self, Self::Error> {
+        Self::from_str(input)
+    }
+}
+
+impl Display for Annotation {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
diff --git a/src/types/kvp.rs b/src/types/kvp.rs
new file mode 100644
index 000000000..65abb6bfd
--- /dev/null
+++ b/src/types/kvp.rs
@@ -0,0 +1,271 @@
+use std::{fmt::Display, str::FromStr};
+
+use crate::validation;
+
+#[derive(Debug, PartialEq, thiserror::Error)]
+pub enum KeyValuePairParseError {
+    #[error("invalid key/value pair syntax, expected '(prefix/)name=value'")]
+    InvalidSyntax,
+
+    #[error("key/value pair input cannot be empty")]
+    EmptyInput,
+
+    #[error("key/value pair key parse error")]
+    KeyParseError(#[from] KeyParseError),
+}
+
+pub trait KeyValuePairExt: Sized + Clone + FromStr + Display {
+    /// Creates a new key/value pair. The key consists of an optional `prefix`
+    /// and a `name`.
+    ///
+    /// ```
+    /// use stackable_operator::types::{KeyValuePair, KeyValuePairExt};
+    ///
+    /// // stackable.tech/release=23.7
+    /// let kvp = KeyValuePair::new(Some("stackable.tech"), "release", "23.7");
+    /// ```
+    fn new<T>(prefix: Option<T>, name: T, value: T) -> Result<Self, KeyValuePairParseError>
+    where
+        T: Into<String>;
+
+    fn key(&self) -> String;
+    fn value(&self) -> &String;
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct KeyValuePair {
+    key: Key,
+    value: String,
+}
+
+impl FromStr for KeyValuePair {
+    type Err = KeyValuePairParseError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        let input = input.trim();
+
+        if input.is_empty() {
+            return Err(KeyValuePairParseError::EmptyInput);
+        }
+
+        let parts: Vec<_> = input.split('=').collect();
+
+        if parts.len() != 2 {
+            return Err(KeyValuePairParseError::InvalidSyntax);
+        }
+
+        if parts[0].is_empty() || parts[1].is_empty() {
+            return Err(KeyValuePairParseError::InvalidSyntax);
+        }
+
+        Ok(Self {
+            key: Key::from_str(parts[0])?,
+            value: parts[1].to_string(),
+        })
+    }
+}
+
+impl TryFrom<&str> for KeyValuePair {
+    type Error = KeyValuePairParseError;
+
+    fn try_from(input: &str) -> Result<Self, Self::Error> {
+        Self::from_str(input)
+    }
+}
+
+impl Display for KeyValuePair {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}={}", self.key, self.value)
+    }
+}
+
+impl KeyValuePairExt for KeyValuePair {
+    /// Creates a new key/value pair. The key consists of an optional `prefix`
+    /// and a `name`.
+    ///
+    /// ```
+    /// use stackable_operator::types::{KeyValuePair, KeyValuePairExt};
+    ///
+    /// // stackable.tech/release=23.7
+    /// let kvp = KeyValuePair::new(Some("stackable.tech"), "release", "23.7");
+    /// ```
+    fn new<T>(prefix: Option<T>, name: T, value: T) -> Result<Self, KeyValuePairParseError>
+    where
+        T: Into<String>,
+    {
+        let key = Key::new(prefix, name)?;
+        let value = value.into();
+
+        Ok(Self { key, value })
+    }
+
+    fn key(&self) -> String {
+        self.key.to_string()
+    }
+
+    fn value(&self) -> &String {
+        &self.value
+    }
+}
+
+#[derive(Debug, PartialEq, thiserror::Error)]
+pub enum KeyParseError {
+    #[error("invalid key separator count - expected at most '1', got '{0}'")]
+    InvalidSeparatorCount(usize),
+
+    #[error("key name is not a valid RFC1123 label")]
+    InvalidRfc1123LabelError(Vec<String>),
+
+    #[error("key prefix is not a valid RFC1123 subdomain")]
+    InvalidRfc1123SubdomainError(Vec<String>),
+}
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct Key {
+    pub prefix: Option<String>,
+    pub name: String,
+}
+
+impl FromStr for Key {
+    type Err = KeyParseError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        let parts: Vec<_> = input.trim().split('/').collect();
+        let len = parts.len();
+
+        // More then one slash (more than two parts)
+        if len > 2 {
+            return Err(KeyParseError::InvalidSeparatorCount(len - 1));
+        }
+
+        // No prefix, just validate the name segment
+        if len == 1 {
+            validation::is_rfc_1123_label(parts[0])
+                .map_err(KeyParseError::InvalidRfc1123LabelError)?;
+
+            return Ok(Self {
+                prefix: None,
+                name: parts[0].to_string(),
+            });
+        }
+
+        // With prefix, validate both
+        validation::is_rfc_1123_subdomain(parts[0])
+            .map_err(KeyParseError::InvalidRfc1123SubdomainError)?;
+
+        validation::is_rfc_1123_label(parts[1]).map_err(KeyParseError::InvalidRfc1123LabelError)?;
+
+        Ok(Self {
+            prefix: Some(parts[0].to_string()),
+            name: parts[1].to_string(),
+        })
+    }
+}
+
+impl TryFrom<&str> for Key {
+    type Error = KeyParseError;
+
+    fn try_from(input: &str) -> Result<Self, Self::Error> {
+        Self::from_str(input)
+    }
+}
+
+impl Display for Key {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match &self.prefix {
+            Some(prefix) => write!(f, "{}/{}", prefix, self.name),
+            None => write!(f, "{}", self.name),
+        }
+    }
+}
+
+impl Key {
+    pub fn new<T>(prefix: Option<T>, name: T) -> Result<Self, KeyParseError>
+    where
+        T: Into<String>,
+    {
+        let prefix = prefix.map(Into::into);
+        let name = name.into();
+
+        if let Some(prefix) = &prefix {
+            validation::is_rfc_1123_label(prefix)
+                .map_err(KeyParseError::InvalidRfc1123LabelError)?;
+        }
+
+        validation::is_rfc_1123_label(&name).map_err(KeyParseError::InvalidRfc1123LabelError)?;
+
+        Ok(Self { prefix, name })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use rstest::*;
+
+    #[fixture]
+    fn long_prefix<'a>() -> &'a str {
+        "this.is.my.super.long.label.prefix.which.should.break.this.test.case.with.ease.but.this.is.super.tedious.to.write.out.maaaaaaaaaaaaan.this.is.getting.ridiculous.why.are.such.long.domains.even.allowed.who.uses.these.for.real.and.we.are.done.and.reached.the.maximum.length/env=prod"
+    }
+
+    #[fixture]
+    fn long_name<'a>() -> &'a str {
+        "stackable.tech/this.is.a.super.loooooooooong.label.name.which.is.way.too.long.to.be.valid"
+    }
+
+    #[rstest]
+    #[case("stackable.tech/env=prod", "stackable.tech/env", "prod")]
+    #[case("env=prod", "env", "prod")]
+    fn parse_label_valid(#[case] input: &str, #[case] key: &str, #[case] value: &str) {
+        let label = KeyValuePair::from_str(input).unwrap();
+
+        assert_eq!(label.key.to_string(), key.to_string());
+        assert_eq!(label.value, value.to_string());
+        assert_eq!(label.to_string(), input.to_string());
+    }
+
+    #[rstest]
+    #[case("stackable.tech/env=")]
+    #[case("stackable.tech/env")]
+    #[case("env=")]
+    #[case("env")]
+    fn parse_label_missing_value(#[case] input: &str) {
+        let result = KeyValuePair::from_str(input);
+        assert!(result.is_err());
+    }
+
+    #[rstest]
+    #[case("=prod")]
+    #[case("prod")]
+    fn parse_label_missing_key(#[case] input: &str) {
+        let result = KeyValuePair::from_str(input);
+        assert!(result.is_err());
+    }
+
+    #[rstest]
+    #[case("stackable.tech/env=prod=invalid")]
+    #[case("env=prod=invalid")]
+    #[case("=prod=invalid")]
+    fn parse_label_too_many_equal_signs(#[case] input: &str) {
+        let result = KeyValuePair::from_str(input);
+        assert!(result.is_err());
+    }
+
+    #[rstest]
+    fn parse_label_too_long(long_prefix: &str, long_name: &str) {
+        let result = KeyValuePair::from_str(long_prefix);
+        assert!(result.is_err());
+
+        let result = KeyValuePair::from_str(long_name);
+        assert!(result.is_err());
+    }
+
+    #[rstest]
+    #[case(" = ")]
+    #[case("  ")]
+    #[case("")]
+    fn parse_label_empty_input(#[case] input: &str) {
+        let result = KeyValuePair::from_str(input);
+        assert!(result.is_err());
+    }
+}
diff --git a/src/types/label.rs b/src/types/label.rs
new file mode 100644
index 000000000..2205a00f7
--- /dev/null
+++ b/src/types/label.rs
@@ -0,0 +1,84 @@
+use std::{fmt::Display, str::FromStr};
+
+use crate::types::{KeyParseError, KeyValuePair, KeyValuePairExt, KeyValuePairParseError};
+
+pub type LabelParseError = KeyValuePairParseError;
+pub type LabelKeyParseError = KeyParseError;
+
+/// Labels are key/value pairs attached to K8s objects like pods. It is modeled
+/// after the following [specs][1] It is highly recommended to use [`Label::new`]
+/// to create a new label. This method ensures that no maximum length restrictions
+/// are violated. [`Label`] also implements [`FromStr`], which allows parsing a
+/// label from a string.
+///
+/// Additionally, [`Label`] implements [`Display`], which formats the label using
+/// the following format: `(<prefix>/)<name>=<value>`.
+///
+/// ### Examples
+///
+/// ```
+/// use stackable_operator::types::{Label, KeyValuePairExt};
+/// use std::str::FromStr;
+///
+/// let label = Label::new(Some("stackable.tech"), "release", "23.7");
+/// let label = Label::from_str("stackable.tech/release=23.7").unwrap();
+/// let label = Label::try_from("stackable.tech/release=23.7").unwrap();
+/// ```
+///
+/// [1]: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+#[derive(Clone, Debug)]
+pub struct Label(KeyValuePair);
+
+impl KeyValuePairExt for Label {
+    /// Creates a new label (key/value pair). The key consists of an optional
+    /// `prefix` and a name.
+    ///
+    /// ```
+    /// use stackable_operator::types::{Label, KeyValuePairExt};
+    ///
+    /// // stackable.tech/release=23.7
+    /// let label = Label::new(Some("stackable.tech"), "release", "23.7");
+    /// ```
+    fn new<T>(prefix: Option<T>, name: T, value: T) -> Result<Self, LabelParseError>
+    where
+        T: Into<String>,
+    {
+        let kvp = KeyValuePair::new(prefix, name, value)?;
+        Ok(Self(kvp))
+    }
+
+    /// Returns the label key as a formatted [`String`]. If the key contains
+    /// a prefix, the key has a format like `<prefix>/<name>`. If not, the key
+    /// only consists of a name.
+    fn key(&self) -> String {
+        self.0.key()
+    }
+
+    /// Returns the label value as a [`String`].
+    fn value(&self) -> &String {
+        self.0.value()
+    }
+}
+
+impl FromStr for Label {
+    type Err = LabelParseError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        let kvp = KeyValuePair::from_str(input)?;
+        Ok(Self(kvp))
+    }
+}
+
+impl TryFrom<&str> for Label {
+    type Error = LabelParseError;
+
+    fn try_from(input: &str) -> Result<Self, Self::Error> {
+        Self::from_str(input)
+    }
+}
+
+impl Display for Label {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
diff --git a/src/types/mod.rs b/src/types/mod.rs
new file mode 100644
index 000000000..52c475209
--- /dev/null
+++ b/src/types/mod.rs
@@ -0,0 +1,7 @@
+mod annotation;
+mod kvp;
+mod label;
+
+pub use annotation::*;
+pub use kvp::*;
+pub use label::*;
diff --git a/src/validation.rs b/src/validation.rs
index 969b61cd0..3b96a6baf 100644
--- a/src/validation.rs
+++ b/src/validation.rs
@@ -9,26 +9,14 @@
 // This is adapted from Kubernetes.
 // See apimachinery/pkg/util/validation/validation.go, apimachinery/pkg/api/validation/generic.go and pkg/apis/core/validation/validation.go in the Kubernetes source
 
-use const_format::concatcp;
 use lazy_static::lazy_static;
 use regex::Regex;
 
-const RFC_1123_LABEL_FMT: &str = "[a-z0-9]([-a-z0-9]*[a-z0-9])?";
-const RFC_1123_SUBDOMAIN_FMT: &str =
-    concatcp!(RFC_1123_LABEL_FMT, "(\\.", RFC_1123_LABEL_FMT, ")*");
-const RFC_1123_SUBDOMAIN_ERROR_MSG: &str = "a lowercase RFC 1123 subdomain must consist of lower case alphanumeric characters, '-' or '.', and must start and end with an alphanumeric character";
-const RFC_1123_LABEL_ERROR_MSG: &str = "a lowercase RFC 1123 label must consist of lower case alphanumeric characters, '-' or '.', and must start and end with an alphanumeric character";
-
-// This is a subdomain's max length in DNS (RFC 1123)
-const RFC_1123_SUBDOMAIN_MAX_LENGTH: usize = 253;
-// Minimal length reuquired by RFC 1123 is 63. Up to 255 allowed, unsupported by k8s.
-const RFC_1123_LABEL_MAX_LENGTH: usize = 63;
-
-const RFC_1035_LABEL_FMT: &str = "[a-z]([-a-z0-9]*[a-z0-9])?";
-const RFC_1035_LABEL_ERR_MSG: &str = "a DNS-1035 label must consist of lower case alphanumeric characters or '-', start with an alphabetic character, and end with an alphanumeric character";
-
-// This is a label's max length in DNS (RFC 1035)
-const RFC_1035_LABEL_MAX_LENGTH: usize = 63;
+use crate::constants::validation::{
+    RFC_1035_LABEL_ERR_MSG, RFC_1035_LABEL_FMT, RFC_1035_LABEL_MAX_LENGTH,
+    RFC_1123_LABEL_ERROR_MSG, RFC_1123_LABEL_MAX_LENGTH, RFC_1123_SUBDOMAIN_ERROR_MSG,
+    RFC_1123_SUBDOMAIN_FMT, RFC_1123_SUBDOMAIN_MAX_LENGTH,
+};
 
 lazy_static! {
     static ref RFC_1123_SUBDOMAIN_REGEX: Regex =