epic/play-with-docker
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add storage API abstraction

Browse Source
This commit is contained in:
Jonathan Leibiusky @xetorthio
2017-06-22 09:16:49 -03:00
committed by Marcos Lilljedahl
parent 824d5334df e0f0e47df7
commit 08b64dadde
19 changed files with 277 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
pwd/client.go
View File

@@ -2,6 +2,7 @@ package pwd
import (
"log"
"sync/atomic"
"time"
"github.com/play-with-docker/play-with-docker/pwd/types"
@@ -11,6 +12,7 @@ func (p *pwd) ClientNew(id string, session *types.Session) *types.Client {
defer observeAction("ClientNew", time.Now())
c := &types.Client{Id: id, Session: session}
session.Clients = append(session.Clients, c)
p.clientCount = atomic.AddInt32(&p.clientCount, 1)
return c
}
@@ -29,6 +31,7 @@ func (p *pwd) ClientClose(client *types.Client) {
for i, cl := range session.Clients {
if cl.Id == client.Id {
session.Clients = append(session.Clients[:i], session.Clients[i+1:]...)
p.clientCount = atomic.AddInt32(&p.clientCount, -1)
break
}
}
@@ -38,6 +41,10 @@ func (p *pwd) ClientClose(client *types.Client) {
p.setGauges()
}
func (p *pwd) ClientCount() int {
return int(atomic.LoadInt32(&p.clientCount))
}
func (p *pwd) notifyClientSmallestViewPort(session *types.Session) {
vp := p.SessionGetSmallestViewPort(session)
// Resize all terminals in the session

15
pwd/client_test.go
View File

@@ -24,6 +24,21 @@ func TestClientNew(t *testing.T) {
assert.Equal(t, types.Client{Id: "foobar", Session: session, ViewPort: types.ViewPort{Cols: 0, Rows: 0}}, *client)
assert.Contains(t, session.Clients, client)
}
func TestClientCount(t *testing.T) {
docker := &mockDocker{}
tasks := &mockTasks{}
broadcast := &mockBroadcast{}
storage := &mockStorage{}
p := NewPWD(docker, tasks, broadcast, storage)
session, err := p.SessionNew(time.Hour, "", "", "")
assert.Nil(t, err)
p.ClientNew("foobar", session)
assert.Equal(t, 1, p.ClientCount())
}
func TestClientResizeViewPort(t *testing.T) {
docker := &mockDocker{}

48
pwd/instance.go
View File

@@ -4,6 +4,7 @@ import (
"fmt"
"io"
"log"
"net"
"net/http"
"path/filepath"
"strings"
@@ -22,6 +23,8 @@ type sessionWriter struct {
broadcast BroadcastApi
}
var terms = make(map[string]map[string]net.Conn)
func (s *sessionWriter) Write(p []byte) (n int, err error) {
s.broadcast.BroadcastTo(s.sessionId, "terminal out", s.instanceName, string(p))
return len(p), nil
@@ -45,6 +48,10 @@ func (p *pwd) InstanceResizeTerminal(instance *types.Instance, rows, cols uint)
}
func (p *pwd) InstanceAttachTerminal(instance *types.Instance) error {
// already have a connection for this instance
if getInstanceTermConn(instance.SessionId, instance.Name) != nil {
return nil
}
conn, err := p.docker.CreateAttachConnection(instance.Name)
if err != nil {
@@ -53,9 +60,12 @@ func (p *pwd) InstanceAttachTerminal(instance *types.Instance) error {
encoder := encoding.Replacement.NewEncoder()
sw := &sessionWriter{sessionId: instance.Session.Id, instanceName: instance.Name, broadcast: p.broadcast}
instance.Terminal = conn
if terms[instance.SessionId] == nil {
terms[instance.SessionId] = map[string]net.Conn{instance.Name: conn}
} else {
terms[instance.SessionId][instance.Name] = conn
}
io.Copy(encoder.Writer(sw), conn)
return nil
}
@@ -82,6 +92,18 @@ func (p *pwd) InstanceUploadFromUrl(instance *types.Instance, url string) error
return nil
}
func (p *pwd) InstanceUploadFromReader(instance *types.Instance, fileName string, reader io.Reader) error {
defer observeAction("InstanceUploadFromReader", time.Now())
copyErr := p.docker.CopyToContainer(instance.Name, "/var/run/pwd/uploads", fileName, reader)
if copyErr != nil {
return fmt.Errorf("Error while uploading file [%s]. Error: %s\n", fileName, copyErr)
}
return nil
}
func (p *pwd) InstanceGet(session *types.Session, name string) *types.Instance {
defer observeAction("InstanceGet", time.Now())
return session.Instances[name]
@@ -118,8 +140,10 @@ func (p *pwd) InstanceFindByAlias(sessionPrefix, alias string) *types.Instance {
func (p *pwd) InstanceDelete(session *types.Session, instance *types.Instance) error {
defer observeAction("InstanceDelete", time.Now())
if instance.Terminal != nil {
instance.Terminal.Close()
conn := getInstanceTermConn(session.Id, instance.Name)
if conn != nil {
conn.Close()
delete(terms[instance.SessionId], instance.Name)
}
err := p.docker.DeleteContainer(instance.Name)
if err != nil && !strings.Contains(err.Error(), "No such container") {
@@ -130,7 +154,7 @@ func (p *pwd) InstanceDelete(session *types.Session, instance *types.Instance) e
p.broadcast.BroadcastTo(session.Id, "delete instance", instance.Name)
delete(session.Instances, instance.Name)
if err := p.storage.SessionPut(session); err != nil {
if err := p.storage.InstanceDelete(session.Id, instance.Name); err != nil {
return err
}
@@ -202,6 +226,7 @@ func (p *pwd) InstanceNew(session *types.Session, conf InstanceConfig) (*types.I
instance := &types.Instance{}
instance.Image = opts.Image
instance.IP = ip
instance.SessionId = session.Id
instance.Name = containerName
instance.Hostname = conf.Hostname
instance.Alias = conf.Alias
@@ -221,7 +246,7 @@ func (p *pwd) InstanceNew(session *types.Session, conf InstanceConfig) (*types.I
go p.InstanceAttachTerminal(instance)
err = p.storage.SessionPut(session)
err = p.storage.InstanceCreate(session.Id, instance)
if err != nil {
return nil, err
}
@@ -233,10 +258,11 @@ func (p *pwd) InstanceNew(session *types.Session, conf InstanceConfig) (*types.I
return instance, nil
}
func (p *pwd) InstanceWriteToTerminal(instance *types.Instance, data string) {
func (p *pwd) InstanceWriteToTerminal(sessionId, instanceName string, data string) {
defer observeAction("InstanceWriteToTerminal", time.Now())
if instance != nil && instance.Terminal != nil && len(data) > 0 {
instance.Terminal.Write([]byte(data))
conn := getInstanceTermConn(sessionId, instanceName)
if conn != nil && len(data) > 0 {
conn.Write([]byte(data))
}
}
@@ -255,3 +281,7 @@ func (p *pwd) InstanceExec(instance *types.Instance, cmd []string) (int, error)
defer observeAction("InstanceExec", time.Now())
return p.docker.Exec(instance.Name, cmd)
}
func getInstanceTermConn(sessionId, instanceName string) net.Conn {
return terms[sessionId][instanceName]
}

41
pwd/instance_test.go
View File

@@ -1,7 +1,9 @@
package pwd
import (
"errors"
"fmt"
"net"
"sync"
"testing"
"time"
@@ -69,6 +71,7 @@ func TestInstanceNew(t *testing.T) {
Alias: "",
Image: config.GetDindImageName(),
IsDockerHost: true,
SessionId: session.Id,
Session: session,
}
@@ -159,6 +162,7 @@ func TestInstanceNew_WithNotAllowedImage(t *testing.T) {
IP: "10.0.0.1",
Alias: "",
Image: "redis",
SessionId: session.Id,
IsDockerHost: false,
Session: session,
}
@@ -209,6 +213,7 @@ func TestInstanceNew_WithCustomHostname(t *testing.T) {
Image: "redis",
IsDockerHost: false,
Session: session,
SessionId: session.Id,
}
assert.Equal(t, expectedInstance, *instance)
@@ -239,3 +244,39 @@ func TestInstanceAllowedImages(t *testing.T) {
assert.Equal(t, expectedImages, p.InstanceAllowedImages())
}
type errConn struct {
*mockConn
}
func (ec errConn) Read(b []byte) (int, error) {
return 0, errors.New("Error")
}
func TestTermConnAssignment(t *testing.T) {
dock := &mockDocker{}
tasks := &mockTasks{}
broadcast := &mockBroadcast{}
storage := &mockStorage{}
dock.createAttachConnection = func(name string) (net.Conn, error) {
// return error connection to unlock the goroutine
return errConn{}, nil
}
p := NewPWD(dock, tasks, broadcast, storage)
session, _ := p.SessionNew(time.Hour, "", "", "")
mockInstance := &types.Instance{
Name: fmt.Sprintf("%s_redis-master", session.Id[:8]),
Hostname: "redis-master",
IP: "10.0.0.1",
Alias: "",
SessionId: session.Id,
Image: "redis",
IsDockerHost: false,
Session: session,
}
p.InstanceAttachTerminal(mockInstance)
assert.NotNil(t, getInstanceTermConn(session.Id, mockInstance.Name))
}

17
pwd/pwd.go
View File

@@ -1,6 +1,7 @@
package pwd
import (
"io"
"time"
"github.com/play-with-docker/play-with-docker/docker"
@@ -42,10 +43,11 @@ func init() {
}
type pwd struct {
docker docker.DockerApi
tasks SchedulerApi
broadcast BroadcastApi
storage storage.StorageApi
docker docker.DockerApi
tasks SchedulerApi
broadcast BroadcastApi
storage storage.StorageApi
clientCount int32
}
type PWDApi interface {
@@ -60,18 +62,21 @@ type PWDApi interface {
InstanceResizeTerminal(instance *types.Instance, cols, rows uint) error
InstanceAttachTerminal(instance *types.Instance) error
InstanceUploadFromUrl(instance *types.Instance, url string) error
InstanceUploadFromReader(instance *types.Instance, filename string, reader io.Reader) error
InstanceGet(session *types.Session, name string) *types.Instance
// TODO remove this function when we add the session prefix to the PWD url
InstanceFindByIP(ip string) *types.Instance
InstanceFindByAlias(sessionPrefix, alias string) *types.Instance
InstanceFindByIPAndSession(sessionPrefix, ip string) *types.Instance
InstanceDelete(session *types.Session, instance *types.Instance) error
InstanceWriteToTerminal(instance *types.Instance, data string)
InstanceWriteToTerminal(sessionId, instanceName string, data string)
InstanceAllowedImages() []string
InstanceExec(instance *types.Instance, cmd []string) (int, error)
ClientNew(id string, session *types.Session) *types.Client
ClientResizeViewPort(client *types.Client, cols, rows uint)
ClientClose(client *types.Client)
ClientCount() int
}
func NewPWD(d docker.DockerApi, t SchedulerApi, b BroadcastApi, s storage.StorageApi) *pwd {
@@ -83,7 +88,7 @@ func (p *pwd) setGauges() {
ses := float64(s)
i, _ := p.storage.InstanceCount()
ins := float64(i)
c, _ := p.storage.ClientCount()
c := p.ClientCount()
cli := float64(c)
clientsGauge.Set(cli)

23
pwd/session.go
View File

@@ -15,6 +15,8 @@ import (
"github.com/twinj/uuid"
)
var preparedSessions = map[string]bool{}
type sessionBuilderWriter struct {
sessionId string
broadcast BroadcastApi
@@ -64,7 +66,7 @@ func (p *pwd) SessionNew(duration time.Duration, stack, stackName, imageName str
}
log.Printf("Network [%s] created for session [%s]\n", s.Id, s.Id)
if err := p.prepareSession(s); err != nil {
if _, err := p.prepareSession(s); err != nil {
log.Println(err)
return nil, err
}
@@ -183,7 +185,7 @@ func (p *pwd) SessionGet(sessionId string) *types.Session {
s, _ := p.storage.SessionGet(sessionId)
if err := p.prepareSession(s); err != nil {
if _, err := p.prepareSession(s); err != nil {
log.Println(err)
return nil
}
@@ -280,22 +282,27 @@ func (p *pwd) SessionSetup(session *types.Session, conf SessionSetupConf) error
return nil
}
func isSessionPrepared(sessionId string) bool {
_, ok := preparedSessions[sessionId]
return ok
}
// This function should be called any time a session needs to be prepared:
// 1. Like when it is created
// 2. When it was loaded from storage
func (p *pwd) prepareSession(session *types.Session) error {
func (p *pwd) prepareSession(session *types.Session) (bool, error) {
session.Lock()
defer session.Unlock()
if session.IsPrepared() {
return nil
if isSessionPrepared(session.Id) {
return false, nil
}
p.scheduleSessionClose(session)
// Connect PWD daemon to the new network
if err := p.connectToNetwork(session); err != nil {
return err
return false, err
}
// Schedule periodic tasks
@@ -306,9 +313,9 @@ func (p *pwd) prepareSession(session *types.Session) error {
i.Session = session
go p.InstanceAttachTerminal(i)
}
session.SetPrepared()
preparedSessions[session.Id] = true
return nil
return true, nil
}
func (p *pwd) scheduleSessionClose(s *types.Session) {

27
pwd/session_test.go
View File

@@ -209,10 +209,10 @@ func TestSessionSetup(t *testing.T) {
Image: "franela/dind",
Hostname: "manager1",
IP: "10.0.0.1",
SessionId: s.Id,
Alias: "",
IsDockerHost: true,
Session: s,
Terminal: manager1Received.Terminal,
Docker: manager1Received.Docker,
}, manager1Received)
@@ -225,8 +225,8 @@ func TestSessionSetup(t *testing.T) {
IP: "10.0.0.2",
Alias: "",
IsDockerHost: true,
SessionId: s.Id,
Session: s,
Terminal: manager2Received.Terminal,
Docker: manager2Received.Docker,
}, manager2Received)
@@ -238,9 +238,9 @@ func TestSessionSetup(t *testing.T) {
Hostname: "manager3",
IP: "10.0.0.3",
Alias: "",
SessionId: s.Id,
IsDockerHost: true,
Session: s,
Terminal: manager3Received.Terminal,
Docker: manager3Received.Docker,
}, manager3Received)
@@ -252,9 +252,9 @@ func TestSessionSetup(t *testing.T) {
Hostname: "worker1",
IP: "10.0.0.4",
Alias: "",
SessionId: s.Id,
IsDockerHost: true,
Session: s,
Terminal: worker1Received.Terminal,
Docker: worker1Received.Docker,
}, worker1Received)
@@ -266,9 +266,9 @@ func TestSessionSetup(t *testing.T) {
Hostname: "other",
IP: "10.0.0.5",
Alias: "",
SessionId: s.Id,
IsDockerHost: true,
Session: s,
Terminal: otherReceived.Terminal,
Docker: otherReceived.Docker,
}, otherReceived)
@@ -277,3 +277,20 @@ func TestSessionSetup(t *testing.T) {
assert.True(t, manager3JoinedHasManager)
assert.True(t, worker1JoinedHasWorker)
}
func TestSessionPrepareOnce(t *testing.T) {
dock := &mockDocker{}
tasks := &mockTasks{}
broadcast := &mockBroadcast{}
storage := &mockStorage{}
p := NewPWD(dock, tasks, broadcast, storage)
session := &types.Session{Id: "1234"}
prepared, err := p.prepareSession(session)
assert.True(t, preparedSessions[session.Id])
assert.True(t, prepared)
prepared, err = p.prepareSession(session)
assert.Nil(t, err)
assert.False(t, prepared)
}

14
pwd/storage_mock_test.go
View File

@@ -10,6 +10,8 @@ type mockStorage struct {
instanceFindByAlias func(sessionPrefix, alias string) (*types.Instance, error)
instanceFindByIP func(ip string) (*types.Instance, error)
instanceFindByIPAndSession func(sessionPrefix, ip string) (*types.Instance, error)
instanceCreate func(string, *types.Instance) error
instanceDelete func(sessionId string, instanceName string) error
instanceCount func() (int, error)
clientCount func() (int, error)
}
@@ -56,6 +58,18 @@ func (m *mockStorage) InstanceFindByIPAndSession(sessionPrefix, ip string) (*typ
}
return nil, nil
}
func (m *mockStorage) InstanceCreate(sessionId string, instance *types.Instance) error {
if m.instanceCreate != nil {
return m.instanceCreate(sessionId, instance)
}
return nil
}
func (m *mockStorage) InstanceDelete(sessionId, instanceName string) error {
if m.instanceDelete != nil {
return m.instanceDelete(sessionId, instanceName)
}
return nil
}
func (m *mockStorage) InstanceCount() (int, error) {
if m.instanceCount != nil {
return m.instanceCount()

2
pwd/tasks.go
View File

@@ -33,7 +33,7 @@ type scheduler struct {
}
func (sch *scheduler) Schedule(s *types.Session) {
if s.IsPrepared() {
if isSessionPrepared(s.Id) {
return
}

44
pwd/types/instance.go
View File

@@ -2,7 +2,6 @@ package types
import (
"context"
"net"
"sync"
"github.com/play-with-docker/play-with-docker/docker"
@@ -15,27 +14,28 @@ func (p UInt16Slice) Less(i, j int) bool { return p[i] < p[j] }
func (p UInt16Slice) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
type Instance struct {
Image string `json:"image"`
Name string `json:"name"`
Hostname string `json:"hostname"`
IP string `json:"ip"`
IsManager *bool `json:"is_manager"`
Mem string `json:"mem"`
Cpu string `json:"cpu"`
Alias string `json:"alias"`
ServerCert []byte `json:"server_cert"`
ServerKey []byte `json:"server_key"`
CACert []byte `json:"ca_cert"`
Cert []byte `json:"cert"`
Key []byte `json:"key"`
IsDockerHost bool `json:"is_docker_host"`
Docker docker.DockerApi `json:"-"`
Session *Session `json:"-"`
Terminal net.Conn `json:"-"`
ctx context.Context `json:"-"`
tempPorts []uint16 `json:"-"`
Ports UInt16Slice
rw sync.Mutex
Image string `json:"image" bson:"image"`
Name string `json:"name" bson:"name"`
Hostname string `json:"hostname" bson:"hostname"`
IP string `json:"ip" bson:"ip"`
IsManager *bool `json:"is_manager" bson:"is_manager"`
Mem string `json:"mem" bson:"mem"`
Cpu string `json:"cpu" bson:"cpu"`
Alias string `json:"alias" bson:"alias"`
ServerCert []byte `json:"server_cert" bson:"server_cert"`
ServerKey []byte `json:"server_key" bson:"server_key"`
CACert []byte `json:"ca_cert" bson:"ca_cert"`
Cert []byte `json:"cert" bson:"cert"`
Key []byte `json:"key" bson:"key"`
IsDockerHost bool `json:"is_docker_host" bson:"is_docker_host"`
SessionId string `json:"session_id" bson:"session_id"`
SessionPrefix string `json:"session_prefix" bson:"session_prefix"`
Docker docker.DockerApi `json:"-"`
Session *Session `json:"-" bson:"-"`
ctx context.Context `json:"-" bson:"-"`
tempPorts []uint16 `json:"-" bson:"-"`
Ports UInt16Slice
rw sync.Mutex
}
func (i *Instance) SetUsedPort(port uint16) {

13
pwd/types/session.go
View File

@@ -7,7 +7,7 @@ import (
type Session struct {
Id string `json:"id"`
Instances map[string]*Instance `json:"instances"`
Instances map[string]*Instance `json:"instances" bson:"-"`
CreatedAt time.Time `json:"created_at"`
ExpiresAt time.Time `json:"expires_at"`
PwdIpAddress string `json:"pwd_ip_address"`
@@ -16,12 +16,11 @@ type Session struct {
StackName string `json:"stack_name"`
ImageName string `json:"image_name"`
Host string `json:"host"`
Clients []*Client `json:"-"`
Clients []*Client `json:"-" bson:"-"`
closingTimer *time.Timer `json:"-"`
scheduled bool `json:"-"`
ticker *time.Ticker `json:"-"`
rw sync.Mutex `json:"-"`
prepared bool `json:"-"`
}
func (s *Session) Lock() {
@@ -46,11 +45,3 @@ func (s *Session) SetClosingTimer(t *time.Timer) {
func (s *Session) ClosingTimer() *time.Timer {
return s.closingTimer
}
func (s *Session) IsPrepared() bool {
return s.prepared
}
func (s *Session) SetPrepared() {
s.prepared = true
}