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

Exit the for-loop after 5 gRPC communication errors #218

Merged
merged 4 commits into from
Sep 3, 2024
Merged

Exit the for-loop after 5 gRPC communication errors #218

Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
b153005
Exit the for-loop after 5 gRPC communication errors
shackra Aug 23, 2024
8ae346d
Reset counter and ignore unknown errors
shackra Aug 24, 2024
2d66da3
Return an error if we exit the for-loop
shackra Aug 24, 2024
02a98f4
Search for the error message we care to stop the for-loop
shackra Aug 28, 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
19 changes: 18 additions & 1 deletion pkg/connectorrunner/runner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@ import (
"github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"google.golang.org/protobuf/types/known/durationpb"

v1 "github.com/conductorone/baton-sdk/pb/c1/connectorapi/baton/v1"
Expand Down Expand Up @@ -128,8 +130,9 @@ func (c *connectorRunner) run(ctx context.Context) error {

waitDuration := time.Second * 0
errCount := 0
commErrCount := 0
var err error
for {
for commErrCount < 5 {
shackra marked this conversation as resolved.
Show resolved Hide resolved
shackra marked this conversation as resolved.
Show resolved Hide resolved
select {
case <-ctx.Done():
return c.handleContextCancel(ctx)
Expand Down Expand Up @@ -194,14 +197,28 @@ func (c *connectorRunner) run(ctx context.Context) error {
defer sem.Release(1)
err := c.processTask(ctx, t)
if err != nil {
switch code := status.Code(err); code {
case codes.Canceled:
commErrCount++
default:
l.Info("Received gRPC error code", zap.Uint32("grpc_code", uint32(code)))
}
l.Error("runner: error processing task", zap.Error(err), zap.String("task_id", t.Id), zap.String("task_type", tasks.GetType(t).String()))
}
// reset the counter
commErrCount = 0
l.Debug("runner: task processed", zap.String("task_id", t.Id), zap.String("task_type", tasks.GetType(t).String()))
}(nextTask)

l.Debug("runner: dispatched task, waiting for next task", zap.Duration("wait_duration", waitDuration))
}
}

if commErrCount > 0 {
return fmt.Errorf("Unable to communicate with gRPC server")
}

return nil
}

func (c *connectorRunner) Close(ctx context.Context) error {
Expand Down
Loading