Skip to content

K
k3s
Commit ac7bf050 authored by Eric Tune's avatar Eric Tune
Browse files
Options

Kubelet has not even heard of etcd.

parent 7d53425b
Hide whitespace changes
Inline Side-by-side
Showing with 4 additions and 277 deletions
cmd/integration/integration.go
View file @ ac7bf050
......@@ -214,13 +214,13 @@ func startComponents(manifestURL string) (apiServerURL string) {
// Kubelet (localhost)
testRootDir := makeTempDirOrDie("kubelet_integ_1.")
glog.Infof("Using %s as root dir for kubelet #1", testRootDir)
kubeletapp.SimpleRunKubelet(cl, nil, &fakeDocker1, machineList[0], testRootDir, manifestURL, "127.0.0.1", 10250, api.NamespaceDefault, empty_dir.ProbeVolumePlugins(), nil)
kubeletapp.SimpleRunKubelet(cl, &fakeDocker1, machineList[0], testRootDir, manifestURL, "127.0.0.1", 10250, api.NamespaceDefault, empty_dir.ProbeVolumePlugins(), nil)
// Kubelet (machine)
// Create a second kubelet so that the guestbook example's two redis slaves both
// have a place they can schedule.
testRootDir = makeTempDirOrDie("kubelet_integ_2.")
glog.Infof("Using %s as root dir for kubelet #2", testRootDir)
kubeletapp.SimpleRunKubelet(cl, nil, &fakeDocker2, machineList[1], testRootDir, "", "127.0.0.1", 10251, api.NamespaceDefault, empty_dir.ProbeVolumePlugins(), nil)
kubeletapp.SimpleRunKubelet(cl, &fakeDocker2, machineList[1], testRootDir, "", "127.0.0.1", 10251, api.NamespaceDefault, empty_dir.ProbeVolumePlugins(), nil)
return apiServer.URL
}
......
cmd/kubelet/app/server.go
View file @ ac7bf050
......@@ -36,7 +36,6 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/volume"
"github.com/GoogleCloudPlatform/kubernetes/pkg/master/ports"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/golang/glog"
......@@ -231,7 +230,6 @@ func (s *KubeletServer) createAPIServerClient() (*client.Client, error) {
// SimpleRunKubelet is a simple way to start a Kubelet talking to dockerEndpoint, using an etcdClient.
// Under the hood it calls RunKubelet (below)
func SimpleRunKubelet(client *client.Client,
etcdClient tools.EtcdClient,
dockerClient dockertools.DockerInterface,
hostname, rootDir, manifestURL, address string,
port uint,
......@@ -240,7 +238,6 @@ func SimpleRunKubelet(client *client.Client,
tlsOptions *kubelet.TLSOptions) {
kcfg := KubeletConfig{
KubeClient: client,
EtcdClient: etcdClient,
DockerClient: dockerClient,
HostnameOverride: hostname,
RootDirectory: rootDir,
......@@ -321,10 +318,6 @@ func makePodSourceConfig(kc *KubeletConfig) *config.PodConfig {
glog.Infof("Adding manifest url: %v", kc.ManifestURL)
config.NewSourceURL(kc.ManifestURL, kc.HTTPCheckFrequency, cfg.Channel(kubelet.HTTPSource))
}
if kc.EtcdClient != nil {
glog.Infof("Watching for etcd configs at %v", kc.EtcdClient.GetCluster())
config.NewSourceEtcd(config.EtcdKeyForHost(kc.Hostname), kc.EtcdClient, cfg.Channel(kubelet.EtcdSource))
}
if kc.KubeClient != nil {
glog.Infof("Watching apiserver")
config.NewSourceApiserver(kc.KubeClient, kc.Hostname, cfg.Channel(kubelet.ApiserverSource))
......@@ -335,7 +328,6 @@ func makePodSourceConfig(kc *KubeletConfig) *config.PodConfig {
// KubeletConfig is all of the parameters necessary for running a kubelet.
// TODO: This should probably be merged with KubeletServer. The extra object is a consequence of refactoring.
type KubeletConfig struct {
EtcdClient tools.EtcdClient
KubeClient *client.Client
DockerClient dockertools.DockerInterface
CAdvisorPort uint
......@@ -392,7 +384,6 @@ func createAndInitKubelet(kc *KubeletConfig, pc *config.PodConfig) (*kubelet.Kub
k, err := kubelet.NewMainKubelet(
kc.Hostname,
kc.DockerClient,
kc.EtcdClient,
kubeClient,
kc.RootDirectory,
kc.PodInfraContainerImage,
......
cmd/kubernetes/kubernetes.go
View file @ ac7bf050
......@@ -144,7 +144,7 @@ func startComponents(etcdClient tools.EtcdClient, cl *client.Client, addr net.IP
runControllerManager(machineList, cl, *nodeMilliCPU, *nodeMemory)
dockerClient := dockertools.ConnectToDockerOrDie(*dockerEndpoint)
kubeletapp.SimpleRunKubelet(cl, nil, dockerClient, machineList[0], "/tmp/kubernetes", "", "127.0.0.1", 10250, *masterServiceNamespace, kubeletapp.ProbeVolumePlugins(), nil)
kubeletapp.SimpleRunKubelet(cl, dockerClient, machineList[0], "/tmp/kubernetes", "", "127.0.0.1", 10250, *masterServiceNamespace, kubeletapp.ProbeVolumePlugins(), nil)
}
func newApiClient(addr net.IP, port int) *client.Client {
......
pkg/kubelet/config/etcd.go deleted 100644 → 0
View file @ 7d53425b
/*
Copyright 2014 Google Inc. 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.
*/
// Reads the pod configuration from etcd using the Kubernetes etcd schema.
package config
import (
"errors"
"path"
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/golang/glog"
)
func EtcdKeyForHost(hostname string) string {
return path.Join("/", "registry", "nodes", hostname, "boundpods")
}
type sourceEtcd struct {
key string
helper tools.EtcdHelper
updates chan<- interface{}
}
// NewSourceEtcd creates a config source that watches and pulls from a key in etcd
func NewSourceEtcd(key string, client tools.EtcdClient, updates chan<- interface{}) {
helper := tools.EtcdHelper{
client,
latest.Codec,
tools.RuntimeVersionAdapter{latest.ResourceVersioner},
}
source := &sourceEtcd{
key: key,
helper: helper,
updates: updates,
}
glog.V(1).Infof("Watching etcd for %s", key)
go util.Forever(source.run, time.Second)
}
func (s *sourceEtcd) run() {
boundPods := api.BoundPods{}
err := s.helper.ExtractObj(s.key, &boundPods, false)
if err != nil {
if tools.IsEtcdNotFound(err) {
glog.V(4).Infof("etcd failed to retrieve the value for the key %q. Error: %v", s.key, err)
return
}
glog.Errorf("etcd failed to retrieve the value for the key %q. Error: %v", s.key, err)
return
}
// Push update. Maybe an empty PodList to allow EtcdSource to be marked as seen
s.updates <- kubelet.PodUpdate{boundPods.Items, kubelet.SET, kubelet.EtcdSource}
index, _ := s.helper.ResourceVersioner.ResourceVersion(&boundPods)
watching := s.helper.Watch(s.key, index)
for {
select {
case event, ok := <-watching.ResultChan():
if !ok {
return
}
if event.Type == watch.Error {
glog.Infof("Watch closed (%#v). Reopening.", event.Object)
watching.Stop()
return
}
pods, err := eventToPods(event)
if err != nil {
glog.Errorf("Failed to parse result from etcd watch: %v", err)
continue
}
glog.V(4).Infof("Received state from etcd watch: %+v", pods)
s.updates <- kubelet.PodUpdate{pods, kubelet.SET, kubelet.EtcdSource}
}
}
}
// eventToPods takes a watch.Event object, and turns it into a structured list of pods.
// It returns a list of containers, or an error if one occurs.
func eventToPods(ev watch.Event) ([]api.BoundPod, error) {
pods := []api.BoundPod{}
if ev.Object == nil {
return pods, nil
}
boundPods, ok := ev.Object.(*api.BoundPods)
if !ok {
return pods, errors.New("unable to parse response as BoundPods")
}
for _, pod := range boundPods.Items {
// Always overrides the namespace provided by the etcd event.
pod.Namespace = kubelet.NamespaceDefault
pods = append(pods, pod)
}
return pods, nil
}
pkg/kubelet/config/etcd_test.go deleted 100644 → 0
View file @ 7d53425b
/*
Copyright 2014 Google Inc. 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 config
import (
"reflect"
"testing"
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api/latest"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
func TestEtcdSourceExistingBoundPods(t *testing.T) {
// Arrange
key := "/registry/nodes/machine/boundpods"
fakeEtcdClient := tools.NewFakeEtcdClient(t)
updates := make(chan interface{})
fakeEtcdClient.Set(
key,
runtime.EncodeOrDie(latest.Codec, &api.BoundPods{
Items: []api.BoundPod{
{
ObjectMeta: api.ObjectMeta{
Name: "foo",
Namespace: "default"},
Spec: api.PodSpec{
Containers: []api.Container{
{
Image: "foo:v1",
}}}},
{
ObjectMeta: api.ObjectMeta{
Name: "bar",
Namespace: "default"},
Spec: api.PodSpec{
Containers: []api.Container{
{
Image: "foo:v1",
}}}}}}),
0)
// Act
NewSourceEtcd(key, fakeEtcdClient, updates)
// Assert
select {
case got := <-updates:
update := got.(kubelet.PodUpdate)
if len(update.Pods) != 2 ||
update.Pods[0].ObjectMeta.Name != "foo" ||
update.Pods[1].ObjectMeta.Name != "bar" {
t.Errorf("Unexpected update response: %#v", update)
}
case <-time.After(200 * time.Millisecond):
t.Errorf("Expected update, timeout instead")
}
}
func TestEventToPods(t *testing.T) {
tests := []struct {
input watch.Event
pods []api.BoundPod
fail bool
}{
{
input: watch.Event{Object: nil},
pods: []api.BoundPod{},
fail: false,
},
{
input: watch.Event{Object: &api.BoundPods{}},
pods: []api.BoundPod{},
fail: false,
},
{
input: watch.Event{
Object: &api.BoundPods{
Items: []api.BoundPod{
{ObjectMeta: api.ObjectMeta{UID: "111", Name: "foo", Namespace: "foo"}},
{ObjectMeta: api.ObjectMeta{UID: "222", Name: "bar", Namespace: "bar"}},
},
},
},
pods: []api.BoundPod{
{
ObjectMeta: api.ObjectMeta{UID: "111", Name: "foo", Namespace: kubelet.NamespaceDefault},
Spec: api.PodSpec{},
},
{
ObjectMeta: api.ObjectMeta{UID: "222", Name: "bar", Namespace: kubelet.NamespaceDefault},
Spec: api.PodSpec{},
},
},
fail: false,
},
{
input: watch.Event{
Object: &api.BoundPods{
Items: []api.BoundPod{
{ObjectMeta: api.ObjectMeta{UID: "111", Name: "foo"}},
},
},
},
pods: []api.BoundPod{
{
ObjectMeta: api.ObjectMeta{UID: "111", Name: "foo", Namespace: kubelet.NamespaceDefault},
Spec: api.PodSpec{}},
},
fail: false,
},
}
for i, tt := range tests {
pods, err := eventToPods(tt.input)
if !reflect.DeepEqual(tt.pods, pods) {
t.Errorf("case %d: expected output %#v, got %#v", i, tt.pods, pods)
}
if tt.fail != (err != nil) {
t.Errorf("case %d: got fail=%t but err=%v", i, tt.fail, err)
}
}
}
pkg/kubelet/kubelet.go
View file @ ac7bf050
......@@ -45,7 +45,6 @@ import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/probe"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
utilErrors "github.com/GoogleCloudPlatform/kubernetes/pkg/util/errors"
......@@ -93,7 +92,6 @@ type volumeMap map[string]volume.Interface
func NewMainKubelet(
hostname string,
dockerClient dockertools.DockerInterface,
etcdClient tools.EtcdClient,
kubeClient client.Interface,
rootDirectory string,
podInfraContainerImage string,
......@@ -156,7 +154,6 @@ func NewMainKubelet(
klet := &Kubelet{
hostname: hostname,
dockerClient: dockerClient,
etcdClient: etcdClient,
kubeClient: kubeClient,
rootDirectory: rootDirectory,
resyncInterval: resyncInterval,
......@@ -232,8 +229,6 @@ type Kubelet struct {
dockerIDToRef map[dockertools.DockerID]*api.ObjectReference
refLock sync.RWMutex
// Optional, no events will be sent without it
etcdClient tools.EtcdClient
// Optional, defaults to simple Docker implementation
dockerPuller dockertools.DockerPuller
// Optional, defaults to /logs/ from /var/log
......
pkg/kubelet/types.go
View file @ ac7bf050
......@@ -41,7 +41,7 @@ const (
// Updates from a file
FileSource = "file"
// Updates from etcd
EtcdSource = "etcd"
EtcdSource = "etcd" // Not supported by current kubelets.
// Updates from querying a web page
HTTPSource = "http"
// Updates received to the kubelet server
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment