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

feat(performance): improve Log Processing performance #5647

Merged
merged 7 commits into from
Jul 10, 2024
Merged

feat(performance): improve Log Processing performance #5647

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b131a55
feat(performance): buffer the logs sent from the container, to avoid …
rangoo94 Jul 9, 2024
fdc56a0
feat(performance): batch Test Workflow's result updates
rangoo94 Jul 9, 2024
dc437a8
fix(testworkflows): handle getting long container logs after the log …
rangoo94 Jul 10, 2024
791b20d
feat(testworkflows): optimize reading timestamp from Kubernetes logs
rangoo94 Jul 10, 2024
bc5cc9e
feat(testworkflows): optimize buffering logs
rangoo94 Jul 10, 2024
ead5e95
feat(testworkflows): use native channel instead of heavier Channels f…
rangoo94 Jul 10, 2024
5ff180f
feat(testworkflows): increase buffer size for logs buffering
rangoo94 Jul 10, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions pkg/testworkflows/testworkflowcontroller/channel.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ type Channel[T any] interface {
Channel() <-chan ChannelMessage[T]
Close()
Done() <-chan struct{}
CtxErr() error
}

type channel[T any] struct {
Expand Down Expand Up @@ -168,3 +169,7 @@ func (c *channel[T]) Close() {
func (c *channel[T]) Done() <-chan struct{} {
return c.ctx.Done()
}

func (c *channel[T]) CtxErr() error {
return c.ctx.Err()
}
8 changes: 4 additions & 4 deletions pkg/testworkflows/testworkflowcontroller/controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,7 +212,7 @@ func (c *controller) StopController() {
}

func (c *controller) Watch(parentCtx context.Context) <-chan ChannelMessage[Notification] {
w, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
ch, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
JobEvents: c.jobEvents,
Job: c.job,
})
Expand All @@ -222,7 +222,7 @@ func (c *controller) Watch(parentCtx context.Context) <-chan ChannelMessage[Noti
v.Close()
return v.Channel()
}
return w.Channel()
return ch
}

// TODO: Make it actually light
Expand Down Expand Up @@ -281,15 +281,15 @@ func (c *controller) Logs(parentCtx context.Context, follow bool) io.Reader {
case <-c.podEvents.Peek(parentCtx):
case <-alignTimeoutCh:
}
w, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
ch, err := WatchInstrumentedPod(parentCtx, c.clientSet, c.signature, c.scheduledAt, c.pod, c.podEvents, WatchInstrumentedPodOptions{
JobEvents: c.jobEvents,
Job: c.job,
Follow: common.Ptr(follow),
})
if err != nil {
return
}
for v := range w.Channel() {
for v := range ch {
if v.Error == nil && v.Value.Log != "" && !v.Value.Temporary {
if ref != v.Value.Ref && v.Value.Ref != "" {
ref = v.Value.Ref
Expand Down
Loading
Loading