[VOL-4756] Cleanup TODO context
Change-Id: I21d5ec8cc015154bc893e54c652d31562d8da5d9
diff --git a/internal/pkg/application/application.go b/internal/pkg/application/application.go
index 9476e55..e7d2d04 100644
--- a/internal/pkg/application/application.go
+++ b/internal/pkg/application/application.go
@@ -84,7 +84,7 @@
var PacketHandlers map[string]CallBack
// CallBack : registered call back function for different protocol packets
-type CallBack func(device string, port string, pkt gopacket.Packet)
+type CallBack func(cntx context.Context, device string, port string, pkt gopacket.Packet)
const (
// ARP packet
@@ -338,7 +338,7 @@
}
// pushFlowsForUnis to send port-up-indication for uni ports.
-func (d *VoltDevice) pushFlowsForUnis() {
+func (d *VoltDevice) pushFlowsForUnis(cntx context.Context) {
logger.Info(ctx, "NNI Discovered, Sending Port UP Ind for UNIs")
d.Ports.Range(func(key, value interface{}) bool {
@@ -359,7 +359,7 @@
for _, vpv := range vnets.([]*VoltPortVnet) {
vpv.VpvLock.Lock()
- vpv.PortUpInd(d, port)
+ vpv.PortUpInd(cntx, d, port)
vpv.VpvLock.Unlock()
}
@@ -447,12 +447,12 @@
}
// RestoreNbDeviceFromDb restores the NB Device in case of VGC pod restart.
-func (va *VoltApplication) RestoreNbDeviceFromDb(deviceID string) *NbDevice {
+func (va *VoltApplication) RestoreNbDeviceFromDb(cntx context.Context, deviceID string) *NbDevice {
nbDevice := NewNbDevice()
nbDevice.SouthBoundID = deviceID
- nbPorts, _ := db.GetAllNbPorts(deviceID)
+ nbPorts, _ := db.GetAllNbPorts(cntx, deviceID)
for key, p := range nbPorts {
b, ok := p.Value.([]byte)
@@ -481,17 +481,17 @@
}
// WriteToDb writes nb device port config to kv store
-func (nbd *NbDevice) WriteToDb(portID uint32, ponPort *PonPortCfg) {
+func (nbd *NbDevice) WriteToDb(cntx context.Context, portID uint32, ponPort *PonPortCfg) {
b, err := json.Marshal(ponPort)
if err != nil {
logger.Errorw(ctx, "PonPortConfig-marshal-failed", log.Fields{"err": err})
return
}
- db.PutNbDevicePort(nbd.SouthBoundID, portID, string(b))
+ db.PutNbDevicePort(cntx, nbd.SouthBoundID, portID, string(b))
}
// AddPortToNbDevice Adds pon port to NB Device and DB
-func (nbd *NbDevice) AddPortToNbDevice(portID, allowedChannels uint32,
+func (nbd *NbDevice) AddPortToNbDevice(cntx context.Context, portID, allowedChannels uint32,
enableMulticastKPI bool, portAlarmProfileID string) *PonPortCfg {
ponPort := &PonPortCfg{
@@ -501,12 +501,12 @@
PortAlarmProfileID: portAlarmProfileID,
}
nbd.PonPorts.Store(portID, ponPort)
- nbd.WriteToDb(portID, ponPort)
+ nbd.WriteToDb(cntx, portID, ponPort)
return ponPort
}
// UpdatePortToNbDevice Adds pon port to NB Device and DB
-func (nbd *NbDevice) UpdatePortToNbDevice(portID, allowedChannels uint32, enableMulticastKPI bool, portAlarmProfileID string) *PonPortCfg {
+func (nbd *NbDevice) UpdatePortToNbDevice(cntx context.Context, portID, allowedChannels uint32, enableMulticastKPI bool, portAlarmProfileID string) *PonPortCfg {
p, exists := nbd.PonPorts.Load(portID)
if !exists {
@@ -521,17 +521,17 @@
}
nbd.PonPorts.Store(portID, port)
- nbd.WriteToDb(portID, port)
+ nbd.WriteToDb(cntx, portID, port)
return port
}
// DeletePortFromNbDevice Deletes pon port from NB Device and DB
-func (nbd *NbDevice) DeletePortFromNbDevice(portID uint32) {
+func (nbd *NbDevice) DeletePortFromNbDevice(cntx context.Context, portID uint32) {
if _, ok := nbd.PonPorts.Load(portID); ok {
nbd.PonPorts.Delete(portID)
}
- db.DelNbDevicePort(nbd.SouthBoundID, portID)
+ db.DelNbDevicePort(cntx, nbd.SouthBoundID, portID)
}
// GetApplication : Interface to access the singleton object
@@ -559,8 +559,8 @@
va.VnetsToDelete = make(map[string]bool)
va.ServicesToDelete = make(map[string]bool)
va.VoltPortVnetsToDelete = make(map[*VoltPortVnet]bool)
- go va.Start(TimerCfg{tick: 100 * time.Millisecond}, tickTimer)
- go va.Start(TimerCfg{tick: time.Duration(GroupExpiryTime) * time.Minute}, pendingPoolTimer)
+ go va.Start(context.Background(), TimerCfg{tick: 100 * time.Millisecond}, tickTimer)
+ go va.Start(context.Background(), TimerCfg{tick: time.Duration(GroupExpiryTime) * time.Minute}, pendingPoolTimer)
InitEventFuncMapper()
db = database.GetDatabase()
return &va
@@ -647,9 +647,9 @@
}
//RestoreUpgradeStatus - gets upgrade/migration status from DB and updates local flags
-func (va *VoltApplication) RestoreUpgradeStatus() {
+func (va *VoltApplication) RestoreUpgradeStatus(cntx context.Context) {
Migrate := new(DataMigration)
- if err := GetMigrationInfo(Migrate); err == nil {
+ if err := GetMigrationInfo(cntx, Migrate); err == nil {
if Migrate.Status == MigrationInProgress {
isUpgradeComplete = false
return
@@ -662,25 +662,25 @@
// ReadAllFromDb : If we are restarted, learn from the database the current execution
// stage
-func (va *VoltApplication) ReadAllFromDb() {
+func (va *VoltApplication) ReadAllFromDb(cntx context.Context) {
logger.Info(ctx, "Reading the meters from DB")
- va.RestoreMetersFromDb()
+ va.RestoreMetersFromDb(cntx)
logger.Info(ctx, "Reading the VNETs from DB")
- va.RestoreVnetsFromDb()
+ va.RestoreVnetsFromDb(cntx)
logger.Info(ctx, "Reading the VPVs from DB")
- va.RestoreVpvsFromDb()
+ va.RestoreVpvsFromDb(cntx)
logger.Info(ctx, "Reading the Services from DB")
- va.RestoreSvcsFromDb()
+ va.RestoreSvcsFromDb(cntx)
logger.Info(ctx, "Reading the MVLANs from DB")
- va.RestoreMvlansFromDb()
+ va.RestoreMvlansFromDb(cntx)
logger.Info(ctx, "Reading the IGMP profiles from DB")
- va.RestoreIGMPProfilesFromDb()
+ va.RestoreIGMPProfilesFromDb(cntx)
logger.Info(ctx, "Reading the Mcast configs from DB")
- va.RestoreMcastConfigsFromDb()
+ va.RestoreMcastConfigsFromDb(cntx)
logger.Info(ctx, "Reading the IGMP groups for DB")
- va.RestoreIgmpGroupsFromDb()
+ va.RestoreIgmpGroupsFromDb(cntx)
logger.Info(ctx, "Reading Upgrade status from DB")
- va.RestoreUpgradeStatus()
+ va.RestoreUpgradeStatus(cntx)
logger.Info(ctx, "Reconciled from DB")
}
@@ -723,7 +723,7 @@
// a single NNI port per OLT. This is true whether the network uses any
// protection mechanism (LAG, ERPS, etc.). The aggregate of the such protection
// is represented by a single NNI port
-func (va *VoltApplication) AddDevice(device string, slno, southBoundID string) {
+func (va *VoltApplication) AddDevice(cntx context.Context, device string, slno, southBoundID string) {
logger.Warnw(ctx, "Received Device Ind: Add", log.Fields{"Device": device, "SrNo": slno})
if _, ok := va.DevicesDisc.Load(device); ok {
logger.Warnw(ctx, "Device Exists", log.Fields{"Device": device})
@@ -742,7 +742,7 @@
nbDevice.(*NbDevice).PonPorts.Range(addPort)
} else {
// Check if NbPort exists in DB. VGC restart case.
- nbd := va.RestoreNbDeviceFromDb(southBoundID)
+ nbd := va.RestoreNbDeviceFromDb(cntx, southBoundID)
nbd.PonPorts.Range(addPort)
}
va.DevicesDisc.Store(device, d)
@@ -757,16 +757,16 @@
}
// DelDevice to delete a device.
-func (va *VoltApplication) DelDevice(device string) {
+func (va *VoltApplication) DelDevice(cntx context.Context, device string) {
logger.Warnw(ctx, "Received Device Ind: Delete", log.Fields{"Device": device})
if vdIntf, ok := va.DevicesDisc.Load(device); ok {
vd := vdIntf.(*VoltDevice)
va.DevicesDisc.Delete(device)
- _ = db.DelAllRoutesForDevice(device)
- va.HandleFlowClearFlag(device, vd.SerialNum, vd.SouthBoundID)
- _ = db.DelAllGroup(device)
- _ = db.DelAllMeter(device)
- _ = db.DelAllPorts(device)
+ _ = db.DelAllRoutesForDevice(cntx, device)
+ va.HandleFlowClearFlag(cntx, device, vd.SerialNum, vd.SouthBoundID)
+ _ = db.DelAllGroup(cntx, device)
+ _ = db.DelAllMeter(cntx, device)
+ _ = db.DelAllPorts(cntx, device)
logger.Debugw(ctx, "Device deleted", log.Fields{"Device": device})
} else {
logger.Warnw(ctx, "Device Doesn't Exist", log.Fields{"Device": device})
@@ -788,7 +788,7 @@
// PortAddInd : This is a PORT add indication coming from the VPAgent, which is essentially
// a request coming from VOLTHA. The device and identity of the port is provided
// in this request. Add them to the application for further use
-func (va *VoltApplication) PortAddInd(device string, id uint32, portName string) {
+func (va *VoltApplication) PortAddInd(cntx context.Context, device string, id uint32, portName string) {
logger.Infow(ctx, "Received Port Ind: Add", log.Fields{"Device": device, "Port": portName})
va.portLock.Lock()
if d := va.GetDevice(device); d != nil {
@@ -797,7 +797,7 @@
va.portLock.Unlock()
nni, _ := va.GetNniPort(device)
if nni == portName {
- d.pushFlowsForUnis()
+ d.pushFlowsForUnis(cntx)
}
} else {
va.portLock.Unlock()
@@ -807,13 +807,13 @@
// PortDelInd : Only the NNI ports are recorded in the device for now. When port delete
// arrives, only the NNI ports need adjustments.
-func (va *VoltApplication) PortDelInd(device string, port string) {
+func (va *VoltApplication) PortDelInd(cntx context.Context, device string, port string) {
logger.Infow(ctx, "Received Port Ind: Delete", log.Fields{"Device": device, "Port": port})
if d := va.GetDevice(device); d != nil {
p := d.GetPort(port)
if p != nil && p.State == PortStateUp {
logger.Infow(ctx, "Port state is UP. Trigerring Port Down Ind before deleting", log.Fields{"Port": p})
- va.PortDownInd(device, port)
+ va.PortDownInd(cntx, device, port)
}
va.portLock.Lock()
defer va.portLock.Unlock()
@@ -840,7 +840,7 @@
}
// AddNbPonPort Add pon port to nbDevice
-func (va *VoltApplication) AddNbPonPort(oltSbID string, portID, maxAllowedChannels uint32,
+func (va *VoltApplication) AddNbPonPort(cntx context.Context, oltSbID string, portID, maxAllowedChannels uint32,
enableMulticastKPI bool, portAlarmProfileID string) error {
var nbd *NbDevice
@@ -852,7 +852,7 @@
} else {
nbd = nbDevice.(*NbDevice)
}
- port := nbd.AddPortToNbDevice(portID, maxAllowedChannels, enableMulticastKPI, portAlarmProfileID)
+ port := nbd.AddPortToNbDevice(cntx, portID, maxAllowedChannels, enableMulticastKPI, portAlarmProfileID)
// Add this port to voltDevice
addPort := func(key, value interface{}) bool {
@@ -872,7 +872,7 @@
}
// UpdateNbPonPort update pon port to nbDevice
-func (va *VoltApplication) UpdateNbPonPort(oltSbID string, portID, maxAllowedChannels uint32, enableMulticastKPI bool, portAlarmProfileID string) error {
+func (va *VoltApplication) UpdateNbPonPort(cntx context.Context, oltSbID string, portID, maxAllowedChannels uint32, enableMulticastKPI bool, portAlarmProfileID string) error {
var nbd *NbDevice
nbDevice, ok := va.NbDevice.Load(oltSbID)
@@ -883,7 +883,7 @@
}
nbd = nbDevice.(*NbDevice)
- port := nbd.UpdatePortToNbDevice(portID, maxAllowedChannels, enableMulticastKPI, portAlarmProfileID)
+ port := nbd.UpdatePortToNbDevice(cntx, portID, maxAllowedChannels, enableMulticastKPI, portAlarmProfileID)
if port == nil {
return fmt.Errorf("Port-doesn't-exists-%v", portID)
}
@@ -913,10 +913,10 @@
}
// DeleteNbPonPort Delete pon port to nbDevice
-func (va *VoltApplication) DeleteNbPonPort(oltSbID string, portID uint32) error {
+func (va *VoltApplication) DeleteNbPonPort(cntx context.Context, oltSbID string, portID uint32) error {
nbDevice, ok := va.NbDevice.Load(oltSbID)
if ok {
- nbDevice.(*NbDevice).DeletePortFromNbDevice(portID)
+ nbDevice.(*NbDevice).DeletePortFromNbDevice(cntx, portID)
va.NbDevice.Store(oltSbID, nbDevice.(*NbDevice))
} else {
logger.Warnw(ctx, "Delete pon received for unknown device", log.Fields{"oltSbID": oltSbID})
@@ -954,19 +954,19 @@
}
// NniDownInd process for Nni down indication.
-func (va *VoltApplication) NniDownInd(deviceID string, devSrNo string) {
+func (va *VoltApplication) NniDownInd(cntx context.Context, deviceID string, devSrNo string) {
logger.Debugw(ctx, "NNI Down Ind", log.Fields{"device": devSrNo})
handleIgmpDsFlows := func(key interface{}, value interface{}) bool {
mvProfile := value.(*MvlanProfile)
- mvProfile.removeIgmpMcastFlows(devSrNo)
+ mvProfile.removeIgmpMcastFlows(cntx, devSrNo)
return true
}
va.MvlanProfilesByName.Range(handleIgmpDsFlows)
//Clear Static Group
- va.ReceiverDownInd(deviceID, StaticPort)
+ va.ReceiverDownInd(cntx, deviceID, StaticPort)
}
// DeviceUpInd changes device state to up.
@@ -990,7 +990,7 @@
}
// DeviceRebootInd process for handling flow clear flag for device reboot
-func (va *VoltApplication) DeviceRebootInd(device string, serialNum string, southBoundID string) {
+func (va *VoltApplication) DeviceRebootInd(cntx context.Context, device string, serialNum string, southBoundID string) {
logger.Warnw(ctx, "Received Device Ind: Reboot", log.Fields{"Device": device, "SerialNumber": serialNum})
if d := va.GetDevice(device); d != nil {
@@ -1000,12 +1000,12 @@
}
d.State = controller.DeviceStateREBOOTED
}
- va.HandleFlowClearFlag(device, serialNum, southBoundID)
+ va.HandleFlowClearFlag(cntx, device, serialNum, southBoundID)
}
// DeviceDisableInd handles device deactivation process
-func (va *VoltApplication) DeviceDisableInd(device string) {
+func (va *VoltApplication) DeviceDisableInd(cntx context.Context, device string) {
logger.Warnw(ctx, "Received Device Ind: Disable", log.Fields{"Device": device})
d := va.GetDevice(device)
@@ -1015,11 +1015,11 @@
}
d.State = controller.DeviceStateDISABLED
- va.HandleFlowClearFlag(device, d.SerialNum, d.SouthBoundID)
+ va.HandleFlowClearFlag(cntx, device, d.SerialNum, d.SouthBoundID)
}
// ProcessIgmpDSFlowForMvlan for processing Igmp DS flow for device
-func (va *VoltApplication) ProcessIgmpDSFlowForMvlan(d *VoltDevice, mvp *MvlanProfile, addFlow bool) {
+func (va *VoltApplication) ProcessIgmpDSFlowForMvlan(cntx context.Context, d *VoltDevice, mvp *MvlanProfile, addFlow bool) {
logger.Debugw(ctx, "Process IGMP DS Flows for MVlan", log.Fields{"device": d.Name, "Mvlan": mvp.Mvlan, "addFlow": addFlow})
portState := false
@@ -1030,20 +1030,20 @@
if addFlow {
if portState {
- mvp.pushIgmpMcastFlows(d.SerialNum)
+ mvp.pushIgmpMcastFlows(cntx, d.SerialNum)
}
} else {
- mvp.removeIgmpMcastFlows(d.SerialNum)
+ mvp.removeIgmpMcastFlows(cntx, d.SerialNum)
}
}
// ProcessIgmpDSFlowForDevice for processing Igmp DS flow for device
-func (va *VoltApplication) ProcessIgmpDSFlowForDevice(d *VoltDevice, addFlow bool) {
+func (va *VoltApplication) ProcessIgmpDSFlowForDevice(cntx context.Context, d *VoltDevice, addFlow bool) {
logger.Debugw(ctx, "Process IGMP DS Flows for device", log.Fields{"device": d.Name, "addFlow": addFlow})
handleIgmpDsFlows := func(key interface{}, value interface{}) bool {
mvProfile := value.(*MvlanProfile)
- va.ProcessIgmpDSFlowForMvlan(d, mvProfile, addFlow)
+ va.ProcessIgmpDSFlowForMvlan(cntx, d, mvProfile, addFlow)
return true
}
va.MvlanProfilesByName.Range(handleIgmpDsFlows)
@@ -1166,13 +1166,13 @@
// device - Device Obj
// vnet - vnet profile name
// enabled - vlan enabled/disabled - based on the status, the flow shall be added/removed
-func (va *VoltApplication) ProcessDevFlowForDevice(device *VoltDevice, vnet *VoltVnet, enabled bool) {
+func (va *VoltApplication) ProcessDevFlowForDevice(cntx context.Context, device *VoltDevice, vnet *VoltVnet, enabled bool) {
_, applied := device.ConfiguredVlanForDeviceFlows.Get(VnetKey(vnet.SVlan, vnet.CVlan, 0))
if enabled {
- va.PushDevFlowForVlan(vnet)
+ va.PushDevFlowForVlan(cntx, vnet)
} else if !enabled && applied {
//va.DeleteDevFlowForVlan(vnet)
- va.DeleteDevFlowForVlanFromDevice(vnet, device.SerialNum)
+ va.DeleteDevFlowForVlanFromDevice(cntx, vnet, device.SerialNum)
}
}
@@ -1212,7 +1212,7 @@
// Port UP indication is passed to all services associated with the port
// so that the services can configure flows applicable when the port goes
// up from down state
-func (va *VoltApplication) PortUpInd(device string, port string) {
+func (va *VoltApplication) PortUpInd(cntx context.Context, device string, port string) {
d := va.GetDevice(device)
if d == nil {
@@ -1264,9 +1264,9 @@
for _, vpv := range vpvs.([]*VoltPortVnet) {
vpv.VpvLock.Lock()
logger.Warnw(ctx, "Removing existing VPVs/Services flows for for Subscriber: UNI Detected on wrong PON", log.Fields{"Port": vpv.Port, "Vnet": vpv.VnetName})
- vpv.PortDownInd(device, port)
+ vpv.PortDownInd(cntx, device, port)
if vpv.IgmpEnabled {
- va.ReceiverDownInd(device, port)
+ va.ReceiverDownInd(cntx, device, port)
}
vpv.VpvLock.Unlock()
}
@@ -1288,7 +1288,7 @@
// part of service delete (during the lock wait duration)
// In that case, the services associated wil be zero
if vpv.servicesCount.Load() != 0 {
- vpv.PortUpInd(d, port)
+ vpv.PortUpInd(cntx, d, port)
}
vpv.VpvLock.Unlock()
}
@@ -1342,7 +1342,7 @@
// PortDownInd : Port down indication is passed on to the services so that the services
// can make changes at this transition.
-func (va *VoltApplication) PortDownInd(device string, port string) {
+func (va *VoltApplication) PortDownInd(cntx context.Context, device string, port string) {
logger.Infow(ctx, "Received SouthBound Port Ind: DOWN", log.Fields{"Device": device, "Port": port})
d := va.GetDevice(device)
@@ -1371,9 +1371,9 @@
if p.Type == VoltPortTypeNni {
logger.Warnw(ctx, "Received NNI Port Ind: DOWN", log.Fields{"Device": device, "Port": port})
- va.DeleteDevFlowForDevice(d)
- va.NniDownInd(device, d.SerialNum)
- va.RemovePendingGroups(device, true)
+ va.DeleteDevFlowForDevice(cntx, d)
+ va.NniDownInd(cntx, device, d.SerialNum)
+ va.RemovePendingGroups(cntx, device, true)
}
vpvs, ok := va.VnetsByPort.Load(port)
if !ok || nil == vpvs || len(vpvs.([]*VoltPortVnet)) == 0 {
@@ -1390,9 +1390,9 @@
*/
for _, vpv := range vpvs.([]*VoltPortVnet) {
vpv.VpvLock.Lock()
- vpv.PortDownInd(device, port)
+ vpv.PortDownInd(cntx, device, port)
if vpv.IgmpEnabled {
- va.ReceiverDownInd(device, port)
+ va.ReceiverDownInd(cntx, device, port)
}
vpv.VpvLock.Unlock()
}
@@ -1406,7 +1406,7 @@
// packet is decoded and the right processor is called. Currently, we
// plan to support only DHCP and IGMP. In future, we can add more
// capabilities as needed
-func (va *VoltApplication) PacketInInd(device string, port string, pkt []byte) {
+func (va *VoltApplication) PacketInInd(cntx context.Context, device string, port string, pkt []byte) {
// Decode the incoming packet
packetSide := US
if strings.Contains(port, NNI) {
@@ -1440,7 +1440,7 @@
arpl := gopkt.Layer(layers.LayerTypeARP)
if arpl != nil {
if callBack, ok := PacketHandlers[ARP]; ok {
- callBack(device, port, gopkt)
+ callBack(cntx, device, port, gopkt)
} else {
logger.Debugw(ctx, "ARP handler is not registered, dropping the packet", log.Fields{"Pkt": hex.EncodeToString(gopkt.Data())})
}
@@ -1455,7 +1455,7 @@
dhcpl := gopkt.Layer(layers.LayerTypeDHCPv4)
if dhcpl != nil {
if callBack, ok := PacketHandlers[DHCPv4]; ok {
- callBack(device, port, gopkt)
+ callBack(cntx, device, port, gopkt)
} else {
logger.Debugw(ctx, "DHCPv4 handler is not registered, dropping the packet", log.Fields{"Pkt": hex.EncodeToString(gopkt.Data())})
}
@@ -1463,7 +1463,7 @@
} else if ip.Protocol == layers.IPProtocolIGMP {
logger.Debugw(ctx, "Received Southbound IGMP packet in", log.Fields{"StreamSide": packetSide})
if callBack, ok := PacketHandlers[IGMP]; ok {
- callBack(device, port, gopkt)
+ callBack(cntx, device, port, gopkt)
} else {
logger.Debugw(ctx, "IGMP handler is not registered, dropping the packet", log.Fields{"Pkt": hex.EncodeToString(gopkt.Data())})
}
@@ -1478,7 +1478,7 @@
dhcpl := gopkt.Layer(layers.LayerTypeDHCPv6)
if dhcpl != nil {
if callBack, ok := PacketHandlers[DHCPv6]; ok {
- callBack(device, port, gopkt)
+ callBack(cntx, device, port, gopkt)
} else {
logger.Debugw(ctx, "DHCPv6 handler is not registered, dropping the packet", log.Fields{"Pkt": hex.EncodeToString(gopkt.Data())})
}
@@ -1491,7 +1491,7 @@
if pppoel != nil {
logger.Debugw(ctx, "Received Southbound PPPoE packet in", log.Fields{"StreamSide": packetSide})
if callBack, ok := PacketHandlers[PPPOE]; ok {
- callBack(device, port, gopkt)
+ callBack(cntx, device, port, gopkt)
} else {
logger.Debugw(ctx, "PPPoE handler is not registered, dropping the packet", log.Fields{"Pkt": hex.EncodeToString(gopkt.Data())})
}
@@ -1528,7 +1528,7 @@
}
// HandleFlowClearFlag to handle flow clear flag during reboot
-func (va *VoltApplication) HandleFlowClearFlag(deviceID string, serialNum, southBoundID string) {
+func (va *VoltApplication) HandleFlowClearFlag(cntx context.Context, deviceID string, serialNum, southBoundID string) {
logger.Warnw(ctx, "Clear All flags for Device", log.Fields{"Device": deviceID, "SerialNum": serialNum, "SBID": southBoundID})
dev, ok := va.DevicesDisc.Load(deviceID)
if ok && dev != nil {
@@ -1557,13 +1557,13 @@
logger.Infow(ctx, "Clear Flags for vpv",
log.Fields{"device": vpv.Device, "port": vpv.Port,
"svlan": vpv.SVlan, "cvlan": vpv.CVlan, "univlan": vpv.UniVlan})
- vpv.ClearAllServiceFlags()
- vpv.ClearAllVpvFlags()
+ vpv.ClearAllServiceFlags(cntx)
+ vpv.ClearAllVpvFlags(cntx)
if vpv.IgmpEnabled {
- va.ReceiverDownInd(vpv.Device, vpv.Port)
+ va.ReceiverDownInd(cntx, vpv.Device, vpv.Port)
//Also clear service igmp stats
- vpv.ClearServiceCounters()
+ vpv.ClearServiceCounters(cntx)
}
}
}
@@ -1572,12 +1572,12 @@
va.VnetsByPort.Range(getVpvs)
//Clear Static Group
- va.ReceiverDownInd(deviceID, StaticPort)
+ va.ReceiverDownInd(cntx, deviceID, StaticPort)
logger.Warnw(ctx, "All flags cleared for device", log.Fields{"Device": deviceID})
//Reset pending group pool
- va.RemovePendingGroups(deviceID, true)
+ va.RemovePendingGroups(cntx, deviceID, true)
//Process all Migrate Service Request - force udpate all profiles since resources are already cleaned up
if dev != nil {
@@ -1585,7 +1585,7 @@
msrList := value.(*util.ConcurrentMap)
forceUpdateServices := func(key, value interface{}) bool {
msr := value.(*MigrateServicesRequest)
- forceUpdateAllServices(msr)
+ forceUpdateAllServices(cntx, msr)
return true
}
msrList.Range(forceUpdateServices)
@@ -1593,7 +1593,7 @@
}
dev.(*VoltDevice).MigratingServices.Range(triggerForceUpdate)
} else {
- va.FetchAndProcessAllMigrateServicesReq(deviceID, forceUpdateAllServices)
+ va.FetchAndProcessAllMigrateServicesReq(cntx, deviceID, forceUpdateAllServices)
}
}
@@ -1604,7 +1604,7 @@
}
//ProcessFlowModResultIndication - Processes Flow mod operation indications from controller
-func (va *VoltApplication) ProcessFlowModResultIndication(flowStatus intf.FlowStatus) {
+func (va *VoltApplication) ProcessFlowModResultIndication(cntx context.Context, flowStatus intf.FlowStatus) {
d := va.GetDevice(flowStatus.Device)
if d == nil {
@@ -1612,7 +1612,7 @@
return
}
- cookieExists := ExecuteFlowEvent(d, flowStatus.Cookie, flowStatus)
+ cookieExists := ExecuteFlowEvent(cntx, d, flowStatus.Cookie, flowStatus)
if flowStatus.Flow != nil {
flowAdd := (flowStatus.FlowModType == of.CommandAdd)
@@ -1670,12 +1670,12 @@
}
//UpdateMvlanProfilesForDevice to update mvlan profile for device
-func (va *VoltApplication) UpdateMvlanProfilesForDevice(device string) {
+func (va *VoltApplication) UpdateMvlanProfilesForDevice(cntx context.Context, device string) {
checkAndAddMvlanUpdateTask := func(key, value interface{}) bool {
mvp := value.(*MvlanProfile)
if mvp.IsUpdateInProgressForDevice(device) {
- mvp.UpdateProfile(device)
+ mvp.UpdateProfile(cntx, device)
}
return true
}
@@ -1867,26 +1867,26 @@
}
//RemoveGroupsFromPendingPool - removes the group from global pending group pool
-func (va *VoltApplication) RemoveGroupsFromPendingPool(device string, mvlan of.VlanType) {
+func (va *VoltApplication) RemoveGroupsFromPendingPool(cntx context.Context, device string, mvlan of.VlanType) {
GetApplication().PendingPoolLock.Lock()
defer GetApplication().PendingPoolLock.Unlock()
logger.Infow(ctx, "Removing IgmpGroups from Global Pending Pool for given Deivce & Mvlan", log.Fields{"Device": device, "Mvlan": mvlan.String()})
key := getPendingPoolKey(mvlan, device)
- va.RemoveGroupListFromPendingPool(key)
+ va.RemoveGroupListFromPendingPool(cntx, key)
}
//RemoveGroupListFromPendingPool - removes the groups for provided key
// 1. Deletes the group from device
// 2. Delete the IgmpGroup obj and release the group ID to pool
// Note: Make sure to obtain PendingPoolLock lock before calling this func
-func (va *VoltApplication) RemoveGroupListFromPendingPool(key string) {
+func (va *VoltApplication) RemoveGroupListFromPendingPool(cntx context.Context, key string) {
if grpMap, ok := va.IgmpPendingPool[key]; ok {
delete(va.IgmpPendingPool, key)
for ig := range grpMap {
for device := range ig.Devices {
- ig.DeleteIgmpGroupDevice(device)
+ ig.DeleteIgmpGroupDevice(cntx, device)
}
}
}
@@ -1942,7 +1942,7 @@
// reference - mvlan/device ID
// isRefDevice - true - Device as reference
// false - Mvlan as reference
-func (va *VoltApplication) RemovePendingGroups(reference string, isRefDevice bool) {
+func (va *VoltApplication) RemovePendingGroups(cntx context.Context, reference string, isRefDevice bool) {
va.PendingPoolLock.Lock()
defer va.PendingPoolLock.Unlock()
@@ -1960,7 +1960,7 @@
for key := range va.IgmpPendingPool {
keyParams := strings.Split(key, "_")
if keyParams[paramPosition] == reference {
- va.RemoveGroupListFromPendingPool(key)
+ va.RemoveGroupListFromPendingPool(cntx, key)
}
}
}
@@ -1969,26 +1969,26 @@
return mvlan.String() + "_" + device
}
-func (va *VoltApplication) removeExpiredGroups() {
+func (va *VoltApplication) removeExpiredGroups(cntx context.Context) {
logger.Debug(ctx, "Check for expired Igmp Groups")
removeExpiredGroups := func(key interface{}, value interface{}) bool {
ig := value.(*IgmpGroup)
- ig.removeExpiredGroupFromDevice()
+ ig.removeExpiredGroupFromDevice(cntx)
return true
}
va.IgmpGroups.Range(removeExpiredGroups)
}
//TriggerPendingProfileDeleteReq - trigger pending profile delete request
-func (va *VoltApplication) TriggerPendingProfileDeleteReq(device string) {
- va.TriggerPendingServiceDeleteReq(device)
- va.TriggerPendingVpvDeleteReq(device)
- va.TriggerPendingVnetDeleteReq(device)
+func (va *VoltApplication) TriggerPendingProfileDeleteReq(cntx context.Context, device string) {
+ va.TriggerPendingServiceDeleteReq(cntx, device)
+ va.TriggerPendingVpvDeleteReq(cntx, device)
+ va.TriggerPendingVnetDeleteReq(cntx, device)
logger.Warnw(ctx, "All Pending Profile Delete triggered for device", log.Fields{"Device": device})
}
//TriggerPendingServiceDeleteReq - trigger pending service delete request
-func (va *VoltApplication) TriggerPendingServiceDeleteReq(device string) {
+func (va *VoltApplication) TriggerPendingServiceDeleteReq(cntx context.Context, device string) {
logger.Warnw(ctx, "Pending Services to be deleted", log.Fields{"Count": len(va.ServicesToDelete)})
for serviceName := range va.ServicesToDelete {
@@ -1996,9 +1996,9 @@
if vs := va.GetService(serviceName); vs != nil {
if vs.Device == device {
logger.Warnw(ctx, "Triggering Pending Service delete", log.Fields{"Service": vs.Name})
- vs.DelHsiaFlows()
+ vs.DelHsiaFlows(cntx)
if vs.ForceDelete {
- vs.DelFromDb()
+ vs.DelFromDb(cntx)
/*
portState := msgbus.PortDown
if d, err := va.GetDeviceFromPort(vs.Port); d != nil {
@@ -2027,19 +2027,19 @@
}
//TriggerPendingVpvDeleteReq - trigger pending VPV delete request
-func (va *VoltApplication) TriggerPendingVpvDeleteReq(device string) {
+func (va *VoltApplication) TriggerPendingVpvDeleteReq(cntx context.Context, device string) {
logger.Warnw(ctx, "Pending VPVs to be deleted", log.Fields{"Count": len(va.VoltPortVnetsToDelete)})
for vpv := range va.VoltPortVnetsToDelete {
if vpv.Device == device {
logger.Warnw(ctx, "Triggering Pending VPv flow delete", log.Fields{"Port": vpv.Port, "Device": vpv.Device, "Vnet": vpv.VnetName})
- va.DelVnetFromPort(vpv.Port, vpv)
+ va.DelVnetFromPort(cntx, vpv.Port, vpv)
}
}
}
//TriggerPendingVnetDeleteReq - trigger pending vnet delete request
-func (va *VoltApplication) TriggerPendingVnetDeleteReq(device string) {
+func (va *VoltApplication) TriggerPendingVnetDeleteReq(cntx context.Context, device string) {
logger.Warnw(ctx, "Pending Vnets to be deleted", log.Fields{"Count": len(va.VnetsToDelete)})
for vnetName := range va.VnetsToDelete {
@@ -2047,7 +2047,7 @@
vnet := vnetIntf.(*VoltVnet)
logger.Warnw(ctx, "Triggering Pending Vnet flows delete", log.Fields{"Vnet": vnet.Name})
if d := va.GetDeviceBySerialNo(vnet.PendingDeviceToDelete); d != nil && d.SerialNum == vnet.PendingDeviceToDelete {
- va.DeleteDevFlowForVlanFromDevice(vnet, vnet.PendingDeviceToDelete)
+ va.DeleteDevFlowForVlanFromDevice(cntx, vnet, vnet.PendingDeviceToDelete)
va.deleteVnetConfig(vnet)
} else {
logger.Warnw(ctx, "Vnet Delete Failed : Device Not Found", log.Fields{"Vnet": vnet.Name, "Device": vnet.PendingDeviceToDelete})
diff --git a/internal/pkg/application/dhcprelay.go b/internal/pkg/application/dhcprelay.go
index c57771c..b779f1f 100644
--- a/internal/pkg/application/dhcprelay.go
+++ b/internal/pkg/application/dhcprelay.go
@@ -17,6 +17,7 @@
import (
"encoding/hex"
+ "context"
"errors"
"net"
"sync"
@@ -93,9 +94,9 @@
GetDhcpv6State() Dhcpv6RelayState
SetDhcpState(DhcpRelayState)
SetDhcpv6State(Dhcpv6RelayState)
- SetMacAddr(net.HardwareAddr)
- DhcpResultInd(*layers.DHCPv4)
- Dhcpv6ResultInd(ipv6Addr net.IP, leaseTime uint32)
+ SetMacAddr(context.Context, net.HardwareAddr)
+ DhcpResultInd(context.Context, *layers.DHCPv4)
+ Dhcpv6ResultInd(cntx context.Context, ipv6Addr net.IP, leaseTime uint32)
}
// DhcpRelayVnet : The DHCP relay sessions are stored in a map to be retrieved from when
@@ -547,7 +548,7 @@
// session is derived from the list of DHCP sessions stored in the
// common map. The key for retrieval includes the VLAN tags in the
// the packet and the MAC address of the client.
-func (va *VoltApplication) ProcessDsDhcpv4Packet(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessDsDhcpv4Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// Retrieve the layers to build the outgoing packet. It is not
// possible to add/remove layers to the existing packet and thus
@@ -600,9 +601,9 @@
// flow installation request, VGC to update US HSIA flow with leanrt MAC.
// separate go rotuine is spawned to avoid drop of ACK packet
// as HSIA flows will be deleted if new MAC is learnt.
- go vpv.SetMacAddr(dhcp4.ClientHWAddr)
+ go vpv.SetMacAddr(cntx, dhcp4.ClientHWAddr)
}
- vpv.DhcpResultInd(dhcp4)
+ vpv.DhcpResultInd(cntx, dhcp4)
}
raiseDHCPv4Indication(msgType, vpv, dhcp4.ClientHWAddr, ipAddr, dsPbit, device, leaseTime)
@@ -761,7 +762,7 @@
// ProcessUsDhcpv4Packet : The US DHCPv4 packet is identified the DHCP OP in the packet. A request is considered upstream
// and the service associated with the packet is located by the port and VLANs in the packet
-func (va *VoltApplication) ProcessUsDhcpv4Packet(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessUsDhcpv4Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// We received the packet on an access port and the service for the packet can be
// gotten from the port and the packet
vpv, svc := va.GetVnetFromPkt(device, port, pkt)
@@ -819,7 +820,7 @@
if NonZeroMacAddress(vpv.MacAddr) && vpv.MacLearning == Learn {
// update learnt mac for debug purpose
vpv.LearntMacAddr = dhcp4.ClientHWAddr
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
logger.Warnw(ctx, "Dropping the packet Mac relearn is disabled",
log.Fields{"vpv.MacAddr": vpv.MacAddr, "LearntMac": dhcp4.ClientHWAddr})
return
@@ -923,13 +924,13 @@
}
// ProcessUDP4Packet : CallBack function registered with application to handle DHCP packetIn
-func ProcessUDP4Packet(device string, port string, pkt gopacket.Packet) {
- GetApplication().ProcessUDP4Packet(device, port, pkt)
+func ProcessUDP4Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) {
+ GetApplication().ProcessUDP4Packet(cntx, device, port, pkt)
}
// ProcessUDP4Packet : The packet is a UDP packet and currently only DHCP relay application is supported
// We determine the packet direction and process it based on the direction
-func (va *VoltApplication) ProcessUDP4Packet(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessUDP4Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// Currently DHCP is the only application supported by the application
// We check for DHCP before proceeding futher. In future, this could be
// based on registration and the callbacks
@@ -943,17 +944,17 @@
// This is treated as an upstream packet in the VOLT application
// as VOLT serves access subscribers who use DHCP to acquire IP
// address and these packets go upstream to the network
- va.ProcessUsDhcpv4Packet(device, port, pkt)
+ va.ProcessUsDhcpv4Packet(cntx, device, port, pkt)
} else {
// This is a downstream packet
- va.ProcessDsDhcpv4Packet(device, port, pkt)
+ va.ProcessDsDhcpv4Packet(cntx, device, port, pkt)
}
}
// ProcessUDP6Packet : CallBack function registered with application to handle DHCPv6 packetIn
-func ProcessUDP6Packet(device string, port string, pkt gopacket.Packet) {
- GetApplication().ProcessUDP6Packet(device, port, pkt)
+func ProcessUDP6Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) {
+ GetApplication().ProcessUDP6Packet(cntx, device, port, pkt)
}
// ProcessUDP6Packet : As a LDRA node, we expect to see only RelayReply from the DHCP server and we always
@@ -962,7 +963,7 @@
// we should also see Renew. However, we should always pack the US message by adding
// additional option that identifies to the server that the DHCP packet is forwarded
// by an LDRA node.
-func (va *VoltApplication) ProcessUDP6Packet(device string, port string, pkt gopacket.Packet) []byte {
+func (va *VoltApplication) ProcessUDP6Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) []byte {
dhcpl := pkt.Layer(layers.LayerTypeDHCPv6)
if dhcpl == nil {
return nil
@@ -973,14 +974,14 @@
case layers.DHCPv6MsgTypeSolicit, layers.DHCPv6MsgTypeRequest, layers.DHCPv6MsgTypeRenew,
layers.DHCPv6MsgTypeRelease, layers.DHCPv6MsgTypeRebind, layers.DHCPv6MsgTypeInformationRequest,
layers.DHCPv6MsgTypeDecline:
- va.ProcessUsDhcpv6Packet(device, port, pkt)
+ va.ProcessUsDhcpv6Packet(cntx, device, port, pkt)
case layers.DHCPv6MsgTypeAdvertise, layers.DHCPv6MsgTypeConfirm, layers.DHCPv6MsgTypeReconfigure:
logger.Warnw(ctx, "SouthBound DHCPv6 DS Messages Expected For a Relay Agent", log.Fields{"Type": dhcpv6.MsgType})
case layers.DHCPv6MsgTypeRelayForward:
logger.Warn(ctx, "As the first DHCPv6 Relay Agent, Unexpected Relay Forward")
case layers.DHCPv6MsgTypeRelayReply:
// We received a response from the server
- va.ProcessDsDhcpv6Packet(device, port, pkt)
+ va.ProcessDsDhcpv6Packet(cntx, device, port, pkt)
}
return nil
}
@@ -1017,7 +1018,7 @@
}
// ProcessUsDhcpv6Packet to rpocess upstream DHCPv6 packet
-func (va *VoltApplication) ProcessUsDhcpv6Packet(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessUsDhcpv6Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// We received the packet on an access port and the service for the packet can be
// gotten from the port and the packet
logger.Infow(ctx, "Processing Southbound US DHCPv6 packet", log.Fields{"Port": port})
@@ -1092,7 +1093,7 @@
if NonZeroMacAddress(vpv.MacAddr) && vpv.MacLearning == Learn {
// update learnt mac for debug purpose
vpv.LearntMacAddr = sourceMac
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
logger.Warnw(ctx, "Dropping the packet Mac relearn is disabled",
log.Fields{"vpv.MacAddr": vpv.MacAddr, "LearntMac": sourceMac})
return
@@ -1191,7 +1192,7 @@
}
// ProcessDsDhcpv6Packet to process downstream dhcpv6 packet
-func (va *VoltApplication) ProcessDsDhcpv6Packet(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessDsDhcpv6Packet(cntx context.Context, device string, port string, pkt gopacket.Packet) {
logger.Infow(ctx, "Processing Southbound DS DHCPv6 packet", log.Fields{"Port": port})
logger.Debugw(ctx, "Packet IN", log.Fields{"Pkt": hex.EncodeToString(pkt.Data())})
@@ -1249,9 +1250,9 @@
// separate go rotuine is spawned to avoid drop of ACK packet
// as HSIA flows will be deleted if new MAC is learnt.
if len(vpvList) == 1 {
- go vpv.SetMacAddr(clientMac)
+ go vpv.SetMacAddr(cntx, clientMac)
}
- vpv.Dhcpv6ResultInd(ipv6Addr, leaseTime)
+ vpv.Dhcpv6ResultInd(cntx, ipv6Addr, leaseTime)
}
raiseDHCPv6Indication(dhcp6.MsgType, vpv, clientMac, ipv6Addr, dsPbit, device, leaseTime)
}
diff --git a/internal/pkg/application/flowevent.go b/internal/pkg/application/flowevent.go
index f6f3584..fe37b74 100644
--- a/internal/pkg/application/flowevent.go
+++ b/internal/pkg/application/flowevent.go
@@ -16,6 +16,8 @@
package application
import (
+ "context"
+
infraerrorcode "voltha-go-controller/internal/pkg/errorcodes/service"
"voltha-go-controller/internal/pkg/intf"
@@ -29,7 +31,7 @@
type FlowEventType string
//FlowEventHandler - Func prototype for flow event handling funcs
-type FlowEventHandler func(*FlowEvent, intf.FlowStatus)
+type FlowEventHandler func(context.Context, *FlowEvent, intf.FlowStatus)
var eventMapper map[FlowEventType]FlowEventHandler
@@ -74,7 +76,7 @@
}
//ExecuteFlowEvent - Process flow based event triggers
-func ExecuteFlowEvent(vd *VoltDevice, cookie string, flowStatus intf.FlowStatus) bool {
+func ExecuteFlowEvent(cntx context.Context, vd *VoltDevice, cookie string, flowStatus intf.FlowStatus) bool {
var event interface{}
flowEventMap, err := vd.GetFlowEventRegister(flowStatus.FlowModType)
@@ -92,12 +94,12 @@
flowEventMap.Remove(cookie)
flowEventMap.MapLock.Unlock()
flowEvent := event.(*FlowEvent)
- eventMapper[flowEvent.eType](flowEvent, flowStatus)
+ eventMapper[flowEvent.eType](cntx, flowEvent, flowStatus)
return true
}
//ProcessUsIgmpFlowAddEvent - Process Us Igmp Flow event trigger
-func ProcessUsIgmpFlowAddEvent(event *FlowEvent, flowStatus intf.FlowStatus) {
+func ProcessUsIgmpFlowAddEvent(cntx context.Context, event *FlowEvent, flowStatus intf.FlowStatus) {
logger.Infow(ctx, "Processing Post Flow Add Event for US Igmp", log.Fields{"Cookie": event.cookie, "event": event})
vpv := event.eventData.(*VoltPortVnet)
@@ -109,19 +111,19 @@
}
//ProcessServiceFlowAddEvent - Process Service Flow event trigger
-func ProcessServiceFlowAddEvent(event *FlowEvent, flowStatus intf.FlowStatus) {
+func ProcessServiceFlowAddEvent(cntx context.Context, event *FlowEvent, flowStatus intf.FlowStatus) {
logger.Infow(ctx, "Processing Post Flow Add Event for Service", log.Fields{"Cookie": event.cookie, "event": event})
vs := event.eventData.(*VoltService)
if isFlowStatusSuccess(flowStatus.Status, true) {
- vs.FlowInstallSuccess(event.cookie, flowStatus.AdditionalData)
+ vs.FlowInstallSuccess(cntx, event.cookie, flowStatus.AdditionalData)
} else {
vs.FlowInstallFailure(event.cookie, flowStatus.Status, flowStatus.Reason)
}
}
//ProcessControlFlowAddEvent - Process Control Flow event trigger
-func ProcessControlFlowAddEvent(event *FlowEvent, flowStatus intf.FlowStatus) {
+func ProcessControlFlowAddEvent(cntx context.Context, event *FlowEvent, flowStatus intf.FlowStatus) {
logger.Infow(ctx, "Processing Post Flow Add Event for VPV", log.Fields{"Cookie": event.cookie, "event": event})
vpv := event.eventData.(*VoltPortVnet)
@@ -131,50 +133,50 @@
}
//ProcessServiceFlowDelEvent - Process Service Flow event trigger
-func ProcessServiceFlowDelEvent(event *FlowEvent, flowStatus intf.FlowStatus) {
+func ProcessServiceFlowDelEvent(cntx context.Context, event *FlowEvent, flowStatus intf.FlowStatus) {
logger.Infow(ctx, "Processing Post Flow Remove Event for Service", log.Fields{"Cookie": event.cookie, "event": event})
vs := event.eventData.(*VoltService)
if isFlowStatusSuccess(flowStatus.Status, false) {
- vs.FlowRemoveSuccess(event.cookie)
+ vs.FlowRemoveSuccess(cntx, event.cookie)
} else {
- vs.FlowRemoveFailure(event.cookie, flowStatus.Status, flowStatus.Reason)
+ vs.FlowRemoveFailure(cntx, event.cookie, flowStatus.Status, flowStatus.Reason)
}
}
//ProcessControlFlowDelEvent - Process Control Flow event trigger
-func ProcessControlFlowDelEvent(event *FlowEvent, flowStatus intf.FlowStatus) {
+func ProcessControlFlowDelEvent(cntx context.Context, event *FlowEvent, flowStatus intf.FlowStatus) {
logger.Infow(ctx, "Processing Post Flow Remove Event for VPV", log.Fields{"Cookie": event.cookie, "event": event})
vpv := event.eventData.(*VoltPortVnet)
if isFlowStatusSuccess(flowStatus.Status, false) {
- vpv.FlowRemoveSuccess(event.cookie, event.device)
+ vpv.FlowRemoveSuccess(cntx, event.cookie, event.device)
} else {
- vpv.FlowRemoveFailure(event.cookie, event.device, flowStatus.Status, flowStatus.Reason)
+ vpv.FlowRemoveFailure(cntx, event.cookie, event.device, flowStatus.Status, flowStatus.Reason)
}
}
//ProcessMcastFlowDelEvent - Process Control Flow event trigger
-func ProcessMcastFlowDelEvent(event *FlowEvent, flowStatus intf.FlowStatus) {
+func ProcessMcastFlowDelEvent(cntx context.Context, event *FlowEvent, flowStatus intf.FlowStatus) {
logger.Infow(ctx, "Processing Post Flow Remove Event for Mcast/Igmp", log.Fields{"Cookie": event.cookie, "event": event})
mvp := event.eventData.(*MvlanProfile)
if isFlowStatusSuccess(flowStatus.Status, false) {
- mvp.FlowRemoveSuccess(event.cookie, event.device)
+ mvp.FlowRemoveSuccess(cntx, event.cookie, event.device)
} else {
mvp.FlowRemoveFailure(event.cookie, event.device, flowStatus.Status, flowStatus.Reason)
}
}
//ProcessDeviceFlowDelEvent - Process Control Flow event trigger
-func ProcessDeviceFlowDelEvent(event *FlowEvent, flowStatus intf.FlowStatus) {
+func ProcessDeviceFlowDelEvent(cntx context.Context, event *FlowEvent, flowStatus intf.FlowStatus) {
logger.Infow(ctx, "Processing Post Flow Remove Event for VNET", log.Fields{"Cookie": event.cookie, "event": event})
vnet := event.eventData.(*VoltVnet)
if isFlowStatusSuccess(flowStatus.Status, false) {
- vnet.FlowRemoveSuccess(event.cookie, event.device)
+ vnet.FlowRemoveSuccess(cntx, event.cookie, event.device)
} else {
- vnet.FlowRemoveFailure(event.cookie, event.device, flowStatus.Status, flowStatus.Reason)
+ vnet.FlowRemoveFailure(cntx, event.cookie, event.device, flowStatus.Status, flowStatus.Reason)
}
}
diff --git a/internal/pkg/application/igmp.go b/internal/pkg/application/igmp.go
index 32eed0d..2e49a41 100644
--- a/internal/pkg/application/igmp.go
+++ b/internal/pkg/application/igmp.go
@@ -16,6 +16,7 @@
package application
import (
+ "context"
"encoding/json"
"errors"
"net"
@@ -106,7 +107,7 @@
}
// ProcessIgmpPacket : CallBack function registered with application to handle IGMP packetIn
-func ProcessIgmpPacket(device string, port string, pkt gopacket.Packet) {
+func ProcessIgmpPacket(cntx context.Context, device string, port string, pkt gopacket.Packet) {
GetApplication().IgmpPacketInd(device, port, pkt)
}
@@ -420,7 +421,7 @@
}
//AddToPendingPool - adds Igmp Device obj to pending pool
-func AddToPendingPool(device string, groupKey string) bool {
+func AddToPendingPool(cntx context.Context, device string, groupKey string) bool {
logger.Infow(ctx, "Add Device to IgmpGroup Pending Pool", log.Fields{"Device": device, "GroupKey": groupKey})
if grp, ok := GetApplication().IgmpGroups.Load(groupKey); ok {
@@ -429,7 +430,7 @@
logger.Infow(ctx, "Adding Device to IgmpGroup Pending Pool", log.Fields{"Device": device, "GroupID": ig.GroupID, "GroupName": ig.GroupName, "GroupAddr": ig.GroupAddr.String()})
ig.PendingGroupForDevice[device] = time.Now().Add(time.Duration(GroupExpiryTime) * time.Minute)
ig.PendingPoolLock.Unlock()
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
return true
@@ -504,9 +505,9 @@
}
// RestoreIgmpGroupsFromDb to restore igmp groups from database
-func (va *VoltApplication) RestoreIgmpGroupsFromDb() {
+func (va *VoltApplication) RestoreIgmpGroupsFromDb(cntx context.Context) {
- groups, _ := db.GetIgmpGroups()
+ groups, _ := db.GetIgmpGroups(cntx)
for _, group := range groups {
b, ok := group.Value.([]byte)
if !ok {
@@ -532,7 +533,7 @@
if _, err := va.GetIgmpGroupID(ig.GroupID); err != nil {
logger.Warnw(ctx, "GetIgmpGroupID Failed", log.Fields{"igGroupID": ig.GroupID, "Error": err})
}
- ig.RestoreDevices()
+ ig.RestoreDevices(cntx)
if ig.NumDevicesActive() == 0 {
va.AddGroupToPendingPool(&ig)
@@ -544,16 +545,16 @@
// AddIgmpGroup : When the first IGMP packet is received, the MVLAN profile is identified
// for the IGMP group and grp obj is obtained from the available pending pool of groups.
// If not, new group obj will be created based on available group IDs
-func (va *VoltApplication) AddIgmpGroup(mvpName string, gip net.IP, device string) *IgmpGroup {
+func (va *VoltApplication) AddIgmpGroup(cntx context.Context, mvpName string, gip net.IP, device string) *IgmpGroup {
var ig *IgmpGroup
if mvp, grpName := va.GetMvlanProfileForMcIP(mvpName, gip); mvp != nil {
if ig = va.GetGroupFromPendingPool(mvp.Mvlan, device); ig != nil {
logger.Infow(ctx, "Igmp Group obtained from global pending pool", log.Fields{"MvlanProfile": mvpName, "GroupID": ig.GroupID, "Device": device, "GroupName": ig.GroupName, "GroupAddr": ig.GroupAddr.String()})
oldKey := mvp.generateGroupKey(ig.GroupName, ig.GroupAddr.String())
- ig.IgmpGroupReInit(grpName, gip)
+ ig.IgmpGroupReInit(cntx, grpName, gip)
ig.IsGroupStatic = mvp.Groups[grpName].IsStatic
- ig.UpdateIgmpGroup(oldKey, ig.getKey())
+ ig.UpdateIgmpGroup(cntx, oldKey, ig.getKey())
} else {
logger.Infow(ctx, "No Igmp Group available in global pending pool. Creating new Igmp Group", log.Fields{"MvlanProfile": mvpName, "Device": device, "GroupAddr": gip.String()})
if ig = va.GetAvailIgmpGroupID(); ig == nil {
@@ -564,7 +565,7 @@
grpKey := ig.getKey()
va.IgmpGroups.Store(grpKey, ig)
}
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
return ig
@@ -606,7 +607,7 @@
// DelIgmpGroup : When the last subscriber leaves the IGMP group across all the devices
// the IGMP group is removed.
-func (va *VoltApplication) DelIgmpGroup(ig *IgmpGroup) {
+func (va *VoltApplication) DelIgmpGroup(cntx context.Context, ig *IgmpGroup) {
profile, found := GetApplication().MvlanProfilesByTag.Load(ig.Mvlan)
if found {
@@ -621,11 +622,11 @@
logger.Debugw(ctx, "Deleting IGMP Group", log.Fields{"Group": grpKey})
va.PutIgmpGroupID(ig)
va.IgmpGroups.Delete(grpKey)
- _ = db.DelIgmpGroup(grpKey)
+ _ = db.DelIgmpGroup(cntx, grpKey)
} else {
logger.Infow(ctx, "Skipping IgmpGroup Device. Pending Igmp Group Devices present", log.Fields{"GroupID": ig.GroupID, "GroupName": ig.GroupName, "GroupAddr": ig.GroupAddr.String(), "PendingDevices": len(ig.Devices)})
va.AddGroupToPendingPool(ig)
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
}
@@ -807,7 +808,7 @@
}
// ProcessIgmpv2Pkt : This is IGMPv2 packet.
-func (va *VoltApplication) ProcessIgmpv2Pkt(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessIgmpv2Pkt(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// First get the layers of interest
dot1Q := pkt.Layer(layers.LayerTypeDot1Q).(*layers.Dot1Q)
pktVlan := of.VlanType(dot1Q.VLANIdentifier)
@@ -868,11 +869,11 @@
ig.IgmpGroupLock.Unlock()
return
}
- ig.AddReceiver(device, port, igmpv2.GroupAddress, nil, IgmpVersion2, dot1Q.VLANIdentifier, dot1Q.Priority, ponPortID)
+ ig.AddReceiver(cntx, device, port, igmpv2.GroupAddress, nil, IgmpVersion2, dot1Q.VLANIdentifier, dot1Q.Priority, ponPortID)
ig.IgmpGroupLock.Unlock()
} else {
// Create the IGMP group and then add the receiver to the group
- if ig := va.AddIgmpGroup(vpv.MvlanProfileName, igmpv2.GroupAddress, device); ig != nil {
+ if ig := va.AddIgmpGroup(cntx, vpv.MvlanProfileName, igmpv2.GroupAddress, device); ig != nil {
logger.Infow(ctx, "New IGMP Group", log.Fields{"Group": ig.GroupID, "devices": ig.Devices})
ig.IgmpGroupLock.Lock()
// Check for port state to avoid race condition where PortDown event
@@ -885,7 +886,7 @@
ig.IgmpGroupLock.Unlock()
return
}
- ig.AddReceiver(device, port, igmpv2.GroupAddress, nil, IgmpVersion2, dot1Q.VLANIdentifier, dot1Q.Priority, ponPortID)
+ ig.AddReceiver(cntx, device, port, igmpv2.GroupAddress, nil, IgmpVersion2, dot1Q.VLANIdentifier, dot1Q.Priority, ponPortID)
ig.IgmpGroupLock.Unlock()
} else {
logger.Errorw(ctx, "IGMP Group Creation Failed", log.Fields{"Addr": igmpv2.GroupAddress})
@@ -914,10 +915,10 @@
if ig := va.GetIgmpGroup(mvlan, igmpv2.GroupAddress); ig != nil {
ig.IgmpGroupLock.Lock()
// Delete the receiver once the IgmpGroup is identified
- ig.DelReceiver(device, port, igmpv2.GroupAddress, nil, ponPortID)
+ ig.DelReceiver(cntx, device, port, igmpv2.GroupAddress, nil, ponPortID)
ig.IgmpGroupLock.Unlock()
if ig.NumDevicesActive() == 0 {
- va.DelIgmpGroup(ig)
+ va.DelIgmpGroup(cntx, ig)
}
}
} else {
@@ -936,13 +937,13 @@
defer mvp.mvpLock.RUnlock()
if net.ParseIP("0.0.0.0").Equal(igmpv2.GroupAddress) {
- va.processIgmpQueries(device, pktVlan, IgmpVersion2)
+ va.processIgmpQueries(cntx, device, pktVlan, IgmpVersion2)
} else {
if ig := va.GetIgmpGroup(pktVlan, igmpv2.GroupAddress); ig != nil {
ig.IgmpGroupLock.Lock()
igd, ok := ig.Devices[device]
if ok {
- igd.ProcessQuery(igmpv2.GroupAddress, IgmpVersion2)
+ igd.ProcessQuery(cntx, igmpv2.GroupAddress, IgmpVersion2)
} else {
logger.Warnw(ctx, "IGMP Device not found", log.Fields{"Device": device, "Group": igmpv2.GroupAddress})
}
@@ -953,7 +954,7 @@
}
// ProcessIgmpv3Pkt : Process IGMPv3 packet
-func (va *VoltApplication) ProcessIgmpv3Pkt(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessIgmpv3Pkt(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// First get the layers of interest
dot1QLayer := pkt.Layer(layers.LayerTypeDot1Q)
@@ -1018,13 +1019,13 @@
ig.IgmpGroupLock.Unlock()
return
}
- ig.AddReceiver(device, port, group.MulticastAddress, &group, IgmpVersion3,
+ ig.AddReceiver(cntx, device, port, group.MulticastAddress, &group, IgmpVersion3,
dot1Q.VLANIdentifier, dot1Q.Priority, ponPortID)
ig.IgmpGroupLock.Unlock()
} else {
// Create the IGMP group and then add the receiver to the group
logger.Infow(ctx, "IGMP Join received for new group", log.Fields{"Addr": group.MulticastAddress, "Port": port})
- if ig := va.AddIgmpGroup(vpv.MvlanProfileName, group.MulticastAddress, device); ig != nil {
+ if ig := va.AddIgmpGroup(cntx, vpv.MvlanProfileName, group.MulticastAddress, device); ig != nil {
ig.IgmpGroupLock.Lock()
// Check for port state to avoid race condition where PortDown event
// acquired lock before packet processing
@@ -1036,7 +1037,7 @@
ig.IgmpGroupLock.Unlock()
return
}
- ig.AddReceiver(device, port, group.MulticastAddress, &group, IgmpVersion3,
+ ig.AddReceiver(cntx, device, port, group.MulticastAddress, &group, IgmpVersion3,
dot1Q.VLANIdentifier, dot1Q.Priority, ponPortID)
ig.IgmpGroupLock.Unlock()
} else {
@@ -1046,10 +1047,10 @@
} else if ig != nil {
logger.Infow(ctx, "IGMP Leave received for existing group", log.Fields{"Addr": group.MulticastAddress, "Port": port})
ig.IgmpGroupLock.Lock()
- ig.DelReceiver(device, port, group.MulticastAddress, &group, ponPortID)
+ ig.DelReceiver(cntx, device, port, group.MulticastAddress, &group, ponPortID)
ig.IgmpGroupLock.Unlock()
if ig.NumDevicesActive() == 0 {
- va.DelIgmpGroup(ig)
+ va.DelIgmpGroup(cntx, ig)
}
} else {
logger.Warnw(ctx, "IGMP Leave received for unknown group", log.Fields{"Addr": group.MulticastAddress})
@@ -1071,13 +1072,13 @@
defer mvp.mvpLock.RUnlock()
if net.ParseIP("0.0.0.0").Equal(igmpv3.GroupAddress) {
- va.processIgmpQueries(device, pktVlan, IgmpVersion3)
+ va.processIgmpQueries(cntx, device, pktVlan, IgmpVersion3)
} else {
if ig := va.GetIgmpGroup(pktVlan, igmpv3.GroupAddress); ig != nil {
ig.IgmpGroupLock.Lock()
igd, ok := ig.Devices[device]
if ok {
- igd.ProcessQuery(igmpv3.GroupAddress, IgmpVersion3)
+ igd.ProcessQuery(cntx, igmpv3.GroupAddress, IgmpVersion3)
} else {
logger.Warnw(ctx, "IGMP Device not found", log.Fields{"Device": device, "Group": igmpv3.GroupAddress})
}
@@ -1088,7 +1089,7 @@
}
// processIgmpQueries to process the igmp queries
-func (va *VoltApplication) processIgmpQueries(device string, pktVlan of.VlanType, version uint8) {
+func (va *VoltApplication) processIgmpQueries(cntx context.Context, device string, pktVlan of.VlanType, version uint8) {
// This is a generic query and respond with all the groups channels in currently being viewed.
processquery := func(key interface{}, value interface{}) bool {
ig := value.(*IgmpGroup)
@@ -1105,7 +1106,7 @@
}
processQueryForEachChannel := func(key interface{}, value interface{}) bool {
groupAddr := key.(string)
- igd.ProcessQuery(net.ParseIP(groupAddr), version)
+ igd.ProcessQuery(cntx, net.ParseIP(groupAddr), version)
return true
}
igd.GroupChannels.Range(processQueryForEachChannel)
@@ -1143,7 +1144,7 @@
// IgmpProcessPkt to process the IGMP packet received. The packet received brings along with it
// the port on which the packet is received and the device the port is in.
-func (va *VoltApplication) IgmpProcessPkt(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) IgmpProcessPkt(cntx context.Context, device string, port string, pkt gopacket.Packet) {
igmpl := pkt.Layer(layers.LayerTypeIGMP)
if igmpl == nil {
logger.Error(ctx, "Invalid IGMP packet arrived as IGMP packet")
@@ -1152,12 +1153,12 @@
if igmp, ok := igmpl.(*layers.IGMPv1or2); ok {
// This is an IGMPv2 packet.
logger.Debugw(ctx, "IGMPv2 Packet Received", log.Fields{"IPAddr": igmp.GroupAddress})
- va.ProcessIgmpv2Pkt(device, port, pkt)
+ va.ProcessIgmpv2Pkt(cntx, device, port, pkt)
return
}
if igmpv3, ok := igmpl.(*layers.IGMP); ok {
logger.Debugw(ctx, "IGMPv3 Packet Received", log.Fields{"NumOfGroups": igmpv3.NumberOfGroupRecords})
- va.ProcessIgmpv3Pkt(device, port, pkt)
+ va.ProcessIgmpv3Pkt(cntx, device, port, pkt)
}
}
@@ -1180,8 +1181,8 @@
}
// RestoreMvlansFromDb to read from the DB and restore all the MVLANs
-func (va *VoltApplication) RestoreMvlansFromDb() {
- mvlans, _ := db.GetMvlans()
+func (va *VoltApplication) RestoreMvlansFromDb(cntx context.Context) {
+ mvlans, _ := db.GetMvlans(cntx)
for _, mvlan := range mvlans {
b, ok := mvlan.Value.([]byte)
if !ok {
@@ -1223,7 +1224,7 @@
}
//UpdateMvlanProfile - only channel groups be updated
-func (va *VoltApplication) UpdateMvlanProfile(name string, vlan of.VlanType, groups map[string][]string, activeChannelCount int, proxy map[string]common.MulticastGroupProxy) error {
+func (va *VoltApplication) UpdateMvlanProfile(cntx context.Context, name string, vlan of.VlanType, groups map[string][]string, activeChannelCount int, proxy map[string]common.MulticastGroupProxy) error {
mvpIntf, ok := va.MvlanProfilesByName.Load(name)
if !ok {
@@ -1256,7 +1257,7 @@
logger.Info(ctx, "No change in groups config")
if uint32(activeChannelCount) != mvp.MaxActiveChannels {
mvp.MaxActiveChannels = uint32(activeChannelCount)
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile Write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
if prevMaxActiveChannels != mvp.MaxActiveChannels {
@@ -1275,7 +1276,7 @@
mvp.oldGroups = existingGroup
mvp.oldProxy = existingProxy
va.storeMvlansMap(vlan, name, mvp)
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile Write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
if prevMaxActiveChannels != mvp.MaxActiveChannels {
@@ -1321,7 +1322,7 @@
}
// AddMcastConfig for addition of a MVLAN profile
-func (va *VoltApplication) AddMcastConfig(MvlanProfileID string, IgmpProfileID string, IgmpProxyIP string, OltSerialNum string) error {
+func (va *VoltApplication) AddMcastConfig(cntx context.Context, MvlanProfileID string, IgmpProfileID string, IgmpProxyIP string, OltSerialNum string) error {
var mcastCfg *McastConfig
mcastCfg = va.GetMcastConfig(OltSerialNum, MvlanProfileID)
@@ -1371,61 +1372,61 @@
va.IgmpGroups.Range(iterIgmpGroups)
va.storeMcastConfig(OltSerialNum, MvlanProfileID, mcastCfg)
- if err := mcastCfg.WriteToDb(); err != nil {
+ if err := mcastCfg.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "McastConfig Write to DB failed", log.Fields{"OltSerialNum": mcastCfg.OltSerialNum, "MvlanProfileID": mcastCfg.MvlanProfileID})
}
- va.addOltToMvlan(MvlanProfileID, OltSerialNum)
+ va.addOltToMvlan(cntx, MvlanProfileID, OltSerialNum)
return nil
}
-func (va *VoltApplication) addOltToMvlan(MvlanProfileID string, OltSerialNum string) {
+func (va *VoltApplication) addOltToMvlan(cntx context.Context, MvlanProfileID string, OltSerialNum string) {
var mvp *MvlanProfile
if mvpIntf, ok := va.MvlanProfilesByName.Load(MvlanProfileID); ok {
servVersion := IgmpVersion0
mvp = mvpIntf.(*MvlanProfile)
mvp.DevicesList[OltSerialNum] = NoOp
mvp.IgmpServVersion[OltSerialNum] = &servVersion
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile Write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
- mvp.pushIgmpMcastFlows(OltSerialNum)
+ mvp.pushIgmpMcastFlows(cntx, OltSerialNum)
}
}
-func (va *VoltApplication) delOltFromMvlan(MvlanProfileID string, OltSerialNum string) {
+func (va *VoltApplication) delOltFromMvlan(cntx context.Context, MvlanProfileID string, OltSerialNum string) {
var mvp *MvlanProfile
if mvpIntf, ok := va.MvlanProfilesByName.Load(MvlanProfileID); ok {
mvp = mvpIntf.(*MvlanProfile)
//Delete from mvp list
- mvp.removeIgmpMcastFlows(OltSerialNum)
+ mvp.removeIgmpMcastFlows(cntx, OltSerialNum)
delete(mvp.DevicesList, OltSerialNum)
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile Write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
}
}
// DelMcastConfig for addition of a MVLAN profile
-func (va *VoltApplication) DelMcastConfig(MvlanProfileID string, IgmpProfileID string, IgmpProxyIP string, OltSerialNum string) {
+func (va *VoltApplication) DelMcastConfig(cntx context.Context, MvlanProfileID string, IgmpProfileID string, IgmpProxyIP string, OltSerialNum string) {
- va.delOltFromMvlan(MvlanProfileID, OltSerialNum)
+ va.delOltFromMvlan(cntx, MvlanProfileID, OltSerialNum)
va.deleteMcastConfig(OltSerialNum, MvlanProfileID)
- _ = db.DelMcastConfig(McastConfigKey(OltSerialNum, MvlanProfileID))
+ _ = db.DelMcastConfig(cntx, McastConfigKey(OltSerialNum, MvlanProfileID))
if d := va.GetDeviceBySerialNo(OltSerialNum); d != nil {
if mvp := va.GetMvlanProfileByName(MvlanProfileID); mvp != nil {
- va.RemoveGroupsFromPendingPool(d.Name, mvp.Mvlan)
+ va.RemoveGroupsFromPendingPool(cntx, d.Name, mvp.Mvlan)
}
}
}
// DelAllMcastConfig for deletion of all mcast config
-func (va *VoltApplication) DelAllMcastConfig(OltSerialNum string) error {
+func (va *VoltApplication) DelAllMcastConfig(cntx context.Context, OltSerialNum string) error {
deleteIndividualMcastConfig := func(key interface{}, value interface{}) bool {
mcastCfg := value.(*McastConfig)
if mcastCfg.OltSerialNum == OltSerialNum {
- va.DelMcastConfig(mcastCfg.MvlanProfileID, mcastCfg.IgmpProfileID, mcastCfg.IgmpProxyIP.String(), mcastCfg.OltSerialNum)
+ va.DelMcastConfig(cntx, mcastCfg.MvlanProfileID, mcastCfg.IgmpProfileID, mcastCfg.IgmpProxyIP.String(), mcastCfg.OltSerialNum)
}
return true
}
@@ -1434,7 +1435,7 @@
}
// UpdateMcastConfig for addition of a MVLAN profile
-func (va *VoltApplication) UpdateMcastConfig(MvlanProfileID string, IgmpProfileID string, IgmpProxyIP string, OltSerialNum string) error {
+func (va *VoltApplication) UpdateMcastConfig(cntx context.Context, MvlanProfileID string, IgmpProfileID string, IgmpProxyIP string, OltSerialNum string) error {
mcastCfg := va.GetMcastConfig(OltSerialNum, MvlanProfileID)
if mcastCfg == nil {
@@ -1463,7 +1464,7 @@
mcastCfg.IgmpGroupDevices.Range(updateIgdProxyCfg)
}
- if err := mcastCfg.WriteToDb(); err != nil {
+ if err := mcastCfg.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "McastConfig Write to DB failed", log.Fields{"OltSerialNum": mcastCfg.OltSerialNum, "MvlanProfileID": mcastCfg.MvlanProfileID})
}
@@ -1471,21 +1472,21 @@
}
// WriteToDb is utility to write Mcast config Info to database
-func (mc *McastConfig) WriteToDb() error {
+func (mc *McastConfig) WriteToDb(cntx context.Context) error {
mc.Version = database.PresentVersionMap[database.McastConfigPath]
b, err := json.Marshal(mc)
if err != nil {
return err
}
- if err1 := db.PutMcastConfig(McastConfigKey(mc.OltSerialNum, mc.MvlanProfileID), string(b)); err1 != nil {
+ if err1 := db.PutMcastConfig(cntx, McastConfigKey(mc.OltSerialNum, mc.MvlanProfileID), string(b)); err1 != nil {
return err1
}
return nil
}
// RestoreMcastConfigsFromDb to read from the DB and restore Mcast configs
-func (va *VoltApplication) RestoreMcastConfigsFromDb() {
- mcastConfigs, _ := db.GetMcastConfigs()
+func (va *VoltApplication) RestoreMcastConfigsFromDb(cntx context.Context) {
+ mcastConfigs, _ := db.GetMcastConfigs(cntx)
for hash, mcastConfig := range mcastConfigs {
b, ok := mcastConfig.Value.([]byte)
if !ok {
@@ -1504,7 +1505,7 @@
}
// AddMvlanProfile for addition of a MVLAN profile
-func (va *VoltApplication) AddMvlanProfile(name string, mvlan of.VlanType, ponVlan of.VlanType,
+func (va *VoltApplication) AddMvlanProfile(cntx context.Context, name string, mvlan of.VlanType, ponVlan of.VlanType,
groups map[string][]string, isChannelBasedGroup bool, OLTSerialNum []string, activeChannelsPerPon int, proxy map[string]common.MulticastGroupProxy) error {
var mvp *MvlanProfile
@@ -1518,7 +1519,7 @@
if mvp.DevicesList[serialNum] != Nil {
//This is backup restore scenario, just update the profile
logger.Info(ctx, "Add Mvlan : Profile Name already exists, update-the-profile")
- return va.UpdateMvlanProfile(name, mvlan, groups, activeChannelsPerPon, proxy)
+ return va.UpdateMvlanProfile(cntx, name, mvlan, groups, activeChannelsPerPon, proxy)
}
}
}
@@ -1546,7 +1547,7 @@
logger.Debugw(ctx, "Added MVLAN Profile", log.Fields{"MVLAN": mvp.Mvlan, "PonVlan": mvp.PonVlan, "Name": mvp.Name, "Grp IPs": mvp.Groups, "IsPonVlanPresent": mvp.IsPonVlanPresent})
mvp.mvpLock.Unlock()
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile Write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
@@ -1570,7 +1571,7 @@
}
// IgmpTick for igmp tick info
-func (va *VoltApplication) IgmpTick() {
+func (va *VoltApplication) IgmpTick(cntx context.Context) {
tickCount++
if (tickCount % 1000) == 0 {
logger.Debugw(ctx, "Time @ Tick", log.Fields{"Tick": tickCount, "Time": time.Now()})
@@ -1580,10 +1581,10 @@
if ig.NumDevicesActive() != 0 {
if tickCount%10 == ig.Hash()%10 {
ig.IgmpGroupLock.Lock()
- ig.Tick()
+ ig.Tick(cntx)
ig.IgmpGroupLock.Unlock()
if ig.NumDevicesActive() == 0 {
- va.DelIgmpGroup(ig)
+ va.DelIgmpGroup(cntx, ig)
}
}
}
@@ -1600,12 +1601,12 @@
}
//AddIgmpProfile for addition of IGMP Profile
-func (va *VoltApplication) AddIgmpProfile(igmpProfileConfig *common.IGMPConfig) error {
+func (va *VoltApplication) AddIgmpProfile(cntx context.Context, igmpProfileConfig *common.IGMPConfig) error {
var igmpProfile *IgmpProfile
if igmpProfileConfig.ProfileID == DefaultIgmpProfID {
logger.Info(ctx, "Updating default IGMP profile")
- return va.UpdateIgmpProfile(igmpProfileConfig)
+ return va.UpdateIgmpProfile(cntx, igmpProfileConfig)
}
igmpProfile = va.checkIgmpProfileMap(igmpProfileConfig.ProfileID)
@@ -1618,7 +1619,7 @@
va.storeIgmpProfileMap(igmpProfileConfig.ProfileID, igmpProfile)
- if err := igmpProfile.WriteToDb(); err != nil {
+ if err := igmpProfile.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp profile Write to DB failed", log.Fields{"profileID": igmpProfile.ProfileID})
}
@@ -1633,7 +1634,7 @@
return nil
}
-func (va *VoltApplication) resetIgmpProfileToDefault() {
+func (va *VoltApplication) resetIgmpProfileToDefault(cntx context.Context) {
igmpProf := va.getIgmpProfileMap(DefaultIgmpProfID)
defIgmpProf := newDefaultIgmpProfile()
@@ -1651,7 +1652,7 @@
igmpProf.IgmpVerToServer = defIgmpProf.IgmpVerToServer
igmpProf.IgmpSourceIP = defIgmpProf.IgmpSourceIP
- if err := igmpProf.WriteToDb(); err != nil {
+ if err := igmpProf.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp profile Write to DB failed", log.Fields{"profileID": igmpProf.ProfileID})
}
}
@@ -1678,11 +1679,11 @@
}
//DelIgmpProfile for addition of IGMP Profile
-func (va *VoltApplication) DelIgmpProfile(igmpProfileConfig *common.IGMPConfig) error {
+func (va *VoltApplication) DelIgmpProfile(cntx context.Context, igmpProfileConfig *common.IGMPConfig) error {
// Deletion of default igmp profile is blocked from submgr. Keeping additional check for safety.
if igmpProfileConfig.ProfileID == DefaultIgmpProfID {
logger.Info(ctx, "Resetting default IGMP profile")
- va.resetIgmpProfileToDefault()
+ va.resetIgmpProfileToDefault(cntx)
return nil
}
igmpProfile := va.checkIgmpProfileMap(igmpProfileConfig.ProfileID)
@@ -1693,13 +1694,13 @@
va.deleteIgmpProfileMap(igmpProfileConfig.ProfileID)
- _ = db.DelIgmpProfile(igmpProfileConfig.ProfileID)
+ _ = db.DelIgmpProfile(cntx, igmpProfileConfig.ProfileID)
return nil
}
//UpdateIgmpProfile for addition of IGMP Profile
-func (va *VoltApplication) UpdateIgmpProfile(igmpProfileConfig *common.IGMPConfig) error {
+func (va *VoltApplication) UpdateIgmpProfile(cntx context.Context, igmpProfileConfig *common.IGMPConfig) error {
igmpProfile := va.checkIgmpProfileMap(igmpProfileConfig.ProfileID)
if igmpProfile == nil {
logger.Errorw(ctx, "Igmp Profile not found. Unable to update", log.Fields{"Profile ID": igmpProfileConfig.ProfileID})
@@ -1738,7 +1739,7 @@
igmpProfile.IgmpSourceIP = net.ParseIP(igmpProfileConfig.IgmpSourceIP)
}
- if err := igmpProfile.WriteToDb(); err != nil {
+ if err := igmpProfile.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp profile Write to DB failed", log.Fields{"profileID": igmpProfile.ProfileID})
}
@@ -1746,9 +1747,9 @@
}
// RestoreIGMPProfilesFromDb to read from the DB and restore IGMP Profiles
-func (va *VoltApplication) RestoreIGMPProfilesFromDb() {
+func (va *VoltApplication) RestoreIGMPProfilesFromDb(cntx context.Context) {
// Loading IGMP profiles
- igmpProfiles, _ := db.GetIgmpProfiles()
+ igmpProfiles, _ := db.GetIgmpProfiles(cntx)
for _, igmpProfile := range igmpProfiles {
b, ok := igmpProfile.Value.([]byte)
if !ok {
@@ -1777,13 +1778,13 @@
}
// DelMvlanProfile for deletion of a MVLAN group
-func (va *VoltApplication) DelMvlanProfile(name string) error {
+func (va *VoltApplication) DelMvlanProfile(cntx context.Context, name string) error {
if mvpIntf, ok := va.MvlanProfilesByName.Load(name); ok {
mvp := mvpIntf.(*MvlanProfile)
if len(mvp.DevicesList) == 0 {
mvp.DeleteInProgress = true
- mvp.DelFromDb()
+ mvp.DelFromDb(cntx)
va.deleteMvlansMap(mvp.Mvlan, name)
logger.Debugw(ctx, "Deleted MVLAN Profile", log.Fields{"Name": mvp.Name})
} else {
@@ -1839,7 +1840,7 @@
}
// ReceiverDownInd to send receiver down indication
-func (va *VoltApplication) ReceiverDownInd(device string, port string) {
+func (va *VoltApplication) ReceiverDownInd(cntx context.Context, device string, port string) {
logger.Infow(ctx, " Receiver Indication: DOWN", log.Fields{"device": device, "port": port})
ponPortID := va.GetPonPortID(device, port)
@@ -1847,10 +1848,10 @@
del := func(key interface{}, value interface{}) bool {
ig := value.(*IgmpGroup)
ig.IgmpGroupLock.Lock()
- ig.DelReceiveronDownInd(device, port, ponPortID)
+ ig.DelReceiveronDownInd(cntx, device, port, ponPortID)
ig.IgmpGroupLock.Unlock()
if ig.NumDevicesActive() == 0 {
- va.DelIgmpGroup(ig)
+ va.DelIgmpGroup(cntx, ig)
}
return true
}
diff --git a/internal/pkg/application/igmpgroup.go b/internal/pkg/application/igmpgroup.go
index 629d92c..a7107cd 100644
--- a/internal/pkg/application/igmpgroup.go
+++ b/internal/pkg/application/igmpgroup.go
@@ -16,6 +16,7 @@
package application
import (
+ "context"
"encoding/json"
"net"
"sync"
@@ -76,7 +77,7 @@
}
// IgmpGroupReInit to re-initialize igmp group members
-func (ig *IgmpGroup) IgmpGroupReInit(name string, gip net.IP) {
+func (ig *IgmpGroup) IgmpGroupReInit(cntx context.Context, name string, gip net.IP) {
logger.Infow(ctx, "Reinitialize Igmp Group", log.Fields{"GroupID": ig.GroupID, "OldName": ig.GroupName, "Name": name, "OldAddr": ig.GroupAddr.String(), "GroupAddr": gip.String()})
@@ -88,12 +89,12 @@
}
for _, igd := range ig.Devices {
- igd.IgmpGroupDeviceReInit(ig)
+ igd.IgmpGroupDeviceReInit(cntx, ig)
}
}
// updateGroupName to update group name
-func (ig *IgmpGroup) updateGroupName(newGroupName string) {
+func (ig *IgmpGroup) updateGroupName(cntx context.Context, newGroupName string) {
if !ig.IsChannelBasedGroup {
logger.Errorw(ctx, "Group name update not supported for GroupChannel based group", log.Fields{"Ig": ig})
return
@@ -101,36 +102,36 @@
oldKey := ig.getKey()
ig.GroupName = newGroupName
for _, igd := range ig.Devices {
- igd.updateGroupName(newGroupName)
+ igd.updateGroupName(cntx, newGroupName)
}
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
if !ig.IsChannelBasedGroup {
- _ = db.DelIgmpGroup(oldKey)
+ _ = db.DelIgmpGroup(cntx, oldKey)
}
}
//HandleGroupMigration - handles migration of group members between static & dynamic
-func (ig *IgmpGroup) HandleGroupMigration(deviceID string, groupAddr net.IP) {
+func (ig *IgmpGroup) HandleGroupMigration(cntx context.Context, deviceID string, groupAddr net.IP) {
var group *layers.IGMPv3GroupRecord
app := GetApplication()
if deviceID == "" {
logger.Infow(ctx, "Handle Group Migration Request for all devices", log.Fields{"DeviceID": deviceID, "GroupAddr": groupAddr, "IG": ig.GroupName, "Mvlan": ig.Mvlan})
for device := range ig.Devices {
- ig.HandleGroupMigration(device, groupAddr)
+ ig.HandleGroupMigration(cntx, device, groupAddr)
}
} else {
logger.Infow(ctx, "Handle Group Migration Request", log.Fields{"DeviceID": deviceID, "GroupAddr": groupAddr, "IG": ig.GroupName})
var newIg *IgmpGroup
- receivers := ig.DelIgmpChannel(deviceID, groupAddr)
+ receivers := ig.DelIgmpChannel(cntx, deviceID, groupAddr)
if ig.NumDevicesActive() == 0 {
- app.DelIgmpGroup(ig)
+ app.DelIgmpGroup(cntx, ig)
}
if newIg = app.GetIgmpGroup(ig.Mvlan, groupAddr); newIg == nil {
logger.Infow(ctx, "IG Group doesn't exist, creating new group", log.Fields{"DeviceID": deviceID, "GroupAddr": groupAddr, "IG": ig.GroupName, "Mvlan": ig.Mvlan})
- if newIg = app.AddIgmpGroup(app.GetMvlanProfileByTag(ig.Mvlan).Name, groupAddr, deviceID); newIg == nil {
+ if newIg = app.AddIgmpGroup(cntx, app.GetMvlanProfileByTag(ig.Mvlan).Name, groupAddr, deviceID); newIg == nil {
logger.Errorw(ctx, "Group Creation failed during group migration", log.Fields{"DeviceID": deviceID, "GroupAddr": groupAddr})
return
}
@@ -161,7 +162,7 @@
}
logger.Infow(ctx, "Adding receiver to new group", log.Fields{"DeviceID": deviceID, "GroupAddr": groupAddr, "newIg": newIg.GroupName, "IGP": igp})
ponPort := GetApplication().GetPonPortID(deviceID, port)
- newIg.AddReceiver(deviceID, port, groupAddr, group, igp.Version, igp.CVlan, igp.Pbit, ponPort)
+ newIg.AddReceiver(cntx, deviceID, port, groupAddr, group, igp.Version, igp.CVlan, igp.Pbit, ponPort)
}
newIg.IgmpGroupLock.Unlock()
}
@@ -169,11 +170,11 @@
// AddIgmpGroupDevice add a device to the group which happens when the first receiver of the device
// is added to the IGMP group.
-func (ig *IgmpGroup) AddIgmpGroupDevice(device string, id uint32, version uint8) *IgmpGroupDevice {
+func (ig *IgmpGroup) AddIgmpGroupDevice(cntx context.Context, device string, id uint32, version uint8) *IgmpGroupDevice {
logger.Infow(ctx, "Adding Device to IGMP group", log.Fields{"Device": device, "GroupName": ig.GroupName})
igd := NewIgmpGroupDevice(device, ig, id, version)
ig.Devices[device] = igd
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device, "GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
return igd
@@ -181,14 +182,14 @@
// DelIgmpGroupDevice delete the device from the group which happens when we receive a leave or when
// there is not response for IGMP query from the receiver
-func (ig *IgmpGroup) DelIgmpGroupDevice(igd *IgmpGroupDevice) {
+func (ig *IgmpGroup) DelIgmpGroupDevice(cntx context.Context, igd *IgmpGroupDevice) {
logger.Infow(ctx, "Deleting Device from IGMP group", log.Fields{"Device": igd.Device, "Name": ig.GroupName})
va := GetApplication()
countersToBeUpdated := false
if igd.NumReceivers() != 0 {
countersToBeUpdated = true
}
- igd.DelAllChannels()
+ igd.DelAllChannels(cntx)
//Clear all internal maps so that the groups can be reused
igd.PortChannelMap.Range(func(key, value interface{}) bool {
@@ -215,34 +216,34 @@
logger.Debugw(ctx, "Igd deleted from mcast config", log.Fields{"mvlan": mcastCfg.MvlanProfileID, "groupId": igd.GroupID})
}
if !igd.GroupInstalled {
- _ = db.DelIgmpDevice(igd.Mvlan, ig.GroupName, ig.GroupAddr, igd.Device)
+ _ = db.DelIgmpDevice(cntx, igd.Mvlan, ig.GroupName, ig.GroupAddr, igd.Device)
delete(ig.Devices, igd.Device)
}
}
// AddReceiver delete the device from the group which happens when we receive a leave or when
// there is not response for IGMP query from the receiver
-func (ig *IgmpGroup) AddReceiver(device string, port string, groupIP net.IP,
+func (ig *IgmpGroup) AddReceiver(cntx context.Context, device string, port string, groupIP net.IP,
group *layers.IGMPv3GroupRecord, ver uint8, cvlan uint16, pbit uint8, ponPort uint32) {
logger.Debugw(ctx, "Adding Receiver", log.Fields{"Port": port})
- if igd, ok := ig.getIgmpGroupDevice(device); !ok {
- igd = ig.AddIgmpGroupDevice(device, ig.GroupID, ver)
- igd.AddReceiver(port, groupIP, group, ver, cvlan, pbit, ponPort)
+ if igd, ok := ig.getIgmpGroupDevice(cntx, device); !ok {
+ igd = ig.AddIgmpGroupDevice(cntx, device, ig.GroupID, ver)
+ igd.AddReceiver(cntx, port, groupIP, group, ver, cvlan, pbit, ponPort)
} else {
logger.Infow(ctx, "IGMP Group Receiver", log.Fields{"IGD": igd.Device})
- igd.AddReceiver(port, groupIP, group, ver, cvlan, pbit, ponPort)
+ igd.AddReceiver(cntx, port, groupIP, group, ver, cvlan, pbit, ponPort)
}
}
-func (ig *IgmpGroup) getIgmpGroupDevice(device string) (*IgmpGroupDevice, bool) {
+func (ig *IgmpGroup) getIgmpGroupDevice(cntx context.Context, device string) (*IgmpGroupDevice, bool) {
ig.PendingPoolLock.Lock()
defer ig.PendingPoolLock.Unlock()
if _, ok := ig.PendingGroupForDevice[device]; ok {
logger.Infow(ctx, "Removing the IgmpGroupDevice from pending pool", log.Fields{"GroupID": ig.GroupID, "Device": device, "GroupName": ig.GroupName, "GroupAddr": ig.GroupAddr.String()})
delete(ig.PendingGroupForDevice, device)
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
}
@@ -252,7 +253,7 @@
// DelReceiveronDownInd deletes a receiver which is the combination of device (OLT)
// and port on Port Down event
-func (ig *IgmpGroup) DelReceiveronDownInd(device string, port string, ponPortID uint32) {
+func (ig *IgmpGroup) DelReceiveronDownInd(cntx context.Context, device string, port string, ponPortID uint32) {
logger.Debugw(ctx, "Deleting Receiver for Group", log.Fields{"Device": device, "port": port})
mvp := GetApplication().GetMvlanProfileByTag(ig.Mvlan)
@@ -274,23 +275,23 @@
for _, groupAddr := range ipsList {
logger.Debugw(ctx, "Port Channels", log.Fields{"Port": port, "IPsList": ipsList, "GroupAddr": groupAddr, "Len": len(ipsList)})
- igd.DelReceiver(groupAddr, port, nil, ponPortID)
+ igd.DelReceiver(cntx, groupAddr, port, nil, ponPortID)
}
if igd.NumReceivers() == 0 {
- ig.DelIgmpGroupDevice(igd)
+ ig.DelIgmpGroupDevice(cntx, igd)
}
}
// DelReceiver deletes a receiver which is the combination of device (OLT)
// and port
-func (ig *IgmpGroup) DelReceiver(device string, port string, groupAddr net.IP, group *layers.IGMPv3GroupRecord, ponPortID uint32) {
+func (ig *IgmpGroup) DelReceiver(cntx context.Context, device string, port string, groupAddr net.IP, group *layers.IGMPv3GroupRecord, ponPortID uint32) {
logger.Debugw(ctx, "Deleting Receiver for Group", log.Fields{"Device": device, "port": port, "GroupIP": groupAddr.String()})
if igd, ok := ig.Devices[device]; ok {
//igd.DelReceiverForGroupAddr(groupAddr, port)
- igd.DelReceiver(groupAddr, port, group, ponPortID)
+ igd.DelReceiver(cntx, groupAddr, port, group, ponPortID)
if igd.NumReceivers() == 0 {
- ig.DelIgmpGroupDevice(igd)
+ ig.DelIgmpGroupDevice(cntx, igd)
}
}
}
@@ -329,18 +330,18 @@
}
// DelIgmpChannel deletes all receivers for the provided igmp group channel for the given device
-func (ig *IgmpGroup) DelIgmpChannel(deviceID string, groupAddr net.IP) map[string]*IgmpGroupPort {
+func (ig *IgmpGroup) DelIgmpChannel(cntx context.Context, deviceID string, groupAddr net.IP) map[string]*IgmpGroupPort {
logger.Infow(ctx, "Deleting Channel from devices", log.Fields{"Device": deviceID, "Group": ig.GroupName, "Channel": groupAddr.String()})
if deviceID == "" {
for device := range ig.Devices {
- ig.DelIgmpChannel(device, groupAddr)
+ ig.DelIgmpChannel(cntx, device, groupAddr)
}
return nil
}
igd := ig.Devices[deviceID]
- receivers := igd.DelChannelReceiver(groupAddr)
+ receivers := igd.DelChannelReceiver(cntx, groupAddr)
if igd.NumReceivers() == 0 {
- ig.DelIgmpGroupDevice(igd)
+ ig.DelIgmpGroupDevice(cntx, igd)
}
return receivers
}
@@ -373,7 +374,7 @@
}
// Tick for Addition of groups to an MVLAN profile
-func (ig *IgmpGroup) Tick() {
+func (ig *IgmpGroup) Tick(cntx context.Context) {
now := time.Now()
for _, igd := range ig.Devices {
var igdChangeCnt uint8
@@ -399,14 +400,14 @@
igd.GroupChannels.Range(sendQueryForAllChannels)
}
if now.After(igd.QueryExpiryTime) {
- igd.QueryExpiry()
+ igd.QueryExpiry(cntx)
// This will keep it quiet till the next query time and then
// it will be reset to a value after the query initiation time
igd.QueryExpiryTime = igd.NextQueryTime
logger.Debugw(ctx, "Expiry", log.Fields{"NextQuery": igd.NextQueryTime, "Expiry": igd.QueryExpiryTime})
igdChangeCnt++
if igd.NumReceivers() == 0 {
- ig.DelIgmpGroupDevice(igd)
+ ig.DelIgmpGroupDevice(cntx, igd)
continue
}
}
@@ -414,7 +415,7 @@
igdChangeCnt += igd.Tick()
if igdChangeCnt > 0 {
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device,
"GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
@@ -426,12 +427,12 @@
// expiry, process the consolidated response for each of the devices participating
// in the MC stream. When a device has no receivers, the device is deleted
// from the group.
-func (ig *IgmpGroup) QueryExpiry() {
+func (ig *IgmpGroup) QueryExpiry(cntx context.Context) {
for _, igd := range ig.Devices {
if _, ok := GetApplication().DevicesDisc.Load(igd.Device); ok {
- igd.QueryExpiry()
+ igd.QueryExpiry(cntx)
if igd.NumReceivers() == 0 {
- ig.DelIgmpGroupDevice(igd)
+ ig.DelIgmpGroupDevice(cntx, igd)
}
} else {
@@ -496,10 +497,10 @@
}
// RestoreDevices : IGMP group write to DB
-func (ig *IgmpGroup) RestoreDevices() {
+func (ig *IgmpGroup) RestoreDevices(cntx context.Context) {
- ig.migrateIgmpDevices()
- devices, _ := db.GetIgmpDevices(ig.Mvlan, ig.GroupName, ig.GroupAddr)
+ ig.migrateIgmpDevices(cntx)
+ devices, _ := db.GetIgmpDevices(cntx, ig.Mvlan, ig.GroupName, ig.GroupAddr)
for _, device := range devices {
b, ok := device.Value.([]byte)
if !ok {
@@ -526,7 +527,7 @@
logger.Debugw(ctx, "VGC igd upgrade", log.Fields{"igd grp name": igd.GroupName})
igd.NextQueryTime = time.Now().Add(time.Duration(igd.proxyCfg.KeepAliveInterval) * time.Second)
igd.QueryExpiryTime = time.Now().Add(time.Duration(igd.proxyCfg.KeepAliveInterval) * time.Second)
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device,
"GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
@@ -534,10 +535,10 @@
ig.Devices[igd.Device] = igd
if ig.IsChannelBasedGroup {
- channel, _ := db.GetIgmpChannel(igd.Mvlan, igd.GroupName, igd.Device, igd.GroupAddr)
- igd.RestoreChannel([]byte(channel))
+ channel, _ := db.GetIgmpChannel(cntx, igd.Mvlan, igd.GroupName, igd.Device, igd.GroupAddr)
+ igd.RestoreChannel(cntx, []byte(channel))
} else {
- igd.RestoreChannels()
+ igd.RestoreChannels(cntx)
}
igd.PortChannelMap.Range(printPortChannel)
logger.Infow(ctx, "Group Device Restored", log.Fields{"IGD": igd})
@@ -558,20 +559,20 @@
}
// WriteToDb is utility to write Igmp Group Info to database
-func (ig *IgmpGroup) WriteToDb() error {
+func (ig *IgmpGroup) WriteToDb(cntx context.Context) error {
ig.Version = database.PresentVersionMap[database.IgmpGroupPath]
b, err := json.Marshal(ig)
if err != nil {
return err
}
- if err1 := db.PutIgmpGroup(ig.getKey(), string(b)); err1 != nil {
+ if err1 := db.PutIgmpGroup(cntx, ig.getKey(), string(b)); err1 != nil {
return err1
}
return nil
}
// UpdateIgmpGroup : When the pending group is allocated to new
-func (ig *IgmpGroup) UpdateIgmpGroup(oldKey, newKey string) {
+func (ig *IgmpGroup) UpdateIgmpGroup(cntx context.Context, oldKey, newKey string) {
//If the group is allocated to same McastGroup, no need to update the
//IgmpGroups map
@@ -581,15 +582,15 @@
logger.Infow(ctx, "Updating Igmp Group with new MVP Group Info", log.Fields{"OldKey": oldKey, "NewKey": newKey, "GroupID": ig.GroupID})
GetApplication().IgmpGroups.Delete(oldKey)
- _ = db.DelIgmpGroup(oldKey)
+ _ = db.DelIgmpGroup(cntx, oldKey)
GetApplication().IgmpGroups.Store(newKey, ig)
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
}
-func (ig *IgmpGroup) removeExpiredGroupFromDevice() {
+func (ig *IgmpGroup) removeExpiredGroupFromDevice(cntx context.Context) {
ig.PendingPoolLock.Lock()
defer ig.PendingPoolLock.Unlock()
@@ -613,13 +614,13 @@
// Remove the group entry from device and remove the IgmpDev Obj
// from IgmpGrp Pending pool
if groupExistsInPendingPool {
- ig.DeleteIgmpGroupDevice(device)
+ ig.DeleteIgmpGroupDevice(cntx, device)
}
}
}
//DeleteIgmpGroupDevice - removes the IgmpGroupDevice obj from IgmpGroup and database
-func (ig *IgmpGroup) DeleteIgmpGroupDevice(device string) {
+func (ig *IgmpGroup) DeleteIgmpGroupDevice(cntx context.Context, device string) {
logger.Infow(ctx, "Deleting IgmpGroupDevice from IG Pending Pool", log.Fields{"Device": device, "GroupID": ig.GroupID, "GroupName": ig.GroupName, "GroupAddr": ig.GroupAddr.String(), "PendingDevices": len(ig.Devices)})
@@ -627,24 +628,24 @@
igd.DelMcGroup(true)
delete(ig.Devices, device)
delete(ig.PendingGroupForDevice, device)
- _ = db.DelIgmpDevice(igd.Mvlan, igd.GroupName, igd.GroupAddr, igd.Device)
+ _ = db.DelIgmpDevice(cntx, igd.Mvlan, igd.GroupName, igd.GroupAddr, igd.Device)
//If the group is not associated to any other device, then the entire Igmp Group obj itself can be removed
if ig.NumDevicesAll() == 0 {
logger.Infow(ctx, "Deleting IgmpGroup as all pending groups has expired", log.Fields{"Device": device, "GroupID": ig.GroupID, "GroupName": ig.GroupName, "GroupAddr": ig.GroupAddr.String(), "PendingDevices": len(ig.Devices)})
- GetApplication().DelIgmpGroup(ig)
+ GetApplication().DelIgmpGroup(cntx, ig)
return
}
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
}
// DelIgmpGroup deletes all devices for the provided igmp group
-func (ig *IgmpGroup) DelIgmpGroup() {
+func (ig *IgmpGroup) DelIgmpGroup(cntx context.Context) {
logger.Infow(ctx, "Deleting All Device for Group", log.Fields{"Group": ig.GroupName})
for _, igd := range ig.Devices {
- ig.DelIgmpGroupDevice(igd)
+ ig.DelIgmpGroupDevice(cntx, igd)
}
- GetApplication().DelIgmpGroup(ig)
+ GetApplication().DelIgmpGroup(cntx, ig)
}
diff --git a/internal/pkg/application/igmpgroupchannel.go b/internal/pkg/application/igmpgroupchannel.go
index ed39d23..d17e209 100644
--- a/internal/pkg/application/igmpgroupchannel.go
+++ b/internal/pkg/application/igmpgroupchannel.go
@@ -16,6 +16,7 @@
package application
import (
+ "context"
"encoding/json"
"net"
@@ -77,10 +78,10 @@
}
// RestorePorts to restore ports
-func (igc *IgmpGroupChannel) RestorePorts() {
+func (igc *IgmpGroupChannel) RestorePorts(cntx context.Context) {
- igc.migrateIgmpPorts()
- ports, _ := db.GetIgmpRcvrs(igc.Mvlan, igc.GroupAddr, igc.Device)
+ igc.migrateIgmpPorts(cntx)
+ ports, _ := db.GetIgmpRcvrs(cntx, igc.Mvlan, igc.GroupAddr, igc.Device)
for _, port := range ports {
b, ok := port.Value.([]byte)
if !ok {
@@ -94,18 +95,18 @@
logger.Warn(ctx, "Failed to decode port from DB")
}
}
- if err := igc.WriteToDb(); err != nil {
+ if err := igc.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group channel Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
}
// WriteToDb is utility to write IGMPGroupChannel Info to database
-func (igc *IgmpGroupChannel) WriteToDb() error {
+func (igc *IgmpGroupChannel) WriteToDb(cntx context.Context) error {
b, err := json.Marshal(igc)
if err != nil {
return err
}
- if err1 := db.PutIgmpChannel(igc.Mvlan, igc.GroupName, igc.Device, igc.GroupAddr, string(b)); err1 != nil {
+ if err1 := db.PutIgmpChannel(cntx, igc.Mvlan, igc.GroupName, igc.Device, igc.GroupAddr, string(b)); err1 != nil {
return err1
}
logger.Info(ctx, "IGC Updated")
@@ -246,7 +247,7 @@
// ProcessSources process the received list of either included sources or the excluded sources
// The return value indicate sif the group is modified and needs to be informed
// to the upstream multicast servers
-func (igc *IgmpGroupChannel) ProcessSources(port string, ip []net.IP, incl bool) (bool, bool) {
+func (igc *IgmpGroupChannel) ProcessSources(cntx context.Context, port string, ip []net.IP, incl bool) (bool, bool) {
groupChanged := false
groupExclUpdated := false
receiverSrcListEmpty := false
@@ -341,7 +342,7 @@
}
groupExclUpdated = igc.UpdateExclSource(ip)
}
- if err := igp.WriteToDb(igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
+ if err := igp.WriteToDb(cntx, igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
logger.Errorw(ctx, "Igmp group port Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
return (groupChanged || groupExclUpdated), receiverSrcListEmpty
@@ -359,7 +360,7 @@
// AddReceiver add the receiver to the device and perform other actions such as adding the group
// to the physical device, add members, add flows to point the MC packets to the
// group. Also, send a IGMP report upstream if there is a change in the group
-func (igc *IgmpGroupChannel) AddReceiver(port string, group *layers.IGMPv3GroupRecord, cvlan uint16, pbit uint8) bool {
+func (igc *IgmpGroupChannel) AddReceiver(cntx context.Context, port string, group *layers.IGMPv3GroupRecord, cvlan uint16, pbit uint8) bool {
var igp *IgmpGroupPort
var groupModified = false
@@ -409,7 +410,7 @@
logger.Debugw(ctx, "New IGMP receiver", log.Fields{"Group": igc.GroupAddr.String(), "Port": port})
if len(igc.NewReceivers) == 1 && len(igc.CurReceivers) == 0 {
groupModified = true
- igc.AddMcFlow()
+ igc.AddMcFlow(cntx)
logger.Debugw(ctx, "Added New Flow", log.Fields{"Group": igc.GroupAddr.String(), "Port": port})
}
if !incl {
@@ -419,7 +420,7 @@
}
// Process the include/exclude list which may end up modifying the group
- if change, _ := igc.ProcessSources(port, ip, incl); change {
+ if change, _ := igc.ProcessSources(cntx, port, ip, incl); change {
groupModified = true
}
igc.ProcessMode(port, incl)
@@ -435,10 +436,10 @@
logger.Debugw(ctx, "Channel Receiver Added", log.Fields{"Group Channel": igc.GroupAddr, "Group Port": igp})
- if err := igc.WriteToDb(); err != nil {
+ if err := igc.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group channel Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
- if err := igp.WriteToDb(igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
+ if err := igp.WriteToDb(cntx, igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
logger.Errorw(ctx, "Igmp group port Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
return isNewReceiver
@@ -446,7 +447,7 @@
// DelReceiver is called when Query expiry happened for a receiver. This removes the receiver from the
// the group
-func (igc *IgmpGroupChannel) DelReceiver(port string, incl bool, srcList []net.IP) bool {
+func (igc *IgmpGroupChannel) DelReceiver(cntx context.Context, port string, incl bool, srcList []net.IP) bool {
// The receiver may exist either in NewReceiver list or
// the CurReceivers list. Find and remove it from either
// of the lists.
@@ -455,7 +456,7 @@
logger.Debugw(ctx, "Current Receivers", log.Fields{"Current": igc.CurReceivers})
receiversUpdated := false
- groupModified, receiverSrcListEmpty := igc.ProcessSources(port, srcList, incl)
+ groupModified, receiverSrcListEmpty := igc.ProcessSources(cntx, port, srcList, incl)
if len(srcList) == 0 || len(igc.IncludeList) == 0 || receiverSrcListEmpty {
if igp, ok := igc.NewReceivers[port]; ok {
@@ -478,11 +479,11 @@
return false
}
}
- _ = db.DelIgmpRcvr(igc.Mvlan, igc.GroupAddr, igc.Device, port)
+ _ = db.DelIgmpRcvr(cntx, igc.Mvlan, igc.GroupAddr, igc.Device, port)
}
if igc.NumReceivers() == 0 {
- igc.DelMcFlow()
+ igc.DelMcFlow(cntx)
mvp := GetApplication().GetMvlanProfileByTag(igc.Mvlan)
/* If proxy is configured and NumReceivers is 0, then we can reset the igc src list so that we send leave */
if _, ok := mvp.Proxy[igc.GroupName]; ok {
@@ -496,7 +497,7 @@
igc.SendReport(false)
logger.Infow(ctx, "Updated SourceList for Channel", log.Fields{"Current": igc.CurReceivers, "New": igc.NewReceivers})
}
- if err := igc.WriteToDb(); err != nil {
+ if err := igc.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group channel Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
logger.Infow(ctx, "Updated Receiver info for Channel", log.Fields{"Current": igc.CurReceivers, "New": igc.NewReceivers})
@@ -505,11 +506,11 @@
}
// DelAllReceivers deletes all receiver for the provided igmp device
-func (igc *IgmpGroupChannel) DelAllReceivers() {
+func (igc *IgmpGroupChannel) DelAllReceivers(cntx context.Context) {
logger.Infow(ctx, "Deleting All Receiver for Channel", log.Fields{"Device": igc.Device, "Channel": igc.GroupAddr.String()})
- _ = db.DelAllIgmpRcvr(igc.Mvlan, igc.GroupAddr, igc.Device)
+ _ = db.DelAllIgmpRcvr(cntx, igc.Mvlan, igc.GroupAddr, igc.Device)
igc.Exclude = 0
- igc.DelMcFlow()
+ igc.DelMcFlow(cntx)
igc.SendLeaveToServer()
logger.Infow(ctx, "MC Flow deleted and Leave sent", log.Fields{"Channel": igc.GroupAddr.String(), "Device": igc.Device})
}
@@ -581,18 +582,18 @@
}
// AddMcFlow adds flow to the device when the first receiver joins
-func (igc *IgmpGroupChannel) AddMcFlow() {
+func (igc *IgmpGroupChannel) AddMcFlow(cntx context.Context) {
flow, err := igc.BuildMcFlow()
if err != nil {
logger.Warnw(ctx, "MC Flow Build Failed", log.Fields{"Reason": err.Error()})
return
}
port, _ := GetApplication().GetNniPort(igc.Device)
- _ = cntlr.GetController().AddFlows(port, igc.Device, flow)
+ _ = cntlr.GetController().AddFlows(cntx, port, igc.Device, flow)
}
// DelMcFlow deletes flow from the device when the last receiver leaves
-func (igc *IgmpGroupChannel) DelMcFlow() {
+func (igc *IgmpGroupChannel) DelMcFlow(cntx context.Context) {
flow, err := igc.BuildMcFlow()
if err != nil {
logger.Warnw(ctx, "MC Flow Build Failed", log.Fields{"Reason": err.Error()})
@@ -603,7 +604,7 @@
if mvpIntf, _ := GetApplication().MvlanProfilesByTag.Load(igc.Mvlan); mvpIntf != nil {
mvp := mvpIntf.(*MvlanProfile)
- err := mvp.DelFlows(device, flow)
+ err := mvp.DelFlows(cntx, device, flow)
if err != nil {
logger.Warnw(ctx, "Delering IGMP Flow for device failed ", log.Fields{"Device": device, "err": err})
}
diff --git a/internal/pkg/application/igmpgroupdevice.go b/internal/pkg/application/igmpgroupdevice.go
index 07c2f82..0d59747 100644
--- a/internal/pkg/application/igmpgroupdevice.go
+++ b/internal/pkg/application/igmpgroupdevice.go
@@ -16,6 +16,7 @@
package application
import (
+ "context"
"encoding/json"
"net"
"sync"
@@ -103,12 +104,12 @@
// IgmpGroupDeviceReInit is re-initializer for a device. The default IGMP version is set to 3
// as the protocol defines the way to manage backward compatibility
-func (igd *IgmpGroupDevice) IgmpGroupDeviceReInit(ig *IgmpGroup) {
+func (igd *IgmpGroupDevice) IgmpGroupDeviceReInit(cntx context.Context, ig *IgmpGroup) {
logger.Infow(ctx, "Reinitialize Igmp Group Device", log.Fields{"Device": igd.Device, "GroupID": ig.GroupID, "OldName": igd.GroupName, "Name": ig.GroupName, "OldAddr": igd.GroupAddr.String(), "GroupAddr": ig.GroupAddr.String()})
if (igd.GroupName != ig.GroupName) || !igd.GroupAddr.Equal(ig.GroupAddr) {
- _ = db.DelIgmpDevice(igd.Mvlan, igd.GroupName, igd.GroupAddr, igd.Device)
+ _ = db.DelIgmpDevice(cntx, igd.Mvlan, igd.GroupName, igd.GroupAddr, igd.Device)
igd.GroupName = ig.GroupName
igd.GroupAddr = ig.GroupAddr
}
@@ -126,7 +127,7 @@
mcastCfg.IgmpGroupDevices.Store(ig.GroupID, igd)
logger.Debugw(ctx, "Igd added to mcast config", log.Fields{"mvlan": mcastCfg.MvlanProfileID, "groupId": ig.GroupID})
}
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device, "GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
}
@@ -144,24 +145,24 @@
}
// updateGroupName to update the group name
-func (igd *IgmpGroupDevice) updateGroupName(newGroupName string) {
+func (igd *IgmpGroupDevice) updateGroupName(cntx context.Context, newGroupName string) {
oldName := igd.GroupName
igd.GroupName = newGroupName
updateGroupName := func(key, value interface{}) bool {
igc := value.(*IgmpGroupChannel)
igc.GroupName = newGroupName
- if err := igc.WriteToDb(); err != nil {
+ if err := igc.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group channel Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
- _ = db.DelIgmpChannel(igc.Mvlan, oldName, igc.Device, igc.GroupAddr)
+ _ = db.DelIgmpChannel(cntx, igc.Mvlan, oldName, igc.Device, igc.GroupAddr)
return true
}
igd.GroupChannels.Range(updateGroupName)
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device, "GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
- _ = db.DelIgmpDevice(igd.Mvlan, oldName, igd.GroupAddr, igd.Device)
+ _ = db.DelIgmpDevice(cntx, igd.Mvlan, oldName, igd.GroupAddr, igd.Device)
}
// NewIgmpGroupDeviceFromBytes is to create the IGMP group port from a byte slice
@@ -184,14 +185,14 @@
}
// RestoreChannel to restore channel
-func (igd *IgmpGroupDevice) RestoreChannel(igmpGroupChannel []byte) {
+func (igd *IgmpGroupDevice) RestoreChannel(cntx context.Context, igmpGroupChannel []byte) {
if igc, err := NewIgmpGroupChannelFromBytes(igmpGroupChannel); err == nil {
igc.ServVersion = igd.ServVersion
igc.IgmpProxyIP = &igd.IgmpProxyIP
igc.proxyCfg = &igd.proxyCfg
igd.GroupChannels.Store(igc.GroupAddr.String(), igc)
- igc.RestorePorts()
+ igc.RestorePorts(cntx)
for port, igp := range igc.NewReceivers {
ipsList := []net.IP{}
@@ -215,10 +216,10 @@
}
// RestoreChannels to restore channels
-func (igd *IgmpGroupDevice) RestoreChannels() {
+func (igd *IgmpGroupDevice) RestoreChannels(cntx context.Context) {
- igd.migrateIgmpChannels()
- channels, _ := db.GetIgmpChannels(igd.Mvlan, igd.GroupName, igd.Device)
+ igd.migrateIgmpChannels(cntx)
+ channels, _ := db.GetIgmpChannels(cntx, igd.Mvlan, igd.GroupName, igd.Device)
for _, channel := range channels {
b, ok := channel.Value.([]byte)
@@ -226,19 +227,19 @@
logger.Warn(ctx, "The value type is not []byte")
continue
}
- igd.RestoreChannel(b)
+ igd.RestoreChannel(cntx, b)
}
}
// WriteToDb is utility to write IGMP Group Device Info to the database
-func (igd *IgmpGroupDevice) WriteToDb() error {
+func (igd *IgmpGroupDevice) WriteToDb(cntx context.Context) error {
b, err := json.Marshal(igd)
if err != nil {
return err
}
- if err1 := db.PutIgmpDevice(igd.Mvlan, igd.GroupName, igd.GroupAddr, igd.Device, string(b)); err1 != nil {
+ if err1 := db.PutIgmpDevice(cntx, igd.Mvlan, igd.GroupName, igd.GroupAddr, igd.Device, string(b)); err1 != nil {
return err1
}
logger.Info(ctx, "IGD Updated")
@@ -332,7 +333,7 @@
// AddReceiver add the receiver to the device and perform other actions such as adding the group
// to the physical device, add members, add flows to point the MC packets to the
// group. Also, send a IGMP report upstream if there is a change in the group
-func (igd *IgmpGroupDevice) AddReceiver(port string, groupAddr net.IP,
+func (igd *IgmpGroupDevice) AddReceiver(cntx context.Context, port string, groupAddr net.IP,
group *layers.IGMPv3GroupRecord, version uint8, cvlan uint16, pbit uint8, ponPortID uint32) {
var igc *IgmpGroupChannel
@@ -347,11 +348,11 @@
}
if !igd.GroupInstalled {
- igd.AddNewReceiver(port, groupAddr, group, cvlan, pbit, ponPortID)
+ igd.AddNewReceiver(cntx, port, groupAddr, group, cvlan, pbit, ponPortID)
return
}
- isNewReceiver := igc.AddReceiver(port, group, cvlan, pbit)
+ isNewReceiver := igc.AddReceiver(cntx, port, group, cvlan, pbit)
if isNewReceiver {
ipsList := []net.IP{}
ipsIntf, _ := igd.PortChannelMap.Load(port)
@@ -368,13 +369,13 @@
igd.ModMcGroup()
}
}
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device, "GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
}
// AddNewReceiver to add new receiver
-func (igd *IgmpGroupDevice) AddNewReceiver(port string, groupAddr net.IP, group *layers.IGMPv3GroupRecord, cvlan uint16, pbit uint8, ponPortID uint32) {
+func (igd *IgmpGroupDevice) AddNewReceiver(cntx context.Context, port string, groupAddr net.IP, group *layers.IGMPv3GroupRecord, cvlan uint16, pbit uint8, ponPortID uint32) {
logger.Debugw(ctx, "Adding New Device Receiver", log.Fields{"Channel": groupAddr, "Port": port, "Device": igd.Device})
igcIntf, _ := igd.GroupChannels.Load(groupAddr.String())
@@ -395,8 +396,8 @@
logger.Debugw(ctx, "Port Channel Updated", log.Fields{"Port": port, "NewChannelList": ipsList, "Addr": groupAddr})
igd.AddMcGroup()
- igc.AddReceiver(port, group, cvlan, pbit)
- if err := igd.WriteToDb(); err != nil {
+ igc.AddReceiver(cntx, port, group, cvlan, pbit)
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device, "GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
}
@@ -415,7 +416,7 @@
// DelReceiver is called when Query expiry happened for a receiver. This removes the receiver from the
// the group
-func (igd *IgmpGroupDevice) DelReceiver(groupAddr net.IP, port string, group *layers.IGMPv3GroupRecord, ponPortID uint32) {
+func (igd *IgmpGroupDevice) DelReceiver(cntx context.Context, groupAddr net.IP, port string, group *layers.IGMPv3GroupRecord, ponPortID uint32) {
logger.Debugw(ctx, "Deleting Receiver for Device", log.Fields{"port": port, "GroupIP": groupAddr.String()})
var igc *IgmpGroupChannel
@@ -437,12 +438,12 @@
return
}
igc = igcIntf.(*IgmpGroupChannel)
- if ok := igc.DelReceiver(port, incl, srcList); !ok {
+ if ok := igc.DelReceiver(cntx, port, incl, srcList); !ok {
return
}
if igc.NumReceivers() == 0 {
- igd.DelIgmpGroupChannel(igc)
+ igd.DelIgmpGroupChannel(cntx, igc)
}
igd.DelPortFromChannel(port, groupAddr)
isGroupModified := igd.RemoveChannelFromChannelsPerPon(port, groupAddr, ponPortID)
@@ -451,14 +452,14 @@
if isGroupModified {
igd.ModMcGroup()
}
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device, "GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
}
// DelChannelReceiver is called when Query expiry happened for a receiver. This removes the receiver from the
// the group
-func (igd *IgmpGroupDevice) DelChannelReceiver(groupAddr net.IP) map[string]*IgmpGroupPort {
+func (igd *IgmpGroupDevice) DelChannelReceiver(cntx context.Context, groupAddr net.IP) map[string]*IgmpGroupPort {
portsRemoved := make(map[string]*IgmpGroupPort)
groupModified := false
@@ -471,7 +472,7 @@
igc := igcIntf.(*IgmpGroupChannel)
for port, igp := range igc.NewReceivers {
- _ = db.DelIgmpRcvr(igc.Mvlan, igc.GroupAddr, igc.Device, port) //TODO: Y not here
+ _ = db.DelIgmpRcvr(cntx, igc.Mvlan, igc.GroupAddr, igc.Device, port) //TODO: Y not here
igd.DelPortFromChannel(port, igc.GroupAddr)
ponPortID := GetApplication().GetPonPortID(igd.Device, port)
groupModified = igd.RemoveChannelFromChannelsPerPon(port, igc.GroupAddr, ponPortID)
@@ -479,7 +480,7 @@
portsRemoved[port] = igp
}
for port, igp := range igc.CurReceivers {
- _ = db.DelIgmpRcvr(igc.Mvlan, igc.GroupAddr, igc.Device, port)
+ _ = db.DelIgmpRcvr(cntx, igc.Mvlan, igc.GroupAddr, igc.Device, port)
igd.DelPortFromChannel(port, igc.GroupAddr)
ponPortID := GetApplication().GetPonPortID(igd.Device, port)
groupModified = igd.RemoveChannelFromChannelsPerPon(port, igc.GroupAddr, ponPortID)
@@ -487,15 +488,15 @@
portsRemoved[port] = igp
}
- igc.DelMcFlow()
- igd.DelIgmpGroupChannel(igc)
+ igc.DelMcFlow(cntx)
+ igd.DelIgmpGroupChannel(cntx, igc)
igc.Exclude = 0
igc.SendLeaveToServer()
if groupModified {
igd.ModMcGroup()
}
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device, "GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
logger.Debugw(ctx, "Deleted the receiver Flow", log.Fields{"Num Receivers": igc.NumReceivers()})
@@ -503,12 +504,12 @@
}
// DelIgmpGroupChannel to delete igmp group channel
-func (igd *IgmpGroupDevice) DelIgmpGroupChannel(igc *IgmpGroupChannel) {
+func (igd *IgmpGroupDevice) DelIgmpGroupChannel(cntx context.Context, igc *IgmpGroupChannel) {
if igc.NumReceivers() != 0 {
- igc.DelAllReceivers()
+ igc.DelAllReceivers(cntx)
}
- _ = db.DelIgmpChannel(igc.Mvlan, igc.GroupName, igc.Device, igc.GroupAddr)
+ _ = db.DelIgmpChannel(cntx, igc.Mvlan, igc.GroupName, igc.Device, igc.GroupAddr)
igd.GroupChannels.Delete(igc.GroupAddr.String())
logger.Infow(ctx, "Deleted the Channel from Device", log.Fields{"Channel": igc.GroupAddr.String()})
isLenZero := true
@@ -554,24 +555,24 @@
}
// DelAllChannels deletes all receiver for the provided igmp device
-func (igd *IgmpGroupDevice) DelAllChannels() {
+func (igd *IgmpGroupDevice) DelAllChannels(cntx context.Context) {
logger.Infow(ctx, "Deleting All Channel for Device", log.Fields{"Device": igd.Device, "Group": igd.GroupName})
delGroupChannels := func(key interface{}, value interface{}) bool {
igc := value.(*IgmpGroupChannel)
- igd.DelIgmpGroupChannel(igc)
+ igd.DelIgmpGroupChannel(cntx, igc)
return true
}
igd.GroupChannels.Range(delGroupChannels)
}
// ProcessQuery process query received from the upstream IGMP server
-func (igd *IgmpGroupDevice) ProcessQuery(groupAddr net.IP, ver uint8) {
+func (igd *IgmpGroupDevice) ProcessQuery(cntx context.Context, groupAddr net.IP, ver uint8) {
logger.Debugw(ctx, "Received Query From Server", log.Fields{"Version": ver})
if ver != *igd.ServVersion {
igd.ServVersionExpiry = time.Now().Add(time.Duration(2*igd.proxyCfg.KeepAliveInterval) * time.Second)
*igd.ServVersion = ver
mvp := GetApplication().GetMvlanProfileByTag(igd.Mvlan)
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
}
@@ -676,7 +677,7 @@
// QueryExpiry processes query expiry. Upon expiry, take stock of the situation
// add either retain/release the group based on number of receivers left
-func (igd *IgmpGroupDevice) QueryExpiry() {
+func (igd *IgmpGroupDevice) QueryExpiry(cntx context.Context) {
logger.Debugw(ctx, "Query Expiry", log.Fields{"Device": igd.Device})
@@ -695,7 +696,7 @@
logger.Debugw(ctx, "Expired Member Port State", log.Fields{"state": state})
ponPortID := GetApplication().GetPonPortID(igd.Device, portKey)
if err == nil && state == cntlr.PortStateDown {
- igd.DelReceiver(igc.GroupAddr, portKey, nil, ponPortID)
+ igd.DelReceiver(cntx, igc.GroupAddr, portKey, nil, ponPortID)
}
port.QueryTimeoutCount++
@@ -703,10 +704,10 @@
if port.QueryTimeoutCount >= (*igc.proxyCfg).KeepAliveCount {
logger.Errorw(ctx, "Expiry Timeout count exceeded. Trigger delete receiver", log.Fields{"PortKey": portKey,
"GroupAddr": igc.GroupAddr, "Count": port.QueryTimeoutCount})
- igd.DelReceiver(igc.GroupAddr, portKey, nil, ponPortID)
+ igd.DelReceiver(cntx, igc.GroupAddr, portKey, nil, ponPortID)
SendQueryExpiredEventGroupSpecific(portKey, igd, igc)
} else {
- _ = port.WriteToDb(igc.Mvlan, igc.GroupAddr, igc.Device)
+ _ = port.WriteToDb(cntx, igc.Mvlan, igc.GroupAddr, igc.Device)
}
}
return true
diff --git a/internal/pkg/application/igmpport.go b/internal/pkg/application/igmpport.go
index 5fe88e0..b22b901 100644
--- a/internal/pkg/application/igmpport.go
+++ b/internal/pkg/application/igmpport.go
@@ -16,6 +16,7 @@
package application
import (
+ "context"
"encoding/json"
"net"
@@ -100,12 +101,12 @@
}
// WriteToDb is utility to write IGMP Group Port Info to database
-func (igp *IgmpGroupPort) WriteToDb(mvlan of.VlanType, gip net.IP, device string) error {
+func (igp *IgmpGroupPort) WriteToDb(cntx context.Context, mvlan of.VlanType, gip net.IP, device string) error {
b, err := json.Marshal(igp)
if err != nil {
return err
}
- if err1 := db.PutIgmpRcvr(mvlan, gip, device, igp.Port, string(b)); err1 != nil {
+ if err1 := db.PutIgmpRcvr(cntx, mvlan, gip, device, igp.Port, string(b)); err1 != nil {
return err1
}
return nil
diff --git a/internal/pkg/application/igmpprofiles.go b/internal/pkg/application/igmpprofiles.go
index b5b24d4..192c8e2 100644
--- a/internal/pkg/application/igmpprofiles.go
+++ b/internal/pkg/application/igmpprofiles.go
@@ -16,6 +16,7 @@
package application
import (
+ "context"
"encoding/json"
"errors"
"net"
@@ -157,7 +158,7 @@
}
// WriteToDb is utility to write Mvlan Profile Info to database
-func (mvp *MvlanProfile) WriteToDb() error {
+func (mvp *MvlanProfile) WriteToDb(cntx context.Context) error {
if mvp.DeleteInProgress {
logger.Warnw(ctx, "Skipping Redis Update for MvlanProfile, MvlanProfile delete in progress", log.Fields{"Mvlan": mvp.Mvlan})
@@ -169,7 +170,7 @@
if err != nil {
return err
}
- if err1 := db.PutMvlan(uint16(mvp.Mvlan), string(b)); err1 != nil {
+ if err1 := db.PutMvlan(cntx, uint16(mvp.Mvlan), string(b)); err1 != nil {
return err1
}
return nil
@@ -288,12 +289,12 @@
}
// DelFromDb to delere mvlan from database
-func (mvp *MvlanProfile) DelFromDb() {
- _ = db.DelMvlan(uint16(mvp.Mvlan))
+func (mvp *MvlanProfile) DelFromDb(cntx context.Context) {
+ _ = db.DelMvlan(cntx, uint16(mvp.Mvlan))
}
//DelFlows - Triggers flow deletion after registering for flow indication event
-func (mvp *MvlanProfile) DelFlows(device *VoltDevice, flow *of.VoltFlow) error {
+func (mvp *MvlanProfile) DelFlows(cntx context.Context, device *VoltDevice, flow *of.VoltFlow) error {
mvp.mvpFlowLock.Lock()
defer mvp.mvpFlowLock.Unlock()
@@ -316,14 +317,14 @@
flowMap[cookie] = true
mvp.PendingDeleteFlow[device.Name] = flowMap
}
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
- return cntlr.GetController().DelFlows(device.NniPort, device.Name, flow)
+ return cntlr.GetController().DelFlows(cntx, device.NniPort, device.Name, flow)
}
//FlowRemoveSuccess - Process flow success indication
-func (mvp *MvlanProfile) FlowRemoveSuccess(cookie string, device string) {
+func (mvp *MvlanProfile) FlowRemoveSuccess(cntx context.Context, cookie string, device string) {
mvp.mvpFlowLock.Lock()
defer mvp.mvpFlowLock.Unlock()
@@ -333,7 +334,7 @@
delete(mvp.PendingDeleteFlow[device], cookie)
}
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
}
@@ -394,7 +395,7 @@
}
//pushIgmpMcastFlows - Adds all IGMP related flows (generic DS flow & static group flows)
-func (mvp *MvlanProfile) pushIgmpMcastFlows(OLTSerialNum string) {
+func (mvp *MvlanProfile) pushIgmpMcastFlows(cntx context.Context, OLTSerialNum string) {
mvp.mvpLock.RLock()
defer mvp.mvpLock.RUnlock()
@@ -416,7 +417,7 @@
logger.Infow(ctx, "NNI Port Status is: UP & Vlan Enabled", log.Fields{"Device": d, "port": p})
//Push Igmp DS Control Flows
- err := mvp.ApplyIgmpDSFlowForMvp(d.Name)
+ err := mvp.ApplyIgmpDSFlowForMvp(cntx, d.Name)
if err != nil {
logger.Errorw(ctx, "DS IGMP Flow Add Failed for device",
log.Fields{"Reason": err.Error(), "device": d.Name})
@@ -424,14 +425,14 @@
//Trigger Join for static channels
if channelList, containsStatic := mvp.getAllStaticChannels(); containsStatic {
- mvp.ProcessStaticGroup(d.Name, channelList, true)
+ mvp.ProcessStaticGroup(cntx, d.Name, channelList, true)
} else {
logger.Infow(ctx, "No Static Channels Present", log.Fields{"mvp": mvp.Name, "Mvlan": mvp.Mvlan})
}
}
}
//removeIgmpMcastFlows - Removes all IGMP related flows (generic DS flow & static group flows)
-func (mvp *MvlanProfile) removeIgmpMcastFlows(oltSerialNum string) {
+func (mvp *MvlanProfile) removeIgmpMcastFlows(cntx context.Context, oltSerialNum string) {
mvp.mvpLock.RLock()
defer mvp.mvpLock.RUnlock()
@@ -446,7 +447,7 @@
//Trigger Leave for static channels
if channelList, containsStatic := mvp.getAllStaticChannels(); containsStatic {
- mvp.ProcessStaticGroup(d.Name, channelList, false)
+ mvp.ProcessStaticGroup(cntx, d.Name, channelList, false)
} else {
logger.Infow(ctx, "No Static Channels Present", log.Fields{"mvp": mvp.Name, "Mvlan": mvp.Mvlan})
}
@@ -456,16 +457,16 @@
ig := value.(*IgmpGroup)
if ig.Mvlan == mvp.Mvlan {
igd := ig.Devices[d.Name]
- ig.DelIgmpGroupDevice(igd)
+ ig.DelIgmpGroupDevice(cntx, igd)
if ig.NumDevicesActive() == 0 {
- GetApplication().DelIgmpGroup(ig)
+ GetApplication().DelIgmpGroup(cntx, ig)
}
}
return true
})
//Remove DS Igmp trap flow
- err := mvp.RemoveIgmpDSFlowForMvp(d.Name)
+ err := mvp.RemoveIgmpDSFlowForMvp(cntx, d.Name)
if err != nil {
logger.Errorw(ctx, "DS IGMP Flow Del Failed", log.Fields{"Reason": err.Error(), "device": d.Name})
}
@@ -474,7 +475,7 @@
}
// ApplyIgmpDSFlowForMvp to apply Igmp DS flow for mvlan.
-func (mvp *MvlanProfile) ApplyIgmpDSFlowForMvp(device string) error {
+func (mvp *MvlanProfile) ApplyIgmpDSFlowForMvp(cntx context.Context, device string) error {
va := GetApplication()
dIntf, ok := va.DevicesDisc.Load(device)
if !ok {
@@ -487,7 +488,7 @@
if !ok || !flowAlreadyApplied {
flows, err := mvp.BuildIgmpDSFlows(device)
if err == nil {
- err = cntlr.GetController().AddFlows(d.NniPort, device, flows)
+ err = cntlr.GetController().AddFlows(cntx, d.NniPort, device, flows)
if err != nil {
logger.Warnw(ctx, "Configuring IGMP Flow for device failed ", log.Fields{"Device": device, "err": err})
return err
@@ -504,7 +505,7 @@
}
// RemoveIgmpDSFlowForMvp to remove Igmp DS flow for mvlan.
-func (mvp *MvlanProfile) RemoveIgmpDSFlowForMvp(device string) error {
+func (mvp *MvlanProfile) RemoveIgmpDSFlowForMvp(cntx context.Context, device string) error {
va := GetApplication()
mvlan := mvp.Mvlan
@@ -522,7 +523,7 @@
if err == nil {
flows.ForceAction = true
- err = mvp.DelFlows(d, flows)
+ err = mvp.DelFlows(cntx, d, flows)
if err != nil {
logger.Warnw(ctx, "De-Configuring IGMP Flow for device failed ", log.Fields{"Device": device, "err": err})
return err
@@ -572,7 +573,7 @@
}
//updateStaticGroups - Generates static joins & leaves for newly added and removed static channels respectively
-func (mvp *MvlanProfile) updateStaticGroups(deviceID string, added []net.IP, removed []net.IP) {
+func (mvp *MvlanProfile) updateStaticGroups(cntx context.Context, deviceID string, added []net.IP, removed []net.IP) {
//Update static group configs for all associated devices
updateGroups := func(key interface{}, value interface{}) bool {
@@ -583,8 +584,8 @@
return true
}
//TODO if mvp.IsChannelBasedGroup {
- mvp.ProcessStaticGroup(d.Name, added, true)
- mvp.ProcessStaticGroup(d.Name, removed, false)
+ mvp.ProcessStaticGroup(cntx, d.Name, added, true)
+ mvp.ProcessStaticGroup(cntx, d.Name, removed, false)
//}
return true
}
@@ -598,7 +599,7 @@
}
//updateDynamicGroups - Generates joins with updated sources for existing channels
-func (mvp *MvlanProfile) updateDynamicGroups(deviceID string, added []net.IP, removed []net.IP) {
+func (mvp *MvlanProfile) updateDynamicGroups(cntx context.Context, deviceID string, added []net.IP, removed []net.IP) {
//mvlan := mvp.Mvlan
va := GetApplication()
@@ -617,7 +618,7 @@
logger.Debugw(ctx, "IGMP Group", log.Fields{"Group": grpKey, "groupAddr": groupAddr})
if igIntf, ok := va.IgmpGroups.Load(grpKey); ok {
ig := igIntf.(*IgmpGroup)
- if igd, ok := ig.getIgmpGroupDevice(d.Name); ok {
+ if igd, ok := ig.getIgmpGroupDevice(cntx, d.Name); ok {
if igcIntf, ok := igd.GroupChannels.Load(groupAddr.String()); ok {
igc := igcIntf.(*IgmpGroupChannel)
incl := false
@@ -631,12 +632,12 @@
}
for port, igp := range igc.NewReceivers {
// Process the include/exclude list which may end up modifying the group
- if change, _ := igc.ProcessSources(port, ip, incl); change {
+ if change, _ := igc.ProcessSources(cntx, port, ip, incl); change {
groupModified = true
}
igc.ProcessMode(port, incl)
- if err := igp.WriteToDb(igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
+ if err := igp.WriteToDb(cntx, igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
logger.Errorw(ctx, "Igmp group port Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
}
@@ -646,7 +647,7 @@
logger.Debug(ctx, "Group Modified and IGMP report sent to the upstream server")
igc.SendReport(false)
}
- if err := igc.WriteToDb(); err != nil {
+ if err := igc.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group channel Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
}
@@ -667,7 +668,7 @@
//GroupsUpdated - Handles removing of Igmp Groups, flows & group table entries for
//channels removed as part of update
-func (mvp *MvlanProfile) GroupsUpdated(deviceID string) {
+func (mvp *MvlanProfile) GroupsUpdated(cntx context.Context, deviceID string) {
deleteChannelIfRemoved := func(key interface{}, value interface{}) bool {
ig := value.(*IgmpGroup)
@@ -703,32 +704,32 @@
// If channel is not Static but existing Group is static - Migrate (from static to dynamic)
// (Channel removed from satic but part of dynamic)
if (staticChannel != mvp.IsStaticGroup(ig.GroupName)) || (ig.IsGroupStatic != mvp.IsStaticGroup(ig.GroupName)) { // Equivalent of XOR
- ig.HandleGroupMigration(deviceID, channelIP)
+ ig.HandleGroupMigration(cntx, deviceID, channelIP)
} else {
if (ig.IsGroupStatic) && mvp.IsStaticGroup(ig.GroupName) {
if ig.GroupName != mvp.GetStaticGroupName(channelIP) {
- ig.HandleGroupMigration(deviceID, channelIP)
+ ig.HandleGroupMigration(cntx, deviceID, channelIP)
}
}
continue
}
} else {
logger.Debugw(ctx, "Channel Removed", log.Fields{"Channel": channel, "Group": grpName})
- ig.DelIgmpChannel(deviceID, net.ParseIP(channel))
+ ig.DelIgmpChannel(cntx, deviceID, net.ParseIP(channel))
if ig.NumDevicesActive() == 0 {
- GetApplication().DelIgmpGroup(ig)
+ GetApplication().DelIgmpGroup(cntx, ig)
}
}
}
ig.IsGroupStatic = mvp.IsStaticGroup(ig.GroupName)
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
return true
}
}
logger.Debugw(ctx, "Group Removed", log.Fields{"Channel": ig.GroupAddr, "Group": grpName, "ChannelBasedGroup": ig.IsChannelBasedGroup})
- ig.DelIgmpGroup()
+ ig.DelIgmpGroup(cntx)
logger.Debugw(ctx, "Removed Igmp Group", log.Fields{"Channel": ig.GroupAddr, "Group": grpName})
return true
}
@@ -771,7 +772,7 @@
}
// ProcessStaticGroup - Process Static Join/Leave Req for static channels
-func (mvp *MvlanProfile) ProcessStaticGroup(device string, groupAddresses []net.IP, isJoin bool) {
+func (mvp *MvlanProfile) ProcessStaticGroup(cntx context.Context, device string, groupAddresses []net.IP, isJoin bool) {
logger.Debugw(ctx, "Received Static Group Request", log.Fields{"Device": device, "Join": isJoin, "Group Address List": groupAddresses})
@@ -790,9 +791,9 @@
if ig == nil {
// First time group Creation: Create the IGMP group and then add the receiver to the group
logger.Infow(ctx, "Static IGMP Add received for new group", log.Fields{"Addr": groupAddr, "Port": StaticPort})
- if ig := GetApplication().AddIgmpGroup(mvp.Name, groupAddr, device); ig != nil {
+ if ig := GetApplication().AddIgmpGroup(cntx, mvp.Name, groupAddr, device); ig != nil {
ig.IgmpGroupLock.Lock()
- ig.AddReceiver(device, StaticPort, groupAddr, nil, getVersion(ver),
+ ig.AddReceiver(cntx, device, StaticPort, groupAddr, nil, getVersion(ver),
0, 0, 0xFF)
ig.IgmpGroupLock.Unlock()
} else {
@@ -801,12 +802,12 @@
} else {
//Converting existing dynamic group to static group
if !mvp.IsStaticGroup(ig.GroupName) {
- ig.updateGroupName(ig.GroupName)
+ ig.updateGroupName(cntx, ig.GroupName)
}
// Update case: If the IGMP group is already created. just add the receiver
logger.Infow(ctx, "Static IGMP Add received for existing group", log.Fields{"Addr": groupAddr, "Port": StaticPort})
ig.IgmpGroupLock.Lock()
- ig.AddReceiver(device, StaticPort, groupAddr, nil, getVersion(ver),
+ ig.AddReceiver(cntx, device, StaticPort, groupAddr, nil, getVersion(ver),
0, 0, 0xFF)
ig.IgmpGroupLock.Unlock()
}
@@ -817,19 +818,19 @@
grpName := mvp.GetMvlanGroup(ig.GroupAddr)
if grpName != "" {
ig.IgmpGroupLock.Lock()
- ig.DelReceiver(device, StaticPort, groupAddr, nil, 0xFF)
+ ig.DelReceiver(cntx, device, StaticPort, groupAddr, nil, 0xFF)
ig.IgmpGroupLock.Unlock()
- ig.updateGroupName(grpName)
+ ig.updateGroupName(cntx, grpName)
} else {
- ig.DelIgmpGroup()
+ ig.DelIgmpGroup(cntx)
}
} else {
ig.IgmpGroupLock.Lock()
- ig.DelReceiver(device, StaticPort, groupAddr, nil, 0xFF)
+ ig.DelReceiver(cntx, device, StaticPort, groupAddr, nil, 0xFF)
ig.IgmpGroupLock.Unlock()
}
if ig.NumDevicesActive() == 0 {
- GetApplication().DelIgmpGroup(ig)
+ GetApplication().DelIgmpGroup(cntx, ig)
}
} else {
logger.Warnw(ctx, "Static IGMP Del received for unknown group", log.Fields{"Addr": groupAddr})
@@ -894,7 +895,7 @@
}
// UpdateProfile - Updates the group & member info w.r.t the mvlan profile for the given device
-func (mvp *MvlanProfile) UpdateProfile(deviceID string) {
+func (mvp *MvlanProfile) UpdateProfile(cntx context.Context, deviceID string) {
logger.Infow(ctx, "Update Mvlan Profile task triggered", log.Fields{"Mvlan": mvp.Mvlan})
var removedStaticChannels []net.IP
addedStaticChannels := []net.IP{}
@@ -922,12 +923,12 @@
logger.Debugw(ctx, "Update Task - Static Group Changes", log.Fields{"Added": addedStaticChannels, "Removed": removedStaticChannels})
if len(addedStaticChannels) > 0 || len(removedStaticChannels) > 0 {
- mvp.updateStaticGroups(deviceID, []net.IP{}, removedStaticChannels)
+ mvp.updateStaticGroups(cntx, deviceID, []net.IP{}, removedStaticChannels)
}
}
- mvp.GroupsUpdated(deviceID)
+ mvp.GroupsUpdated(cntx, deviceID)
if len(addedStaticChannels) > 0 {
- mvp.updateStaticGroups(deviceID, addedStaticChannels, []net.IP{})
+ mvp.updateStaticGroups(cntx, deviceID, addedStaticChannels, []net.IP{})
}
/* Need to handle if SSM params are modified for groups */
@@ -936,10 +937,10 @@
if mvp.checkStaticGrpSSMProxyDiff(mvp.oldProxy[key], mvp.Proxy[key]) {
if mvp.Groups[key].IsStatic {
/* Static group proxy modified, need to trigger membership report with new mode/src-list for existing channels */
- mvp.updateStaticGroups(deviceID, commonChannels, []net.IP{})
+ mvp.updateStaticGroups(cntx, deviceID, commonChannels, []net.IP{})
} else {
/* Dynamic group proxy modified, need to trigger membership report with new mode/src-list for existing channels */
- mvp.updateDynamicGroups(deviceID, commonChannels, []net.IP{})
+ mvp.updateDynamicGroups(cntx, deviceID, commonChannels, []net.IP{})
}
}
}
@@ -949,7 +950,7 @@
if deviceID == "" || !mvp.isUpdateInProgress() {
mvp.oldGroups = nil
}
- if err := mvp.WriteToDb(); err != nil {
+ if err := mvp.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile write to DB failed", log.Fields{"ProfileName": mvp.Name})
}
logger.Debugw(ctx, "Updated MVLAN Profile", log.Fields{"VLAN": mvp.Mvlan, "Name": mvp.Name, "Grp IPs": mvp.Groups})
@@ -1055,7 +1056,7 @@
}
//TriggerAssociatedFlowDelete - Re-trigger delete for pending delete flows
-func (mvp *MvlanProfile) TriggerAssociatedFlowDelete(device string) bool {
+func (mvp *MvlanProfile) TriggerAssociatedFlowDelete(cntx context.Context, device string) bool {
mvp.mvpFlowLock.Lock()
cookieList := []uint64{}
@@ -1078,7 +1079,7 @@
subFlow.Cookie = cookie
flow.SubFlows[cookie] = subFlow
logger.Infow(ctx, "Retriggering Vnet Delete Flow", log.Fields{"Device": device, "Mvlan": mvp.Mvlan.String(), "Cookie": cookie})
- err := mvp.DelFlows(vd, flow)
+ err := mvp.DelFlows(cntx, vd, flow)
if err != nil {
logger.Warnw(ctx, "De-Configuring IGMP Flow for device failed ", log.Fields{"Device": device, "err": err})
}
@@ -1216,13 +1217,13 @@
}
// WriteToDb is utility to write Igmp Config Info to database
-func (igmpProfile *IgmpProfile) WriteToDb() error {
+func (igmpProfile *IgmpProfile) WriteToDb(cntx context.Context) error {
igmpProfile.Version = database.PresentVersionMap[database.IgmpProfPath]
b, err := json.Marshal(igmpProfile)
if err != nil {
return err
}
- if err1 := db.PutIgmpProfile(igmpProfile.ProfileID, string(b)); err1 != nil {
+ if err1 := db.PutIgmpProfile(cntx, igmpProfile.ProfileID, string(b)); err1 != nil {
return err1
}
return nil
diff --git a/internal/pkg/application/igmptasks.go b/internal/pkg/application/igmptasks.go
index af42562..268453c 100644
--- a/internal/pkg/application/igmptasks.go
+++ b/internal/pkg/application/igmptasks.go
@@ -72,7 +72,7 @@
func (tt *TickTask) Start(ctx context.Context, taskID uint8) error {
tt.taskID = taskID
tt.ctx = ctx
- GetApplication().IgmpTick()
+ GetApplication().IgmpTick(ctx)
return nil
}
@@ -124,7 +124,7 @@
func (pt *IgmpPacketTask) Start(ctx context.Context, taskID uint8) error {
pt.taskID = taskID
pt.ctx = ctx
- GetApplication().IgmpProcessPkt(pt.Device, pt.Port, pt.Pkt)
+ GetApplication().IgmpProcessPkt(ctx, pt.Device, pt.Port, pt.Pkt)
return nil
}
@@ -170,6 +170,6 @@
mt.taskID = taskID
mt.ctx = ctx
mvp := mt.mvp
- mvp.UpdateProfile(mt.DeviceID)
+ mvp.UpdateProfile(ctx, mt.DeviceID)
return nil
}
diff --git a/internal/pkg/application/major_upgrade.go b/internal/pkg/application/major_upgrade.go
index 639822b..b423b58 100644
--- a/internal/pkg/application/major_upgrade.go
+++ b/internal/pkg/application/major_upgrade.go
@@ -48,7 +48,7 @@
ModuleVer map[string]string // eg. "service": "v1"
}
-type paramsMigrationFunc func([]byte) string
+type paramsMigrationFunc func(context.Context, []byte) string
//map to store conversion functions
var migrationMap = map[string]paramsMigrationFunc{
@@ -79,32 +79,32 @@
}
// WriteToDb write a meter profile to DB
-func (md *DataMigration) WriteToDb() error {
+func (md *DataMigration) WriteToDb(cntx context.Context) error {
b, err := json.Marshal(md)
if err != nil {
return err
}
- if err1 := db.PutMigrationInfo(string(b)); err1 != nil {
+ if err1 := db.PutMigrationInfo(cntx, string(b)); err1 != nil {
return err1
}
return nil
}
// DelFromDb delete a meter profile from DB
-func (md *DataMigration) DelFromDb() {
- if err := db.DelMigrationInfo(); err != nil {
+func (md *DataMigration) DelFromDb(cntx context.Context) {
+ if err := db.DelMigrationInfo(cntx); err != nil {
logger.Warnw(ctx, "DelMigrationInfo Failed", log.Fields{"Error": err})
}
}
// GetMigrationInfo to get data migration info
-func GetMigrationInfo(dmInfo *DataMigration) error {
+func GetMigrationInfo(cntx context.Context, dmInfo *DataMigration) error {
var migrationInfo string
var err error
if db == nil {
db = database.GetDatabase()
}
- if migrationInfo, err = db.GetMigrationInfo(); err != nil {
+ if migrationInfo, err = db.GetMigrationInfo(cntx); err != nil {
return err
}
err = json.Unmarshal([]byte(migrationInfo), &dmInfo)
@@ -122,7 +122,7 @@
func CheckIfMigrationRequired(ctx context.Context) bool {
Migrate := new(DataMigration)
var NoDataInDB bool
- err := GetMigrationInfo(Migrate)
+ err := GetMigrationInfo(ctx, Migrate)
logger.Debugw(ctx, "Migration data", log.Fields{"DataMigration": Migrate})
// No DB entry represents N verison Bring Up for the First time
if err != nil {
@@ -135,7 +135,7 @@
Migrate.Version = database.PresentVersion
Migrate.Status = MigrationComplete
Migrate.ModuleVer = database.PresentVersionMap
- if err := Migrate.WriteToDb(); err != nil {
+ if err := Migrate.WriteToDb(ctx); err != nil {
logger.Errorw(ctx, "DB Write failed for Migration Path", log.Fields{"error": err})
}
//MigrateProbestatus has to be Updated to Complete when No Migration is Required
@@ -191,11 +191,11 @@
go func() {
logger.Debug(ctx, "Started Go Routine for data migration")
- err = MigrateDBData()
+ err = MigrateDBData(ctx)
if err != nil {
logger.Errorw(ctx, "Failed to Migrate the Data", log.Fields{"error": err})
Migrate.Status = MigrationFailed
- if err := Migrate.WriteToDb(); err != nil {
+ if err := Migrate.WriteToDb(ctx); err != nil {
logger.Errorw(ctx, "DB Write failed to Migration Path", log.Fields{"error": err})
}
}
@@ -205,7 +205,7 @@
Migrate.Version = database.PresentVersion
Migrate.Status = MigrationInProgress
Migrate.ModuleVer = database.PresentVersionMap
- if err = Migrate.WriteToDb(); err != nil {
+ if err = Migrate.WriteToDb(ctx); err != nil {
logger.Errorw(ctx, "DB Write failed for Migration Path", log.Fields{"error": err})
return
}
@@ -215,7 +215,7 @@
if err := recover(); err != nil {
logger.Errorw(ctx, "Migration failure due to Exception happend", log.Fields{"reason": err})
Migrate.Status = MigrationFailed
- if err := Migrate.WriteToDb(); err != nil {
+ if err := Migrate.WriteToDb(ctx); err != nil {
logger.Errorw(ctx, "DB Write failed for Migration Path", log.Fields{"error": err})
}
//probe.UpdateDBMigrationStatus(ctx, false)
@@ -226,14 +226,14 @@
migrationWG.Wait()
//probe.UpdateDBMigrationStatus(ctx, true)
Migrate.Status = MigrationComplete
- if err := Migrate.WriteToDb(); err != nil {
+ if err := Migrate.WriteToDb(ctx); err != nil {
logger.Errorw(ctx, "DB Write failed for Migration Path", log.Fields{"error": err})
}
logger.Infow(ctx, "Migration completed successfully", log.Fields{"Status": Migrate.Status})
}
// MigrateDBData to migrate database data
-func MigrateDBData() error {
+func MigrateDBData(cntx context.Context) error {
var err error
for module, currentVersion := range database.PresentVersionMap {
@@ -251,7 +251,7 @@
database.DeviceFlowHashPath:
err = FetchAndMigrateDeviceDBData(module)
default:
- err = FetchAndMigrateDBData(module)
+ err = FetchAndMigrateDBData(cntx, module)
}
} else {
logger.Infow(ctx, "No Data Migration handling found for module", log.Fields{"Table": module, "Version": currentVersion})
@@ -272,10 +272,10 @@
}
//FetchAndMigrateDBData fetchs the data from database and migrte the same to latest versions and store ot back ot database
-func FetchAndMigrateDBData(module string) error {
+func FetchAndMigrateDBData(cntx context.Context, module string) error {
previousPath := database.GetModuleKeypath(module, database.PreviousVersionMap[module])
- dbPathKeysValueMap, err := db.List(previousPath)
+ dbPathKeysValueMap, err := db.List(cntx, previousPath)
if err != nil {
logger.Errorw(ctx, "failed to Fetch the Keys from Redis", log.Fields{"error": err})
//No return required, Data might not be present in DB
@@ -296,7 +296,7 @@
return errors.New("Error-in-migration")
}
- presentParams := migrationMap[module](b)
+ presentParams := migrationMap[module](cntx, b)
logger.Infow(ctx, "Migrated data", log.Fields{"presentParams": presentParams})
if "" == presentParams {
logger.Error(ctx, "Error in migrating data\n")
@@ -306,7 +306,7 @@
}
presentPath := database.GetKeyPath(module) + hash
logger.Infow(ctx, "Before writing to DB", log.Fields{"presentParams": presentParams})
- if err := db.Put(presentPath, presentParams); err != nil {
+ if err := db.Put(cntx, presentPath, presentParams); err != nil {
logger.Errorw(ctx, "Update Params failed", log.Fields{"key": presentPath, "presentparams": presentParams})
return err
}
@@ -315,7 +315,7 @@
}
//MigrateServices modifyies the old data as per current version requirement and updates the database
-func MigrateServices(data []byte) string {
+func MigrateServices(cntx context.Context, data []byte) string {
var vs VoltService
var updatedData, updatedData1 []byte
var vsmap map[string]interface{}
@@ -356,43 +356,43 @@
}
//MigrateDevices modifyies the old data as per current version requirement and updates the database
-func MigrateDevices(data []byte) string {
+func MigrateDevices(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Devices")
return ""
}
//MigrateDevicePorts modifyies the old data as per current version requirement and updates the database
-func MigrateDevicePorts(data []byte) string {
+func MigrateDevicePorts(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Ports")
return ""
}
//MigrateDeviceFlows modifyies the old data as per current version requirement and updates the database
-func MigrateDeviceFlows(data []byte) string {
+func MigrateDeviceFlows(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Flows")
return ""
}
//MigrateDeviceGroups modifyies the old data as per current version requirement and updates the database
-func MigrateDeviceGroups(data []byte) string {
+func MigrateDeviceGroups(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Groups")
return ""
}
//MigrateDeviceMeters modifyies the old data as per current version requirement and updates the database
-func MigrateDeviceMeters(data []byte) string {
+func MigrateDeviceMeters(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Meters")
return ""
}
//MigrateDeviceFlowHash modifyies the old data as per current version requirement and updates the database
-func MigrateDeviceFlowHash(data []byte) string {
+func MigrateDeviceFlowHash(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for FlowHash")
return ""
}
//MigrateVnets modifyies the old data as per current version requirement and updates the database
-func MigrateVnets(data []byte) string {
+func MigrateVnets(cntx context.Context, data []byte) string {
var vnet VoltVnet
var updatedData []byte
@@ -423,7 +423,7 @@
}
//MigrateVpvs modifyies the old data as per current version requirement and updates the database
-func MigrateVpvs(data []byte) string {
+func MigrateVpvs(cntx context.Context, data []byte) string {
var vpv VoltPortVnet
var updatedData, updatedData1 []byte
var vpvmap map[string]interface{}
@@ -479,7 +479,7 @@
}
//MigrateMvlans modifyies the old data as per current version requirement and updates the database
-func MigrateMvlans(data []byte) string {
+func MigrateMvlans(cntx context.Context, data []byte) string {
var mvp MvlanProfile
var updatedData []byte
@@ -504,13 +504,13 @@
}
//MigrateMeters modifyies the old data as per current version requirement and updates the database
-func MigrateMeters(data []byte) string {
+func MigrateMeters(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Meters")
return ""
}
//MigrateIgmpConfs modifyies the old data as per current version requirement and updates the database
-func MigrateIgmpConfs(data []byte) string {
+func MigrateIgmpConfs(cntx context.Context, data []byte) string {
var igmpProfile IgmpProfile
err := json.Unmarshal(data, &igmpProfile)
@@ -518,7 +518,7 @@
logger.Warn(ctx, "Unmarshal of IGMP failed")
return ""
}
- if err := igmpProfile.WriteToDb(); err != nil {
+ if err := igmpProfile.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp profile Write to DB failed", log.Fields{"profileID": igmpProfile.ProfileID})
}
@@ -527,73 +527,73 @@
}
//MigrateIgmpGroups modifyies the old data as per current version requirement and updates the database
-func MigrateIgmpGroups(data []byte) string {
+func MigrateIgmpGroups(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for IGMP Groups")
return ""
}
//MigrateIgmpDevices modifyies the old data as per current version requirement and updates the database
-func MigrateIgmpDevices(data []byte) string {
+func MigrateIgmpDevices(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for IGMP Device")
return ""
}
//MigrateIgmpChannels modifyies the old data as per current version requirement and updates the database
-func MigrateIgmpChannels(data []byte) string {
+func MigrateIgmpChannels(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for IGMP Channels")
return ""
}
//MigrateIgmpPorts modifyies the old data as per current version requirement and updates the database
-func MigrateIgmpPorts(data []byte) string {
+func MigrateIgmpPorts(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for IGMP Ports")
return ""
}
//MigrateIgmpProfs modifyies the old data as per current version requirement and updates the database
-func MigrateIgmpProfs(data []byte) string {
+func MigrateIgmpProfs(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for IGMP Profs")
return ""
}
//MigrateMcastConfs modifyies the old data as per current version requirement and updates the database
-func MigrateMcastConfs(data []byte) string {
+func MigrateMcastConfs(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Mcast Confs")
return ""
}
//MigrateLogLevels modifyies the old data as per current version requirement and updates the database
-func MigrateLogLevels(data []byte) string {
+func MigrateLogLevels(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Log Levels")
return ""
}
//MigrateHealth modifyies the old data as per current version requirement and updates the database
-func MigrateHealth(data []byte) string {
+func MigrateHealth(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Health")
return ""
}
//MigratePonCounters modifyies the old data as per current version requirement and updates the database
-func MigratePonCounters(data []byte) string {
+func MigratePonCounters(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Pon Counters")
return ""
}
//MigrateChannelCounters modifyies the old data as per current version requirement and updates the database
-func MigrateChannelCounters(data []byte) string {
+func MigrateChannelCounters(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Channel Counters")
return ""
}
//MigrateServiceCounters modifyies the old data as per current version requirement and updates the database
-func MigrateServiceCounters(data []byte) string {
+func MigrateServiceCounters(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for Service Counters")
return ""
}
//MigrateNbDevices modifyies the old data as per current version requirement and updates the database
-func MigrateNbDevices(data []byte) string {
+func MigrateNbDevices(cntx context.Context, data []byte) string {
logger.Error(ctx, "Data Migration not implemented for NB Devices")
return ""
}
@@ -605,11 +605,11 @@
}
//DeleteDbPathKeys Deleted the paths from DB
-func DeleteDbPathKeys(keyPath string) error {
+func DeleteDbPathKeys(cntx context.Context, keyPath string) error {
logger.Debugw(ctx, "Deleting paths for version", log.Fields{"Path": keyPath})
// Delete all the keys
- err := db.DeleteAll(keyPath)
+ err := db.DeleteAll(cntx, keyPath)
if err != nil && err.Error() != common.ErrEntryNotFound.Error() {
logger.Errorw(ctx, "Delete Key failed", log.Fields{"error": err})
return err
diff --git a/internal/pkg/application/meters.go b/internal/pkg/application/meters.go
index cd3e724..276fe99 100644
--- a/internal/pkg/application/meters.go
+++ b/internal/pkg/application/meters.go
@@ -18,6 +18,7 @@
import (
"encoding/json"
"errors"
+ "context"
"sync"
cntlr "voltha-go-controller/internal/pkg/controller"
@@ -87,21 +88,21 @@
}
// WriteToDb to write a meter profile to DB
-func (vm *VoltMeter) WriteToDb() error {
+func (vm *VoltMeter) WriteToDb(cntx context.Context) error {
vm.Version = database.PresentVersionMap[database.MeterPath]
b, err := json.Marshal(vm)
if err != nil {
return err
}
- if err1 := db.PutMeter(vm.Name, string(b)); err1 != nil {
+ if err1 := db.PutMeter(cntx, vm.Name, string(b)); err1 != nil {
return err1
}
return nil
}
// DelFromDb to delete a meter profile from DB
-func (vm *VoltMeter) DelFromDb() {
- _ = db.DelMeter(vm.Name)
+func (vm *VoltMeter) DelFromDb(cntx context.Context) {
+ _ = db.DelMeter(cntx, vm.Name)
}
// GetMeterByName to get meter by name
@@ -215,10 +216,10 @@
}
// RestoreMetersFromDb to read from the DB and restore all the services
-func (m *MeterMgr) RestoreMetersFromDb() {
+func (m *MeterMgr) RestoreMetersFromDb(cntx context.Context) {
// VNETS must be learnt first
logger.Infow(ctx, "LastMeterID on restart", log.Fields{"LastMeterID": m.LastMeterID})
- ms, _ := db.GetMeters()
+ ms, _ := db.GetMeters(cntx)
for _, mt := range ms {
b, ok := mt.Value.([]byte)
if !ok {
@@ -241,7 +242,7 @@
}
// AddMeterProf to add the meter profile name as key
-func (va *VoltApplication) AddMeterProf(cfg VoltMeter) {
+func (va *VoltApplication) AddMeterProf(cntx context.Context, cfg VoltMeter) {
mm := &va.MeterMgr
if _, ok := mm.GetMeterByName(cfg.Name); ok {
@@ -256,20 +257,20 @@
id := mm.LastMeterID
cfg.ID = id
mm.AddMeter(&cfg)
- if err := cfg.WriteToDb(); err != nil {
+ if err := cfg.WriteToDb(cntx); err != nil {
logger.Warnw(ctx, "MeterProf Write to DB Failed", log.Fields{"MeterConfig": cfg, "Error": err})
}
}
// UpdateMeterProf to update the meter profile
-func (va *VoltApplication) UpdateMeterProf(cfg VoltMeter) {
+func (va *VoltApplication) UpdateMeterProf(cntx context.Context, cfg VoltMeter) {
mm := &va.MeterMgr
if _, ok := mm.GetMeterByName(cfg.Name); !ok {
logger.Warnw(ctx, "Meter profile does not exist", log.Fields{"Name": cfg.Name})
return
}
mm.AddMeter(&cfg)
- if err := cfg.WriteToDb(); err != nil {
+ if err := cfg.WriteToDb(cntx); err != nil {
logger.Warnw(ctx, "MeterProf Write to DB Failed", log.Fields{"MeterConfig": cfg, "Error": err})
}
}
@@ -304,7 +305,7 @@
}
// DelMeterProf to delete meter profile
-func (va *VoltApplication) DelMeterProf(name string) error {
+func (va *VoltApplication) DelMeterProf(cntx context.Context, name string) error {
mm := &va.MeterMgr
if _, ok := mm.GetMeterByName(name); !ok {
logger.Warnw(ctx, "Meter profile does not exist", log.Fields{"Name": name})
@@ -324,7 +325,7 @@
return true
}
va.DevicesDisc.Range(delmeterFromDevice)
- cfg.DelFromDb()
+ cfg.DelFromDb(cntx)
//Delete meter from device will be invoked by caller separately
mm.DelMeter(cfg)
return nil
diff --git a/internal/pkg/application/minor_upgrade.go b/internal/pkg/application/minor_upgrade.go
index d8e7094..a0bba15 100644
--- a/internal/pkg/application/minor_upgrade.go
+++ b/internal/pkg/application/minor_upgrade.go
@@ -17,6 +17,7 @@
import (
"errors"
+ "context"
"net"
"voltha-go-controller/internal/pkg/types"
@@ -27,7 +28,7 @@
"voltha-go-controller/log"
)
-type paramsUpdationFunc func(hash string, value interface{}) error
+type paramsUpdationFunc func(cntx context.Context, hash string, value interface{}) error
//map to store conversion functions
var updationMap = map[string]paramsUpdationFunc{
@@ -41,9 +42,9 @@
}
// UpdateDbData to update database data
-func UpdateDbData(dbPath, hash string, value interface{}) error {
+func UpdateDbData(cntx context.Context, dbPath, hash string, value interface{}) error {
if migrationFunc, ok := updationMap[dbPath]; ok {
- err := migrationFunc(hash, value)
+ err := migrationFunc(cntx, hash, value)
if err != nil {
logger.Error(ctx, "Error in migrating data\n")
return errors.New("Error-in-migration")
@@ -54,7 +55,7 @@
//This function modifyies the old data as per current version requirement and also
//returns the new path on which the modified data has to be written
-func updateServices(hash string, value interface{}) error {
+func updateServices(cntx context.Context, hash string, value interface{}) error {
param := value.(*VoltService)
param.VnetID = VnetKey(param.SVlan, param.CVlan, param.UniVlan)
return nil
@@ -62,12 +63,12 @@
//This function modifyies the old data as per current version requirement and also
//returns the new path on which the modified data has to be written
-func updateVnets(hash string, value interface{}) error {
+func updateVnets(cntx context.Context, hash string, value interface{}) error {
param := value.(*VoltVnet)
newKey := VnetKey(param.SVlan, param.CVlan, param.UniVlan)
if newKey != hash {
//Delete the older key
- _ = db.DelVnet(hash)
+ _ = db.DelVnet(cntx, hash)
} else {
//Update SVlan Tag Protocol id param with default valud if not present
if param.SVlanTpid == 0 {
@@ -83,7 +84,7 @@
//This function modifyies the old data as per current version requirement and also
//returns the new path on which the modified data has to be written
-func updateVpvs(hash string, value interface{}) error {
+func updateVpvs(cntx context.Context, hash string, value interface{}) error {
//var param VoltPortVnet
param := value.(*VoltPortVnet)
@@ -99,20 +100,20 @@
}
//Add the vpv under new path
- param.WriteToDb()
+ param.WriteToDb(cntx)
//delete the older path
fullPath := database.BasePath + database.VpvPath + hash
- if err := db.Del(fullPath); err != nil {
+ if err := db.Del(cntx, fullPath); err != nil {
logger.Errorw(ctx, "Vpv Delete from DB failed", log.Fields{"Error": err, "key": fullPath})
}
return nil
}
-func updateMvlans(hash string, value interface{}) error {
+func updateMvlans(cntx context.Context, hash string, value interface{}) error {
param := value.(*MvlanProfile)
if param.DevicesList == nil || len(param.DevicesList) == 0 {
param.DevicesList = make(map[string]OperInProgress) //Empty OLT serial number as of now since submgr won't have proper serial num
- if err := param.WriteToDb(); err != nil {
+ if err := param.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Mvlan profile write to DB failed", log.Fields{"ProfileName": param.Name})
}
@@ -125,14 +126,14 @@
//This function modifyies the old Igmp Group data as per current version requirement and also
//returns the new path on which the modified data has to be written
-func updateIgmpGroups(hash string, value interface{}) error {
+func updateIgmpGroups(cntx context.Context, hash string, value interface{}) error {
ig := value.(*IgmpGroup)
logger.Infow(ctx, "Group Data Migration", log.Fields{"ig": ig, "GroupAddr": ig.GroupAddr, "hash": hash})
if ig.GroupAddr == nil {
ig.GroupAddr = net.ParseIP("0.0.0.0")
}
- if err := ig.WriteToDb(); err != nil {
+ if err := ig.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group Write to DB failed", log.Fields{"groupName": ig.GroupName})
}
@@ -141,13 +142,13 @@
//This function modifyies the old Igmp Device data as per current version requirement and also
//returns the new path on which the modified data has to be written
-func updateIgmpDevices(hash string, value interface{}) error {
+func updateIgmpDevices(cntx context.Context, hash string, value interface{}) error {
igd := value.(*IgmpGroupDevice)
logger.Infow(ctx, "Group Device Migration", log.Fields{"igd": igd, "GroupAddr": igd.GroupAddr, "hash": hash})
if igd.GroupAddr == nil {
igd.GroupAddr = net.ParseIP("0.0.0.0")
}
- if err := igd.WriteToDb(); err != nil {
+ if err := igd.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group device Write to DB failed", log.Fields{"Device": igd.Device,
"GroupName": igd.GroupName, "GroupAddr": igd.GroupAddr.String()})
}
@@ -157,15 +158,15 @@
//This function modifyies the old Igmp Profile data as per current version requirement and also
//returns the new path on which the modified data has to be written
-func updateIgmpProfiles(hash string, value interface{}) error {
+func updateIgmpProfiles(cntx context.Context, hash string, value interface{}) error {
igmpProfile := value.(*IgmpProfile)
logger.Infow(ctx, "IGMP Profile Migration", log.Fields{"igmpProfile": igmpProfile, "hash": hash})
return nil
}
-func (ig *IgmpGroup) migrateIgmpDevices() {
+func (ig *IgmpGroup) migrateIgmpDevices(cntx context.Context) {
- devices, _ := db.GetPrevIgmpDevices(ig.Mvlan, ig.GroupName)
+ devices, _ := db.GetPrevIgmpDevices(cntx, ig.Mvlan, ig.GroupName)
logger.Infow(ctx, "Migratable Devices", log.Fields{"Devices": devices})
for _, device := range devices {
b, ok := device.Value.([]byte)
@@ -176,10 +177,10 @@
if igd, err := NewIgmpGroupDeviceFromBytes(b); err == nil {
key := database.BasePath + database.IgmpDevicePath + igd.Mvlan.String() + "/" + igd.GroupName + "/" + igd.Device
logger.Infow(ctx, "Deleting old entry", log.Fields{"Path": key, "igd": igd})
- if err := db.Del(key); err != nil {
+ if err := db.Del(cntx, key); err != nil {
logger.Errorw(ctx, "Igmp Group Delete from DB failed", log.Fields{"Error": err, "key": key})
}
- if err := UpdateDbData(database.IgmpDevicePath, key, igd); err != nil {
+ if err := UpdateDbData(cntx, database.IgmpDevicePath, key, igd); err != nil {
logger.Warnw(ctx, "Group Device Migration failed", log.Fields{"IGD": igd, "Error": err})
} else {
logger.Infow(ctx, "Group Device Migrated", log.Fields{"IGD": igd})
@@ -190,9 +191,9 @@
}
}
-func (igd *IgmpGroupDevice) migrateIgmpChannels() {
+func (igd *IgmpGroupDevice) migrateIgmpChannels(cntx context.Context) {
- channels, _ := db.GetPrevIgmpChannels(igd.GroupName, igd.Device)
+ channels, _ := db.GetPrevIgmpChannels(cntx, igd.GroupName, igd.Device)
logger.Infow(ctx, "Migratable Channels", log.Fields{"Channels": channels})
for _, channel := range channels {
@@ -204,10 +205,10 @@
if igc, err := NewIgmpGroupChannelFromBytes(b); err == nil {
key := database.BasePath + database.IgmpChannelPath + igc.GroupName + "/" + igc.Device + "/" + igc.GroupAddr.String()
logger.Infow(ctx, "Deleting old entry", log.Fields{"Path": key, "igc": igc})
- if err := db.Del(key); err != nil {
+ if err := db.Del(cntx, key); err != nil {
logger.Errorw(ctx, "Igmp Group Delete from DB failed", log.Fields{"Error": err, "key": key})
}
- if err := igc.WriteToDb(); err != nil {
+ if err := igc.WriteToDb(cntx); err != nil {
logger.Errorw(ctx, "Igmp group channel Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
@@ -218,9 +219,9 @@
}
}
-func (igc *IgmpGroupChannel) migrateIgmpPorts() {
+func (igc *IgmpGroupChannel) migrateIgmpPorts(cntx context.Context) {
- ports, _ := db.GetPrevIgmpRcvrs(igc.GroupAddr, igc.Device)
+ ports, _ := db.GetPrevIgmpRcvrs(cntx, igc.GroupAddr, igc.Device)
logger.Infow(ctx, "Migratable Ports", log.Fields{"Ports": ports})
for _, port := range ports {
@@ -232,10 +233,10 @@
if igp, err := NewIgmpGroupPortFromBytes(b); err == nil {
key := database.BasePath + database.IgmpPortPath + igc.GroupAddr.String() + "/" + igc.Device + "/" + igp.Port
logger.Infow(ctx, "Deleting old entry", log.Fields{"Key": key, "Igp": igp})
- if err := db.Del(key); err != nil {
+ if err := db.Del(cntx, key); err != nil {
logger.Errorw(ctx, "Igmp Group port Delete from DB failed", log.Fields{"Error": err, "key": key})
}
- if err := igp.WriteToDb(igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
+ if err := igp.WriteToDb(cntx, igc.Mvlan, igc.GroupAddr, igc.Device); err != nil {
logger.Errorw(ctx, "Igmp group port Write to DB failed", log.Fields{"mvlan": igc.Mvlan, "GroupAddr": igc.GroupAddr})
}
diff --git a/internal/pkg/application/pppoeia.go b/internal/pkg/application/pppoeia.go
index 7c55b79..2d50033 100644
--- a/internal/pkg/application/pppoeia.go
+++ b/internal/pkg/application/pppoeia.go
@@ -77,7 +77,7 @@
GetNniVlans() (uint16, uint16)
GetPppoeIaState() PppoeIaState
SetPppoeIaState(PppoeIaState)
- SetMacAddr(net.HardwareAddr)
+ SetMacAddr(context.Context, net.HardwareAddr)
}
// PppoeIaRelayVnet : The PppoeIa relay sessions are stored in a map to be retrieved from when
@@ -280,7 +280,7 @@
// session is derived from the list of PppoeIa sessions stored in the
// common map. The key for retrieval includes the VLAN tags in the
// the packet and the MAC address of the client.
-func (va *VoltApplication) ProcessDsPppoeIaPacket(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessDsPppoeIaPacket(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// Retrieve the layers to build the outgoing packet. It is not
// possible to add/remove layers to the existing packet and thus
@@ -319,7 +319,7 @@
} else if pppoe.Code == layers.PPPoECodePADT {
vpv.SetPppoeIaState(PppoeIaStatePADT)
}
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
// Create the outgoing bufer and set the checksum in the packet
buff := gopacket.NewSerializeBuffer()
@@ -394,7 +394,7 @@
// ProcessUsPppoeIaPacket : The US PppoeIa packet is identified the PppoeIa OP in the packet. A request is considered upstream
// and the service associated with the packet is located by the port and VLANs in the packet
-func (va *VoltApplication) ProcessUsPppoeIaPacket(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessUsPppoeIaPacket(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// We received the packet on an access port and the service for the packet can be
// gotten from the port and the packet
vpv, svc := va.GetVnetFromPkt(device, port, pkt)
@@ -443,7 +443,7 @@
return
}
}
- vpv.SetMacAddr(eth.SrcMAC)
+ vpv.SetMacAddr(cntx, eth.SrcMAC)
}
if pppoe.Code == layers.PPPoECodePADI {
@@ -451,7 +451,7 @@
} else if pppoe.Code == layers.PPPoECodePADR {
vpv.SetPppoeIaState(PppoeIaStatePADR)
}
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
buff := gopacket.NewSerializeBuffer()
@@ -521,7 +521,7 @@
}
// ProcessPPPoEIaPacket to process Pppoeia packet
-func (va *VoltApplication) ProcessPPPoEIaPacket(device string, port string, pkt gopacket.Packet) {
+func (va *VoltApplication) ProcessPPPoEIaPacket(cntx context.Context, device string, port string, pkt gopacket.Packet) {
// Make some error checks before proceeding
pppoel := pkt.Layer(layers.LayerTypePPPoE)
if pppoel == nil {
@@ -544,10 +544,10 @@
// This is treated as an upstream packet in the VOLT application
// as VOLT serves access subscribers who use DHCP to acquire IP
// address and these packets go upstream to the network
- va.ProcessUsPppoeIaPacket(device, port, pkt)
+ va.ProcessUsPppoeIaPacket(cntx, device, port, pkt)
} else {
// This is a downstream packet
- va.ProcessDsPppoeIaPacket(device, port, pkt)
+ va.ProcessDsPppoeIaPacket(cntx, device, port, pkt)
}
}
@@ -566,7 +566,7 @@
}
// ProcessPPPoEPacket : CallBack function registered with application to handle PPPoE packetIn
-func ProcessPPPoEPacket(device string, port string, pkt gopacket.Packet) {
+func ProcessPPPoEPacket(cntx context.Context, device string, port string, pkt gopacket.Packet) {
GetApplication().ProcessPPPoEPacket(device, port, pkt)
}
@@ -613,6 +613,6 @@
func (dpt *PppoeIaPacketTask) Start(ctx context.Context, taskID uint8) error {
dpt.taskID = taskID
dpt.ctx = ctx
- GetApplication().ProcessPPPoEIaPacket(dpt.device, dpt.port, dpt.pkt)
+ GetApplication().ProcessPPPoEIaPacket(ctx, dpt.device, dpt.port, dpt.pkt)
return nil
}
diff --git a/internal/pkg/application/service.go b/internal/pkg/application/service.go
index b9ff4b2..690f348 100644
--- a/internal/pkg/application/service.go
+++ b/internal/pkg/application/service.go
@@ -19,6 +19,7 @@
"bytes"
"encoding/json"
"errors"
+ "context"
"net"
"reflect"
infraerrorCodes "voltha-go-controller/internal/pkg/errorcodes"
@@ -179,7 +180,7 @@
}
// WriteToDb commit a service to the DB if service delete is not in-progress
-func (vs *VoltService) WriteToDb() {
+func (vs *VoltService) WriteToDb(cntx context.Context) {
vs.ServiceLock.RLock()
defer vs.ServiceLock.RUnlock()
@@ -188,18 +189,18 @@
logger.Warnw(ctx, "Skipping Redis Update for Service, Service delete in progress", log.Fields{"Service": vs.Name})
return
}
- vs.ForceWriteToDb()
+ vs.ForceWriteToDb(cntx)
}
//ForceWriteToDb force commit a service to the DB
-func (vs *VoltService) ForceWriteToDb() {
+func (vs *VoltService) ForceWriteToDb(cntx context.Context) {
b, err := json.Marshal(vs)
if err != nil {
logger.Errorw(ctx, "Json Marshal Failed for Service", log.Fields{"Service": vs.Name})
return
}
- if err1 := db.PutService(vs.Name, string(b)); err1 != nil {
+ if err1 := db.PutService(cntx, vs.Name, string(b)); err1 != nil {
logger.Errorw(ctx, "DB write oper failed for Service", log.Fields{"Service": vs.Name})
}
}
@@ -210,12 +211,12 @@
}
// DelFromDb delete a service from DB
-func (vs *VoltService) DelFromDb() {
+func (vs *VoltService) DelFromDb(cntx context.Context) {
logger.Debugw(ctx, "Deleting Service from DB", log.Fields{"Name": vs.Name})
//TODO - Need to understand and delete the second call
//Calling twice has worked though don't know why
- _ = db.DelService(vs.Name)
- _ = db.DelService(vs.Name)
+ _ = db.DelService(cntx, vs.Name)
+ _ = db.DelService(cntx, vs.Name)
}
// MatchesVlans find the service that matches the VLANs. In this case it is
@@ -258,32 +259,32 @@
}
// AddHsiaFlows - Adds US & DS HSIA Flows for the service
-func (vs *VoltService) AddHsiaFlows() {
- if err := vs.AddUsHsiaFlows(); err != nil {
+func (vs *VoltService) AddHsiaFlows(cntx context.Context) {
+ if err := vs.AddUsHsiaFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
- if err := vs.AddDsHsiaFlows(); err != nil {
+ if err := vs.AddDsHsiaFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
}
//DelHsiaFlows - Deletes US & DS HSIA Flows for the service
-func (vs *VoltService) DelHsiaFlows() {
- if err := vs.DelUsHsiaFlows(); err != nil {
+func (vs *VoltService) DelHsiaFlows(cntx context.Context) {
+ if err := vs.DelUsHsiaFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
- if err := vs.DelDsHsiaFlows(); err != nil {
+ if err := vs.DelDsHsiaFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
}
// AddUsHsiaFlows - Add US HSIA Flows for the service
-func (vs *VoltService) AddUsHsiaFlows() error {
+func (vs *VoltService) AddUsHsiaFlows(cntx context.Context) error {
if vs.DeleteInProgress || vs.UpdateInProgress {
logger.Errorw(ctx, "Ignoring US HSIA Flow Push, Service deleteion In-Progress", log.Fields{"Device": vs.Device, "Service": vs.Name})
@@ -322,7 +323,7 @@
continue
}
usflows.MigrateCookie = vgcRebooted
- if err := vs.AddFlows(device, usflows); err != nil {
+ if err := vs.AddFlows(cntx, device, usflows); err != nil {
logger.Errorw(ctx, "Error adding HSIA US flows", log.Fields{"Reason": err.Error()})
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
@@ -331,12 +332,12 @@
vs.UsHSIAFlowsApplied = true
logger.Infow(ctx, "Pushed US HSIA Service Flows", log.Fields{"ServiceName": vs.Name})
}
- vs.WriteToDb()
+ vs.WriteToDb(cntx)
return nil
}
// AddDsHsiaFlows - Add DS HSIA Flows for the service
-func (vs *VoltService) AddDsHsiaFlows() error {
+func (vs *VoltService) AddDsHsiaFlows(cntx context.Context) error {
if vs.DeleteInProgress {
logger.Errorw(ctx, "Ignoring DS HSIA Flow Push, Service deleteion In-Progress", log.Fields{"Device": vs.Device, "Service": vs.Name})
return nil
@@ -365,7 +366,7 @@
return err
}
dsflows.MigrateCookie = vgcRebooted
- if err = vs.AddFlows(device, dsflows); err != nil {
+ if err = vs.AddFlows(cntx, device, dsflows); err != nil {
logger.Errorw(ctx, "Failed to add HSIA DS flows", log.Fields{"Reason": err})
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
@@ -380,7 +381,7 @@
}
logger.Debug(ctx, "Add-one-match-all-pbit-flow")
dsflows.MigrateCookie = vgcRebooted
- if err := vs.AddFlows(device, dsflows); err != nil {
+ if err := vs.AddFlows(cntx, device, dsflows); err != nil {
logger.Errorw(ctx, "Failed to add HSIA DS flows", log.Fields{"Reason": err})
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
@@ -395,7 +396,7 @@
continue
}
dsflows.MigrateCookie = vgcRebooted
- if err := vs.AddFlows(device, dsflows); err != nil {
+ if err := vs.AddFlows(cntx, device, dsflows); err != nil {
logger.Errorw(ctx, "Failed to Add HSIA DS flows", log.Fields{"Reason": err})
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
@@ -406,12 +407,12 @@
vs.DsHSIAFlowsApplied = true
logger.Infow(ctx, "Pushed DS HSIA Service Flows", log.Fields{"ServiceName": vs.Name})
}
- vs.WriteToDb()
+ vs.WriteToDb(cntx)
return nil
}
// DelUsHsiaFlows - Deletes US HSIA Flows for the service
-func (vs *VoltService) DelUsHsiaFlows() error {
+func (vs *VoltService) DelUsHsiaFlows(cntx context.Context) error {
logger.Infow(ctx, "Removing US HSIA Services", log.Fields{"Services": vs.Name})
if vs.UsHSIAFlowsApplied || vgcRebooted {
@@ -437,19 +438,19 @@
continue
}
usflows.MigrateCookie = vgcRebooted
- if err = vs.DelFlows(device, usflows); err != nil {
+ if err = vs.DelFlows(cntx, device, usflows); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
}
vs.UsHSIAFlowsApplied = false
}
- vs.WriteToDb()
+ vs.WriteToDb(cntx)
return nil
}
// DelDsHsiaFlows - Deletes DS HSIA Flows for the service
-func (vs *VoltService) DelDsHsiaFlows() error {
+func (vs *VoltService) DelDsHsiaFlows(cntx context.Context) error {
logger.Infow(ctx, "Removing DS HSIA Services", log.Fields{"Services": vs.Name})
if vs.DsHSIAFlowsApplied || vgcRebooted {
@@ -469,7 +470,7 @@
return err
}
dsflows.MigrateCookie = vgcRebooted
- if err = vs.DelFlows(device, dsflows); err != nil {
+ if err = vs.DelFlows(cntx, device, dsflows); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
@@ -480,7 +481,7 @@
return err
}
dsflows.MigrateCookie = vgcRebooted
- if err = vs.DelFlows(device, dsflows); err != nil {
+ if err = vs.DelFlows(cntx, device, dsflows); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
@@ -494,7 +495,7 @@
continue
}
dsflows.MigrateCookie = vgcRebooted
- if err = vs.DelFlows(device, dsflows); err != nil {
+ if err = vs.DelFlows(cntx, device, dsflows); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vs.triggerServiceFailureInd(statusCode, statusMessage)
}
@@ -504,7 +505,7 @@
}
logger.Infow(ctx, "Deleted HSIA DS flows from DB successfuly", log.Fields{"ServiceName": vs.Name})
// Post HSIA configuration success indication on message bus
- vs.WriteToDb()
+ vs.WriteToDb(cntx)
return nil
}
@@ -949,13 +950,13 @@
}
// SvcUpInd for service up indication
-func (vs *VoltService) SvcUpInd() {
- vs.AddHsiaFlows()
+func (vs *VoltService) SvcUpInd(cntx context.Context) {
+ vs.AddHsiaFlows(cntx)
}
// SvcDownInd for service down indication
-func (vs *VoltService) SvcDownInd() {
- vs.DelHsiaFlows()
+func (vs *VoltService) SvcDownInd(cntx context.Context) {
+ vs.DelHsiaFlows(cntx)
}
// SetIpv4Addr to set ipv4 address
@@ -987,7 +988,7 @@
// current implementation, a service is an entity that is identified by a
// unique L2 (MAC address + VLANs) or unique L3 (VLANs + IP address)
// FUNC: Add Service
-func (va *VoltApplication) AddService(cfg VoltServiceCfg, oper *VoltServiceOper) error {
+func (va *VoltApplication) AddService(cntx context.Context, cfg VoltServiceCfg, oper *VoltServiceOper) error {
var mmUs, mmDs *VoltMeter
var err error
@@ -1059,10 +1060,10 @@
if vnet != nil {
if vpv := va.GetVnetByPort(vs.Port, cfg.SVlan, cfg.CVlan, cfg.UniVlan); vpv != nil {
vpv.VpvLock.Lock()
- vpv.AddSvc(vs)
+ vpv.AddSvc(cntx, vs)
vpv.VpvLock.Unlock()
} else {
- va.AddVnetToPort(vs.Port, vnet, vs)
+ va.AddVnetToPort(cntx, vs.Port, vnet, vs)
}
} else {
logger.Errorw(ctx, "VNET-does-not-exist-for-service", log.Fields{"ServiceName": cfg.Name})
@@ -1072,7 +1073,7 @@
vs.Version = database.PresentVersionMap[database.ServicePath]
// Add the service to the volt application
va.ServiceByName.Store(vs.Name, vs)
- vs.WriteToDb()
+ vs.WriteToDb(cntx)
if nil == oper {
@@ -1082,10 +1083,10 @@
//Update meter profiles service count if service is being added from northbound
mmDs.AssociatedServices++
- va.UpdateMeterProf(*mmDs)
+ va.UpdateMeterProf(cntx, *mmDs)
if mmUs != nil {
mmUs.AssociatedServices++
- va.UpdateMeterProf(*mmUs)
+ va.UpdateMeterProf(cntx, *mmUs)
}
//mmAg.AssociatedServices++
//va.UpdateMeterProf(*mmAg)
@@ -1098,18 +1099,18 @@
//DelServiceWithPrefix - Deletes service with the provided prefix.
// Added for DT/TT usecase with sadis replica interface
-func (va *VoltApplication) DelServiceWithPrefix(prefix string) {
+func (va *VoltApplication) DelServiceWithPrefix(cntx context.Context, prefix string) {
va.ServiceByName.Range(func(key, value interface{}) bool {
srvName := key.(string)
vs := value.(*VoltService)
if strings.Contains(srvName, prefix) {
- va.DelService(srvName, true, nil, false)
+ va.DelService(cntx, srvName, true, nil, false)
vnetName := strconv.FormatUint(uint64(vs.SVlan), 10) + "-"
vnetName = vnetName + strconv.FormatUint(uint64(vs.CVlan), 10) + "-"
vnetName = vnetName + strconv.FormatUint(uint64(vs.UniVlan), 10)
- if err := va.DelVnet(vnetName, ""); err != nil {
+ if err := va.DelVnet(cntx, vnetName, ""); err != nil {
logger.Warnw(ctx, "Delete Vnet Failed", log.Fields{"Name": vnetName, "Error": err})
}
}
@@ -1118,7 +1119,7 @@
}
// DelService delete a service form the application
-func (va *VoltApplication) DelService(name string, forceDelete bool, newSvc *VoltServiceCfg, serviceMigration bool) {
+func (va *VoltApplication) DelService(cntx context.Context, name string, forceDelete bool, newSvc *VoltServiceCfg, serviceMigration bool) {
AppMutex.ServiceDataMutex.Lock()
defer AppMutex.ServiceDataMutex.Unlock()
@@ -1141,7 +1142,7 @@
//Set this to avoid race-condition during flow result processing
vs.DeleteInProgress = true
vs.ForceDelete = forceDelete
- vs.ForceWriteToDb()
+ vs.ForceWriteToDb(cntx)
if len(vs.AssociatedFlows) == 0 {
noFlowsPresent = true
@@ -1149,22 +1150,22 @@
vpv.VpvLock.Lock()
defer vpv.VpvLock.Unlock()
- vs.DelHsiaFlows()
+ vs.DelHsiaFlows(cntx)
if vpv.IgmpEnabled {
- va.ReceiverDownInd(vpv.Device, vpv.Port)
+ va.ReceiverDownInd(cntx, vpv.Device, vpv.Port)
}
logger.Infow(ctx, "Delete Service from VPV", log.Fields{"VPV_Port": vpv.Port, "VPV_SVlan": vpv.SVlan, "VPV_CVlan": vpv.CVlan, "VPV_UniVlan": vpv.UniVlan, "ServiceName": name})
- vpv.DelService(vs)
+ vpv.DelService(cntx, vs)
if vpv.servicesCount.Load() == 0 {
- va.DelVnetFromPort(vs.Port, vpv)
+ va.DelVnetFromPort(cntx, vs.Port, vpv)
}
// Delete the service immediately in case of Force Delete
// This will be enabled when profile reconciliation happens after restore
// of backedup data
if vs.ForceDelete {
- vs.DelFromDb()
+ vs.DelFromDb(cntx)
GetApplication().ServiceByName.Delete(vs.Name)
logger.Warnw(ctx, "Deleted service from DB/Cache successfully", log.Fields{"serviceName": vs.Name})
}
@@ -1209,25 +1210,25 @@
meter.AssociatedServices--
if meter.AssociatedServices == 0 && !skipMeterDeletion {
logger.Infow(ctx, "Meter should be deleted now\n", log.Fields{"MeterID": meter})
- va.UpdateMeterProf(*meter)
+ va.UpdateMeterProf(cntx, *meter)
}
}
}
if noFlowsPresent || vs.ForceDelete {
- vs.CheckAndDeleteService()
+ vs.CheckAndDeleteService(cntx)
}
//Delete the per service counter too
va.ServiceCounters.Delete(name)
if vs.IgmpEnabled && vs.EnableMulticastKPI {
- _ = db.DelAllServiceChannelCounter(name)
+ _ = db.DelAllServiceChannelCounter(cntx, name)
}
}
//AddFlows - Adds the flow to the service
// Triggers flow addition after registering for flow indication event
-func (vs *VoltService) AddFlows(device *VoltDevice, flow *of.VoltFlow) error {
+func (vs *VoltService) AddFlows(cntx context.Context, device *VoltDevice, flow *of.VoltFlow) error {
// Using locks instead of concurrent map for PendingFlows to avoid
// race condition during flow response indication processing
@@ -1245,12 +1246,12 @@
device.RegisterFlowAddEvent(cookie, fe)
vs.PendingFlows[cookie] = true
}
- return cntlr.GetController().AddFlows(vs.Port, device.Name, flow)
+ return cntlr.GetController().AddFlows(cntx, vs.Port, device.Name, flow)
}
//FlowInstallSuccess - Called when corresponding service flow installation is success
// If no more pending flows, HSIA indication wil be triggered
-func (vs *VoltService) FlowInstallSuccess(cookie string, bwAvailInfo of.BwAvailDetails) {
+func (vs *VoltService) FlowInstallSuccess(cntx context.Context, cookie string, bwAvailInfo of.BwAvailDetails) {
if vs.DeleteInProgress {
logger.Warnw(ctx, "Skipping Flow Add Success Notification. Service deletion in-progress", log.Fields{"Cookie": cookie, "Service": vs.Name})
return
@@ -1273,7 +1274,7 @@
vs.BwAvailInfo = prevBwAvail + "," + presentBwAvail
logger.Debugw(ctx, "Bandwidth-value-formed", log.Fields{"BwAvailInfo": vs.BwAvailInfo})
}
- vs.WriteToDb()
+ vs.WriteToDb(cntx)
if len(vs.PendingFlows) == 0 && vs.DsHSIAFlowsApplied {
@@ -1288,7 +1289,7 @@
if vs.Trigger == ServiceVlanUpdate {
vs.Trigger = NBActivate
- defer vs.WriteToDb()
+ defer vs.WriteToDb(cntx)
}
logger.Infow(ctx, "All Flows installed for Service", log.Fields{"Service": vs.Name})
return
@@ -1313,7 +1314,7 @@
//DelFlows - Deletes the flow from the service
// Triggers flow deletion after registering for flow indication event
-func (vs *VoltService) DelFlows(device *VoltDevice, flow *of.VoltFlow) error {
+func (vs *VoltService) DelFlows(cntx context.Context, device *VoltDevice, flow *of.VoltFlow) error {
if !vs.ForceDelete {
// Using locks instead of concurrent map for AssociatedFlows to avoid
@@ -1331,13 +1332,13 @@
device.RegisterFlowDelEvent(cookie, fe)
}
}
- return cntlr.GetController().DelFlows(vs.Port, device.Name, flow)
+ return cntlr.GetController().DelFlows(cntx, vs.Port, device.Name, flow)
}
//CheckAndDeleteService - remove service from DB is there are no pending flows to be removed
-func (vs *VoltService) CheckAndDeleteService() {
+func (vs *VoltService) CheckAndDeleteService(cntx context.Context) {
if vs.DeleteInProgress && len(vs.AssociatedFlows) == 0 && !vs.DsHSIAFlowsApplied {
- vs.DelFromDb()
+ vs.DelFromDb(cntx)
GetApplication().ServiceByName.Delete(vs.Name)
logger.Warnw(ctx, "Deleted service from DB/Cache successfully", log.Fields{"serviceName": vs.Name})
}
@@ -1345,7 +1346,7 @@
//FlowRemoveSuccess - Called when corresponding service flow removal is success
// If no more associated flows, DelHSIA indication wil be triggered
-func (vs *VoltService) FlowRemoveSuccess(cookie string) {
+func (vs *VoltService) FlowRemoveSuccess(cntx context.Context, cookie string) {
// if vs.DeleteInProgress {
// logger.Warnw(ctx, "Skipping Flow Remove Success Notification. Service deletion in-progress", log.Fields{"Cookie": cookie, "Service": vs.Name})
@@ -1364,7 +1365,7 @@
vs.ServiceLock.Unlock()
- vs.WriteToDb()
+ vs.WriteToDb(cntx)
if len(vs.AssociatedFlows) == 0 && !vs.DsHSIAFlowsApplied {
@@ -1378,12 +1379,12 @@
}
if vs.UpdateInProgress {
- vs.updateVnetProfile(vs.Device)
+ vs.updateVnetProfile(cntx, vs.Device)
//Not sending DEL_HSIA Indication since it wil be generated internally by SubMgr
return
}
logger.Infow(ctx, "All Flows removed for Service. Triggering Service De-activation Success indication to NB", log.Fields{"Service": vs.Name, "DeleteFlag": vs.DeleteInProgress})
- vs.CheckAndDeleteService()
+ vs.CheckAndDeleteService(cntx)
return
}
@@ -1392,7 +1393,7 @@
//FlowRemoveFailure - Called when corresponding service flow installation is failed
// Trigger service failure indication to NB
-func (vs *VoltService) FlowRemoveFailure(cookie string, errorCode uint32, errReason string) {
+func (vs *VoltService) FlowRemoveFailure(cntx context.Context, cookie string, errorCode uint32, errReason string) {
vs.ServiceLock.RLock()
if _, ok := vs.AssociatedFlows[cookie]; !ok {
@@ -1407,7 +1408,7 @@
logger.Errorw(ctx, "Service Flow Remove Failure Notification", log.Fields{"uniPort": vs.Port, "Cookie": cookie, "Service": vs.Name, "ErrorCode": errorCode, "ErrorReason": errReason})
vs.triggerServiceFailureInd(errorCode, errReason)
- vs.CheckAndDeleteService()
+ vs.CheckAndDeleteService(cntx)
}
func (vs *VoltService) triggerServiceFailureInd(errorCode uint32, errReason string) {
@@ -1422,9 +1423,9 @@
}
// RestoreSvcsFromDb read from the DB and restore all the services
-func (va *VoltApplication) RestoreSvcsFromDb() {
+func (va *VoltApplication) RestoreSvcsFromDb(cntx context.Context) {
// VNETS must be learnt first
- vss, _ := db.GetServices()
+ vss, _ := db.GetServices(cntx)
for _, vs := range vss {
b, ok := vs.Value.([]byte)
if !ok {
@@ -1438,7 +1439,7 @@
continue
}
logger.Debugw(ctx, "Retrieved Service", log.Fields{"Service": vvs.VoltServiceCfg})
- if err := va.AddService(vvs.VoltServiceCfg, &vvs.VoltServiceOper); err != nil {
+ if err := va.AddService(cntx, vvs.VoltServiceCfg, &vvs.VoltServiceOper); err != nil {
logger.Warnw(ctx, "Add New Service Failed", log.Fields{"Service": vvs.Name, "Error": err})
}
@@ -1551,10 +1552,10 @@
// }
//WriteToDB - writes the udpate vnet request details ot DB
-func (msr *MigrateServicesRequest) WriteToDB() {
+func (msr *MigrateServicesRequest) WriteToDB(cntx context.Context) {
logger.Debugw(ctx, "Adding Migrate Service Request to DB", log.Fields{"OldVnet": msr.OldVnetID, "NewVnet": msr.NewVnetID, "Device": msr.DeviceID, "RequestID": msr.ID, "ServiceCount": len(msr.ServicesList)})
if b, err := json.Marshal(msr); err == nil {
- if err = db.PutMigrateServicesReq(msr.DeviceID, msr.GetMsrKey(), string(b)); err != nil {
+ if err = db.PutMigrateServicesReq(cntx, msr.DeviceID, msr.GetMsrKey(), string(b)); err != nil {
logger.Warnw(ctx, "PutMigrateServicesReq Failed", log.Fields{"OldVnet": msr.OldVnetID, "NewVnet": msr.NewVnetID,
"Device": msr.DeviceID, "Error": err})
}
@@ -1562,7 +1563,7 @@
}
//MigrateServices - updated vnet profile for services
-func (va *VoltApplication) MigrateServices(serialNum string, reqID string, oldVnetID, newVnetID string, serviceList []string) error {
+func (va *VoltApplication) MigrateServices(cntx context.Context, serialNum string, reqID string, oldVnetID, newVnetID string, serviceList []string) error {
logger.Warnw(ctx, "Migrate Serviec Request Received", log.Fields{"SerialNum": serialNum, "RequestID": reqID, "OldVnet": oldVnetID, "NewVnet": newVnetID, "ServiceList": serviceList})
if _, ok := va.VnetsByName.Load(oldVnetID); !ok {
@@ -1584,22 +1585,22 @@
serviceMap[service] = false
}
msr := newMigrateServicesRequest(reqID, oldVnetID, newVnetID, serviceMap, d.Name)
- msr.WriteToDB()
+ msr.WriteToDB(cntx)
d.AddMigratingServices(msr)
- go msr.ProcessMigrateServicesProfRequest()
+ go msr.ProcessMigrateServicesProfRequest(cntx)
return nil
}
//ProcessMigrateServicesProfRequest - collects all associated profiles
-func (msr *MigrateServicesRequest) ProcessMigrateServicesProfRequest() {
+func (msr *MigrateServicesRequest) ProcessMigrateServicesProfRequest(cntx context.Context) {
va := GetApplication()
for srv, processed := range msr.ServicesList {
//Indicates new service is already created and only deletion of old one is pending
if processed {
- va.DelService(srv, true, nil, true)
- msr.serviceMigrated(srv)
+ va.DelService(cntx, srv, true, nil, true)
+ msr.serviceMigrated(cntx, srv)
continue
}
@@ -1628,11 +1629,11 @@
//vpv flows will be removed when last service is removed from it and
// new vpv flows will be installed when new service is added
if vs.UsHSIAFlowsApplied {
- vpv.DelTrapFlows()
- vs.DelHsiaFlows()
+ vpv.DelTrapFlows(cntx)
+ vs.DelHsiaFlows(cntx)
logger.Infow(ctx, "Remove Service Flows Triggered", log.Fields{"Service": srv, "US": vs.UsHSIAFlowsApplied, "DS": vs.DsHSIAFlowsApplied})
} else {
- vs.updateVnetProfile(msr.DeviceID)
+ vs.updateVnetProfile(cntx, msr.DeviceID)
}
} else {
logger.Warnw(ctx, "Migrate Service Failed: Service Not Found", log.Fields{"Service": srv, "Vnet": msr.OldVnetID})
@@ -1688,7 +1689,7 @@
//updateVnetProfile - Called on flow process completion
// Removes old service and creates new VPV & service with udpated vnet profile
-func (vs *VoltService) updateVnetProfile(deviceID string) {
+func (vs *VoltService) updateVnetProfile(cntx context.Context, deviceID string) {
logger.Infow(ctx, "Update Vnet Profile Triggering", log.Fields{"Service": vs.Name, "US": vs.UsHSIAFlowsApplied, "DS": vs.DsHSIAFlowsApplied})
@@ -1746,24 +1747,24 @@
//TODO:Nav Pass a copy, not the pointer
logger.Infow(ctx, "Add New Service Triggering", log.Fields{"Service": nvs.Name, "US": nvs.UsHSIAFlowsApplied, "DS": nvs.DsHSIAFlowsApplied, "DelFlag": nvs.DeleteInProgress})
- if err := va.AddService(nvs.VoltServiceCfg, &nvs.VoltServiceOper); err != nil {
+ if err := va.AddService(cntx, nvs.VoltServiceCfg, &nvs.VoltServiceOper); err != nil {
logger.Warnw(ctx, "Add New Service Failed", log.Fields{"Service": nvs.Name, "Error": err})
}
logger.Infow(ctx, "Add New Service Triggered", log.Fields{"Service": nvs.Name, "US": nvs.UsHSIAFlowsApplied, "DS": nvs.DsHSIAFlowsApplied, "DelFlag": nvs.DeleteInProgress})
msr.ServicesList[oldSrvName] = true
va.updateMigrateServicesRequest(deviceID, oldVnetID, id, msr)
- msr.WriteToDB()
+ msr.WriteToDB(cntx)
logger.Infow(ctx, "Del Old Service Triggering", log.Fields{"Service": oldSrvName, "US": vs.UsHSIAFlowsApplied, "DS": vs.DsHSIAFlowsApplied, "DelFlag": vs.DeleteInProgress})
- va.DelService(oldSrvName, true, nil, true)
+ va.DelService(cntx, oldSrvName, true, nil, true)
logger.Infow(ctx, "Del Old Service Triggered", log.Fields{"Service": oldSrvName, "US": vs.UsHSIAFlowsApplied, "DS": vs.DsHSIAFlowsApplied, "DelFlag": vs.DeleteInProgress})
- msr.serviceMigrated(oldSrvName)
+ msr.serviceMigrated(cntx, oldSrvName)
}
//serviceMigrated - called on successful service updation
// Removes the service entry from servicelist and deletes the request on process completion
-func (msr *MigrateServicesRequest) serviceMigrated(serviceName string) {
+func (msr *MigrateServicesRequest) serviceMigrated(cntx context.Context, serviceName string) {
msr.MigrateServicesLock.Lock()
defer msr.MigrateServicesLock.Unlock()
@@ -1771,22 +1772,22 @@
delete(msr.ServicesList, serviceName)
if len(msr.ServicesList) == 0 {
- _ = db.DelMigrateServicesReq(msr.DeviceID, msr.GetMsrKey())
+ _ = db.DelMigrateServicesReq(cntx, msr.DeviceID, msr.GetMsrKey())
return
}
- msr.WriteToDB()
+ msr.WriteToDB(cntx)
//TODO:Nav - Need for any Response to SubMgr?
}
//TriggerPendingMigrateServicesReq - trigger pending service request
-func (va *VoltApplication) TriggerPendingMigrateServicesReq(device string) {
- va.FetchAndProcessAllMigrateServicesReq(device, storeAndProcessMigrateSrvRequest)
+func (va *VoltApplication) TriggerPendingMigrateServicesReq(cntx context.Context, device string) {
+ va.FetchAndProcessAllMigrateServicesReq(cntx, device, storeAndProcessMigrateSrvRequest)
}
//FetchAndProcessAllMigrateServicesReq - fetch all pending migrate services req from DB and process based on provided func
-func (va *VoltApplication) FetchAndProcessAllMigrateServicesReq(device string, msrAction func(*MigrateServicesRequest)) {
+func (va *VoltApplication) FetchAndProcessAllMigrateServicesReq(cntx context.Context, device string, msrAction func(context.Context, *MigrateServicesRequest)) {
- msrList, _ := db.GetAllMigrateServicesReq(device)
+ msrList, _ := db.GetAllMigrateServicesReq(cntx, device)
for _, msr := range msrList {
b, ok := msr.Value.([]byte)
if !ok {
@@ -1794,7 +1795,7 @@
continue
}
msr := va.createMigrateServicesFromString(b)
- msrAction(msr)
+ msrAction(cntx, msr)
logger.Warnw(ctx, "Triggering Pending Migrate Services Req", log.Fields{"OldVnet": msr.OldVnetID, "NewVnet": msr.NewVnetID, "Device": device, "PendingProfiles": len(msr.ServicesList)})
}
@@ -1813,20 +1814,20 @@
}
//storeAndProcessMigrateSrvRequest - stores the msr info in device obj and triggers req
-func storeAndProcessMigrateSrvRequest(msr *MigrateServicesRequest) {
+func storeAndProcessMigrateSrvRequest(cntx context.Context, msr *MigrateServicesRequest) {
d := GetApplication().GetDevice(msr.DeviceID)
d.AddMigratingServices(msr)
- msr.ProcessMigrateServicesProfRequest()
+ msr.ProcessMigrateServicesProfRequest(cntx)
}
//forceUpdateAllServices - force udpate services with new vnet profile
-func forceUpdateAllServices(msr *MigrateServicesRequest) {
+func forceUpdateAllServices(cntx context.Context, msr *MigrateServicesRequest) {
for srv := range msr.ServicesList {
if vsIntf, ok := GetApplication().ServiceByName.Load(srv); ok {
- vsIntf.(*VoltService).updateVnetProfile(msr.DeviceID)
+ vsIntf.(*VoltService).updateVnetProfile(cntx, msr.DeviceID)
}
}
- _ = db.DelMigrateServicesReq(msr.DeviceID, msr.GetMsrKey())
+ _ = db.DelMigrateServicesReq(cntx, msr.DeviceID, msr.GetMsrKey())
}
//DeepEqualServicecfg - checks if the given service cfgs are same
@@ -1929,18 +1930,18 @@
}
//TriggerAssociatedFlowDelete - re-trigger service flow delete for pending delete flows
-func (vs *VoltService) TriggerAssociatedFlowDelete() bool {
+func (vs *VoltService) TriggerAssociatedFlowDelete(cntx context.Context) bool {
//Clear the Flows flag if already set
//This case happens only in case of some race condition
if vs.UsHSIAFlowsApplied {
- if err := vs.DelUsHsiaFlows(); err != nil {
+ if err := vs.DelUsHsiaFlows(cntx); err != nil {
logger.Errorw(ctx, "DelUsHsiaFlows Failed", log.Fields{"Device": vs.Device, "Service": vs.Name, "Error": err})
}
}
if vs.DsHSIAFlowsApplied {
- if err := vs.DelDsHsiaFlows(); err != nil {
+ if err := vs.DelDsHsiaFlows(cntx); err != nil {
logger.Errorw(ctx, "DelDsHsiaFlows Failed", log.Fields{"Device": vs.Device, "Service": vs.Name, "Error": err})
}
}
@@ -1965,7 +1966,7 @@
subFlow.Cookie = cookie
flow.SubFlows[cookie] = subFlow
logger.Infow(ctx, "Retriggering Service Delete Flow", log.Fields{"Device": vs.Device, "Service": vs.Name, "Cookie": cookie})
- if err := vs.DelFlows(vd, flow); err != nil {
+ if err := vs.DelFlows(cntx, vd, flow); err != nil {
logger.Errorw(ctx, "DelFlows Failed", log.Fields{"Device": vs.Device, "Service": vs.Name, "Cookie": cookie, "Error": err})
}
}
diff --git a/internal/pkg/application/timer.go b/internal/pkg/application/timer.go
index 1a80cb5..aa156a8 100644
--- a/internal/pkg/application/timer.go
+++ b/internal/pkg/application/timer.go
@@ -16,6 +16,7 @@
package application
import (
+ "context"
"time"
)
@@ -40,7 +41,7 @@
}
// Start to start timer
-func (va *VoltApplication) Start(cfg TimerCfg, timerType TimerType) {
+func (va *VoltApplication) Start(cntx context.Context, cfg TimerCfg, timerType TimerType) {
if timerMap[timerType] {
logger.Warn(ctx, "Duplicate Timer!!! Timer already running")
return
@@ -54,7 +55,7 @@
case tickTimer:
va.Tick()
case pendingPoolTimer:
- va.removeExpiredGroups()
+ va.removeExpiredGroups(cntx)
}
case <- timerChannels[timerType]:
return
diff --git a/internal/pkg/application/vnets.go b/internal/pkg/application/vnets.go
index 78fe54f..e11960c 100644
--- a/internal/pkg/application/vnets.go
+++ b/internal/pkg/application/vnets.go
@@ -18,6 +18,7 @@
import (
"encoding/json"
"errors"
+ "context"
"net"
infraerrorCodes "voltha-go-controller/internal/pkg/errorcodes"
"strconv"
@@ -162,7 +163,7 @@
}
//disassociatePortFromVnet - disassociate a port from Vnet and return true if the association map is empty
-func (vv *VoltVnet) disassociatePortFromVnet(device string, port string) {
+func (vv *VoltVnet) disassociatePortFromVnet(cntx context.Context, device string, port string) {
vv.VnetPortLock.Lock()
delete(vv.AssociatedPorts, port)
logger.Infow(ctx, "Disassociated Port from Vnet", log.Fields{"Device": device, "Port": port, "Vnet": vv.Name, "PendingDeleteFlow": vv.PendingDeleteFlow, "AssociatedPorts": vv.AssociatedPorts, "DeleteFlag": vv.DeleteInProgress})
@@ -173,7 +174,7 @@
if len(vv.PendingDeleteFlow[device]) == 0 {
logger.Warnw(ctx, "Deleting Vnet", log.Fields{"Name": vv.Name})
GetApplication().deleteVnetConfig(vv)
- _ = db.DelVnet(vv.Name)
+ _ = db.DelVnet(cntx, vv.Name)
} else {
logger.Warnw(ctx, "Skipping Del Vnet", log.Fields{"Name": vv.Name, "PendingDeleteFlow": vv.PendingDeleteFlow})
}
@@ -192,23 +193,23 @@
}
// WriteToDb commit the VNET to the database
-func (vv *VoltVnet) WriteToDb() {
+func (vv *VoltVnet) WriteToDb(cntx context.Context) {
if vv.DeleteInProgress {
logger.Warnw(ctx, "Skipping Redis Update for Vnet, Vnet delete in progress", log.Fields{"Vnet": vv.Name})
return
}
- vv.ForceWriteToDb()
+ vv.ForceWriteToDb(cntx)
}
//ForceWriteToDb force commit a vnet to the DB
-func (vv *VoltVnet) ForceWriteToDb() {
+func (vv *VoltVnet) ForceWriteToDb(cntx context.Context) {
vv.VnetPortLock.RLock()
defer vv.VnetPortLock.RUnlock()
vv.Version = database.PresentVersionMap[database.VnetPath]
logger.Debugw(ctx, "Updating VNET....", log.Fields{"vnet": vv})
if b, err := json.Marshal(vv); err == nil {
- if err:= db.PutVnet(vv.Name, string(b)); err != nil {
+ if err:= db.PutVnet(cntx, vv.Name, string(b)); err != nil {
logger.Warnw(ctx, "Add Vnet to DB failed", log.Fields{"vnet name": vv.Name, "Error": err})
}
}
@@ -275,7 +276,7 @@
}
// AddVnet to add a VNET to the list of VNETs configured.
-func (va *VoltApplication) AddVnet(cfg VnetConfig, oper *VnetOper) error {
+func (va *VoltApplication) AddVnet(cntx context.Context, cfg VnetConfig, oper *VnetOper) error {
AppMutex.VnetMutex.Lock()
var vv *VoltVnet
@@ -312,7 +313,7 @@
}
va.storeVnetConfig(cfg, vv)
- vv.WriteToDb()
+ vv.WriteToDb(cntx)
logger.Infow(ctx, "Added VNET TO DB", log.Fields{"cfg": cfg, "devicesToHandle": devicesToHandle})
@@ -322,7 +323,7 @@
}
// DelVnet to delete a VNET from the list of VNETs configured
-func (va *VoltApplication) DelVnet(name, deviceSerialNum string) error {
+func (va *VoltApplication) DelVnet(cntx context.Context, name, deviceSerialNum string) error {
logger.Infow(ctx, "Deleting Vnet", log.Fields{"Vnet": name})
AppMutex.VnetMutex.Lock()
if vnetIntf, ok := va.VnetsByName.Load(name); ok {
@@ -330,23 +331,23 @@
//Delete from mvp list
vnet.DevicesList = util.RemoveFromSlice(vnet.DevicesList, deviceSerialNum)
- va.DeleteDevFlowForVlanFromDevice(vnet, deviceSerialNum)
+ va.DeleteDevFlowForVlanFromDevice(cntx, vnet, deviceSerialNum)
if len(vnet.DevicesList) == 0 {
vnet.DeleteInProgress = true
vnet.PendingDeviceToDelete = deviceSerialNum
- vnet.ForceWriteToDb()
+ vnet.ForceWriteToDb(cntx)
vnet.VnetPortLock.RLock()
if len(vnet.PendingDeleteFlow) == 0 && !vnet.isAssociatedPortsPresent() {
logger.Warnw(ctx, "Deleting Vnet", log.Fields{"Name": vnet.Name, "AssociatedPorts": vnet.AssociatedPorts, "PendingDelFlows": vnet.PendingDeleteFlow})
va.deleteVnetConfig(vnet)
- _ = db.DelVnet(vnet.Name)
+ _ = db.DelVnet(cntx, vnet.Name)
} else {
logger.Warnw(ctx, "Skipping Del Vnet", log.Fields{"Name": vnet.Name, "AssociatedPorts": vnet.AssociatedPorts, "PendingDelFlows": vnet.PendingDeleteFlow})
}
vnet.VnetPortLock.RUnlock()
} else {
//Update the devicelist in db
- vnet.WriteToDb()
+ vnet.WriteToDb(cntx)
}
}
//TODO: if no vnets are present on device remove icmpv6 group from device
@@ -355,9 +356,9 @@
}
// UpdateVnet to update the VNET with associated service count
-func (va *VoltApplication) UpdateVnet(vv *VoltVnet) error {
+func (va *VoltApplication) UpdateVnet(cntx context.Context, vv *VoltVnet) error {
va.storeVnetConfig(vv.VnetConfig, vv)
- vv.WriteToDb()
+ vv.WriteToDb(cntx)
logger.Infow(ctx, "Updated VNET TO DB", log.Fields{"vv": vv.VnetConfig})
return nil
}
@@ -577,13 +578,13 @@
}
// DhcpResultInd for dhcp result indication
-func (vpv *VoltPortVnet) DhcpResultInd(res *layers.DHCPv4) {
- vpv.ProcessDhcpResult(res)
+func (vpv *VoltPortVnet) DhcpResultInd(cntx context.Context, res *layers.DHCPv4) {
+ vpv.ProcessDhcpResult(cntx, res)
}
// Dhcpv6ResultInd for dhcpv6 result indication
-func (vpv *VoltPortVnet) Dhcpv6ResultInd(ipv6Addr net.IP, leaseTime uint32) {
- vpv.ProcessDhcpv6Result(ipv6Addr, leaseTime)
+func (vpv *VoltPortVnet) Dhcpv6ResultInd(cntx context.Context, ipv6Addr net.IP, leaseTime uint32) {
+ vpv.ProcessDhcpv6Result(cntx, ipv6Addr, leaseTime)
}
// GetNniVlans to get nni vlans
@@ -611,20 +612,20 @@
}
// AddService to add service
-func (vpv *VoltPortVnet) AddService(service *VoltService) {
+func (vpv *VoltPortVnet) AddService(cntx context.Context, service *VoltService) {
vpv.services.Store(service.Name, service)
vpv.servicesCount.Inc()
logger.Infow(ctx, "Service added/updated to VPV", log.Fields{"Port": vpv.Port, "SVLAN": vpv.SVlan, "CVLAN": vpv.CVlan, "UNIVlan": vpv.UniVlan, "Service": service.Name, "Count": vpv.servicesCount.Load()})
}
// DelService to delete service
-func (vpv *VoltPortVnet) DelService(service *VoltService) {
+func (vpv *VoltPortVnet) DelService(cntx context.Context, service *VoltService) {
vpv.services.Delete(service.Name)
vpv.servicesCount.Dec()
// If the only Igmp Enabled service is removed, remove the Igmp trap flow along with it
if service.IgmpEnabled {
- if err := vpv.DelIgmpFlows(); err != nil {
+ if err := vpv.DelIgmpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
@@ -635,26 +636,33 @@
}
// ProcessDhcpResult to process dhcp results
-func (vpv *VoltPortVnet) ProcessDhcpResult(res *layers.DHCPv4) {
+func (vpv *VoltPortVnet) ProcessDhcpResult(cntx context.Context, res *layers.DHCPv4) {
msgType := DhcpMsgType(res)
if msgType == layers.DHCPMsgTypeAck {
- vpv.ProcessDhcpSuccess(res)
+ vpv.ProcessDhcpSuccess(cntx, res)
} else if msgType == layers.DHCPMsgTypeNak {
vpv.DhcpStatus = DhcpStatusNacked
}
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
+}
+
+// RangeOnServices to call a function on all services on the vpv
+func (vpv *VoltPortVnet) RangeOnServices(cntx context.Context, callback func(cntx context.Context, key, value interface{}) bool) {
+ vpv.services.Range(func(key, value interface{}) bool {
+ return callback(cntx, key, value)
+ })
}
// ProcessDhcpSuccess : Learn the IPv4 address allocated to the services and update the
// the services with the same. This also calls for adding flows
// for the services as the DHCP procedure is completed
-func (vpv *VoltPortVnet) ProcessDhcpSuccess(res *layers.DHCPv4) {
+func (vpv *VoltPortVnet) ProcessDhcpSuccess(cntx context.Context, res *layers.DHCPv4) {
vpv.DhcpStatus = DhcpStatusAcked
vpv.Ipv4Addr, _ = GetIpv4Addr(res)
logger.Infow(ctx, "Received IPv4 Address", log.Fields{"IP Address": vpv.Ipv4Addr.String()})
logger.Infow(ctx, "Services Configured", log.Fields{"Count": vpv.servicesCount.Load()})
- vpv.services.Range(vpv.updateIPv4AndProvisionFlows)
+ vpv.RangeOnServices(cntx, vpv.updateIPv4AndProvisionFlows)
vpv.ProcessDhcpv4Options(res)
}
@@ -675,17 +683,17 @@
// VNET. The same IPv6 address is also passed to the services. When a
// service is fetched all the associated information such as MAC address,
// IPv4 address and IPv6 addresses can be provided.
-func (vpv *VoltPortVnet) ProcessDhcpv6Result(ipv6Addr net.IP, leaseTime uint32) {
+func (vpv *VoltPortVnet) ProcessDhcpv6Result(cntx context.Context, ipv6Addr net.IP, leaseTime uint32) {
// TODO: Status based hanlding of flows
vpv.Dhcp6ExpiryTime = time.Now().Add((time.Duration(leaseTime) * time.Second))
vpv.Ipv6Addr = ipv6Addr
- vpv.services.Range(vpv.updateIPv6AndProvisionFlows)
- vpv.WriteToDb()
+ vpv.RangeOnServices(cntx, vpv.updateIPv6AndProvisionFlows)
+ vpv.WriteToDb(cntx)
}
// AddSvcUsMeterToDevice to add service upstream meter info to device
-func AddSvcUsMeterToDevice(key, value interface{}) bool {
+func AddSvcUsMeterToDevice(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
logger.Infow(ctx, "Adding upstream meter profile to device", log.Fields{"ServiceName": svc.Name})
if device, _ := GetApplication().GetDeviceFromPort(svc.Port); device != nil {
@@ -697,7 +705,7 @@
}
// PushFlowsForPortVnet - triggers flow construction and push for provided VPV
-func (vpv *VoltPortVnet) PushFlowsForPortVnet(d *VoltDevice) {
+func (vpv *VoltPortVnet) PushFlowsForPortVnet(cntx context.Context, d *VoltDevice) {
vp := d.GetPort(vpv.Port)
@@ -717,7 +725,7 @@
// vpv.DsFlowsApplied = false
// vpv.UsFlowsApplied = false
vpv.VpvLock.Lock()
- vpv.PortUpInd(d, vpv.Port)
+ vpv.PortUpInd(cntx, d, vpv.Port)
vpv.VpvLock.Unlock()
}
@@ -726,7 +734,7 @@
// again here to apply the latest configuration if the configuration
// changed. Thus, a reboot of ONT forces the new configuration to get
// applied.
-func (vpv *VoltPortVnet) PortUpInd(device *VoltDevice, port string) {
+func (vpv *VoltPortVnet) PortUpInd(cntx context.Context, device *VoltDevice, port string) {
if vpv.DeleteInProgress {
logger.Errorw(ctx, "Ignoring VPV Port UP Ind, VPV deleteion In-Progress", log.Fields{"Device": device, "Port": port, "Vnet": vpv.VnetName})
@@ -760,16 +768,16 @@
logger.Infow(ctx, "Port Up - Trap Flows", log.Fields{"Device": device.Name, "Port": port})
// no HSIA flows for multicast service
if !vpv.McastService {
- vpv.services.Range(AddUsHsiaFlows)
+ vpv.RangeOnServices(cntx, AddUsHsiaFlows)
}
- vpv.AddTrapFlows()
+ vpv.AddTrapFlows(cntx)
if vpv.MacLearning == MacLearningNone || NonZeroMacAddress(vpv.MacAddr) {
logger.Infow(ctx, "Port Up - DS Flows", log.Fields{"Device": device.Name, "Port": port})
// US & DS DHCP, US HSIA flows are already installed
// install only DS HSIA flow here.
// no HSIA flows for multicast service
if !vpv.McastService {
- vpv.services.Range(AddDsHsiaFlows)
+ vpv.RangeOnServices(cntx, AddDsHsiaFlows)
}
}
@@ -781,50 +789,50 @@
// however is not seen as a real use case.
logger.Infow(ctx, "Port Up - Service Flows", log.Fields{"Device": device.Name, "Port": port})
if !vpv.McastService {
- vpv.services.Range(AddUsHsiaFlows)
+ vpv.RangeOnServices(cntx, AddUsHsiaFlows)
}
- vpv.AddTrapFlows()
+ vpv.AddTrapFlows(cntx)
if !vpv.McastService {
- vpv.services.Range(AddDsHsiaFlows)
+ vpv.RangeOnServices(cntx, AddDsHsiaFlows)
}
}
// Process IGMP proxy - install IGMP trap rules before DHCP trap rules
if vpv.IgmpEnabled {
logger.Infow(ctx, "Port Up - IGMP Flows", log.Fields{"Device": device.Name, "Port": port})
- vpv.services.Range(AddSvcUsMeterToDevice)
- if err := vpv.AddIgmpFlows(); err != nil {
+ vpv.RangeOnServices(cntx, AddSvcUsMeterToDevice)
+ if err := vpv.AddIgmpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
if vpv.McastService {
- vpv.services.Range(PostAccessConfigSuccessInd)
+ vpv.RangeOnServices(cntx, PostAccessConfigSuccessInd)
}
}
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
// PortDownInd : When the port status changes to down, we delete all configured flows
// The same indication is also passed to the services enqueued for them
// to take appropriate actions
-func (vpv *VoltPortVnet) PortDownInd(device string, port string) {
+func (vpv *VoltPortVnet) PortDownInd(cntx context.Context, device string, port string) {
logger.Infow(ctx, "VPV Port DOWN Ind, deleting all flows for services",
log.Fields{"service count": vpv.servicesCount.Load()})
- //vpv.services.Range(DelAllFlows)
- vpv.DelTrapFlows()
- vpv.DelHsiaFlows()
- vpv.WriteToDb()
- vpv.ClearServiceCounters()
+ //vpv.RangeOnServices(cntx, DelAllFlows)
+ vpv.DelTrapFlows(cntx)
+ vpv.DelHsiaFlows(cntx)
+ vpv.WriteToDb(cntx)
+ vpv.ClearServiceCounters(cntx)
}
// SetMacAddr : The MAC address is set when a MAC address is learnt through the
// packets received from the network. Currently, DHCP packets are
// only packets we learn the MAC address from
-func (vpv *VoltPortVnet) SetMacAddr(addr net.HardwareAddr) {
+func (vpv *VoltPortVnet) SetMacAddr(cntx context.Context, addr net.HardwareAddr) {
//Store Learnt MAC address and return if MACLearning is not enabled
vpv.LearntMacAddr = addr
@@ -852,15 +860,15 @@
// may have been changed
// Atleast one HSIA flow should be present in adapter to retain the TP and GEM
// hence delete one after the other
- vpv.services.Range(DelUsHsiaFlows)
+ vpv.RangeOnServices(cntx, DelUsHsiaFlows)
vpv.MacAddr = addr
- vpv.services.Range(vpv.setLearntMAC)
- vpv.services.Range(AddUsHsiaFlows)
- vpv.services.Range(DelDsHsiaFlows)
+ vpv.RangeOnServices(cntx, vpv.setLearntMAC)
+ vpv.RangeOnServices(cntx, AddUsHsiaFlows)
+ vpv.RangeOnServices(cntx, DelDsHsiaFlows)
GetApplication().DeleteMacInPortMap(vpv.MacAddr)
} else {
vpv.MacAddr = addr
- vpv.services.Range(vpv.setLearntMAC)
+ vpv.RangeOnServices(cntx, vpv.setLearntMAC)
logger.Infow(ctx, "MAC Address learnt from DHCP or ARP", log.Fields{"Learnt MAC": addr.String(), "Port": vpv.Port})
}
GetApplication().UpdateMacInPortMap(vpv.MacAddr, vpv.Port)
@@ -879,10 +887,10 @@
if vpv.FlowsApplied {
// no HSIA flows for multicast service
if !vpv.McastService {
- vpv.services.Range(AddDsHsiaFlows)
+ vpv.RangeOnServices(cntx, AddDsHsiaFlows)
}
}
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
// MatchesVlans : If the VNET matches both S and C VLANs, return true. Else, return false
@@ -954,10 +962,10 @@
// AddSvc adds a service on the VNET on a port. The addition is
// triggered when NB requests for service addition
-func (vpv *VoltPortVnet) AddSvc(svc *VoltService) {
+func (vpv *VoltPortVnet) AddSvc(cntx context.Context, svc *VoltService) {
//vpv.services = append(vpv.services, svc)
- vpv.AddService(svc)
+ vpv.AddService(cntx, svc)
logger.Debugw(ctx, "Added Service to VPV", log.Fields{"Num of SVCs": vpv.servicesCount.Load(), "SVC": svc})
// Learn the circuit-id and remote-id from the service
@@ -1034,9 +1042,9 @@
//to which the serivce is associated
if vpv.FlowsApplied {
if NonZeroMacAddress(vpv.MacAddr) || svc.MacLearning == MacLearningNone {
- svc.AddHsiaFlows()
+ svc.AddHsiaFlows(cntx)
} else {
- if err:= svc.AddUsHsiaFlows(); err != nil {
+ if err:= svc.AddUsHsiaFlows(cntx); err != nil {
logger.Warnw(ctx, "Add US hsia flow failed", log.Fields{"service": svc.Name, "Error": err})
}
}
@@ -1047,71 +1055,71 @@
// service with Igmp Enabled needs to be installed
if svc.IgmpEnabled && vpv.FlowsApplied {
logger.Infow(ctx, "Add Service - IGMP Flows", log.Fields{"Device": vpv.Device, "Port": vpv.Port})
- if err := vpv.AddIgmpFlows(); err != nil {
+ if err := vpv.AddIgmpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
if vpv.McastService {
//For McastService, send Service Activated indication once IGMP US flow is pushed
- vpv.services.Range(PostAccessConfigSuccessInd)
+ vpv.RangeOnServices(cntx, PostAccessConfigSuccessInd)
}
}
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
// setLearntMAC to set learnt mac
-func (vpv *VoltPortVnet) setLearntMAC(key, value interface{}) bool {
+func (vpv *VoltPortVnet) setLearntMAC(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
svc.SetMacAddr(vpv.MacAddr)
- svc.WriteToDb()
+ svc.WriteToDb(cntx)
return true
}
// PostAccessConfigSuccessInd for posting access config success indication
-func PostAccessConfigSuccessInd(key, value interface{}) bool {
+func PostAccessConfigSuccessInd(cntx context.Context, key, value interface{}) bool {
return true
}
// updateIPv4AndProvisionFlows to update ipv4 and provisional flows
-func (vpv *VoltPortVnet) updateIPv4AndProvisionFlows(key, value interface{}) bool {
+func (vpv *VoltPortVnet) updateIPv4AndProvisionFlows(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
logger.Infow(ctx, "Updating Ipv4 address for service", log.Fields{"ServiceName": svc.Name})
svc.SetIpv4Addr(vpv.Ipv4Addr)
- svc.WriteToDb()
+ svc.WriteToDb(cntx)
return true
}
// updateIPv6AndProvisionFlows to update ipv6 and provisional flow
-func (vpv *VoltPortVnet) updateIPv6AndProvisionFlows(key, value interface{}) bool {
+func (vpv *VoltPortVnet) updateIPv6AndProvisionFlows(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
svc.SetIpv6Addr(vpv.Ipv6Addr)
- svc.WriteToDb()
+ svc.WriteToDb(cntx)
return true
}
// AddUsHsiaFlows to add upstream hsia flows
-func AddUsHsiaFlows(key, value interface{}) bool {
+func AddUsHsiaFlows(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
- if err:= svc.AddUsHsiaFlows(); err != nil {
+ if err:= svc.AddUsHsiaFlows(cntx); err != nil {
logger.Warnw(ctx, "Add US hsia flow failed", log.Fields{"service": svc.Name, "Error": err})
}
return true
}
// AddDsHsiaFlows to add downstream hsia flows
-func AddDsHsiaFlows(key, value interface{}) bool {
+func AddDsHsiaFlows(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
- if err:= svc.AddDsHsiaFlows(); err != nil {
+ if err:= svc.AddDsHsiaFlows(cntx); err != nil {
logger.Warnw(ctx, "Add DS hsia flow failed", log.Fields{"service": svc.Name, "Error": err})
}
return true
}
// ClearFlagsInService to clear the flags used in service
-func ClearFlagsInService(key, value interface{}) bool {
+func ClearFlagsInService(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
svc.ServiceLock.Lock()
svc.IgmpFlowsApplied = false
@@ -1123,50 +1131,50 @@
svc.PendingFlows = make(map[string]bool)
svc.AssociatedFlows = make(map[string]bool)
svc.ServiceLock.Unlock()
- svc.WriteToDb()
+ svc.WriteToDb(cntx)
logger.Debugw(ctx, "Cleared Flow Flags for service", log.Fields{"name": svc.Name})
return true
}
// DelDsHsiaFlows to delete hsia flows
-func DelDsHsiaFlows(key, value interface{}) bool {
+func DelDsHsiaFlows(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
- if err:= svc.DelDsHsiaFlows(); err != nil {
+ if err:= svc.DelDsHsiaFlows(cntx); err != nil {
logger.Warnw(ctx, "Delete DS hsia flow failed", log.Fields{"service": svc.Name, "Error": err})
}
return true
}
// DelUsHsiaFlows to delete upstream hsia flows
-func DelUsHsiaFlows(key, value interface{}) bool {
+func DelUsHsiaFlows(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
- if err:= svc.DelUsHsiaFlows(); err != nil {
+ if err:= svc.DelUsHsiaFlows(cntx); err != nil {
logger.Warnw(ctx, "Delete US hsia flow failed", log.Fields{"service": svc.Name, "Error": err})
}
return true
}
// ClearServiceCounters to clear the service counters
-func ClearServiceCounters(key, value interface{}) bool {
+func ClearServiceCounters(cntx context.Context, key, value interface{}) bool {
svc := value.(*VoltService)
//Delete the per service counter too
GetApplication().ServiceCounters.Delete(svc.Name)
if svc.IgmpEnabled && svc.EnableMulticastKPI {
- _ = db.DelAllServiceChannelCounter(svc.Name)
+ _ = db.DelAllServiceChannelCounter(cntx, svc.Name)
}
return true
}
//AddTrapFlows - Adds US & DS Trap flows
-func (vpv *VoltPortVnet) AddTrapFlows() {
+func (vpv *VoltPortVnet) AddTrapFlows(cntx context.Context) {
if !vpv.FlowsApplied || vgcRebooted {
if vpv.DhcpRelay {
- if err := vpv.AddUsDhcpFlows(); err != nil {
+ if err := vpv.AddUsDhcpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
- if err := vpv.AddDsDhcpFlows(); err != nil {
+ if err := vpv.AddDsDhcpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
@@ -1174,85 +1182,85 @@
log.Fields{"port": vpv.Port})
//vpv.updateICMPv6McGroup(true)
} else if vpv.ArpRelay {
- if err := vpv.AddUsArpFlows(); err != nil {
+ if err := vpv.AddUsArpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
logger.Info(ctx, "ARP trap rules not added in downstream direction")
} else if vpv.PppoeIa {
- if err := vpv.AddUsPppoeFlows(); err != nil {
+ if err := vpv.AddUsPppoeFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
- if err := vpv.AddDsPppoeFlows(); err != nil {
+ if err := vpv.AddDsPppoeFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
}
vpv.FlowsApplied = true
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
}
//DelTrapFlows - Removes all US & DS DHCP, IGMP trap flows.
-func (vpv *VoltPortVnet) DelTrapFlows() {
+func (vpv *VoltPortVnet) DelTrapFlows(cntx context.Context) {
// Delete HSIA & DHCP flows before deleting IGMP flows
if vpv.FlowsApplied || vgcRebooted {
if vpv.DhcpRelay {
- if err:= vpv.DelUsDhcpFlows(); err != nil {
+ if err:= vpv.DelUsDhcpFlows(cntx); err != nil {
logger.Warnw(ctx, "Delete US hsia flow failed", log.Fields{"port": vpv.Port, "SVlan": vpv.SVlan, "CVlan": vpv.CVlan,
"UniVlan": vpv.UniVlan, "Error": err})
}
logger.Infow(ctx, "ICMPv6 MC Group modification will not be triggered to rwcore for ",
log.Fields{"port": vpv.Port})
- if err := vpv.DelDsDhcpFlows(); err != nil {
+ if err := vpv.DelDsDhcpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
//vpv.updateICMPv6McGroup(false)
} else if vpv.ArpRelay {
- if err := vpv.DelUsArpFlows(); err != nil {
+ if err := vpv.DelUsArpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
} else if vpv.PppoeIa {
- if err := vpv.DelUsPppoeFlows(); err != nil {
+ if err := vpv.DelUsPppoeFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
- if err := vpv.DelDsPppoeFlows(); err != nil {
+ if err := vpv.DelDsPppoeFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
}
vpv.FlowsApplied = false
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
- if err:= vpv.DelIgmpFlows(); err != nil {
+ if err:= vpv.DelIgmpFlows(cntx); err != nil {
logger.Warnw(ctx, "Delete igmp flow failed", log.Fields{"port": vpv.Port, "SVlan": vpv.SVlan, "CVlan": vpv.CVlan,
"UniVlan": vpv.UniVlan, "Error": err})
}
}
// DelHsiaFlows deletes the service flows
-func (vpv *VoltPortVnet) DelHsiaFlows() {
+func (vpv *VoltPortVnet) DelHsiaFlows(cntx context.Context) {
// no HSIA flows for multicast service
if !vpv.McastService {
- vpv.services.Range(DelUsHsiaFlows)
- vpv.services.Range(DelDsHsiaFlows)
+ vpv.RangeOnServices(cntx, DelUsHsiaFlows)
+ vpv.RangeOnServices(cntx, DelDsHsiaFlows)
}
}
//ClearServiceCounters - Removes all igmp counters for a service
-func (vpv *VoltPortVnet) ClearServiceCounters() {
+func (vpv *VoltPortVnet) ClearServiceCounters(cntx context.Context) {
//send flows deleted indication to submgr
- vpv.services.Range(ClearServiceCounters)
+ vpv.RangeOnServices(cntx, ClearServiceCounters)
}
// AddUsDhcpFlows pushes the DHCP flows to the VOLTHA via the controller
-func (vpv *VoltPortVnet) AddUsDhcpFlows() error {
+func (vpv *VoltPortVnet) AddUsDhcpFlows(cntx context.Context) error {
var vd *VoltDevice
device := vpv.Device
@@ -1269,7 +1277,7 @@
flows, err := vpv.BuildUsDhcpFlows()
if err == nil {
logger.Debugw(ctx, "Adding US DHCP flows", log.Fields{"Device": device})
- if err1 := vpv.PushFlows(vd, flows); err1 != nil {
+ if err1 := vpv.PushFlows(cntx, vd, flows); err1 != nil {
//push ind here ABHI
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err1)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
@@ -1302,7 +1310,7 @@
}
// AddDsDhcpFlows function pushes the DHCP flows to the VOLTHA via the controller
-func (vpv *VoltPortVnet) AddDsDhcpFlows() error {
+func (vpv *VoltPortVnet) AddDsDhcpFlows(cntx context.Context) error {
var vd *VoltDevice
device := vpv.Device
@@ -1322,7 +1330,7 @@
flows, err := vpv.BuildDsDhcpFlows()
if err == nil {
- if err1 := vpv.PushFlows(vd, flows); err1 != nil {
+ if err1 := vpv.PushFlows(cntx, vd, flows); err1 != nil {
//push ind here and procced
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err1)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
@@ -1358,13 +1366,13 @@
}
// DelDhcpFlows deletes both US & DS DHCP flows applied for this Vnet instantiated on the port
-func (vpv *VoltPortVnet) DelDhcpFlows() {
- if err := vpv.DelUsDhcpFlows(); err != nil {
+func (vpv *VoltPortVnet) DelDhcpFlows(cntx context.Context) {
+ if err := vpv.DelUsDhcpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
- if err := vpv.DelDsDhcpFlows(); err != nil {
+ if err := vpv.DelDsDhcpFlows(cntx); err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
}
@@ -1373,13 +1381,13 @@
// DelUsDhcpFlows delete the DHCP flows applied for this Vnet instantiated on the port
// Write the status of the VPV to the DB once the delete is scheduled
// for dispatch
-func (vpv *VoltPortVnet) DelUsDhcpFlows() error {
+func (vpv *VoltPortVnet) DelUsDhcpFlows(cntx context.Context) error {
device, err := GetApplication().GetDeviceFromPort(vpv.Port)
if err != nil {
return err
}
- err = vpv.delDhcp4Flows(device)
+ err = vpv.delDhcp4Flows(cntx, device)
if err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
@@ -1393,10 +1401,10 @@
return nil
}
-func (vpv *VoltPortVnet) delDhcp4Flows(device *VoltDevice) error {
+func (vpv *VoltPortVnet) delDhcp4Flows(cntx context.Context, device *VoltDevice) error {
flows, err := vpv.BuildUsDhcpFlows()
if err == nil {
- return vpv.RemoveFlows(device, flows)
+ return vpv.RemoveFlows(cntx, device, flows)
}
logger.Errorw(ctx, "US DHCP Flow Delete Failed", log.Fields{"Reason": err.Error()})
return err
@@ -1415,12 +1423,12 @@
// DelDsDhcpFlows delete the DHCP flows applied for this Vnet instantiated on the port
// Write the status of the VPV to the DB once the delete is scheduled
// for dispatch
-func (vpv *VoltPortVnet) DelDsDhcpFlows() error {
+func (vpv *VoltPortVnet) DelDsDhcpFlows(cntx context.Context) error {
device, err := GetApplication().GetDeviceFromPort(vpv.Port)
if err != nil {
return err
}
- err = vpv.delDsDhcp4Flows(device)
+ err = vpv.delDsDhcp4Flows(cntx, device)
if err != nil {
statusCode, statusMessage := infraerrorCodes.GetErrorInfo(err)
vpv.FlowInstallFailure("VGC processing failure", statusCode, statusMessage)
@@ -1434,10 +1442,10 @@
return nil
}
-func (vpv *VoltPortVnet) delDsDhcp4Flows(device *VoltDevice) error {
+func (vpv *VoltPortVnet) delDsDhcp4Flows(cntx context.Context, device *VoltDevice) error {
flows, err := vpv.BuildDsDhcpFlows()
if err == nil {
- return vpv.RemoveFlows(device, flows)
+ return vpv.RemoveFlows(cntx, device, flows)
}
logger.Errorw(ctx, "DS DHCP Flow Delete Failed", log.Fields{"Reason": err.Error()})
return err
@@ -1454,7 +1462,7 @@
}*/
// AddUsArpFlows pushes the ARP flows to the VOLTHA via the controller
-func (vpv *VoltPortVnet) AddUsArpFlows() error {
+func (vpv *VoltPortVnet) AddUsArpFlows(cntx context.Context) error {
var vd *VoltDevice
device := vpv.Device
@@ -1471,7 +1479,7 @@
flows, err := vpv.BuildUsArpFlows()
if err == nil {
logger.Debugw(ctx, "Adding US ARP flows", log.Fields{"Device": device})
- if err1 := vpv.PushFlows(vd, flows); err1 != nil {
+ if err1 := vpv.PushFlows(cntx, vd, flows); err1 != nil {
return err1
}
} else {
@@ -1484,21 +1492,21 @@
// DelUsArpFlows delete the ARP flows applied for this Vnet instantiated on the port
// Write the status of the VPV to the DB once the delete is scheduled
// for dispatch
-func (vpv *VoltPortVnet) DelUsArpFlows() error {
+func (vpv *VoltPortVnet) DelUsArpFlows(cntx context.Context) error {
device, err := GetApplication().GetDeviceFromPort(vpv.Port)
if err != nil {
return err
}
flows, err := vpv.BuildUsArpFlows()
if err == nil {
- return vpv.RemoveFlows(device, flows)
+ return vpv.RemoveFlows(cntx, device, flows)
}
logger.Errorw(ctx, "US ARP Flow Delete Failed", log.Fields{"Reason": err.Error()})
return err
}
// AddUsPppoeFlows pushes the PPPoE flows to the VOLTHA via the controller
-func (vpv *VoltPortVnet) AddUsPppoeFlows() error {
+func (vpv *VoltPortVnet) AddUsPppoeFlows(cntx context.Context) error {
logger.Debugw(ctx, "Adding US PPPoE flows", log.Fields{"STAG": vpv.SVlan, "CTAG": vpv.CVlan, "Device": vpv.Device})
var vd *VoltDevice
@@ -1517,7 +1525,7 @@
if flows, err := vpv.BuildUsPppoeFlows(); err == nil {
logger.Debugw(ctx, "Adding US PPPoE flows", log.Fields{"Device": device})
- if err1 := vpv.PushFlows(vd, flows); err1 != nil {
+ if err1 := vpv.PushFlows(cntx, vd, flows); err1 != nil {
return err1
}
} else {
@@ -1528,7 +1536,7 @@
}
// AddDsPppoeFlows to add downstream pppoe flows
-func (vpv *VoltPortVnet) AddDsPppoeFlows() error {
+func (vpv *VoltPortVnet) AddDsPppoeFlows(cntx context.Context) error {
logger.Debugw(ctx, "Adding DS PPPoE flows", log.Fields{"STAG": vpv.SVlan, "CTAG": vpv.CVlan, "Device": vpv.Device})
var vd *VoltDevice
device := vpv.Device
@@ -1546,7 +1554,7 @@
flows, err := vpv.BuildDsPppoeFlows()
if err == nil {
- if err1 := vpv.PushFlows(vd, flows); err1 != nil {
+ if err1 := vpv.PushFlows(cntx, vd, flows); err1 != nil {
return err1
}
} else {
@@ -1559,7 +1567,7 @@
// DelUsPppoeFlows delete the PPPoE flows applied for this Vnet instantiated on the port
// Write the status of the VPV to the DB once the delete is scheduled
// for dispatch
-func (vpv *VoltPortVnet) DelUsPppoeFlows() error {
+func (vpv *VoltPortVnet) DelUsPppoeFlows(cntx context.Context) error {
logger.Debugw(ctx, "Deleting US PPPoE flows", log.Fields{"STAG": vpv.SVlan, "CTAG": vpv.CVlan, "Device": vpv.Device})
device, err := GetApplication().GetDeviceFromPort(vpv.Port)
if err != nil {
@@ -1567,7 +1575,7 @@
}
flows, err := vpv.BuildUsPppoeFlows()
if err == nil {
- return vpv.RemoveFlows(device, flows)
+ return vpv.RemoveFlows(cntx, device, flows)
}
logger.Errorw(ctx, "US PPPoE Flow Delete Failed", log.Fields{"Reason": err.Error()})
return err
@@ -1576,7 +1584,7 @@
// DelDsPppoeFlows delete the PPPoE flows applied for this Vnet instantiated on the port
// Write the status of the VPV to the DB once the delete is scheduled
// for dispatch
-func (vpv *VoltPortVnet) DelDsPppoeFlows() error {
+func (vpv *VoltPortVnet) DelDsPppoeFlows(cntx context.Context) error {
logger.Debugw(ctx, "Deleting DS PPPoE flows", log.Fields{"STAG": vpv.SVlan, "CTAG": vpv.CVlan, "Device": vpv.Device})
device, err := GetApplication().GetDeviceFromPort(vpv.Port)
if err != nil {
@@ -1584,14 +1592,14 @@
}
flows, err := vpv.BuildDsPppoeFlows()
if err == nil {
- return vpv.RemoveFlows(device, flows)
+ return vpv.RemoveFlows(cntx, device, flows)
}
logger.Errorw(ctx, "DS PPPoE Flow Delete Failed", log.Fields{"Reason": err.Error()})
return err
}
// AddIgmpFlows function pushes the IGMP flows to the VOLTHA via the controller
-func (vpv *VoltPortVnet) AddIgmpFlows() error {
+func (vpv *VoltPortVnet) AddIgmpFlows(cntx context.Context) error {
if !vpv.IgmpFlowsApplied || vgcRebooted {
if vpv.MvlanProfileName == "" {
@@ -1619,7 +1627,7 @@
vd.RegisterFlowAddEvent(cookie, fe)
}
}
- if err1 := cntlr.GetController().AddFlows(vpv.Port, device.Name, flows); err1 != nil {
+ if err1 := cntlr.GetController().AddFlows(cntx, vpv.Port, device.Name, flows); err1 != nil {
return err1
}
} else {
@@ -1627,7 +1635,7 @@
return err
}
vpv.IgmpFlowsApplied = true
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
return nil
}
@@ -1635,7 +1643,7 @@
// DelIgmpFlows delete the IGMP flows applied for this Vnet instantiated on the port
// Write the status of the VPV to the DB once the delete is scheduled
// for dispatch
-func (vpv *VoltPortVnet) DelIgmpFlows() error {
+func (vpv *VoltPortVnet) DelIgmpFlows(cntx context.Context) error {
if vpv.IgmpFlowsApplied || vgcRebooted {
device, err := GetApplication().GetDeviceFromPort(vpv.Port)
@@ -1645,7 +1653,7 @@
}
flows, err := vpv.BuildIgmpFlows()
if err == nil {
- if err1 := vpv.RemoveFlows(device, flows); err1 != nil {
+ if err1 := vpv.RemoveFlows(cntx, device, flows); err1 != nil {
return err1
}
} else {
@@ -1653,7 +1661,7 @@
return err
}
vpv.IgmpFlowsApplied = false
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
return nil
}
@@ -2111,21 +2119,21 @@
}
// WriteToDb for writing to database
-func (vpv *VoltPortVnet) WriteToDb() {
+func (vpv *VoltPortVnet) WriteToDb(cntx context.Context) {
if vpv.DeleteInProgress {
logger.Warnw(ctx, "Skipping Redis Update for VPV, VPV delete in progress", log.Fields{"Vnet": vpv.VnetName, "Port": vpv.Port})
return
}
- vpv.ForceWriteToDb()
+ vpv.ForceWriteToDb(cntx)
}
//ForceWriteToDb force commit a VPV to the DB
-func (vpv *VoltPortVnet) ForceWriteToDb() {
+func (vpv *VoltPortVnet) ForceWriteToDb(cntx context.Context) {
vpv.PendingFlowLock.RLock()
defer vpv.PendingFlowLock.RUnlock()
vpv.Version = database.PresentVersionMap[database.VpvPath]
if b, err := json.Marshal(vpv); err == nil {
- if err := db.PutVpv(vpv.Port, uint16(vpv.SVlan), uint16(vpv.CVlan), uint16(vpv.UniVlan), string(b)); err != nil {
+ if err := db.PutVpv(cntx, vpv.Port, uint16(vpv.SVlan), uint16(vpv.CVlan), uint16(vpv.UniVlan), string(b)); err != nil {
logger.Warnw(ctx, "VPV write to DB failed", log.Fields{"port": vpv.Port, "SVlan": vpv.SVlan, "CVlan": vpv.CVlan,
"UniVlan": vpv.UniVlan, "Error": err})
}
@@ -2133,24 +2141,24 @@
}
// DelFromDb for deleting from database
-func (vpv *VoltPortVnet) DelFromDb() {
+func (vpv *VoltPortVnet) DelFromDb(cntx context.Context) {
logger.Debugw(ctx, "Deleting VPV from DB", log.Fields{"Port": vpv.Port, "SVLAN": vpv.SVlan, "CVLAN": vpv.CVlan})
- _ = db.DelVpv(vpv.Port, uint16(vpv.SVlan), uint16(vpv.CVlan), uint16(vpv.UniVlan))
+ _ = db.DelVpv(cntx, vpv.Port, uint16(vpv.SVlan), uint16(vpv.CVlan), uint16(vpv.UniVlan))
}
// ClearAllServiceFlags to clear all service flags
-func (vpv *VoltPortVnet) ClearAllServiceFlags() {
- vpv.services.Range(ClearFlagsInService)
+func (vpv *VoltPortVnet) ClearAllServiceFlags(cntx context.Context) {
+ vpv.RangeOnServices(cntx, ClearFlagsInService)
}
// ClearAllVpvFlags to clear all vpv flags
-func (vpv *VoltPortVnet) ClearAllVpvFlags() {
+func (vpv *VoltPortVnet) ClearAllVpvFlags(cntx context.Context) {
vpv.PendingFlowLock.Lock()
vpv.FlowsApplied = false
vpv.IgmpFlowsApplied = false
vpv.PendingDeleteFlow = make(map[string]bool)
vpv.PendingFlowLock.Unlock()
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
logger.Debugw(ctx, "Cleared Flow Flags for VPV",
log.Fields{"device": vpv.Device, "port": vpv.Port,
"svlan": vpv.SVlan, "cvlan": vpv.CVlan, "univlan": vpv.UniVlan})
@@ -2183,9 +2191,9 @@
}
// RestoreVpvsFromDb to restore vpvs from database
-func (va *VoltApplication) RestoreVpvsFromDb() {
+func (va *VoltApplication) RestoreVpvsFromDb(cntx context.Context) {
// VNETS must be learnt first
- vpvs, _ := db.GetVpvs()
+ vpvs, _ := db.GetVpvs(cntx)
for hash, vpv := range vpvs {
b, ok := vpv.Value.([]byte)
if !ok {
@@ -2214,7 +2222,7 @@
}
// AddVnetToPort to add vnet to port
-func (va *VoltApplication) AddVnetToPort(port string, vvnet *VoltVnet, vs *VoltService) *VoltPortVnet {
+func (va *VoltApplication) AddVnetToPort(cntx context.Context, port string, vvnet *VoltVnet, vs *VoltService) *VoltPortVnet {
// The VNET is not on the port and is to be added
logger.Debugw(ctx, "Adding VNET to Port", log.Fields{"Port": port, "VNET": vvnet.Name})
vpv := NewVoltPortVnet(vvnet)
@@ -2234,7 +2242,7 @@
defer vpv.VpvLock.Unlock()
// Add the service that is causing the VNET to be added to the port
- vpv.AddSvc(vs)
+ vpv.AddSvc(cntx, vs)
// Process the PORT UP if the port is already up
d, err := va.GetDeviceFromPort(port)
@@ -2248,17 +2256,17 @@
} else {
logger.Infow(ctx, "Checking UNI port state", log.Fields{"State": p.State})
if d.State == controller.DeviceStateUP && p.State == PortStateUp {
- vpv.PortUpInd(d, port)
+ vpv.PortUpInd(cntx, d, port)
}
}
}
}
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
return vpv
}
// DelVnetFromPort for deleting vnet from port
-func (va *VoltApplication) DelVnetFromPort(port string, vpv *VoltPortVnet) {
+func (va *VoltApplication) DelVnetFromPort(cntx context.Context, port string, vpv *VoltPortVnet) {
//Delete DHCP Session
delDhcpSessions(vpv.LearntMacAddr, vpv.SVlan, vpv.CVlan, vpv.DHCPv6DUID)
@@ -2283,18 +2291,18 @@
vpvs = append(vpvs[0:i], vpvs[i+1:]...)
vpv.DeleteInProgress = true
- vpv.ForceWriteToDb()
+ vpv.ForceWriteToDb(cntx)
va.VnetsByPort.Store(port, vpvs)
- vpv.DelTrapFlows()
- vpv.DelHsiaFlows()
+ vpv.DelTrapFlows(cntx)
+ vpv.DelHsiaFlows(cntx)
va.DisassociateVpvsFromDevice(vpv.Device, vpv)
vpv.PendingFlowLock.RLock()
if len(vpv.PendingDeleteFlow) == 0 {
- vpv.DelFromDb()
+ vpv.DelFromDb(cntx)
}
if vnet := va.GetVnetByName(vpv.VnetName); vnet != nil {
- vnet.disassociatePortFromVnet(vpv.Device, vpv.Port)
+ vnet.disassociatePortFromVnet(cntx, vpv.Device, vpv.Port)
}
vpv.PendingFlowLock.RUnlock()
return
@@ -2303,9 +2311,9 @@
}
// RestoreVnetsFromDb to restore vnet from port
-func (va *VoltApplication) RestoreVnetsFromDb() {
+func (va *VoltApplication) RestoreVnetsFromDb(cntx context.Context) {
// VNETS must be learnt first
- vnets, _ := db.GetVnets()
+ vnets, _ := db.GetVnets(cntx)
for _, net := range vnets {
b, ok := net.Value.([]byte)
if !ok {
@@ -2319,7 +2327,7 @@
continue
}
logger.Debugw(ctx, "Retrieved VNET", log.Fields{"VNET": vnet.VnetConfig})
- if err := va.AddVnet(vnet.VnetConfig, &vnet.VnetOper); err != nil {
+ if err := va.AddVnet(cntx, vnet.VnetConfig, &vnet.VnetOper); err != nil {
logger.Warnw(ctx, "Add Vnet Failed", log.Fields{"Config": vnet.VnetConfig, "Error": err})
}
@@ -2454,7 +2462,7 @@
}
// PushDevFlowForVlan to push icmpv6 flows for vlan
-func (va *VoltApplication) PushDevFlowForVlan(vnet *VoltVnet) {
+func (va *VoltApplication) PushDevFlowForVlan(cntx context.Context, vnet *VoltVnet) {
logger.Infow(ctx, "PushDevFlowForVlan", log.Fields{"SVlan": vnet.SVlan, "CVlan": vnet.CVlan})
pushflow := func(key interface{}, value interface{}) bool {
device := value.(*VoltDevice)
@@ -2492,7 +2500,7 @@
//Pushing ICMPv6 Flow
flow := BuildICMPv6Flow(portID, vnet)
- err = cntlr.GetController().AddFlows(device.NniPort, device.Name, flow)
+ err = cntlr.GetController().AddFlows(cntx, device.NniPort, device.Name, flow)
if err != nil {
logger.Warnw(ctx, "Configuring ICMPv6 Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2501,7 +2509,7 @@
// Pushing ARP Flow
flow = BuildDSArpFlow(portID, vnet)
- err = cntlr.GetController().AddFlows(device.NniPort, device.Name, flow)
+ err = cntlr.GetController().AddFlows(cntx, device.NniPort, device.Name, flow)
if err != nil {
logger.Warnw(ctx, "Configuring ARP Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2518,7 +2526,7 @@
}
// PushDevFlowForDevice to push icmpv6 flows for device
-func (va *VoltApplication) PushDevFlowForDevice(device *VoltDevice) {
+func (va *VoltApplication) PushDevFlowForDevice(cntx context.Context, device *VoltDevice) {
logger.Infow(ctx, "PushDevFlowForDevice", log.Fields{"device": device})
logger.Debugw(ctx, "Configuring ICMPv6 Group for device ", log.Fields{"Device": device.Name})
@@ -2545,7 +2553,7 @@
return true
}
flow := BuildICMPv6Flow(nniPortID, vnet)
- err = cntlr.GetController().AddFlows(device.NniPort, device.Name, flow)
+ err = cntlr.GetController().AddFlows(cntx, device.NniPort, device.Name, flow)
if err != nil {
logger.Warnw(ctx, "Configuring ICMPv6 Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2553,7 +2561,7 @@
logger.Infow(ctx, "ICMP Flow Added to Queue", log.Fields{"flow": flow})
flow = BuildDSArpFlow(nniPortID, vnet)
- err = cntlr.GetController().AddFlows(device.NniPort, device.Name, flow)
+ err = cntlr.GetController().AddFlows(cntx, device.NniPort, device.Name, flow)
if err != nil {
logger.Warnw(ctx, "Configuring ARP Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2569,7 +2577,7 @@
}
// DeleteDevFlowForVlan to delete icmpv6 flow for vlan
-func (va *VoltApplication) DeleteDevFlowForVlan(vnet *VoltVnet) {
+func (va *VoltApplication) DeleteDevFlowForVlan(cntx context.Context, vnet *VoltVnet) {
logger.Infow(ctx, "DeleteDevFlowForVlan", log.Fields{"SVlan": vnet.SVlan, "CVlan": vnet.CVlan})
delflows := func(key interface{}, value interface{}) bool {
device := value.(*VoltDevice)
@@ -2591,7 +2599,7 @@
//Pushing ICMPv6 Flow
flow := BuildICMPv6Flow(portID, vnet)
flow.ForceAction = true
- err := vnet.RemoveFlows(device, flow)
+ err := vnet.RemoveFlows(cntx, device, flow)
if err != nil {
logger.Warnw(ctx, "De-Configuring ICMPv6 Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2601,7 +2609,7 @@
//Pushing ARP Flow
flow = BuildDSArpFlow(portID, vnet)
flow.ForceAction = true
- err = vnet.RemoveFlows(device, flow)
+ err = vnet.RemoveFlows(cntx, device, flow)
if err != nil {
logger.Warnw(ctx, "De-Configuring ARP Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2616,7 +2624,7 @@
}
// DeleteDevFlowForDevice to delete icmpv6 flow for device
-func (va *VoltApplication) DeleteDevFlowForDevice(device *VoltDevice) {
+func (va *VoltApplication) DeleteDevFlowForDevice(cntx context.Context, device *VoltDevice) {
logger.Infow(ctx, "DeleteDevFlowForDevice", log.Fields{"Device": device})
delicmpv6 := func(key, value interface{}) bool {
vnet := value.(*VoltVnet)
@@ -2638,7 +2646,7 @@
}
flow := BuildICMPv6Flow(nniPortID, vnet)
flow.ForceAction = true
- err = vnet.RemoveFlows(device, flow)
+ err = vnet.RemoveFlows(cntx, device, flow)
if err != nil {
logger.Warnw(ctx, "De-Configuring ICMPv6 Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2646,7 +2654,7 @@
flow = BuildDSArpFlow(nniPortID, vnet)
flow.ForceAction = true
- err = vnet.RemoveFlows(device, flow)
+ err = vnet.RemoveFlows(cntx, device, flow)
if err != nil {
logger.Warnw(ctx, "De-Configuring ARP Flow for device failed ", log.Fields{"Device": device.Name, "err": err})
return true
@@ -2666,7 +2674,7 @@
}
// DeleteDevFlowForVlanFromDevice to delete icmpv6 flow for vlan from device
-func (va *VoltApplication) DeleteDevFlowForVlanFromDevice(vnet *VoltVnet, deviceSerialNum string) {
+func (va *VoltApplication) DeleteDevFlowForVlanFromDevice(cntx context.Context, vnet *VoltVnet, deviceSerialNum string) {
logger.Infow(ctx, "DeleteDevFlowForVlanFromDevice", log.Fields{"Device-serialNum": deviceSerialNum, "SVlan": vnet.SVlan, "CVlan": vnet.CVlan})
delflows := func(key interface{}, value interface{}) bool {
device := value.(*VoltDevice)
@@ -2701,14 +2709,14 @@
}
flow := BuildICMPv6Flow(portID, vnet)
flow.ForceAction = true
- if err := vnet.RemoveFlows(device, flow); err != nil {
+ if err := vnet.RemoveFlows(cntx, device, flow); err != nil {
logger.Warnw(ctx, "Delete Flow Failed", log.Fields{"Device": device, "Flow": flow, "Error": err})
}
logger.Infow(ctx, "ICMP Flow Delete Added to Queue", log.Fields{"flow": flow})
flow = BuildDSArpFlow(portID, vnet)
flow.ForceAction = true
- if err := vnet.RemoveFlows(device, flow); err != nil {
+ if err := vnet.RemoveFlows(cntx, device, flow); err != nil {
logger.Warnw(ctx, "Delete Flow Failed", log.Fields{"Device": device, "Flow": flow, "Error": err})
}
logger.Infow(ctx, "ARP Flow Delete Added to Queue", log.Fields{"flow": flow})
@@ -2859,7 +2867,7 @@
}
//PushFlows - Triggers flow addition after registering for flow indication event
-func (vpv *VoltPortVnet) PushFlows(device *VoltDevice, flow *of.VoltFlow) error {
+func (vpv *VoltPortVnet) PushFlows(cntx context.Context, device *VoltDevice, flow *of.VoltFlow) error {
for cookie := range flow.SubFlows {
cookie := strconv.FormatUint(cookie, 10)
@@ -2870,7 +2878,7 @@
}
device.RegisterFlowAddEvent(cookie, fe)
}
- return cntlr.GetController().AddFlows(vpv.Port, device.Name, flow)
+ return cntlr.GetController().AddFlows(cntx, vpv.Port, device.Name, flow)
}
//FlowInstallFailure - Process flow failure indication and triggers HSIA failure for all associated services
@@ -2886,7 +2894,7 @@
}
//RemoveFlows - Triggers flow deletion after registering for flow indication event
-func (vpv *VoltPortVnet) RemoveFlows(device *VoltDevice, flow *of.VoltFlow) error {
+func (vpv *VoltPortVnet) RemoveFlows(cntx context.Context, device *VoltDevice, flow *of.VoltFlow) error {
vpv.PendingFlowLock.Lock()
defer vpv.PendingFlowLock.Unlock()
@@ -2902,11 +2910,11 @@
device.RegisterFlowDelEvent(cookie, fe)
vpv.PendingDeleteFlow[cookie] = true
}
- return cntlr.GetController().DelFlows(vpv.Port, device.Name, flow)
+ return cntlr.GetController().DelFlows(cntx, vpv.Port, device.Name, flow)
}
//CheckAndDeleteVpv - remove VPV from DB is there are no pending flows to be removed
-func (vpv *VoltPortVnet) CheckAndDeleteVpv() {
+func (vpv *VoltPortVnet) CheckAndDeleteVpv(cntx context.Context) {
vpv.PendingFlowLock.RLock()
defer vpv.PendingFlowLock.RUnlock()
if !vpv.DeleteInProgress {
@@ -2914,24 +2922,24 @@
}
if len(vpv.PendingDeleteFlow) == 0 && !vpv.FlowsApplied {
logger.Infow(ctx, "All Flows removed for VPV. Triggering VPV Deletion from DB", log.Fields{"VPV Port": vpv.Port, "Device": vpv.Device, "Vnet": vpv.VnetName})
- vpv.DelFromDb()
+ vpv.DelFromDb(cntx)
logger.Infow(ctx, "Deleted VPV from DB/Cache successfully", log.Fields{"VPV Port": vpv.Port, "Device": vpv.Device, "Vnet": vpv.VnetName})
}
}
//FlowRemoveSuccess - Process flow success indication
-func (vpv *VoltPortVnet) FlowRemoveSuccess(cookie string, device string) {
+func (vpv *VoltPortVnet) FlowRemoveSuccess(cntx context.Context, cookie string, device string) {
vpv.PendingFlowLock.Lock()
logger.Infow(ctx, "VPV Flow Remove Success Notification", log.Fields{"Port": vpv.Port, "Cookie": cookie, "Device": device})
delete(vpv.PendingDeleteFlow, cookie)
vpv.PendingFlowLock.Unlock()
- vpv.CheckAndDeleteVpv()
- vpv.WriteToDb()
+ vpv.CheckAndDeleteVpv(cntx)
+ vpv.WriteToDb(cntx)
}
//FlowRemoveFailure - Process flow failure indication and triggers Del HSIA failure for all associated services
-func (vpv *VoltPortVnet) FlowRemoveFailure(cookie string, device string, errorCode uint32, errReason string) {
+func (vpv *VoltPortVnet) FlowRemoveFailure(cntx context.Context, cookie string, device string, errorCode uint32, errReason string) {
vpv.PendingFlowLock.Lock()
logger.Errorw(ctx, "VPV Flow Remove Failure Notification", log.Fields{"Port": vpv.Port, "Cookie": cookie, "ErrorCode": errorCode, "ErrorReason": errReason, "Device": device})
@@ -2947,15 +2955,15 @@
if vpv.DeleteInProgress {
delete(vpv.PendingDeleteFlow, cookie)
vpv.PendingFlowLock.Unlock()
- vpv.CheckAndDeleteVpv()
+ vpv.CheckAndDeleteVpv(cntx)
} else {
vpv.PendingFlowLock.Unlock()
- vpv.WriteToDb()
+ vpv.WriteToDb(cntx)
}
}
//RemoveFlows - Triggers flow deletion after registering for flow indication event
-func (vv *VoltVnet) RemoveFlows(device *VoltDevice, flow *of.VoltFlow) error {
+func (vv *VoltVnet) RemoveFlows(cntx context.Context, device *VoltDevice, flow *of.VoltFlow) error {
vv.VnetLock.Lock()
defer vv.VnetLock.Unlock()
@@ -2978,12 +2986,12 @@
flowMap[cookie] = true
vv.PendingDeleteFlow[device.Name] = flowMap
}
- vv.WriteToDb()
- return cntlr.GetController().DelFlows(device.NniPort, device.Name, flow)
+ vv.WriteToDb(cntx)
+ return cntlr.GetController().DelFlows(cntx, device.NniPort, device.Name, flow)
}
//CheckAndDeleteVnet - remove Vnet from DB is there are no pending flows to be removed
-func (vv *VoltVnet) CheckAndDeleteVnet(device string) {
+func (vv *VoltVnet) CheckAndDeleteVnet(cntx context.Context, device string) {
if !vv.DeleteInProgress {
return
}
@@ -2991,7 +2999,7 @@
if len(vv.PendingDeleteFlow[device]) == 0 && !vv.isAssociatedPortsPresent() {
logger.Warnw(ctx, "Deleting Vnet : All flows removed", log.Fields{"Name": vv.Name, "AssociatedPorts": vv.AssociatedPorts, "Device": device})
GetApplication().deleteVnetConfig(vv)
- _ = db.DelVnet(vv.Name)
+ _ = db.DelVnet(cntx, vv.Name)
logger.Infow(ctx, "Deleted Vnet from DB/Cache successfully", log.Fields{"Device": device, "Vnet": vv.Name})
} else {
logger.Warnw(ctx, "Skipping Del Vnet", log.Fields{"Name": vv.Name, "AssociatedPorts": vv.AssociatedPorts, "PendingDelFlows": vv.PendingDeleteFlow[device]})
@@ -3000,7 +3008,7 @@
}
//FlowRemoveSuccess - Process flow success indication
-func (vv *VoltVnet) FlowRemoveSuccess(cookie string, device string) {
+func (vv *VoltVnet) FlowRemoveSuccess(cntx context.Context, cookie string, device string) {
vv.VnetLock.Lock()
defer vv.VnetLock.Unlock()
@@ -3014,14 +3022,14 @@
if d := GetApplication().GetDevice(device); d != nil {
_, present := d.ConfiguredVlanForDeviceFlows.Get(VnetKey(vv.SVlan, vv.CVlan, 0))
if !present && len(vv.PendingDeleteFlow[device]) == 0 {
- vv.CheckAndDeleteVnet(device)
+ vv.CheckAndDeleteVnet(cntx, device)
}
}
- vv.WriteToDb()
+ vv.WriteToDb(cntx)
}
//FlowRemoveFailure - Process flow failure indication
-func (vv *VoltVnet) FlowRemoveFailure(cookie string, device string, errorCode uint32, errReason string) {
+func (vv *VoltVnet) FlowRemoveFailure(cntx context.Context, cookie string, device string, errorCode uint32, errReason string) {
vv.VnetLock.Lock()
defer vv.VnetLock.Unlock()
@@ -3032,7 +3040,7 @@
if vv.DeleteInProgress {
delete(vv.PendingDeleteFlow[device], cookie)
- vv.CheckAndDeleteVnet(device)
+ vv.CheckAndDeleteVnet(cntx, device)
}
return
}
@@ -3107,7 +3115,7 @@
}
//TriggerAssociatedFlowDelete - Re-trigger delete for pending delete flows
-func (vv *VoltVnet) TriggerAssociatedFlowDelete(device string) bool {
+func (vv *VoltVnet) TriggerAssociatedFlowDelete(cntx context.Context, device string) bool {
vv.VnetLock.Lock()
cookieList := []uint64{}
flowMap := vv.PendingDeleteFlow[device]
@@ -3129,7 +3137,7 @@
subFlow.Cookie = cookie
flow.SubFlows[cookie] = subFlow
logger.Infow(ctx, "Retriggering Vnet Delete Flow", log.Fields{"Device": device, "Vnet": vv.Name, "Cookie": cookie})
- if err := vv.RemoveFlows(vd, flow); err != nil {
+ if err := vv.RemoveFlows(cntx, vd, flow); err != nil {
logger.Warnw(ctx, "Vnet Delete Flow Failed", log.Fields{"Device": device, "Vnet": vv.Name, "Cookie": cookie, "Error": err})
}
}