Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
105 changes: 90 additions & 15 deletions rust/operator-binary/src/csi_server/node.rs
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
use std::{
fs::Permissions,
os::unix::prelude::PermissionsExt,
os::unix::{fs::MetadataExt, prelude::PermissionsExt},
path::{Component, Path, PathBuf},
};

use chrono::{DateTime, FixedOffset};
use openssl::sha::Sha256;
use serde::{Deserialize, de::IntoDeserializer};
use snafu::{ResultExt, Snafu, ensure};
Expand Down Expand Up @@ -66,6 +67,12 @@ enum PublishError {
path: PathBuf,
},

#[snafu(display("failed to check whether {path:?} is already a mount point"))]
CheckMountPoint {
source: std::io::Error,
path: PathBuf,
},

#[snafu(display("failed to mount volume mount directory {path:?}"))]
Mount {
source: std::io::Error,
Expand Down Expand Up @@ -122,6 +129,7 @@ impl From<PublishError> for Status {
Status::new(source.grpc_code(), full_msg)
}
PublishError::CreateDir { .. } => Status::unavailable(full_msg),
PublishError::CheckMountPoint { .. } => Status::unavailable(full_msg),
PublishError::Mount { .. } => Status::unavailable(full_msg),
PublishError::FormatData { .. } => Status::unavailable(full_msg),
PublishError::SetDirPermissions { .. } => Status::unavailable(full_msg),
Expand Down Expand Up @@ -163,6 +171,23 @@ impl From<UnpublishError> for Status {
}
}

/// Returns whether something is mounted at `path`.
///
/// A mount point is on a different device than its parent directory. A path that does not exist
/// is not a mount point.
fn is_mount_point(path: &Path) -> std::io::Result<bool> {
let metadata = match std::fs::metadata(path) {
Ok(metadata) => metadata,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(false),
Err(err) => return Err(err),
};
// The root directory is always a mount point
let Some(parent) = path.parent() else {
return Ok(true);
};
Ok(metadata.dev() != std::fs::metadata(parent)?.dev())
}

// The actual provisioner that is run on all nodes and in charge of provisioning and storing
// secrets for pods that get scheduled on that node.
pub struct SecretProvisionerNode {
Expand Down Expand Up @@ -289,12 +314,35 @@ impl SecretProvisionerNode {
Ok(())
}

/// Provisions a new volume at `target_path`.
///
/// The Pod is only tagged with the expiry once the files are in place. Otherwise, the restart
/// controller could be told to wait for a certificate that the Pod never received.
async fn publish_new_volume(
&self,
target_path: &Path,
volume_id: &str,
selector: SecretVolumeSelector,
data: SecretContents,
) -> Result<(), PublishError> {
// `save_secret_data` consumes both, but they are still needed to tag the Pod afterwards
let pod = selector.pod.clone();
let namespace = selector.namespace.clone();
let expires_after = data.expires_after;

self.prepare_secret_dir(target_path).await?;
self.save_secret_data(target_path, data, selector).await?;
self.tag_pod(&self.client, volume_id, &pod, &namespace, expires_after)
.await
}

async fn tag_pod(
&self,
client: &stackable_operator::client::Client,
volume_id: &str,
selector: &SecretVolumeSelector,
data: &SecretContents,
pod: &str,
namespace: &str,
expires_after: Option<DateTime<FixedOffset>>,
) -> Result<(), PublishError> {
// Each volume must have a unique tag, so that multiple markers of the same type can coexist on the same pod
// Each tag needs to be simple and unique-ish per volume
Expand All @@ -309,7 +357,7 @@ impl SecretProvisionerNode {

let mut annotations = Annotations::new();

if let Some(expires_after) = data.expires_after {
if let Some(expires_after) = expires_after {
annotations
.parse_insert((
format!(
Expand All @@ -324,8 +372,8 @@ impl SecretProvisionerNode {
if !annotations.is_empty() {
let tagged_pod = Pod {
metadata: ObjectMetaBuilder::new()
.name(&selector.pod)
.namespace(&selector.namespace)
.name(pod)
.namespace(namespace)
.annotations(annotations)
.build(),
..Pod::default()
Expand Down Expand Up @@ -408,6 +456,23 @@ impl Node for SecretProvisionerNode {
volume.path = %target_path.display(),
"Received NodePublishVolume request"
);
// NodePublishVolume must be idempotent. The Kubelet calls it again for volumes that
// are already in use (for example after a restart of the Kubelet). The container
// keeps the mount from the first call, so a second tmpfs mounted on top of it
// would never be seen by the Pod, while the Pod annotation would already point
// to the expiry of the new, undelivered certificate.
// This relies on failed publishes being cleaned up (see below), so that a mount
// point always belongs to a complete volume.
// There is no mount in unprivileged mode, so there is nothing to detect there.
if self.privileged && is_mount_point(&target_path).context(
publish_error::CheckMountPointSnafu { path: &target_path },
)? {
tracing::info!(
volume.path = %target_path.display(),
"Volume is already published, keeping the existing secret"
);
return Ok(Response::new(NodePublishVolumeResponse {}));
}
let selector =
SecretVolumeSelector::deserialize(request.volume_context.into_deserializer())
.context(publish_error::InvalidSelectorSnafu)?;
Expand All @@ -421,15 +486,25 @@ impl Node for SecretProvisionerNode {
.get_secret_data(&selector, pod_info)
.await
.context(publish_error::BackendGetSecretDataSnafu)?;
self.tag_pod(&self.client, &request.volume_id, &selector, &data)
.await?;
self.prepare_secret_dir(&target_path).await?;
self.save_secret_data(
&target_path,
data,
selector
)
.await?;
if let Err(err) = self
.publish_new_volume(&target_path, &request.volume_id, selector, data)
.await
{
// Don't leave a partially provisioned volume behind. It would be mistaken
// for a complete one by the next call, which would then skip provisioning.
// In unprivileged mode, the directory might be a previously provisioned
// volume that is still in use, which must not be deleted.
if self.privileged
&& let Err(cleanup_err) = self.clean_secret_dir(&target_path).await
{
tracing::warn!(
volume.path = %target_path.display(),
error = %error_full_message(&cleanup_err),
"Failed to clean up volume after a failed publish"
);
}
return Err(err.into());
}
Ok(Response::new(NodePublishVolumeResponse {}))
}
.await,
Expand Down
Loading