v1alpha2: Add implementation

.travis.yml

@@ -21,6 +21,7 @@ install:
 
 script:
   - go build -o tf-operator github.com/kubeflow/tf-operator/cmd/tf-operator
+  - go build -o tf-operator github.com/kubeflow/tf-operator/cmd/tf-operator.v2
   - gometalinter --config=linter_config.json ./pkg/...
   # We customize the build step because by default
   # Travis runs go test -v ./... which will include the vendor

cmd/tf-operator.v2/app/options/options.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 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.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", false,
+		"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,164 @@
+// 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"
+)
+
+const (
+	apiVersion = "v1alpha2"
+)
+
+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(apiVersion)
+	}
+
+	namespace := os.Getenv(v1alpha2.EnvKubeflowNamespace)
+	if len(namespace) == 0 {
+		log.Infof("EnvKubeflowNamespace not set, use default namespace")
+		namespace = metav1.NamespaceDefault
+	}
+
+	// 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, 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)
+	}
+
+}

cmd/tf-operator/app/server.go

@@ -41,6 +41,10 @@ import (
 	"github.com/kubeflow/tf-operator/version"
 )
 
+const (
+	apiVersion = "v1alpha1"
+)
+
 var (
 	leaseDuration = 15 * time.Second
 	renewDuration = 5 * time.Second
@@ -52,7 +56,7 @@ 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()
+		version.PrintVersionAndExit(apiVersion)
 	}
 
 	namespace := os.Getenv(util.EnvKubeflowNamespace)
@@ -62,7 +66,7 @@ func Run(opt *options.ServerOption) error {
 	}
 
 	// To help debugging, immediately log version
-	log.Infof("%+v", version.Info())
+	log.Infof("%+v", version.Info(apiVersion))
 
 	config, err := k8sutil.GetClusterConfig()
 	if err != nil {

linter_config.json

@@ -30,7 +30,7 @@
     "pkg/apis/tensorflow/validation/validation_test.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"
   ],
   "Deadline": "300s",
   "Skip": ["pkg/client"]