Sync minimal flow data

[dwsutherland]

addresses #585
addresses #568

This change aims to:

• interrogate the graphql subscription for data requirements.
• Sync only the data required.
• Reduce the protobuf subscription topics and all data dump size.
• Handle data requirements for queries.

This first attempt is functional, reads the fragments requested by the UI with
{'AddedDelta', 'WorkflowData', 'UpdatedDelta'}
being the workflow summary information of the dashboard subscription.

The UI doesn't seem to reduce the subscription if you go from tree view to dashboard, however, closing the UI will reduce it back down.

One thing we need to be able to handle is general queries hitting the UIS, as we'll want CLI via UIS in at some point and to be able to handle random queries.. For this I think we can use that timer concept, keeping a full sync of workflows queried for some amount of time.. Done

Check List

• I have read CONTRIBUTING.md and added my name as a Code Contributor.
• Contains logically grouped changes (else tidy your branch by rebase).
• Does not contain off-topic changes (use other PRs for other changes).
• Applied any dependency changes to both setup.cfg (and conda-environment.yml if present).
• Tests are included (or explain why tests are not needed).
• CHANGES.md entry included if this is a change that can affect users
• Cylc-Doc pull request opened if required at cylc/cylc-doc/pull/XXXX.
• If this is a bug fix, PR should be raised against the relevant ?.?.x branch.

[dwsutherland]

more updates inc..

[dwsutherland]

Ok I've added more features and taken some ideas from #568 ..
Turns out you can subscribe/unsubscribe from topics without restarting the connection:
#568

If you print the topic near the top of the data_store_mgr.py _update_workflow_data method, you can see it changing when you open and close the graph view:

It's even more noticeable with multiple workflows running, and switching between them in the UI.

[dwsutherland]

Not sure the reason for this mypy failure:

new_data[field.name].CopyFrom(value) has always been there... and passes locally

Also, I get a lot of this:

ERROR cylc/uiserver/tests/test_workflows_mgr_data_store.py::test_data_store_changes[active-active-actions8] - AttributeError: 'CylcUIServer' object has no attribute 'profiler'. Did you mean: 'profile'?

ERROR cylc/uiserver/utils.py::uiserver.utils.fmt_call - AttributeError: 'CylcUIServer' object has no attribute 'profiler'. Did you mean: 'profile'?

from pytest locally.. I assume I'm missing some sort of profiler

[oliver-sanders]

@dwsutherland, I don't think we are going to be able to get this ready for 1.5.0, however, provided we don't need to change much of the cylc-flow code we can make a UIS release for this change at any point so we don't necessarily need to tether this work to cylc-flow milestones.

There are two technical barriers to cross with this work:

1. Determining topics we must subscribe to in order to satisfy a query/subscription.
2. Determining when we no longer require to subscribe to a topic (as a result of query completion / subscription completion).

I think we need to really pin these two mechanisms down before proceeding as they are quite involved and very fundamental to UIS functionality.

[oliver-sanders]

I don't think that we can reliably determine the topic(s) we need to subscribe to by looking at the GraphQL fragments alone:

• We have to handle queries as well as subscriptions.
• Subscriptions might not use fragments at all.
• Without looking inside the fragments, we can't tell what they are looking at anyway.
• We can have subscriptions that request only data present at the UIS but require no data from the scheduler (e.g. analysis view queries and log file subscriptions).

The difficulty in coming up with a reliable mechanism for this is why I didn't go ahead with the code I wrote for #568 (which worked, but just wasn't production grade). I think there are fundamentally two approaches:

1. Parse the query before it is resolved (inefficient and has security implications).
2. Add hooks into the resolvers (awkward).

To achieve (2) we could "spy" on the resolvers to see which resolvers a query/subscription is going to hit, then subscribe to the corresponding topics, wait for the data to arrive, and return the response. I.E:

async def resolve_task_proxies(...):
    await subscribe_topic('tasks')
    ...

[oliver-sanders]

@dwsutherland, I had a poke at the info object and found that it contains the complete parsed query.

This should allow us to determine what the query/subscription is looking at without having to rely on the client/user writing their query to use fragments named in a particular way or for the UIS to treat queries differently to subscriptions. It should also allow for finer grained protobuf subscriptions if we decide to go down that route (e.g. only subscribe to tasks & families to satisfy the tree view, add edges to the subscription to satisfy the graph view).

I had a quick go at using this to extract the requested field types. It's crude, but it seems to be good enough to determine whether a query/subscription is requesting top-level Cylc types e.g. tasks, taskProxies, edges which is all we need to know for this purpose.

from graphql import language


TOP_LEVEL_THINGGIES = {
    # TODO: populate from somewhere else?
    'tasks',
    'taskProxies',
    'families',
    'familyProxies',
    'edges',
}


def iter_top_level_types(info):
    """Loop through top-level Cylc types requested in a query/subscription."""
    ret = set()
    for selection_set in iter_selection_sets(info):
        for selection in selection_set.selections:
            # TODO: check this handles query nesting correctly, it appears to
            if isinstance(selection, language.ast.Field):
                # TODO: it looks like selection items are always fields?
                field_name = selection.name.value
                if field_name in TOP_LEVEL_THINGGIES:
                    if field_name not in ret:
                        ret.add(field_name)
                        yield field_name


def iter_selection_sets(info):
    """Loop through top-level selection sets in the query definitions."""
    if hasattr(info, 'operation'):
        yield info.operation.selection_set
    if hasattr(info, 'fragments'):
        for fragment in info.fragments.values():
            yield fragment.selection_set

[dwsutherland]

I agree we can do this...
IMO - this can just be a refinement change, the bones are ready in the PR, and we can just build on it by changing the criteria from fragments to field sets.

But we're running out of time for 8.3.0.. so going to have to bump the cylc-flow end to 8.4

[oliver-sanders]

So long as this doesn't require changes to cylc-flow (which is likely necessary for back-compat), then this can be put into a cylc-uiserver release at any point (no need to wait for any particular cylc-flow release).

[oliver-sanders]

I'm not sure how Protobuf subscriptions could be escalated back to the minimum level after GraphQL subscriptions complete under this model.

[dwsutherland]

They already do, once the subscription ends, switching between workflows in the UI will do it.. or closing the browser (obviously)

[dwsutherland]

https://github.com/dwsutherland/cylc-uiserver/blob/2605026328a6168664c6a32d27131c1652bc12af/cylc/uiserver/data_store_mgr.py#L624-L630

[dwsutherland]

provided we don't need to change much of the cylc-flow code

It can go in now, and not break anything.. So if we can at least do that for 8.3.0, then we can move forward with this later.

I don't think that we can reliably determine the topic(s) we need to subscribe to by looking at the GraphQL fragments

We can, and it works, but yes not reliable (if UI graphql changes, it could break it).. if we could pick up the fragments from the associated UI code then maybe.. For any subscription that doesn't contain those fragments (i.e. a bespoke subscription) it will default to max.
Parsing for every field requiring resolvers into sets could be another way..
Should only need to happen at the start of a new subscription, so we can probably get away with something more intensive then.

[dwsutherland]

Have a solution for queries, hopefully post it up tomorrow..

[dwsutherland]

Query sync level change and expire implemented, and functional:

The expiry is set to 60sec, with a periodic callback every 30sec to check and demote. (~30sec expiry in the above example)

The expiry will take precedence over subscription start/stops, so the level will remain.
Otherwise the levels promote/demote immediately in accordance with subscription start/stop.

[dwsutherland]

provided we don't need to change much of the cylc-flow code

It can go in now, and not break anything.. So if we can at least do that for 8.3.0, then we can move forward with this later.

I was wrong on this, I think the current UIS will break with the cylc-flow end in first (because the data-store manager on the UIS doesn't contain the methods that will be called by the resolvers)

So they both need to go in .. which is a shame .. because I assume the next 8.x release will be some way off..

[dwsutherland]

I think I may have found an easier way to hook into the query/(subscriptions) before execution .. Actually, at other points of the execution too:
https://github.com/graphql-python/graphene-tornado/blob/master/graphene_tornado/graphql_extension.py

So, probably a good thing it doesn't go in yet

[oliver-sanders]

because I assume the next 8.x release will be some way off..

8.3.0 was a monster, we're hoping to resume a more regular release cycle once this is out the way, 8.4.0 could be ~1 month away.

However, back compat is important, otherwise the GUI won't be able to connect to older workflows, so a cylc-uiserver side approach is definitely preferable. Note, if we do need to cut off back-compat, the way we do this is to increment the API version number in cylc-flow (currently 5). There is an API number filter in the scan mechanism in cylc-uiserver which will filter out older or newer workflows avoiding runtime problems.