diff --git a/elasticsearch.go b/elasticsearch.go index a626853..2635b89 100644 --- a/elasticsearch.go +++ b/elasticsearch.go @@ -84,9 +84,11 @@ func (p *ESPlugin) Init(URI string) { p.done = make(chan bool) p.indexor.Start() - // Only start the ErrorHandler goroutine when in verbose mode - // no need to burn ressources otherwise - go p.ErrorHandler() + if Settings.verbose { + // Only start the ErrorHandler goroutine when in verbose mode + // no need to burn ressources otherwise + go p.ErrorHandler() + } log.Println("Initialized Elasticsearch Plugin") return diff --git a/examples/server.go b/examples/server.go new file mode 100644 index 0000000..8be30c6 --- /dev/null +++ b/examples/server.go @@ -0,0 +1,15 @@ +package main + +import ( + "io" + "net/http" +) + +func hello(w http.ResponseWriter, r *http.Request) { + io.WriteString(w, "Hello world!") +} + +func main() { + http.HandleFunc("/", hello) + http.ListenAndServe(":8000", nil) +} diff --git a/output_kafka.go b/output_kafka.go index cf2ea1f..9af4ca0 100644 --- a/output_kafka.go +++ b/output_kafka.go @@ -1,7 +1,10 @@ package main import ( + "encoding/json" "github.com/Shopify/sarama" + "github.com/buger/gor/proto" + "io" "log" "strings" "time" @@ -10,25 +13,41 @@ import ( // KafkaConfig should contains required information to // build producers. type KafkaConfig struct { - zookeeper string - topic string + host string + topic string } // KafkaOutput should make producer client. type KafkaOutput struct { - address string config *KafkaConfig producer sarama.AsyncProducer } +// KafkaMessage should contains catched request information that should be +// passed as Json to Apache Kafka. +type KafkaMessage struct { + ReqURL string `json:"Req_URL"` + ReqMethod string `json:"Req_Method"` + ReqUserAgent string `json:"Req_User-Agent"` + ReqAcceptLanguage string `json:"Req_Accept-Language,omitempty"` + ReqAccept string `json:"Req_Accept,omitempty"` + ReqAcceptEncoding string `json:"Req_Accept-Encoding,omitempty"` + ReqIfModifiedSince string `json:"Req_If-Modified-Since,omitempty"` + ReqConnection string `json:"Req_Connection,omitempty"` + ReqCookies string `json:"Req_Cookies,omitempty"` +} + +// KafkaOutputFrequency in milliseconds +const KafkaOutputFrequency = 500 + // NewKafkaOutput creates instance of kafka producer client. -func NewKafkaOutput(address string, config *KafkaConfig) *KafkaOutput { +func NewKafkaOutput(address string, config *KafkaConfig) io.Writer { c := sarama.NewConfig() c.Producer.RequiredAcks = sarama.WaitForLocal c.Producer.Compression = sarama.CompressionSnappy - c.Producer.Flush.Frequency = 500 * time.Millisecond + c.Producer.Flush.Frequency = KafkaOutputFrequency * time.Millisecond - brokerList := strings.Split(config.zookeeper, ",") + brokerList := strings.Split(config.host, ",") producer, err := sarama.NewAsyncProducer(brokerList, c) if err != nil { @@ -36,13 +55,14 @@ func NewKafkaOutput(address string, config *KafkaConfig) *KafkaOutput { } o := &KafkaOutput{ - address: address, config: config, producer: producer, } - // Start infinite loop for tracking errors for kafka producer. - go o.ErrorHandler() + if Settings.verbose { + // Start infinite loop for tracking errors for kafka producer. + go o.ErrorHandler() + } return o } @@ -55,14 +75,24 @@ func (o *KafkaOutput) ErrorHandler() { } func (o *KafkaOutput) Write(data []byte) (n int, err error) { - buf := make(sarama.ByteEncoder, len(data)) - copy(buf, data) + kafkaMessage := KafkaMessage{ + ReqURL: string(proto.Path(data)), + ReqMethod: string(proto.Method(data)), + ReqUserAgent: string(proto.Header(data, []byte("User-Agent"))), + ReqAcceptLanguage: string(proto.Header(data, []byte("Accept-Language"))), + ReqAccept: string(proto.Header(data, []byte("Accept"))), + ReqAcceptEncoding: string(proto.Header(data, []byte("Accept-Encoding"))), + ReqIfModifiedSince: string(proto.Header(data, []byte("If-Modified-Since"))), + ReqConnection: string(proto.Header(data, []byte("Connection"))), + ReqCookies: string(proto.Header(data, []byte("Cookie"))), + } + jsonMessage, _ := json.Marshal(&kafkaMessage) + message := sarama.StringEncoder(jsonMessage) o.producer.Input() <- &sarama.ProducerMessage{ Topic: o.config.topic, - Key: sarama.StringEncoder(o.address), - Value: buf, + Value: message, } - return len(data), nil + return len(message), nil } diff --git a/plugins.go b/plugins.go index 1d11971..a625f9b 100644 --- a/plugins.go +++ b/plugins.go @@ -143,5 +143,5 @@ func InitPlugins() { registerPlugin(NewHTTPOutput, options, &Settings.outputHTTPConfig) } - registerPlugin(NewKafkaOutput, &Settings.outputKafkaConfig) + registerPlugin(NewKafkaOutput, "", &Settings.outputKafkaConfig) } diff --git a/settings.go b/settings.go index 0b4973b..ff417ca 100644 --- a/settings.go +++ b/settings.go @@ -127,7 +127,8 @@ func init() { flag.BoolVar(&Settings.outputHTTPConfig.Debug, "output-http-debug", false, "Enables http debug output.") flag.StringVar(&Settings.outputHTTPConfig.elasticSearch, "output-http-elasticsearch", "", "Send request and response stats to ElasticSearch:\n\tgor --input-raw :8080 --output-http staging.com --output-http-elasticsearch 'es_host:api_port/index_name'") - flag.StringVar(&Settings.outputKafkaConfig.zookeeper, "output-kafka-zookeeper", "", "Send request and response stats to Kafka:\n\tgor --input-raw :8080 --output-kafka-zookeeper '192.168.0.1:2181,192.168.0.2:2181'") + + flag.StringVar(&Settings.outputKafkaConfig.host, "output-kafka-host", "", "Send request and response stats to Kafka:\n\tgor --input-raw :8080 --output-kafka-host '192.168.0.1:2181,192.168.0.2:2181'") flag.StringVar(&Settings.outputKafkaConfig.topic, "output-kafka-topic", "", "Send request and response stats to Kafka:\n\tgor --input-raw :8080 --output-kafka-topic 'kafka-log'") flag.Var(&Settings.modifierConfig.headers, "http-set-header", "Inject additional headers to http reqest:\n\tgor --input-raw :8080 --output-http staging.com --http-set-header 'User-Agent: Gor'")