Merge pull request #36 from contiv/mapuri/resourceAllocator

Resource allocator interface changes

Author: mapuri

Makefile

@@ -27,10 +27,16 @@ clean-demo:
 unit-test: build
 	CONTIV_HOST_GOBIN=$(HOST_GOBIN) CONTIV_HOST_GOROOT=$(HOST_GOROOT) ./scripts/unittests -vagrant
 
+# setting CONTIV_SOE=1 while calling 'make system-test' will stop the test
+# on first failure and leave setup in that state. This can be useful for debugging
+# as part of development.
 system-test: build
 	go test -v -run "sanity" github.com/contiv/netplugin/systemtests/singlehost 
 	go test --timeout 20m -v -run "sanity" github.com/contiv/netplugin/systemtests/twohosts
 
+# setting CONTIV_SOE=1 while calling 'make regress-test' will stop the test
+# on first failure and leave setup in that state. This can be useful for debugging
+# as part of development.
 regress-test: build
 	go test -v -run "regress" github.com/contiv/netplugin/systemtests/singlehost 
 	go test --timeout 60m -v -run "regress" github.com/contiv/netplugin/systemtests/twohosts

core/core.go

@@ -27,12 +27,6 @@ type Address struct {
 	addr string
 }
 
-type State interface {
-	Write() error
-	Read(id string) error
-	Clear() error
-}
-
 type Config struct {
 	// Config object parsed from a json styled config
 	V interface{}
@@ -103,5 +97,30 @@ type StateDriver interface {
 		marshal func(interface{}) ([]byte, error)) error
 	ReadState(key string, value State,
 		unmarshal func([]byte, interface{}) error) error
+	ReadAllState(baseKey string, stateType State,
+		unmarshal func([]byte, interface{}) error) ([]State, error)
 	ClearState(key string) error
 }
+
+type Resource interface {
+	// Resource defines a allocatable unit. A resource is uniquely identified
+	// by 'Id'. A resource description identifies the nature of the resource.
+	State
+	Init(rsrcCfg interface{}) error
+	Deinit()
+	Description() string
+	Allocate() (interface{}, error)
+	Deallocate(interface{}) error
+}
+
+type ResourceManager interface {
+	// A resource manager provides mechanism to manage (define/undefine,
+	// allocate/deallocate) resources. Example, it may provide management in
+	// logically centralized manner in a distributed system
+	Init() error
+	Deinit()
+	DefineResource(id, desc string, rsrcCfg interface{}) error
+	UndefineResource(id, desc string) error
+	AllocateResourceVal(id, desc string) (interface{}, error)
+	DeallocateResourceVal(id, desc string, value interface{}) error
+}

core/state.go

@@ -0,0 +1,16 @@
+package core
+
+type State interface {
+	Read(id string) error
+	ReadAll() ([]State, error)
+	Write() error
+	Clear() error
+}
+
+// CommonState defines the fields common to all core.State implementations.
+// This struct shall be embedded as anonymous field in all structs that
+// implement core.State
+type CommonState struct {
+	StateDriver StateDriver `json:"-"`
+	Id          string      `json:"id"`
+}

drivers/etcdstatedriver.go

@@ -17,8 +17,9 @@ package drivers
 
 import (
 	"fmt"
-	"github.com/contiv/go-etcd/etcd"
+	"reflect"
 
+	"github.com/contiv/go-etcd/etcd"
 	"github.com/contiv/netplugin/core"
 )
 
@@ -103,6 +104,48 @@ func (d *EtcdStateDriver) ReadState(key string, value core.State,
 	return nil
 }
 
+// XXX: move this to some common file
+func ReadAllStateCommon(d core.StateDriver, baseKey string, sType core.State,
+	unmarshal func([]byte, interface{}) error) ([]core.State, error) {
+	stateType := reflect.TypeOf(sType)
+	sliceType := reflect.SliceOf(stateType)
+	values := reflect.MakeSlice(sliceType, 0, 1)
+
+	byteValues, err := d.ReadAll(baseKey)
+	if err != nil {
+		return nil, err
+	}
+	for _, byteValue := range byteValues {
+		value := reflect.New(stateType)
+		err = unmarshal(byteValue, value.Interface())
+		if err != nil {
+			return nil, err
+		}
+		values = reflect.Append(values, value.Elem())
+	}
+
+	stateValues := []core.State{}
+	for i := 0; i < values.Len(); i++ {
+		// sanity checks
+		if !values.Index(i).Elem().FieldByName("CommonState").IsValid() {
+			panic(fmt.Sprintf("The state structure %v is missing core.CommonState",
+				stateType))
+			return nil, &core.Error{Desc: fmt.Sprintf("The state structure %v is missing core.CommonState",
+				stateType)}
+		}
+		//the following works as every core.State is expected to embed core.CommonState struct
+		values.Index(i).Elem().FieldByName("CommonState").FieldByName("StateDriver").Set(reflect.ValueOf(d))
+		stateValue := values.Index(i).Interface().(core.State)
+		stateValues = append(stateValues, stateValue)
+	}
+	return stateValues, nil
+}
+
+func (d *EtcdStateDriver) ReadAllState(baseKey string, sType core.State,
+	unmarshal func([]byte, interface{}) error) ([]core.State, error) {
+	return ReadAllStateCommon(d, baseKey, sType, unmarshal)
+}
+
 func (d *EtcdStateDriver) WriteState(key string, value core.State,
 	marshal func(interface{}) ([]byte, error)) error {
 	encodedState, err := marshal(value)

drivers/etcdstatedriver_test.go

@@ -23,26 +23,6 @@ import (
 	"github.com/contiv/netplugin/core"
 )
 
-// setup a etcd cluster, run tests and then cleanup the cluster
-// XXX: enabled once I upgrade to golang 1.4
-//func TestMain(m *testing.M) {
-//
-//	// start etcd
-//	proc, err := os.StartProcess("etcd", []string{}, nil)
-//	if err != nil {
-//		log.Printf("failed to start etcd. Error: %s", err)
-//		os.Exit(-1)
-//	}
-//
-//	//run the tests
-//	exitC := m.Run()
-//
-//	// stop etcd
-//	proc.Kill()
-//
-//	os.Exit(exitC)
-//}
-
 func setupDriver(t *testing.T) *EtcdStateDriver {
 	etcdConfig := &EtcdStateDriverConfig{}
 	etcdConfig.Etcd.Machines = []string{"http://127.0.0.1:4001"}
@@ -125,6 +105,10 @@ func (s *testState) Read(id string) error {
 	return &core.Error{Desc: "Should not be called!!"}
 }
 
+func (s *testState) ReadAll() ([]core.State, error) {
+	return nil, &core.Error{Desc: "Should not be called!!"}
+}
+
 func (s *testState) Clear() error {
 	return &core.Error{Desc: "Should not be called!!"}
 }

drivers/fakestatedriver.go

@@ -0,0 +1,103 @@
+package drivers
+
+import (
+	"log"
+	"strings"
+
+	"github.com/contiv/netplugin/core"
+)
+
+// The FakeStateDriver implements core.StateDriver interface for use with
+// unit-tests
+
+type ValueData struct {
+	value []byte
+}
+
+type FakeStateDriver struct {
+	TestState map[string]ValueData
+}
+
+func (d *FakeStateDriver) Init(config *core.Config) error {
+	d.TestState = make(map[string]ValueData)
+
+	return nil
+}
+
+func (d *FakeStateDriver) Deinit() {
+	d.TestState = nil
+}
+
+func (d *FakeStateDriver) Write(key string, value []byte) error {
+	val := ValueData{value: value}
+	d.TestState[key] = val
+
+	return nil
+}
+
+func (d *FakeStateDriver) Read(key string) ([]byte, error) {
+	if val, ok := d.TestState[key]; ok {
+		return val.value, nil
+	}
+
+	return []byte{}, &core.Error{Desc: "Key not found!"}
+}
+
+func (d *FakeStateDriver) ReadAll(baseKey string) ([][]byte, error) {
+	values := [][]byte{}
+
+	for key, val := range d.TestState {
+		if strings.Contains(key, baseKey) {
+			values = append(values, val.value)
+		}
+	}
+	return values, nil
+}
+
+func (d *FakeStateDriver) ClearState(key string) error {
+	if _, ok := d.TestState[key]; ok {
+		delete(d.TestState, key)
+	}
+	return nil
+}
+
+func (d *FakeStateDriver) ReadState(key string, value core.State,
+	unmarshal func([]byte, interface{}) error) error {
+	encodedState, err := d.Read(key)
+	if err != nil {
+		return err
+	}
+
+	err = unmarshal(encodedState, value)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (d *FakeStateDriver) ReadAllState(baseKey string, sType core.State,
+	unmarshal func([]byte, interface{}) error) ([]core.State, error) {
+	return ReadAllStateCommon(d, baseKey, sType, unmarshal)
+}
+
+func (d *FakeStateDriver) WriteState(key string, value core.State,
+	marshal func(interface{}) ([]byte, error)) error {
+	encodedState, err := marshal(value)
+	if err != nil {
+		return err
+	}
+
+	err = d.Write(key, encodedState)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (d *FakeStateDriver) DumpState() {
+	for key, _ := range d.TestState {
+		log.Printf("key: %q\n", key)
+	}
+}

drivers/ovsdriver.go

@@ -328,7 +328,8 @@ func vxlanIfName(netId, vtepIp string) string {
 }
 
 func (d *OvsDriver) createVtep(epCfg *OvsCfgEndpointState) error {
-	cfgNw := OvsCfgNetworkState{StateDriver: d.stateDriver}
+	cfgNw := OvsCfgNetworkState{}
+	cfgNw.StateDriver = d.stateDriver
 	err := cfgNw.Read(epCfg.NetId)
 	if err != nil {
 		return err
@@ -352,7 +353,8 @@ func (d *OvsDriver) createVtep(epCfg *OvsCfgEndpointState) error {
 
 func (d *OvsDriver) deleteVtep(epOper *OvsOperEndpointState) error {
 
-	cfgNw := OvsCfgNetworkState{StateDriver: d.stateDriver}
+	cfgNw := OvsCfgNetworkState{}
+	cfgNw.StateDriver = d.stateDriver
 	err := cfgNw.Read(epOper.NetId)
 	if err != nil {
 		return err
@@ -423,7 +425,8 @@ func (d *OvsDriver) Deinit() {
 }
 
 func (d *OvsDriver) CreateNetwork(id string) error {
-	cfgNw := OvsCfgNetworkState{StateDriver: d.stateDriver}
+	cfgNw := OvsCfgNetworkState{}
+	cfgNw.StateDriver = d.stateDriver
 	err := cfgNw.Read(id)
 	if err != nil {
 		log.Printf("Failed to read net %s \n", cfgNw.Id)
@@ -450,7 +453,8 @@ func (d *OvsDriver) CreateEndpoint(id string) error {
 	intfName := portName
 	intfType := "internal"
 
-	epCfg := OvsCfgEndpointState{StateDriver: d.stateDriver}
+	epCfg := OvsCfgEndpointState{}
+	epCfg.StateDriver = d.stateDriver
 	err = epCfg.Read(id)
 	if err != nil {
 		return err
@@ -476,7 +480,8 @@ func (d *OvsDriver) CreateEndpoint(id string) error {
 		intfType = ""
 	}
 
-	cfgNw := OvsCfgNetworkState{StateDriver: d.stateDriver}
+	cfgNw := OvsCfgNetworkState{}
+	cfgNw.StateDriver = d.stateDriver
 	err = cfgNw.Read(epCfg.NetId)
 	if err != nil {
 		return err
@@ -495,16 +500,16 @@ func (d *OvsDriver) CreateEndpoint(id string) error {
 	}()
 
 	operEp := OvsOperEndpointState{
-		StateDriver: d.stateDriver,
-		Id:          id,
-		PortName:    portName,
-		NetId:       epCfg.NetId,
-		AttachUUID:  epCfg.AttachUUID,
-		ContName:    epCfg.ContName,
-		IpAddress:   epCfg.IpAddress,
-		IntfName:    intfName,
-		HomingHost:  epCfg.HomingHost,
-		VtepIp:      epCfg.VtepIp}
+		PortName:   portName,
+		NetId:      epCfg.NetId,
+		AttachUUID: epCfg.AttachUUID,
+		ContName:   epCfg.ContName,
+		IpAddress:  epCfg.IpAddress,
+		IntfName:   intfName,
+		HomingHost: epCfg.HomingHost,
+		VtepIp:     epCfg.VtepIp}
+	operEp.StateDriver = d.stateDriver
+	operEp.Id = id
 	err = operEp.Write()
 	if err != nil {
 		return err
@@ -520,7 +525,8 @@ func (d *OvsDriver) CreateEndpoint(id string) error {
 
 func (d *OvsDriver) DeleteEndpoint(id string) (err error) {
 
-	epOper := OvsOperEndpointState{StateDriver: d.stateDriver}
+	epOper := OvsOperEndpointState{}
+	epOper.StateDriver = d.stateDriver
 	err = epOper.Read(id)
 	if err != nil {
 		return err

drivers/ovsdriver_test.go

@@ -172,6 +172,11 @@ func (d *testOvsStateDriver) ReadState(key string, value core.State,
 	return &core.Error{Desc: fmt.Sprintf("unknown key! %s", key)}
 }
 
+func (d *testOvsStateDriver) ReadAllState(key string, value core.State,
+	unmarshal func([]byte, interface{}) error) ([]core.State, error) {
+	return nil, &core.Error{Desc: "shouldn't be called!"}
+}
+
 func (d *testOvsStateDriver) WriteState(key string, value core.State,
 	marshal func(interface{}) ([]byte, error)) error {
 	return nil