VOL-3318: PM Framework changes to support configuration updates.
Change-Id: I10690991ae21990cd2d7277b1277b911d331152d
diff --git a/internal/pkg/onuadaptercore/device_handler.go b/internal/pkg/onuadaptercore/device_handler.go
index 64e8f24..7405f20 100644
--- a/internal/pkg/onuadaptercore/device_handler.go
+++ b/internal/pkg/onuadaptercore/device_handler.go
@@ -31,7 +31,6 @@
"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"
@@ -125,19 +124,6 @@
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
@@ -154,7 +140,7 @@
AdapterProxy adapterif.AdapterProxy
EventProxy eventif.EventProxy
- pmMetrics *common.PmMetrics
+ pmConfigs *voltha.PmConfigs
pOpenOnuAc *OpenONUAC
pDeviceStateFsm *fsm.FSM
@@ -210,32 +196,13 @@
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)
- }
+ if dh.device.PmConfigs != nil { // can happen after onu adapter restart
+ dh.pmConfigs = cloned.PmConfigs
+ } /* else {
+ // will be populated when onu_metrics_mananger is initialized.
+ }*/
- 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,
@@ -289,9 +256,11 @@
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})
+ if device.PmConfigs == nil { // device.PmConfigs is not nil in cases when adapter restarts. We should not re-set the core again.
+ // Now, set the initial PM configuration for that device
+ if err := dh.coreProxy.DevicePMConfigUpdate(ctx, dh.pmConfigs); 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})
@@ -2553,42 +2522,151 @@
}
// 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})
+func (dh *deviceHandler) updatePmConfig(ctx context.Context, pmConfigs *voltha.PmConfigs) error {
+ var errorsList []error
+ logger.Infow(ctx, "update-pm-config", log.Fields{"device-id": dh.device.Id, "new-pm-configs": pmConfigs, "old-pm-config": dh.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})
+ errorsList = append(dh.handleGlobalPmConfigUpdates(ctx, pmConfigs), errorsList...)
+ errorsList = append(dh.handleGroupPmConfigUpdates(ctx, pmConfigs), errorsList...)
+ errorsList = append(dh.handleStandalonePmConfigUpdates(ctx, pmConfigs), errorsList...)
+
+ // Note that if more than one pm config field is updated in a given call, it is possible that partial pm config is handled
+ // successfully.
+ // TODO: Although it is possible to revert to old config in case of partial failure, the code becomes quite complex. Needs more investigation
+ // Is it possible the rw-core reverts to old config on partial failure but adapter retains a partial new config?
+ if len(errorsList) > 0 {
+ logger.Errorw(ctx, "one-or-more-pm-config-failed", log.Fields{"device-id": dh.deviceID, "pmConfig": dh.pmConfigs})
+ return fmt.Errorf("errors-handling-one-or-more-pm-config, errors:%v", errorsList)
}
+ logger.Infow(ctx, "pm-config-updated", log.Fields{"device-id": dh.deviceID, "pmConfig": dh.pmConfigs})
+ return nil
}
+func (dh *deviceHandler) handleGlobalPmConfigUpdates(ctx context.Context, pmConfigs *voltha.PmConfigs) []error {
+ var err error
+ var errorsList []error
+ logger.Infow(ctx, "handling-global-pm-config-params", log.Fields{"device-id": dh.device.Id})
+
+ if pmConfigs.DefaultFreq != dh.pmConfigs.DefaultFreq {
+ if err = dh.pOnuMetricsMgr.updateDefaultFrequency(ctx, pmConfigs); err != nil {
+ errorsList = append(errorsList, err)
+ }
+ }
+
+ return errorsList
+}
+
+func (dh *deviceHandler) handleGroupPmConfigUpdates(ctx context.Context, pmConfigs *voltha.PmConfigs) []error {
+ var err error
+ var errorsList []error
+ logger.Debugw(ctx, "handling-group-pm-config-params", log.Fields{"device-id": dh.device.Id})
+ // Check if group metric related config is updated
+ for _, v := range pmConfigs.Groups {
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.RLock()
+ m, ok := dh.pOnuMetricsMgr.groupMetricMap[v.GroupName]
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.RUnlock()
+
+ if ok && m.frequency != v.GroupFreq {
+ if err = dh.pOnuMetricsMgr.updateGroupFreq(ctx, v.GroupName, pmConfigs); err != nil {
+ errorsList = append(errorsList, err)
+ }
+ }
+ if ok && m.enabled != v.Enabled {
+ if err = dh.pOnuMetricsMgr.updateGroupSupport(ctx, v.GroupName, pmConfigs); err != nil {
+ errorsList = append(errorsList, err)
+ }
+ }
+ }
+ return errorsList
+}
+
+func (dh *deviceHandler) handleStandalonePmConfigUpdates(ctx context.Context, pmConfigs *voltha.PmConfigs) []error {
+ var err error
+ var errorsList []error
+ logger.Debugw(ctx, "handling-individual-pm-config-params", log.Fields{"device-id": dh.device.Id})
+ // Check if standalone metric related config is updated
+ for _, v := range pmConfigs.Metrics {
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.RLock()
+ m, ok := dh.pOnuMetricsMgr.groupMetricMap[v.Name]
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.RUnlock()
+
+ if ok && m.frequency != v.SampleFreq {
+ if err = dh.pOnuMetricsMgr.updateMetricFreq(ctx, v.Name, pmConfigs); err != nil {
+ errorsList = append(errorsList, err)
+ }
+ }
+ if ok && m.enabled != v.Enabled {
+ if err = dh.pOnuMetricsMgr.updateMetricSupport(ctx, v.Name, pmConfigs); err != nil {
+ errorsList = append(errorsList, err)
+ }
+ }
+ }
+ return errorsList
+}
+
+// nolint: gocyclo
func (dh *deviceHandler) startCollector(ctx context.Context) {
logger.Debugf(ctx, "startingCollector")
// Start routine to process OMCI GET Responses
go dh.pOnuMetricsMgr.processOmciMessages(ctx)
-
+ // Initialize the next metric collection time.
+ // Normally done when the onu_metrics_manager is initialized the first time, but needed again later when ONU is
+ // reset like onu rebooted.
+ dh.pOnuMetricsMgr.initializeMetricCollectionTime(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)
- }()
+ case <-time.After(time.Duration(FrequencyGranularity) * time.Second): // Check every FrequencyGranularity to see if it is time for collecting metrics
+ if !dh.pmConfigs.FreqOverride { // If FreqOverride is false, then nextGlobalMetricCollectionTime applies
+ // If the current time is eqaul to or greater than the nextGlobalMetricCollectionTime, collect the group and standalone metrics
+ if time.Now().Equal(dh.pOnuMetricsMgr.nextGlobalMetricCollectionTime) || time.Now().After(dh.pOnuMetricsMgr.nextGlobalMetricCollectionTime) {
+ go dh.pOnuMetricsMgr.collectAllGroupAndStandaloneMetrics(ctx)
+ }
+ // Update the next metric collection time.
+ dh.pOnuMetricsMgr.nextGlobalMetricCollectionTime = time.Now().Add(time.Duration(dh.pmConfigs.DefaultFreq) * time.Second)
+ } else {
+ if dh.pmConfigs.Grouped { // metrics are managed as a group
+ // parse through the group and standalone metrics to see it is time to collect their metrics
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.RLock() // Rlock as we are reading groupMetricMap and standaloneMetricMap
- go func() {
- logger.Debug(ctx, "startCollector before collecting uni metrics")
- metricInfo := dh.pOnuMetricsMgr.collectUniStatusMetrics(ctx)
- dh.pOnuMetricsMgr.publishMetrics(ctx, metricInfo)
- }()
+ for n, g := range dh.pOnuMetricsMgr.groupMetricMap {
+ // If the group is enabled AND (current time is equal to OR after nextCollectionInterval, collect the group metric)
+ if g.enabled && (time.Now().Equal(g.nextCollectionInterval) || time.Now().After(g.nextCollectionInterval)) {
+ go dh.pOnuMetricsMgr.collectGroupMetric(ctx, n)
+ }
+ }
+ for n, m := range dh.pOnuMetricsMgr.standaloneMetricMap {
+ // If the standalone is enabled AND (current time is equal to OR after nextCollectionInterval, collect the metric)
+ if m.enabled && (time.Now().Equal(m.nextCollectionInterval) || time.Now().After(m.nextCollectionInterval)) {
+ go dh.pOnuMetricsMgr.collectStandaloneMetric(ctx, n)
+ }
+ }
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.RUnlock()
+
+ // parse through the group and update the next metric collection time
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.Lock() // Lock as we are writing the next metric collection time
+ for _, g := range dh.pOnuMetricsMgr.groupMetricMap {
+ // If group enabled, and the nextCollectionInterval is old (before or equal to current time), update the next collection time stamp
+ if g.enabled && (g.nextCollectionInterval.Before(time.Now()) || g.nextCollectionInterval.Equal(time.Now())) {
+ g.nextCollectionInterval = time.Now().Add(time.Duration(g.frequency) * time.Second)
+ }
+ }
+ // parse through the standalone metrics and update the next metric collection time
+ for _, m := range dh.pOnuMetricsMgr.standaloneMetricMap {
+ // If standalone metrics enabled, and the nextCollectionInterval is old (before or equal to current time), update the next collection time stamp
+ if m.enabled && (m.nextCollectionInterval.Before(time.Now()) || m.nextCollectionInterval.Equal(time.Now())) {
+ m.nextCollectionInterval = time.Now().Add(time.Duration(m.frequency) * time.Second)
+ }
+ }
+ dh.pOnuMetricsMgr.onuMetricsManagerLock.Unlock()
+ } /* else { // metrics are not managed as a group
+ // TODO: We currently do not have standalone metrics. When available, add code here to fetch the metric.
+ } */
+ }
}
}
}
diff --git a/internal/pkg/onuadaptercore/onu_metrics_manager.go b/internal/pkg/onuadaptercore/onu_metrics_manager.go
index 3e43275..4316186 100644
--- a/internal/pkg/onuadaptercore/onu_metrics_manager.go
+++ b/internal/pkg/onuadaptercore/onu_metrics_manager.go
@@ -25,9 +25,62 @@
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"
+ "sync"
"time"
)
+// general constants used for overall Metric Collection management
+const (
+ DefaultMetricCollectionFrequency = 15 * 60 // unit in seconds. This setting can be changed from voltha NBI PmConfig configuration
+ GroupMetricEnabled = true // This is READONLY and cannot be changed from VOLTHA NBI
+ DefaultFrequencyOverrideEnabled = true // This is READONLY and cannot be changed from VOLTHA NBI
+ FrequencyGranularity = 5 // The frequency (in seconds) has to be multiple of 5. This setting cannot changed later.
+)
+
+// OpticalPowerGroupMetrics are supported optical pm names
+var OpticalPowerGroupMetrics = map[string]voltha.PmConfig_PmType{
+ "ani_g_instance_id": voltha.PmConfig_CONTEXT,
+ "transmit_power": voltha.PmConfig_GAUGE,
+ "receive_power": voltha.PmConfig_GAUGE,
+}
+
+// OpticalPowerGroupMetrics specific constants
+const (
+ OpticalPowerGroupMetricName = "OpticalPower"
+ OpticalPowerGroupMetricEnabled = true // This setting can be changed from voltha NBI PmConfig configuration
+ OpticalPowerMetricGroupCollectionFrequency = 5 * 60 // unit in seconds. This setting can be changed from voltha NBI PmConfig configuration
+)
+
+// UniStatusGroupMetrics are supported UNI status names
+var UniStatusGroupMetrics = map[string]voltha.PmConfig_PmType{
+ "uni_port_no": voltha.PmConfig_CONTEXT,
+ "ethernet_type": voltha.PmConfig_GAUGE,
+ "oper_status": voltha.PmConfig_GAUGE,
+ "uni_admin_state": voltha.PmConfig_GAUGE,
+}
+
+// UniStatusGroupMetrics specific constants
+const (
+ UniStatusGroupMetricName = "UniStatus"
+ UniStatusGroupMetricEnabled = true // This setting can be changed from voltha NBI PmConfig configuration
+ UniStatusMetricGroupCollectionFrequency = 5 * 60 // unit in seconds. This setting can be changed from voltha NBI PmConfig configuration
+)
+
+type groupMetric struct {
+ groupName string
+ enabled bool
+ frequency uint32 // valid only if FrequencyOverride is enabled.
+ metricMap map[string]voltha.PmConfig_PmType
+ nextCollectionInterval time.Time // valid only if FrequencyOverride is enabled.
+}
+
+type standaloneMetric struct {
+ metricName string
+ enabled bool
+ frequency uint32 // valid only if FrequencyOverride is enabled.
+ nextCollectionInterval time.Time // valid only if FrequencyOverride is enabled.
+}
+
type onuMetricsManager struct {
pDeviceHandler *deviceHandler
@@ -35,10 +88,21 @@
opticalMetricsChan chan me.AttributeValueMap
uniStatusMetricsChan chan me.AttributeValueMap
+ groupMetricMap map[string]*groupMetric
+ standaloneMetricMap map[string]*standaloneMetric
+
stopProcessingOmciResponses chan bool
+
+ nextGlobalMetricCollectionTime time.Time // valid only if pmConfig.FreqOverride is set to false.
+
+ onuMetricsManagerLock sync.RWMutex
}
// newonuMetricsManager returns a new instance of the newonuMetricsManager
+// Note that none of the context stored internally in onuMetricsManager is backed up on KV store for resiliency.
+// Metric collection is not a critical operation that needs support for resiliency. On adapter restart, some context
+// could be lost (except for Device.PmConfigs which is backed up the rw-core on KV store). An example of information
+// that is lost on adapter restart is nextCollectionInterval time.
func newonuMetricsManager(ctx context.Context, dh *deviceHandler) *onuMetricsManager {
var metricsManager onuMetricsManager
@@ -50,20 +114,358 @@
metricsManager.uniStatusMetricsChan = make(chan me.AttributeValueMap)
metricsManager.stopProcessingOmciResponses = make(chan bool)
+ metricsManager.groupMetricMap = make(map[string]*groupMetric)
+ metricsManager.standaloneMetricMap = make(map[string]*standaloneMetric)
+
+ if dh.pmConfigs == nil { // dh.pmConfigs is NOT nil if adapter comes back from a restart. We should NOT go back to defaults in this case
+ dh.pmConfigs = &voltha.PmConfigs{}
+ dh.pmConfigs.Id = dh.deviceID
+ dh.pmConfigs.DefaultFreq = DefaultMetricCollectionFrequency
+ dh.pmConfigs.Grouped = GroupMetricEnabled
+ dh.pmConfigs.FreqOverride = DefaultFrequencyOverrideEnabled
+
+ // Populate group metrics.
+ // Lets populate irrespective of GroupMetricEnabled is true or not.
+ // The group metrics collection will decided on this flag later
+
+ // Populate optical power group metrics
+ var opPmConfigSlice []*voltha.PmConfig
+ for k, v := range OpticalPowerGroupMetrics {
+ opPmConfigSlice = append(opPmConfigSlice, &voltha.PmConfig{Name: k, Type: v})
+ }
+ opticalPowerGroupMetric := voltha.PmGroupConfig{
+ GroupName: OpticalPowerGroupMetricName,
+ Enabled: OpticalPowerGroupMetricEnabled,
+ GroupFreq: OpticalPowerMetricGroupCollectionFrequency,
+ Metrics: opPmConfigSlice,
+ }
+ dh.pmConfigs.Groups = append(dh.pmConfigs.Groups, &opticalPowerGroupMetric)
+
+ // Populate uni status group metrics
+ var uniStPmConfigSlice []*voltha.PmConfig
+ for k, v := range UniStatusGroupMetrics {
+ uniStPmConfigSlice = append(uniStPmConfigSlice, &voltha.PmConfig{Name: k, Type: v})
+ }
+ uniStatusGroupMetric := voltha.PmGroupConfig{
+ GroupName: UniStatusGroupMetricName,
+ Enabled: UniStatusGroupMetricEnabled,
+ GroupFreq: UniStatusMetricGroupCollectionFrequency,
+ Metrics: uniStPmConfigSlice,
+ }
+ dh.pmConfigs.Groups = append(dh.pmConfigs.Groups, &uniStatusGroupMetric)
+
+ // Add standalone metric (if present) after this (will be added to dh.pmConfigs.Metrics)
+ }
+
+ // Populate local group metric structures
+ for _, g := range dh.pmConfigs.Groups {
+ metricsManager.groupMetricMap[g.GroupName] = &groupMetric{
+ groupName: g.GroupName,
+ enabled: g.Enabled,
+ frequency: g.GroupFreq,
+ }
+ switch g.GroupName {
+ case OpticalPowerGroupMetricName:
+ metricsManager.groupMetricMap[g.GroupName].metricMap = OpticalPowerGroupMetrics
+ case UniStatusGroupMetricName:
+ metricsManager.groupMetricMap[g.GroupName].metricMap = UniStatusGroupMetrics
+
+ default:
+ logger.Errorw(ctx, "unhandled-group-name", log.Fields{"groupName": g.GroupName})
+ }
+ }
+
+ // Populate local standalone metric structures
+ for _, m := range dh.pmConfigs.Metrics {
+ metricsManager.standaloneMetricMap[m.Name] = &standaloneMetric{
+ metricName: m.Name,
+ enabled: m.Enabled,
+ frequency: m.SampleFreq,
+ }
+ switch m.Name {
+ // None exist as of now. Add when available.
+ default:
+ logger.Errorw(ctx, "unhandled-metric-name", log.Fields{"metricName": m.Name})
+ }
+ }
+
+ // initialize the next metric collection intervals.
+ metricsManager.initializeMetricCollectionTime(ctx)
+ logger.Info(ctx, "init-onuMetricsManager completed", log.Fields{"device-id": dh.deviceID})
return &metricsManager
}
+func (mm *onuMetricsManager) initializeMetricCollectionTime(ctx context.Context) {
+ if mm.pDeviceHandler.pmConfigs.FreqOverride {
+ // If mm.pDeviceHandler.pmConfigs.FreqOverride is set to true, then group/standalone metric specific interval applies
+ mm.onuMetricsManagerLock.Lock()
+ defer mm.onuMetricsManagerLock.Unlock()
+ for _, v := range mm.groupMetricMap {
+ if v.enabled {
+ v.nextCollectionInterval = time.Now().Add(time.Duration(v.frequency) * time.Second)
+ }
+ }
+
+ for _, v := range mm.standaloneMetricMap {
+ if v.enabled {
+ v.nextCollectionInterval = time.Now().Add(time.Duration(v.frequency) * time.Second)
+ }
+ }
+ } else {
+ // If mm.pDeviceHandler.pmConfigs.FreqOverride is set to false, then overall metric specific interval applies
+ mm.nextGlobalMetricCollectionTime = time.Now().Add(time.Duration(mm.pDeviceHandler.pmConfigs.DefaultFreq) * time.Second)
+ }
+ logger.Infow(ctx, "initialized standalone group/metric collection time", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+}
+
+func (mm *onuMetricsManager) updateDefaultFrequency(ctx context.Context, pmConfigs *voltha.PmConfigs) error {
+ // Verify that the configured DefaultFrequency is > 0 and is a multiple of FrequencyGranularity
+ if pmConfigs.DefaultFreq == 0 && pmConfigs.DefaultFreq%FrequencyGranularity != 0 {
+ logger.Errorf(ctx, "frequency-%u-should-be-a-multiple-of-%u", pmConfigs.DefaultFreq, FrequencyGranularity)
+ return fmt.Errorf("frequency-%d-should-be-a-multiple-of-%d", pmConfigs.DefaultFreq, FrequencyGranularity)
+ }
+ mm.pDeviceHandler.pmConfigs.DefaultFreq = pmConfigs.DefaultFreq
+ // re-set the nextGlobalMetricCollectionTime based on the new DefaultFreq
+ mm.nextGlobalMetricCollectionTime = time.Now().Add(time.Duration(mm.pDeviceHandler.pmConfigs.DefaultFreq) * time.Second)
+ logger.Debugw(ctx, "frequency-updated--new-frequency", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "frequency": mm.pDeviceHandler.pmConfigs.DefaultFreq})
+ return nil
+}
+
+func (mm *onuMetricsManager) updateGroupFreq(ctx context.Context, aGroupName string, pmConfigs *voltha.PmConfigs) error {
+ var newGroupFreq uint32
+ found := false
+ groupSliceIdx := 0
+ var group *voltha.PmGroupConfig
+ for groupSliceIdx, group = range pmConfigs.Groups {
+ if group.GroupName == aGroupName {
+ if group.GroupFreq != 0 { // freq 0 not allowed
+ newGroupFreq = group.GroupFreq
+ found = true
+ break
+ }
+ }
+ }
+ // if not found update group freq and next collection interval for the group
+ if !found {
+ logger.Errorw(ctx, "group name not found", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "groupName": aGroupName})
+ return fmt.Errorf("group-name-not-found-%v", aGroupName)
+ }
+
+ updated := false
+ mm.onuMetricsManagerLock.Lock()
+ defer mm.onuMetricsManagerLock.Unlock()
+ for k, v := range mm.groupMetricMap {
+ if k == aGroupName && newGroupFreq != 0 { // freq 0 not allowed
+ v.frequency = newGroupFreq
+ // update internal pm config
+ mm.pDeviceHandler.pmConfigs.Groups[groupSliceIdx].GroupFreq = newGroupFreq
+ // Also updated the next group metric collection time from now
+ v.nextCollectionInterval = time.Now().Add(time.Duration(newGroupFreq) * time.Second)
+ updated = true
+ logger.Infow(ctx, "group frequency updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "newGroupFreq": newGroupFreq, "groupName": aGroupName})
+ }
+ }
+ if !updated {
+ logger.Errorw(ctx, "group frequency not updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "newGroupFreq": newGroupFreq, "groupName": aGroupName})
+ return fmt.Errorf("internal-error-during-group-freq-update--groupname-%s-freq-%d", aGroupName, newGroupFreq)
+ }
+ return nil
+}
+
+func (mm *onuMetricsManager) updateMetricFreq(ctx context.Context, aMetricName string, pmConfigs *voltha.PmConfigs) error {
+ var newMetricFreq uint32
+ found := false
+ metricSliceIdx := 0
+ var metric *voltha.PmConfig
+ for metricSliceIdx, metric = range pmConfigs.Metrics {
+ if metric.Name == aMetricName {
+ if metric.SampleFreq != 0 { // freq 0 not allowed
+ newMetricFreq = metric.SampleFreq
+ found = true
+ break
+ }
+ }
+ }
+ if !found {
+ logger.Errorw(ctx, "metric name not found", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "metricName": aMetricName})
+ return fmt.Errorf("metric-name-not-found-%v", aMetricName)
+ }
+
+ updated := false
+ mm.onuMetricsManagerLock.Lock()
+ defer mm.onuMetricsManagerLock.Unlock()
+ for k, v := range mm.groupMetricMap {
+ if k == aMetricName && newMetricFreq != 0 {
+ v.frequency = newMetricFreq
+ // update internal pm config
+ mm.pDeviceHandler.pmConfigs.Metrics[metricSliceIdx].SampleFreq = newMetricFreq
+ // Also updated the next standalone metric collection time from now
+ v.nextCollectionInterval = time.Now().Add(time.Duration(newMetricFreq) * time.Second)
+ updated = true
+ logger.Infow(ctx, "metric frequency updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "newMetricFreq": newMetricFreq, "aMetricName": aMetricName})
+ }
+ }
+ if !updated {
+ logger.Errorw(ctx, "metric frequency not updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "newMetricFreq": newMetricFreq, "aMetricName": aMetricName})
+ return fmt.Errorf("internal-error-during-standalone-metric-update--matricnane-%s-freq-%d", aMetricName, newMetricFreq)
+ }
+ return nil
+}
+
+func (mm *onuMetricsManager) updateGroupSupport(ctx context.Context, aGroupName string, pmConfigs *voltha.PmConfigs) error {
+ groupSliceIdx := 0
+ var group *voltha.PmGroupConfig
+
+ for groupSliceIdx, group = range pmConfigs.Groups {
+ if group.GroupName == aGroupName {
+ break
+ }
+ }
+ if group == nil {
+ logger.Errorw(ctx, "group metric not found", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "groupName": aGroupName})
+ return fmt.Errorf("group-not-found--groupName-%s", aGroupName)
+ }
+
+ updated := false
+ mm.onuMetricsManagerLock.Lock()
+ defer mm.onuMetricsManagerLock.Unlock()
+ for k, v := range mm.groupMetricMap {
+ if k == aGroupName && v.enabled != group.Enabled {
+ mm.pDeviceHandler.pmConfigs.Groups[groupSliceIdx].Enabled = group.Enabled
+ v.enabled = group.Enabled
+ // If the group is now enabled and frequency override is enabled, set the next group metric collection time
+ if group.Enabled && mm.pDeviceHandler.pmConfigs.FreqOverride {
+ v.nextCollectionInterval = time.Now().Add(time.Duration(v.frequency) * time.Second)
+ }
+ updated = true
+ logger.Infow(ctx, "group metric support updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "groupName": aGroupName, "enabled": group.Enabled})
+ }
+ }
+
+ if !updated {
+ logger.Errorw(ctx, "group metric support not updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "groupName": aGroupName})
+ return fmt.Errorf("internal-error-during-group-support-update--groupName-%s", aGroupName)
+ }
+ return nil
+}
+
+func (mm *onuMetricsManager) updateMetricSupport(ctx context.Context, aMetricName string, pmConfigs *voltha.PmConfigs) error {
+ metricSliceIdx := 0
+ var metric *voltha.PmConfig
+
+ for metricSliceIdx, metric = range pmConfigs.Metrics {
+ if metric.Name == aMetricName {
+ break
+ }
+ }
+
+ if metric == nil {
+ logger.Errorw(ctx, "standalone metric not found", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "metricName": aMetricName})
+ return fmt.Errorf("metric-not-found--metricname-%s", aMetricName)
+ }
+
+ updated := false
+ mm.onuMetricsManagerLock.Lock()
+ defer mm.onuMetricsManagerLock.Unlock()
+ for k, v := range mm.standaloneMetricMap {
+ if k == aMetricName && v.enabled != metric.Enabled {
+ mm.pDeviceHandler.pmConfigs.Metrics[metricSliceIdx].Enabled = metric.Enabled
+ v.enabled = metric.Enabled
+ // If the standalone metric is now enabled and frequency override is enabled, set the next metric collection time
+ if metric.Enabled && mm.pDeviceHandler.pmConfigs.FreqOverride {
+ v.nextCollectionInterval = time.Now().Add(time.Duration(v.frequency) * time.Second)
+ }
+ updated = true
+ logger.Infow(ctx, "standalone metric support updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "metricName": aMetricName, "enabled": metric.Enabled})
+ }
+ }
+ if !updated {
+ logger.Errorw(ctx, "standalone metric support not updated", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "metricName": aMetricName})
+ return fmt.Errorf("internal-error-during-standalone-support-update--metricname-%s", aMetricName)
+ }
+ return nil
+}
+
+func (mm *onuMetricsManager) collectAllGroupAndStandaloneMetrics(ctx context.Context) {
+ if mm.pDeviceHandler.pmConfigs.Grouped { // metrics are managed as a group.
+ go mm.collectAllGroupMetrics(ctx)
+ } else {
+ go mm.collectAllStandaloneMetrics(ctx)
+ }
+}
+
+func (mm *onuMetricsManager) collectAllGroupMetrics(ctx context.Context) {
+ go func() {
+ logger.Debug(ctx, "startCollector before collecting optical metrics")
+ metricInfo := mm.collectOpticalMetrics(ctx)
+ if metricInfo != nil {
+ mm.publishMetrics(ctx, metricInfo)
+ }
+ }()
+
+ go func() {
+ logger.Debug(ctx, "startCollector before collecting uni metrics")
+ metricInfo := mm.collectUniStatusMetrics(ctx)
+ if metricInfo != nil {
+ mm.publishMetrics(ctx, metricInfo)
+ }
+ }()
+
+ // Add more here
+}
+
+func (mm *onuMetricsManager) collectAllStandaloneMetrics(ctx context.Context) {
+ // None exists as of now, add when available here
+}
+
+func (mm *onuMetricsManager) collectGroupMetric(ctx context.Context, groupName string) {
+ switch groupName {
+ case OpticalPowerGroupMetricName:
+ go func() {
+ if mi := mm.collectOpticalMetrics(ctx); mm != nil {
+ mm.publishMetrics(ctx, mi)
+ }
+ }()
+ case UniStatusGroupMetricName:
+ go func() {
+ if mi := mm.collectUniStatusMetrics(ctx); mm != nil {
+ mm.publishMetrics(ctx, mi)
+ }
+ }()
+ default:
+ logger.Errorw(ctx, "unhandled group metric name", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "groupName": groupName})
+ }
+}
+
+func (mm *onuMetricsManager) collectStandaloneMetric(ctx context.Context, metricName string) {
+ switch metricName {
+ // None exist as of now, add when available
+ default:
+ logger.Errorw(ctx, "unhandled standalone metric name", log.Fields{"device-id": mm.pDeviceHandler.deviceID, "metricName": metricName})
+ }
+}
+
+// collectOpticalMetrics collects groups metrics related to optical power from ani-g ME.
func (mm *onuMetricsManager) collectOpticalMetrics(ctx context.Context) []*voltha.MetricInformation {
logger.Debugw(ctx, "collectOpticalMetrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+
+ mm.onuMetricsManagerLock.RLock()
+ if !mm.groupMetricMap[OpticalPowerGroupMetricName].enabled {
+ mm.onuMetricsManagerLock.RUnlock()
+ logger.Debugw(ctx, "optical power group metric is not enabled", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ return nil
+ }
+ mm.onuMetricsManagerLock.RUnlock()
+
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()
+ raisedTs := time.Now().Unix()
mmd := voltha.MetricMetaData{
- Title: "OpticalMetrics",
+ Title: OpticalPowerGroupMetricName,
Ts: float64(raisedTs),
Context: metricsContext,
DeviceId: mm.pDeviceHandler.deviceID,
@@ -71,14 +473,6 @@
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:
@@ -97,12 +491,20 @@
break loop
}
// Populate metric only if it was enabled.
- for _, v := range enabledMetrics {
- switch v {
+ for k := range OpticalPowerGroupMetrics {
+ switch k {
+ case "ani_g_instance_id":
+ if val, ok := meAttributes["ManagedEntityId"]; ok && val != nil {
+ opticalMetrics[k] = float32(val.(uint16))
+ }
case "transmit_power":
- opticalMetrics["transmit_power"] = float32(meAttributes["TransmitOpticalLevel"].(uint16))
+ if val, ok := meAttributes["TransmitOpticalLevel"]; ok && val != nil {
+ opticalMetrics[k] = float32(val.(uint16))
+ }
case "receive_power":
- opticalMetrics["receive_power"] = float32(meAttributes["OpticalSignalLevel"].(uint16))
+ if val, ok := meAttributes["OpticalSignalLevel"]; ok && val != nil {
+ opticalMetrics[k] = float32(val.(uint16))
+ }
default:
// do nothing
}
@@ -117,17 +519,25 @@
return metricInfoSlice
}
-// Note: UNI status does not seem to be a metric, but this is being treated as metric in Python implementation
+// collectUniStatusMetrics collects UNI status group metric from various MEs (uni-g, pptp and veip).
// nolint: gocyclo
func (mm *onuMetricsManager) collectUniStatusMetrics(ctx context.Context) []*voltha.MetricInformation {
logger.Debugw(ctx, "collectUniStatusMetrics", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ mm.onuMetricsManagerLock.RLock()
+ if !mm.groupMetricMap[UniStatusGroupMetricName].enabled {
+ mm.onuMetricsManagerLock.RUnlock()
+ logger.Debugw(ctx, "uni status group metric is not enabled", log.Fields{"device-id": mm.pDeviceHandler.deviceID})
+ return nil
+ }
+ mm.onuMetricsManagerLock.RUnlock()
+
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()
+ raisedTs := time.Now().Unix()
mmd := voltha.MetricMetaData{
Title: "UniStatus", // Is this ok to hard code?
Ts: float64(raisedTs),
@@ -137,15 +547,6 @@
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:
@@ -167,14 +568,24 @@
break loop1
}
// Populate metric only if it was enabled.
- for _, v := range enabledMetrics {
- switch v {
+ for k := range UniStatusGroupMetrics {
+ switch k {
case "uni_admin_state":
- unigMetrics["uni_admin_state"] = float32(meAttributes["AdministrativeState"].(byte))
+ if val, ok := meAttributes["AdministrativeState"]; ok && val != nil {
+ unigMetrics[k] = float32(val.(byte))
+ }
default:
// do nothing
}
}
+ var entityID uint32
+ if val, ok := meAttributes["ManagedEntityId"]; ok && val != nil {
+ entityID = uint32(val.(uint16))
+ }
+ // TODO: Rlock needed for reading uniEntityMap?
+ if uniPort, ok := mm.pDeviceHandler.uniEntityMap[entityID]; ok && uniPort != nil {
+ unigMetrics["uni_port_no"] = float32(uniPort.portNo)
+ }
// 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)
@@ -203,19 +614,34 @@
}
// Populate metric only if it was enabled.
- for _, v := range enabledMetrics {
- switch v {
+ for k := range UniStatusGroupMetrics {
+ switch k {
case "ethernet_type":
- pptpMetrics["ethernet_type"] = float32(meAttributes["SensedType"].(byte))
+ if val, ok := meAttributes["SensedType"]; ok && val != nil {
+ pptpMetrics[k] = float32(val.(byte))
+ }
case "oper_status":
- pptpMetrics["oper_status"] = float32(meAttributes["OperationalState"].(byte))
+ if val, ok := meAttributes["OperationalState"]; ok && val != nil {
+ pptpMetrics[k] = float32(val.(byte))
+ }
case "uni_admin_state":
- pptpMetrics["uni_admin_state"] = float32(meAttributes["AdministrativeState"].(byte))
+ if val, ok := meAttributes["AdministrativeState"]; ok && val != nil {
+ pptpMetrics[k] = float32(val.(byte))
+ }
default:
// do nothing
}
}
}
+ var entityID uint32
+ if val, ok := meAttributes["ManagedEntityId"]; ok && val != nil {
+ entityID = uint32(val.(uint16))
+ }
+ // TODO: Rlock needed for reading uniEntityMap?
+ if uniPort, ok := mm.pDeviceHandler.uniEntityMap[entityID]; ok && uniPort != nil {
+ pptpMetrics["uni_port_no"] = float32(uniPort.portNo)
+ }
+
// 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)
@@ -228,7 +654,7 @@
// 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)
+ veipMetrics := 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 {
@@ -243,19 +669,33 @@
}
// Populate metric only if it was enabled.
- for _, v := range enabledMetrics {
- switch v {
+ for k := range UniStatusGroupMetrics {
+ switch k {
case "oper_status":
- pptpMetrics["oper_status"] = float32(meAttributes["OperationalState"].(byte))
+ if val, ok := meAttributes["OperationalState"]; ok && val != nil {
+ veipMetrics[k] = float32(val.(byte))
+ }
case "uni_admin_state":
- pptpMetrics["uni_admin_state"] = float32(meAttributes["AdministrativeState"].(byte))
+ if val, ok := meAttributes["AdministrativeState"]; ok && val != nil {
+ veipMetrics[k] = float32(val.(byte))
+ }
default:
// do nothing
}
}
}
+
+ var entityID uint32
+ if val, ok := meAttributes["ManagedEntityId"]; ok && val != nil {
+ entityID = uint32(meAttributes["ManagedEntityId"].(uint16))
+ }
+ // TODO: Rlock needed for reading uniEntityMap?
+ if uniPort, ok := mm.pDeviceHandler.uniEntityMap[entityID]; ok && uniPort != nil {
+ veipMetrics["uni_port_no"] = float32(uniPort.portNo)
+ }
+
// create slice of metrics given that there could be more than one VEIP instance
- metricInfo := voltha.MetricInformation{Metadata: &mmd, Metrics: pptpMetrics}
+ metricInfo := voltha.MetricInformation{Metadata: &mmd, Metrics: veipMetrics}
metricInfoSlice = append(metricInfoSlice, &metricInfo)
}
@@ -265,7 +705,7 @@
// publishMetrics publishes the metrics on kafka
func (mm *onuMetricsManager) publishMetrics(ctx context.Context, metricInfo []*voltha.MetricInformation) {
var ke voltha.KpiEvent2
- ts := time.Now().UnixNano()
+ ts := time.Now().Unix()
ke.SliceData = metricInfo
ke.Type = voltha.KpiEventType_slice
ke.Ts = float64(ts)
diff --git a/internal/pkg/onuadaptercore/openonu.go b/internal/pkg/onuadaptercore/openonu.go
index 6ab0f78..2bfb3c9 100644
--- a/internal/pkg/onuadaptercore/openonu.go
+++ b/internal/pkg/onuadaptercore/openonu.go
@@ -426,12 +426,10 @@
func (oo *OpenONUAC) Update_pm_config(ctx context.Context, device *voltha.Device, pmConfigs *voltha.PmConfigs) error {
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 handler.updatePmConfig(ctx, pmConfigs)
}
- return nil
+ 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))
}
//Receive_packet_out sends packet out to the device