From e33c8aaafc8f156394bdb3ec0f1cfab56cd96eb0 Mon Sep 17 00:00:00 2001 From: gdupont Date: Mon, 8 Oct 2018 23:49:11 +0200 Subject: [PATCH 1/3] simple http producer to kafka topic --- README.md | 46 ++++++++++++++++++++++++++++++++++++++- cons/consumerTest.go | 34 +++++++++++++++++++++++++++++ dummyRequest.json | 3 +++ testserver4g.go | 52 +++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 133 insertions(+), 2 deletions(-) create mode 100644 cons/consumerTest.go create mode 100644 dummyRequest.json diff --git a/README.md b/README.md index f6853ad..ca4120c 100644 --- a/README.md +++ b/README.md @@ -1 +1,45 @@ -`go run testserver4g.go` \ No newline at end of file +# Golan go + +From basic HTTP server to gateway for kafka. + +## Run the server + +TODO: + +- install go: https://golang.org/doc/install +- compile +- run + +## Launch kafka + +### From source + +https://kafka.apache.org/quickstart + +### With docker-compose + +https://github.com/simplesteph/kafka-stack-docker-compose + +## Testing Confluent's Golang Client for Apache KafkaTM + +From https://github.com/confluentinc/confluent-kafka-go + +1) Install librdkafka: + +``` shell +git clone https://github.com/edenhill/librdkafka.git +cd librdkafka +./configure --prefix /usr +make +sudo make install +``` +2) Install go client + +```shell +go get -u github.com/confluentinc/confluent-kafka-go/kafka +``` +## Query the HTTP server + +```shell +curl --header "Content-Type: application/json" --request POST --data '{"message":"everything is awesome"}' 'http://localhost:8080/' +``` \ No newline at end of file diff --git a/cons/consumerTest.go b/cons/consumerTest.go new file mode 100644 index 0000000..eb5d087 --- /dev/null +++ b/cons/consumerTest.go @@ -0,0 +1,34 @@ +package main + +import ( + "fmt" + + "github.com/confluentinc/confluent-kafka-go/kafka" +) + +func main() { + + c, err := kafka.NewConsumer(&kafka.ConfigMap{ + "bootstrap.servers": "localhost", + "group.id": "myGroup", + "auto.offset.reset": "earliest", + }) + + if err != nil { + panic(err) + } + + c.SubscribeTopics([]string{"incoming_req", "^aRegex.*[Tt]opic"}, nil) + + for { + msg, err := c.ReadMessage(-1) + if err == nil { + fmt.Printf("Message on %s: %s\n", msg.TopicPartition, string(msg.Value)) + } else { + fmt.Printf("Consumer error: %v (%v)\n", err, msg) + break + } + } + + c.Close() +} diff --git a/dummyRequest.json b/dummyRequest.json new file mode 100644 index 0000000..c07492e --- /dev/null +++ b/dummyRequest.json @@ -0,0 +1,3 @@ +{ + "message" : "everything is awesome!" +} \ No newline at end of file diff --git a/testserver4g.go b/testserver4g.go index ca1759c..98c0d54 100644 --- a/testserver4g.go +++ b/testserver4g.go @@ -2,11 +2,16 @@ package main import ( "encoding/json" + "fmt" "log" "net/http" + "net/url" + + "github.com/confluentinc/confluent-kafka-go/kafka" ) func main() { + // start HTTP server log.Print("Starting test server for G") http.HandleFunc("/", index) log.Fatal(http.ListenAndServe(":8080", nil)) @@ -20,20 +25,65 @@ type response struct { Ok bool `json:"ok"` } +type click struct { + referer string + URL *url.URL + datetime int64 +} + func index(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") decoder := json.NewDecoder(r.Body) var request request if err := decoder.Decode(&request); err != nil { - http.Error(w, `{"error_message": "internal error"}`, http.StatusBadRequest) + http.Error(w, `{"error_message": "internal error - json request expected"}`, http.StatusBadRequest) return } + + // TODO process request and push to topic log.Printf("host: %v, referer: %v", r.Host, r.Referer()) log.Printf("language: %v", request.Language) log.Printf("headers:") for i := range r.Header { log.Printf("\t%v => %v", i, r.Header.Get(i)) } + + log.Printf("body:") + + // initiate kafka connection + p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost"}) + if err != nil { + panic(err) + } + defer p.Close() + + // push to topic + topic := "incoming_req" + + p.Produce(&kafka.Message{ + TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny}, + Value: []byte(`"referer": "` + r.Host + `"`), + }, nil) + + // Delivery report handler for produced messages + go func() { + for e := range p.Events() { + switch ev := e.(type) { + case *kafka.Message: + if ev.TopicPartition.Error != nil { + fmt.Printf("Delivery failed: %v\n", ev.TopicPartition) + } else { + fmt.Printf("Delivered message to %v\n", ev.TopicPartition) + } + } + } + }() + + // Wait for message deliveries + p.Flush(15 * 1000) + + // write http response payload, err := json.Marshal(response{ Ok: true, }) From c2fd45aa6892b0284c6014e7415ab52b8fcd6729 Mon Sep 17 00:00:00 2001 From: gdupont Date: Mon, 8 Oct 2018 23:57:17 +0200 Subject: [PATCH 2/3] todo list --- README.md | 47 ++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 40 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index ca4120c..48d6717 100644 --- a/README.md +++ b/README.md @@ -2,13 +2,27 @@ From basic HTTP server to gateway for kafka. -## Run the server +## Work + +### DONE + +- [x] launch the http server +- [x] explore request headers +- [x] launch kafka +- [x] create topic, produce/consume messages +- [x] make the server produces message for each incoming request + +### IN PROGRESS -TODO: +- [ ] explore request content +- [ ] define a data model to handle the request data -- install go: https://golang.org/doc/install -- compile -- run +### TODO + +- [ ] test on kafka cluster +- [ ] optimize kafka connectio creation +- [ ] check error handling +- [ ] package in docker + docker-compose (?) ## Launch kafka @@ -16,11 +30,13 @@ TODO: https://kafka.apache.org/quickstart -### With docker-compose +### With docker-compose in cluster mode https://github.com/simplesteph/kafka-stack-docker-compose -## Testing Confluent's Golang Client for Apache KafkaTM +## Run the server + +### Installing Confluent's Golang Client for Apache KafkaTM From https://github.com/confluentinc/confluent-kafka-go @@ -38,6 +54,23 @@ sudo make install ```shell go get -u github.com/confluentinc/confluent-kafka-go/kafka ``` + +### Run baby run! + +```shell +cd golang_helloooo +go build +./golang_helloooo +``` + +### Launch the consumer + +```shell +cd golang_helloooo/cons +go build +./cons +``` + ## Query the HTTP server ```shell From c41578fcdfa573432f0746423de8410c052efe2d Mon Sep 17 00:00:00 2001 From: gdupont Date: Wed, 10 Oct 2018 09:36:27 +0200 Subject: [PATCH 3/3] adding WIP dockerfile --- Dockerfile | 22 ++++++++++++++++++++++ README.md | 6 +++++- cons/consumerTest.go | 2 +- testserver4g.go | 2 +- 4 files changed, 29 insertions(+), 3 deletions(-) create mode 100644 Dockerfile diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..b68d1aa --- /dev/null +++ b/Dockerfile @@ -0,0 +1,22 @@ +FROM ubuntu:18.04 + +RUN apt-get update && apt-get install -y git make g++ + +# Installing Confluent's Golang Client for Apache KafkaTM +RUN \ +git clone https://github.com/edenhill/librdkafka.git && \ +cd librdkafka && \ +./configure --prefix /usr && \ +make && \ +make install + +# Install go client +RUN apt-get install -y golang-go +RUN go get -u github.com/confluentinc/confluent-kafka-go/kafka + +# Install golang_helloooo +RUN \ +git clone https://github.com/ggdupont/golang_helloooo.git && \ +cd golang_helloooo && \ +go build && \ +./golang_helloooo diff --git a/README.md b/README.md index 48d6717..ec39748 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ From basic HTTP server to gateway for kafka. ### IN PROGRESS +- [ ] package in docker + docker-compose (?) - [ ] explore request content - [ ] define a data model to handle the request data @@ -22,7 +23,6 @@ From basic HTTP server to gateway for kafka. - [ ] test on kafka cluster - [ ] optimize kafka connectio creation - [ ] check error handling -- [ ] package in docker + docker-compose (?) ## Launch kafka @@ -34,6 +34,10 @@ https://kafka.apache.org/quickstart https://github.com/simplesteph/kafka-stack-docker-compose +```shell +docker-compose -f zk-single-kafka-single.yml up +``` + ## Run the server ### Installing Confluent's Golang Client for Apache KafkaTM diff --git a/cons/consumerTest.go b/cons/consumerTest.go index eb5d087..57ceed8 100644 --- a/cons/consumerTest.go +++ b/cons/consumerTest.go @@ -9,7 +9,7 @@ import ( func main() { c, err := kafka.NewConsumer(&kafka.ConfigMap{ - "bootstrap.servers": "localhost", + "bootstrap.servers": "localhost:9092", "group.id": "myGroup", "auto.offset.reset": "earliest", }) diff --git a/testserver4g.go b/testserver4g.go index 98c0d54..df62e2b 100644 --- a/testserver4g.go +++ b/testserver4g.go @@ -52,7 +52,7 @@ func index(w http.ResponseWriter, r *http.Request) { log.Printf("body:") // initiate kafka connection - p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost"}) + p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "localhost:9092"}) if err != nil { panic(err) }