.github/issue_label_bot.yaml

@@ -0,0 +1,5 @@
+# for https://mlbot.net a Github bot that labels issues using KubeFlow
+label-alias:
+  bug: 'kind/bug'
+  feature_request: 'improvement/enhancement'
+  question: 'community/question'

.travis.yml

@@ -15,6 +15,7 @@ install:
 script:
   - go build -o tf-operator.v1beta1 github.com/kubeflow/tf-operator/cmd/tf-operator.v1beta1
   - go build -o tf-operator.v1beta2 github.com/kubeflow/tf-operator/cmd/tf-operator.v1beta2
+  - go build -o tf-operator.v1 github.com/kubeflow/tf-operator/cmd/tf-operator.v1
   - go build -o backend github.com/kubeflow/tf-operator/dashboard/backend
   - gometalinter --config=linter_config.json --vendor ./...
   # We customize the build step because by default

OWNERS

@@ -1,17 +1,10 @@
 approvers:
-  - DjangoPeng
   - gaocegege
-  - jlewi
   - johnugeorge
   - richardsliu
   - ScorpioCPH
-  - wbuchwalter
 reviewers:
   - ChanYiLin
   - cheyang
   - codeflitting
-  - ddysher
   - Jimexist
-  - lluunn
-  - willb
-  - yph152

build/images/tf_operator/Dockerfile

@@ -1,10 +1,13 @@
 FROM registry.access.redhat.com/ubi8-dev-preview/ubi:latest
 
 ARG GOLANG_VERSION=1.8.3
-ARG GOLANG_ARCH=amd64
 ENV GOROOT=/usr/local/go PATH=$PATH:/usr/local/go/bin
 
-RUN curl -fksSL https://dl.google.com/go/go$GOLANG_VERSION.linux-$GOLANG_ARCH.tar.gz | tar -xz -C /usr/local/
+RUN if [ "$(uname -m)" = "ppc64le" ]; then \
+        curl -fksSL https://dl.google.com/go/go$GOLANG_VERSION.linux-ppc64le.tar.gz | tar -xz -C /usr/local/; \
+    else \
+        curl -fksSL https://dl.google.com/go/go$GOLANG_VERSION.linux-amd64.tar.gz | tar -xz -C /usr/local/; \
+    fi
 
 # TODO(jlewi): We should probably change the directory to /opt/kubeflow.
 RUN mkdir -p /opt/tensorflow_k8s/dashboard/
@@ -14,11 +17,13 @@ COPY tf_smoke.py /opt/kubeflow/samples/
 RUN chmod a+x /opt/kubeflow/samples/*
 COPY tf-operator.v1beta1 /opt/kubeflow
 COPY tf-operator.v1beta2 /opt/kubeflow
+COPY tf-operator.v1 /opt/kubeflow
 COPY backend /opt/tensorflow_k8s/dashboard/
 COPY build /opt/tensorflow_k8s/dashboard/frontend/build
 
 RUN chmod a+x /opt/kubeflow/tf-operator.v1beta1
 RUN chmod a+x /opt/kubeflow/tf-operator.v1beta2
+RUN chmod a+x /opt/kubeflow/tf-operator.v1
 RUN chmod a+x /opt/tensorflow_k8s/dashboard/backend
 
-ENTRYPOINT ["/opt/kubeflow/tf-operator.v1beta1"]
+ENTRYPOINT ["/opt/kubeflow/tf-operator.v1"]

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

@@ -0,0 +1,58 @@
+// 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"
+
+	"k8s.io/api/core/v1"
+)
+
+// ServerOption is the main context object for the controller manager.
+type ServerOption struct {
+	Kubeconfig           string
+	MasterURL            string
+	Threadiness          int
+	PrintVersion         bool
+	JSONLogFormat        bool
+	EnableGangScheduling bool
+	Namespace            string
+}
+
+// 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.MasterURL, "master", "",
+		`The url of the Kubernetes API server,
+		 will overrides any value in kubeconfig, only required if out-of-cluster.`)
+
+	fs.StringVar(&s.Namespace, "namespace", v1.NamespaceAll,
+		`The namespace to monitor tfjobs. If unset, it monitors all namespaces cluster-wide. 
+                If set, it only monitors tfjobs in the given namespace.`)
+
+	fs.IntVar(&s.Threadiness, "threadiness", 1,
+		`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")
+	fs.BoolVar(&s.EnableGangScheduling, "enable-gang-scheduling", false, "Set true to enable gang scheduling by kube-batch.")
+}

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

@@ -0,0 +1,204 @@
+// 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"
+
+	"github.com/kubeflow/tf-operator/cmd/tf-operator.v1/app/options"
+	"github.com/kubeflow/tf-operator/pkg/apis/tensorflow/v1"
+	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.v1/tensorflow"
+	"github.com/kubeflow/tf-operator/pkg/util/signals"
+	"github.com/kubeflow/tf-operator/pkg/version"
+	kubebatchclient "github.com/kubernetes-sigs/kube-batch/pkg/client/clientset/versioned"
+	log "github.com/sirupsen/logrus"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	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"
+)
+
+const (
+	apiVersion = "v1"
+)
+
+var (
+	// leader election config
+	leaseDuration = 15 * time.Second
+	renewDuration = 5 * time.Second
+	retryPeriod   = 3 * time.Second
+	resyncPeriod  = 30 * 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(apiVersion)
+	}
+
+	namespace := os.Getenv(v1.EnvKubeflowNamespace)
+	if len(namespace) == 0 {
+		log.Infof("EnvKubeflowNamespace not set, use default namespace")
+		namespace = metav1.NamespaceDefault
+	}
+	if opt.Namespace == corev1.NamespaceAll {
+		log.Info("Using cluster scoped operator")
+	} else {
+		log.Infof("Scoping operator to namespace %s", opt.Namespace)
+	}
+
+	// To help debugging, immediately log version.
+	log.Infof("%+v", version.Info(apiVersion))
+
+	// 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, kubeBatchClientSet, err := createClientSets(kcfg)
+	if err != nil {
+		return err
+	}
+	if !checkCRDExists(tfJobClientSet, opt.Namespace) {
+		log.Info("CRD doesn't exist. Exiting")
+		os.Exit(1)
+	}
+	// Create informer factory.
+	kubeInformerFactory := kubeinformers.NewFilteredSharedInformerFactory(kubeClientSet, resyncPeriod, opt.Namespace, nil)
+	tfJobInformerFactory := tfjobinformers.NewSharedInformerFactory(tfJobClientSet, resyncPeriod)
+
+	unstructuredInformer := controller.NewUnstructuredTFJobInformer(kcfg, opt.Namespace)
+
+	// Create tf controller.
+	tc := controller.NewTFController(unstructuredInformer, kubeClientSet, kubeBatchClientSet, tfJobClientSet, kubeInformerFactory, tfJobInformerFactory, *opt)
+
+	// Start informer goroutines.
+	go kubeInformerFactory.Start(stopCh)
+
+	// We do not use the generated informer because of
+	// https://github.com/kubeflow/tf-operator/issues/561
+	// go tfJobInformerFactory.Start(stopCh)
+	go unstructuredInformer.Informer().Run(stopCh)
+
+	// Set leader election start function.
+	run := func(<-chan struct{}) {
+		if err := tc.Run(opt.Threadiness, stopCh); err != nil {
+			log.Errorf("Failed to run the controller: %v", err)
+		}
+	}
+
+	id, err := os.Hostname()
+	if err != nil {
+		return fmt.Errorf("failed to get hostname: %v", err)
+	}
+
+	// Prepare event clients.
+	eventBroadcaster := record.NewBroadcaster()
+	if err = v1.AddToScheme(scheme.Scheme); err != nil {
+		return fmt.Errorf("CoreV1 Add Scheme failed: %v", err)
+	}
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.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, kubebatchclient.Interface, error) {
+
+	kubeClientSet, err := kubeclientset.NewForConfig(restclientset.AddUserAgent(config, "tf-operator"))
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	leaderElectionClientSet, err := kubeclientset.NewForConfig(restclientset.AddUserAgent(config, "leader-election"))
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	tfJobClientSet, err := tfjobclientset.NewForConfig(config)
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+
+	kubeBatchClientSet, err := kubebatchclient.NewForConfig(restclientset.AddUserAgent(config, "kube-batch"))
+	if err != nil {
+		return nil, nil, nil, nil, err
+	}
+	return kubeClientSet, leaderElectionClientSet, tfJobClientSet, kubeBatchClientSet, nil
+}
+
+func checkCRDExists(clientset tfjobclientset.Interface, namespace string) bool {
+	_, err := clientset.KubeflowV1().TFJobs(namespace).List(metav1.ListOptions{})
+
+	if err != nil {
+		log.Error(err)
+		if _, ok := err.(*errors.StatusError); ok {
+			if errors.IsNotFound(err) {
+				return false
+			}
+		}
+	}
+	return true
+}

cmd/tf-operator.v1/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.v1/app"
+	"github.com/kubeflow/tf-operator/cmd/tf-operator.v1/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/distribution_strategy/distributed_tfjob.yaml

@@ -7,7 +7,7 @@ spec:
   cleanPodPolicy: None
   tfReplicaSpecs:
     Worker:
-      replicas: 3 
+      replicas: 3
       restartPolicy: Never
       template:
         metadata:
@@ -16,4 +16,4 @@ spec:
         spec:
           containers:
             - name: tensorflow
-              image: gcr.io/kubeflow-examples/distributed_worker:v20181031-513e107c 
+              image: gcr.io/kubeflow-examples/distributed_worker:v20181031-513e107c