This repository has been archived by the owner on Nov 22, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 90
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Browse files
Browse the repository at this point in the history
* Added support back in for --volume-mount. Added test. Some refactoring, added check for missing volume mounts. Better error message. Cleanup. Fix for compile. Added a comment. * Added the Cloudflow CRD. (#1074) * Cloudflow Custom Resource Definition (CRD). * Comment to remove endpoint statuses from CRD status, not used. * Added docs to CRD fields. * Remove tmp file.
- Loading branch information
1 parent
120cefa
commit 6c3f7f5
Showing
8 changed files
with
623 additions
and
81 deletions.
There are no files selected for viewing
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
116 changes: 116 additions & 0 deletions
116
core/cloudflow-cli/src/main/scala/akka/cli/cloudflow/execution/WithUpdateVolumeMounts.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| /* | ||
| * Copyright (C) 2021 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.cli.cloudflow.execution | ||
|
|
||
| import akka.cli.cloudflow.CliException | ||
| import akka.datap.crd.App | ||
|
|
||
| import scala.util.{ Failure, Success, Try } | ||
|
|
||
| // This can be deprecated when VolumeMount API is deprecated. | ||
| trait WithUpdateVolumeMounts { | ||
| def updateVolumeMounts( | ||
| crApp: App.Cr, | ||
| volumeMountsArgs: Map[String, String], | ||
| pvcs: () => Try[List[String]]): Try[App.Cr] = { | ||
| for { | ||
| streamletVolumeNameToPvc <- streamletVolumeNameToPvcMap(crApp, volumeMountsArgs, pvcs) | ||
| _ <- missingStreamletVolumeMountNames(crApp, streamletVolumeNameToPvc.keys) | ||
| } yield crApp.copy(spec = crApp.spec.copy( | ||
| deployments = updatedDeployments(crApp, streamletVolumeNameToPvc), | ||
| streamlets = updatedStreamlets(crApp, streamletVolumeNameToPvc))) | ||
| } | ||
|
|
||
| private def streamletVolumeNameToPvcMap( | ||
| crApp: App.Cr, | ||
| volumeMountsArgs: Map[String, String], | ||
| pvcs: () => Try[List[String]]): Try[Map[(String, String), String]] = { | ||
| for { | ||
| existingPvcs <- pvcs() | ||
| map <- Try { | ||
| volumeMountsArgs.map { | ||
| case (streamletVolumeNamePath, pvcName) => | ||
| if (!existingPvcs.contains(pvcName)) { | ||
| throw new CliException( | ||
| s"Cannot find persistent volume claim '$pvcName' specified via --volume-mount argument.") | ||
| } | ||
| // volumeMounts Map is "<streamlet-name>.<volume-mount-name>" -> pvc-name | ||
| val parts = streamletVolumeNamePath.split("\\.").toList | ||
| if (parts.size != 2) { | ||
| throw new CliException( | ||
| "--volume-mount argument is invalid, please provide as --volume-mount <streamlet-name>.<volume-mount-name>=<pvc-name>") | ||
| } | ||
| val streamletName = parts(0) | ||
| val volumeMountName = parts(1) | ||
| if (crApp.spec.deployments.find(_.streamletName == streamletName).isEmpty) { | ||
| throw new CliException(s"Cannot find streamlet '$streamletName' in --volume-mount argument") | ||
| } | ||
|
|
||
| if (crApp.spec.deployments | ||
| .filter(_.streamletName == streamletName) | ||
| .flatMap(_.volumeMounts) | ||
| .find(_.name == volumeMountName) | ||
| .isEmpty) { | ||
| throw new CliException( | ||
| s"Cannot find volume mount name '$volumeMountName' for streamlet '$streamletName' in --volume-mount argument") | ||
| } | ||
| (streamletName, volumeMountName) -> pvcName | ||
| }.toMap | ||
| } | ||
| } yield map | ||
| } | ||
|
|
||
| private def missingStreamletVolumeMountNames( | ||
| crApp: App.Cr, | ||
| streamletVolumeNamesFromArgs: Iterable[(String, String)]): Try[Unit] = { | ||
| Try { | ||
| val missing = (streamletVolumeMountNamesInCr(crApp) -- streamletVolumeNamesFromArgs.toSet) | ||
| .map { case (streamletName, volumeName) => s"$streamletName.$volumeName" } | ||
| .toSeq | ||
| .sorted | ||
| if (missing.nonEmpty) { | ||
| def plural = s"""${if (missing.size > 1) "s" else ""}""" | ||
| throw new CliException( | ||
| s"""Please provide persistent volume name$plural with --volume-mount argument$plural (replace 'pvc-name'$plural with correct value$plural):\n | ||
| |${missing.map(m => s"--volume-mount $m=pvc-name").mkString("\n")} | ||
| """.stripMargin) | ||
| } | ||
| () | ||
| } | ||
| } | ||
|
|
||
| private def streamletVolumeMountNamesInCr(crApp: App.Cr): Set[(String, String)] = { | ||
| crApp.spec.deployments | ||
| .flatMap(deployment => deployment.volumeMounts.map(vm => deployment.streamletName -> vm.name)) | ||
| .toSet | ||
| } | ||
|
|
||
| private def updatedDeployments( | ||
| crApp: App.Cr, | ||
| streamletVolumeNameToPvc: Map[(String, String), String]): Seq[App.Deployment] = { | ||
| crApp.spec.deployments.map { deployment => | ||
| deployment.copy(volumeMounts = deployment.volumeMounts.map { vmd => | ||
| streamletVolumeNameToPvc | ||
| .get((deployment.streamletName, vmd.name)) | ||
| .map(pvcName => vmd.copy(pvcName = Some(pvcName))) | ||
| .getOrElse(vmd) | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| private def updatedStreamlets( | ||
| crApp: App.Cr, | ||
| streamletVolumeNameToPvc: Map[(String, String), String]): Seq[App.Streamlet] = { | ||
| crApp.spec.streamlets.map { streamlet => | ||
| streamlet.copy(descriptor = streamlet.descriptor.copy(volumeMounts = streamlet.descriptor.volumeMounts.map { | ||
| vmd => | ||
| streamletVolumeNameToPvc | ||
| .get((streamlet.name, vmd.name)) | ||
| .map(pvcName => vmd.copy(pvcName = Some(pvcName))) | ||
| .getOrElse(vmd) | ||
| })) | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.