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.

(cherry picked from commit 68f44f6)

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(