75f3163ed35ce03a7bf3d3121c77f88f19954d91
[nonrtric.git] / test / kafka-procon / main.go
1 // -
2 //   ========================LICENSE_START=================================
3 //   O-RAN-SC
4 //   %%
5 //   Copyright (C) 2020-2022: Nordix Foundation
6 //   %%
7 //   Licensed under the Apache License, Version 2.0 (the "License");
8 //   you may not use this file except in compliance with the License.
9 //   You may obtain a copy of the License at
10 //
11 //        http://www.apache.org/licenses/LICENSE-2.0
12 //
13 //   Unless required by applicable law or agreed to in writing, software
14 //   distributed under the License is distributed on an "AS IS" BASIS,
15 //   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16 //   See the License for the specific language governing permissions and
17 //   limitations under the License.
18 //   ========================LICENSE_END===================================
19 //
20
21 // Writing a basic HTTP server is easy using the
22 // `net/http` package.
23 package main
24
25 import (
26         "context"
27         "encoding/json"
28         "fmt"
29         "io/ioutil"
30         "net/http"
31         "os"
32         "strconv"
33         "sync/atomic"
34         "time"
35
36         "github.com/confluentinc/confluent-kafka-go/kafka"
37         "github.com/enriquebris/goconcurrentqueue"
38         "github.com/gorilla/mux"
39 )
40
41 // Note: consumer 'group' and consumer 'user' both uses hardcoded values specific to this interface
42 //    globalCounters      var holding the "global counters"
43 //      received          number of received messages from all topics                             (int)
44 //      sent              number of sent messages to all topics                                   (int)
45 //    topics              var holding all topic related info
46 //      <topic-name>      name of a topic (present after topic is created)
47 //        content-type    data type of the topic                                                  (string)
48 //        counters
49 //          received      number of received messages from the topic                              (int)
50 //          sent          number of sent messages to the topic                                    (int)
51 //        messages
52 //          send          messages waiting to be sent (set when sending is started)               (fifo)
53 //          received      received messages waiting to be fetched (set when reception is started) (fifo)
54
55 type counter struct {
56         c uint64
57 }
58
59 func (c *counter) step() {
60         atomic.AddUint64(&c.c, 1)
61 }
62
63 func (c counter) get() uint64 {
64         return atomic.LoadUint64(&c.c)
65 }
66
67 type counters struct {
68         received counter
69         sent     counter
70 }
71
72 func newCounters() counters {
73         return counters{
74                 received: counter{},
75                 sent:     counter{},
76         }
77 }
78
79 type messages struct {
80         send     *goconcurrentqueue.FIFO
81         received *goconcurrentqueue.FIFO
82 }
83
84 func (m *messages) startSend() bool {
85         if m.send == nil {
86                 m.send = goconcurrentqueue.NewFIFO()
87                 return true
88         }
89         return false
90 }
91
92 func (m *messages) stopSend() {
93         m.send = nil
94 }
95
96 func (m *messages) addToSend(msg string) error {
97         if m.send == nil {
98                 return fmt.Errorf("sending not started")
99         }
100         m.send.Lock()
101         defer m.send.Unlock()
102         return m.send.Enqueue(msg)
103 }
104
105 func (m *messages) getToSend() (interface{}, error) {
106         if m.send == nil {
107                 return "", fmt.Errorf("sending not started")
108         }
109         m.send.Lock()
110         defer m.send.Unlock()
111         return m.send.Dequeue()
112 }
113
114 func (m *messages) startReceive() bool {
115         if m.received == nil {
116                 m.received = goconcurrentqueue.NewFIFO()
117                 return true
118         }
119         return false
120 }
121
122 func (m *messages) stopReceive() {
123         m.send = nil
124 }
125
126 type topic struct {
127         contentType string
128         counters    counters
129         messages    messages
130 }
131
132 func newTopic(ct string) *topic {
133         return &topic{
134                 contentType: ct,
135                 counters:    counters{},
136                 messages:    messages{},
137         }
138 }
139
140 var globalCounters counters
141 var topics map[string]*topic = make(map[string]*topic)
142
143 var bootstrapserver = ""
144
145 func initApp() {
146         bootstrapserver = os.Getenv("KAFKA_BOOTSTRAP_SERVER")
147         if len(bootstrapserver) == 0 {
148                 fmt.Println("Fatal error: env var KAFKA_BOOTSTRAP_SERVER not set")
149                 fmt.Println("Exiting... ")
150                 os.Exit(1)
151         }
152         fmt.Println("Using KAFKA_BOOTSTRAP_SERVER=" + bootstrapserver)
153 }
154
155 //Helper function to get a created topic, if it exists
156 func getTopicFromRequest(w http.ResponseWriter, req *http.Request) (*topic, string, bool) {
157         topicId := mux.Vars(req)["topic"]
158         t, exist := topics[topicId]
159         if exist == false {
160                 w.WriteHeader(http.StatusNotFound)
161                 fmt.Fprintf(w, "Topic %v does not exist", topicId)
162                 return nil, "", false
163         }
164         return t, topicId, true
165 }
166
167 // Alive check
168 // GET on /
169 func healthCheck(w http.ResponseWriter, req *http.Request) {
170         fmt.Fprintf(w, "OK")
171 }
172
173 // Deep reset of this interface stub - no removal of msgs or topics in kafka
174 // POST on /reset
175 func allreset(w http.ResponseWriter, req *http.Request) {
176         for _, v := range topics {
177                 v.messages.stopSend()
178                 v.messages.stopReceive()
179         }
180         time.Sleep(5 * time.Second) //Allow producers/consumers to shut down
181         globalCounters = newCounters()
182         topics = make(map[string]*topic)
183         fmt.Fprintf(w, "OK")
184 }
185
186 // Get topics, return json array of strings of topics created by this interface stub
187 // Returns json array of strings, array is empty if no topics exist
188 // GET on /topics
189
190 func getTopics(w http.ResponseWriter, req *http.Request) {
191         topicKeys := make([]string, 0, len(topics))
192         fmt.Printf("len topics: %v\n", len(topics))
193         for k := range topics {
194                 topicKeys = append(topicKeys, k)
195         }
196         var j, err = json.Marshal(topicKeys)
197         if err != nil {
198                 w.WriteHeader(http.StatusInternalServerError)
199                 fmt.Fprintf(w, "Cannot convert list of topics to json, error details: %v", err)
200                 return
201         } else {
202                 w.Header().Set("Content-Type", "application/json")
203                 w.WriteHeader(http.StatusOK)
204                 w.Write(j)
205         }
206 }
207
208 func writeOkRepsonse(w http.ResponseWriter, httpStatus int, msg string) {
209         w.WriteHeader(httpStatus)
210         w.Header().Set("Content-Type", "text/plain")
211         fmt.Fprintf(w, msg)
212 }
213
214 // Get a counter value
215 // GET /topics/counters/{counter}
216 func getCounter(w http.ResponseWriter, req *http.Request) {
217         ctr := mux.Vars(req)["counter"]
218         var ctrvalue = -1
219         if ctr == "received" {
220                 ctrvalue = int(globalCounters.received.get())
221         } else if ctr == "sent" {
222                 ctrvalue = int(globalCounters.sent.get())
223         }
224
225         if ctrvalue == -1 {
226                 w.WriteHeader(http.StatusBadRequest)
227                 fmt.Fprintf(w, "Counter %v does not exist", ctr)
228                 return
229         }
230         writeOkRepsonse(w, http.StatusOK, strconv.Itoa(ctrvalue))
231         return
232
233 }
234
235 // Create a topic
236 // PUT on /topics/<topic>?type=<type>    type shall be 'application/json' or 'text/plain'
237 func createTopic(w http.ResponseWriter, req *http.Request) {
238         topicId := mux.Vars(req)["topic"]
239         topicType := mux.Vars(req)["type"]
240
241         fmt.Printf("Creating topic: %v, content type: %v\n", topicId, topicType)
242
243         if len(topicType) == 0 {
244                 w.WriteHeader(http.StatusBadRequest)
245                 fmt.Fprintf(w, "Type not specified")
246                 return
247         }
248
249         tid, exist := topics[topicId]
250         if exist == true {
251                 if tid.contentType != topicType {
252                         w.WriteHeader(http.StatusBadRequest)
253                         fmt.Fprintf(w, "Topic type exist but type is different, queue content type: %v, requested content type: %v", tid.contentType, topicType)
254                         return
255                 }
256                 writeOkRepsonse(w, http.StatusOK, "Topic exist")
257                 return
258         }
259
260         t := newTopic(topicType)
261
262         a, err := kafka.NewAdminClient(&kafka.ConfigMap{"bootstrap.servers": bootstrapserver})
263         defer func() { a.Close() }()
264         if err != nil {
265                 w.WriteHeader(http.StatusInternalServerError)
266                 fmt.Fprintf(w, "Cannot create client to bootstrap server: "+bootstrapserver+", error details: %v", err)
267                 return
268         }
269
270         ctx, cancel := context.WithCancel(context.Background())
271         defer cancel()
272
273         maxDur := 10 * time.Second
274
275         _, err = a.CreateTopics(
276                 ctx,
277                 []kafka.TopicSpecification{{
278                         Topic:             topicId,
279                         NumPartitions:     1,
280                         ReplicationFactor: 1}},
281                 kafka.SetAdminOperationTimeout(maxDur))
282
283         if err != nil {
284                 w.WriteHeader(http.StatusInternalServerError)
285                 fmt.Fprintf(w, "Failed to create topic: %v, error details: %v", topicId, err)
286                 return
287         }
288         topics[topicId] = t
289         w.WriteHeader(http.StatusCreated)
290         fmt.Fprintf(w, "Topic created")
291 }
292
293 // Get topic type
294 // GET on /topic/<topic>
295 func getTopic(w http.ResponseWriter, req *http.Request) {
296         t, _, exist := getTopicFromRequest(w, req)
297         if !exist {
298                 return
299         }
300         w.WriteHeader(http.StatusOK)
301         fmt.Fprintf(w, t.contentType)
302 }
303
304 // Get a topics counter value
305 // GET /topics/{topic}/counters/{counter}
306 func getTopicCounter(w http.ResponseWriter, req *http.Request) {
307         t, _, exist := getTopicFromRequest(w, req)
308         if !exist {
309                 return
310         }
311         ctr := mux.Vars(req)["counter"]
312
313         var ctrvalue = -1
314         if ctr == "received" {
315                 ctrvalue = int(t.counters.received.get())
316         } else if ctr == "sent" {
317                 ctrvalue = int(t.counters.sent.get())
318         }
319
320         if ctrvalue == -1 {
321                 w.WriteHeader(http.StatusBadRequest)
322                 fmt.Fprintf(w, "Counter %v does not exist", ctr)
323                 return
324         }
325         w.WriteHeader(http.StatusOK)
326         fmt.Fprintf(w, strconv.Itoa(ctrvalue))
327         return
328 }
329
330 func startToSend(w http.ResponseWriter, req *http.Request) {
331         t, topicId, exist := getTopicFromRequest(w, req)
332         fmt.Printf("Start to send to topic: %v\n", topicId)
333         if !exist {
334                 return
335         }
336
337         if !t.messages.startSend() {
338                 w.WriteHeader(http.StatusOK)
339                 fmt.Fprintf(w, "Already started sending")
340                 return
341         }
342         go func() {
343                 p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": bootstrapserver})
344                 if err != nil {
345                         fmt.Printf("Cannot create producer for topic: %v, error details: %v\n", topicId, err)
346                         return
347                 }
348                 defer func() {
349                         fmt.Printf("Closing producer for topic: %v\n", topicId)
350                         p.Close()
351                 }()
352                 for {
353                         q := t.messages.send
354                         if q == nil {
355                                 return
356                         }
357                         m, err := q.Get(0)
358                         if err == nil {
359                                 err = p.Produce(&kafka.Message{
360                                         TopicPartition: kafka.TopicPartition{Topic: &topicId, Partition: kafka.PartitionAny},
361                                         Value:          []byte(fmt.Sprintf("%v", m)),
362                                 }, nil)
363                                 if err == nil {
364                                         q.Remove(0)
365                                         t.counters.sent.step()
366                                         globalCounters.sent.step()
367                                         msg := fmt.Sprintf("%v", m)
368                                         if len(msg) < 500 {
369                                                 fmt.Printf("Message sent on topic: %v, len: %v, msg: %v\n", topicId, len(msg), msg)
370                                         } else {
371                                                 fmt.Printf("Message sent on topic: %v, len: %v, is larger than 500...not printed\n", topicId, len(msg))
372                                         }
373                                 } else {
374                                         fmt.Printf("Failed to send message on topic: %v. Discarded. Error details: %v\n", topicId, err)
375                                         q.Remove(0)
376                                 }
377                         } else {
378                                 time.Sleep(10 * time.Millisecond)
379                         }
380                 }
381         }()
382
383         w.WriteHeader(http.StatusOK)
384         fmt.Fprintf(w, "Sending started")
385 }
386
387 func startToReceive(w http.ResponseWriter, req *http.Request) {
388         t, topicId, exist := getTopicFromRequest(w, req)
389         if !exist {
390                 return
391         }
392
393         if !t.messages.startReceive() {
394                 w.WriteHeader(http.StatusOK)
395                 fmt.Fprintf(w, "Already started receiving")
396                 return
397         }
398
399         go func() {
400
401                 defer func() { t.messages.stopReceive() }()
402
403                 groudId := "kafkaprocon"
404
405                 c, err := kafka.NewConsumer(&kafka.ConfigMap{
406                         "bootstrap.servers":       bootstrapserver,
407                         "group.id":                groudId,
408                         "auto.offset.reset":       "earliest",
409                         "enable.auto.commit":      true,
410                         "auto.commit.interval.ms": 5000,
411                 })
412                 if err != nil {
413                         fmt.Printf("Cannot create consumer for topic: %v, error details: %v\n", topicId, err)
414                         t.messages.stopReceive()
415                         return
416                 }
417                 c.Commit()
418                 defer func() { c.Close() }()
419                 for {
420                         que := t.messages.received
421                         if que == nil {
422                                 fmt.Printf("Cannot start receiving on topic: %v, queue does not exist\n", topicId)
423                                 return
424                         }
425                         fmt.Printf("Start subscribing on topic: %v\n", topicId)
426                         err = c.SubscribeTopics([]string{topicId}, nil)
427                         if err != nil {
428                                 fmt.Printf("Cannot start subscribing on topic: %v, error details: %v\n", topicId, err)
429                                 return
430                         }
431                         maxDur := 1 * time.Second
432                         for {
433                                 msg, err := c.ReadMessage(maxDur)
434                                 if err == nil {
435                                         if len(msg.Value) < 500 {
436                                                 fmt.Printf("Message received on topic: %v, partion: %v, len: %v, msg: %v\n", topicId, msg.TopicPartition, len(msg.Value), string(msg.Value))
437                                         } else {
438                                                 fmt.Printf("Message received on topic: %v, partion: %v, len: %v is larger than 500...not printed\n", topicId, msg.TopicPartition, len(msg.Value))
439                                         }
440                                         err = t.messages.received.Enqueue(string(msg.Value))
441                                         if err != nil {
442                                                 w.WriteHeader(http.StatusInternalServerError)
443                                                 fmt.Fprintf(w, "Received message topic: %v, cannot be put in queue, %v", topicId, err)
444                                                 return
445                                         }
446                                         t.counters.received.step()
447                                         globalCounters.received.step()
448                                 } else {
449                                         fmt.Printf("Nothing to consume on topic: %v, reason: %v\n", topicId, err)
450                                 }
451                         }
452                 }
453         }()
454
455         w.WriteHeader(http.StatusOK)
456         fmt.Fprintf(w, "Receiving started")
457 }
458
459 // Post a message to a topic
460 // POST /send    content type is specified in content type
461 func sendToTopic(w http.ResponseWriter, req *http.Request) {
462
463         t, topicId, exist := getTopicFromRequest(w, req)
464         fmt.Printf("Send to topic: %v\n", topicId)
465         if !exist {
466                 return
467         }
468         q := t.messages.send
469         if q == nil {
470                 w.WriteHeader(http.StatusBadRequest)
471                 fmt.Fprintf(w, "Sending not initiated on topic: %v", topicId)
472                 return
473         }
474         ct := req.Header.Get("Content-Type")
475         if ct != t.contentType {
476                 w.WriteHeader(http.StatusBadRequest)
477                 fmt.Fprintf(w, "Message to send content type: %v on topic: %v does not match queue content type: %v", ct, topicId, t.contentType)
478                 return
479         }
480
481         if ct == "application/json" {
482                 // decoder := json.NewDecoder(req.Body)
483                 // var j :=
484                 // err := decoder.Decode(&j)
485                 // if err != nil {
486                 //      w.WriteHeader(http.StatusBadRequest)
487                 //      w.Header().Set("Content-Type", "text/plain")
488                 //      fmt.Fprintf(w, "Json payload cannot be decoded, error details %v\n", err)
489                 //      return
490                 // }
491                 //m = mux.Vars(req)[""]
492                 if err := req.ParseForm(); err != nil {
493                         w.WriteHeader(http.StatusBadRequest)
494                         fmt.Fprintf(w, "Json payload cannot be decoded on topic: %v, error details %v", topicId, err)
495                         return
496                 }
497                 b, err := ioutil.ReadAll(req.Body)
498                 if err == nil {
499                         if len(b) < 500 {
500                                 fmt.Printf("Json payload to send on topic: %v, msg: %v\n", topicId, string(b))
501                         } else {
502                                 fmt.Printf("Json payload to send on topic: %v larger than 500 bytes, not printed...\n", topicId)
503                         }
504                 } else {
505                         fmt.Printf("Json payload to send on topic: %v cannnot be decoded, err: %v\n", topicId, err)
506                 }
507                 err = q.Enqueue(string(b))
508                 if err != nil {
509                         w.WriteHeader(http.StatusInternalServerError)
510                         fmt.Fprintf(w, "Json message to send cannot be put in queue")
511                         return
512                 }
513         } else if ct == "text/plain" {
514                 if err := req.ParseForm(); err != nil {
515                         w.WriteHeader(http.StatusBadRequest)
516                         fmt.Fprintf(w, "Text payload to send on topic: %v cannot be decoded, error details %v\n", topicId, err)
517                         return
518                 }
519                 b, err := ioutil.ReadAll(req.Body)
520                 if err == nil {
521                         if len(b) < 500 {
522                                 fmt.Printf("Text payload to send on topic: %v, msg: %v\n", topicId, string(b))
523                         } else {
524                                 fmt.Printf("Text payload to send on topic: %v larger than 500 bytes, not printed...\n", topicId)
525                         }
526                 } else {
527                         fmt.Printf("Text payload to send on topic: %v cannnot be decoded, err: %v\n", topicId, err)
528                 }
529                 err = q.Enqueue(string(b))
530                 if err != nil {
531                         w.WriteHeader(http.StatusInternalServerError)
532                         fmt.Fprintf(w, "Text message to send cannot be put in queue")
533                         return
534                 }
535         } else {
536                 w.WriteHeader(http.StatusBadRequest)
537                 fmt.Fprintf(w, "Message to send, unknown content type %v", ct)
538                 return
539         }
540
541         w.WriteHeader(http.StatusOK)
542         w.Header().Set("Content-Type", "text/plain")
543         fmt.Fprintf(w, "Message to send put in queue")
544 }
545
546 // Get zero or one message from a topic
547 // GET /receive
548 func receiveFromTopic(w http.ResponseWriter, req *http.Request) {
549         t, topicId, exist := getTopicFromRequest(w, req)
550         if !exist {
551                 return
552         }
553         if t.messages.received == nil {
554                 w.WriteHeader(http.StatusBadRequest)
555                 fmt.Fprintf(w, "Receiving not initiated on topic %v", topicId)
556                 return
557         }
558
559         m, err := t.messages.received.Dequeue()
560         if err != nil {
561                 w.WriteHeader(http.StatusNoContent)
562                 return
563         }
564
565         w.Header().Set("Content-Type", t.contentType)
566         w.WriteHeader(http.StatusOK)
567         fmt.Fprintf(w, "%v", m)
568 }
569
570 // Remove the send queue to stop sending
571 func stopToSend(w http.ResponseWriter, req *http.Request) {
572         fmt.Printf("Stop sending\n")
573         t, _, exist := getTopicFromRequest(w, req)
574         if !exist {
575                 return
576         }
577         t.messages.stopSend()
578         w.WriteHeader(http.StatusNoContent)
579 }
580
581 // Remove the receive queue to stop receiving
582 func stopToReceive(w http.ResponseWriter, req *http.Request) {
583         fmt.Printf("Stop receiving\n")
584         t, _, exist := getTopicFromRequest(w, req)
585         if !exist {
586                 return
587         }
588         t.messages.stopReceive()
589         w.WriteHeader(http.StatusNoContent)
590 }
591
592 func HelloServer(w http.ResponseWriter, r *http.Request) {
593         fmt.Fprintf(w, "Hello, %s!", r.URL.Path[1:])
594 }
595
596 func main() {
597
598         initApp()
599
600         r := mux.NewRouter()
601
602         r.HandleFunc("/", healthCheck).Methods("GET")
603         r.HandleFunc("/reset", allreset).Methods("POST")
604         r.HandleFunc("/counters/{counter}", getCounter).Methods("GET")
605         r.HandleFunc("/topics", getTopics).Methods("GET")
606         r.HandleFunc("/topics/{topic}/counters/{counter}", getTopicCounter).Methods("GET")
607         r.HandleFunc("/topics/{topic}", createTopic).Methods("PUT").Queries("type", "{type}")
608         r.HandleFunc("/topics/{topic}", getTopic).Methods("GET")
609         r.HandleFunc("/topics/{topic}/startsend", startToSend).Methods("POST")
610         r.HandleFunc("/topics/{topic}/startreceive", startToReceive).Methods("POST")
611         r.HandleFunc("/topics/{topic}/stopsend", stopToSend).Methods("POST")
612         r.HandleFunc("/topics/{topic}/stopreceive", stopToReceive).Methods("POST")
613         r.HandleFunc("/topics/{topic}/msg", sendToTopic).Methods("POST")
614         r.HandleFunc("/topics/{topic}/msg", receiveFromTopic).Methods("GET")
615
616         port := "8090"
617         srv := &http.Server{
618                 Handler:      r,
619                 Addr:         ":" + port,
620                 WriteTimeout: 15 * time.Second,
621                 ReadTimeout:  15 * time.Second,
622         }
623         fmt.Println("Running on port: " + port)
624         fmt.Printf(srv.ListenAndServe().Error())
625 }