Limit relevant pods to those on the same node as kube2iam (#108)

In clusters with 1000's of pods the memory used by each kube2iam
instance explodes because each kube2iam instance is storing information
about all pods on all nodes in memory.

This adds a flag `--node` for specifying the node name of kube2iam. If
set, it will limit the pods it's watching to those only on the same
node as the kube2iam instance.

README.md

@@ -102,6 +102,12 @@ spec:
           name: kube2iam
           args:
             - "--base-role-arn=arn:aws:iam::123456789012:role/"
+            - "--node=$(NODE_NAME)"
+          env:
+            - name: NODE_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
           ports:
             - containerPort: 8181
               hostPort: 8181
@@ -159,11 +165,16 @@ spec:
             - "--base-role-arn=arn:aws:iam::123456789012:role/"
             - "--iptables=true"
             - "--host-ip=$(HOST_IP)"
+            - "--node=$(NODE_NAME)"
           env:
             - name: HOST_IP
               valueFrom:
                 fieldRef:
                   fieldPath: status.podIP
+            - name: NODE_NAME
+              valueFrom:
+                fieldRef:
+                  fieldPath: spec.nodeName
           ports:
             - containerPort: 8181
               hostPort: 8181

cmd/main.go

@@ -30,6 +30,7 @@ func addFlags(s *server.Server, fs *pflag.FlagSet) {
 	fs.BoolVar(&s.NamespaceRestriction, "namespace-restrictions", false, "Enable namespace restrictions")
 	fs.StringVar(&s.NamespaceKey, "namespace-key", s.NamespaceKey, "Namespace annotation key used to retrieve the IAM roles allowed (value in annotation should be json array)")
 	fs.StringVar(&s.HostIP, "host-ip", s.HostIP, "IP address of host")
+	fs.StringVar(&s.NodeName, "node", s.NodeName, "Name of the node where kube2iam is running")
 	fs.DurationVar(&s.BackoffMaxInterval, "backoff-max-interval", s.BackoffMaxInterval, "Max interval for backoff when querying for role.")
 	fs.DurationVar(&s.BackoffMaxElapsedTime, "backoff-max-elapsed-time", s.BackoffMaxElapsedTime, "Max elapsed time for backoff when querying for role.")
 	fs.StringVar(&s.LogFormat, "log-format", s.LogFormat, "Log format (text/json)")
@@ -106,7 +107,7 @@ func main() {
 		}
 	}
 
-	if err := s.Run(s.APIServer, s.APIToken, s.Insecure); err != nil {
+	if err := s.Run(s.APIServer, s.APIToken, s.NodeName, s.Insecure); err != nil {
 		log.Fatalf("%s", err)
 	}
 }

k8s/k8s.go

@@ -27,11 +27,16 @@ type Client struct {
 	namespaceIndexer    cache.Indexer
 	podController       *cache.Controller
 	podIndexer          cache.Indexer
+	nodeName            string
 }
 
 // Returns a cache.ListWatch that gets all changes to pods.
 func (k8s *Client) createPodLW() *cache.ListWatch {
-	return cache.NewListWatchFromClient(k8s.CoreV1().RESTClient(), "pods", v1.NamespaceAll, selector.Everything())
+	fieldSelector := selector.Everything()
+	if k8s.nodeName != "" {
+		fieldSelector = selector.OneTermEqualSelector("spec.nodeName", k8s.nodeName)
+	}
+	return cache.NewListWatchFromClient(k8s.CoreV1().RESTClient(), "pods", v1.NamespaceAll, fieldSelector)
 }
 
 // WatchForPods watches for pod changes.
@@ -119,7 +124,7 @@ func (k8s *Client) NamespaceByName(namespaceName string) (*v1.Namespace, error)
 }
 
 // NewClient returns a new kubernetes client.
-func NewClient(host, token string, insecure bool) (*Client, error) {
+func NewClient(host, token, nodeName string, insecure bool) (*Client, error) {
 	var config *rest.Config
 	var err error
 	if host != "" && token != "" {
@@ -138,5 +143,5 @@ func NewClient(host, token string, insecure bool) (*Client, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Client{Clientset: client}, nil
+	return &Client{Clientset: client, nodeName: nodeName}, nil
 }

server/server.go

@@ -47,6 +47,7 @@ type Server struct {
 	MetadataAddress         string
 	HostInterface           string
 	HostIP                  string
+	NodeName                string
 	NamespaceKey            string
 	LogLevel                string
 	LogFormat               string
@@ -261,8 +262,8 @@ func write(logger *log.Entry, w http.ResponseWriter, s string) {
 }
 
 // Run runs the specified Server.
-func (s *Server) Run(host, token string, insecure bool) error {
-	k, err := k8s.NewClient(host, token, insecure)
+func (s *Server) Run(host, token, nodeName string, insecure bool) error {
+	k, err := k8s.NewClient(host, token, nodeName, insecure)
 	if err != nil {
 		return err
 	}