Do replay when watching topo

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/onosproject/onos-api/go v0.7.0
 	github.com/onosproject/onos-lib-go v0.7.0
 	github.com/onosproject/onos-test v0.6.4
-	github.com/onosproject/onos-topo v0.7.0
+	github.com/onosproject/onos-topo v0.7.0 // indirect
 	github.com/openconfig/gnmi v0.0.0-20200617225440-d2b4e6a45802
 	github.com/openconfig/goyang v0.2.1
 	github.com/openconfig/ygot v0.8.12

go.sum

@@ -712,6 +712,7 @@ go.mongodb.org/mongo-driver v1.0.3/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qL
 go.mongodb.org/mongo-driver v1.1.1/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.mongodb.org/mongo-driver v1.1.2/go.mod h1:u7ryQJ+DOzQmeO7zB6MHyr8jkEQvC8vH7qLUO4lqsUM=
 go.opencensus.io v0.21.0/go.mod h1:mSImk1erAIZhrmZN+AvHh14ztQfjbGwt4TtuofqLduU=
+go.opencensus.io v0.22.0 h1:C9hSCOW830chIVkdja34wa6Ky+IzWllkUinR+BtRZd4=
 go.opencensus.io v0.22.0/go.mod h1:+kGneAE2xo2IficOXnaByMWTGM9T73dGwxeWcUqIpI8=
 go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
 go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
@@ -721,6 +722,7 @@ go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/
 go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
 go.uber.org/multierr v1.4.0 h1:f3WCSC2KzAcBXGATIxAB1E2XuCpNU255wNKZ505qi3E=
 go.uber.org/multierr v1.4.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4=
+go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee h1:0mgffUl7nfd+FpvXMVz4IDEaUSmT1ysygQC7qYo7sG4=
 go.uber.org/tools v0.0.0-20190618225709-2cfd321de3ee/go.mod h1:vJERXedbb3MVM5f9Ejo0C68/HhF8uaILCdgjnY+goOA=
 go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.13.0 h1:nR6NoDBgAf67s68NhaXbsojM+2gxp3S1hWkHDl27pVU=
@@ -755,6 +757,7 @@ golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvx
 golang.org/x/lint v0.0.0-20190301231843-5614ed5bae6f/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
 golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/lint v0.0.0-20190409202823-959b441ac422/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
+golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
 golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
 golang.org/x/mobile v0.0.0-20190312151609-d3739f865fa6/go.mod h1:z+o9i4GpDbdi3rU15maQ/Ox0txvL9dWGYEHz965HBQE=
 golang.org/x/mobile v0.0.0-20190806162312-597adff16ade/go.mod h1:AlhUtkH4DA4asiFC5RgK7ZKmauvtkAVcy9L0epCzlWo=
@@ -864,9 +867,11 @@ golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtn
 golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20200113040837-eac381796e91/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/tools v0.0.0-20200522201501-cb1345f3a375 h1:SjQ2+AKWgZLc1xej6WSzL+Dfs5Uyd5xcZH1mGC411IA=
 golang.org/x/tools v0.0.0-20200522201501-cb1345f3a375/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 gonum.org/v1/gonum v0.0.0-20190331200053-3d26580ed485/go.mod h1:2ltnJ7xHfj0zHS40VVPYEAAMTa3ZGguvHGBSJeRWqE0=
 gonum.org/v1/netlib v0.0.0-20190313105609-8cb42192e0e0/go.mod h1:wa6Ws7BG/ESfp6dHfk7C6KdzKA7wR7u/rKwOGE66zvw=
@@ -877,6 +882,7 @@ google.golang.org/api v0.7.0/go.mod h1:WtwebWUNSVBH/HAw79HIFXZNqEvBhG+Ra+ax0hx3E
 google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
 google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
 google.golang.org/appengine v1.5.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
+google.golang.org/appengine v1.6.1 h1:QzqyMA1tlu6CgqCDUtU9V+ZKhLFT2dkJuANu5QaxI3I=
 google.golang.org/appengine v1.6.1/go.mod h1:i06prIuMbXzDqacNJfV5OdTW448YApPu5ww/cMBSeb0=
 google.golang.org/cloud v0.0.0-20151119220103-975617b05ea8/go.mod h1:0H1ncTHf11KCFhTc/+EFRbzSCOZx+VUbRMk55Yv5MYk=
 google.golang.org/genproto v0.0.0-20180817151627-c66870c02cf8/go.mod h1:JiN7NxoALGmiZfu7CAH4rXhgtRTLTxftemlI0sWmxmc=
@@ -961,6 +967,7 @@ honnef.co/go/tools v0.0.0-20190106161140-3f1c8253044a/go.mod h1:rf3lG4BRIbNafJWh
 honnef.co/go/tools v0.0.0-20190418001031-e561f6794a2a/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
 honnef.co/go/tools v0.0.1-2019.2.2/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
+honnef.co/go/tools v0.0.1-2019.2.3 h1:3JgtbtFHMiCmsznwGVTUWbgGov+pVqnlf1dEJTNAXeM=
 honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt0JzvZhAg=
 k8s.io/api v0.17.2/go.mod h1:BS9fjjLc4CMuqfSO8vgbHPKMt5+SF0ET6u/RVDihTo4=
 k8s.io/api v0.17.3 h1:XAm3PZp3wnEdzekNkcmj/9Y1zdmQYJ1I4GKSBBZ8aG0=

pkg/controller/change/device/controller_test.go

@@ -48,7 +48,9 @@ import (
 
 const (
 	device1     = device.ID("device-1")
+	device1Addr = "device-1:5150"
 	device2     = device.ID("device-2")
+	device2Addr = "device-2:5150"
 	dcDevice    = device.ID("disconnected")
 	v1          = "1.0.0"
 	stratumType = "Stratum"
@@ -475,7 +477,10 @@ func newStores(t *testing.T) (devicestore.Store, devicechanges.Store) {
 
 	devices := map[topodevice.ID]*topodevice.Device{
 		topodevice.ID(device1): {
-			ID: topodevice.ID(device1),
+			ID:      topodevice.ID(device1),
+			Version: v1,
+			Type:    stratumType,
+			Address: device1Addr,
 			Protocols: []*topo.ProtocolState{
 				{
 					Protocol:          topo.Protocol_GNMI,
@@ -485,7 +490,10 @@ func newStores(t *testing.T) (devicestore.Store, devicechanges.Store) {
 			},
 		},
 		topodevice.ID(device2): {
-			ID: topodevice.ID(device2),
+			ID:      topodevice.ID(device2),
+			Version: v1,
+			Type:    stratumType,
+			Address: device2Addr,
 			Protocols: []*topo.ProtocolState{
 				{
 					Protocol:          topo.Protocol_GNMI,
@@ -495,7 +503,10 @@ func newStores(t *testing.T) (devicestore.Store, devicechanges.Store) {
 			},
 		},
 		topodevice.ID(dcDevice): {
-			ID: topodevice.ID(dcDevice),
+			ID:      topodevice.ID(dcDevice),
+			Version: v1,
+			Type:    stratumType,
+			Address: "",
 			Protocols: []*topo.ProtocolState{
 				{
 					Protocol:          topo.Protocol_GNMI,

pkg/controller/change/network/controller_test.go

@@ -16,6 +16,7 @@ package network
 
 import (
 	"context"
+	"fmt"
 	"github.com/golang/mock/gomock"
 	types "github.com/onosproject/onos-api/go/onos/config"
 	"github.com/onosproject/onos-api/go/onos/config/change"
@@ -33,8 +34,10 @@ import (
 )
 
 const (
-	device1 = device.ID("device-1")
-	device2 = device.ID("device-2")
+	device1     = device.ID("device-1")
+	device2     = device.ID("device-2")
+	v1          = "1.0.0"
+	stratumType = "Stratum"
 )
 
 const (
@@ -321,7 +324,10 @@ func newStores(t *testing.T) (networkchanges.Store, devicechanges.Store, devices
 	client.EXPECT().Get(gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, request *topo.GetRequest) (*topo.GetResponse, error) {
 		return &topo.GetResponse{
 			Object: devicetopo.ToObject(&devicetopo.Device{
-				ID: devicetopo.ID(request.ID),
+				ID:      devicetopo.ID(request.ID),
+				Version: v1,
+				Type:    stratumType,
+				Address: fmt.Sprintf("%s:5150", request.ID),
 				Protocols: []*topo.ProtocolState{
 					{
 						Protocol:          topo.Protocol_GNMI,

pkg/controller/change/network/watcher_test.go

@@ -16,6 +16,7 @@ package network
 
 import (
 	"context"
+	"fmt"
 	"github.com/golang/mock/gomock"
 	types "github.com/onosproject/onos-api/go/onos/config"
 	devicechange "github.com/onosproject/onos-api/go/onos/config/change/device"
@@ -58,8 +59,8 @@ func TestNetworkWatcher(t *testing.T) {
 		Changes: []*devicechange.Change{
 			{
 
-				DeviceID:      "device-1",
-				DeviceVersion: "1.0.0",
+				DeviceID:      device1,
+				DeviceVersion: v1,
 				Values: []*devicechange.ChangeValue{
 					{
 						Path: "foo",
@@ -78,8 +79,8 @@ func TestNetworkWatcher(t *testing.T) {
 				},
 			},
 			{
-				DeviceID:      "device-2",
-				DeviceVersion: "1.0.0",
+				DeviceID:      device2,
+				DeviceVersion: v1,
 				Values: []*devicechange.ChangeValue{
 					{
 						Path: "baz",
@@ -107,8 +108,8 @@ func TestNetworkWatcher(t *testing.T) {
 		ID: "change-2",
 		Changes: []*devicechange.Change{
 			{
-				DeviceID:      "device-1",
-				DeviceVersion: "1.0.0",
+				DeviceID:      device1,
+				DeviceVersion: v1,
 				Values: []*devicechange.ChangeValue{
 					{
 						Path:    "foo",
@@ -145,14 +146,14 @@ func TestDeviceWatcher(t *testing.T) {
 	t.Log("Testing")
 	cachedDevices := []*cache.Info{
 		{
-			DeviceID: "device-1",
-			Type:     "DeviceSim",
-			Version:  "1.0.0",
+			DeviceID: device1,
+			Type:     v1,
+			Version:  v1,
 		},
 		{
-			DeviceID: "device-2",
-			Type:     "DeviceSim",
-			Version:  "1.0.0",
+			DeviceID: device2,
+			Type:     v1,
+			Version:  v1,
 		},
 	}
 
@@ -163,9 +164,10 @@ func TestDeviceWatcher(t *testing.T) {
 			Event: topo.Event{
 				Type: topo.EventType_NONE,
 				Object: *devicetopo.ToObject(&devicetopo.Device{
-					ID:      "device-1",
-					Type:    "DeviceSim",
-					Version: "1.0.0",
+					ID:      devicetopo.ID(device1),
+					Type:    stratumType,
+					Version: v1,
+					Address: fmt.Sprintf("%s:1234", device1),
 					Protocols: []*topo.ProtocolState{
 						{
 							Protocol:          topo.Protocol_GNMI,
@@ -180,9 +182,10 @@ func TestDeviceWatcher(t *testing.T) {
 			Event: topo.Event{
 				Type: topo.EventType_NONE,
 				Object: *devicetopo.ToObject(&devicetopo.Device{
-					ID:      "device-2",
-					Type:    "DeviceSim",
-					Version: "1.0.0",
+					ID:      devicetopo.ID(device2),
+					Type:    stratumType,
+					Version: v1,
+					Address: fmt.Sprintf("%s:1234", device2),
 					Protocols: []*topo.ProtocolState{
 						{
 							Protocol:          topo.Protocol_GNMI,

pkg/device/device.go

@@ -17,6 +17,7 @@
 package device
 
 import (
+	"fmt"
 	"github.com/onosproject/onos-api/go/onos/topo"
 	"time"
 )
@@ -155,16 +156,31 @@ func ToObject(device *Device) *topo.Object {
 }
 
 // ToDevice converts local device structure to topology object entity
-func ToDevice(object *topo.Object) *Device {
+func ToDevice(object *topo.Object) (*Device, error) {
+	if object.Type != topo.Object_ENTITY {
+		return nil, fmt.Errorf("object is not a topo entity %v+", object)
+	}
+	version, ok := object.Attributes[topo.Version]
+	if !ok {
+		return nil, fmt.Errorf("topo entity %s must have 'version' attribute to work with onos-config", object.ID)
+	}
+	address, ok := object.Attributes[topo.Address]
+	if !ok {
+		return nil, fmt.Errorf("topo entity %s must have 'address' attribute to work with onos-config", object.ID)
+	}
+	typeKindID := Type(object.GetEntity().KindID)
+	if len(typeKindID) == 0 {
+		return nil, fmt.Errorf("topo entity %s must have a 'kindid' to work with onos-config", object.ID)
+	}
 	d := &Device{
 		ID:        ID(object.ID),
 		Revision:  object.Revision,
 		Protocols: object.GetEntity().Protocols,
-		Type:      Type(object.GetEntity().KindID),
+		Type:      typeKindID,
 		Role:      Role(object.Attributes[topo.Role]),
-		Address:   object.Attributes[topo.Address],
+		Address:   address,
 		Target:    object.Attributes[topo.Target],
-		Version:   object.Attributes[topo.Version],
+		Version:   version,
 		TLS: TLSConfig{
 			Plain:    object.Attributes[topo.TLSPlain] == "true",
 			Insecure: object.Attributes[topo.TLSInsecure] == "true",
@@ -174,5 +190,5 @@ func ToDevice(object *topo.Object) *Device {
 		},
 		Attributes: object.Attributes,
 	}
-	return d
+	return d, nil
 }

pkg/southbound/synchronizer/device_update.go

@@ -40,7 +40,7 @@ func (s *Session) getTermPerDevice(device *topodevice.Device) (int, error) {
 
 func (s *Session) updateDevice(connectivity topo.ConnectivityState, channel topo.ChannelState,
 	service topo.ServiceState) error {
-	log.Info("Update device %s state", s.device.ID)
+	log.Infof("Update device %s state", s.device.ID)
 
 	id := s.device.ID
 	topoDevice, err := s.deviceStore.Get(id)

pkg/southbound/synchronizer/device_update_test.go

@@ -31,6 +31,7 @@ import (
 const (
 	device1        = "device1"
 	deviceVersion1 = "1.0.0"
+	stratumType    = "Stratum"
 )
 
 type AllMocks struct {
@@ -67,6 +68,7 @@ func TestUpdateDisconnectedDevice(t *testing.T) {
 		Revision:   1,
 		Address:    "device1:1234",
 		Version:    deviceVersion1,
+		Type:       stratumType,
 		Attributes: make(map[string]string),
 	}
 
@@ -106,6 +108,7 @@ func TestUpdateConnectedDevice(t *testing.T) {
 		Revision:   1,
 		Address:    "device1:1234",
 		Version:    deviceVersion1,
+		Type:       stratumType,
 		Attributes: make(map[string]string),
 	}
 

pkg/southbound/synchronizer/session_manager.go

@@ -137,12 +137,13 @@ func WithDeviceChangeStore(deviceChangeStore device.Store) func(*SessionManager)
 // Start starts session manager
 func (sm *SessionManager) Start() error {
 	log.Info("Session manager started")
+	go sm.processDeviceEvents(sm.topoChannel)
+
 	err := sm.deviceStore.Watch(sm.topoChannel)
 	if err != nil {
 		return err
 	}
 
-	go sm.processDeviceEvents(sm.topoChannel)
 	return nil
 }