mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-07-29 22:46:12 +00:00
remove the rs&subrs logic from cluster controller
This commit is contained in:
huangyuqi
parent
649b2c6e20
commit
21fe26bd07
35
federation/client/cache/cluster_cache.go
vendored
Normal file
35
federation/client/cache/cluster_cache.go
vendored
Normal file
@ -0,0 +1,35 @@
|
||||
/*
|
||||
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 cache
|
||||
|
||||
import (
|
||||
federation_v1alpha1 "k8s.io/kubernetes/federation/apis/federation/v1alpha1"
|
||||
kubeCache "k8s.io/kubernetes/pkg/client/cache"
|
||||
)
|
||||
|
||||
// StoreToClusterLister makes a Store have the List method of the unversioned.ClusterInterface
|
||||
// The Store must contain (only) clusters.
|
||||
type StoreToClusterLister struct {
|
||||
kubeCache.Store
|
||||
}
|
||||
|
||||
func (s *StoreToClusterLister) List() (clusters federation_v1alpha1.ClusterList, err error) {
|
||||
for _, m := range s.Store.List() {
|
||||
clusters.Items = append(clusters.Items, *(m.(*federation_v1alpha1.Cluster)))
|
||||
}
|
||||
return clusters, nil
|
||||
}
|
||||
@ -1,4 +0,0 @@
|
||||
assignees:
|
||||
- davidopp
|
||||
- lavalamp
|
||||
- mikedanese
|
||||
4
federation/cmd/federation-controller-manager/OWNERS
Normal file
4
federation/cmd/federation-controller-manager/OWNERS
Normal file
@ -0,0 +1,4 @@
|
||||
assignees:
|
||||
- quinton-hoole
|
||||
- nikhiljindal
|
||||
- madhusundancs
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
Copyright 2014 The Kubernetes Authors All rights reserved.
|
||||
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.
|
||||
@ -15,11 +15,8 @@ limitations under the License.
|
||||
*/
|
||||
|
||||
// Package app implements a server that runs a set of active
|
||||
// components. This includes replication controllers, service endpoints and
|
||||
// nodes.
|
||||
//
|
||||
// CAUTION: If you update code in this file, you may need to also update code
|
||||
// in contrib/mesos/pkg/controllermanager/controllermanager.go
|
||||
// components. This includes cluster controller
|
||||
|
||||
package app
|
||||
|
||||
import (
|
||||
@ -28,15 +25,12 @@ import (
|
||||
"net/http/pprof"
|
||||
"strconv"
|
||||
|
||||
"k8s.io/kubernetes/federation/cmd/federated-controller-manager/app/options"
|
||||
"k8s.io/kubernetes/federation/cmd/federation-controller-manager/app/options"
|
||||
"k8s.io/kubernetes/pkg/client/restclient"
|
||||
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/internalclientset"
|
||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_3"
|
||||
clustercontroller "k8s.io/kubernetes/federation/pkg/federation-controller/cluster"
|
||||
"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
|
||||
|
||||
clustercontroller "k8s.io/kubernetes/federation/pkg/federated-controller/cluster"
|
||||
|
||||
"k8s.io/kubernetes/pkg/healthz"
|
||||
"k8s.io/kubernetes/pkg/util/configz"
|
||||
|
||||
@ -44,7 +38,6 @@ import (
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/spf13/cobra"
|
||||
"github.com/spf13/pflag"
|
||||
"k8s.io/kubernetes/pkg/util/wait"
|
||||
)
|
||||
|
||||
// NewControllerManagerCommand creates a *cobra.Command object with default parameters
|
||||
@ -52,15 +45,14 @@ func NewControllerManagerCommand() *cobra.Command {
|
||||
s := options.NewCMServer()
|
||||
s.AddFlags(pflag.CommandLine)
|
||||
cmd := &cobra.Command{
|
||||
Use: "ube-controller-manager",
|
||||
Long: `The ubernetes controller manager is a daemon that embeds
|
||||
the core control loops shipped with ubernetes. In applications of robotics and
|
||||
Use: "federation-controller-manager",
|
||||
Long: `The federation controller manager is a daemon that embeds
|
||||
the core control loops shipped with federation. In applications of robotics and
|
||||
automation, a control loop is a non-terminating loop that regulates the state of
|
||||
the system. In ubernetes, a controller is a control loop that watches the shared
|
||||
state of the cluster sub-replication constroller through the apiserver and makes
|
||||
changes attempting to move the current state towards the desired state. Examples
|
||||
of controllers that ship with ubernetes today are the cluster controller, service
|
||||
controller.`,
|
||||
the system. In federation, a controller is a control loop that watches the shared
|
||||
state of the federation cluster through the apiserver and makes changes attempting
|
||||
to move the current state towards the desired state. Examples of controllers that
|
||||
ship with federation today is the cluster controller.`,
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
},
|
||||
}
|
||||
@ -75,14 +67,14 @@ func Run(s *options.CMServer) error {
|
||||
} else {
|
||||
glog.Errorf("unable to register configz: %s", err)
|
||||
}
|
||||
restClientCfg, err := clientcmd.BuildConfigFromFlags(s.Master, s.ApiServerconfig)
|
||||
restClientCfg, err := clientcmd.BuildConfigFromFlags(s.Master, s.Kubeconfig)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Override restClientCfg qps/burst settings from flags
|
||||
restClientCfg.QPS = s.UberAPIQPS
|
||||
restClientCfg.Burst = s.UberAPIBurst
|
||||
restClientCfg.QPS = s.APIServerQPS
|
||||
restClientCfg.Burst = s.APIServerBurst
|
||||
|
||||
go func() {
|
||||
mux := http.NewServeMux()
|
||||
@ -101,22 +93,17 @@ func Run(s *options.CMServer) error {
|
||||
glog.Fatal(server.ListenAndServe())
|
||||
}()
|
||||
|
||||
run := func(stop <-chan struct{}) {
|
||||
err := StartControllers(s, restClientCfg, stop)
|
||||
run := func() {
|
||||
err := StartControllers(s, restClientCfg)
|
||||
glog.Fatalf("error running controllers: %v", err)
|
||||
panic("unreachable")
|
||||
}
|
||||
run(nil)
|
||||
run()
|
||||
panic("unreachable")
|
||||
}
|
||||
|
||||
func StartControllers(s *options.CMServer, restClientCfg *restclient.Config, stop <-chan struct{}) error {
|
||||
kubernetesClientSet := clientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "cluster-controller"))
|
||||
func StartControllers(s *options.CMServer, restClientCfg *restclient.Config) error {
|
||||
federationClientSet := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "cluster-controller"))
|
||||
go clustercontroller.NewclusterController(
|
||||
kubernetesClientSet,
|
||||
federationClientSet,
|
||||
s.ClusterMonitorPeriod.Duration,
|
||||
).Run(s.ConcurrentSubRCSyncs, wait.NeverStop)
|
||||
go clustercontroller.NewclusterController(federationClientSet, s.ClusterMonitorPeriod.Duration).Run()
|
||||
select {}
|
||||
}
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
Copyright 2014 The Kubernetes Authors All rights reserved.
|
||||
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.
|
||||
@ -15,9 +15,7 @@ limitations under the License.
|
||||
*/
|
||||
|
||||
// Package options provides the flags used for the controller manager.
|
||||
//
|
||||
// CAUTION: If you update code in this file, you may need to also update code
|
||||
// in contrib/mesos/pkg/controllermanager/controllermanager.go
|
||||
|
||||
package options
|
||||
|
||||
import (
|
||||
@ -30,52 +28,46 @@ import (
|
||||
)
|
||||
|
||||
type ControllerManagerConfiguration struct {
|
||||
unversioned.TypeMeta
|
||||
|
||||
// port is the port that the controller-manager's http service runs on.
|
||||
Port int `json:"port"`
|
||||
// address is the IP address to serve on (set to 0.0.0.0 for all interfaces).
|
||||
Address string `json:"address"`
|
||||
// concurrentSubRCSyncs is the number of sub replication controllers that are
|
||||
// allowed to sync concurrently. Larger number = more responsive replica
|
||||
// management, but more CPU (and network) load.
|
||||
ConcurrentSubRCSyncs int `json:"concurrentSubRCSyncs"`
|
||||
// clusterMonitorPeriod is the period for syncing ClusterStatus in cluster controller.
|
||||
ClusterMonitorPeriod unversioned.Duration `json:"clusterMonitorPeriod"`
|
||||
// uberAPIQPS is the QPS to use while talking with ubernetes apiserver.
|
||||
UberAPIQPS float32 `json:"uberAPIQPS"`
|
||||
// uberAPIBurst is the burst to use while talking with ubernetes apiserver.
|
||||
UberAPIBurst int `json:"uberAPIBurst"`
|
||||
// APIServerQPS is the QPS to use while talking with federation apiserver.
|
||||
APIServerQPS float32 `json:"federatedAPIQPS"`
|
||||
// APIServerBurst is the burst to use while talking with federation apiserver.
|
||||
APIServerBurst int `json:"federatedAPIBurst"`
|
||||
// enableProfiling enables profiling via web interface host:port/debug/pprof/
|
||||
EnableProfiling bool `json:"enableProfiling"`
|
||||
// leaderElection defines the configuration of leader election client.
|
||||
LeaderElection componentconfig.LeaderElectionConfiguration `json:"leaderElection"`
|
||||
// contentType is contentType of requests sent to apiserver.
|
||||
ContentType string `json:"contentType"`
|
||||
}
|
||||
|
||||
// CMServer is the main context object for the controller manager.
|
||||
type CMServer struct {
|
||||
ControllerManagerConfiguration
|
||||
|
||||
Master string
|
||||
ApiServerconfig string
|
||||
Master string
|
||||
Kubeconfig string
|
||||
}
|
||||
|
||||
const (
|
||||
// UberControllerManagerPort is the default port for the ubernetes controller manager status server.
|
||||
// FederatedControllerManagerPort is the default port for the federation controller manager status server.
|
||||
// May be overridden by a flag at startup.
|
||||
UberControllerManagerPort = 10252
|
||||
FederatedControllerManagerPort = 10253
|
||||
)
|
||||
|
||||
// NewCMServer creates a new CMServer with a default config.
|
||||
func NewCMServer() *CMServer {
|
||||
s := CMServer{
|
||||
ControllerManagerConfiguration: ControllerManagerConfiguration{
|
||||
Port: UberControllerManagerPort,
|
||||
Port: FederatedControllerManagerPort,
|
||||
Address: "0.0.0.0",
|
||||
ConcurrentSubRCSyncs: 5,
|
||||
ClusterMonitorPeriod: unversioned.Duration{40 * time.Second},
|
||||
UberAPIQPS: 20.0,
|
||||
UberAPIBurst: 30,
|
||||
ClusterMonitorPeriod: unversioned.Duration{Duration: 40 * time.Second},
|
||||
APIServerQPS: 20.0,
|
||||
APIServerBurst: 30,
|
||||
LeaderElection: leaderelection.DefaultLeaderElectionConfiguration(),
|
||||
},
|
||||
}
|
||||
@ -85,14 +77,13 @@ func NewCMServer() *CMServer {
|
||||
// AddFlags adds flags for a specific CMServer to the specified FlagSet
|
||||
func (s *CMServer) AddFlags(fs *pflag.FlagSet) {
|
||||
fs.IntVar(&s.Port, "port", s.Port, "The port that the controller-manager's http service runs on")
|
||||
fs.Var(componentconfig.IPVar{&s.Address}, "address", "The IP address to serve on (set to 0.0.0.0 for all interfaces)")
|
||||
fs.IntVar(&s.ConcurrentSubRCSyncs, "concurrent-subRc-syncs", s.ConcurrentSubRCSyncs, "The number of subRC syncing operations that will be done concurrently. Larger number = faster endpoint updating, but more CPU (and network) load")
|
||||
|
||||
fs.Var(componentconfig.IPVar{Val: &s.Address}, "address", "The IP address to serve on (set to 0.0.0.0 for all interfaces)")
|
||||
fs.DurationVar(&s.ClusterMonitorPeriod.Duration, "cluster-monitor-period", s.ClusterMonitorPeriod.Duration, "The period for syncing ClusterStatus in ClusterController.")
|
||||
fs.BoolVar(&s.EnableProfiling, "profiling", true, "Enable profiling via web interface host:port/debug/pprof/")
|
||||
fs.StringVar(&s.Master, "master", s.Master, "The address of the Kubernetes API server (overrides any value in kubeconfig)")
|
||||
fs.StringVar(&s.ApiServerconfig, "uberconfig", s.ApiServerconfig, "Path to ApiServerconfig file with authorization and master location information.")
|
||||
fs.Float32Var(&s.UberAPIQPS, "uber-api-qps", s.UberAPIQPS, "QPS to use while talking with ubernetes apiserver")
|
||||
fs.IntVar(&s.UberAPIBurst, "uber-api-burst", s.UberAPIBurst, "Burst to use while talking with ubernetes apiserver")
|
||||
fs.StringVar(&s.Master, "master", s.Master, "The address of the federation API server (overrides any value in kubeconfig)")
|
||||
fs.StringVar(&s.Kubeconfig, "kubeconfig", s.Kubeconfig, "Path to kubeconfig file with authorization and master location information.")
|
||||
fs.StringVar(&s.ContentType, "kube-api-content-type", s.ContentType, "ContentType of requests sent to apiserver. Passing application/vnd.kubernetes.protobuf is an experimental feature now.")
|
||||
fs.Float32Var(&s.APIServerQPS, "federated-api-qps", s.APIServerQPS, "QPS to use while talking with federation apiserver")
|
||||
fs.IntVar(&s.APIServerBurst, "federated-api-burst", s.APIServerBurst, "Burst to use while talking with federation apiserver")
|
||||
leaderelection.BindFlags(&s.LeaderElection, fs)
|
||||
}
|
||||
@ -22,8 +22,8 @@ import (
|
||||
"runtime"
|
||||
|
||||
"github.com/spf13/pflag"
|
||||
"k8s.io/kubernetes/federation/cmd/federated-controller-manager/app"
|
||||
"k8s.io/kubernetes/federation/cmd/federated-controller-manager/app/options"
|
||||
"k8s.io/kubernetes/federation/cmd/federation-controller-manager/app"
|
||||
"k8s.io/kubernetes/federation/cmd/federation-controller-manager/app/options"
|
||||
"k8s.io/kubernetes/pkg/healthz"
|
||||
"k8s.io/kubernetes/pkg/util"
|
||||
"k8s.io/kubernetes/pkg/util/flag"
|
||||
@ -1,5 +0,0 @@
|
||||
assignees:
|
||||
- bprashanth
|
||||
- davidopp
|
||||
- derekwaynecarr
|
||||
- mikedanese
|
||||
@ -1,87 +0,0 @@
|
||||
/*
|
||||
Copyright 2014 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 cluster
|
||||
|
||||
import (
|
||||
"k8s.io/kubernetes/pkg/apis/extensions"
|
||||
|
||||
"k8s.io/kubernetes/federation/apis/federation"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||
"k8s.io/kubernetes/pkg/client/restclient"
|
||||
"k8s.io/kubernetes/pkg/client/typed/discovery"
|
||||
"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
|
||||
"strings"
|
||||
)
|
||||
|
||||
const (
|
||||
UserAgentName = "Cluster-Controller"
|
||||
KubeAPIQPS = 20.0
|
||||
KubeAPIBurst = 30
|
||||
)
|
||||
|
||||
type ClusterClient struct {
|
||||
clientSet clientset.Interface
|
||||
discoveryClient *discovery.DiscoveryClient
|
||||
}
|
||||
|
||||
func NewClusterClientSet(c *federation.Cluster) (*ClusterClient, error) {
|
||||
//TODO:How to get cluster IP(huangyuqi)
|
||||
var clusterClientSet = ClusterClient{}
|
||||
clusterConfig, err := clientcmd.BuildConfigFromFlags(c.Spec.ServerAddressByClientCIDRs[0].ServerAddress, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// clusterConfig.ContentConfig.GroupVersion.Version = "extensions"
|
||||
clusterConfig.QPS = KubeAPIQPS
|
||||
clusterConfig.Burst = KubeAPIBurst
|
||||
clusterClientSet.clientSet = clientset.NewForConfigOrDie(restclient.AddUserAgent(clusterConfig, UserAgentName))
|
||||
clusterClientSet.discoveryClient = discovery.NewDiscoveryClientForConfigOrDie((restclient.AddUserAgent(clusterConfig, UserAgentName)))
|
||||
return &clusterClientSet, err
|
||||
}
|
||||
|
||||
// GetReplicaSetFromCluster get the replicaset from the kubernetes cluster
|
||||
func (self *ClusterClient) GetReplicaSetFromCluster(subRsName string, subRsNameSpace string) (*extensions.ReplicaSet, error) {
|
||||
return self.clientSet.Extensions().ReplicaSets(subRsNameSpace).Get(subRsName)
|
||||
}
|
||||
|
||||
// CreateReplicaSetToCluster create replicaset to the kubernetes cluster
|
||||
func (self *ClusterClient) CreateReplicaSetToCluster(subRs *extensions.ReplicaSet) (*extensions.ReplicaSet, error) {
|
||||
return self.clientSet.Extensions().ReplicaSets(subRs.Namespace).Create(subRs)
|
||||
}
|
||||
|
||||
// UpdateReplicaSetToCluster update replicaset to the kubernetes cluster
|
||||
func (self *ClusterClient) UpdateReplicaSetToCluster(subRs *extensions.ReplicaSet) (*extensions.ReplicaSet, error) {
|
||||
return self.clientSet.Extensions().ReplicaSets(subRs.Namespace).Update(subRs)
|
||||
}
|
||||
|
||||
// DeleteReplicasetFromCluster delete the replicaset from the kubernetes cluster
|
||||
func (self *ClusterClient) DeleteReplicasetFromCluster(subRs *extensions.ReplicaSet) error {
|
||||
return self.clientSet.Extensions().ReplicaSets(subRs.Namespace).Delete(subRs.Name, &api.DeleteOptions{})
|
||||
}
|
||||
|
||||
// GetClusterHealthStatus get the kubernetes cluster health status
|
||||
func (self *ClusterClient) GetClusterHealthStatus() federation.ClusterPhase {
|
||||
body, err := self.discoveryClient.Get().AbsPath("/healthz").Do().Raw()
|
||||
if err != nil {
|
||||
return federation.ClusterOffline
|
||||
}
|
||||
if !strings.EqualFold(string(body), "ok") {
|
||||
return federation.ClusterPending
|
||||
}
|
||||
return federation.ClusterRunning
|
||||
}
|
||||
@ -1,403 +0,0 @@
|
||||
/*
|
||||
Copyright 2014 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 cluster
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/golang/glog"
|
||||
"k8s.io/kubernetes/federation/apis/federation"
|
||||
federationcache "k8s.io/kubernetes/federation/client/cache"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/internalclientset"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
"k8s.io/kubernetes/pkg/api/meta"
|
||||
"k8s.io/kubernetes/pkg/apis/extensions"
|
||||
"k8s.io/kubernetes/pkg/client/cache"
|
||||
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
|
||||
"k8s.io/kubernetes/pkg/controller"
|
||||
"k8s.io/kubernetes/pkg/controller/framework"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
utilruntime "k8s.io/kubernetes/pkg/util/runtime"
|
||||
"k8s.io/kubernetes/pkg/util/sets"
|
||||
"k8s.io/kubernetes/pkg/util/wait"
|
||||
"k8s.io/kubernetes/pkg/util/workqueue"
|
||||
"k8s.io/kubernetes/pkg/watch"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
AnnotationKeyOfTargetCluster = "kubernetes.io/target-cluster"
|
||||
AnnotationKeyOfFederationReplicaSet = "kubernetes.io/created-by"
|
||||
)
|
||||
|
||||
type ClusterController struct {
|
||||
knownClusterSet sets.String
|
||||
|
||||
//federationClient used to operate cluster and subrs
|
||||
federationClient federationclientset.Interface
|
||||
|
||||
//client used to operate rs
|
||||
client clientset.Interface
|
||||
|
||||
// To allow injection of syncSubRC for testing.
|
||||
syncHandler func(subRcKey string) error
|
||||
|
||||
//clusterMonitorPeriod is the period for updating status of cluster
|
||||
clusterMonitorPeriod time.Duration
|
||||
//clusterKubeClientMap is a mapping of clusterName and restclient
|
||||
clusterKubeClientMap map[string]ClusterClient
|
||||
|
||||
// subRc framework and store
|
||||
subReplicaSetController *framework.Controller
|
||||
subReplicaSetStore federationcache.StoreToSubReplicaSetLister
|
||||
|
||||
// cluster framework and store
|
||||
clusterController *framework.Controller
|
||||
clusterStore federationcache.StoreToClusterLister
|
||||
|
||||
// UberRc framework and store
|
||||
replicaSetStore cache.StoreToReplicationControllerLister
|
||||
replicaSetController *framework.Controller
|
||||
|
||||
// subRC that have been queued up for processing by workers
|
||||
queue *workqueue.Type
|
||||
}
|
||||
|
||||
// NewclusterController returns a new cluster controller
|
||||
func NewclusterController(client clientset.Interface, federationClient federationclientset.Interface, clusterMonitorPeriod time.Duration) *ClusterController {
|
||||
cc := &ClusterController{
|
||||
knownClusterSet: make(sets.String),
|
||||
federationClient: federationClient,
|
||||
client: client,
|
||||
clusterMonitorPeriod: clusterMonitorPeriod,
|
||||
clusterKubeClientMap: make(map[string]ClusterClient),
|
||||
queue: workqueue.New(),
|
||||
}
|
||||
|
||||
cc.subReplicaSetStore.Store, cc.subReplicaSetController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
|
||||
return cc.federationClient.Federation().SubReplicaSets(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
|
||||
return cc.federationClient.Federation().SubReplicaSets(api.NamespaceAll).Watch(options)
|
||||
},
|
||||
},
|
||||
&federation.SubReplicaSet{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
framework.ResourceEventHandlerFuncs{
|
||||
AddFunc: func(obj interface{}) {
|
||||
subRc := obj.(*federation.SubReplicaSet)
|
||||
cc.enqueueSubRc(subRc)
|
||||
},
|
||||
UpdateFunc: func(oldObj, newObj interface{}) {
|
||||
subRc := newObj.(*federation.SubReplicaSet)
|
||||
cc.enqueueSubRc(subRc)
|
||||
},
|
||||
},
|
||||
)
|
||||
|
||||
cc.clusterStore.Store, cc.clusterController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
|
||||
return cc.federationClient.Federation().Clusters().List(options)
|
||||
},
|
||||
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
|
||||
return cc.federationClient.Federation().Clusters().Watch(options)
|
||||
},
|
||||
},
|
||||
&federation.Cluster{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
framework.ResourceEventHandlerFuncs{
|
||||
DeleteFunc: cc.delFromClusterSet,
|
||||
AddFunc: cc.addToClusterSet,
|
||||
},
|
||||
)
|
||||
|
||||
cc.replicaSetStore.Store, cc.replicaSetController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
|
||||
return cc.client.Extensions().ReplicaSets(api.NamespaceAll).List(options)
|
||||
},
|
||||
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
|
||||
return cc.client.Extensions().ReplicaSets(api.NamespaceAll).Watch(options)
|
||||
},
|
||||
},
|
||||
&api.ReplicationController{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
framework.ResourceEventHandlerFuncs{
|
||||
DeleteFunc: cc.deleteSubRs,
|
||||
},
|
||||
)
|
||||
cc.syncHandler = cc.syncSubReplicaSet
|
||||
return cc
|
||||
}
|
||||
|
||||
//delFromClusterSet delete a cluster from clusterSet and
|
||||
//delete the corresponding restclient from the map clusterKubeClientMap
|
||||
func (cc *ClusterController) delFromClusterSet(obj interface{}) {
|
||||
cluster := obj.(*federation.Cluster)
|
||||
cc.knownClusterSet.Delete(cluster.Name)
|
||||
delete(cc.clusterKubeClientMap, cluster.Name)
|
||||
}
|
||||
|
||||
//addToClusterSet insert the new cluster to clusterSet and creat a corresponding
|
||||
//restclient to map clusterKubeClientMap
|
||||
func (cc *ClusterController) addToClusterSet(obj interface{}) {
|
||||
cluster := obj.(*federation.Cluster)
|
||||
cc.knownClusterSet.Insert(cluster.Name)
|
||||
//create the restclient of cluster
|
||||
restClient, err := NewClusterClientSet(cluster)
|
||||
if err != nil {
|
||||
glog.Errorf("Failed to create corresponding restclient of kubernetes cluster: %v", err)
|
||||
}
|
||||
cc.clusterKubeClientMap[cluster.Name] = *restClient
|
||||
}
|
||||
|
||||
// Run begins watching and syncing.
|
||||
func (cc *ClusterController) Run(workers int, stopCh <-chan struct{}) {
|
||||
defer utilruntime.HandleCrash()
|
||||
go cc.clusterController.Run(wait.NeverStop)
|
||||
go cc.replicaSetController.Run(wait.NeverStop)
|
||||
go cc.subReplicaSetController.Run(wait.NeverStop)
|
||||
// monitor cluster status periodically, in phase 1 we just get the health state from "/healthz"
|
||||
go wait.Until(func() {
|
||||
if err := cc.UpdateClusterStatus(); err != nil {
|
||||
glog.Errorf("Error monitoring cluster status: %v", err)
|
||||
}
|
||||
}, cc.clusterMonitorPeriod, wait.NeverStop)
|
||||
for i := 0; i < workers; i++ {
|
||||
go wait.Until(cc.worker, time.Second, stopCh)
|
||||
}
|
||||
<-stopCh
|
||||
glog.Infof("Shutting down ClusterController")
|
||||
cc.queue.ShutDown()
|
||||
|
||||
}
|
||||
|
||||
// enqueueSubRc adds an object to the controller work queue
|
||||
// obj could be an *federation.SubReplicaSet, or a DeletionFinalStateUnknown item.
|
||||
func (cc *ClusterController) enqueueSubRc(obj interface{}) {
|
||||
key, err := controller.KeyFunc(obj)
|
||||
if err != nil {
|
||||
glog.Errorf("Couldn't get key for object %+v: %v", obj, err)
|
||||
return
|
||||
}
|
||||
cc.queue.Add(key)
|
||||
}
|
||||
|
||||
func (cc *ClusterController) worker() {
|
||||
for {
|
||||
func() {
|
||||
key, quit := cc.queue.Get()
|
||||
if quit {
|
||||
return
|
||||
}
|
||||
defer cc.queue.Done(key)
|
||||
err := cc.syncHandler(key.(string))
|
||||
if err != nil {
|
||||
glog.Errorf("Error syncing cluster controller: %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
// syncSubReplicaSet will sync the subrc with the given key,
|
||||
func (cc *ClusterController) syncSubReplicaSet(key string) error {
|
||||
startTime := time.Now()
|
||||
defer func() {
|
||||
glog.V(4).Infof("Finished syncing controller %q (%v)", key, time.Now().Sub(startTime))
|
||||
}()
|
||||
obj, exists, err := cc.subReplicaSetStore.Store.GetByKey(key)
|
||||
if !exists {
|
||||
glog.Infof("sub replicaset: %v has been deleted", key)
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
glog.Infof("Unable to retrieve sub replicaset %v from store: %v", key, err)
|
||||
cc.queue.Add(key)
|
||||
return err
|
||||
}
|
||||
subRs := obj.(*federation.SubReplicaSet)
|
||||
err = cc.manageSubReplicaSet(subRs)
|
||||
if err != nil {
|
||||
glog.Infof("Unable to manage subRs in kubernetes cluster: %v", key, err)
|
||||
cc.queue.Add(key)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
//getBindingClusterOfSubRS get the target cluster(scheduled by federation scheduler) of subRS
|
||||
//return the targetCluster name
|
||||
func (cc *ClusterController) getBindingClusterOfSubRS(subRs *federation.SubReplicaSet) (string, error) {
|
||||
accessor, err := meta.Accessor(subRs)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
annotations := accessor.GetAnnotations()
|
||||
if annotations == nil {
|
||||
return "", fmt.Errorf("Failed to get target cluster from the annotation of subreplicaset")
|
||||
}
|
||||
targetCluster, found := annotations[AnnotationKeyOfTargetCluster]
|
||||
if !found {
|
||||
return "", fmt.Errorf("Failed to get target cluster from the annotation of subreplicaset")
|
||||
}
|
||||
return targetCluster, nil
|
||||
}
|
||||
|
||||
//getFederateRsCreateBy get the federation ReplicaSet created by of subRS
|
||||
//return the replica set name
|
||||
func (cc *ClusterController) getFederateRsCreateBy(subRs *federation.SubReplicaSet) (string, error) {
|
||||
accessor, err := meta.Accessor(subRs)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
annotations := accessor.GetAnnotations()
|
||||
if annotations == nil {
|
||||
return "", fmt.Errorf("Failed to get Federate Rs Create By from the annotation of subreplicaset")
|
||||
}
|
||||
rsCreateBy, found := annotations[AnnotationKeyOfFederationReplicaSet]
|
||||
if !found {
|
||||
return "", fmt.Errorf("Failed to get Federate Rs Create By from the annotation of subreplicaset")
|
||||
}
|
||||
return rsCreateBy, nil
|
||||
}
|
||||
|
||||
// manageSubReplicaSet will sync the sub replicaset with the given key,and then create
|
||||
// or update replicaset to kubernetes cluster
|
||||
func (cc *ClusterController) manageSubReplicaSet(subRs *federation.SubReplicaSet) error {
|
||||
|
||||
targetClusterName, err := cc.getBindingClusterOfSubRS(subRs)
|
||||
if targetClusterName == "" || err != nil {
|
||||
glog.Infof("Failed to get target cluster of SubRS: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
clusterClient, found := cc.clusterKubeClientMap[targetClusterName]
|
||||
if !found {
|
||||
glog.Infof("Failed to get restclient of target cluster")
|
||||
return fmt.Errorf("Failed to get restclient of target cluster")
|
||||
}
|
||||
// check the sub replicaset already exists in kubernetes cluster or not
|
||||
replicaSet, err := clusterClient.GetReplicaSetFromCluster(subRs.Name, subRs.Namespace)
|
||||
if err != nil {
|
||||
glog.Infof("Failed to get RC in kubernetes cluster: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
rs := extensions.ReplicaSet(*subRs)
|
||||
//if not exist, means that this sub replicaset need to be created
|
||||
if replicaSet == nil {
|
||||
// create the sub replicaset to kubernetes cluster
|
||||
replicaSet, err := clusterClient.CreateReplicaSetToCluster(&rs)
|
||||
if err != nil || replicaSet == nil {
|
||||
glog.Infof("Failed to create sub replicaset in kubernetes cluster: %v", err)
|
||||
return err
|
||||
}
|
||||
}
|
||||
// if exists, then update it
|
||||
replicaSet, err = clusterClient.UpdateReplicaSetToCluster(&rs)
|
||||
if err != nil || replicaSet == nil {
|
||||
glog.Infof("Failed to update sub replicaset in kubernetes cluster: %v", err)
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cc *ClusterController) GetClusterStatus(cluster *federation.Cluster) (*federation.ClusterStatus, error) {
|
||||
// just get the status of cluster, by getting the version of kubernetes cluster api-server
|
||||
var clusterStatus federation.ClusterStatus
|
||||
clusterClient, found := cc.clusterKubeClientMap[cluster.Name]
|
||||
if !found {
|
||||
glog.Infof("Failed to get restclient of target cluster")
|
||||
return nil, fmt.Errorf("Failed to get restclient of target cluster")
|
||||
}
|
||||
clusterStatus.Phase = clusterClient.GetClusterHealthStatus()
|
||||
return &clusterStatus, nil
|
||||
}
|
||||
|
||||
// monitorClusterStatus checks cluster status and get the metrics from cluster's restapi and RC state
|
||||
func (cc *ClusterController) UpdateClusterStatus() error {
|
||||
clusters, err := cc.federationClient.Federation().Clusters().List(api.ListOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, cluster := range clusters.Items {
|
||||
if !cc.knownClusterSet.Has(cluster.Name) {
|
||||
glog.V(1).Infof("ClusterController observed a new cluster: %#v", cluster)
|
||||
cc.knownClusterSet.Insert(cluster.Name)
|
||||
}
|
||||
}
|
||||
|
||||
// If there's a difference between lengths of known clusters and observed clusters
|
||||
// we must have removed some clusters, and evict the subRs belong to the cluster
|
||||
if len(cc.knownClusterSet) != len(clusters.Items) {
|
||||
observedSet := make(sets.String)
|
||||
for _, cluster := range clusters.Items {
|
||||
observedSet.Insert(cluster.Name)
|
||||
}
|
||||
deleted := cc.knownClusterSet.Difference(observedSet)
|
||||
for clusterName := range deleted {
|
||||
glog.V(1).Infof("ClusterController observed a Cluster deletion: %v", clusterName)
|
||||
//TODO: evict the subRS
|
||||
cc.knownClusterSet.Delete(clusterName)
|
||||
}
|
||||
}
|
||||
for _, cluster := range clusters.Items {
|
||||
ClusterStatus, err := cc.GetClusterStatus(&cluster)
|
||||
if err == nil {
|
||||
continue
|
||||
}
|
||||
cluster.Status.Phase = ClusterStatus.Phase
|
||||
_, err = cc.federationClient.Federation().Clusters().Update(&cluster)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (cc *ClusterController) deleteSubRs(cur interface{}) {
|
||||
rs := cur.(*extensions.ReplicaSet)
|
||||
// get the corresponing subrs from the cache
|
||||
subRSList, err := cc.federationClient.Federation().SubReplicaSets(api.NamespaceAll).List(api.ListOptions{})
|
||||
if err != nil || len(subRSList.Items) == 0 {
|
||||
glog.Infof("Couldn't get subRS to delete : %+v", cur)
|
||||
return
|
||||
}
|
||||
|
||||
// get the related subRS created by the replicaset
|
||||
for _, subRs := range subRSList.Items {
|
||||
name, err := cc.getFederateRsCreateBy(&subRs)
|
||||
if err != nil || !strings.EqualFold(rs.Name, name){
|
||||
continue
|
||||
}
|
||||
targetClusterName, err := cc.getBindingClusterOfSubRS(&subRs)
|
||||
if targetClusterName == "" || err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
clusterClient, found := cc.clusterKubeClientMap[targetClusterName]
|
||||
if !found {
|
||||
continue
|
||||
}
|
||||
rs := extensions.ReplicaSet(subRs)
|
||||
err = clusterClient.DeleteReplicasetFromCluster(&rs)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@ -1,90 +0,0 @@
|
||||
/*
|
||||
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 controller
|
||||
|
||||
import (
|
||||
"hash/adler32"
|
||||
"sync"
|
||||
|
||||
"github.com/golang/groupcache/lru"
|
||||
"k8s.io/kubernetes/pkg/api/meta"
|
||||
hashutil "k8s.io/kubernetes/pkg/util/hash"
|
||||
)
|
||||
|
||||
type objectWithMeta interface {
|
||||
meta.Object
|
||||
}
|
||||
|
||||
// keyFunc returns the key of an object, which is used to look up in the cache for it's matching object.
|
||||
// Since we match objects by namespace and Labels/Selector, so if two objects have the same namespace and labels,
|
||||
// they will have the same key.
|
||||
func keyFunc(obj objectWithMeta) uint64 {
|
||||
hash := adler32.New()
|
||||
hashutil.DeepHashObject(hash, &equivalenceLabelObj{
|
||||
namespace: obj.GetNamespace(),
|
||||
labels: obj.GetLabels(),
|
||||
})
|
||||
return uint64(hash.Sum32())
|
||||
}
|
||||
|
||||
type equivalenceLabelObj struct {
|
||||
namespace string
|
||||
labels map[string]string
|
||||
}
|
||||
|
||||
// MatchingCache save label and selector matching relationship
|
||||
type MatchingCache struct {
|
||||
mutex sync.RWMutex
|
||||
cache *lru.Cache
|
||||
}
|
||||
|
||||
// NewMatchingCache return a NewMatchingCache, which save label and selector matching relationship.
|
||||
func NewMatchingCache(maxCacheEntries int) *MatchingCache {
|
||||
return &MatchingCache{
|
||||
cache: lru.New(maxCacheEntries),
|
||||
}
|
||||
}
|
||||
|
||||
// Add will add matching information to the cache.
|
||||
func (c *MatchingCache) Add(labelObj objectWithMeta, selectorObj objectWithMeta) {
|
||||
key := keyFunc(labelObj)
|
||||
c.mutex.Lock()
|
||||
defer c.mutex.Unlock()
|
||||
c.cache.Add(key, selectorObj)
|
||||
}
|
||||
|
||||
// GetMatchingObject lookup the matching object for a given object.
|
||||
// Note: the cache information may be invalid since the controller may be deleted or updated,
|
||||
// we need check in the external request to ensure the cache data is not dirty.
|
||||
func (c *MatchingCache) GetMatchingObject(labelObj objectWithMeta) (controller interface{}, exists bool) {
|
||||
key := keyFunc(labelObj)
|
||||
c.mutex.Lock()
|
||||
defer c.mutex.Unlock()
|
||||
return c.cache.Get(key)
|
||||
}
|
||||
|
||||
// Update update the cached matching information.
|
||||
func (c *MatchingCache) Update(labelObj objectWithMeta, selectorObj objectWithMeta) {
|
||||
c.Add(labelObj, selectorObj)
|
||||
}
|
||||
|
||||
// InvalidateAll invalidate the whole cache.
|
||||
func (c *MatchingCache) InvalidateAll() {
|
||||
c.mutex.Lock()
|
||||
defer c.mutex.Unlock()
|
||||
c.cache = lru.New(c.cache.MaxEntries)
|
||||
}
|
||||
4
federation/pkg/federation-controller/OWNERS
Normal file
4
federation/pkg/federation-controller/OWNERS
Normal file
@ -0,0 +1,4 @@
|
||||
assignees:
|
||||
- quinton-hoole
|
||||
- nikhiljindal
|
||||
- madhusundancs
|
||||
123
federation/pkg/federation-controller/cluster/cluster_client.go
Normal file
123
federation/pkg/federation-controller/cluster/cluster_client.go
Normal file
@ -0,0 +1,123 @@
|
||||
/*
|
||||
Copyright 2014 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 cluster
|
||||
|
||||
import (
|
||||
"net"
|
||||
"strings"
|
||||
|
||||
federation_v1alpha1 "k8s.io/kubernetes/federation/apis/federation/v1alpha1"
|
||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||
"k8s.io/kubernetes/pkg/api/v1"
|
||||
"k8s.io/kubernetes/pkg/client/restclient"
|
||||
"k8s.io/kubernetes/pkg/client/typed/discovery"
|
||||
"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
|
||||
utilnet "k8s.io/kubernetes/pkg/util/net"
|
||||
)
|
||||
|
||||
const (
|
||||
UserAgentName = "Cluster-Controller"
|
||||
KubeAPIQPS = 20.0
|
||||
KubeAPIBurst = 30
|
||||
)
|
||||
|
||||
type ClusterClient struct {
|
||||
discoveryClient *discovery.DiscoveryClient
|
||||
}
|
||||
|
||||
func NewClusterClientSet(c *federation_v1alpha1.Cluster) (*ClusterClient, error) {
|
||||
var serverAddress string
|
||||
hostIP, err := utilnet.ChooseHostInterface()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for _, item := range c.Spec.ServerAddressByClientCIDRs {
|
||||
_, cidrnet, err := net.ParseCIDR(item.ClientCIDR)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
myaddr := net.ParseIP(hostIP.String())
|
||||
if cidrnet.Contains(myaddr) == true {
|
||||
serverAddress = item.ServerAddress
|
||||
break
|
||||
}
|
||||
}
|
||||
var clusterClientSet = ClusterClient{}
|
||||
if serverAddress != "" {
|
||||
clusterConfig, err := clientcmd.BuildConfigFromFlags(serverAddress, "")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
clusterConfig.QPS = KubeAPIQPS
|
||||
clusterConfig.Burst = KubeAPIBurst
|
||||
clusterClientSet.discoveryClient = discovery.NewDiscoveryClientForConfigOrDie((restclient.AddUserAgent(clusterConfig, UserAgentName)))
|
||||
if clusterClientSet.discoveryClient == nil {
|
||||
return nil, nil
|
||||
}
|
||||
}
|
||||
return &clusterClientSet, err
|
||||
}
|
||||
|
||||
// GetClusterHealthStatus gets the kubernetes cluster health status by requesting "/healthz"
|
||||
func (self *ClusterClient) GetClusterHealthStatus() *federation_v1alpha1.ClusterStatus {
|
||||
clusterStatus := federation_v1alpha1.ClusterStatus{}
|
||||
currentTime := unversioned.Now()
|
||||
newClusterReadyCondition := federation_v1alpha1.ClusterCondition{
|
||||
Type: federation_v1alpha1.ClusterReady,
|
||||
Status: v1.ConditionTrue,
|
||||
Reason: "ClusterReady",
|
||||
Message: "/healthz responded with ok",
|
||||
LastProbeTime: currentTime,
|
||||
LastTransitionTime: currentTime,
|
||||
}
|
||||
newClusterNotReadyCondition := federation_v1alpha1.ClusterCondition{
|
||||
Type: federation_v1alpha1.ClusterReady,
|
||||
Status: v1.ConditionFalse,
|
||||
Reason: "ClusterNotReady",
|
||||
Message: "/healthz responded without ok",
|
||||
LastProbeTime: currentTime,
|
||||
LastTransitionTime: currentTime,
|
||||
}
|
||||
newNodeOfflineCondition := federation_v1alpha1.ClusterCondition{
|
||||
Type: federation_v1alpha1.ClusterOffline,
|
||||
Status: v1.ConditionTrue,
|
||||
Reason: "ClusterNotReachable",
|
||||
Message: "cluster is not reachable",
|
||||
LastProbeTime: currentTime,
|
||||
LastTransitionTime: currentTime,
|
||||
}
|
||||
newNodeNotOfflineCondition := federation_v1alpha1.ClusterCondition{
|
||||
Type: federation_v1alpha1.ClusterOffline,
|
||||
Status: v1.ConditionFalse,
|
||||
Reason: "ClusterReachable",
|
||||
Message: "cluster is reachable",
|
||||
LastProbeTime: currentTime,
|
||||
LastTransitionTime: currentTime,
|
||||
}
|
||||
body, err := self.discoveryClient.Get().AbsPath("/healthz").Do().Raw()
|
||||
if err != nil {
|
||||
clusterStatus.Conditions = append(clusterStatus.Conditions, newNodeOfflineCondition)
|
||||
} else {
|
||||
if !strings.EqualFold(string(body), "ok") {
|
||||
clusterStatus.Conditions = append(clusterStatus.Conditions, newClusterNotReadyCondition, newNodeNotOfflineCondition)
|
||||
} else {
|
||||
clusterStatus.Conditions = append(clusterStatus.Conditions, newClusterReadyCondition)
|
||||
}
|
||||
}
|
||||
return &clusterStatus
|
||||
}
|
||||
@ -0,0 +1,196 @@
|
||||
/*
|
||||
Copyright 2014 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 cluster
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/golang/glog"
|
||||
"k8s.io/kubernetes/federation/apis/federation"
|
||||
federation_v1alpha1 "k8s.io/kubernetes/federation/apis/federation/v1alpha1"
|
||||
cluster_cache "k8s.io/kubernetes/federation/client/cache"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_3"
|
||||
"k8s.io/kubernetes/pkg/api"
|
||||
"k8s.io/kubernetes/pkg/client/cache"
|
||||
"k8s.io/kubernetes/pkg/controller"
|
||||
"k8s.io/kubernetes/pkg/controller/framework"
|
||||
"k8s.io/kubernetes/pkg/runtime"
|
||||
utilruntime "k8s.io/kubernetes/pkg/util/runtime"
|
||||
"k8s.io/kubernetes/pkg/util/sets"
|
||||
"k8s.io/kubernetes/pkg/util/wait"
|
||||
"k8s.io/kubernetes/pkg/watch"
|
||||
)
|
||||
|
||||
type ClusterController struct {
|
||||
knownClusterSet sets.String
|
||||
|
||||
// federationClient used to operate cluster
|
||||
federationClient federationclientset.Interface
|
||||
|
||||
// clusterMonitorPeriod is the period for updating status of cluster
|
||||
clusterMonitorPeriod time.Duration
|
||||
// clusterClusterStatusMap is a mapping of clusterName and cluster status of last sampling
|
||||
clusterClusterStatusMap map[string]federation_v1alpha1.ClusterStatus
|
||||
|
||||
// clusterKubeClientMap is a mapping of clusterName and restclient
|
||||
clusterKubeClientMap map[string]ClusterClient
|
||||
|
||||
// cluster framework and store
|
||||
clusterController *framework.Controller
|
||||
clusterStore cluster_cache.StoreToClusterLister
|
||||
}
|
||||
|
||||
// NewclusterController returns a new cluster controller
|
||||
func NewclusterController(federationClient federationclientset.Interface, clusterMonitorPeriod time.Duration) *ClusterController {
|
||||
cc := &ClusterController{
|
||||
knownClusterSet: make(sets.String),
|
||||
federationClient: federationClient,
|
||||
clusterMonitorPeriod: clusterMonitorPeriod,
|
||||
clusterClusterStatusMap: make(map[string]federation_v1alpha1.ClusterStatus),
|
||||
clusterKubeClientMap: make(map[string]ClusterClient),
|
||||
}
|
||||
cc.clusterStore.Store, cc.clusterController = framework.NewInformer(
|
||||
&cache.ListWatch{
|
||||
ListFunc: func(options api.ListOptions) (runtime.Object, error) {
|
||||
return cc.federationClient.Federation().Clusters().List(options)
|
||||
},
|
||||
WatchFunc: func(options api.ListOptions) (watch.Interface, error) {
|
||||
return cc.federationClient.Federation().Clusters().Watch(options)
|
||||
},
|
||||
},
|
||||
&federation.Cluster{},
|
||||
controller.NoResyncPeriodFunc(),
|
||||
framework.ResourceEventHandlerFuncs{
|
||||
DeleteFunc: cc.delFromClusterSet,
|
||||
AddFunc: cc.addToClusterSet,
|
||||
},
|
||||
)
|
||||
return cc
|
||||
}
|
||||
|
||||
// delFromClusterSet delete a cluster from clusterSet and
|
||||
// delete the corresponding restclient from the map clusterKubeClientMap
|
||||
func (cc *ClusterController) delFromClusterSet(obj interface{}) {
|
||||
cluster := obj.(*federation_v1alpha1.Cluster)
|
||||
cc.knownClusterSet.Delete(cluster.Name)
|
||||
delete(cc.clusterKubeClientMap, cluster.Name)
|
||||
}
|
||||
|
||||
// addToClusterSet insert the new cluster to clusterSet and create a corresponding
|
||||
// restclient to map clusterKubeClientMap
|
||||
func (cc *ClusterController) addToClusterSet(obj interface{}) {
|
||||
cluster := obj.(*federation_v1alpha1.Cluster)
|
||||
cc.knownClusterSet.Insert(cluster.Name)
|
||||
// create the restclient of cluster
|
||||
restClient, err := NewClusterClientSet(cluster)
|
||||
if err != nil || restClient == nil {
|
||||
glog.Errorf("Failed to create corresponding restclient of kubernetes cluster: %v", err)
|
||||
return
|
||||
}
|
||||
cc.clusterKubeClientMap[cluster.Name] = *restClient
|
||||
}
|
||||
|
||||
// Run begins watching and syncing.
|
||||
func (cc *ClusterController) Run() {
|
||||
defer utilruntime.HandleCrash()
|
||||
go cc.clusterController.Run(wait.NeverStop)
|
||||
// monitor cluster status periodically, in phase 1 we just get the health state from "/healthz"
|
||||
go wait.Until(func() {
|
||||
if err := cc.UpdateClusterStatus(); err != nil {
|
||||
glog.Errorf("Error monitoring cluster status: %v", err)
|
||||
}
|
||||
}, cc.clusterMonitorPeriod, wait.NeverStop)
|
||||
}
|
||||
|
||||
func (cc *ClusterController) GetClusterStatus(cluster *federation_v1alpha1.Cluster) (*federation_v1alpha1.ClusterStatus, error) {
|
||||
// just get the status of cluster, by requesting the restapi "/healthz"
|
||||
clusterClient, found := cc.clusterKubeClientMap[cluster.Name]
|
||||
if !found {
|
||||
glog.Infof("It's a new cluster, a cluster client will be created")
|
||||
client, err := NewClusterClientSet(cluster)
|
||||
if err != nil || client == nil {
|
||||
glog.Infof("Failed to create cluster client, err: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
clusterClient = *client
|
||||
cc.clusterKubeClientMap[cluster.Name] = clusterClient
|
||||
}
|
||||
clusterStatus := clusterClient.GetClusterHealthStatus()
|
||||
return clusterStatus, nil
|
||||
}
|
||||
|
||||
// UpdateClusterStatus checks cluster status and get the metrics from cluster's restapi
|
||||
func (cc *ClusterController) UpdateClusterStatus() error {
|
||||
clusters, err := cc.federationClient.Federation().Clusters().List(api.ListOptions{})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
for _, cluster := range clusters.Items {
|
||||
if !cc.knownClusterSet.Has(cluster.Name) {
|
||||
glog.V(1).Infof("ClusterController observed a new cluster: %#v", cluster)
|
||||
cc.knownClusterSet.Insert(cluster.Name)
|
||||
}
|
||||
}
|
||||
|
||||
// If there's a difference between lengths of known clusters and observed clusters
|
||||
if len(cc.knownClusterSet) != len(clusters.Items) {
|
||||
observedSet := make(sets.String)
|
||||
for _, cluster := range clusters.Items {
|
||||
observedSet.Insert(cluster.Name)
|
||||
}
|
||||
deleted := cc.knownClusterSet.Difference(observedSet)
|
||||
for clusterName := range deleted {
|
||||
glog.V(1).Infof("ClusterController observed a Cluster deletion: %v", clusterName)
|
||||
cc.knownClusterSet.Delete(clusterName)
|
||||
}
|
||||
}
|
||||
for _, cluster := range clusters.Items {
|
||||
clusterStatusNew, err := cc.GetClusterStatus(&cluster)
|
||||
if err != nil {
|
||||
glog.Infof("Failed to Get the status of cluster: %v", cluster.Name)
|
||||
continue
|
||||
}
|
||||
clusterStatusOld, found := cc.clusterClusterStatusMap[cluster.Name]
|
||||
if !found {
|
||||
glog.Infof("There is no status stored for cluster: %v before", cluster.Name)
|
||||
|
||||
} else {
|
||||
hasTransition := false
|
||||
for i := 0; i < len(clusterStatusNew.Conditions); i++ {
|
||||
if !(strings.EqualFold(string(clusterStatusNew.Conditions[i].Type), string(clusterStatusOld.Conditions[i].Type)) &&
|
||||
strings.EqualFold(string(clusterStatusNew.Conditions[i].Status), string(clusterStatusOld.Conditions[i].Status))) {
|
||||
hasTransition = true
|
||||
break
|
||||
}
|
||||
}
|
||||
if !hasTransition {
|
||||
for j := 0; j < len(clusterStatusNew.Conditions); j++ {
|
||||
clusterStatusNew.Conditions[j].LastTransitionTime = clusterStatusOld.Conditions[j].LastTransitionTime
|
||||
}
|
||||
}
|
||||
}
|
||||
cc.clusterClusterStatusMap[cluster.Name] = *clusterStatusNew
|
||||
cluster.Status = *clusterStatusNew
|
||||
cluster, err := cc.federationClient.Federation().Clusters().UpdateStatus(&cluster)
|
||||
if err != nil {
|
||||
glog.Infof("Failed to update the status of cluster: %v ,error is : %v", cluster.Name, err)
|
||||
continue
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@ -0,0 +1,138 @@
|
||||
/*
|
||||
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 cluster
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
federation_v1alpha1 "k8s.io/kubernetes/federation/apis/federation/v1alpha1"
|
||||
federationclientset "k8s.io/kubernetes/federation/client/clientset_generated/federation_release_1_3"
|
||||
"k8s.io/kubernetes/pkg/api/testapi"
|
||||
"k8s.io/kubernetes/pkg/api/unversioned"
|
||||
"k8s.io/kubernetes/pkg/api/v1"
|
||||
"k8s.io/kubernetes/pkg/client/restclient"
|
||||
"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
|
||||
"k8s.io/kubernetes/pkg/util"
|
||||
)
|
||||
|
||||
func newCluster(clusterName string, serverUrl string) *federation_v1alpha1.Cluster {
|
||||
cluster := federation_v1alpha1.Cluster{
|
||||
TypeMeta: unversioned.TypeMeta{APIVersion: testapi.Federation.GroupVersion().String()},
|
||||
ObjectMeta: v1.ObjectMeta{
|
||||
UID: util.NewUUID(),
|
||||
Name: clusterName,
|
||||
},
|
||||
Spec: federation_v1alpha1.ClusterSpec{
|
||||
ServerAddressByClientCIDRs: []federation_v1alpha1.ServerAddressByClientCIDR{
|
||||
{
|
||||
ClientCIDR: "0.0.0.0/0",
|
||||
ServerAddress: serverUrl,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
return &cluster
|
||||
}
|
||||
|
||||
func newClusterList(cluster *federation_v1alpha1.Cluster) *federation_v1alpha1.ClusterList {
|
||||
clusterList := federation_v1alpha1.ClusterList{
|
||||
TypeMeta: unversioned.TypeMeta{APIVersion: testapi.Federation.GroupVersion().String()},
|
||||
ListMeta: unversioned.ListMeta{
|
||||
SelfLink: "foobar",
|
||||
},
|
||||
Items: []federation_v1alpha1.Cluster{},
|
||||
}
|
||||
clusterList.Items = append(clusterList.Items, *cluster)
|
||||
return &clusterList
|
||||
}
|
||||
|
||||
// init a fake http handler, simulate a federation apiserver, response the "DELETE" "PUT" "GET" "UPDATE"
|
||||
// when "canBeGotten" is false, means that user can not get the cluster cluster from apiserver
|
||||
func createHttptestFakeHandlerForFederation(clusterList *federation_v1alpha1.ClusterList, canBeGotten bool) *http.HandlerFunc {
|
||||
fakeHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
clusterListString, _ := json.Marshal(*clusterList)
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch r.Method {
|
||||
case "PUT":
|
||||
fmt.Fprintln(w, string(clusterListString))
|
||||
case "GET":
|
||||
if canBeGotten {
|
||||
fmt.Fprintln(w, string(clusterListString))
|
||||
} else {
|
||||
fmt.Fprintln(w, "")
|
||||
}
|
||||
default:
|
||||
fmt.Fprintln(w, "")
|
||||
}
|
||||
})
|
||||
return &fakeHandler
|
||||
}
|
||||
|
||||
// init a fake http handler, simulate a cluster apiserver, response the "/healthz"
|
||||
// when "canBeGotten" is false, means that user can not get response from apiserver
|
||||
func createHttptestFakeHandlerForCluster(canBeGotten bool) *http.HandlerFunc {
|
||||
fakeHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
switch r.Method {
|
||||
case "GET":
|
||||
if canBeGotten {
|
||||
fmt.Fprintln(w, "ok")
|
||||
} else {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
default:
|
||||
fmt.Fprintln(w, "")
|
||||
}
|
||||
})
|
||||
return &fakeHandler
|
||||
}
|
||||
|
||||
func TestUpdateClusterStatusOK(t *testing.T) {
|
||||
clusterName := "foobarCluster"
|
||||
// create dummy httpserver
|
||||
testClusterServer := httptest.NewServer(createHttptestFakeHandlerForCluster(true))
|
||||
defer testClusterServer.Close()
|
||||
federationCluster := newCluster(clusterName, testClusterServer.URL)
|
||||
federationClusterList := newClusterList(federationCluster)
|
||||
|
||||
testFederationServer := httptest.NewServer(createHttptestFakeHandlerForFederation(federationClusterList, true))
|
||||
defer testFederationServer.Close()
|
||||
|
||||
restClientCfg, err := clientcmd.BuildConfigFromFlags(testFederationServer.URL, "")
|
||||
if err != nil {
|
||||
t.Errorf("Failed to build client config")
|
||||
}
|
||||
federationClientSet := federationclientset.NewForConfigOrDie(restclient.AddUserAgent(restClientCfg, "cluster-controller"))
|
||||
|
||||
manager := NewclusterController(federationClientSet, 5)
|
||||
err = manager.UpdateClusterStatus()
|
||||
if err != nil {
|
||||
t.Errorf("Failed to Update Cluster Status: %v", err)
|
||||
}
|
||||
clusterStatus, found := manager.clusterClusterStatusMap[clusterName]
|
||||
if !found {
|
||||
t.Errorf("Failed to Update Cluster Status")
|
||||
} else {
|
||||
if (clusterStatus.Conditions[1].Status != v1.ConditionFalse) || (clusterStatus.Conditions[1].Type != federation_v1alpha1.ClusterOffline) {
|
||||
t.Errorf("Failed to Update Cluster Status")
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,5 +1,5 @@
|
||||
/*
|
||||
Copyright 2014 The Kubernetes Authors All rights reserved.
|
||||
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.
|
||||
@ -14,6 +14,5 @@ See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
// Package cluster contains code for syncing cloud instances with
|
||||
// node registry
|
||||
// Package cluster contains code for syncing cluster
|
||||
package cluster
|
||||
@ -14,6 +14,6 @@ See the License for the specific language governing permissions and
|
||||
limitations under the License.
|
||||
*/
|
||||
|
||||
// Package controller contains code for controllers (like the replication
|
||||
// Package federation_controller contains code for controllers (like the cluster
|
||||
// controller).
|
||||
package controller
|
||||
package federation_controller
|
||||
@ -37,6 +37,7 @@ kube::golang::server_targets() {
|
||||
cmd/kubemark
|
||||
cmd/hyperkube
|
||||
federation/cmd/federated-apiserver
|
||||
federation/cmd/federation-controller-manager
|
||||
plugin/cmd/kube-scheduler
|
||||
)
|
||||
if [ -n "${KUBERNETES_CONTRIB:-}" ]; then
|
||||
|
||||
@ -54,6 +54,7 @@ cluster-dns
|
||||
cluster-domain
|
||||
cluster-name
|
||||
cluster-tag
|
||||
cluster-monitor-period
|
||||
concurrent-deployment-syncs
|
||||
concurrent-endpoint-syncs
|
||||
concurrent-namespace-syncs
|
||||
@ -136,6 +137,8 @@ external-ip
|
||||
failover-timeout
|
||||
failure-domains
|
||||
fake-clientset
|
||||
federated-api-burst
|
||||
federated-api-qps
|
||||
file-check-frequency
|
||||
file-suffix
|
||||
file_content_in_loop
|
||||
|
||||
Loading…
Reference in New Issue
Block a user