WIP

.gitignore

@@ -1,3 +1,4 @@
 tmp
 dist
 paddle
+vendor

README.md

@@ -14,6 +14,13 @@ git clone git@github.com:deliveroo/paddle.git
 cd paddle
 ```
 
+Install dependencies:
+
+```
+brew install glide
+glide up -v
+```
+
 You will need create a `$HOME/.paddle.yaml` that contains the bucket name, e.g:
 
 ```

cli/pipeline/cmd.go

@@ -0,0 +1,28 @@
+// Copyright © 2017 RooFoods LTD
+// 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 pipeline
+
+import (
+	"github.com/spf13/cobra"
+)
+
+var PipelineCmd = &cobra.Command{
+	Use:   "pipeline",
+	Short: "Manage Canoe pipelines",
+	Long:  "Run and control Canoe pipelines",
+}
+
+func init() {
+	PipelineCmd.AddCommand(runCmd)
+}

cli/pipeline/kubernetes.go

@@ -0,0 +1,33 @@
+package pipeline
+
+import (
+	"flag"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"os"
+	"path/filepath"
+)
+
+func getKubernetesConfig() (*rest.Config, error) {
+	var config *rest.Config
+	var kubeconfig *string
+	if home := homeDir(); home != "" {
+		kubeconfig = flag.String("kubeconfig", filepath.Join(home, ".kube", "config"), "(optional) absolute path to the kubeconfig file")
+	} else {
+		kubeconfig = flag.String("kubeconfig", "", "absolute path to the kubeconfig file")
+	}
+	flag.Parse()
+
+	config, err := clientcmd.BuildConfigFromFlags("", *kubeconfig)
+	if err != nil {
+		config, err = rest.InClusterConfig()
+	}
+	return config, err
+}
+
+func homeDir() string {
+	if h := os.Getenv("HOME"); h != "" {
+		return h
+	}
+	return os.Getenv("USERPROFILE") // windows
+}

cli/pipeline/pipeline_definition.go

@@ -0,0 +1,41 @@
+package pipeline
+
+import (
+	"gopkg.in/yaml.v2"
+	"log"
+)
+
+type PipelineDefinitionStep struct {
+	Step    string `yaml:"step"`
+	Version string `yaml:"version"`
+	Branch  string `yaml:"branch"`
+	Image   string `yaml:"image"`
+	Inputs  []struct {
+		Step    string `yaml:"step"`
+		Version string `yaml:"version"`
+		Branch  string `yaml:"branch"`
+		Path    string `yaml:"path"`
+	} `yaml:"inputs"`
+	Commands  []string `yaml:"commands"`
+	Resources struct {
+		CPU    int    `yaml:"cpu"`
+		Memory string `yaml:"memory"`
+	} `yaml:"resources"`
+}
+
+type PipelineDefinition struct {
+	Pipeline  string                   `yaml:"pipeline"`
+	Bucket    string                   `yaml:"bucket"`
+	Namespace string                   `yaml:"namespace"`
+	Steps     []PipelineDefinitionStep `yaml:"steps"`
+}
+
+func parsePipeline(data []byte) *PipelineDefinition {
+	pipeline := PipelineDefinition{}
+
+	err := yaml.Unmarshal(data, &pipeline)
+	if err != nil {
+		log.Fatalf("error: %v", err)
+	}
+	return &pipeline
+}

cli/pipeline/run.go

@@ -0,0 +1,68 @@
+// Copyright © 2017 RooFoods LTD
+// 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 pipeline
+
+import (
+	"fmt"
+	"github.com/spf13/cobra"
+	"io/ioutil"
+	"k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+)
+
+var stepName string
+
+var runCmd = &cobra.Command{
+	Use:   "run [pipeline_yaml] [-s step_name]",
+	Short: "Run a pipeline or a pipeline step",
+	Args:  cobra.ExactArgs(1),
+	Long: `Store data into S3 under a versioned path, and update HEAD.
+
+Example:
+
+$ paddle pipeline run test_pipeline.yaml
+`,
+	Run: func(cmd *cobra.Command, args []string) {
+		runPipeline(args[0])
+	},
+}
+
+func init() {
+	runCmd.Flags().StringVarP(&stepName, "step", "s", "", "Single step to execute")
+}
+
+func runPipeline(path string) {
+	config, err := getKubernetesConfig()
+	if err != nil {
+		panic(err.Error())
+	}
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		panic(err.Error())
+	}
+	data, err := ioutil.ReadFile(path)
+	if err != nil {
+		panic(err.Error())
+	}
+	pipeline := parsePipeline(data)
+	namespace := pipeline.Namespace
+	list, err := clientset.CoreV1().Pods(namespace).List(v1.ListOptions{})
+	if err != nil {
+		panic(err.Error())
+	}
+	fmt.Printf("{}", list)
+	for _, step := range pipeline.Steps {
+		compilePodTemplate(pipeline, &step)
+	}
+}

cli/pipeline/template.go

@@ -0,0 +1,154 @@
+package pipeline
+
+import (
+	"fmt"
+	"os"
+	"strings"
+	"text/template"
+)
+
+type PodTemplateData struct {
+	PodName    string
+	StepName   string
+	BranchName string
+	Namespace  string
+	Bucket     string
+
+	Step PipelineDefinitionStep
+}
+
+const podTemplate = `
+apiVersion: v1
+kind: Pod
+metadata:
+  name: "{{ .PodName }}"
+  namespace: {{ .Namespace }}
+  labels:
+    pipeline: canoe
+spec:
+  restartPolicy: Never
+  volumes:
+    -
+      name: shared-data
+      emptyDir:
+        medium: ''
+  containers:
+    -
+      name: main
+      image: "{{ .Step.Image }}"
+      volumeMounts:
+        -
+          name: shared-data
+          mountPath: /data
+      resources:
+        limits:
+          cpu: "{{ .Step.Resources.CPU }}"
+          memory: "{{ .Step.Resources.Memory }}"
+      command:
+        - "/bin/sh"
+        - "-c"
+        - "while true; do if [ -e /data/first-step.txt ]; then ((
+          {{ range $index, $command := .Step.Commands }}
+          ({{ $command }}) &&
+          {{ end }}
+          touch /data/main-passed.txt) || (touch /data/main-failed.txt && exit 1)) && touch /data/main.txt; break; fi; done"
+      env:
+        -
+          name: INPUT_PATH
+          value: /data/input
+        -
+          name: OUTPUT_PATH
+          value: /data/output
+        -
+          name: AWS_ACCESS_KEY_ID
+          valueFrom:
+            secretKeyRef:
+              name: aws-credentials-training
+              key: aws-access-key-id
+        -
+          name: AWS_SECRET_ACCESS_KEY
+          valueFrom:
+            secretKeyRef:
+              name: aws-credentials-training
+              key: aws-secret-access-key
+    -
+      name: paddle
+      image: "219541440308.dkr.ecr.eu-west-1.amazonaws.com/paddlecontainer:latest"
+      volumeMounts:
+        -
+          name: shared-data
+          mountPath: /data
+      command:
+        - "/bin/sh"
+        - "-c"
+        - "mkdir -p $INPUT_PATH $OUTPUT_PATH &&
+          {{ range $index, $input := .Step.Inputs }}
+          paddle data get {{ $input.Step }}/{{ $input.Version }} $INPUT_PATH -b {{ $input.Branch | sanitizeName }} -p {{ $input.Path }} &&
+          {{ end }}
+          touch /data/first-step.txt &&
+          echo first step finished &&
+          (while true; do
+            if [ -e /data/main-failed.txt ]; then
+              exit 1;
+            fi;
+            if [ -e /data/main-passed.txt ]; then
+              paddle data commit $OUTPUT_PATH {{ .StepName }}/{{ .Step.Version }} -b {{ .BranchName }};
+              exit 0;
+            fi;
+          done)"
+      env:
+        -
+          name: BUCKET
+          value: "{{ .Bucket }}"
+        -
+          name: AWS_REGION
+          value: eu-west-1
+        -
+          name: INPUT_PATH
+          value: /data/input
+        -
+          name: OUTPUT_PATH
+          value: /data/output
+        -
+          name: AWS_ACCESS_KEY_ID
+          valueFrom:
+            secretKeyRef:
+              name: aws-credentials
+              key: aws-access-key-id
+        -
+          name: AWS_SECRET_ACCESS_KEY
+          valueFrom:
+            secretKeyRef:
+              name: aws-credentials
+              key: aws-secret-access-key
+`
+
+func compilePodTemplate(pipelineDefinition *PipelineDefinition, pipelineDefinitionStep *PipelineDefinitionStep) string {
+	fmap := template.FuncMap{
+		"sanitizeName": sanitizeName,
+	}
+	stepName := sanitizeName(pipelineDefinitionStep.Step)
+	branchName := sanitizeName(pipelineDefinitionStep.Branch)
+	podName := fmt.Sprintf("%s-%s-%s", sanitizeName(pipelineDefinition.Pipeline), stepName, branchName)
+	templateData := PodTemplateData{
+		PodName:    podName,
+		Namespace:  pipelineDefinition.Namespace,
+		Step:       *pipelineDefinitionStep,
+		Bucket:     pipelineDefinition.Bucket,
+		StepName:   stepName,
+		BranchName: branchName,
+	}
+	tmpl := template.Must(template.New("podTemplate").Funcs(fmap).Parse(podTemplate))
+	err := tmpl.Execute(os.Stdout, templateData)
+	if err != nil {
+		panic(err.Error())
+	}
+	return ""
+}
+
+func sanitizeName(name string) string {
+	str := strings.ToLower(name)
+	str = strings.Replace(str, "_", "-", -1)
+	str = strings.Replace(str, "/", "-", -1)
+	return str
+}

cli/root.go

@@ -18,6 +18,7 @@ import (
 	"os"
 
 	"github.com/deliveroo/paddle/cli/data"
+	"github.com/deliveroo/paddle/cli/pipeline"
 	homedir "github.com/mitchellh/go-homedir"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
@@ -54,7 +55,7 @@ func init() {
 	// RootCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
 
 	RootCmd.AddCommand(data.DataCmd)
-
+	RootCmd.AddCommand(pipeline.PipelineCmd)
 }
 
 // initConfig reads in config file and ENV variables if set.