blob: 19a4f2657c7a77e3e8f51db2a888726e71a83676 [file] [log] [blame]
Scott Baker2c1c4822019-10-16 11:02:41 -07001/*
Himani Chawla4d2eb5d2020-11-12 17:19:20 +05302 * Copyright 2020-present Open Networking Foundation
Scott Baker2c1c4822019-10-16 11:02:41 -07003
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
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053017package events
Scott Baker2c1c4822019-10-16 11:02:41 -070018
19import (
Himani Chawla6c38e572021-03-23 19:44:25 +053020 "container/ring"
Neha Sharma94f16a92020-06-26 04:17:55 +000021 "context"
Scott Baker2c1c4822019-10-16 11:02:41 -070022 "errors"
23 "fmt"
24 "strconv"
25 "strings"
Himani Chawla6c38e572021-03-23 19:44:25 +053026 "sync"
Scott Baker2c1c4822019-10-16 11:02:41 -070027 "time"
28
Scott Baker8e2be6b2020-02-10 17:27:15 -080029 "github.com/golang/protobuf/ptypes"
Girish Gowdra248971a2021-06-01 15:14:15 -070030 "github.com/opencord/voltha-lib-go/v5/pkg/events/eventif"
31 "github.com/opencord/voltha-lib-go/v5/pkg/kafka"
32 "github.com/opencord/voltha-lib-go/v5/pkg/log"
Girish Gowdra89c985b2020-10-14 15:02:09 -070033 "github.com/opencord/voltha-protos/v4/go/voltha"
Scott Baker2c1c4822019-10-16 11:02:41 -070034)
35
Himani Chawla6c38e572021-03-23 19:44:25 +053036// TODO: Make configurable through helm chart
37const EVENT_THRESHOLD = 1000
38
39type lastEvent struct{}
40
Scott Baker2c1c4822019-10-16 11:02:41 -070041type EventProxy struct {
Himani Chawla6c38e572021-03-23 19:44:25 +053042 kafkaClient kafka.Client
43 eventTopic kafka.Topic
44 eventQueue *EventQueue
45 queueCtx context.Context
46 queueCancelCtx context.CancelFunc
Scott Baker2c1c4822019-10-16 11:02:41 -070047}
48
49func NewEventProxy(opts ...EventProxyOption) *EventProxy {
50 var proxy EventProxy
51 for _, option := range opts {
52 option(&proxy)
53 }
Himani Chawla6c38e572021-03-23 19:44:25 +053054 proxy.eventQueue = newEventQueue()
55 proxy.queueCtx, proxy.queueCancelCtx = context.WithCancel(context.Background())
Scott Baker2c1c4822019-10-16 11:02:41 -070056 return &proxy
57}
58
59type EventProxyOption func(*EventProxy)
60
61func MsgClient(client kafka.Client) EventProxyOption {
62 return func(args *EventProxy) {
63 args.kafkaClient = client
64 }
65}
66
67func MsgTopic(topic kafka.Topic) EventProxyOption {
68 return func(args *EventProxy) {
69 args.eventTopic = topic
70 }
71}
72
73func (ep *EventProxy) formatId(eventName string) string {
74 return fmt.Sprintf("Voltha.openolt.%s.%s", eventName, strconv.FormatInt(time.Now().UnixNano(), 10))
75}
76
Scott Baker8e2be6b2020-02-10 17:27:15 -080077func (ep *EventProxy) getEventHeader(eventName string,
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053078 category eventif.EventCategory,
79 subCategory *eventif.EventSubCategory,
80 eventType eventif.EventType,
Scott Baker8e2be6b2020-02-10 17:27:15 -080081 raisedTs int64) (*voltha.EventHeader, error) {
Scott Baker2c1c4822019-10-16 11:02:41 -070082 var header voltha.EventHeader
83 if strings.Contains(eventName, "_") {
84 eventName = strings.Join(strings.Split(eventName, "_")[:len(strings.Split(eventName, "_"))-2], "_")
85 } else {
86 eventName = "UNKNOWN_EVENT"
87 }
88 /* Populating event header */
89 header.Id = ep.formatId(eventName)
90 header.Category = category
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053091 if subCategory != nil {
92 header.SubCategory = *subCategory
Himani Chawlaadc1b312021-02-04 13:12:33 +053093 } else {
94 header.SubCategory = voltha.EventSubCategory_NONE
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053095 }
Scott Baker2c1c4822019-10-16 11:02:41 -070096 header.Type = eventType
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053097 header.TypeVersion = eventif.EventTypeVersion
Scott Baker8e2be6b2020-02-10 17:27:15 -080098
Girish Gowdradcd54062021-04-22 12:49:17 -070099 // raisedTs is in seconds
100 timestamp, err := ptypes.TimestampProto(time.Unix(raisedTs, 0))
Scott Baker8e2be6b2020-02-10 17:27:15 -0800101 if err != nil {
102 return nil, err
103 }
104 header.RaisedTs = timestamp
105
106 timestamp, err = ptypes.TimestampProto(time.Now())
107 if err != nil {
108 return nil, err
109 }
110 header.ReportedTs = timestamp
111
112 return &header, nil
Scott Baker2c1c4822019-10-16 11:02:41 -0700113}
114
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530115/* Send out rpc events*/
116func (ep *EventProxy) SendRPCEvent(ctx context.Context, id string, rpcEvent *voltha.RPCEvent, category eventif.EventCategory, subCategory *eventif.EventSubCategory, raisedTs int64) error {
117 if rpcEvent == nil {
118 logger.Error(ctx, "Received empty rpc event")
119 return errors.New("rpc event nil")
120 }
121 var event voltha.Event
122 var err error
123 if event.Header, err = ep.getEventHeader(id, category, subCategory, voltha.EventType_RPC_EVENT, raisedTs); err != nil {
124 return err
125 }
126 event.EventType = &voltha.Event_RpcEvent{RpcEvent: rpcEvent}
Himani Chawla6c38e572021-03-23 19:44:25 +0530127 ep.eventQueue.push(&event)
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530128 return nil
129
130}
131
Scott Baker2c1c4822019-10-16 11:02:41 -0700132/* Send out device events*/
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530133func (ep *EventProxy) SendDeviceEvent(ctx context.Context, deviceEvent *voltha.DeviceEvent, category eventif.EventCategory, subCategory eventif.EventSubCategory, raisedTs int64) error {
Scott Baker2c1c4822019-10-16 11:02:41 -0700134 if deviceEvent == nil {
Neha Sharma94f16a92020-06-26 04:17:55 +0000135 logger.Error(ctx, "Recieved empty device event")
Scott Baker2c1c4822019-10-16 11:02:41 -0700136 return errors.New("Device event nil")
137 }
138 var event voltha.Event
139 var de voltha.Event_DeviceEvent
Scott Baker8e2be6b2020-02-10 17:27:15 -0800140 var err error
Scott Baker2c1c4822019-10-16 11:02:41 -0700141 de.DeviceEvent = deviceEvent
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530142 if event.Header, err = ep.getEventHeader(deviceEvent.DeviceEventName, category, &subCategory, voltha.EventType_DEVICE_EVENT, raisedTs); err != nil {
Scott Baker8e2be6b2020-02-10 17:27:15 -0800143 return err
144 }
Scott Baker2c1c4822019-10-16 11:02:41 -0700145 event.EventType = &de
Neha Sharma94f16a92020-06-26 04:17:55 +0000146 if err := ep.sendEvent(ctx, &event); err != nil {
147 logger.Errorw(ctx, "Failed to send device event to KAFKA bus", log.Fields{"device-event": deviceEvent})
Scott Baker2c1c4822019-10-16 11:02:41 -0700148 return err
149 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000150 logger.Infow(ctx, "Successfully sent device event KAFKA", log.Fields{"Id": event.Header.Id, "Category": event.Header.Category,
Scott Baker2c1c4822019-10-16 11:02:41 -0700151 "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
152 "ReportedTs": event.Header.ReportedTs, "ResourceId": deviceEvent.ResourceId, "Context": deviceEvent.Context,
153 "DeviceEventName": deviceEvent.DeviceEventName})
154
155 return nil
156
157}
158
Naga Manjunatha59c9152019-10-30 12:48:49 +0530159// SendKpiEvent is to send kpi events to voltha.event topic
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530160func (ep *EventProxy) SendKpiEvent(ctx context.Context, id string, kpiEvent *voltha.KpiEvent2, category eventif.EventCategory, subCategory eventif.EventSubCategory, raisedTs int64) error {
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530161 if kpiEvent == nil {
Neha Sharma94f16a92020-06-26 04:17:55 +0000162 logger.Error(ctx, "Recieved empty kpi event")
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530163 return errors.New("KPI event nil")
164 }
165 var event voltha.Event
166 var de voltha.Event_KpiEvent2
Scott Baker8e2be6b2020-02-10 17:27:15 -0800167 var err error
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530168 de.KpiEvent2 = kpiEvent
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530169 if event.Header, err = ep.getEventHeader(id, category, &subCategory, voltha.EventType_KPI_EVENT2, raisedTs); err != nil {
Scott Baker8e2be6b2020-02-10 17:27:15 -0800170 return err
171 }
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530172 event.EventType = &de
Neha Sharma94f16a92020-06-26 04:17:55 +0000173 if err := ep.sendEvent(ctx, &event); err != nil {
174 logger.Errorw(ctx, "Failed to send kpi event to KAFKA bus", log.Fields{"device-event": kpiEvent})
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530175 return err
176 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000177 logger.Infow(ctx, "Successfully sent kpi event to KAFKA", log.Fields{"Id": event.Header.Id, "Category": event.Header.Category,
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530178 "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
179 "ReportedTs": event.Header.ReportedTs, "KpiEventName": "STATS_EVENT"})
180
181 return nil
182
183}
184
Neha Sharma94f16a92020-06-26 04:17:55 +0000185func (ep *EventProxy) sendEvent(ctx context.Context, event *voltha.Event) error {
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530186 logger.Debugw(ctx, "Send event to kafka", log.Fields{"event": event})
Neha Sharma94f16a92020-06-26 04:17:55 +0000187 if err := ep.kafkaClient.Send(ctx, event, &ep.eventTopic); err != nil {
Scott Baker2c1c4822019-10-16 11:02:41 -0700188 return err
189 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000190 logger.Debugw(ctx, "Sent event to kafka", log.Fields{"event": event})
Scott Baker2c1c4822019-10-16 11:02:41 -0700191
192 return nil
193}
Matteo Scandolo2ca74462021-03-01 14:03:17 -0800194
195func (ep *EventProxy) EnableLivenessChannel(ctx context.Context, enable bool) chan bool {
196 return ep.kafkaClient.EnableLivenessChannel(ctx, enable)
197}
198
199func (ep *EventProxy) SendLiveness(ctx context.Context) error {
200 return ep.kafkaClient.SendLiveness(ctx)
201}
Himani Chawla6c38e572021-03-23 19:44:25 +0530202
203// Start the event proxy
204func (ep *EventProxy) Start() {
205 eq := ep.eventQueue
206 go eq.start(ep.queueCtx)
207 logger.Debugw(context.Background(), "event-proxy-starting...", log.Fields{"events-threashold": EVENT_THRESHOLD})
208 for {
209 // Notify the queue I am ready
210 eq.readyToSendToKafkaCh <- struct{}{}
211 // Wait for an event
212 elem, ok := <-eq.eventChannel
213 if !ok {
214 logger.Debug(context.Background(), "event-channel-closed-exiting")
215 break
216 }
217 // Check for last event
218 if _, ok := elem.(*lastEvent); ok {
219 // close the queuing loop
220 logger.Info(context.Background(), "received-last-event")
221 ep.queueCancelCtx()
222 break
223 }
224 ctx := context.Background()
225 event, ok := elem.(*voltha.Event)
226 if !ok {
227 logger.Warnw(ctx, "invalid-event", log.Fields{"element": elem})
228 continue
229 }
230 if err := ep.sendEvent(ctx, event); err != nil {
231 logger.Errorw(ctx, "failed-to-send-event-to-kafka-bus", log.Fields{"event": event})
232 } else {
233 logger.Debugw(ctx, "successfully-sent-rpc-event-to-kafka-bus", log.Fields{"id": event.Header.Id, "category": event.Header.Category,
234 "sub-category": event.Header.SubCategory, "type": event.Header.Type, "type-version": event.Header.TypeVersion,
235 "reported-ts": event.Header.ReportedTs, "event-type": event.EventType})
236 }
237 }
238}
239
240func (ep *EventProxy) Stop() {
241 ep.eventQueue.stop()
242}
243
244type EventQueue struct {
245 mutex sync.RWMutex
246 eventChannel chan interface{}
247 insertPosition *ring.Ring
248 popPosition *ring.Ring
249 dataToSendAvailable chan struct{}
250 readyToSendToKafkaCh chan struct{}
251 eventQueueStopped chan struct{}
252}
253
254func newEventQueue() *EventQueue {
255 ev := &EventQueue{
256 eventChannel: make(chan interface{}),
257 insertPosition: ring.New(EVENT_THRESHOLD),
258 dataToSendAvailable: make(chan struct{}),
259 readyToSendToKafkaCh: make(chan struct{}),
260 eventQueueStopped: make(chan struct{}),
261 }
262 ev.popPosition = ev.insertPosition
263 return ev
264}
265
266// push is invoked to push an event at the back of a queue
267func (eq *EventQueue) push(event interface{}) {
268 eq.mutex.Lock()
269
270 if eq.insertPosition != nil {
271 // Handle Queue is full.
272 // TODO: Current default is to overwrite old data if queue is full. Is there a need to
273 // block caller if max threshold is reached?
274 if eq.insertPosition.Value != nil && eq.insertPosition == eq.popPosition {
275 eq.popPosition = eq.popPosition.Next()
276 }
277
278 // Insert data and move pointer to next empty position
279 eq.insertPosition.Value = event
280 eq.insertPosition = eq.insertPosition.Next()
281
282 // Check for last event
283 if _, ok := event.(*lastEvent); ok {
284 eq.insertPosition = nil
285 }
286 eq.mutex.Unlock()
287 // Notify waiting thread of data availability
288 eq.dataToSendAvailable <- struct{}{}
289
290 } else {
291 logger.Debug(context.Background(), "event-queue-is-closed-as-insert-position-is-cleared")
292 eq.mutex.Unlock()
293 }
294}
295
296// start starts the routine that extracts an element from the event queue and
297// send it to the kafka sending routine to process.
298func (eq *EventQueue) start(ctx context.Context) {
299 logger.Info(ctx, "starting-event-queue")
300loop:
301 for {
302 select {
303 case <-eq.dataToSendAvailable:
304 // Do nothing - use to prevent caller pushing data to block
305 case <-eq.readyToSendToKafkaCh:
306 {
307 // Kafka sending routine is ready to process an event
308 eq.mutex.Lock()
309 element := eq.popPosition.Value
310 if element == nil {
311 // No events to send. Wait
312 eq.mutex.Unlock()
313 select {
314 case _, ok := <-eq.dataToSendAvailable:
315 if !ok {
316 // channel closed
317 eq.eventQueueStopped <- struct{}{}
318 return
319 }
320 case <-ctx.Done():
321 logger.Info(ctx, "event-queue-context-done")
322 eq.eventQueueStopped <- struct{}{}
323 return
324 }
325 eq.mutex.Lock()
326 element = eq.popPosition.Value
327 }
328 eq.popPosition.Value = nil
329 eq.popPosition = eq.popPosition.Next()
330 eq.mutex.Unlock()
331 eq.eventChannel <- element
332 }
333 case <-ctx.Done():
334 logger.Info(ctx, "event-queue-context-done")
335 eq.eventQueueStopped <- struct{}{}
336 break loop
337 }
338 }
339 logger.Info(ctx, "event-queue-stopped")
340
341}
342
343func (eq *EventQueue) stop() {
344 // Flush all
345 eq.push(&lastEvent{})
346 <-eq.eventQueueStopped
347 eq.mutex.Lock()
348 close(eq.readyToSendToKafkaCh)
349 close(eq.dataToSendAvailable)
350 close(eq.eventChannel)
351 eq.mutex.Unlock()
352
353}