Cloud spanner client does not seem to be retrying broken streams

Documentation for snapshot transactions mentions:

If streaming a chunk fails due to a “resumable” error, Session.read() retries the StreamingRead API reqeust, passing the resume_token from the last partial result streamed.

This is the correct thing to do but the client actually does not seem to be doing this. I looked around a little bit in the code and I could find anyplace where it is retrying the read with the resume token.
This is the place where the stream would throw an error if it broke:
https://github.com/GoogleCloudPlatform/google-cloud-python/blob/9d9b6c0708ae6553458ad6107e5ad8efb23762e8/spanner/google/cloud/spanner/streamed.py#L132

I dont see any exception handling happening here and it looks like that the error would just be propagated back to the user.

@bjwatson

cc @bjwatson @tseaver @lukesneeringer

@tseaver

Our retry logic is in GAX, but leaving this open for @tseaver to confirm that it is applicable here.

This needs to be done at a higher level than GAX, since we send resume tokens along with the responses. So if a stream breaks, instead of retrying the whole stream, the library should just pass the resume token back in the request which will cause the stream to be resumed from that point onwards. It is further complicated since cloud spanner might not send resume tokens with all responses so the client library needs to buffer responses till it sees one with resume token and only then yield them to the user. This will allow it to safely resume in case the stream breaks.
This is the Java implementation of the same logic:
https://github.com/GoogleCloudPlatform/google-cloud-java/blob/master/google-cloud-spanner/src/main/java/com/google/cloud/spanner/SpannerImpl.java#L2166

What does GAX do for streaming calls? Does it retry the whole stream if it fails? What if the stream had yielded all but the last 1 response before failing? In that case retrying the whole stream is very wasteful. If GAX does do retries for streaming calls, we need to disable them.

What does GAX do for streaming calls? Does it retry the whole stream if it fails? What if the stream had yielded all but the last 1 response before failing? In that case retrying the whole stream is very wasteful. If GAX does do retries for streaming calls, we need to disable them.

gax is pretty naive in what it does for anything other than simple RPCs. I'm writing a doc internally on this now. :)

@lukesneeringer

@lukesneeringer @vkedia Does this need to be Beta blocking?

Till this is resolved can we fix the documentation to indicate that users would need to resume the stream themselves by passing back the resume token if available?

Snapshot.read and Snapshot.execute_sql return an instance of StreamedResultSet. That instance does not contain the information needed to reconstruct the original call inside its consume_next, which means that we cannot make handling such errors seamless to the user out-of-the box.

We could attempt to add enough information to the StreamedResultSet (a curried partial, maybe?) to allow such retries. E.g., from read():

    retry = functools.partial(self.read, table, columns, keyset, index, limit)
    return StreamedResultSet(iterator, retry=retry)

or from execute_sql:

    retry = functools.partial(self.execute_sql, params, param_types, query_mode)
    return StreamedResultSet(iterator, retry=retry)

The only other way I can see to hide such errors from the user would be to add callback-based methods which took the arguments to be passed to read / execute_sql, executed them, and then iterated the result set, invoking the callback for each row successfully pulled from the stream.

E.g.:

    def handle_row(row):
        ...

    snapshot.read_cb(table, keyset, ..., callback=handle_row)

    snapshot.execute_sql(sql, ..., callback=handle_row)

I don't understand how the callback solution fixes this. Can you please elaborate? How would you retry in that case?
Your option 1 is similar to what we do in Java as well. Note that when you retry you will have to pass back the resume_token, so the retry function could possible take a resume token as an argument.

Also note this is more complex because we do not guarantee that all PartialResultSet would have a resume token. So if you yield a Row to the user but the PRS that contained that Row did not have a resume_token, you cannot retry safely. So what you will need to do is store PRS's in a buffer till you see one with resume_token. Only then yield them to the user. So essentially you guarantee that every time you yield something to the user it is at a resumable boundary.

The curried partial is the pattern we'll use in google.api.core. Instead of passing the curried partial to StreamedResultSet, can you construct an intermediary iterator wrapper for the request?

class _RetryingStreamIterator(object):
    def __init__(self, iter, request):
        self._iter = iter
        self._request = request
        self._resume_token = None

    def __iter__(self):
        return self

    def __next__(self):
        try:
            next = six.next(self._iter)
            self._resume_token = next.resumeToken
        except grpc.RpcError as exc:
            if (exc.status_code == grpc.StatusCode.RESUMABLE
                    and self.resumeToken is not None):
                self._iter = self._request(resume_token=self._resume_token)
                return six.next(self._iter)
            else:
                raise

...


# read/execute_sql:

request = functools.partial(api.streaming_read,
    self._session.name, table, columns, keyset.to_pb(),
    transaction=transaction, index=index, limit=limit,
    resume_token=resume_token, options=options)

iterator = _RetryingStreamIterator(request(), request)

if self._multi_use:
    return StreamedResultSet(iterator, source=self)
else:
    return StreamedResultSet(iterator)

I applied the release blocking label temporarily, to make sure we reach some kind of resolution on this before Beta. Either there's a quick and easy fix, or we update the docs so that we're not misleading users. In the latter case, we'll want to fix this shortly after declaring Beta.

Document retry of exceptions while streaming 'read'/'execute_sql' res…

@jonparrott

@jonparrott the "stock" bits in google.api.core.retry don't really fit this case: I need to inject a "fixup" when the appropriate exception (UNAVAILABLE) is raised, before retrying the stream iteration. Something like:

    def consume_next(self):
        """Consume the next partial result set from the stream.

        Parse the result set into new/existing rows in :attr:`_rows`
        """
        sleep_generator = retry.exponential_sleep_generator(
            initial=retry._DEFAULT_INITIAL_DELAY,
            maximum=retry._DEFAULT_MAXIMUM_DELAY)

        for sleep in sleep_generator:
            try:
                response = six.next(self._response_iterator)
            except exceptions.ServiceUnavailable:

                if self._resume_token in (None, b''):
                    raise

                self._response_iterator = self._retry(
                    resume_token=self._resume_token)
            else:
                break

@tseaver

@tseaver that seems pretty reasonable. (it's one of the reasons I made exponential_sleep_generator public)

You should set a deadline of some sort, though.

Does the Python client allow users to specify deadline? If it does then the same deadline should be set adjusting for the time time elapsed so far.

…

On Mon, Aug 28, 2017 at 10:43 AM, Jon Wayne Parrott < ***@***.***> wrote: @tseaver <https://github.com/tseaver> that seems pretty reasonable. (it's one of the reasons I made exponential_sleep_generator public) You should set a deadline of some sort, though. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#3775 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ATdef1-qD59qWEW1G7iw2pW8ElDJyjstks5scvxNgaJpZM4OyakU> .

And of course I have to add my own deadline handling.

BTW, why does google.api.core.retry.retry_target use the more convoluted datetime stuff to compute the deadline, rather than just time.time plus a constant number of seconds?

BTW, why does google.api.core.retry.retry_target use the more convoluted datetime stuff to compute the deadline, rather than just time.time plus a constant number of seconds?

No particular reason other than being functionally equivalent and having on single source of truth for time in tests (datetime_helpers.utcnow()).

And of course I have to add my own deadline handling.

How so?

@jonparrott

@jonparrott:

And of course I have to add my own deadline handling.

How so?

As you pointed out above, I'm doing the low level work myself.

ISTM that the retry_target (and therefore Retry) could grow some feature to signal a "fixup" to happen between the retries, to accommodate this kind of stateful behavior.

I'm still not sure what you mean? Just pass the user's deadline into retry_target?

I can't use retry_target: it doesn't let me issue the "restart" request and capture the new iterator. And so I need to do my own deadline handling.

Thanks for the PR - I think I see now, so it seems like if you did this:

retry_target(functools.partial(six.next, self._response_iterator))

It wouldn't let you catch the error and replace self._response_iterator.

I could imagine a hack using the predicate and some getattr magic, but that sounds gross to me.

Is there a minimal change you could make to retry_target to make it work for you- perhaps an optional on_exception callback? If so, I'm happy to review and merge that.

Also - I've thought about a retry_iterator function - maybe it makes more sense to just go ahead and add that to core?

Use '_restart_on_unavailable' wrapper in 'SRS.{read,execute_sql}.

Retry streaming exceptions (this time for sure, Rocky!) (#4016)

@tseaver

Woo hoo! Thanks @tseaver!

Retry streaming exceptions (this time for sure, Rocky!) (#4016)

Navigation Menu

Repository navigation

Cloud spanner client does not seem to be retrying broken streams #3775

23 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Cloud spanner client does not seem to be retrying broken streams #3775

Description

Activity

vkedia commented on Aug 9, 2017

lukesneeringer commented on Aug 9, 2017

vkedia commented on Aug 9, 2017

theacodes commented on Aug 9, 2017

bjwatson commented on Aug 9, 2017

vkedia commented on Aug 10, 2017

tseaver commented on Aug 10, 2017

vkedia commented on Aug 10, 2017

theacodes commented on Aug 10, 2017

bjwatson commented on Aug 11, 2017

23 remaining items

tseaver commented on Aug 28, 2017

theacodes commented on Aug 28, 2017

vkedia commented on Aug 28, 2017

tseaver commented on Aug 28, 2017

theacodes commented on Aug 28, 2017

tseaver commented on Aug 28, 2017

theacodes commented on Aug 28, 2017

tseaver commented on Aug 28, 2017

theacodes commented on Aug 28, 2017

theacodes commented on Aug 28, 2017

bjwatson commented on Sep 22, 2017

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions