Support keyword arguments when creating Pub/Sub subscriptions

[hernandezc1]

Users will be interested in filtering Pub/Sub messages either by using the message attributes and/or the content of the message payload itself. It would be great to support the passing of keyword arguments to enable this filtering when creating the subscription.

[troyraen]

I agree, it would be really helpful for us to support the creation of a subscription with an attached filter (ie, https://docs.cloud.google.com/pubsub/docs/subscription-message-filter - but recall that this type of filter can only access message attributes[*]). We could easily support this by accepting the filter string as a kwarg. The user would have to construct a valid filter string. The docs linked above look pretty clear, so we should point users there.

We might want to think about how to help the user validate a filter string. I don't think the Pub/Sub service provides a way to do this other than to try it on a real subscription (I doubt pubsub has enough info to validate a priori). We could consider maintaining a list in this repo of the attribute names for each topic. A more complicated option would be to incorporate attributes into pittgoogle.schema.Schema -- that would require us to maintain a schema for every topic rather than only for every survey, but could have the added benefit of making it easy for our broker modules to transform the incoming schema to the outgoing one (could solve #97 and #89 and potentially be done in combination with the idea in mwvgroup/Pitt-Google-Broker#271 (comment)).

But we shouldn't let validation hold us back from doing the easy thing of supporting creation of subscriptions with attached filters.

[*] For filtering on the message payload, we do support that in pittgoogle.pubsub.Consumer() with the msg_callback and batch_callback kwargs. This type of filter is applied after a message is pulled from a subscription. (By contrast, messages that do not pass a filter attached to the subscription are never delivered to the client.)

[hernandezc1]

@troyraen I've gotten a bit confused with how Google's documentation defines/describes "filtering". You're right that Pub/Sub's built-in filters can only filter messages by their attributes and not by the data in the message, but Pub/Sub does support filtering on the message payload through User Defined Functions. It's implementation is just as straightforward as defining the Pub/Sub message attribute filter, and I think that many users would want to do both. The following documentation outlines how SMTs interact with other subscription features. The important thing to note here is that the message attribute filter is applied before the SMT.

Here is an example that I think would be useful:

# Topic that the subscription should be connected to
topic = pittgoogle.Topic(name="lsst-supernnova", projectid=pittgoogle.ProjectIds().pittgoogle)

# Specify filters (Optional)
filterByClassAndHealpixIndex = "attributes.pg_supernnova_class = '0' AND attributes.healpix9 = '3105751'"
filterByClassProbability = """
        function filterByClassProbability(message, metadata) {
            const data = JSON.parse(message.data);

            // Find and extract the value_added dictionary
            const valueAddedKey = Object.keys(data).find(k => k !== "alert_lite");
            const valueAddedData = valueAddedKey ? (data[valueAddedKey] || {}) : {};

            // Extract the predicted class probability
            const classProbability = Number(valueAddedData["prob_class0"]);
            return (classProbability >= 0.90) ? message : null;
        }
        """

# Create the subscription
subscription = pittgoogle.Subscription(
        name="my-subscription",
        topic=topic,
        schema_name="lsst",
        attribute_filter= filterByClassAndHealpixIndex,
        udf= filterByClassProbability
    )
subscription.touch()

In the example above, the user creates a subscription to the lsst-supernnova topic. The specified attribute filter ensures that only alerts with a SuperNNova-predicted class of "0" (Type Ia candidates) and belonging to the specified HEALPix order-9 index (3105751) are passed to the UDF. All other messages are automatically acknowledged and dropped by Pub/Sub. The user-defined function (UDF) then performs an additional layer of filtering, keeping only those alerts whose SuperNNova-predicted class probability (prob_class0) is at least 0.9, and discarding the rest.

[troyraen]

@troyraen I've gotten a bit confused with how Google's documentation defines/describes "filtering".

I look at it like this. There's only one thing that GCP calls a subscription "filter" (noun) and it's this: https://docs.cloud.google.com/pubsub/docs/subscription-message-filter. The only thing that will do is filter (verb). By contrast, the thing GCP calls an "SMT" can also filter (verb) but that's not the only kind of transformation it can do.