v1alpha2: Implementation

.travis.yml

@@ -29,4 +29,4 @@ script:
   # For now though we just run all tests in pkg.
   # And we can not use ** because goveralls uses filepath.Match
   # to match ignore files and it does not support it.
-  - goveralls -service=travis-ci -v -package ./pkg/... -ignore "pkg/client/*/*.go,pkg/client/*/*/*.go,pkg/client/*/*/*/*.go,pkg/client/*/*/*/*/*.go,pkg/client/*/*/*/*/*/*.go,pkg/client/*/*/*/*/*/*/*.go,pkg/apis/tensorflow/*/zz_generated.*.go"
+  - goveralls -service=travis-ci -v -package ./pkg/... -ignore "pkg/client/*/*.go,pkg/client/*/*/*.go,pkg/client/*/*/*/*.go,pkg/client/*/*/*/*/*.go,pkg/client/*/*/*/*/*/*.go,pkg/client/*/*/*/*/*/*/*.go,pkg/apis/tensorflow/*/zz_generated.*.go,pkg/controller/controller_utils.go,pkg/controller/controller_ref_manager.go"

cmd/tf-operator.v2/app/options/options.go

@@ -0,0 +1,52 @@
+// Copyright 2018 The Kubeflow 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 options
+
+import (
+	"flag"
+)
+
+// ServerOption is the main context object for the controller manager.
+type ServerOption struct {
+	Kubeconfig    string
+	MasterURL     string
+	Threadiness   int
+	PrintVersion  bool
+	JSONLogFormat bool
+}
+
+// NewServerOption creates a new CMServer with a default config.
+func NewServerOption() *ServerOption {
+	s := ServerOption{}
+	return &s
+}
+
+// AddFlags adds flags for a specific CMServer to the specified FlagSet.
+func (s *ServerOption) AddFlags(fs *flag.FlagSet) {
+	fs.StringVar(&s.Kubeconfig, "kubeconfig", "~/.kube/config",
+		`Path to a kubeconfig, only required if out-of-cluster.`)
+
+	fs.StringVar(&s.MasterURL, "master", "",
+		`The url of the Kubernetes API server,
+		 will overrides any value in kubeconfig, only required if out-of-cluster.`)
+
+	fs.IntVar(&s.Threadiness, "threadiness", 2,
+		`How many threads to process the main logic`)
+
+	fs.BoolVar(&s.PrintVersion, "version", false, "Show version and quit")
+
+	fs.BoolVar(&s.JSONLogFormat, "json-log-format", true,
+		"Set true to use json style log format. Set false to use plaintext style log format")
+}

cmd/tf-operator.v2/app/server.go

@@ -0,0 +1,162 @@
+// Copyright 2018 The Kubeflow 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 app
+
+import (
+	"fmt"
+	"os"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	kubeinformers "k8s.io/client-go/informers"
+	kubeclientset "k8s.io/client-go/kubernetes"
+	restclientset "k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	election "k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/client-go/tools/record"
+
+	"github.com/kubeflow/tf-operator/cmd/tf-operator.v2/app/options"
+	"github.com/kubeflow/tf-operator/pkg/apis/tensorflow/v1alpha2"
+	tfjobclientset "github.com/kubeflow/tf-operator/pkg/client/clientset/versioned"
+	"github.com/kubeflow/tf-operator/pkg/client/clientset/versioned/scheme"
+	tfjobinformers "github.com/kubeflow/tf-operator/pkg/client/informers/externalversions"
+	controller "github.com/kubeflow/tf-operator/pkg/controller.v2"
+	"github.com/kubeflow/tf-operator/pkg/util/signals"
+	"github.com/kubeflow/tf-operator/version"
+)
+
+var (
+	// leader election config
+	leaseDuration = 15 * time.Second
+	renewDuration = 5 * time.Second
+	retryPeriod   = 3 * time.Second
+)
+
+const RecommendedKubeConfigPathEnv = "KUBECONFIG"
+
+func Run(opt *options.ServerOption) error {
+
+	// Check if the -version flag was passed and, if so, print the version and exit.
+	if opt.PrintVersion {
+		version.PrintVersionAndExit()
+	}
+
+	namespace := os.Getenv(v1alpha2.EnvKubeflowNamespace)
+	if len(namespace) == 0 {
+		log.Infof("KUBEFLOW_NAMESPACE not set, using default namespace")
+		namespace = metav1.NamespaceDefault
+	}
+
+	// To help debugging, immediately log version.
+	log.Infof("%+v", version.Info())
+
+	// Set up signals so we handle the first shutdown signal gracefully.
+	stopCh := signals.SetupSignalHandler()
+
+	// Note: ENV KUBECONFIG will overwrite user defined Kubeconfig option.
+	if len(os.Getenv(RecommendedKubeConfigPathEnv)) > 0 {
+		// use the current context in kubeconfig
+		// This is very useful for running locally.
+		opt.Kubeconfig = os.Getenv(RecommendedKubeConfigPathEnv)
+	}
+
+	// Get kubernetes config.
+	kcfg, err := clientcmd.BuildConfigFromFlags(opt.MasterURL, opt.Kubeconfig)
+	if err != nil {
+		log.Fatalf("Error building kubeconfig: %s", err.Error())
+	}
+
+	// Create clients.
+	kubeClientSet, leaderElectionClientSet, tfJobClientSet, err := createClientSets(kcfg)
+	if err != nil {
+		return err
+	}
+
+	// Create informer factory.
+	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClientSet, time.Second*30)
+	tfJobInformerFactory := tfjobinformers.NewSharedInformerFactory(tfJobClientSet, time.Second*30)
+
+	// Create tf controller.
+	tc := controller.NewTFJobController(kubeClientSet, tfJobClientSet, kubeInformerFactory, tfJobInformerFactory)
+
+	// Start informer goroutines.
+	go kubeInformerFactory.Start(stopCh)
+	go tfJobInformerFactory.Start(stopCh)
+
+	// Set leader election start function.
+	run := func(<-chan struct{}) {
+		tc.Run(opt.Threadiness, stopCh)
+	}
+
+	id, err := os.Hostname()
+	if err != nil {
+		return fmt.Errorf("Failed to get hostname: %v", err)
+	}
+
+	// Prepare event clients.
+	eventBroadcaster := record.NewBroadcaster()
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "tf-operator"})
+
+	rl := &resourcelock.EndpointsLock{
+		EndpointsMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      "tf-operator",
+		},
+		Client: leaderElectionClientSet.CoreV1(),
+		LockConfig: resourcelock.ResourceLockConfig{
+			Identity:      id,
+			EventRecorder: recorder,
+		},
+	}
+
+	// Start leader election.
+	election.RunOrDie(election.LeaderElectionConfig{
+		Lock:          rl,
+		LeaseDuration: leaseDuration,
+		RenewDeadline: renewDuration,
+		RetryPeriod:   retryPeriod,
+		Callbacks: election.LeaderCallbacks{
+			OnStartedLeading: run,
+			OnStoppedLeading: func() {
+				log.Fatalf("leader election lost")
+			},
+		},
+	})
+
+	return nil
+}
+
+func createClientSets(config *restclientset.Config) (kubeclientset.Interface, kubeclientset.Interface, tfjobclientset.Interface, error) {
+	kubeClientSet, err := kubeclientset.NewForConfig(restclientset.AddUserAgent(config, "tf-operator"))
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	leaderElectionClientSet, err := kubeclientset.NewForConfig(restclientset.AddUserAgent(config, "leader-election"))
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	tfJobClientSet, err := tfjobclientset.NewForConfig(config)
+	if err != nil {
+		return nil, nil, nil, err
+	}
+
+	return kubeClientSet, leaderElectionClientSet, tfJobClientSet, nil
+}

cmd/tf-operator.v2/main.go

@@ -0,0 +1,49 @@
+// Copyright 2018 The Kubeflow 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 main
+
+import (
+	"flag"
+
+	"github.com/onrik/logrus/filename"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/kubeflow/tf-operator/cmd/tf-operator.v2/app"
+	"github.com/kubeflow/tf-operator/cmd/tf-operator.v2/app/options"
+)
+
+func init() {
+	// Add filename as one of the fields of the structured log message.
+	filenameHook := filename.NewHook()
+	filenameHook.Field = "filename"
+	log.AddHook(filenameHook)
+}
+
+func main() {
+	s := options.NewServerOption()
+	s.AddFlags(flag.CommandLine)
+
+	flag.Parse()
+
+	if s.JSONLogFormat {
+		// Output logs in a json format so that it can be parsed by services like Stackdriver.
+		log.SetFormatter(&log.JSONFormatter{})
+	}
+
+	if err := app.Run(s); err != nil {
+		log.Fatalf("%v\n", err)
+	}
+
+}

examples/crd/v1alpha2.yaml

@@ -0,0 +1,11 @@
+apiVersion: apiextensions.k8s.io/v1beta1
+kind: CustomResourceDefinition
+metadata:
+  name: tfjobs.kubeflow.org
+spec:
+  group: kubeflow.org
+  version: v1alpha2
+  names:
+    kind: TFJob
+    singular: tfjob
+    plural: tfjobs

examples/simple_tf_job.yaml

@@ -0,0 +1,13 @@
+apiVersion: "kubeflow.org/v1alpha2"
+kind: "TFJob"
+metadata:
+  name: "simple-job"
+spec:
+  tfReplicaSpecs:
+    Worker:
+      template:
+        spec:
+          containers:
+            - name: worker-busybox
+              image: busybox
+              command: ["sleep", "30000"]

examples/tf_job_v1alpha2.yaml

@@ -0,0 +1,24 @@
+apiVersion: "kubeflow.org/v1alpha2"
+kind: "TFJob"
+metadata:
+  name: "example-job-1"
+spec:
+  tfReplicaSpecs:
+    PS:
+      replicas: 2
+      template:
+        spec:
+          containers:
+            - name: ps-busybox
+              image: busybox
+              command: ["sleep", "30000"]
+          restartPolicy: OnFailure
+    Worker:
+      replicas: 4
+      template:
+        spec:
+          containers:
+            - name: worker-busybox
+              image: busybox
+              command: ["sleep", "30000"]
+          restartPolicy: OnFailure

hack/update-codegen.sh

@@ -28,7 +28,15 @@ CODEGEN_PKG=${CODEGEN_PKG:-$(cd ${SCRIPT_ROOT}; ls -d -1 ./vendor/k8s.io/code-ge
 # --output-base    because this script should also be able to run inside the vendor dir of
 #                  k8s.io/kubernetes. The output-base is needed for the generators to output into the vendor dir
 #                  instead of the $GOPATH directly. For normal projects this can be dropped.
-${CODEGEN_PKG}/generate-groups.sh "defaulter,deepcopy,client,informer,lister" \
+${CODEGEN_PKG}/generate-groups.sh "all" \
  github.com/kubeflow/tf-operator/pkg/client github.com/kubeflow/tf-operator/pkg/apis \
- tensorflow:v1alpha1 \
+ tensorflow:v1alpha1,v1alpha2 \
  --go-header-file ${SCRIPT_ROOT}/hack/boilerplate/boilerplate.go.txt
+
+# Notice: The code in code-generator does not generate defaulter by default.
+echo "Generating defaulters"
+${GOPATH}/bin/defaulter-gen  --input-dirs github.com/kubeflow/tf-operator/pkg/apis/tensorflow/v1alpha1 -O zz_generated.defaults "$@"
+
+# Notice: The code in code-generator does not generate defaulter by default.
+echo "Generating defaulters"
+${GOPATH}/bin/defaulter-gen  --input-dirs github.com/kubeflow/tf-operator/pkg/apis/tensorflow/v1alpha2 -O zz_generated.defaults "$@"

linter_config.json

@@ -25,6 +25,10 @@
     "comment on exported",
     "pkg/apis/tensorflow/v1alpha1/zz_generated.deepcopy.go",
     "pkg/apis/tensorflow/v1alpha1/zz_generated.defaults.go",
+    "pkg/apis/tensorflow/v1alpha2/zz_generated.deepcopy.go",
+    "pkg/apis/tensorflow/v1alpha2/zz_generated.defaults.go",
+    "pkg/controller/controller_utils.go",
+    "pkg/controller.v2/controller_utils.go",
     "pkg/apis/tensorflow/v1alpha1/defaults.go",
     "pkg/apis/tensorflow/v1alpha1/defaults_test.go",
     "pkg/apis/tensorflow/validation/validation_test.go"

pkg/apis/tensorflow/v1alpha2/constants.go

@@ -0,0 +1,23 @@
+// Copyright 2018 The Kubeflow 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 v1alpha2
+
+const (
+	// EnvKubeflowNamespace is ENV for kubeflow namespace specified by user.
+	EnvKubeflowNamespace = "KUBEFLOW_NAMESPACE"
+
+	defaultPortName = "tfjob-port"
+	defaultPort     = 2222
+)