switch build logs to use client, not storage

Author: deads2k

pkg/build/apiserver/apiserver.go

@@ -161,12 +161,12 @@ func (c *BuildServerConfig) newV1RESTStorage() (map[string]rest.Storage, error)
 	v1Storage := map[string]rest.Storage{}
 	v1Storage["builds"] = buildStorage
 	v1Storage["builds/clone"] = buildclone.NewStorage(buildGenerator)
-	v1Storage["builds/log"] = buildlogregistry.NewREST(buildStorage, buildStorage, kubeInternalClient.Core(), nodeConnectionInfoGetter)
+	v1Storage["builds/log"] = buildlogregistry.NewREST(buildClient.Build(), kubeInternalClient.Core(), nodeConnectionInfoGetter)
 	v1Storage["builds/details"] = buildDetailsStorage
 
 	v1Storage["buildConfigs"] = buildConfigStorage
 	v1Storage["buildConfigs/webhooks"] = buildConfigWebHooks
 	v1Storage["buildConfigs/instantiate"] = buildconfiginstantiate.NewStorage(buildGenerator)
-	v1Storage["buildConfigs/instantiatebinary"] = buildconfiginstantiate.NewBinaryStorage(buildGenerator, buildStorage, kubeInternalClient.Core(), nodeConnectionInfoGetter)
+	v1Storage["buildConfigs/instantiatebinary"] = buildconfiginstantiate.NewBinaryStorage(buildGenerator, buildClient.Build(), kubeInternalClient.Core(), nodeConnectionInfoGetter)
 	return v1Storage, nil
 }

pkg/build/registry/buildconfiginstantiate/rest.go

@@ -27,6 +27,7 @@ import (
 	buildapi "github.com/openshift/origin/pkg/build/apis/build"
 	buildapiv1 "github.com/openshift/origin/pkg/build/apis/build/v1"
 	buildstrategy "github.com/openshift/origin/pkg/build/controller/strategy"
+	buildtypedclient "github.com/openshift/origin/pkg/build/generated/internalclientset/typed/build/internalversion"
 	"github.com/openshift/origin/pkg/build/generator"
 	"github.com/openshift/origin/pkg/build/registry"
 	buildutil "github.com/openshift/origin/pkg/build/util"
@@ -84,10 +85,10 @@ func (s *InstantiateREST) ProducesMIMETypes(verb string) []string {
 
 var _ rest.StorageMetadata = &InstantiateREST{}
 
-func NewBinaryStorage(generator *generator.BuildGenerator, watcher rest.Watcher, podClient kcoreclient.PodsGetter, info kubeletclient.ConnectionInfoGetter) *BinaryInstantiateREST {
+func NewBinaryStorage(generator *generator.BuildGenerator, buildClient buildtypedclient.BuildsGetter, podClient kcoreclient.PodsGetter, info kubeletclient.ConnectionInfoGetter) *BinaryInstantiateREST {
 	return &BinaryInstantiateREST{
 		Generator:      generator,
-		Watcher:        watcher,
+		BuildClient:    buildClient,
 		PodGetter:      &podGetter{podClient},
 		ConnectionInfo: info,
 		Timeout:        5 * time.Minute,
@@ -96,7 +97,7 @@ func NewBinaryStorage(generator *generator.BuildGenerator, watcher rest.Watcher,
 
 type BinaryInstantiateREST struct {
 	Generator      *generator.BuildGenerator
-	Watcher        rest.Watcher
+	BuildClient    buildtypedclient.BuildsGetter
 	PodGetter      pod.ResourceGetter
 	ConnectionInfo kubeletclient.ConnectionInfoGetter
 	Timeout        time.Duration
@@ -224,7 +225,7 @@ func (h *binaryInstantiateHandler) handle(r io.Reader) (runtime.Object, error) {
 		h.cancelBuild(build)
 	}()
 
-	latest, ok, err := registry.WaitForRunningBuild(h.r.Watcher, h.ctx, build, remaining)
+	latest, ok, err := registry.WaitForRunningBuild(h.r.BuildClient, build, remaining)
 
 	switch {
 	case latest.Status.Phase == buildapi.BuildPhaseError:

pkg/build/registry/buildlog/rest.go

@@ -23,19 +23,20 @@ import (
 
 	buildapi "github.com/openshift/origin/pkg/build/apis/build"
 	"github.com/openshift/origin/pkg/build/apis/build/validation"
+	buildtypedclient "github.com/openshift/origin/pkg/build/generated/internalclientset/typed/build/internalversion"
 	"github.com/openshift/origin/pkg/build/registry"
 	buildutil "github.com/openshift/origin/pkg/build/util"
 )
 
 // REST is an implementation of RESTStorage for the api server.
 type REST struct {
-	Getter         rest.Getter
-	Watcher        rest.Watcher
+	BuildClient    buildtypedclient.BuildsGetter
 	PodGetter      pod.ResourceGetter
 	ConnectionInfo kubeletclient.ConnectionInfoGetter
 	Timeout        time.Duration
 }
 
+// TODO these wrapers shouldb e removed
 type podGetter struct {
 	kcoreclient.PodsGetter
 }
@@ -53,10 +54,9 @@ const defaultTimeout time.Duration = 10 * time.Second
 // NewREST creates a new REST for BuildLog
 // Takes build registry and pod client to get necessary attributes to assemble
 // URL to which the request shall be redirected in order to get build logs.
-func NewREST(getter rest.Getter, watcher rest.Watcher, pn kcoreclient.PodsGetter, connectionInfo kubeletclient.ConnectionInfoGetter) *REST {
+func NewREST(buildClient buildtypedclient.BuildsGetter, pn kcoreclient.PodsGetter, connectionInfo kubeletclient.ConnectionInfoGetter) *REST {
 	return &REST{
-		Getter:         getter,
-		Watcher:        watcher,
+		BuildClient:    buildClient,
 		PodGetter:      &podGetter{pn},
 		ConnectionInfo: connectionInfo,
 		Timeout:        defaultTimeout,
@@ -74,21 +74,20 @@ func (r *REST) Get(ctx apirequest.Context, name string, opts runtime.Object) (ru
 	if errs := validation.ValidateBuildLogOptions(buildLogOpts); len(errs) > 0 {
 		return nil, errors.NewInvalid(buildapi.Kind("BuildLogOptions"), "", errs)
 	}
-	obj, err := r.Getter.Get(ctx, name, &metav1.GetOptions{})
+	build, err := r.BuildClient.Builds(apirequest.NamespaceValue(ctx)).Get(name, metav1.GetOptions{})
 	if err != nil {
 		return nil, err
 	}
-	build := obj.(*buildapi.Build)
 	if buildLogOpts.Previous {
 		version := buildutil.VersionForBuild(build)
 		// Use the previous version
 		version--
 		previousBuildName := buildutil.BuildNameForConfigVersion(buildutil.ConfigNameForBuild(build), version)
-		previous, err := r.Getter.Get(ctx, previousBuildName, &metav1.GetOptions{})
+		previous, err := r.BuildClient.Builds(apirequest.NamespaceValue(ctx)).Get(previousBuildName, metav1.GetOptions{})
 		if err != nil {
 			return nil, err
 		}
-		build = previous.(*buildapi.Build)
+		build = previous
 	}
 	switch build.Status.Phase {
 	// Build has not launched, wait until it runs
@@ -99,7 +98,7 @@ func (r *REST) Get(ctx apirequest.Context, name string, opts runtime.Object) (ru
 			return &genericrest.LocationStreamer{}, nil
 		}
 		glog.V(4).Infof("Build %s/%s is in %s state, waiting for Build to start", build.Namespace, build.Name, build.Status.Phase)
-		latest, ok, err := registry.WaitForRunningBuild(r.Watcher, ctx, build, r.Timeout)
+		latest, ok, err := registry.WaitForRunningBuild(r.BuildClient, build, r.Timeout)
 		if err != nil {
 			return nil, errors.NewBadRequest(fmt.Sprintf("unable to wait for build %s to run: %v", build.Name, err))
 		}
@@ -126,7 +125,7 @@ func (r *REST) Get(ctx apirequest.Context, name string, opts runtime.Object) (ru
 
 	// if we can't at least get the build pod, we're not going to get very far, so
 	// error out now.
-	obj, err = r.PodGetter.Get(ctx, buildPodName, &metav1.GetOptions{})
+	obj, err := r.PodGetter.Get(ctx, buildPodName, &metav1.GetOptions{})
 	if err != nil {
 		return nil, errors.NewBadRequest(err.Error())
 	}

pkg/build/registry/buildlog/rest_test.go

@@ -8,19 +8,19 @@ import (
 	"testing"
 	"time"
 
-	metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/watch"
 	apirequest "k8s.io/apiserver/pkg/endpoints/request"
 	genericrest "k8s.io/apiserver/pkg/registry/generic/rest"
 	"k8s.io/apiserver/pkg/registry/rest"
+	clientgotesting "k8s.io/client-go/testing"
 	kapi "k8s.io/kubernetes/pkg/api"
 	kubeletclient "k8s.io/kubernetes/pkg/kubelet/client"
 
 	buildapi "github.com/openshift/origin/pkg/build/apis/build"
-	"github.com/openshift/origin/pkg/build/registry/test"
+	buildfakeclient "github.com/openshift/origin/pkg/build/generated/internalclientset/fake"
 )
 
 type testPodGetter struct{}
@@ -139,26 +139,20 @@ func TestWaitForBuild(t *testing.T) {
 
 	for _, tt := range tests {
 		build := mockBuild(buildapi.BuildPhasePending, "running", 1)
-		ch := make(chan watch.Event)
-		watcher := &buildWatcher{
-			Build: build,
-			Watcher: &fakeWatch{
-				Channel: ch,
-			},
-		}
+		buildClient := buildfakeclient.NewSimpleClientset(build)
+		fakeWatcher := watch.NewFake()
+		buildClient.PrependWatchReactor("builds", func(action clientgotesting.Action) (handled bool, ret watch.Interface, err error) {
+			return true, fakeWatcher, nil
+		})
 		storage := REST{
-			Getter:         watcher,
-			Watcher:        watcher,
+			BuildClient:    buildClient.Build(),
 			PodGetter:      &testPodGetter{},
 			ConnectionInfo: &fakeConnectionInfoGetter{},
 			Timeout:        defaultTimeout,
 		}
 		go func() {
 			for _, status := range tt.status {
-				ch <- watch.Event{
-					Type:   watch.Modified,
-					Object: mockBuild(status, "running", 1),
-				}
+				fakeWatcher.Modify(mockBuild(status, "running", 1))
 			}
 		}()
 		_, err := storage.Get(ctx, build.Name, &buildapi.BuildLogOptions{})
@@ -172,18 +166,11 @@ func TestWaitForBuild(t *testing.T) {
 }
 
 func TestWaitForBuildTimeout(t *testing.T) {
-	ctx := apirequest.NewDefaultContext()
 	build := mockBuild(buildapi.BuildPhasePending, "running", 1)
-	ch := make(chan watch.Event)
-	watcher := &buildWatcher{
-		Build: build,
-		Watcher: &fakeWatch{
-			Channel: ch,
-		},
-	}
+	buildClient := buildfakeclient.NewSimpleClientset(build)
+	ctx := apirequest.NewDefaultContext()
 	storage := REST{
-		Getter:         watcher,
-		Watcher:        watcher,
+		BuildClient:    buildClient.Build(),
 		PodGetter:      &testPodGetter{},
 		ConnectionInfo: &fakeConnectionInfoGetter{},
 		Timeout:        100 * time.Millisecond,
@@ -194,44 +181,18 @@ func TestWaitForBuildTimeout(t *testing.T) {
 	}
 }
 
-type buildWatcher struct {
-	Build   *buildapi.Build
-	Watcher watch.Interface
-	Err     error
-}
-
-func (r *buildWatcher) Get(ctx apirequest.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
-	return r.Build, nil
-}
-
-func (r *buildWatcher) Watch(ctx apirequest.Context, options *metainternal.ListOptions) (watch.Interface, error) {
-	return r.Watcher, r.Err
-}
-
-type fakeWatch struct {
-	Channel chan watch.Event
-}
-
-func (w *fakeWatch) Stop() {
-	close(w.Channel)
-}
-
-func (w *fakeWatch) ResultChan() <-chan watch.Event {
-	return w.Channel
-}
-
 func resourceLocationHelper(BuildPhase buildapi.BuildPhase, podPhase string, ctx apirequest.Context, version int) (string, error) {
 	expectedBuild := mockBuild(BuildPhase, podPhase, version)
-	internal := &test.BuildStorage{Build: expectedBuild}
+	buildClient := buildfakeclient.NewSimpleClientset(expectedBuild)
 
 	storage := &REST{
-		Getter:         internal,
+		BuildClient:    buildClient.Build(),
 		PodGetter:      &testPodGetter{},
 		ConnectionInfo: &fakeConnectionInfoGetter{},
 		Timeout:        defaultTimeout,
 	}
 	getter := rest.GetterWithOptions(storage)
-	obj, err := getter.Get(ctx, "foo-build", &buildapi.BuildLogOptions{NoWait: true})
+	obj, err := getter.Get(ctx, expectedBuild.Name, &buildapi.BuildLogOptions{NoWait: true})
 	if err != nil {
 		return "", err
 	}
@@ -269,7 +230,8 @@ func mockPod(podPhase kapi.PodPhase, podName string) *kapi.Pod {
 func mockBuild(status buildapi.BuildPhase, podName string, version int) *buildapi.Build {
 	return &buildapi.Build{
 		ObjectMeta: metav1.ObjectMeta{
-			Name: podName,
+			Namespace: "default",
+			Name:      podName,
 			Annotations: map[string]string{
 				buildapi.BuildNumberAnnotation: strconv.Itoa(version),
 			},
@@ -303,10 +265,10 @@ func TestPreviousBuildLogs(t *testing.T) {
 	first := mockBuild(buildapi.BuildPhaseComplete, "bc-1", 1)
 	second := mockBuild(buildapi.BuildPhaseComplete, "bc-2", 2)
 	third := mockBuild(buildapi.BuildPhaseComplete, "bc-3", 3)
-	internal := &test.BuildStorage{Builds: &buildapi.BuildList{Items: []buildapi.Build{*first, *second, *third}}}
+	buildClient := buildfakeclient.NewSimpleClientset(first, second, third)
 
 	storage := &REST{
-		Getter:         internal,
+		BuildClient:    buildClient.Build(),
 		PodGetter:      &anotherTestPodGetter{},
 		ConnectionInfo: &fakeConnectionInfoGetter{},
 		Timeout:        defaultTimeout,

pkg/build/registry/rest.go

@@ -4,13 +4,12 @@ import (
 	"fmt"
 	"time"
 
-	metainternal "k8s.io/apimachinery/pkg/apis/meta/internalversion"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/watch"
-	apirequest "k8s.io/apiserver/pkg/endpoints/request"
-	"k8s.io/apiserver/pkg/registry/rest"
 
 	buildapi "github.com/openshift/origin/pkg/build/apis/build"
+	buildtypedclient "github.com/openshift/origin/pkg/build/generated/internalclientset/typed/build/internalversion"
 )
 
 var (
@@ -22,39 +21,38 @@ var (
 // WaitForRunningBuild waits until the specified build is no longer New or Pending. Returns true if
 // the build ran within timeout, false if it did not, and an error if any other error state occurred.
 // The last observed Build state is returned.
-func WaitForRunningBuild(watcher rest.Watcher, ctx apirequest.Context, build *buildapi.Build, timeout time.Duration) (*buildapi.Build, bool, error) {
+func WaitForRunningBuild(buildClient buildtypedclient.BuildsGetter, build *buildapi.Build, timeout time.Duration) (*buildapi.Build, bool, error) {
 	fieldSelector := fields.OneTermEqualSelector("metadata.name", build.Name)
-	options := &metainternal.ListOptions{FieldSelector: fieldSelector, ResourceVersion: build.ResourceVersion}
-	w, err := watcher.Watch(ctx, options)
+	options := metav1.ListOptions{FieldSelector: fieldSelector.String(), ResourceVersion: build.ResourceVersion}
+	w, err := buildClient.Builds(build.Namespace).Watch(options)
 	if err != nil {
 		return build, false, err
 	}
-	defer w.Stop()
 
 	observed := build
-	ch := w.ResultChan()
-	expire := time.After(timeout)
-	for {
-		select {
-		case event := <-ch:
-			obj, ok := event.Object.(*buildapi.Build)
-			if !ok {
-				return observed, false, fmt.Errorf("received unknown object while watching for builds")
-			}
-			observed = obj
-
-			if event.Type == watch.Deleted {
-				return observed, false, ErrBuildDeleted
-			}
-			switch obj.Status.Phase {
-			case buildapi.BuildPhaseRunning, buildapi.BuildPhaseComplete, buildapi.BuildPhaseFailed, buildapi.BuildPhaseError, buildapi.BuildPhaseCancelled:
-				return observed, true, nil
-			case buildapi.BuildPhaseNew, buildapi.BuildPhasePending:
-			default:
-				return observed, false, ErrUnknownBuildPhase
-			}
-		case <-expire:
-			return observed, false, nil
+	_, err = watch.Until(timeout, w, func(event watch.Event) (bool, error) {
+		obj, ok := event.Object.(*buildapi.Build)
+		if !ok {
+			return false, fmt.Errorf("received unknown object while watching for builds: %T", event.Object)
 		}
+		observed = obj
+
+		if event.Type == watch.Deleted {
+			return false, ErrBuildDeleted
+		}
+		switch obj.Status.Phase {
+		case buildapi.BuildPhaseRunning, buildapi.BuildPhaseComplete, buildapi.BuildPhaseFailed, buildapi.BuildPhaseError, buildapi.BuildPhaseCancelled:
+			return true, nil
+		case buildapi.BuildPhaseNew, buildapi.BuildPhasePending:
+		default:
+			return false, ErrUnknownBuildPhase
+		}
+
+		return false, nil
+	})
+	if err != nil {
+		return nil, false, err
 	}
+
+	return observed, true, nil
 }