blob: 4ad82993b85da3d18a4045cf134a87ac7d823537 [file] [log] [blame]
Scott Bakered4efab2020-01-13 19:12:25 -08001/*
2 * Copyright 2019-present Ciena Corporation
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 */
16package commands
17
18import (
19 "encoding/json"
20 "errors"
21 "fmt"
22 "github.com/Shopify/sarama"
Scott Baker9173ed82020-05-19 08:30:12 -070023 "github.com/golang/protobuf/jsonpb"
24 "github.com/golang/protobuf/proto"
Scott Baker2fe436a2020-02-10 17:21:47 -080025 "github.com/golang/protobuf/ptypes"
26 "github.com/golang/protobuf/ptypes/timestamp"
Scott Bakered4efab2020-01-13 19:12:25 -080027 flags "github.com/jessevdk/go-flags"
Scott Bakered4efab2020-01-13 19:12:25 -080028 "github.com/opencord/voltctl/pkg/filter"
29 "github.com/opencord/voltctl/pkg/format"
Scott Baker9173ed82020-05-19 08:30:12 -070030 "github.com/opencord/voltha-protos/v3/go/voltha"
Scott Bakered4efab2020-01-13 19:12:25 -080031 "log"
32 "os"
33 "os/signal"
34 "strings"
35 "time"
36)
37
38const (
39 DEFAULT_EVENT_FORMAT = "table{{.Category}}\t{{.SubCategory}}\t{{.Type}}\t{{.Timestamp}}\t{{.Device_ids}}\t{{.Titles}}"
40)
41
42type EventListenOpts struct {
David K. Bainbridge2b627612020-02-18 14:50:13 -080043 Format string `long:"format" value-name:"FORMAT" default:"" description:"Format to use to output structured data"`
44 // nolint: staticcheck
Scott Bakered4efab2020-01-13 19:12:25 -080045 OutputAs string `short:"o" long:"outputas" default:"table" choice:"table" choice:"json" choice:"yaml" description:"Type of output to generate"`
46 Filter string `short:"f" long:"filter" default:"" value-name:"FILTER" description:"Only display results that match filter"`
47 Follow bool `short:"F" long:"follow" description:"Continue to consume until CTRL-C is pressed"`
48 ShowBody bool `short:"b" long:"show-body" description:"Show body of events rather than only a header summary"`
49 Count int `short:"c" long:"count" default:"-1" value-name:"LIMIT" description:"Limit the count of messages that will be printed"`
50 Now bool `short:"n" long:"now" description:"Stop printing events when current time is reached"`
51 Timeout int `short:"t" long:"idle" default:"900" value-name:"SECONDS" description:"Timeout if no message received within specified seconds"`
Scott Bakerf05e60a2020-02-02 21:53:57 -080052 Since string `short:"s" long:"since" default:"" value-name:"TIMESTAMP" description:"Do not show entries before timestamp"`
Scott Bakered4efab2020-01-13 19:12:25 -080053}
54
55type EventOpts struct {
56 EventListen EventListenOpts `command:"listen"`
57}
58
59var eventOpts = EventOpts{}
60
61type EventHeader struct {
Scott Baker2fe436a2020-02-10 17:21:47 -080062 Category string `json:"category"`
63 SubCategory string `json:"sub_category"`
64 Type string `json:"type"`
65 Raised_ts time.Time `json:"raised_ts"`
66 Reported_ts time.Time `json:"reported_ts"`
67 Device_ids []string `json:"device_ids"` // Opportunistically collected list of device_ids
68 Titles []string `json:"titles"` // Opportunistically collected list of titles
69 Timestamp time.Time `json:"timestamp"` // Timestamp from Kafka
Scott Bakered4efab2020-01-13 19:12:25 -080070}
71
72type EventHeaderWidths struct {
73 Category int
74 SubCategory int
75 Type int
76 Raised_ts int
77 Reported_ts int
78 Device_ids int
79 Titles int
80 Timestamp int
81}
82
83var DefaultWidths EventHeaderWidths = EventHeaderWidths{
84 Category: 13,
85 SubCategory: 3,
86 Type: 12,
87 Raised_ts: 10,
88 Reported_ts: 10,
89 Device_ids: 40,
90 Titles: 40,
91 Timestamp: 10,
92}
93
94func RegisterEventCommands(parent *flags.Parser) {
95 _, err := parent.AddCommand("event", "event commands", "Commands for observing events", &eventOpts)
96 if err != nil {
97 Error.Fatalf("Unable to register event commands with voltctl command parser: %s", err.Error())
98 }
99}
100
Scott Bakerf05e60a2020-02-02 21:53:57 -0800101func ParseSince(s string) (*time.Time, error) {
102 if strings.EqualFold(s, "now") {
103 since := time.Now()
104 return &since, nil
105 }
106
107 rfc3339Time, err := time.Parse(time.RFC3339, s)
108 if err == nil {
109 return &rfc3339Time, nil
110 }
111
112 duration, err := time.ParseDuration(s)
113 if err == nil {
114 since := time.Now().Add(-duration)
115 return &since, nil
116 }
117
118 return nil, fmt.Errorf("Unable to parse time specification `%s`. Please use either `now`, a duration, or an RFC3339-compliant string", s)
119}
120
Scott Baker2fe436a2020-02-10 17:21:47 -0800121// Convert a timestamp field in an event to a time.Time
122func DecodeTimestamp(tsIntf interface{}) (time.Time, error) {
123 ts, okay := tsIntf.(*timestamp.Timestamp)
124 if okay {
125 // Voltha-Protos 3.2.3 and above
126 result, err := ptypes.Timestamp(ts)
127 return result, err
128 }
129 tsFloat, okay := tsIntf.(float32)
130 if okay {
131 // Voltha-Protos 3.2.2 and below
132 return time.Unix(int64(tsFloat), 0), nil
133 }
Scott Bakera1e53fa2020-04-01 11:13:34 -0700134 tsInt64, okay := tsIntf.(int64)
135 if okay {
136 if tsInt64 > 10000000000000 {
137 // sometimes it's in nanoseconds
138 return time.Unix(tsInt64/1000000000, tsInt64%1000000000), nil
139 } else {
140 // sometimes it's in seconds
141 return time.Unix(tsInt64/1000, 0), nil
142 }
143 }
Scott Baker2fe436a2020-02-10 17:21:47 -0800144 return time.Time{}, errors.New("Failed to decode timestamp")
145}
146
Scott Bakered4efab2020-01-13 19:12:25 -0800147// Extract the header, as well as a few other items that might be of interest
Scott Baker9173ed82020-05-19 08:30:12 -0700148func DecodeHeader(b []byte, ts time.Time) (*EventHeader, error) {
149 m := &voltha.Event{}
150 if err := proto.Unmarshal(b, m); err != nil {
Scott Bakered4efab2020-01-13 19:12:25 -0800151 return nil, err
152 }
153
Scott Baker9173ed82020-05-19 08:30:12 -0700154 header := m.Header
155 cat := voltha.EventCategory_Types_name[int32(header.Category)]
156 subCat := voltha.EventSubCategory_Types_name[int32(header.SubCategory)]
157 evType := voltha.EventType_Types_name[int32(header.Type)]
158 raised, err := DecodeTimestamp(header.RaisedTs)
Scott Bakered4efab2020-01-13 19:12:25 -0800159 if err != nil {
160 return nil, err
161 }
Scott Baker9173ed82020-05-19 08:30:12 -0700162 reported, err := DecodeTimestamp(header.ReportedTs)
Scott Baker2fe436a2020-02-10 17:21:47 -0800163 if err != nil {
164 return nil, err
165 }
Scott Bakered4efab2020-01-13 19:12:25 -0800166
167 // Opportunistically try to extract the device_id and title from a kpi_event2
168 // note that there might actually be multiple_slice data, so there could
169 // be multiple device_id, multiple title, etc.
170 device_ids := make(map[string]interface{})
171 titles := make(map[string]interface{})
172
Scott Baker9173ed82020-05-19 08:30:12 -0700173 kpi := m.GetKpiEvent2()
174 if kpi != nil {
175 sliceList := kpi.SliceData
176 if len(sliceList) > 0 {
177 slice := sliceList[0]
178 if slice != nil {
179 metadata := slice.Metadata
180 deviceId := metadata.DeviceId
181 device_ids[deviceId] = slice
182
183 title := metadata.Title
184 titles[title] = slice
Scott Bakered4efab2020-01-13 19:12:25 -0800185 }
186 }
187 }
188
189 // Opportunistically try to pull a resource_id and title from a DeviceEvent
190 // There can only be one resource_id and title from a DeviceEvent, so it's easier
191 // than dealing with KPI_EVENT2.
Scott Baker9173ed82020-05-19 08:30:12 -0700192 deviceEvent := m.GetDeviceEvent()
193 if deviceEvent != nil {
194 titles[deviceEvent.DeviceEventName] = deviceEvent
195 device_ids[deviceEvent.ResourceId] = deviceEvent
Scott Bakered4efab2020-01-13 19:12:25 -0800196 }
197
198 device_id_keys := make([]string, len(device_ids))
199 i := 0
David K. Bainbridge2b627612020-02-18 14:50:13 -0800200 for k := range device_ids {
Scott Bakered4efab2020-01-13 19:12:25 -0800201 device_id_keys[i] = k
202 i++
203 }
204
205 title_keys := make([]string, len(titles))
206 i = 0
David K. Bainbridge2b627612020-02-18 14:50:13 -0800207 for k := range titles {
Scott Bakered4efab2020-01-13 19:12:25 -0800208 title_keys[i] = k
209 i++
210 }
211
Scott Baker9173ed82020-05-19 08:30:12 -0700212 evHeader := EventHeader{Category: cat,
213 SubCategory: subCat,
214 Type: evType,
Scott Bakered4efab2020-01-13 19:12:25 -0800215 Raised_ts: raised,
216 Reported_ts: reported,
217 Device_ids: device_id_keys,
Scott Baker2fe436a2020-02-10 17:21:47 -0800218 Timestamp: ts,
Scott Bakered4efab2020-01-13 19:12:25 -0800219 Titles: title_keys}
220
221 return &evHeader, nil
222}
223
224// Print the full message, either in JSON or in GRPCURL-human-readable format,
225// depending on which grpcurl formatter is passed in.
Scott Baker9173ed82020-05-19 08:30:12 -0700226func PrintMessage(outputAs string, b []byte) error {
227 ms := &voltha.Event{}
228 if err := proto.Unmarshal(b, ms); err != nil {
Scott Bakered4efab2020-01-13 19:12:25 -0800229 return err
230 }
Scott Baker9173ed82020-05-19 08:30:12 -0700231
232 if outputAs == "json" {
233 marshaler := jsonpb.Marshaler{EmitDefaults: true, AnyResolver: &VolthaAnyResolver{}}
234 asJson, err := marshaler.MarshalToString(ms)
235 if err != nil {
236 return fmt.Errorf("Failed to marshal the json: %s", err)
237 }
238 fmt.Println(asJson)
239 } else {
240 // print in golang native format
241 fmt.Printf("%v\n", ms)
Scott Bakered4efab2020-01-13 19:12:25 -0800242 }
Scott Baker9173ed82020-05-19 08:30:12 -0700243
Scott Bakered4efab2020-01-13 19:12:25 -0800244 return nil
245}
246
247// Print just the enriched EventHeader. This is either in JSON format, or in
248// table format.
249func PrintEventHeader(outputAs string, outputFormat string, hdr *EventHeader) error {
250 if outputAs == "json" {
251 asJson, err := json.Marshal(hdr)
252 if err != nil {
253 return fmt.Errorf("Error marshalling JSON: %v", err)
254 } else {
255 fmt.Printf("%s\n", asJson)
256 }
257 } else {
258 f := format.Format(outputFormat)
259 output, err := f.ExecuteFixedWidth(DefaultWidths, false, *hdr)
260 if err != nil {
261 return err
262 }
263 fmt.Printf("%s\n", output)
264 }
265 return nil
266}
267
Scott Bakered4efab2020-01-13 19:12:25 -0800268// Start output, print any column headers or other start characters
269func (options *EventListenOpts) StartOutput(outputFormat string) error {
270 if options.OutputAs == "json" {
271 fmt.Println("[")
272 } else if (options.OutputAs == "table") && !options.ShowBody {
273 f := format.Format(outputFormat)
274 output, err := f.ExecuteFixedWidth(DefaultWidths, true, nil)
275 if err != nil {
276 return err
277 }
278 fmt.Println(output)
279 }
280 return nil
281}
282
283// Finish output, print any column footers or other end characters
284func (options *EventListenOpts) FinishOutput() {
285 if options.OutputAs == "json" {
286 fmt.Println("]")
287 }
288}
289
290func (options *EventListenOpts) Execute(args []string) error {
291 ProcessGlobalOptions()
292 if GlobalConfig.Kafka == "" {
293 return errors.New("Kafka address is not specified")
294 }
295
Scott Bakered4efab2020-01-13 19:12:25 -0800296 config := sarama.NewConfig()
297 config.ClientID = "go-kafka-consumer"
298 config.Consumer.Return.Errors = true
299 config.Version = sarama.V1_0_0_0
300 brokers := []string{GlobalConfig.Kafka}
301
302 client, err := sarama.NewClient(brokers, config)
303 if err != nil {
304 return err
305 }
306
307 defer func() {
308 if err := client.Close(); err != nil {
309 panic(err)
310 }
311 }()
312
313 consumer, consumerErrors, highwaterMarks, err := startConsumer([]string{"voltha.events"}, client)
314 if err != nil {
315 return err
316 }
317
318 highwater := highwaterMarks["voltha.events"]
319
320 signals := make(chan os.Signal, 1)
321 signal.Notify(signals, os.Interrupt)
322
323 // Count how many message processed
324 consumeCount := 0
325
326 // Count how many messages were printed
327 count := 0
328
Scott Bakered4efab2020-01-13 19:12:25 -0800329 var headerFilter *filter.Filter
330 if options.Filter != "" {
331 headerFilterVal, err := filter.Parse(options.Filter)
332 if err != nil {
333 return fmt.Errorf("Failed to parse filter: %v", err)
334 }
335 headerFilter = &headerFilterVal
336 }
337
338 outputFormat := CharReplacer.Replace(options.Format)
339 if outputFormat == "" {
340 outputFormat = GetCommandOptionWithDefault("events-listen", "format", DEFAULT_EVENT_FORMAT)
341 }
342
343 err = options.StartOutput(outputFormat)
344 if err != nil {
345 return err
346 }
347
Scott Bakerf05e60a2020-02-02 21:53:57 -0800348 var since *time.Time
349 if options.Since != "" {
350 since, err = ParseSince(options.Since)
351 if err != nil {
352 return err
353 }
354 }
355
Scott Bakered4efab2020-01-13 19:12:25 -0800356 // Get signnal for finish
357 doneCh := make(chan struct{})
358 go func() {
Scott Baker2fe436a2020-02-10 17:21:47 -0800359 tStart := time.Now()
Scott Bakered4efab2020-01-13 19:12:25 -0800360 Loop:
361 for {
362 // Initialize the idle timeout timer
363 timeoutTimer := time.NewTimer(time.Duration(options.Timeout) * time.Second)
364 select {
365 case msg := <-consumer:
366 consumeCount++
Scott Baker9173ed82020-05-19 08:30:12 -0700367 hdr, err := DecodeHeader(msg.Value, msg.Timestamp)
Scott Bakered4efab2020-01-13 19:12:25 -0800368 if err != nil {
369 log.Printf("Error decoding header %v\n", err)
370 continue
371 }
372 if headerFilter != nil && !headerFilter.Evaluate(*hdr) {
373 // skip printing message
Scott Baker2fe436a2020-02-10 17:21:47 -0800374 } else if since != nil && hdr.Timestamp.Before(*since) {
Scott Bakerf05e60a2020-02-02 21:53:57 -0800375 // it's too old
Scott Bakered4efab2020-01-13 19:12:25 -0800376 } else {
377 // comma separated between this message and predecessor
378 if count > 0 {
379 if options.OutputAs == "json" {
380 fmt.Println(",")
381 }
382 }
383
384 // Print it
385 if options.ShowBody {
Scott Baker9173ed82020-05-19 08:30:12 -0700386 if err := PrintMessage(options.OutputAs, msg.Value); err != nil {
David K. Bainbridge2b627612020-02-18 14:50:13 -0800387 log.Printf("%v\n", err)
388 }
Scott Bakered4efab2020-01-13 19:12:25 -0800389 } else {
David K. Bainbridge2b627612020-02-18 14:50:13 -0800390 if err := PrintEventHeader(options.OutputAs, outputFormat, hdr); err != nil {
Scott Bakered4efab2020-01-13 19:12:25 -0800391 log.Printf("%v\n", err)
392 }
393 }
394
395 // Check to see if we've hit the "count" threshold the user specified
396 count++
397 if (options.Count > 0) && (count >= options.Count) {
398 log.Println("Count reached")
399 doneCh <- struct{}{}
400 break Loop
401 }
402
403 // Check to see if we've hit the "now" threshold the user specified
Scott Baker2fe436a2020-02-10 17:21:47 -0800404 if (options.Now) && (!hdr.Timestamp.Before(tStart)) {
Scott Bakered4efab2020-01-13 19:12:25 -0800405 log.Println("Now timestamp reached")
406 doneCh <- struct{}{}
407 break Loop
408 }
409 }
410
411 // If we're not in follow mode, see if we hit the highwater mark
412 if !options.Follow && !options.Now && (msg.Offset >= highwater) {
413 log.Println("High water reached")
414 doneCh <- struct{}{}
415 break Loop
416 }
417
418 // Reset the timeout timer
419 if !timeoutTimer.Stop() {
420 <-timeoutTimer.C
421 }
422 case consumerError := <-consumerErrors:
423 log.Printf("Received consumerError topic=%v, partition=%v, err=%v\n", string(consumerError.Topic), string(consumerError.Partition), consumerError.Err)
424 doneCh <- struct{}{}
425 case <-signals:
426 doneCh <- struct{}{}
427 case <-timeoutTimer.C:
428 log.Printf("Idle timeout\n")
429 doneCh <- struct{}{}
430 }
431 }
432 }()
433
434 <-doneCh
435
436 options.FinishOutput()
437
438 log.Printf("Consumed %d messages. Printed %d messages", consumeCount, count)
439
440 return nil
441}
442
443// Consume message from Sarama and send them out on a channel.
444// Supports multiple topics.
445// Taken from Sarama example consumer.
446func startConsumer(topics []string, client sarama.Client) (chan *sarama.ConsumerMessage, chan *sarama.ConsumerError, map[string]int64, error) {
447 master, err := sarama.NewConsumerFromClient(client)
448 if err != nil {
449 return nil, nil, nil, err
450 }
451
452 consumers := make(chan *sarama.ConsumerMessage)
453 errors := make(chan *sarama.ConsumerError)
454 highwater := make(map[string]int64)
455 for _, topic := range topics {
456 if strings.Contains(topic, "__consumer_offsets") {
457 continue
458 }
459 partitions, _ := master.Partitions(topic)
460
461 // TODO: Add support for multiple partitions
462 if len(partitions) > 1 {
463 log.Printf("WARNING: %d partitions on topic %s but we only listen to the first one\n", len(partitions), topic)
464 }
465
466 hw, err := client.GetOffset("voltha.events", partitions[0], sarama.OffsetNewest)
467 if err != nil {
468 return nil, nil, nil, fmt.Errorf("Error in consume() getting highwater: Topic %v Partitions: %v", topic, partitions)
469 }
470 highwater[topic] = hw
471
472 consumer, err := master.ConsumePartition(topic, partitions[0], sarama.OffsetOldest)
473 if nil != err {
474 return nil, nil, nil, fmt.Errorf("Error in consume(): Topic %v Partitions: %v", topic, partitions)
475 }
476 log.Println(" Start consuming topic ", topic)
477 go func(topic string, consumer sarama.PartitionConsumer) {
478 for {
479 select {
480 case consumerError := <-consumer.Errors():
481 errors <- consumerError
482
483 case msg := <-consumer.Messages():
484 consumers <- msg
485 }
486 }
487 }(topic, consumer)
488 }
489
490 return consumers, errors, highwater, nil
491}