/// This RPC is called by the CO when a workload that wants to use the
/// specified volume is placed (scheduled) on a node. The Plugin SHALL
/// assume that this RPC will be executed on the node where the volume will
/// be used. If the corresponding Controller Plugin has
/// PUBLISH_UNPUBLISH_VOLUME controller capability, the CO MUST guarantee
/// that this RPC is called after ControllerPublishVolume is called for the
/// given volume on the given node and returns a success. This operation
/// MUST be idempotent. If the volume corresponding to the volume_id has
/// already been published at the specified target_path, and is compatible
/// with the specified volume_capability and readonly flag, the Plugin MUST
/// reply 0 OK. If this RPC failed, or the CO does not know if it failed or
/// not, it MAY choose to call NodePublishVolume again, or choose to call
/// NodeUnpublishVolume. This RPC MAY be called by the CO multiple times on
/// the same node for the same volume with possibly different target_path
/// and/or other arguments if the volume has MULTI_NODE capability (i.e.,
/// access_mode is either MULTI_NODE_READER_ONLY, MULTI_NODE_SINGLE_WRITER
/// or MULTI_NODE_MULTI_WRITER).
This RPC is called by the CO when a workload that wants to use the
specified volume is placed (scheduled) on a node. The Plugin SHALL
assume that this RPC will be executed on the node where the volume will
be used. If the corresponding Controller Plugin has
PUBLISH_UNPUBLISH_VOLUME controller capability, the CO MUST guarantee
that this RPC is called after ControllerPublishVolume is called for the
given volume on the given node and returns a success. This operation
MUST be idempotent. If the volume corresponding to the volume_id has
already been published at the specified target_path, and is compatible
with the specified volume_capability and readonly flag, the Plugin MUST
reply 0 OK. If this RPC failed, or the CO does not know if it failed or
not, it MAY choose to call NodePublishVolume again, or choose to call
NodeUnpublishVolume. This RPC MAY be called by the CO multiple times on
the same node for the same volume with possibly different target_path
and/or other arguments if the volume has MULTI_NODE capability .
This RPC is called by the CO when a workload that wants to use the specified volume is placed ( scheduled ) on a node . The Plugin SHALL assume that this RPC will be executed on the node where the volume will be used . If the corresponding Controller Plugin has PUBLISH_UNPUBLISH_VOLUME controller capability the CO MUST guarantee that this RPC is called after ControllerPublishVolume is called for the given volume on the given node and returns a success . This operation MUST be idempotent . If the volume corresponding to the volume_id has already been published at the specified target_path and is compatible with the specified volume_capability and readonly flag the Plugin MUST reply 0 OK . If this RPC failed or the CO does not know if it failed or not it MAY choose to call NodePublishVolume again or choose to call NodeUnpublishVolume . This RPC MAY be called by the CO multiple times on the same node for the same volume with possibly different target_path and / or other arguments if the volume has MULTI_NODE capability .
async fn node_publish_volume(
        &self,
        request: Request<NodePublishVolumeRequest>,
    ) -> Result<Response<NodePublishVolumeResponse>, Status> {
        let msg = request.into_inner();
        trace!("node_publish_volume {:?}", msg);
        let staging_path = &msg.staging_target_path;
        let target_path = &msg.target_path;
        let volume_id = &msg.volume_id;
        if volume_id.is_empty() {
            return Err(failure!(
                Code::InvalidArgument,
                "Failed to publish volume: missing volume id"
            ));
        }
        if target_path.is_empty() {
            return Err(failure!(
                Code::InvalidArgument,
                "Failed to publish volume {}: missing target path",
                volume_id
            ));
        }
        // Note that the staging path is NOT optional,
        // as we advertise StageUnstageVolume.
        if staging_path.is_empty() {
            return Err(failure!(
                Code::InvalidArgument,
                "Failed to publish volume {}: missing staging path",
                volume_id
            ));
        }
        // TODO: Support block volumes
        let mnt =
            get_access_mount(&msg.volume_capability).map_err(|error| {
                failure!(
                    Code::InvalidArgument,
                    "Failed to publish volume {}: {}",
                    volume_id,
                    error
                )
            })?;
        debug!(
            "Publishing volume {} from {} to {}",
            volume_id, staging_path, target_path
        );
        let staged =
            mount::find_mount(None, Some(staging_path)).ok_or_else(|| {
                failure!(
                    Code::InvalidArgument,
                    "Failed to publish volume {}: no mount for staging path {}",
                    volume_id,
                    staging_path
                )
            })?;
        // TODO: Should also check that the staged "device"
        // corresponds to the the volume uuid
        if let Err(error) =
            check_access_mode(&msg.volume_capability, msg.readonly)
        {
            return Err(failure!(
                Code::InvalidArgument,
                "Failed to publish volume {}: {}",
                volume_id,
                error
            ));
        }
        if mnt.fs_type != "" && mnt.fs_type != staged.fstype {
            return Err(failure!(
                Code::InvalidArgument,
                "Failed to publish volume {}: filesystem type ({}) does not match staged volume ({})",
                volume_id,
                mnt.fs_type,
                staged.fstype
            ));
        }
        if self
            .filesystems
            .iter()
            .find(|&entry| entry == &staged.fstype)
            .is_none()
        {
            return Err(failure!(
                Code::InvalidArgument,
                "Failed to publish volume {}: unsupported filesystem type: {}",
                volume_id,
                staged.fstype
            ));
        }
        let readonly = staged.options.readonly();
        if readonly && !msg.readonly {
            return Err(failure!(
                Code::InvalidArgument,
                "Failed to publish volume {}: volume is staged as \"ro\" but publish requires \"rw\"",
                volume_id
            ));
        }
        if let Some(mount) = mount::find_mount(None, Some(target_path)) {
            if mount.source != staged.source {
                return Err(failure!(
                    Code::AlreadyExists,
                    "Failed to publish volume {}: directory {} is already in use",
                    volume_id,
                    target_path
                ));
            }
            if !subset(&mnt.mount_flags, &mount.options)
                || msg.readonly != mount.options.readonly()
            {
                return Err(failure!(
                    Code::AlreadyExists,
                    "Failed to publish volume {}: directory {} is already mounted but with incompatible flags",
                    volume_id,
                    target_path
                ));
            }
            info!(
                "Volume {} is already published to {}",
                volume_id, target_path
            );
            return Ok(Response::new(NodePublishVolumeResponse {}));
        }
        debug!("Creating directory {}", target_path);
        if let Err(error) = fs::create_dir_all(PathBuf::from(target_path)) {
            if error.kind() != ErrorKind::AlreadyExists {
                return Err(failure!(
                    Code::Internal,
                    "Failed to publish volume {}: failed to create directory {}: {}",
                    volume_id,
                    target_path,
                    error
                ));
            }
        }
        debug!("Mounting {} to {}", staging_path, target_path);
        if let Err(error) =
            mount::bind_mount(&staging_path, &target_path, false)
        {
            return Err(failure!(
                Code::Internal,
                "Failed to publish volume {}: failed to mount {} to {}: {}",
                volume_id,
                staging_path,
                target_path,
                error
            ));
        }
        if msg.readonly && !readonly {
            let mut options = mnt.mount_flags.clone();
            options.push(String::from("ro"));
            debug!("Remounting {} as readonly", target_path);
            if let Err(error) = mount::bind_remount(&target_path, &options) {
                let message = format!(
                    "Failed to publish volume {}: failed to mount {} to {} as readonly: {}",
                    volume_id,
                    staging_path,
                    target_path,
                    error
                );
                error!("Failed to remount {}: {}", target_path, error);
                debug!("Unmounting {}", target_path);
                if let Err(error) = mount::bind_unmount(&target_path) {
                    error!("Failed to unmount {}: {}", target_path, error);
                }
                return Err(Status::new(Code::Internal, message));
            }
        }
        info!("Volume {} published to {}", volume_id, target_path);
        Ok(Response::new(NodePublishVolumeResponse {}))
    }
async fn node_publish_volume ( & self , request : Request < NodePublishVolumeRequest > , ) -> Result < Response < NodePublishVolumeResponse > , Status > { let msg = request . into_inner ( ) ; trace ! ( " " msg ) ; let staging_path = & msg . staging_target_path ; let target_path = & msg . target_path ; let volume_id = & msg . volume_id ; if volume_id . is_empty ( ) { return Err ( failure ! ( Code InvalidArgument " " ) ) ; } if target_path . is_empty ( ) { return Err ( failure ! ( Code InvalidArgument " " volume_id ) ) ; } // Note that the staging path is NOT optional, // as we advertise StageUnstageVolume. if staging_path . is_empty ( ) { return Err ( failure ! ( Code InvalidArgument " " volume_id ) ) ; } // TODO: Support block volumes let mnt = get_access_mount ( & msg . volume_capability ) . map_err ( | error | { failure ! ( Code InvalidArgument " " volume_id error ) } ) ? ; debug ! ( " " volume_id staging_path target_path ) ; let staged = mount :: find_mount ( None , Some ( staging_path ) ) . ok_or_else ( | | { failure ! ( Code InvalidArgument " " volume_id staging_path ) } ) ? ; // TODO: Should also check that the staged "device" // corresponds to the the volume uuid if let Err ( error ) = check_access_mode ( & msg . volume_capability , msg . readonly ) { return Err ( failure ! ( Code InvalidArgument " " volume_id error ) ) ; } if mnt . fs_type != " " && mnt . fs_type != staged . fstype { return Err ( failure ! ( Code InvalidArgument " " volume_id mnt fs_type staged fstype ) ) ; } if self . filesystems . iter ( ) . find ( | & entry | entry == & staged . fstype ) . is_none ( ) { return Err ( failure ! ( Code InvalidArgument " " volume_id staged fstype ) ) ; } let readonly = staged . options . readonly ( ) ; if readonly && ! msg . readonly { return Err ( failure ! ( Code InvalidArgument " \" \" \" \" " volume_id ) ) ; } if let Some ( mount ) = mount :: find_mount ( None , Some ( target_path ) ) { if mount . source != staged . source { return Err ( failure ! ( Code AlreadyExists " " volume_id target_path ) ) ; } if ! subset ( & mnt . mount_flags , & mount . options ) || msg . readonly != mount . options . readonly ( ) { return Err ( failure ! ( Code AlreadyExists " " volume_id target_path ) ) ; } info ! ( " " volume_id target_path ) ; return Ok ( Response :: new ( NodePublishVolumeResponse { } ) ) ; } debug ! ( " " target_path ) ; if let Err ( error ) = fs :: create_dir_all ( PathBuf :: from ( target_path ) ) { if error . kind ( ) != ErrorKind :: AlreadyExists { return Err ( failure ! ( Code Internal " " volume_id target_path error ) ) ; } } debug ! ( " " staging_path target_path ) ; if let Err ( error ) = mount :: bind_mount ( & staging_path , & target_path , false ) { return Err ( failure ! ( Code Internal " " volume_id staging_path target_path error ) ) ; } if msg . readonly && ! readonly { let mut options = mnt . mount_flags . clone ( ) ; options . push ( String :: from ( " " ) ) ; debug ! ( " " target_path ) ; if let Err ( error ) = mount :: bind_remount ( & target_path , & options ) { let message = format ! ( " " volume_id staging_path target_path error ) ; error ! ( " " target_path error ) ; debug ! ( " " target_path ) ; if let Err ( error ) = mount :: bind_unmount ( & target_path ) { error ! ( " " target_path error ) ; } return Err ( Status :: new ( Code :: Internal , message ) ) ; } } info ! ( " " volume_id target_path ) ; Ok ( Response :: new ( NodePublishVolumeResponse { } ) ) }
This RPC is called by the CO when a workload that wants to use the
specified volume is placed (scheduled) on a node.
This RPC is called by the CO when a workload that wants to use the specified volume is placed ( scheduled ) on a node .
&self
request
Request<NodePublishVolumeRequest>