Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(backend): Add Semaphore and Mutex fields to Workflow CR #11370

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
+101 −26
Open

feat(backend): Add Semaphore and Mutex fields to Workflow CR #11370

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 35 additions & 16 deletions backend/src/apiserver/template/v2_template.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,34 @@
Launcher = ""
)

func getKubernetesSpec(platformSpec map[string]*pipelinespec.SinglePlatformSpec) *pipelinespec.SinglePlatformSpec {
var kubernetesSpec *pipelinespec.SinglePlatformSpec

// Check for "kubernetes" key in the platformSpec map
if platformSpec != nil {
if platform, ok := platformSpec["kubernetes"]; ok && platform != nil {
kubernetesSpec = platform
}
}
return kubernetesSpec
}

func getPipelineOptions(platform *pipelinespec.SinglePlatformSpec) *argocompiler.Options {
var pipelineOptions *argocompiler.Options

if platform != nil && platform.PipelineConfig != nil {

Check failure on line 63 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / run-go-unittests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)

Check failure on line 63 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / backend-tests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)
pipelineOptions = &argocompiler.Options{}
if platform.PipelineConfig.SemaphoreKey != "" {

Check failure on line 65 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / run-go-unittests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)

Check failure on line 65 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / backend-tests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)
pipelineOptions.SemaphoreKey = platform.PipelineConfig.SemaphoreKey

Check failure on line 66 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / run-go-unittests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)

Check failure on line 66 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / backend-tests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)
}
if platform.PipelineConfig.MutexName != "" {

Check failure on line 68 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / run-go-unittests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)

Check failure on line 68 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / backend-tests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)
pipelineOptions.MutexName = platform.PipelineConfig.MutexName

Check failure on line 69 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / run-go-unittests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)

Check failure on line 69 in backend/src/apiserver/template/v2_template.go

View workflow job for this annotation

GitHub Actions / backend-tests 

platform.PipelineConfig undefined (type *pipelinespec.SinglePlatformSpec has no field or method PipelineConfig)
}
}
return pipelineOptions
}


// Converts modelJob to ScheduledWorkflow.
func (t *V2Spec) ScheduledWorkflow(modelJob *model.Job) (*scheduledworkflow.ScheduledWorkflow, error) {
job := &pipelinespec.PipelineJob{}
Expand All @@ -69,17 +97,12 @@
return nil, util.Wrap(err, "invalid pipeline job inputs")
}

// Pick out Kubernetes platform configs
var kubernetesSpec *pipelinespec.SinglePlatformSpec
if t.platformSpec != nil {
if _, ok := t.platformSpec.Platforms["kubernetes"]; ok {
kubernetesSpec = t.platformSpec.Platforms["kubernetes"]
}
}
kubernetesSpec := getKubernetesSpec(t.platformSpec.Platforms)
pipelineOptions := getPipelineOptions(kubernetesSpec)

var obj interface{}
if util.CurrentExecutionType() == util.ArgoWorkflow {
obj, err = argocompiler.Compile(job, kubernetesSpec, nil)
obj, err = argocompiler.Compile(job, kubernetesSpec, pipelineOptions)
} else if util.CurrentExecutionType() == util.TektonPipelineRun {
obj, err = tektoncompiler.Compile(job, kubernetesSpec, &tektoncompiler.Options{LauncherImage: Launcher})
}
Expand Down Expand Up @@ -292,17 +315,13 @@
if err = t.validatePipelineJobInputs(job); err != nil {
return nil, util.Wrap(err, "invalid pipeline job inputs")
}
// Pick out Kubernetes platform configs
var kubernetesSpec *pipelinespec.SinglePlatformSpec
if t.platformSpec != nil {
if _, ok := t.platformSpec.Platforms["kubernetes"]; ok {
kubernetesSpec = t.platformSpec.Platforms["kubernetes"]
}
}

kubernetesSpec := getKubernetesSpec(t.platformSpec.Platforms)
pipelineOptions := getPipelineOptions(kubernetesSpec)

var obj interface{}
if util.CurrentExecutionType() == util.ArgoWorkflow {
obj, err = argocompiler.Compile(job, kubernetesSpec, nil)
obj, err = argocompiler.Compile(job, kubernetesSpec, pipelineOptions)
} else if util.CurrentExecutionType() == util.TektonPipelineRun {
obj, err = tektoncompiler.Compile(job, kubernetesSpec, nil)
}
Expand Down
42 changes: 33 additions & 9 deletions backend/src/v2/compiler/argocompiler/argo.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
"encoding/json"
"fmt"
"strings"
"os"

wfapi "github.com/argoproj/argo-workflows/v3/pkg/apis/workflow/v1alpha1"
"github.com/kubeflow/pipelines/api/v2alpha1/go/pipelinespec"
Expand All @@ -40,6 +41,16 @@ type Options struct {
// optional
PipelineRoot string
// TODO(Bobgy): add an option -- dev mode, ImagePullPolicy should only be Always in dev mode.
SemaphoreKey string
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't love using this Options struct for these fields, but I'm not sure what an alternative would be. I guess it's fine for now and maybe we can figure out a cleaner way down the line.

MutexName string
}

func getSemaphoreConfigMapName() string {
const defaultConfigMapName = "semaphore-config"
if name := os.Getenv("SEMAPHORE_CONFIGMAP_NAME"); name != "" {
return name
}
return defaultConfigMapName
}

func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.SinglePlatformSpec, opts *Options) (*wfapi.Workflow, error) {
Expand Down Expand Up @@ -86,22 +97,13 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
}
}

// initialization
wf := &wfapi.Workflow{
TypeMeta: k8smeta.TypeMeta{
APIVersion: "argoproj.io/v1alpha1",
Kind: "Workflow",
},
ObjectMeta: k8smeta.ObjectMeta{
GenerateName: retrieveLastValidString(spec.GetPipelineInfo().GetName()) + "-",
// Note, uncomment the following during development to view argo inputs/outputs in KFP UI.
// TODO(Bobgy): figure out what annotations we should use for v2 engine.
// For now, comment this annotation, so that in KFP UI, it shows argo input/output params/artifacts
// suitable for debugging.
//
// Annotations: map[string]string{
// "pipelines.kubeflow.org/v2_pipeline": "true",
// },
},
Spec: wfapi.WorkflowSpec{
PodMetadata: &wfapi.Metadata{
Expand All @@ -119,6 +121,28 @@ func Compile(jobArg *pipelinespec.PipelineJob, kubernetesSpecArg *pipelinespec.S
Entrypoint: tmplEntrypoint,
},
}

if opts != nil && opts.SemaphoreKey != "" {
wf.Spec.Synchronization = &wfapi.Synchronization{
Semaphore: &wfapi.SemaphoreRef{
ConfigMapKeyRef: &k8score.ConfigMapKeySelector{
LocalObjectReference: k8score.LocalObjectReference{
Name: getSemaphoreConfigMapName(),
},
Key: opts.SemaphoreKey,
},
},
}
}

if opts != nil && opts.MutexName != "" {
wf.Spec.Synchronization = &wfapi.Synchronization{
Mutex: &wfapi.Mutex{
Name: opts.MutexName,
},
}
}

c := &workflowCompiler{
wf: wf,
templates: make(map[string]*wfapi.Template),
Expand Down
2 changes: 1 addition & 1 deletion backend/src/v2/compiler/visitor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ func (state *pipelineDFS) dfs(name string, component *pipelinespec.ComponentSpec
}

// Add kubernetes spec to annotation
if state.kubernetesSpec != nil {
if state.kubernetesSpec != nil && state.kubernetesSpec.DeploymentSpec != nil {
kubernetesExecSpec, ok := state.kubernetesSpec.DeploymentSpec.Executors[executorLabel]
if ok {
state.visitor.AddKubernetesSpec(name, kubernetesExecSpec)
Expand Down
1 change: 1 addition & 0 deletions manifests/kustomize/base/pipeline/kustomization.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ resources:
- ml-pipeline-scheduledworkflow-role.yaml
- ml-pipeline-scheduledworkflow-rolebinding.yaml
- ml-pipeline-scheduledworkflow-sa.yaml
- ml-pipeline-semaphore-configmap.yaml
- ml-pipeline-ui-deployment.yaml
- ml-pipeline-ui-configmap.yaml
- ml-pipeline-ui-role.yaml
Expand Down
2 changes: 2 additions & 0 deletions manifests/kustomize/base/pipeline/ml-pipeline-apiserver-deployment.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,8 @@ spec:
secretKeyRef:
name: mlpipeline-minio-artifact
key: secretkey
- name: SEMAPHORE_CONFIGMAP_NAME
value: "semaphore-config"
image: gcr.io/ml-pipeline/api-server:dummy
imagePullPolicy: IfNotPresent
name: ml-pipeline-api-server
Expand Down
29 changes: 29 additions & 0 deletions manifests/kustomize/base/pipeline/ml-pipeline-semaphore-configmap.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,29 @@
kind: ConfigMap
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this lgtm, but I just had a thought. What happens if:

  1. I install kfp and I get this configmap created
  2. I customize it by adding keys to it
  3. I upgrade to the next version of kfp

Is that down the line upgrade going to overwrite my customized configmap with this blank one? It'd be cool if we could test that somehow -- perhaps by manually creating the configmap on a cluster, putting data in it, and then installing kfp.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Makes sense.
To resolve this, I've added a job to the configmap, semaphore-configmap-init that runs during the Kustomize deployment process.
It checks for the existence of the semaphore-config ConfigMap using kubectl get.
If the ConfigMap does not exist, it creates one with an empty init key.
If the ConfigMap already exists, the Job skips creation and exits successfully.

apiVersion: v1
metadata:
name: semaphore-config
data: {}
---
apiVersion: batch/v1
kind: Job
metadata:
name: semaphore-configmap-init
namespace: kubeflow
spec:
template:
spec:
containers:
- name: create-configmap
image: bitnami/kubectl:latest
command:
- /bin/sh
- -c
- |
if ! kubectl get configmap semaphore-config -n kubeflow > /dev/null 2>&1; then
echo "Creating semaphore-config ConfigMap..."
kubectl create configmap semaphore-config -n kubeflow --from-literal=init=""
else
echo "ConfigMap semaphore-config already exists. Skipping creation."
fi
restartPolicy: OnFailure
backoffLimit: 3
Loading