kraken

1
// Copyright (c) 2016-2019 Uber Technologies, Inc.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12
// See the License for the specific language governing permissions and
13
// limitations under the License.
14
package networkevent
15

16
import (
17
	"encoding/json"
18
	"errors"
19
	"fmt"
20
	"os"
21

22
	"github.com/uber/kraken/utils/log"
23
)
24

25
// Producer emits events.
26
type Producer interface {
27
	Produce(e *Event)
28
	Close() error
29
}
30

31
type producer struct {
32
	file *os.File
33
}
34

35
// NewProducer creates a new Producer.
36
func NewProducer(config Config) (Producer, error) {
37
	var f *os.File
38
	if config.Enabled {
39
		if config.LogPath == "" {
40
			return nil, errors.New("no log path supplied")
41
		}
42
		var flag int
43
		if _, err := os.Stat(config.LogPath); err != nil {
44
			if os.IsNotExist(err) {
45
				flag = os.O_WRONLY | os.O_CREATE | os.O_EXCL
46
			} else {
47
				return nil, fmt.Errorf("stat: %s", err)
48
			}
49
		} else {
50
			flag = os.O_WRONLY | os.O_APPEND
51
		}
52
		var err error
53
		f, err = os.OpenFile(config.LogPath, flag, 0775)
54
		if err != nil {
55
			return nil, fmt.Errorf("open %d: %s", flag, err)
56
		}
57
	} else {
58
		log.Warn("Kafka network events disabled")
59
	}
60
	return &producer{f}, nil
61
}
62

63
// Produce emits a network event.
64
func (p *producer) Produce(e *Event) {
65
	if p.file == nil {
66
		return
67
	}
68
	b, err := json.Marshal(e)
69
	if err != nil {
70
		log.Errorf("Error serializing network event to json: %s", err)
71
		return
72
	}
73
	line := append(b, byte('\n'))
74
	if _, err := p.file.Write(line); err != nil {
75
		log.Errorf("Error writing network event: %s", err)
76
		return
77
	}
78
}
79

80
// Close closes the producer.
81
func (p *producer) Close() error {
82
	if p.file == nil {
83
		return nil
84
	}
85
	return p.file.Close()
86
}
87