Skip to content

Add ops the ability to clean orphan commands #1434

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

Merged
pipecd-bot merged 10 commits into from
Jan 15, 2021
Merged

Add ops the ability to clean orphan commands #1434

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
dea8bbc
Create orphan commands cleaner that find orphan commands and update i…
sanposhiho Jan 14, 2021
fabf13a
Format unfmt-ed files
Jan 14, 2021
c7d7148
Fix BUILD.bazel
sanposhiho Jan 14, 2021
cf8d0dc
Merge branch 'orphan-commands-timeout' of https://github.com/pipe-cd/…
sanposhiho Jan 14, 2021
04a5385
Fix commandTimeout and interval
sanposhiho Jan 15, 2021
105bbaa
Fix typo
sanposhiho Jan 15, 2021
c0aaaf5
Handle errors correctly
sanposhiho Jan 15, 2021
bd2af05
Add context.Done
sanposhiho Jan 15, 2021
0512b9f
Add some information on error message
sanposhiho Jan 15, 2021
68021e7
Logging only when successfuly processed, and not return even if some …
sanposhiho Jan 15, 2021
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
1 change: 1 addition & 0 deletions cmd/pipecd/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ go_library(
"//pkg/app/api/stagelogstore:go_default_library",
"//pkg/app/ops/handler:go_default_library",
"//pkg/app/ops/insightcollector:go_default_library",
"//pkg/app/ops/orphancommandcleaner:go_default_library",
"//pkg/backoff:go_default_library",
"//pkg/cache/rediscache:go_default_library",
"//pkg/cli:go_default_library",
Expand Down
5 changes: 5 additions & 0 deletions cmd/pipecd/ops.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ import (
"github.com/pipe-cd/pipe/pkg/admin"
"github.com/pipe-cd/pipe/pkg/app/ops/handler"
"github.com/pipe-cd/pipe/pkg/app/ops/insightcollector"
"github.com/pipe-cd/pipe/pkg/app/ops/orphancommandcleaner"
"github.com/pipe-cd/pipe/pkg/backoff"
"github.com/pipe-cd/pipe/pkg/cli"
"github.com/pipe-cd/pipe/pkg/datastore"
Expand Down Expand Up @@ -98,6 +99,10 @@ func (s *ops) run(ctx context.Context, t cli.Telemetry) error {
}
}()

// Starting orphan commands cleaner
cleaner := orphancommandcleaner.NewOrphanCommandCleaner(ds, t.Logger)
cleaner.Run(ctx)

@nghialv nghialv Jan 15, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

This should be wrapped by errgroup to terminate ops when cleaner exit because of an error.

group.Go(func() error {
	return cleaner.Run(ctx)
})

@sanposhiho sanposhiho Jan 15, 2021 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

IMO, there was no need to stop ops because errors that occur in cleaner is probably caused by a network error when communicating with datastore.

Of course, it is necessary to output the error log to show that an error has occurred. However, I don't think it is necessary to stop ops because updateOrphanCommandsStatus will include the update that should have been done with the previous process if the previous process failed.

Or how about stopping ops if it fails several times in a row?

What do you think about it? 🤔

@nghialv nghialv Jan 15, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Nice comment. I got your points.
Agree with you that the cleaner may have some errors while executing its jobs.

But that is the problem of cleaner, not this run function.
This run should not care about what each component is doing, it just cares that they are fine or not. If any of them had a problem, ops should be stopped to notify that there was a problem from one of its components.
In other words, run should see all of its components (cleaner, http-server, insightcollector...) in the same way.

About the network problem or something that can cause the error while cleaner executing the job,
cleaner can ignore them, just log the error message, and does not return.
By that way, the ops will not be stopped by those errors and we can keep run to be nice too.

@sanposhiho sanposhiho Jan 15, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

About the network problem or something that can cause the error while cleaner executing the job,
cleaner can ignore them, just log the error message, and does not return.
By that way, the ops will not be stopped by those errors and we can keep run to be nice too.

OK, I see 👍


// Starting a cron job for insight collector.
if s.enableInsightCollector {
collector := insightcollector.NewInsightCollector(ds, fs, t.Logger)
Expand Down
13 changes: 13 additions & 0 deletions pkg/app/ops/orphancommandcleaner/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
load("@io_bazel_rules_go//go:def.bzl", "go_library")

go_library(
name = "go_default_library",
srcs = ["orphancommandcleaner.go"],
importpath = "github.com/pipe-cd/pipe/pkg/app/ops/orphancommandcleaner",
visibility = ["//visibility:public"],
deps = [
"//pkg/datastore:go_default_library",
"//pkg/model:go_default_library",
"@org_uber_go_zap//:go_default_library",
],
)
71 changes: 71 additions & 0 deletions pkg/app/ops/orphancommandcleaner/orphancommandcleaner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,71 @@
package orphancommandcleaner

import (
"context"
"time"

"go.uber.org/zap"

"github.com/pipe-cd/pipe/pkg/datastore"
"github.com/pipe-cd/pipe/pkg/model"
)

var (
commandTimeOut = 7 * 24 * time.Hour
interval = 24 * time.Hour

@sanposhiho sanposhiho Jan 14, 2021

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

What values should be set?

@nghialv nghialv Jan 15, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

commandTimeOut = 24 * time.Hour
interval = 6 * time.Hour

)

type OrphanCommandCleaner struct {
commandstore datastore.CommandStore
logger *zap.Logger
}

func NewOrphanCommandCleaner(
ds datastore.DataStore,
logger *zap.Logger,
) *OrphanCommandCleaner {
return &OrphanCommandCleaner{
commandstore: datastore.NewCommandStore(ds),
logger: logger.Named("orphan-command-finder"),

@nghialv nghialv Jan 15, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

orphan-command-cleaner

}
}

func (o *OrphanCommandCleaner) Run(ctx context.Context) {

@nghialv nghialv Jan 15, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

nit: func(c * because it is a cleaner.

for {
if err := o.updateOrphanCommandsStatus(ctx); err != nil {
o.logger.Error("failed to update orphan commands", zap.Error(err))
}
time.Sleep(interval)

@nghialv nghialv Jan 15, 2021 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Let's use time.Ticker instead of sleep.

}
}

func (o *OrphanCommandCleaner) updateOrphanCommandsStatus(ctx context.Context) error {
timeoutedTime := time.Now().Add(-commandTimeOut).Unix()

@nghialv nghialv Jan 15, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

nit: timedout or just timeout

opts := datastore.ListOptions{
Filters: []datastore.ListFilter{
{
Field: "Status",
Operator: "==",
Value: model.CommandStatus_COMMAND_NOT_HANDLED_YET,
},
{
Field: "CreatedAt",
Operator: "<=",
Value: timeoutedTime,
},
},
}
commands, err := o.commandstore.ListCommands(ctx, opts)
if err != nil {
return err
}

for _, c := range commands {
o.commandstore.UpdateCommand(ctx, c.Id, func(c *model.Command) error {

@nghialv nghialv Jan 15, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

Can you add a log to show that the command was successfully updated or not?

c.Status = model.CommandStatus_COMMAND_TIMEOUT
return nil
})
}

return nil
}