support creating and updating lease's begin_time

this allows to create scheduled leases that will start at the requested time. At begin_time the lease will start acquisition that may be delayed or fail if the selected exporter is not available, following the lease acquisition timeout setting.

Author: michalskrivanek

packages/jumpstarter-cli/jumpstarter_cli/common.py

@@ -1,4 +1,4 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 from functools import partial
 
 import click
@@ -44,3 +44,31 @@ def convert(self, value, param, ctx):
 See https://docs.rs/speedate/latest/speedate/ for details
 """,
 )
+
+
+class DateTimeParamType(click.ParamType):
+    name = "datetime"
+
+    def convert(self, value, param, ctx):
+        if isinstance(value, datetime):
+            return value
+
+        try:
+            return TypeAdapter(datetime).validate_python(value)
+        except ValueError:
+            self.fail(f"{value!r} is not a valid datetime", param, ctx)
+
+
+DATETIME = DateTimeParamType()
+
+opt_begin_time = click.option(
+    "--begin-time",
+    "begin_time",
+    type=DATETIME,
+    default=None,
+    help="""
+Begin time for the lease in ISO 8601 format (e.g., 2024-01-01T12:00:00 or 2024-01-01T12:00:00Z).
+If not specified, the lease tries to be acquired immediately. The lease duration always starts
+at the actual time of acquisition.
+""",
+)

packages/jumpstarter-cli/jumpstarter_cli/create.py

@@ -1,12 +1,12 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 import click
 from jumpstarter_cli_common.config import opt_config
 from jumpstarter_cli_common.exceptions import handle_exceptions_with_reauthentication
 from jumpstarter_cli_common.opt import OutputType, opt_output_all
 from jumpstarter_cli_common.print import model_print
 
-from .common import opt_duration_partial, opt_selector
+from .common import opt_begin_time, opt_duration_partial, opt_selector
 from .login import relogin_client
 
 
@@ -21,9 +21,10 @@ def create():
 @opt_config(exporter=False)
 @opt_selector
 @opt_duration_partial(required=True)
+@opt_begin_time
 @opt_output_all
 @handle_exceptions_with_reauthentication(relogin_client)
-def create_lease(config, selector: str, duration: timedelta, output: OutputType):
+def create_lease(config, selector: str, duration: timedelta, begin_time: datetime | None, output: OutputType):
     """
     Create a lease
 
@@ -49,6 +50,6 @@ def create_lease(config, selector: str, duration: timedelta, output: OutputType)
 
     """
 
-    lease = config.create_lease(selector=selector, duration=duration)
+    lease = config.create_lease(selector=selector, duration=duration, begin_time=begin_time)
 
     model_print(lease, output)

packages/jumpstarter-cli/jumpstarter_cli/update.py

@@ -1,12 +1,12 @@
-from datetime import timedelta
+from datetime import datetime, timedelta
 
 import click
 from jumpstarter_cli_common.config import opt_config
 from jumpstarter_cli_common.exceptions import handle_exceptions_with_reauthentication
 from jumpstarter_cli_common.opt import OutputType, opt_output_all
 from jumpstarter_cli_common.print import model_print
 
-from .common import opt_duration_partial
+from .common import opt_begin_time, opt_duration_partial
 from .login import relogin_client
 
 
@@ -20,14 +20,22 @@ def update():
 @update.command(name="lease")
 @opt_config(exporter=False)
 @click.argument("name")
-@opt_duration_partial(required=True)
+@opt_duration_partial(required=False)
+@opt_begin_time
 @opt_output_all
 @handle_exceptions_with_reauthentication(relogin_client)
-def update_lease(config, name: str, duration: timedelta, output: OutputType):
+def update_lease(config, name: str, duration: timedelta | None, begin_time: datetime | None, output: OutputType):
     """
     Update a lease
+
+    Update the duration and/or begin time of an existing lease.
+    At least one of --duration or --begin-time must be specified.
+    Updating the begin time of an already active lease is not allowed.
     """
 
-    lease = config.update_lease(name, duration)
+    if duration is None and begin_time is None:
+        raise click.UsageError("At least one of --duration or --begin-time must be specified")
+
+    lease = config.update_lease(name, duration=duration, begin_time=begin_time)
 
     model_print(lease, output)

packages/jumpstarter/jumpstarter/client/grpc.py

@@ -7,7 +7,7 @@
 from types import SimpleNamespace
 from typing import Any
 
-from google.protobuf import duration_pb2, field_mask_pb2, json_format
+from google.protobuf import duration_pb2, field_mask_pb2, json_format, timestamp_pb2
 from grpc import ChannelConnectivity
 from grpc.aio import Channel
 from jumpstarter_protocol import client_pb2, client_pb2_grpc, jumpstarter_pb2_grpc, kubernetes_pb2, router_pb2_grpc
@@ -346,18 +346,26 @@ async def CreateLease(
         *,
         selector: str,
         duration: timedelta,
+        begin_time: datetime | None = None,
     ):
         duration_pb = duration_pb2.Duration()
         duration_pb.FromTimedelta(duration)
 
+        lease_pb = client_pb2.Lease(
+            duration=duration_pb,
+            selector=selector,
+        )
+
+        if begin_time:
+            timestamp_pb = timestamp_pb2.Timestamp()
+            timestamp_pb.FromDatetime(begin_time)
+            lease_pb.begin_time.CopyFrom(timestamp_pb)
+
         with translate_grpc_exceptions():
             lease = await self.stub.CreateLease(
                 client_pb2.CreateLeaseRequest(
                     parent="namespaces/{}".format(self.namespace),
-                    lease=client_pb2.Lease(
-                        duration=duration_pb,
-                        selector=selector,
-                    ),
+                    lease=lease_pb,
                 )
             )
         return Lease.from_protobuf(lease)
@@ -366,21 +374,37 @@ async def UpdateLease(
         self,
         *,
         name: str,
-        duration: timedelta,
+        duration: timedelta | None = None,
+        begin_time: datetime | None = None,
     ):
-        duration_pb = duration_pb2.Duration()
-        duration_pb.FromTimedelta(duration)
+        lease_pb = client_pb2.Lease(
+            name="namespaces/{}/leases/{}".format(self.namespace, name),
+        )
+
+        update_fields = []
+
+        if duration is not None:
+            duration_pb = duration_pb2.Duration()
+            duration_pb.FromTimedelta(duration)
+            lease_pb.duration.CopyFrom(duration_pb)
+            update_fields.append("duration")
+
+        if begin_time is not None:
+            timestamp_pb = timestamp_pb2.Timestamp()
+            timestamp_pb.FromDatetime(begin_time)
+            lease_pb.begin_time.CopyFrom(timestamp_pb)
+            update_fields.append("begin_time")
+
+        if not update_fields:
+            raise ValueError("At least one of duration or begin_time must be provided")
 
         update_mask = field_mask_pb2.FieldMask()
-        update_mask.FromJsonString("duration")
+        update_mask.FromJsonString(",".join(update_fields))
 
         with translate_grpc_exceptions():
             lease = await self.stub.UpdateLease(
                 client_pb2.UpdateLeaseRequest(
-                    lease=client_pb2.Lease(
-                        name="namespaces/{}/leases/{}".format(self.namespace, name),
-                        duration=duration_pb,
-                    ),
+                    lease=lease_pb,
                     update_mask=update_mask,
                 )
             )

packages/jumpstarter/jumpstarter/config/client.py

@@ -3,7 +3,7 @@
 import asyncio
 import os
 from contextlib import asynccontextmanager, contextmanager
-from datetime import timedelta
+from datetime import datetime, timedelta
 from functools import wraps
 from pathlib import Path
 from typing import Annotated, ClassVar, Literal, Optional, Self
@@ -200,11 +200,13 @@ async def create_lease(
         self,
         selector: str,
         duration: timedelta,
+        begin_time: datetime | None = None,
     ):
         svc = ClientService(channel=await self.channel(), namespace=self.metadata.namespace)
         return await svc.CreateLease(
             selector=selector,
             duration=duration,
+            begin_time=begin_time,
         )
 
     @_blocking_compat
@@ -237,11 +239,12 @@ async def list_leases(
     @_handle_connection_error
     async def update_lease(
         self,
-        name,
-        duration: timedelta,
+        name: str,
+        duration: timedelta | None = None,
+        begin_time: datetime | None = None,
     ):
         svc = ClientService(channel=await self.channel(), namespace=self.metadata.namespace)
-        return await svc.UpdateLease(name=name, duration=duration)
+        return await svc.UpdateLease(name=name, duration=duration, begin_time=begin_time)
 
     @asynccontextmanager
     async def lease_async(