/
main.go
117 lines (96 loc) · 4.16 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
package main
import (
"os"
"os/signal"
"runtime/pprof"
"strings"
"syscall"
"time"
"code.cloudfoundry.org/lager"
"github.com/cloudfoundry-community/go-cfclient"
"github.com/humio/cloudfoundry2humio/caching"
"github.com/humio/cloudfoundry2humio/humio"
"github.com/humio/cloudfoundry2humio/nozzle"
"gopkg.in/alecthomas/kingpin.v2"
)
const (
firehoseSubscriptionID = "humio-nozzle"
version = "0.1.0"
)
var (
apiAddress = kingpin.Flag("api-addr", "Api URL").OverrideDefaultFromEnvar("API_ADDR").Required().String()
dopplerAddress = kingpin.Flag("doppler-addr", "Traffic controller URL").OverrideDefaultFromEnvar("DOPPLER_ADDR").Required().String()
cfUser = kingpin.Flag("firehose-user", "CF user with admin and firehose access").OverrideDefaultFromEnvar("FIREHOSE_USER").Required().String()
cfPassword = kingpin.Flag("firehose-user-password", "Password of the CF user").OverrideDefaultFromEnvar("FIREHOSE_USER_PASSWORD").Required().String()
environment = kingpin.Flag("cf-environment", "CF environment name").OverrideDefaultFromEnvar("CF_ENVIRONMENT").Default("cf").String()
// comma separated list of types to exclude. For now use metric,log,http and revisit later
eventFilter = kingpin.Flag("eventFilter", "Comma separated list of types to exclude").Default("").OverrideDefaultFromEnvar("EVENT_FILTER").String()
skipSslValidation = kingpin.Flag("skip-ssl-validation", "Skip SSL validation").Default("false").OverrideDefaultFromEnvar("SKIP_SSL_VALIDATION").Bool()
idleTimeout = kingpin.Flag("idle-timeout", "Keep Alive duration for the firehose consumer").Default("25s").OverrideDefaultFromEnvar("IDLE_TIMEOUT").Duration()
logLevel = kingpin.Flag("log-level", "Log level: DEBUG, INFO, ERROR").Default("INFO").OverrideDefaultFromEnvar("LOG_LEVEL").String()
// Humio endpoint info
humioHost = kingpin.Flag("humio-host", "Humio host endpoint").OverrideDefaultFromEnvar("HUMIO_HOST").Required().String()
humioDataspace = kingpin.Flag("humio-dataspace", "Humio dataspace to push logs to").OverrideDefaultFromEnvar("HUMIO_DATASPACE").Required().String()
humioIngestToken = kingpin.Flag("humio-ingest-token", "Humio ingest token").OverrideDefaultFromEnvar("HUMIO_INGEST_TOKEN").Required().String()
)
func main() {
kingpin.Version(version)
kingpin.Parse()
logger := lager.NewLogger("humio-nozzle")
level := lager.INFO
switch strings.ToUpper(*logLevel) {
case "DEBUG":
level = lager.DEBUG
case "ERROR":
level = lager.ERROR
}
logger.RegisterSink(lager.NewWriterSink(os.Stdout, level))
// enable thread dump
threadDumpChan := registerGoRoutineDumpSignalChannel()
defer close(threadDumpChan)
go dumpGoRoutine(threadDumpChan)
cachingCFClientConfig := &cfclient.Config{
ApiAddress: *apiAddress,
Username: *cfUser,
Password: *cfPassword,
SkipSslValidation: *skipSslValidation,
}
cachingClient := caching.NewCaching(cachingCFClientConfig, logger, *environment)
firehoseCFClientConfig := &cfclient.Config{
ApiAddress: *apiAddress,
Username: *cfUser,
Password: *cfPassword,
SkipSslValidation: *skipSslValidation,
}
firehoseConfig := &nozzle.FirehoseConfig{
SubscriptionId: firehoseSubscriptionID,
TrafficControllerUrl: *dopplerAddress,
IdleTimeout: *idleTimeout,
}
firehoseClient := nozzle.NewFirehoseClient(firehoseCFClientConfig, firehoseConfig, logger)
humioConfig := &humio.HumioConfig{
Host: *humioHost,
Dataspace: *humioDataspace,
Token: *humioIngestToken,
}
humioClient := humio.NewHumioClient(humioConfig, logger)
nozzleConfig := &nozzle.NozzleConfig{
HumioBatchTime: 5 * time.Second,
HumioMaxMsgNumPerBatch: 500,
}
nozzleApp := nozzle.NewHumioNozzle(logger, firehoseClient, nozzleConfig, humioClient, cachingClient)
nozzleApp.Start()
}
func registerGoRoutineDumpSignalChannel() chan os.Signal {
threadDumpChan := make(chan os.Signal, 1)
signal.Notify(threadDumpChan, syscall.SIGUSR1)
return threadDumpChan
}
func dumpGoRoutine(dumpChan chan os.Signal) {
for range dumpChan {
goRoutineProfiles := pprof.Lookup("goroutine")
if goRoutineProfiles != nil {
goRoutineProfiles.WriteTo(os.Stdout, 2)
}
}
}