diff --git a/CHANGELOG.md b/CHANGELOG.md
index 1f5cc5c4f..468f63ded 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,19 @@ All notable changes to this project will be documented in this file.
 
 ## [Unreleased]
 
+### Added
+
+- Add `stackble_operator::kvp` module and types to allow validated construction of key/value pairs, like labels and
+  annotations. Most users want to use the exported type aliases `Label` and `Annotation` ([#684]).
+
+### Changed
+
+- Move `stackable_operator::label_selector::convert_label_selector_to_query_string` into `kvp` module. The conversion
+  functionality now is encapsulated in a new trait `LabelSelectorExt`. An instance of a `LabelSelector` can now be
+  converted into a query string by calling the associated function `ls.to_query_string()` ([#684]).
+
+[#684]: https://github.com/stackabletech/operator-rs/pull/684
+
 ## [0.58.1] - 2023-12-12
 
 ### Added
diff --git a/src/builder/meta.rs b/src/builder/meta.rs
index dcee94da2..17f9419c4 100644
--- a/src/builder/meta.rs
+++ b/src/builder/meta.rs
@@ -1,10 +1,20 @@
-use crate::error::{Error, OperatorResult};
-use crate::labels::{self, ObjectLabels};
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::{ObjectMeta, OwnerReference};
 use kube::{Resource, ResourceExt};
-use std::collections::BTreeMap;
+use snafu::{ResultExt, Snafu};
 use tracing::warn;
 
+use crate::{
+    error::{Error, OperatorResult},
+    kvp::{Annotation, Annotations, Label, LabelError, Labels, ObjectLabels},
+};
+
+// NOTE (Techassi): Think about that name
+#[derive(Debug, Snafu)]
+pub enum ObjectMetaBuilderError {
+    #[snafu(display("failed to set recommended labels"))]
+    RecommendedLabels { source: LabelError },
+}
+
 /// A builder to build [`ObjectMeta`] objects.
 ///
 /// Of special interest is the [`Self::ownerreference_from_resource()`] function.
@@ -13,12 +23,12 @@ use tracing::warn;
 /// It is strongly recommended to always call [`Self::with_recommended_labels()`]!
 #[derive(Clone, Default)]
 pub struct ObjectMetaBuilder {
-    name: Option<String>,
+    ownerreference: Option<OwnerReference>,
+    annotations: Option<Annotations>,
     generate_name: Option<String>,
     namespace: Option<String>,
-    ownerreference: Option<OwnerReference>,
-    labels: Option<BTreeMap<String, String>>,
-    annotations: Option<BTreeMap<String, String>>,
+    labels: Option<Labels>,
+    name: Option<String>,
 }
 
 impl ObjectMetaBuilder {
@@ -92,74 +102,70 @@ 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 {
+    pub fn with_annotation(&mut self, annotation: Annotation) -> &mut Self {
         self.annotations
-            .get_or_insert_with(BTreeMap::new)
-            .insert(annotation_key.into(), annotation_value.into());
+            .get_or_insert(Annotations::new())
+            .insert(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 {
+    pub fn with_annotations(&mut self, annotations: Annotations) -> &mut Self {
         self.annotations
-            .get_or_insert_with(BTreeMap::new)
+            .get_or_insert(Annotations::new())
             .extend(annotations);
         self
     }
 
     /// This will replace all existing annotations
-    pub fn annotations(&mut self, annotations: BTreeMap<String, String>) -> &mut Self {
+    pub fn annotations(&mut self, annotations: Annotations) -> &mut Self {
         self.annotations = Some(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) -> &mut Self {
+        self.labels.get_or_insert(Labels::new()).insert(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: Labels) -> &mut Self {
+        self.labels.get_or_insert(Labels::new()).extend(labels);
         self
     }
 
     /// This will replace all existing labels
-    pub fn labels(&mut self, labels: BTreeMap<String, String>) -> &mut Self {
+    pub fn labels(&mut self, labels: Labels) -> &mut Self {
         self.labels = Some(labels);
         self
     }
 
-    /// This sets the common recommended labels (in the `app.kubernetes.io` namespace).
-    /// It is recommended to always call this method.
-    /// The only reasons it is not _required_ is to make testing easier and to allow for more
-    /// flexibility if needed.
+    /// This sets the common recommended labels (in the `app.kubernetes.io`
+    /// namespace). It is recommended to always call this method. 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,
         object_labels: ObjectLabels<T>,
-    ) -> &mut Self {
-        let recommended_labels = labels::get_recommended_labels(object_labels);
+    ) -> Result<&mut Self, ObjectMetaBuilderError> {
+        let recommended_labels =
+            Labels::recommended(object_labels).context(RecommendedLabelsSnafu)?;
+
         self.labels
-            .get_or_insert_with(BTreeMap::new)
+            .get_or_insert(Labels::new())
             .extend(recommended_labels);
-        self
+
+        Ok(self)
     }
 
     pub fn build(&self) -> ObjectMeta {
+        // NOTE (Techassi): Shouldn't this take self instead of &self to consume
+        // the builder and build ObjectMeta without cloning?
+
         // 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) {
@@ -178,8 +184,8 @@ impl ObjectMetaBuilder {
                 .ownerreference
                 .as_ref()
                 .map(|ownerreference| vec![ownerreference.clone()]),
-            labels: self.labels.clone(),
-            annotations: self.annotations.clone(),
+            labels: self.labels.clone().map(|l| l.into()),
+            annotations: self.annotations.clone().map(|a| a.into()),
             ..ObjectMeta::default()
         }
     }
@@ -329,7 +335,8 @@ mod tests {
                 role: "role",
                 role_group: "rolegroup",
             })
-            .with_annotation("foo", "bar")
+            .unwrap()
+            .with_annotation(("foo", "bar").try_into().unwrap())
             .build();
 
         assert_eq!(meta.generate_name, Some("generate_foo".to_string()));
diff --git a/src/builder/pdb.rs b/src/builder/pdb.rs
index e567a96db..d0c6aba14 100644
--- a/src/builder/pdb.rs
+++ b/src/builder/pdb.rs
@@ -1,9 +1,3 @@
-use crate::{
-    builder::ObjectMetaBuilder,
-    error::OperatorResult,
-    labels::{role_selector_labels, APP_MANAGED_BY_LABEL},
-    utils::format_full_controller_name,
-};
 use k8s_openapi::{
     api::policy::v1::{PodDisruptionBudget, PodDisruptionBudgetSpec},
     apimachinery::pkg::{
@@ -13,6 +7,12 @@ use k8s_openapi::{
 };
 use kube::{Resource, ResourceExt};
 
+use crate::{
+    builder::ObjectMetaBuilder,
+    error::OperatorResult,
+    kvp::{Label, Labels},
+};
+
 /// This builder is used to construct [`PodDisruptionBudget`]s.
 /// If you are using this to create [`PodDisruptionBudget`]s according to [ADR 30 on Allowed Pod disruptions][adr],
 /// the use of [`PodDisruptionBudgetBuilder::new_with_role`] is recommended.
@@ -48,15 +48,22 @@ impl PodDisruptionBudgetBuilder<(), (), ()> {
         PodDisruptionBudgetBuilder::default()
     }
 
-    /// This method populates [`PodDisruptionBudget::metadata`] and [`PodDisruptionBudgetSpec::selector`] from the give role
-    /// (not roleGroup!).
+    /// This method populates [`PodDisruptionBudget::metadata`] and
+    /// [`PodDisruptionBudgetSpec::selector`] from the give role (not roleGroup!).
     ///
-    /// The parameters are the same as the fields from [`crate::labels::ObjectLabels`]:
-    /// * `owner` - Reference to the k8s object owning the created resource, such as `HdfsCluster` or `TrinoCluster`.
-    /// * `app_name` - The name of the app being managed, such as `hdfs` or `trino`.
-    /// * `role` - The role that this object belongs to, e.g. `datanode` or `worker`.
-    /// * `operator_name` - The DNS-style name of the operator managing the object (such as `hdfs.stackable.tech`).
-    /// * `controller_name` - The name of the controller inside of the operator managing the object (such as `hdfscluster`)
+    /// The parameters are the same as the fields from
+    /// [`ObjectLabels`][crate::kvp::ObjectLabels]:
+    ///
+    /// * `owner` - Reference to the k8s object owning the created resource,
+    ///   such as `HdfsCluster` or `TrinoCluster`.
+    /// * `app_name` - The name of the app being managed, such as `hdfs` or
+    ///   `trino`.
+    /// * `role` - The role that this object belongs to, e.g. `datanode` or
+    ///   `worker`.
+    /// * `operator_name` - The DNS-style name of the operator managing the
+    ///   object (such as `hdfs.stackable.tech`).
+    /// * `controller_name` - The name of the controller inside of the operator
+    ///   managing the object (such as `hdfscluster`)
     pub fn new_with_role<T: Resource<DynamicType = ()>>(
         owner: &T,
         app_name: &str,
@@ -64,23 +71,20 @@ impl PodDisruptionBudgetBuilder<(), (), ()> {
         operator_name: &str,
         controller_name: &str,
     ) -> OperatorResult<PodDisruptionBudgetBuilder<ObjectMeta, LabelSelector, ()>> {
-        let role_selector_labels = role_selector_labels(owner, app_name, role);
+        let role_selector_labels = Labels::role_selector(owner, app_name, role)?;
         let metadata = ObjectMetaBuilder::new()
             .namespace_opt(owner.namespace())
             .name(format!("{}-{}", owner.name_any(), role))
             .ownerreference_from_resource(owner, None, Some(true))?
             .with_labels(role_selector_labels.clone())
-            .with_label(
-                APP_MANAGED_BY_LABEL.to_string(),
-                format_full_controller_name(operator_name, controller_name),
-            )
+            .with_label(Label::managed_by(operator_name, controller_name)?)
             .build();
 
         Ok(PodDisruptionBudgetBuilder {
             metadata,
             selector: LabelSelector {
                 match_expressions: None,
-                match_labels: Some(role_selector_labels),
+                match_labels: Some(role_selector_labels.into()),
             },
             ..PodDisruptionBudgetBuilder::default()
         })
diff --git a/src/builder/pod/mod.rs b/src/builder/pod/mod.rs
index f2e892eb2..8636dba56 100644
--- a/src/builder/pod/mod.rs
+++ b/src/builder/pod/mod.rs
@@ -1,15 +1,20 @@
-pub mod container;
-pub mod resources;
-pub mod security;
-pub mod volume;
+use std::{collections::BTreeMap, num::TryFromIntError};
 
-use std::collections::BTreeMap;
-use std::num::TryFromIntError;
+use k8s_openapi::{
+    api::core::v1::{
+        Affinity, Container, LocalObjectReference, NodeAffinity, Pod, PodAffinity, PodAntiAffinity,
+        PodCondition, PodSecurityContext, PodSpec, PodStatus, PodTemplateSpec,
+        ResourceRequirements, Toleration, Volume,
+    },
+    apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::ObjectMeta},
+};
+use snafu::{ResultExt, Snafu};
+use tracing::warn;
 
 use crate::{
     builder::{
-        meta::ObjectMetaBuilder, ListenerOperatorVolumeSourceBuilder, ListenerReference,
-        VolumeBuilder,
+        meta::ObjectMetaBuilder, ListenerOperatorVolumeSourceBuilder,
+        ListenerOperatorVolumeSourceBuilderError, ListenerReference, VolumeBuilder,
     },
     commons::{
         affinity::StackableAffinity,
@@ -23,23 +28,24 @@ use crate::{
     time::Duration,
 };
 
-use k8s_openapi::{
-    api::core::v1::{
-        Affinity, Container, LocalObjectReference, NodeAffinity, Pod, PodAffinity, PodAntiAffinity,
-        PodCondition, PodSecurityContext, PodSpec, PodStatus, PodTemplateSpec,
-        ResourceRequirements, Toleration, Volume,
-    },
-    apimachinery::pkg::{api::resource::Quantity, apis::meta::v1::ObjectMeta},
-};
-use tracing::warn;
+pub mod container;
+pub mod resources;
+pub mod security;
+pub mod volume;
 
-#[derive(Debug, thiserror::Error)]
+#[derive(Debug, Snafu)]
 pub enum Error {
-    #[error("termination grace period is too long (got {duration}, maximum allowed is {max})", max = Duration::from_secs(i64::MAX as u64))]
+    #[snafu(display("termination grace period is too long (got {duration}, maximum allowed is {max})", max = Duration::from_secs(i64::MAX as u64)))]
     TerminationGracePeriodTooLong {
         source: TryFromIntError,
         duration: Duration,
     },
+
+    #[snafu(display("failed to add listener volume '{name}' to the pod"))]
+    ListenerVolume {
+        source: ListenerOperatorVolumeSourceBuilderError,
+        name: String,
+    },
 }
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -299,6 +305,7 @@ impl PodBuilder {
     ///             .build(),
     ///     )
     ///     .add_listener_volume_by_listener_class("listener", "nodeport")
+    ///     .unwrap()
     ///     .build()
     ///     .unwrap();
     ///
@@ -341,18 +348,19 @@ impl PodBuilder {
         &mut self,
         volume_name: &str,
         listener_class: &str,
-    ) -> &mut Self {
+    ) -> Result<&mut Self> {
+        let listener_reference = ListenerReference::ListenerClass(listener_class.to_string());
+        let volume = ListenerOperatorVolumeSourceBuilder::new(&listener_reference)
+            .build()
+            .context(ListenerVolumeSnafu { name: volume_name })?;
+
         self.add_volume(Volume {
             name: volume_name.into(),
-            ephemeral: Some(
-                ListenerOperatorVolumeSourceBuilder::new(&ListenerReference::ListenerClass(
-                    listener_class.into(),
-                ))
-                .build(),
-            ),
+            ephemeral: Some(volume),
             ..Volume::default()
         });
-        self
+
+        Ok(self)
     }
 
     /// Add a [`Volume`] for the storage class `listeners.stackable.tech` with the given listener
@@ -386,6 +394,7 @@ impl PodBuilder {
     ///             .build(),
     ///     )
     ///     .add_listener_volume_by_listener_name("listener", "preprovisioned-listener")
+    ///     .unwrap()
     ///     .build()
     ///     .unwrap();
     ///
@@ -428,18 +437,19 @@ impl PodBuilder {
         &mut self,
         volume_name: &str,
         listener_name: &str,
-    ) -> &mut Self {
+    ) -> Result<&mut Self> {
+        let listener_reference = ListenerReference::ListenerName(listener_name.to_string());
+        let volume = ListenerOperatorVolumeSourceBuilder::new(&listener_reference)
+            .build()
+            .context(ListenerVolumeSnafu { name: volume_name })?;
+
         self.add_volume(Volume {
             name: volume_name.into(),
-            ephemeral: Some(
-                ListenerOperatorVolumeSourceBuilder::new(&ListenerReference::ListenerName(
-                    listener_name.into(),
-                ))
-                .build(),
-            ),
+            ephemeral: Some(volume),
             ..Volume::default()
         });
-        self
+
+        Ok(self)
     }
 
     pub fn image_pull_secrets(
diff --git a/src/builder/pod/volume.rs b/src/builder/pod/volume.rs
index e17e0b5bf..f13ab0d98 100644
--- a/src/builder/pod/volume.rs
+++ b/src/builder/pod/volume.rs
@@ -1,23 +1,23 @@
-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 snafu::{ResultExt, Snafu};
 use tracing::warn;
 
-use crate::builder::ObjectMetaBuilder;
+use crate::{
+    builder::ObjectMetaBuilder,
+    kvp::{Annotation, AnnotationError, Annotations},
+};
 
-/// A builder to build [`Volume`] objects.
-/// May only contain one `volume_source` at a time.
-/// E.g. a call like `secret` after `empty_dir` will overwrite the `empty_dir`.
+/// A builder to build [`Volume`] objects. May only contain one `volume_source`
+/// at a time. E.g. a call like `secret` after `empty_dir` will overwrite the
+/// `empty_dir`.
 #[derive(Clone, Default)]
 pub struct VolumeBuilder {
     name: String,
@@ -209,7 +209,6 @@ impl VolumeBuilder {
 }
 
 /// A builder to build [`VolumeMount`] objects.
-///
 #[derive(Clone, Default)]
 pub struct VolumeMountBuilder {
     mount_path: String,
@@ -262,6 +261,12 @@ impl VolumeMountBuilder {
     }
 }
 
+#[derive(Debug, Snafu)]
+pub enum SecretOperatorVolumeSourceBuilderError {
+    #[snafu(display("failed to parse secret operator volume annotation"))]
+    ParseAnnotation { source: AnnotationError },
+}
+
 #[derive(Clone)]
 pub struct SecretOperatorVolumeSourceBuilder {
     secret_class: String,
@@ -313,41 +318,26 @@ impl SecretOperatorVolumeSourceBuilder {
         self
     }
 
-    pub fn build(&self) -> EphemeralVolumeSource {
-        let mut attrs = BTreeMap::from([(
-            "secrets.stackable.tech/class".to_string(),
-            self.secret_class.clone(),
-        )]);
+    pub fn build(&self) -> Result<EphemeralVolumeSource, SecretOperatorVolumeSourceBuilderError> {
+        let mut annotations = Annotations::new();
+
+        annotations
+            .insert(Annotation::secret_class(&self.secret_class).context(ParseAnnotationSnafu)?);
 
         if !self.scopes.is_empty() {
-            let mut scopes = String::new();
-            for scope in self.scopes.iter() {
-                if !scopes.is_empty() {
-                    scopes.push(',');
-                };
-                match scope {
-                    SecretOperatorVolumeScope::Node => scopes.push_str("node"),
-                    SecretOperatorVolumeScope::Pod => scopes.push_str("pod"),
-                    SecretOperatorVolumeScope::Service { name } => {
-                        scopes.push_str("service=");
-                        scopes.push_str(name);
-                    }
-                }
-            }
-            attrs.insert("secrets.stackable.tech/scope".to_string(), scopes);
+            annotations
+                .insert(Annotation::secret_scope(&self.scopes).context(ParseAnnotationSnafu)?);
         }
 
         if let Some(format) = &self.format {
-            attrs.insert(
-                "secrets.stackable.tech/format".to_string(),
-                format.as_ref().to_string(),
-            );
+            annotations
+                .insert(Annotation::secret_format(format.as_ref()).context(ParseAnnotationSnafu)?);
         }
 
         if !self.kerberos_service_names.is_empty() {
-            attrs.insert(
-                "secrets.stackable.tech/kerberos.service.names".to_string(),
-                self.kerberos_service_names.join(","),
+            annotations.insert(
+                Annotation::kerberos_service_names(&self.kerberos_service_names)
+                    .context(ParseAnnotationSnafu)?,
             );
         }
 
@@ -356,16 +346,15 @@ impl SecretOperatorVolumeSourceBuilder {
             if Some(SecretFormat::TlsPkcs12) != self.format {
                 warn!(format.actual = ?self.format, format.expected = ?Some(SecretFormat::TlsPkcs12), "A TLS PKCS12 password was set but ignored because another format was requested")
             } else {
-                attrs.insert(
-                    "secrets.stackable.tech/format.compatibility.tls-pkcs12.password".to_string(),
-                    password.to_string(),
+                annotations.insert(
+                    Annotation::tls_pkcs12_password(password).context(ParseAnnotationSnafu)?,
                 );
             }
         }
 
-        EphemeralVolumeSource {
+        Ok(EphemeralVolumeSource {
             volume_claim_template: Some(PersistentVolumeClaimTemplate {
-                metadata: Some(ObjectMetaBuilder::new().annotations(attrs).build()),
+                metadata: Some(ObjectMetaBuilder::new().annotations(annotations).build()),
                 spec: PersistentVolumeClaimSpec {
                     storage_class_name: Some("secrets.stackable.tech".to_string()),
                     resources: Some(ResourceRequirements {
@@ -376,7 +365,7 @@ impl SecretOperatorVolumeSourceBuilder {
                     ..PersistentVolumeClaimSpec::default()
                 },
             }),
-        }
+        })
     }
 }
 
@@ -395,7 +384,7 @@ pub enum SecretFormat {
 }
 
 #[derive(Clone)]
-enum SecretOperatorVolumeScope {
+pub enum SecretOperatorVolumeScope {
     Node,
     Pod,
     Service { name: String },
@@ -410,20 +399,26 @@ 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, AnnotationError> {
         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(class) => {
+                Annotation::try_from(("listeners.stackable.tech/listener-class", class.as_str()))
+            }
+            ListenerReference::ListenerName(name) => {
+                Annotation::try_from(("listeners.stackable.tech/listener-name", name.as_str()))
+            }
         }
     }
 }
 
+// NOTE (Techassi): We might want to think about these names and how long they
+// are getting.
+#[derive(Debug, Snafu)]
+pub enum ListenerOperatorVolumeSourceBuilderError {
+    #[snafu(display("failed to convert listener reference into Kubernetes annotation"))]
+    ListenerReferenceAnnotation { source: AnnotationError },
+}
+
 /// Builder for an [`EphemeralVolumeSource`] containing the listener configuration
 ///
 /// # Example
@@ -435,10 +430,13 @@ impl ListenerReference {
 /// # use stackable_operator::builder::PodBuilder;
 /// let mut pod_builder = PodBuilder::new();
 ///
-/// let volume_source = ListenerOperatorVolumeSourceBuilder::new(
+/// let volume_source =
+///     ListenerOperatorVolumeSourceBuilder::new(
 ///         &ListenerReference::ListenerClass("nodeport".into()),
 ///     )
-///     .build();
+///     .build()
+///     .unwrap();
+///
 /// pod_builder
 ///     .add_volume(Volume {
 ///         name: "listener".to_string(),
@@ -464,12 +462,17 @@ impl ListenerOperatorVolumeSourceBuilder {
     }
 
     /// Build an [`EphemeralVolumeSource`] from the builder
-    pub fn build(&self) -> EphemeralVolumeSource {
-        EphemeralVolumeSource {
+    pub fn build(&self) -> Result<EphemeralVolumeSource, ListenerOperatorVolumeSourceBuilderError> {
+        let listener_reference_annotation = self
+            .listener_reference
+            .to_annotation()
+            .context(ListenerReferenceAnnotationSnafu)?;
+
+        Ok(EphemeralVolumeSource {
             volume_claim_template: Some(PersistentVolumeClaimTemplate {
                 metadata: Some(
                     ObjectMetaBuilder::new()
-                        .annotations([self.listener_reference.to_annotation()].into())
+                        .with_annotation(listener_reference_annotation)
                         .build(),
                 ),
                 spec: PersistentVolumeClaimSpec {
@@ -482,7 +485,7 @@ impl ListenerOperatorVolumeSourceBuilder {
                     ..PersistentVolumeClaimSpec::default()
                 },
             }),
-        }
+        })
     }
 }
 
@@ -553,7 +556,7 @@ mod tests {
             "public".into(),
         ));
 
-        let volume_source = builder.build();
+        let volume_source = builder.build().unwrap();
 
         let volume_claim_template = volume_source.volume_claim_template;
         let annotations = volume_claim_template
diff --git a/src/client.rs b/src/client.rs
index bff7c17b7..7a821baab 100644
--- a/src/client.rs
+++ b/src/client.rs
@@ -1,5 +1,5 @@
 use crate::error::{Error, OperatorResult};
-use crate::label_selector;
+use crate::kvp::LabelSelectorExt;
 
 use either::Either;
 use futures::StreamExt;
@@ -128,7 +128,7 @@ impl Client {
 
     /// Lists resources from the API using a LabelSelector.
     ///
-    /// This takes a LabelSelector and converts it into a query string using [`label_selector::convert_label_selector_to_query_string`].
+    /// This takes a LabelSelector and converts it into a query string using [`LabelSelectorExt`].
     ///
     /// # Arguments
     ///
@@ -143,7 +143,7 @@ impl Client {
         T: Clone + Debug + DeserializeOwned + Resource + GetApi,
         <T as Resource>::DynamicType: Default,
     {
-        let selector_string = label_selector::convert_label_selector_to_query_string(selector)?;
+        let selector_string = selector.to_query_string()?;
         trace!("Listing for LabelSelector [{}]", selector_string);
         let list_params = ListParams {
             label_selector: Some(selector_string),
diff --git a/src/cluster_resources.rs b/src/cluster_resources.rs
index 0e986fdeb..ce85762fc 100644
--- a/src/cluster_resources.rs
+++ b/src/cluster_resources.rs
@@ -10,7 +10,10 @@ use crate::{
         },
     },
     error::{Error, OperatorResult},
-    labels::{APP_INSTANCE_LABEL, APP_MANAGED_BY_LABEL, APP_NAME_LABEL},
+    kvp::{
+        consts::{K8S_APP_INSTANCE_KEY, K8S_APP_MANAGED_BY_KEY, K8S_APP_NAME_KEY},
+        Label, LabelError, Labels,
+    },
     utils::format_full_controller_name,
 };
 
@@ -342,14 +345,25 @@ impl ClusterResource for DaemonSet {
 pub struct ClusterResources {
     /// The namespace of the cluster
     namespace: String,
+
     /// The name of the cluster
     app_instance: String,
+
     /// The name of the application
     app_name: String,
-    /// The manager of the cluster resources, e.g. the controller
+
+    // TODO (Techassi): Add doc comments
+    operator_name: String,
+
+    // TODO (Techassi): Add doc comments
+    controller_name: String,
+
+    // TODO (Techassi): Add doc comment
     manager: String,
+
     /// The unique IDs of the cluster resources
     resource_ids: HashSet<String>,
+
     /// Strategy to manage how cluster resources are applied. Resources could be patched, merged
     /// or not applied at all depending on the strategy.
     apply_strategy: ClusterResourceApplyStrategy,
@@ -395,6 +409,8 @@ impl ClusterResources {
             namespace,
             app_instance,
             app_name: app_name.into(),
+            operator_name: operator_name.into(),
+            controller_name: controller_name.into(),
             manager: format_full_controller_name(operator_name, controller_name),
             resource_ids: Default::default(),
             apply_strategy,
@@ -403,17 +419,15 @@ 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<Labels, LabelError> {
+        let mut labels = Labels::common(&self.app_name, &self.app_instance)?;
+
+        labels.insert(Label::managed_by(
+            &self.operator_name,
+            &self.controller_name,
+        )?);
+
+        Ok(labels)
     }
 
     /// Adds a resource to the cluster resources.
@@ -442,7 +456,11 @@ impl ClusterResources {
     ) -> OperatorResult<T> {
         Self::check_labels(
             resource.labels(),
-            &[APP_INSTANCE_LABEL, APP_MANAGED_BY_LABEL, APP_NAME_LABEL],
+            &[
+                K8S_APP_INSTANCE_KEY,
+                K8S_APP_MANAGED_BY_KEY,
+                K8S_APP_NAME_KEY,
+            ],
             &[&self.app_instance, &self.manager, &self.app_name],
         )?;
 
@@ -664,17 +682,17 @@ impl ClusterResources {
         let label_selector = LabelSelector {
             match_expressions: Some(vec![
                 LabelSelectorRequirement {
-                    key: APP_INSTANCE_LABEL.into(),
+                    key: K8S_APP_INSTANCE_KEY.into(),
                     operator: "In".into(),
                     values: Some(vec![self.app_instance.to_owned()]),
                 },
                 LabelSelectorRequirement {
-                    key: APP_NAME_LABEL.into(),
+                    key: K8S_APP_NAME_KEY.into(),
                     operator: "In".into(),
                     values: Some(vec![self.app_name.to_owned()]),
                 },
                 LabelSelectorRequirement {
-                    key: APP_MANAGED_BY_LABEL.into(),
+                    key: K8S_APP_MANAGED_BY_KEY.into(),
                     operator: "In".into(),
                     values: Some(vec![self.manager.to_owned()]),
                 },
diff --git a/src/commons/affinity.rs b/src/commons/affinity.rs
index d3b1699ff..5c94db2aa 100644
--- a/src/commons/affinity.rs
+++ b/src/commons/affinity.rs
@@ -13,7 +13,7 @@ use stackable_operator_derive::Fragment;
 
 use crate::{
     config::merge::{Atomic, Merge},
-    labels::{APP_COMPONENT_LABEL, APP_INSTANCE_LABEL, APP_NAME_LABEL},
+    kvp::consts::{K8S_APP_COMPONENT_KEY, K8S_APP_INSTANCE_KEY, K8S_APP_NAME_KEY},
 };
 
 pub const TOPOLOGY_KEY_HOSTNAME: &str = "kubernetes.io/hostname";
@@ -140,9 +140,9 @@ pub fn affinity_between_role_pods(
             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()),
+                    (K8S_APP_NAME_KEY.to_string(), app_name.to_string()),
+                    (K8S_APP_INSTANCE_KEY.to_string(), cluster_name.to_string()),
+                    (K8S_APP_COMPONENT_KEY.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
                 ])),
             }),
@@ -167,8 +167,8 @@ pub fn affinity_between_cluster_pods(
             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()),
+                    (K8S_APP_NAME_KEY.to_string(), app_name.to_string()),
+                    (K8S_APP_INSTANCE_KEY.to_string(), cluster_name.to_string()),
                 ])),
             }),
             namespace_selector: None,
diff --git a/src/commons/authentication/ldap.rs b/src/commons/authentication/ldap.rs
index 069140b23..b9f713d8e 100644
--- a/src/commons/authentication/ldap.rs
+++ b/src/commons/authentication/ldap.rs
@@ -1,15 +1,24 @@
 use k8s_openapi::api::core::v1::{Volume, VolumeMount};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use snafu::{ResultExt, Snafu};
 
 use crate::{
     builder::{ContainerBuilder, PodBuilder, VolumeMountBuilder},
     commons::{
         authentication::{tls::TlsClientDetails, SECRET_BASE_PATH},
-        secret_class::SecretClassVolume,
+        secret_class::{SecretClassVolume, SecretClassVolumeError},
     },
 };
 
+#[derive(Debug, Snafu)]
+pub enum AuthenticationProviderError {
+    #[snafu(display(
+        "failed to convert bind credentials (secret class volume) into named Kubernetes volume"
+    ))]
+    BindCredentials { source: SecretClassVolumeError },
+}
+
 #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct AuthenticationProvider {
@@ -58,32 +67,40 @@ impl AuthenticationProvider {
         &self,
         pod_builder: &mut PodBuilder,
         container_builders: Vec<&mut ContainerBuilder>,
-    ) {
-        let (volumes, mounts) = self.volumes_and_mounts();
+    ) -> Result<(), AuthenticationProviderError> {
+        let (volumes, mounts) = self.volumes_and_mounts()?;
         pod_builder.add_volumes(volumes);
+
         for cb in container_builders {
             cb.add_volume_mounts(mounts.clone());
         }
+
+        Ok(())
     }
 
     /// It is recommended to use [`Self::add_volumes_and_mounts`], this function returns you the
     /// volumes and mounts in case you need to add them by yourself.
-    pub fn volumes_and_mounts(&self) -> (Vec<Volume>, Vec<VolumeMount>) {
+    pub fn volumes_and_mounts(
+        &self,
+    ) -> Result<(Vec<Volume>, Vec<VolumeMount>), AuthenticationProviderError> {
         let mut volumes = Vec::new();
         let mut mounts = Vec::new();
 
         if let Some(bind_credentials) = &self.bind_credentials {
             let secret_class = &bind_credentials.secret_class;
             let volume_name = format!("{secret_class}-bind-credentials");
+            let volume = bind_credentials
+                .to_volume(&volume_name)
+                .context(BindCredentialsSnafu)?;
 
-            volumes.push(bind_credentials.to_volume(&volume_name));
+            volumes.push(volume);
             mounts.push(
                 VolumeMountBuilder::new(volume_name, format!("{SECRET_BASE_PATH}/{secret_class}"))
                     .build(),
             );
         }
 
-        (volumes, mounts)
+        Ok((volumes, mounts))
     }
 
     /// Returns the path of the files containing bind user and password.
@@ -213,14 +230,15 @@ mod test {
             ldap.tls.tls_ca_cert_mount_path(),
             Some("/stackable/secrets/ldap-ca-cert/ca.crt".to_string())
         );
-        let (tls_volumes, tls_mounts) = ldap.tls.volumes_and_mounts();
+        let (tls_volumes, tls_mounts) = ldap.tls.volumes_and_mounts().unwrap();
         assert_eq!(
             tls_volumes,
             vec![SecretClassVolume {
                 secret_class: "ldap-ca-cert".to_string(),
                 scope: None,
             }
-            .to_volume("ldap-ca-cert-ca-cert")]
+            .to_volume("ldap-ca-cert-ca-cert")
+            .unwrap()]
         );
         assert_eq!(
             tls_mounts,
@@ -238,14 +256,15 @@ mod test {
                 "/stackable/secrets/openldap-bind-credentials/password".to_string()
             ))
         );
-        let (bind_volumes, bind_mounts) = ldap.volumes_and_mounts();
+        let (bind_volumes, bind_mounts) = ldap.volumes_and_mounts().unwrap();
         assert_eq!(
             bind_volumes,
             vec![SecretClassVolume {
                 secret_class: "openldap-bind-credentials".to_string(),
                 scope: None,
             }
-            .to_volume("openldap-bind-credentials-bind-credentials")]
+            .to_volume("openldap-bind-credentials-bind-credentials")
+            .unwrap()]
         );
         assert_eq!(
             bind_mounts,
diff --git a/src/commons/authentication/tls.rs b/src/commons/authentication/tls.rs
index be51ca24b..d980b72d1 100644
--- a/src/commons/authentication/tls.rs
+++ b/src/commons/authentication/tls.rs
@@ -1,10 +1,14 @@
 use k8s_openapi::api::core::v1::{Volume, VolumeMount};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use snafu::{ResultExt, Snafu};
 
 use crate::{
     builder::{ContainerBuilder, PodBuilder, VolumeMountBuilder},
-    commons::{authentication::SECRET_BASE_PATH, secret_class::SecretClassVolume},
+    commons::{
+        authentication::SECRET_BASE_PATH,
+        secret_class::{SecretClassVolume, SecretClassVolumeError},
+    },
 };
 
 #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
@@ -17,6 +21,12 @@ pub struct AuthenticationProvider {
     pub client_cert_secret_class: Option<String>,
 }
 
+#[derive(Debug, Snafu)]
+pub enum TlsClientDetailsError {
+    #[snafu(display("failed to convert secret class volume into named Kubernetes volume"))]
+    SecretClassVolume { source: SecretClassVolumeError },
+}
+
 #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct TlsClientDetails {
@@ -37,36 +47,40 @@ impl TlsClientDetails {
         &self,
         pod_builder: &mut PodBuilder,
         container_builders: Vec<&mut ContainerBuilder>,
-    ) {
-        let (volumes, mounts) = self.volumes_and_mounts();
+    ) -> Result<(), TlsClientDetailsError> {
+        let (volumes, mounts) = self.volumes_and_mounts()?;
         pod_builder.add_volumes(volumes);
+
         for cb in container_builders {
             cb.add_volume_mounts(mounts.clone());
         }
+
+        Ok(())
     }
 
     /// It is recommended to use [`Self::add_volumes_and_mounts`], this function returns you the
     /// volumes and mounts in case you need to add them by yourself.
-    pub fn volumes_and_mounts(&self) -> (Vec<Volume>, Vec<VolumeMount>) {
+    pub fn volumes_and_mounts(
+        &self,
+    ) -> Result<(Vec<Volume>, Vec<VolumeMount>), TlsClientDetailsError> {
         let mut volumes = Vec::new();
         let mut mounts = Vec::new();
 
         if let Some(secret_class) = self.tls_ca_cert_secret_class() {
             let volume_name = format!("{secret_class}-ca-cert");
-            volumes.push(
-                SecretClassVolume {
-                    secret_class: secret_class.to_string(),
-                    scope: None,
-                }
-                .to_volume(&volume_name),
-            );
+            let secret_class_volume = SecretClassVolume::new(secret_class.clone(), None);
+            let volume = secret_class_volume
+                .to_volume(&volume_name)
+                .context(SecretClassVolumeSnafu)?;
+
+            volumes.push(volume);
             mounts.push(
                 VolumeMountBuilder::new(volume_name, format!("{SECRET_BASE_PATH}/{secret_class}"))
                     .build(),
             );
         }
 
-        (volumes, mounts)
+        Ok((volumes, mounts))
     }
 
     /// Whether TLS is configured
diff --git a/src/commons/product_image_selection.rs b/src/commons/product_image_selection.rs
index 70926f195..17eb58c8d 100644
--- a/src/commons/product_image_selection.rs
+++ b/src/commons/product_image_selection.rs
@@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
 use strum::AsRefStr;
 
 #[cfg(doc)]
-use crate::labels::get_recommended_labels;
+use crate::kvp::Labels;
 
 pub const STACKABLE_DOCKER_REPO: &str = "docker.stackable.tech/stackable";
 
@@ -66,12 +66,16 @@ pub struct ProductImageStackableVersion {
 pub struct ResolvedProductImage {
     /// Version of the product, e.g. `1.4.1`.
     pub product_version: String,
-    /// App version as formatted for [`get_recommended_labels`]
+
+    /// App version as formatted for [`Labels::recommended`]
     pub app_version_label: String,
+
     /// Image to be used for the product image e.g. `docker.stackable.tech/stackable/superset:1.4.1-stackable2.1.0`
     pub image: String,
+
     /// Image pull policy for the containers using the product image
     pub image_pull_policy: String,
+
     /// Image pull secrets for the containers using the product image
     pub pull_secrets: Option<Vec<LocalObjectReference>>,
 }
diff --git a/src/commons/rbac.rs b/src/commons/rbac.rs
index b0f51f8a1..425c6fe49 100644
--- a/src/commons/rbac.rs
+++ b/src/commons/rbac.rs
@@ -1,9 +1,14 @@
-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 crate::{
+    builder::ObjectMetaBuilder,
+    error::OperatorResult,
+    k8s_openapi::api::{
+        core::v1::ServiceAccount,
+        rbac::v1::{RoleBinding, RoleRef, Subject},
+    },
+    kvp::Labels,
+};
 
 /// Build RBAC objects for the product workloads.
 /// The `rbac_prefix` is meant to be the product name, for example: zookeeper, airflow, etc.
@@ -11,7 +16,7 @@ use std::collections::BTreeMap;
 pub fn build_rbac_resources<T: Clone + Resource<DynamicType = ()>>(
     resource: &T,
     rbac_prefix: &str,
-    labels: BTreeMap<String, String>,
+    labels: Labels,
 ) -> OperatorResult<(ServiceAccount, RoleBinding)> {
     let sa_name = service_account_name(rbac_prefix);
     let service_account = ServiceAccount {
@@ -61,11 +66,14 @@ pub fn role_binding_name(rbac_prefix: &str) -> String {
 
 #[cfg(test)]
 mod tests {
-    use crate::commons::rbac::{build_rbac_resources, role_binding_name, service_account_name};
     use kube::CustomResource;
     use schemars::{self, JsonSchema};
     use serde::{Deserialize, Serialize};
-    use std::collections::BTreeMap;
+
+    use crate::{
+        commons::rbac::{build_rbac_resources, role_binding_name, service_account_name},
+        kvp::Labels,
+    };
 
     const CLUSTER_NAME: &str = "simple-cluster";
     const RESOURCE_NAME: &str = "test-resource";
@@ -96,7 +104,7 @@ mod tests {
     fn test_build_rbac() {
         let cluster = build_test_resource();
         let (rbac_sa, rbac_rolebinding) =
-            build_rbac_resources(&cluster, RESOURCE_NAME, BTreeMap::new()).unwrap();
+            build_rbac_resources(&cluster, RESOURCE_NAME, Labels::new()).unwrap();
 
         assert_eq!(
             Some(service_account_name(RESOURCE_NAME)),
diff --git a/src/commons/secret_class.rs b/src/commons/secret_class.rs
index b8fc56afc..7e5fde8ca 100644
--- a/src/commons/secret_class.rs
+++ b/src/commons/secret_class.rs
@@ -1,20 +1,42 @@
-use crate::builder::{SecretOperatorVolumeSourceBuilder, VolumeBuilder};
 use k8s_openapi::api::core::v1::{EphemeralVolumeSource, Volume};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
+use snafu::{ResultExt, Snafu};
+
+use crate::builder::{
+    SecretOperatorVolumeSourceBuilder, SecretOperatorVolumeSourceBuilderError, VolumeBuilder,
+};
+
+#[derive(Debug, Snafu)]
+pub enum SecretClassVolumeError {
+    #[snafu(display("failed to build secret operator volume"))]
+    SecretOperatorVolume {
+        source: SecretOperatorVolumeSourceBuilderError,
+    },
+}
 
 #[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct SecretClassVolume {
     /// [SecretClass](DOCS_BASE_URL_PLACEHOLDER/secret-operator/secretclass) containing the LDAP bind credentials.
     pub secret_class: String,
+
     /// [Scope](DOCS_BASE_URL_PLACEHOLDER/secret-operator/scope) of the
     /// [SecretClass](DOCS_BASE_URL_PLACEHOLDER/secret-operator/secretclass).
     pub scope: Option<SecretClassVolumeScope>,
 }
 
 impl SecretClassVolume {
-    pub fn to_ephemeral_volume_source(&self) -> EphemeralVolumeSource {
+    pub fn new(secret_class: String, scope: Option<SecretClassVolumeScope>) -> Self {
+        Self {
+            secret_class,
+            scope,
+        }
+    }
+
+    pub fn to_ephemeral_volume_source(
+        &self,
+    ) -> Result<EphemeralVolumeSource, SecretClassVolumeError> {
         let mut secret_operator_volume_builder =
             SecretOperatorVolumeSourceBuilder::new(&self.secret_class);
 
@@ -30,13 +52,14 @@ impl SecretClassVolume {
             }
         }
 
-        secret_operator_volume_builder.build()
+        secret_operator_volume_builder
+            .build()
+            .context(SecretOperatorVolumeSnafu)
     }
 
-    pub fn to_volume(&self, volume_name: &str) -> Volume {
-        VolumeBuilder::new(volume_name)
-            .ephemeral(self.to_ephemeral_volume_source())
-            .build()
+    pub fn to_volume(&self, volume_name: &str) -> Result<Volume, SecretClassVolumeError> {
+        let ephemeral = self.to_ephemeral_volume_source()?;
+        Ok(VolumeBuilder::new(volume_name).ephemeral(ephemeral).build())
     }
 }
 
@@ -74,7 +97,8 @@ mod tests {
                 services: vec!["myservice".to_string()],
             }),
         }
-        .to_ephemeral_volume_source();
+        .to_ephemeral_volume_source()
+        .unwrap();
 
         let expected_volume_attributes = BTreeMap::from([
             (
diff --git a/src/error.rs b/src/error.rs
index 482495b46..f00a6e4f6 100644
--- a/src/error.rs
+++ b/src/error.rs
@@ -1,4 +1,4 @@
-use crate::product_config_utils;
+use crate::{kvp::LabelError, product_config_utils};
 use std::path::PathBuf;
 
 #[derive(Debug, thiserror::Error)]
@@ -120,6 +120,12 @@ pub enum Error {
 
     #[error("OIDC authentication details not specified. The AuthenticationClass {auth_class_name:?} uses an OIDC provider, you need to specify OIDC authentication details (such as client credentials) as well")]
     OidcAuthenticationDetailsNotSpecified { auth_class_name: String },
+
+    #[error("failed to parse label: {source}")]
+    InvalidLabel {
+        #[from]
+        source: LabelError,
+    },
 }
 
 pub type OperatorResult<T> = std::result::Result<T, Error>;
diff --git a/src/kvp/annotation/mod.rs b/src/kvp/annotation/mod.rs
new file mode 100644
index 000000000..4cb1d7f54
--- /dev/null
+++ b/src/kvp/annotation/mod.rs
@@ -0,0 +1,218 @@
+//! This module provides various types and functions to construct valid Kubernetes
+//! annotations. Annotations are key/value pairs, where the key must meet certain
+//! requirementens regarding length and character set. The value can contain
+//! **any** valid UTF-8 data.
+//!
+//! Additionally, the [`Annotation`] struct provides various helper functions to
+//! construct commonly used annotations across the Stackable Data Platform, like
+//! the secret scope or class.
+//!
+//! See <https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/>
+//! for more information on Kubernetes annotations.
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    convert::Infallible,
+    fmt::Display,
+};
+
+use crate::{
+    builder::SecretOperatorVolumeScope,
+    kvp::{Key, KeyValuePair, KeyValuePairError, KeyValuePairs, KeyValuePairsError},
+};
+
+mod value;
+
+pub use value::*;
+
+/// A type alias for errors returned when construction of an annotation fails.
+pub type AnnotationsError = KeyValuePairsError<Infallible>;
+
+/// A type alias for errors returned when construction or manipulation of a set
+/// of annotations fails.
+pub type AnnotationError = KeyValuePairError<Infallible>;
+
+/// A specialized implementation of a key/value pair representing Kubernetes
+/// annotations.
+///
+/// The validation of the annotation value can **never** fail, as [`str`] is
+/// guaranteed  to only contain valid UTF-8 data - which is the only
+/// requirement for a valid Kubernetes annotation value.
+///
+/// See <https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/>
+/// for more information on Kubernetes annotations.
+#[derive(Debug)]
+pub struct Annotation(KeyValuePair<AnnotationValue>);
+
+impl<T> TryFrom<(T, T)> for Annotation
+where
+    T: AsRef<str>,
+{
+    type Error = AnnotationError;
+
+    fn try_from(value: (T, T)) -> Result<Self, Self::Error> {
+        let kvp = KeyValuePair::try_from(value)?;
+        Ok(Self(kvp))
+    }
+}
+
+impl Display for Annotation {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl Annotation {
+    /// Returns an immutable reference to the annotation's [`Key`].
+    pub fn key(&self) -> &Key {
+        self.0.key()
+    }
+
+    /// Returns an immutable reference to the annotation's value.
+    pub fn value(&self) -> &AnnotationValue {
+        self.0.value()
+    }
+
+    /// Consumes self and returns the inner [`KeyValuePair<AnnotationValue>`].
+    pub fn into_inner(self) -> KeyValuePair<AnnotationValue> {
+        self.0
+    }
+
+    /// Constructs a `secrets.stackable.tech/class` annotation.
+    pub fn secret_class(secret_class: &str) -> Result<Self, AnnotationError> {
+        let kvp = KeyValuePair::try_from(("secrets.stackable.tech/class", secret_class))?;
+        Ok(Self(kvp))
+    }
+
+    /// Constructs a `secrets.stackable.tech/scope` annotation.
+    pub fn secret_scope(
+        scopes: impl AsRef<[SecretOperatorVolumeScope]>,
+    ) -> Result<Self, AnnotationError> {
+        let mut value = String::new();
+
+        for scope in scopes.as_ref() {
+            if !value.is_empty() {
+                value.push(',');
+            }
+
+            match scope {
+                SecretOperatorVolumeScope::Node => value.push_str("node"),
+                SecretOperatorVolumeScope::Pod => value.push_str("pod"),
+                SecretOperatorVolumeScope::Service { name } => {
+                    value.push_str("service=");
+                    value.push_str(name);
+                }
+            }
+        }
+
+        let kvp = KeyValuePair::try_from(("secrets.stackable.tech/scope", value))?;
+        Ok(Self(kvp))
+    }
+
+    /// Constructs a `secrets.stackable.tech/format` annotation.
+    pub fn secret_format(format: &str) -> Result<Self, AnnotationError> {
+        let kvp = KeyValuePair::try_from(("secrets.stackable.tech/format", format))?;
+        Ok(Self(kvp))
+    }
+
+    /// Constructs a `secrets.stackable.tech/kerberos.service.names` annotation.
+    pub fn kerberos_service_names(names: impl AsRef<[String]>) -> Result<Self, AnnotationError> {
+        let names = names.as_ref().join(",");
+        let kvp = KeyValuePair::try_from(("secrets.stackable.tech/kerberos.service.names", names))?;
+        Ok(Self(kvp))
+    }
+
+    /// Constructs a `secrets.stackable.tech/format.compatibility.tls-pkcs12.password`
+    /// annotation.
+    pub fn tls_pkcs12_password(password: &str) -> Result<Self, AnnotationError> {
+        let kvp = KeyValuePair::try_from((
+            "secrets.stackable.tech/format.compatibility.tls-pkcs12.password",
+            password,
+        ))?;
+        Ok(Self(kvp))
+    }
+}
+
+/// A validated set/list of Kubernetes annotations.
+///
+/// It provides selected associated functions to manipulate the set of
+/// annotations, like inserting or extending.
+#[derive(Clone, Debug, Default)]
+pub struct Annotations(KeyValuePairs<AnnotationValue>);
+
+impl TryFrom<BTreeMap<String, String>> for Annotations {
+    type Error = AnnotationError;
+
+    fn try_from(value: BTreeMap<String, String>) -> Result<Self, Self::Error> {
+        let kvps = KeyValuePairs::try_from(value)?;
+        Ok(Self(kvps))
+    }
+}
+
+impl FromIterator<KeyValuePair<AnnotationValue>> for Annotations {
+    fn from_iter<T: IntoIterator<Item = KeyValuePair<AnnotationValue>>>(iter: T) -> Self {
+        let kvps = KeyValuePairs::from_iter(iter);
+        Self(kvps)
+    }
+}
+
+impl From<Annotations> for BTreeMap<String, String> {
+    fn from(value: Annotations) -> Self {
+        value.0.into()
+    }
+}
+
+// TODO (Techassi): Use https://crates.io/crates/delegate to forward function impls
+impl Annotations {
+    /// Creates a new empty list of [`Annotations`].
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Creates a new list of [`Annotations`] from `pairs`.
+    pub fn new_with(pairs: BTreeSet<KeyValuePair<AnnotationValue>>) -> Self {
+        Self(KeyValuePairs::new_with(pairs))
+    }
+
+    /// Tries to insert a new [`Annotation`]. It ensures there are no duplicate
+    /// entries. Trying to insert duplicated data returns an error. If no such
+    /// check is required, use the `insert` function instead.
+    pub fn try_insert(&mut self, annotation: Annotation) -> Result<&mut Self, AnnotationsError> {
+        self.0.try_insert(annotation.0)?;
+        Ok(self)
+    }
+
+    /// Inserts a new [`Annotation`]. This function will overide any existing
+    /// annotation already present. If this behaviour is not desired, use the
+    /// `try_insert` function instead.
+    pub fn insert(&mut self, annotation: Annotation) -> &mut Self {
+        self.0.insert(annotation.0);
+        self
+    }
+
+    /// Extends `self` with `other`.
+    pub fn extend(&mut self, other: Self) {
+        self.0.extend(other.0)
+    }
+
+    /// Returns the number of annotations.
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    /// Returns if the set of annotations is empty.
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
+    /// Returns if the set of annotations contains the provided `label`.
+    /// Failure to parse/validate the [`KeyValuePair`] will return `false`.
+    pub fn contains(&self, annotation: impl TryInto<KeyValuePair<AnnotationValue>>) -> bool {
+        self.0.contains(annotation)
+    }
+
+    /// Returns if the set of annotations contains a label with the provided
+    /// `key`. Failure to parse/validate the [`Key`] will return `false`.
+    pub fn contains_key(&self, key: impl TryInto<Key>) -> bool {
+        self.0.contains_key(key)
+    }
+}
diff --git a/src/kvp/annotation/value.rs b/src/kvp/annotation/value.rs
new file mode 100644
index 000000000..3071e92e9
--- /dev/null
+++ b/src/kvp/annotation/value.rs
@@ -0,0 +1,39 @@
+use std::{convert::Infallible, fmt::Display, ops::Deref, str::FromStr};
+
+use crate::kvp::Value;
+
+/// A validated Kubernetes annotation value, which only requires valid UTF-8
+/// data.
+///
+/// Since [`str`] and [`String`] are guaranteed to be valid UTF-8 data, we
+/// don't perform any additional validation.
+///
+/// This wrapper type solely exists to mirror the label value type.
+#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
+pub struct AnnotationValue(String);
+
+impl Value for AnnotationValue {
+    type Error = Infallible;
+}
+
+impl FromStr for AnnotationValue {
+    type Err = Infallible;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        Ok(Self(input.to_owned()))
+    }
+}
+
+impl Deref for AnnotationValue {
+    type Target = str;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl Display for AnnotationValue {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
diff --git a/src/kvp/consts.rs b/src/kvp/consts.rs
new file mode 100644
index 000000000..a8e9ecce3
--- /dev/null
+++ b/src/kvp/consts.rs
@@ -0,0 +1,40 @@
+//! This module contains various label and annotation related constants used by
+//! Kubernetes. Most constants define well-known `app.kubernetes.io/<NAME>`
+//! keys. These constants can be used to construct various labels or annotations
+//! without sprinkling magic values all over the code.
+use const_format::concatcp;
+
+/// The well-known Kubernetes app key prefix.
+const K8S_APP_KEY_PREFIX: &str = "app.kubernetes.io/";
+
+/// The well-known Kubernetes app name key `app.kubernetes.io/name`. It is used
+/// to label the application with a name, e.g. `mysql`.
+pub const K8S_APP_NAME_KEY: &str = concatcp!(K8S_APP_KEY_PREFIX, "name");
+
+/// The well-known Kubernetes app instance key `app.kubernetes.io/instance`. It
+/// is used to identify the instance of an application, e.g. `mysql-abcxyz`.
+pub const K8S_APP_INSTANCE_KEY: &str = concatcp!(K8S_APP_KEY_PREFIX, "instance");
+
+/// The well-known Kubernetes app version key `app.kubernetes.io/version`. It is
+/// used to indicate the current version of the application. The value can
+/// represent a semantic version or a revision, e.g. `5.7.21`.
+pub const K8S_APP_VERSION_KEY: &str = concatcp!(K8S_APP_KEY_PREFIX, "version");
+
+/// The well-known Kubernetes app component key `app.kubernetes.io/component`.
+/// It is used to specify the compoent within the architecture, e.g. `database`.
+pub const K8S_APP_COMPONENT_KEY: &str = concatcp!(K8S_APP_KEY_PREFIX, "component");
+
+/// The well-known Kubernetes app part-of key `app.kubernetes.io/part-of`. It is
+/// used to specify the name of a higher level application this one is part of,
+/// e.g. `wordpress`.
+pub const K8S_APP_PART_OF_KEY: &str = concatcp!(K8S_APP_KEY_PREFIX, "part-of");
+
+/// The well-known Kubernetes app managed-by key `app.kubernetes.io/managed-by`.
+/// It is used to indicate what tool is being used to manage the operation of
+/// an application, e.g. `helm`.
+pub const K8S_APP_MANAGED_BY_KEY: &str = concatcp!(K8S_APP_KEY_PREFIX, "managed-by");
+
+/// The well-kown Kubernetes app role-group key `app.kubernetes.io/role-group`.
+/// It is used to specify to which role group this application belongs to, e.g.
+/// `worker`.
+pub const K8S_APP_ROLE_GROUP_KEY: &str = concatcp!(K8S_APP_KEY_PREFIX, "role-group");
diff --git a/src/kvp/key.rs b/src/kvp/key.rs
new file mode 100644
index 000000000..2d99f4a07
--- /dev/null
+++ b/src/kvp/key.rs
@@ -0,0 +1,368 @@
+use std::{fmt::Display, ops::Deref, str::FromStr};
+
+use lazy_static::lazy_static;
+use regex::Regex;
+use snafu::{ensure, ResultExt, Snafu};
+
+const KEY_PREFIX_MAX_LEN: usize = 253;
+const KEY_NAME_MAX_LEN: usize = 63;
+
+lazy_static! {
+    static ref KEY_PREFIX_REGEX: Regex =
+        Regex::new(r"^[a-zA-Z](\.?[a-zA-Z0-9-])*\.[a-zA-Z]{2,}\.?$").unwrap();
+    static ref KEY_NAME_REGEX: Regex =
+        Regex::new(r"^[a-z0-9A-Z]([a-z0-9A-Z-_.]*[a-z0-9A-Z]+)?$").unwrap();
+}
+
+/// The error type for key parsing/validation operations.
+///
+/// This error will be returned if the input is empty, the parser encounters
+/// multiple prefixes or any deeper errors occur during key prefix and key name
+/// parsing.
+#[derive(Debug, PartialEq, Snafu)]
+pub enum KeyError {
+    /// Indicates that the input is empty. The key must at least contain a name.
+    /// The prefix is optional.
+    #[snafu(display("key input cannot be empty"))]
+    EmptyInput,
+
+    /// Indicates that the input contains multiple nested prefixes, e.g.
+    /// `app.kubernetes.io/nested/name`. Valid keys only contain one prefix
+    /// like `app.kubernetes.io/name`.
+    #[snafu(display("key prefixes cannot be nested, only use a single slash"))]
+    NestedPrefix,
+
+    /// Indicates that the key prefix failed to parse. See [`KeyPrefixError`]
+    /// for more information about error causes.
+    #[snafu(display("failed to parse key prefix"))]
+    KeyPrefixError { source: KeyPrefixError },
+
+    /// Indicates that the key name failed to parse. See [`KeyNameError`] for
+    /// more information about error causes.
+    #[snafu(display("failed to parse key name"))]
+    KeyNameError { source: KeyNameError },
+}
+
+/// The key of a a key/value pair. It contains an optional prefix, and a
+/// required name.
+///
+/// The general format is `(<PREFIX>/)<NAME>`. Further, the Kubernetes
+/// documentation defines the format and allowed characters in more detail
+/// [here][k8s-labels]. A [`Key`] is always validated. It also doesn't provide
+/// any associated functions which enable unvalidated manipulation of the inner
+/// values.
+///
+/// [k8s-labels]: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub struct Key {
+    prefix: Option<KeyPrefix>,
+    name: KeyName,
+}
+
+impl FromStr for Key {
+    type Err = KeyError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        let input = input.trim();
+
+        // The input cannot be empty
+        ensure!(!input.is_empty(), EmptyInputSnafu);
+
+        // Split the input up into the optional prefix and name
+        let parts = input.split('/').collect::<Vec<_>>();
+
+        let (prefix, name) = match parts[..] {
+            [name] => (None, name),
+            [prefix, name] => (Some(prefix), name),
+            _ => return NestedPrefixSnafu.fail(),
+        };
+
+        let key = Self {
+            prefix: prefix
+                .map(KeyPrefix::from_str)
+                .transpose()
+                .context(KeyPrefixSnafu)?,
+            name: KeyName::from_str(name).context(KeyNameSnafu)?,
+        };
+
+        Ok(key)
+    }
+}
+
+impl TryFrom<&str> for Key {
+    type Error = KeyError;
+
+    fn try_from(value: &str) -> Result<Self, Self::Error> {
+        Self::from_str(value)
+    }
+}
+
+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 {
+    /// Retrieves the key's prefix.
+    ///
+    /// ```
+    /// use std::str::FromStr;
+    /// use stackable_operator::kvp::{Key, KeyPrefix};
+    ///
+    /// let key = Key::from_str("stackable.tech/vendor").unwrap();
+    /// let prefix = KeyPrefix::from_str("stackable.tech").unwrap();
+    ///
+    /// assert_eq!(key.prefix(), Some(&prefix));
+    /// ```
+    pub fn prefix(&self) -> Option<&KeyPrefix> {
+        self.prefix.as_ref()
+    }
+
+    /// Adds or replaces the key prefix. This takes a parsed and validated
+    /// [`KeyPrefix`] as a parameter. If instead you want to use a raw value,
+    /// use the [`Key::try_add_prefix()`] function instead.
+    pub fn add_prefix(&mut self, prefix: KeyPrefix) {
+        self.prefix = Some(prefix)
+    }
+
+    /// Adds or replaces the key prefix by parsing and validation raw input. If
+    /// instead you already have a parsed and validated [`KeyPrefix`], use the
+    /// [`Key::add_prefix()`] function instead.
+    pub fn try_add_prefix(&mut self, prefix: impl AsRef<str>) -> Result<&mut Self, KeyError> {
+        self.prefix = Some(KeyPrefix::from_str(prefix.as_ref()).context(KeyPrefixSnafu)?);
+        Ok(self)
+    }
+
+    /// Retrieves the key's name.
+    ///
+    /// ```
+    /// use std::str::FromStr;
+    /// use stackable_operator::kvp::{Key, KeyName};
+    ///
+    /// let key = Key::from_str("stackable.tech/vendor").unwrap();
+    /// let name = KeyName::from_str("vendor").unwrap();
+    ///
+    /// assert_eq!(key.name(), &name);
+    /// ```
+    pub fn name(&self) -> &KeyName {
+        &self.name
+    }
+
+    /// Sets the key name. This takes a parsed and validated [`KeyName`] as a
+    /// parameter. If instead you want to use a raw value, use the
+    /// [`Key::try_set_name()`] function instead.
+    pub fn set_name(&mut self, name: KeyName) {
+        self.name = name
+    }
+
+    /// Sets the key name by parsing and validation raw input. If instead you
+    /// already have a parsed and validated [`KeyName`], use the
+    /// [`Key::set_name()`] function instead.
+    pub fn try_set_name(&mut self, name: impl AsRef<str>) -> Result<&mut Self, KeyError> {
+        self.name = KeyName::from_str(name.as_ref()).context(KeyNameSnafu)?;
+        Ok(self)
+    }
+}
+
+/// The error type for key prefix parsing/validation operations.
+#[derive(Debug, PartialEq, Snafu)]
+pub enum KeyPrefixError {
+    /// Indicates that the key prefix segment is empty, which is not permitted
+    /// when the key indicates that a prefix is present (via a slash). This
+    /// prevents keys like `/name`.
+    #[snafu(display("prefix segment of key cannot be empty"))]
+    PrefixEmpty,
+
+    /// Indicates that the key prefix segment exceeds the mamximum length of
+    /// 253 ASCII characters. It additionally reports how many characters were
+    /// encountered during parsing / validation.
+    #[snafu(display("prefix segment of key exceeds the maximum length - expected 253 characters or less, got {length}"))]
+    PrefixTooLong { length: usize },
+
+    /// Indidcates that the key prefix segment contains non-ASCII characters
+    /// which the Kubernetes spec does not permit.
+    #[snafu(display("prefix segment of key contains non-ascii characters"))]
+    PrefixNotAscii,
+
+    /// Indicates that the key prefix segment violates the specified Kubernetes
+    /// format.
+    #[snafu(display("prefix segment of key violates kubernetes format"))]
+    PrefixInvalid,
+}
+
+/// A validated optional key prefix segment of a key.
+///
+/// Instances of this struct are always valid. [`KeyPrefix`] implements
+/// [`Deref`], which enables read-only access to the inner value (a [`String`]).
+/// It, however, does not implement [`DerefMut`](std::ops::DerefMut) which would
+/// enable unvalidated mutable access to inner values.
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub struct KeyPrefix(String);
+
+impl FromStr for KeyPrefix {
+    type Err = KeyPrefixError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        // The prefix cannot be empty when one is provided
+        ensure!(!input.is_empty(), PrefixEmptySnafu);
+
+        // The length of the prefix cannot exceed 253 characters
+        ensure!(
+            input.len() <= KEY_PREFIX_MAX_LEN,
+            PrefixTooLongSnafu {
+                length: input.len()
+            }
+        );
+
+        // The prefix cannot contain non-ascii characters
+        ensure!(input.is_ascii(), PrefixNotAsciiSnafu);
+
+        // The prefix must use the format specified by Kubernetes
+        ensure!(KEY_PREFIX_REGEX.is_match(input), PrefixInvalidSnafu);
+
+        Ok(Self(input.to_string()))
+    }
+}
+
+impl Deref for KeyPrefix {
+    type Target = String;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl Display for KeyPrefix {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+/// The error type for key name parsing/validation operations.
+#[derive(Debug, PartialEq, Snafu)]
+pub enum KeyNameError {
+    /// Indicates that the key name segment is empty. The key name is required
+    /// and therefore cannot be empty.
+    #[snafu(display("name segment of key cannot be empty"))]
+    NameEmpty,
+
+    /// Indicates that the key name sgement exceeds the maximum length of 63
+    /// ASCII characters. It additionally reports how many characters were
+    /// encountered during parsing / validation.
+    #[snafu(display("name segment of key exceeds the maximum length - expected 63 characters or less, got {length}"))]
+    NameTooLong { length: usize },
+
+    /// Indidcates that the key name segment contains non-ASCII characters
+    /// which the Kubernetes spec does not permit.
+    #[snafu(display("name segment of key contains non-ascii characters"))]
+    NameNotAscii,
+
+    /// Indicates that the key name segment violates the specified Kubernetes
+    /// format.
+    #[snafu(display("name segment of key violates kubernetes format"))]
+    NameInvalid,
+}
+
+/// A validated name segement of a key. This part of the key is required.
+///
+/// Instances of this struct are always valid. It also implements [`Deref`],
+/// which enables read-only access to the inner value (a [`String`]). It,
+/// however, does not implement [`DerefMut`](std::ops::DerefMut) which would
+/// enable unvalidated mutable access to inner values.
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub struct KeyName(String);
+
+impl FromStr for KeyName {
+    type Err = KeyNameError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        // The name cannot be empty
+        ensure!(!input.is_empty(), NameEmptySnafu);
+
+        // The length of the name cannot exceed 63 characters
+        ensure!(
+            input.len() <= KEY_NAME_MAX_LEN,
+            NameTooLongSnafu {
+                length: input.len()
+            }
+        );
+
+        // The name cannot contain non-ascii characters
+        ensure!(input.is_ascii(), NameNotAsciiSnafu);
+
+        // The name must use the format specified by Kubernetes
+        ensure!(KEY_NAME_REGEX.is_match(input), NameInvalidSnafu);
+
+        Ok(Self(input.to_string()))
+    }
+}
+
+impl Deref for KeyName {
+    type Target = String;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl Display for KeyName {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use rstest::rstest;
+
+    #[test]
+    fn key_with_prefix() {
+        let key = Key::from_str("stackable.tech/vendor").unwrap();
+
+        assert_eq!(key.prefix, Some(KeyPrefix("stackable.tech".into())));
+        assert_eq!(key.name, KeyName("vendor".into()));
+        assert_eq!(key.to_string(), "stackable.tech/vendor");
+    }
+
+    #[test]
+    fn key_without_prefix() {
+        let key = Key::from_str("vendor").unwrap();
+
+        assert_eq!(key.prefix, None);
+        assert_eq!(key.name, KeyName("vendor".into()));
+        assert_eq!(key.to_string(), "vendor");
+    }
+
+    #[rstest]
+    #[case("foo/bar/baz", KeyError::NestedPrefix)]
+    #[case("", KeyError::EmptyInput)]
+    fn invalid_key(#[case] input: &str, #[case] error: KeyError) {
+        let err = Key::from_str(input).unwrap_err();
+        assert_eq!(err, error);
+    }
+
+    #[rstest]
+    #[case("a".repeat(254), KeyPrefixError::PrefixTooLong { length: 254 })]
+    #[case("foo.", KeyPrefixError::PrefixInvalid)]
+    #[case("ä", KeyPrefixError::PrefixNotAscii)]
+    #[case("", KeyPrefixError::PrefixEmpty)]
+    fn invalid_key_prefix(#[case] input: String, #[case] error: KeyPrefixError) {
+        let err = KeyPrefix::from_str(&input).unwrap_err();
+        assert_eq!(err, error);
+    }
+
+    #[rstest]
+    #[case("a".repeat(64), KeyNameError::NameTooLong { length: 64 })]
+    #[case("foo-", KeyNameError::NameInvalid)]
+    #[case("ä", KeyNameError::NameNotAscii)]
+    #[case("", KeyNameError::NameEmpty)]
+    fn invalid_key_name(#[case] input: String, #[case] error: KeyNameError) {
+        let err = KeyName::from_str(&input).unwrap_err();
+        assert_eq!(err, error);
+    }
+}
diff --git a/src/kvp/label/mod.rs b/src/kvp/label/mod.rs
new file mode 100644
index 000000000..832b35306
--- /dev/null
+++ b/src/kvp/label/mod.rs
@@ -0,0 +1,308 @@
+//! This module provides various types and functions to construct valid
+//! Kubernetes labels. Labels are key/value pairs, where the key must meet
+//! certain requirementens regarding length and character set. The value can
+//! contain a limited set of ASCII characters.
+//!
+//! Additionally, the [`Label`] struct provides various helper functions to
+//! construct commonly used labels across the Stackable Data Platform, like
+//! the role_group or component.
+//!
+//! See <https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/>
+//! for more information on Kubernetes labels.
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    fmt::Display,
+};
+
+use kube::{Resource, ResourceExt};
+
+use crate::{
+    kvp::{
+        consts::{
+            K8S_APP_COMPONENT_KEY, K8S_APP_INSTANCE_KEY, K8S_APP_MANAGED_BY_KEY, K8S_APP_NAME_KEY,
+            K8S_APP_ROLE_GROUP_KEY, K8S_APP_VERSION_KEY,
+        },
+        Key, KeyValuePair, KeyValuePairError, KeyValuePairs, KeyValuePairsError, ObjectLabels,
+    },
+    utils::format_full_controller_name,
+};
+
+mod selector;
+mod value;
+
+pub use selector::*;
+pub use value::*;
+
+/// A type alias for errors returned when construction of a label fails.
+pub type LabelsError = KeyValuePairsError<LabelValueError>;
+
+/// A type alias for errors returned when construction or manipulation of a set
+/// of labels fails.
+pub type LabelError = KeyValuePairError<LabelValueError>;
+
+/// A specialized implementation of a key/value pair representing Kubernetes
+/// labels.
+///
+/// ```
+/// # use stackable_operator::kvp::Label;
+/// let label = Label::try_from(("stackable.tech/vendor", "Stackable")).unwrap();
+/// assert_eq!(label.to_string(), "stackable.tech/vendor=Stackable");
+/// ```
+///
+/// The validation of the label value can fail due to multiple reasons. It can
+/// only contain a limited set and combination of ASCII characters. See
+/// <https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/>
+/// for more information on Kubernetes labels.
+#[derive(Clone, Debug)]
+pub struct Label(KeyValuePair<LabelValue>);
+
+impl<T> TryFrom<(T, T)> for Label
+where
+    T: AsRef<str>,
+{
+    type Error = LabelError;
+
+    fn try_from(value: (T, T)) -> Result<Self, Self::Error> {
+        let kvp = KeyValuePair::try_from(value)?;
+        Ok(Self(kvp))
+    }
+}
+
+impl Display for Label {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl Label {
+    /// Returns an immutable reference to the label's [`Key`].
+    ///
+    /// ```
+    /// # use stackable_operator::kvp::Label;
+    /// let label = Label::try_from(("stackable.tech/vendor", "Stackable")).unwrap();
+    /// assert_eq!(label.key().to_string(), "stackable.tech/vendor");
+    /// ```
+    pub fn key(&self) -> &Key {
+        self.0.key()
+    }
+
+    /// Returns an immutable reference to the label's value.
+    pub fn value(&self) -> &LabelValue {
+        self.0.value()
+    }
+
+    /// Consumes self and returns the inner [`KeyValuePair<LabelValue>`].
+    pub fn into_inner(self) -> KeyValuePair<LabelValue> {
+        self.0
+    }
+
+    /// Creates the `app.kubernetes.io/component` label with `role` as the
+    /// value. This function will return an error if `role` violates the required
+    /// Kubernetes restrictions.
+    pub fn component(component: &str) -> Result<Self, LabelError> {
+        let kvp = KeyValuePair::try_from((K8S_APP_COMPONENT_KEY, component))?;
+        Ok(Self(kvp))
+    }
+
+    /// Creates the `app.kubernetes.io/role-group` label with `role_group` as
+    /// the value. This function will return an error if `role_group` violates
+    /// the required Kubernetes restrictions.
+    pub fn role_group(role_group: &str) -> Result<Self, LabelError> {
+        let kvp = KeyValuePair::try_from((K8S_APP_ROLE_GROUP_KEY, role_group))?;
+        Ok(Self(kvp))
+    }
+
+    /// Creates the `app.kubernetes.io/managed-by` label with the formated
+    /// full controller name based on `operator_name` and `controller_name` as
+    /// the value. This function will return an error if the formatted controller
+    /// name violates the required Kubernetes restrictions.
+    pub fn managed_by(operator_name: &str, controller_name: &str) -> Result<Self, LabelError> {
+        let kvp = KeyValuePair::try_from((
+            K8S_APP_MANAGED_BY_KEY,
+            format_full_controller_name(operator_name, controller_name).as_str(),
+        ))?;
+        Ok(Self(kvp))
+    }
+
+    /// Creates the `app.kubernetes.io/version` label with `version` as the
+    /// value. This function will return an error if `role_group` violates the
+    /// required Kubernetes restrictions.
+    pub fn version(version: &str) -> Result<Self, LabelError> {
+        // NOTE (Techassi): Maybe use semver::Version
+        let kvp = KeyValuePair::try_from((K8S_APP_VERSION_KEY, version))?;
+        Ok(Self(kvp))
+    }
+}
+
+/// A validated set/list of Kubernetes labels.
+///
+/// It provides selected associated functions to manipulate the set of labels,
+/// like inserting or extending.
+#[derive(Clone, Debug, Default)]
+pub struct Labels(KeyValuePairs<LabelValue>);
+
+impl TryFrom<BTreeMap<String, String>> for Labels {
+    type Error = LabelError;
+
+    fn try_from(value: BTreeMap<String, String>) -> Result<Self, Self::Error> {
+        let kvps = KeyValuePairs::try_from(value)?;
+        Ok(Self(kvps))
+    }
+}
+
+impl FromIterator<KeyValuePair<LabelValue>> for Labels {
+    fn from_iter<T: IntoIterator<Item = KeyValuePair<LabelValue>>>(iter: T) -> Self {
+        let kvps = KeyValuePairs::from_iter(iter);
+        Self(kvps)
+    }
+}
+
+impl From<Labels> for BTreeMap<String, String> {
+    fn from(value: Labels) -> Self {
+        value.0.into()
+    }
+}
+
+// TODO (Techassi): Use https://crates.io/crates/delegate to forward function impls
+impl Labels {
+    /// Creates a new empty list of [`Labels`].
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Creates a new list of [`Labels`] from `pairs`.
+    pub fn new_with(pairs: BTreeSet<KeyValuePair<LabelValue>>) -> Self {
+        Self(KeyValuePairs::new_with(pairs))
+    }
+
+    /// Tries to insert a new [`Label`]. It ensures there are no duplicate
+    /// entries. Trying to insert duplicated data returns an error. If no such
+    /// check is required, use the `insert` function instead.
+    pub fn try_insert(&mut self, label: Label) -> Result<&mut Self, LabelsError> {
+        self.0.try_insert(label.0)?;
+        Ok(self)
+    }
+
+    /// Inserts a new [`Label`]. This function will overide any existing label
+    /// already present. If this behaviour is not desired, use the `try_insert`
+    /// function instead.
+    pub fn insert(&mut self, label: Label) -> &mut Self {
+        self.0.insert(label.0);
+        self
+    }
+
+    /// Extends `self` with `other`.
+    pub fn extend(&mut self, other: Self) {
+        self.0.extend(other.0)
+    }
+
+    /// Returns the number of labels.
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    /// Returns if the set of labels is empty.
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
+
+    /// Returns if the set of labels contains the provided `label`. Failure to
+    /// parse/validate the [`KeyValuePair`] will return `false`.
+    pub fn contains(&self, label: impl TryInto<KeyValuePair<LabelValue>>) -> bool {
+        self.0.contains(label)
+    }
+
+    /// Returns if the set of labels contains a label with the provided `key`.
+    /// Failure to parse/validate the [`Key`] will return `false`.
+    pub fn contains_key(&self, key: impl TryInto<Key>) -> bool {
+        self.0.contains_key(key)
+    }
+
+    /// Returns the recommended set of labels. The set includes these well-known
+    /// labels:
+    ///
+    /// - `app.kubernetes.io/role-group`
+    /// - `app.kubernetes.io/managed-by`
+    /// - `app.kubernetes.io/component`
+    /// - `app.kubernetes.io/instance`
+    /// - `app.kubernetes.io/version`
+    /// - `app.kubernetes.io/name`
+    ///
+    /// This function returns a result, because the parameter `object_labels`
+    /// can contain invalid data or can exceed the maximum allowed number of
+    /// characters.
+    pub fn recommended<R>(object_labels: ObjectLabels<R>) -> Result<Self, LabelError>
+    where
+        R: Resource,
+    {
+        let mut labels = Self::role_group_selector(
+            object_labels.owner,
+            object_labels.app_name,
+            object_labels.role,
+            object_labels.role_group,
+        )?;
+
+        let managed_by =
+            Label::managed_by(object_labels.operator_name, object_labels.controller_name)?;
+        let version = Label::version(object_labels.app_version)?;
+
+        labels.insert(managed_by);
+        labels.insert(version);
+
+        Ok(labels)
+    }
+
+    /// Returns the set of labels required to select the resource based on the
+    /// role group. The set contains role selector labels, see
+    /// [`Labels::role_selector`] for more details. Additionally, it contains
+    /// the `app.kubernetes.io/role-group` label with `role_group` as the value.
+    pub fn role_group_selector<R>(
+        owner: &R,
+        app_name: &str,
+        role: &str,
+        role_group: &str,
+    ) -> Result<Self, LabelError>
+    where
+        R: Resource,
+    {
+        let mut labels = Self::role_selector(owner, app_name, role)?;
+        labels.insert(Label::role_group(role_group)?);
+        Ok(labels)
+    }
+
+    /// Returns the set of labels required to select the resource based on the
+    /// role. The set contains the common labels, see [`Labels::common`] for
+    /// more details. Additionally, it contains the `app.kubernetes.io/component`
+    /// label with `role` as the value.
+    ///
+    /// This function returns a result, because the parameters `owner`, `app_name`,
+    /// and `role` can contain invalid data or can exceed the maximum allowed
+    /// number fo characters.
+    pub fn role_selector<R>(owner: &R, app_name: &str, role: &str) -> Result<Self, LabelError>
+    where
+        R: Resource,
+    {
+        let mut labels = Self::common(app_name, owner.name_any().as_str())?;
+        labels.insert(Label::component(role)?);
+        Ok(labels)
+    }
+
+    /// Returns a common set of labels, which are required to identify resources
+    /// that belong to a certain owner object, for example a `ZookeeperCluster`.
+    /// The set contains these well-known labels:
+    ///
+    /// - `app.kubernetes.io/instance` and
+    /// - `app.kubernetes.io/name`
+    ///
+    /// This function returns a result, because the parameters `app_name` and
+    /// `app_instance` can contain invalid data or can exceed the maximum
+    /// allowed number of characters.
+    pub fn common(app_name: &str, app_instance: &str) -> Result<Self, LabelError> {
+        let mut labels = Self::new();
+
+        labels.insert((K8S_APP_INSTANCE_KEY, app_instance).try_into()?);
+        labels.insert((K8S_APP_NAME_KEY, app_name).try_into()?);
+
+        Ok(labels)
+    }
+}
diff --git a/src/label_selector.rs b/src/kvp/label/selector.rs
similarity index 70%
rename from src/label_selector.rs
rename to src/kvp/label/selector.rs
index bb1992f26..4656e9fcc 100644
--- a/src/label_selector.rs
+++ b/src/kvp/label/selector.rs
@@ -1,32 +1,43 @@
-use crate::error::{Error, OperatorResult};
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;
 
-/// Takes a [`LabelSelector`] and converts it to a String that can be used in Kubernetes API calls.
-/// It will return an error if the LabelSelector contains illegal things (e.g. an `Exists` operator
-/// with a value).
-pub fn convert_label_selector_to_query_string(
-    label_selector: &LabelSelector,
-) -> OperatorResult<String> {
-    let mut query_string = String::new();
-
-    // match_labels are the "old" part of LabelSelectors.
-    // They are the equivalent for the "In" operator in match_expressions
-    // In a query string each key-value pair will be separated by an "=" and the pairs
-    // are then joined on commas.
-    // The whole match_labels part is optional so we only do this if there are match labels.
-    if let Some(label_map) = &label_selector.match_labels {
-        query_string.push_str(
-            &label_map
-                .iter()
-                .map(|(key, value)| format!("{key}={value}"))
-                .collect::<Vec<_>>()
-                .join(","),
-        );
-    }
+/// This trait extends the functionality of [`LabelSelector`].
+///
+/// Implementing this trait for any other type other than [`LabelSelector`]
+/// can result in unndefined behaviour.
+pub trait LabelSelectorExt {
+    type Error: std::error::Error;
+
+    /// Takes a [`LabelSelector`] and converts it to a String that can be used
+    /// in Kubernetes API calls. It will return an error if the LabelSelector
+    /// contains illegal things (e.g. an `Exists` operator with a value).
+    fn to_query_string(&self) -> Result<String, Self::Error>;
+}
+
+impl LabelSelectorExt for LabelSelector {
+    // NOTE (Techassi): This should be its own error
+    type Error = crate::error::Error;
+
+    fn to_query_string(&self) -> Result<String, Self::Error> {
+        let mut query_string = String::new();
+
+        // match_labels are the "old" part of LabelSelectors.
+        // They are the equivalent for the "In" operator in match_expressions
+        // In a query string each key-value pair will be separated by an "=" and the pairs
+        // are then joined on commas.
+        // The whole match_labels part is optional so we only do this if there are match labels.
+        if let Some(label_map) = &self.match_labels {
+            query_string.push_str(
+                &label_map
+                    .iter()
+                    .map(|(key, value)| format!("{key}={value}"))
+                    .collect::<Vec<_>>()
+                    .join(","),
+            );
+        }
 
-    // Match expressions are more complex than match labels, both can appear in the same API call
-    // They support these operators: "In", "NotIn", "Exists" and "DoesNotExist"
-    let expressions = label_selector.match_expressions.as_ref().map(|requirements| {
+        // Match expressions are more complex than match labels, both can appear in the same API call
+        // They support these operators: "In", "NotIn", "Exists" and "DoesNotExist"
+        let expressions = self.match_expressions.as_ref().map(|requirements| {
         // If we had match_labels AND we have match_expressions we need to separate those two
         // with a comma.
         if !requirements.is_empty() && !query_string.is_empty() {
@@ -38,7 +49,7 @@ pub fn convert_label_selector_to_query_string(
         // We then collect those Results into a single Result with the Error being the _first_ error.
         // This, unfortunately means, that we'll throw away all but one error.
         // TODO: Return all errors in one go: https://github.com/stackabletech/operator-rs/issues/127
-        let expression_string: Result<Vec<String>, Error> = requirements
+        let expression_string: Result<Vec<String>, crate::error::Error> = requirements
             .iter()
             .map(|requirement| match requirement.operator.as_str() {
                 // In and NotIn can be handled the same, they both map to a simple "key OPERATOR (values)" string
@@ -49,7 +60,7 @@ pub fn convert_label_selector_to_query_string(
                         operator.to_ascii_lowercase(),
                         values.join(", ")
                     )),
-                    _ => Err(Error::InvalidLabelSelector {
+                    _ => Err(crate::error::Error::InvalidLabelSelector {
                         message: format!(
                             "LabelSelector has no or empty values for [{operator}] operator"
                         ),
@@ -58,7 +69,7 @@ pub fn convert_label_selector_to_query_string(
                 // "Exists" is just the key and nothing else, if values have been specified it's an error
                 "Exists" => match &requirement.values {
                     Some(values) if !values.is_empty() => Err(
-                        Error::InvalidLabelSelector {
+                        crate::error::Error::InvalidLabelSelector {
                             message: "LabelSelector has [Exists] operator with values, this is not legal".to_string(),
                     }),
                     _ => Ok(requirement.key.to_string()),
@@ -66,14 +77,14 @@ pub fn convert_label_selector_to_query_string(
                 // "DoesNotExist" is similar to "Exists" but it is preceded by an exclamation mark
                 "DoesNotExist" => match &requirement.values {
                     Some(values) if !values.is_empty() => Err(
-                        Error::InvalidLabelSelector {
+                        crate::error::Error::InvalidLabelSelector {
                             message: "LabelSelector has [DoesNotExist] operator with values, this is not legal".to_string(),
                         }),
                     _ => Ok(format!("!{}", requirement.key))
                 }
                 op => {
                     Err(
-                        Error::InvalidLabelSelector {
+                        crate::error::Error::InvalidLabelSelector {
                             message: format!("LabelSelector has illegal/unknown operator [{op}]")
                         })
                 }
@@ -84,11 +95,12 @@ pub fn convert_label_selector_to_query_string(
 
     });
 
-    if let Some(expressions) = expressions.transpose()? {
-        query_string.push_str(&expressions.join(","));
-    };
+        if let Some(expressions) = expressions.transpose()? {
+            query_string.push_str(&expressions.join(","));
+        };
 
-    Ok(query_string)
+        Ok(query_string)
+    }
 }
 
 #[cfg(test)]
@@ -136,24 +148,21 @@ mod tests {
             match_labels: Some(match_labels.clone()),
         };
         assert_eq!(
+            ls.to_query_string().unwrap(),
             "foo=bar,hui=buh,foo in (bar),foo in (quick, bar),foo notin (quick, bar),foo,!foo",
-            convert_label_selector_to_query_string(&ls).unwrap()
         );
 
         let ls = LabelSelector {
             match_expressions: None,
             match_labels: Some(match_labels),
         };
-        assert_eq!(
-            "foo=bar,hui=buh",
-            convert_label_selector_to_query_string(&ls).unwrap()
-        );
+        assert_eq!(ls.to_query_string().unwrap(), "foo=bar,hui=buh",);
 
         let ls = LabelSelector {
             match_expressions: None,
             match_labels: None,
         };
-        assert_eq!("", convert_label_selector_to_query_string(&ls).unwrap());
+        assert_eq!(ls.to_query_string().unwrap(), "");
     }
 
     #[test]
@@ -170,7 +179,7 @@ mod tests {
             match_labels: None,
         };
 
-        convert_label_selector_to_query_string(&ls).unwrap();
+        ls.to_query_string().unwrap();
     }
 
     #[test]
@@ -187,7 +196,7 @@ mod tests {
             match_labels: None,
         };
 
-        convert_label_selector_to_query_string(&ls).unwrap();
+        ls.to_query_string().unwrap();
     }
 
     #[test]
@@ -204,6 +213,6 @@ mod tests {
             match_labels: None,
         };
 
-        convert_label_selector_to_query_string(&ls).unwrap();
+        ls.to_query_string().unwrap();
     }
 }
diff --git a/src/kvp/label/value.rs b/src/kvp/label/value.rs
new file mode 100644
index 000000000..4190b0545
--- /dev/null
+++ b/src/kvp/label/value.rs
@@ -0,0 +1,103 @@
+use std::{fmt::Display, ops::Deref, str::FromStr};
+
+use lazy_static::lazy_static;
+use regex::Regex;
+use snafu::{ensure, Snafu};
+
+use crate::kvp::Value;
+
+const LABEL_VALUE_MAX_LEN: usize = 63;
+
+lazy_static! {
+    static ref LABEL_VALUE_REGEX: Regex =
+        Regex::new(r"^[a-z0-9A-Z]([a-z0-9A-Z-_.]*[a-z0-9A-Z]+)?$").unwrap();
+}
+
+/// The error type for label value parse/validation operations.
+#[derive(Debug, PartialEq, Snafu)]
+pub enum LabelValueError {
+    /// Indicates that the label value exceeds the maximum length of 63 ASCII
+    /// characters. It additionally reports how many characters were
+    /// encountered during parsing / validation.
+    #[snafu(display(
+        "value exceeds the maximum length - expected 63 characters or less, got {length}"
+    ))]
+    ValueTooLong { length: usize },
+
+    /// Indicates that the label value contains non-ASCII characters which the
+    /// Kubernetes spec does not permit.
+    #[snafu(display("value contains non-ascii characters"))]
+    ValueNotAscii,
+
+    /// Indicates that the label value violates the specified Kubernetes format.
+    #[snafu(display("value violates kubernetes format"))]
+    ValueInvalid,
+}
+
+/// A validated Kubernetes label value.
+///
+/// Instances of this struct are always valid. The format and valid characters
+/// are described [here][k8s-labels]. It also implements [`Deref`], which
+/// enables read-only access to the inner value (a [`String`]). It, however,
+/// does not implement [`DerefMut`](std::ops::DerefMut) which would enable
+/// unvalidated mutable access to inner values.
+///
+/// [k8s-labels]: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/
+#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord)]
+pub struct LabelValue(String);
+
+impl Value for LabelValue {
+    type Error = LabelValueError;
+}
+
+impl FromStr for LabelValue {
+    type Err = LabelValueError;
+
+    fn from_str(input: &str) -> Result<Self, Self::Err> {
+        // The length of the value cannot exceed 63 characters, but can be
+        // empty
+        ensure!(
+            input.len() <= LABEL_VALUE_MAX_LEN,
+            ValueTooLongSnafu {
+                length: input.len()
+            }
+        );
+
+        // The value cannot contain non-ascii characters
+        ensure!(input.is_ascii(), ValueNotAsciiSnafu);
+
+        // The value must use the format specified by Kubernetes
+        ensure!(LABEL_VALUE_REGEX.is_match(input), ValueInvalidSnafu);
+
+        Ok(Self(input.to_string()))
+    }
+}
+
+impl Deref for LabelValue {
+    type Target = str;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl Display for LabelValue {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+    use rstest::rstest;
+
+    #[rstest]
+    #[case("a".repeat(64), LabelValueError::ValueTooLong { length: 64 })]
+    #[case("foo-", LabelValueError::ValueInvalid)]
+    #[case("ä", LabelValueError::ValueNotAscii)]
+    fn invalid_value(#[case] input: String, #[case] error: LabelValueError) {
+        let err = LabelValue::from_str(&input).unwrap_err();
+        assert_eq!(err, error);
+    }
+}
diff --git a/src/kvp/mod.rs b/src/kvp/mod.rs
new file mode 100644
index 000000000..1f0d5e4bf
--- /dev/null
+++ b/src/kvp/mod.rs
@@ -0,0 +1,359 @@
+//! Utility functions and data structures the create and manage Kubernetes
+//! key/value pairs, like labels and annotations.
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    fmt::Display,
+    ops::Deref,
+    str::FromStr,
+};
+
+use snafu::{ensure, ResultExt, Snafu};
+
+mod annotation;
+pub mod consts;
+mod key;
+mod label;
+mod value;
+
+pub use annotation::*;
+pub use key::*;
+pub use label::*;
+pub use value::*;
+
+/// The error type for key/value pair parsing/validating operations.
+#[derive(Debug, PartialEq, Snafu)]
+pub enum KeyValuePairError<E>
+where
+    E: std::error::Error + 'static,
+{
+    /// Indicates that the key failed to parse. See [`KeyError`] for more
+    /// information about the error causes.
+    #[snafu(display("failed to parse key of key/value pair"))]
+    InvalidKey { source: KeyError },
+
+    /// Indicates that the value failed to parse.
+    #[snafu(display("failed to parse value of key/value pair"))]
+    InvalidValue { source: E },
+}
+
+/// A validated Kubernetes key/value pair.
+///
+/// These pairs can be used as Kubernetes labels or annotations. A pair can be
+/// parsed from a `(str, str)` tuple.
+///
+/// ### Examples
+///
+/// This example describes the usage of [`Label`], which is a specialized
+/// [`KeyValuePair`]. The implementation makes sure that both the key (comprised
+/// of optional prefix and name) and the value are validated according to the
+/// Kubernetes spec linked [below](#links).
+///
+/// ```
+/// # use stackable_operator::kvp::Label;
+/// let label = Label::try_from(("stackable.tech/vendor", "Stackable")).unwrap();
+/// assert_eq!(label.to_string(), "stackable.tech/vendor=Stackable");
+/// ```
+///
+/// ---
+///
+/// [`KeyValuePair`] is generic over the value. This allows implementors to
+/// write custom validation logic for different value requirements. This
+/// library provides two implementations out of the box: [`AnnotationValue`]
+/// and [`LabelValue`]. Custom implementations need to implement the required
+/// trait [`Value`].
+///
+/// ```ignore
+/// use stackable_operator::kvp::{KeyValuePair, Value};
+/// use serde::Serialize;
+///
+/// #[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Serialize)]
+/// struct MyValue(String);
+///
+/// impl Value for MyValue {
+///     // Implementation omitted for brevity
+/// }
+///
+/// let kvp = KeyValuePair::<MyValue>::try_from(("key", "my_custom_value"));
+/// ```
+///
+/// Implementing [`Value`] requires various other trait implementations like
+/// [`Deref`] and [`FromStr`]. Check out the documentation for the [`Value`]
+/// trait for a more detailed implementation guide.
+///
+/// ### Links
+///
+/// - <https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/>
+/// - <https://kubernetes.io/docs/concepts/overview/working-with-objects/annotations/>
+#[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord)]
+pub struct KeyValuePair<V>
+where
+    V: Value,
+{
+    key: Key,
+    value: V,
+}
+
+impl<T, K, V> TryFrom<(T, K)> for KeyValuePair<V>
+where
+    T: AsRef<str>,
+    K: AsRef<str>,
+    V: Value,
+{
+    type Error = KeyValuePairError<V::Error>;
+
+    fn try_from(value: (T, K)) -> Result<Self, Self::Error> {
+        let key = Key::from_str(value.0.as_ref()).context(InvalidKeySnafu)?;
+        let value = V::from_str(value.1.as_ref()).context(InvalidValueSnafu)?;
+
+        Ok(Self { key, value })
+    }
+}
+
+impl<V> Display for KeyValuePair<V>
+where
+    V: Value,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{}={}", self.key, self.value)
+    }
+}
+
+impl<V> KeyValuePair<V>
+where
+    V: Value,
+{
+    /// Creates a new [`KeyValuePair`] from a validated [`Key`] and value.
+    pub fn new(key: Key, value: V) -> Self {
+        Self { key, value }
+    }
+
+    /// Returns an immutable reference to the pair's [`Key`].
+    pub fn key(&self) -> &Key {
+        &self.key
+    }
+
+    /// Returns an immutable reference to the pair's value.
+    pub fn value(&self) -> &V {
+        &self.value
+    }
+}
+
+#[derive(Debug, Snafu)]
+pub enum KeyValuePairsError<E>
+where
+    E: std::error::Error + 'static,
+{
+    #[snafu(display("key/value pair already present"))]
+    AlreadyPresent,
+
+    #[snafu(display("failed to parse key/value pair"))]
+    KeyValuePairParse { source: KeyValuePairError<E> },
+}
+
+/// A validated set/list of Kubernetes key/value pairs.
+#[derive(Clone, Debug, Default)]
+pub struct KeyValuePairs<V: Value>(BTreeSet<KeyValuePair<V>>);
+
+impl<V> TryFrom<BTreeMap<String, String>> for KeyValuePairs<V>
+where
+    V: Value,
+{
+    type Error = KeyValuePairError<V::Error>;
+
+    fn try_from(map: BTreeMap<String, String>) -> Result<Self, Self::Error> {
+        let pairs = map
+            .into_iter()
+            .map(KeyValuePair::try_from)
+            .collect::<Result<BTreeSet<_>, KeyValuePairError<V::Error>>>()?;
+
+        Ok(Self(pairs))
+    }
+}
+
+impl<V> FromIterator<KeyValuePair<V>> for KeyValuePairs<V>
+where
+    V: Value,
+{
+    fn from_iter<T: IntoIterator<Item = KeyValuePair<V>>>(iter: T) -> Self {
+        Self(iter.into_iter().collect())
+    }
+}
+
+impl<V> From<KeyValuePairs<V>> for BTreeMap<String, String>
+where
+    V: Value,
+{
+    fn from(value: KeyValuePairs<V>) -> Self {
+        value
+            .iter()
+            .map(|pair| (pair.key().to_string(), pair.value().to_string()))
+            .collect()
+    }
+}
+
+impl<V> Deref for KeyValuePairs<V>
+where
+    V: Value,
+{
+    type Target = BTreeSet<KeyValuePair<V>>;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl<V> KeyValuePairs<V>
+where
+    V: Value + std::default::Default,
+{
+    /// Creates a new empty list of [`KeyValuePair`]s.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Creates a new list of [`KeyValuePair`]s from `pairs`.
+    pub fn new_with(pairs: BTreeSet<KeyValuePair<V>>) -> Self {
+        Self(pairs)
+    }
+
+    /// Extends `self` with `other`.
+    pub fn extend(&mut self, other: Self) {
+        self.0.extend(other.0);
+    }
+
+    pub fn try_insert(
+        &mut self,
+        kvp: KeyValuePair<V>,
+    ) -> Result<&mut Self, KeyValuePairsError<V::Error>> {
+        ensure!(!self.0.contains(&kvp), AlreadyPresentSnafu);
+
+        self.0.insert(kvp);
+        Ok(self)
+    }
+
+    pub fn insert(&mut self, kvp: KeyValuePair<V>) -> &mut Self {
+        self.0.insert(kvp);
+        self
+    }
+
+    pub fn contains(&self, kvp: impl TryInto<KeyValuePair<V>>) -> bool {
+        let Ok(kvp) = kvp.try_into() else {return false};
+        self.0.contains(&kvp)
+    }
+
+    pub fn contains_key(&self, key: impl TryInto<Key>) -> bool {
+        let Ok(key) = key.try_into() else {return false};
+
+        for kvp in &self.0 {
+            if kvp.key == key {
+                return true;
+            }
+        }
+
+        false
+    }
+}
+
+/// A recommended set of labels to set on objects created by Stackable
+/// operators or management tools.
+#[derive(Debug, Clone, Copy)]
+pub struct ObjectLabels<'a, T> {
+    /// Reference to the k8s object owning the created resource, such as
+    /// `HdfsCluster` or `TrinoCluster`.
+    pub owner: &'a T,
+
+    /// The name of the app being managed, such as `zookeeper`.
+    pub app_name: &'a str,
+
+    /// The version of the app being managed (not of the operator).
+    ///
+    /// If setting this label on a Stackable product then please use
+    /// [`ResolvedProductImage::app_version_label`][avl].
+    ///
+    /// This version should include the Stackable version, such as
+    /// `3.0.0-stackable23.11`. If the Stackable version is not known, then
+    /// the product version should be used together with a suffix (if possible).
+    /// If a custom product image is provided by the user (in which case only
+    /// the product version is known), then the format `3.0.0-<tag-of-custom-image>`
+    /// should be used.
+    ///
+    /// However, this is pure documentation and should not be parsed.
+    ///
+    /// [avl]: crate::commons::product_image_selection::ResolvedProductImage::app_version_label
+    pub app_version: &'a str,
+
+    /// The DNS-style name of the operator managing the object (such as `zookeeper.stackable.tech`)
+    pub operator_name: &'a str,
+
+    /// The name of the controller inside of the operator managing the object (such as `zookeepercluster`)
+    pub controller_name: &'a str,
+
+    /// The role that this object belongs to
+    pub role: &'a str,
+
+    /// The role group that this object belongs to
+    pub role_group: &'a str,
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn try_from_tuple() {
+        let label = Label::try_from(("stackable.tech/vendor", "Stackable")).unwrap();
+
+        assert_eq!(
+            label.key(),
+            &Key::from_str("stackable.tech/vendor").unwrap()
+        );
+        assert_eq!(label.value(), &LabelValue::from_str("Stackable").unwrap());
+
+        assert_eq!(label.to_string(), "stackable.tech/vendor=Stackable");
+    }
+
+    #[test]
+    fn labels_from_iter() {
+        let labels = Labels::from_iter([
+            KeyValuePair::try_from(("stackable.tech/managed-by", "stackablectl")).unwrap(),
+            KeyValuePair::try_from(("stackable.tech/vendor", "Stackable")).unwrap(),
+        ]);
+
+        assert_eq!(labels.len(), 2);
+    }
+
+    #[test]
+    fn labels_try_from_map() {
+        let map = BTreeMap::from([
+            ("stackable.tech/vendor".to_string(), "Stackable".to_string()),
+            (
+                "stackable.tech/managed-by".to_string(),
+                "stackablectl".to_string(),
+            ),
+        ]);
+
+        let labels = Labels::try_from(map).unwrap();
+        assert_eq!(labels.len(), 2);
+    }
+
+    #[test]
+    fn labels_into_map() {
+        let pairs = BTreeSet::from([
+            KeyValuePair::try_from(("stackable.tech/managed-by", "stackablectl")).unwrap(),
+            KeyValuePair::try_from(("stackable.tech/vendor", "Stackable")).unwrap(),
+        ]);
+
+        let labels = Labels::new_with(pairs);
+        let map: BTreeMap<String, String> = labels.into();
+
+        assert_eq!(map.len(), 2);
+    }
+
+    #[test]
+    fn contains() {
+        let labels = Labels::common("test", "test-01").unwrap();
+
+        assert!(labels.contains(("app.kubernetes.io/name", "test")));
+        assert!(labels.contains_key("app.kubernetes.io/instance"))
+    }
+}
diff --git a/src/kvp/value.rs b/src/kvp/value.rs
new file mode 100644
index 000000000..b80b2adec
--- /dev/null
+++ b/src/kvp/value.rs
@@ -0,0 +1,15 @@
+use std::{
+    error::Error,
+    fmt::{Debug, Display},
+    ops::Deref,
+    str::FromStr,
+};
+
+/// Trait which ensures the value of [`KeyValuePair`][crate::kvp::KeyValuePair]
+/// is validated. Different value implementations should use [`FromStr`] to
+/// parse and validate the value based on the requirements.
+pub trait Value:
+    Deref<Target = str> + FromStr<Err = Self::Error> + Clone + Display + Eq + Ord
+{
+    type Error: Error + Debug + 'static;
+}
diff --git a/src/labels.rs b/src/labels.rs
deleted file mode 100644
index a788a8ff2..000000000
--- a/src/labels.rs
+++ /dev/null
@@ -1,120 +0,0 @@
-use crate::utils::format_full_controller_name;
-use const_format::concatcp;
-use kube::api::{Resource, ResourceExt};
-use std::collections::BTreeMap;
-
-#[cfg(doc)]
-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`].
-#[derive(Debug, Clone, Copy)]
-pub struct ObjectLabels<'a, T> {
-    /// Reference to the k8s object owning the created resource, such as `HdfsCluster` or `TrinoCluster`.
-    pub owner: &'a T,
-    /// The name of the app being managed, such as `zookeeper`
-    pub app_name: &'a str,
-    /// The version of the app being managed (not of the operator).
-    ///
-    /// If setting this label on a Stackable product then please use [`ResolvedProductImage::app_version_label`]
-    ///
-    /// This version should include the Stackable version, such as `3.0.0-stackable0.1.0`.
-    /// If the Stackable version is not known, then the product version should be used together with a suffix (if possible).
-    /// If a custom product image is provided by the user (in which case only the product version is known),
-    /// then the format `3.0.0-<tag-of-custom-image>` should be used.
-    ///
-    /// However, this is pure documentation and should not be parsed.
-    pub app_version: &'a str,
-    /// The DNS-style name of the operator managing the object (such as `zookeeper.stackable.tech`)
-    pub operator_name: &'a str,
-    /// The name of the controller inside of the operator managing the object (such as `zookeepercluster`)
-    pub controller_name: &'a str,
-    /// The role that this object belongs to
-    pub role: &'a str,
-    /// The role group that this object belongs to
-    pub role_group: &'a str,
-}
-
-/// Create kubernetes recommended labels
-pub fn get_recommended_labels<T>(
-    ObjectLabels {
-        owner,
-        app_name,
-        app_version,
-        operator_name,
-        controller_name,
-        role,
-        role_group,
-    }: ObjectLabels<T>,
-) -> BTreeMap<String, String>
-where
-    T: Resource,
-{
-    let mut labels = role_group_selector_labels(owner, app_name, role, role_group);
-
-    // 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
-}
diff --git a/src/lib.rs b/src/lib.rs
index a12995a2f..4d3e5116c 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -8,8 +8,7 @@ pub mod cpu;
 pub mod crd;
 pub mod error;
 pub mod iter;
-pub mod label_selector;
-pub mod labels;
+pub mod kvp;
 pub mod logging;
 pub mod memory;
 pub mod namespace;