astrobase/lcproc/awsrun.py
def runcp_consumer_loop(
        in_queue_url,
        workdir,
        lclist_pkl_s3url,
        lc_altexts=('',),
        wait_time_seconds=5,
        cache_clean_timer_seconds=3600.0,
        shutdown_check_timer_seconds=60.0,
        sqs_client=None,
        s3_client=None
):

    """This runs checkplot pickle making in a loop until interrupted.

    Consumes work task items from an input queue set up by `runcp_producer_loop`
    above. For the moment, we don't generate neighbor light curves since this
    would require a lot more S3 calls.

    Parameters
    ----------

    in_queue_url : str
        The SQS URL of the input queue to listen to for work assignment
        messages. The task orders will include the input and output S3 bucket
        names, as well as the URL of the output queue to where this function
        will report its work-complete or work-failed status.

    workdir : str
        The directory on the local machine where this worker loop will download
        the input light curves and associated period-finder results (if any),
        process them, and produce its output checkplot pickles. These will then
        be uploaded to the specified S3 output bucket and then deleted from the
        workdir when the upload is confirmed to make it safely to S3.

    lclist_pkl : str
        S3 URL of a catalog pickle generated by `lcproc.catalogs.make_lclist`
        that contains objectids and coordinates, as well as a kdtree for all of
        the objects in the current light curve collection being processed. This
        is used to look up neighbors for each object being processed.

    lc_altexts : sequence of str
        If not None, this is a sequence of alternate extensions to try for the
        input light curve file other than the one provided in the input task
        order. For example, to get anything that's an .sqlite where .sqlite.gz
        is expected, use altexts=[''] to strip the .gz.

    wait_time_seconds : int
        The amount of time to wait in the input SQS queue for an input task
        order. If this timeout expires and no task has been received, this
        function goes back to the top of the work loop.

    cache_clean_timer_seconds : float
        The amount of time in seconds to wait before periodically removing old
        files (such as finder chart FITS, external service result pickles) from
        the astrobase cache directory. These accumulate as the work items are
        processed, and take up significant space, so must be removed
        periodically.

    shutdown_check_timer_seconds : float
        The amount of time to wait before checking for a pending EC2 shutdown
        message for the instance this worker loop is operating on. If a shutdown
        is noticed, the worker loop is cancelled in preparation for instance
        shutdown.

    sqs_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its SQS operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    s3_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its S3 operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    Returns
    -------

    Nothing.

    """

    if not sqs_client:
        sqs_client = boto3.client('sqs')
    if not s3_client:
        s3_client = boto3.client('s3')

    lclist_pklf = lclist_pkl_s3url.split('/')[-1]

    if not os.path.exists(lclist_pklf):

        # get the lclist pickle from S3 to help with neighbor queries
        lclist_pklf = awsutils.s3_get_url(
            lclist_pkl_s3url,
            client=s3_client
        )

    with open(lclist_pklf,'rb') as infd:
        lclistpkl = pickle.load(infd)

    # listen to the kill and term signals and raise KeyboardInterrupt when
    # called
    signal.signal(signal.SIGINT, kill_handler)
    signal.signal(signal.SIGTERM, kill_handler)

    shutdown_last_time = time.monotonic()
    diskspace_last_time = time.monotonic()

    while True:

        curr_time = time.monotonic()

        if (curr_time - shutdown_last_time) > shutdown_check_timer_seconds:
            shutdown_check = shutdown_check_handler()
            if shutdown_check:
                LOGWARNING('instance will die soon, breaking loop')
                break
            shutdown_last_time = time.monotonic()

        if (curr_time - diskspace_last_time) > cache_clean_timer_seconds:
            cache_clean_handler()
            diskspace_last_time = time.monotonic()

        try:

            # receive a single message from the inqueue
            work = awsutils.sqs_get_item(in_queue_url,
                                         client=sqs_client,
                                         raiseonfail=True)

            # JSON deserialize the work item
            if work is not None and len(work) > 0:

                recv = work[0]

                # skip any messages that don't tell us to runcp
                # FIXME: use the MessageAttributes for setting topics instead
                action = recv['item']['action']
                if action != 'runcp':
                    continue

                target = recv['item']['target']
                args = recv['item']['args']
                kwargs = recv['item']['kwargs']
                outbucket = recv['item']['outbucket']

                if 'outqueue' in recv['item']:
                    out_queue_url = recv['item']['outqueue']
                else:
                    out_queue_url = None

                receipt = recv['receipt_handle']

                # download the target from S3 to a file in the work directory
                try:

                    lc_filename = awsutils.s3_get_url(
                        target,
                        altexts=lc_altexts,
                        client=s3_client,
                    )

                    # get the period-finder pickle if present in args
                    if len(args) > 0 and args[0] is not None:

                        pf_pickle = awsutils.s3_get_url(
                            args[0],
                            client=s3_client
                        )

                    else:

                        pf_pickle = None

                    # now runcp
                    cpfs = runcp(
                        pf_pickle,
                        workdir,
                        workdir,
                        lcfname=lc_filename,
                        lclistpkl=lclistpkl,
                        makeneighborlcs=False,
                        **kwargs
                    )

                    if cpfs and all(os.path.exists(x) for x in cpfs):

                        LOGINFO('runcp OK for LC: %s, PF: %s -> %s' %
                                (lc_filename, pf_pickle, cpfs))

                        # check if the file exists already because it's been
                        # processed somewhere else
                        resp = s3_client.list_objects_v2(
                            Bucket=outbucket,
                            MaxKeys=1,
                            Prefix=cpfs[0]
                        )
                        outbucket_list = resp.get('Contents',[])

                        if outbucket_list and len(outbucket_list) > 0:

                            LOGWARNING(
                                'not uploading runcp results for %s because '
                                'they exist in the output bucket already'
                                % target
                            )
                            awsutils.sqs_delete_item(in_queue_url, receipt)
                            continue

                        for cpf in cpfs:

                            put_url = awsutils.s3_put_file(cpf,
                                                           outbucket,
                                                           client=s3_client)

                            if put_url is not None:

                                LOGINFO('result uploaded to %s' % put_url)

                                # put the S3 URL of the output into the output
                                # queue if requested
                                if out_queue_url is not None:

                                    awsutils.sqs_put_item(
                                        out_queue_url,
                                        {'cpf':put_url,
                                         'target': target,
                                         'lc_filename':lc_filename,
                                         'lclistpkl':lclist_pklf,
                                         'kwargs':kwargs},
                                        raiseonfail=True
                                    )

                                # delete the result from the local directory
                                os.remove(cpf)

                            # if the upload fails, don't acknowledge the
                            # message. might be a temporary S3 failure, so
                            # another worker might succeed later.
                            else:
                                LOGERROR('failed to upload %s to S3' % cpf)

                        # delete the input item from the input queue to
                        # acknowledge its receipt and indicate that
                        # processing is done and successful
                        awsutils.sqs_delete_item(in_queue_url,
                                                 receipt)

                        # delete the light curve file when we're done with it
                        if ( (lc_filename is not None) and
                             (os.path.exists(lc_filename)) ):
                            os.remove(lc_filename)

                    # if runcp failed outright, don't requeue. instead, write a
                    # ('failed-checkplot-%s.pkl' % lc_filename) file to the
                    # output S3 bucket.
                    else:

                        LOGWARNING('runcp failed for LC: %s, PF: %s' %
                                   (lc_filename, pf_pickle))

                        with open('failed-checkplot-%s.pkl' %
                                  lc_filename, 'wb') as outfd:
                            pickle.dump(
                                {'in_queue_url':in_queue_url,
                                 'target':target,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs,
                                 'outbucket':outbucket,
                                 'out_queue_url':out_queue_url},
                                outfd, pickle.HIGHEST_PROTOCOL
                            )

                        put_url = awsutils.s3_put_file(
                            'failed-checkplot-%s.pkl' % lc_filename,
                            outbucket,
                            client=s3_client
                        )

                        # put the S3 URL of the output into the output
                        # queue if requested
                        if out_queue_url is not None:

                            awsutils.sqs_put_item(
                                out_queue_url,
                                {'cpf':put_url,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs},
                                raiseonfail=True
                            )

                        # delete the input item from the input queue to
                        # acknowledge its receipt and indicate that
                        # processing is done
                        awsutils.sqs_delete_item(in_queue_url,
                                                 receipt,
                                                 raiseonfail=True)

                        # delete the light curve file when we're done with it
                        if ( (lc_filename is not None) and
                             (os.path.exists(lc_filename)) ):
                            os.remove(lc_filename)


                except ClientError as e:

                    LOGWARNING('queues have disappeared. stopping worker loop')
                    break


                # if there's any other exception, put a failed response into the
                # output bucket and queue
                except Exception as e:

                    LOGEXCEPTION('could not process input from queue')

                    if 'lc_filename' in locals():

                        with open('failed-checkplot-%s.pkl' %
                                  lc_filename,'wb') as outfd:
                            pickle.dump(
                                {'in_queue_url':in_queue_url,
                                 'target':target,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs,
                                 'outbucket':outbucket,
                                 'out_queue_url':out_queue_url},
                                outfd, pickle.HIGHEST_PROTOCOL
                            )

                        put_url = awsutils.s3_put_file(
                            'failed-checkplot-%s.pkl' % lc_filename,
                            outbucket,
                            client=s3_client
                        )


                        # put the S3 URL of the output into the output
                        # queue if requested
                        if out_queue_url is not None:

                            awsutils.sqs_put_item(
                                out_queue_url,
                                {'cpf':put_url,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs},
                                raiseonfail=True
                            )

                        if ( (lc_filename is not None) and
                             (os.path.exists(lc_filename)) ):
                            os.remove(lc_filename)

                    # delete the input item from the input queue to
                    # acknowledge its receipt and indicate that
                    # processing is done
                    awsutils.sqs_delete_item(in_queue_url,
                                             receipt,
                                             raiseonfail=True)


        # a keyboard interrupt kills the loop
        except KeyboardInterrupt:

            LOGWARNING('breaking out of the processing loop.')
            break


        # if the queues disappear, then the producer loop is done and we should
        # exit
        except ClientError as e:

            LOGWARNING('queues have disappeared. stopping worker loop')
            break


        # any other exception continues the loop we'll write the output file to
        # the output S3 bucket (and any optional output queue), but add a
        # failed-* prefix to it to indicate that processing failed. FIXME: could
        # use a dead-letter queue for this instead
        except Exception as e:

            LOGEXCEPTION('could not process input from queue')

            if 'lc_filename' in locals():

                with open('failed-checkplot-%s.pkl' %
                          lc_filename,'wb') as outfd:
                    pickle.dump(
                        {'in_queue_url':in_queue_url,
                         'target':target,
                         'lclistpkl':lclist_pklf,
                         'kwargs':kwargs,
                         'outbucket':outbucket,
                         'out_queue_url':out_queue_url},
                        outfd, pickle.HIGHEST_PROTOCOL
                    )

                put_url = awsutils.s3_put_file(
                    'failed-checkplot-%s.pkl' % lc_filename,
                    outbucket,
                    client=s3_client
                )


                # put the S3 URL of the output into the output
                # queue if requested
                if out_queue_url is not None:

                    awsutils.sqs_put_item(
                        out_queue_url,
                        {'cpf':put_url,
                         'lclistpkl':lclist_pklf,
                         'kwargs':kwargs},
                        raiseonfail=True
                    )

                if ( (lc_filename is not None) and
                     (os.path.exists(lc_filename)) ):
                    os.remove(lc_filename)

            # delete the input item from the input queue to
            # acknowledge its receipt and indicate that
            # processing is done
            awsutils.sqs_delete_item(in_queue_url, receipt, raiseonfail=True)
def runcp_consumer_loop(
        in_queue_url,
        workdir,
        lclist_pkl_s3url,
        lc_altexts=('',),
        wait_time_seconds=5,
        cache_clean_timer_seconds=3600.0,
        shutdown_check_timer_seconds=60.0,
        sqs_client=None,
        s3_client=None
):

    """This runs checkplot pickle making in a loop until interrupted.

    Consumes work task items from an input queue set up by `runcp_producer_loop`
    above. For the moment, we don't generate neighbor light curves since this
    would require a lot more S3 calls.

    Parameters
    ----------

    in_queue_url : str
        The SQS URL of the input queue to listen to for work assignment
        messages. The task orders will include the input and output S3 bucket
        names, as well as the URL of the output queue to where this function
        will report its work-complete or work-failed status.

    workdir : str
        The directory on the local machine where this worker loop will download
        the input light curves and associated period-finder results (if any),
        process them, and produce its output checkplot pickles. These will then
        be uploaded to the specified S3 output bucket and then deleted from the
        workdir when the upload is confirmed to make it safely to S3.

    lclist_pkl : str
        S3 URL of a catalog pickle generated by `lcproc.catalogs.make_lclist`
        that contains objectids and coordinates, as well as a kdtree for all of
        the objects in the current light curve collection being processed. This
        is used to look up neighbors for each object being processed.

    lc_altexts : sequence of str
        If not None, this is a sequence of alternate extensions to try for the
        input light curve file other than the one provided in the input task
        order. For example, to get anything that's an .sqlite where .sqlite.gz
        is expected, use altexts=[''] to strip the .gz.

    wait_time_seconds : int
        The amount of time to wait in the input SQS queue for an input task
        order. If this timeout expires and no task has been received, this
        function goes back to the top of the work loop.

    cache_clean_timer_seconds : float
        The amount of time in seconds to wait before periodically removing old
        files (such as finder chart FITS, external service result pickles) from
        the astrobase cache directory. These accumulate as the work items are
        processed, and take up significant space, so must be removed
        periodically.

    shutdown_check_timer_seconds : float
        The amount of time to wait before checking for a pending EC2 shutdown
        message for the instance this worker loop is operating on. If a shutdown
        is noticed, the worker loop is cancelled in preparation for instance
        shutdown.

    sqs_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its SQS operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    s3_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its S3 operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    Returns
    -------

    Nothing.

    """

    if not sqs_client:
        sqs_client = boto3.client('sqs')
    if not s3_client:
        s3_client = boto3.client('s3')

    lclist_pklf = lclist_pkl_s3url.split('/')[-1]

    if not os.path.exists(lclist_pklf):

        # get the lclist pickle from S3 to help with neighbor queries
        lclist_pklf = awsutils.s3_get_url(
            lclist_pkl_s3url,
            client=s3_client
        )

    with open(lclist_pklf,'rb') as infd:
        lclistpkl = pickle.load(infd)

    # listen to the kill and term signals and raise KeyboardInterrupt when
    # called
    signal.signal(signal.SIGINT, kill_handler)
    signal.signal(signal.SIGTERM, kill_handler)

    shutdown_last_time = time.monotonic()
    diskspace_last_time = time.monotonic()

    while True:

        curr_time = time.monotonic()

        if (curr_time - shutdown_last_time) > shutdown_check_timer_seconds:
            shutdown_check = shutdown_check_handler()
            if shutdown_check:
                LOGWARNING('instance will die soon, breaking loop')
                break
            shutdown_last_time = time.monotonic()

        if (curr_time - diskspace_last_time) > cache_clean_timer_seconds:
            cache_clean_handler()
            diskspace_last_time = time.monotonic()

        try:

            # receive a single message from the inqueue
            work = awsutils.sqs_get_item(in_queue_url,
                                         client=sqs_client,
                                         raiseonfail=True)

            # JSON deserialize the work item
            if work is not None and len(work) > 0:

                recv = work[0]

                # skip any messages that don't tell us to runcp
                # FIXME: use the MessageAttributes for setting topics instead
                action = recv['item']['action']
                if action != 'runcp':
                    continue

                target = recv['item']['target']
                args = recv['item']['args']
                kwargs = recv['item']['kwargs']
                outbucket = recv['item']['outbucket']

                if 'outqueue' in recv['item']:
                    out_queue_url = recv['item']['outqueue']
                else:
                    out_queue_url = None

                receipt = recv['receipt_handle']

                # download the target from S3 to a file in the work directory
                try:

                    lc_filename = awsutils.s3_get_url(
                        target,
                        altexts=lc_altexts,
                        client=s3_client,
                    )

                    # get the period-finder pickle if present in args
                    if len(args) > 0 and args[0] is not None:

                        pf_pickle = awsutils.s3_get_url(
                            args[0],
                            client=s3_client
                        )

                    else:

                        pf_pickle = None

                    # now runcp
                    cpfs = runcp(
                        pf_pickle,
                        workdir,
                        workdir,
                        lcfname=lc_filename,
                        lclistpkl=lclistpkl,
                        makeneighborlcs=False,
                        **kwargs
                    )

                    if cpfs and all(os.path.exists(x) for x in cpfs):

                        LOGINFO('runcp OK for LC: %s, PF: %s -> %s' %
                                (lc_filename, pf_pickle, cpfs))

                        # check if the file exists already because it's been
                        # processed somewhere else
                        resp = s3_client.list_objects_v2(
                            Bucket=outbucket,
                            MaxKeys=1,
                            Prefix=cpfs[0]
                        )
                        outbucket_list = resp.get('Contents',[])

                        if outbucket_list and len(outbucket_list) > 0:

                            LOGWARNING(
                                'not uploading runcp results for %s because '
                                'they exist in the output bucket already'
                                % target
                            )
                            awsutils.sqs_delete_item(in_queue_url, receipt)
                            continue

                        for cpf in cpfs:

                            put_url = awsutils.s3_put_file(cpf,
                                                           outbucket,
                                                           client=s3_client)

                            if put_url is not None:

                                LOGINFO('result uploaded to %s' % put_url)

                                # put the S3 URL of the output into the output
                                # queue if requested
                                if out_queue_url is not None:

                                    awsutils.sqs_put_item(
                                        out_queue_url,
                                        {'cpf':put_url,
                                         'target': target,
                                         'lc_filename':lc_filename,
                                         'lclistpkl':lclist_pklf,
                                         'kwargs':kwargs},
                                        raiseonfail=True
                                    )

                                # delete the result from the local directory
                                os.remove(cpf)

                            # if the upload fails, don't acknowledge the
                            # message. might be a temporary S3 failure, so
                            # another worker might succeed later.
                            else:
                                LOGERROR('failed to upload %s to S3' % cpf)

                        # delete the input item from the input queue to
                        # acknowledge its receipt and indicate that
                        # processing is done and successful
                        awsutils.sqs_delete_item(in_queue_url,
                                                 receipt)

                        # delete the light curve file when we're done with it
                        if ( (lc_filename is not None) and
                             (os.path.exists(lc_filename)) ):
                            os.remove(lc_filename)

                    # if runcp failed outright, don't requeue. instead, write a
                    # ('failed-checkplot-%s.pkl' % lc_filename) file to the
                    # output S3 bucket.
                    else:

                        LOGWARNING('runcp failed for LC: %s, PF: %s' %
                                   (lc_filename, pf_pickle))

                        with open('failed-checkplot-%s.pkl' %
                                  lc_filename, 'wb') as outfd:
                            pickle.dump(
                                {'in_queue_url':in_queue_url,
                                 'target':target,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs,
                                 'outbucket':outbucket,
                                 'out_queue_url':out_queue_url},
                                outfd, pickle.HIGHEST_PROTOCOL
                            )

                        put_url = awsutils.s3_put_file(
                            'failed-checkplot-%s.pkl' % lc_filename,
                            outbucket,
                            client=s3_client
                        )

                        # put the S3 URL of the output into the output
                        # queue if requested
                        if out_queue_url is not None:

                            awsutils.sqs_put_item(
                                out_queue_url,
                                {'cpf':put_url,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs},
                                raiseonfail=True
                            )

                        # delete the input item from the input queue to
                        # acknowledge its receipt and indicate that
                        # processing is done
                        awsutils.sqs_delete_item(in_queue_url,
                                                 receipt,
                                                 raiseonfail=True)

                        # delete the light curve file when we're done with it
                        if ( (lc_filename is not None) and
                             (os.path.exists(lc_filename)) ):
                            os.remove(lc_filename)


                except ClientError as e:

                    LOGWARNING('queues have disappeared. stopping worker loop')
                    break


                # if there's any other exception, put a failed response into the
                # output bucket and queue
                except Exception as e:

                    LOGEXCEPTION('could not process input from queue')

                    if 'lc_filename' in locals():

                        with open('failed-checkplot-%s.pkl' %
                                  lc_filename,'wb') as outfd:
                            pickle.dump(
                                {'in_queue_url':in_queue_url,
                                 'target':target,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs,
                                 'outbucket':outbucket,
                                 'out_queue_url':out_queue_url},
                                outfd, pickle.HIGHEST_PROTOCOL
                            )

                        put_url = awsutils.s3_put_file(
                            'failed-checkplot-%s.pkl' % lc_filename,
                            outbucket,
                            client=s3_client
                        )


                        # put the S3 URL of the output into the output
                        # queue if requested
                        if out_queue_url is not None:

                            awsutils.sqs_put_item(
                                out_queue_url,
                                {'cpf':put_url,
                                 'lc_filename':lc_filename,
                                 'lclistpkl':lclist_pklf,
                                 'kwargs':kwargs},
                                raiseonfail=True
                            )

                        if ( (lc_filename is not None) and
                             (os.path.exists(lc_filename)) ):
                            os.remove(lc_filename)

                    # delete the input item from the input queue to
                    # acknowledge its receipt and indicate that
                    # processing is done
                    awsutils.sqs_delete_item(in_queue_url,
                                             receipt,
                                             raiseonfail=True)


        # a keyboard interrupt kills the loop
        except KeyboardInterrupt:

            LOGWARNING('breaking out of the processing loop.')
            break


        # if the queues disappear, then the producer loop is done and we should
        # exit
        except ClientError as e:

            LOGWARNING('queues have disappeared. stopping worker loop')
            break


        # any other exception continues the loop we'll write the output file to
        # the output S3 bucket (and any optional output queue), but add a
        # failed-* prefix to it to indicate that processing failed. FIXME: could
        # use a dead-letter queue for this instead
        except Exception as e:

            LOGEXCEPTION('could not process input from queue')

            if 'lc_filename' in locals():

                with open('failed-checkplot-%s.pkl' %
                          lc_filename,'wb') as outfd:
                    pickle.dump(
                        {'in_queue_url':in_queue_url,
                         'target':target,
                         'lclistpkl':lclist_pklf,
                         'kwargs':kwargs,
                         'outbucket':outbucket,
                         'out_queue_url':out_queue_url},
                        outfd, pickle.HIGHEST_PROTOCOL
                    )

                put_url = awsutils.s3_put_file(
                    'failed-checkplot-%s.pkl' % lc_filename,
                    outbucket,
                    client=s3_client
                )


                # put the S3 URL of the output into the output
                # queue if requested
                if out_queue_url is not None:

                    awsutils.sqs_put_item(
                        out_queue_url,
                        {'cpf':put_url,
                         'lclistpkl':lclist_pklf,
                         'kwargs':kwargs},
                        raiseonfail=True
                    )

                if ( (lc_filename is not None) and
                     (os.path.exists(lc_filename)) ):
                    os.remove(lc_filename)

            # delete the input item from the input queue to
            # acknowledge its receipt and indicate that
            # processing is done
            awsutils.sqs_delete_item(in_queue_url, receipt, raiseonfail=True)
def runcp_consumer_loop ( in_queue_url , workdir , lclist_pkl_s3url , lc_altexts = ( '' , ) , wait_time_seconds = 5 , cache_clean_timer_seconds = 3600.0 , shutdown_check_timer_seconds = 60.0 , sqs_client = None , s3_client = None ) : if not sqs_client : sqs_client = boto3 . client ( 'sqs' ) if not s3_client : s3_client = boto3 . client ( 's3' ) lclist_pklf = lclist_pkl_s3url . split ( '/' ) [ - 1 ] if not os . path . exists ( lclist_pklf ) : # get the lclist pickle from S3 to help with neighbor queries lclist_pklf = awsutils . s3_get_url ( lclist_pkl_s3url , client = s3_client ) with open ( lclist_pklf , 'rb' ) as infd : lclistpkl = pickle . load ( infd ) # listen to the kill and term signals and raise KeyboardInterrupt when # called signal . signal ( signal . SIGINT , kill_handler ) signal . signal ( signal . SIGTERM , kill_handler ) shutdown_last_time = time . monotonic ( ) diskspace_last_time = time . monotonic ( ) while True : curr_time = time . monotonic ( ) if ( curr_time - shutdown_last_time ) > shutdown_check_timer_seconds : shutdown_check = shutdown_check_handler ( ) if shutdown_check : LOGWARNING ( 'instance will die soon, breaking loop' ) break shutdown_last_time = time . monotonic ( ) if ( curr_time - diskspace_last_time ) > cache_clean_timer_seconds : cache_clean_handler ( ) diskspace_last_time = time . monotonic ( ) try : # receive a single message from the inqueue work = awsutils . sqs_get_item ( in_queue_url , client = sqs_client , raiseonfail = True ) # JSON deserialize the work item if work is not None and len ( work ) > 0 : recv = work [ 0 ] # skip any messages that don't tell us to runcp # FIXME: use the MessageAttributes for setting topics instead action = recv [ 'item' ] [ 'action' ] if action != 'runcp' : continue target = recv [ 'item' ] [ 'target' ] args = recv [ 'item' ] [ 'args' ] kwargs = recv [ 'item' ] [ 'kwargs' ] outbucket = recv [ 'item' ] [ 'outbucket' ] if 'outqueue' in recv [ 'item' ] : out_queue_url = recv [ 'item' ] [ 'outqueue' ] else : out_queue_url = None receipt = recv [ 'receipt_handle' ] # download the target from S3 to a file in the work directory try : lc_filename = awsutils . s3_get_url ( target , altexts = lc_altexts , client = s3_client , ) # get the period-finder pickle if present in args if len ( args ) > 0 and args [ 0 ] is not None : pf_pickle = awsutils . s3_get_url ( args [ 0 ] , client = s3_client ) else : pf_pickle = None # now runcp cpfs = runcp ( pf_pickle , workdir , workdir , lcfname = lc_filename , lclistpkl = lclistpkl , makeneighborlcs = False , * * kwargs ) if cpfs and all ( os . path . exists ( x ) for x in cpfs ) : LOGINFO ( 'runcp OK for LC: %s, PF: %s -> %s' % ( lc_filename , pf_pickle , cpfs ) ) # check if the file exists already because it's been # processed somewhere else resp = s3_client . list_objects_v2 ( Bucket = outbucket , MaxKeys = 1 , Prefix = cpfs [ 0 ] ) outbucket_list = resp . get ( 'Contents' , [ ] ) if outbucket_list and len ( outbucket_list ) > 0 : LOGWARNING ( 'not uploading runcp results for %s because ' 'they exist in the output bucket already' % target ) awsutils . sqs_delete_item ( in_queue_url , receipt ) continue for cpf in cpfs : put_url = awsutils . s3_put_file ( cpf , outbucket , client = s3_client ) if put_url is not None : LOGINFO ( 'result uploaded to %s' % put_url ) # put the S3 URL of the output into the output # queue if requested if out_queue_url is not None : awsutils . sqs_put_item ( out_queue_url , { 'cpf' : put_url , 'target' : target , 'lc_filename' : lc_filename , 'lclistpkl' : lclist_pklf , 'kwargs' : kwargs } , raiseonfail = True ) # delete the result from the local directory os . remove ( cpf ) # if the upload fails, don't acknowledge the # message. might be a temporary S3 failure, so # another worker might succeed later. else : LOGERROR ( 'failed to upload %s to S3' % cpf ) # delete the input item from the input queue to # acknowledge its receipt and indicate that # processing is done and successful awsutils . sqs_delete_item ( in_queue_url , receipt ) # delete the light curve file when we're done with it if ( ( lc_filename is not None ) and ( os . path . exists ( lc_filename ) ) ) : os . remove ( lc_filename ) # if runcp failed outright, don't requeue. instead, write a # ('failed-checkplot-%s.pkl' % lc_filename) file to the # output S3 bucket. else : LOGWARNING ( 'runcp failed for LC: %s, PF: %s' % ( lc_filename , pf_pickle ) ) with open ( 'failed-checkplot-%s.pkl' % lc_filename , 'wb' ) as outfd : pickle . dump ( { 'in_queue_url' : in_queue_url , 'target' : target , 'lc_filename' : lc_filename , 'lclistpkl' : lclist_pklf , 'kwargs' : kwargs , 'outbucket' : outbucket , 'out_queue_url' : out_queue_url } , outfd , pickle . HIGHEST_PROTOCOL ) put_url = awsutils . s3_put_file ( 'failed-checkplot-%s.pkl' % lc_filename , outbucket , client = s3_client ) # put the S3 URL of the output into the output # queue if requested if out_queue_url is not None : awsutils . sqs_put_item ( out_queue_url , { 'cpf' : put_url , 'lc_filename' : lc_filename , 'lclistpkl' : lclist_pklf , 'kwargs' : kwargs } , raiseonfail = True ) # delete the input item from the input queue to # acknowledge its receipt and indicate that # processing is done awsutils . sqs_delete_item ( in_queue_url , receipt , raiseonfail = True ) # delete the light curve file when we're done with it if ( ( lc_filename is not None ) and ( os . path . exists ( lc_filename ) ) ) : os . remove ( lc_filename ) except ClientError as e : LOGWARNING ( 'queues have disappeared. stopping worker loop' ) break # if there's any other exception, put a failed response into the # output bucket and queue except Exception as e : LOGEXCEPTION ( 'could not process input from queue' ) if 'lc_filename' in locals ( ) : with open ( 'failed-checkplot-%s.pkl' % lc_filename , 'wb' ) as outfd : pickle . dump ( { 'in_queue_url' : in_queue_url , 'target' : target , 'lc_filename' : lc_filename , 'lclistpkl' : lclist_pklf , 'kwargs' : kwargs , 'outbucket' : outbucket , 'out_queue_url' : out_queue_url } , outfd , pickle . HIGHEST_PROTOCOL ) put_url = awsutils . s3_put_file ( 'failed-checkplot-%s.pkl' % lc_filename , outbucket , client = s3_client ) # put the S3 URL of the output into the output # queue if requested if out_queue_url is not None : awsutils . sqs_put_item ( out_queue_url , { 'cpf' : put_url , 'lc_filename' : lc_filename , 'lclistpkl' : lclist_pklf , 'kwargs' : kwargs } , raiseonfail = True ) if ( ( lc_filename is not None ) and ( os . path . exists ( lc_filename ) ) ) : os . remove ( lc_filename ) # delete the input item from the input queue to # acknowledge its receipt and indicate that # processing is done awsutils . sqs_delete_item ( in_queue_url , receipt , raiseonfail = True ) # a keyboard interrupt kills the loop except KeyboardInterrupt : LOGWARNING ( 'breaking out of the processing loop.' ) break # if the queues disappear, then the producer loop is done and we should # exit except ClientError as e : LOGWARNING ( 'queues have disappeared. stopping worker loop' ) break # any other exception continues the loop we'll write the output file to # the output S3 bucket (and any optional output queue), but add a # failed-* prefix to it to indicate that processing failed. FIXME: could # use a dead-letter queue for this instead except Exception as e : LOGEXCEPTION ( 'could not process input from queue' ) if 'lc_filename' in locals ( ) : with open ( 'failed-checkplot-%s.pkl' % lc_filename , 'wb' ) as outfd : pickle . dump ( { 'in_queue_url' : in_queue_url , 'target' : target , 'lclistpkl' : lclist_pklf , 'kwargs' : kwargs , 'outbucket' : outbucket , 'out_queue_url' : out_queue_url } , outfd , pickle . HIGHEST_PROTOCOL ) put_url = awsutils . s3_put_file ( 'failed-checkplot-%s.pkl' % lc_filename , outbucket , client = s3_client ) # put the S3 URL of the output into the output # queue if requested if out_queue_url is not None : awsutils . sqs_put_item ( out_queue_url , { 'cpf' : put_url , 'lclistpkl' : lclist_pklf , 'kwargs' : kwargs } , raiseonfail = True ) if ( ( lc_filename is not None ) and ( os . path . exists ( lc_filename ) ) ) : os . remove ( lc_filename ) # delete the input item from the input queue to # acknowledge its receipt and indicate that # processing is done awsutils . sqs_delete_item ( in_queue_url , receipt , raiseonfail = True )
This runs checkplot pickle making in a loop until interrupted.

    Consumes work task items from an input queue set up by `runcp_producer_loop`
    above. For the moment, we don't generate neighbor light curves since this
    would require a lot more S3 calls.

    Parameters
    ----------

    in_queue_url : str
        The SQS URL of the input queue to listen to for work assignment
        messages. The task orders will include the input and output S3 bucket
        names, as well as the URL of the output queue to where this function
        will report its work-complete or work-failed status.

    workdir : str
        The directory on the local machine where this worker loop will download
        the input light curves and associated period-finder results (if any),
        process them, and produce its output checkplot pickles. These will then
        be uploaded to the specified S3 output bucket and then deleted from the
        workdir when the upload is confirmed to make it safely to S3.

    lclist_pkl : str
        S3 URL of a catalog pickle generated by `lcproc.catalogs.make_lclist`
        that contains objectids and coordinates, as well as a kdtree for all of
        the objects in the current light curve collection being processed. This
        is used to look up neighbors for each object being processed.

    lc_altexts : sequence of str
        If not None, this is a sequence of alternate extensions to try for the
        input light curve file other than the one provided in the input task
        order. For example, to get anything that's an .sqlite where .sqlite.gz
        is expected, use altexts=[''] to strip the .gz.

    wait_time_seconds : int
        The amount of time to wait in the input SQS queue for an input task
        order. If this timeout expires and no task has been received, this
        function goes back to the top of the work loop.

    cache_clean_timer_seconds : float
        The amount of time in seconds to wait before periodically removing old
        files (such as finder chart FITS, external service result pickles) from
        the astrobase cache directory. These accumulate as the work items are
        processed, and take up significant space, so must be removed
        periodically.

    shutdown_check_timer_seconds : float
        The amount of time to wait before checking for a pending EC2 shutdown
        message for the instance this worker loop is operating on. If a shutdown
        is noticed, the worker loop is cancelled in preparation for instance
        shutdown.

    sqs_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its SQS operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    s3_client : boto3.Client or None
        If None, this function will instantiate a new `boto3.Client` object to
        use in its S3 operations. Alternatively, pass in an existing
        `boto3.Client` instance to re-use it here.

    Returns
    -------

    Nothing.
This runs checkplot pickle making in a loop until interrupted .