| 282 | |
| 283 | |
| 284 | Future<VolumeInfo> VolumeManagerProcess::createVolume( |
| 285 | const string& name, |
| 286 | const Bytes& capacity, |
| 287 | const CSIVolume::VolumeCapability& capability, |
| 288 | const Map<string, string>& parameters) |
| 289 | { |
| 290 | if (!controllerCapabilities->createDeleteVolume) { |
| 291 | return Failure( |
| 292 | "CREATE_DELETE_VOLUME controller capability is not supported for CSI " |
| 293 | "plugin type '" + info.type() + "' and name '" + info.name()); |
| 294 | } |
| 295 | |
| 296 | LOG(INFO) << "Creating volume with name '" << name << "'"; |
| 297 | |
| 298 | CreateVolumeRequest request; |
| 299 | request.set_name(name); |
| 300 | request.mutable_capacity_range()->set_required_bytes(capacity.bytes()); |
| 301 | request.mutable_capacity_range()->set_limit_bytes(capacity.bytes()); |
| 302 | *request.add_volume_capabilities() = evolve(capability); |
| 303 | *request.mutable_parameters() = parameters; |
| 304 | |
| 305 | // We retry the `CreateVolume` call for MESOS-9517. |
| 306 | return call( |
| 307 | CONTROLLER_SERVICE, &Client::createVolume, std::move(request), true) |
| 308 | .then(process::defer(self(), [=]( |
| 309 | const CreateVolumeResponse& response) -> Future<VolumeInfo> { |
| 310 | const string& volumeId = response.volume().id(); |
| 311 | |
| 312 | // NOTE: If the volume is already tracked, there might already be |
| 313 | // operations running in its sequence. Since this continuation runs |
| 314 | // outside the sequence, we fail the call here to avoid any race issue. |
| 315 | // This also means that this call is not idempotent. |
| 316 | if (volumes.contains(volumeId)) { |
| 317 | return Failure("Volume with name '" + name + "' already exists"); |
| 318 | } |
| 319 | |
| 320 | VolumeState volumeState; |
| 321 | volumeState.set_state(VolumeState::CREATED); |
| 322 | *volumeState.mutable_volume_capability() = capability; |
| 323 | *volumeState.mutable_parameters() = parameters; |
| 324 | *volumeState.mutable_volume_context() = response.volume().attributes(); |
| 325 | |
| 326 | volumes.put(volumeId, std::move(volumeState)); |
| 327 | checkpointVolumeState(volumeId); |
| 328 | |
| 329 | return VolumeInfo{capacity, volumeId, response.volume().attributes()}; |
| 330 | })); |
| 331 | } |
| 332 | |
| 333 | |
| 334 | Future<Option<Error>> VolumeManagerProcess::validateVolume( |