leak.go

package main

import (
	"fmt"
	"log"
	"math/rand"
	"net/http"
	"strings"
	"time"
)

var alphabet []rune = []rune("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")

func randomString(n int) string {
	alphabetSize := len(alphabet)
	var sb strings.Builder

	for i := 0; i < n; i++ {
		ch := alphabet[rand.Intn(alphabetSize)]
		sb.WriteRune(ch)
	}

	s := sb.String()
	return s
}

func main() {
	rand.Seed(time.Now().UnixNano())

	client := &http.Client{}

	upCfgTmpl := `
{
  "scheme": "http",
  "nodes": [
    {
      "host": "localhost",
      "weight": 100,
      "priority": 0,
      "port": 10000
    }
  ],
  "pass_host": "pass",
  "desc": "Created by apisix-ingress-controller, DO NOT modify it manually",
  "name": "ns_deployment1_8001",
  "hash_on": "vars",
  "labels": {
    "managed-by": "apisix-ingress-controller"
  },
  "type": "roundrobin"
}
    `

	rtCfgTmpl := `
{
  "uris": [
    "/http/%s ",
    "/http/%s/*"
  ],
  "priority": 0,
  "name": "http-test",
  "status": 1,
  "upstream_id": "%s",
  "labels": {
    "svcPort": "8001",
    "sourceName": "deployment1",
    "serviceName": "deployment1",
    "sourceType": "ingress",
    "managed-by": "apisix-ingress-controller",
    "path": "/http/%s"
  }
}
    `

	adminUpUrl := "http://localhost:9180/apisix/admin/upstreams/%s"
	adminRouteUrl := "http://localhost:9180/apisix/admin/routes/%s"
	urlTmpl := "http://localhost:9080/http/%s/foo"

	{
		data := `
{
    "log_format": {
        "host": "$host",
        "@timestamp": "$time_iso8601",
        "client_ip": "$remote_addr"
    }
}
        `
		req, err := http.NewRequest(http.MethodPut,
			"http://localhost:9180/apisix/admin/plugin_metadata/http-logger",
			strings.NewReader(data))
		req.Header.Add("X-API-KEY", "edd1c9f034335f136f87ad84b625c8f1")
		if err != nil {
			log.Fatal(err)
		}
		resp, err := client.Do(req)
		resp.Body.Close()
		if err != nil {
			log.Fatal(err)
		}
		if resp.StatusCode != 200 && resp.StatusCode != 201 {
			log.Fatal("set upstream failed", resp)
		}
	}

	{
		data := `
{
  "plugins": {
    "http-logger": {
      "name": "http logger",
      "uri": "http://localhost:10000/logs",
      "auth_header": "",
      "timeout": 3,
      "batch_max_size": 1000,
      "max_retry_count": 0,
      "retry_delay": 1,
      "buffer_duration": 60,
      "inactive_timeout": 5,
      "include_req_body": false,
      "concat_method": "json"
    },
    "prometheus": {
      "prefer_name": false
    }
  }
}
        `
		req, err := http.NewRequest(http.MethodPut,
			"http://127.0.0.1:9180/apisix/admin/global_rules/1",
			strings.NewReader(data))
		req.Header.Add("X-API-KEY", "edd1c9f034335f136f87ad84b625c8f1")
		if err != nil {
			log.Fatal(err)
		}
		resp, err := client.Do(req)
		resp.Body.Close()
		if err != nil {
			log.Fatal(err)
		}
		if resp.StatusCode != 200 && resp.StatusCode != 201 {
			log.Fatal("set upstream failed", resp)
		}
	}

	for i := 0; i < 10000000; i++ {
		if i%100 == 0 {
			fmt.Println("iter ", i)
		}
		upstreamId := randomString(12)
		randomId := randomString(12)
		randomPath := randomString(6)
		upUrl := fmt.Sprintf(adminUpUrl, upstreamId)
		rtUrl := fmt.Sprintf(adminRouteUrl, randomId)

		{
			req, err := http.NewRequest(http.MethodPut, upUrl, strings.NewReader(upCfgTmpl))
			req.Header.Add("X-API-KEY", "edd1c9f034335f136f87ad84b625c8f1")
			if err != nil {
				log.Fatal(err)
			}
			resp, err := client.Do(req)
			resp.Body.Close()
			if err != nil {
				log.Fatal(err)
			}
			if resp.StatusCode != 200 && resp.StatusCode != 201 {
				log.Fatal("set upstream failed", resp)
			}
		}

		{
			data := fmt.Sprintf(rtCfgTmpl, randomPath, randomPath, upstreamId, randomPath)
			req, err := http.NewRequest(http.MethodPut, rtUrl, strings.NewReader(data))
			req.Header.Add("X-API-KEY", "edd1c9f034335f136f87ad84b625c8f1")
			if err != nil {
				log.Fatal(err)
			}
			resp, err := client.Do(req)
			resp.Body.Close()
			if err != nil {
				log.Fatal(err)
			}
			if resp.StatusCode != 200 && resp.StatusCode != 201 {
				log.Fatal("set route failed", resp)
			}
		}

		time.Sleep(50 * time.Millisecond)

		for {
			url := fmt.Sprintf(urlTmpl, randomPath)
			resp, err := http.Get(url)
			resp.Body.Close()
			if err != nil {
				log.Fatal(err)
			}
			if resp.StatusCode != 200 {
				fmt.Println("get", url, resp)
				time.Sleep(100 * time.Millisecond)
			} else {
				break
			}
		}

		{
			req, err := http.NewRequest(http.MethodDelete, rtUrl, nil)
			req.Header.Add("X-API-KEY", "edd1c9f034335f136f87ad84b625c8f1")
			if err != nil {
				log.Fatal(err)
			}
			resp, err := client.Do(req)
			resp.Body.Close()
			if err != nil {
				log.Fatal(err)
			}
			if resp.StatusCode != 200 {
				panic("admin route failed")
			}
		}

		time.Sleep(50 * time.Millisecond)

		for {
			req, err := http.NewRequest(http.MethodDelete, upUrl, nil)
			req.Header.Add("X-API-KEY", "edd1c9f034335f136f87ad84b625c8f1")
			if err != nil {
				log.Fatal(err)
			}
			resp, err := client.Do(req)
			resp.Body.Close()
			if err != nil {
				log.Fatal(err)
			}
			if resp.StatusCode != 200 {
				log.Println(req, resp)
				time.Sleep(100 * time.Millisecond)
			} else {
				break
			}
		}
	}
}

lua-leak.sxx

#!/usr/bin/env stap++

@use nginx.lua
@use luajit

global ptr2bt
global ptr2size
global bt_stats
global sum
global alloc_cnt = 0
global quit = 0
global gc_size_begin = 0
global gc_size_end = 0

function get_bt()
{
    if (@defined(@var("globalL", "$^exec_path"))) {
        mL = @var("globalL", "$^exec_path")

    } else {
        mL = ngx_lua_get_main_lua_vm()
    }

    g = luajit_G(mL)
    L = luajit_cur_thread(g)

    return luajit_backtrace(L, g, $^arg_detailed :default(0) ? 0 : 1)
}

function get_gc_size()
{
    if (@defined(@var("globalL", "$^exec_path"))) {
        mL = @var("globalL", "$^exec_path")

    } else {
        mL = ngx_lua_get_main_lua_vm()
    }

    G = luajit_G(mL)
    $*G := @cast(G, "global_State", "$^libluajit_path")
    return $*G->gc->total
}

function report()
{
    foreach (bt in bt_stats) {
        sum[bt] = @sum(bt_stats[bt])
    }
    foreach (bt in sum- limit $^arg_limit :default(5)) {
        print_ustack(bt)
        printf("\ttotal %d bytes\n", sum[bt])
    }
    delete ptr2bt
    delete ptr2size
    delete bt_stats
    delete sum
    alloc_cnt = 0
    printf("-------\n")
}

probe begin
{
    warn(sprintf("Start tracing %d ($^exec_path)...", target()))
    %( "$^arg_time" != ""  %?
    warn(sprintf("Please wait for $^arg_time seconds...\n"))
    %:
    warn("Hit Ctrl-C to end.\n")
    %)
}

probe end
{
    report()
    printf("\nGC size change: %d -> %d bytes\n", gc_size_begin, gc_size_end)
}

probe process("$^libluajit_path").function("lj_alloc_realloc").return,
      process("$^libluajit_path").function("lj_alloc_malloc").return
{
    if (gc_size_begin == 0) {
        gc_size_begin = get_gc_size()
    }

    gc_size_end = get_gc_size()

    if (tid() == target() && !quit) {
        ptr = returnval()
        bt = get_bt()
        if (ptr && bt != "") {
            if (alloc_cnt >= 100000) {
                report()
            }

            if (ptr2bt[ptr] == "") {
                alloc_cnt++
            }

            size = @entry($nsize)
            ptr2bt[ptr] = bt
            ptr2size[ptr] = size
            bt_stats[bt] <<< size
        }
    }
}

probe process("$^libluajit_path").function("lj_alloc_free")
{
    ptr = pointer_arg(2)
    if (tid() == target() && ptr && !quit) {
        bt = ptr2bt[ptr]
        delete ptr2bt[ptr]

        bytes = ptr2size[ptr]
        delete ptr2size[ptr]

        if (bt != "") {
            alloc_cnt--
            bt_stats[bt] <<< -bytes
            if (@sum(bt_stats[bt]) == 0) {
                delete bt_stats[bt]
            }
        }
    }
}

probe timer.s($^arg_time)
{
    quit = 1
    printf("exit...\n")
    exit()
}

nginx.conf

error_log /dev/stderr info;
worker_processes 1;

events {}

http {
    server {
        listen 10000;

        location / {
            content_by_lua_block {
                ngx.say("ok")
            }
        }
    }
}

test.md

# run nginx as upstream
/usr/local/openresty/nginx/sbin/nginx -p $PWD -c nginx.conf -g "daemon off;"

# run client workload
go run leak.go

# watch the memory leak
ps auxf |grep nginx
watch -n 3 'ps -p 91107 -o cmd,vsz,rss'

# use systemtap to analyze
./samples/lua-leaks.sxx -D STP_NO_OVERLOAD -D MAXMAPENTRIES=150000 -D MAXACTION=10000 --skip-badvars -arg time=60 -x 91107