diff --git a/input_kafka.go b/input_kafka.go new file mode 100644 index 0000000..5420340 --- /dev/null +++ b/input_kafka.go @@ -0,0 +1,93 @@ +package main + +import ( + "encoding/json" + "log" + "time" + + "github.com/Shopify/sarama" +) + +// KafkaInput should make consumer client. +type KafkaInput struct { + config *KafkaConfig + consumers []sarama.PartitionConsumer + messages chan *sarama.ConsumerMessage +} + +// NewKafkaInput constructor for KafkaInput +func NewKafkaInput(address string, config *KafkaConfig) *KafkaInput { + c := sarama.NewConfig() + // c.Consumer. + + con, err := sarama.NewConsumer([]string{config.host}, c) + if err != nil { + log.Fatalln("Failed to start Sarama(Kafka) consumer:", err) + } + + partitions, err := con.Partitions(config.topic) + if err != nil { + log.Fatalln("Failed to collect Sarama(Kafka) partitions:", err) + } + + i := &KafkaInput{ + config: config, + consumers: make([]sarama.PartitionConsumer, len(partitions)), + messages: make(chan *sarama.ConsumerMessage, 256), + } + + for index, partition := range partitions { + consumer, err := con.ConsumePartition(config.topic, partition, sarama.OffsetNewest) + if err != nil { + log.Fatalln("Failed to start Sarama(Kafka) partition consumer:", err) + } + + go func(consumer sarama.PartitionConsumer) { + defer consumer.Close() + + for message := range consumer.Messages() { + i.messages <- message + } + }(consumer) + + if Settings.verbose { + // Start infinite loop for tracking errors for kafka producer. + go i.ErrorHandler(consumer) + } + + i.consumers[index] = consumer + } + + return i +} + +// ErrorHandler should receive errors +func (i *KafkaInput) ErrorHandler(consumer sarama.PartitionConsumer) { + for err := range consumer.Errors() { + log.Println("Failed to read access log entry:", err) + } +} + +func (i *KafkaInput) Read(data []byte) (int, error) { + message := <-i.messages + + var kafkaMessage KafkaMessage + json.Unmarshal(message.Value, &kafkaMessage) + + buf, err := kafkaMessage.Dump() + if err != nil { + log.Println("Failed to decode access log entry:", err) + return 0, err + } + + header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1) + + copy(data[0:len(header)], header) + copy(data[len(header):], buf) + + return len(buf) + len(header), nil +} + +func (i *KafkaInput) String() string { + return "Kafka Input: " + i.config.host + "/" + i.config.topic +} diff --git a/kafka.go b/kafka.go new file mode 100644 index 0000000..ef0849d --- /dev/null +++ b/kafka.go @@ -0,0 +1,42 @@ +package main + +import ( + "bytes" + "fmt" + + "github.com/buger/gor/proto" +) + +// KafkaConfig should contains required information to +// build producers. +type KafkaConfig struct { + host string + topic string +} + +// 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"` + ReqBody string `json:"Req_Body,omitempty"` + ReqHeaders map[string]string `json:"Req_Headers,omitempty"` +} + +// Dump returns the given request in its HTTP/1.x wire +// representation. +func (m KafkaMessage) Dump() ([]byte, error) { + var b bytes.Buffer + + b.WriteString(fmt.Sprintf("%s %s HTTP/1.1", m.ReqMethod, m.ReqURL)) + b.Write(proto.CLRF) + for key, value := range m.ReqHeaders { + b.WriteString(fmt.Sprintf("%s: %s", key, value)) + b.Write(proto.CLRF) + } + + b.Write(proto.CLRF) + b.WriteString(m.ReqBody) + + return b.Bytes(), nil +} diff --git a/output_kafka.go b/output_kafka.go index 4ebfc4c..77c4a35 100644 --- a/output_kafka.go +++ b/output_kafka.go @@ -2,20 +2,14 @@ package main import ( "encoding/json" - "github.com/Shopify/sarama" - "github.com/buger/gor/proto" "io" "log" "strings" "time" -) -// KafkaConfig should contains required information to -// build producers. -type KafkaConfig struct { - host string - topic string -} + "github.com/Shopify/sarama" + "github.com/buger/gor/proto" +) // KafkaOutput should make producer client. type KafkaOutput struct { @@ -23,15 +17,6 @@ type KafkaOutput struct { 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"` - ReqBody string `json:"Req_Body,omitempty"` - ReqHeaders map[string]string `json:"Req_Headers,omitempty"` -} - // KafkaOutputFrequency in milliseconds const KafkaOutputFrequency = 500 diff --git a/plugins.go b/plugins.go index 6ee1516..a5a9113 100644 --- a/plugins.go +++ b/plugins.go @@ -146,4 +146,8 @@ func InitPlugins() { if Settings.outputKafkaConfig.host != "" && Settings.outputKafkaConfig.topic != "" { registerPlugin(NewKafkaOutput, "", &Settings.outputKafkaConfig) } + + if Settings.inputKafkaConfig.host != "" && Settings.inputKafkaConfig.topic != "" { + registerPlugin(NewKafkaInput, "", &Settings.inputKafkaConfig) + } } diff --git a/settings.go b/settings.go index ac62f7c..274ffff 100644 --- a/settings.go +++ b/settings.go @@ -59,6 +59,7 @@ type AppSettings struct { outputHTTPConfig HTTPOutputConfig modifierConfig HTTPModifierConfig + inputKafkaConfig KafkaConfig outputKafkaConfig KafkaConfig } @@ -128,8 +129,11 @@ func init() { 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.host, "output-kafka-host", "", "Send request and response stats to Kafka:\n\tgor --input-raw :8080 --output-kafka-host '192.168.0.1:9092,192.168.0.2:9092'") - 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.StringVar(&Settings.outputKafkaConfig.host, "output-kafka-host", "", "Read request and response stats from Kafka:\n\tgor --input-raw :8080 --output-kafka-host '192.168.0.1:9092,192.168.0.2:9092'") + flag.StringVar(&Settings.outputKafkaConfig.topic, "output-kafka-topic", "", "Read request and response stats from Kafka:\n\tgor --input-raw :8080 --output-kafka-topic 'kafka-log'") + + flag.StringVar(&Settings.inputKafkaConfig.host, "input-kafka-host", "", "Send request and response stats to Kafka:\n\tgor --output-stdout --input-kafka-host '192.168.0.1:9092,192.168.0.2:9092'") + flag.StringVar(&Settings.inputKafkaConfig.topic, "input-kafka-topic", "", "Send request and response stats to Kafka:\n\tgor --output-stdout --input-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'") flag.Var(&Settings.modifierConfig.headers, "output-http-header", "WARNING: `--output-http-header` DEPRECATED, use `--http-set-header` instead")