blob: b22770caf641b1f5052921b0fd5ed1d8e768ad77 [file] [log] [blame]
Naveen Sampath04696f72022-06-13 15:19:14 +05301/*
2* Copyright 2022-present Open Networking Foundation
3* Licensed under the Apache License, Version 2.0 (the "License");
4* you may not use this file except in compliance with the License.
5* You may obtain a copy of the License at
6*
7* http://www.apache.org/licenses/LICENSE-2.0
8*
9* Unless required by applicable law or agreed to in writing, software
10* distributed under the License is distributed on an "AS IS" BASIS,
11* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12* See the License for the specific language governing permissions and
13* limitations under the License.
vinokuma926cb3e2023-03-29 11:41:06 +053014 */
Naveen Sampath04696f72022-06-13 15:19:14 +053015
16package controller
17
18import (
19 "context"
20 "strconv"
21 "time"
22
23 "voltha-go-controller/internal/pkg/intf"
24 "voltha-go-controller/internal/pkg/of"
25 "voltha-go-controller/internal/pkg/tasks"
26 "voltha-go-controller/internal/pkg/util"
Tinoj Joseph1d108322022-07-13 10:07:39 +053027 "voltha-go-controller/log"
vinokuma926cb3e2023-03-29 11:41:06 +053028
Naveen Sampath04696f72022-06-13 15:19:14 +053029 "github.com/opencord/voltha-protos/v5/go/common"
30 ofp "github.com/opencord/voltha-protos/v5/go/openflow_13"
31 "github.com/opencord/voltha-protos/v5/go/voltha"
32)
33
34var (
35 rcvdGroups map[uint32]*ofp.OfpGroupDesc
36 groupsToAdd []*of.Group
37 groupsToMod []*of.Group
38)
39
40// AuditTablesTask structure
41type AuditTablesTask struct {
Naveen Sampath04696f72022-06-13 15:19:14 +053042 ctx context.Context
43 device *Device
Naveen Sampath04696f72022-06-13 15:19:14 +053044 timestamp string
vinokuma926cb3e2023-03-29 11:41:06 +053045 taskID uint8
46 stop bool
Naveen Sampath04696f72022-06-13 15:19:14 +053047}
48
49// NewAuditTablesTask is constructor for AuditTablesTask
50func NewAuditTablesTask(device *Device) *AuditTablesTask {
51 var att AuditTablesTask
52 att.device = device
53 att.stop = false
54 tstamp := (time.Now()).Format(time.RFC3339Nano)
55 att.timestamp = tstamp
56 return &att
57}
58
59// Name returns name of the task
60func (att *AuditTablesTask) Name() string {
61 return "Audit Table Task"
62}
63
64// TaskID to return task id of the task
65func (att *AuditTablesTask) TaskID() uint8 {
66 return att.taskID
67}
68
69// Timestamp to return timestamp for the task
70func (att *AuditTablesTask) Timestamp() string {
71 return att.timestamp
72}
73
74// Stop to stop the task
75func (att *AuditTablesTask) Stop() {
76 att.stop = true
77}
78
79// Start is called by the framework and is responsible for implementing
80// the actual task.
81func (att *AuditTablesTask) Start(ctx context.Context, taskID uint8) error {
Akash Soni6168f312023-05-18 20:57:33 +053082 logger.Infow(ctx, "Audit Table Task Triggered", log.Fields{"Context": ctx, "taskId": taskID, "Device": att.device.ID})
Naveen Sampath04696f72022-06-13 15:19:14 +053083 att.taskID = taskID
84 att.ctx = ctx
85 var errInfo error
86 var err error
87
Tinoj Josephaf37ce82022-12-28 11:59:43 +053088 // Audit ports
89 if err = att.AuditPorts(); err != nil {
90 logger.Errorw(ctx, "Audit Ports Failed", log.Fields{"Reason": err.Error()})
91 errInfo = err
92 }
93
Naveen Sampath04696f72022-06-13 15:19:14 +053094 // Audit the meters
95 if err = att.AuditMeters(); err != nil {
96 logger.Errorw(ctx, "Audit Meters Failed", log.Fields{"Reason": err.Error()})
97 errInfo = err
98 }
99
100 // Audit the Groups
101 if rcvdGroups, err = att.AuditGroups(); err != nil {
102 logger.Errorw(ctx, "Audit Groups Failed", log.Fields{"Reason": err.Error()})
103 errInfo = err
104 }
105
106 // Audit the flows
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530107 if err = att.AuditFlows(ctx); err != nil {
Naveen Sampath04696f72022-06-13 15:19:14 +0530108 logger.Errorw(ctx, "Audit Flows Failed", log.Fields{"Reason": err.Error()})
109 errInfo = err
110 }
111
112 // Triggering deletion of excess groups from device after the corresponding flows are removed
113 // to avoid flow dependency error during group deletion
Akash Soni6168f312023-05-18 20:57:33 +0530114 logger.Debugw(ctx, "Excess Groups", log.Fields{"Groups": rcvdGroups})
Naveen Sampath04696f72022-06-13 15:19:14 +0530115 att.DelExcessGroups(rcvdGroups)
Akash Soni6168f312023-05-18 20:57:33 +0530116 logger.Infow(ctx, "Audit Table Task Completed", log.Fields{"Context": ctx, "taskId": taskID, "Device": att.device.ID})
Naveen Sampath04696f72022-06-13 15:19:14 +0530117 return errInfo
Naveen Sampath04696f72022-06-13 15:19:14 +0530118}
119
120// AuditMeters : Audit the meters which includes fetching the existing meters at the
121// voltha and identifying the delta between the ones held here and the
122// ones held at VOLTHA. The delta must be cleaned up to keep both the
123// components in sync
124func (att *AuditTablesTask) AuditMeters() error {
Naveen Sampath04696f72022-06-13 15:19:14 +0530125 if att.stop {
126 return tasks.ErrTaskCancelError
127 }
128 var vc voltha.VolthaServiceClient
129 if vc = att.device.VolthaClient(); vc == nil {
130 logger.Error(ctx, "Fetch Device Meters Failed: Voltha Client Unavailable")
131 return nil
132 }
133
134 //-----------------------------
135 // Perform the audit of meters
136 // Fetch the meters
137 ms, err := vc.ListLogicalDeviceMeters(att.ctx, &voltha.ID{Id: att.device.ID})
138 if err != nil {
139 logger.Warnw(ctx, "Audit of flows failed", log.Fields{"Reason": err.Error()})
140 return err
141 }
142
143 // Build the map for easy and faster processing
144 rcvdMeters := make(map[uint32]*ofp.OfpMeterStats)
145 for _, m := range ms.Items {
146 rcvdMeters[m.Stats.MeterId] = m.Stats
147 }
148
149 // Verify all meters that are in the controller but not in the device
150 missingMeters := []*of.Meter{}
151 for _, meter := range att.device.meters {
Naveen Sampath04696f72022-06-13 15:19:14 +0530152 if att.stop {
153 break
154 }
155 logger.Debugw(ctx, "Auditing Meter", log.Fields{"Id": meter.ID})
156
157 if _, ok := rcvdMeters[meter.ID]; ok {
158 // The meter exists in the device too. Just remove it from
159 // the received meters
160 delete(rcvdMeters, meter.ID)
161 } else {
162 // The flow exists at the controller but not at the device
163 // Push the flow to the device
164 logger.Debugw(ctx, "Adding Meter To Missing Meters", log.Fields{"Id": meter.ID})
165 missingMeters = append(missingMeters, meter)
166 }
167 }
168 if !att.stop {
169 att.AddMissingMeters(missingMeters)
170 att.DelExcessMeters(rcvdMeters)
171 } else {
172 err = tasks.ErrTaskCancelError
173 }
174 return err
175}
176
177// AddMissingMeters adds the missing meters detected by AuditMeters
178func (att *AuditTablesTask) AddMissingMeters(meters []*of.Meter) {
179 logger.Debugw(ctx, "Adding missing meters", log.Fields{"Number": len(meters)})
180 for _, meter := range meters {
181 meterMod, err := of.MeterUpdate(att.device.ID, of.MeterCommandAdd, meter)
182 if err != nil {
183 logger.Errorw(ctx, "Update Meter Table Failed", log.Fields{"Reason": err.Error()})
184 continue
185 }
186 if vc := att.device.VolthaClient(); vc != nil {
187 if _, err = vc.UpdateLogicalDeviceMeterTable(att.ctx, meterMod); err != nil {
188 logger.Errorw(ctx, "Update Meter Table Failed", log.Fields{"Reason": err.Error()})
189 }
190 } else {
191 logger.Error(ctx, "Update Meter Table Failed: Voltha Client Unavailable")
192 }
193 }
194}
195
196// DelExcessMeters to delete excess meters
197func (att *AuditTablesTask) DelExcessMeters(meters map[uint32]*ofp.OfpMeterStats) {
198 logger.Debugw(ctx, "Deleting Excess Meters", log.Fields{"Number": len(meters)})
199 for _, meter := range meters {
200 meterMod := &ofp.OfpMeterMod{}
201 meterMod.Command = ofp.OfpMeterModCommand_OFPMC_DELETE
202 meterMod.MeterId = meter.MeterId
203 meterUpd := &ofp.MeterModUpdate{Id: att.device.ID, MeterMod: meterMod}
204 if vc := att.device.VolthaClient(); vc != nil {
205 if _, err := vc.UpdateLogicalDeviceMeterTable(att.ctx, meterUpd); err != nil {
206 logger.Errorw(ctx, "Update Meter Table Failed", log.Fields{"Reason": err.Error()})
207 }
208 } else {
209 logger.Error(ctx, "Update Meter Table Failed: Voltha Client Unavailable")
210 }
211 }
212}
213
214// AuditFlows audit the flows which includes fetching the existing meters at the
215// voltha and identifying the delta between the ones held here and the
216// ones held at VOLTHA. The delta must be cleaned up to keep both the
217// components in sync
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530218func (att *AuditTablesTask) AuditFlows(cntx context.Context) error {
Naveen Sampath04696f72022-06-13 15:19:14 +0530219 if att.stop {
220 return tasks.ErrTaskCancelError
221 }
222
223 var vc voltha.VolthaServiceClient
224 if vc = att.device.VolthaClient(); vc == nil {
225 logger.Error(ctx, "Flow Audit Failed: Voltha Client Unavailable")
226 return nil
227 }
228
229 // ---------------------------------
230 // Perform the audit of flows first
231 // Retrieve the flows from the device
232 f, err := vc.ListLogicalDeviceFlows(att.ctx, &common.ID{Id: att.device.ID})
233 if err != nil {
234 logger.Warnw(ctx, "Audit of flows failed", log.Fields{"Reason": err.Error()})
235 return err
236 }
237
238 defaultSuccessFlowStatus := intf.FlowStatus{
239 Device: att.device.ID,
240 FlowModType: of.CommandAdd,
241 Status: 0,
242 Reason: "",
243 }
244
245 // Build the map for easy and faster processing
246 rcvdFlows := make(map[uint64]*ofp.OfpFlowStats)
247 flowsToAdd := &of.VoltFlow{}
248 flowsToAdd.SubFlows = make(map[uint64]*of.VoltSubFlow)
249 for _, flow := range f.Items {
250 rcvdFlows[flow.Cookie] = flow
251 }
252
253 att.device.flowLock.Lock()
254 // Verify all flows that are in the controller but not in the device
255 for _, flow := range att.device.flows {
Naveen Sampath04696f72022-06-13 15:19:14 +0530256 if att.stop {
257 break
258 }
259
260 logger.Debugw(ctx, "Auditing Flow", log.Fields{"Cookie": flow.Cookie})
261 if _, ok := rcvdFlows[flow.Cookie]; ok {
262 // The flow exists in the device too. Just remove it from
263 // the received flows & trigger flow success indication unless
264 // the flow in del failure/pending state
265
266 if flow.State != of.FlowDelFailure && flow.State != of.FlowDelPending {
267 delete(rcvdFlows, flow.Cookie)
268 }
269 defaultSuccessFlowStatus.Cookie = strconv.FormatUint(flow.Cookie, 10)
270
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530271 GetController().ProcessFlowModResultIndication(cntx, defaultSuccessFlowStatus)
Naveen Sampath04696f72022-06-13 15:19:14 +0530272 } else {
273 // The flow exists at the controller but not at the device
274 // Push the flow to the device
275 logger.Debugw(ctx, "Adding Flow To Missing Flows", log.Fields{"Cookie": flow.Cookie})
276 flowsToAdd.SubFlows[flow.Cookie] = flow
277 }
278 }
279 att.device.flowLock.Unlock()
280
281 if !att.stop {
282 // The flows remaining in the received flows are the excess flows at
283 // the device. Delete those flows
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530284 att.DelExcessFlows(cntx, rcvdFlows)
Naveen Sampath04696f72022-06-13 15:19:14 +0530285 // Add the flows missing at the device
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530286 att.AddMissingFlows(cntx, flowsToAdd)
Naveen Sampath04696f72022-06-13 15:19:14 +0530287 } else {
288 err = tasks.ErrTaskCancelError
289 }
290 return err
291}
292
293// AddMissingFlows : The flows missing from the device are reinstalled att the audit
294// The flows are added into a VoltFlow structure.
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530295func (att *AuditTablesTask) AddMissingFlows(cntx context.Context, mflow *of.VoltFlow) {
Naveen Sampath04696f72022-06-13 15:19:14 +0530296 logger.Debugw(ctx, "Add Missing Flows", log.Fields{"Number": len(mflow.SubFlows)})
297 mflow.Command = of.CommandAdd
298 ofFlows := of.ProcessVoltFlow(att.device.ID, mflow.Command, mflow.SubFlows)
299 var vc voltha.VolthaServiceClient
300 var bwConsumedInfo of.BwAvailDetails
301 if vc = att.device.VolthaClient(); vc == nil {
302 logger.Error(ctx, "Update Flow Table Failed: Voltha Client Unavailable")
303 return
304 }
305 for _, flow := range ofFlows {
306 var dbFlow *of.VoltSubFlow
307 var present bool
308 if flow.FlowMod != nil {
309 if dbFlow, present = att.device.GetFlow(flow.FlowMod.Cookie); !present {
Tinoj Joseph1d108322022-07-13 10:07:39 +0530310 logger.Warnw(ctx, "Flow Removed from DB. Ignoring Add Missing Flow", log.Fields{"Device": att.device.ID, "Cookie": flow.FlowMod.Cookie})
Naveen Sampath04696f72022-06-13 15:19:14 +0530311 continue
312 }
313 }
314 var err error
315 if _, err = vc.UpdateLogicalDeviceFlowTable(att.ctx, flow); err != nil {
316 logger.Errorw(ctx, "Update Flow Table Failed", log.Fields{"Reason": err.Error()})
317 }
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530318 att.device.triggerFlowResultNotification(cntx, flow.FlowMod.Cookie, dbFlow, of.CommandAdd, bwConsumedInfo, err)
Naveen Sampath04696f72022-06-13 15:19:14 +0530319 }
320}
321
322// DelExcessFlows delete the excess flows held at the VOLTHA
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530323func (att *AuditTablesTask) DelExcessFlows(cntx context.Context, flows map[uint64]*ofp.OfpFlowStats) {
Naveen Sampath04696f72022-06-13 15:19:14 +0530324 logger.Debugw(ctx, "Deleting Excess Flows", log.Fields{"Number of Flows": len(flows)})
325
326 var vc voltha.VolthaServiceClient
327 if vc = att.device.VolthaClient(); vc == nil {
328 logger.Error(ctx, "Delete Excess Flows Failed: Voltha Client Unavailable")
329 return
330 }
331
332 // Let's cycle through the flows to delete the excess flows
333 for _, flow := range flows {
Naveen Sampath04696f72022-06-13 15:19:14 +0530334 if _, present := att.device.GetFlow(flow.Cookie); present {
Tinoj Joseph1d108322022-07-13 10:07:39 +0530335 logger.Warnw(ctx, "Flow Present in DB. Ignoring Delete Excess Flow", log.Fields{"Device": att.device.ID, "Cookie": flow.Cookie})
Naveen Sampath04696f72022-06-13 15:19:14 +0530336 continue
337 }
338
339 logger.Debugw(ctx, "Deleting Flow", log.Fields{"Cookie": flow.Cookie})
340 // Create the flowMod structure and fill it out
341 flowMod := &ofp.OfpFlowMod{}
342 flowMod.Cookie = flow.Cookie
343 flowMod.TableId = flow.TableId
344 flowMod.Command = ofp.OfpFlowModCommand_OFPFC_DELETE_STRICT
345 flowMod.IdleTimeout = flow.IdleTimeout
346 flowMod.HardTimeout = flow.HardTimeout
347 flowMod.Priority = flow.Priority
348 flowMod.BufferId = of.DefaultBufferID
349 flowMod.OutPort = of.DefaultOutPort
350 flowMod.OutGroup = of.DefaultOutGroup
351 flowMod.Flags = flow.Flags
352 flowMod.Match = flow.Match
353 flowMod.Instructions = flow.Instructions
354
355 // Create FlowTableUpdate
356 flowUpdate := &ofp.FlowTableUpdate{
357 Id: att.device.ID,
358 FlowMod: flowMod,
359 }
360
361 var err error
362 if _, err = vc.UpdateLogicalDeviceFlowTable(att.ctx, flowUpdate); err != nil {
363 logger.Errorw(ctx, "Flow Audit Delete Failed", log.Fields{"Reason": err.Error()})
364 }
Tinoj Joseph07cc5372022-07-18 22:53:51 +0530365 att.device.triggerFlowResultNotification(cntx, flow.Cookie, nil, of.CommandDel, of.BwAvailDetails{}, err)
Naveen Sampath04696f72022-06-13 15:19:14 +0530366 }
367}
368
369// AuditGroups audit the groups which includes fetching the existing groups at the
370// voltha and identifying the delta between the ones held here and the
371// ones held at VOLTHA. The delta must be cleaned up to keep both the
372// components in sync
373func (att *AuditTablesTask) AuditGroups() (map[uint32]*ofp.OfpGroupDesc, error) {
Naveen Sampath04696f72022-06-13 15:19:14 +0530374 // Build the map for easy and faster processing
375 rcvdGroups = make(map[uint32]*ofp.OfpGroupDesc)
376
377 if att.stop {
378 return rcvdGroups, tasks.ErrTaskCancelError
379 }
380
381 var vc voltha.VolthaServiceClient
382 if vc = att.device.VolthaClient(); vc == nil {
383 logger.Error(ctx, "Group Audit Failed: Voltha Client Unavailable")
384 return rcvdGroups, nil
385 }
386
387 // ---------------------------------
388 // Perform the audit of groups first
389 // Retrieve the groups from the device
390 g, err := vc.ListLogicalDeviceFlowGroups(att.ctx, &common.ID{Id: att.device.ID})
391 if err != nil {
392 logger.Warnw(ctx, "Audit of groups failed", log.Fields{"Reason": err.Error()})
393 return rcvdGroups, err
394 }
395
396 groupsToAdd = []*of.Group{}
397 groupsToMod = []*of.Group{}
398 for _, group := range g.Items {
399 rcvdGroups[group.Desc.GroupId] = group.Desc
400 }
401 logger.Infow(ctx, "Received Groups", log.Fields{"Groups": rcvdGroups})
402
403 // Verify all groups that are in the controller but not in the device
404 att.device.groups.Range(att.compareGroupEntries)
405
406 if !att.stop {
407 // Add the groups missing at the device
408 logger.Infow(ctx, "Missing Groups", log.Fields{"Groups": groupsToAdd})
409 att.AddMissingGroups(groupsToAdd)
410
411 // Update groups with group member mismatch
412 logger.Infow(ctx, "Modify Groups", log.Fields{"Groups": groupsToMod})
413 att.UpdateMismatchGroups(groupsToMod)
414
415 // Note: Excess groups will be deleted after ensuring the connected
416 // flows are also removed as part fo audit flows
417 } else {
418 err = tasks.ErrTaskCancelError
419 }
420 // The groups remaining in the received groups are the excess groups at
421 // the device
422 return rcvdGroups, err
423}
424
425// compareGroupEntries to compare the group entries
426func (att *AuditTablesTask) compareGroupEntries(key, value interface{}) bool {
Naveen Sampath04696f72022-06-13 15:19:14 +0530427 if att.stop {
428 return false
429 }
430
431 groupID := key.(uint32)
432 dbGroup := value.(*of.Group)
433 logger.Debugw(ctx, "Auditing Group", log.Fields{"Groupid": groupID})
434 if rcvdGrp, ok := rcvdGroups[groupID]; ok {
435 // The group exists in the device too.
436 // Compare the group members and add to modify list if required
437 compareGroupMembers(dbGroup, rcvdGrp)
438 delete(rcvdGroups, groupID)
439 } else {
440 // The group exists at the controller but not at the device
441 // Push the group to the device
442 logger.Debugw(ctx, "Adding Group To Missing Groups", log.Fields{"GroupId": groupID})
443 groupsToAdd = append(groupsToAdd, value.(*of.Group))
444 }
445 return true
446}
447
448func compareGroupMembers(refGroup *of.Group, rcvdGroup *ofp.OfpGroupDesc) {
Naveen Sampath04696f72022-06-13 15:19:14 +0530449 portList := []uint32{}
450 refPortList := []uint32{}
451
vinokuma926cb3e2023-03-29 11:41:06 +0530452 // Collect port list from response Group Mod structure
453 // If PON is configured even for one group, then only PON shall be considered for compared for all groups
Naveen Sampath04696f72022-06-13 15:19:14 +0530454 for _, bucket := range rcvdGroup.Buckets {
455 for _, actionBucket := range bucket.Actions {
456 if actionBucket.Type == ofp.OfpActionType_OFPAT_OUTPUT {
457 action := actionBucket.GetOutput()
458 portList = append(portList, action.Port)
459 }
460 }
461 }
462
463 refPortList = append(refPortList, refGroup.Buckets...)
464
vinokuma926cb3e2023-03-29 11:41:06 +0530465 // Is port list differs, trigger group update
Naveen Sampath04696f72022-06-13 15:19:14 +0530466 if !util.IsSliceSame(refPortList, portList) {
467 groupsToMod = append(groupsToMod, refGroup)
468 }
469}
470
vinokuma926cb3e2023-03-29 11:41:06 +0530471// AddMissingGroups - addmissing groups to Voltha
Naveen Sampath04696f72022-06-13 15:19:14 +0530472func (att *AuditTablesTask) AddMissingGroups(groupList []*of.Group) {
473 att.PushGroups(groupList, of.GroupCommandAdd)
474}
475
vinokuma926cb3e2023-03-29 11:41:06 +0530476// UpdateMismatchGroups - updates mismatched groups to Voltha
Naveen Sampath04696f72022-06-13 15:19:14 +0530477func (att *AuditTablesTask) UpdateMismatchGroups(groupList []*of.Group) {
478 att.PushGroups(groupList, of.GroupCommandMod)
479}
480
481// PushGroups - The groups missing/to be updated in the device are reinstalled att the audit
482func (att *AuditTablesTask) PushGroups(groupList []*of.Group, grpCommand of.GroupCommand) {
483 logger.Debugw(ctx, "Pushing Groups", log.Fields{"Number": len(groupList), "Command": grpCommand})
484
485 var vc voltha.VolthaServiceClient
486 if vc = att.device.VolthaClient(); vc == nil {
487 logger.Error(ctx, "Update Group Table Failed: Voltha Client Unavailable")
488 return
489 }
490 for _, group := range groupList {
491 group.Command = grpCommand
492 groupUpdate := of.CreateGroupTableUpdate(group)
493 if _, err := vc.UpdateLogicalDeviceFlowGroupTable(att.ctx, groupUpdate); err != nil {
494 logger.Errorw(ctx, "Update Group Table Failed", log.Fields{"Reason": err.Error()})
495 }
496 }
497}
498
499// DelExcessGroups - Delete the excess groups held at the VOLTHA
500func (att *AuditTablesTask) DelExcessGroups(groups map[uint32]*ofp.OfpGroupDesc) {
501 logger.Debugw(ctx, "Deleting Excess Groups", log.Fields{"Number of Groups": len(groups)})
502
503 var vc voltha.VolthaServiceClient
504 if vc = att.device.VolthaClient(); vc == nil {
505 logger.Error(ctx, "Delete Excess Groups Failed: Voltha Client Unavailable")
506 return
507 }
508
509 // Let's cycle through the groups to delete the excess groups
510 for _, groupDesc := range groups {
511 logger.Debugw(ctx, "Deleting Group", log.Fields{"GroupId": groupDesc.GroupId})
512 group := &of.Group{}
513 group.Device = att.device.ID
514 group.GroupID = groupDesc.GroupId
515
vinokuma926cb3e2023-03-29 11:41:06 +0530516 // Group Members should be deleted before triggered group delete
Naveen Sampath04696f72022-06-13 15:19:14 +0530517 group.Command = of.GroupCommandMod
518 groupUpdate := of.CreateGroupTableUpdate(group)
519 if _, err := vc.UpdateLogicalDeviceFlowGroupTable(att.ctx, groupUpdate); err != nil {
520 logger.Errorw(ctx, "Update Group Table Failed", log.Fields{"Reason": err.Error()})
521 }
522
523 group.Command = of.GroupCommandDel
524 groupUpdate = of.CreateGroupTableUpdate(group)
525 if _, err := vc.UpdateLogicalDeviceFlowGroupTable(att.ctx, groupUpdate); err != nil {
526 logger.Errorw(ctx, "Update Group Table Failed", log.Fields{"Reason": err.Error()})
527 }
528 }
529}
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530530
531func (att *AuditTablesTask) AuditPorts() error {
vinokuma926cb3e2023-03-29 11:41:06 +0530532 if att.stop {
533 return tasks.ErrTaskCancelError
534 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530535
vinokuma926cb3e2023-03-29 11:41:06 +0530536 var vc voltha.VolthaServiceClient
537 if vc = att.device.VolthaClient(); vc == nil {
538 logger.Error(ctx, "Flow Audit Failed: Voltha Client Unavailable")
539 return nil
540 }
541 ofpps, err := vc.ListLogicalDevicePorts(att.ctx, &common.ID{Id: att.device.ID})
542 if err != nil {
543 return err
544 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530545
vinokuma926cb3e2023-03-29 11:41:06 +0530546 // Compute the difference between the ports received and ports at VGC
547 // First build a map of all the received ports under missing ports. We
548 // will eliminate the ports that are in the device from the missing ports
549 // so that the elements remaining are missing ports. The ones that are
550 // not in missing ports are added to excess ports which should be deleted
551 // from the VGC.
552 missingPorts := make(map[uint32]*ofp.OfpPort)
553 for _, ofpp := range ofpps.Items {
554 missingPorts[ofpp.OfpPort.PortNo] = ofpp.OfpPort
555 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530556
vinokuma926cb3e2023-03-29 11:41:06 +0530557 var excessPorts []uint32
558 processPortState := func(id uint32, vgcPort *DevicePort) {
559 logger.Debugw(ctx, "Process Port State Ind", log.Fields{"Port No": vgcPort.ID, "Port Name": vgcPort.Name})
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530560
vinokuma926cb3e2023-03-29 11:41:06 +0530561 if ofpPort, ok := missingPorts[id]; ok {
562 if ((vgcPort.State == PortStateDown) && (ofpPort.State == uint32(ofp.OfpPortState_OFPPS_LIVE))) || ((vgcPort.State == PortStateUp) && (ofpPort.State != uint32(ofp.OfpPortState_OFPPS_LIVE))) {
563 // This port exists in the received list and the map at
564 // VGC. This is common so delete it
565 logger.Infow(ctx, "Port State Mismatch", log.Fields{"Port": vgcPort.ID, "OfpPort": ofpPort.PortNo, "ReceivedState": ofpPort.State, "CurrentState": vgcPort.State})
566 att.device.ProcessPortState(ctx, ofpPort.PortNo, ofpPort.State)
567 }
568 delete(missingPorts, id)
569 } else {
570 // This port is missing from the received list. This is an
571 // excess port at VGC. This must be added to excess ports
572 excessPorts = append(excessPorts, id)
573 }
574 logger.Debugw(ctx, "Processed Port State Ind", log.Fields{"Port No": vgcPort.ID, "Port Name": vgcPort.Name})
575 }
576 // 1st process the NNI port before all other ports so that the device state can be updated.
577 if vgcPort, ok := att.device.PortsByID[NNIPortID]; ok {
Akash Soni6168f312023-05-18 20:57:33 +0530578 logger.Debugw(ctx, "Processing NNI port state", log.Fields{"Port ID": vgcPort.ID, "Port Name": vgcPort.Name})
vinokuma926cb3e2023-03-29 11:41:06 +0530579 processPortState(NNIPortID, vgcPort)
580 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530581
vinokuma926cb3e2023-03-29 11:41:06 +0530582 for id, vgcPort := range att.device.PortsByID {
583 if id == NNIPortID {
584 // NNI port already processed
585 continue
586 }
587 if att.stop {
588 break
589 }
590 processPortState(id, vgcPort)
591 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530592
593 if att.stop {
Akash Soni6168f312023-05-18 20:57:33 +0530594 logger.Warnw(ctx, "Audit Device Task Canceled", log.Fields{"Context": att.ctx, "Task": att.taskID})
vinokuma926cb3e2023-03-29 11:41:06 +0530595 return tasks.ErrTaskCancelError
596 }
597 att.AddMissingPorts(ctx, missingPorts)
598 att.DelExcessPorts(ctx, excessPorts)
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530599 return nil
600}
601
602// AddMissingPorts to add the missing ports
603func (att *AuditTablesTask) AddMissingPorts(cntx context.Context, mps map[uint32]*ofp.OfpPort) {
Akash Soni6168f312023-05-18 20:57:33 +0530604 logger.Infow(ctx, "Device Audit - Add Missing Ports", log.Fields{"NumPorts": len(mps)})
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530605
vinokuma926cb3e2023-03-29 11:41:06 +0530606 addMissingPort := func(mp *ofp.OfpPort) {
607 logger.Debugw(ctx, "Process Port Add Ind", log.Fields{"Port No": mp.PortNo, "Port Name": mp.Name})
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530608
vinokuma926cb3e2023-03-29 11:41:06 +0530609 // Error is ignored as it only drops duplicate ports
610 logger.Infow(ctx, "Calling AddPort", log.Fields{"No": mp.PortNo, "Name": mp.Name})
611 if err := att.device.AddPort(cntx, mp); err != nil {
612 logger.Warnw(ctx, "AddPort Failed", log.Fields{"No": mp.PortNo, "Name": mp.Name, "Reason": err})
613 }
614 if mp.State == uint32(ofp.OfpPortState_OFPPS_LIVE) {
615 att.device.ProcessPortState(cntx, mp.PortNo, mp.State)
616 }
617 logger.Debugw(ctx, "Processed Port Add Ind", log.Fields{"Port No": mp.PortNo, "Port Name": mp.Name})
618 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530619
vinokuma926cb3e2023-03-29 11:41:06 +0530620 // 1st process the NNI port before all other ports so that the flow provisioning for UNIs can be enabled
621 if mp, ok := mps[NNIPortID]; ok {
Akash Soni6168f312023-05-18 20:57:33 +0530622 logger.Debugw(ctx, "Adding Missing NNI port", log.Fields{"PortNo": mp.PortNo})
vinokuma926cb3e2023-03-29 11:41:06 +0530623 addMissingPort(mp)
624 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530625
vinokuma926cb3e2023-03-29 11:41:06 +0530626 for portNo, mp := range mps {
627 if portNo != NNIPortID {
628 addMissingPort(mp)
629 }
630 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530631}
632
633// DelExcessPorts to delete the excess ports
634func (att *AuditTablesTask) DelExcessPorts(cntx context.Context, eps []uint32) {
vinokuma926cb3e2023-03-29 11:41:06 +0530635 logger.Debugw(ctx, "Device Audit - Delete Excess Ports", log.Fields{"NumPorts": len(eps)})
636 for _, id := range eps {
637 // Now delete the port from the device @ VGC
638 logger.Infow(ctx, "Device Audit - Deleting Port", log.Fields{"PortId": id})
639 if err := att.device.DelPort(cntx, id); err != nil {
640 logger.Warnw(ctx, "DelPort Failed", log.Fields{"PortId": id, "Reason": err})
641 }
642 }
Tinoj Josephaf37ce82022-12-28 11:59:43 +0530643}