blob: 2301f43ea87284070e4d95d0b8ed922158f7b524 [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 (
Neha Sharma94f16a92020-06-26 04:17:55 +000020 "context"
Scott Baker2c1c4822019-10-16 11:02:41 -070021 "errors"
22 "fmt"
23 "strconv"
24 "strings"
25 "time"
26
Scott Baker8e2be6b2020-02-10 17:27:15 -080027 "github.com/golang/protobuf/ptypes"
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053028 "github.com/opencord/voltha-lib-go/v4/pkg/events/eventif"
Girish Gowdra89c985b2020-10-14 15:02:09 -070029 "github.com/opencord/voltha-lib-go/v4/pkg/kafka"
30 "github.com/opencord/voltha-lib-go/v4/pkg/log"
31 "github.com/opencord/voltha-protos/v4/go/voltha"
Scott Baker2c1c4822019-10-16 11:02:41 -070032)
33
34type EventProxy struct {
35 kafkaClient kafka.Client
36 eventTopic kafka.Topic
37}
38
39func NewEventProxy(opts ...EventProxyOption) *EventProxy {
40 var proxy EventProxy
41 for _, option := range opts {
42 option(&proxy)
43 }
44 return &proxy
45}
46
47type EventProxyOption func(*EventProxy)
48
49func MsgClient(client kafka.Client) EventProxyOption {
50 return func(args *EventProxy) {
51 args.kafkaClient = client
52 }
53}
54
55func MsgTopic(topic kafka.Topic) EventProxyOption {
56 return func(args *EventProxy) {
57 args.eventTopic = topic
58 }
59}
60
61func (ep *EventProxy) formatId(eventName string) string {
62 return fmt.Sprintf("Voltha.openolt.%s.%s", eventName, strconv.FormatInt(time.Now().UnixNano(), 10))
63}
64
Scott Baker8e2be6b2020-02-10 17:27:15 -080065func (ep *EventProxy) getEventHeader(eventName string,
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053066 category eventif.EventCategory,
67 subCategory *eventif.EventSubCategory,
68 eventType eventif.EventType,
Scott Baker8e2be6b2020-02-10 17:27:15 -080069 raisedTs int64) (*voltha.EventHeader, error) {
Scott Baker2c1c4822019-10-16 11:02:41 -070070 var header voltha.EventHeader
71 if strings.Contains(eventName, "_") {
72 eventName = strings.Join(strings.Split(eventName, "_")[:len(strings.Split(eventName, "_"))-2], "_")
73 } else {
74 eventName = "UNKNOWN_EVENT"
75 }
76 /* Populating event header */
77 header.Id = ep.formatId(eventName)
78 header.Category = category
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053079 if subCategory != nil {
80 header.SubCategory = *subCategory
Himani Chawlaadc1b312021-02-04 13:12:33 +053081 } else {
82 header.SubCategory = voltha.EventSubCategory_NONE
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053083 }
Scott Baker2c1c4822019-10-16 11:02:41 -070084 header.Type = eventType
Himani Chawla4d2eb5d2020-11-12 17:19:20 +053085 header.TypeVersion = eventif.EventTypeVersion
Scott Baker8e2be6b2020-02-10 17:27:15 -080086
87 // raisedTs is in nanoseconds
88 timestamp, err := ptypes.TimestampProto(time.Unix(0, raisedTs))
89 if err != nil {
90 return nil, err
91 }
92 header.RaisedTs = timestamp
93
94 timestamp, err = ptypes.TimestampProto(time.Now())
95 if err != nil {
96 return nil, err
97 }
98 header.ReportedTs = timestamp
99
100 return &header, nil
Scott Baker2c1c4822019-10-16 11:02:41 -0700101}
102
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530103/* Send out rpc events*/
104func (ep *EventProxy) SendRPCEvent(ctx context.Context, id string, rpcEvent *voltha.RPCEvent, category eventif.EventCategory, subCategory *eventif.EventSubCategory, raisedTs int64) error {
105 if rpcEvent == nil {
106 logger.Error(ctx, "Received empty rpc event")
107 return errors.New("rpc event nil")
108 }
109 var event voltha.Event
110 var err error
111 if event.Header, err = ep.getEventHeader(id, category, subCategory, voltha.EventType_RPC_EVENT, raisedTs); err != nil {
112 return err
113 }
114 event.EventType = &voltha.Event_RpcEvent{RpcEvent: rpcEvent}
115 if err := ep.sendEvent(ctx, &event); err != nil {
116 logger.Errorw(ctx, "Failed to send rpc event to KAFKA bus", log.Fields{"rpc-event": rpcEvent})
117 return err
118 }
119 logger.Debugw(ctx, "Successfully sent RPC event to KAFKA bus", log.Fields{"Id": event.Header.Id, "Category": event.Header.Category,
120 "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
121 "ReportedTs": event.Header.ReportedTs, "ResourceId": rpcEvent.ResourceId, "Context": rpcEvent.Context,
122 "RPCEventName": id})
123
124 return nil
125
126}
127
Scott Baker2c1c4822019-10-16 11:02:41 -0700128/* Send out device events*/
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530129func (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 -0700130 if deviceEvent == nil {
Neha Sharma94f16a92020-06-26 04:17:55 +0000131 logger.Error(ctx, "Recieved empty device event")
Scott Baker2c1c4822019-10-16 11:02:41 -0700132 return errors.New("Device event nil")
133 }
134 var event voltha.Event
135 var de voltha.Event_DeviceEvent
Scott Baker8e2be6b2020-02-10 17:27:15 -0800136 var err error
Scott Baker2c1c4822019-10-16 11:02:41 -0700137 de.DeviceEvent = deviceEvent
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530138 if event.Header, err = ep.getEventHeader(deviceEvent.DeviceEventName, category, &subCategory, voltha.EventType_DEVICE_EVENT, raisedTs); err != nil {
Scott Baker8e2be6b2020-02-10 17:27:15 -0800139 return err
140 }
Scott Baker2c1c4822019-10-16 11:02:41 -0700141 event.EventType = &de
Neha Sharma94f16a92020-06-26 04:17:55 +0000142 if err := ep.sendEvent(ctx, &event); err != nil {
143 logger.Errorw(ctx, "Failed to send device event to KAFKA bus", log.Fields{"device-event": deviceEvent})
Scott Baker2c1c4822019-10-16 11:02:41 -0700144 return err
145 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000146 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 -0700147 "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
148 "ReportedTs": event.Header.ReportedTs, "ResourceId": deviceEvent.ResourceId, "Context": deviceEvent.Context,
149 "DeviceEventName": deviceEvent.DeviceEventName})
150
151 return nil
152
153}
154
Naga Manjunatha59c9152019-10-30 12:48:49 +0530155// SendKpiEvent is to send kpi events to voltha.event topic
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530156func (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 +0530157 if kpiEvent == nil {
Neha Sharma94f16a92020-06-26 04:17:55 +0000158 logger.Error(ctx, "Recieved empty kpi event")
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530159 return errors.New("KPI event nil")
160 }
161 var event voltha.Event
162 var de voltha.Event_KpiEvent2
Scott Baker8e2be6b2020-02-10 17:27:15 -0800163 var err error
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530164 de.KpiEvent2 = kpiEvent
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530165 if event.Header, err = ep.getEventHeader(id, category, &subCategory, voltha.EventType_KPI_EVENT2, raisedTs); err != nil {
Scott Baker8e2be6b2020-02-10 17:27:15 -0800166 return err
167 }
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530168 event.EventType = &de
Neha Sharma94f16a92020-06-26 04:17:55 +0000169 if err := ep.sendEvent(ctx, &event); err != nil {
170 logger.Errorw(ctx, "Failed to send kpi event to KAFKA bus", log.Fields{"device-event": kpiEvent})
Naga Manjunath86e9d2e2019-10-25 15:21:49 +0530171 return err
172 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000173 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 +0530174 "SubCategory": event.Header.SubCategory, "Type": event.Header.Type, "TypeVersion": event.Header.TypeVersion,
175 "ReportedTs": event.Header.ReportedTs, "KpiEventName": "STATS_EVENT"})
176
177 return nil
178
179}
180
Neha Sharma94f16a92020-06-26 04:17:55 +0000181func (ep *EventProxy) sendEvent(ctx context.Context, event *voltha.Event) error {
Himani Chawla4d2eb5d2020-11-12 17:19:20 +0530182 logger.Debugw(ctx, "Send event to kafka", log.Fields{"event": event})
Neha Sharma94f16a92020-06-26 04:17:55 +0000183 if err := ep.kafkaClient.Send(ctx, event, &ep.eventTopic); err != nil {
Scott Baker2c1c4822019-10-16 11:02:41 -0700184 return err
185 }
Neha Sharma94f16a92020-06-26 04:17:55 +0000186 logger.Debugw(ctx, "Sent event to kafka", log.Fields{"event": event})
Scott Baker2c1c4822019-10-16 11:02:41 -0700187
188 return nil
189}
Matteo Scandolo2ca74462021-03-01 14:03:17 -0800190
191func (ep *EventProxy) EnableLivenessChannel(ctx context.Context, enable bool) chan bool {
192 return ep.kafkaClient.EnableLivenessChannel(ctx, enable)
193}
194
195func (ep *EventProxy) SendLiveness(ctx context.Context) error {
196 return ep.kafkaClient.SendLiveness(ctx)
197}