blob: e4493f91bb9ebc5a80d13ff1b71d37f13d7f0cbb [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
khenaidoo26721882021-08-11 17:42:52 -040029 "github.com/opencord/voltha-lib-go/v7/pkg/events/eventif"
30 "github.com/opencord/voltha-lib-go/v7/pkg/kafka"
31 "github.com/opencord/voltha-lib-go/v7/pkg/log"
32 "github.com/opencord/voltha-protos/v5/go/voltha"
33 "google.golang.org/protobuf/types/known/timestamppb"
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
khenaidoo26721882021-08-11 17:42:52 -0400100 header.RaisedTs = timestamppb.New(time.Unix(raisedTs, 0))
101 header.ReportedTs = timestamppb.New(time.Now())
Scott Baker8e2be6b2020-02-10 17:27:15 -0800102
103 return &header, nil
Scott Baker2c1c4822019-10-16 11:02:41 -0700104}
105
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530106/* Send out rpc events*/
107func (ep *EventProxy) SendRPCEvent(ctx context.Context, id string, rpcEvent *voltha.RPCEvent, category eventif.EventCategory, subCategory *eventif.EventSubCategory, raisedTs int64) error {
108 if rpcEvent == nil {
109 logger.Error(ctx, "Received empty rpc event")
110 return errors.New("rpc event nil")
111 }
112 var event voltha.Event
113 var err error
114 if event.Header, err = ep.getEventHeader(id, category, subCategory, voltha.EventType_RPC_EVENT, raisedTs); err != nil {
115 return err
116 }
117 event.EventType = &voltha.Event_RpcEvent{RpcEvent: rpcEvent}
Himani Chawla6c38e572021-03-23 19:44:25 +0530118 ep.eventQueue.push(&event)
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530119 return nil
120
121}
122
Scott Baker2c1c4822019-10-16 11:02:41 -0700123/* Send out device events*/
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530124func (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 -0700125 if deviceEvent == nil {
Neha Sharma94f16a92020-06-26 04:17:55 +0000126 logger.Error(ctx, "Recieved empty device event")
Scott Baker2c1c4822019-10-16 11:02:41 -0700127 return errors.New("Device event nil")
128 }
129 var event voltha.Event
130 var de voltha.Event_DeviceEvent
Scott Baker8e2be6b2020-02-10 17:27:15 -0800131 var err error
Scott Baker2c1c4822019-10-16 11:02:41 -0700132 de.DeviceEvent = deviceEvent
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530133 if event.Header, err = ep.getEventHeader(deviceEvent.DeviceEventName, category, &subCategory, voltha.EventType_DEVICE_EVENT, raisedTs); err != nil {
Scott Baker8e2be6b2020-02-10 17:27:15 -0800134 return err
135 }
Scott Baker2c1c4822019-10-16 11:02:41 -0700136 event.EventType = &de
Neha Sharma94f16a92020-06-26 04:17:55 +0000137 if err := ep.sendEvent(ctx, &event); err != nil {
138 logger.Errorw(ctx, "Failed to send device event to KAFKA bus", log.Fields{"device-event": deviceEvent})
Scott Baker2c1c4822019-10-16 11:02:41 -0700139 return err
140 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000141 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 -0700142 "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
143 "ReportedTs": event.Header.ReportedTs, "ResourceId": deviceEvent.ResourceId, "Context": deviceEvent.Context,
144 "DeviceEventName": deviceEvent.DeviceEventName})
145
146 return nil
147
148}
149
Naga Manjunatha59c9152019-10-30 12:48:49 +0530150// SendKpiEvent is to send kpi events to voltha.event topic
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530151func (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 +0530152 if kpiEvent == nil {
Neha Sharma94f16a92020-06-26 04:17:55 +0000153 logger.Error(ctx, "Recieved empty kpi event")
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530154 return errors.New("KPI event nil")
155 }
156 var event voltha.Event
157 var de voltha.Event_KpiEvent2
Scott Baker8e2be6b2020-02-10 17:27:15 -0800158 var err error
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530159 de.KpiEvent2 = kpiEvent
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530160 if event.Header, err = ep.getEventHeader(id, category, &subCategory, voltha.EventType_KPI_EVENT2, raisedTs); err != nil {
Scott Baker8e2be6b2020-02-10 17:27:15 -0800161 return err
162 }
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530163 event.EventType = &de
Neha Sharma94f16a92020-06-26 04:17:55 +0000164 if err := ep.sendEvent(ctx, &event); err != nil {
165 logger.Errorw(ctx, "Failed to send kpi event to KAFKA bus", log.Fields{"device-event": kpiEvent})
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530166 return err
167 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000168 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 +0530169 "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
170 "ReportedTs": event.Header.ReportedTs, "KpiEventName": "STATS_EVENT"})
171
172 return nil
173
174}
175
Neha Sharma94f16a92020-06-26 04:17:55 +0000176func (ep *EventProxy) sendEvent(ctx context.Context, event *voltha.Event) error {
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530177 logger.Debugw(ctx, "Send event to kafka", log.Fields{"event": event})
Neha Sharma94f16a92020-06-26 04:17:55 +0000178 if err := ep.kafkaClient.Send(ctx, event, &ep.eventTopic); err != nil {
Scott Baker2c1c4822019-10-16 11:02:41 -0700179 return err
180 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000181 logger.Debugw(ctx, "Sent event to kafka", log.Fields{"event": event})
Scott Baker2c1c4822019-10-16 11:02:41 -0700182
183 return nil
184}
Matteo Scandolo2ca74462021-03-01 14:03:17 -0800185
186func (ep *EventProxy) EnableLivenessChannel(ctx context.Context, enable bool) chan bool {
187 return ep.kafkaClient.EnableLivenessChannel(ctx, enable)
188}
189
190func (ep *EventProxy) SendLiveness(ctx context.Context) error {
191 return ep.kafkaClient.SendLiveness(ctx)
192}
Himani Chawla6c38e572021-03-23 19:44:25 +0530193
194// Start the event proxy
khenaidoo26721882021-08-11 17:42:52 -0400195func (ep *EventProxy) Start() error {
Himani Chawla6c38e572021-03-23 19:44:25 +0530196 eq := ep.eventQueue
197 go eq.start(ep.queueCtx)
198 logger.Debugw(context.Background(), "event-proxy-starting...", log.Fields{"events-threashold": EVENT_THRESHOLD})
199 for {
200 // Notify the queue I am ready
201 eq.readyToSendToKafkaCh <- struct{}{}
202 // Wait for an event
203 elem, ok := <-eq.eventChannel
204 if !ok {
205 logger.Debug(context.Background(), "event-channel-closed-exiting")
206 break
207 }
208 // Check for last event
209 if _, ok := elem.(*lastEvent); ok {
210 // close the queuing loop
211 logger.Info(context.Background(), "received-last-event")
212 ep.queueCancelCtx()
213 break
214 }
215 ctx := context.Background()
216 event, ok := elem.(*voltha.Event)
217 if !ok {
218 logger.Warnw(ctx, "invalid-event", log.Fields{"element": elem})
219 continue
220 }
221 if err := ep.sendEvent(ctx, event); err != nil {
222 logger.Errorw(ctx, "failed-to-send-event-to-kafka-bus", log.Fields{"event": event})
223 } else {
224 logger.Debugw(ctx, "successfully-sent-rpc-event-to-kafka-bus", log.Fields{"id": event.Header.Id, "category": event.Header.Category,
225 "sub-category": event.Header.SubCategory, "type": event.Header.Type, "type-version": event.Header.TypeVersion,
226 "reported-ts": event.Header.ReportedTs, "event-type": event.EventType})
227 }
228 }
khenaidoo26721882021-08-11 17:42:52 -0400229 return nil
Himani Chawla6c38e572021-03-23 19:44:25 +0530230}
231
232func (ep *EventProxy) Stop() {
khenaidoo26721882021-08-11 17:42:52 -0400233 if ep.eventQueue != nil {
234 ep.eventQueue.stop()
235 }
Himani Chawla6c38e572021-03-23 19:44:25 +0530236}
237
238type EventQueue struct {
239 mutex sync.RWMutex
240 eventChannel chan interface{}
241 insertPosition *ring.Ring
242 popPosition *ring.Ring
243 dataToSendAvailable chan struct{}
244 readyToSendToKafkaCh chan struct{}
245 eventQueueStopped chan struct{}
246}
247
248func newEventQueue() *EventQueue {
249 ev := &EventQueue{
250 eventChannel: make(chan interface{}),
251 insertPosition: ring.New(EVENT_THRESHOLD),
252 dataToSendAvailable: make(chan struct{}),
253 readyToSendToKafkaCh: make(chan struct{}),
254 eventQueueStopped: make(chan struct{}),
255 }
256 ev.popPosition = ev.insertPosition
257 return ev
258}
259
260// push is invoked to push an event at the back of a queue
261func (eq *EventQueue) push(event interface{}) {
262 eq.mutex.Lock()
263
264 if eq.insertPosition != nil {
265 // Handle Queue is full.
266 // TODO: Current default is to overwrite old data if queue is full. Is there a need to
267 // block caller if max threshold is reached?
268 if eq.insertPosition.Value != nil && eq.insertPosition == eq.popPosition {
269 eq.popPosition = eq.popPosition.Next()
270 }
271
272 // Insert data and move pointer to next empty position
273 eq.insertPosition.Value = event
274 eq.insertPosition = eq.insertPosition.Next()
275
276 // Check for last event
277 if _, ok := event.(*lastEvent); ok {
278 eq.insertPosition = nil
279 }
280 eq.mutex.Unlock()
281 // Notify waiting thread of data availability
282 eq.dataToSendAvailable <- struct{}{}
283
284 } else {
285 logger.Debug(context.Background(), "event-queue-is-closed-as-insert-position-is-cleared")
286 eq.mutex.Unlock()
287 }
288}
289
290// start starts the routine that extracts an element from the event queue and
291// send it to the kafka sending routine to process.
292func (eq *EventQueue) start(ctx context.Context) {
293 logger.Info(ctx, "starting-event-queue")
294loop:
295 for {
296 select {
297 case <-eq.dataToSendAvailable:
298 // Do nothing - use to prevent caller pushing data to block
299 case <-eq.readyToSendToKafkaCh:
300 {
301 // Kafka sending routine is ready to process an event
302 eq.mutex.Lock()
303 element := eq.popPosition.Value
304 if element == nil {
305 // No events to send. Wait
306 eq.mutex.Unlock()
307 select {
308 case _, ok := <-eq.dataToSendAvailable:
309 if !ok {
310 // channel closed
311 eq.eventQueueStopped <- struct{}{}
312 return
313 }
314 case <-ctx.Done():
315 logger.Info(ctx, "event-queue-context-done")
316 eq.eventQueueStopped <- struct{}{}
317 return
318 }
319 eq.mutex.Lock()
320 element = eq.popPosition.Value
321 }
322 eq.popPosition.Value = nil
323 eq.popPosition = eq.popPosition.Next()
324 eq.mutex.Unlock()
325 eq.eventChannel <- element
326 }
327 case <-ctx.Done():
328 logger.Info(ctx, "event-queue-context-done")
329 eq.eventQueueStopped <- struct{}{}
330 break loop
331 }
332 }
333 logger.Info(ctx, "event-queue-stopped")
334
335}
336
337func (eq *EventQueue) stop() {
338 // Flush all
339 eq.push(&lastEvent{})
340 <-eq.eventQueueStopped
341 eq.mutex.Lock()
342 close(eq.readyToSendToKafkaCh)
343 close(eq.dataToSendAvailable)
344 close(eq.eventChannel)
345 eq.mutex.Unlock()
346
347}