Skip to content

Commit

Permalink
Add E2E for TensorFlow integration with Volcano
Browse files Browse the repository at this point in the history
  • Loading branch information
thandayuthapani committed Jul 26, 2019
1 parent 8748d21 commit 475f054
Show file tree
Hide file tree
Showing 4 changed files with 136 additions and 9 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ spec:
WORKER_HOST=`cat /etc/volcano/worker.host | sed 's/$/&:2222/g' | sed 's/^/"/;s/$/"/' | tr "\n" ","`;
export TF_CONFIG={\"cluster\":{\"ps\":[${PS_HOST}],\"worker\":[${WORKER_HOST}]},\"task\":{\"type\":\"ps\",\"index\":${VK_TASK_INDEX}},\"environment\":\"cloud\"};
python /var/tf_dist_mnist/dist_mnist.py
image: thanda/tf-operator-example:1.0
image: volcanosh/dist-mnist-tf-example:0.0.1
name: tensorflow
ports:
- containerPort: 2222
Expand All @@ -48,7 +48,7 @@ spec:
WORKER_HOST=`cat /etc/volcano/worker.host | sed 's/$/&:2222/g' | sed 's/^/"/;s/$/"/' | tr "\n" ","`;
export TF_CONFIG={\"cluster\":{\"ps\":[${PS_HOST}],\"worker\":[${WORKER_HOST}]},\"task\":{\"type\":\"worker\",\"index\":${VK_TASK_INDEX}},\"environment\":\"cloud\"};
python /var/tf_dist_mnist/dist_mnist.py
image: thanda/tf-operator-example:1.0
image: volcanosh/dist-mnist-tf-example:0.0.1
name: tensorflow
ports:
- containerPort: 2222
Expand Down
3 changes: 3 additions & 0 deletions hack/run-e2e-kind.sh
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ export LOG_LEVEL=3
export SHOW_VOLCANO_LOGS=${SHOW_VOLCANO_LOGS:-1}
export CLEANUP_CLUSTER=${CLEANUP_CLUSTER:-1}
export MPI_EXAMPLE_IMAGE=${MPI_EXAMPLE_IMAGE:-"volcanosh/example-mpi:0.0.1"}
export TF_EXAMPLE_IMAGE=${TF_EXAMPLE_IMAGE:-"volcanosh/dist-mnist-tf-example:0.0.1"}

if [[ "${CLUSTER_NAME}xxx" == "xxx" ]];then
CLUSTER_NAME="integration"
Expand Down Expand Up @@ -33,9 +34,11 @@ function install-volcano {

echo "Pulling required docker images"
docker pull ${MPI_EXAMPLE_IMAGE}
docker pull ${TF_EXAMPLE_IMAGE}

echo "Loading docker images into kind cluster"
kind load docker-image ${MPI_EXAMPLE_IMAGE} ${CLUSTER_CONTEXT}
kind load docker-image ${TF_EXAMPLE_IMAGE} ${CLUSTER_CONTEXT}

echo "Install volcano chart"
helm install installer/helm/chart/volcano --namespace kube-system --name ${CLUSTER_NAME} --kubeconfig ${KUBECONFIG} --set basic.image_tag_version=${TAG} --set basic.scheduler_config_file=volcano-scheduler-ci.conf --wait
Expand Down
123 changes: 123 additions & 0 deletions test/e2e/tensorflow.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,123 @@
/*
Copyright 2019 The Volcano Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package e2e

import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"

"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

vkv1 "volcano.sh/volcano/pkg/apis/batch/v1alpha1"
)

var _ = Describe("TensorFlow E2E Test", func() {
It("Will Start in pending state and goes through other phases to get complete phase", func() {
context := initTestContext()
defer cleanupTestContext(context)

jobName := "tensorflow-dist-mnist"

job := &vkv1.Job{
ObjectMeta: metav1.ObjectMeta{
Name: jobName,
},
Spec: vkv1.JobSpec{
MinAvailable: int32(3),
SchedulerName: schedulerName,
Plugins: map[string][]string{
"svc": {},
"env": {},
},
Policies: []vkv1.LifecyclePolicy{
{
Event: vkv1.PodEvictedEvent,
Action: vkv1.RestartJobAction,
},
},
Tasks: []vkv1.TaskSpec{
{
Replicas: int32(1),
Name: "ps",
Template: v1.PodTemplateSpec{
Spec: v1.PodSpec{
RestartPolicy: v1.RestartPolicyNever,
Containers: []v1.Container{
{
Command: []string{
"sh",
"-c",
"PS_HOST=`cat /etc/volcano/ps.host | sed 's/$/&:2222/g' | sed 's/^/\"/;s/$/\"/' | tr \"\n\" \",\"`; WORKER_HOST=`cat /etc/volcano/worker.host | sed 's/$/&:2222/g' | sed 's/^/\"/;s/$/\"/' | tr \"\n\" \",\"`; export TF_CONFIG={\\\"cluster\\\":{\\\"ps\\\":[${PS_HOST}],\\\"worker\\\":[${WORKER_HOST}]},\\\"task\\\":{\\\"type\\\":\\\"ps\\\",\\\"index\\\":${VK_TASK_INDEX}},\\\"environment\\\":\\\"cloud\\\"}; echo ${TF_CONFIG}; python /var/tf_dist_mnist/dist_mnist.py --train_steps 1000",
},
Image: defaultTFImage,
Name: "tensorflow",
Ports: []v1.ContainerPort{
{
Name: "tfjob-port",
ContainerPort: int32(2222),
},
},
},
},
},
},
},
{
Replicas: int32(2),
Name: "worker",
Policies: []vkv1.LifecyclePolicy{
{
Event: vkv1.TaskCompletedEvent,
Action: vkv1.CompleteJobAction,
},
},
Template: v1.PodTemplateSpec{
Spec: v1.PodSpec{
RestartPolicy: v1.RestartPolicyNever,
Containers: []v1.Container{
{
Command: []string{
"sh",
"-c",
"PS_HOST=`cat /etc/volcano/ps.host | sed 's/$/&:2222/g' | sed 's/^/\"/;s/$/\"/' | tr \"\n\" \",\"`; WORKER_HOST=`cat /etc/volcano/worker.host | sed 's/$/&:2222/g' | sed 's/^/\"/;s/$/\"/' | tr \"\n\" \",\"`; export TF_CONFIG={\\\"cluster\\\":{\\\"ps\\\":[${PS_HOST}],\\\"worker\\\":[${WORKER_HOST}]},\\\"task\\\":{\\\"type\\\":\\\"worker\\\",\\\"index\\\":${VK_TASK_INDEX}},\\\"environment\\\":\\\"cloud\\\"}; echo ${TF_CONFIG}; python /var/tf_dist_mnist/dist_mnist.py --train_steps 1000",
},
Image: defaultTFImage,
Name: "tensorflow",
Ports: []v1.ContainerPort{
{
Name: "tfjob-port",
ContainerPort: int32(2222),
},
},
},
},
},
},
},
},
},
}

created, err := context.vkclient.BatchV1alpha1().Jobs("test").Create(job)
Expect(err).NotTo(HaveOccurred())

err = waitJobStates(context, created, []vkv1.JobPhase{vkv1.Pending, vkv1.Inqueue, vkv1.Running, vkv1.Completed}, twoMinute)
Expect(err).NotTo(HaveOccurred())
})

})
15 changes: 8 additions & 7 deletions test/e2e/util.go
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ const (
defaultBusyBoxImage = "busybox:1.24"
defaultMPIImage = "volcanosh/example-mpi:0.0.1"
schedulerName = "volcano"
defaultTFImage = "volcanosh/dist-mnist-tf-example:0.0.1"

defaultNamespace = "test"
defaultQueue1 = "q1"
Expand Down Expand Up @@ -575,9 +576,9 @@ func waitJobPhases(ctx *context, job *batchv1alpha1.Job, phases []batchv1alpha1.
return nil
}

func waitJobStates(ctx *context, job *batchv1alpha1.Job, phases []batchv1alpha1.JobPhase) error {
func waitJobStates(ctx *context, job *batchv1alpha1.Job, phases []batchv1alpha1.JobPhase, waitTime time.Duration) error {
for _, phase := range phases {
err := waitJobPhaseExpect(ctx, job, phase)
err := waitJobPhaseExpect(ctx, job, phase, waitTime)
if err != nil {
return err
}
Expand Down Expand Up @@ -665,22 +666,22 @@ func waitTasksPending(ctx *context, job *batchv1alpha1.Job, taskNum int) error {
}

func waitJobStateReady(ctx *context, job *batchv1alpha1.Job) error {
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Running)
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Running, oneMinute)
}

func waitJobStatePending(ctx *context, job *batchv1alpha1.Job) error {
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Pending)
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Pending, oneMinute)
}

func waitJobStateInqueue(ctx *context, job *batchv1alpha1.Job) error {
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Inqueue)
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Inqueue, oneMinute)
}

func waitJobStateAborted(ctx *context, job *batchv1alpha1.Job) error {
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Aborted)
return waitJobPhaseExpect(ctx, job, batchv1alpha1.Aborted, oneMinute)
}

func waitJobPhaseExpect(ctx *context, job *batchv1alpha1.Job, state batchv1alpha1.JobPhase) error {
func waitJobPhaseExpect(ctx *context, job *batchv1alpha1.Job, state batchv1alpha1.JobPhase, waitTime time.Duration) error {
var additionalError error
err := wait.Poll(100*time.Millisecond, oneMinute, func() (bool, error) {
job, err := ctx.vcclient.BatchV1alpha1().Jobs(job.Namespace).Get(job.Name, metav1.GetOptions{})
Expand Down

0 comments on commit 475f054

Please sign in to comment.