Merge pull request #20253 from mikedanese/flock

cmd/kubelet/app/options/options.go

@@ -105,6 +105,7 @@ func NewKubeletServer() *KubeletServer {
 			NodeStatusUpdateFrequency:   unversioned.Duration{10 * time.Second},
 			NodeLabels:                  make(map[string]string),
 			OOMScoreAdj:                 qos.KubeletOOMScoreAdj,
+			LockFilePath:                "/var/run/lock/kubelet.lock",
 			PodInfraContainerImage:      kubetypes.PodInfraContainerImage,
 			Port:                           ports.KubeletPort,
 			ReadOnlyPort:                   ports.KubeletReadOnlyPort,
@@ -194,6 +195,7 @@ func (s *KubeletServer) AddFlags(fs *pflag.FlagSet) {
 	fs.StringVar(&s.ResourceContainer, "resource-container", s.ResourceContainer, "Absolute name of the resource-only container to create and run the Kubelet in (Default: /kubelet).")
 	fs.StringVar(&s.CgroupRoot, "cgroup-root", s.CgroupRoot, "Optional root cgroup to use for pods. This is handled by the container runtime on a best effort basis. Default: '', which means use the container runtime default.")
 	fs.StringVar(&s.ContainerRuntime, "container-runtime", s.ContainerRuntime, "The container runtime to use. Possible values: 'docker', 'rkt'. Default: 'docker'.")
+	fs.StringVar(&s.LockFilePath, "lock-file", s.LockFilePath, "<Warning: Alpha feature> The path to file for kubelet to use as a lock file.")
 	fs.StringVar(&s.RktPath, "rkt-path", s.RktPath, "Path of rkt binary. Leave empty to use the first rkt in $PATH.  Only used if --container-runtime='rkt'")
 	fs.StringVar(&s.RktStage1Image, "rkt-stage1-image", s.RktStage1Image, "image to use as stage1. Local paths and http/https URLs are supported. If empty, the 'stage1.aci' in the same directory as '--rkt-path' will be used")
 	fs.StringVar(&s.SystemContainer, "system-container", s.SystemContainer, "Optional resource-only container in which to place all non-kernel processes that are not already in a container. Empty for no container. Rolling back the flag requires a reboot. (Default: \"\").")

cmd/kubelet/app/server.go

@@ -58,6 +58,7 @@ import (
 	"k8s.io/kubernetes/pkg/kubelet/server"
 	kubetypes "k8s.io/kubernetes/pkg/kubelet/types"
 	"k8s.io/kubernetes/pkg/util"
+	"k8s.io/kubernetes/pkg/util/flock"
 	"k8s.io/kubernetes/pkg/util/io"
 	"k8s.io/kubernetes/pkg/util/mount"
 	nodeutil "k8s.io/kubernetes/pkg/util/node"
@@ -262,6 +263,12 @@ func UnsecuredKubeletConfig(s *options.KubeletServer) (*KubeletConfig, error) {
 // will be ignored.
 func Run(s *options.KubeletServer, kcfg *KubeletConfig) error {
 	var err error
+	if s.LockFilePath != "" {
+		glog.Infof("aquiring lock on %q", s.LockFilePath)
+		if err := flock.Acquire(s.LockFilePath); err != nil {
+			return fmt.Errorf("unable to aquire file lock on %q: %v", s.LockFilePath, err)
+		}
+	}
 	if kcfg == nil {
 		cfg, err := UnsecuredKubeletConfig(s)
 		if err != nil {

docs/admin/kubelet.md

@@ -104,6 +104,7 @@ kubelet
       --kube-api-qps=5: QPS to use while talking with kubernetes apiserver
       --kube-reserved=: A set of ResourceName=ResourceQuantity (e.g. cpu=200m,memory=150G) pairs that describe resources reserved for kubernetes system components. Currently only cpu and memory are supported. See http://releases.k8s.io/HEAD/docs/user-guide/compute-resources.html for more detail. [default=none]
       --kubeconfig="/var/lib/kubelet/kubeconfig": Path to a kubeconfig file, specifying how to authenticate to API server (the master location is set by the api-servers flag).
+      --lock-file="/var/run/lock/kubelet.lock": <Warning: Alpha feature> The path to file for kubelet to use as a lock file.
       --log-flush-frequency=5s: Maximum number of seconds between log flushes
       --low-diskspace-threshold-mb=256: The absolute free disk space, in MB, to maintain. When disk space falls below this threshold, new pods would be rejected. Default: 256
       --manifest-url="": URL for accessing the container manifest
@@ -149,7 +150,7 @@ kubelet
       --volume-stats-agg-period=1m0s: Specifies interval for kubelet to calculate and cache the volume disk usage for all pods and volumes.  To disable volume calculations, set to 0.  Default: '1m'
 ```
 
-###### Auto generated by spf13/cobra on 9-Feb-2016
+###### Auto generated by spf13/cobra on 10-Feb-2016
 
 
 <!-- BEGIN MUNGE: GENERATED_ANALYTICS -->

hack/verify-flags/known-flags.txt

@@ -383,3 +383,4 @@ leader-elect-renew-deadline
 leader-elect-retry-period
 watch-cache-sizes
 configure-hairpin-mode
+lock-file

pkg/apis/componentconfig/types.go

@@ -234,6 +234,10 @@ type KubeletConfiguration struct {
 	// rktPath is hte path of rkt binary. Leave empty to use the first rkt in
 	// $PATH.
 	RktPath string `json:"rktPath,omitempty"`
+	// lockFilePath is the path that kubelet will use to as a lock file.
+	// It uses this file as a lock to synchronize with other kubelet processes
+	// that may be running.
+	LockFilePath string `json:"lockFilePath"`
 	// rktStage1Image is the image to use as stage1. Local paths and
 	// http/https URLs are supported.
 	RktStage1Image string `json:"rktStage1Image,omitempty"`

pkg/util/flock/flock_other.go

+// +build !linux,!darwin,!freebsd,!openbsd,!netbsd,!dragonfly
+
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+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 flock
+
+// Acquire is not implemented on non-unix systems.
+func Acquire(path string) error {
+	return nil
+}

pkg/util/flock/flock_unix.go

+// +build linux darwin freebsd openbsd netbsd dragonfly
+
+/*
+Copyright 2016 The Kubernetes Authors All rights reserved.
+
+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 flock
+
+import (
+	"os"
+	"sync"
+
+	"golang.org/x/sys/unix"
+)
+
+var (
+	// lock guards lockfile. Assignment is not atomic.
+	lock sync.Mutex
+	// os.File has a runtime.Finalizer so the fd will be closed if the struct
+	// is garbage collected. Let's hold onto a reference so that doesn't happen.
+	lockfile *os.File
+)
+
+// Acquire acquires a lock on a file for the duration of the process. This method
+// is reentrant.
+func Acquire(path string) error {
+	lock.Lock()
+	defer lock.Unlock()
+	var err error
+	if lockfile, err = os.OpenFile(path, os.O_RDWR|os.O_CREATE, 0600); err != nil {
+		return err
+	}
+
+	opts := unix.Flock_t{Type: unix.F_WRLCK}
+	if err := unix.FcntlFlock(lockfile.Fd(), unix.F_SETLKW, &opts); err != nil {
+		return err
+	}
+	return nil
+}