mirror of
				https://github.com/optim-enterprises-bv/kubernetes.git
				synced 2025-10-31 18:28:13 +00:00 
			
		
		
		
	
		
			
				
	
	
		
			782 lines
		
	
	
		
			24 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
			
		
		
	
	
			782 lines
		
	
	
		
			24 KiB
		
	
	
	
		
			Go
		
	
	
	
	
	
| /*
 | |
| Copyright 2014 The Kubernetes Authors 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 kubelet
 | |
| 
 | |
| import (
 | |
| 	"crypto/tls"
 | |
| 	"encoding/json"
 | |
| 	"errors"
 | |
| 	"fmt"
 | |
| 	"io"
 | |
| 	"net"
 | |
| 	"net/http"
 | |
| 	"net/http/pprof"
 | |
| 	"net/url"
 | |
| 	"path"
 | |
| 	"strconv"
 | |
| 	"strings"
 | |
| 	"sync"
 | |
| 	"time"
 | |
| 
 | |
| 	"github.com/golang/glog"
 | |
| 	cadvisorApi "github.com/google/cadvisor/info/v1"
 | |
| 	"github.com/prometheus/client_golang/prometheus"
 | |
| 	"k8s.io/kubernetes/pkg/api"
 | |
| 	"k8s.io/kubernetes/pkg/api/latest"
 | |
| 	"k8s.io/kubernetes/pkg/healthz"
 | |
| 	"k8s.io/kubernetes/pkg/httplog"
 | |
| 	kubecontainer "k8s.io/kubernetes/pkg/kubelet/container"
 | |
| 	"k8s.io/kubernetes/pkg/types"
 | |
| 	"k8s.io/kubernetes/pkg/util/flushwriter"
 | |
| 	"k8s.io/kubernetes/pkg/util/httpstream"
 | |
| 	"k8s.io/kubernetes/pkg/util/httpstream/spdy"
 | |
| )
 | |
| 
 | |
| // Server is a http.Handler which exposes kubelet functionality over HTTP.
 | |
| type Server struct {
 | |
| 	host HostInterface
 | |
| 	mux  *http.ServeMux
 | |
| }
 | |
| 
 | |
| type TLSOptions struct {
 | |
| 	Config   *tls.Config
 | |
| 	CertFile string
 | |
| 	KeyFile  string
 | |
| }
 | |
| 
 | |
| // ListenAndServeKubeletServer initializes a server to respond to HTTP network requests on the Kubelet.
 | |
| func ListenAndServeKubeletServer(host HostInterface, address net.IP, port uint, tlsOptions *TLSOptions, enableDebuggingHandlers bool) {
 | |
| 	glog.Infof("Starting to listen on %s:%d", address, port)
 | |
| 	handler := NewServer(host, enableDebuggingHandlers)
 | |
| 	s := &http.Server{
 | |
| 		Addr:           net.JoinHostPort(address.String(), strconv.FormatUint(uint64(port), 10)),
 | |
| 		Handler:        &handler,
 | |
| 		MaxHeaderBytes: 1 << 20,
 | |
| 	}
 | |
| 	if tlsOptions != nil {
 | |
| 		s.TLSConfig = tlsOptions.Config
 | |
| 		glog.Fatal(s.ListenAndServeTLS(tlsOptions.CertFile, tlsOptions.KeyFile))
 | |
| 	} else {
 | |
| 		glog.Fatal(s.ListenAndServe())
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // ListenAndServeKubeletReadOnlyServer initializes a server to respond to HTTP network requests on the Kubelet.
 | |
| func ListenAndServeKubeletReadOnlyServer(host HostInterface, address net.IP, port uint) {
 | |
| 	glog.V(1).Infof("Starting to listen read-only on %s:%d", address, port)
 | |
| 	s := NewServer(host, false)
 | |
| 	s.mux.Handle("/metrics", prometheus.Handler())
 | |
| 
 | |
| 	server := &http.Server{
 | |
| 		Addr:           net.JoinHostPort(address.String(), strconv.FormatUint(uint64(port), 10)),
 | |
| 		Handler:        &s,
 | |
| 		MaxHeaderBytes: 1 << 20,
 | |
| 	}
 | |
| 	glog.Fatal(server.ListenAndServe())
 | |
| }
 | |
| 
 | |
| // HostInterface contains all the kubelet methods required by the server.
 | |
| // For testablitiy.
 | |
| type HostInterface interface {
 | |
| 	GetContainerInfo(podFullName string, uid types.UID, containerName string, req *cadvisorApi.ContainerInfoRequest) (*cadvisorApi.ContainerInfo, error)
 | |
| 	GetContainerRuntimeVersion() (kubecontainer.Version, error)
 | |
| 	GetRawContainerInfo(containerName string, req *cadvisorApi.ContainerInfoRequest, subcontainers bool) (map[string]*cadvisorApi.ContainerInfo, error)
 | |
| 	GetCachedMachineInfo() (*cadvisorApi.MachineInfo, error)
 | |
| 	GetPods() []*api.Pod
 | |
| 	GetRunningPods() ([]*api.Pod, error)
 | |
| 	GetPodByName(namespace, name string) (*api.Pod, bool)
 | |
| 	RunInContainer(name string, uid types.UID, container string, cmd []string) ([]byte, error)
 | |
| 	ExecInContainer(name string, uid types.UID, container string, cmd []string, in io.Reader, out, err io.WriteCloser, tty bool) error
 | |
| 	AttachContainer(name string, uid types.UID, container string, in io.Reader, out, err io.WriteCloser, tty bool) error
 | |
| 	GetKubeletContainerLogs(podFullName, containerName, tail string, follow, previous bool, stdout, stderr io.Writer) error
 | |
| 	ServeLogs(w http.ResponseWriter, req *http.Request)
 | |
| 	PortForward(name string, uid types.UID, port uint16, stream io.ReadWriteCloser) error
 | |
| 	StreamingConnectionIdleTimeout() time.Duration
 | |
| 	ResyncInterval() time.Duration
 | |
| 	GetHostname() string
 | |
| 	LatestLoopEntryTime() time.Time
 | |
| }
 | |
| 
 | |
| // NewServer initializes and configures a kubelet.Server object to handle HTTP requests.
 | |
| func NewServer(host HostInterface, enableDebuggingHandlers bool) Server {
 | |
| 	server := Server{
 | |
| 		host: host,
 | |
| 		mux:  http.NewServeMux(),
 | |
| 	}
 | |
| 	server.InstallDefaultHandlers()
 | |
| 	if enableDebuggingHandlers {
 | |
| 		server.InstallDebuggingHandlers()
 | |
| 	}
 | |
| 	return server
 | |
| }
 | |
| 
 | |
| // InstallDefaultHandlers registers the default set of supported HTTP request patterns with the mux.
 | |
| func (s *Server) InstallDefaultHandlers() {
 | |
| 	healthz.InstallHandler(s.mux,
 | |
| 		healthz.PingHealthz,
 | |
| 		healthz.NamedCheck("docker", s.dockerHealthCheck),
 | |
| 		healthz.NamedCheck("hostname", s.hostnameHealthCheck),
 | |
| 		healthz.NamedCheck("syncloop", s.syncLoopHealthCheck),
 | |
| 	)
 | |
| 	s.mux.HandleFunc("/pods", s.handlePods)
 | |
| 	s.mux.HandleFunc("/stats/", s.handleStats)
 | |
| 	s.mux.HandleFunc("/spec/", s.handleSpec)
 | |
| }
 | |
| 
 | |
| // InstallDeguggingHandlers registers the HTTP request patterns that serve logs or run commands/containers
 | |
| func (s *Server) InstallDebuggingHandlers() {
 | |
| 	s.mux.HandleFunc("/run/", s.handleRun)
 | |
| 	s.mux.HandleFunc("/exec/", s.handleExec)
 | |
| 	s.mux.HandleFunc("/attach/", s.handleAttach)
 | |
| 	s.mux.HandleFunc("/portForward/", s.handlePortForward)
 | |
| 
 | |
| 	s.mux.HandleFunc("/logs/", s.handleLogs)
 | |
| 	s.mux.HandleFunc("/containerLogs/", s.handleContainerLogs)
 | |
| 	s.mux.Handle("/metrics", prometheus.Handler())
 | |
| 	// The /runningpods endpoint is used for testing only.
 | |
| 	s.mux.HandleFunc("/runningpods", s.handleRunningPods)
 | |
| 
 | |
| 	s.mux.HandleFunc("/debug/pprof/", pprof.Index)
 | |
| 	s.mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
 | |
| 	s.mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
 | |
| }
 | |
| 
 | |
| // error serializes an error object into an HTTP response.
 | |
| func (s *Server) error(w http.ResponseWriter, err error) {
 | |
| 	msg := fmt.Sprintf("Internal Error: %v", err)
 | |
| 	glog.Infof("HTTP InternalServerError: %s", msg)
 | |
| 	http.Error(w, msg, http.StatusInternalServerError)
 | |
| }
 | |
| 
 | |
| func (s *Server) dockerHealthCheck(req *http.Request) error {
 | |
| 	version, err := s.host.GetContainerRuntimeVersion()
 | |
| 	if err != nil {
 | |
| 		return errors.New("unknown Docker version")
 | |
| 	}
 | |
| 	// Verify the docker version.
 | |
| 	result, err := version.Compare("1.15")
 | |
| 	if err != nil {
 | |
| 		return err
 | |
| 	}
 | |
| 	if result < 0 {
 | |
| 		return fmt.Errorf("Docker version is too old: %q", version.String())
 | |
| 	}
 | |
| 	return nil
 | |
| }
 | |
| 
 | |
| func (s *Server) hostnameHealthCheck(req *http.Request) error {
 | |
| 	masterHostname, _, err := net.SplitHostPort(req.Host)
 | |
| 	if err != nil {
 | |
| 		if !strings.Contains(req.Host, ":") {
 | |
| 			masterHostname = req.Host
 | |
| 		} else {
 | |
| 			return fmt.Errorf("Could not parse hostname from http request: %v", err)
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	// Check that the hostname known by the master matches the hostname
 | |
| 	// the kubelet knows
 | |
| 	hostname := s.host.GetHostname()
 | |
| 	if masterHostname != hostname && masterHostname != "127.0.0.1" && masterHostname != "localhost" {
 | |
| 		return fmt.Errorf("Kubelet hostname \"%v\" does not match the hostname expected by the master \"%v\"", hostname, masterHostname)
 | |
| 	}
 | |
| 	return nil
 | |
| }
 | |
| 
 | |
| // Checks if kubelet's sync loop  that updates containers is working.
 | |
| func (s *Server) syncLoopHealthCheck(req *http.Request) error {
 | |
| 	duration := s.host.ResyncInterval() * 2
 | |
| 	minDuration := time.Minute * 5
 | |
| 	if duration < minDuration {
 | |
| 		duration = minDuration
 | |
| 	}
 | |
| 	enterLoopTime := s.host.LatestLoopEntryTime()
 | |
| 	if !enterLoopTime.IsZero() && time.Now().After(enterLoopTime.Add(duration)) {
 | |
| 		return fmt.Errorf("Sync Loop took longer than expected.")
 | |
| 	}
 | |
| 	return nil
 | |
| }
 | |
| 
 | |
| // handleContainerLogs handles containerLogs request against the Kubelet
 | |
| func (s *Server) handleContainerLogs(w http.ResponseWriter, req *http.Request) {
 | |
| 	defer req.Body.Close()
 | |
| 	u, err := url.ParseRequestURI(req.RequestURI)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	parts := strings.Split(u.Path, "/")
 | |
| 
 | |
| 	// req URI: /containerLogs/<podNamespace>/<podID>/<containerName>
 | |
| 	var podNamespace, podID, containerName string
 | |
| 	if len(parts) == 5 {
 | |
| 		podNamespace = parts[2]
 | |
| 		podID = parts[3]
 | |
| 		containerName = parts[4]
 | |
| 	} else {
 | |
| 		http.Error(w, "Unexpected path for command running", http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if len(podID) == 0 {
 | |
| 		http.Error(w, `{"message": "Missing podID."}`, http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 	if len(containerName) == 0 {
 | |
| 		http.Error(w, `{"message": "Missing container name."}`, http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 	if len(podNamespace) == 0 {
 | |
| 		http.Error(w, `{"message": "Missing podNamespace."}`, http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	uriValues := u.Query()
 | |
| 	follow, _ := strconv.ParseBool(uriValues.Get("follow"))
 | |
| 	previous, _ := strconv.ParseBool(uriValues.Get("previous"))
 | |
| 	tail := uriValues.Get("tail")
 | |
| 
 | |
| 	pod, ok := s.host.GetPodByName(podNamespace, podID)
 | |
| 	if !ok {
 | |
| 		http.Error(w, fmt.Sprintf("Pod %q does not exist", podID), http.StatusNotFound)
 | |
| 		return
 | |
| 	}
 | |
| 	// Check if containerName is valid.
 | |
| 	containerExists := false
 | |
| 	for _, container := range pod.Spec.Containers {
 | |
| 		if container.Name == containerName {
 | |
| 			containerExists = true
 | |
| 		}
 | |
| 	}
 | |
| 	if !containerExists {
 | |
| 		http.Error(w, fmt.Sprintf("Container %q not found in Pod %q", containerName, podID), http.StatusNotFound)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if _, ok := w.(http.Flusher); !ok {
 | |
| 		s.error(w, fmt.Errorf("unable to convert %v into http.Flusher", w))
 | |
| 		return
 | |
| 	}
 | |
| 	fw := flushwriter.Wrap(w)
 | |
| 	w.Header().Set("Transfer-Encoding", "chunked")
 | |
| 	w.WriteHeader(http.StatusOK)
 | |
| 	err = s.host.GetKubeletContainerLogs(kubecontainer.GetPodFullName(pod), containerName, tail, follow, previous, fw, fw)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // encodePods creates an api.PodList object from pods and returns the encoded
 | |
| // PodList.
 | |
| func encodePods(pods []*api.Pod) (data []byte, err error) {
 | |
| 	podList := new(api.PodList)
 | |
| 	for _, pod := range pods {
 | |
| 		podList.Items = append(podList.Items, *pod)
 | |
| 	}
 | |
| 	return latest.Codec.Encode(podList)
 | |
| }
 | |
| 
 | |
| // handlePods returns a list of pods bound to the Kubelet and their spec.
 | |
| func (s *Server) handlePods(w http.ResponseWriter, req *http.Request) {
 | |
| 	pods := s.host.GetPods()
 | |
| 	data, err := encodePods(pods)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	w.Header().Add("Content-type", "application/json")
 | |
| 	w.Write(data)
 | |
| }
 | |
| 
 | |
| // handleRunningPods returns a list of pods running on Kubelet. The list is
 | |
| // provided by the container runtime, and is different from the list returned
 | |
| // by handlePods, which is a set of desired pods to run.
 | |
| func (s *Server) handleRunningPods(w http.ResponseWriter, req *http.Request) {
 | |
| 	pods, err := s.host.GetRunningPods()
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	data, err := encodePods(pods)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	w.Header().Add("Content-type", "application/json")
 | |
| 	w.Write(data)
 | |
| }
 | |
| 
 | |
| // handleStats handles stats requests against the Kubelet.
 | |
| func (s *Server) handleStats(w http.ResponseWriter, req *http.Request) {
 | |
| 	s.serveStats(w, req)
 | |
| }
 | |
| 
 | |
| // handleLogs handles logs requests against the Kubelet.
 | |
| func (s *Server) handleLogs(w http.ResponseWriter, req *http.Request) {
 | |
| 	s.host.ServeLogs(w, req)
 | |
| }
 | |
| 
 | |
| // handleSpec handles spec requests against the Kubelet.
 | |
| func (s *Server) handleSpec(w http.ResponseWriter, req *http.Request) {
 | |
| 	info, err := s.host.GetCachedMachineInfo()
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	data, err := json.Marshal(info)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	w.Header().Add("Content-type", "application/json")
 | |
| 	w.Write(data)
 | |
| }
 | |
| 
 | |
| func parseContainerCoordinates(path string) (namespace, pod string, uid types.UID, container string, err error) {
 | |
| 	parts := strings.Split(path, "/")
 | |
| 
 | |
| 	if len(parts) == 5 {
 | |
| 		namespace = parts[2]
 | |
| 		pod = parts[3]
 | |
| 		container = parts[4]
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if len(parts) == 6 {
 | |
| 		namespace = parts[2]
 | |
| 		pod = parts[3]
 | |
| 		uid = types.UID(parts[4])
 | |
| 		container = parts[5]
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	err = fmt.Errorf("Unexpected path %s. Expected /.../.../<namespace>/<pod>/<container> or /.../.../<namespace>/<pod>/<uid>/<container>", path)
 | |
| 	return
 | |
| }
 | |
| 
 | |
| const streamCreationTimeout = 30 * time.Second
 | |
| 
 | |
| func (s *Server) handleAttach(w http.ResponseWriter, req *http.Request) {
 | |
| 	u, err := url.ParseRequestURI(req.RequestURI)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	podNamespace, podID, uid, container, err := parseContainerCoordinates(u.Path)
 | |
| 	if err != nil {
 | |
| 		http.Error(w, err.Error(), http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 	pod, ok := s.host.GetPodByName(podNamespace, podID)
 | |
| 	if !ok {
 | |
| 		http.Error(w, "Pod does not exist", http.StatusNotFound)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	stdinStream, stdoutStream, stderrStream, errorStream, conn, tty, ok := s.createStreams(w, req)
 | |
| 	if conn != nil {
 | |
| 		defer conn.Close()
 | |
| 	}
 | |
| 	if !ok {
 | |
| 		// error is handled in the createStreams function
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	err = s.host.AttachContainer(kubecontainer.GetPodFullName(pod), uid, container, stdinStream, stdoutStream, stderrStream, tty)
 | |
| 	if err != nil {
 | |
| 		msg := fmt.Sprintf("Error executing command in container: %v", err)
 | |
| 		glog.Error(msg)
 | |
| 		errorStream.Write([]byte(msg))
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // handleRun handles requests to run a command inside a container.
 | |
| func (s *Server) handleRun(w http.ResponseWriter, req *http.Request) {
 | |
| 	u, err := url.ParseRequestURI(req.RequestURI)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	podNamespace, podID, uid, container, err := parseContainerCoordinates(u.Path)
 | |
| 	if err != nil {
 | |
| 		http.Error(w, err.Error(), http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 	pod, ok := s.host.GetPodByName(podNamespace, podID)
 | |
| 	if !ok {
 | |
| 		http.Error(w, "Pod does not exist", http.StatusNotFound)
 | |
| 		return
 | |
| 	}
 | |
| 	command := strings.Split(u.Query().Get("cmd"), " ")
 | |
| 	data, err := s.host.RunInContainer(kubecontainer.GetPodFullName(pod), uid, container, command)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	w.Header().Add("Content-type", "text/plain")
 | |
| 	w.Write(data)
 | |
| }
 | |
| 
 | |
| // handleExec handles requests to run a command inside a container.
 | |
| func (s *Server) handleExec(w http.ResponseWriter, req *http.Request) {
 | |
| 	u, err := url.ParseRequestURI(req.RequestURI)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	podNamespace, podID, uid, container, err := parseContainerCoordinates(u.Path)
 | |
| 	if err != nil {
 | |
| 		http.Error(w, err.Error(), http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 	pod, ok := s.host.GetPodByName(podNamespace, podID)
 | |
| 	if !ok {
 | |
| 		http.Error(w, "Pod does not exist", http.StatusNotFound)
 | |
| 		return
 | |
| 	}
 | |
| 	stdinStream, stdoutStream, stderrStream, errorStream, conn, tty, ok := s.createStreams(w, req)
 | |
| 	if conn != nil {
 | |
| 		defer conn.Close()
 | |
| 	}
 | |
| 	if !ok {
 | |
| 		return
 | |
| 	}
 | |
| 	err = s.host.ExecInContainer(kubecontainer.GetPodFullName(pod), uid, container, u.Query()[api.ExecCommandParamm], stdinStream, stdoutStream, stderrStream, tty)
 | |
| 	if err != nil {
 | |
| 		msg := fmt.Sprintf("Error executing command in container: %v", err)
 | |
| 		glog.Error(msg)
 | |
| 		errorStream.Write([]byte(msg))
 | |
| 	}
 | |
| }
 | |
| 
 | |
| func (s *Server) createStreams(w http.ResponseWriter, req *http.Request) (io.Reader, io.WriteCloser, io.WriteCloser, io.WriteCloser, httpstream.Connection, bool, bool) {
 | |
| 	req.ParseForm()
 | |
| 	// start at 1 for error stream
 | |
| 	expectedStreams := 1
 | |
| 	if req.FormValue(api.ExecStdinParam) == "1" {
 | |
| 		expectedStreams++
 | |
| 	}
 | |
| 	if req.FormValue(api.ExecStdoutParam) == "1" {
 | |
| 		expectedStreams++
 | |
| 	}
 | |
| 	tty := req.FormValue(api.ExecTTYParam) == "1"
 | |
| 	if !tty && req.FormValue(api.ExecStderrParam) == "1" {
 | |
| 		expectedStreams++
 | |
| 	}
 | |
| 
 | |
| 	if expectedStreams == 1 {
 | |
| 		http.Error(w, "You must specify at least 1 of stdin, stdout, stderr", http.StatusBadRequest)
 | |
| 		return nil, nil, nil, nil, nil, false, false
 | |
| 	}
 | |
| 
 | |
| 	streamCh := make(chan httpstream.Stream)
 | |
| 
 | |
| 	upgrader := spdy.NewResponseUpgrader()
 | |
| 	conn := upgrader.UpgradeResponse(w, req, func(stream httpstream.Stream) error {
 | |
| 		streamCh <- stream
 | |
| 		return nil
 | |
| 	})
 | |
| 	// from this point on, we can no longer call methods on w
 | |
| 	if conn == nil {
 | |
| 		// The upgrader is responsible for notifying the client of any errors that
 | |
| 		// occurred during upgrading. All we can do is return here at this point
 | |
| 		// if we weren't successful in upgrading.
 | |
| 		return nil, nil, nil, nil, nil, false, false
 | |
| 	}
 | |
| 
 | |
| 	conn.SetIdleTimeout(s.host.StreamingConnectionIdleTimeout())
 | |
| 
 | |
| 	// TODO make it configurable?
 | |
| 	expired := time.NewTimer(streamCreationTimeout)
 | |
| 
 | |
| 	var errorStream, stdinStream, stdoutStream, stderrStream httpstream.Stream
 | |
| 	receivedStreams := 0
 | |
| WaitForStreams:
 | |
| 	for {
 | |
| 		select {
 | |
| 		case stream := <-streamCh:
 | |
| 			streamType := stream.Headers().Get(api.StreamType)
 | |
| 			switch streamType {
 | |
| 			case api.StreamTypeError:
 | |
| 				errorStream = stream
 | |
| 				defer errorStream.Reset()
 | |
| 				receivedStreams++
 | |
| 			case api.StreamTypeStdin:
 | |
| 				stdinStream = stream
 | |
| 				receivedStreams++
 | |
| 			case api.StreamTypeStdout:
 | |
| 				stdoutStream = stream
 | |
| 				receivedStreams++
 | |
| 			case api.StreamTypeStderr:
 | |
| 				stderrStream = stream
 | |
| 				receivedStreams++
 | |
| 			default:
 | |
| 				glog.Errorf("Unexpected stream type: '%s'", streamType)
 | |
| 			}
 | |
| 			if receivedStreams == expectedStreams {
 | |
| 				break WaitForStreams
 | |
| 			}
 | |
| 		case <-expired.C:
 | |
| 			// TODO find a way to return the error to the user. Maybe use a separate
 | |
| 			// stream to report errors?
 | |
| 			glog.Error("Timed out waiting for client to create streams")
 | |
| 			return nil, nil, nil, nil, nil, false, false
 | |
| 		}
 | |
| 	}
 | |
| 
 | |
| 	if stdinStream != nil {
 | |
| 		// close our half of the input stream, since we won't be writing to it
 | |
| 		stdinStream.Close()
 | |
| 	}
 | |
| 
 | |
| 	return stdinStream, stdoutStream, stderrStream, errorStream, conn, tty, true
 | |
| }
 | |
| 
 | |
| func parsePodCoordinates(path string) (namespace, pod string, uid types.UID, err error) {
 | |
| 	parts := strings.Split(path, "/")
 | |
| 
 | |
| 	if len(parts) == 4 {
 | |
| 		namespace = parts[2]
 | |
| 		pod = parts[3]
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	if len(parts) == 5 {
 | |
| 		namespace = parts[2]
 | |
| 		pod = parts[3]
 | |
| 		uid = types.UID(parts[4])
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	err = fmt.Errorf("Unexpected path %s. Expected /.../.../<namespace>/<pod> or /.../.../<namespace>/<pod>/<uid>", path)
 | |
| 	return
 | |
| }
 | |
| 
 | |
| func (s *Server) handlePortForward(w http.ResponseWriter, req *http.Request) {
 | |
| 	u, err := url.ParseRequestURI(req.RequestURI)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	podNamespace, podID, uid, err := parsePodCoordinates(u.Path)
 | |
| 	if err != nil {
 | |
| 		http.Error(w, err.Error(), http.StatusBadRequest)
 | |
| 		return
 | |
| 	}
 | |
| 	pod, ok := s.host.GetPodByName(podNamespace, podID)
 | |
| 	if !ok {
 | |
| 		http.Error(w, "Pod does not exist", http.StatusNotFound)
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	streamChan := make(chan httpstream.Stream, 1)
 | |
| 	upgrader := spdy.NewResponseUpgrader()
 | |
| 	conn := upgrader.UpgradeResponse(w, req, func(stream httpstream.Stream) error {
 | |
| 		portString := stream.Headers().Get(api.PortHeader)
 | |
| 		port, err := strconv.ParseUint(portString, 10, 16)
 | |
| 		if err != nil {
 | |
| 			return fmt.Errorf("Unable to parse '%s' as a port: %v", portString, err)
 | |
| 		}
 | |
| 		if port < 1 {
 | |
| 			return fmt.Errorf("Port '%d' must be greater than 0", port)
 | |
| 		}
 | |
| 		streamChan <- stream
 | |
| 		return nil
 | |
| 	})
 | |
| 	if conn == nil {
 | |
| 		return
 | |
| 	}
 | |
| 	defer conn.Close()
 | |
| 	conn.SetIdleTimeout(s.host.StreamingConnectionIdleTimeout())
 | |
| 
 | |
| 	var dataStreamLock sync.Mutex
 | |
| 	dataStreamChans := make(map[string]chan httpstream.Stream)
 | |
| 
 | |
| Loop:
 | |
| 	for {
 | |
| 		select {
 | |
| 		case <-conn.CloseChan():
 | |
| 			break Loop
 | |
| 		case stream := <-streamChan:
 | |
| 			streamType := stream.Headers().Get(api.StreamType)
 | |
| 			port := stream.Headers().Get(api.PortHeader)
 | |
| 			dataStreamLock.Lock()
 | |
| 			switch streamType {
 | |
| 			case "error":
 | |
| 				ch := make(chan httpstream.Stream)
 | |
| 				dataStreamChans[port] = ch
 | |
| 				go waitForPortForwardDataStreamAndRun(kubecontainer.GetPodFullName(pod), uid, stream, ch, s.host)
 | |
| 			case "data":
 | |
| 				ch, ok := dataStreamChans[port]
 | |
| 				if ok {
 | |
| 					ch <- stream
 | |
| 					delete(dataStreamChans, port)
 | |
| 				} else {
 | |
| 					glog.Errorf("Unable to locate data stream channel for port %s", port)
 | |
| 				}
 | |
| 			default:
 | |
| 				glog.Errorf("streamType header must be 'error' or 'data', got: '%s'", streamType)
 | |
| 				stream.Reset()
 | |
| 			}
 | |
| 			dataStreamLock.Unlock()
 | |
| 		}
 | |
| 	}
 | |
| }
 | |
| 
 | |
| func waitForPortForwardDataStreamAndRun(pod string, uid types.UID, errorStream httpstream.Stream, dataStreamChan chan httpstream.Stream, host HostInterface) {
 | |
| 	defer errorStream.Reset()
 | |
| 
 | |
| 	var dataStream httpstream.Stream
 | |
| 
 | |
| 	select {
 | |
| 	case dataStream = <-dataStreamChan:
 | |
| 	case <-time.After(streamCreationTimeout):
 | |
| 		errorStream.Write([]byte("Timed out waiting for data stream"))
 | |
| 		//TODO delete from dataStreamChans[port]
 | |
| 		return
 | |
| 	}
 | |
| 
 | |
| 	portString := dataStream.Headers().Get(api.PortHeader)
 | |
| 	port, _ := strconv.ParseUint(portString, 10, 16)
 | |
| 	err := host.PortForward(pod, uid, uint16(port), dataStream)
 | |
| 	if err != nil {
 | |
| 		msg := fmt.Errorf("Error forwarding port %d to pod %s, uid %v: %v", port, pod, uid, err)
 | |
| 		glog.Error(msg)
 | |
| 		errorStream.Write([]byte(msg.Error()))
 | |
| 	}
 | |
| }
 | |
| 
 | |
| // ServeHTTP responds to HTTP requests on the Kubelet.
 | |
| func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 | |
| 	defer httplog.NewLogged(req, &w).StacktraceWhen(
 | |
| 		httplog.StatusIsNot(
 | |
| 			http.StatusOK,
 | |
| 			http.StatusMovedPermanently,
 | |
| 			http.StatusTemporaryRedirect,
 | |
| 			http.StatusNotFound,
 | |
| 			http.StatusSwitchingProtocols,
 | |
| 		),
 | |
| 	).Log()
 | |
| 	s.mux.ServeHTTP(w, req)
 | |
| }
 | |
| 
 | |
| type StatsRequest struct {
 | |
| 	// The name of the container for which to request stats.
 | |
| 	// Default: /
 | |
| 	ContainerName string `json:"containerName,omitempty"`
 | |
| 
 | |
| 	// Max number of stats to return.
 | |
| 	// If start and end time are specified this limit is ignored.
 | |
| 	// Default: 60
 | |
| 	NumStats int `json:"num_stats,omitempty"`
 | |
| 
 | |
| 	// Start time for which to query information.
 | |
| 	// If omitted, the beginning of time is assumed.
 | |
| 	Start time.Time `json:"start,omitempty"`
 | |
| 
 | |
| 	// End time for which to query information.
 | |
| 	// If omitted, current time is assumed.
 | |
| 	End time.Time `json:"end,omitempty"`
 | |
| 
 | |
| 	// Whether to also include information from subcontainers.
 | |
| 	// Default: false.
 | |
| 	Subcontainers bool `json:"subcontainers,omitempty"`
 | |
| }
 | |
| 
 | |
| // serveStats implements stats logic.
 | |
| func (s *Server) serveStats(w http.ResponseWriter, req *http.Request) {
 | |
| 	// Stats requests are in the following forms:
 | |
| 	//
 | |
| 	// /stats/                                              : Root container stats
 | |
| 	// /stats/container/                                    : Non-Kubernetes container stats (returns a map)
 | |
| 	// /stats/<pod name>/<container name>                   : Stats for Kubernetes pod/container
 | |
| 	// /stats/<namespace>/<pod name>/<uid>/<container name> : Stats for Kubernetes namespace/pod/uid/container
 | |
| 	components := strings.Split(strings.TrimPrefix(path.Clean(req.URL.Path), "/"), "/")
 | |
| 	var stats interface{}
 | |
| 	var err error
 | |
| 	var query StatsRequest
 | |
| 	query.NumStats = 60
 | |
| 
 | |
| 	err = json.NewDecoder(req.Body).Decode(&query)
 | |
| 	if err != nil && err != io.EOF {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	cadvisorRequest := cadvisorApi.ContainerInfoRequest{
 | |
| 		NumStats: query.NumStats,
 | |
| 		Start:    query.Start,
 | |
| 		End:      query.End,
 | |
| 	}
 | |
| 
 | |
| 	switch len(components) {
 | |
| 	case 1:
 | |
| 		// Root container stats.
 | |
| 		var statsMap map[string]*cadvisorApi.ContainerInfo
 | |
| 		statsMap, err = s.host.GetRawContainerInfo("/", &cadvisorRequest, false)
 | |
| 		stats = statsMap["/"]
 | |
| 	case 2:
 | |
| 		// Non-Kubernetes container stats.
 | |
| 		if components[1] != "container" {
 | |
| 			http.Error(w, fmt.Sprintf("unknown stats request type %q", components[1]), http.StatusNotFound)
 | |
| 			return
 | |
| 		}
 | |
| 		containerName := path.Join("/", query.ContainerName)
 | |
| 		stats, err = s.host.GetRawContainerInfo(containerName, &cadvisorRequest, query.Subcontainers)
 | |
| 	case 3:
 | |
| 		// Backward compatibility without uid information, does not support namespace
 | |
| 		pod, ok := s.host.GetPodByName(api.NamespaceDefault, components[1])
 | |
| 		if !ok {
 | |
| 			http.Error(w, "Pod does not exist", http.StatusNotFound)
 | |
| 			return
 | |
| 		}
 | |
| 		stats, err = s.host.GetContainerInfo(kubecontainer.GetPodFullName(pod), "", components[2], &cadvisorRequest)
 | |
| 	case 5:
 | |
| 		pod, ok := s.host.GetPodByName(components[1], components[2])
 | |
| 		if !ok {
 | |
| 			http.Error(w, "Pod does not exist", http.StatusNotFound)
 | |
| 			return
 | |
| 		}
 | |
| 		stats, err = s.host.GetContainerInfo(kubecontainer.GetPodFullName(pod), types.UID(components[3]), components[4], &cadvisorRequest)
 | |
| 	default:
 | |
| 		http.Error(w, fmt.Sprintf("Unknown resource: %v", components), http.StatusNotFound)
 | |
| 		return
 | |
| 	}
 | |
| 	switch err {
 | |
| 	case nil:
 | |
| 		break
 | |
| 	case ErrContainerNotFound:
 | |
| 		http.Error(w, err.Error(), http.StatusNotFound)
 | |
| 		return
 | |
| 	default:
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	if stats == nil {
 | |
| 		fmt.Fprint(w, "{}")
 | |
| 		return
 | |
| 	}
 | |
| 	data, err := json.Marshal(stats)
 | |
| 	if err != nil {
 | |
| 		s.error(w, err)
 | |
| 		return
 | |
| 	}
 | |
| 	w.Header().Add("Content-type", "application/json")
 | |
| 	w.Write(data)
 | |
| 	return
 | |
| }
 | 
