Merge pull request kubernetes#3083 from lavalamp/optimize

Make list pods constant time

cmd/integration/integration.go

@@ -63,6 +63,7 @@ var (
 type fakeKubeletClient struct{}
 
 func (fakeKubeletClient) GetPodInfo(host, podNamespace, podID string) (api.PodContainerInfo, error) {
+	glog.V(3).Infof("Trying to get container info for %v/%v/%v", host, podNamespace, podID)
 	// This is a horrible hack to get around the fact that we can't provide
 	// different port numbers per kubelet...
 	var c client.PodInfoGetter

pkg/master/ip_cache.go

@@ -0,0 +1,90 @@
+/*
+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 master
+
+import (
+	"sync"
+	"time"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+	"github.com/golang/glog"
+)
+
+type ipCacheEntry struct {
+	ip         string
+	lastUpdate time.Time
+}
+
+type ipCache struct {
+	clock         util.Clock
+	cloudProvider cloudprovider.Interface
+	cache         map[string]ipCacheEntry
+	lock          sync.Mutex
+	ttl           time.Duration
+}
+
+// NewIPCache makes a new ip caching layer, which will get IP addresses from cp,
+// and use clock for deciding when to re-get an IP address.
+// Thread-safe.
+//
+// TODO: when we switch to go1.4, this class would be a good candidate for something
+// that could be produced from a template and a type via `go generate`.
+func NewIPCache(cp cloudprovider.Interface, clock util.Clock, ttl time.Duration) *ipCache {
+	return &ipCache{
+		clock:         clock,
+		cloudProvider: cp,
+		cache:         map[string]ipCacheEntry{},
+		ttl:           ttl,
+	}
+}
+
+// GetInstanceIP returns the IP address of host, from the cache
+// if possible, otherwise it asks the cloud provider.
+func (c *ipCache) GetInstanceIP(host string) string {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	data, ok := c.cache[host]
+	now := c.clock.Now()
+
+	if !ok || now.Sub(data.lastUpdate) > c.ttl {
+		ip := getInstanceIPFromCloud(c.cloudProvider, host)
+		data = ipCacheEntry{
+			ip:         ip,
+			lastUpdate: now,
+		}
+		c.cache[host] = data
+	}
+	return data.ip
+}
+
+func getInstanceIPFromCloud(cloud cloudprovider.Interface, host string) string {
+	if cloud == nil {
+		return ""
+	}
+	instances, ok := cloud.Instances()
+	if instances == nil || !ok {
+		return ""
+	}
+	addr, err := instances.IPAddress(host)
+	if err != nil {
+		glog.Errorf("Error getting instance IP for %q: %v", host, err)
+		return ""
+	}
+	return addr.String()
+}

pkg/master/ip_cache_test.go

@@ -0,0 +1,59 @@
+/*
+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 master
+
+import (
+	"testing"
+	"time"
+
+	fake_cloud "github.com/GoogleCloudPlatform/kubernetes/pkg/cloudprovider/fake"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+)
+
+func TestCacheExpire(t *testing.T) {
+	fakeCloud := &fake_cloud.FakeCloud{}
+	clock := &util.FakeClock{time.Now()}
+
+	c := NewIPCache(fakeCloud, clock, 60*time.Second)
+
+	_ = c.GetInstanceIP("foo")
+	// This call should hit the cache, so we expect no additional calls to the cloud
+	_ = c.GetInstanceIP("foo")
+	// Advance the clock, this call should miss the cache, so expect one more call.
+	clock.Time = clock.Time.Add(61 * time.Second)
+	_ = c.GetInstanceIP("foo")
+
+	if len(fakeCloud.Calls) != 2 || fakeCloud.Calls[1] != "ip-address" || fakeCloud.Calls[0] != "ip-address" {
+		t.Errorf("Unexpected calls: %+v", fakeCloud.Calls)
+	}
+}
+
+func TestCacheNotExpire(t *testing.T) {
+	fakeCloud := &fake_cloud.FakeCloud{}
+	clock := &util.FakeClock{time.Now()}
+
+	c := NewIPCache(fakeCloud, clock, 60*time.Second)
+
+	_ = c.GetInstanceIP("foo")
+	// This call should hit the cache, so we expect no additional calls to the cloud
+	clock.Time = clock.Time.Add(60 * time.Second)
+	_ = c.GetInstanceIP("foo")
+
+	if len(fakeCloud.Calls) != 1 || fakeCloud.Calls[0] != "ip-address" {
+		t.Errorf("Unexpected calls: %+v", fakeCloud.Calls)
+	}
+}

pkg/master/master.go

@@ -319,28 +319,24 @@ func makeMinionRegistry(c *Config) minion.Registry {
 
 // init initializes master.
 func (m *Master) init(c *Config) {
-	podCache := NewPodCache(c.KubeletClient, m.podRegistry)
-	go util.Forever(func() { podCache.UpdateAllContainers() }, time.Second*30)
-
 	var userContexts = handlers.NewUserRequestContext()
 	var authenticator = c.Authenticator
 
 	nodeRESTStorage := minion.NewREST(m.minionRegistry)
+	ipCache := NewIPCache(c.Cloud, util.RealClock{}, 30*time.Second)
+	podCache := NewPodCache(
+		ipCache,
+		c.KubeletClient,
+		RESTStorageToNodes(nodeRESTStorage).Nodes(),
+		m.podRegistry,
+	)
+	go util.Forever(func() { podCache.UpdateAllContainers() }, time.Second*30)
 
 	// TODO: Factor out the core API registration
 	m.storage = map[string]apiserver.RESTStorage{
 		"pods": pod.NewREST(&pod.RESTConfig{
-			CloudProvider: c.Cloud,
-			PodCache:      podCache,
-			PodInfoGetter: c.KubeletClient,
-			Registry:      m.podRegistry,
-			// Note: this allows the pod rest object to directly call
-			// the node rest object without going through the network &
-			// apiserver. This arrangement should be temporary, nodes
-			// shouldn't really need this at all. Once we add more auth in,
-			// we need to consider carefully if this sort of shortcut is a
-			// good idea.
-			Nodes: RESTStorageToNodes(nodeRESTStorage).Nodes(),
+			PodCache: podCache,
+			Registry: m.podRegistry,
 		}),
 		"replicationControllers": controller.NewREST(m.controllerRegistry, m.podRegistry),
 		"services":               service.NewREST(m.serviceRegistry, c.Cloud, m.minionRegistry, m.portalNet),