[VOL-3318] : Framework for collecting Metrics from the ONU
- Currently implements collecting of OpticalPower Metrics and UNI status metrics
- Supports changing Metric collection frequency
- Use voltha-lib-go version 4.0.6
Change-Id: I9bd1ec8d8af9d739db96ae0303b6702dd3ce8520
diff --git a/VERSION b/VERSION
index 80eab4f..e80532c 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-1.1.0-dev156
+1.1.0-dev157
diff --git a/go.mod b/go.mod
index 6c99589..82b9486 100644
--- a/go.mod
+++ b/go.mod
@@ -9,7 +9,7 @@
github.com/google/gopacket v1.1.17
github.com/looplab/fsm v0.1.0
github.com/opencord/omci-lib-go v0.13.4
- github.com/opencord/voltha-lib-go/v4 v4.0.5
+ github.com/opencord/voltha-lib-go/v4 v4.0.6
github.com/opencord/voltha-protos/v4 v4.0.8
github.com/stretchr/testify v1.6.1
google.golang.org/grpc v1.25.1 // indirect
diff --git a/go.sum b/go.sum
index cbfa3ff..93b5de6 100644
--- a/go.sum
+++ b/go.sum
@@ -219,8 +219,8 @@
github.com/onsi/gomega v1.4.2/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
github.com/opencord/omci-lib-go v0.13.4 h1:L3GpFHrx3DASixkCvztnJbcLGUTU/21wJYdE/qG9/XA=
github.com/opencord/omci-lib-go v0.13.4/go.mod h1:6OIHB14Ch5qGgHzwSWlMACtk5KFoLzQ4LAhdcy4jwvo=
-github.com/opencord/voltha-lib-go/v4 v4.0.5 h1:tIJHjIxIExwCEvlY7wnSqERnoaDn7uOaQgL9t3SKvQQ=
-github.com/opencord/voltha-lib-go/v4 v4.0.5/go.mod h1:8NFUZz/mp4OvRmilBRhkLOUrw4G01ruSAVdzQu2ivPc=
+github.com/opencord/voltha-lib-go/v4 v4.0.6 h1:Mc/bkYDR3YcpWPeW35ju9h3sO2cGByz8XM0ik5ikJkc=
+github.com/opencord/voltha-lib-go/v4 v4.0.6/go.mod h1:8NFUZz/mp4OvRmilBRhkLOUrw4G01ruSAVdzQu2ivPc=
github.com/opencord/voltha-protos/v4 v4.0.8 h1:/P9IYuWPTp/aabS5n1fakQgHCFNzKjW1JHw3TOLAfKE=
github.com/opencord/voltha-protos/v4 v4.0.8/go.mod h1:W/OIFIyvFh/C0vchRUuarIsMylEhzCRM9pNxLvkPtKc=
github.com/opentracing/opentracing-go v1.1.0 h1:pWlfV3Bxv7k65HYwkikxat0+s3pV4bsqf19k25Ur8rU=
diff --git a/internal/pkg/onuadaptercore/device_handler.go b/internal/pkg/onuadaptercore/device_handler.go
index 0e0c1af..758bf9e 100644
--- a/internal/pkg/onuadaptercore/device_handler.go
+++ b/internal/pkg/onuadaptercore/device_handler.go
@@ -31,6 +31,7 @@
"github.com/looplab/fsm"
me "github.com/opencord/omci-lib-go/generated"
"github.com/opencord/voltha-lib-go/v4/pkg/adapters/adapterif"
+ "github.com/opencord/voltha-lib-go/v4/pkg/adapters/common"
"github.com/opencord/voltha-lib-go/v4/pkg/db"
"github.com/opencord/voltha-lib-go/v4/pkg/events/eventif"
flow "github.com/opencord/voltha-lib-go/v4/pkg/flows"
@@ -124,6 +125,19 @@
drTechProfileConfigDeleteSuccess: "tech-profile-config-delete-success",
}
+// OmciOpticalMetricsNames are supported optical pm names
+var OmciOpticalMetricsNames = map[string]voltha.PmConfig_PmType{
+ "transmit_power": voltha.PmConfig_GAUGE,
+ "receive_power": voltha.PmConfig_GAUGE,
+}
+
+// OmciUniMetricsNames are supported UNI status names
+var OmciUniMetricsNames = map[string]voltha.PmConfig_PmType{
+ "ethernet_type": voltha.PmConfig_GAUGE,
+ "oper_status": voltha.PmConfig_GAUGE,
+ "uni_admin_state": voltha.PmConfig_GAUGE,
+}
+
//deviceHandler will interact with the ONU ? device.
type deviceHandler struct {
deviceID string
@@ -140,12 +154,15 @@
AdapterProxy adapterif.AdapterProxy
EventProxy eventif.EventProxy
+ pmMetrics *common.PmMetrics
+
pOpenOnuAc *OpenONUAC
pDeviceStateFsm *fsm.FSM
//pPonPort *voltha.Port
deviceEntrySet chan bool //channel for DeviceEntry set event
pOnuOmciDevice *OnuDeviceEntry
pOnuTP *onuUniTechProf
+ pOnuMetricsMgr *onuMetricsManager
exitChannel chan int
lockDevice sync.RWMutex
pOnuIndication *oop.OnuIndication
@@ -160,7 +177,6 @@
//discOnus sync.Map
//onus sync.Map
//portStats *OpenOltStatisticsMgr
- //metrics *pmmetrics.PmMetrics
stopCollector chan bool
stopHeartbeatCheck chan bool
uniEntityMap map[uint32]*onuUniPort
@@ -194,7 +210,32 @@
dh.UniVlanConfigFsmMap = make(map[uint8]*UniVlanConfigFsm)
dh.reconciling = false
dh.ReadyForSpecificOmciConfig = false
+ metricNames := make([]string, len(OmciOpticalMetricsNames)+len(OmciUniMetricsNames))
+ for k := range OmciOpticalMetricsNames {
+ metricNames = append(metricNames, k)
+ }
+
+ for k := range OmciUniMetricsNames {
+ metricNames = append(metricNames, k)
+ }
+
+ // The frequency is in seconds.
+ dh.pmMetrics = common.NewPmMetrics(cloned.Id, common.Frequency(150), common.FrequencyOverride(false), common.Grouped(false), common.Metrics(metricNames))
+ for pmName, pmType := range OmciOpticalMetricsNames {
+ dh.pmMetrics.ToPmConfigs().Metrics = append(dh.pmMetrics.ToPmConfigs().Metrics, &voltha.PmConfig{
+ Name: pmName,
+ Type: pmType,
+ Enabled: true,
+ })
+ }
+ for pmName, pmType := range OmciUniMetricsNames {
+ dh.pmMetrics.ToPmConfigs().Metrics = append(dh.pmMetrics.ToPmConfigs().Metrics, &voltha.PmConfig{
+ Name: pmName,
+ Type: pmType,
+ Enabled: true,
+ })
+ }
// Device related state machine
dh.pDeviceStateFsm = fsm.NewFSM(
devStNull,
@@ -248,6 +289,10 @@
logger.Errorw(ctx, "Device FSM: Can't go to state DeviceInit", log.Fields{"err": err})
}
logger.Debugw(ctx, "Device FSM: ", log.Fields{"state": string(dh.pDeviceStateFsm.Current())})
+ // Now, set the initial PM configuration for that device
+ if err := dh.coreProxy.DevicePMConfigUpdate(ctx, dh.pmMetrics.ToPmConfigs()); err != nil {
+ logger.Errorw(ctx, "error updating pm config to core", log.Fields{"device-id": dh.deviceID, "err": err})
+ }
} else {
logger.Debugw(ctx, "AdoptOrReconcileDevice: Agent/device init already done", log.Fields{"device-id": device.Id})
}
@@ -1015,6 +1060,7 @@
go dh.reconcileDeviceOnuInd(ctx)
// reconcilement will be continued after mib download is done
}
+
/*
############################################################################
# Setup Alarm handler
@@ -1058,6 +1104,7 @@
else:
self.logger.info('onu-already-activated')
*/
+
logger.Debug(ctx, "postInit-done")
}
@@ -1187,11 +1234,12 @@
//setOnuDeviceEntry sets the ONU device entry within the handler
func (dh *deviceHandler) setOnuDeviceEntry(
- apDeviceEntry *OnuDeviceEntry, apOnuTp *onuUniTechProf) {
+ apDeviceEntry *OnuDeviceEntry, apOnuTp *onuUniTechProf, apOnuMetricsMgr *onuMetricsManager) {
dh.lockDevice.Lock()
defer dh.lockDevice.Unlock()
dh.pOnuOmciDevice = apDeviceEntry
dh.pOnuTP = apOnuTp
+ dh.pOnuMetricsMgr = apOnuMetricsMgr
}
//addOnuDeviceEntry creates a new ONU device or returns the existing
@@ -1206,8 +1254,9 @@
/* and no alarm_db yet (oo.alarm_db) */
deviceEntry = newOnuDeviceEntry(ctx, dh)
onuTechProfProc := newOnuUniTechProf(ctx, dh)
+ onuMetricsMgr := newonuMetricsManager(ctx, dh)
//error treatment possible //TODO!!!
- dh.setOnuDeviceEntry(deviceEntry, onuTechProfProc)
+ dh.setOnuDeviceEntry(deviceEntry, onuTechProfProc, onuMetricsMgr)
// fire deviceEntry ready event to spread to possibly waiting processing
dh.deviceEntrySet <- true
logger.Debugw(ctx, "onuDeviceEntry-added", log.Fields{"device-id": dh.deviceID})
@@ -1398,6 +1447,10 @@
logger.Errorw(ctx, "MibSyncFsm invalid - cannot be executed!!", log.Fields{"device-id": dh.deviceID})
return fmt.Errorf("can't execute MibSync: %s", dh.deviceID)
}
+
+ // Start PM collector routine
+ go dh.startCollector(ctx)
+
return nil
}
@@ -1516,7 +1569,9 @@
}
}
}
- //TODO!!! care about PM/Alarm processing once started
+ // Stop collector routine for PM Counters
+ dh.stopCollector <- true
+
return nil
}
@@ -2492,3 +2547,44 @@
}
return 0, errors.New("error-fetching-uni-port")
}
+
+// updatePmConfig updates the pm metrics config.
+func (dh *deviceHandler) updatePmConfig(ctx context.Context, pmConfigs *voltha.PmConfigs) {
+ logger.Infow(ctx, "update-pm-config", log.Fields{"device-id": dh.device.Id, "pm-configs": pmConfigs})
+
+ // TODO: Currently support only updating the PM Sampling Frequency
+ if pmConfigs.DefaultFreq != dh.pmMetrics.ToPmConfigs().DefaultFreq {
+ dh.pmMetrics.UpdateFrequency(pmConfigs.DefaultFreq)
+ logger.Debugw(ctx, "frequency-updated--new-frequency", log.Fields{"device-id": dh.deviceID, "frequency": dh.pmMetrics.ToPmConfigs().DefaultFreq})
+ } else {
+ logger.Debugw(ctx, "new-frequency-same-as-old--not-updating", log.Fields{"frequency": pmConfigs.DefaultFreq})
+ }
+}
+
+func (dh *deviceHandler) startCollector(ctx context.Context) {
+ logger.Debugf(ctx, "startingCollector")
+
+ // Start routine to process OMCI GET Responses
+ go dh.pOnuMetricsMgr.processOmciMessages(ctx)
+
+ for {
+ select {
+ case <-dh.stopCollector:
+ logger.Debugw(ctx, "stopping-collector-for-onu", log.Fields{"device-id": dh.device.Id})
+ dh.pOnuMetricsMgr.stopProcessingOmciResponses <- true // Stop the OMCI GET response processing routine
+ return
+ case <-time.After(time.Duration(dh.pmMetrics.ToPmConfigs().DefaultFreq) * time.Second):
+ go func() {
+ logger.Debug(ctx, "startCollector before collecting optical metrics")
+ metricInfo := dh.pOnuMetricsMgr.collectOpticalMetrics(ctx)
+ dh.pOnuMetricsMgr.publishMetrics(ctx, metricInfo)
+ }()
+
+ go func() {
+ logger.Debug(ctx, "startCollector before collecting uni metrics")
+ metricInfo := dh.pOnuMetricsMgr.collectUniStatusMetrics(ctx)
+ dh.pOnuMetricsMgr.publishMetrics(ctx, metricInfo)
+ }()
+ }
+ }
+}
diff --git a/internal/pkg/onuadaptercore/mib_sync.go b/internal/pkg/onuadaptercore/mib_sync.go
index 833425c..4b6b86d 100644
--- a/internal/pkg/onuadaptercore/mib_sync.go
+++ b/internal/pkg/onuadaptercore/mib_sync.go
@@ -83,7 +83,7 @@
func (oo *OnuDeviceEntry) enterGettingVendorAndSerialState(ctx context.Context, e *fsm.Event) {
logger.Debugw(ctx, "MibSync FSM", log.Fields{"Start getting VendorId and SerialNumber in State": e.FSM.Current(), "device-id": oo.deviceID})
requestedAttributes := me.AttributeValueMap{"VendorId": "", "SerialNumber": 0}
- meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.OnuGClassID, onugMeID, requestedAttributes, ConstDefaultOmciTimeout, true)
+ meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.OnuGClassID, onugMeID, requestedAttributes, ConstDefaultOmciTimeout, true, oo.pMibUploadFsm.commChan)
//accept also nil as (error) return value for writing to LastTx
// - this avoids misinterpretation of new received OMCI messages
oo.PDevOmciCC.pLastTxMeInstance = meInstance
@@ -92,7 +92,7 @@
func (oo *OnuDeviceEntry) enterGettingEquipmentIDState(ctx context.Context, e *fsm.Event) {
logger.Debugw(ctx, "MibSync FSM", log.Fields{"Start getting EquipmentId in State": e.FSM.Current(), "device-id": oo.deviceID})
requestedAttributes := me.AttributeValueMap{"EquipmentId": ""}
- meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.Onu2GClassID, onu2gMeID, requestedAttributes, ConstDefaultOmciTimeout, true)
+ meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.Onu2GClassID, onu2gMeID, requestedAttributes, ConstDefaultOmciTimeout, true, oo.pMibUploadFsm.commChan)
//accept also nil as (error) return value for writing to LastTx
// - this avoids misinterpretation of new received OMCI messages
oo.PDevOmciCC.pLastTxMeInstance = meInstance
@@ -101,7 +101,7 @@
func (oo *OnuDeviceEntry) enterGettingFirstSwVersionState(ctx context.Context, e *fsm.Event) {
logger.Debugw(ctx, "MibSync FSM", log.Fields{"Start getting IsActive and Version of first SW-image in State": e.FSM.Current(), "device-id": oo.deviceID})
requestedAttributes := me.AttributeValueMap{"IsActive": 0, "Version": ""}
- meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.SoftwareImageClassID, firstSwImageMeID, requestedAttributes, ConstDefaultOmciTimeout, true)
+ meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.SoftwareImageClassID, firstSwImageMeID, requestedAttributes, ConstDefaultOmciTimeout, true, oo.pMibUploadFsm.commChan)
//accept also nil as (error) return value for writing to LastTx
// - this avoids misinterpretation of new received OMCI messages
oo.PDevOmciCC.pLastTxMeInstance = meInstance
@@ -110,7 +110,7 @@
func (oo *OnuDeviceEntry) enterGettingSecondSwVersionState(ctx context.Context, e *fsm.Event) {
logger.Debugw(ctx, "MibSync FSM", log.Fields{"Start getting IsActive and Version of second SW-image in State": e.FSM.Current(), "device-id": oo.deviceID})
requestedAttributes := me.AttributeValueMap{"IsActive": 0, "Version": ""}
- meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.SoftwareImageClassID, secondSwImageMeID, requestedAttributes, ConstDefaultOmciTimeout, true)
+ meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.SoftwareImageClassID, secondSwImageMeID, requestedAttributes, ConstDefaultOmciTimeout, true, oo.pMibUploadFsm.commChan)
//accept also nil as (error) return value for writing to LastTx
// - this avoids misinterpretation of new received OMCI messages
oo.PDevOmciCC.pLastTxMeInstance = meInstance
@@ -119,7 +119,7 @@
func (oo *OnuDeviceEntry) enterGettingMacAddressState(ctx context.Context, e *fsm.Event) {
logger.Debugw(ctx, "MibSync FSM", log.Fields{"Start getting MacAddress in State": e.FSM.Current(), "device-id": oo.deviceID})
requestedAttributes := me.AttributeValueMap{"MacAddress": ""}
- meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.IpHostConfigDataClassID, ipHostConfigDataMeID, requestedAttributes, ConstDefaultOmciTimeout, true)
+ meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx), me.IpHostConfigDataClassID, ipHostConfigDataMeID, requestedAttributes, ConstDefaultOmciTimeout, true, oo.pMibUploadFsm.commChan)
//accept also nil as (error) return value for writing to LastTx
// - this avoids misinterpretation of new received OMCI messages
oo.PDevOmciCC.pLastTxMeInstance = meInstance
@@ -637,7 +637,7 @@
logger.Debugw(ctx, "Request MDS value", log.Fields{"device-id": oo.deviceID})
requestedAttributes := me.AttributeValueMap{"MibDataSync": ""}
meInstance := oo.PDevOmciCC.sendGetMe(log.WithSpanFromContext(context.TODO(), ctx),
- me.OnuDataClassID, onuDataMeID, requestedAttributes, ConstDefaultOmciTimeout, true)
+ me.OnuDataClassID, onuDataMeID, requestedAttributes, ConstDefaultOmciTimeout, true, oo.pMibUploadFsm.commChan)
//accept also nil as (error) return value for writing to LastTx
// - this avoids misinterpretation of new received OMCI messages
oo.PDevOmciCC.pLastTxMeInstance = meInstance
diff --git a/internal/pkg/onuadaptercore/omci_cc.go b/internal/pkg/onuadaptercore/omci_cc.go
index cd91e77..5b0e448 100644
--- a/internal/pkg/onuadaptercore/omci_cc.go
+++ b/internal/pkg/onuadaptercore/omci_cc.go
@@ -1189,7 +1189,7 @@
}
func (oo *omciCC) sendGetMe(ctx context.Context, classID me.ClassID, entityID uint16, requestedAttributes me.AttributeValueMap,
- timeout int, highPrio bool) *me.ManagedEntity {
+ timeout int, highPrio bool, rxChan chan Message) *me.ManagedEntity {
tid := oo.getNextTid(highPrio)
logger.Debugw(ctx, "send get-request-msg", log.Fields{"classID": classID, "device-id": oo.deviceID,
@@ -1214,7 +1214,7 @@
}
omciRxCallbackPair := callbackPair{
cbKey: tid,
- cbEntry: callbackPairEntry{(*oo.pOnuDeviceEntry).pMibUploadFsm.commChan, oo.receiveOmciResponse},
+ cbEntry: callbackPairEntry{rxChan, oo.receiveOmciResponse},
}
err = oo.send(ctx, pkt, timeout, 0, highPrio, omciRxCallbackPair)
if err != nil {
diff --git a/internal/pkg/onuadaptercore/onu_metrics_manager.go b/internal/pkg/onuadaptercore/onu_metrics_manager.go
new file mode 100644
index 0000000..3e43275
--- /dev/null
+++ b/internal/pkg/onuadaptercore/onu_metrics_manager.go
@@ -0,0 +1,382 @@
+/*
+ * Copyright 2021-present Open Networking Foundation
+
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+
+ * http://www.apache.org/licenses/LICENSE-2.0
+
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+//Package adaptercoreonu provides the utility for onu devices, flows and statistics
+package adaptercoreonu
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "github.com/opencord/omci-lib-go"
+ me "github.com/opencord/omci-lib-go/generated"
+ "github.com/opencord/voltha-lib-go/v4/pkg/log"
+ "github.com/opencord/voltha-protos/v4/go/voltha"
+ "time"
+)
+
+type onuMetricsManager struct {
+ pDeviceHandler *deviceHandler
+
+ commMetricsChan chan Message
+ opticalMetricsChan chan me.AttributeValueMap
+ uniStatusMetricsChan chan me.AttributeValueMap
+
+ stopProcessingOmciResponses chan bool
+}
+
+// newonuMetricsManager returns a new instance of the newonuMetricsManager
+func newonuMetricsManager(ctx context.Context, dh *deviceHandler) *onuMetricsManager {
+
+ var metricsManager onuMetricsManager
+ logger.Debugw(ctx, "init-onuMetricsManager", log.Fields{"device-id": dh.deviceID})
+ metricsManager.pDeviceHandler = dh
+
+ metricsManager.commMetricsChan = make(chan Message)
+ metricsManager.opticalMetricsChan = make(chan me.AttributeValueMap)
+ metricsManager.uniStatusMetricsChan = make(chan me.AttributeValueMap)
+ metricsManager.stopProcessingOmciResponses = make(chan bool)
+
+ return &metricsManager
+}
+
+func (mm *onuMetricsManager) collectOpticalMetrics(ctx context.Context) []*voltha.MetricInformation {
+ logger.Debugw(ctx, "collectOpticalMetrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ var metricInfoSlice []*voltha.MetricInformation
+ metricsContext := make(map[string]string)
+ metricsContext["onuID"] = fmt.Sprintf("%d", mm.pDeviceHandler.device.ProxyAddress.OnuId)
+ metricsContext["intfID"] = fmt.Sprintf("%d", mm.pDeviceHandler.device.ProxyAddress.ChannelId)
+ metricsContext["devicetype"] = mm.pDeviceHandler.DeviceType
+
+ raisedTs := time.Now().UnixNano()
+ mmd := voltha.MetricMetaData{
+ Title: "OpticalMetrics",
+ Ts: float64(raisedTs),
+ Context: metricsContext,
+ DeviceId: mm.pDeviceHandler.deviceID,
+ LogicalDeviceId: mm.pDeviceHandler.logicalDeviceID,
+ SerialNo: mm.pDeviceHandler.device.SerialNumber,
+ }
+
+ enabledMetrics := make([]string, 0)
+ // Populate enabled metrics
+ for _, m := range mm.pDeviceHandler.pmMetrics.ToPmConfigs().Metrics {
+ if m.Enabled {
+ enabledMetrics = append(enabledMetrics, m.Name)
+ }
+ }
+ logger.Debugw(ctx, "enabled metrics", log.Fields{"enabledMetrics": enabledMetrics})
+ // get the ANI-G instance IDs
+ anigInstKeys := mm.pDeviceHandler.pOnuOmciDevice.pOnuDB.getSortedInstKeys(ctx, me.AniGClassID)
+loop:
+ for _, anigInstID := range anigInstKeys {
+ var meAttributes me.AttributeValueMap
+ opticalMetrics := make(map[string]float32)
+ // Get the ANI-G instance optical power attributes
+ requestedAttributes := me.AttributeValueMap{"OpticalSignalLevel": 0, "TransmitOpticalLevel": 0}
+ if meInstance := mm.pDeviceHandler.pOnuOmciDevice.PDevOmciCC.sendGetMe(ctx, me.AniGClassID, anigInstID, requestedAttributes, ConstDefaultOmciTimeout, true, mm.commMetricsChan); meInstance != nil {
+ select {
+ case meAttributes = <-mm.opticalMetricsChan:
+ logger.Debugw(ctx, "received optical metrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ case <-time.After(time.Duration(ConstDefaultOmciTimeout) * time.Second):
+ logger.Errorw(ctx, "timeout waiting for omci-get response for uni status", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ // The metrics will be empty in this case
+ break loop
+ }
+ // Populate metric only if it was enabled.
+ for _, v := range enabledMetrics {
+ switch v {
+ case "transmit_power":
+ opticalMetrics["transmit_power"] = float32(meAttributes["TransmitOpticalLevel"].(uint16))
+ case "receive_power":
+ opticalMetrics["receive_power"] = float32(meAttributes["OpticalSignalLevel"].(uint16))
+ default:
+ // do nothing
+ }
+ }
+ }
+ // create slice of metrics given that there could be more than one ANI-G instance and
+ // optical metrics are collected per ANI-G instance
+ metricInfo := voltha.MetricInformation{Metadata: &mmd, Metrics: opticalMetrics}
+ metricInfoSlice = append(metricInfoSlice, &metricInfo)
+ }
+
+ return metricInfoSlice
+}
+
+// Note: UNI status does not seem to be a metric, but this is being treated as metric in Python implementation
+// nolint: gocyclo
+func (mm *onuMetricsManager) collectUniStatusMetrics(ctx context.Context) []*voltha.MetricInformation {
+ logger.Debugw(ctx, "collectUniStatusMetrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ var metricInfoSlice []*voltha.MetricInformation
+ metricsContext := make(map[string]string)
+ metricsContext["onuID"] = fmt.Sprintf("%d", mm.pDeviceHandler.device.ProxyAddress.OnuId)
+ metricsContext["intfID"] = fmt.Sprintf("%d", mm.pDeviceHandler.device.ProxyAddress.ChannelId)
+ metricsContext["devicetype"] = mm.pDeviceHandler.DeviceType
+
+ raisedTs := time.Now().UnixNano()
+ mmd := voltha.MetricMetaData{
+ Title: "UniStatus", // Is this ok to hard code?
+ Ts: float64(raisedTs),
+ Context: metricsContext,
+ DeviceId: mm.pDeviceHandler.deviceID,
+ LogicalDeviceId: mm.pDeviceHandler.logicalDeviceID,
+ SerialNo: mm.pDeviceHandler.device.SerialNumber,
+ }
+
+ enabledMetrics := make([]string, 0)
+ // Populate enabled metrics
+ for _, m := range mm.pDeviceHandler.pmMetrics.ToPmConfigs().Metrics {
+ if m.Enabled {
+ enabledMetrics = append(enabledMetrics, m.Name)
+ }
+ }
+ logger.Debugw(ctx, "enabled metrics", log.Fields{"enabledMetrics": enabledMetrics})
+
+ // get the UNI-G instance IDs
+ unigInstKeys := mm.pDeviceHandler.pOnuOmciDevice.pOnuDB.getSortedInstKeys(ctx, me.UniGClassID)
+loop1:
+ for _, unigInstID := range unigInstKeys {
+ // TODO: Include additional information in the voltha.MetricMetaData - like portno, uni-id, instance-id
+ // to uniquely identify this ME instance and also to correlate the ME instance to physical instance
+ unigMetrics := make(map[string]float32)
+ var meAttributes me.AttributeValueMap
+ // Get the UNI-G instance optical power attributes
+ requestedAttributes := me.AttributeValueMap{"AdministrativeState": 0}
+ if meInstance := mm.pDeviceHandler.pOnuOmciDevice.PDevOmciCC.sendGetMe(ctx, me.UniGClassID, unigInstID, requestedAttributes, ConstDefaultOmciTimeout, true, mm.commMetricsChan); meInstance != nil {
+ // Wait for metrics or timeout
+ select {
+ case meAttributes = <-mm.uniStatusMetricsChan:
+ logger.Debugw(ctx, "received uni-g metrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ case <-time.After(time.Duration(ConstDefaultOmciTimeout) * time.Second):
+ logger.Errorw(ctx, "timeout waiting for omci-get response for uni status", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ // The metrics could be empty in this case
+ break loop1
+ }
+ // Populate metric only if it was enabled.
+ for _, v := range enabledMetrics {
+ switch v {
+ case "uni_admin_state":
+ unigMetrics["uni_admin_state"] = float32(meAttributes["AdministrativeState"].(byte))
+ default:
+ // do nothing
+ }
+ }
+ // create slice of metrics given that there could be more than one UNI-G instance
+ metricInfo := voltha.MetricInformation{Metadata: &mmd, Metrics: unigMetrics}
+ metricInfoSlice = append(metricInfoSlice, &metricInfo)
+ }
+ }
+
+ // get the PPTP instance IDs
+ pptpInstKeys := mm.pDeviceHandler.pOnuOmciDevice.pOnuDB.getSortedInstKeys(ctx, me.PhysicalPathTerminationPointEthernetUniClassID)
+loop2:
+ for _, pptpInstID := range pptpInstKeys {
+ // TODO: Include additional information in the voltha.MetricMetaData - like portno, uni-id, instance-id
+ // to uniquely identify this ME instance and also to correlate the ME instance to physical instance
+ var meAttributes me.AttributeValueMap
+ pptpMetrics := make(map[string]float32)
+
+ requestedAttributes := me.AttributeValueMap{"SensedType": 0, "OperationalState": 0, "AdministrativeState": 0}
+ if meInstance := mm.pDeviceHandler.pOnuOmciDevice.PDevOmciCC.sendGetMe(ctx, me.PhysicalPathTerminationPointEthernetUniClassID, pptpInstID, requestedAttributes, ConstDefaultOmciTimeout, true, mm.commMetricsChan); meInstance != nil {
+ // Wait for metrics or timeout
+ select {
+ case meAttributes = <-mm.uniStatusMetricsChan:
+ logger.Debugw(ctx, "received pptp metrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ case <-time.After(time.Duration(ConstDefaultOmciTimeout) * time.Second):
+ logger.Errorw(ctx, "timeout waiting for omci-get response for uni status", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ // The metrics could be empty in this case
+ break loop2
+ }
+
+ // Populate metric only if it was enabled.
+ for _, v := range enabledMetrics {
+ switch v {
+ case "ethernet_type":
+ pptpMetrics["ethernet_type"] = float32(meAttributes["SensedType"].(byte))
+ case "oper_status":
+ pptpMetrics["oper_status"] = float32(meAttributes["OperationalState"].(byte))
+ case "uni_admin_state":
+ pptpMetrics["uni_admin_state"] = float32(meAttributes["AdministrativeState"].(byte))
+ default:
+ // do nothing
+ }
+ }
+ }
+ // create slice of metrics given that there could be more than one PPTP instance and
+ metricInfo := voltha.MetricInformation{Metadata: &mmd, Metrics: pptpMetrics}
+ metricInfoSlice = append(metricInfoSlice, &metricInfo)
+ }
+
+ // get the VEIP instance IDs
+ veipInstKeys := mm.pDeviceHandler.pOnuOmciDevice.pOnuDB.getSortedInstKeys(ctx, me.VirtualEthernetInterfacePointClassID)
+loop3:
+ for _, veipInstID := range veipInstKeys {
+ // TODO: Include additional information in the voltha.MetricMetaData - like portno, uni-id, instance-id
+ // to uniquely identify this ME instance and also to correlate the ME instance to physical instance
+ var meAttributes me.AttributeValueMap
+ pptpMetrics := make(map[string]float32)
+
+ requestedAttributes := me.AttributeValueMap{"OperationalState": 0, "AdministrativeState": 0}
+ if meInstance := mm.pDeviceHandler.pOnuOmciDevice.PDevOmciCC.sendGetMe(ctx, me.VirtualEthernetInterfacePointClassID, veipInstID, requestedAttributes, ConstDefaultOmciTimeout, true, mm.commMetricsChan); meInstance != nil {
+ // Wait for metrics or timeout
+ select {
+ case meAttributes = <-mm.uniStatusMetricsChan:
+ logger.Debugw(ctx, "received veip metrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ case <-time.After(time.Duration(ConstDefaultOmciTimeout) * time.Second):
+ logger.Errorw(ctx, "timeout waiting for omci-get response for uni status", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ // The metrics could be empty in this case
+ break loop3
+ }
+
+ // Populate metric only if it was enabled.
+ for _, v := range enabledMetrics {
+ switch v {
+ case "oper_status":
+ pptpMetrics["oper_status"] = float32(meAttributes["OperationalState"].(byte))
+ case "uni_admin_state":
+ pptpMetrics["uni_admin_state"] = float32(meAttributes["AdministrativeState"].(byte))
+ default:
+ // do nothing
+ }
+ }
+ }
+ // create slice of metrics given that there could be more than one VEIP instance
+ metricInfo := voltha.MetricInformation{Metadata: &mmd, Metrics: pptpMetrics}
+ metricInfoSlice = append(metricInfoSlice, &metricInfo)
+ }
+
+ return metricInfoSlice
+}
+
+// publishMetrics publishes the metrics on kafka
+func (mm *onuMetricsManager) publishMetrics(ctx context.Context, metricInfo []*voltha.MetricInformation) {
+ var ke voltha.KpiEvent2
+ ts := time.Now().UnixNano()
+ ke.SliceData = metricInfo
+ ke.Type = voltha.KpiEventType_slice
+ ke.Ts = float64(ts)
+
+ if err := mm.pDeviceHandler.EventProxy.SendKpiEvent(ctx, "STATS_EVENT", &ke, voltha.EventCategory_EQUIPMENT, voltha.EventSubCategory_ONU, ts); err != nil {
+ logger.Errorw(ctx, "failed-to-send-pon-stats", log.Fields{"err": err})
+ }
+}
+
+func (mm *onuMetricsManager) processOmciMessages(ctx context.Context) {
+ logger.Infow(ctx, "Start routine to process OMCI-GET messages for device-id", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ // Flush metric collection channels to be safe.
+ // It is possible that there is stale data on this channel if the processOmciMessages routine
+ // is stopped right after issuing a OMCI-GET request and started again.
+ // The processOmciMessages routine will get stopped if startCollector routine (in device_handler.go)
+ // is stopped - as a result of ONU going down.
+ mm.flushMetricCollectionChannels(ctx)
+
+ for {
+ select {
+ case <-mm.stopProcessingOmciResponses: // stop this routine
+ logger.Infow(ctx, "Stop routine to process OMCI-GET messages for device-id", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ return
+ case message, ok := <-mm.commMetricsChan:
+ if !ok {
+ logger.Errorw(ctx, "Message couldn't be read from channel", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ continue
+ }
+ logger.Debugw(ctx, "Received message on ONU metrics channel", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+
+ switch message.Type {
+ case OMCI:
+ msg, _ := message.Data.(OmciMessage)
+ mm.handleOmciMessage(ctx, msg)
+ default:
+ logger.Warn(ctx, "Unknown message type received", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "message.Type": message.Type})
+ }
+ }
+ }
+}
+
+func (mm *onuMetricsManager) handleOmciMessage(ctx context.Context, msg OmciMessage) {
+ logger.Debugw(ctx, "omci Msg", log.Fields{"device-id": mm.pDeviceHandler.deviceID,
+ "msgType": msg.OmciMsg.MessageType, "msg": msg})
+ switch msg.OmciMsg.MessageType {
+ case omci.GetResponseType:
+ //TODO: error handling
+ _ = mm.handleOmciGetResponseMessage(ctx, msg)
+
+ default:
+ logger.Warnw(ctx, "Unknown Message Type", log.Fields{"msgType": msg.OmciMsg.MessageType})
+
+ }
+}
+
+func (mm *onuMetricsManager) handleOmciGetResponseMessage(ctx context.Context, msg OmciMessage) error {
+ msgLayer := (*msg.OmciPacket).Layer(omci.LayerTypeGetResponse)
+ if msgLayer == nil {
+ logger.Errorw(ctx, "omci Msg layer could not be detected for GetResponse - handling stopped", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ return fmt.Errorf("omci Msg layer could not be detected for GetResponse - handling stopped: %s", mm.pDeviceHandler.deviceID)
+ }
+ msgObj, msgOk := msgLayer.(*omci.GetResponse)
+ if !msgOk {
+ logger.Errorw(ctx, "omci Msg layer could not be assigned for GetResponse - handling stopped", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ return fmt.Errorf("omci Msg layer could not be assigned for GetResponse - handling stopped: %s", mm.pDeviceHandler.deviceID)
+ }
+ logger.Debugw(ctx, "OMCI GetResponse Data", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "data-fields": msgObj})
+ if msgObj.Result == me.Success {
+ meAttributes := msgObj.Attributes
+ switch msgObj.EntityClass {
+ case me.AniGClassID:
+ mm.opticalMetricsChan <- meAttributes
+ return nil
+ case me.UniGClassID:
+ mm.uniStatusMetricsChan <- meAttributes
+ return nil
+ case me.PhysicalPathTerminationPointEthernetUniClassID:
+ mm.uniStatusMetricsChan <- meAttributes
+ return nil
+ case me.VirtualEthernetInterfacePointClassID:
+ mm.uniStatusMetricsChan <- meAttributes
+ return nil
+ default:
+ logger.Errorw(ctx, "unhandled omci get response message",
+ log.Fields{"device-id": mm.pDeviceHandler.deviceID, "class-id": msgObj.EntityClass})
+ }
+ }
+
+ return errors.New("unhandled-omci-get-response-message")
+}
+
+// flushMetricCollectionChannels flushes all metric collection channels for any stale OMCI responses
+func (mm *onuMetricsManager) flushMetricCollectionChannels(ctx context.Context) {
+ // flush commMetricsChan
+ select {
+ case <-mm.commMetricsChan:
+ logger.Debug(ctx, "flushed common metrics channel")
+ default:
+ }
+
+ // flush opticalMetricsChan
+ select {
+ case <-mm.opticalMetricsChan:
+ logger.Debug(ctx, "flushed optical metrics channel")
+ default:
+ }
+
+ // flush uniStatusMetricsChan
+ select {
+ case <-mm.uniStatusMetricsChan:
+ logger.Debug(ctx, "flushed uni status metrics channel")
+ default:
+ }
+}
diff --git a/internal/pkg/onuadaptercore/openonu.go b/internal/pkg/onuadaptercore/openonu.go
index f670437..c75dc97 100644
--- a/internal/pkg/onuadaptercore/openonu.go
+++ b/internal/pkg/onuadaptercore/openonu.go
@@ -424,7 +424,14 @@
//Update_pm_config returns PmConfigs nil or error
func (oo *OpenONUAC) Update_pm_config(ctx context.Context, device *voltha.Device, pmConfigs *voltha.PmConfigs) error {
- return errors.New("unImplemented")
+ logger.Infow(ctx, "update-pm-config", log.Fields{"device-id": device.Id})
+ if handler := oo.getDeviceHandler(ctx, device.Id, false); handler != nil {
+ handler.updatePmConfig(ctx, pmConfigs)
+ } else {
+ logger.Warnw(ctx, "no handler found for update-pm-config", log.Fields{"device-id": device.Id})
+ return fmt.Errorf(fmt.Sprintf("handler-not-found-%s", device.Id))
+ }
+ return nil
}
//Receive_packet_out sends packet out to the device
@@ -506,7 +513,7 @@
return nil, errors.New("unImplemented")
}
-//Single_get_value_request as needed by voltha-lib-go update to 4.0.xx?
+// Single_get_value_request - unimplemented
func (oo *OpenONUAC) Single_get_value_request(ctx context.Context, request extension.SingleGetValueRequest) (*extension.SingleGetValueResponse, error) {
return nil, errors.New("unImplemented")
}
diff --git a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/adapters/common/performance_metrics.go b/vendor/github.com/opencord/voltha-lib-go/v4/pkg/adapters/common/performance_metrics.go
index 6705c72..3b6d4f9 100644
--- a/vendor/github.com/opencord/voltha-lib-go/v4/pkg/adapters/common/performance_metrics.go
+++ b/vendor/github.com/opencord/voltha-lib-go/v4/pkg/adapters/common/performance_metrics.go
@@ -48,6 +48,11 @@
}
}
+// UpdateFrequency will update the frequency.
+func (pm *PmMetrics) UpdateFrequency(frequency uint32) {
+ pm.frequency = frequency
+}
+
func Metrics(pmNames []string) PmMetricsOption {
return func(args *PmMetrics) {
args.metrics = make(map[string]*voltha.PmConfig)
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 2d33169..d0ec634 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -74,7 +74,7 @@
# github.com/opencord/omci-lib-go v0.13.4
github.com/opencord/omci-lib-go
github.com/opencord/omci-lib-go/generated
-# github.com/opencord/voltha-lib-go/v4 v4.0.5
+# github.com/opencord/voltha-lib-go/v4 v4.0.6
github.com/opencord/voltha-lib-go/v4/pkg/adapters
github.com/opencord/voltha-lib-go/v4/pkg/adapters/adapterif
github.com/opencord/voltha-lib-go/v4/pkg/adapters/common