VOL-3735 Retrieve port counters using the GetExtValue(SingleGetValue) rpc from the openolt adapter
Change-Id: I195f7c9fdc4b677c333ae9e4bb3547bb95312fdf
diff --git a/internal/pkg/core/device_handler.go b/internal/pkg/core/device_handler.go
index fa8dd52..ef1b179 100644
--- a/internal/pkg/core/device_handler.go
+++ b/internal/pkg/core/device_handler.go
@@ -45,6 +45,7 @@
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
"github.com/opencord/voltha-protos/v4/go/common"
+ "github.com/opencord/voltha-protos/v4/go/extension"
ic "github.com/opencord/voltha-protos/v4/go/inter_container"
of "github.com/opencord/voltha-protos/v4/go/openflow_13"
oop "github.com/opencord/voltha-protos/v4/go/openolt"
@@ -62,6 +63,7 @@
McastFlowOrGroupAdd = "McastFlowOrGroupAdd"
McastFlowOrGroupModify = "McastFlowOrGroupModify"
McastFlowOrGroupRemove = "McastFlowOrGroupRemove"
+ oltPortInfoTimeout = 3
)
//DeviceHandler will interact with the OLT device.
@@ -2479,3 +2481,79 @@
}
}
}
+
+func (dh *DeviceHandler) getOltPortCounters(ctx context.Context, oltPortInfo *extension.GetOltPortCounters) *extension.SingleGetValueResponse {
+
+ singleValResp := extension.SingleGetValueResponse{
+ Response: &extension.GetValueResponse{
+ Response: &extension.GetValueResponse_PortCoutners{
+ PortCoutners: &extension.GetOltPortCountersResponse{},
+ },
+ },
+ }
+
+ errResp := func(status extension.GetValueResponse_Status,
+ reason extension.GetValueResponse_ErrorReason) *extension.SingleGetValueResponse {
+ return &extension.SingleGetValueResponse{
+ Response: &extension.GetValueResponse{
+ Status: status,
+ ErrReason: reason,
+ },
+ }
+ }
+
+ if oltPortInfo.PortType != extension.GetOltPortCounters_Port_ETHERNET_NNI &&
+ oltPortInfo.PortType != extension.GetOltPortCounters_Port_PON_OLT {
+ //send error response
+ logger.Debugw(ctx, "getOltPortCounters invalid portType", log.Fields{"oltPortInfo": oltPortInfo.PortType})
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INVALID_PORT_TYPE)
+ }
+ statIndChn := make(chan bool, 1)
+ dh.portStats.RegisterForStatIndication(ctx, portStatsType, statIndChn, oltPortInfo.PortNo, oltPortInfo.PortType)
+ defer dh.portStats.DeRegisterFromStatIndication(ctx, portStatsType, statIndChn)
+ //request openOlt agent to send the the port statistics indication
+
+ go func() {
+ _, err := dh.Client.CollectStatistics(ctx, new(oop.Empty))
+ if err != nil {
+ logger.Errorw(ctx, "getOltPortCounters CollectStatistics failed ", log.Fields{"err": err})
+ }
+ }()
+ select {
+ case <-statIndChn:
+ //indication received for ports stats
+ logger.Debugw(ctx, "getOltPortCounters recvd statIndChn", log.Fields{"oltPortInfo": oltPortInfo})
+ case <-time.After(oltPortInfoTimeout * time.Second):
+ logger.Debugw(ctx, "getOltPortCounters timeout happened", log.Fields{"oltPortInfo": oltPortInfo})
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_TIMEOUT)
+ case <-ctx.Done():
+ logger.Debugw(ctx, "getOltPortCounters ctx Done ", log.Fields{"oltPortInfo": oltPortInfo})
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_TIMEOUT)
+ }
+ if oltPortInfo.PortType == extension.GetOltPortCounters_Port_ETHERNET_NNI {
+ //get nni stats
+ intfID := PortNoToIntfID(oltPortInfo.PortNo, voltha.Port_ETHERNET_NNI)
+ logger.Debugw(ctx, "getOltPortCounters intfID ", log.Fields{"intfID": intfID})
+ cmnni := dh.portStats.collectNNIMetrics(intfID)
+ if cmnni == nil {
+ //TODO define the error reason
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INTERNAL_ERROR)
+ }
+ dh.portStats.updateGetOltPortCountersResponse(ctx, &singleValResp, cmnni)
+ return &singleValResp
+
+ } else if oltPortInfo.PortType == extension.GetOltPortCounters_Port_PON_OLT {
+ // get pon stats
+ intfID := PortNoToIntfID(oltPortInfo.PortNo, voltha.Port_PON_OLT)
+ if val, ok := dh.activePorts.Load(intfID); ok && val == true {
+ cmpon := dh.portStats.collectPONMetrics(intfID)
+ if cmpon == nil {
+ //TODO define the error reason
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INTERNAL_ERROR)
+ }
+ dh.portStats.updateGetOltPortCountersResponse(ctx, &singleValResp, cmpon)
+ return &singleValResp
+ }
+ }
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INTERNAL_ERROR)
+}
diff --git a/internal/pkg/core/openolt.go b/internal/pkg/core/openolt.go
index 49c5675..eebed78 100644
--- a/internal/pkg/core/openolt.go
+++ b/internal/pkg/core/openolt.go
@@ -29,6 +29,7 @@
"github.com/opencord/voltha-lib-go/v4/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/config"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
+ "github.com/opencord/voltha-protos/v4/go/extension"
ic "github.com/opencord/voltha-protos/v4/go/inter_container"
"github.com/opencord/voltha-protos/v4/go/openflow_13"
"github.com/opencord/voltha-protos/v4/go/voltha"
@@ -376,3 +377,29 @@
}
return resp, nil
}
+
+//Single_get_value_request handles get uni status on ONU and ondemand metric on OLT
+func (oo *OpenOLT) Single_get_value_request(ctx context.Context, request extension.SingleGetValueRequest) (*extension.SingleGetValueResponse, error) {
+ logger.Infow(ctx, "Single_get_value_request", log.Fields{"request": request})
+
+ errResp := func(status extension.GetValueResponse_Status,
+ reason extension.GetValueResponse_ErrorReason) *extension.SingleGetValueResponse {
+ return &extension.SingleGetValueResponse{
+ Response: &extension.GetValueResponse{
+ Status: status,
+ ErrReason: reason,
+ },
+ }
+ }
+ if handler := oo.getDeviceHandler(request.TargetId); handler != nil {
+ switch reqType := request.GetRequest().GetRequest().(type) {
+ case *extension.GetValueRequest_OltPortInfo:
+ return handler.getOltPortCounters(ctx, reqType.OltPortInfo), nil
+ default:
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_UNSUPPORTED), nil
+ }
+ }
+
+ logger.Infow(ctx, "Single_get_value_request failed ", log.Fields{"request": request})
+ return errResp(extension.GetValueResponse_ERROR, extension.GetValueResponse_INVALID_DEVICE_ID), nil
+}
diff --git a/internal/pkg/core/statsmanager.go b/internal/pkg/core/statsmanager.go
index 1b52e7c..e13abed 100755
--- a/internal/pkg/core/statsmanager.go
+++ b/internal/pkg/core/statsmanager.go
@@ -18,6 +18,7 @@
package core
import (
+ "container/list"
"context"
"fmt"
rsrcMgr "github.com/opencord/voltha-openolt-adapter/internal/pkg/resourcemanager"
@@ -26,6 +27,7 @@
"github.com/opencord/voltha-lib-go/v4/pkg/log"
"github.com/opencord/voltha-openolt-adapter/internal/pkg/olterrors"
+ "github.com/opencord/voltha-protos/v4/go/extension"
"github.com/opencord/voltha-protos/v4/go/openolt"
"github.com/opencord/voltha-protos/v4/go/voltha"
)
@@ -93,6 +95,14 @@
var onuStats = make(chan *openolt.OnuStatistics, 100)
var gemStats = make(chan *openolt.GemPortStatistics, 100)
+//statRegInfo is used to register for notifications
+//on receiving port stats and flow stats indication
+type statRegInfo struct {
+ chn chan bool
+ portNo uint32
+ portType extension.GetOltPortCounters_PortType
+}
+
// PonPort representation
type PonPort struct {
/*
@@ -242,12 +252,23 @@
return &NNI
}
+//StatType defines portStatsType and flowStatsType types
+type StatType int
+
+const (
+ portStatsType StatType = iota
+ flowStatsType
+)
+
// OpenOltStatisticsMgr structure
type OpenOltStatisticsMgr struct {
Device *DeviceHandler
NorthBoundPort map[uint32]*NniPort
SouthBoundPort map[uint32]*PonPort
// TODO PMMetrics Metrics
+ //statIndListners is the list of requests to be notified when port and flow stats indication is received
+ statIndListnerMu sync.Mutex
+ statIndListners map[StatType]*list.List
}
// NewOpenOltStatsMgr returns a new instance of the OpenOltStatisticsMgr
@@ -271,6 +292,9 @@
if StatMgr.Device.openOLT.enableGemStats {
go StatMgr.publishGemStats()
}
+ StatMgr.statIndListners = make(map[StatType]*list.List)
+ StatMgr.statIndListners[portStatsType] = list.New()
+ StatMgr.statIndListners[flowStatsType] = list.New()
return &StatMgr
}
@@ -529,7 +553,7 @@
}
// publishMetrics will publish the pon port metrics
-func (StatMgr OpenOltStatisticsMgr) publishMetrics(ctx context.Context, statType string, val map[string]float32,
+func (StatMgr *OpenOltStatisticsMgr) publishMetrics(ctx context.Context, statType string, val map[string]float32,
port *voltha.Port, devID string, devType string) {
logger.Debugw(ctx, "publish-metrics",
log.Fields{
@@ -576,6 +600,9 @@
func (StatMgr *OpenOltStatisticsMgr) PortStatisticsIndication(ctx context.Context, PortStats *openolt.PortStatistics, NumPonPorts uint32) {
StatMgr.PortsStatisticsKpis(ctx, PortStats, NumPonPorts)
logger.Debugw(ctx, "received-port-stats-indication", log.Fields{"port-stats": PortStats})
+ //Indicate that PortStatisticsIndication is handled
+ //PortStats.IntfId is actually the port number
+ StatMgr.processStatIndication(ctx, portStatsType, PortStats.IntfId)
// TODO send stats to core topic to the voltha kafka or a different kafka ?
}
@@ -673,3 +700,81 @@
*/
}
+
+func (StatMgr *OpenOltStatisticsMgr) updateGetOltPortCountersResponse(ctx context.Context, singleValResp *extension.SingleGetValueResponse, stats map[string]float32) {
+
+ metrics := singleValResp.GetResponse().GetPortCoutners()
+ metrics.TxBytes = uint64(stats["TxBytes"])
+ metrics.RxBytes = uint64(stats["RxBytes"])
+ metrics.TxPackets = uint64(stats["TxPackets"])
+ metrics.RxPackets = uint64(stats["RxPackets"])
+ metrics.TxErrorPackets = uint64(stats["TxErrorPackets"])
+ metrics.RxErrorPackets = uint64(stats["RxErrorPackets"])
+ metrics.TxBcastPackets = uint64(stats["TxBcastPackets"])
+ metrics.RxBcastPackets = uint64(stats["RxBcastPackets"])
+ metrics.TxUcastPackets = uint64(stats["TxUcastPackets"])
+ metrics.RxUcastPackets = uint64(stats["RxUcastPackets"])
+ metrics.TxMcastPackets = uint64(stats["TxMcastPackets"])
+ metrics.RxMcastPackets = uint64(stats["RxMcastPackets"])
+
+ singleValResp.Response.Status = extension.GetValueResponse_OK
+ logger.Debugw(ctx, "updateGetOltPortCountersResponse", log.Fields{"resp": singleValResp})
+}
+
+//RegisterForStatIndication registers ch as a channel on which indication is sent when statistics of type t is received
+func (StatMgr *OpenOltStatisticsMgr) RegisterForStatIndication(ctx context.Context, t StatType, ch chan bool, portNo uint32, portType extension.GetOltPortCounters_PortType) {
+ statInd := statRegInfo{
+ chn: ch,
+ portNo: portNo,
+ portType: portType,
+ }
+
+ logger.Debugf(ctx, "RegisterForStatIndication stat type %v portno %v porttype %v chan %v", t, portNo, portType, ch)
+ StatMgr.statIndListnerMu.Lock()
+ StatMgr.statIndListners[t].PushBack(statInd)
+ StatMgr.statIndListnerMu.Unlock()
+
+}
+
+//DeRegisterFromStatIndication removes the previously registered channel ch for type t of statistics
+func (StatMgr *OpenOltStatisticsMgr) DeRegisterFromStatIndication(ctx context.Context, t StatType, ch chan bool) {
+ StatMgr.statIndListnerMu.Lock()
+ defer StatMgr.statIndListnerMu.Unlock()
+
+ for e := StatMgr.statIndListners[t].Front(); e != nil; e = e.Next() {
+ statInd := e.Value.(statRegInfo)
+ if statInd.chn == ch {
+ StatMgr.statIndListners[t].Remove(e)
+ return
+ }
+ }
+}
+
+func (StatMgr *OpenOltStatisticsMgr) processStatIndication(ctx context.Context, t StatType, portNo uint32) {
+ var deRegList []*list.Element
+ var statInd statRegInfo
+
+ StatMgr.statIndListnerMu.Lock()
+ defer StatMgr.statIndListnerMu.Unlock()
+
+ if StatMgr.statIndListners[t] == nil || StatMgr.statIndListners[t].Len() == 0 {
+ logger.Debugf(ctx, "processStatIndication %v list is empty ", t)
+ return
+ }
+
+ for e := StatMgr.statIndListners[t].Front(); e != nil; e = e.Next() {
+ statInd = e.Value.(statRegInfo)
+ if statInd.portNo != portNo {
+ fmt.Printf("Skipping %v\n", e.Value)
+ continue
+ }
+ // message sent
+ statInd.chn <- true
+ deRegList = append(deRegList, e)
+
+ }
+ for _, e := range deRegList {
+ StatMgr.statIndListners[t].Remove(e)
+ }
+
+}