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

fix: spurious cancelation of async webhooks, better tracing #2969

Merged
merged 3 commits into from
Dec 20, 2022
Merged

fix: spurious cancelation of async webhooks, better tracing #2969

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c2adc6b
fix: add failing webhook async dispatch test
alnr Dec 19, 2022
134295e
fix: spurious cancelation of async webhooks, better tracing
alnr Dec 19, 2022
1da1cd1
chore: code review
ory-bot Dec 20, 2022
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
52 changes: 30 additions & 22 deletions selfservice/hook/web_hook.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -274,29 +274,36 @@ func (e *WebHook) execute(ctx context.Context, data *templateContext) error {
attribute.String("webhook.identity.nid", data.Identity.NID.String()),
)
}

var (
httpClient = e.deps.HTTPClient(ctx)
async = gjson.GetBytes(e.conf, "response.ignore").Bool()
parseResponse = gjson.GetBytes(e.conf, "can_interrupt").Bool()
tracer = trace.SpanFromContext(ctx).TracerProvider().Tracer("kratos-webhooks")
cancel context.CancelFunc = func() {}
Copy link
Member

Choose a reason for hiding this comment

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

The cancel function was not needed IMO. We only used it to cancel the time out, but the underlying HTTP client already has a timeout itself (it's 30 seconds)

spanOpts = []trace.SpanStartOption{trace.WithAttributes(attrs...)}
errChan = make(chan error, 1)
httpClient = e.deps.HTTPClient(ctx)
ignoreResponse = gjson.GetBytes(e.conf, "response.ignore").Bool()
canInterrupt = gjson.GetBytes(e.conf, "can_interrupt").Bool()
tracer = trace.SpanFromContext(ctx).TracerProvider().Tracer("kratos-webhooks")
spanOpts = []trace.SpanStartOption{trace.WithAttributes(attrs...)}
errChan = make(chan error, 1)
)
if async {
// dissociate the context from the one passed into this function
ctx, cancel = context.WithTimeout(context.Background(), 5*time.Minute)
spanOpts = append(spanOpts, trace.WithNewRoot())
}

ctx, span := tracer.Start(ctx, "Webhook", spanOpts...)
Copy link
Member

Choose a reason for hiding this comment

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

Here for example is the first instance where the incorrect context is used if async is true. The trace would be lost and without context!

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The code is correct. For async webhooks, we start a new root span which is not associated with the incoming span.

Copy link
Member

Choose a reason for hiding this comment

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

What do you think of doing something like this:

req := // ...
var cancel // ...
if (async) {
  makeRequestContext, cancel = context.WithTimeout(context.Background(), time.Minute())
  req = req.WithContext(ctx)
}

that way, we ensure that the context is never leaving the async execution path :)

Copy link
Member

Choose a reason for hiding this comment

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

The code is correct. For async webhooks, we start a new root span which is not associated with the incoming span.

But then the span would not appear as part of the request, and it would not be possible to correlate it to an incoming HTTP request? Is that really intentional?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes, the span would be orphaned in that sense. But still easy enough to find.

If you made the async webhook a child span of the incoming Kratos request span, the child will outlive the parent span. That will trip up many tools and generally violate OpenTelemetry semantics.

OpenTracing used to have a special trace-trace relationship (FollowsFrom) which was perfect for cases like this. AFAIK, that did not make it into the OpenTelemetry specification, though, and tooling support was sketchy.

Copy link
Member

Choose a reason for hiding this comment

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

Oh I see, that is a very good explanation - thank you!

So I did not include this in the merge, would you be open to make a follow-up PR with the span fixed?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

The merged patch is fine as-is. The OpenTelemetry people themselves are not sure how they want to models this: open-telemetry/opentelemetry-specification#65

I'd suggest looking at some traces in Jaeger and Tempo, and deciding whether we're OK with how those tools interpret the data in practice. If there's a major hiccup, we can then change the instrumentation.

Copy link
Member

Choose a reason for hiding this comment

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

Perfect, that sounds like a plan!

e.deps.Logger().WithRequest(req.Request).Info("Dispatching webhook")
t0 := time.Now()

req = req.WithContext(ctx)
if ignoreResponse {
// This is one of the few places where spawning a context.Background() is ok. We need to do this
// because the function runs asynchronously and we don't want to cancel the request if the
// incoming request context is cancelled.
//
// The webhook will still cancel after 30 seconds as that is the configured timeout for the HTTP client.
req = req.WithContext(context.Background())
Copy link
Member

Choose a reason for hiding this comment

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

Instead of overshadowing the context, we now use context.Background() only where it is needed - in the request context. An alternative would be to not include a context if the request is async:

 	if !ignoreResponse {
 	 	req = req.WithContext(ctx)
 	}

// spanOpts = append(spanOpts, trace.WithNewRoot())
}

startTime := time.Now()
go func() {
defer close(errChan)
defer cancel()
defer span.End()

resp, err := httpClient.Do(req.WithContext(ctx))
resp, err := httpClient.Do(req)
if err != nil {
span.SetStatus(codes.Error, err.Error())
errChan <- errors.WithStack(err)
Expand All @@ -307,7 +314,7 @@ func (e *WebHook) execute(ctx context.Context, data *templateContext) error {

if resp.StatusCode >= http.StatusBadRequest {
span.SetStatus(codes.Error, "HTTP status code >= 400")
if parseResponse {
if canInterrupt {
if err := parseWebhookResponse(resp); err != nil {
span.SetStatus(codes.Error, err.Error())
errChan <- err
Expand All @@ -320,16 +327,17 @@ func (e *WebHook) execute(ctx context.Context, data *templateContext) error {
errChan <- nil
}()

if async {
if ignoreResponse {
traceID, spanID := span.SpanContext().TraceID(), span.SpanContext().SpanID()
logger := e.deps.Logger().WithField("otel", map[string]string{
"trace_id": traceID.String(),
"span_id": spanID.String(),
})
Comment on lines +332 to +335
Copy link
Member

Choose a reason for hiding this comment

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

Always include trace info

go func() {
if err := <-errChan; err != nil {
e.deps.Logger().WithField("otel", map[string]string{
"trace_id": traceID.String(),
"span_id": spanID.String(),
}).WithError(err).Warning("Webhook request failed but the error was ignored because the configuration indicated that the upstream response should be ignored.")
logger.WithField("duration", time.Since(startTime)).WithError(err).Warning("Webhook request failed but the error was ignored because the configuration indicated that the upstream response should be ignored.")
} else {
e.deps.Logger().WithField("duration", time.Since(t0)).Info("Webhook request succeeded")
logger.WithField("duration", time.Since(startTime)).Info("Webhook request succeeded")
}
}()
return nil
Expand Down
33 changes: 11 additions & 22 deletions selfservice/hook/web_hook_integration_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -861,13 +861,15 @@ func TestAsyncWebhook(t *testing.T) {
URL: &url.URL{Path: "/some_end_point"},
Method: http.MethodPost,
}

incomingCtx, incomingCancel := context.WithCancel(context.Background())
if deadline, ok := t.Deadline(); ok {
// cancel this context one second before test timeout for clean shutdown
var cleanup context.CancelFunc
incomingCtx, cleanup = context.WithDeadline(incomingCtx, deadline.Add(-time.Second))
defer cleanup()
}

req = req.WithContext(incomingCtx)
s := &session.Session{ID: x.NewUUID(), Identity: &identity.Identity{ID: x.NewUUID()}}
f := &login.Flow{ID: x.NewUUID()}
Expand Down Expand Up @@ -899,36 +901,23 @@ func TestAsyncWebhook(t *testing.T) {
}
// at this point, a goroutine is in the middle of the call to our test handler and waiting for a response
incomingCancel() // simulate the incoming Kratos request having finished
close(blockHandlerOnExit)
timeout := time.After(200 * time.Millisecond)
for done := false; !done; {
Copy link
Member

Choose a reason for hiding this comment

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

I don't think this was needed, as we already wait for handlerEntered

if last := logHook.LastEntry(); last != nil {
msg, err := last.String()
require.NoError(t, err)
assert.Contains(t, msg, "Dispatching webhook")
var found bool
for !found {
for _, entry := range logHook.AllEntries() {
if entry.Message == "Webhook request succeeded" {
Copy link
Member

@aeneasr aeneasr Dec 20, 2022 •
edited
Loading

Choose a reason for hiding this comment

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

Make this independent of ordering and just look whether we find the succeeded log entry :)

found = true
break
}
}

select {
case <-timeout:
done = true
case <-time.After(50 * time.Millisecond):
// continue loop
}
}
logHook.Reset()
close(blockHandlerOnExit)
timeout = time.After(200 * time.Millisecond)
for {
if last := logHook.LastEntry(); last != nil {
msg, err := last.String()
require.NoError(t, err)
assert.Contains(t, msg, "Webhook request succeeded")
break
}
select {
case <-timeout:
t.Fatal("timed out waiting for successful webhook completion")
case <-time.After(50 * time.Millisecond):
// continue loop
}
}
require.True(t, found)
}