Custom Tasks: question with taskSpec

[Tomcli]

related #3463

The current custom task can run in a pipeline using pipeline.taskRef. Will there be any support for pipeline.taskSpec in the roadmap?

/cc @GregDritschler @fenglixa

[imjasonh]

Very possibly! I think we should take it slow to get there and collect user feedback along the way, but there's nothing technically difficult about supporting it, and I think it could help reduce complexity if done well.

Sorry if that "maybe" is not a terribly definitive answer. 🙃

[Tomcli]

Thanks Jason, we are building the a new custom task for our pipeline. Our use case is to have multiple users in the same namespace, so our users sometimes create the custom task with the same name.

Right now we try to work around with appending a unique user string to each custom task name, so the name can be unique for each user. We would like to provide more feedback as custom task is getting more task feature in the upcoming releases.

[bobcatfish]

My 2 cents is that I think this is totally reasonable to add - esp since we support this for other Tasks (I think I've been assuming we'd support this for sure)

[fenglixa]

Thanks @bobcatfish.
Great that we will support pipeline.taskSpec for custom task defination!

[Tomcli]

/cc @afrittoli @animeshsingh @pritidesai

[pritidesai]

/assign

yup, this is reasonable and we have a use case.

Custom task is specified using taskRef:

spec:
  tasks:
    - name: run-my-custom-task
      taskRef:
        apiVersion: example.tekton.dev/v1alpha1
        kind: Example

And its classified as custom task if taskRef.APIVersion and taskRef.Kind are populated:

isCustomTask := cfg.FeatureFlags.EnableCustomTasks && t.TaskRef != nil && t.TaskRef.APIVersion != "" && t.TaskRef.Kind != ""

@Tomcli what is your expected usage with taskSpec? Please help me while I am trying to understand how taskSpec can be supported.

[pritidesai]

/remove-kind question
/kind feature

[imjasonh]

I think what's being asked for is support for inlining the custom task CR definition in the PipelineSpec, instead of only being able to reference the definition elsewhere. Something along the lines of:

spec:
  tasks:
    - name: run-my-custom-task
      taskSpec:
        apiVersion: example.tekton.dev/v1alpha1  # <-- this is a new field
        kind: Example                            # <-- this is a new field
        spec:                                    # <-- this is a new field, of type json.RawMessage?
          customDetailsOfTheExample: ...

(Please correct me if I'm wrong!)

This is similar to how taskSpec lets you inline the Task definition (steps, etc.) in the PipelineSpec, instead of only referencing external Tasks.

When run, the PipelineRun controller could create a Run that includes the inlined taskSpec information:

apiVersion: tekton.dev/v1alpha1
kind: Run
...
spec:
  spec:  # <-- this is new json.RawMessage field, *and* its name is bad :(
    customDetailsOfExample: ...

And in turn the Custom Task controller for Examples would have to understand how to interpret that and perform the custom task behavior.

I don't think it's a very complicated change, but it's kind of a lot to add to the API, in the name of simplifying a case that's possible today just by referencing external CRs instead of inlining them. I'd prefer to start with a prototype at least of the controller you intend to build to take advantage of this new way to express custom tasks in a pipeline. This would help us more clearly identify the use case, and identify exactly where we can improve it.

If you already have a custom task controller you're working on I'd love to hear more about it!

[Tomcli]

We already have our own custom task controller and spec ready, here is one of our examples.
https://github.com/kubeflow/kfp-tekton/blob/master/tekton-catalog/pipeline-loops/examples/loop-example-basic.yaml

The use case we have is to run the custom task inside a pipelineRun using the Tekton pipelineRun go client. We are using Kubeflow Pipeline to do this which assumes all the information for displaying the graph is under one Kubernetes resource (for tekton is the pipelineRun).

For taskSpec, we want to figure out where we should add the apiVersion and kind field. The solution you have above is a great option. We probably need to update the Tekton pipelinerun reconciler to see can we detect it as a custom task without too much api spec changes. It looks like the core logic is under

pipeline/pkg/reconciler/pipelinerun/resources/pipelinerunresolution.go

Lines 356 to 366 in 08177fa

	rprt.CustomTask = cfg.FeatureFlags.EnableCustomTasks && rprt.PipelineTask.TaskRef != nil &&
	rprt.PipelineTask.TaskRef.APIVersion != "" && rprt.PipelineTask.TaskRef.Kind != ""
	if rprt.IsCustomTask() {
	rprt.RunName = GetRunName(pipelineRun.Status.Runs, task.Name, pipelineRun.Name)
	run, err := getRun(rprt.RunName)
	if err != nil && !errors.IsNotFound(err) {
	return nil, fmt.Errorf("error retrieving Run %s: %w", rprt.RunName, err)
	}
	rprt.Run = run
	} else {

[vincent-pli]

@imjasonh
I think runtime.RawExtension could satisfy the requirement, user could address any information in the run to shape the behavior of custom controller and avoid to create a template CRD in advance.

As @Tomcli mentioned the loop is just one case we still have Exception handler and some incoming components adopt the custom task, they hit the same issue, so hope we can start the task.

[afrittoli]

Being able to embed custom task specs in pipelines would be consistent with the existing API; to having it instead could inhibit usability of custom tasks instead. We would need to define the name of the field use to embed the spec. The only question I have, if whether support for this new field would be optional or mandatory for custom controllers.
In any case this is an API change, so it will require a small TEP :)

[pritidesai]

@imjasonh please confirm 🙏

So what we are proposing here is to mainly add a new field in the RunSpec to hold the specification in raw bytes (RawExtension) similar to taskSpec (which is of type TaskSpec) and expect the custom controller to interpret those bytes to discover the specifications from the RunSpec?

pipeline/pkg/apis/pipeline/v1alpha1/run_types.go

Lines 40 to 43 in 1cd612f

	type RunSpec struct {
	// +optional
	Ref *TaskRef `json:"ref,omitempty"`

pipeline/pkg/apis/pipeline/v1beta1/taskrun_types.go

Lines 44 to 55 in 1cd612f

	type TaskRunSpec struct {
	// +optional
	Params []Param `json:"params,omitempty"`
	// +optional
	Resources *TaskRunResources `json:"resources,omitempty"`
	// +optional
	ServiceAccountName string `json:"serviceAccountName"`
	// no more than one of the TaskRef and TaskSpec may be specified.
	// +optional
	TaskRef *TaskRef `json:"taskRef,omitempty"`
	// +optional
	TaskSpec *TaskSpec `json:"taskSpec,omitempty"`

https://github.com/kubernetes/apimachinery/blob/03ac7a9ade429d715a1a46ceaa3724c18ebae54f/pkg/runtime/types.go#L94

@Tomcli I am trying to understand what value is this feature adding? It will be great to be aligned with the existing API but what kind of issues would this solve? The biggest advantage of this would be to not deal with the management nightmare of those custom resources 🤔

[Tomcli]

Thanks @pritidesai. This feature can help us

1. Users can put all the information in one pipelineRun CR.
2. Users don't have to manage the custom task CR. Since custom task CR is namespace scope, if there are multiple users in the same namespace, they will have conflicts when they are using the same name for their custom task CR.
3. The pipeline developer might use several different custom task controllers and APIs. The Kubernetes/Tekton admins and operators will have a hard time to learn and manage these new CRDs

[Tomcli]

In our use case, KFP is the admin for storing and managing all the PipelineRuns. We are having a hard time to track all the custom resources because they all have different APIs.

[imjasonh]

@imjasonh please confirm 🙏

Yep, that seems right to me.

[pritidesai]

Alright, I took sometime to put the puzzle pieces together before writing a TEP for it. Let me know if its headed in right direction:

Introduce a new field customSpec in pipelineTask which can be merged with existing taskSpec if needed. Chose to create a new field just for the PoC:

type PipelineTask struct {
	TaskRef *TaskRef `json:"taskRef,omitempty"`
	CustomSpec *CustomInlinedSpec `json:"customSpec,omitempty"`
...

customeSpec is of type CustomInlinedSpec which holds APIVersion, Kind, and Spec:

type CustomInlinedSpec struct {
	runtime.TypeMeta `json:",inline,omitempty"`
	Spec             runtime.RawExtension `json:"spec,omitempty"`
}

Updating the RunSpec to include this new type:

type RunSpec struct {
	Ref *TaskRef `json:"ref,omitempty"`
	CustomSpec *v1beta1.CustomInlinedSpec `json:"customSpec,omitempty"`
...

Now, while creating a Run object, initialize this new field customSpec with the specified APIVersion and Kind. Also, marshal the customSpec.spec into Run object:

		run.Spec.CustomSpec.Spec = runtime.RawExtension{
			Raw: json.Marshal(rprt.PipelineTask.CustomSpec.Spec),
		}

These changes ships the entire inlined specification with the Run object. Its the custom controller's job to interpret this specification and use it the way it desires.

I took an example of taskLoop controller to experiment with this. The way its implemented, getTaskLoop will have to adapt to this new model and initialize taskLoopSpec:

func (c *Reconciler) getTaskLoop(ctx context.Context, run *v1alpha1.Run) (*metav1.ObjectMeta, *taskloopv1alpha1.TaskLoopSpec, error) {
	taskLoopMeta := metav1.ObjectMeta{}
	taskLoopSpec := taskloopv1alpha1.TaskLoopSpec{}
	if run.Spec.CustomSpec != nil {
		s := run.Spec.CustomSpec
		# s.APIVersion holds the APIVersion specified under customSpec
		# s.Kind holds Kind specified under customSpec
                 # either decode it using the GroupVersion or unmarshal directly into taskLoopSpec
		err := json.Unmarshal(s.Spec.Raw, &taskLoopSpec)
		if err != nil {
                   return nil, nil, err
		}
	}
...

So, the following sample pipeline:

apiVersion: tekton.dev/v1beta1
kind: Pipeline
metadata:
  name: testpipeline
spec:
  tasks:
    - name: run-my-tests
      customSpec:
        apiVersion: custom.tekton.dev/v1alpha1
        kind: TaskLoop
        spec:
          taskRef:
            name: simpletask
          iterateParam: word
          timeout: 60s
          retries: 2

Initializes taskLoopSpec in taskLoop controller with:

(v1alpha1.TaskLoopSpec) {
 TaskRef: (*v1beta1.TaskRef)(0xc0007bf4c0)({
  Name: (string) (len=10) "simpletask",
 }),
 IterateParam: (string) (len=4) "word",
 Timeout: (*v1.Duration)(0xc000ebfa10)(&Duration{Duration:1m0s,}),
 Retries: (int) 2,
}

Anything missing here? I have the pipeline controller changes here.