blob: c64646ea2e5166e1fc9c74e6bec5db754e592fbc [file] [log] [blame]
Pragya Arya324337e2020-02-20 14:35:08 +05301/*
2 * Copyright 2018-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
17package common
18
19import (
20 "encoding/json"
21 "strconv"
22 "time"
23
24 "github.com/Shopify/sarama"
25 log "github.com/sirupsen/logrus"
26)
27
28var producer sarama.AsyncProducer
29var topic string
30
31// Event defines structure for bbsim events
32type Event struct {
33 EventType string
34 OnuSerial string
35 OltID int
36 IntfID int32
37 OnuID int32
38 EpochTime int64
39 Timestamp string
40}
41
42// InitializePublisher initalizes kafka publisher
Shrey Baid64cda472020-04-24 18:58:18 +053043func InitializePublisher(NewAsyncProducer func([]string, *sarama.Config) (sarama.AsyncProducer, error), oltID int) error {
Pragya Arya324337e2020-02-20 14:35:08 +053044
45 var err error
46 sarama.Logger = log.New()
47 // producer config
48 config := sarama.NewConfig()
49 config.Producer.Retry.Max = 5
50 config.Metadata.Retry.Max = 10
51 config.Metadata.Retry.Backoff = 10 * time.Second
52 config.ClientID = "BBSim-OLT-" + strconv.Itoa(oltID)
Matteo Scandolo4a036262020-08-17 15:56:13 -070053 if len(Config.BBSim.KafkaEventTopic) > 0 {
54 topic = Config.BBSim.KafkaEventTopic
Shrey Baid64cda472020-04-24 18:58:18 +053055 } else {
56 topic = "BBSim-OLT-" + strconv.Itoa(oltID) + "-Events"
57 }
Pragya Arya324337e2020-02-20 14:35:08 +053058
Matteo Scandolo4a036262020-08-17 15:56:13 -070059 producer, err = NewAsyncProducer([]string{Config.BBSim.KafkaAddress}, config)
Pragya Arya324337e2020-02-20 14:35:08 +053060 return err
61}
62
63// KafkaPublisher receives messages on eventChannel and publish them to kafka
64func KafkaPublisher(eventChannel chan Event) {
65 defer log.Debugf("KafkaPublisher stopped")
66 for {
Shrey Baid688b4242020-07-10 20:40:10 +053067 event := <-eventChannel
68 log.WithFields(log.Fields{
69 "EventType": event.EventType,
70 "OnuSerial": event.OnuSerial,
71 "OltID": event.OltID,
72 "IntfID": event.IntfID,
73 "OnuID": event.OnuID,
74 "EpochTime": event.EpochTime,
75 "Timestamp": event.Timestamp,
76 }).Trace("Received event on channel")
77 jsonEvent, err := json.Marshal(event)
78 if err != nil {
79 log.Errorf("Failed to get json event %v", err)
80 continue
Pragya Arya324337e2020-02-20 14:35:08 +053081 }
Shrey Baid688b4242020-07-10 20:40:10 +053082 producer.Input() <- &sarama.ProducerMessage{
83 Topic: topic,
84 Value: sarama.ByteEncoder(jsonEvent),
85 }
86 log.WithFields(log.Fields{
87 "EventType": event.EventType,
88 "OnuSerial": event.OnuSerial,
89 "OltID": event.OltID,
90 "IntfID": event.IntfID,
91 "OnuID": event.OnuID,
92 "EpochTime": event.EpochTime,
93 "Timestamp": event.Timestamp,
94 }).Debug("Event sent on kafka")
Pragya Arya324337e2020-02-20 14:35:08 +053095 }
Shrey Baid688b4242020-07-10 20:40:10 +053096
Pragya Arya324337e2020-02-20 14:35:08 +053097}