feat: persist scheduledAt in the Job/Pod annotations

cmd/tcl/testworkflow-toolkit/commands/parallel.go

@@ -218,10 +218,11 @@ func NewParallelCmd() *cobra.Command {
  // Deploy the resource
  scheduledAt := time.Now()
  result, err := spawn.ExecutionWorker().Execute(context.Background(), executionworker.ExecuteRequest{
+ ResourceId: cfg.Resource.Id,
  Execution: cfg.Execution,
  Workflow: testworkflowsv1.TestWorkflow{Spec: *spec},
+ ScheduledAt: &scheduledAt,
  ControlPlane: cfg.ControlPlane,
- ResourceId: cfg.Resource.Id,
  ArtifactsPathPrefix: cfg.Resource.FsPrefix,
  })
  if err != nil {

cmd/tcl/testworkflow-toolkit/commands/services.go

@@ -268,14 +268,15 @@ func NewServicesCmd() *cobra.Command {
  // Build the resources bundle
  scheduledAt := time.Now()
  result, err := spawn.ExecutionWorker().Service(context.Background(), executionworker.ServiceRequest{
- Execution: cfg.Execution,
- Workflow: testworkflowsv1.TestWorkflow{Spec: instance.Spec},
- ControlPlane: cfg.ControlPlane,
- ResourceId: cfg.Resource.Id,
- ArtifactsPathPrefix: cfg.Resource.FsPrefix,
-
+ ResourceId: cfg.Resource.Id,
+ Execution: cfg.Execution,
+ Workflow: testworkflowsv1.TestWorkflow{Spec: instance.Spec},
+ ScheduledAt: &scheduledAt,
  RestartPolicy: string(instance.RestartPolicy),
  ReadinessProbe: common.MapPtr(instance.ReadinessProbe, testworkflows.MapProbeKubeToAPI),
+
+ ControlPlane: cfg.ControlPlane,
+ ArtifactsPathPrefix: cfg.Resource.FsPrefix,
  })
  if err != nil {
  fmt.Printf("%d: failed to prepare resources: %s\n", index, err.Error())

pkg/testworkflows/executionworker/controller/controller.go

@@ -92,6 +92,9 @@ func New(parentCtx context.Context, clientSet kubernetes.Interface, namespace, i
  return nil, errors.Wrap(err, "invalid job signature")
  }
 
+ // Obtain the scheduled at timestamp
+ scheduledAt = watcher.State().ScheduledAt()
+
  // Build accessible controller
  return &controller{
  id: id,

pkg/testworkflows/executionworker/controller/watchers/executionstate.go

@@ -53,6 +53,7 @@ type ExecutionState interface {
  ContainerFailed(name string) bool
  Signature() ([]stage.Signature, error)
  ActionGroups() (actiontypes.ActionGroups, error)
+ ScheduledAt() time.Time
 
  ExecutionError() string
  JobExecutionError() string
@@ -252,6 +253,22 @@ func (e *executionState) Signature() ([]stage.Signature, error) {
  return nil, ErrMissingData
 }
 
+func (e *executionState) ScheduledAt() time.Time {
+ if e.job != nil {
+ v, err := e.job.ScheduledAt()
+ if err == nil {
+ return v
+ }
+ }
+ if e.pod != nil {
+ v, err := e.pod.ScheduledAt()
+ if err == nil {
+ return v
+ }
+ }
+ return e.options.ScheduledAt
+}
+
 func (e *executionState) ActionGroups() (actiontypes.ActionGroups, error) {
  if e.job != nil {
  return e.job.ActionGroups()
@@ -289,8 +306,8 @@ func (e *executionState) EstimatedJobCreationTimestamp() time.Time {
  if e.job != nil {
  return e.job.CreationTimestamp()
  }
- if !e.options.ScheduledAt.IsZero() {
- return e.options.ScheduledAt
+ if !e.ScheduledAt().IsZero() {
+ return e.ScheduledAt()
  }
  ts := e.jobEvents.FirstTimestamp()
  if e.podEvents.FirstTimestamp().Before(ts) {

pkg/testworkflows/executionworker/controller/watchers/job.go

@@ -26,6 +26,7 @@ type Job interface {
  Finished() bool
  ActionGroups() (actiontypes.ActionGroups, error)
  Signature() ([]stage.Signature, error)
+ ScheduledAt() (time.Time, error)
  ExecutionError() string
 }
 
@@ -77,6 +78,10 @@ func (j *job) Signature() ([]stage.Signature, error) {
  return stage.GetSignatureFromJSON([]byte(j.original.Spec.Template.Annotations[constants.SignatureAnnotationName]))
 }
 
+func (j *job) ScheduledAt() (time.Time, error) {
+ return time.Parse(time.RFC3339Nano, j.original.Spec.Template.Annotations[constants.ScheduledAtAnnotationName])
+}
+
 func (j *job) ExecutionError() string {
  return GetJobError(j.original)
 }

pkg/testworkflows/executionworker/controller/watchers/pod.go

@@ -30,6 +30,7 @@ type Pod interface {
  Finished() bool
  ActionGroups() (actiontypes.ActionGroups, error)
  Signature() ([]stage.Signature, error)
+ ScheduledAt() (time.Time, error)
  ContainerStarted(name string) bool
  ContainerFinished(name string) bool
  ContainerFailed(name string) bool
@@ -116,6 +117,10 @@ func (p *pod) Signature() ([]stage.Signature, error) {
  return stage.GetSignatureFromJSON([]byte(p.original.Annotations[constants.SignatureAnnotationName]))
 }
 
+func (p *pod) ScheduledAt() (time.Time, error) {
+ return time.Parse(time.RFC3339Nano, p.original.Annotations[constants.ScheduledAtAnnotationName])
+}
+
 func (p *pod) ContainerStarted(name string) bool {
  return IsContainerStarted(p.original, name)
 }

pkg/testworkflows/executionworker/interface.go

@@ -39,10 +39,11 @@ type Config struct {
 
 // TODO: Consider some context data
 type ExecuteRequest struct {
- ResourceId string // defaults to execution ID
- GroupId string
- Workflow testworkflowsv1.TestWorkflow // TODO: Use OpenAPI object
- Secrets map[string]map[string]string
+ ResourceId string // defaults to execution ID
+ GroupId string
+ Workflow testworkflowsv1.TestWorkflow // TODO: Use OpenAPI object
+ Secrets map[string]map[string]string
+ ScheduledAt *time.Time
 
  Execution testworkflowconfig.ExecutionConfig
  ControlPlane testworkflowconfig.ControlPlaneConfig // TODO: Think if it's required
@@ -54,6 +55,7 @@ type ServiceRequest struct {
  GroupId string
  Workflow testworkflowsv1.TestWorkflow // TODO: Use OpenAPI object
  Secrets map[string]map[string]string
+ ScheduledAt *time.Time
  ReadinessProbe *testkube.Probe
  RestartPolicy string
 
@@ -74,13 +76,17 @@ type Hints struct {
 type ExecuteResult struct {
  // Signature for the deployed resource.
  Signature []testkube.TestWorkflowSignature
+ // ScheduledAt informs about scheduled time.
+ ScheduledAt time.Time
  // Namespace where it has been scheduled.
  Namespace string
 }
 
 type ServiceResult struct {
  // Signature for the deployed resource.
  Signature []testkube.TestWorkflowSignature
+ // ScheduledAt informs about scheduled time.
+ ScheduledAt time.Time
  // Namespace where it has been scheduled.
  Namespace string
 }

pkg/testworkflows/executionworker/worker.go

@@ -99,6 +99,12 @@ func (w *worker) Execute(ctx context.Context, request ExecuteRequest) (*ExecuteR
  if resourceId == "" {
  resourceId = request.Execution.Id
  }
+ scheduledAt := time.Now()
+ if request.ScheduledAt != nil {
+ scheduledAt = *request.ScheduledAt
+ } else if resourceId == request.Execution.Id && !request.Execution.ScheduledAt.IsZero() {
+ scheduledAt = request.Execution.ScheduledAt
+ }
  cfg := w.buildInternalConfig(resourceId, request.ArtifactsPathPrefix, request.Execution, request.ControlPlane, request.Workflow)
  secrets := w.buildSecrets(request.Secrets)
 
@@ -108,7 +114,7 @@ func (w *worker) Execute(ctx context.Context, request ExecuteRequest) (*ExecuteR
  }
 
  // Process the Test Workflow
- bundle, err := w.processor.Bundle(ctx, &request.Workflow, testworkflowprocessor.BundleOptions{Config: cfg, Secrets: secrets})
+ bundle, err := w.processor.Bundle(ctx, &request.Workflow, testworkflowprocessor.BundleOptions{Config: cfg, Secrets: secrets, ScheduledAt: scheduledAt})
  if err != nil {
  return nil, errors.Wrap(err, "failed to process test workflow")
  }
@@ -128,8 +134,9 @@ func (w *worker) Execute(ctx context.Context, request ExecuteRequest) (*ExecuteR
  }
 
  return &ExecuteResult{
- Signature: stage.MapSignatureListToInternal(bundle.Signature),
- Namespace: bundle.Job.Namespace,
+ Signature: stage.MapSignatureListToInternal(bundle.Signature),
+ ScheduledAt: scheduledAt,
+ Namespace: bundle.Job.Namespace,
  }, nil
 }
 
@@ -139,6 +146,12 @@ func (w *worker) Service(ctx context.Context, request ServiceRequest) (*ServiceR
  if resourceId == "" {
  resourceId = request.Execution.Id
  }
+ scheduledAt := time.Now()
+ if request.ScheduledAt != nil {
+ scheduledAt = *request.ScheduledAt
+ } else if resourceId == request.Execution.Id && !request.Execution.ScheduledAt.IsZero() {
+ scheduledAt = request.Execution.ScheduledAt
+ }
  cfg := w.buildInternalConfig(resourceId, "", request.Execution, request.ControlPlane, request.Workflow)
  secrets := w.buildSecrets(request.Secrets)
 
@@ -148,7 +161,7 @@ func (w *worker) Service(ctx context.Context, request ServiceRequest) (*ServiceR
  }
 
  // Process the Test Workflow
- bundle, err := w.processor.Bundle(ctx, &request.Workflow, testworkflowprocessor.BundleOptions{Config: cfg, Secrets: secrets})
+ bundle, err := w.processor.Bundle(ctx, &request.Workflow, testworkflowprocessor.BundleOptions{Config: cfg, Secrets: secrets, ScheduledAt: scheduledAt})
  if err != nil {
  return nil, errors.Wrap(err, "failed to process test workflow")
  }
@@ -182,8 +195,9 @@ func (w *worker) Service(ctx context.Context, request ServiceRequest) (*ServiceR
  }
 
  return &ServiceResult{
- Signature: stage.MapSignatureListToInternal(bundle.Signature),
- Namespace: bundle.Job.Namespace,
+ Signature: stage.MapSignatureListToInternal(bundle.Signature),
+ ScheduledAt: scheduledAt,
+ Namespace: bundle.Job.Namespace,
  }, nil
 }
 

pkg/testworkflows/testworkflowprocessor/bundle.go

@@ -3,6 +3,7 @@ package testworkflowprocessor
 import (
  "context"
  "encoding/json"
+ "time"
 
  "github.com/pkg/errors"
  batchv1 "k8s.io/api/batch/v1"
@@ -18,8 +19,9 @@ import (
 )
 
 type BundleOptions struct {
- Secrets []corev1.Secret
- Config testworkflowconfig.InternalConfig
+ Secrets []corev1.Secret
+ Config testworkflowconfig.InternalConfig
+ ScheduledAt time.Time
 }
 
 type Bundle struct {

pkg/testworkflows/testworkflowprocessor/constants/constants.go

@@ -23,6 +23,7 @@ const (
  RootResourceIdLabelName = "testkube.io/root"
  GroupIdLabelName = "testkube.io/contextGroup"
  SignatureAnnotationName = "testkube.io/signature"
+ ScheduledAtAnnotationName = "testkube.io/at"
  SpecAnnotationName = "testkube.io/spec"
  SpecAnnotationFieldPath = "metadata.annotations['" + SpecAnnotationName + "']"
  InternalAnnotationName = "testkube.io/config"

pkg/testworkflows/testworkflowprocessor/processor.go

@@ -6,6 +6,7 @@ import (
  "fmt"
  "maps"
  "path/filepath"
+ "time"
 
  "github.com/pkg/errors"
  batchv1 "k8s.io/api/batch/v1"
@@ -408,9 +409,10 @@ func (p *processor) Bundle(ctx context.Context, workflow *testworkflowsv1.TestWo
  podAnnotations := make(map[string]string)
  maps.Copy(podAnnotations, jobSpec.Spec.Template.Annotations)
  maps.Copy(podAnnotations, map[string]string{
- constants.SignatureAnnotationName: string(sigSerialized),
- constants.SpecAnnotationName: string(actionGroupsSerialized),
- constants.InternalAnnotationName: string(internalConfigSerialized),
+ constants.SignatureAnnotationName: string(sigSerialized),
+ constants.SpecAnnotationName: string(actionGroupsSerialized),
+ constants.InternalAnnotationName: string(internalConfigSerialized),
+ constants.ScheduledAtAnnotationName: options.ScheduledAt.UTC().Format(time.RFC3339Nano),
  })
  jobSpec.Spec.Template.Annotations = podAnnotations