Commit 52db5766 authored by Abhi Shah's avatar Abhi Shah

Merge pull request #8882 from mesosphere/upstream_k8sm

Upstream Kubernetes-Mesos framework
parents 0ca96c3a 6436c4a3
...@@ -359,6 +359,10 @@ ...@@ -359,6 +359,10 @@
"Rev": "afde71eb1740fd763ab9450e1f700ba0e53c36d0" "Rev": "afde71eb1740fd763ab9450e1f700ba0e53c36d0"
}, },
{ {
"ImportPath": "github.com/kardianos/osext",
"Rev": "8fef92e41e22a70e700a96b29f066cda30ea24ef"
},
{
"ImportPath": "github.com/kr/pty", "ImportPath": "github.com/kr/pty",
"Comment": "release.r56-25-g05017fc", "Comment": "release.r56-25-g05017fc",
"Rev": "05017fcccf23c823bfdea560dcc958a136e54fb7" "Rev": "05017fcccf23c823bfdea560dcc958a136e54fb7"
...@@ -368,10 +372,18 @@ ...@@ -368,10 +372,18 @@
"Rev": "fc2b8d3a73c4867e51861bbdd5ae3c1f0869dd6a" "Rev": "fc2b8d3a73c4867e51861bbdd5ae3c1f0869dd6a"
}, },
{ {
"ImportPath": "github.com/mesos/mesos-go/auth",
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
},
{
"ImportPath": "github.com/mesos/mesos-go/detector", "ImportPath": "github.com/mesos/mesos-go/detector",
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a" "Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
}, },
{ {
"ImportPath": "github.com/mesos/mesos-go/executor",
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
},
{
"ImportPath": "github.com/mesos/mesos-go/mesosproto", "ImportPath": "github.com/mesos/mesos-go/mesosproto",
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a" "Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
}, },
...@@ -380,6 +392,14 @@ ...@@ -380,6 +392,14 @@
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a" "Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
}, },
{ {
"ImportPath": "github.com/mesos/mesos-go/messenger",
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
},
{
"ImportPath": "github.com/mesos/mesos-go/scheduler",
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
},
{
"ImportPath": "github.com/mesos/mesos-go/upid", "ImportPath": "github.com/mesos/mesos-go/upid",
"Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a" "Rev": "4b1767c0dfc51020e01f35da5b38472f40ce572a"
}, },
......
Copyright (c) 2012 The Go Authors. All rights reserved.
Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:
* Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.
* Redistributions in binary form must reproduce the above
copyright notice, this list of conditions and the following disclaimer
in the documentation and/or other materials provided with the
distribution.
* Neither the name of Google Inc. nor the names of its
contributors may be used to endorse or promote products derived from
this software without specific prior written permission.
THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
### Extensions to the "os" package.
## Find the current Executable and ExecutableFolder.
There is sometimes utility in finding the current executable file
that is running. This can be used for upgrading the current executable
or finding resources located relative to the executable file.
Multi-platform and supports:
* Linux
* OS X
* Windows
* Plan 9
* BSDs.
// Copyright 2012 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.
// Extensions to the standard "os" package.
package osext
import "path/filepath"
// Executable returns an absolute path that can be used to
// re-invoke the current program.
// It may not be valid after the current program exits.
func Executable() (string, error) {
p, err := executable()
return filepath.Clean(p), err
}
// Returns same path as Executable, returns just the folder
// path. Excludes the executable name.
func ExecutableFolder() (string, error) {
p, err := Executable()
if err != nil {
return "", err
}
folder, _ := filepath.Split(p)
return folder, nil
}
// Copyright 2012 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 osext
import (
"os"
"strconv"
"syscall"
)
func executable() (string, error) {
f, err := os.Open("/proc/" + strconv.Itoa(os.Getpid()) + "/text")
if err != nil {
return "", err
}
defer f.Close()
return syscall.Fd2path(int(f.Fd()))
}
// Copyright 2012 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.
// +build linux netbsd openbsd solaris dragonfly
package osext
import (
"errors"
"fmt"
"os"
"runtime"
"strings"
)
func executable() (string, error) {
switch runtime.GOOS {
case "linux":
const deletedTag = " (deleted)"
execpath, err := os.Readlink("/proc/self/exe")
if err != nil {
return execpath, err
}
execpath = strings.TrimSuffix(execpath, deletedTag)
execpath = strings.TrimPrefix(execpath, deletedTag)
return execpath, nil
case "netbsd":
return os.Readlink("/proc/curproc/exe")
case "openbsd", "dragonfly":
return os.Readlink("/proc/curproc/file")
case "solaris":
return os.Readlink(fmt.Sprintf("/proc/%d/path/a.out", os.Getpid()))
}
return "", errors.New("ExecPath not implemented for " + runtime.GOOS)
}
// Copyright 2012 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.
// +build darwin freebsd
package osext
import (
"os"
"path/filepath"
"runtime"
"syscall"
"unsafe"
)
var initCwd, initCwdErr = os.Getwd()
func executable() (string, error) {
var mib [4]int32
switch runtime.GOOS {
case "freebsd":
mib = [4]int32{1 /* CTL_KERN */, 14 /* KERN_PROC */, 12 /* KERN_PROC_PATHNAME */, -1}
case "darwin":
mib = [4]int32{1 /* CTL_KERN */, 38 /* KERN_PROCARGS */, int32(os.Getpid()), -1}
}
n := uintptr(0)
// Get length.
_, _, errNum := syscall.Syscall6(syscall.SYS___SYSCTL, uintptr(unsafe.Pointer(&mib[0])), 4, 0, uintptr(unsafe.Pointer(&n)), 0, 0)
if errNum != 0 {
return "", errNum
}
if n == 0 { // This shouldn't happen.
return "", nil
}
buf := make([]byte, n)
_, _, errNum = syscall.Syscall6(syscall.SYS___SYSCTL, uintptr(unsafe.Pointer(&mib[0])), 4, uintptr(unsafe.Pointer(&buf[0])), uintptr(unsafe.Pointer(&n)), 0, 0)
if errNum != 0 {
return "", errNum
}
if n == 0 { // This shouldn't happen.
return "", nil
}
for i, v := range buf {
if v == 0 {
buf = buf[:i]
break
}
}
var err error
execPath := string(buf)
// execPath will not be empty due to above checks.
// Try to get the absolute path if the execPath is not rooted.
if execPath[0] != '/' {
execPath, err = getAbs(execPath)
if err != nil {
return execPath, err
}
}
// For darwin KERN_PROCARGS may return the path to a symlink rather than the
// actual executable.
if runtime.GOOS == "darwin" {
if execPath, err = filepath.EvalSymlinks(execPath); err != nil {
return execPath, err
}
}
return execPath, nil
}
func getAbs(execPath string) (string, error) {
if initCwdErr != nil {
return execPath, initCwdErr
}
// The execPath may begin with a "../" or a "./" so clean it first.
// Join the two paths, trailing and starting slashes undetermined, so use
// the generic Join function.
return filepath.Join(initCwd, filepath.Clean(execPath)), nil
}
// Copyright 2012 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.
// +build darwin linux freebsd netbsd windows
package osext
import (
"bytes"
"fmt"
"io"
"os"
"os/exec"
"path/filepath"
"runtime"
"testing"
)
const (
executableEnvVar = "OSTEST_OUTPUT_EXECUTABLE"
executableEnvValueMatch = "match"
executableEnvValueDelete = "delete"
)
func TestExecutableMatch(t *testing.T) {
ep, err := Executable()
if err != nil {
t.Fatalf("Executable failed: %v", err)
}
// fullpath to be of the form "dir/prog".
dir := filepath.Dir(filepath.Dir(ep))
fullpath, err := filepath.Rel(dir, ep)
if err != nil {
t.Fatalf("filepath.Rel: %v", err)
}
// Make child start with a relative program path.
// Alter argv[0] for child to verify getting real path without argv[0].
cmd := &exec.Cmd{
Dir: dir,
Path: fullpath,
Env: []string{fmt.Sprintf("%s=%s", executableEnvVar, executableEnvValueMatch)},
}
out, err := cmd.CombinedOutput()
if err != nil {
t.Fatalf("exec(self) failed: %v", err)
}
outs := string(out)
if !filepath.IsAbs(outs) {
t.Fatalf("Child returned %q, want an absolute path", out)
}
if !sameFile(outs, ep) {
t.Fatalf("Child returned %q, not the same file as %q", out, ep)
}
}
func TestExecutableDelete(t *testing.T) {
if runtime.GOOS != "linux" {
t.Skip()
}
fpath, err := Executable()
if err != nil {
t.Fatalf("Executable failed: %v", err)
}
r, w := io.Pipe()
stderrBuff := &bytes.Buffer{}
stdoutBuff := &bytes.Buffer{}
cmd := &exec.Cmd{
Path: fpath,
Env: []string{fmt.Sprintf("%s=%s", executableEnvVar, executableEnvValueDelete)},
Stdin: r,
Stderr: stderrBuff,
Stdout: stdoutBuff,
}
err = cmd.Start()
if err != nil {
t.Fatalf("exec(self) start failed: %v", err)
}
tempPath := fpath + "_copy"
_ = os.Remove(tempPath)
err = copyFile(tempPath, fpath)
if err != nil {
t.Fatalf("copy file failed: %v", err)
}
err = os.Remove(fpath)
if err != nil {
t.Fatalf("remove running test file failed: %v", err)
}
err = os.Rename(tempPath, fpath)
if err != nil {
t.Fatalf("rename copy to previous name failed: %v", err)
}
w.Write([]byte{0})
w.Close()
err = cmd.Wait()
if err != nil {
t.Fatalf("exec wait failed: %v", err)
}
childPath := stderrBuff.String()
if !filepath.IsAbs(childPath) {
t.Fatalf("Child returned %q, want an absolute path", childPath)
}
if !sameFile(childPath, fpath) {
t.Fatalf("Child returned %q, not the same file as %q", childPath, fpath)
}
}
func sameFile(fn1, fn2 string) bool {
fi1, err := os.Stat(fn1)
if err != nil {
return false
}
fi2, err := os.Stat(fn2)
if err != nil {
return false
}
return os.SameFile(fi1, fi2)
}
func copyFile(dest, src string) error {
df, err := os.Create(dest)
if err != nil {
return err
}
defer df.Close()
sf, err := os.Open(src)
if err != nil {
return err
}
defer sf.Close()
_, err = io.Copy(df, sf)
return err
}
func TestMain(m *testing.M) {
env := os.Getenv(executableEnvVar)
switch env {
case "":
os.Exit(m.Run())
case executableEnvValueMatch:
// First chdir to another path.
dir := "/"
if runtime.GOOS == "windows" {
dir = filepath.VolumeName(".")
}
os.Chdir(dir)
if ep, err := Executable(); err != nil {
fmt.Fprint(os.Stderr, "ERROR: ", err)
} else {
fmt.Fprint(os.Stderr, ep)
}
case executableEnvValueDelete:
bb := make([]byte, 1)
var err error
n, err := os.Stdin.Read(bb)
if err != nil {
fmt.Fprint(os.Stderr, "ERROR: ", err)
os.Exit(2)
}
if n != 1 {
fmt.Fprint(os.Stderr, "ERROR: n != 1, n == ", n)
os.Exit(2)
}
if ep, err := Executable(); err != nil {
fmt.Fprint(os.Stderr, "ERROR: ", err)
} else {
fmt.Fprint(os.Stderr, ep)
}
}
os.Exit(0)
}
// Copyright 2012 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 osext
import (
"syscall"
"unicode/utf16"
"unsafe"
)
var (
kernel = syscall.MustLoadDLL("kernel32.dll")
getModuleFileNameProc = kernel.MustFindProc("GetModuleFileNameW")
)
// GetModuleFileName() with hModule = NULL
func executable() (exePath string, err error) {
return getModuleFileName()
}
func getModuleFileName() (string, error) {
var n uint32
b := make([]uint16, syscall.MAX_PATH)
size := uint32(len(b))
r0, _, e1 := getModuleFileNameProc.Call(0, uintptr(unsafe.Pointer(&b[0])), uintptr(size))
n = uint32(r0)
if n == 0 {
return "", e1
}
return string(utf16.Decode(b[0:n])), nil
}
package callback
import (
"fmt"
)
type Unsupported struct {
Callback Interface
}
func (uc *Unsupported) Error() string {
return fmt.Sprintf("Unsupported callback <%T>: %v", uc.Callback, uc.Callback)
}
type Interface interface {
// marker interface
}
type Handler interface {
// may return an Unsupported error on failure
Handle(callbacks ...Interface) error
}
type HandlerFunc func(callbacks ...Interface) error
func (f HandlerFunc) Handle(callbacks ...Interface) error {
return f(callbacks...)
}
package callback
import (
"github.com/mesos/mesos-go/upid"
)
type Interprocess struct {
client upid.UPID
server upid.UPID
}
func NewInterprocess() *Interprocess {
return &Interprocess{}
}
func (cb *Interprocess) Client() upid.UPID {
return cb.client
}
func (cb *Interprocess) Server() upid.UPID {
return cb.server
}
func (cb *Interprocess) Set(server, client upid.UPID) {
cb.server = server
cb.client = client
}
package callback
type Name struct {
name string
}
func NewName() *Name {
return &Name{}
}
func (cb *Name) Get() string {
return cb.name
}
func (cb *Name) Set(name string) {
cb.name = name
}
package callback
type Password struct {
password []byte
}
func NewPassword() *Password {
return &Password{}
}
func (cb *Password) Get() []byte {
clone := make([]byte, len(cb.password))
copy(clone, cb.password)
return clone
}
func (cb *Password) Set(password []byte) {
cb.password = make([]byte, len(password))
copy(cb.password, password)
}
package auth
import (
"errors"
"fmt"
"sync"
log "github.com/golang/glog"
"github.com/mesos/mesos-go/auth/callback"
"golang.org/x/net/context"
)
// SPI interface: login provider implementations support this interface, clients
// do not authenticate against this directly, instead they should use Login()
type Authenticatee interface {
// Returns no errors if successfully authenticated, otherwise a single
// error.
Authenticate(ctx context.Context, handler callback.Handler) error
}
// Func adapter for interface: allow func's to implement the Authenticatee interface
// as long as the func signature matches
type AuthenticateeFunc func(ctx context.Context, handler callback.Handler) error
func (f AuthenticateeFunc) Authenticate(ctx context.Context, handler callback.Handler) error {
return f(ctx, handler)
}
var (
// Authentication was attempted and failed (likely due to incorrect credentials, too
// many retries within a time window, etc). Distinctly different from authentication
// errors (e.g. network errors, configuration errors, etc).
AuthenticationFailed = errors.New("authentication failed")
authenticateeProviders = make(map[string]Authenticatee) // authentication providers dict
providerLock sync.Mutex
)
// Register an authentication provider (aka "login provider"). packages that
// provide Authenticatee implementations should invoke this func in their
// init() to register.
func RegisterAuthenticateeProvider(name string, auth Authenticatee) (err error) {
providerLock.Lock()
defer providerLock.Unlock()
if _, found := authenticateeProviders[name]; found {
err = fmt.Errorf("authentication provider already registered: %v", name)
} else {
authenticateeProviders[name] = auth
log.V(1).Infof("registered authentication provider: %v", name)
}
return
}
// Look up an authentication provider by name, returns non-nil and true if such
// a provider is found.
func getAuthenticateeProvider(name string) (provider Authenticatee, ok bool) {
providerLock.Lock()
defer providerLock.Unlock()
provider, ok = authenticateeProviders[name]
return
}
package auth
import (
"errors"
"fmt"
"github.com/mesos/mesos-go/auth/callback"
"github.com/mesos/mesos-go/upid"
"golang.org/x/net/context"
)
var (
// No login provider name has been specified in a context.Context
NoLoginProviderName = errors.New("missing login provider name in context")
)
// Main client entrypoint into the authentication APIs: clients are expected to
// invoke this func with a context containing a login provider name value.
// This may be written as:
// providerName := ... // the user has probably configured this via some flag
// handler := ... // handlers provide data like usernames and passwords
// ctx := ... // obtain some initial or timed context
// err := auth.Login(auth.WithLoginProvider(ctx, providerName), handler)
func Login(ctx context.Context, handler callback.Handler) error {
name, ok := LoginProviderFrom(ctx)
if !ok {
return NoLoginProviderName
}
provider, ok := getAuthenticateeProvider(name)
if !ok {
return fmt.Errorf("unrecognized login provider name in context: %s", name)
}
return provider.Authenticate(ctx, handler)
}
// Unexported key type, avoids conflicts with other context-using packages. All
// context items registered from this package should use keys of this type.
type loginKeyType int
const (
loginProviderNameKey loginKeyType = iota // name of login provider to use
parentUpidKey // upid.UPID of some parent process
)
// Return a context that inherits all values from the parent ctx and specifies
// the login provider name given here. Intended to be invoked before calls to
// Login().
func WithLoginProvider(ctx context.Context, providerName string) context.Context {
return context.WithValue(ctx, loginProviderNameKey, providerName)
}
// Return the name of the login provider specified in this context.
func LoginProviderFrom(ctx context.Context) (name string, ok bool) {
name, ok = ctx.Value(loginProviderNameKey).(string)
return
}
// Return the name of the login provider specified in this context, or empty
// string if none.
func LoginProvider(ctx context.Context) string {
name, _ := LoginProviderFrom(ctx)
return name
}
func WithParentUPID(ctx context.Context, pid upid.UPID) context.Context {
return context.WithValue(ctx, parentUpidKey, pid)
}
func ParentUPIDFrom(ctx context.Context) (pid upid.UPID, ok bool) {
pid, ok = ctx.Value(parentUpidKey).(upid.UPID)
return
}
func ParentUPID(ctx context.Context) (upid *upid.UPID) {
if upid, ok := ParentUPIDFrom(ctx); ok {
return &upid
} else {
return nil
}
}
package sasl
import (
"testing"
"time"
"github.com/gogo/protobuf/proto"
"github.com/mesos/mesos-go/auth/callback"
"github.com/mesos/mesos-go/auth/sasl/mech/crammd5"
mesos "github.com/mesos/mesos-go/mesosproto"
"github.com/mesos/mesos-go/messenger"
"github.com/mesos/mesos-go/upid"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
"golang.org/x/net/context"
)
type MockTransport struct {
*messenger.MockedMessenger
}
func (m *MockTransport) Send(ctx context.Context, upid *upid.UPID, msg proto.Message) error {
return m.Called(mock.Anything, upid, msg).Error(0)
}
func TestAuthticatee_validLogin(t *testing.T) {
assert := assert.New(t)
ctx := context.TODO()
client := upid.UPID{
ID: "someFramework",
Host: "b.net",
Port: "789",
}
server := upid.UPID{
ID: "serv",
Host: "a.com",
Port: "123",
}
tpid := upid.UPID{
ID: "sasl_transport",
Host: "g.org",
Port: "456",
}
handler := callback.HandlerFunc(func(cb ...callback.Interface) error {
for _, c := range cb {
switch c := c.(type) {
case *callback.Name:
c.Set("foo")
case *callback.Password:
c.Set([]byte("bar"))
case *callback.Interprocess:
c.Set(server, client)
default:
return &callback.Unsupported{Callback: c}
}
}
return nil
})
var transport *MockTransport
factory := transportFactoryFunc(func() messenger.Messenger {
transport = &MockTransport{messenger.NewMockedMessenger()}
transport.On("Install").Return(nil)
transport.On("UPID").Return(&tpid)
transport.On("Start").Return(nil)
transport.On("Stop").Return(nil)
transport.On("Send", mock.Anything, &server, &mesos.AuthenticateMessage{
Pid: proto.String(client.String()),
}).Return(nil).Once()
transport.On("Send", mock.Anything, &server, &mesos.AuthenticationStartMessage{
Mechanism: proto.String(crammd5.Name),
Data: proto.String(""), // may be nil, depends on init step
}).Return(nil).Once()
transport.On("Send", mock.Anything, &server, &mesos.AuthenticationStepMessage{
Data: []byte(`foo cc7fd96cd80123ea844a7dba29a594ed`),
}).Return(nil).Once()
go func() {
transport.Recv(&server, &mesos.AuthenticationMechanismsMessage{
Mechanisms: []string{crammd5.Name},
})
transport.Recv(&server, &mesos.AuthenticationStepMessage{
Data: []byte(`lsd;lfkgjs;dlfkgjs;dfklg`),
})
transport.Recv(&server, &mesos.AuthenticationCompletedMessage{})
}()
return transport
})
login, err := makeAuthenticatee(handler, factory)
assert.Nil(err)
err = login.Authenticate(ctx, handler)
assert.Nil(err)
assert.NotNil(transport)
time.Sleep(1 * time.Second) // wait for the authenticator to shut down
transport.AssertExpectations(t)
}
package sasl
import (
"net"
"golang.org/x/net/context"
)
// unexported to prevent collisions with context keys defined in
// other packages.
type _key int
// If this package defined other context keys, they would have
// different integer values.
const (
statusKey _key = iota
bindingAddressKey // bind address for login-related network ops
)
func withStatus(ctx context.Context, s statusType) context.Context {
return context.WithValue(ctx, statusKey, s)
}
func statusFrom(ctx context.Context) statusType {
s, ok := ctx.Value(statusKey).(statusType)
if !ok {
panic("missing status in context")
}
return s
}
func WithBindingAddress(ctx context.Context, address net.IP) context.Context {
return context.WithValue(ctx, bindingAddressKey, address)
}
func BindingAddressFrom(ctx context.Context) net.IP {
obj := ctx.Value(bindingAddressKey)
if addr, ok := obj.(net.IP); ok {
return addr
} else {
return nil
}
}
package crammd5
import (
"crypto/hmac"
"crypto/md5"
"encoding/hex"
"errors"
"io"
log "github.com/golang/glog"
"github.com/mesos/mesos-go/auth/callback"
"github.com/mesos/mesos-go/auth/sasl/mech"
)
var (
Name = "CRAM-MD5" // name this mechanism is registered with
//TODO(jdef) is this a generic SASL error? if so, move it up to mech
challengeDataRequired = errors.New("challenge data may not be empty")
)
func init() {
mech.Register(Name, newInstance)
}
type mechanism struct {
handler callback.Handler
}
func (m *mechanism) Handler() callback.Handler {
return m.handler
}
func (m *mechanism) Discard() {
// noop
}
func newInstance(h callback.Handler) (mech.Interface, mech.StepFunc, error) {
m := &mechanism{
handler: h,
}
fn := func(m mech.Interface, data []byte) (mech.StepFunc, []byte, error) {
// noop: no initialization needed
return challengeResponse, nil, nil
}
return m, fn, nil
}
// algorithm lifted from wikipedia: http://en.wikipedia.org/wiki/CRAM-MD5
// except that the SASL mechanism used by Mesos doesn't leverage base64 encoding
func challengeResponse(m mech.Interface, data []byte) (mech.StepFunc, []byte, error) {
if len(data) == 0 {
return mech.IllegalState, nil, challengeDataRequired
}
decoded := string(data)
log.V(4).Infof("challenge(decoded): %s", decoded) // for deep debugging only
username := callback.NewName()
secret := callback.NewPassword()
if err := m.Handler().Handle(username, secret); err != nil {
return mech.IllegalState, nil, err
}
hash := hmac.New(md5.New, secret.Get())
if _, err := io.WriteString(hash, decoded); err != nil {
return mech.IllegalState, nil, err
}
codes := hex.EncodeToString(hash.Sum(nil))
msg := username.Get() + " " + codes
return nil, []byte(msg), nil
}
package mech
import (
"errors"
"github.com/mesos/mesos-go/auth/callback"
)
var (
IllegalStateErr = errors.New("illegal mechanism state")
)
type Interface interface {
Handler() callback.Handler
Discard() // clean up resources or sensitive information; idempotent
}
// return a mechanism and it's initialization step (may be a noop that returns
// a nil data blob and handle to the first "real" challenge step).
type Factory func(h callback.Handler) (Interface, StepFunc, error)
// StepFunc implementations should never return a nil StepFunc result. This
// helps keep the logic in the SASL authticatee simpler: step functions are
// never nil. Mechanisms that end up an error state (for example, some decoding
// logic fails...) should return a StepFunc that represents an error state.
// Some mechanisms may be able to recover from such.
type StepFunc func(m Interface, data []byte) (StepFunc, []byte, error)
// reflects an unrecoverable, illegal mechanism state; always returns IllegalState
// as the next step along with an IllegalStateErr
func IllegalState(m Interface, data []byte) (StepFunc, []byte, error) {
return IllegalState, nil, IllegalStateErr
}
package mech
import (
"fmt"
"sync"
log "github.com/golang/glog"
)
var (
mechLock sync.Mutex
supportedMechs = make(map[string]Factory)
)
func Register(name string, f Factory) error {
mechLock.Lock()
defer mechLock.Unlock()
if _, found := supportedMechs[name]; found {
return fmt.Errorf("Mechanism registered twice: %s", name)
}
supportedMechs[name] = f
log.V(1).Infof("Registered mechanism %s", name)
return nil
}
func ListSupported() (list []string) {
mechLock.Lock()
defer mechLock.Unlock()
for mechname := range supportedMechs {
list = append(list, mechname)
}
return list
}
func SelectSupported(mechanisms []string) (selectedMech string, factory Factory) {
mechLock.Lock()
defer mechLock.Unlock()
for _, m := range mechanisms {
if f, ok := supportedMechs[m]; ok {
selectedMech = m
factory = f
break
}
}
return
}
/*
Package executor includes the interfaces of the mesos executor and
the mesos executor driver, as well as an implementation of the driver.
*/
package executor
package executor
import (
"github.com/mesos/mesos-go/mesosproto"
)
/**
* Executor callback interface to be implemented by frameworks' executors. Note
* that only one callback will be invoked at a time, so it is not
* recommended that you block within a callback because it may cause a
* deadlock.
*
* Each callback includes an instance to the executor driver that was
* used to run this executor. The driver will not change for the
* duration of an executor (i.e., from the point you do
* ExecutorDriver.Start() to the point that ExecutorDriver.Join()
* returns). This is intended for convenience so that an executor
* doesn't need to store a pointer to the driver itself.
*/
type Executor interface {
/**
* Invoked once the executor driver has been able to successfully
* connect with Mesos. In particular, a scheduler can pass some
* data to its executors through the FrameworkInfo.ExecutorInfo's
* data field.
*/
Registered(ExecutorDriver, *mesosproto.ExecutorInfo, *mesosproto.FrameworkInfo, *mesosproto.SlaveInfo)
/**
* Invoked when the executor re-registers with a restarted slave.
*/
Reregistered(ExecutorDriver, *mesosproto.SlaveInfo)
/**
* Invoked when the executor becomes "disconnected" from the slave
* (e.g., the slave is being restarted due to an upgrade).
*/
Disconnected(ExecutorDriver)
/**
* Invoked when a task has been launched on this executor (initiated
* via SchedulerDriver.LaunchTasks). Note that this task can be realized
* with a goroutine, an external process, or some simple computation, however,
* no other callbacks will be invoked on this executor until this
* callback has returned.
*/
LaunchTask(ExecutorDriver, *mesosproto.TaskInfo)
/**
* Invoked when a task running within this executor has been killed
* (via SchedulerDriver.KillTask). Note that no status update will
* be sent on behalf of the executor, the executor is responsible
* for creating a new TaskStatus (i.e., with TASK_KILLED) and
* invoking ExecutorDriver.SendStatusUpdate.
*/
KillTask(ExecutorDriver, *mesosproto.TaskID)
/**
* Invoked when a framework message has arrived for this
* executor. These messages are best effort; do not expect a
* framework message to be retransmitted in any reliable fashion.
*/
FrameworkMessage(ExecutorDriver, string)
/**
* Invoked when the executor should terminate all of its currently
* running tasks. Note that after Mesos has determined that an
* executor has terminated, any tasks that the executor did not send
* terminal status updates for (e.g., TASK_KILLED, TASK_FINISHED,
* TASK_FAILED, etc) a TASK_LOST status update will be created.
*/
Shutdown(ExecutorDriver)
/**
* Invoked when a fatal error has occured with the executor and/or
* executor driver. The driver will be aborted BEFORE invoking this
* callback.
*/
Error(ExecutorDriver, string)
}
/**
* ExecutorDriver interface for connecting an executor to Mesos. This
* interface is used both to manage the executor's lifecycle (start
* it, stop it, or wait for it to finish) and to interact with Mesos
* (e.g., send status updates, send framework messages, etc.).
* A driver method is expected to fail-fast and return an error when possible.
* Other internal errors (or remote error) that occur asynchronously are handled
* using the the Executor.Error() callback.
*/
type ExecutorDriver interface {
/**
* Starts the executor driver. This needs to be called before any
* other driver calls are made.
*/
Start() (mesosproto.Status, error)
/**
* Stops the executor driver.
*/
Stop() (mesosproto.Status, error)
/**
* Aborts the driver so that no more callbacks can be made to the
* executor. The semantics of abort and stop have deliberately been
* separated so that code can detect an aborted driver (i.e., via
* the return status of ExecutorDriver.Join, see below), and
* instantiate and start another driver if desired (from within the
* same process ... although this functionality is currently not
* supported for executors).
*/
Abort() (mesosproto.Status, error)
/**
* Waits for the driver to be stopped or aborted, possibly
* blocking the calling goroutine indefinitely. The return status of
* this function can be used to determine if the driver was aborted
* (see package mesosproto for a description of Status).
*/
Join() (mesosproto.Status, error)
/**
* Starts and immediately joins (i.e., blocks on) the driver.
*/
Run() (mesosproto.Status, error)
/**
* Sends a status update to the framework scheduler, retrying as
* necessary until an acknowledgement has been received or the
* executor is terminated (in which case, a TASK_LOST status update
* will be sent). See Scheduler.StatusUpdate for more information
* about status update acknowledgements.
*/
SendStatusUpdate(*mesosproto.TaskStatus) (mesosproto.Status, error)
/**
* Sends a message to the framework scheduler. These messages are
* best effort; do not expect a framework message to be
* retransmitted in any reliable fashion.
*/
SendFrameworkMessage(string) (mesosproto.Status, error)
}
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 executor
import (
"github.com/mesos/mesos-go/mesosproto"
"github.com/stretchr/testify/mock"
)
// MockedExecutor is used for testing the executor driver.
type MockedExecutor struct {
mock.Mock
}
// NewMockedExecutor returns a mocked executor.
func NewMockedExecutor() *MockedExecutor {
return &MockedExecutor{}
}
// Registered implements the Registered handler.
func (e *MockedExecutor) Registered(ExecutorDriver, *mesosproto.ExecutorInfo, *mesosproto.FrameworkInfo, *mesosproto.SlaveInfo) {
e.Called()
}
// Reregistered implements the Reregistered handler.
func (e *MockedExecutor) Reregistered(ExecutorDriver, *mesosproto.SlaveInfo) {
e.Called()
}
// Disconnected implements the Disconnected handler.
func (e *MockedExecutor) Disconnected(ExecutorDriver) {
e.Called()
}
// LaunchTask implements the LaunchTask handler.
func (e *MockedExecutor) LaunchTask(ExecutorDriver, *mesosproto.TaskInfo) {
e.Called()
}
// KillTask implements the KillTask handler.
func (e *MockedExecutor) KillTask(ExecutorDriver, *mesosproto.TaskID) {
e.Called()
}
// FrameworkMessage implements the FrameworkMessage handler.
func (e *MockedExecutor) FrameworkMessage(ExecutorDriver, string) {
e.Called()
}
// Shutdown implements the Shutdown handler.
func (e *MockedExecutor) Shutdown(ExecutorDriver) {
e.Called()
}
// Error implements the Error handler.
func (e *MockedExecutor) Error(ExecutorDriver, string) {
e.Called()
}
####Benchmark of the messenger.
```shell
$ go test -v -run=Benckmark* -bench=.
PASS
BenchmarkMessengerSendSmallMessage 50000 70568 ns/op
BenchmarkMessengerSendMediumMessage 50000 70265 ns/op
BenchmarkMessengerSendBigMessage 50000 72693 ns/op
BenchmarkMessengerSendLargeMessage 50000 72896 ns/op
BenchmarkMessengerSendMixedMessage 50000 72631 ns/op
BenchmarkMessengerSendRecvSmallMessage 20000 78409 ns/op
BenchmarkMessengerSendRecvMediumMessage 20000 80471 ns/op
BenchmarkMessengerSendRecvBigMessage 20000 82629 ns/op
BenchmarkMessengerSendRecvLargeMessage 20000 85987 ns/op
BenchmarkMessengerSendRecvMixedMessage 20000 83678 ns/op
ok github.com/mesos/mesos-go/messenger 115.135s
$ go test -v -run=Benckmark* -bench=. -cpu=4 -send-routines=4 2>/dev/null
PASS
BenchmarkMessengerSendSmallMessage-4 50000 35529 ns/op
BenchmarkMessengerSendMediumMessage-4 50000 35997 ns/op
BenchmarkMessengerSendBigMessage-4 50000 36871 ns/op
BenchmarkMessengerSendLargeMessage-4 50000 37310 ns/op
BenchmarkMessengerSendMixedMessage-4 50000 37419 ns/op
BenchmarkMessengerSendRecvSmallMessage-4 50000 39320 ns/op
BenchmarkMessengerSendRecvMediumMessage-4 50000 41990 ns/op
BenchmarkMessengerSendRecvBigMessage-4 50000 42157 ns/op
BenchmarkMessengerSendRecvLargeMessage-4 50000 45472 ns/op
BenchmarkMessengerSendRecvMixedMessage-4 50000 47393 ns/op
ok github.com/mesos/mesos-go/messenger 105.173s
```
####environment:
```
OS: Linux yifan-laptop 3.13.0-32-generic #57-Ubuntu SMP Tue Jul 15 03:51:08 UTC 2014 x86_64 x86_64 x86_64 GNU/Linux
CPU: Intel(R) Core(TM) i5-3210M CPU @ 2.50GHz
MEM: 4G DDR3 1600MHz
```
/*
Package messenger includes a messenger and a transporter.
The messenger provides interfaces to send a protobuf message
through the underlying transporter. It also dispatches messages
to installed handlers.
*/
package messenger
package messenger
import (
"fmt"
"net/http"
"net/http/httptest"
"regexp"
"strconv"
"testing"
"time"
"github.com/mesos/mesos-go/messenger/testmessage"
"github.com/mesos/mesos-go/upid"
"github.com/stretchr/testify/assert"
"golang.org/x/net/context"
)
func TestTransporterNew(t *testing.T) {
id, err := upid.Parse(fmt.Sprintf("mesos1@localhost:%d", getNewPort()))
assert.NoError(t, err)
trans := NewHTTPTransporter(id, nil)
assert.NotNil(t, trans)
assert.NotNil(t, trans.upid)
assert.NotNil(t, trans.messageQueue)
assert.NotNil(t, trans.client)
}
func TestTransporterSend(t *testing.T) {
idreg := regexp.MustCompile(`[A-Za-z0-9_\-]+@[A-Za-z0-9_\-\.]+:[0-9]+`)
serverId := "testserver"
// setup mesos client-side
fromUpid, err := upid.Parse(fmt.Sprintf("mesos1@localhost:%d", getNewPort()))
assert.NoError(t, err)
protoMsg := testmessage.GenerateSmallMessage()
msgName := getMessageName(protoMsg)
msg := &Message{
Name: msgName,
ProtoMessage: protoMsg,
}
requestURI := fmt.Sprintf("/%s/%s", serverId, msgName)
// setup server-side
msgReceived := make(chan struct{})
srv := makeMockServer(requestURI, func(rsp http.ResponseWriter, req *http.Request) {
defer close(msgReceived)
from := req.Header.Get("Libprocess-From")
assert.NotEmpty(t, from)
assert.True(t, idreg.MatchString(from), fmt.Sprintf("regexp failed for '%v'", from))
})
defer srv.Close()
toUpid, err := upid.Parse(fmt.Sprintf("%s@%s", serverId, srv.Listener.Addr().String()))
assert.NoError(t, err)
// make transport call.
transport := NewHTTPTransporter(fromUpid, nil)
errch := transport.Start()
defer transport.Stop(false)
msg.UPID = toUpid
err = transport.Send(context.TODO(), msg)
assert.NoError(t, err)
select {
case <-time.After(2 * time.Second):
t.Fatalf("timed out waiting for message receipt")
case <-msgReceived:
case err := <-errch:
if err != nil {
t.Fatalf(err.Error())
}
}
}
func TestTransporter_DiscardedSend(t *testing.T) {
serverId := "testserver"
// setup mesos client-side
fromUpid, err := upid.Parse(fmt.Sprintf("mesos1@localhost:%d", getNewPort()))
assert.NoError(t, err)
protoMsg := testmessage.GenerateSmallMessage()
msgName := getMessageName(protoMsg)
msg := &Message{
Name: msgName,
ProtoMessage: protoMsg,
}
requestURI := fmt.Sprintf("/%s/%s", serverId, msgName)
// setup server-side
msgReceived := make(chan struct{})
srv := makeMockServer(requestURI, func(rsp http.ResponseWriter, req *http.Request) {
close(msgReceived)
time.Sleep(2 * time.Second) // long enough that we should be able to stop it
})
defer srv.Close()
toUpid, err := upid.Parse(fmt.Sprintf("%s@%s", serverId, srv.Listener.Addr().String()))
assert.NoError(t, err)
// make transport call.
transport := NewHTTPTransporter(fromUpid, nil)
errch := transport.Start()
defer transport.Stop(false)
msg.UPID = toUpid
senderr := make(chan struct{})
go func() {
defer close(senderr)
err = transport.Send(context.TODO(), msg)
assert.NotNil(t, err)
assert.Equal(t, discardOnStopError, err)
}()
// wait for message to be received
select {
case <-time.After(2 * time.Second):
t.Fatalf("timed out waiting for message receipt")
return
case <-msgReceived:
transport.Stop(false)
case err := <-errch:
if err != nil {
t.Fatalf(err.Error())
return
}
}
// wait for send() to process discarded-error
select {
case <-time.After(5 * time.Second):
t.Fatalf("timed out waiting for aborted send")
return
case <-senderr: // continue
}
}
func TestTransporterStartAndRcvd(t *testing.T) {
serverId := "testserver"
serverPort := getNewPort()
serverAddr := "127.0.0.1:" + strconv.Itoa(serverPort)
protoMsg := testmessage.GenerateSmallMessage()
msgName := getMessageName(protoMsg)
ctrl := make(chan struct{})
// setup receiver (server) process
rcvPid, err := upid.Parse(fmt.Sprintf("%s@%s", serverId, serverAddr))
assert.NoError(t, err)
receiver := NewHTTPTransporter(rcvPid, nil)
receiver.Install(msgName)
go func() {
defer close(ctrl)
msg, err := receiver.Recv()
assert.Nil(t, err)
assert.NotNil(t, msg)
if msg != nil {
assert.Equal(t, msgName, msg.Name)
}
}()
errch := receiver.Start()
defer receiver.Stop(false)
assert.NotNil(t, errch)
time.Sleep(time.Millisecond * 7) // time to catchup
// setup sender (client) process
sndUpid, err := upid.Parse(fmt.Sprintf("mesos1@localhost:%d", getNewPort()))
assert.NoError(t, err)
sender := NewHTTPTransporter(sndUpid, nil)
msg := &Message{
UPID: rcvPid,
Name: msgName,
ProtoMessage: protoMsg,
}
errch2 := sender.Start()
defer sender.Stop(false)
sender.Send(context.TODO(), msg)
select {
case <-time.After(time.Second * 5):
t.Fatalf("Timeout")
case <-ctrl:
case err := <-errch:
if err != nil {
t.Fatalf(err.Error())
}
case err := <-errch2:
if err != nil {
t.Fatalf(err.Error())
}
}
}
func TestTransporterStartAndInject(t *testing.T) {
serverId := "testserver"
serverPort := getNewPort()
serverAddr := "127.0.0.1:" + strconv.Itoa(serverPort)
protoMsg := testmessage.GenerateSmallMessage()
msgName := getMessageName(protoMsg)
ctrl := make(chan struct{})
// setup receiver (server) process
rcvPid, err := upid.Parse(fmt.Sprintf("%s@%s", serverId, serverAddr))
assert.NoError(t, err)
receiver := NewHTTPTransporter(rcvPid, nil)
receiver.Install(msgName)
errch := receiver.Start()
defer receiver.Stop(false)
msg := &Message{
UPID: rcvPid,
Name: msgName,
ProtoMessage: protoMsg,
}
receiver.Inject(context.TODO(), msg)
go func() {
defer close(ctrl)
msg, err := receiver.Recv()
assert.Nil(t, err)
assert.NotNil(t, msg)
if msg != nil {
assert.Equal(t, msgName, msg.Name)
}
}()
select {
case <-time.After(time.Second * 1):
t.Fatalf("Timeout")
case <-ctrl:
case err := <-errch:
if err != nil {
t.Fatalf(err.Error())
}
}
}
func TestTransporterStartAndStop(t *testing.T) {
serverId := "testserver"
serverPort := getNewPort()
serverAddr := "127.0.0.1:" + strconv.Itoa(serverPort)
// setup receiver (server) process
rcvPid, err := upid.Parse(fmt.Sprintf("%s@%s", serverId, serverAddr))
assert.NoError(t, err)
receiver := NewHTTPTransporter(rcvPid, nil)
errch := receiver.Start()
assert.NotNil(t, errch)
time.Sleep(1 * time.Second)
receiver.Stop(false)
select {
case <-time.After(2 * time.Second):
t.Fatalf("timed out waiting for transport to stop")
case err := <-errch:
if err != nil {
t.Fatalf(err.Error())
}
}
}
func makeMockServer(path string, handler func(rsp http.ResponseWriter, req *http.Request)) *httptest.Server {
mux := http.NewServeMux()
mux.HandleFunc(path, handler)
return httptest.NewServer(mux)
}
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 messenger
import (
"fmt"
"strings"
"github.com/gogo/protobuf/proto"
"github.com/mesos/mesos-go/upid"
)
// Message defines the type that passes in the Messenger.
type Message struct {
UPID *upid.UPID
Name string
ProtoMessage proto.Message
Bytes []byte
}
// RequestURI returns the request URI of the message.
func (m *Message) RequestURI() string {
return fmt.Sprintf("/%s/%s", m.UPID.ID, m.Name)
}
// NOTE: This should not fail or panic.
func extractNameFromRequestURI(requestURI string) string {
return strings.Split(requestURI, "/")[2]
}
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 messenger
import (
"reflect"
"github.com/gogo/protobuf/proto"
"github.com/mesos/mesos-go/upid"
"github.com/stretchr/testify/mock"
"golang.org/x/net/context"
)
type message struct {
from *upid.UPID
msg proto.Message
}
// MockedMessenger is a messenger that returns error on every operation.
type MockedMessenger struct {
mock.Mock
messageQueue chan *message
handlers map[string]MessageHandler
stop chan struct{}
}
// NewMockedMessenger returns a mocked messenger used for testing.
func NewMockedMessenger() *MockedMessenger {
return &MockedMessenger{
messageQueue: make(chan *message, 1),
handlers: make(map[string]MessageHandler),
stop: make(chan struct{}),
}
}
// Install is a mocked implementation.
func (m *MockedMessenger) Install(handler MessageHandler, msg proto.Message) error {
m.handlers[reflect.TypeOf(msg).Elem().Name()] = handler
return m.Called().Error(0)
}
// Send is a mocked implementation.
func (m *MockedMessenger) Send(ctx context.Context, upid *upid.UPID, msg proto.Message) error {
return m.Called().Error(0)
}
func (m *MockedMessenger) Route(ctx context.Context, upid *upid.UPID, msg proto.Message) error {
return m.Called().Error(0)
}
// Start is a mocked implementation.
func (m *MockedMessenger) Start() error {
go m.recvLoop()
return m.Called().Error(0)
}
// Stop is a mocked implementation.
func (m *MockedMessenger) Stop() error {
// don't close an already-closed channel
select {
case <-m.stop:
// noop
default:
close(m.stop)
}
return m.Called().Error(0)
}
// UPID is a mocked implementation.
func (m *MockedMessenger) UPID() *upid.UPID {
return m.Called().Get(0).(*upid.UPID)
}
func (m *MockedMessenger) recvLoop() {
for {
select {
case <-m.stop:
return
case msg := <-m.messageQueue:
name := reflect.TypeOf(msg.msg).Elem().Name()
m.handlers[name](msg.from, msg.msg)
}
}
}
// Recv receives a upid and a message, it will dispatch the message to its handler
// with the upid. This is for testing.
func (m *MockedMessenger) Recv(from *upid.UPID, msg proto.Message) {
m.messageQueue <- &message{from, msg}
}
all: testmessage.proto
protoc --proto_path=${GOPATH}/src:${GOPATH}/src/github.com/gogo/protobuf/protobuf:. --gogo_out=. testmessage.proto
package testmessage
import (
"math/rand"
)
func generateRandomString(length int) string {
b := make([]byte, length)
for i := range b {
b[i] = byte(rand.Int())
}
return string(b)
}
// GenerateSmallMessage generates a small size message.
func GenerateSmallMessage() *SmallMessage {
v := make([]string, 3)
for i := range v {
v[i] = generateRandomString(5)
}
return &SmallMessage{Values: v}
}
// GenerateMediumMessage generates a medium size message.
func GenerateMediumMessage() *MediumMessage {
v := make([]string, 10)
for i := range v {
v[i] = generateRandomString(10)
}
return &MediumMessage{Values: v}
}
// GenerateBigMessage generates a big size message.
func GenerateBigMessage() *BigMessage {
v := make([]string, 20)
for i := range v {
v[i] = generateRandomString(20)
}
return &BigMessage{Values: v}
}
// GenerateLargeMessage generates a large size message.
func GenerateLargeMessage() *LargeMessage {
v := make([]string, 30)
for i := range v {
v[i] = generateRandomString(30)
}
return &LargeMessage{Values: v}
}
package testmessage;
import "github.com/gogo/protobuf/gogoproto/gogo.proto";
option (gogoproto.gostring_all) = true;
option (gogoproto.equal_all) = true;
option (gogoproto.verbose_equal_all) = true;
option (gogoproto.goproto_stringer_all) = false;
option (gogoproto.stringer_all) = true;
option (gogoproto.populate_all) = true;
option (gogoproto.testgen_all) = false;
option (gogoproto.benchgen_all) = false;
option (gogoproto.marshaler_all) = true;
option (gogoproto.sizer_all) = true;
option (gogoproto.unmarshaler_all) = true;
message SmallMessage {
repeated string Values = 1;
}
message MediumMessage {
repeated string Values = 1;
}
message BigMessage {
repeated string Values = 1;
}
message LargeMessage {
repeated string Values = 1;
}
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 messenger
import (
"github.com/mesos/mesos-go/upid"
"golang.org/x/net/context"
)
// Transporter defines methods for communicating with remote processes.
type Transporter interface {
//Send sends message to remote process. Must use context to determine
//cancelled requests. Will stop sending when transport is stopped.
Send(ctx context.Context, msg *Message) error
//Rcvd receives and delegate message handling to installed handlers.
//Will stop receiving when transport is stopped.
Recv() (*Message, error)
//Inject injects a message to the incoming queue. Must use context to
//determine cancelled requests. Injection is aborted if the transport
//is stopped.
Inject(ctx context.Context, msg *Message) error
//Install mount an handler based on incoming message name.
Install(messageName string)
//Start starts the transporter and returns immediately. The error chan
//is never nil.
Start() <-chan error
//Stop kills the transporter.
Stop(graceful bool) error
//UPID returns the PID for transporter.
UPID() *upid.UPID
}
/*
Package scheduler includes the interfaces for the mesos scheduler and
the mesos executor driver. It also contains as well as an implementation
of the driver that you can use in your code.
*/
package scheduler
package scheduler
import (
"github.com/mesos/mesos-go/auth/callback"
mesos "github.com/mesos/mesos-go/mesosproto"
"github.com/mesos/mesos-go/upid"
)
type CredentialHandler struct {
pid *upid.UPID // the process to authenticate against (master)
client *upid.UPID // the process to be authenticated (slave / framework)
credential *mesos.Credential
}
func (h *CredentialHandler) Handle(callbacks ...callback.Interface) error {
for _, cb := range callbacks {
switch cb := cb.(type) {
case *callback.Name:
cb.Set(h.credential.GetPrincipal())
case *callback.Password:
cb.Set(h.credential.GetSecret())
case *callback.Interprocess:
cb.Set(*(h.pid), *(h.client))
default:
return &callback.Unsupported{Callback: cb}
}
}
return nil
}
package scheduler
import (
log "github.com/golang/glog"
mesos "github.com/mesos/mesos-go/mesosproto"
"github.com/stretchr/testify/mock"
)
type MockScheduler struct {
mock.Mock
}
func NewMockScheduler() *MockScheduler {
return &MockScheduler{}
}
func (sched *MockScheduler) Registered(SchedulerDriver, *mesos.FrameworkID, *mesos.MasterInfo) {
sched.Called()
}
func (sched *MockScheduler) Reregistered(SchedulerDriver, *mesos.MasterInfo) {
sched.Called()
}
func (sched *MockScheduler) Disconnected(SchedulerDriver) {
sched.Called()
}
func (sched *MockScheduler) ResourceOffers(SchedulerDriver, []*mesos.Offer) {
sched.Called()
}
func (sched *MockScheduler) OfferRescinded(SchedulerDriver, *mesos.OfferID) {
sched.Called()
}
func (sched *MockScheduler) StatusUpdate(SchedulerDriver, *mesos.TaskStatus) {
sched.Called()
}
func (sched *MockScheduler) FrameworkMessage(SchedulerDriver, *mesos.ExecutorID, *mesos.SlaveID, string) {
sched.Called()
}
func (sched *MockScheduler) SlaveLost(SchedulerDriver, *mesos.SlaveID) {
sched.Called()
}
func (sched *MockScheduler) ExecutorLost(SchedulerDriver, *mesos.ExecutorID, *mesos.SlaveID, int) {
sched.Called()
}
func (sched *MockScheduler) Error(d SchedulerDriver, msg string) {
log.Error(msg)
sched.Called()
}
package scheduler
import (
_ "github.com/mesos/mesos-go/auth/sasl"
_ "github.com/mesos/mesos-go/auth/sasl/mech/crammd5"
_ "github.com/mesos/mesos-go/detector/zoo"
)
package scheduler
import (
log "github.com/golang/glog"
mesos "github.com/mesos/mesos-go/mesosproto"
"github.com/mesos/mesos-go/upid"
"sync"
)
type cachedOffer struct {
offer *mesos.Offer
slavePid *upid.UPID
}
func newCachedOffer(offer *mesos.Offer, slavePid *upid.UPID) *cachedOffer {
return &cachedOffer{offer: offer, slavePid: slavePid}
}
// schedCache a managed cache with backing maps to store offeres
// and tasked slaves.
type schedCache struct {
lock sync.RWMutex
savedOffers map[string]*cachedOffer // current offers key:OfferID
savedSlavePids map[string]*upid.UPID // Current saved slaves, key:slaveId
}
func newSchedCache() *schedCache {
return &schedCache{
savedOffers: make(map[string]*cachedOffer),
savedSlavePids: make(map[string]*upid.UPID),
}
}
// putOffer stores an offer and the slavePID associated with offer.
func (cache *schedCache) putOffer(offer *mesos.Offer, pid *upid.UPID) {
if offer == nil || pid == nil {
log.V(3).Infoln("WARN: Offer not cached. The offer or pid cannot be nil")
return
}
log.V(3).Infoln("Caching offer ", offer.Id.GetValue(), " with slavePID ", pid.String())
cache.lock.Lock()
cache.savedOffers[offer.Id.GetValue()] = &cachedOffer{offer: offer, slavePid: pid}
cache.lock.Unlock()
}
// getOffer returns cached offer
func (cache *schedCache) getOffer(offerId *mesos.OfferID) *cachedOffer {
if offerId == nil {
log.V(3).Infoln("WARN: OfferId == nil, returning nil")
return nil
}
cache.lock.RLock()
defer cache.lock.RUnlock()
return cache.savedOffers[offerId.GetValue()]
}
// containsOff test cache for offer(offerId)
func (cache *schedCache) containsOffer(offerId *mesos.OfferID) bool {
cache.lock.RLock()
defer cache.lock.RUnlock()
_, ok := cache.savedOffers[offerId.GetValue()]
return ok
}
func (cache *schedCache) removeOffer(offerId *mesos.OfferID) {
cache.lock.Lock()
delete(cache.savedOffers, offerId.GetValue())
cache.lock.Unlock()
}
func (cache *schedCache) putSlavePid(slaveId *mesos.SlaveID, pid *upid.UPID) {
cache.lock.Lock()
cache.savedSlavePids[slaveId.GetValue()] = pid
cache.lock.Unlock()
}
func (cache *schedCache) getSlavePid(slaveId *mesos.SlaveID) *upid.UPID {
if slaveId == nil {
log.V(3).Infoln("SlaveId == nil, returning empty UPID")
return nil
}
return cache.savedSlavePids[slaveId.GetValue()]
}
func (cache *schedCache) containsSlavePid(slaveId *mesos.SlaveID) bool {
cache.lock.RLock()
defer cache.lock.RUnlock()
_, ok := cache.savedSlavePids[slaveId.GetValue()]
return ok
}
func (cache *schedCache) removeSlavePid(slaveId *mesos.SlaveID) {
cache.lock.Lock()
delete(cache.savedSlavePids, slaveId.GetValue())
cache.lock.Unlock()
}
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 scheduler
import (
mesos "github.com/mesos/mesos-go/mesosproto"
util "github.com/mesos/mesos-go/mesosutil"
"github.com/stretchr/testify/assert"
"testing"
"github.com/mesos/mesos-go/upid"
)
func TestSchedCacheNew(t *testing.T) {
cache := newSchedCache()
assert.NotNil(t, cache)
assert.NotNil(t, cache.savedOffers)
assert.NotNil(t, cache.savedSlavePids)
}
func TestSchedCachePutOffer(t *testing.T) {
cache := newSchedCache()
offer01 := createTestOffer("01")
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
cache.putOffer(offer01, pid01)
offer02 := createTestOffer("02")
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
cache.putOffer(offer02, pid02)
assert.Equal(t, len(cache.savedOffers), 2)
cachedOffer1, ok := cache.savedOffers["test-offer-01"]
assert.True(t, ok)
cachedOffer2, ok := cache.savedOffers["test-offer-02"]
assert.True(t, ok)
assert.NotNil(t, cachedOffer1.offer)
assert.Equal(t, "test-offer-01", cachedOffer1.offer.Id.GetValue())
assert.NotNil(t, cachedOffer2.offer)
assert.Equal(t, "test-offer-02", cachedOffer2.offer.Id.GetValue())
assert.NotNil(t, cachedOffer1.slavePid)
assert.Equal(t, "slave01@127.0.0.1:5050", cachedOffer1.slavePid.String())
assert.NotNil(t, cachedOffer2.slavePid)
assert.Equal(t, "slave02@127.0.0.1:5050", cachedOffer2.slavePid.String())
}
func TestSchedCacheGetOffer(t *testing.T) {
cache := newSchedCache()
offer01 := createTestOffer("01")
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
offer02 := createTestOffer("02")
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
cache.putOffer(offer01, pid01)
cache.putOffer(offer02, pid02)
cachedOffer01 := cache.getOffer(util.NewOfferID("test-offer-01")).offer
cachedOffer02 := cache.getOffer(util.NewOfferID("test-offer-02")).offer
assert.NotEqual(t, offer01, cachedOffer02)
assert.Equal(t, offer01, cachedOffer01)
assert.Equal(t, offer02, cachedOffer02)
}
func TestSchedCacheContainsOffer(t *testing.T) {
cache := newSchedCache()
offer01 := createTestOffer("01")
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
offer02 := createTestOffer("02")
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
cache.putOffer(offer01, pid01)
cache.putOffer(offer02, pid02)
assert.True(t, cache.containsOffer(util.NewOfferID("test-offer-01")))
assert.True(t, cache.containsOffer(util.NewOfferID("test-offer-02")))
assert.False(t, cache.containsOffer(util.NewOfferID("test-offer-05")))
}
func TestSchedCacheRemoveOffer(t *testing.T) {
cache := newSchedCache()
offer01 := createTestOffer("01")
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
offer02 := createTestOffer("02")
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
cache.putOffer(offer01, pid01)
cache.putOffer(offer02, pid02)
cache.removeOffer(util.NewOfferID("test-offer-01"))
assert.Equal(t, 1, len(cache.savedOffers))
assert.True(t, cache.containsOffer(util.NewOfferID("test-offer-02")))
assert.False(t, cache.containsOffer(util.NewOfferID("test-offer-01")))
}
func TestSchedCachePutSlavePid(t *testing.T) {
cache := newSchedCache()
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
pid03, err := upid.Parse("slave03@127.0.0.1:5050")
assert.NoError(t, err)
cache.putSlavePid(util.NewSlaveID("slave01"), pid01)
cache.putSlavePid(util.NewSlaveID("slave02"), pid02)
cache.putSlavePid(util.NewSlaveID("slave03"), pid03)
assert.Equal(t, len(cache.savedSlavePids), 3)
cachedSlavePid1, ok := cache.savedSlavePids["slave01"]
assert.True(t, ok)
cachedSlavePid2, ok := cache.savedSlavePids["slave02"]
assert.True(t, ok)
cachedSlavePid3, ok := cache.savedSlavePids["slave03"]
assert.True(t, ok)
assert.True(t, cachedSlavePid1.Equal(pid01))
assert.True(t, cachedSlavePid2.Equal(pid02))
assert.True(t, cachedSlavePid3.Equal(pid03))
}
func TestSchedCacheGetSlavePid(t *testing.T) {
cache := newSchedCache()
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
cache.putSlavePid(util.NewSlaveID("slave01"), pid01)
cache.putSlavePid(util.NewSlaveID("slave02"), pid02)
cachedSlavePid1 := cache.getSlavePid(util.NewSlaveID("slave01"))
cachedSlavePid2 := cache.getSlavePid(util.NewSlaveID("slave02"))
assert.NotNil(t, cachedSlavePid1)
assert.NotNil(t, cachedSlavePid2)
assert.True(t, pid01.Equal(cachedSlavePid1))
assert.True(t, pid02.Equal(cachedSlavePid2))
assert.False(t, pid01.Equal(cachedSlavePid2))
}
func TestSchedCacheContainsSlavePid(t *testing.T) {
cache := newSchedCache()
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
cache.putSlavePid(util.NewSlaveID("slave01"), pid01)
cache.putSlavePid(util.NewSlaveID("slave02"), pid02)
assert.True(t, cache.containsSlavePid(util.NewSlaveID("slave01")))
assert.True(t, cache.containsSlavePid(util.NewSlaveID("slave02")))
assert.False(t, cache.containsSlavePid(util.NewSlaveID("slave05")))
}
func TestSchedCacheRemoveSlavePid(t *testing.T) {
cache := newSchedCache()
pid01, err := upid.Parse("slave01@127.0.0.1:5050")
assert.NoError(t, err)
pid02, err := upid.Parse("slave02@127.0.0.1:5050")
assert.NoError(t, err)
cache.putSlavePid(util.NewSlaveID("slave01"), pid01)
cache.putSlavePid(util.NewSlaveID("slave02"), pid02)
assert.True(t, cache.containsSlavePid(util.NewSlaveID("slave01")))
assert.True(t, cache.containsSlavePid(util.NewSlaveID("slave02")))
assert.False(t, cache.containsSlavePid(util.NewSlaveID("slave05")))
cache.removeSlavePid(util.NewSlaveID("slave01"))
assert.Equal(t, 1, len(cache.savedSlavePids))
assert.False(t, cache.containsSlavePid(util.NewSlaveID("slave01")))
}
func createTestOffer(idSuffix string) *mesos.Offer {
return util.NewOffer(
util.NewOfferID("test-offer-"+idSuffix),
util.NewFrameworkID("test-framework-"+idSuffix),
util.NewSlaveID("test-slave-"+idSuffix),
"localhost."+idSuffix,
)
}
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you 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 scheduler
import (
mesos "github.com/mesos/mesos-go/mesosproto"
)
// Interface for connecting a scheduler to Mesos. This
// interface is used both to manage the scheduler's lifecycle (start
// it, stop it, or wait for it to finish) and to interact with Mesos
// (e.g., launch tasks, kill tasks, etc.).
// See the MesosSchedulerDriver type for a concrete
// impl of a SchedulerDriver.
type SchedulerDriver interface {
// Starts the scheduler driver. This needs to be called before any
// other driver calls are made.
Start() (mesos.Status, error)
// Stops the scheduler driver. If the 'failover' flag is set to
// false then it is expected that this framework will never
// reconnect to Mesos and all of its executors and tasks can be
// terminated. Otherwise, all executors and tasks will remain
// running (for some framework specific failover timeout) allowing the
// scheduler to reconnect (possibly in the same process, or from a
// different process, for example, on a different machine).
Stop(failover bool) (mesos.Status, error)
// Aborts the driver so that no more callbacks can be made to the
// scheduler. The semantics of abort and stop have deliberately been
// separated so that code can detect an aborted driver (i.e., via
// the return status of SchedulerDriver::join, see below), and
// instantiate and start another driver if desired (from within the
// same process). Note that 'Stop()' is not automatically called
// inside 'Abort()'.
Abort() (mesos.Status, error)
// Waits for the driver to be stopped or aborted, possibly
// _blocking_ the current thread indefinitely. The return status of
// this function can be used to determine if the driver was aborted
// (see mesos.proto for a description of Status).
Join() (mesos.Status, error)
// Starts and immediately joins (i.e., blocks on) the driver.
Run() (mesos.Status, error)
// Requests resources from Mesos (see mesos.proto for a description
// of Request and how, for example, to request resources
// from specific slaves). Any resources available are offered to the
// framework via Scheduler.ResourceOffers callback, asynchronously.
RequestResources(requests []*mesos.Request) (mesos.Status, error)
// Launches the given set of tasks. Any resources remaining (i.e.,
// not used by the tasks or their executors) will be considered
// declined. The specified filters are applied on all unused
// resources (see mesos.proto for a description of Filters).
// Available resources are aggregated when mutiple offers are
// provided. Note that all offers must belong to the same slave.
// Invoking this function with an empty collection of tasks declines
// offers in their entirety (see Scheduler::declineOffer).
LaunchTasks(offerIDs []*mesos.OfferID, tasks []*mesos.TaskInfo, filters *mesos.Filters) (mesos.Status, error)
// Kills the specified task. Note that attempting to kill a task is
// currently not reliable. If, for example, a scheduler fails over
// while it was attempting to kill a task it will need to retry in
// the future. Likewise, if unregistered / disconnected, the request
// will be dropped (these semantics may be changed in the future).
KillTask(taskID *mesos.TaskID) (mesos.Status, error)
// Declines an offer in its entirety and applies the specified
// filters on the resources (see mesos.proto for a description of
// Filters). Note that this can be done at any time, it is not
// necessary to do this within the Scheduler::resourceOffers
// callback.
DeclineOffer(offerID *mesos.OfferID, filters *mesos.Filters) (mesos.Status, error)
// Removes all filters previously set by the framework (via
// LaunchTasks()). This enables the framework to receive offers from
// those filtered slaves.
ReviveOffers() (mesos.Status, error)
// Sends a message from the framework to one of its executors. These
// messages are best effort; do not expect a framework message to be
// retransmitted in any reliable fashion.
SendFrameworkMessage(executorID *mesos.ExecutorID, slaveID *mesos.SlaveID, data string) (mesos.Status, error)
// Allows the framework to query the status for non-terminal tasks.
// This causes the master to send back the latest task status for
// each task in 'statuses', if possible. Tasks that are no longer
// known will result in a TASK_LOST update. If statuses is empty,
// then the master will send the latest status for each task
// currently known.
ReconcileTasks(statuses []*mesos.TaskStatus) (mesos.Status, error)
}
// Scheduler a type with callback attributes to be provided by frameworks
// schedulers.
//
// Each callback includes a reference to the scheduler driver that was
// used to run this scheduler. The pointer will not change for the
// duration of a scheduler (i.e., from the point you do
// SchedulerDriver.Start() to the point that SchedulerDriver.Stop()
// returns). This is intended for convenience so that a scheduler
// doesn't need to store a reference to the driver itself.
type Scheduler interface {
// Invoked when the scheduler successfully registers with a Mesos
// master. A unique ID (generated by the master) used for
// distinguishing this framework from others and MasterInfo
// with the ip and port of the current master are provided as arguments.
Registered(SchedulerDriver, *mesos.FrameworkID, *mesos.MasterInfo)
// Invoked when the scheduler re-registers with a newly elected Mesos master.
// This is only called when the scheduler has previously been registered.
// MasterInfo containing the updated information about the elected master
// is provided as an argument.
Reregistered(SchedulerDriver, *mesos.MasterInfo)
// Invoked when the scheduler becomes "disconnected" from the master
// (e.g., the master fails and another is taking over).
Disconnected(SchedulerDriver)
// Invoked when resources have been offered to this framework. A
// single offer will only contain resources from a single slave.
// Resources associated with an offer will not be re-offered to
// _this_ framework until either (a) this framework has rejected
// those resources (see SchedulerDriver::launchTasks) or (b) those
// resources have been rescinded (see Scheduler::offerRescinded).
// Note that resources may be concurrently offered to more than one
// framework at a time (depending on the allocator being used). In
// that case, the first framework to launch tasks using those
// resources will be able to use them while the other frameworks
// will have those resources rescinded (or if a framework has
// already launched tasks with those resources then those tasks will
// fail with a TASK_LOST status and a message saying as much).
ResourceOffers(SchedulerDriver, []*mesos.Offer)
// Invoked when an offer is no longer valid (e.g., the slave was
// lost or another framework used resources in the offer). If for
// whatever reason an offer is never rescinded (e.g., dropped
// message, failing over framework, etc.), a framwork that attempts
// to launch tasks using an invalid offer will receive TASK_LOST
// status updates for those tasks (see Scheduler::resourceOffers).
OfferRescinded(SchedulerDriver, *mesos.OfferID)
// Invoked when the status of a task has changed (e.g., a slave is
// lost and so the task is lost, a task finishes and an executor
// sends a status update saying so, etc). Note that returning from
// this callback _acknowledges_ receipt of this status update! If
// for whatever reason the scheduler aborts during this callback (or
// the process exits) another status update will be delivered (note,
// however, that this is currently not true if the slave sending the
// status update is lost/fails during that time).
StatusUpdate(SchedulerDriver, *mesos.TaskStatus)
// Invoked when an executor sends a message. These messages are best
// effort; do not expect a framework message to be retransmitted in
// any reliable fashion.
FrameworkMessage(SchedulerDriver, *mesos.ExecutorID, *mesos.SlaveID, string)
// Invoked when a slave has been determined unreachable (e.g.,
// machine failure, network partition). Most frameworks will need to
// reschedule any tasks launched on this slave on a new slave.
SlaveLost(SchedulerDriver, *mesos.SlaveID)
// Invoked when an executor has exited/terminated. Note that any
// tasks running will have TASK_LOST status updates automagically
// generated.
ExecutorLost(SchedulerDriver, *mesos.ExecutorID, *mesos.SlaveID, int)
// Invoked when there is an unrecoverable error in the scheduler or
// scheduler driver. The driver will be aborted BEFORE invoking this
// callback.
Error(SchedulerDriver, string)
}
...@@ -304,10 +304,9 @@ function kube::build::ensure_golang() { ...@@ -304,10 +304,9 @@ function kube::build::ensure_golang() {
} }
} }
# Set up the context directory for the kube-build image and build it. # The set of source targets to include in the kube-build image
function kube::build::build_image() { function kube::build::source_targets() {
local -r build_context_dir="${LOCAL_OUTPUT_IMAGE_STAGING}/${KUBE_BUILD_IMAGE}" local targets=(
local -r source=(
api api
build build
cmd cmd
...@@ -323,11 +322,22 @@ function kube::build::build_image() { ...@@ -323,11 +322,22 @@ function kube::build::build_image() {
test test
third_party third_party
) )
if [ -n "${KUBERNETES_CONTRIB:-}" ]; then
for contrib in "${KUBERNETES_CONTRIB}"; do
targets+=($(eval "kube::contrib::${contrib}::source_targets"))
done
fi
echo "${targets[@]}"
}
# Set up the context directory for the kube-build image and build it.
function kube::build::build_image() {
local -r build_context_dir="${LOCAL_OUTPUT_IMAGE_STAGING}/${KUBE_BUILD_IMAGE}"
kube::build::build_image_cross kube::build::build_image_cross
mkdir -p "${build_context_dir}" mkdir -p "${build_context_dir}"
tar czf "${build_context_dir}/kube-source.tar.gz" "${source[@]}" tar czf "${build_context_dir}/kube-source.tar.gz" $(kube::build::source_targets)
kube::version::get_version_vars kube::version::get_version_vars
kube::version::save_version_vars "${build_context_dir}/kube-version-defs" kube::version::save_version_vars "${build_context_dir}/kube-version-defs"
...@@ -412,8 +422,12 @@ function kube::build::run_build_command() { ...@@ -412,8 +422,12 @@ function kube::build::run_build_command() {
local -a docker_run_opts=( local -a docker_run_opts=(
"--name=${KUBE_BUILD_CONTAINER_NAME}" "--name=${KUBE_BUILD_CONTAINER_NAME}"
"${DOCKER_MOUNT_ARGS[@]}" "${DOCKER_MOUNT_ARGS[@]}"
) )
if [ -n "${KUBERNETES_CONTRIB:-}" ]; then
docker_run_opts+=(-e "KUBERNETES_CONTRIB=${KUBERNETES_CONTRIB}")
fi
# If we have stdin we can run interactive. This allows things like 'shell.sh' # If we have stdin we can run interactive. This allows things like 'shell.sh'
# to work. However, if we run this way and don't have stdin, then it ends up # to work. However, if we run this way and don't have stdin, then it ends up
......
...@@ -17,6 +17,9 @@ limitations under the License. ...@@ -17,6 +17,9 @@ limitations under the License.
// Package app makes it easy to create a kubelet server for various contexts. // Package app makes it easy to create a kubelet server for various contexts.
package app package app
// Note: if you change code in this file, you might need to change code in
// contrib/mesos/pkg/executor/service/.
import ( import (
"crypto/tls" "crypto/tls"
"fmt" "fmt"
......
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// This package main implements the executable Kubernetes Mesos executor.
package main
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package main
import (
"fmt"
"os"
"runtime"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/executor/service"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/hyperkube"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/version/verflag"
"github.com/spf13/pflag"
)
func main() {
runtime.GOMAXPROCS(runtime.NumCPU())
s := service.NewKubeletExecutorServer()
s.AddStandaloneFlags(pflag.CommandLine)
util.InitFlags()
util.InitLogs()
defer util.FlushLogs()
verflag.PrintAndExitIfRequested()
if err := s.Run(hyperkube.Nil(), pflag.CommandLine.Args()); err != nil {
fmt.Fprintf(os.Stderr, err.Error())
os.Exit(1)
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// This package main is used for testing the redirfd package.
// Inspired by http://skarnet.org/software/execline/redirfd.html.
// Usage:
// k8sm-redirfb [-n] [-b] {mode} {fd} {file} {prog...}
package main
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package main
import (
"flag"
"fmt"
"os"
"os/exec"
"syscall"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/redirfd"
)
func main() {
nonblock := flag.Bool("n", false, "open file in non-blocking mode")
changemode := flag.Bool("b", false, "change mode of file after opening it: to non-blocking mode if the -n option was not given, to blocking mode if it was")
flag.Parse()
args := flag.Args()
if len(args) < 4 {
fmt.Fprintf(os.Stderr, "expected {mode} {fd} {file} instead of: %v\n", args)
os.Exit(1)
}
var mode redirfd.RedirectMode
switch m := args[0]; m {
case "r":
mode = redirfd.Read
case "w":
mode = redirfd.Write
case "u":
mode = redirfd.Update
case "a":
mode = redirfd.Append
case "c":
mode = redirfd.AppendExisting
case "x":
mode = redirfd.WriteNew
default:
fmt.Fprintf(os.Stderr, "unrecognized mode %q\n", mode)
os.Exit(1)
}
fd, err := redirfd.ParseFileDescriptor(args[1])
if err != nil {
fmt.Fprintf(os.Stderr, "failed to parse file descriptor: %v\n", err)
os.Exit(1)
}
file := args[2]
f, err := mode.Redirect(*nonblock, *changemode, fd, file)
if err != nil {
fmt.Fprintf(os.Stderr, "redirect failed: %q, %v\n", args[1], err)
os.Exit(1)
}
var pargs []string
if len(args) > 4 {
pargs = args[4:]
}
cmd := exec.Command(args[3], pargs...)
cmd.Stdin = os.Stdin
cmd.Stdout = os.Stdout
cmd.Stderr = os.Stderr
switch fd {
case redirfd.Stdin:
cmd.Stdin = f
case redirfd.Stdout:
cmd.Stdout = f
case redirfd.Stderr:
cmd.Stderr = f
default:
cmd.ExtraFiles = []*os.File{f}
}
defer f.Close()
if err = cmd.Run(); err != nil {
exiterr := err.(*exec.ExitError)
state := exiterr.ProcessState
if state != nil {
sys := state.Sys()
if waitStatus, ok := sys.(syscall.WaitStatus); ok {
if waitStatus.Signaled() {
os.Exit(256 + int(waitStatus.Signal()))
} else {
os.Exit(waitStatus.ExitStatus())
}
}
}
os.Exit(3)
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// This package main implements the executable Kubernetes Mesos scheduler.
package main
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package main
import (
"fmt"
"os"
"runtime"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/hyperkube"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/scheduler/service"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/version/verflag"
"github.com/spf13/pflag"
)
func main() {
runtime.GOMAXPROCS(runtime.NumCPU())
s := service.NewSchedulerServer()
s.AddStandaloneFlags(pflag.CommandLine)
util.InitFlags()
util.InitLogs()
defer util.FlushLogs()
verflag.PrintAndExitIfRequested()
if err := s.Run(hyperkube.Nil(), pflag.CommandLine.Args()); err != nil {
fmt.Fprintf(os.Stderr, err.Error())
os.Exit(1)
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package assert
import (
"testing"
"time"
"github.com/stretchr/testify/assert"
)
// EventuallyTrue asserts that the given predicate becomes true within the given timeout. It
// checks the predicate regularly each 100ms.
func EventuallyTrue(t *testing.T, timeout time.Duration, fn func() bool, msgAndArgs ...interface{}) bool {
start := time.Now()
for {
if fn() {
return true
}
if time.Now().Sub(start) > timeout {
if len(msgAndArgs) > 0 {
return assert.Fail(t, msgAndArgs[0].(string), msgAndArgs[1:]...)
} else {
return assert.Fail(t, "predicate fn has not been true after %v", timeout.String())
}
}
time.Sleep(100 * time.Millisecond)
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package assert is an utility package containing reusable testing functionality
// extending github.com/stretchr/testify/assert
package assert
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package backoff
import (
"math/rand"
"sync"
"time"
log "github.com/golang/glog"
)
type clock interface {
Now() time.Time
}
type realClock struct{}
func (realClock) Now() time.Time {
return time.Now()
}
type backoffEntry struct {
backoff time.Duration
lastUpdate time.Time
}
type Backoff struct {
perItemBackoff map[string]*backoffEntry
lock sync.Mutex
clock clock
defaultDuration time.Duration
maxDuration time.Duration
}
func New(initial, max time.Duration) *Backoff {
return &Backoff{
perItemBackoff: map[string]*backoffEntry{},
clock: realClock{},
defaultDuration: initial,
maxDuration: max,
}
}
func (p *Backoff) getEntry(id string) *backoffEntry {
p.lock.Lock()
defer p.lock.Unlock()
entry, ok := p.perItemBackoff[id]
if !ok {
entry = &backoffEntry{backoff: p.defaultDuration}
p.perItemBackoff[id] = entry
}
entry.lastUpdate = p.clock.Now()
return entry
}
func (p *Backoff) Get(id string) time.Duration {
entry := p.getEntry(id)
duration := entry.backoff
entry.backoff *= 2
if entry.backoff > p.maxDuration {
entry.backoff = p.maxDuration
}
//TODO(jdef) parameterize use of jitter?
// add jitter, get better backoff distribution
duration = time.Duration(rand.Int63n(int64(duration)))
log.V(3).Infof("Backing off %v for pod %s", duration, id)
return duration
}
// Garbage collect records that have aged past maxDuration. Backoff users are expected
// to invoke this periodically.
func (p *Backoff) GC() {
p.lock.Lock()
defer p.lock.Unlock()
now := p.clock.Now()
for id, entry := range p.perItemBackoff {
if now.Sub(entry.lastUpdate) > p.maxDuration {
delete(p.perItemBackoff, id)
}
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package backoff provides backoff functionality with a simple API.
// Originally copied from Kubernetes: plugin/pkg/scheduler/factory/factory.go
package backoff
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package election provides interfaces used for master election.
package election
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package election
import (
"fmt"
"time"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/coreos/go-etcd/etcd"
"github.com/golang/glog"
)
// Master is used to announce the current elected master.
type Master string
// IsAnAPIObject is used solely so we can work with the watch package.
// TODO(k8s): Either fix watch so this isn't necessary, or make this a real API Object.
// TODO(k8s): when it becomes clear how this package will be used, move these declarations to
// to the proper place.
func (Master) IsAnAPIObject() {}
// NewEtcdMasterElector returns an implementation of election.MasterElector backed by etcd.
func NewEtcdMasterElector(h tools.EtcdGetSet) MasterElector {
return &etcdMasterElector{etcd: h}
}
type empty struct{}
// internal implementation struct
type etcdMasterElector struct {
etcd tools.EtcdGetSet
done chan empty
events chan watch.Event
}
// Elect implements the election.MasterElector interface.
func (e *etcdMasterElector) Elect(path, id string) watch.Interface {
e.done = make(chan empty)
e.events = make(chan watch.Event)
go util.Forever(func() { e.run(path, id) }, time.Second*5)
return e
}
func (e *etcdMasterElector) run(path, id string) {
masters := make(chan string)
errors := make(chan error)
go e.master(path, id, 30, masters, errors, e.done) // TODO(jdef) extract constant
for {
select {
case m := <-masters:
e.events <- watch.Event{
Type: watch.Modified,
Object: Master(m),
}
case e := <-errors:
glog.Errorf("error in election: %v", e)
}
}
}
// ResultChan implements the watch.Interface interface.
func (e *etcdMasterElector) ResultChan() <-chan watch.Event {
return e.events
}
// extendMaster attempts to extend ownership of a master lock for TTL seconds.
// returns "", nil if extension failed
// returns id, nil if extension succeeded
// returns "", err if an error occurred
func (e *etcdMasterElector) extendMaster(path, id string, ttl uint64, res *etcd.Response) (string, error) {
// If it matches the passed in id, extend the lease by writing a new entry.
// Uses compare and swap, so that if we TTL out in the meantime, the write will fail.
// We don't handle the TTL delete w/o a write case here, it's handled in the next loop
// iteration.
_, err := e.etcd.CompareAndSwap(path, id, ttl, "", res.Node.ModifiedIndex)
if err != nil && !tools.IsEtcdTestFailed(err) {
return "", err
}
if err != nil && tools.IsEtcdTestFailed(err) {
return "", nil
}
return id, nil
}
// becomeMaster attempts to become the master for this lock.
// returns "", nil if the attempt failed
// returns id, nil if the attempt succeeded
// returns "", err if an error occurred
func (e *etcdMasterElector) becomeMaster(path, id string, ttl uint64) (string, error) {
_, err := e.etcd.Create(path, id, ttl)
if err != nil && !tools.IsEtcdNodeExist(err) {
// unexpected error
return "", err
}
if err != nil && tools.IsEtcdNodeExist(err) {
return "", nil
}
return id, nil
}
// handleMaster performs one loop of master locking.
// on success it returns <master>, nil
// on error it returns "", err
// in situations where you should try again due to concurrent state changes (e.g. another actor simultaneously acquiring the lock)
// it returns "", nil
func (e *etcdMasterElector) handleMaster(path, id string, ttl uint64) (string, error) {
res, err := e.etcd.Get(path, false, false)
// Unexpected error, bail out
if err != nil && !tools.IsEtcdNotFound(err) {
return "", err
}
// There is no master, try to become the master.
if err != nil && tools.IsEtcdNotFound(err) {
return e.becomeMaster(path, id, ttl)
}
// This should never happen.
if res.Node == nil {
return "", fmt.Errorf("unexpected response: %#v", res)
}
// We're not the master, just return the current value
if res.Node.Value != id {
return res.Node.Value, nil
}
// We are the master, try to extend out lease
return e.extendMaster(path, id, ttl, res)
}
// master provices a distributed master election lock, maintains lock until failure, or someone sends something in the done channel.
// The basic algorithm is:
// while !done
// Get the current master
// If there is no current master
// Try to become the master
// Otherwise
// If we are the master, extend the lease
// If the master is different than the last time through the loop, report the master
// Sleep 80% of TTL
func (e *etcdMasterElector) master(path, id string, ttl uint64, masters chan<- string, errors chan<- error, done <-chan empty) {
lastMaster := ""
for {
master, err := e.handleMaster(path, id, ttl)
if err != nil {
errors <- err
} else if len(master) == 0 {
continue
} else if master != lastMaster {
lastMaster = master
masters <- master
}
// TODO(k8s): Add Watch here, skip the polling for faster reactions
// If done is closed, break out.
select {
case <-done:
return
case <-time.After(time.Duration((ttl*8)/10) * time.Second):
}
}
}
// ResultChan implements the watch.Interface interface
func (e *etcdMasterElector) Stop() {
close(e.done)
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package election
import (
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/coreos/go-etcd/etcd"
)
func TestEtcdMasterOther(t *testing.T) {
path := "foo"
etcd := tools.NewFakeEtcdClient(t)
etcd.Set(path, "baz", 0)
master := NewEtcdMasterElector(etcd)
w := master.Elect(path, "bar")
result := <-w.ResultChan()
if result.Type != watch.Modified || result.Object.(Master) != "baz" {
t.Errorf("unexpected event: %#v", result)
}
w.Stop()
}
func TestEtcdMasterNoOther(t *testing.T) {
path := "foo"
e := tools.NewFakeEtcdClient(t)
e.TestIndex = true
e.Data["foo"] = tools.EtcdResponseWithError{
R: &etcd.Response{
Node: nil,
},
E: &etcd.EtcdError{
ErrorCode: tools.EtcdErrorCodeNotFound,
},
}
master := NewEtcdMasterElector(e)
w := master.Elect(path, "bar")
result := <-w.ResultChan()
if result.Type != watch.Modified || result.Object.(Master) != "bar" {
t.Errorf("unexpected event: %#v", result)
}
w.Stop()
}
func TestEtcdMasterNoOtherThenConflict(t *testing.T) {
path := "foo"
e := tools.NewFakeEtcdClient(t)
e.TestIndex = true
// Ok, so we set up a chain of responses from etcd:
// 1) Nothing there
// 2) conflict (someone else wrote)
// 3) new value (the data they wrote)
empty := tools.EtcdResponseWithError{
R: &etcd.Response{
Node: nil,
},
E: &etcd.EtcdError{
ErrorCode: tools.EtcdErrorCodeNotFound,
},
}
empty.N = &tools.EtcdResponseWithError{
R: &etcd.Response{},
E: &etcd.EtcdError{
ErrorCode: tools.EtcdErrorCodeNodeExist,
},
}
empty.N.N = &tools.EtcdResponseWithError{
R: &etcd.Response{
Node: &etcd.Node{
Value: "baz",
},
},
}
e.Data["foo"] = empty
master := NewEtcdMasterElector(e)
w := master.Elect(path, "bar")
result := <-w.ResultChan()
if result.Type != watch.Modified || result.Object.(Master) != "bar" {
t.Errorf("unexpected event: %#v", result)
}
w.Stop()
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package election
import (
"sync"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
// Fake allows for testing of anything consuming a MasterElector.
type Fake struct {
mux *watch.Broadcaster
currentMaster Master
lock sync.Mutex // Protect access of currentMaster
}
// NewFake makes a new fake MasterElector.
func NewFake() *Fake {
// 0 means block for clients.
return &Fake{mux: watch.NewBroadcaster(0, watch.WaitIfChannelFull)}
}
func (f *Fake) ChangeMaster(newMaster Master) {
f.lock.Lock()
defer f.lock.Unlock()
f.mux.Action(watch.Modified, newMaster)
f.currentMaster = newMaster
}
func (f *Fake) Elect(path, id string) watch.Interface {
f.lock.Lock()
defer f.lock.Unlock()
w := f.mux.Watch()
if f.currentMaster != "" {
f.mux.Action(watch.Modified, f.currentMaster)
}
return w
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package election
import (
"sync"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/runtime"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/golang/glog"
)
// MasterElector is an interface for services that can elect masters.
// Important Note: MasterElectors are not inter-operable, all participants in the election need to be
// using the same underlying implementation of this interface for correct behavior.
type MasterElector interface {
// Elect makes the caller represented by 'id' enter into a master election for the
// distributed lock defined by 'path'
// The returned watch.Interface provides a stream of Master objects which
// contain the current master.
// Calling Stop on the returned interface relinquishes ownership (if currently possesed)
// and removes the caller from the election
Elect(path, id string) watch.Interface
}
// Service represents anything that can start and stop on demand.
type Service interface {
Validate(desired, current Master)
Start()
Stop()
}
type notifier struct {
lock sync.Mutex
cond *sync.Cond
// desired is updated with every change, current is updated after
// Start()/Stop() finishes. 'cond' is used to signal that a change
// might be needed. This handles the case where mastership flops
// around without calling Start()/Stop() excessively.
desired, current Master
// for comparison, to see if we are master.
id Master
service Service
}
// Notify runs Elect() on m, and calls Start()/Stop() on s when the
// elected master starts/stops matching 'id'. Never returns.
func Notify(m MasterElector, path, id string, s Service, abort <-chan struct{}) {
n := &notifier{id: Master(id), service: s}
n.cond = sync.NewCond(&n.lock)
finished := runtime.After(func() {
runtime.Until(func() {
for {
w := m.Elect(path, id)
for {
select {
case <-abort:
return
case event, open := <-w.ResultChan():
if !open {
break
}
if event.Type != watch.Modified {
continue
}
electedMaster, ok := event.Object.(Master)
if !ok {
glog.Errorf("Unexpected object from election channel: %v", event.Object)
break
}
func() {
n.lock.Lock()
defer n.lock.Unlock()
n.desired = electedMaster
if n.desired != n.current {
n.cond.Signal()
}
}()
}
}
}
}, 0, abort)
})
runtime.Until(func() { n.serviceLoop(finished) }, 0, abort)
}
// serviceLoop waits for changes, and calls Start()/Stop() as needed.
func (n *notifier) serviceLoop(abort <-chan struct{}) {
n.lock.Lock()
defer n.lock.Unlock()
for {
select {
case <-abort:
return
default:
for n.desired == n.current {
ch := runtime.After(n.cond.Wait)
select {
case <-abort:
n.cond.Signal() // ensure that Wait() returns
<-ch
return
case <-ch:
// we were notified and have the lock, proceed..
}
}
if n.current != n.id && n.desired == n.id {
n.service.Validate(n.desired, n.current)
n.service.Start()
} else if n.current == n.id && n.desired != n.id {
n.service.Stop()
}
n.current = n.desired
}
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package election
import (
"testing"
"time"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/runtime"
)
type slowService struct {
t *testing.T
on bool
// We explicitly have no lock to prove that
// Start and Stop are not called concurrently.
changes chan<- bool
done <-chan struct{}
}
func (s *slowService) Validate(d, c Master) {
// noop
}
func (s *slowService) Start() {
select {
case <-s.done:
return // avoid writing to closed changes chan
default:
}
if s.on {
s.t.Errorf("started already on service")
}
time.Sleep(2 * time.Millisecond)
s.on = true
s.changes <- true
}
func (s *slowService) Stop() {
select {
case <-s.done:
return // avoid writing to closed changes chan
default:
}
if !s.on {
s.t.Errorf("stopped already off service")
}
time.Sleep(2 * time.Millisecond)
s.on = false
s.changes <- false
}
func Test(t *testing.T) {
m := NewFake()
changes := make(chan bool, 1500)
done := make(chan struct{})
s := &slowService{t: t, changes: changes, done: done}
notifyDone := runtime.After(func() { Notify(m, "", "me", s, done) })
go func() {
defer close(done)
for i := 0; i < 500; i++ {
for _, key := range []string{"me", "notme", "alsonotme"} {
m.ChangeMaster(Master(key))
}
}
}()
<-notifyDone
close(changes)
changeList := []bool{}
for {
change, ok := <-changes
if !ok {
break
}
changeList = append(changeList, change)
}
if len(changeList) > 1000 {
t.Errorf("unexpected number of changes: %v", len(changeList))
}
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package config
import (
"time"
)
// default values to use when constructing mesos ExecutorInfo messages
const (
DefaultInfoID = "k8sm-executor"
DefaultInfoSource = "kubernetes"
DefaultInfoName = "Kubelet-Executor"
DefaultSuicideTimeout = 20 * time.Minute
)
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package config contains executor configuration constants.
package config
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
/*
Package executor includes a mesos executor, which contains
a kubelet as its member to manage containers.
*/
package executor
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package messages exposes executor event/message names as constants.
package messages
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package messages
// messages that ship with TaskStatus objects
const (
ContainersDisappeared = "containers-disappeared"
CreateBindingFailure = "create-binding-failure"
CreateBindingSuccess = "create-binding-success"
ExecutorUnregistered = "executor-unregistered"
ExecutorShutdown = "executor-shutdown"
LaunchTaskFailed = "launch-task-failed"
TaskKilled = "task-killed"
UnmarshalTaskDataFailure = "unmarshal-task-data-failure"
TaskLostAck = "task-lost-ack" // executor acknowledgement of forwarded TASK_LOST framework message
Kamikaze = "kamikaze"
)
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package executor
import (
"testing"
"github.com/GoogleCloudPlatform/kubernetes/pkg/kubelet/dockertools"
"github.com/mesos/mesos-go/mesosproto"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/mock"
)
type MockExecutorDriver struct {
mock.Mock
}
func (m *MockExecutorDriver) Start() (mesosproto.Status, error) {
args := m.Called()
return args.Get(0).(mesosproto.Status), args.Error(1)
}
func (m *MockExecutorDriver) Stop() (mesosproto.Status, error) {
args := m.Called()
return args.Get(0).(mesosproto.Status), args.Error(1)
}
func (m *MockExecutorDriver) Abort() (mesosproto.Status, error) {
args := m.Called()
return args.Get(0).(mesosproto.Status), args.Error(1)
}
func (m *MockExecutorDriver) Join() (mesosproto.Status, error) {
args := m.Called()
return args.Get(0).(mesosproto.Status), args.Error(1)
}
func (m *MockExecutorDriver) Run() (mesosproto.Status, error) {
args := m.Called()
return args.Get(0).(mesosproto.Status), args.Error(1)
}
func (m *MockExecutorDriver) SendStatusUpdate(taskStatus *mesosproto.TaskStatus) (mesosproto.Status, error) {
args := m.Called(*taskStatus.State)
return args.Get(0).(mesosproto.Status), args.Error(1)
}
func (m *MockExecutorDriver) SendFrameworkMessage(msg string) (mesosproto.Status, error) {
args := m.Called(msg)
return args.Get(0).(mesosproto.Status), args.Error(1)
}
func NewTestKubernetesExecutor() *KubernetesExecutor {
return New(Config{
Docker: dockertools.ConnectToDockerOrDie("fake://"),
Updates: make(chan interface{}, 1024),
})
}
func TestExecutorNew(t *testing.T) {
mockDriver := &MockExecutorDriver{}
executor := NewTestKubernetesExecutor()
executor.Init(mockDriver)
assert.Equal(t, executor.isDone(), false, "executor should not be in Done state on initialization")
assert.Equal(t, executor.isConnected(), false, "executor should not be connected on initialization")
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package service contains the cmd/k8sm-executor glue code.
package service
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package hyperkube facilitates the combination of multiple
// kubernetes-mesos components into a single binary form, providing a
// simple mechanism for intra-component discovery as per the original
// Kubernetes hyperkube package.
package hyperkube
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package hyperkube
import (
"github.com/spf13/pflag"
)
var (
nilKube = &nilKubeType{}
)
type Interface interface {
// FindServer will find a specific server named name.
FindServer(name string) bool
// The executable name, used for help and soft-link invocation
Name() string
// Flags returns a flagset for "global" flags.
Flags() *pflag.FlagSet
}
type nilKubeType struct{}
func (n *nilKubeType) FindServer(_ string) bool {
return false
}
func (n *nilKubeType) Name() string {
return ""
}
func (n *nilKubeType) Flags() *pflag.FlagSet {
return nil
}
func Nil() Interface {
return nilKube
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package offers contains code that manages Mesos offers.
package offers
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package metrics defines and exposes instrumentation metrics related to
// Mesos offers.
package metrics
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package metrics
import (
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
)
const (
offerSubsystem = "mesos_offers"
)
type OfferDeclinedReason string
const (
OfferExpired = OfferDeclinedReason("expired")
OfferRescinded = OfferDeclinedReason("rescinded")
OfferCompat = OfferDeclinedReason("compat")
)
var (
OffersReceived = prometheus.NewCounterVec(
prometheus.CounterOpts{
Subsystem: offerSubsystem,
Name: "received",
Help: "Counter of offers received from Mesos broken out by slave host.",
},
[]string{"hostname"},
)
OffersDeclined = prometheus.NewCounterVec(
prometheus.CounterOpts{
Subsystem: offerSubsystem,
Name: "declined",
Help: "Counter of offers declined by the framework broken out by slave host.",
},
[]string{"hostname", "reason"},
)
OffersAcquired = prometheus.NewCounterVec(
prometheus.CounterOpts{
Subsystem: offerSubsystem,
Name: "acquired",
Help: "Counter of offers acquired for task launch broken out by slave host.",
},
[]string{"hostname"},
)
OffersReleased = prometheus.NewCounterVec(
prometheus.CounterOpts{
Subsystem: offerSubsystem,
Name: "released",
Help: "Counter of previously-acquired offers later released, broken out by slave host.",
},
[]string{"hostname"},
)
)
var registerMetrics sync.Once
func Register() {
registerMetrics.Do(func() {
prometheus.MustRegister(OffersReceived)
prometheus.MustRegister(OffersDeclined)
prometheus.MustRegister(OffersAcquired)
prometheus.MustRegister(OffersReleased)
})
}
func InMicroseconds(d time.Duration) float64 {
return float64(d.Nanoseconds() / time.Microsecond.Nanoseconds())
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package proc provides opinionated utilities for processing background
// operations and future errors, somewhat inspired by libprocess.
package proc
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package proc
import (
"errors"
)
var (
errProcessTerminated = errors.New("cannot execute action because process has terminated")
errIllegalState = errors.New("illegal state, cannot execute action")
)
func IsProcessTerminated(err error) bool {
return err == errProcessTerminated
}
func IsIllegalState(err error) bool {
return err == errIllegalState
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package proc
import (
"fmt"
"sync"
"sync/atomic"
"time"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/runtime"
log "github.com/golang/glog"
)
const (
// if the action processor crashes (if some Action panics) then we
// wait this long before spinning up the action processor again.
defaultActionHandlerCrashDelay = 100 * time.Millisecond
// how many actions we can store in the backlog
defaultActionQueueDepth = 1024
)
type procImpl struct {
Config
backlog chan Action // action queue
terminate chan struct{} // signaled via close()
wg sync.WaitGroup // End() terminates when the wait is over
done runtime.Signal
state *stateType
pid uint32
writeLock sync.Mutex // avoid data race between write and close of backlog
changed *sync.Cond // wait/signal for backlog changes
engine DoerFunc // isolated this for easier unit testing later on
running chan struct{} // closes once event loop processing starts
dead chan struct{} // closes upon completion of process termination
}
type Config struct {
// cooldown period in between deferred action crashes
actionHandlerCrashDelay time.Duration
// determines the size of the deferred action backlog
actionQueueDepth uint32
}
var (
defaultConfig = Config{
actionHandlerCrashDelay: defaultActionHandlerCrashDelay,
actionQueueDepth: defaultActionQueueDepth,
}
pid uint32
closedErrChan <-chan error
)
func init() {
ch := make(chan error)
close(ch)
closedErrChan = ch
}
func New() Process {
return newConfigured(defaultConfig)
}
func newConfigured(config Config) Process {
state := stateNew
pi := &procImpl{
Config: config,
backlog: make(chan Action, config.actionQueueDepth),
terminate: make(chan struct{}),
state: &state,
pid: atomic.AddUint32(&pid, 1),
running: make(chan struct{}),
dead: make(chan struct{}),
}
pi.engine = DoerFunc(pi.doLater)
pi.changed = sync.NewCond(&pi.writeLock)
pi.wg.Add(1) // symmetrical to wg.Done() in End()
pi.done = pi.begin()
return pi
}
// returns a chan that closes upon termination of the action processing loop
func (self *procImpl) Done() <-chan struct{} {
return self.done
}
func (self *procImpl) Running() <-chan struct{} {
return self.running
}
func (self *procImpl) begin() runtime.Signal {
if !self.state.transition(stateNew, stateRunning) {
panic(fmt.Errorf("failed to transition from New to Idle state"))
}
defer log.V(2).Infof("started process %d", self.pid)
var entered runtime.Latch
// execute actions on the backlog chan
return runtime.After(func() {
runtime.Until(func() {
if entered.Acquire() {
close(self.running)
self.wg.Add(1)
}
for action := range self.backlog {
select {
case <-self.terminate:
return
default:
// signal to indicate there's room in the backlog now
self.changed.Broadcast()
// rely on Until to handle action panics
action()
}
}
}, self.actionHandlerCrashDelay, self.terminate)
}).Then(func() {
log.V(2).Infof("finished processing action backlog for process %d", self.pid)
if !entered.Acquire() {
self.wg.Done()
}
})
}
// execute some action in the context of the current process. Actions
// executed via this func are to be executed in a concurrency-safe manner:
// no two actions should execute at the same time. invocations of this func
// should not block for very long, unless the action backlog is full or the
// process is terminating.
// returns errProcessTerminated if the process already ended.
func (self *procImpl) doLater(deferredAction Action) (err <-chan error) {
a := Action(func() {
self.wg.Add(1)
defer self.wg.Done()
deferredAction()
})
scheduled := false
self.writeLock.Lock()
defer self.writeLock.Unlock()
for err == nil && !scheduled {
switch s := self.state.get(); s {
case stateRunning:
select {
case self.backlog <- a:
scheduled = true
default:
self.changed.Wait()
}
case stateTerminal:
err = ErrorChan(errProcessTerminated)
default:
err = ErrorChan(errIllegalState)
}
}
return
}
// implementation of Doer interface, schedules some action to be executed via
// the current execution engine
func (self *procImpl) Do(a Action) <-chan error {
return self.engine(a)
}
// spawn a goroutine that waits for an error. if a non-nil error is read from the
// channel then the handler func is invoked, otherwise (nil error or closed chan)
// the handler is skipped. if a nil handler is specified then it's not invoked.
// the signal chan that's returned closes once the error process logic (and handler,
// if any) has completed.
func OnError(ch <-chan error, f func(error), abort <-chan struct{}) <-chan struct{} {
return runtime.After(func() {
if ch == nil {
return
}
select {
case err, ok := <-ch:
if ok && err != nil && f != nil {
f(err)
}
case <-abort:
if f != nil {
f(errProcessTerminated)
}
}
})
}
func (self *procImpl) OnError(ch <-chan error, f func(error)) <-chan struct{} {
return OnError(ch, f, self.Done())
}
func (self *procImpl) flush() {
log.V(2).Infof("flushing action backlog for process %d", self.pid)
i := 0
//TODO: replace with `for range self.backlog` once Go 1.3 support is dropped
for {
_, open := <-self.backlog
if !open {
break
}
i++
}
log.V(2).Infof("flushed %d backlog actions for process %d", i, self.pid)
}
func (self *procImpl) End() <-chan struct{} {
if self.state.transitionTo(stateTerminal, stateTerminal) {
go func() {
defer close(self.dead)
self.writeLock.Lock()
defer self.writeLock.Unlock()
log.V(2).Infof("terminating process %d", self.pid)
close(self.backlog)
close(self.terminate)
self.wg.Done()
self.changed.Broadcast()
log.V(2).Infof("waiting for deferred actions to complete")
// wait for all pending actions to complete, then flush the backlog
self.wg.Wait()
self.flush()
}()
}
return self.dead
}
type errorOnce struct {
once sync.Once
err chan error
abort <-chan struct{}
}
func NewErrorOnce(abort <-chan struct{}) ErrorOnce {
return &errorOnce{
err: make(chan error, 1),
abort: abort,
}
}
func (b *errorOnce) Err() <-chan error {
return b.err
}
func (b *errorOnce) Reportf(msg string, args ...interface{}) {
b.Report(fmt.Errorf(msg, args...))
}
func (b *errorOnce) Report(err error) {
b.once.Do(func() {
select {
case b.err <- err:
default:
}
})
}
func (b *errorOnce) Send(errIn <-chan error) ErrorOnce {
go b.forward(errIn)
return b
}
func (b *errorOnce) forward(errIn <-chan error) {
if errIn == nil {
b.Report(nil)
return
}
select {
case err, _ := <-errIn:
b.Report(err)
case <-b.abort:
b.Report(errProcessTerminated)
}
}
type processAdapter struct {
parent Process
delegate Doer
}
func (p *processAdapter) Do(a Action) <-chan error {
if p == nil || p.parent == nil || p.delegate == nil {
return ErrorChan(errIllegalState)
}
errCh := NewErrorOnce(p.Done())
go func() {
errOuter := p.parent.Do(func() {
errInner := p.delegate.Do(a)
errCh.forward(errInner)
})
// if the outer err is !nil then either the parent failed to schedule the
// the action, or else it backgrounded the scheduling task.
if errOuter != nil {
errCh.forward(errOuter)
}
}()
return errCh.Err()
}
func (p *processAdapter) End() <-chan struct{} {
if p != nil && p.parent != nil {
return p.parent.End()
}
return nil
}
func (p *processAdapter) Done() <-chan struct{} {
if p != nil && p.parent != nil {
return p.parent.Done()
}
return nil
}
func (p *processAdapter) Running() <-chan struct{} {
if p != nil && p.parent != nil {
return p.parent.Running()
}
return nil
}
func (p *processAdapter) OnError(ch <-chan error, f func(error)) <-chan struct{} {
if p != nil && p.parent != nil {
return p.parent.OnError(ch, f)
}
return nil
}
// returns a process that, within its execution context, delegates to the specified Doer.
// if the given Doer instance is nil, a valid Process is still returned though calls to its
// Do() implementation will always return errIllegalState.
// if the given Process instance is nil then in addition to the behavior in the prior sentence,
// calls to End() and Done() are effectively noops.
func DoWith(other Process, d Doer) Process {
return &processAdapter{
parent: other,
delegate: d,
}
}
func ErrorChanf(msg string, args ...interface{}) <-chan error {
return ErrorChan(fmt.Errorf(msg, args...))
}
func ErrorChan(err error) <-chan error {
if err == nil {
return closedErrChan
}
ch := make(chan error, 1)
ch <- err
return ch
}
// invoke the f on action a. returns an illegal state error if f is nil.
func (f DoerFunc) Do(a Action) <-chan error {
if f != nil {
return f(a)
}
return ErrorChan(errIllegalState)
}
/*
Copyright 2015 The Kubernetes Authors All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package proc
import (
"fmt"
"sync"
"testing"
"time"
"github.com/GoogleCloudPlatform/kubernetes/contrib/mesos/pkg/runtime"
log "github.com/golang/glog"
)
// logs a testing.Fatalf if the elapsed time d passes before signal chan done is closed
func fatalAfter(t *testing.T, done <-chan struct{}, d time.Duration, msg string, args ...interface{}) {
select {
case <-done:
case <-time.After(d):
t.Fatalf(msg, args...)
}
}
func errorAfter(errOnce ErrorOnce, done <-chan struct{}, d time.Duration, msg string, args ...interface{}) {
select {
case <-done:
case <-time.After(d):
errOnce.Reportf(msg, args...)
}
}
// logs a testing.Fatalf if the signal chan closes before the elapsed time d passes
func fatalOn(t *testing.T, done <-chan struct{}, d time.Duration, msg string, args ...interface{}) {
select {
case <-done:
t.Fatalf(msg, args...)
case <-time.After(d):
}
}
func TestProc_manyEndings(t *testing.T) {
p := New()
const COUNT = 20
var wg sync.WaitGroup
wg.Add(COUNT)
for i := 0; i < COUNT; i++ {
runtime.On(p.End(), wg.Done)
}
fatalAfter(t, runtime.After(wg.Wait), 5*time.Second, "timed out waiting for loose End()s")
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_singleAction(t *testing.T) {
p := New()
scheduled := make(chan struct{})
called := make(chan struct{})
go func() {
log.Infof("do'ing deferred action")
defer close(scheduled)
err := p.Do(func() {
defer close(called)
log.Infof("deferred action invoked")
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
}()
fatalAfter(t, scheduled, 5*time.Second, "timed out waiting for deferred action to be scheduled")
fatalAfter(t, called, 5*time.Second, "timed out waiting for deferred action to be invoked")
p.End()
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_singleActionEnd(t *testing.T) {
p := New()
scheduled := make(chan struct{})
called := make(chan struct{})
go func() {
log.Infof("do'ing deferred action")
defer close(scheduled)
err := p.Do(func() {
defer close(called)
log.Infof("deferred action invoked")
p.End()
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
}()
fatalAfter(t, scheduled, 5*time.Second, "timed out waiting for deferred action to be scheduled")
fatalAfter(t, called, 5*time.Second, "timed out waiting for deferred action to be invoked")
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_multiAction(t *testing.T) {
p := New()
const COUNT = 10
var called sync.WaitGroup
called.Add(COUNT)
// test FIFO property
next := 0
for i := 0; i < COUNT; i++ {
log.Infof("do'ing deferred action %d", i)
idx := i
err := p.Do(func() {
defer called.Done()
log.Infof("deferred action invoked")
if next != idx {
t.Fatalf("expected index %d instead of %d", idx, next)
}
next++
})
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
}
fatalAfter(t, runtime.After(called.Wait), 2*time.Second, "timed out waiting for deferred actions to be invoked")
p.End()
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_goodLifecycle(t *testing.T) {
p := New()
p.End()
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_doWithDeadProc(t *testing.T) {
p := New()
p.End()
time.Sleep(100 * time.Millisecond)
errUnexpected := fmt.Errorf("unexpected execution of delegated action")
decorated := DoWith(p, DoerFunc(func(_ Action) <-chan error {
return ErrorChan(errUnexpected)
}))
decorated.Do(func() {})
fatalAfter(t, decorated.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_doWith(t *testing.T) {
p := New()
delegated := false
decorated := DoWith(p, DoerFunc(func(a Action) <-chan error {
delegated = true
a()
return nil
}))
executed := make(chan struct{})
err := decorated.Do(func() {
defer close(executed)
if !delegated {
t.Fatalf("expected delegated execution")
}
})
if err == nil {
t.Fatalf("expected !nil error chan")
}
fatalAfter(t, executed, 5*time.Second, "timed out waiting deferred execution")
fatalAfter(t, decorated.OnError(err, func(e error) {
t.Fatalf("unexpected error: %v", err)
}), 1*time.Second, "timed out waiting for doer result")
decorated.End()
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_doWithNestedTwice(t *testing.T) {
p := New()
delegated := false
decorated := DoWith(p, DoerFunc(func(a Action) <-chan error {
a()
return nil
}))
decorated2 := DoWith(decorated, DoerFunc(func(a Action) <-chan error {
delegated = true
a()
return nil
}))
executed := make(chan struct{})
err := decorated2.Do(func() {
defer close(executed)
if !delegated {
t.Fatalf("expected delegated execution")
}
})
if err == nil {
t.Fatalf("expected !nil error chan")
}
fatalAfter(t, executed, 5*time.Second, "timed out waiting deferred execution")
fatalAfter(t, decorated2.OnError(err, func(e error) {
t.Fatalf("unexpected error: %v", err)
}), 1*time.Second, "timed out waiting for doer result")
decorated2.End()
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func TestProc_doWithNestedErrorPropagation(t *testing.T) {
p := New()
delegated := false
decorated := DoWith(p, DoerFunc(func(a Action) <-chan error {
a()
return nil
}))
expectedErr := fmt.Errorf("expecting this")
errOnce := NewErrorOnce(p.Done())
decorated2 := DoWith(decorated, DoerFunc(func(a Action) <-chan error {
delegated = true
a()
errOnce.Reportf("unexpected error in decorator2")
return ErrorChanf("another unexpected error in decorator2")
}))
executed := make(chan struct{})
err := decorated2.Do(func() {
defer close(executed)
if !delegated {
t.Fatalf("expected delegated execution")
}
errOnce.Report(expectedErr)
})
if err == nil {
t.Fatalf("expected !nil error chan")
}
errOnce.Send(err)
foundError := false
fatalAfter(t, executed, 1*time.Second, "timed out waiting deferred execution")
fatalAfter(t, decorated2.OnError(errOnce.Err(), func(e error) {
if e != expectedErr {
t.Fatalf("unexpected error: %v", err)
} else {
foundError = true
}
}), 1*time.Second, "timed out waiting for doer result")
if !foundError {
t.Fatalf("expected a propagated error")
}
decorated2.End()
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
func runDelegationTest(t *testing.T, p Process, name string, errOnce ErrorOnce) {
defer func() {
t.Logf("runDelegationTest finished at " + time.Now().String())
}()
var decorated Process
decorated = p
const DEPTH = 100
var wg sync.WaitGroup
wg.Add(DEPTH)
y := 0
for x := 1; x <= DEPTH; x++ {
x := x
nextp := DoWith(decorated, DoerFunc(func(a Action) <-chan error {
if x == 1 {
t.Logf("delegate chain invoked for " + name)
}
y++
if y != x {
return ErrorChanf("out of order delegated execution")
}
defer wg.Done()
a()
return nil
}))
decorated = nextp
}
executed := make(chan struct{})
errCh := decorated.Do(func() {
defer close(executed)
if y != DEPTH {
errOnce.Reportf("expected delegated execution")
}
t.Logf("executing deferred action: " + name + " at " + time.Now().String())
errOnce.Send(nil) // we completed without error, let the listener know
})
if errCh == nil {
t.Fatalf("expected !nil error chan")
}
// forward any scheduling errors to the listener; NOTHING else should attempt to read
// from errCh after this point
errOnce.Send(errCh)
errorAfter(errOnce, executed, 5*time.Second, "timed out waiting deferred execution")
t.Logf("runDelegationTest received executed signal at " + time.Now().String())
}
func TestProc_doWithNestedX(t *testing.T) {
t.Logf("starting test case at " + time.Now().String())
p := New()
errOnce := NewErrorOnce(p.Done())
runDelegationTest(t, p, "nested", errOnce)
<-p.End()
select {
case err := <-errOnce.Err():
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
case <-time.After(5 * time.Second):
t.Fatalf("timed out waiting for doer result")
}
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
// intended to be run with -race
func TestProc_doWithNestedXConcurrent(t *testing.T) {
p := New()
errOnce := NewErrorOnce(p.Done())
var wg sync.WaitGroup
const CONC = 20
wg.Add(CONC)
for i := 0; i < CONC; i++ {
i := i
runtime.After(func() { runDelegationTest(t, p, fmt.Sprintf("nested%d", i), errOnce) }).Then(wg.Done)
}
ch := runtime.After(wg.Wait)
fatalAfter(t, ch, 10*time.Second, "timed out waiting for concurrent delegates")
<-p.End()
select {
case err := <-errOnce.Err():
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
case <-time.After(5 * time.Second):
t.Fatalf("timed out waiting for doer result")
}
fatalAfter(t, p.Done(), 5*time.Second, "timed out waiting for process death")
}
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment