Volumes

Argo Workflows allows you to mount volumes from many different sources. Hera provides wrappper classes for all of the first-party Kubernetes volumes (such as Volume and EmptyDirVolume), as well as third-party volumes (such as CinderVolume).

The hera.workflows.volume module provides all Argo volume types that can be used via Hera.

Volume

Volume represents a basic, dynamic, volume representation.

This Volume cannot only be instantiated to be used for mounting purposes but also for dynamically privisioning volumes in K8s. When the volume is used a corresponding persistent volume claim is also created on workflow submission.

Source code in src/hera/workflows/volume.py

@dataclass(kw_only=True)
class Volume(_BaseVolume):
    """Volume represents a basic, dynamic, volume representation.

    This `Volume` cannot only be instantiated to be used for mounting purposes but also for dynamically privisioning
    volumes in K8s. When the volume is used a corresponding persistent volume claim is also created on workflow
    submission.
    """

    data_source: Optional[TypedLocalObjectReference] = None
    data_source_ref: Optional[TypedObjectReference] = None
    resources: Optional[VolumeResourceRequirements] = None
    selector: Optional[v1.LabelSelector] = None
    storage_class_name: Optional[str] = None
    volume_attributes_class_name: Optional[str] = None
    volume_mode: Optional[str] = None
    volume_name: Optional[str] = None
    size: Optional[str] = None
    metadata: Optional[ObjectMeta] = None
    access_modes: List[str | AccessMode] = field(default_factory=lambda: [AccessMode.read_write_once])

    def __post_init__(self):
        if not self.name:
            self.name = str(uuid.uuid4())

        if not self.access_modes:
            self.access_modes = [AccessMode.read_write_once]
        else:
            result: List[str | AccessMode] = []
            for mode in self.access_modes:
                if isinstance(mode, AccessMode):
                    result.append(mode)
                elif isinstance(mode, str):
                    result.append(AccessMode(mode))

            self.access_modes = result

        if self.size and self.resources:
            if self.resources.requests is not None and "storage" not in self.resources.requests:
                self.resources.requests["storage"] = resource.Quantity(root=self.size)
        elif self.resources is None:
            assert self.size, "at least one of `size` or `resources` must be specified"
            validate_storage_units(self.size)
            self.resources = VolumeResourceRequirements(requests={"storage": resource.Quantity(root=self.size)})
        else:  # self.resources is not None
            assert self.resources.requests is not None, "Resource requests are required"
            storage = self.resources.requests.get("storage")
            assert storage is not None, "At least one of `size` or `resources.requests.storage` must be specified"
            validate_storage_units(cast(str, storage))

    def _build_persistent_volume_claim(self) -> _ModelPersistentVolumeClaim:
        return _ModelPersistentVolumeClaim(
            metadata=self.metadata or ObjectMeta(name=self.name),
            spec=_ModelPersistentVolumeClaimSpec(
                access_modes=[str(cast(AccessMode, am).value) for am in self.access_modes]
                if self.access_modes is not None
                else None,
                data_source=self.data_source,
                data_source_ref=self.data_source_ref,
                resources=self.resources,
                selector=self.selector,
                storage_class_name=self.storage_class_name,
                volume_mode=self.volume_mode,
                volume_name=self.volume_name,
            ),
        )

    def _build_volume(self) -> _ModelVolume:
        claim = self._build_persistent_volume_claim()
        assert claim.metadata is not None, "claim metadata is required"
        assert self.name
        return _ModelVolume(
            name=self.name,
            persistent_volume_claim=_ModelPersistentVolumeClaimVolumeSource(
                claim_name=cast(str, claim.metadata.name),
                read_only=self.read_only,
            ),
        )

access_modes

access_modes: List[str | AccessMode] = field(
    default_factory=lambda: [read_write_once]
)

data_source

data_source: Optional[TypedLocalObjectReference] = None

data_source_ref

data_source_ref: Optional[TypedObjectReference] = None

metadata

metadata: Optional[ObjectMeta] = None

mount_path

mount_path: Optional[str] = None

mount_propagation

mount_propagation: Optional[str] = None

name

name: Optional[str] = None

read_only

read_only: Optional[bool] = None

recursive_read_only

recursive_read_only: Optional[str] = None

resources

resources: Optional[VolumeResourceRequirements] = None

selector

selector: Optional[LabelSelector] = None

size

size: Optional[str] = None

storage_class_name

storage_class_name: Optional[str] = None

sub_path

sub_path: Optional[str] = None

sub_path_expr

sub_path_expr: Optional[str] = None

volume_attributes_class_name

volume_attributes_class_name: Optional[str] = None

volume_mode

volume_mode: Optional[str] = None

volume_name

volume_name: Optional[str] = None

AccessMode

A representations of the volume access modes for Kubernetes.

Notes

See: access modes docs for more information.

Source code in src/hera/workflows/volume.py

class AccessMode(Enum):
    """A representations of the volume access modes for Kubernetes.

    Notes:
        See: [access modes docs](https://kubernetes.io/docs/concepts/storage/persistent-volumes/#access-modes) for
        more information.
    """

    read_write_once = "ReadWriteOnce"
    """
    The volume can be mounted as read-write by a single node. ReadWriteOnce access mode still can allow multiple
    pods to access the volume when the pods are running on the same node
    """

    read_only_many = "ReadOnlyMany"
    """The volume can be mounted as read-only by many nodes"""

    read_write_many = "ReadWriteMany"
    """The volume can be mounted as read-write by many nodes"""

    read_write_once_pod = "ReadWriteOncePod"
    """
    The volume can be mounted as read-write by a single Pod. Use ReadWriteOncePod access mode if you want to
    ensure that only one pod across whole cluster can read that PVC or write to it. This is only supported for CSI
    volumes and Kubernetes version 1.22+.
    """

    def __str__(self) -> str:
        """Returns the value representation of the enum in the form of a string."""
        return str(self.value)

read_only_many

read_only_many = 'ReadOnlyMany'

The volume can be mounted as read-only by many nodes

read_write_many

read_write_many = 'ReadWriteMany'

The volume can be mounted as read-write by many nodes

read_write_once

read_write_once = 'ReadWriteOnce'

The volume can be mounted as read-write by a single node. ReadWriteOnce access mode still can allow multiple pods to access the volume when the pods are running on the same node

read_write_once_pod

read_write_once_pod = 'ReadWriteOncePod'

The volume can be mounted as read-write by a single Pod. Use ReadWriteOncePod access mode if you want to ensure that only one pod across whole cluster can read that PVC or write to it. This is only supported for CSI volumes and Kubernetes version 1.22+.

Comments