github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-31 15:25:57 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #1452 from derekwaynecarr/introduce_context_obj

Browse Source 
Introduce a context object on RESTStorage to prepare for future changes
This commit is contained in:
Daniel Smith 2014-09-26 10:09:07 -07:00
commit 1170f4bea3
28 changed files with 1219 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Godeps/Godeps.json generated
View File

@ -14,6 +14,11 @@
"Comment": "null-12",
"Rev": "7dda39b2e7d5e265014674c5af696ba4186679e9"
},
{
"ImportPath": "code.google.com/p/go.net/context",
"Comment": "null-144",
"Rev": "ad01a6fcc8a19d3a4478c836895ffe883bd2ceab"
},
{
"ImportPath": "code.google.com/p/go.net/html",
"Comment": "null-144",

431
Godeps/_workspace/src/code.google.com/p/go.net/context/context.go generated vendored Normal file
View File

@ -0,0 +1,431 @@
// Copyright 2014 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Package context defines the Context type, which carries deadlines,
// cancelation signals, and other request-scoped values across API boundaries
// and between processes.
//
// Incoming requests to a server should create a Context, and outgoing calls to
// servers should accept a Context. The chain of function calls between must
// propagate the Context, optionally replacing it with a modified copy created
// using WithDeadline, WithTimeout, WithCancel, or WithValue.
//
// Programs that use Contexts should follow these rules to keep interfaces
// consistent across packages and enable static analysis tools to check context
// propagation:
//
// Do not store Contexts inside a struct type; instead, pass a Context
// explicitly to each function that needs it. The Context should be the first
// parameter, typically named ctx:
//
// func DoSomething(ctx context.Context, arg Arg) error {
// // ... use ctx ...
// }
//
// Do not pass a nil Context, even if a function permits it. Pass context.TODO
// if you are unsure about which Context to use.
//
// Use context Values only for request-scoped data that transits processes and
// APIs, not for passing optional parameters to functions.
//
// The same Context may be passed to functions running in different goroutines;
// Contexts are safe for simultaneous use by multiple goroutines.
//
// See http://blog.golang.org/context for example code for a server that uses
// Contexts.
package context
import (
"errors"
"fmt"
"sync"
"time"
)
// A Context carries a deadline, a cancelation signal, and other values across
// API boundaries.
//
// Context's methods may be called by multiple goroutines simultaneously.
type Context interface {
// Deadline returns the time when work done on behalf of this context
// should be canceled. Deadline returns ok==false when no deadline is
// set. Successive calls to Deadline return the same results.
Deadline() (deadline time.Time, ok bool)
// Done returns a channel that's closed when work done on behalf of this
// context should be canceled. Done may return nil if this context can
// never be canceled. Successive calls to Done return the same value.
//
// WithCancel arranges for Done to be closed when cancel is called;
// WithDeadline arranges for Done to be closed when the deadline
// expires; WithTimeout arranges for Done to be closed when the timeout
// elapses.
//
// Done is provided for use in select statements:
//
// // DoSomething calls DoSomethingSlow and returns as soon as
// // it returns or ctx.Done is closed.
// func DoSomething(ctx context.Context) (Result, error) {
// c := make(chan Result, 1)
// go func() { c <- DoSomethingSlow(ctx) }()
// select {
// case res := <-c:
// return res, nil
// case <-ctx.Done():
// return nil, ctx.Err()
// }
// }
//
// See http://blog.golang.org/pipelines for more examples of how to use
// a Done channel for cancelation.
Done() <-chan struct{}
// Err returns a non-nil error value after Done is closed. Err returns
// Canceled if the context was canceled or DeadlineExceeded if the
// context's deadline passed. No other values for Err are defined.
// After Done is closed, successive calls to Err return the same value.
Err() error
// Value returns the value associated with this context for key, or nil
// if no value is associated with key. Successive calls to Value with
// the same key returns the same result.
//
// Use context values only for request-scoped data that transits
// processes and API boundaries, not for passing optional parameters to
// functions.
//
// A key identifies a specific value in a Context. Functions that wish
// to store values in Context typically allocate a key in a global
// variable then use that key as the argument to context.WithValue and
// Context.Value. A key can be any type that supports equality;
// packages should define keys as an unexported type to avoid
// collisions.
//
// Packages that define a Context key should provide type-safe accessors
// for the values stores using that key:
//
// // Package user defines a User type that's stored in Contexts.
// package user
//
// import "code.google.com/p/go.net/context"
//
// // User is the type of value stored in the Contexts.
// type User struct {...}
//
// // key is an unexported type for keys defined in this package.
// // This prevents collisions with keys defined in other packages.
// type key int
//
// // userKey is the key for user.User values in Contexts. It is
// // unexported; clients use user.NewContext and user.FromContext
// // instead of using this key directly.
// var userKey key = 0
//
// // NewContext returns a new Context that carries value u.
// func NewContext(ctx context.Context, u *User) context.Context {
// return context.WithValue(userKey, u)
// }
//
// // FromContext returns the User value stored in ctx, if any.
// func FromContext(ctx context.Context) (*User, bool) {
// u, ok := ctx.Value(userKey).(*User)
// return u, ok
// }
Value(key interface{}) interface{}
}
// Canceled is the error returned by Context.Err when the context is canceled.
var Canceled = errors.New("context canceled")
// DeadlineExceeded is the error returned by Context.Err when the context's
// deadline passes.
var DeadlineExceeded = errors.New("context deadline exceeded")
// An emptyCtx is never canceled, has no values, and has no deadline.
type emptyCtx int
func (emptyCtx) Deadline() (deadline time.Time, ok bool) {
return
}
func (emptyCtx) Done() <-chan struct{} {
return nil
}
func (emptyCtx) Err() error {
return nil
}
func (emptyCtx) Value(key interface{}) interface{} {
return nil
}
func (n emptyCtx) String() string {
switch n {
case background:
return "context.Background"
case todo:
return "context.TODO"
}
return "unknown empty Context"
}
const (
background emptyCtx = 1
todo emptyCtx = 2
)
// Background returns a non-nil, empty Context. It is never canceled, has no
// values, and has no deadline. It is typically used by the main function,
// initialization, and tests, and as the top-level Context for incoming
// requests.
func Background() Context {
return background
}
// TODO returns a non-nil, empty Context. Code should use context.TODO when
// it's unclear which Context to use or it's is not yet available (because the
// surrounding function has not yet been extended to accept a Context
// parameter). TODO is recognized by static analysis tools that determine
// whether Contexts are propagated correctly in a program.
func TODO() Context {
return todo
}
// A CancelFunc tells an operation to abandon its work.
// A CancelFunc does not wait for the work to stop.
// After the first call, subsequent calls to a CancelFunc do nothing.
type CancelFunc func()
// WithCancel returns a copy of parent with a new Done channel. The returned
// context's Done channel is closed when the returned cancel function is called
// or when the parent context's Done channel is closed, whichever happens first.
func WithCancel(parent Context) (ctx Context, cancel CancelFunc) {
c := newCancelCtx(parent)
propagateCancel(parent, &c)
return &c, func() { c.cancel(true, Canceled) }
}
// newCancelCtx returns an initialized cancelCtx.
func newCancelCtx(parent Context) cancelCtx {
return cancelCtx{
Context: parent,
done: make(chan struct{}),
}
}
// propagateCancel arranges for child to be canceled when parent is.
func propagateCancel(parent Context, child canceler) {
if parent.Done() == nil {
return // parent is never canceled
}
if p, ok := parentCancelCtx(parent); ok {
p.mu.Lock()
if p.err != nil {
// parent has already been canceled
child.cancel(false, p.err)
} else {
if p.children == nil {
p.children = make(map[canceler]bool)
}
p.children[child] = true
}
p.mu.Unlock()
} else {
go func() {
select {
case <-parent.Done():
child.cancel(false, parent.Err())
case <-child.Done():
}
}()
}
}
// parentCancelCtx follows a chain of parent references until it finds a
// *cancelCtx. This function understands how each of the concrete types in this
// package represents its parent.
func parentCancelCtx(parent Context) (*cancelCtx, bool) {
for {
switch c := parent.(type) {
case *cancelCtx:
return c, true
case *timerCtx:
return &c.cancelCtx, true
case *valueCtx:
parent = c.Context
default:
return nil, false
}
}
}
// A canceler is a context type that can be canceled directly. The
// implementations are *cancelCtx and *timerCtx.
type canceler interface {
cancel(removeFromParent bool, err error)
Done() <-chan struct{}
}
// A cancelCtx can be canceled. When canceled, it also cancels any children
// that implement canceler.
type cancelCtx struct {
Context
done chan struct{} // closed by the first cancel call.
mu sync.Mutex
children map[canceler]bool // set to nil by the first cancel call
err error // set to non-nil by the first cancel call
}
func (c *cancelCtx) Done() <-chan struct{} {
return c.done
}
func (c *cancelCtx) Err() error {
c.mu.Lock()
defer c.mu.Unlock()
return c.err
}
func (c *cancelCtx) String() string {
return fmt.Sprintf("%v.WithCancel", c.Context)
}
// cancel closes c.done, cancels each of c's children, and, if
// removeFromParent is true, removes c from its parent's children.
func (c *cancelCtx) cancel(removeFromParent bool, err error) {
if err == nil {
panic("context: internal error: missing cancel error")
}
c.mu.Lock()
if c.err != nil {
c.mu.Unlock()
return // already canceled
}
c.err = err
close(c.done)
for child := range c.children {
// NOTE: acquiring the child's lock while holding parent's lock.
child.cancel(false, err)
}
c.children = nil
c.mu.Unlock()
if removeFromParent {
if p, ok := parentCancelCtx(c.Context); ok {
p.mu.Lock()
if p.children != nil {
delete(p.children, c)
}
p.mu.Unlock()
}
}
}
// WithDeadline returns a copy of the parent context with the deadline adjusted
// to be no later than d. If the parent's deadline is already earlier than d,
// WithDeadline(parent, d) is semantically equivalent to parent. The returned
// context's Done channel is closed when the deadline expires, when the returned
// cancel function is called, or when the parent context's Done channel is
// closed, whichever happens first.
//
// Canceling this context releases resources associated with the deadline
// timer, so code should call cancel as soon as the operations running in this
// Context complete.
func WithDeadline(parent Context, deadline time.Time) (Context, CancelFunc) {
if cur, ok := parent.Deadline(); ok && cur.Before(deadline) {
// The current deadline is already sooner than the new one.
return WithCancel(parent)
}
c := &timerCtx{
cancelCtx: newCancelCtx(parent),
deadline: deadline,
}
propagateCancel(parent, c)
d := deadline.Sub(time.Now())
if d <= 0 {
c.cancel(true, DeadlineExceeded) // deadline has already passed
return c, func() { c.cancel(true, Canceled) }
}
c.mu.Lock()
defer c.mu.Unlock()
if c.err == nil {
c.timer = time.AfterFunc(d, func() {
c.cancel(true, DeadlineExceeded)
})
}
return c, func() { c.cancel(true, Canceled) }
}
// A timerCtx carries a timer and a deadline. It embeds a cancelCtx to
// implement Done and Err. It implements cancel by stopping its timer then
// delegating to cancelCtx.cancel.
type timerCtx struct {
cancelCtx
timer *time.Timer // Under cancelCtx.mu.
deadline time.Time
}
func (c *timerCtx) Deadline() (deadline time.Time, ok bool) {
return c.deadline, true
}
func (c *timerCtx) String() string {
return fmt.Sprintf("%v.WithDeadline(%s [%s])", c.cancelCtx.Context, c.deadline, c.deadline.Sub(time.Now()))
}
func (c *timerCtx) cancel(removeFromParent bool, err error) {
c.cancelCtx.cancel(removeFromParent, err)
c.mu.Lock()
if c.timer != nil {
c.timer.Stop()
c.timer = nil
}
c.mu.Unlock()
}
// WithTimeout returns WithDeadline(parent, time.Now().Add(timeout)).
//
// Canceling this context releases resources associated with the deadline
// timer, so code should call cancel as soon as the operations running in this
// Context complete:
//
// func slowOperationWithTimeout(ctx context.Context) (Result, error) {
// ctx, cancel := context.WithTimeout(ctx, 100*time.Millisecond)
// defer cancel() // releases resources if slowOperation completes before timeout elapses
// return slowOperation(ctx)
// }
func WithTimeout(parent Context, timeout time.Duration) (Context, CancelFunc) {
return WithDeadline(parent, time.Now().Add(timeout))
}
// WithValue returns a copy of parent in which the value associated with key is
// val.
//
// Use context Values only for request-scoped data that transits processes and
// APIs, not for passing optional parameters to functions.
func WithValue(parent Context, key interface{}, val interface{}) Context {
return &valueCtx{parent, key, val}
}
// A valueCtx carries a key-value pair. It implements Value for that key and
// delegates all other calls to the embedded Context.
type valueCtx struct {
Context
key, val interface{}
}
func (c *valueCtx) String() string {
return fmt.Sprintf("%v.WithValue(%#v, %#v)", c.Context, c.key, c.val)
}
func (c *valueCtx) Value(key interface{}) interface{} {
if c.key == key {
return c.val
}
return c.Context.Value(key)
}

553
Godeps/_workspace/src/code.google.com/p/go.net/context/context_test.go generated vendored Normal file
View File

@ -0,0 +1,553 @@
// Copyright 2014 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package context
import (
"fmt"
"math/rand"
"runtime"
"strings"
"sync"
"testing"
"time"
)
// otherContext is a Context that's not one of the types defined in context.go.
// This lets us test code paths that differ based on the underlying type of the
// Context.
type otherContext struct {
Context
}
func TestBackground(t *testing.T) {
c := Background()
if c == nil {
t.Fatalf("Background returned nil")
}
select {
case x := <-c.Done():
t.Errorf("<-c.Done() == %v want nothing (it should block)", x)
default:
}
if got, want := fmt.Sprint(c), "context.Background"; got != want {
t.Errorf("Background().String() = %q want %q", got, want)
}
}
func TestTODO(t *testing.T) {
c := TODO()
if c == nil {
t.Fatalf("TODO returned nil")
}
select {
case x := <-c.Done():
t.Errorf("<-c.Done() == %v want nothing (it should block)", x)
default:
}
if got, want := fmt.Sprint(c), "context.TODO"; got != want {
t.Errorf("TODO().String() = %q want %q", got, want)
}
}
func TestWithCancel(t *testing.T) {
c1, cancel := WithCancel(Background())
if got, want := fmt.Sprint(c1), "context.Background.WithCancel"; got != want {
t.Errorf("c1.String() = %q want %q", got, want)
}
o := otherContext{c1}
c2, _ := WithCancel(o)
contexts := []Context{c1, o, c2}
for i, c := range contexts {
if d := c.Done(); d == nil {
t.Errorf("c[%d].Done() == %v want non-nil", i, d)
}
if e := c.Err(); e != nil {
t.Errorf("c[%d].Err() == %v want nil", i, e)
}
select {
case x := <-c.Done():
t.Errorf("<-c.Done() == %v want nothing (it should block)", x)
default:
}
}
cancel()
time.Sleep(100 * time.Millisecond) // let cancelation propagate
for i, c := range contexts {
select {
case <-c.Done():
default:
t.Errorf("<-c[%d].Done() blocked, but shouldn't have", i)
}
if e := c.Err(); e != Canceled {
t.Errorf("c[%d].Err() == %v want %v", i, e, Canceled)
}
}
}
func TestParentFinishesChild(t *testing.T) {
// Context tree:
// parent -> cancelChild
// parent -> valueChild -> timerChild
parent, cancel := WithCancel(Background())
cancelChild, stop := WithCancel(parent)
defer stop()
valueChild := WithValue(parent, "key", "value")
timerChild, stop := WithTimeout(valueChild, 10000*time.Hour)
defer stop()
select {
case x := <-parent.Done():
t.Errorf("<-parent.Done() == %v want nothing (it should block)", x)
case x := <-cancelChild.Done():
t.Errorf("<-cancelChild.Done() == %v want nothing (it should block)", x)
case x := <-timerChild.Done():
t.Errorf("<-timerChild.Done() == %v want nothing (it should block)", x)
case x := <-valueChild.Done():
t.Errorf("<-valueChild.Done() == %v want nothing (it should block)", x)
default:
}
// The parent's children should contain the two cancelable children.
pc := parent.(*cancelCtx)
cc := cancelChild.(*cancelCtx)
tc := timerChild.(*timerCtx)
pc.mu.Lock()
if len(pc.children) != 2 || !pc.children[cc] || !pc.children[tc] {
t.Errorf("bad linkage: pc.children = %v, want %v and %v",
pc.children, cc, tc)
}
pc.mu.Unlock()
if p, ok := parentCancelCtx(cc.Context); !ok || p != pc {
t.Errorf("bad linkage: parentCancelCtx(cancelChild.Context) = %v, %v want %v, true", p, ok, pc)
}
if p, ok := parentCancelCtx(tc.Context); !ok || p != pc {
t.Errorf("bad linkage: parentCancelCtx(timerChild.Context) = %v, %v want %v, true", p, ok, pc)
}
cancel()
pc.mu.Lock()
if len(pc.children) != 0 {
t.Errorf("pc.cancel didn't clear pc.children = %v", pc.children)
}
pc.mu.Unlock()
// parent and children should all be finished.
check := func(ctx Context, name string) {
select {
case <-ctx.Done():
default:
t.Errorf("<-%s.Done() blocked, but shouldn't have", name)
}
if e := ctx.Err(); e != Canceled {
t.Errorf("%s.Err() == %v want %v", name, e, Canceled)
}
}
check(parent, "parent")
check(cancelChild, "cancelChild")
check(valueChild, "valueChild")
check(timerChild, "timerChild")
// WithCancel should return a canceled context on a canceled parent.
precanceledChild := WithValue(parent, "key", "value")
select {
case <-precanceledChild.Done():
default:
t.Errorf("<-precanceledChild.Done() blocked, but shouldn't have")
}
if e := precanceledChild.Err(); e != Canceled {
t.Errorf("precanceledChild.Err() == %v want %v", e, Canceled)
}
}
func TestChildFinishesFirst(t *testing.T) {
cancelable, stop := WithCancel(Background())
defer stop()
for _, parent := range []Context{Background(), cancelable} {
child, cancel := WithCancel(parent)
select {
case x := <-parent.Done():
t.Errorf("<-parent.Done() == %v want nothing (it should block)", x)
case x := <-child.Done():
t.Errorf("<-child.Done() == %v want nothing (it should block)", x)
default:
}
cc := child.(*cancelCtx)
pc, pcok := parent.(*cancelCtx) // pcok == false when parent == Background()
if p, ok := parentCancelCtx(cc.Context); ok != pcok || (ok && pc != p) {
t.Errorf("bad linkage: parentCancelCtx(cc.Context) = %v, %v want %v, %v", p, ok, pc, pcok)
}
if pcok {
pc.mu.Lock()
if len(pc.children) != 1 || !pc.children[cc] {
t.Errorf("bad linkage: pc.children = %v, cc = %v", pc.children, cc)
}
pc.mu.Unlock()
}
cancel()
if pcok {
pc.mu.Lock()
if len(pc.children) != 0 {
t.Errorf("child's cancel didn't remove self from pc.children = %v", pc.children)
}
pc.mu.Unlock()
}
// child should be finished.
select {
case <-child.Done():
default:
t.Errorf("<-child.Done() blocked, but shouldn't have")
}
if e := child.Err(); e != Canceled {
t.Errorf("child.Err() == %v want %v", e, Canceled)
}
// parent should not be finished.
select {
case x := <-parent.Done():
t.Errorf("<-parent.Done() == %v want nothing (it should block)", x)
default:
}
if e := parent.Err(); e != nil {
t.Errorf("parent.Err() == %v want nil", e)
}
}
}
func testDeadline(c Context, wait time.Duration, t *testing.T) {
select {
case <-time.After(wait):
t.Fatalf("context should have timed out")
case <-c.Done():
}
if e := c.Err(); e != DeadlineExceeded {
t.Errorf("c.Err() == %v want %v", e, DeadlineExceeded)
}
}
func TestDeadline(t *testing.T) {
c, _ := WithDeadline(Background(), time.Now().Add(100*time.Millisecond))
if got, prefix := fmt.Sprint(c), "context.Background.WithDeadline("; !strings.HasPrefix(got, prefix) {
t.Errorf("c.String() = %q want prefix %q", got, prefix)
}
testDeadline(c, 200*time.Millisecond, t)
c, _ = WithDeadline(Background(), time.Now().Add(100*time.Millisecond))
o := otherContext{c}
testDeadline(o, 200*time.Millisecond, t)
c, _ = WithDeadline(Background(), time.Now().Add(100*time.Millisecond))
o = otherContext{c}
c, _ = WithDeadline(o, time.Now().Add(300*time.Millisecond))
testDeadline(c, 200*time.Millisecond, t)
}
func TestTimeout(t *testing.T) {
c, _ := WithTimeout(Background(), 100*time.Millisecond)
if got, prefix := fmt.Sprint(c), "context.Background.WithDeadline("; !strings.HasPrefix(got, prefix) {
t.Errorf("c.String() = %q want prefix %q", got, prefix)
}
testDeadline(c, 200*time.Millisecond, t)
c, _ = WithTimeout(Background(), 100*time.Millisecond)
o := otherContext{c}
testDeadline(o, 200*time.Millisecond, t)
c, _ = WithTimeout(Background(), 100*time.Millisecond)
o = otherContext{c}
c, _ = WithTimeout(o, 300*time.Millisecond)
testDeadline(c, 200*time.Millisecond, t)
}
func TestCanceledTimeout(t *testing.T) {
c, _ := WithTimeout(Background(), 200*time.Millisecond)
o := otherContext{c}
c, cancel := WithTimeout(o, 400*time.Millisecond)
cancel()
time.Sleep(100 * time.Millisecond) // let cancelation propagate
select {
case <-c.Done():
default:
t.Errorf("<-c.Done() blocked, but shouldn't have")
}
if e := c.Err(); e != Canceled {
t.Errorf("c.Err() == %v want %v", e, Canceled)
}
}
type key1 int
type key2 int
var k1 = key1(1)
var k2 = key2(1) // same int as k1, different type
var k3 = key2(3) // same type as k2, different int
func TestValues(t *testing.T) {
check := func(c Context, nm, v1, v2, v3 string) {
if v, ok := c.Value(k1).(string); ok == (len(v1) == 0) || v != v1 {
t.Errorf(`%s.Value(k1).(string) = %q, %t want %q, %t`, nm, v, ok, v1, len(v1) != 0)
}
if v, ok := c.Value(k2).(string); ok == (len(v2) == 0) || v != v2 {
t.Errorf(`%s.Value(k2).(string) = %q, %t want %q, %t`, nm, v, ok, v2, len(v2) != 0)
}
if v, ok := c.Value(k3).(string); ok == (len(v3) == 0) || v != v3 {
t.Errorf(`%s.Value(k3).(string) = %q, %t want %q, %t`, nm, v, ok, v3, len(v3) != 0)
}
}
c0 := Background()
check(c0, "c0", "", "", "")
c1 := WithValue(Background(), k1, "c1k1")
check(c1, "c1", "c1k1", "", "")
if got, want := fmt.Sprint(c1), `context.Background.WithValue(1, "c1k1")`; got != want {
t.Errorf("c.String() = %q want %q", got, want)
}
c2 := WithValue(c1, k2, "c2k2")
check(c2, "c2", "c1k1", "c2k2", "")
c3 := WithValue(c2, k3, "c3k3")
check(c3, "c2", "c1k1", "c2k2", "c3k3")
c4 := WithValue(c3, k1, nil)
check(c4, "c4", "", "c2k2", "c3k3")
o0 := otherContext{Background()}
check(o0, "o0", "", "", "")
o1 := otherContext{WithValue(Background(), k1, "c1k1")}
check(o1, "o1", "c1k1", "", "")
o2 := WithValue(o1, k2, "o2k2")
check(o2, "o2", "c1k1", "o2k2", "")
o3 := otherContext{c4}
check(o3, "o3", "", "c2k2", "c3k3")
o4 := WithValue(o3, k3, nil)
check(o4, "o4", "", "c2k2", "")
}
func TestAllocs(t *testing.T) {
bg := Background()
for _, test := range []struct {
desc string
f func()
limit float64
gccgoLimit float64
}{
{
desc: "Background()",
f: func() { Background() },
limit: 0,
gccgoLimit: 0,
},
{
desc: fmt.Sprintf("WithValue(bg, %v, nil)", k1),
f: func() {
c := WithValue(bg, k1, nil)
c.Value(k1)
},
limit: 1,
gccgoLimit: 3,
},
{
desc: "WithTimeout(bg, 15*time.Millisecond)",
f: func() {
c, _ := WithTimeout(bg, 15*time.Millisecond)
<-c.Done()
},
limit: 8,
gccgoLimit: 13,
},
{
desc: "WithCancel(bg)",
f: func() {
c, cancel := WithCancel(bg)
cancel()
<-c.Done()
},
limit: 5,
gccgoLimit: 8,
},
{
desc: "WithTimeout(bg, 100*time.Millisecond)",
f: func() {
c, cancel := WithTimeout(bg, 100*time.Millisecond)
cancel()
<-c.Done()
},
limit: 8,
gccgoLimit: 25,
},
} {
limit := test.limit
if runtime.Compiler == "gccgo" {
// gccgo does not yet do escape analysis.
// TOOD(iant): Remove this when gccgo does do escape analysis.
limit = test.gccgoLimit
}
if n := testing.AllocsPerRun(100, test.f); n > limit {
t.Errorf("%s allocs = %f want %d", test.desc, n, int(limit))
}
}
}
func TestSimultaneousCancels(t *testing.T) {
root, cancel := WithCancel(Background())
m := map[Context]CancelFunc{root: cancel}
q := []Context{root}
// Create a tree of contexts.
for len(q) != 0 && len(m) < 100 {
parent := q[0]
q = q[1:]
for i := 0; i < 4; i++ {
ctx, cancel := WithCancel(parent)
m[ctx] = cancel
q = append(q, ctx)
}
}
// Start all the cancels in a random order.
var wg sync.WaitGroup
wg.Add(len(m))
for _, cancel := range m {
go func(cancel CancelFunc) {
cancel()
wg.Done()
}(cancel)
}
// Wait on all the contexts in a random order.
for ctx := range m {
select {
case <-ctx.Done():
case <-time.After(1 * time.Second):
buf := make([]byte, 10<<10)
n := runtime.Stack(buf, true)
t.Fatalf("timed out waiting for <-ctx.Done(); stacks:\n%s", buf[:n])
}
}
// Wait for all the cancel functions to return.
done := make(chan struct{})
go func() {
wg.Wait()
close(done)
}()
select {
case <-done:
case <-time.After(1 * time.Second):
buf := make([]byte, 10<<10)
n := runtime.Stack(buf, true)
t.Fatalf("timed out waiting for cancel functions; stacks:\n%s", buf[:n])
}
}
func TestInterlockedCancels(t *testing.T) {
parent, cancelParent := WithCancel(Background())
child, cancelChild := WithCancel(parent)
go func() {
parent.Done()
cancelChild()
}()
cancelParent()
select {
case <-child.Done():
case <-time.After(1 * time.Second):
buf := make([]byte, 10<<10)
n := runtime.Stack(buf, true)
t.Fatalf("timed out waiting for child.Done(); stacks:\n%s", buf[:n])
}
}
func TestLayersCancel(t *testing.T) {
testLayers(t, time.Now().UnixNano(), false)
}
func TestLayersTimeout(t *testing.T) {
testLayers(t, time.Now().UnixNano(), true)
}
func testLayers(t *testing.T, seed int64, testTimeout bool) {
rand.Seed(seed)
errorf := func(format string, a ...interface{}) {
t.Errorf(fmt.Sprintf("seed=%d: %s", seed, format), a...)
}
const (
timeout = 200 * time.Millisecond
minLayers = 30
)
type value int
var (
vals []*value
cancels []CancelFunc
numTimers int
ctx = Background()
)
for i := 0; i < minLayers || numTimers == 0 || len(cancels) == 0 || len(vals) == 0; i++ {
switch rand.Intn(3) {
case 0:
v := new(value)
ctx = WithValue(ctx, v, v)
vals = append(vals, v)
case 1:
var cancel CancelFunc
ctx, cancel = WithCancel(ctx)
cancels = append(cancels, cancel)
case 2:
var cancel CancelFunc
ctx, cancel = WithTimeout(ctx, timeout)
cancels = append(cancels, cancel)
numTimers++
}
}
checkValues := func(when string) {
for _, key := range vals {
if val := ctx.Value(key).(*value); key != val {
errorf("%s: ctx.Value(%p) = %p want %p", when, key, val, key)
}
}
}
select {
case <-ctx.Done():
errorf("ctx should not be canceled yet")
default:
}
if s, prefix := fmt.Sprint(ctx), "context.Background."; !strings.HasPrefix(s, prefix) {
t.Errorf("ctx.String() = %q want prefix %q", s, prefix)
}
t.Log(ctx)
checkValues("before cancel")
if testTimeout {
select {
case <-ctx.Done():
case <-time.After(timeout + timeout/10):
errorf("ctx should have timed out")
}
checkValues("after timeout")
} else {
cancel := cancels[rand.Intn(len(cancels))]
cancel()
select {
case <-ctx.Done():
default:
errorf("ctx should be canceled")
}
checkValues("after cancel")
}
}

26
Godeps/_workspace/src/code.google.com/p/go.net/context/withtimeout_test.go generated vendored Normal file
View File

@ -0,0 +1,26 @@
// Copyright 2014 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package context_test
import (
"fmt"
"time"
"code.google.com/p/go.net/context"
)
func ExampleWithTimeout() {
// Pass a context with a timeout to tell a blocking function that it
// should abandon its work after the timeout elapses.
ctx, _ := context.WithTimeout(context.Background(), 100*time.Millisecond)
select {
case <-time.After(200 * time.Millisecond):
fmt.Println("overslept")
case <-ctx.Done():
fmt.Println(ctx.Err()) // prints "context deadline exceeded"
}
// Output:
// context deadline exceeded
}

31
pkg/api/context.go Normal file
View File

@ -0,0 +1,31 @@
/*
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.
*/
package api
import (
"code.google.com/p/go.net/context"
)
// Context carries values across API boundaries.
type Context interface {
Value(key interface{}) interface{}
}
// NewContext instantiates a base context object for request flows
func NewContext() Context {
return context.TODO()
}

14
pkg/apiserver/apiserver_test.go
View File

@ -88,18 +88,18 @@ type SimpleRESTStorage struct {
injectedFunction func(obj runtime.Object) (returnObj runtime.Object, err error)
}
func (storage *SimpleRESTStorage) List(label, field labels.Selector) (runtime.Object, error) {
func (storage *SimpleRESTStorage) List(ctx api.Context, label, field labels.Selector) (runtime.Object, error) {
result := &SimpleList{
Items: storage.list,
}
return result, storage.errors["list"]
}
func (storage *SimpleRESTStorage) Get(id string) (runtime.Object, error) {
func (storage *SimpleRESTStorage) Get(ctx api.Context, id string) (runtime.Object, error) {
return api.Scheme.CopyOrDie(&storage.item), storage.errors["get"]
}
func (storage *SimpleRESTStorage) Delete(id string) (<-chan runtime.Object, error) {
func (storage *SimpleRESTStorage) Delete(ctx api.Context, id string) (<-chan runtime.Object, error) {
storage.deleted = id
if err := storage.errors["delete"]; err != nil {
return nil, err
@ -116,7 +116,7 @@ func (storage *SimpleRESTStorage) New() runtime.Object {
return &Simple{}
}
func (storage *SimpleRESTStorage) Create(obj runtime.Object) (<-chan runtime.Object, error) {
func (storage *SimpleRESTStorage) Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
storage.created = obj.(*Simple)
if err := storage.errors["create"]; err != nil {
return nil, err
@ -129,7 +129,7 @@ func (storage *SimpleRESTStorage) Create(obj runtime.Object) (<-chan runtime.Obj
}), nil
}
func (storage *SimpleRESTStorage) Update(obj runtime.Object) (<-chan runtime.Object, error) {
func (storage *SimpleRESTStorage) Update(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
storage.updated = obj.(*Simple)
if err := storage.errors["update"]; err != nil {
return nil, err
@ -143,7 +143,7 @@ func (storage *SimpleRESTStorage) Update(obj runtime.Object) (<-chan runtime.Obj
}
// Implement ResourceWatcher.
func (storage *SimpleRESTStorage) Watch(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
func (storage *SimpleRESTStorage) Watch(ctx api.Context, label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
storage.requestedLabelSelector = label
storage.requestedFieldSelector = field
storage.requestedResourceVersion = resourceVersion
@ -155,7 +155,7 @@ func (storage *SimpleRESTStorage) Watch(label, field labels.Selector, resourceVe
}
// Implement Redirector.
func (storage *SimpleRESTStorage) ResourceLocation(id string) (string, error) {
func (storage *SimpleRESTStorage) ResourceLocation(ctx api.Context, id string) (string, error) {
storage.requestedResourceLocationID = id
if err := storage.errors["resourceLocation"]; err != nil {
return "", err

15
pkg/apiserver/interfaces.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package apiserver
import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
@ -30,20 +31,20 @@ type RESTStorage interface {
New() runtime.Object
// List selects resources in the storage which match to the selector.
List(label, field labels.Selector) (runtime.Object, error)
List(ctx api.Context, label, field labels.Selector) (runtime.Object, error)
// Get finds a resource in the storage by id and returns it.
// Although it can return an arbitrary error value, IsNotFound(err) is true for the
// returned error value err when the specified resource is not found.
Get(id string) (runtime.Object, error)
Get(ctx api.Context, id string) (runtime.Object, error)
// Delete finds a resource in the storage and deletes it.
// Although it can return an arbitrary error value, IsNotFound(err) is true for the
// returned error value err when the specified resource is not found.
Delete(id string) (<-chan runtime.Object, error)
Delete(ctx api.Context, id string) (<-chan runtime.Object, error)
Create(runtime.Object) (<-chan runtime.Object, error)
Update(runtime.Object) (<-chan runtime.Object, error)
Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error)
Update(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error)
}
// ResourceWatcher should be implemented by all RESTStorage objects that
@ -53,11 +54,11 @@ type ResourceWatcher interface {
// are supported; an error should be returned if 'field' tries to select on a field that
// isn't supported. 'resourceVersion' allows for continuing/starting a watch at a
// particular version.
Watch(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error)
Watch(ctx api.Context, label, field labels.Selector, resourceVersion uint64) (watch.Interface, error)
}
// Redirector know how to return a remote resource's location.
type Redirector interface {
// ResourceLocation should return the remote location of the given resource, or an error.
ResourceLocation(id string) (remoteLocation string, err error)
ResourceLocation(ctx api.Context, id string) (remoteLocation string, err error)
}

4
pkg/apiserver/proxy.go
View File

@ -26,6 +26,7 @@ import (
"path"
"strings"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/httplog"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
@ -76,6 +77,7 @@ type ProxyHandler struct {
}
func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
ctx := api.NewContext()
parts := strings.SplitN(req.URL.Path, "/", 3)
if len(parts) < 2 {
notFound(w, req)
@ -101,7 +103,7 @@ func (r *ProxyHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
return
}
location, err := redirector.ResourceLocation(id)
location, err := redirector.ResourceLocation(ctx, id)
if err != nil {
status := errToAPIStatus(err)
writeJSON(status.Code, r.codec, status, w)

4
pkg/apiserver/redirect.go
View File

@ -19,6 +19,7 @@ package apiserver
import (
"net/http"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/httplog"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
)
@ -29,6 +30,7 @@ type RedirectHandler struct {
}
func (r *RedirectHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
ctx := api.NewContext()
parts := splitPath(req.URL.Path)
if len(parts) != 2 || req.Method != "GET" {
notFound(w, req)
@ -50,7 +52,7 @@ func (r *RedirectHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
return
}
location, err := redirector.ResourceLocation(id)
location, err := redirector.ResourceLocation(ctx, id)
if err != nil {
status := errToAPIStatus(err)
writeJSON(status.Code, r.codec, status, w)

11
pkg/apiserver/resthandler.go
View File

@ -64,6 +64,7 @@ func (h *RESTHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// timeout=<duration> Timeout for synchronous requests, only applies if sync=true
// labels=<label-selector> Used for filtering list operations
func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w http.ResponseWriter, storage RESTStorage) {
ctx := api.NewContext()
sync := req.URL.Query().Get("sync") == "true"
timeout := parseTimeout(req.URL.Query().Get("timeout"))
switch req.Method {
@ -80,14 +81,14 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt
errorJSON(err, h.codec, w)
return
}
list, err := storage.List(label, field)
list, err := storage.List(ctx, label, field)
if err != nil {
errorJSON(err, h.codec, w)
return
}
writeJSON(http.StatusOK, h.codec, list, w)
case 2:
item, err := storage.Get(parts[1])
item, err := storage.Get(ctx, parts[1])
if err != nil {
errorJSON(err, h.codec, w)
return
@ -113,7 +114,7 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt
errorJSON(err, h.codec, w)
return
}
out, err := storage.Create(obj)
out, err := storage.Create(ctx, obj)
if err != nil {
errorJSON(err, h.codec, w)
return
@ -126,7 +127,7 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt
notFound(w, req)
return
}
out, err := storage.Delete(parts[1])
out, err := storage.Delete(ctx, parts[1])
if err != nil {
errorJSON(err, h.codec, w)
return
@ -150,7 +151,7 @@ func (h *RESTHandler) handleRESTStorage(parts []string, req *http.Request, w htt
errorJSON(err, h.codec, w)
return
}
out, err := storage.Update(obj)
out, err := storage.Update(ctx, obj)
if err != nil {
errorJSON(err, h.codec, w)
return

4
pkg/apiserver/watch.go
View File

@ -24,6 +24,7 @@ import (
"strings"
"code.google.com/p/go.net/websocket"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/httplog"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
@ -61,6 +62,7 @@ func isWebsocketRequest(req *http.Request) bool {
// ServeHTTP processes watch requests.
func (h *WatchHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
ctx := api.NewContext()
parts := splitPath(req.URL.Path)
if len(parts) < 1 || req.Method != "GET" {
notFound(w, req)
@ -73,7 +75,7 @@ func (h *WatchHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
}
if watcher, ok := storage.(ResourceWatcher); ok {
label, field, resourceVersion := getWatchParams(req.URL.Query())
watching, err := watcher.Watch(label, field, resourceVersion)
watching, err := watcher.Watch(ctx, label, field, resourceVersion)
if err != nil {
errorJSON(err, h.codec, w)
return

3
pkg/master/pod_cache.go
View File

@ -72,7 +72,8 @@ func (p *PodCache) updatePodInfo(host, id string) error {
// UpdateAllContainers updates information about all containers. Either called by Loop() below, or one-off.
func (p *PodCache) UpdateAllContainers() {
pods, err := p.pods.ListPods(labels.Everything())
var ctx api.Context
pods, err := p.pods.ListPods(ctx, labels.Everything())
if err != nil {
glog.Errorf("Error synchronizing container list: %v", err)
return

10
pkg/registry/binding/rest.go
View File

@ -41,17 +41,17 @@ func NewREST(bindingRegistry Registry) *REST {
}
// List returns an error because bindings are write-only objects.
func (*REST) List(label, field labels.Selector) (runtime.Object, error) {
func (*REST) List(ctx api.Context, label, field labels.Selector) (runtime.Object, error) {
return nil, errors.NewNotFound("binding", "list")
}
// Get returns an error because bindings are write-only objects.
func (*REST) Get(id string) (runtime.Object, error) {
func (*REST) Get(ctx api.Context, id string) (runtime.Object, error) {
return nil, errors.NewNotFound("binding", id)
}
// Delete returns an error because bindings are write-only objects.
func (*REST) Delete(id string) (<-chan runtime.Object, error) {
func (*REST) Delete(ctx api.Context, id string) (<-chan runtime.Object, error) {
return nil, errors.NewNotFound("binding", id)
}
@ -61,7 +61,7 @@ func (*REST) New() runtime.Object {
}
// Create attempts to make the assignment indicated by the binding it recieves.
func (b *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
func (b *REST) Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
binding, ok := obj.(*api.Binding)
if !ok {
return nil, fmt.Errorf("incorrect type: %#v", obj)
@ -75,6 +75,6 @@ func (b *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
}
// Update returns an error-- this object may not be updated.
func (b *REST) Update(obj runtime.Object) (<-chan runtime.Object, error) {
func (b *REST) Update(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
return nil, fmt.Errorf("Bindings may not be changed.")
}

14
pkg/registry/binding/rest_test.go
View File

@ -52,24 +52,25 @@ func TestNewREST(t *testing.T) {
}
func TestRESTUnsupported(t *testing.T) {
var ctx api.Context
mockRegistry := MockRegistry{
OnApplyBinding: func(b *api.Binding) error { return nil },
}
b := NewREST(mockRegistry)
if _, err := b.Delete("binding id"); err == nil {
if _, err := b.Delete(ctx, "binding id"); err == nil {
t.Errorf("unexpected non-error")
}
if _, err := b.Update(&api.Binding{PodID: "foo", Host: "new machine"}); err == nil {
if _, err := b.Update(ctx, &api.Binding{PodID: "foo", Host: "new machine"}); err == nil {
t.Errorf("unexpected non-error")
}
if _, err := b.Get("binding id"); err == nil {
if _, err := b.Get(ctx, "binding id"); err == nil {
t.Errorf("unexpected non-error")
}
if _, err := b.List(labels.Set{"name": "foo"}.AsSelector(), labels.Everything()); err == nil {
if _, err := b.List(ctx, labels.Set{"name": "foo"}.AsSelector(), labels.Everything()); err == nil {
t.Errorf("unexpected non-error")
}
// Try sending wrong object just to get 100% coverage
if _, err := b.Create(&api.Pod{}); err == nil {
if _, err := b.Create(ctx, &api.Pod{}); err == nil {
t.Errorf("unexpected non-error")
}
}
@ -93,8 +94,9 @@ func TestRESTPost(t *testing.T) {
return item.err
},
}
ctx := api.NewContext()
b := NewREST(mockRegistry)
resultChan, err := b.Create(item.b)
resultChan, err := b.Create(ctx, item.b)
if err != nil {
t.Errorf("Unexpected error %v", err)
continue

28
pkg/registry/controller/rest.go
View File

@ -34,7 +34,7 @@ import (
// PodLister is anything that knows how to list pods.
type PodLister interface {
ListPods(labels.Selector) (*api.PodList, error)
ListPods(ctx api.Context, labels labels.Selector) (*api.PodList, error)
}
// REST implements apiserver.RESTStorage for the replication controller service.
@ -54,7 +54,7 @@ func NewREST(registry Registry, podLister PodLister) *REST {
}
// Create registers the given ReplicationController.
func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
controller, ok := obj.(*api.ReplicationController)
if !ok {
return nil, fmt.Errorf("not a replication controller: %#v", obj)
@ -80,24 +80,24 @@ func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
}
// Delete asynchronously deletes the ReplicationController specified by its id.
func (rs *REST) Delete(id string) (<-chan runtime.Object, error) {
func (rs *REST) Delete(ctx api.Context, id string) (<-chan runtime.Object, error) {
return apiserver.MakeAsync(func() (runtime.Object, error) {
return &api.Status{Status: api.StatusSuccess}, rs.registry.DeleteController(id)
}), nil
}
// Get obtains the ReplicationController specified by its id.
func (rs *REST) Get(id string) (runtime.Object, error) {
func (rs *REST) Get(ctx api.Context, id string) (runtime.Object, error) {
controller, err := rs.registry.GetController(id)
if err != nil {
return nil, err
}
rs.fillCurrentState(controller)
rs.fillCurrentState(ctx, controller)
return controller, err
}
// List obtains a list of ReplicationControllers that match selector.
func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
func (rs *REST) List(ctx api.Context, label, field labels.Selector) (runtime.Object, error) {
if !field.Empty() {
return nil, fmt.Errorf("field selector not supported yet")
}
@ -108,7 +108,7 @@ func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
filtered := []api.ReplicationController{}
for _, controller := range controllers.Items {
if label.Matches(labels.Set(controller.Labels)) {
rs.fillCurrentState(&controller)
rs.fillCurrentState(ctx, &controller)
filtered = append(filtered, controller)
}
}
@ -123,7 +123,7 @@ func (*REST) New() runtime.Object {
// Update replaces a given ReplicationController instance with an existing
// instance in storage.registry.
func (rs *REST) Update(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
controller, ok := obj.(*api.ReplicationController)
if !ok {
return nil, fmt.Errorf("not a replication controller: %#v", obj)
@ -142,7 +142,7 @@ func (rs *REST) Update(obj runtime.Object) (<-chan runtime.Object, error) {
// Watch returns ReplicationController events via a watch.Interface.
// It implements apiserver.ResourceWatcher.
func (rs *REST) Watch(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
func (rs *REST) Watch(ctx api.Context, label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
if !field.Empty() {
return nil, fmt.Errorf("no field selector implemented for controllers")
}
@ -160,15 +160,15 @@ func (rs *REST) Watch(label, field labels.Selector, resourceVersion uint64) (wat
}
match := label.Matches(labels.Set(repController.Labels))
if match {
rs.fillCurrentState(repController)
rs.fillCurrentState(ctx, repController)
}
return e, match
}), nil
}
func (rs *REST) waitForController(ctrl *api.ReplicationController) (runtime.Object, error) {
func (rs *REST) waitForController(ctx api.Context, ctrl *api.ReplicationController) (runtime.Object, error) {
for {
pods, err := rs.podLister.ListPods(labels.Set(ctrl.DesiredState.ReplicaSelector).AsSelector())
pods, err := rs.podLister.ListPods(ctx, labels.Set(ctrl.DesiredState.ReplicaSelector).AsSelector())
if err != nil {
return ctrl, err
}
@ -180,11 +180,11 @@ func (rs *REST) waitForController(ctrl *api.ReplicationController) (runtime.Obje
return ctrl, nil
}
func (rs *REST) fillCurrentState(ctrl *api.ReplicationController) error {
func (rs *REST) fillCurrentState(ctx api.Context, ctrl *api.ReplicationController) error {
if rs.podLister == nil {
return nil
}
list, err := rs.podLister.ListPods(labels.Set(ctrl.DesiredState.ReplicaSelector).AsSelector())
list, err := rs.podLister.ListPods(ctx, labels.Set(ctrl.DesiredState.ReplicaSelector).AsSelector())
if err != nil {
return err
}

24
pkg/registry/controller/rest_test.go
View File

@ -39,7 +39,8 @@ func TestListControllersError(t *testing.T) {
storage := REST{
registry: &mockRegistry,
}
controllers, err := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
controllers, err := storage.List(ctx, labels.Everything(), labels.Everything())
if err != mockRegistry.Err {
t.Errorf("Expected %#v, Got %#v", mockRegistry.Err, err)
}
@ -53,7 +54,8 @@ func TestListEmptyControllerList(t *testing.T) {
storage := REST{
registry: &mockRegistry,
}
controllers, err := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
controllers, err := storage.List(ctx, labels.Everything(), labels.Everything())
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -86,7 +88,8 @@ func TestListControllerList(t *testing.T) {
storage := REST{
registry: &mockRegistry,
}
controllersObj, err := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
controllersObj, err := storage.List(ctx, labels.Everything(), labels.Everything())
controllers := controllersObj.(*api.ReplicationControllerList)
if err != nil {
t.Errorf("unexpected error: %v", err)
@ -240,7 +243,8 @@ func TestCreateController(t *testing.T) {
PodTemplate: validPodTemplate,
},
}
channel, err := storage.Create(controller)
ctx := api.NewContext()
channel, err := storage.Create(ctx, controller)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
@ -263,7 +267,6 @@ func TestControllerStorageValidatesCreate(t *testing.T) {
podLister: nil,
pollPeriod: time.Millisecond * 1,
}
failureCases := map[string]api.ReplicationController{
"empty ID": {
JSONBase: api.JSONBase{ID: ""},
@ -277,7 +280,8 @@ func TestControllerStorageValidatesCreate(t *testing.T) {
},
}
for _, failureCase := range failureCases {
c, err := storage.Create(&failureCase)
ctx := api.NewContext()
c, err := storage.Create(ctx, &failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -307,7 +311,8 @@ func TestControllerStorageValidatesUpdate(t *testing.T) {
},
}
for _, failureCase := range failureCases {
c, err := storage.Update(&failureCase)
ctx := api.NewContext()
c, err := storage.Update(ctx, &failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -323,7 +328,7 @@ type fakePodLister struct {
s labels.Selector
}
func (f *fakePodLister) ListPods(s labels.Selector) (*api.PodList, error) {
func (f *fakePodLister) ListPods(ctx api.Context, s labels.Selector) (*api.PodList, error) {
f.s = s
return &f.l, f.e
}
@ -349,7 +354,8 @@ func TestFillCurrentState(t *testing.T) {
},
},
}
storage.fillCurrentState(&controller)
ctx := api.NewContext()
storage.fillCurrentState(ctx, &controller)
if controller.CurrentState.Replicas != 2 {
t.Errorf("expected 2, got: %d", controller.CurrentState.Replicas)
}

12
pkg/registry/endpoint/rest.go
View File

@ -38,12 +38,12 @@ func NewREST(registry Registry) *REST {
}
// Get satisfies the RESTStorage interface.
func (rs *REST) Get(id string) (runtime.Object, error) {
func (rs *REST) Get(ctx api.Context, id string) (runtime.Object, error) {
return rs.registry.GetEndpoints(id)
}
// List satisfies the RESTStorage interface.
func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
func (rs *REST) List(ctx api.Context, label, field labels.Selector) (runtime.Object, error) {
if !label.Empty() || !field.Empty() {
return nil, errors.New("label/field selectors are not supported on endpoints")
}
@ -52,22 +52,22 @@ func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
// Watch returns Endpoint events via a watch.Interface.
// It implements apiserver.ResourceWatcher.
func (rs *REST) Watch(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
func (rs *REST) Watch(ctx api.Context, label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
return rs.registry.WatchEndpoints(label, field, resourceVersion)
}
// Create satisfies the RESTStorage interface but is unimplemented.
func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
return nil, errors.New("unimplemented")
}
// Update satisfies the RESTStorage interface but is unimplemented.
func (rs *REST) Update(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
return nil, errors.New("unimplemented")
}
// Delete satisfies the RESTStorage interface but is unimplemented.
func (rs *REST) Delete(id string) (<-chan runtime.Object, error) {
func (rs *REST) Delete(ctx api.Context, id string) (<-chan runtime.Object, error) {
return nil, errors.New("unimplemented")
}

12
pkg/registry/endpoint/rest_test.go
View File

@ -34,7 +34,8 @@ func TestGetEndpoints(t *testing.T) {
},
}
storage := NewREST(registry)
obj, err := storage.Get("foo")
ctx := api.NewContext()
obj, err := storage.Get(ctx, "foo")
if err != nil {
t.Fatalf("unexpected error: %#v", err)
}
@ -48,9 +49,9 @@ func TestGetEndpointsMissingService(t *testing.T) {
Err: errors.NewNotFound("service", "foo"),
}
storage := NewREST(registry)
ctx := api.NewContext()
// returns service not found
_, err := storage.Get("foo")
_, err := storage.Get(ctx, "foo")
if !errors.IsNotFound(err) || !reflect.DeepEqual(err, errors.NewNotFound("service", "foo")) {
t.Errorf("expected NotFound error, got %#v", err)
}
@ -60,7 +61,7 @@ func TestGetEndpointsMissingService(t *testing.T) {
registry.Service = &api.Service{
JSONBase: api.JSONBase{ID: "foo"},
}
obj, err := storage.Get("foo")
obj, err := storage.Get(ctx, "foo")
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
@ -79,7 +80,8 @@ func TestEndpointsRegistryList(t *testing.T) {
{JSONBase: api.JSONBase{ID: "bar"}},
},
}
s, _ := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
s, _ := storage.List(ctx, labels.Everything(), labels.Everything())
sl := s.(*api.EndpointsList)
if len(sl.Items) != 2 {
t.Fatalf("Expected 2 endpoints, but got %v", len(sl.Items))

2
pkg/registry/etcd/etcd.go
View File

@ -55,7 +55,7 @@ func makePodKey(podID string) string {
}
// ListPods obtains a list of pods with labels that match selector.
func (r *Registry) ListPods(selector labels.Selector) (*api.PodList, error) {
func (r *Registry) ListPods(ctx api.Context, selector labels.Selector) (*api.PodList, error) {
return r.ListPodsPredicate(func(pod *api.Pod) bool {
return selector.Matches(labels.Set(pod.Labels))
})

9
pkg/registry/etcd/etcd_test.go
View File

@ -410,7 +410,8 @@ func TestEtcdEmptyListPods(t *testing.T) {
E: nil,
}
registry := NewTestEtcdRegistry(fakeClient)
pods, err := registry.ListPods(labels.Everything())
ctx := api.NewContext()
pods, err := registry.ListPods(ctx, labels.Everything())
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -428,7 +429,8 @@ func TestEtcdListPodsNotFound(t *testing.T) {
E: tools.EtcdErrorNotFound,
}
registry := NewTestEtcdRegistry(fakeClient)
pods, err := registry.ListPods(labels.Everything())
ctx := api.NewContext()
pods, err := registry.ListPods(ctx, labels.Everything())
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -463,7 +465,8 @@ func TestEtcdListPods(t *testing.T) {
E: nil,
}
registry := NewTestEtcdRegistry(fakeClient)
pods, err := registry.ListPods(labels.Everything())
ctx := api.NewContext()
pods, err := registry.ListPods(ctx, labels.Everything())
if err != nil {
t.Errorf("unexpected error: %v", err)
}

10
pkg/registry/minion/rest.go
View File

@ -38,7 +38,7 @@ func NewREST(m Registry) *REST {
}
}
func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
minion, ok := obj.(*api.Minion)
if !ok {
return nil, fmt.Errorf("not a minion: %#v", obj)
@ -65,7 +65,7 @@ func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
}), nil
}
func (rs *REST) Delete(id string) (<-chan runtime.Object, error) {
func (rs *REST) Delete(ctx api.Context, id string) (<-chan runtime.Object, error) {
exists, err := rs.registry.Contains(id)
if !exists {
return nil, ErrDoesNotExist
@ -78,7 +78,7 @@ func (rs *REST) Delete(id string) (<-chan runtime.Object, error) {
}), nil
}
func (rs *REST) Get(id string) (runtime.Object, error) {
func (rs *REST) Get(ctx api.Context, id string) (runtime.Object, error) {
exists, err := rs.registry.Contains(id)
if !exists {
return nil, ErrDoesNotExist
@ -86,7 +86,7 @@ func (rs *REST) Get(id string) (runtime.Object, error) {
return rs.toApiMinion(id), err
}
func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
func (rs *REST) List(ctx api.Context, label, field labels.Selector) (runtime.Object, error) {
nameList, err := rs.registry.List()
if err != nil {
return nil, err
@ -102,7 +102,7 @@ func (*REST) New() runtime.Object {
return &api.Minion{}
}
func (rs *REST) Update(minion runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Update(ctx api.Context, minion runtime.Object) (<-chan runtime.Object, error) {
return nil, fmt.Errorf("Minions can only be created (inserted) and deleted.")
}

20
pkg/registry/minion/rest_test.go
View File

@ -27,18 +27,18 @@ import (
func TestMinionREST(t *testing.T) {
m := NewRegistry([]string{"foo", "bar"})
ms := NewREST(m)
if obj, err := ms.Get("foo"); err != nil || obj.(*api.Minion).ID != "foo" {
ctx := api.NewContext()
if obj, err := ms.Get(ctx, "foo"); err != nil || obj.(*api.Minion).ID != "foo" {
t.Errorf("missing expected object")
}
if obj, err := ms.Get("bar"); err != nil || obj.(*api.Minion).ID != "bar" {
if obj, err := ms.Get(ctx, "bar"); err != nil || obj.(*api.Minion).ID != "bar" {
t.Errorf("missing expected object")
}
if _, err := ms.Get("baz"); err != ErrDoesNotExist {
if _, err := ms.Get(ctx, "baz"); err != ErrDoesNotExist {
t.Errorf("has unexpected object")
}
c, err := ms.Create(&api.Minion{JSONBase: api.JSONBase{ID: "baz"}})
c, err := ms.Create(ctx, &api.Minion{JSONBase: api.JSONBase{ID: "baz"}})
if err != nil {
t.Errorf("insert failed")
}
@ -46,11 +46,11 @@ func TestMinionREST(t *testing.T) {
if m, ok := obj.(*api.Minion); !ok || m.ID != "baz" {
t.Errorf("insert return value was weird: %#v", obj)
}
if obj, err := ms.Get("baz"); err != nil || obj.(*api.Minion).ID != "baz" {
if obj, err := ms.Get(ctx, "baz"); err != nil || obj.(*api.Minion).ID != "baz" {
t.Errorf("insert didn't actually insert")
}
c, err = ms.Delete("bar")
c, err = ms.Delete(ctx, "bar")
if err != nil {
t.Errorf("delete failed")
}
@ -58,16 +58,16 @@ func TestMinionREST(t *testing.T) {
if s, ok := obj.(*api.Status); !ok || s.Status != api.StatusSuccess {
t.Errorf("delete return value was weird: %#v", obj)
}
if _, err := ms.Get("bar"); err != ErrDoesNotExist {
if _, err := ms.Get(ctx, "bar"); err != ErrDoesNotExist {
t.Errorf("delete didn't actually delete")
}
_, err = ms.Delete("bar")
_, err = ms.Delete(ctx, "bar")
if err != ErrDoesNotExist {
t.Errorf("delete returned wrong error")
}
list, err := ms.List(labels.Everything(), labels.Everything())
list, err := ms.List(ctx, labels.Everything(), labels.Everything())
if err != nil {
t.Errorf("got error calling List")
}

2
pkg/registry/pod/registry.go
View File

@ -25,7 +25,7 @@ import (
// Registry is an interface implemented by things that know how to store Pod objects.
type Registry interface {
// ListPods obtains a list of pods having labels which match selector.
ListPods(selector labels.Selector) (*api.PodList, error)
ListPods(ctx api.Context, selector labels.Selector) (*api.PodList, error)
// ListPodsPredicate obtains a list of pods for which filter returns true.
ListPodsPredicate(filter func(*api.Pod) bool) (*api.PodList, error)
// Watch for new/changed/deleted pods

16
pkg/registry/pod/rest.go
View File

@ -67,7 +67,7 @@ func NewREST(config *RESTConfig) *REST {
}
}
func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
pod := obj.(*api.Pod)
pod.DesiredState.Manifest.UUID = uuid.NewUUID().String()
if len(pod.ID) == 0 {
@ -88,13 +88,13 @@ func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
}), nil
}
func (rs *REST) Delete(id string) (<-chan runtime.Object, error) {
func (rs *REST) Delete(ctx api.Context, id string) (<-chan runtime.Object, error) {
return apiserver.MakeAsync(func() (runtime.Object, error) {
return &api.Status{Status: api.StatusSuccess}, rs.registry.DeletePod(id)
}), nil
}
func (rs *REST) Get(id string) (runtime.Object, error) {
func (rs *REST) Get(ctx api.Context, id string) (runtime.Object, error) {
pod, err := rs.registry.GetPod(id)
if err != nil {
return pod, err
@ -131,7 +131,7 @@ func (rs *REST) filterFunc(label, field labels.Selector) func(*api.Pod) bool {
}
}
func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
func (rs *REST) List(ctx api.Context, label, field labels.Selector) (runtime.Object, error) {
pods, err := rs.registry.ListPodsPredicate(rs.filterFunc(label, field))
if err == nil {
for i := range pods.Items {
@ -149,7 +149,7 @@ func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
}
// Watch begins watching for new, changed, or deleted pods.
func (rs *REST) Watch(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
func (rs *REST) Watch(ctx api.Context, label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
return rs.registry.WatchPods(resourceVersion, rs.filterFunc(label, field))
}
@ -157,7 +157,7 @@ func (*REST) New() runtime.Object {
return &api.Pod{}
}
func (rs *REST) Update(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
pod := obj.(*api.Pod)
if errs := validation.ValidatePod(pod); len(errs) > 0 {
return nil, errors.NewInvalid("pod", pod.ID, errs)
@ -277,9 +277,9 @@ func getPodStatus(pod *api.Pod, minions client.MinionInterface) (api.PodStatus,
}
}
func (rs *REST) waitForPodRunning(pod *api.Pod) (runtime.Object, error) {
func (rs *REST) waitForPodRunning(ctx api.Context, pod *api.Pod) (runtime.Object, error) {
for {
podObj, err := rs.Get(pod.ID)
podObj, err := rs.Get(ctx, pod.ID)
if err != nil || podObj == nil {
return nil, err
}

36
pkg/registry/pod/rest_test.go
View File

@ -69,7 +69,8 @@ func TestCreatePodRegistryError(t *testing.T) {
},
}
pod := &api.Pod{DesiredState: desiredState}
ch, err := storage.Create(pod)
ctx := api.NewContext()
ch, err := storage.Create(ctx, pod)
if err != nil {
t.Errorf("Expected %#v, Got %#v", nil, err)
}
@ -88,7 +89,8 @@ func TestCreatePodSetsIds(t *testing.T) {
},
}
pod := &api.Pod{DesiredState: desiredState}
ch, err := storage.Create(pod)
ctx := api.NewContext()
ch, err := storage.Create(ctx, pod)
if err != nil {
t.Errorf("Expected %#v, Got %#v", nil, err)
}
@ -114,7 +116,8 @@ func TestCreatePodSetsUUIDs(t *testing.T) {
},
}
pod := &api.Pod{DesiredState: desiredState}
ch, err := storage.Create(pod)
ctx := api.NewContext()
ch, err := storage.Create(ctx, pod)
if err != nil {
t.Errorf("Expected %#v, Got %#v", nil, err)
}
@ -131,7 +134,8 @@ func TestListPodsError(t *testing.T) {
storage := REST{
registry: podRegistry,
}
pods, err := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
pods, err := storage.List(ctx, labels.Everything(), labels.Everything())
if err != podRegistry.Err {
t.Errorf("Expected %#v, Got %#v", podRegistry.Err, err)
}
@ -145,7 +149,8 @@ func TestListEmptyPodList(t *testing.T) {
storage := REST{
registry: podRegistry,
}
pods, err := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
pods, err := storage.List(ctx, labels.Everything(), labels.Everything())
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -177,7 +182,8 @@ func TestListPodList(t *testing.T) {
storage := REST{
registry: podRegistry,
}
podsObj, err := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
podsObj, err := storage.List(ctx, labels.Everything(), labels.Everything())
pods := podsObj.(*api.PodList)
if err != nil {
t.Errorf("unexpected error: %v", err)
@ -217,6 +223,7 @@ func TestListPodListSelection(t *testing.T) {
storage := REST{
registry: podRegistry,
}
ctx := api.NewContext()
table := []struct {
label, field string
@ -256,7 +263,7 @@ func TestListPodListSelection(t *testing.T) {
t.Errorf("unexpected error: %v", err)
continue
}
podsObj, err := storage.List(label, field)
podsObj, err := storage.List(ctx, label, field)
if err != nil {
t.Errorf("unexpected error: %v", err)
}
@ -305,7 +312,8 @@ func TestGetPod(t *testing.T) {
storage := REST{
registry: podRegistry,
}
obj, err := storage.Get("foo")
ctx := api.NewContext()
obj, err := storage.Get(ctx, "foo")
pod := obj.(*api.Pod)
if err != nil {
t.Errorf("unexpected error: %v", err)
@ -324,7 +332,8 @@ func TestGetPodCloud(t *testing.T) {
registry: podRegistry,
cloudProvider: fakeCloud,
}
obj, err := storage.Get("foo")
ctx := api.NewContext()
obj, err := storage.Get(ctx, "foo")
pod := obj.(*api.Pod)
if err != nil {
t.Errorf("unexpected error: %v", err)
@ -487,8 +496,9 @@ func TestPodStorageValidatesCreate(t *testing.T) {
storage := REST{
registry: podRegistry,
}
ctx := api.NewContext()
pod := &api.Pod{}
c, err := storage.Create(pod)
c, err := storage.Create(ctx, pod)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -503,8 +513,9 @@ func TestPodStorageValidatesUpdate(t *testing.T) {
storage := REST{
registry: podRegistry,
}
ctx := api.NewContext()
pod := &api.Pod{}
c, err := storage.Update(pod)
c, err := storage.Update(ctx, pod)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -534,7 +545,8 @@ func TestCreatePod(t *testing.T) {
JSONBase: api.JSONBase{ID: "foo"},
DesiredState: desiredState,
}
channel, err := storage.Create(pod)
ctx := api.NewContext()
channel, err := storage.Create(ctx, pod)
if err != nil {
t.Errorf("unexpected error: %v", err)
}

2
pkg/registry/registrytest/pod.go
View File

@ -57,7 +57,7 @@ func (r *PodRegistry) ListPodsPredicate(filter func(*api.Pod) bool) (*api.PodLis
return &pods, nil
}
func (r *PodRegistry) ListPods(selector labels.Selector) (*api.PodList, error) {
func (r *PodRegistry) ListPods(ctx api.Context, selector labels.Selector) (*api.PodList, error) {
return r.ListPodsPredicate(func(pod *api.Pod) bool {
return selector.Matches(labels.Set(pod.Labels))
})

14
pkg/registry/service/rest.go
View File

@ -50,7 +50,7 @@ func NewREST(registry Registry, cloud cloudprovider.Interface, machines minion.R
}
}
func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Create(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
srv := obj.(*api.Service)
if errs := validation.ValidateService(srv); len(errs) > 0 {
return nil, errors.NewInvalid("service", srv.ID, errs)
@ -94,7 +94,7 @@ func (rs *REST) Create(obj runtime.Object) (<-chan runtime.Object, error) {
}), nil
}
func (rs *REST) Delete(id string) (<-chan runtime.Object, error) {
func (rs *REST) Delete(ctx api.Context, id string) (<-chan runtime.Object, error) {
service, err := rs.registry.GetService(id)
if err != nil {
return nil, err
@ -105,7 +105,7 @@ func (rs *REST) Delete(id string) (<-chan runtime.Object, error) {
}), nil
}
func (rs *REST) Get(id string) (runtime.Object, error) {
func (rs *REST) Get(ctx api.Context, id string) (runtime.Object, error) {
s, err := rs.registry.GetService(id)
if err != nil {
return nil, err
@ -114,7 +114,7 @@ func (rs *REST) Get(id string) (runtime.Object, error) {
}
// TODO: implement field selector?
func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
func (rs *REST) List(ctx api.Context, label, field labels.Selector) (runtime.Object, error) {
list, err := rs.registry.ListServices()
if err != nil {
return nil, err
@ -131,7 +131,7 @@ func (rs *REST) List(label, field labels.Selector) (runtime.Object, error) {
// Watch returns Services events via a watch.Interface.
// It implements apiserver.ResourceWatcher.
func (rs *REST) Watch(label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
func (rs *REST) Watch(ctx api.Context, label, field labels.Selector, resourceVersion uint64) (watch.Interface, error) {
return rs.registry.WatchServices(label, field, resourceVersion)
}
@ -163,7 +163,7 @@ func GetServiceEnvironmentVariables(registry Registry, machine string) ([]api.En
return result, nil
}
func (rs *REST) Update(obj runtime.Object) (<-chan runtime.Object, error) {
func (rs *REST) Update(ctx api.Context, obj runtime.Object) (<-chan runtime.Object, error) {
srv := obj.(*api.Service)
if errs := validation.ValidateService(srv); len(errs) > 0 {
return nil, errors.NewInvalid("service", srv.ID, errs)
@ -179,7 +179,7 @@ func (rs *REST) Update(obj runtime.Object) (<-chan runtime.Object, error) {
}
// ResourceLocation returns a URL to which one can send traffic for the specified service.
func (rs *REST) ResourceLocation(id string) (string, error) {
func (rs *REST) ResourceLocation(ctx api.Context, id string) (string, error) {
e, err := rs.registry.GetEndpoints(id)
if err != nil {
return "", err

35
pkg/registry/service/rest_test.go
View File

@ -40,7 +40,8 @@ func TestServiceRegistryCreate(t *testing.T) {
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz"},
}
c, _ := storage.Create(svc)
ctx := api.NewContext()
c, _ := storage.Create(ctx, svc)
created_svc := <-c
created_service := created_svc.(*api.Service)
if created_service.ID != "foo" {
@ -75,8 +76,9 @@ func TestServiceStorageValidatesCreate(t *testing.T) {
Selector: map[string]string{},
},
}
ctx := api.NewContext()
for _, failureCase := range failureCases {
c, err := storage.Create(&failureCase)
c, err := storage.Create(ctx, &failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -95,7 +97,8 @@ func TestServiceRegistryUpdate(t *testing.T) {
Selector: map[string]string{"bar": "baz1"},
})
storage := NewREST(registry, nil, nil)
c, err := storage.Update(&api.Service{
ctx := api.NewContext()
c, err := storage.Update(ctx, &api.Service{
Port: 6502,
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz2"},
@ -136,8 +139,9 @@ func TestServiceStorageValidatesUpdate(t *testing.T) {
Selector: map[string]string{},
},
}
ctx := api.NewContext()
for _, failureCase := range failureCases {
c, err := storage.Update(&failureCase)
c, err := storage.Update(ctx, &failureCase)
if c != nil {
t.Errorf("Expected nil channel")
}
@ -158,7 +162,8 @@ func TestServiceRegistryExternalService(t *testing.T) {
Selector: map[string]string{"bar": "baz"},
CreateExternalLoadBalancer: true,
}
c, _ := storage.Create(svc)
ctx := api.NewContext()
c, _ := storage.Create(ctx, svc)
<-c
if len(fakeCloud.Calls) != 2 || fakeCloud.Calls[0] != "get-zone" || fakeCloud.Calls[1] != "create" {
t.Errorf("Unexpected call(s): %#v", fakeCloud.Calls)
@ -185,7 +190,8 @@ func TestServiceRegistryExternalServiceError(t *testing.T) {
Selector: map[string]string{"bar": "baz"},
CreateExternalLoadBalancer: true,
}
c, _ := storage.Create(svc)
ctx := api.NewContext()
c, _ := storage.Create(ctx, svc)
<-c
if len(fakeCloud.Calls) != 1 || fakeCloud.Calls[0] != "get-zone" {
t.Errorf("Unexpected call(s): %#v", fakeCloud.Calls)
@ -204,8 +210,9 @@ func TestServiceRegistryDelete(t *testing.T) {
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz"},
}
ctx := api.NewContext()
registry.CreateService(svc)
c, _ := storage.Delete(svc.ID)
c, _ := storage.Delete(ctx, svc.ID)
<-c
if len(fakeCloud.Calls) != 0 {
t.Errorf("Unexpected call(s): %#v", fakeCloud.Calls)
@ -225,8 +232,9 @@ func TestServiceRegistryDeleteExternal(t *testing.T) {
Selector: map[string]string{"bar": "baz"},
CreateExternalLoadBalancer: true,
}
ctx := api.NewContext()
registry.CreateService(svc)
c, _ := storage.Delete(svc.ID)
c, _ := storage.Delete(ctx, svc.ID)
<-c
if len(fakeCloud.Calls) != 2 || fakeCloud.Calls[0] != "get-zone" || fakeCloud.Calls[1] != "delete" {
t.Errorf("Unexpected call(s): %#v", fakeCloud.Calls)
@ -309,7 +317,8 @@ func TestServiceRegistryGet(t *testing.T) {
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz"},
})
storage.Get("foo")
ctx := api.NewContext()
storage.Get(ctx, "foo")
if len(fakeCloud.Calls) != 0 {
t.Errorf("Unexpected call(s): %#v", fakeCloud.Calls)
}
@ -324,12 +333,13 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
fakeCloud := &cloud.FakeCloud{}
machines := []string{"foo", "bar", "baz"}
storage := NewREST(registry, fakeCloud, minion.NewRegistry(machines))
ctx := api.NewContext()
registry.CreateService(&api.Service{
JSONBase: api.JSONBase{ID: "foo"},
Selector: map[string]string{"bar": "baz"},
})
redirector := apiserver.Redirector(storage)
location, err := redirector.ResourceLocation("foo")
location, err := redirector.ResourceLocation(ctx, "foo")
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
@ -342,7 +352,7 @@ func TestServiceRegistryResourceLocation(t *testing.T) {
// Test error path
registry.Err = fmt.Errorf("fake error")
if _, err = redirector.ResourceLocation("foo"); err == nil {
if _, err = redirector.ResourceLocation(ctx, "foo"); err == nil {
t.Errorf("unexpected nil error")
}
}
@ -361,7 +371,8 @@ func TestServiceRegistryList(t *testing.T) {
Selector: map[string]string{"bar2": "baz2"},
})
registry.List.ResourceVersion = 1
s, _ := storage.List(labels.Everything(), labels.Everything())
ctx := api.NewContext()
s, _ := storage.List(ctx, labels.Everything(), labels.Everything())
sl := s.(*api.ServiceList)
if len(fakeCloud.Calls) != 0 {
t.Errorf("Unexpected call(s): %#v", fakeCloud.Calls)