googleapis / googleapis/google-cloud-go

pubsub: allow timing out publish requests and retrying

Open
#11,942 1 comment 0 reactions 1 assignee Claimed by @hongalex View on GitHub
api: pubsub type: feature request
Dominant language
Go
Stars
4.5k
Forks
1.6k
Avg merge
1d 13h
Merged PRs (30d)
109

Description

## Is your feature request related to a problem? Please describe.

Every day we get a couple of >1000ms (sometimes up to 10 seconds) spikes on the `rpc_client_duration_milliseconds_bucket` `Publish` method metric from on many of our services (running in GKE) publishing to pubsub. The amount of spikes is low, and in higher traffic it usually falls into the p99.99 and higher percentiles, however during periods of low traffic, it affects the lower percentiles as well (p99), causing us to breach our SLAs:

![Image](https://github.com/user-attachments/assets/76babd6d-5d7f-4c04-91ac-7f59c2917732)

The spikes are also visible in the GCP metrics explorer and we couldn't see anything wrong on our end that could be causing this issue, network-wise:

![Image](https://github.com/user-attachments/assets/6087b4d7-7e40-4f77-888b-9d794c852d76)

We use the default publish settings and our throughput is low, usually we publish only 1 to 2 messages per batch. We've tried many things and we're quite confident that we're using the library correctly (e.g. we're not creating a new client on each call, we're making sure topic objects are cached etc.)

After a while, we thought we could try timing out early and retrying. To try this, we forked the library, changed the [publisher_client Publish method](https://github.com/googleapis/google-cloud-go/blob/99bc72b20779654226392070a46e1371a89e73b0/pubsub/apiv1/publisher_client.go#L668) to a hardcoded timeout of 200ms and made it retry if it reaches this timeout. It seemed to completely eliminate the spikes without any other side effects (other than probably slightly increasing odds for duplicate messages, but we're fine with that) - most of the time it would succeed on the 1st reattempt, although it would sometimes take a few more tries to get it through.

This is what the Publish method looks like with our very simple retry loop. The code is not sophisticated and doesn't handle retrying for other cases such as network errors. We just wanted to check if this helps our spikes or not:

```
func (c *publisherGRPCClient) Publish(ctx context.Context, req *pubsubpb.PublishRequest, opts ...gax.CallOption) (*pubsubpb.PublishResponse, error) {
hds := []string{"x-goog-request-params", fmt.Sprintf("%s=%v", "topic", url.QueryEscape(req.GetTopic()))}

hds = append(c.xGoogHeaders, hds...)
ctx = gax.InsertMetadataIntoOutgoingContext(ctx, hds...)
opts = append((*c.CallOptions).Publish[0:len((*c.CallOptions).Publish):len((*c.CallOptions).Publish)], opts...)
var resp *pubsubpb.PublishResponse
var err error
attempts := 0
var settings gax.CallSettings

for {
attempts++
if attempts == 50 {
fmt.Println(fmt.Sprintf("maximum attempts exceeded while publishing to topic %s", req.Topic))
break
}

ctx, _ := context.WithDeadline(ctx, time.Now().Add(200*time.Millisecond))
resp, err = executeRPC(ctx, c.publisherClient.Publish, req, settings.GRPC, c.logger, "Publish")
if err != nil {
errCode := status.Code(err)
if errCode == codes.DeadlineExceeded {
fmt.Println(fmt.Sprintf("attempt number %d, msg to topic %s took longer than 200ms", attempts, req.Topic))
continue
}
fmt.Println(fmt.Sprintf("non-deadline error while publishing to topic %s occurred %s", req.Topic, err.Error()))
return nil, err
}

break
}

return resp, nil
}
```

## Describe the solution you'd like
We would like a way to timeout publish requests early and reattempt. Ideally, we would want to set the publish timeout in the publisher settings, as well as the maximum number of retries. If this number is breached, we'd still like to publish but without timing out early - to ensure the message is published.

## Describe alternatives you've considered
As far as we understand it's not possible to achieve this behaviour with the way the package currently works. `topic.Publish` just adds a message to the scheduler, which has its own context - meaning we couldn't pass in our own context. There is an option to set a Timeout on the publisher settings, but it seems like [gax's retry mechanism doesn't work with context timeouts](https://github.com/googleapis/google-cloud-go/issues/1941), meaning we would never retry if this timeout is reached.

## Additional context

Might be similar to https://github.com/googleapis/google-cloud-go/issues/11449.

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.