-
Notifications
You must be signed in to change notification settings - Fork 2
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: add bytesize, queue, mem, profiler and sync packages (#50)
- Loading branch information
Showing
46 changed files
with
1,544 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| package bytesize | ||
|
|
||
| const ( | ||
| B int64 = 1 | ||
| KB int64 = 1 << (10 * iota) | ||
| MB | ||
| GB | ||
| TB | ||
| PB | ||
| EB | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,42 @@ | ||
| package httputil | ||
|
|
||
| import ( | ||
| "context" | ||
| "net" | ||
| "net/http" | ||
| "time" | ||
| ) | ||
|
|
||
| func ListenAndServe(ctx context.Context, server *http.Server, shutdownTimeout ...time.Duration) error { | ||
| return gracefulFunc(ctx, server, server.ListenAndServe, shutdownTimeout...) | ||
| } | ||
|
|
||
| func Serve(ctx context.Context, server *http.Server, l net.Listener, shutdownTimeout ...time.Duration) error { | ||
| fn := func() error { | ||
| return server.Serve(l) | ||
| } | ||
| return gracefulFunc(ctx, server, fn, shutdownTimeout...) | ||
| } | ||
|
|
||
| func gracefulFunc(ctx context.Context, server *http.Server, fn func() error, shutdownTimeout ...time.Duration) error { | ||
| errCh := make(chan error, 1) | ||
| go func() { | ||
| errCh <- fn() | ||
| }() | ||
| select { | ||
| case err := <-errCh: | ||
| return err | ||
| case <-ctx.Done(): | ||
| switch { | ||
| case len(shutdownTimeout) == 0: | ||
| return server.Shutdown(context.Background()) | ||
| case shutdownTimeout[0] == 0: | ||
| return server.Close() | ||
| default: | ||
| ctx, cancel := context.WithTimeout(context.Background(), shutdownTimeout[0]) | ||
| defer cancel() | ||
|
|
||
| return server.Shutdown(ctx) | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,124 @@ | ||
| package cgroup | ||
|
|
||
| import ( | ||
| "strconv" | ||
| ) | ||
|
|
||
| // GetMemoryUsage returns cgroup (v1 or v2) memory usage | ||
| func GetMemoryUsage(basePath string) int64 { | ||
| n, err := getMemStatCgroup1(basePath, "memory.usage_in_bytes") | ||
| if err == nil { | ||
| wss := getWSSMemoryCgroup1(basePath, n) | ||
| rss := getRSSMemoryCgroup1(basePath) | ||
| if wss > rss { | ||
| return wss | ||
| } | ||
| return rss | ||
| } | ||
| n, err = getMemStatCgroup2(basePath, "memory.current") | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| return getWSSMemoryCgroup2(basePath, n) | ||
| } | ||
|
|
||
| // GetMemoryLimit returns the cgroup's (v1 or v2) memory limit, or [totalMem] if there is no limit set. | ||
| // If using cgroups v1, hierarchical memory limit is also taken into consideration if there is no limit set. | ||
| // | ||
| // - https://www.kernel.org/doc/Documentation/cgroup-v1/memory.txt | ||
| // | ||
| // - https://www.kernel.org/doc/html/latest/admin-guide/cgroup-v2.html#memory-interface-files | ||
| func GetMemoryLimit(basePath string, totalMem int) int { | ||
| getLimit := func() int64 { | ||
| // cgroups v1 | ||
| n, err := getMemStatCgroup1(basePath, "memory.limit_in_bytes") | ||
| if err == nil { | ||
| if n <= 0 || int64(int(n)) != n || int(n) > totalMem { | ||
| // try to get hierarchical limit | ||
| n = GetHierarchicalMemoryLimitCgroup1(basePath) | ||
| } | ||
| return n | ||
| } | ||
|
|
||
| // cgroups v2 | ||
| n, err = getMemStatCgroup2(basePath, "memory.max") | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| return n | ||
| } | ||
| limit := getLimit() | ||
|
|
||
| // if the number in not within expected boundaries, return totalMem | ||
| if limit <= 0 || int64(int(limit)) != limit || int(limit) > totalMem { | ||
| return totalMem | ||
| } | ||
| return int(limit) | ||
| } | ||
|
|
||
| func getMemStatCgroup2(basePath, statName string) (int64, error) { | ||
| // See https: //www.kernel.org/doc/html/latest/admin-guide/cgroup-v2.html#memory-interface-files | ||
| return getStatGeneric(statName, basePath+"/sys/fs/cgroup", basePath+"/proc/self/cgroup", "") | ||
| } | ||
|
|
||
| func getMemStatCgroup1(basePath, statName string) (int64, error) { | ||
| return getStatGeneric(statName, basePath+"/sys/fs/cgroup/memory", basePath+"/proc/self/cgroup", "memory") | ||
| } | ||
|
|
||
| // GetHierarchicalMemoryLimitCgroup1 returns hierarchical memory limit | ||
| // https://www.kernel.org/doc/Documentation/cgroup-v1/memory.txt | ||
| func GetHierarchicalMemoryLimitCgroup1(basePath string) int64 { | ||
| return memStatCgroup1(basePath, "hierarchical_memory_limit") | ||
| } | ||
|
|
||
| func getRSSMemoryCgroup1(basePath string) int64 { | ||
| return memStatCgroup1(basePath, "total_rss") | ||
| } | ||
|
|
||
| func getWSSMemoryCgroup1(basePath string, used int64) int64 { | ||
| inactive := memStatCgroup1(basePath, "total_inactive_file") | ||
| if used < inactive { | ||
| return 0 | ||
| } | ||
| return used - inactive | ||
| } | ||
|
|
||
| func getWSSMemoryCgroup2(basePath string, used int64) int64 { | ||
| inactive := memStatCgroup2(basePath, "inactive_file") | ||
| if used < inactive { | ||
| return 0 | ||
| } | ||
| return used - inactive | ||
| } | ||
|
|
||
| func memStatCgroup1(basePath, key string) int64 { | ||
| data, err := getFileContents("memory.stat", basePath+"/sys/fs/cgroup/memory", basePath+"/proc/self/cgroup", "memory") | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| memStat, err := grepFirstMatch(data, key, 1, " ") | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| n, err := strconv.ParseInt(memStat, 10, 64) | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| return n | ||
| } | ||
|
|
||
| func memStatCgroup2(basePath, key string) int64 { | ||
| data, err := getFileContents("memory.stat", basePath+"/sys/fs/cgroup", basePath+"/proc/self/cgroup", "") | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| memStat, err := grepFirstMatch(data, key, 1, " ") | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| n, err := strconv.ParseInt(memStat, 10, 64) | ||
| if err != nil { | ||
| return 0 | ||
| } | ||
| return n | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,74 @@ | ||
| package cgroup_test | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| "github.com/rudderlabs/rudder-go-kit/bytesize" | ||
| "github.com/rudderlabs/rudder-go-kit/mem/internal/cgroup" | ||
| "github.com/stretchr/testify/require" | ||
| ) | ||
|
|
||
| func TestCgroupMemory(t *testing.T) { | ||
| t.Run("cgroups v1 with limit", func(t *testing.T) { | ||
| basePath := "testdata/cgroups_v1_mem_limit" | ||
| totalMem := int(100 * bytesize.GB) | ||
| limit := cgroup.GetMemoryLimit(basePath, totalMem) | ||
|
|
||
| require.EqualValues(t, 25*bytesize.GB, limit, "when a limit is set, this limit should be returned") | ||
| require.EqualValues(t, 7873486848, cgroup.GetMemoryUsage(basePath)) | ||
| }) | ||
|
|
||
| t.Run("cgroups v1 with self limit", func(t *testing.T) { | ||
| basePath := "testdata/cgroups_v1_mem_limit_proc_self" | ||
| totalMem := int(100 * bytesize.GB) | ||
| limit := cgroup.GetMemoryLimit(basePath, totalMem) | ||
|
|
||
| require.EqualValues(t, 25*bytesize.GB, limit, "when a limit is set, this limit should be returned") | ||
| require.EqualValues(t, 9456156572, cgroup.GetMemoryUsage(basePath)) | ||
| }) | ||
|
|
||
| t.Run("cgroups v1 with hierarchical limit", func(t *testing.T) { | ||
| basePath := "testdata/cgroups_v1_mem_hierarchy" | ||
| totalMem := int(100 * bytesize.GB) | ||
| limit := cgroup.GetMemoryLimit(basePath, totalMem) | ||
|
|
||
| require.EqualValues(t, 25*bytesize.GB, limit, "when a hierarchical limit is set, this limit should be returned") | ||
| require.EqualValues(t, 7873486848, cgroup.GetMemoryUsage(basePath)) | ||
| }) | ||
|
|
||
| t.Run("cgroups v1 no limit", func(t *testing.T) { | ||
| basePath := "testdata/cgroups_v1_mem_no_limit" | ||
| totalMem := int(100 * bytesize.GB) | ||
| limit := cgroup.GetMemoryLimit(basePath, totalMem) | ||
|
|
||
| require.EqualValues(t, totalMem, limit, "when no limit is set, total memory should be returned") | ||
| require.EqualValues(t, 7873486848, cgroup.GetMemoryUsage(basePath)) | ||
| }) | ||
|
|
||
| t.Run("cgroups v2 with limit", func(t *testing.T) { | ||
| basePath := "testdata/cgroups_v2_mem_limit" | ||
| totalMem := int(100 * bytesize.GB) | ||
| limit := cgroup.GetMemoryLimit(basePath, totalMem) | ||
|
|
||
| require.EqualValues(t, 32*bytesize.GB, limit, "when a limit is set, this limit should be returned") | ||
| require.EqualValues(t, 26071040, cgroup.GetMemoryUsage(basePath)) | ||
| }) | ||
|
|
||
| t.Run("cgroups v2 no limit", func(t *testing.T) { | ||
| basePath := "testdata/cgroups_v2_mem_no_limit" | ||
| totalMem := int(100 * bytesize.GB) | ||
| limit := cgroup.GetMemoryLimit(basePath, totalMem) | ||
|
|
||
| require.EqualValues(t, totalMem, limit, "when no limit is set, total memory should be returned") | ||
| require.EqualValues(t, 26071040, cgroup.GetMemoryUsage(basePath)) | ||
| }) | ||
|
|
||
| t.Run("no cgroups info", func(t *testing.T) { | ||
| basePath := "testdata/invalid_path" | ||
| totalMem := int(100 * bytesize.GB) | ||
| limit := cgroup.GetMemoryLimit(basePath, totalMem) | ||
|
|
||
| require.EqualValues(t, limit, limit, "when no cgroups info is available, this limit should be returned") | ||
| require.EqualValues(t, 0, cgroup.GetMemoryUsage(basePath)) | ||
| }) | ||
| } |
1 change: 1 addition & 0 deletions
1
mem/internal/cgroup/testdata/cgroups_v1_mem_hierarchy/README.md
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| Using cgroup v1 with current memory usage at 9456156672, where no memory limit is set but a hierarchical memory limit is set at 26843545600 |
1 change: 1 addition & 0 deletions
1
...ernal/cgroup/testdata/cgroups_v1_mem_hierarchy/sys/fs/cgroup/memory/memory.limit_in_bytes
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1 @@ | ||
| 9223372036854771712 |
Oops, something went wrong.