Make edl controller jobparser workable

Dockerfile

@@ -2,8 +2,14 @@ FROM    golang:1.8
 RUN     go get github.com/Masterminds/glide
 RUN     apt-get update && apt-get install -y git
 WORKDIR $GOPATH/src/github.com/paddlepaddle
-RUN     git clone https://github.com/paddlepaddle/edl.git
+RUN     mkdir -p $GOPATH/src/github.com/paddlepaddle/edl
+# Add ENV http_proxy=[your proxy server] if needed
+# run glide install before building go sources, so that
+# if we change the code and rebuild the image can cost little time
+ADD     ./glide.yaml ./glide.lock $GOPATH/src/github.com/paddlepaddle/edl/
 WORKDIR $GOPATH/src/github.com/paddlepaddle/edl
 RUN     glide install --strip-vendor
+ADD     . $GOPATH/src/github.com/paddlepaddle/edl
 RUN     go build -o /usr/local/bin/edl github.com/paddlepaddle/edl/cmd/edl
+RUN     rm -rf $GOPATH/src/github.com/paddlepaddle/edl
 CMD     ["edl"]

cmd/edl/edl.go

@@ -24,7 +24,7 @@ func main() {
 	candy.Must(err)
 
 	log.Root().SetHandler(
-		log.LvlFilterHandler(lvl, log.CallerStackHandler("%+v", log.StderrHandler)),
+		log.LvlFilterHandler(lvl, log.CallerFileHandler(log.StderrHandler)),
 	)
 
 	// Create the client config. Use kubeconfig if given, otherwise assume in-cluster.

edl_controller.yaml

@@ -0,0 +1,21 @@
+apiVersion: extensions/v1beta1
+kind: Deployment
+metadata:
+  name: training-job-controller
+  namespace: paddlecloud
+spec:
+  replicas: 1
+  template:
+    metadata:
+      labels:
+        name: training-job-controller
+    spec:
+      containers:
+      - name: training-job-controller
+        image: paddlepaddle/edl-controller
+        env:
+        - name: https_proxy
+          value: ""
+        - name: http_proxy
+          value: ""
+        command: ["/usr/local/bin/edl", "-logtostderr", "-log_level", "info"]

example/Dockerfile

@@ -0,0 +1,4 @@
+FROM paddlepaddle/paddlecloud-job
+RUN mkdir -p /workspace
+ADD train_ft.py /workspace
+

example/examplejob.yaml

@@ -0,0 +1,34 @@
+apiVersion: paddlepaddle.org/v1
+kind: TrainingJob
+metadata:
+  name: example
+spec:
+  image: "paddlepaddle/paddlecloud-job"
+  port: 7164
+  ports_num: 1
+  ports_num_for_sparse: 1
+  fault_tolerant: true
+  trainer:
+    entrypoint: "python /workspace/vgg16_v2.py"
+    workspace: "/workspace"
+    passes: 50
+    min-instance: 2
+    max-instance: 6
+    resources:
+      limits:
+        #alpha.kubernetes.io/nvidia-gpu: 1
+        cpu: "200m"
+        memory: "200Mi"
+      requests:
+        cpu: "200m"
+        memory: "200Mi"
+  pserver:
+    min-instance: 2
+    max-instance: 2
+    resources:
+      limits:
+        cpu: "800m"
+        memory: "1Gi"
+      requests:
+        cpu: "500m"
+        memory: "600Mi"

example/train_ft.py

@@ -0,0 +1,140 @@
+#   Copyright (c) 2018 PaddlePaddle Authors. All Rights Reserved.
+#
+# 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.
+
+from PIL import Image
+import numpy as np
+import paddle.v2 as paddle
+import paddle.v2.dataset.common as common
+import paddle.v2.dataset as dataset
+import os
+import sys
+
+TRAINER_ID = int(os.getenv("PADDLE_INIT_TRAINER_ID", "-1"))
+TOTAL_TRAINERS = int(os.getenv("PADDLE_INIT_NUM_GRADIENT_SERVERS", "-1"))
+
+
+def softmax_regression(img):
+    predict = paddle.layer.fc(input=img,
+                              size=10,
+                              act=paddle.activation.Softmax())
+    return predict
+
+
+def multilayer_perceptron(img):
+    # The first fully-connected layer
+    hidden1 = paddle.layer.fc(input=img,
+                              size=128,
+                              act=paddle.activation.Relu())
+    # The second fully-connected layer and the according activation function
+    hidden2 = paddle.layer.fc(input=hidden1,
+                              size=64,
+                              act=paddle.activation.Relu())
+    # The thrid fully-connected layer, note that the hidden size should be 10,
+    # which is the number of unique digits
+    predict = paddle.layer.fc(input=hidden2,
+                              size=10,
+                              act=paddle.activation.Softmax())
+    return predict
+
+
+def convolutional_neural_network(img):
+    # first conv layer
+    conv_pool_1 = paddle.networks.simple_img_conv_pool(
+        input=img,
+        filter_size=5,
+        num_filters=20,
+        num_channel=1,
+        pool_size=2,
+        pool_stride=2,
+        act=paddle.activation.Relu())
+    # second conv layer
+    conv_pool_2 = paddle.networks.simple_img_conv_pool(
+        input=conv_pool_1,
+        filter_size=5,
+        num_filters=50,
+        num_channel=20,
+        pool_size=2,
+        pool_stride=2,
+        act=paddle.activation.Relu())
+    # fully-connected layer
+    predict = paddle.layer.fc(input=conv_pool_2,
+                              size=10,
+                              act=paddle.activation.Softmax())
+    return predict
+
+
+def main():
+    etcd_ip = os.getenv("ETCD_IP")
+    etcd_endpoint = "http://" + etcd_ip + ":" + "2379"
+    paddle.init()
+
+    # define network topology
+    images = paddle.layer.data(
+        name='pixel', type=paddle.data_type.dense_vector(784))
+    label = paddle.layer.data(
+        name='label', type=paddle.data_type.integer_value(10))
+
+    # Here we can build the prediction network in different ways. Please
+    # choose one by uncomment corresponding line.
+    # predict = softmax_regression(images)
+    # predict = multilayer_perceptron(images)
+    predict = convolutional_neural_network(images)
+
+    cost = paddle.layer.classification_cost(input=predict, label=label)
+
+    parameters = paddle.parameters.create(cost)
+
+    optimizer = paddle.optimizer.Momentum(
+        learning_rate=0.1 / 128.0,
+        momentum=0.9,
+        regularization=paddle.optimizer.L2Regularization(rate=0.0005 * 128))
+
+    trainer = paddle.trainer.SGD(cost=cost,
+                                 parameters=parameters,
+                                 update_equation=optimizer,
+                                 is_local=False,
+                                 pserver_spec=etcd_endpoint,
+                                 use_etcd=True)
+
+    def event_handler(event):
+        if isinstance(event, paddle.event.EndIteration):
+            if event.batch_id % 100 == 0:
+                print "Pass %d, Batch %d, Cost %f, %s" % (
+                    event.pass_id, event.batch_id, event.cost, event.metrics)
+        if isinstance(event, paddle.event.EndPass):
+            result = trainer.test(reader=paddle.batch(
+                dataset.mnist.test(),
+                batch_size=2))
+            print "Test with Pass %d, Cost %f, %s\n" % (
+                event.pass_id, result.cost, result.metrics)
+
+    trainer.train(
+        reader=paddle.batch(
+            dataset.mnist.train(),
+            batch_size=128),
+        event_handler=event_handler,
+        num_passes=5)
+
+
+if __name__ == '__main__':
+    usage = "python train.py [prepare|train]"
+    if len(sys.argv) != 2:
+        print usage
+        exit(1)
+
+    if TRAINER_ID == -1 or TOTAL_TRAINERS == -1:
+        print "no cloud environ found, must run on cloud"
+        exit(1)
+    if sys.argv[1] == "train":
+        main()

pkg/controller.go

@@ -117,20 +117,34 @@ func (c *Controller) onAdd(obj interface{}) {
 	//  create trainjob from paddlectl
 	//  scheduler can schedule trainjobs
 	var parser DefaultJobParser
+	m := parser.ParseToMaster(job)
 	p := parser.ParseToPserver(job)
 	t := parser.ParseToTrainer(job)
-	m := parser.ParseToMaster(job)
 
-	b, _ := json.MarshalIndent(p, "", "   ")
+	b, _ := json.MarshalIndent(m, "", "   ")
+	log.Debug("create master:" + string(b))
+
+	b, _ = json.MarshalIndent(p, "", "   ")
 	log.Debug("create pserver:" + string(b))
 
 	b, _ = json.MarshalIndent(t, "", "   ")
 	log.Debug("create trainer-job:" + string(b))
 
-	b, _ = json.MarshalIndent(m, "", "   ")
-	log.Debug("create master:" + string(b))
-
-	// TODO(gongwb): create them
+	// create all resources
+	_, err := c.clientset.ExtensionsV1beta1().ReplicaSets(m.ObjectMeta.Namespace).Create(m)
+	if err != nil {
+		log.Error("create master", "error", err)
+	}
+
+	_, err = c.clientset.ExtensionsV1beta1().ReplicaSets(m.ObjectMeta.Namespace).Create(p)
+	if err != nil {
+		log.Error("create pserver", "error", err)
+	}
+
+	_, err = c.clientset.BatchV1().Jobs(t.ObjectMeta.Namespace).Create(t)
+	if err != nil {
+		log.Error("create trainer", "error", err)
+	}
 }
 
 func (c *Controller) onUpdate(oldObj, newObj interface{}) {

pkg/jobparser.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"strconv"
 
+	log "github.com/inconshreveable/log15"
 	edlresource "github.com/paddlepaddle/edl/pkg/resource"
 	batchv1 "k8s.io/api/batch/v1"
 	"k8s.io/api/core/v1"
@@ -86,7 +87,10 @@ func (p *DefaultJobParser) ParseToPserver(job *edlresource.TrainingJob) *v1beta1
 			Kind:       "extensions/v1beta1",
 			APIVersion: "ReplicaSet",
 		},
-		ObjectMeta: job.ObjectMeta,
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      job.ObjectMeta.Name + "-pserver",
+			Namespace: job.ObjectMeta.Namespace,
+		},
 		Spec: v1beta1.ReplicaSetSpec{
 			Replicas: &replicas,
 			Template: v1.PodTemplateSpec{
@@ -235,14 +239,17 @@ func (p *DefaultJobParser) ParseToMaster(job *edlresource.TrainingJob) *v1beta1.
 // general functions that pserver, trainer use the same
 // -----------------------------------------------------------------------
 func podPorts(job *edlresource.TrainingJob) []v1.ContainerPort {
+	log.Debug("get pod ports", "portsnum", job.Spec.PortsNum, "sparse", job.Spec.PortsNumForSparse)
 	portsTotal := job.Spec.PortsNum + job.Spec.PortsNumForSparse
-	ports := make([]v1.ContainerPort, 8)
+	ports := make([]v1.ContainerPort, portsTotal)
 	basePort := int32(job.Spec.Port)
 	for i := 0; i < portsTotal; i++ {
-		ports = append(ports, v1.ContainerPort{
+		log.Debug("adding port ", "base", basePort,
+			" total ", portsTotal)
+		ports[i] = v1.ContainerPort{
 			Name:          fmt.Sprintf("jobport-%d", basePort),
 			ContainerPort: basePort,
-		})
+		}
 		basePort++
 	}
 	return ports