diff --git a/agent/rpc/client_grpc.go b/agent/rpc/client_grpc.go index 4e80787ec..191b75f6e 100644 --- a/agent/rpc/client_grpc.go +++ b/agent/rpc/client_grpc.go @@ -60,6 +60,9 @@ type client struct { // restored before the agent gives up and exits. Zero means infinite. // Maps directly onto backoff.WithMaxElapsedTime. connectionRetryTimeout time.Duration + // logEntryBufferSize sets the buffer size before the log reader is blocking. + // If you have enough memory and log bursts, set it higher. + logEntryBufferSize int } // NewGrpcClient returns a new grpc Client. @@ -67,12 +70,14 @@ func NewGrpcClient(ctx context.Context, conn *grpc.ClientConn, opts ...ClientOpt client := new(client) client.client = proto.NewWoodpeckerClient(conn) client.conn = conn - client.logs = make(chan *proto.LogEntry, 10) // max memory use: 10 lines * 1 MiB + client.logEntryBufferSize = 100 for _, opt := range opts { opt(client) } + client.logs = make(chan *proto.LogEntry, client.logEntryBufferSize) // max memory use: buffer count * 1 MiB + go client.processLogs(ctx) return client } @@ -88,6 +93,17 @@ func SetConnectionRetryTimeout(d time.Duration) ClientOption { } } +func SetLogEntryBufferSize(count int) ClientOption { + return func(c *client) { + if count < 0 { + log.Error().Msgf("LogEntry Buffer can not be negative") + return + } + log.Info().Msgf("log-entry stream buffer size set to max %dMb", count) + c.logEntryBufferSize = count + } +} + // IsConnected reports whether the underlying gRPC connection is currently up. // It is a pure observer with no side effects. func (c *client) IsConnected() bool { diff --git a/cmd/agent/core/agent.go b/cmd/agent/core/agent.go index 92abe2719..9d99330ed 100644 --- a/cmd/agent/core/agent.go +++ b/cmd/agent/core/agent.go @@ -137,6 +137,7 @@ func run(ctx context.Context, c *cli.Command, backends []types.Backend) error { client := agent_rpc.NewGrpcClient( ctx, agentConn.MainConn, agent_rpc.SetConnectionRetryTimeout(c.Duration("retry-timeout")), + agent_rpc.SetLogEntryBufferSize(c.Int("log-entry-stream-buffer-size")), ) agentConfigPersisted := atomic.Bool{} diff --git a/cmd/agent/core/flags.go b/cmd/agent/core/flags.go index d880f073f..ec452f6fc 100644 --- a/cmd/agent/core/flags.go +++ b/cmd/agent/core/flags.go @@ -58,6 +58,12 @@ var flags = []cli.Flag{ Usage: "how long the agent keeps retrying to reconnect to the server after the gRPC connection is lost before giving up, set to 0 to retry forever", Value: 2 * time.Minute, }, + &cli.IntFlag{ + Sources: cli.EnvVars("WOODPECKER_LOG_ENTRY_STREAM_BUFFER_SIZE"), + Name: "log-entry-stream-buffer-size", + Usage: "how many log lines an agent can buffer before it blocks io.Pipe, expect one log-entry to be 1 MB in worst case.", + Value: 100, + }, &cli.StringFlag{ Sources: cli.EnvVars("WOODPECKER_HOSTNAME"), Name: "hostname", diff --git a/docs/docs/30-administration/10-configuration/30-agent.md b/docs/docs/30-administration/10-configuration/30-agent.md index b16c361e1..bb7dc3783 100644 --- a/docs/docs/30-administration/10-configuration/30-agent.md +++ b/docs/docs/30-administration/10-configuration/30-agent.md @@ -241,6 +241,20 @@ If set to 0 we retry forever. --- +## LOG_ENTRY_STREAM_BUFFER_SIZE + +- Name: `WOODPECKER_LOG_ENTRY_STREAM_BUFFER_SIZE` +- Default: `100` + +Set how many log lines an agent can buffer before it blocks io.Pipe, expect logentries to reach 1 MB in worst case. +If used with local backend, tis can increase your performance in special cases significantly. + +:::warning +If set to 0 we are always blocking. +::: + +--- + ### BACKEND - Name: `WOODPECKER_BACKEND`