// WithTaskCheckpoint allows a task to be created with live runtime and memory data from a
// previous checkpoint. Additional software such as CRIU may be required to
// restore a task from a checkpoint
431764
func WithTaskCheckpoint(im Image) NewTaskOpts {
	return func(ctx context.Context, c *Client, info *TaskInfo) error {
		desc := im.Target()
		id := desc.Digest
		index, err := decodeIndex(ctx, c.ContentStore(), desc)
		if err != nil {
			return err
		}
		for _, m := range index.Manifests {
			if m.MediaType == images.MediaTypeContainerd1Checkpoint {
				info.Checkpoint = &types.Descriptor{
					MediaType:   m.MediaType,
					Size_:       m.Size,
					Digest:      m.Digest,
					Annotations: m.Annotations,
				}
				return nil
			}
		}
		return fmt.Errorf("checkpoint not found in index %s", id)
	}
}
964121
func MarkTaskSucceeded(t *automationpb.Task, output string) {
	t.State = automationpb.TaskState_DONE
	t.Output = output
}
322121
func (c *Coordinator) claimExistingTasks() {
	tasks, _, err := c.TaskService.FindTasks(context.Background(), platform.TaskFilter{})
	if err != nil {
		return
	}
	newLatestCompleted := time.Now().UTC().Format(time.RFC3339)
	for len(tasks) > 0 {
		for _, task := range tasks {

			task, err := c.TaskService.UpdateTask(context.Background(), task.ID, platform.TaskUpdate{LatestCompleted: &newLatestCompleted})
			if err != nil {
				c.logger.Error("failed to set latestCompleted", zap.Error(err))
			}

			if task.Status != string(backend.TaskActive) {
				// Don't claim inactive tasks at startup.
				continue
			}

			// I may need a context with an auth here
			if err := c.sch.ClaimTask(context.Background(), task); err != nil {
				c.logger.Error("failed claim task", zap.Error(err))
				continue
			}
		}
		tasks, _, err = c.TaskService.FindTasks(context.Background(), platform.TaskFilter{
			After: &tasks[len(tasks)-1].ID,
		})
		if err != nil {
			c.logger.Error("failed list additional tasks", zap.Error(err))
			return
		}
	}
}
675745
func WithCheckpointImage(ctx context.Context, client *Client, c *containers.Container, index *imagespec.Index, copts *options.CheckpointOptions) error {
	ir, err := client.ImageService().Get(ctx, c.Image)
	if err != nil {
		return err
	}
	index.Manifests = append(index.Manifests, ir.Target)
	return nil
}
1254699
func getCheckpointDir(checkDir, checkpointID, ctrName, ctrID, ctrCheckpointDir string, create bool) (string, error) {
	var checkpointDir string
	var err2 error
	if checkDir != "" {
		checkpointDir = checkDir
	} else {
		checkpointDir = ctrCheckpointDir
	}
	checkpointAbsDir := filepath.Join(checkpointDir, checkpointID)
	stat, err := os.Stat(checkpointAbsDir)
	if create {
		switch {
		case err == nil && stat.IsDir():
			err2 = fmt.Errorf("checkpoint with name %s already exists for container %s", checkpointID, ctrName)
		case err != nil && os.IsNotExist(err):
			err2 = os.MkdirAll(checkpointAbsDir, 0700)
		case err != nil:
			err2 = err
		case err == nil:
			err2 = fmt.Errorf("%s exists and is not a directory", checkpointAbsDir)
		}
	} else {
		switch {
		case err != nil:
			err2 = fmt.Errorf("checkpoint %s does not exist for container %s", checkpointID, ctrName)
		case err == nil && stat.IsDir():
			err2 = nil
		case err == nil:
			err2 = fmt.Errorf("%s exists and is not a directory", checkpointAbsDir)
		}
	}
	return checkpointAbsDir, err2
}
392522
func (ar *allocRunner) RestartTask(taskName string, taskEvent *structs.TaskEvent) error {
	tr, ok := ar.tasks[taskName]
	if !ok {
		return fmt.Errorf("Could not find task runner for task: %s", taskName)
	}

	return tr.Restart(context.TODO(), taskEvent, false)
}
855644
func (sm *schedulerMetrics) ClaimTask(succeeded bool) {
	status := statusString(succeeded)

	sm.claimsComplete.WithLabelValues(status).Inc()
	if succeeded {
		sm.claimsActive.Inc()
	}
}
405629
func withCheckpointOpts(rt string, context *cli.Context) containerd.CheckpointTaskOpts {
	return func(r *containerd.CheckpointTaskInfo) error {
		imagePath := context.String("image-path")
		workPath := context.String("work-path")

		switch rt {
		case plugin.RuntimeRuncV1, plugin.RuntimeRuncV2:
			if r.Options == nil {
				r.Options = &options.CheckpointOptions{}
			}
			opts, _ := r.Options.(*options.CheckpointOptions)

			if context.Bool("exit") {
				opts.Exit = true
			}
			if imagePath != "" {
				opts.ImagePath = imagePath
			}
			if workPath != "" {
				opts.WorkPath = workPath
			}
		case plugin.RuntimeLinuxV1:
			if r.Options == nil {
				r.Options = &runctypes.CheckpointOptions{}
			}
			opts, _ := r.Options.(*runctypes.CheckpointOptions)

			if context.Bool("exit") {
				opts.Exit = true
			}
			if imagePath != "" {
				opts.ImagePath = imagePath
			}
			if workPath != "" {
				opts.WorkPath = workPath
			}
		}
		return nil
	}
}
992431
func (manager *impl) GetCheckpoint(checkpointKey string, checkpoint Checkpoint) error {
	manager.mutex.Lock()
	defer manager.mutex.Unlock()
	blob, err := manager.store.Read(checkpointKey)
	if err != nil {
		if err == utilstore.ErrKeyNotFound {
			return errors.ErrCheckpointNotFound
		}
		return err
	}
	err = checkpoint.UnmarshalCheckpoint(blob)
	if err == nil {
		err = checkpoint.VerifyChecksum()
	}
	return err
}
100724
func loadCheckpoint(checkpointFile string) (string, error) {
	content, err := ioutil.ReadFile(checkpointFile)
	if err != nil {
		if os.IsNotExist(err) {
			return "", nil
		}
		return "", err
	}
	return string(content), nil
}
377307
func (cp *CPUManagerCheckpoint) MarshalCheckpoint() ([]byte, error) {
	// make sure checksum wasn't set before so it doesn't affect output checksum
	cp.Checksum = 0
	cp.Checksum = checksum.New(cp)
	return json.Marshal(*cp)
}
1240958
func (pdev podDevices) fromCheckpointData(data []checkpoint.PodDevicesEntry) {
	for _, entry := range data {
		klog.V(2).Infof("Get checkpoint entry: %v %v %v %v %v\n",
			entry.PodUID, entry.ContainerName, entry.ResourceName, entry.DeviceIDs, entry.AllocResp)
		devIDs := sets.NewString()
		for _, devID := range entry.DeviceIDs {
			devIDs.Insert(devID)
		}
		allocResp := &pluginapi.ContainerAllocateResponse{}
		err := allocResp.Unmarshal(entry.AllocResp)
		if err != nil {
			klog.Errorf("Can't unmarshal allocResp for %v %v %v: %v", entry.PodUID, entry.ContainerName, entry.ResourceName, err)
			continue
		}
		pdev.insert(entry.PodUID, entry.ContainerName, entry.ResourceName, devIDs, allocResp)
	}
}
850683
func (s *CheckpointConfigurationDescription) SetCheckpointInterval(v int64) *CheckpointConfigurationDescription {
	s.CheckpointInterval = &v
	return s
}
711444
func (c *CheckpointWriter) UpdateTask(taskID string, status workflowpb.TaskState, err error) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	errorMessage := ""
	if err != nil {
		errorMessage = err.Error()
	}

	t := c.checkpoint.Tasks[taskID]
	t.State = status
	t.Error = errorMessage
	return c.saveLocked()
}
828340
func (s *FlinkApplicationConfigurationDescription) SetCheckpointConfigurationDescription(v *CheckpointConfigurationDescription) *FlinkApplicationConfigurationDescription {
	s.CheckpointConfigurationDescription = v
	return s
}
328833
func TaskCheckStateGreaterThan(t1, t2 *api.Task) bool {
	return t2.Status.State > t1.Status.State
}
505423
func (s *globalTaskService) TaskTrack(ctx types.Context) *types.Task {
	return &s.taskTrack(ctx).Task
}