OpenResty as redirect server for logging sent to Kafka

Purpose

A redirect server for logging is used to trace various urls user clicked. The redirect server will log any query parameters in url and redirect to url parameter finally. The logging should be doing asynchronously to minimize redirect time.

OpenResty is a combination of Nginx and Lua. I won’t explain that here, you could google it to know more.

Environment

  • MacOS
  • Docker Desktop
  • Download lua-resty-kafka for lua kafka operation
  • Lua to get query paramters and send to kafka
  • Encode query parameters using cjson as message
  • Send message to kafka

Download lua-resty-kafka

1
2
3
4
5
mkdir openresty
cd openresty
wget https://github.com/doujiang24/lua-resty-kafka/archive/v0.07.zip
unzip v0.07.zip
mkdir logs

Directory layout

We will mount docker host files(nginx.conf) and directory(logs, lua-resty-kafka-0.07) to container

1
2
3
4
5
6
7
8
9
(base) kevin.luo➜~/dev/openresty» ls -al
total 32
drwxr-xr-x   8 kevin.luo  staff   256 Nov  6 12:59 .
drwxr-xr-x@ 24 kevin.luo  staff   768 Nov  5 20:10 ..
drwxr-xr-x   5 kevin.luo  staff   160 Nov  6 10:47 logs
drwxr-xr-x@  9 kevin.luo  staff   288 Aug 31 08:48 lua-resty-kafka-0.07
-rw-r--r--   1 kevin.luo  staff  4490 Nov  6 12:59 nginx.conf
-rwxr-xr-x   1 kevin.luo  staff   874 Nov  6 11:19 run.sh
-rwxr-xr-x   1 kevin.luo  staff    58 Nov  5 20:29 stop.sh

Files description and content

run.sh is to run this rd service

1
2
3
#!/usr/bin/env bash

docker run -d --name="rd" -p 8089:8089 -v $PWD/nginx.conf:/usr/local/openresty/nginx/conf/nginx.conf:ro -v $PWD/logs:/usr/local/openresty/nginx/logs -v $PWD/lua-resty-kafka-0.07/lib/resty/kafka:/usr/local/openresty/lualib/resty/kafka openresty/openresty:1.15.8.2-4-xenial-nosse42

stop.sh is to stop rd service and remove docker container

1
2
3
#!/usr/bin/env bash

docker kill nginx && docker rm nginx

nginx.conf is the main nginx + lua configuration file.

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
worker_processes  1;
events {
    use epoll;
    worker_connections 1024;
}

error_log logs/error.log;

http {
    include       mime.types;
    default_type  application/octet-stream;

    server_tokens off;
    more_clear_headers 'Server';

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent" "$http_x_forwarded_for"';

    access_log  logs/access.log  main;

    sendfile        on;
    keepalive_timeout  65;

    lua_package_path "/usr/local/openresty/lualib/resty/kafka/?.lua;;";
    # without resolver setting, it will fail in DNS lookup
    # either 8.8.8.8 or using container's resolver
    # resolver 8.8.8.8
    resolver local=on ipv6=off;
    resolver_timeout 5s;

    server {
        listen       8089;
        server_name  localhost;

        location / {
            root   html;
            index  index.html index.htm;
        }

        # Make sure Lua is working
        location /hello {
            default_type text/html;
            content_by_lua '
                ngx.say("Lua: hello world!")
            ';
        }

        # The default 1by1 gif
        location /empty_gif {
           empty_gif;
        }

        location /rd {
            default_type text/html;
		    content_by_lua '
                local cjson = require "cjson"
                local producer = require "resty.kafka.producer"
                -- local client = require "resty.kafka.client"

                local broker_list = {
                    { host = "192.168.0.16", port = 9092},
                    { host = "192.168.0.17", port = 9092},
                    { host = "192.168.0.18", port = 9092}
                }

                -- Parsing request args
                local params_json = {}
                local args, err = ngx.req.get_uri_args()

                if err == "truncated" then
                    -- one can choose to ignore or reject the current request here
                end

                for key, val in pairs(args) do
                    params_json[key] = val
                end
                ngx.log(ngx.ERR, "url: ", params_json["d"])

                local topic = "redirect-server"
                local params_message = cjson.encode(params_json)

                -- Set producer async
                local bp = producer:new(broker_list, { producer_type = "async" })

                -- The second parameter of send method is to control kafka routing.
                -- When it is nill, it will write message to same partition
                -- With designated key，it will write message to hash of key of partition
                local ok, err = bp:send(topic, nil, params_message)

                -- For debugging
                ngx.log(ngx.ERR, "Message in json: ", params_message)

                if not ok then
                    ngx.log(ngx.ERR, "kafka send err:", err)
                    return
                end

                if params_json["url"] ~= nil then
                    return ngx.redirect(params_json["url"], 301)
                end

                return ngx.exec("/empty_gif")
            ';

        }
    }
}

Testing

Check you logs/error.log file, there should not have any kafka send err:

1
2
2019/11/06 04:59:44 [error] 6#6: *1 [lua] content_by_lua(nginx.conf:119):25: url: nil, client: 172.17.0.1, server: localhost, request: "GET /rd?a=b&open=false&run=success&yeild=ok&url=http://tw.news.yahoo.com HTTP/1.1", host: "127.0.0.1:8089"
2019/11/06 04:59:44 [error] 6#6: *1 [lua] content_by_lua(nginx.conf:119):56: Message in json: {"run":"success","yeild":"ok","url":"http:\/\/tw.news.yahoo.com","a":"b","open":"false"}, client: 172.17.0.1, server: localhost, request: "GET /rd?a=b&open=false&run=success&yeild=ok&url=http://tw.news.yahoo.com HTTP/1.1", host: "127.0.0.1:8089"

Verify kafka message queue

1
2
3
kafka-console-consumer --broker-list 192.168.0.16:9092 --topic redirect-server --from-beginning

kafka-console-producer --broker-list 192.168.0.16:9092 --topic redirect-server