Read Api cleanup

core/core.go

@@ -98,7 +98,6 @@ type StateDriver interface {
 	Deinit()
 	Write(key string, value []byte) error
 	Read(key string) ([]byte, error)
-	ReadRecursive(baseKey string) ([]string, error)
 	WriteState(key string, value State,
 		marshal func(interface{}) ([]byte, error)) error
 	ReadState(key string, value State,

drivers/etcdstatedriver.go

@@ -60,28 +60,28 @@ func (d *EtcdStateDriver) Write(key string, value []byte) error {
 	return err
 }
 
-func (d *EtcdStateDriver) ReadRecursive(baseKey string) ([]string, error) {
-	resp, err := d.Client.Get(baseKey, true, false)
+func (d *EtcdStateDriver) Read(key string) ([]byte, error) {
+	resp, err := d.Client.Get(key, false, false)
 	if err != nil {
-		return []string{}, err
-	}
-
-	keys := make([]string, len(resp.Node.Nodes))
-
-	for idx, respNode := range resp.Node.Nodes {
-		keys[idx] = respNode.Key
+		return []byte{}, err
 	}
 
-	return keys, err
+	return []byte(resp.Node.Value), err
 }
 
-func (d *EtcdStateDriver) Read(key string) ([]byte, error) {
-	resp, err := d.Client.Get(key, false, false)
+func ReadAll(d core.StateDriver, baseKey string) ([][]byte, error) {
+	etcdDriver := d.(*EtcdStateDriver)
+	resp, err := etcdDriver.Client.Get(baseKey, true, false)
 	if err != nil {
-		return []byte{}, err
+		return nil, err
 	}
 
-	return []byte(resp.Node.Value), err
+	values := [][]byte{}
+	for _, node := range resp.Node.Nodes {
+		values = append(values, []byte(node.Value))
+	}
+
+	return values, nil
 }
 
 func (d *EtcdStateDriver) ClearState(key string) error {

drivers/ovsdriver_test.go

@@ -66,10 +66,6 @@ func (d *testOvsStateDriver) Read(key string) ([]byte, error) {
 	return []byte{}, &core.Error{Desc: "Shouldn't be called!"}
 }
 
-func (d *testOvsStateDriver) ReadRecursive(baseKey string) ([]string, error) {
-	return []string{}, &core.Error{Desc: "Shouldn't be called!"}
-}
-
 func (d *testOvsStateDriver) ClearState(key string) error {
 	return nil
 }

drivers/ovsendpointstate.go

@@ -18,7 +18,6 @@ package drivers
 import (
 	"encoding/json"
 	"fmt"
-	"log"
 
 	"github.com/contiv/netplugin/core"
 )
@@ -52,33 +51,21 @@ func (s *OvsCfgEndpointState) Clear() error {
 	return s.StateDriver.ClearState(key)
 }
 
-func (s *OvsCfgEndpointState) Unmarshal(value string) error {
-	return json.Unmarshal([]byte(value), s)
-}
-
-func (s *OvsCfgEndpointState) Marshal() (string, error) {
-	bytes, err := json.Marshal(s)
-	return string(bytes[:]), err
-}
-
-func ReadAllEpsCfg(sd *core.StateDriver) (epState []OvsCfgEndpointState, err error) {
-
-	keys, err := (*sd).ReadRecursive(EP_CFG_PATH_PREFIX)
+func ReadAllOvsCfgEndpoints(d core.StateDriver) ([]*OvsCfgEndpointState, error) {
+	values := []*OvsCfgEndpointState{}
+	byteValues, err := ReadAll(d, EP_CFG_PATH_PREFIX)
 	if err != nil {
-		return
+		return nil, err
 	}
-
-	epState = make([]OvsCfgEndpointState, len(keys))
-	for idx, key := range keys {
-		err = (*sd).ReadState(key, &epState[idx], json.Unmarshal)
+	for _, byteValue := range byteValues {
+		value := &OvsCfgEndpointState{StateDriver: d}
+		err = json.Unmarshal(byteValue, value)
 		if err != nil {
-			log.Printf("error '%s' reading oper state of key %s \n",
-				err, key)
-			return
+			return nil, err
 		}
+		values = append(values, value)
 	}
-
-	return
+	return values, nil
 }
 
 type OvsOperEndpointState struct {
@@ -103,26 +90,6 @@ func (s *OvsOperEndpointState) Read(id string) error {
 	return s.StateDriver.ReadState(key, s, json.Unmarshal)
 }
 
-func ReadAllEpsOper(sd core.StateDriver) (epState []OvsOperEndpointState, err error) {
-
-	keys, err := sd.ReadRecursive(EP_OPER_PATH_PREFIX)
-	if err != nil {
-		return
-	}
-
-	epState = make([]OvsOperEndpointState, len(keys))
-	for idx, key := range keys {
-		err = sd.ReadState(key, &epState[idx], json.Unmarshal)
-		if err != nil {
-			log.Printf("error '%s' reading oper state of key %s \n",
-				err, key)
-			return
-		}
-	}
-
-	return
-}
-
 func (s *OvsOperEndpointState) Clear() error {
 	key := fmt.Sprintf(EP_OPER_PATH, s.Id)
 	return s.StateDriver.ClearState(key)

drivers/ovsendpointstate_test.go

@@ -48,10 +48,6 @@ func (d *testEpStateDriver) Read(key string) ([]byte, error) {
 	return []byte{}, &core.Error{Desc: "Shouldn't be called!"}
 }
 
-func (d *testEpStateDriver) ReadRecursive(baseKey string) ([]string, error) {
-	return []string{}, &core.Error{Desc: "Shouldn't be called!"}
-}
-
 func (d *testEpStateDriver) validateKey(key string) error {
 	if key != epCfgKey && key != epOperKey {
 		return &core.Error{Desc: fmt.Sprintf("Unexpected key. recvd: %s expected: %s or %s ",

drivers/ovsnetworkstate.go

@@ -63,6 +63,23 @@ func (s *OvsCfgNetworkState) Read(id string) error {
 	return s.StateDriver.ReadState(key, s, json.Unmarshal)
 }
 
+func ReadAllOvsCfgNetworks(d core.StateDriver) ([]*OvsCfgNetworkState, error) {
+	values := []*OvsCfgNetworkState{}
+	byteValues, err := ReadAll(d, NW_CFG_PATH_PREFIX)
+	if err != nil {
+		return nil, err
+	}
+	for _, byteValue := range byteValues {
+		value := &OvsCfgNetworkState{StateDriver: d}
+		err = json.Unmarshal(byteValue, value)
+		if err != nil {
+			return nil, err
+		}
+		values = append(values, value)
+	}
+	return values, nil
+}
+
 func (s *OvsCfgNetworkState) Clear() error {
 	key := fmt.Sprintf(NW_CFG_PATH, s.Id)
 	return s.StateDriver.ClearState(key)

drivers/ovsnetworkstate_test.go

@@ -47,10 +47,6 @@ func (d *testNwStateDriver) Read(key string) ([]byte, error) {
 	return []byte{}, &core.Error{Desc: "Shouldn't be called!"}
 }
 
-func (d *testNwStateDriver) ReadRecursive(baseKey string) ([]string, error) {
-	return []string{}, &core.Error{Desc: "Shouldn't be called!"}
-}
-
 func (d *testNwStateDriver) validateKey(key string) error {
 	if key != nwCfgKey {
 		return &core.Error{Desc: fmt.Sprintf("Unexpected key. "+

gstate/gstate.go

@@ -25,6 +25,7 @@ import (
 	"github.com/jainvipin/bitset"
 
 	"github.com/contiv/netplugin/core"
+	"github.com/contiv/netplugin/drivers"
 	"github.com/contiv/netplugin/netutils"
 )
 
@@ -58,13 +59,15 @@ type DeployParams struct {
 
 // global state of the network plugin
 type Cfg struct {
-	Version string
-	Tenant  string
-	Auto    AutoParams
-	Deploy  DeployParams
+	StateDriver core.StateDriver
+	Version     string
+	Tenant      string
+	Auto        AutoParams
+	Deploy      DeployParams
 }
 
 type Oper struct {
+	StateDriver     core.StateDriver
 	Tenant          string
 	DefaultNetType  string
 	SubnetPool      string
@@ -77,20 +80,6 @@ type Oper struct {
 	FreeVxlans      bitset.BitSet
 }
 
-func (gc *Cfg) UnMarshal(data string) error {
-	err := json.Unmarshal([]byte(data), &gc)
-	if err != nil {
-		return err
-	}
-
-	return nil
-}
-
-func (gc *Cfg) Marshal() (string, error) {
-	b, err := json.Marshal(gc)
-	return string(b[:]), err
-}
-
 func (gc *Cfg) Dump() error {
 	log.Printf("Global State %v \n", gc)
 	return nil
@@ -144,64 +133,51 @@ func Parse(configBytes []byte) (*Cfg, error) {
 	return &gc, err
 }
 
-func (gc *Cfg) Update(d core.StateDriver) error {
+func (gc *Cfg) Write() error {
 	key := fmt.Sprintf(CFG_GLOBAL_PATH, gc.Tenant)
-	value, err := json.Marshal(gc)
-	if err != nil {
-		return err
-	}
-
-	return d.Write(key, value)
+	return gc.StateDriver.WriteState(key, gc, json.Marshal)
 }
 
-func (gc *Cfg) Read(d core.StateDriver, tenant string) error {
+func (gc *Cfg) Read(tenant string) error {
 	key := fmt.Sprintf(CFG_GLOBAL_PATH, tenant)
-	value, err := d.Read(key)
+	return gc.StateDriver.ReadState(key, gc, json.Unmarshal)
+}
+
+func ReadAllGlobalCfg(d core.StateDriver) ([]*Cfg, error) {
+	values := []*Cfg{}
+	byteValues, err := drivers.ReadAll(d, CFG_GLOBAL_PREFIX)
 	if err != nil {
-		return err
+		return nil, err
 	}
-	err = json.Unmarshal(value, &gc)
-	if err != nil {
-		return err
+	for _, byteValue := range byteValues {
+		value := &Cfg{StateDriver: d}
+		err = json.Unmarshal(byteValue, value)
+		if err != nil {
+			return nil, err
+		}
+		values = append(values, value)
 	}
-
-	return nil
+	return values, nil
 }
 
-func (gc *Cfg) Clear(d core.StateDriver) error {
+func (gc *Cfg) Clear() error {
 	key := fmt.Sprintf(CFG_GLOBAL_PATH, gc.Tenant)
-
-	return d.ClearState(key)
+	return gc.StateDriver.ClearState(key)
 }
 
-func (g *Oper) Update(d core.StateDriver) error {
+func (g *Oper) Write() error {
 	key := fmt.Sprintf(OPER_GLOBAL_PATH, g.Tenant)
-	value, err := json.Marshal(g)
-	if err != nil {
-		return err
-	}
-
-	return d.Write(key, value)
+	return g.StateDriver.WriteState(key, g, json.Marshal)
 }
 
-func (g *Oper) Read(d core.StateDriver, tenant string) error {
+func (g *Oper) Read(tenant string) error {
 	key := fmt.Sprintf(OPER_GLOBAL_PATH, tenant)
-	value, err := d.Read(key)
-	if err != nil {
-		return err
-	}
-	err = json.Unmarshal(value, &g)
-	if err != nil {
-		return err
-	}
-
-	return nil
+	return g.StateDriver.ReadState(key, g, json.Unmarshal)
 }
 
-func (g *Oper) Clear(d core.StateDriver) error {
+func (g *Oper) Clear() error {
 	key := fmt.Sprintf(OPER_GLOBAL_PATH, g.Tenant)
-
-	return d.ClearState(key)
+	return g.StateDriver.ClearState(key)
 }
 
 func (g *Oper) initVxlanBitset(vxlans string, vlans string,
@@ -405,6 +381,7 @@ func (gc *Cfg) Process() (*Oper, error) {
 	}
 
 	g := &Oper{
+		StateDriver:    gc.StateDriver,
 		Tenant:         gc.Tenant,
 		SubnetLen:      gc.Auto.SubnetLen,
 		DefaultNetType: gc.Deploy.DefaultNetType,