apiservers: add synchronous shutdown mechanism on SIGTERM+INT

staging/src/k8s.io/apiextensions-apiserver/BUILD

@@ -15,8 +15,9 @@ go_library(
     name = "go_default_library",
     srcs = ["main.go"],
     deps = [
+        "//vendor/github.com/golang/glog:go_default_library",
         "//vendor/k8s.io/apiextensions-apiserver/pkg/cmd/server:go_default_library",
-        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/server:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/logs:go_default_library",
     ],
 )

staging/src/k8s.io/apiextensions-apiserver/main.go

@@ -21,8 +21,10 @@ import (
 	"os"
 	"runtime"
 
+	"github.com/golang/glog"
+
 	"k8s.io/apiextensions-apiserver/pkg/cmd/server"
-	"k8s.io/apimachinery/pkg/util/wait"
+	genericapiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/util/logs"
 )
 
@@ -34,9 +36,10 @@ func main() {
 		runtime.GOMAXPROCS(runtime.NumCPU())
 	}
 
-	cmd := server.NewCommandStartCustomResourceDefinitionsServer(os.Stdout, os.Stderr, wait.NeverStop)
+	stopCh := genericapiserver.SetupSignalHandler()
+	cmd := server.NewCommandStartCustomResourceDefinitionsServer(os.Stdout, os.Stderr, stopCh)
 	cmd.Flags().AddGoFlagSet(flag.CommandLine)
 	if err := cmd.Execute(); err != nil {
-		panic(err)
+		glog.Fatal(err)
 	}
 }

staging/src/k8s.io/apiserver/pkg/audit/types.go

@@ -34,4 +34,8 @@ type Backend interface {
 	// Run will initialize the backend. It must not block, but may run go routines in the background. If
 	// stopCh is closed, it is supposed to stop them. Run will be called before the first call to ProcessEvents.
 	Run(stopCh <-chan struct{}) error
+
+	// Shutdown will synchronously shut down the backend while making sure that all pending
+	// events are delivered.
+	Shutdown()
 }

staging/src/k8s.io/apiserver/pkg/audit/union.go

@@ -49,3 +49,9 @@ func (u union) Run(stopCh <-chan struct{}) error {
 	}
 	return errors.AggregateGoroutines(funcs...)
 }
+
+func (u union) Shutdown() {
+	for _, backend := range u.backends {
+		backend.Shutdown()
+	}
+}

staging/src/k8s.io/apiserver/pkg/audit/union_test.go

@@ -36,6 +36,10 @@ func (f *fakeBackend) Run(stopCh <-chan struct{}) error {
 	return nil
 }
 
+func (u *fakeBackend) Shutdown() {
+	// nothing to do here
+}
+
 func TestUnion(t *testing.T) {
 	backends := []Backend{
 		new(fakeBackend),

staging/src/k8s.io/apiserver/pkg/server/BUILD

@@ -52,7 +52,14 @@ go_library(
         "hooks.go",
         "plugins.go",
         "serve.go",
-    ],
+        "signal.go",
+        "signal_posix.go",
+    ] + select({
+        "@io_bazel_rules_go//go/platform:windows_amd64": [
+            "signal_windows.go",
+        ],
+        "//conditions:default": [],
+    }),
     deps = [
         "//vendor/github.com/coreos/go-systemd/daemon:go_default_library",
         "//vendor/github.com/emicklei/go-restful:go_default_library",

staging/src/k8s.io/apiserver/pkg/server/genericapiserver.go

@@ -246,6 +246,11 @@ func (s preparedGenericAPIServer) Run(stopCh <-chan struct{}) error {
 	}
 
 	<-stopCh
+
+	if s.GenericAPIServer.AuditBackend != nil {
+		s.GenericAPIServer.AuditBackend.Shutdown()
+	}
+
 	return nil
 }
 

staging/src/k8s.io/apiserver/pkg/server/signal.go

@@ -0,0 +1,43 @@
+/*
+Copyright 2017 The Kubernetes 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 server
+
+import (
+	"os"
+	"os/signal"
+)
+
+var onlyOneSignalHandler = make(chan struct{})
+
+// SetupSignalHandler registered for SIGTERM and SIGINT. A stop channel is returned
+// which is closed on one of these signals. If a second signal is caught, the program
+// is terminated with exit code 1.
+func SetupSignalHandler() (stopCh <-chan struct{}) {
+	close(onlyOneSignalHandler) // panics when called twice
+
+	stop := make(chan struct{})
+	c := make(chan os.Signal, 2)
+	signal.Notify(c, shutdownSignals...)
+	go func() {
+		<-c
+		close(stop)
+		<-c
+		os.Exit(1) // second signal. Exit directly.
+	}()
+
+	return stop
+}

staging/src/k8s.io/apiserver/pkg/server/signal_posix.go

@@ -0,0 +1,26 @@
+// +build !windows
+
+/*
+Copyright 2017 The Kubernetes 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 server
+
+import (
+	"os"
+	"syscall"
+)
+
+var shutdownSignals = []os.Signal{os.Interrupt, syscall.SIGTERM}

staging/src/k8s.io/apiserver/pkg/server/signal_windows.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2017 The Kubernetes 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 server
+
+import (
+	"os"
+)
+
+var shutdownSignals = []os.Signal{os.Interrupt}

staging/src/k8s.io/apiserver/plugin/pkg/audit/log/backend.go

@@ -85,3 +85,7 @@ func (b *backend) logEvent(ev *auditinternal.Event) {
 func (b *backend) Run(stopCh <-chan struct{}) error {
 	return nil
 }
+
+func (u *backend) Shutdown() {
+	// nothing to do here
+}

staging/src/k8s.io/apiserver/plugin/pkg/audit/webhook/webhook.go

@@ -117,6 +117,10 @@ func (b *blockingBackend) Run(stopCh <-chan struct{}) error {
 	return nil
 }
 
+func (b *blockingBackend) Shutdown() {
+	// nothing to do here
+}
+
 func (b *blockingBackend) ProcessEvents(ev ...*auditinternal.Event) {
 	if err := b.processEvents(ev...); err != nil {
 		audit.HandlePluginError(pluginName, err, ev...)
@@ -203,6 +207,10 @@ func (b *batchBackend) Run(stopCh <-chan struct{}) error {
 	return nil
 }
 
+func (b *batchBackend) Shutdown() {
+	// TODO: send out batched events
+}
+
 // sendBatchEvents attempts to batch some number of events to the backend. It POSTs events
 // in a goroutine and logging any error encountered during the POST.
 //

staging/src/k8s.io/kube-aggregator/BUILD

@@ -23,7 +23,8 @@ go_library(
     name = "go_default_library",
     srcs = ["main.go"],
     deps = [
-        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/server:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/logs:go_default_library",
         "//vendor/k8s.io/kube-aggregator/pkg/apis/apiregistration/install:go_default_library",
         "//vendor/k8s.io/kube-aggregator/pkg/apis/apiregistration/validation:go_default_library",

staging/src/k8s.io/kube-aggregator/main.go

@@ -21,7 +21,9 @@ import (
 	"os"
 	"runtime"
 
-	"k8s.io/apimachinery/pkg/util/wait"
+	"github.com/golang/glog"
+
+	genericapiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/util/logs"
 	"k8s.io/kube-aggregator/pkg/cmd/server"
 
@@ -41,9 +43,10 @@ func main() {
 		runtime.GOMAXPROCS(runtime.NumCPU())
 	}
 
-	cmd := server.NewCommandStartAggregator(os.Stdout, os.Stderr, wait.NeverStop)
+	stopCh := genericapiserver.SetupSignalHandler()
+	cmd := server.NewCommandStartAggregator(os.Stdout, os.Stderr, stopCh)
 	cmd.Flags().AddGoFlagSet(flag.CommandLine)
 	if err := cmd.Execute(); err != nil {
-		panic(err)
+		glog.Fatal(err)
 	}
 }

staging/src/k8s.io/kube-gen/cmd/go-to-protobuf/protobuf/tags.go

@@ -27,7 +27,7 @@ import (
 func extractBoolTagOrDie(key string, lines []string) bool {
 	val, err := types.ExtractSingleBoolCommentTag("+", key, false, lines)
 	if err != nil {
-		glog.Fatalf(err.Error())
+		glog.Fatal(err)
 	}
 	return val
 }

staging/src/k8s.io/kube-gen/cmd/informer-gen/generators/tags.go

@@ -27,7 +27,7 @@ import (
 func extractBoolTagOrDie(key string, lines []string) bool {
 	val, err := types.ExtractSingleBoolCommentTag("+", key, false, lines)
 	if err != nil {
-		glog.Fatalf(err.Error())
+		glog.Fatal(err)
 	}
 	return val
 }

staging/src/k8s.io/kube-gen/cmd/lister-gen/generators/tags.go

@@ -27,7 +27,7 @@ import (
 func extractBoolTagOrDie(key string, lines []string) bool {
 	val, err := types.ExtractSingleBoolCommentTag("+", key, false, lines)
 	if err != nil {
-		glog.Fatalf(err.Error())
+		glog.Fatal(err)
 	}
 	return val
 }

staging/src/k8s.io/sample-apiserver/BUILD

@@ -15,7 +15,8 @@ go_library(
     name = "go_default_library",
     srcs = ["main.go"],
     deps = [
-        "//vendor/k8s.io/apimachinery/pkg/util/wait:go_default_library",
+        "//vendor/github.com/golang/glog:go_default_library",
+        "//vendor/k8s.io/apiserver/pkg/server:go_default_library",
         "//vendor/k8s.io/apiserver/pkg/util/logs:go_default_library",
         "//vendor/k8s.io/sample-apiserver/pkg/cmd/server:go_default_library",
     ],

staging/src/k8s.io/sample-apiserver/main.go

@@ -21,7 +21,9 @@ import (
 	"os"
 	"runtime"
 
-	"k8s.io/apimachinery/pkg/util/wait"
+	"github.com/golang/glog"
+
+	genericapiserver "k8s.io/apiserver/pkg/server"
 	"k8s.io/apiserver/pkg/util/logs"
 	"k8s.io/sample-apiserver/pkg/cmd/server"
 )
@@ -34,9 +36,10 @@ func main() {
 		runtime.GOMAXPROCS(runtime.NumCPU())
 	}
 
-	cmd := server.NewCommandStartWardleServer(os.Stdout, os.Stderr, wait.NeverStop)
+	stopCh := genericapiserver.SetupSignalHandler()
+	cmd := server.NewCommandStartWardleServer(os.Stdout, os.Stderr, stopCh)
 	cmd.Flags().AddGoFlagSet(flag.CommandLine)
 	if err := cmd.Execute(); err != nil {
-		panic(err)
+		glog.Fatal(err)
 	}
 }