Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix goroutine leak: make flushDaemon stoppable #293

Merged
merged 5 commits into from Feb 27, 2022
Merged

Fix goroutine leak: make flushDaemon stoppable #293

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0d5c8b7
Make flushDaemon stoppable and testable
katexochen Feb 10, 2022
7ad18de
fixup! Make flushDaemon stoppable and testable
katexochen Feb 24, 2022
b6757a5
fixup! Make flushDaemon stoppable and testable
katexochen Feb 24, 2022
31a05ea
fixup! Make flushDaemon stoppable and testable
katexochen Feb 26, 2022
ec08484
fixup! Make flushDaemon stoppable and testable
katexochen Feb 26, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
39 changes: 39 additions & 0 deletions examples/flushing/flushing_test.go
View file
Open in desktop
@@ -0,0 +1,39 @@
package main

import (
"flag"
"testing"

"go.uber.org/goleak"
"k8s.io/klog/v2"
)

func main() {
klog.InitFlags(nil)

// By default klog writes to stderr. Setting logtostderr to false makes klog
// write to a log file.
flag.Set("logtostderr", "false")
flag.Set("log_file", "myfile.log")
flag.Parse()

// Info writes the first log message. When the first log file is created,
// a flushDaemon is started to frequently flush bytes to the file.
klog.Info("nice to meet you")

// klog won't ever stop this flushDaemon. To exit without leaking a goroutine,
// the daemon can be stopped manually.
klog.StopFlushDaemon()

// After you stopped the flushDaemon, you can still manually flush.
klog.Info("bye")
klog.Flush()
}

func TestLeakingFlushDaemon(t *testing.T) {
// goleak detects leaking goroutines.
defer goleak.VerifyNone(t)
dims marked this conversation as resolved.
Show resolved Hide resolved

// Without calling StopFlushDaemon in main, this test will fail.
main()
}
1 change: 1 addition & 0 deletions examples/go.mod
View file
Open in desktop
Expand Up @@ -6,6 +6,7 @@ require (
github.com/go-logr/logr v1.2.2
github.com/go-logr/zapr v1.2.2
github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b
go.uber.org/goleak v1.1.12
go.uber.org/zap v1.19.0
k8s.io/klog/v2 v2.30.0
)
Expand Down
45 changes: 43 additions & 2 deletions examples/go.sum
View file
Open in desktop
@@ -1,6 +1,7 @@
github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA=
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
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/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
Expand All @@ -19,38 +20,78 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
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/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k=
go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
go.uber.org/goleak v1.1.10 h1:z+mqJhf6ss6BSfSM671tgKyZBFPTTJM+HLxnhPC3wu0=
go.uber.org/goleak v1.1.10/go.mod h1:8a7PlsEVH3e/a/GLqe5IIrQx6GzcnRmZEufDUTk4A7A=
go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA=
go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ=
go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
go.uber.org/zap v1.19.0 h1:mZQZefskPPCMIBCSEH0v2/iUqqLrYtaeqwD6FUGUnFE=
go.uber.org/zap v1.19.0/go.mod h1:xg/QME4nWcxGxrpdeYfq7UvYrLh66cuVKdrbD1XF/NI=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
golang.org/x/lint v0.0.0-20190930215403-16217165b5de h1:5hukYrvBGR8/eNkX5mdUezrA6JiaEZDtJb9Ei+1LlBs=
golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc=
golang.org/x/mod v0.4.2 h1:Gz96sIWK3OalVv/I/qNygP42zyoKp3xptRVCWRFEBvo=
golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210510120138-977fb7262007 h1:gG67DSER+11cZvqIMb8S8bt0vZtiN6xWYARwirrOSfE=
golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
golang.org/x/tools v0.0.0-20191108193012-7d206e10da11 h1:Yq9t9jnGoR+dBuitxdo9l6Q7xh/zOyNnYUtDKaQ3x0E=
golang.org/x/tools v0.0.0-20191108193012-7d206e10da11/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.1.5 h1:ouewzE6p+/VEB31YYnTbEJdi8pFqKp4P4n85vwo3DHA=
golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 h1:go1bK/D/BFZV2I8cIQd1NKEZ+0owSTG1fDTci4IqFcE=
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.8 h1:obN1ZagJSUGI0Ek/LBmuj4SNLPfIny3KsKFopxRdj10=
gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b h1:h8qDotaEPuJATrMmW04NCwg7v22aHH28wwpauUhK9Oo=
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
5 changes: 4 additions & 1 deletion go.mod
View file
Open in desktop
Expand Up @@ -2,4 +2,7 @@ module k8s.io/klog/v2

go 1.13

require github.com/go-logr/logr v1.2.0
require (
github.com/go-logr/logr v1.2.0
k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this adds a circular dependency... utils already depends on klog, we should avoid making klog also depend on utils

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the test is useful enough that we should not remove it and instead carry a copy of the clock code -> #310

)
11 changes: 11 additions & 0 deletions go.sum
View file
Open in desktop
@@ -1,2 +1,13 @@
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
github.com/go-logr/logr v1.2.0 h1:QK40JKJyMdUDz+h+xvCsru/bJhvG0UxvePV0ufL/AcE=
github.com/go-logr/logr v1.2.0/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/spf13/afero v1.2.2/go.mod h1:9ZxEEn6pIJ8Rxe320qSDBk6AsU0r9pR7Q4OcevTdifk=
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9 h1:HNSDgDCrr/6Ly3WEGKZftiE7IY19Vz2GdbOCyI4qqhc=
k8s.io/utils v0.0.0-20220210201930-3a6ce19ff2f9/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
90 changes: 85 additions & 5 deletions klog.go
View file
Open in desktop
Expand Up @@ -94,6 +94,7 @@ import (
"k8s.io/klog/v2/internal/buffer"
"k8s.io/klog/v2/internal/serialize"
"k8s.io/klog/v2/internal/severity"
"k8s.io/utils/clock"
)

// severityValue identifies the sort of log: info, warning etc. It also implements
Expand Down Expand Up @@ -380,7 +381,7 @@ type flushSyncWriter interface {
io.Writer
}

// init sets up the defaults and runs flushDaemon.
// init sets up the defaults.
func init() {
logging.stderrThreshold = severityValue{
Severity: severity.ErrorLog, // Default stderrThreshold is ERROR.
Expand All @@ -395,7 +396,7 @@ func init() {
logging.addDirHeader = false
logging.skipLogHeaders = false
logging.oneOutput = false
go logging.flushDaemon()
logging.flushD = newFlushDaemon(flushInterval, logging.lockAndFlushAll, nil)
}

// InitFlags is for explicitly initializing the flags.
Expand Down Expand Up @@ -447,6 +448,8 @@ type loggingT struct {
mu sync.Mutex
// file holds writer for each of the log types.
file [severity.NumSeverity]flushSyncWriter
// flushD holds a flushDaemon that frequently flushes log file buffers.
flushD *flushDaemon
// pcs is used in V to avoid an allocation when computing the caller's PC.
pcs [1]uintptr
// vmap is a cache of the V Level for each V() call site, identified by PC.
Expand Down Expand Up @@ -1022,6 +1025,7 @@ const bufferSize = 256 * 1024
// createFiles creates all the log files for severity from sev down to infoLog.
// l.mu is held.
func (l *loggingT) createFiles(sev severity.Severity) error {
l.flushD.run()
now := time.Now()
// Files are created in decreasing severity order, so as soon as we find one
// has already been created, we can stop.
Expand All @@ -1042,10 +1046,86 @@ func (l *loggingT) createFiles(sev severity.Severity) error {
const flushInterval = 5 * time.Second

// flushDaemon periodically flushes the log file buffers.
func (l *loggingT) flushDaemon() {
for range time.NewTicker(flushInterval).C {
l.lockAndFlushAll()
type flushDaemon struct {
mu sync.Mutex
clock clock.WithTicker
interval time.Duration
flush func()
stopC chan struct{}
stopDone chan struct{}
}

// newFlushDaemon returns a new flushDaemon. If the passed clock is nil, a
// clock.RealClock is used.
func newFlushDaemon(interval time.Duration, flush func(), tickClock clock.WithTicker) *flushDaemon {
if tickClock == nil {
tickClock = clock.RealClock{}
}
return &flushDaemon{
interval: interval,
flush: flush,
clock: tickClock,
}
}

// run starts a goroutine that periodically calls the daemons flush function.
// Calling run on an already running daemon will have no effect.
func (f *flushDaemon) run() {
f.mu.Lock()
defer f.mu.Unlock()

if f.stopC != nil { // daemon already running
return
}

f.stopC = make(chan struct{}, 1)
f.stopDone = make(chan struct{}, 1)

ticker := f.clock.NewTicker(f.interval)
go func() {
defer ticker.Stop()
defer func() { f.stopDone <- struct{}{} }()
for {
select {
case <-ticker.C():
f.flush()
case <-f.stopC:
thockin marked this conversation as resolved.
Show resolved Hide resolved
f.flush()
return
}
}
}()
}

// stop stops the running flushDaemon and waits until the daemon has shut down.
// Calling stop on a daemon that isn't running will have no effect.
func (f *flushDaemon) stop() {
f.mu.Lock()
defer f.mu.Unlock()

if f.stopC == nil { // daemon not running
return
}

f.stopC <- struct{}{}
<-f.stopDone

f.stopC = nil
f.stopDone = nil
}

// isRunning returns true if the flush daemon is running.
func (f *flushDaemon) isRunning() bool {
f.mu.Lock()
defer f.mu.Unlock()
return f.stopC != nil
}

// StopFlushDaemon stops the flush daemon, if running.
// This prevents klog from leaking goroutines on shutdown. After stopping
// the daemon, you can still manually flush buffers by calling Flush().
func StopFlushDaemon() {
logging.flushD.stop()
}

// lockAndFlushAll is like flushAll but locks l.mu first.
Expand Down
64 changes: 55 additions & 9 deletions klog_test.go
View file
Open in desktop
Expand Up @@ -39,6 +39,7 @@ import (
"k8s.io/klog/v2/internal/buffer"
"k8s.io/klog/v2/internal/severity"
"k8s.io/klog/v2/internal/test"
testingclock "k8s.io/utils/clock/testing"
)

// TODO: This test package should be refactored so that tests cannot
Expand Down Expand Up @@ -378,10 +379,11 @@ func TestSetOutputDataRace(t *testing.T) {
setFlags()
defer logging.swap(logging.newBuffers())
var wg sync.WaitGroup
var daemons []*flushDaemon
for i := 1; i <= 50; i++ {
go func() {
logging.flushDaemon()
}()
daemon := newFlushDaemon(time.Second, logging.lockAndFlushAll, nil)
daemon.run()
daemons = append(daemons, daemon)
}
for i := 1; i <= 50; i++ {
wg.Add(1)
Expand All @@ -391,9 +393,9 @@ func TestSetOutputDataRace(t *testing.T) {
}()
}
for i := 1; i <= 50; i++ {
go func() {
logging.flushDaemon()
}()
daemon := newFlushDaemon(time.Second, logging.lockAndFlushAll, nil)
daemon.run()
daemons = append(daemons, daemon)
}
for i := 1; i <= 50; i++ {
wg.Add(1)
Expand All @@ -403,11 +405,14 @@ func TestSetOutputDataRace(t *testing.T) {
}()
}
for i := 1; i <= 50; i++ {
go func() {
logging.flushDaemon()
}()
daemon := newFlushDaemon(time.Second, logging.lockAndFlushAll, nil)
daemon.run()
daemons = append(daemons, daemon)
}
wg.Wait()
for _, d := range daemons {
d.stop()
}
}

func TestLogToOutput(t *testing.T) {
Expand Down Expand Up @@ -1852,3 +1857,44 @@ func (s *structWithLock) addWithDefer() {
defer s.m.Unlock()
s.n++
}

func TestFlushDaemon(t *testing.T) {
for sev := severity.InfoLog; sev < severity.FatalLog; sev++ {
flushed := make(chan struct{}, 1)
spyFunc := func() {
flushed <- struct{}{}
}
testClock := testingclock.NewFakeClock(time.Now())
thockin marked this conversation as resolved.
Show resolved Hide resolved
testLog := loggingT{
flushD: newFlushDaemon(time.Second, spyFunc, testClock),
}

// Calling testLog will call createFile, which should start the daemon.
testLog.print(sev, nil, nil, "x")

if !testLog.flushD.isRunning() {
t.Error("expected flushD to be running")
}

testClock.Step(time.Second)
<-flushed
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should this be verifying something? What happens if there was a bug and flush-daemon did not call the func properly, what would this test report?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The statement verifies that flush was called and let's the test wait until the flush execution finished. If the function isn't called, the test will time out. If I don't wait for the function here, the testClock.Step() statements will be all executed directly after each other, without flushDaemon having time to flush in between, so the flushDaemon would only enter the case <-ticker.C() once, regardless of how many steps the clock does.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What I meant is maybe we should control the timeout and produce an error ourselves. The default is something like 10 minutes, right? It seems like we could wrap these reads in a timeout of 10 seconds and produce a better error.

testClock.Step(time.Second)
<-flushed
testLog.flushD.stop()
// wait for stop to trigger one last flush
<-flushed
}
}

func TestStopFlushDaemon(t *testing.T) {
logging.flushD.stop()
logging.flushD = newFlushDaemon(time.Second, func() {}, nil)
logging.flushD.run()
if !logging.flushD.isRunning() {
t.Error("expected flushD to be running")
}
StopFlushDaemon()
if logging.flushD.isRunning() {
t.Error("expected flushD to be stopped")
}
}