Merge "[VOL-3986] Correctly serializing GemPortNetworkCtpPerformanceMonitoringHistoryData ME in GetResponse"
diff --git a/internal/bbr/devices/olt.go b/internal/bbr/devices/olt.go
index b65a174..b40bf58 100644
--- a/internal/bbr/devices/olt.go
+++ b/internal/bbr/devices/olt.go
@@ -48,6 +48,18 @@
CompletedOnus int // Number of ONUs that have received a DHCPAck
}
+type MockStream struct {
+ grpc.ServerStream
+}
+
+func (*MockStream) Send(ind *openolt.Indication) error {
+ return nil
+}
+
+func (*MockStream) Context() context.Context {
+ return context.Background()
+}
+
// trigger an enable call and start the same listeners on the gRPC stream that VOLTHA would create
// this method is blocking
func (o *OltMock) Start() {
@@ -211,7 +223,11 @@
}
ctx, cancel := context.WithCancel(context.TODO())
- go onu.ProcessOnuMessages(ctx, nil, client)
+ // NOTE we need to create a fake stream for ProcessOnuMessages
+ // as it listen on the context to cancel the loop
+ // In the BBR case it's not used for anything else
+ mockStream := MockStream{}
+ go onu.ProcessOnuMessages(ctx, &mockStream, client)
go func() {
diff --git a/internal/bbsim/devices/olt.go b/internal/bbsim/devices/olt.go
index 6ad068d..1c4e099 100644
--- a/internal/bbsim/devices/olt.go
+++ b/internal/bbsim/devices/olt.go
@@ -413,19 +413,20 @@
// new ones
o.Lock()
if o.enableContext != nil && o.enableContextCancel != nil {
- oltLogger.Info("This is an OLT reboot")
+ oltLogger.Info("This is an OLT reboot or a reconcile")
o.enableContextCancel()
rebootFlag = true
+ time.Sleep(1 * time.Second)
}
o.enableContext, o.enableContextCancel = context.WithCancel(context.TODO())
o.Unlock()
wg := sync.WaitGroup{}
- wg.Add(3)
o.OpenoltStream = stream
// create Go routine to process all OLT events
+ wg.Add(1)
go o.processOltMessages(o.enableContext, stream, &wg)
// enable the OLT
@@ -461,6 +462,15 @@
}
o.channel <- msg
}
+ // when the enableContext was canceled the ONUs stopped listening on the channel
+ for _, onu := range pon.Onus {
+ go onu.ProcessOnuMessages(o.enableContext, stream, nil)
+
+ // update the stream on all the services
+ for _, service := range onu.Services {
+ service.UpdateStream(stream)
+ }
+ }
}
} else {
@@ -482,18 +492,22 @@
}
}
- oltLogger.Debug("Enable OLT Done")
-
if !o.enablePerf {
// Start a go routine to send periodic port stats to openolt adapter
- go o.periodicPortStats(o.enableContext)
+ wg.Add(1)
+ go o.periodicPortStats(o.enableContext, &wg, stream)
}
wg.Wait()
+ oltLogger.WithFields(log.Fields{
+ "stream": stream,
+ }).Debug("OpenOLT Stream closed")
}
-func (o *OltDevice) periodicPortStats(ctx context.Context) {
+func (o *OltDevice) periodicPortStats(ctx context.Context, wg *sync.WaitGroup, stream openolt.Openolt_EnableIndicationServer) {
var portStats *openolt.PortStatistics
+
+loop:
for {
select {
case <-time.After(time.Duration(o.PortStatsInterval) * time.Second):
@@ -504,7 +518,7 @@
incrementStat = false
}
portStats, port.PacketCount = getPortStats(port.PacketCount, incrementStat)
- o.sendPortStatsIndication(portStats, port.ID, port.Type)
+ o.sendPortStatsIndication(portStats, port.ID, port.Type, stream)
}
// send PON port stats
@@ -515,14 +529,14 @@
incrementStat = false
}
portStats, port.PacketCount = getPortStats(port.PacketCount, incrementStat)
- o.sendPortStatsIndication(portStats, port.ID, port.Type)
+ o.sendPortStatsIndication(portStats, port.ID, port.Type, stream)
}
case <-ctx.Done():
- log.Debug("Stop sending port stats")
- return
+ oltLogger.Debug("Stop sending port stats")
+ break loop
}
-
}
+ wg.Done()
}
// Helpers method
@@ -695,7 +709,7 @@
}).Debug("Sent Indication_IntfOperInd for PON")
}
-func (o *OltDevice) sendPortStatsIndication(stats *openolt.PortStatistics, portID uint32, portType string) {
+func (o *OltDevice) sendPortStatsIndication(stats *openolt.PortStatistics, portID uint32, portType string, stream openolt.Openolt_EnableIndicationServer) {
if o.InternalState.Current() == "enabled" {
oltLogger.WithFields(log.Fields{
"Type": portType,
@@ -705,7 +719,7 @@
data := &openolt.Indication_PortStats{
PortStats: stats,
}
- stream := o.OpenoltStream
+
if err := stream.Send(&openolt.Indication{Data: data}); err != nil {
oltLogger.Errorf("Failed to send PortStats: %v", err)
return
@@ -714,8 +728,10 @@
}
// processOltMessages handles messages received over the OpenOLT interface
-func (o *OltDevice) processOltMessages(ctx context.Context, stream openolt.Openolt_EnableIndicationServer, wg *sync.WaitGroup) {
- oltLogger.Debug("Starting OLT Indication Channel")
+func (o *OltDevice) processOltMessages(ctx context.Context, stream types.Stream, wg *sync.WaitGroup) {
+ oltLogger.WithFields(log.Fields{
+ "stream": stream,
+ }).Debug("Starting OLT Indication Channel")
ch := o.channel
loop:
@@ -724,9 +740,15 @@
case <-ctx.Done():
oltLogger.Debug("OLT Indication processing canceled via context")
break loop
+ case <-stream.Context().Done():
+ oltLogger.Debug("OLT Indication processing canceled via stream context")
+ break loop
case message, ok := <-ch:
- if !ok || ctx.Err() != nil {
- oltLogger.Debug("OLT Indication processing canceled via closed channel")
+ if !ok {
+ if ctx.Err() != nil {
+ oltLogger.WithField("err", ctx.Err()).Error("OLT EnableContext error")
+ }
+ oltLogger.Warn("OLT Indication processing canceled via closed channel")
break loop
}
@@ -788,7 +810,9 @@
}
}
wg.Done()
- oltLogger.Warn("Stopped handling OLT Indication Channel")
+ oltLogger.WithFields(log.Fields{
+ "stream": stream,
+ }).Warn("Stopped handling OLT Indication Channel")
}
// returns an ONU with a given Serial Number
diff --git a/internal/bbsim/devices/onu.go b/internal/bbsim/devices/onu.go
index f9e2e32..466f383 100644
--- a/internal/bbsim/devices/onu.go
+++ b/internal/bbsim/devices/onu.go
@@ -329,6 +329,7 @@
"onuID": o.ID,
"onuSN": o.Sn(),
"ponPort": o.PonPortID,
+ "stream": stream,
}).Debug("Starting ONU Indication Channel")
loop:
@@ -338,14 +339,20 @@
onuLogger.WithFields(log.Fields{
"onuID": o.ID,
"onuSN": o.Sn(),
- }).Tracef("ONU message handling canceled via context")
+ }).Debug("ONU message handling canceled via context")
+ break loop
+ case <-stream.Context().Done():
+ onuLogger.WithFields(log.Fields{
+ "onuID": o.ID,
+ "onuSN": o.Sn(),
+ }).Debug("ONU message handling canceled via stream context")
break loop
case message, ok := <-o.Channel:
if !ok || ctx.Err() != nil {
onuLogger.WithFields(log.Fields{
"onuID": o.ID,
"onuSN": o.Sn(),
- }).Tracef("ONU message handling canceled via channel close")
+ }).Debug("ONU message handling canceled via channel close")
break loop
}
onuLogger.WithFields(log.Fields{
@@ -477,8 +484,9 @@
}
}
onuLogger.WithFields(log.Fields{
- "onuID": o.ID,
- "onuSN": o.Sn(),
+ "onuID": o.ID,
+ "onuSN": o.Sn(),
+ "stream": stream,
}).Debug("Stopped handling ONU Indication Channel")
}
@@ -741,7 +749,8 @@
case omci.MibUploadNextRequestType:
responsePkt, _ = omcilib.CreateMibUploadNextResponse(msg.OmciPkt, msg.OmciMsg, o.MibDataSync)
case omci.GetRequestType:
- responsePkt, _ = omcilib.CreateGetResponse(msg.OmciPkt, msg.OmciMsg, o.SerialNumber, o.MibDataSync, o.ActiveImageEntityId, o.CommittedImageEntityId)
+ onuDown := o.OperState.Current() == "down"
+ responsePkt, _ = omcilib.CreateGetResponse(msg.OmciPkt, msg.OmciMsg, o.SerialNumber, o.MibDataSync, o.ActiveImageEntityId, o.CommittedImageEntityId, onuDown)
case omci.SetRequestType:
success := true
msgObj, _ := omcilib.ParseSetRequest(msg.OmciPkt)
@@ -759,6 +768,23 @@
raiseOMCIAlarm := false
if adminState == 1 {
raiseOMCIAlarm = true
+ // set the OperState to disabled
+ if err := o.OperState.Event(OnuTxDisable); err != nil {
+ onuLogger.WithFields(log.Fields{
+ "OnuId": o.ID,
+ "IntfId": o.PonPortID,
+ "OnuSn": o.Sn(),
+ }).Errorf("Cannot change ONU OperState to down: %s", err.Error())
+ }
+ } else {
+ // set the OperState to enabled
+ if err := o.OperState.Event(OnuTxEnable); err != nil {
+ onuLogger.WithFields(log.Fields{
+ "OnuId": o.ID,
+ "IntfId": o.PonPortID,
+ "OnuSn": o.Sn(),
+ }).Errorf("Cannot change ONU OperState to up: %s", err.Error())
+ }
}
msg := bbsim.Message{
Type: bbsim.UniStatusAlarm,
diff --git a/internal/bbsim/devices/onu_indications_test.go b/internal/bbsim/devices/onu_indications_test.go
index 04af678..5cc3533 100644
--- a/internal/bbsim/devices/onu_indications_test.go
+++ b/internal/bbsim/devices/onu_indications_test.go
@@ -47,6 +47,10 @@
return nil
}
+func (s *mockStream) Context() context.Context {
+ return context.Background()
+}
+
// test that we're sending a Discovery indication to VOLTHA
func Test_Onu_DiscoverIndication_send_on_discovery(t *testing.T) {
onu := createTestOnu()
diff --git a/internal/bbsim/devices/service_test.go b/internal/bbsim/devices/service_test.go
index a71a50c..ec081db 100644
--- a/internal/bbsim/devices/service_test.go
+++ b/internal/bbsim/devices/service_test.go
@@ -46,8 +46,9 @@
s.HandlePacketsCallCount = s.HandlePacketsCallCount + 1
}
-func (s *mockService) Initialize(stream types.Stream) {}
-func (s *mockService) Disable() {}
+func (s *mockService) Initialize(stream types.Stream) {}
+func (s *mockService) UpdateStream(stream types.Stream) {}
+func (s *mockService) Disable() {}
func createTestService(needsEapol bool, needsDchp bool) (*Service, error) {
diff --git a/internal/bbsim/devices/services.go b/internal/bbsim/devices/services.go
index 4391414..8c4ddfd 100644
--- a/internal/bbsim/devices/services.go
+++ b/internal/bbsim/devices/services.go
@@ -45,6 +45,7 @@
HandleDhcp(pbit uint8, cTag int) // Sends the DHCPDiscover packet
Initialize(stream bbsimTypes.Stream)
+ UpdateStream(stream bbsimTypes.Stream)
Disable()
}
@@ -115,7 +116,7 @@
serviceLogger.Fatal("initialize invoke with wrong arguments")
}
- service.Stream = stream
+ service.UpdateStream(stream)
service.PacketCh = make(chan bbsimTypes.OnuPacketMessage)
service.Channel = make(chan bbsimTypes.Message)
@@ -290,6 +291,10 @@
return &service, nil
}
+func (s *Service) UpdateStream(stream bbsimTypes.Stream) {
+ s.Stream = stream
+}
+
// HandleAuth is used to start EAPOL for a particular Service when the corresponding flow is received
func (s *Service) HandleAuth() {
diff --git a/internal/bbsim/types/interfaces.go b/internal/bbsim/types/interfaces.go
index 380af34..946d507 100644
--- a/internal/bbsim/types/interfaces.go
+++ b/internal/bbsim/types/interfaces.go
@@ -18,9 +18,11 @@
import (
"github.com/opencord/voltha-protos/v4/go/openolt"
+ "google.golang.org/grpc"
)
// represent a gRPC stream
type Stream interface {
Send(*openolt.Indication) error
+ grpc.ServerStream
}
diff --git a/internal/common/omci/get.go b/internal/common/omci/get.go
index efa768f..384fc0b 100644
--- a/internal/common/omci/get.go
+++ b/internal/common/omci/get.go
@@ -45,7 +45,7 @@
return msgObj, nil
}
-func CreateGetResponse(omciPkt gopacket.Packet, omciMsg *omci.OMCI, onuSn *openolt.SerialNumber, mds uint8, activeImageEntityId uint16, committedImageEntityId uint16) ([]byte, error) {
+func CreateGetResponse(omciPkt gopacket.Packet, omciMsg *omci.OMCI, onuSn *openolt.SerialNumber, mds uint8, activeImageEntityId uint16, committedImageEntityId uint16, onuDown bool) ([]byte, error) {
msgObj, err := ParseGetRequest(omciPkt)
@@ -70,9 +70,9 @@
case me.IpHostConfigDataClassID:
response = createIpHostResponse(msgObj.AttributeMask, msgObj.EntityInstance)
case me.UniGClassID:
- response = createUnigResponse(msgObj.AttributeMask, msgObj.EntityInstance)
+ response = createUnigResponse(msgObj.AttributeMask, msgObj.EntityInstance, onuDown)
case me.PhysicalPathTerminationPointEthernetUniClassID:
- response = createPptpResponse(msgObj.AttributeMask, msgObj.EntityInstance)
+ response = createPptpResponse(msgObj.AttributeMask, msgObj.EntityInstance, onuDown)
case me.AniGClassID:
response = createAnigResponse(msgObj.AttributeMask, msgObj.EntityInstance)
case me.OnuDataClassID:
@@ -260,13 +260,18 @@
}
}
-func createUnigResponse(attributeMask uint16, entityID uint16) *omci.GetResponse {
+func createUnigResponse(attributeMask uint16, entityID uint16, onuDown bool) *omci.GetResponse {
+ // Valid values for uni_admin_state are 0 (unlocks) and 1 (locks)
+ omciAdminState := 1
+ if !onuDown {
+ omciAdminState = 0
+ }
managedEntity, meErr := me.NewUniG(me.ParamData{
EntityID: entityID,
Attributes: me.AttributeValueMap{
"ManagedEntityId": entityID,
"Deprecated": 0,
- "AdministrativeState": 0,
+ "AdministrativeState": omciAdminState,
"ManagementCapability": 0,
"NonOmciManagementIdentifier": 1,
"RelayAgentOptions": 1,
@@ -289,7 +294,14 @@
}
}
-func createPptpResponse(attributeMask uint16, entityID uint16) *omci.GetResponse {
+func createPptpResponse(attributeMask uint16, entityID uint16, onuDown bool) *omci.GetResponse {
+ // Valid values for oper_state are 0 (enabled) and 1 (disabled)
+ // Valid values for uni_admin_state are 0 (unlocks) and 1 (locks)
+ onuAdminState := 1
+ if !onuDown {
+ onuAdminState = 0
+ }
+ onuOperState := onuAdminState // For now make the assumption that oper state reflects the admin state
managedEntity, meErr := me.NewPhysicalPathTerminationPointEthernetUni(me.ParamData{
EntityID: entityID,
Attributes: me.AttributeValueMap{
@@ -298,8 +310,8 @@
"SensedType": 0,
"AutoDetectionConfiguration": 0,
"EthernetLoopbackConfiguration": 0,
- "AdministrativeState": 0,
- "OperationalState": 0,
+ "AdministrativeState": onuAdminState,
+ "OperationalState": onuOperState,
"ConfigurationInd": 0,
"MaxFrameSize": 0,
"DteOrDceInd": 0,