Merge pull request #19473 from deads2k/deploy-01-kubelet

remove directly kubelet dependency from deploymentconfigs

Author: openshift-merge-robot

pkg/apps/apiserver/apiserver.go

@@ -1,7 +1,6 @@
 package apiserver
 
 import (
-	"fmt"
 	"sync"
 
 	"k8s.io/apimachinery/pkg/apimachinery/registered"
@@ -10,11 +9,9 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/serializer"
 	"k8s.io/apiserver/pkg/registry/rest"
 	genericapiserver "k8s.io/apiserver/pkg/server"
-	kclientsetexternal "k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/kubernetes/pkg/api/legacyscheme"
 	kclientsetinternal "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset"
-	kubeletclient "k8s.io/kubernetes/pkg/kubelet/client"
 
 	appsapiv1 "github.com/openshift/api/apps/v1"
 	appsclientinternal "github.com/openshift/origin/pkg/apps/generated/internalclientset"
@@ -27,7 +24,6 @@ import (
 
 type ExtraConfig struct {
 	KubeAPIServerClientConfig *restclient.Config
-	KubeletClientConfig       *kubeletclient.KubeletClientConfig
 
 	// TODO these should all become local eventually
 	Scheme   *runtime.Scheme
@@ -121,14 +117,6 @@ func (c *completedConfig) newV1RESTStorage() (map[string]rest.Storage, error) {
 	if err != nil {
 		return nil, err
 	}
-	kubeExternalClient, err := kclientsetexternal.NewForConfig(c.ExtraConfig.KubeAPIServerClientConfig)
-	if err != nil {
-		return nil, err
-	}
-	nodeConnectionInfoGetter, err := kubeletclient.NewNodeConnectionInfoGetter(kubeExternalClient.CoreV1().Nodes(), *c.ExtraConfig.KubeletClientConfig)
-	if err != nil {
-		return nil, fmt.Errorf("unable to configure the node connection info getter: %v", err)
-	}
 
 	deployConfigStorage, deployConfigStatusStorage, deployConfigScaleStorage, err := deployconfigetcd.NewREST(c.GenericConfig.RESTOptionsGetter)
 	if err != nil {
@@ -148,7 +136,7 @@ func (c *completedConfig) newV1RESTStorage() (map[string]rest.Storage, error) {
 	v1Storage["deploymentConfigs/scale"] = deployConfigScaleStorage
 	v1Storage["deploymentConfigs/status"] = deployConfigStatusStorage
 	v1Storage["deploymentConfigs/rollback"] = deployConfigRollbackStorage
-	v1Storage["deploymentConfigs/log"] = deploylogregistry.NewREST(openshiftInternalAppsClient.Apps(), kubeInternalClient.Core(), kubeInternalClient.Core(), nodeConnectionInfoGetter)
+	v1Storage["deploymentConfigs/log"] = deploylogregistry.NewREST(openshiftInternalAppsClient.Apps(), kubeInternalClient.Core(), kubeInternalClient.Core())
 	v1Storage["deploymentConfigs/instantiate"] = dcInstantiateStorage
 	return v1Storage, nil
 }

pkg/apps/registry/deploylog/rest.go

@@ -2,6 +2,7 @@ package deploylog
 
 import (
 	"fmt"
+	"io"
 	"sort"
 	"time"
 
@@ -11,6 +12,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/wait"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericrest "k8s.io/apiserver/pkg/registry/generic/rest"
@@ -19,8 +21,6 @@ import (
 	kcoreclient "k8s.io/kubernetes/pkg/client/clientset_generated/internalclientset/typed/core/internalversion"
 	"k8s.io/kubernetes/pkg/controller"
 	kcmdutil "k8s.io/kubernetes/pkg/kubectl/cmd/util"
-	kubeletclient "k8s.io/kubernetes/pkg/kubelet/client"
-	"k8s.io/kubernetes/pkg/registry/core/pod"
 
 	appsapi "github.com/openshift/origin/pkg/apps/apis/apps"
 	"github.com/openshift/origin/pkg/apps/apis/apps/validation"
@@ -36,29 +36,16 @@ const (
 	defaultInterval = 1 * time.Second
 )
 
-// podGetter implements the ResourceGetter interface. Used by LogLocation to
-// retrieve the deployer pod
-type podGetter struct {
-	pn kcoreclient.PodsGetter
-}
-
-// Get is responsible for retrieving the deployer pod
-func (g *podGetter) Get(ctx apirequest.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
-	namespace, ok := apirequest.NamespaceFrom(ctx)
-	if !ok {
-		return nil, errors.NewBadRequest("namespace parameter required.")
-	}
-	return g.pn.Pods(namespace).Get(name, *options)
-}
-
 // REST is an implementation of RESTStorage for the api server.
 type REST struct {
-	dn       appsclient.DeploymentConfigsGetter
-	rn       kcoreclient.ReplicationControllersGetter
-	pn       kcoreclient.PodsGetter
-	connInfo kubeletclient.ConnectionInfoGetter
-	timeout  time.Duration
-	interval time.Duration
+	dcClient  appsclient.DeploymentConfigsGetter
+	rcClient  kcoreclient.ReplicationControllersGetter
+	podClient kcoreclient.PodsGetter
+	timeout   time.Duration
+	interval  time.Duration
+
+	// for unit testing
+	getLogsFn func(podNamespace, podName string, logOpts *kapi.PodLogOptions) (runtime.Object, error)
 }
 
 // REST implements GetterWithOptions
@@ -68,15 +55,17 @@ var _ = rest.GetterWithOptions(&REST{})
 // one for deployments (replication controllers) and one for pods to get the necessary
 // attributes to assemble the URL to which the request shall be redirected in order to
 // get the deployment logs.
-func NewREST(dn appsclient.DeploymentConfigsGetter, rn kcoreclient.ReplicationControllersGetter, pn kcoreclient.PodsGetter, connectionInfo kubeletclient.ConnectionInfoGetter) *REST {
-	return &REST{
-		dn:       dn,
-		rn:       rn,
-		pn:       pn,
-		connInfo: connectionInfo,
-		timeout:  defaultTimeout,
-		interval: defaultInterval,
+func NewREST(dcClient appsclient.DeploymentConfigsGetter, rcClient kcoreclient.ReplicationControllersGetter, podClient kcoreclient.PodsGetter) *REST {
+	r := &REST{
+		dcClient:  dcClient,
+		rcClient:  rcClient,
+		podClient: podClient,
+		timeout:   defaultTimeout,
+		interval:  defaultInterval,
 	}
+	r.getLogsFn = r.getLogs
+
+	return r
 }
 
 // NewGetOptions returns a new options object for deployment logs
@@ -108,7 +97,7 @@ func (r *REST) Get(ctx apirequest.Context, name string, opts runtime.Object) (ru
 
 	// Fetch deploymentConfig and check latest version; if 0, there are no deployments
 	// for this config
-	config, err := r.dn.DeploymentConfigs(namespace).Get(name, metav1.GetOptions{})
+	config, err := r.dcClient.DeploymentConfigs(namespace).Get(name, metav1.GetOptions{})
 	if err != nil {
 		return nil, errors.NewNotFound(appsapi.Resource("deploymentconfig"), name)
 	}
@@ -154,11 +143,11 @@ func (r *REST) Get(ctx apirequest.Context, name string, opts runtime.Object) (ru
 		}
 		glog.V(4).Infof("Deployment %s is in %s state, waiting for it to start...", appsutil.LabelForDeployment(target), status)
 
-		if err := appsutil.WaitForRunningDeployerPod(r.pn, target, r.timeout); err != nil {
+		if err := appsutil.WaitForRunningDeployerPod(r.podClient, target, r.timeout); err != nil {
 			return nil, errors.NewBadRequest(fmt.Sprintf("failed to run deployer pod %s: %v", podName, err))
 		}
 
-		latest, ok, err := registry.WaitForRunningDeployment(r.rn, target, r.timeout)
+		latest, ok, err := registry.WaitForRunningDeployment(r.rcClient, target, r.timeout)
 		if err != nil {
 			return nil, errors.NewBadRequest(fmt.Sprintf("unable to wait for deployment %s to run: %v", appsutil.LabelForDeployment(target), err))
 		}
@@ -179,20 +168,46 @@ func (r *REST) Get(ctx apirequest.Context, name string, opts runtime.Object) (ru
 	}
 
 	logOpts := appsapi.DeploymentToPodLogOptions(deployLogOpts)
-	location, transport, err := pod.LogLocation(&podGetter{r.pn}, r.connInfo, ctx, podName, logOpts)
+	return r.getLogsFn(namespace, podName, logOpts)
+}
+
+func (r *REST) getLogs(podNamespace, podName string, logOpts *kapi.PodLogOptions) (runtime.Object, error) {
+	logRequest := r.podClient.Pods(podNamespace).GetLogs(podName, logOpts)
+
+	readerCloser, err := logRequest.Stream()
 	if err != nil {
-		return nil, errors.NewBadRequest(err.Error())
+		return nil, err
 	}
 
-	return &genericrest.LocationStreamer{
-		Location:        location,
-		Transport:       transport,
-		ContentType:     "text/plain",
-		Flush:           deployLogOpts.Follow,
-		ResponseChecker: genericrest.NewGenericHttpResponseChecker(kapi.Resource("pod"), podName),
+	return &passThroughStreamer{
+		In:          readerCloser,
+		Flush:       logOpts.Follow,
+		ContentType: "text/plain",
 	}, nil
 }
 
+type passThroughStreamer struct {
+	In          io.ReadCloser
+	Flush       bool
+	ContentType string
+}
+
+// a PipeStreamer must implement a rest.ResourceStreamer
+var _ rest.ResourceStreamer = &passThroughStreamer{}
+
+func (obj *passThroughStreamer) GetObjectKind() schema.ObjectKind {
+	return schema.EmptyObjectKind
+}
+
+func (obj *passThroughStreamer) DeepCopyObject() runtime.Object {
+	panic("passThroughStreamer does not implement DeepCopyObject")
+}
+
+// InputStream returns a stream with the contents of the embedded pipe.
+func (s *passThroughStreamer) InputStream(apiVersion, acceptHeader string) (stream io.ReadCloser, flush bool, contentType string, err error) {
+	return s.In, s.Flush, s.ContentType, nil
+}
+
 // waitForExistingDeployment will use the timeout to wait for a deployment to appear.
 func (r *REST) waitForExistingDeployment(namespace, name string) (*kapi.ReplicationController, error) {
 	var (
@@ -201,7 +216,7 @@ func (r *REST) waitForExistingDeployment(namespace, name string) (*kapi.Replicat
 	)
 
 	condition := func() (bool, error) {
-		target, err = r.rn.ReplicationControllers(namespace).Get(name, metav1.GetOptions{})
+		target, err = r.rcClient.ReplicationControllers(namespace).Get(name, metav1.GetOptions{})
 		switch {
 		case errors.IsNotFound(err):
 			return false, nil
@@ -224,7 +239,7 @@ func (r *REST) returnApplicationPodName(target *kapi.ReplicationController) (str
 	selector := labels.SelectorFromValidatedSet(labels.Set(target.Spec.Selector))
 	sortBy := func(pods []*kapiv1.Pod) sort.Interface { return controller.ByLogging(pods) }
 
-	firstPod, _, err := kcmdutil.GetFirstPod(r.pn, target.Namespace, selector.String(), r.timeout, sortBy)
+	firstPod, _, err := kcmdutil.GetFirstPod(r.podClient, target.Namespace, selector.String(), r.timeout, sortBy)
 	if err != nil {
 		return "", errors.NewInternalError(err)
 	}