Merge pull request #62 from mwvgroup/v/0.5.0/tjr

V/0.5.0/tjr

Author: troyraen

README.md

@@ -1,30 +1,13 @@
+[![python](https://img.shields.io/badge/python-3.7-g.svg)]()
+[![Build Status](https://travis-ci.com/mwvgroup/Pitt-Google-Broker.svg?branch=master)](https://travis-ci.com/mwvgroup/Pitt-Google-Broker)
+[![Documentation Status](https://readthedocs.org/projects/pitt-broker/badge/?version=latest)](https://pitt-broker.readthedocs.io/en/latest/?badge=latest)
+
 # Pitt-Google Alert Broker
 
 The Pitt-Google broker is an astronomical alert broker that is being developed for large scale surveys of the night sky, particularly the upcoming [Vera Rubin Observatory's Legacy Survey of Space and Time](https://www.lsst.org/) (LSST).
 We currently process and serve the [Zwicky Transient Facility](https://www.ztf.caltech.edu/)'s (ZTF) nightly alert stream.
-The broker runs on the Google Cloud Platform ([GCP](https://cloud.google.com)).
-
----
-
-## Access the Data
-
-See [Pitt-Google-Tutorial-Code-Samples.ipynb](https://github.com/mwvgroup/Pitt-Google-Broker/blob/master/pgb_utils/tutorials/Pitt-Google-Tutorial-Code-Samples.ipynb) for a tutorial.
-Data can be accessed using Google's [Cloud SDK](https://cloud.google.com/sdk) (Python, command-line, etc.).
-In addition, we offer the Python package `pgb_utils` which contains wrappers of Cloud SDK methods and other helper functions to facilitate common use cases.
-See the tutorials for details.
-
-If you run into issues or need assistance, please open an Issue on GitHub or contact troy.raen@pitt.edu.
+The broker runs on the [Google Cloud Platform](https://cloud.google.com) (GCP).
 
----
+Documentation is at [pitt-broker.readthedocs.io](https://pitt-broker.readthedocs.io/).
 
-## Run the Alert Broker
-
-See [broker/README.md](broker/README.md) for information about the alert broker software and instructions on running it. The broker will connect to a survey alert stream (e.g., ZTF) and process \& redistribute the data. Those looking to __access__ the data do not need to run the broker; instead see [Access the Data](#access-the-data)
-
-<!-- Full online documentation is available online via [Read the Docs](https://pitt-broker.readthedocs.io/en/latest/index.html). -->
-
----
-
-[![python](https://img.shields.io/badge/python-3.7-g.svg)]()
-[![Build Status](https://travis-ci.com/mwvgroup/Pitt-Google-Broker.svg?branch=master)](https://travis-ci.com/mwvgroup/Pitt-Google-Broker)
-[![Documentation Status](https://readthedocs.org/projects/pitt-broker/badge/?version=latest)](https://pitt-broker.readthedocs.io/en/latest/?badge=latest)
+If you run into issues or need assistance, please [open an Issue](https://github.com/mwvgroup/Pitt-Google-Broker/issues).

broker/broker_utils/MANIFEST.in

@@ -0,0 +1 @@
+include requirements.txt

broker/broker_utils/broker_utils/beam_transforms.py

@@ -52,7 +52,7 @@ def process(self, alertDict):
 
 
 class ExtractDIASource(DoFn):
-    """Extract the DIA source` fields and information needed for provinance
+    """Extract the DIA source` fields and information needed for provenance
     from the alertDict.
     """
     def __init__(self, schema_map):
@@ -74,7 +74,7 @@ def process_decat(self, alertDict):
         sourcename = lambda x: x if x not in dup_cols else f'source_{x}'
         src = {sourcename(k):v for k,v in alertDict['triggersource'].items()}
 
-        # get info for provinance
+        # get info for provenance
         notmetakeys = ['triggersource', 'sources']
         metadict = {k:v for k,v in alertDict.items() if k not in notmetakeys}
 
@@ -94,7 +94,7 @@ def process_ztf(self, alertDict):
         dup_cols = ['candid']  # candid is repeated, drop the one nested here
         cand = {k:v for k,v in alertDict['candidate'].items() if k not in dup_cols}
 
-        # get info for provinance
+        # get info for provenance
         metakeys = ['schemavsn', 'publisher', 'objectId', 'candid']
         metadict = {k:v for k,v in alertDict.items() if k in metakeys}
 

broker/broker_utils/broker_utils/consumer_sim.py

@@ -21,7 +21,8 @@ def publish_stream(
     publish_batch_every: Tuple[int,str] = (5,'sec'),
     sub_id: Optional[str] = None,
     topic_id: Optional[str] = None,
-    nack: bool = False
+    nack: bool = False,
+    auto_confirm: bool = False
     ):
     """Pulls messages from from a Pub/Sub subscription determined by either
     `instance` or `sub_id`, and publishes them to a topic determined by either
@@ -54,6 +55,8 @@ def publish_stream(
                     messages are published to the topic, but they are not
                     dropped from the subscription and so will be delivered again
                     at an arbitrary time in the future.
+
+        auto_confirm:   Whether to automatically answer "Y" to the confirmation prompt.
     """
 
     pbeN, pbeU = publish_batch_every  # shorthand
@@ -69,10 +72,10 @@ def publish_stream(
     print(f"\nPublishing:\n\t{Nbatches} batches\n\teach with {alerts_per_batch} alerts\n\tat a rate of 1 batch per {pbeN} {pbeU} (plus processing time)\n\tfor a total of {Nbatches*alerts_per_batch} alerts")
 
     # publish the stream
-    _do_publish_stream(instance, alerts_per_batch, Nbatches, publish_batch_every, sub_id, topic_id, nack)
+    _do_publish_stream(instance, alerts_per_batch, Nbatches, publish_batch_every, sub_id, topic_id, nack, auto_confirm)
 
 def _do_publish_stream(
-    instance, alerts_per_batch, Nbatches, publish_batch_every, sub_id=None, topic_id=None, nack=False):
+    instance, alerts_per_batch, Nbatches, publish_batch_every, sub_id=None, topic_id=None, nack=False, auto_confirm=False):
 
     # check units
     if publish_batch_every[1] != 'sec':
@@ -85,7 +88,7 @@ def _do_publish_stream(
     print(f"and\n\tPublish to topic: {topic_path}\n")
 
     # make the user confirm
-    _user_confirm()
+    _user_confirm(auto_confirm)
     print(f"\nPublishing...")
 
     b = 0
@@ -105,10 +108,11 @@ def _do_publish_stream(
         b = b+1
         time.sleep(publish_batch_every[0])
 
-def _user_confirm():
-    cont = input('Continue? [Y/n]: ') or 'Y'
-    if cont not in ['Y', 'y']:
-        sys.exit('Exiting consumer simulator.')
+def _user_confirm(auto_confirm=False):
+    if not auto_confirm:
+        cont = input('Continue? [Y/n]: ') or 'Y'
+        if cont not in ['Y', 'y']:
+            sys.exit('Exiting consumer simulator.')
 
 def _setup_subscribe(alerts_per_batch, instance=None, sub_id=None):
     if (instance is None) and (sub_id is None):

broker/broker_utils/schema_maps/README.md

@@ -0,0 +1,7 @@
+# Schema Maps
+
+The files in this directory contain mappings between the schema of an individual survey and a PGB-standardized schema that is used within the broker code.
+
+Note: This directory is __not__ packaged with the `broker_utils` module.
+In order to allow broker instances to use unique schema maps, independent of other instances, each instance uploads this directory to its [`broker_files`] Cloud Storage bucket upon setup.
+The broker code loads the schema maps from the bucket of the appropriate instance at runtime.

broker/broker_utils/schema_maps/decat.yaml

@@ -1,4 +1,5 @@
 SURVEY: decat
+TOPIC_SYNTAX: decat_yyyymmdd_2021A-0113  # replace yyyymmdd with the date
 FILTER_MAP:
   g DECam SDSS c0001 4720.0 1520.0: g
   r DECam SDSS c0002 6415.0 1480.0: r

broker/broker_utils/schema_maps/ztf.yaml

@@ -1,4 +1,5 @@
 SURVEY: ztf
+TOPIC_SYNTAX: ztf_yyyymmdd_programid1  # replace yyyymmdd with the date
 FILTER_MAP:
   1: g
   2: r

broker/broker_utils/setup.py

@@ -31,7 +31,7 @@
 
 setup(
     name='pgb_broker_utils',  # Required
-    version='0.2.7',  # Required
+    version='0.2.8',  # Required
     description='Tools used by the Pitt-Google astronomical alert broker.',
     long_description=long_description,
     long_description_content_type='text/markdown',

broker/cloud_functions/check_cue_response/README.md

@@ -0,0 +1,26 @@
+# Check cue response
+
+This Cloud Function checks whether the broker responded appropriately to the auto-scheduler's cue.
+It does this by first pausing to allow time for the response, and then checking each broker component, such as VMs and Dataflow jobs.
+If a component is found to be in an unexpected state, a "Critical" error is raised which triggers a GCP alerting policy.
+
+This Cloud Function is triggered by the auto-scheduler's Pub/Sub topic. For reference, the auto-scheduling process looks like this (see [Auto-scheduler](auto-scheduler.md)):
+
+Cloud Scheduler cron job -> Pub/Sub -> Cloud Function -> Night Conductor VM startup
+
+
+## Alerting policy
+
+An alerting policy was created manually to notify Troy Raen of anything written to the log named `check-cue-response-cloudfnc` that has severity `'CRITICAL'`.
+Every broker instance has a unique `check_cue_response` Cloud Function, but they all write to the same log.
+Therefore, a new policy does not need to be created with each new broker instance.
+(Also, recall that the auto-scheduler is typically only active in Production instances.)
+
+To update the existing policy, or create a new one, see:
+- [Managing log-based alerts](https://cloud.google.com/logging/docs/alerting/log-based-alerts)
+- [Managing alerting policies by API](https://cloud.google.com/monitoring/alerts/using-alerting-api)
+- [Managing notification channels](https://cloud.google.com/monitoring/support/notification-options)
+
+## Links
+
+- [`googleapiclient.discovery.build` Library reference documentation by API](https://github.com/googleapis/google-api-python-client/blob/master/docs/dyn/index.md)