Merge pull request #5704 from wojtek-t/kubelet_file_read_pod

pkg/kubelet/config/common.go

+/*
+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.
+*/
+
+// Common logic used by both http and file channels.
+package config
+
+import (
+	"crypto/md5"
+	"encoding/hex"
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/v1beta1"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
+	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+
+	"github.com/ghodss/yaml"
+	"github.com/golang/glog"
+)
+
+func applyDefaults(pod *api.Pod, source string, isFile bool) error {
+	if len(pod.UID) == 0 {
+		hasher := md5.New()
+		if isFile {
+			hostname, err := os.Hostname() // TODO: kubelet name would be better
+			if err != nil {
+				return err
+			}
+			hostname = strings.ToLower(hostname)
+			fmt.Fprintf(hasher, "host:%s", hostname)
+			fmt.Fprintf(hasher, "file:%s", source)
+		} else {
+			fmt.Fprintf(hasher, "url:%s", source)
+		}
+		util.DeepHashObject(hasher, pod)
+		pod.UID = types.UID(hex.EncodeToString(hasher.Sum(nil)[0:]))
+		glog.V(5).Infof("Generated UID %q pod %q from %s", pod.UID, pod.Name, source)
+	}
+
+	// This is required for backward compatibility, and should be removed once we
+	// completely deprecate ContainerManifest.
+	var err error
+	if len(pod.Name) == 0 {
+		pod.Name = string(pod.UID)
+	}
+	if pod.Name, err = GeneratePodName(pod.Name); err != nil {
+		return err
+	}
+	glog.V(5).Infof("Generated Name %q for UID %q from URL %s", pod.Name, pod.UID, source)
+
+	if pod.Namespace == "" {
+		pod.Namespace = kubelet.NamespaceDefault
+	}
+	glog.V(5).Infof("Using namespace %q for pod %q from %s", pod.Namespace, pod.Name, source)
+
+	// Currently just simply follow the same format in resthandler.go
+	pod.ObjectMeta.SelfLink =
+		fmt.Sprintf("/api/v1beta2/pods/%s?namespace=%s", pod.Name, pod.Namespace)
+	return nil
+}
+
+func tryDecodeSinglePod(data []byte, source string, isFile bool) (parsed bool, pod api.Pod, err error) {
+	obj, err := api.Scheme.Decode(data)
+	if err != nil {
+		return false, pod, err
+	}
+	// Check whether the object could be converted to single pod.
+	if _, ok := obj.(*api.Pod); !ok {
+		err = fmt.Errorf("invalid pod: %+v", obj)
+		return false, pod, err
+	}
+	newPod := obj.(*api.Pod)
+	// Apply default values and validate the pod.
+	if err = applyDefaults(newPod, source, isFile); err != nil {
+		return true, pod, err
+	}
+	if errs := validation.ValidatePod(newPod); len(errs) > 0 {
+		err = fmt.Errorf("invalid pod: %v", errs)
+		return true, pod, err
+	}
+	return true, *newPod, nil
+}
+
+func tryDecodePodList(data []byte, source string, isFile bool) (parsed bool, pods api.PodList, err error) {
+	obj, err := api.Scheme.Decode(data)
+	if err != nil {
+		return false, pods, err
+	}
+	// Check whether the object could be converted to list of pods.
+	if _, ok := obj.(*api.PodList); !ok {
+		err = fmt.Errorf("invalid pods list: %+v", obj)
+		return false, pods, err
+	}
+	newPods := obj.(*api.PodList)
+	// Apply default values and validate pods.
+	for i := range newPods.Items {
+		newPod := &newPods.Items[i]
+		if err = applyDefaults(newPod, source, isFile); err != nil {
+			return true, pods, err
+		}
+		if errs := validation.ValidatePod(newPod); len(errs) > 0 {
+			err = fmt.Errorf("invalid pod: %v", errs)
+			return true, pods, err
+		}
+	}
+	return true, *newPods, err
+}
+
+func tryDecodeSingleManifest(data []byte) (parsed bool, manifest v1beta1.ContainerManifest, pod api.Pod, err error) {
+	// TODO: should be api.Scheme.Decode
+	// This is awful.  DecodeInto() expects to find an APIObject, which
+	// Manifest is not.  We keep reading manifest for now for compat, but
+	// we will eventually change it to read Pod (at which point this all
+	// becomes nicer).  Until then, we assert that the ContainerManifest
+	// structure on disk is always v1beta1.  Read that, convert it to a
+	// "current" ContainerManifest (should be ~identical), then convert
+	// that to a Pod (which is a well-understood conversion).  This
+	// avoids writing a v1beta1.ContainerManifest -> api.Pod
+	// conversion which would be identical to the api.ContainerManifest ->
+	// api.Pod conversion.
+	if err = yaml.Unmarshal(data, &manifest); err != nil {
+		return false, manifest, pod, err
+	}
+	newManifest := api.ContainerManifest{}
+	if err = api.Scheme.Convert(&manifest, &newManifest); err != nil {
+		return false, manifest, pod, err
+	}
+	if errs := validation.ValidateManifest(&newManifest); len(errs) > 0 {
+		err = fmt.Errorf("invalid manifest: %v", errs)
+		return false, manifest, pod, err
+	}
+	if err = api.Scheme.Convert(&newManifest, &pod); err != nil {
+		return true, manifest, pod, err
+	}
+	// Success.
+	return true, manifest, pod, nil
+}
+
+func tryDecodeManifestList(data []byte) (parsed bool, manifests []v1beta1.ContainerManifest, pods api.PodList, err error) {
+	// TODO: should be api.Scheme.Decode
+	// See the comment in tryDecodeSingle().
+	if err = yaml.Unmarshal(data, &manifests); err != nil {
+		return false, manifests, pods, err
+	}
+	newManifests := []api.ContainerManifest{}
+	if err = api.Scheme.Convert(&manifests, &newManifests); err != nil {
+		return false, manifests, pods, err
+	}
+	for i := range newManifests {
+		manifest := &newManifests[i]
+		if errs := validation.ValidateManifest(manifest); len(errs) > 0 {
+			err = fmt.Errorf("invalid manifest: %v", errs)
+			return false, manifests, pods, err
+		}
+	}
+	list := api.ContainerManifestList{Items: newManifests}
+	if err = api.Scheme.Convert(&list, &pods); err != nil {
+		return true, manifests, pods, err
+	}
+	// Success.
+	return true, manifests, pods, nil
+}

pkg/kubelet/config/file.go

@@ -18,23 +18,17 @@ limitations under the License.
 package config
 
 import (
-	"crypto/md5"
-	"encoding/hex"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sort"
-	"strings"
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/v1beta1"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 
-	"github.com/ghodss/yaml"
 	"github.com/golang/glog"
 )
 
@@ -131,9 +125,7 @@ func extractFromDir(name string) ([]api.Pod, error) {
 	return pods, nil
 }
 
-func extractFromFile(filename string) (api.Pod, error) {
-	var pod api.Pod
-
+func extractFromFile(filename string) (pod api.Pod, err error) {
 	glog.V(3).Infof("Reading config file %q", filename)
 	file, err := os.Open(filename)
 	if err != nil {
@@ -146,65 +138,28 @@ func extractFromFile(filename string) (api.Pod, error) {
 		return pod, err
 	}
 
-	// TODO: use api.Scheme.DecodeInto
-	// This is awful.  DecodeInto() expects to find an APIObject, which
-	// Manifest is not.  We keep reading manifest for now for compat, but
-	// we will eventually change it to read Pod (at which point this all
-	// becomes nicer).  Until then, we assert that the ContainerManifest
-	// structure on disk is always v1beta1.  Read that, convert it to a
-	// "current" ContainerManifest (should be ~identical), then convert
-	// that to a Pod (which is a well-understood conversion).  This
-	// avoids writing a v1beta1.ContainerManifest -> api.Pod
-	// conversion which would be identical to the api.ContainerManifest ->
-	// api.Pod conversion.
-	oldManifest := &v1beta1.ContainerManifest{}
-	if err := yaml.Unmarshal(data, oldManifest); err != nil {
-		return pod, fmt.Errorf("can't unmarshal file %q: %v", filename, err)
-	}
-	newManifest := &api.ContainerManifest{}
-	if err := api.Scheme.Convert(oldManifest, newManifest); err != nil {
-		return pod, fmt.Errorf("can't convert pod from file %q: %v", filename, err)
-	}
-	if err := api.Scheme.Convert(newManifest, &pod); err != nil {
-		return pod, fmt.Errorf("can't convert pod from file %q: %v", filename, err)
+	parsed, _, pod, manifestErr := tryDecodeSingleManifest(data)
+	if parsed {
+		if manifestErr != nil {
+			// It parsed but could not be used.
+			return pod, manifestErr
+		}
+		// It parsed!
+		if err = applyDefaults(&pod, filename, true); err != nil {
+			return pod, err
+		}
+		return pod, nil
 	}
 
-	hostname, err := os.Hostname() //TODO: kubelet name would be better
-	if err != nil {
-		return pod, err
-	}
-	hostname = strings.ToLower(hostname)
-
-	if len(pod.UID) == 0 {
-		hasher := md5.New()
-		fmt.Fprintf(hasher, "host:%s", hostname)
-		fmt.Fprintf(hasher, "file:%s", filename)
-		util.DeepHashObject(hasher, pod)
-		pod.UID = types.UID(hex.EncodeToString(hasher.Sum(nil)[0:]))
-		glog.V(5).Infof("Generated UID %q for pod %q from file %s", pod.UID, pod.Name, filename)
-	}
-	// This is required for backward compatibility, and should be removed once we
-	// completely deprecate ContainerManifest.
-	if len(pod.Name) == 0 {
-		pod.Name = string(pod.UID)
-	}
-	if pod.Name, err = GeneratePodName(pod.Name); err != nil {
-		return pod, err
+	parsed, pod, podErr := tryDecodeSinglePod(data, filename, true)
+	if parsed {
+		if podErr != nil {
+			return pod, podErr
+		}
+		return pod, nil
 	}
-	glog.V(5).Infof("Generated Name %q for UID %q from file %s", pod.Name, pod.UID, filename)
 
-	// Always overrides the namespace provided by the file.
-	pod.Namespace = kubelet.NamespaceDefault
-	glog.V(5).Infof("Using namespace %q for pod %q from file %s", pod.Namespace, pod.Name, filename)
-
-	// Currently just simply follow the same format in resthandler.go
-	pod.ObjectMeta.SelfLink = fmt.Sprintf("/api/v1beta2/pods/%s?namespace=%s",
-		pod.Name, pod.Namespace)
-
-	if glog.V(4) {
-		glog.Infof("Got pod from file %q: %#v", filename, pod)
-	} else {
-		glog.V(5).Infof("Got pod from file %q: %s.%s (%s)", filename, pod.Namespace, pod.Name, pod.UID)
-	}
-	return pod, nil
+	return pod, fmt.Errorf("%v: read '%v', but couldn't parse as neither "+
+		"manifest (%v) nor pod (%v).\n",
+		filename, string(data), manifestErr, podErr)
 }

pkg/kubelet/config/http.go

@@ -19,21 +19,15 @@ package config
 
 import (
 	"bytes"
-	"crypto/md5"
-	"encoding/hex"
 	"fmt"
 	"io/ioutil"
 	"net/http"
 	"time"
 
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/v1beta1"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/api/validation"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet"
-	"github.com/GoogleCloudPlatform/kubernetes/pkg/types"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 
-	"github.com/ghodss/yaml"
 	"github.com/golang/glog"
 )
 
@@ -84,14 +78,14 @@ func (s *sourceURL) extractFromURL() error {
 	s.data = data
 
 	// First try as if it's a single manifest
-	parsed, manifest, pod, singleErr := tryDecodeSingle(data)
+	parsed, manifest, pod, singleErr := tryDecodeSingleManifest(data)
 	if parsed {
 		if singleErr != nil {
 			// It parsed but could not be used.
 			return singleErr
 		}
 		// It parsed!
-		if err = applyDefaults(&pod, s.url); err != nil {
+		if err = applyDefaults(&pod, s.url, false); err != nil {
 			return err
 		}
 		s.updates <- kubelet.PodUpdate{[]api.Pod{pod}, kubelet.SET, kubelet.HTTPSource}
@@ -99,7 +93,7 @@ func (s *sourceURL) extractFromURL() error {
 	}
 
 	// That didn't work, so try an array of manifests.
-	parsed, manifests, pods, multiErr := tryDecodeList(data)
+	parsed, manifests, pods, multiErr := tryDecodeManifestList(data)
 	if parsed {
 		if multiErr != nil {
 			// It parsed but could not be used.
@@ -115,7 +109,7 @@ func (s *sourceURL) extractFromURL() error {
 		// Assume it parsed.
 		for i := range pods.Items {
 			pod := &pods.Items[i]
-			if err = applyDefaults(pod, s.url); err != nil {
+			if err = applyDefaults(pod, s.url, false); err != nil {
 				return err
 			}
 		}
@@ -127,7 +121,7 @@ func (s *sourceURL) extractFromURL() error {
 	// Try to parse it as Pod(s).
 
 	// First try as it is a single pod.
-	parsed, pod, singlePodErr := tryDecodeSinglePod(data, s.url)
+	parsed, pod, singlePodErr := tryDecodeSinglePod(data, s.url, false)
 	if parsed {
 		if singlePodErr != nil {
 			// It parsed but could not be used.
@@ -138,7 +132,7 @@ func (s *sourceURL) extractFromURL() error {
 	}
 
 	// That didn't work, so try a list of pods.
-	parsed, pods, multiPodErr := tryDecodePodList(data, s.url)
+	parsed, pods, multiPodErr := tryDecodePodList(data, s.url, false)
 	if parsed {
 		if multiPodErr != nil {
 			// It parsed but could not be used.
@@ -154,134 +148,3 @@ func (s *sourceURL) extractFromURL() error {
 		s.url, string(data), singleErr, manifest, multiErr, manifests,
 		singlePodErr, multiPodErr)
 }
-
-func tryDecodeSingle(data []byte) (parsed bool, manifest v1beta1.ContainerManifest, pod api.Pod, err error) {
-	// TODO: should be api.Scheme.Decode
-	// This is awful.  DecodeInto() expects to find an APIObject, which
-	// Manifest is not.  We keep reading manifest for now for compat, but
-	// we will eventually change it to read Pod (at which point this all
-	// becomes nicer).  Until then, we assert that the ContainerManifest
-	// structure on disk is always v1beta1.  Read that, convert it to a
-	// "current" ContainerManifest (should be ~identical), then convert
-	// that to a Pod (which is a well-understood conversion).  This
-	// avoids writing a v1beta1.ContainerManifest -> api.Pod
-	// conversion which would be identical to the api.ContainerManifest ->
-	// api.Pod conversion.
-	if err = yaml.Unmarshal(data, &manifest); err != nil {
-		return false, manifest, pod, err
-	}
-	newManifest := api.ContainerManifest{}
-	if err = api.Scheme.Convert(&manifest, &newManifest); err != nil {
-		return false, manifest, pod, err
-	}
-	if errs := validation.ValidateManifest(&newManifest); len(errs) > 0 {
-		err = fmt.Errorf("invalid manifest: %v", errs)
-		return false, manifest, pod, err
-	}
-	if err = api.Scheme.Convert(&newManifest, &pod); err != nil {
-		return true, manifest, pod, err
-	}
-	// Success.
-	return true, manifest, pod, nil
-}
-
-func tryDecodeList(data []byte) (parsed bool, manifests []v1beta1.ContainerManifest, pods api.PodList, err error) {
-	// TODO: should be api.Scheme.Decode
-	// See the comment in tryDecodeSingle().
-	if err = yaml.Unmarshal(data, &manifests); err != nil {
-		return false, manifests, pods, err
-	}
-	newManifests := []api.ContainerManifest{}
-	if err = api.Scheme.Convert(&manifests, &newManifests); err != nil {
-		return false, manifests, pods, err
-	}
-	for i := range newManifests {
-		manifest := &newManifests[i]
-		if errs := validation.ValidateManifest(manifest); len(errs) > 0 {
-			err = fmt.Errorf("invalid manifest: %v", errs)
-			return false, manifests, pods, err
-		}
-	}
-	list := api.ContainerManifestList{Items: newManifests}
-	if err = api.Scheme.Convert(&list, &pods); err != nil {
-		return true, manifests, pods, err
-	}
-	// Success.
-	return true, manifests, pods, nil
-}
-
-func tryDecodeSinglePod(data []byte, url string) (parsed bool, pod api.Pod, err error) {
-	obj, err := api.Scheme.Decode(data)
-	if err != nil {
-		return false, pod, err
-	}
-	// Check whether the object could be converted to single pod.
-	if _, ok := obj.(*api.Pod); !ok {
-		err = fmt.Errorf("invalid pod: %+v", obj)
-		return false, pod, err
-	}
-	newPod := obj.(*api.Pod)
-	// Apply default values and validate the pod.
-	if err = applyDefaults(newPod, url); err != nil {
-		return true, pod, err
-	}
-	if errs := validation.ValidatePod(newPod); len(errs) > 0 {
-		err = fmt.Errorf("invalid pod: %v", errs)
-		return true, pod, err
-	}
-	return true, *newPod, nil
-}
-
-func tryDecodePodList(data []byte, url string) (parsed bool, pods api.PodList, err error) {
-	obj, err := api.Scheme.Decode(data)
-	if err != nil {
-		return false, pods, err
-	}
-	// Check whether the object could be converted to list of pods.
-	if _, ok := obj.(*api.PodList); !ok {
-		err = fmt.Errorf("invalid pods list: %+v", obj)
-		return false, pods, err
-	}
-	newPods := obj.(*api.PodList)
-	// Apply default values and validate pods.
-	for i := range newPods.Items {
-		newPod := &newPods.Items[i]
-		if err = applyDefaults(newPod, url); err != nil {
-			return true, pods, err
-		}
-		if errs := validation.ValidatePod(newPod); len(errs) > 0 {
-			err = fmt.Errorf("invalid pod: %v", errs)
-			return true, pods, err
-		}
-	}
-	return true, *newPods, err
-}
-
-func applyDefaults(pod *api.Pod, url string) error {
-	if len(pod.UID) == 0 {
-		hasher := md5.New()
-		fmt.Fprintf(hasher, "url:%s", url)
-		util.DeepHashObject(hasher, pod)
-		pod.UID = types.UID(hex.EncodeToString(hasher.Sum(nil)[0:]))
-		glog.V(5).Infof("Generated UID %q for pod %q from URL %s", pod.UID, pod.Name, url)
-	}
-	// This is required for backward compatibility, and should be removed once we
-	// completely deprecate ContainerManifest.
-	var err error
-	if len(pod.Name) == 0 {
-		pod.Name = string(pod.UID)
-	}
-	pod.Name, err = GeneratePodName(pod.Name)
-	if err != nil {
-		return err
-	}
-	glog.V(5).Infof("Generated Name %q for UID %q from URL %s", pod.Name, pod.UID, url)
-
-	if pod.Namespace == "" {
-		pod.Namespace = kubelet.NamespaceDefault
-	}
-	glog.V(5).Infof("Using namespace %q for pod %q from URL %s", pod.Namespace, pod.Name, url)
-	pod.ObjectMeta.SelfLink = fmt.Sprintf("/api/v1beta2/pods/%s?namespace=%s",
-		pod.Name, pod.Namespace)
-	return nil
-}