Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 1 | /* |
| 2 | * Copyright 2020-present Open Networking Foundation |
| 3 | * |
| 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | * you may not use this file except in compliance with the License. |
| 6 | * You may obtain a copy of the License at |
| 7 | * |
| 8 | * http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | * |
| 10 | * Unless required by applicable law or agreed to in writing, software |
| 11 | * distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | * See the License for the specific language governing permissions and |
| 14 | * limitations under the License. |
| 15 | */ |
| 16 | |
| 17 | package event |
| 18 | |
| 19 | import ( |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 20 | "context" |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 21 | "encoding/binary" |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 22 | "encoding/hex" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 23 | "fmt" |
Himani Chawla | 606a4f0 | 2021-03-23 19:45:58 +0530 | [diff] [blame] | 24 | "sync" |
| 25 | "time" |
| 26 | |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 27 | "github.com/golang/protobuf/ptypes/empty" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 28 | "github.com/opencord/voltha-go/rw_core/utils" |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 29 | ev "github.com/opencord/voltha-lib-go/v4/pkg/events" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 30 | "github.com/opencord/voltha-lib-go/v4/pkg/events/eventif" |
Maninder | dfadc98 | 2020-10-28 14:04:33 +0530 | [diff] [blame] | 31 | "github.com/opencord/voltha-lib-go/v4/pkg/log" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 32 | "github.com/opencord/voltha-protos/v4/go/common" |
Maninder | dfadc98 | 2020-10-28 14:04:33 +0530 | [diff] [blame] | 33 | "github.com/opencord/voltha-protos/v4/go/openflow_13" |
| 34 | "github.com/opencord/voltha-protos/v4/go/voltha" |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 35 | "github.com/opentracing/opentracing-go" |
| 36 | jtracing "github.com/uber/jaeger-client-go" |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 37 | ) |
| 38 | |
| 39 | type Manager struct { |
| 40 | packetInQueue chan openflow_13.PacketIn |
| 41 | packetInQueueDone chan bool |
| 42 | changeEventQueue chan openflow_13.ChangeEvent |
| 43 | changeEventQueueDone chan bool |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 44 | Agent *Agent |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 45 | } |
| 46 | |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 47 | type Agent struct { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 48 | eventProxy eventif.EventProxy |
| 49 | coreInstanceID string |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame] | 50 | stackID string |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 51 | } |
| 52 | |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 53 | func NewManager(proxyForEvents eventif.EventProxy, instanceID string, stackID string) *Manager { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 54 | return &Manager{ |
| 55 | packetInQueue: make(chan openflow_13.PacketIn, 100), |
| 56 | packetInQueueDone: make(chan bool, 1), |
| 57 | changeEventQueue: make(chan openflow_13.ChangeEvent, 100), |
| 58 | changeEventQueueDone: make(chan bool, 1), |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 59 | Agent: NewAgent(proxyForEvents, instanceID, stackID), |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 60 | } |
| 61 | } |
| 62 | |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 63 | func NewAgent(proxyForEvents eventif.EventProxy, instanceID string, stackID string) *Agent { |
| 64 | return &Agent{ |
| 65 | eventProxy: proxyForEvents, |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 66 | coreInstanceID: instanceID, |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame] | 67 | stackID: stackID, |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 68 | } |
| 69 | } |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 70 | func (q *Manager) SendPacketIn(ctx context.Context, deviceID string, transationID string, packet *openflow_13.OfpPacketIn) { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 71 | // TODO: Augment the OF PacketIn to include the transactionId |
| 72 | packetIn := openflow_13.PacketIn{Id: deviceID, PacketIn: packet} |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 73 | logger.Debugw(ctx, "send-packet-in", log.Fields{"packet-in": packetIn}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 74 | q.packetInQueue <- packetIn |
| 75 | } |
| 76 | |
| 77 | type callTracker struct { |
| 78 | failedPacket interface{} |
| 79 | } |
| 80 | type streamTracker struct { |
| 81 | calls map[string]*callTracker |
| 82 | sync.Mutex |
| 83 | } |
| 84 | |
| 85 | var streamingTracker = &streamTracker{calls: make(map[string]*callTracker)} |
| 86 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 87 | func (q *Manager) getStreamingTracker(ctx context.Context, method string, done chan<- bool) *callTracker { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 88 | streamingTracker.Lock() |
| 89 | defer streamingTracker.Unlock() |
| 90 | if _, ok := streamingTracker.calls[method]; ok { |
| 91 | // bail out the other packet in thread |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 92 | logger.Debugf(ctx, "%s-streaming-call-already-running-exiting-it", method) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 93 | done <- true |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 94 | logger.Debugf(ctx, "last-%s-exited-continuing", method) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 95 | } else { |
| 96 | streamingTracker.calls[method] = &callTracker{failedPacket: nil} |
| 97 | } |
| 98 | return streamingTracker.calls[method] |
| 99 | } |
| 100 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 101 | func (q *Manager) flushFailedPackets(ctx context.Context, tracker *callTracker) error { |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 102 | if tracker.failedPacket != nil { |
| 103 | switch tracker.failedPacket.(type) { |
| 104 | case openflow_13.PacketIn: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 105 | logger.Debug(ctx, "enqueueing-last-failed-packet-in") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 106 | q.packetInQueue <- tracker.failedPacket.(openflow_13.PacketIn) |
| 107 | case openflow_13.ChangeEvent: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 108 | logger.Debug(ctx, "enqueueing-last-failed-change-event") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 109 | q.changeEventQueue <- tracker.failedPacket.(openflow_13.ChangeEvent) |
| 110 | } |
| 111 | } |
| 112 | return nil |
| 113 | } |
| 114 | |
| 115 | // ReceivePacketsIn receives packets from adapter |
| 116 | func (q *Manager) ReceivePacketsIn(_ *empty.Empty, packetsIn voltha.VolthaService_ReceivePacketsInServer) error { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 117 | ctx := context.Background() |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 118 | ctx = utils.WithRPCMetadataContext(ctx, "ReceivePacketsIn") |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 119 | var streamingTracker = q.getStreamingTracker(ctx, "ReceivePacketsIn", q.packetInQueueDone) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 120 | logger.Debugw(ctx, "receive-packets-in-request", log.Fields{"packets-in": packetsIn}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 121 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 122 | err := q.flushFailedPackets(ctx, streamingTracker) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 123 | if err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 124 | logger.Errorw(ctx, "unable-to-flush-failed-packets", log.Fields{"error": err}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 125 | } |
| 126 | |
| 127 | loop: |
| 128 | for { |
| 129 | select { |
| 130 | case packet := <-q.packetInQueue: |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 131 | logger.Debugw(ctx, "sending-packet-in", log.Fields{ |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 132 | "packet": hex.EncodeToString(packet.PacketIn.Data), |
| 133 | }) |
| 134 | if err := packetsIn.Send(&packet); err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 135 | logger.Errorw(ctx, "failed-to-send-packet", log.Fields{"error": err}) |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 136 | go q.Agent.GetAndSendRPCEvent(ctx, packet.Id, err.Error(), |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 137 | nil, "RPC_ERROR_RAISE_EVENT", voltha.EventCategory_COMMUNICATION, |
Himani Chawla | 606a4f0 | 2021-03-23 19:45:58 +0530 | [diff] [blame] | 138 | nil, time.Now().Unix()) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 139 | // save the last failed packet in |
| 140 | streamingTracker.failedPacket = packet |
| 141 | } else { |
| 142 | if streamingTracker.failedPacket != nil { |
| 143 | // reset last failed packet saved to avoid flush |
| 144 | streamingTracker.failedPacket = nil |
| 145 | } |
| 146 | } |
| 147 | case <-q.packetInQueueDone: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 148 | logger.Debug(ctx, "another-receive-packets-in-running-bailing-out") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 149 | break loop |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | //TODO: Find an elegant way to get out of the above loop when the Core is stopped |
| 154 | return nil |
| 155 | } |
| 156 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 157 | func (q *Manager) SendChangeEvent(ctx context.Context, deviceID string, reason openflow_13.OfpPortReason, desc *openflow_13.OfpPort) { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 158 | logger.Debugw(ctx, "send-change-event", log.Fields{"device-id": deviceID, "reason": reason, "desc": desc}) |
Kent Hagerman | fa9d6d4 | 2020-05-25 11:49:40 -0400 | [diff] [blame] | 159 | q.changeEventQueue <- openflow_13.ChangeEvent{ |
| 160 | Id: deviceID, |
| 161 | Event: &openflow_13.ChangeEvent_PortStatus{ |
| 162 | PortStatus: &openflow_13.OfpPortStatus{ |
| 163 | Reason: reason, |
| 164 | Desc: desc, |
| 165 | }, |
| 166 | }, |
| 167 | } |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 168 | } |
| 169 | |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 170 | func (q *Manager) SendFlowChangeEvent(ctx context.Context, deviceID string, res []error, xid uint32, flowCookie uint64) { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 171 | logger.Debugw(ctx, "send-change-event", log.Fields{"device-id": deviceID, |
| 172 | "flow-id": xid, "flow-cookie": flowCookie, "errors": res}) |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 173 | errorType := openflow_13.OfpErrorType_OFPET_FLOW_MOD_FAILED |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 174 | //Manually creating the data payload for the flow error message |
| 175 | bs := make([]byte, 2) |
| 176 | //OF 1.3 |
| 177 | bs[0] = byte(4) |
| 178 | //Flow Mod |
| 179 | bs[1] = byte(14) |
| 180 | //Length of the message |
| 181 | length := make([]byte, 2) |
| 182 | binary.BigEndian.PutUint16(length, 56) |
| 183 | bs = append(bs, length...) |
| 184 | emptyArr := []byte{0, 0, 0, 0} |
| 185 | bs = append(bs, emptyArr...) |
| 186 | //Cookie of the Flow |
| 187 | cookie := make([]byte, 52) |
| 188 | binary.BigEndian.PutUint64(cookie, flowCookie) |
| 189 | bs = append(bs, cookie...) |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 190 | q.changeEventQueue <- openflow_13.ChangeEvent{ |
| 191 | Id: deviceID, |
| 192 | Event: &openflow_13.ChangeEvent_Error{ |
| 193 | Error: &openflow_13.OfpErrorMsg{ |
| 194 | Header: &openflow_13.OfpHeader{ |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 195 | Type: openflow_13.OfpType_OFPT_FLOW_MOD, |
| 196 | Xid: xid, |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 197 | }, |
| 198 | Type: uint32(errorType), |
| 199 | Code: uint32(openflow_13.OfpFlowModFailedCode_OFPFMFC_UNKNOWN), |
Andrea Campanella | 4afb2f0 | 2021-01-29 09:38:57 +0100 | [diff] [blame] | 200 | Data: bs, |
Maninder | f421da6 | 2020-12-04 11:44:58 +0530 | [diff] [blame] | 201 | }, |
| 202 | }, |
| 203 | } |
| 204 | } |
| 205 | |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 206 | // ReceiveChangeEvents receives change in events |
| 207 | func (q *Manager) ReceiveChangeEvents(_ *empty.Empty, changeEvents voltha.VolthaService_ReceiveChangeEventsServer) error { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 208 | ctx := context.Background() |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 209 | ctx = utils.WithRPCMetadataContext(ctx, "ReceiveChangeEvents") |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 210 | var streamingTracker = q.getStreamingTracker(ctx, "ReceiveChangeEvents", q.changeEventQueueDone) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 211 | logger.Debugw(ctx, "receive-change-events-request", log.Fields{"change-events": changeEvents}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 212 | |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 213 | err := q.flushFailedPackets(ctx, streamingTracker) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 214 | if err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 215 | logger.Errorw(ctx, "unable-to-flush-failed-packets", log.Fields{"error": err}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 216 | } |
| 217 | |
| 218 | loop: |
| 219 | for { |
| 220 | select { |
| 221 | // Dequeue a change event |
| 222 | case event := <-q.changeEventQueue: |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 223 | logger.Debugw(ctx, "sending-change-event", log.Fields{"event": event}) |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 224 | if err := changeEvents.Send(&event); err != nil { |
Rohan Agrawal | 31f2180 | 2020-06-12 05:38:46 +0000 | [diff] [blame] | 225 | logger.Errorw(ctx, "failed-to-send-change-event", log.Fields{"error": err}) |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 226 | go q.Agent.GetAndSendRPCEvent(ctx, event.Id, err.Error(), |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 227 | nil, "RPC_ERROR_RAISE_EVENT", voltha.EventCategory_COMMUNICATION, nil, |
Himani Chawla | 606a4f0 | 2021-03-23 19:45:58 +0530 | [diff] [blame] | 228 | time.Now().Unix()) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 229 | // save last failed change event |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 230 | streamingTracker.failedPacket = event |
| 231 | } else { |
| 232 | if streamingTracker.failedPacket != nil { |
| 233 | // reset last failed event saved on success to avoid flushing |
| 234 | streamingTracker.failedPacket = nil |
| 235 | } |
| 236 | } |
| 237 | case <-q.changeEventQueueDone: |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 238 | logger.Debug(ctx, "another-receive-change-events-already-running-bailing-out") |
Kent Hagerman | 45a13e4 | 2020-04-13 12:23:50 -0400 | [diff] [blame] | 239 | break loop |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | return nil |
| 244 | } |
| 245 | |
| 246 | func (q *Manager) GetChangeEventsQueueForTest() <-chan openflow_13.ChangeEvent { |
| 247 | return q.changeEventQueue |
| 248 | } |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 249 | |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 250 | func (q *Agent) NewRPCEvent(ctx context.Context, resourceID, desc string, context map[string]string) *voltha.RPCEvent { |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 251 | logger.Debugw(ctx, "new-rpc-event", log.Fields{"resource-id": resourceID}) |
| 252 | var opID string |
| 253 | var rpc string |
| 254 | |
| 255 | if span := opentracing.SpanFromContext(ctx); span != nil { |
| 256 | if jSpan, ok := span.(*jtracing.Span); ok { |
| 257 | opID = fmt.Sprintf("%016x", jSpan.SpanContext().TraceID().Low) // Using Sprintf to avoid removal of leading 0s |
| 258 | } |
| 259 | } |
| 260 | rpc = utils.GetRPCMetadataFromContext(ctx) |
| 261 | rpcev := &voltha.RPCEvent{ |
| 262 | Rpc: rpc, |
| 263 | OperationId: opID, |
| 264 | ResourceId: resourceID, |
| 265 | Service: q.coreInstanceID, |
Himani Chawla | 9cfc499 | 2021-03-22 12:43:01 +0530 | [diff] [blame] | 266 | StackId: q.stackID, |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 267 | Status: &common.OperationResp{ |
| 268 | Code: common.OperationResp_OPERATION_FAILURE, |
| 269 | }, |
| 270 | Description: desc, |
| 271 | Context: context, |
| 272 | } |
| 273 | return rpcev |
| 274 | } |
| 275 | |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 276 | func (q *Agent) SendRPCEvent(ctx context.Context, id string, rpcEvent *voltha.RPCEvent, category voltha.EventCategory_Types, subCategory *voltha.EventSubCategory_Types, raisedTs int64) { |
| 277 | //TODO Instead of directly sending to the kafka bus, queue the message and send it asynchronously |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 278 | if rpcEvent.Rpc != "" { |
Himani Chawla | 606a4f0 | 2021-03-23 19:45:58 +0530 | [diff] [blame] | 279 | if err := q.eventProxy.SendRPCEvent(ctx, id, rpcEvent, category, subCategory, raisedTs); err != nil { |
| 280 | logger.Errorw(ctx, "failed-to-send-rpc-event", log.Fields{"resource-id": id}) |
| 281 | } |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 282 | } |
| 283 | } |
| 284 | |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 285 | func (q *Agent) GetAndSendRPCEvent(ctx context.Context, resourceID, desc string, context map[string]string, |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 286 | id string, category voltha.EventCategory_Types, subCategory *voltha.EventSubCategory_Types, raisedTs int64) { |
| 287 | rpcEvent := q.NewRPCEvent(ctx, resourceID, desc, context) |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 288 | if rpcEvent.Rpc != "" { |
Himani Chawla | 606a4f0 | 2021-03-23 19:45:58 +0530 | [diff] [blame] | 289 | if err := q.eventProxy.SendRPCEvent(ctx, id, rpcEvent, category, subCategory, raisedTs); err != nil { |
| 290 | logger.Errorw(ctx, "failed-to-send-rpc-event", log.Fields{"resource-id": id}) |
| 291 | } |
Himani Chawla | b4c2591 | 2020-11-12 17:16:38 +0530 | [diff] [blame] | 292 | } |
| 293 | } |
Mahir Gunyel | b0343bf | 2021-05-11 14:14:26 -0700 | [diff] [blame] | 294 | |
| 295 | // SendDeviceStateChangeEvent sends Device State Change Event to message bus |
| 296 | func (q *Agent) SendDeviceStateChangeEvent(ctx context.Context, |
| 297 | prevOperStatus voltha.OperStatus_Types, prevConnStatus voltha.ConnectStatus_Types, prevAdminStatus voltha.AdminState_Types, |
| 298 | device *voltha.Device, raisedTs int64) error { |
| 299 | de := ev.CreateDeviceStateChangeEvent(device.SerialNumber, device.Id, device.ParentId, |
| 300 | prevOperStatus, prevConnStatus, prevAdminStatus, |
| 301 | device.OperStatus, device.ConnectStatus, device.AdminState, |
| 302 | device.ParentPortNo, device.Root) |
| 303 | |
| 304 | subCategory := voltha.EventSubCategory_ONU |
| 305 | if device.Root { |
| 306 | subCategory = voltha.EventSubCategory_OLT |
| 307 | } |
| 308 | if err := q.eventProxy.SendDeviceEvent(ctx, de, voltha.EventCategory_EQUIPMENT, subCategory, raisedTs); err != nil { |
| 309 | logger.Errorw(ctx, "error-sending-device-event", log.Fields{"id": device.Id, "err": err}) |
| 310 | return err |
| 311 | } |
| 312 | logger.Debugw(ctx, "device-state-change-sent", log.Fields{"event": *de}) |
| 313 | return nil |
| 314 | } |