Feat/rate limit write

.gitignore

@@ -1,6 +1,9 @@
 # Prerequisites
 *.d
 
+# Cache
+.cache
+
 # Object files
 *.o
 *.ko

Makefile

@@ -1,6 +1,8 @@
 .PHONY: build clean download-nginx build-nginx configure build-module run test debug
 
-build: clean download-nginx build-nginx
+setup: clean download-nginx configure
+
+build: build-nginx build-module
 
 download-nginx:
 	curl https://nginx.org/download/nginx-1.26.3.tar.gz > nginx.tar.gz
@@ -9,7 +11,7 @@ download-nginx:
 configure:
 	cd nginx-1.26.3 && ./configure --prefix=$(PWD)/build --add-dynamic-module=..
 
-build-nginx: configure
+build-nginx:
 	cd nginx-1.26.3 && make && make install
 
 build-module:
@@ -29,3 +31,9 @@ debug:
 	cd ./reader-go; go build -o debugger main.go; mv ./debugger ..
 	./debugger one.bin
 	./debugger two.bin
+
+log:
+	go run log_zone/*.go log
+
+send:
+	go run log_zone/*.go send

config

@@ -3,7 +3,27 @@ ngx_module_type=HTTP
 ngx_module_name=ngx_http_limit_req_rw_module
 ngx_module_srcs="$ngx_addon_dir/ngx_http_limit_req_rw_module.c"
 
-CFLAGS="$CFLAGS `pkg-config --cflags 'msgpack-c = 6.1.0'`"
-CORE_LIBS="$CORE_LIBS `pkg-config --libs 'msgpack-c = 6.1.0'`"
+# Detect platform
+OS_NAME="$(uname -s)"
+
+# Default values
+MSGPACK_PKG_NAME=""
+MSGPACK_VERSION=""
+
+# Try to detect msgpack library
+if pkg-config --exists 'msgpack-c'; then
+    MSGPACK_PKG_NAME="msgpack-c"
+    MSGPACK_VERSION="6.1.0"  # Optional: only enforce version if strictly necessary
+elif pkg-config --exists 'msgpack'; then
+    MSGPACK_PKG_NAME="msgpack"
+    MSGPACK_VERSION="3.1.0"
+else
+    echo "Error: Neither 'msgpack-c' nor 'msgpack' pkg-config package found."
+    exit 1
+fi
+
+# Add flags
+CFLAGS="$CFLAGS $(pkg-config --cflags "$MSGPACK_PKG_NAME")"
+CORE_LIBS="$CORE_LIBS $(pkg-config --libs "$MSGPACK_PKG_NAME")"
 
 . auto/module

go.mod

@@ -0,0 +1,13 @@
+module github.com/tsuru/ngx-http-limit-req-rw-module
+
+go 1.24.3
+
+require (
+	github.com/sirupsen/logrus v1.9.3
+	github.com/vmihailenco/msgpack/v5 v5.4.1
+)
+
+require (
+	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
+	golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 // indirect
+)

go.sum

@@ -0,0 +1,19 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
+github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8=
+github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok=
+github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g=
+github.com/vmihailenco/tagparser/v2 v2.0.0/go.mod h1:Wri+At7QHww0WTrCBeu4J6bNtoV6mEfg5OIWRZA9qds=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8 h1:0A+M6Uqn+Eje4kHMK80dtF3JCXC4ykBgQG4Fe06QRhQ=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

log_zone/logs.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/vmihailenco/msgpack/v5"
+)
+
+func logs(zone string) {
+	for {
+		zone, err := handleRequest(zone)
+		if err != nil {
+			logrus.Error("Error handling request", "error", err)
+			return
+		}
+		fmt.Print("\033[H\033[2J")
+		logrus.Infof("Zone: %s, RateLimitHeader: %+v, RateLimitEntries: %d",
+			zone.Name, zone.RateLimitHeader, len(zone.RateLimitEntries))
+		for _, entry := range zone.RateLimitEntries {
+			logrus.Infof("Entry Key: %s, Last: %d, Excess: %d",
+				entry.Key.String(zone.RateLimitHeader), entry.Last, entry.Excess)
+		}
+		fmt.Println("\nPress Ctrl+C to exit")
+		time.Sleep(2 * time.Second)
+	}
+}
+
+func handleRequest(zone string) (Zone, error) {
+	endpoint := fmt.Sprintf("http://localhost:9000/api/%s", zone)
+	req, err := http.NewRequest(http.MethodGet, endpoint, nil)
+	if err != nil {
+		return Zone{}, err
+	}
+
+	response, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return Zone{}, fmt.Errorf("error making request to %s: %w", endpoint, err)
+	}
+	defer response.Body.Close()
+	decoder := msgpack.NewDecoder(response.Body)
+	var rateLimitHeader RateLimitHeader
+	rateLimitEntries := []RateLimitEntry{}
+	log := logrus.New()
+	if err := decoder.Decode(&rateLimitHeader); err != nil {
+		if err == io.EOF {
+			return Zone{
+				Name:             zone,
+				RateLimitHeader:  rateLimitHeader,
+				RateLimitEntries: rateLimitEntries,
+			}, nil
+		}
+		log.Error("Error decoding header", "error", err)
+		return Zone{}, err
+	}
+	for {
+		var message RateLimitEntry
+		if err := decoder.Decode(&message); err != nil {
+			if err == io.EOF {
+				break
+			}
+			log.Error("Error decoding entry", "error", err)
+			return Zone{}, err
+		}
+		message.Last = toNonMonotonic(message.Last, rateLimitHeader)
+		rateLimitEntries = append(rateLimitEntries, message)
+	}
+	log.Debug("Received rate limit entries", "zone", zone, "entries", len(rateLimitEntries))
+	return Zone{
+		Name:             zone,
+		RateLimitHeader:  rateLimitHeader,
+		RateLimitEntries: rateLimitEntries,
+	}, nil
+}

log_zone/main.go

@@ -0,0 +1,64 @@
+package main
+
+import (
+	"net"
+	"os"
+)
+
+func main() {
+	arg := os.Args[1:]
+	zone := "one"
+	if arg[0] == "log" {
+		if len(arg) >= 2 {
+			zone = arg[1]
+		}
+		logs(zone)
+		return
+	}
+	if arg[0] == "send" {
+		if len(arg) >= 2 {
+			zone = arg[1]
+		}
+		send(zone)
+	}
+}
+
+func toNonMonotonic(last int64, header RateLimitHeader) int64 {
+	return header.Now - (header.NowMonotonic - last)
+}
+
+type Zone struct {
+	Name             string
+	RateLimitHeader  RateLimitHeader
+	RateLimitEntries []RateLimitEntry
+}
+
+type RateLimitHeader struct {
+	Key          string
+	Now          int64
+	NowMonotonic int64
+}
+
+type RateLimitEntry struct {
+	Key    Key
+	Last   int64
+	Excess int64
+}
+
+const (
+	BinaryRemoteAddress = "$binary_remote_addr"
+	RemoteAddress       = "$remote_addr"
+)
+
+type Key []byte
+
+func (r Key) String(header RateLimitHeader) string {
+	switch header.Key {
+	case BinaryRemoteAddress:
+		return net.IP(r).String()
+	case RemoteAddress:
+		fallthrough
+	default:
+		return string(r)
+	}
+}

log_zone/send.go

@@ -0,0 +1,76 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"github.com/vmihailenco/msgpack/v5"
+)
+
+func send(zone string) {
+	err := sendRequest(
+		zone,
+		RateLimitHeader{
+			Key:          BinaryRemoteAddress,
+			Now:          time.Now().Unix(),
+			NowMonotonic: time.Now().UnixNano() / int64(time.Millisecond),
+		}, []RateLimitEntry{
+			{Key([]byte{127, 0, 0, 0}), 7, 99},
+			{Key([]byte{127, 0, 0, 1}), 7, 12},
+			{Key([]byte{127, 6, 4, 00}), 2, 98},
+			{Key([]byte{127, 0, 0, 99}), 30, 300},
+			{Key([]byte{10, 0, 0, 1}), 444, 21},
+		})
+	if err != nil {
+		logrus.Fatalf("Error sending request: %v", err)
+	}
+}
+
+func sendRequest(zone string, header RateLimitHeader, entries []RateLimitEntry) error {
+	var buf bytes.Buffer
+	encoder := msgpack.NewEncoder(&buf)
+	var values []interface{} = []interface{}{
+		headerToArray(header),
+	}
+	for _, entry := range entries {
+		values = append(values, entryToArray(entry, header))
+	}
+	if err := encoder.Encode(values); err != nil {
+		return fmt.Errorf("error encoding entries: %w", err)
+	}
+	endpoint := fmt.Sprintf("http://localhost:9000/api/%s", zone)
+	req, err := http.NewRequest(http.MethodPost, endpoint, &buf)
+	if err != nil {
+		return fmt.Errorf("error creating request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/x-msgpack")
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("error sending request to %s: %w", endpoint, err)
+	}
+	fmt.Println(resp.Status)
+	defer resp.Body.Close()
+	respBody, _ := io.ReadAll(resp.Body)
+	logrus.Infof("response status: %s, body: %s", resp.Status, respBody)
+	return nil
+}
+
+func headerToArray(header RateLimitHeader) []interface{} {
+	return []interface{}{
+		header.Key,
+		header.Now,
+		header.NowMonotonic,
+	}
+}
+
+func entryToArray(entry RateLimitEntry, header RateLimitHeader) []interface{} {
+	return []interface{}{
+		entry.Key,
+		entry.Last,
+		entry.Excess,
+	}
+}

nginx.conf

@@ -1,6 +1,5 @@
 worker_processes 1;
 master_process on;
-daemon off;
 
 error_log ./error.log debug;