github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-24 12:15:52 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #15427 from brendandburns/controllerversion

Browse Source 
Auto commit by PR queue bot
This commit is contained in:
k8s-merge-robot 2015-10-15 05:38:33 -07:00
commit 7ef2a1b293
9 changed files with 315 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
cluster/saltbase/salt/kube-controller-manager/kube-controller-manager.manifest
View File

@ -1,7 +1,6 @@
{% set cluster_name = "" -%}
{% set cluster_cidr = "" -%}
{% set allocate_node_cidrs = "" -%}
{% set enable_experimental = "" -%}
{% if pillar['instance_prefix'] is defined -%}
{% set cluster_name = "--cluster-name=" + pillar['instance_prefix'] -%}
@ -12,9 +11,6 @@
{% if pillar['allocate_node_cidrs'] is defined -%}
{% set allocate_node_cidrs = "--allocate-node-cidrs=" + pillar['allocate_node_cidrs'] -%}
{% endif -%}
{% if pillar['enable_experimental_api'] is defined -%}
{% set enable_experimental = "--enable-experimental=" + pillar['enable_experimental_api'] -%}
{% endif -%}
{% set cloud_provider = "" -%}
{% set cloud_config = "" -%}
@ -38,7 +34,7 @@
{% set root_ca_file = "--root-ca-file=/srv/kubernetes/ca.crt" -%}
{% endif -%}
{% set params = "--master=127.0.0.1:8080" + " " + cluster_name + " " + cluster_cidr + " " + allocate_node_cidrs + " " + enable_experimental + " " + cloud_provider + " " + cloud_config + service_account_key + pillar['log_level'] + " " + root_ca_file -%}
{% set params = "--master=127.0.0.1:8080" + " " + cluster_name + " " + cluster_cidr + " " + allocate_node_cidrs + " " + cloud_provider + " " + cloud_config + service_account_key + pillar['log_level'] + " " + root_ca_file -%}
# test_args has to be kept at the end, so they'll overwrite any prior configuration
{% if pillar['controller_manager_test_args'] is defined -%}

77
cmd/kube-controller-manager/app/controllermanager.go
View File

@ -32,6 +32,7 @@ import (
"strconv"
"time"
"k8s.io/kubernetes/pkg/api/unversioned"
client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
@ -91,11 +92,10 @@ type CMServer struct {
ServiceAccountKeyFile string
RootCAFile string
ClusterName string
ClusterCIDR net.IPNet
AllocateNodeCIDRs bool
EnableProfiling bool
EnableExperimental bool
ClusterName string
ClusterCIDR net.IPNet
AllocateNodeCIDRs bool
EnableProfiling bool
Master string
Kubeconfig string
@ -196,7 +196,6 @@ func (s *CMServer) AddFlags(fs *pflag.FlagSet) {
fs.StringVar(&s.Master, "master", s.Master, "The address of the Kubernetes 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.RootCAFile, "root-ca-file", s.RootCAFile, "If set, this root certificate authority will be included in service account's token secret. This must be a valid PEM-encoded CA bundle.")
fs.BoolVar(&s.EnableExperimental, "enable-experimental", s.EnableExperimental, "Enables experimental controllers (requires enabling experimental API on apiserver).")
fs.Float32Var(&s.KubeApiQps, "kube-api-qps", s.KubeApiQps, "QPS to use while talking with kubernetes apiserver")
fs.IntVar(&s.KubeApiBurst, "kube-api-burst", s.KubeApiBurst, "Burst to use while talking with kubernetes apiserver")
}
@ -287,20 +286,47 @@ func (s *CMServer) Run(_ []string) error {
resourcequotacontroller.NewResourceQuotaController(kubeClient).Run(s.ResourceQuotaSyncPeriod)
namespacecontroller.NewNamespaceController(kubeClient, s.EnableExperimental, s.NamespaceSyncPeriod).Run()
versionStrings, err := client.ServerAPIVersions(kubeconfig)
if err != nil {
glog.Fatalf("Failed to get api versions from server: %v", err)
}
versions := &unversioned.APIVersions{Versions: versionStrings}
if s.EnableExperimental {
go daemon.NewDaemonSetsController(kubeClient, s.resyncPeriod).
Run(s.ConcurrentDSCSyncs, util.NeverStop)
resourceMap, err := client.SupportedResources(kubeClient, kubeconfig)
if err != nil {
glog.Fatalf("Failed to get supported resources from server: %v", err)
}
go job.NewJobController(kubeClient, s.resyncPeriod).
Run(s.ConcurrentJobSyncs, util.NeverStop)
namespacecontroller.NewNamespaceController(kubeClient, versions, s.NamespaceSyncPeriod).Run()
podautoscaler.NewHorizontalController(kubeClient, metrics.NewHeapsterMetricsClient(kubeClient)).
Run(s.HorizontalPodAutoscalerSyncPeriod)
groupVersion := "extensions/v1beta1"
resources, found := resourceMap[groupVersion]
// TODO: this needs to be dynamic so users don't have to restart their controller manager if they change the apiserver
if containsVersion(versions, groupVersion) && found {
glog.Infof("Starting %s apis", groupVersion)
if containsResource(resources, "horizontalpodautoscalers") {
glog.Infof("Starting horizontal pod controller.")
podautoscaler.NewHorizontalController(kubeClient, metrics.NewHeapsterMetricsClient(kubeClient)).
Run(s.HorizontalPodAutoscalerSyncPeriod)
}
deployment.New(kubeClient).
Run(s.DeploymentControllerSyncPeriod)
if containsResource(resources, "daemonsets") {
glog.Infof("Starting daemon set controller")
go daemon.NewDaemonSetsController(kubeClient, s.resyncPeriod).
Run(s.ConcurrentDSCSyncs, util.NeverStop)
}
if containsResource(resources, "jobs") {
glog.Infof("Starting job controller")
go job.NewJobController(kubeClient, s.resyncPeriod).
Run(s.ConcurrentJobSyncs, util.NeverStop)
}
if containsResource(resources, "deployments") {
glog.Infof("Starting deployment controller")
deployment.New(kubeClient).
Run(s.DeploymentControllerSyncPeriod)
}
}
pvclaimBinder := persistentvolumecontroller.NewPersistentVolumeClaimBinder(kubeClient, s.PVClaimBinderSyncPeriod)
@ -348,3 +374,22 @@ func (s *CMServer) Run(_ []string) error {
select {}
}
func containsVersion(versions *unversioned.APIVersions, version string) bool {
for ix := range versions.Versions {
if versions.Versions[ix] == version {
return true
}
}
return false
}
func containsResource(resources *unversioned.APIResourceList, resourceName string) bool {
for ix := range resources.APIResources {
resource := resources.APIResources[ix]
if resource.Name == resourceName {
return true
}
}
return false
}

3
contrib/mesos/pkg/controllermanager/controllermanager.go
View File

@ -26,6 +26,7 @@ import (
"time"
"k8s.io/kubernetes/cmd/kube-controller-manager/app"
"k8s.io/kubernetes/pkg/api/unversioned"
client "k8s.io/kubernetes/pkg/client/unversioned"
"k8s.io/kubernetes/pkg/client/unversioned/clientcmd"
clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
@ -156,7 +157,7 @@ func (s *CMServer) Run(_ []string) error {
resourceQuotaController := resourcequotacontroller.NewResourceQuotaController(kubeClient)
resourceQuotaController.Run(s.ResourceQuotaSyncPeriod)
namespaceController := namespacecontroller.NewNamespaceController(kubeClient, false, s.NamespaceSyncPeriod)
namespaceController := namespacecontroller.NewNamespaceController(kubeClient, &unversioned.APIVersions{}, s.NamespaceSyncPeriod)
namespaceController.Run()
pvclaimBinder := persistentvolumecontroller.NewPersistentVolumeClaimBinder(kubeClient, s.PVClaimBinderSyncPeriod)

1
hack/verify-flags/known-flags.txt
View File

@ -71,7 +71,6 @@ duration-sec
e2e-verify-service-account
e2e-output-dir
enable-debugging-handlers
enable-experimental
enable-server
etcd-config
etcd-prefix

42
pkg/client/unversioned/client.go
View File

@ -52,6 +52,7 @@ type Interface interface {
ComponentStatusesInterface
SwaggerSchemaInterface
Extensions() ExtensionsInterface
ResourcesInterface
}
func (c *Client) ReplicationControllers(namespace string) ReplicationControllerInterface {
@ -119,6 +120,11 @@ type VersionInterface interface {
ServerAPIVersions() (*unversioned.APIVersions, error)
}
// ResourcesInterface has methods for obtaining supported resources on the API server
type ResourcesInterface interface {
SupportedResourcesForGroupVersion(groupVersion string) (*unversioned.APIResourceList, error)
}
// APIStatus is exposed by errors that can be converted to an api.Status object
// for finer grained details.
type APIStatus interface {
@ -145,6 +151,42 @@ func (c *Client) ServerVersion() (*version.Info, error) {
return &info, nil
}
// SupportedResourcesForGroupVersion retrieves the list of resources supported by the API server for a group version.
func (c *Client) SupportedResourcesForGroupVersion(groupVersion string) (*unversioned.APIResourceList, error) {
var prefix string
if groupVersion == "v1" {
prefix = "/api"
} else {
prefix = "/apis"
}
body, err := c.Get().AbsPath(prefix, groupVersion).Do().Raw()
if err != nil {
return nil, err
}
resources := unversioned.APIResourceList{}
if err := json.Unmarshal(body, &resources); err != nil {
return nil, err
}
return &resources, nil
}
// SupportedResources gets all supported resources for all group versions. The key in the map is an API groupVersion.
func SupportedResources(c Interface, cfg *Config) (map[string]*unversioned.APIResourceList, error) {
apis, err := ServerAPIVersions(cfg)
if err != nil {
return nil, err
}
result := map[string]*unversioned.APIResourceList{}
for _, groupVersion := range apis {
resources, err := c.SupportedResourcesForGroupVersion(groupVersion)
if err != nil {
return nil, err
}
result[groupVersion] = resources
}
return result, nil
}
// ServerAPIVersions retrieves and parses the list of API versions the server supports.
func (c *Client) ServerAPIVersions() (*unversioned.APIVersions, error) {
body, err := c.Get().UnversionedPath("").Do().Raw()

108
pkg/client/unversioned/client_test.go
View File

@ -271,6 +271,114 @@ func TestGetServerVersion(t *testing.T) {
}
}
func TestGetServerResources(t *testing.T) {
stable := unversioned.APIResourceList{
GroupVersion: "v1",
APIResources: []unversioned.APIResource{
{"pods", true},
{"services", true},
{"namespaces", false},
},
}
beta := unversioned.APIResourceList{
GroupVersion: "extensions/v1",
APIResources: []unversioned.APIResource{
{"deployments", true},
{"ingress", true},
{"jobs", true},
},
}
tests := []struct {
resourcesList *unversioned.APIResourceList
path string
request string
expectErr bool
}{
{
resourcesList: &stable,
path: "/api/v1",
request: "v1",
expectErr: false,
},
{
resourcesList: &beta,
path: "/apis/extensions/v1beta1",
request: "extensions/v1beta1",
expectErr: false,
},
{
resourcesList: &stable,
path: "/api/v1",
request: "foobar",
expectErr: true,
},
}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
var list interface{}
switch req.URL.Path {
case "/api/v1":
list = &stable
case "/apis/extensions/v1beta1":
list = &beta
case "/api":
list = &unversioned.APIVersions{
Versions: []string{
"v1",
},
}
case "/apis":
list = &unversioned.APIGroupList{
Groups: []unversioned.APIGroup{
{
Versions: []unversioned.GroupVersion{
{GroupVersion: "extensions/v1beta1"},
},
},
},
}
default:
t.Logf("unexpected request: %s", req.URL.Path)
w.WriteHeader(http.StatusNotFound)
return
}
output, err := json.Marshal(list)
if err != nil {
t.Errorf("unexpected encoding error: %v", err)
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.Write(output)
}))
client := NewOrDie(&Config{Host: server.URL})
for _, test := range tests {
got, err := client.SupportedResourcesForGroupVersion(test.request)
if test.expectErr {
if err == nil {
t.Error("unexpected non-error")
}
continue
}
if err != nil {
t.Errorf("unexpected error: %v", err)
continue
}
if !reflect.DeepEqual(got, test.resourcesList) {
t.Errorf("expected:\n%v\ngot:\n%v\n", test.resourcesList, got)
}
}
resourceMap, err := SupportedResources(client, &Config{Host: server.URL})
if err != nil {
t.Errorf("unexpected error: %v", err)
}
for _, api := range []string{"v1", "extensions/v1beta1"} {
if _, found := resourceMap[api]; !found {
t.Errorf("missing expected api: %s", api)
}
}
}
func TestGetServerAPIVersions(t *testing.T) {
versions := []string{"v1", "v2", "v3"}
expect := unversioned.APIVersions{Versions: versions}

16
pkg/client/unversioned/testclient/testclient.go
View File

@ -60,6 +60,8 @@ type Fake struct {
WatchReactionChain []WatchReactor
// ProxyReactionChain is the list of proxy reactors that will be attempted for every request in the order they are tried
ProxyReactionChain []ProxyReactor
Resources []unversioned.APIResourceList
}
// Reactor is an interface to allow the composition of reaction functions.
@ -272,6 +274,20 @@ func (c *Fake) Extensions() client.ExtensionsInterface {
return &FakeExperimental{c}
}
func (c *Fake) SupportedResourcesForGroupVersion(version string) (*unversioned.APIResourceList, error) {
action := ActionImpl{
Verb: "get",
Resource: "resource",
}
c.Invokes(action, nil)
for _, resource := range c.Resources {
if resource.GroupVersion == version {
return &resource, nil
}
}
return nil, nil
}
func (c *Fake) ServerVersion() (*version.Info, error) {
action := ActionImpl{}
action.Verb = "get"

78
pkg/controller/namespace/namespace_controller.go
View File

@ -43,7 +43,7 @@ type NamespaceController struct {
}
// NewNamespaceController creates a new NamespaceController
func NewNamespaceController(kubeClient client.Interface, experimentalMode bool, resyncPeriod time.Duration) *NamespaceController {
func NewNamespaceController(kubeClient client.Interface, versions *unversioned.APIVersions, resyncPeriod time.Duration) *NamespaceController {
var controller *framework.Controller
_, controller = framework.NewInformer(
&cache.ListWatch{
@ -60,7 +60,7 @@ func NewNamespaceController(kubeClient client.Interface, experimentalMode bool,
framework.ResourceEventHandlerFuncs{
AddFunc: func(obj interface{}) {
namespace := obj.(*api.Namespace)
if err := syncNamespace(kubeClient, experimentalMode, namespace); err != nil {
if err := syncNamespace(kubeClient, versions, namespace); err != nil {
if estimate, ok := err.(*contentRemainingError); ok {
go func() {
// Estimate is the aggregate total of TerminationGracePeriodSeconds, which defaults to 30s
@ -82,7 +82,7 @@ func NewNamespaceController(kubeClient client.Interface, experimentalMode bool,
},
UpdateFunc: func(oldObj, newObj interface{}) {
namespace := newObj.(*api.Namespace)
if err := syncNamespace(kubeClient, experimentalMode, namespace); err != nil {
if err := syncNamespace(kubeClient, versions, namespace); err != nil {
if estimate, ok := err.(*contentRemainingError); ok {
go func() {
t := estimate.Estimate/2 + 1
@ -155,7 +155,7 @@ func (e *contentRemainingError) Error() string {
// deleteAllContent will delete all content known to the system in a namespace. It returns an estimate
// of the time remaining before the remaining resources are deleted. If estimate > 0 not all resources
// are guaranteed to be gone.
func deleteAllContent(kubeClient client.Interface, experimentalMode bool, namespace string, before unversioned.Time) (estimate int64, err error) {
func deleteAllContent(kubeClient client.Interface, versions *unversioned.APIVersions, namespace string, before unversioned.Time) (estimate int64, err error) {
err = deleteServiceAccounts(kubeClient, namespace)
if err != nil {
return estimate, err
@ -193,26 +193,40 @@ func deleteAllContent(kubeClient client.Interface, experimentalMode bool, namesp
return estimate, err
}
// If experimental mode, delete all experimental resources for the namespace.
if experimentalMode {
err = deleteHorizontalPodAutoscalers(kubeClient.Extensions(), namespace)
if containsVersion(versions, "extensions/v1beta1") {
resources, err := kubeClient.SupportedResourcesForGroupVersion("extensions/v1beta1")
if err != nil {
return estimate, err
}
err = deleteDaemonSets(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
if containsResource(resources, "horizontalpodautoscalers") {
err = deleteHorizontalPodAutoscalers(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
}
}
err = deleteJobs(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
if containsResource(resources, "ingress") {
err = deleteIngress(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
}
}
err = deleteDeployments(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
if containsResource(resources, "daemonsets") {
err = deleteDaemonSets(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
}
}
err = deleteIngress(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
if containsResource(resources, "jobs") {
err = deleteJobs(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
}
}
if containsResource(resources, "deployments") {
err = deleteDeployments(kubeClient.Extensions(), namespace)
if err != nil {
return estimate, err
}
}
}
return estimate, nil
@ -254,7 +268,7 @@ func updateNamespaceStatusFunc(kubeClient client.Interface, namespace *api.Names
}
// syncNamespace orchestrates deletion of a Namespace and its associated content.
func syncNamespace(kubeClient client.Interface, experimentalMode bool, namespace *api.Namespace) (err error) {
func syncNamespace(kubeClient client.Interface, versions *unversioned.APIVersions, namespace *api.Namespace) (err error) {
if namespace.DeletionTimestamp == nil {
return nil
}
@ -280,7 +294,7 @@ func syncNamespace(kubeClient client.Interface, experimentalMode bool, namespace
}
// there may still be content for us to remove
estimate, err := deleteAllContent(kubeClient, experimentalMode, namespace.Name, *namespace.DeletionTimestamp)
estimate, err := deleteAllContent(kubeClient, versions, namespace.Name, *namespace.DeletionTimestamp)
if err != nil {
return err
}
@ -515,3 +529,27 @@ func deleteIngress(expClient client.ExtensionsInterface, ns string) error {
}
return nil
}
// TODO: this is duplicated logic. Move it somewhere central?
func containsVersion(versions *unversioned.APIVersions, version string) bool {
for ix := range versions.Versions {
if versions.Versions[ix] == version {
return true
}
}
return false
}
// TODO: this is duplicated logic. Move it somewhere central?
func containsResource(resources *unversioned.APIResourceList, resourceName string) bool {
if resources == nil {
return false
}
for ix := range resources.APIResources {
resource := resources.APIResources[ix]
if resource.Name == resourceName {
return true
}
}
return false
}

38
pkg/controller/namespace/namespace_controller_test.go
View File

@ -73,7 +73,7 @@ func TestFinalizeNamespaceFunc(t *testing.T) {
}
}
func testSyncNamespaceThatIsTerminating(t *testing.T, experimentalMode bool) {
func testSyncNamespaceThatIsTerminating(t *testing.T, versions *unversioned.APIVersions) {
mockClient := &testclient.Fake{}
now := unversioned.Now()
testNamespace := &api.Namespace{
@ -89,7 +89,21 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, experimentalMode bool) {
Phase: api.NamespaceTerminating,
},
}
err := syncNamespace(mockClient, experimentalMode, testNamespace)
if containsVersion(versions, "extensions/v1beta1") {
resources := []unversioned.APIResource{}
for _, resource := range []string{"daemonsets", "deployments", "jobs", "horizontalpodautoscalers", "ingress"} {
resources = append(resources, unversioned.APIResource{Name: resource})
}
mockClient.Resources = []unversioned.APIResourceList{
{
GroupVersion: "extensions/v1beta1",
APIResources: resources,
},
}
}
err := syncNamespace(mockClient, versions, testNamespace)
if err != nil {
t.Errorf("Unexpected error when synching namespace %v", err)
}
@ -108,13 +122,14 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, experimentalMode bool) {
strings.Join([]string{"delete", "namespaces", ""}, "-"),
)
if experimentalMode {
if containsVersion(versions, "extensions/v1beta1") {
expectedActionSet.Insert(
strings.Join([]string{"list", "horizontalpodautoscalers", ""}, "-"),
strings.Join([]string{"list", "daemonsets", ""}, "-"),
strings.Join([]string{"list", "deployments", ""}, "-"),
strings.Join([]string{"list", "jobs", ""}, "-"),
strings.Join([]string{"list", "horizontalpodautoscalers", ""}, "-"),
strings.Join([]string{"list", "ingress", ""}, "-"),
strings.Join([]string{"get", "resource", ""}, "-"),
)
}
@ -123,10 +138,10 @@ func testSyncNamespaceThatIsTerminating(t *testing.T, experimentalMode bool) {
actionSet.Insert(strings.Join([]string{action.GetVerb(), action.GetResource(), action.GetSubresource()}, "-"))
}
if !actionSet.HasAll(expectedActionSet.List()...) {
t.Errorf("Expected actions: %v, but got: %v", expectedActionSet, actionSet)
t.Errorf("Expected actions:\n%v\n but got:\n%v\nDifference:\n%v", expectedActionSet, actionSet, expectedActionSet.Difference(actionSet))
}
if !expectedActionSet.HasAll(actionSet.List()...) {
t.Errorf("Expected actions: %v, but got: %v", expectedActionSet, actionSet)
t.Errorf("Expected actions:\n%v\n but got:\n%v\nDifference:\n%v", expectedActionSet, actionSet, actionSet.Difference(expectedActionSet))
}
}
@ -151,11 +166,11 @@ func TestRetryOnConflictError(t *testing.T) {
}
func TestSyncNamespaceThatIsTerminatingNonExperimental(t *testing.T) {
testSyncNamespaceThatIsTerminating(t, false)
testSyncNamespaceThatIsTerminating(t, &unversioned.APIVersions{})
}
func TestSyncNamespaceThatIsTerminatingExperimental(t *testing.T) {
testSyncNamespaceThatIsTerminating(t, true)
func TestSyncNamespaceThatIsTerminatingV1Beta1(t *testing.T) {
testSyncNamespaceThatIsTerminating(t, &unversioned.APIVersions{Versions: []string{"extensions/v1beta1"}})
}
func TestSyncNamespaceThatIsActive(t *testing.T) {
@ -172,7 +187,7 @@ func TestSyncNamespaceThatIsActive(t *testing.T) {
Phase: api.NamespaceActive,
},
}
err := syncNamespace(mockClient, false, testNamespace)
err := syncNamespace(mockClient, &unversioned.APIVersions{}, testNamespace)
if err != nil {
t.Errorf("Unexpected error when synching namespace %v", err)
}
@ -183,7 +198,8 @@ func TestSyncNamespaceThatIsActive(t *testing.T) {
func TestRunStop(t *testing.T) {
mockClient := &testclient.Fake{}
nsController := NewNamespaceController(mockClient, false, 1*time.Second)
nsController := NewNamespaceController(mockClient, &unversioned.APIVersions{}, 1*time.Second)
if nsController.StopEverything != nil {
t.Errorf("Non-running manager should not have a stop channel. Got %v", nsController.StopEverything)