feat: Redoer middleware (#25)

Author: spompea-cfa

.trivyignore.yaml

@@ -3,5 +3,6 @@ misconfigurations:
   - id: AVD-DS-0026
     paths:
       - Dockerfile.consumer # Jira -67
+      - Dockerfile.redoer # Jira -67
       - Dockerfile.exporter # ephemeral container, healthcheck not necessary
       - Dockerfile.tools

Dockerfile.redoer

@@ -0,0 +1,22 @@
+FROM senzing/senzingsdk-runtime:4.0.0
+
+USER root
+
+RUN apt-get update \
+ && apt-get -y install --no-install-recommends python3 python3-pip python3-boto3 \
+ && apt-get -y autoremove \
+ && apt-get -y clean
+
+WORKDIR /app
+COPY middleware/* .
+
+# Add a new user and switch to it.
+RUN useradd -m -u 1001 senzing
+USER senzing
+
+ENV PYTHONPATH=/opt/senzing/er/sdk/python:/app
+
+# Flush buffer - helps with print statements.
+ENV PYTHONUNBUFFERED=1
+
+CMD ["python3", "redoer.py"]

README.md

@@ -186,6 +186,16 @@ simultaneously as needed):
 
 `LOG_LEVEL` is optional; defaults to `INFO`.
 
+### Redoer
+
+Similar to the consumer, the redoer is also a continually-running process.
+
+    ```bash
+    docker compose run --env AWS_PROFILE=localstack --env LOG_LEVEL=debug redoer
+    ```
+
+`LOG_LEVEL` is optional; defaults to `INFO`.
+
 ### Exporter
 
 Spinning up the exporter middleware (this is intended to be an ephemeral

docker-compose.yaml

@@ -108,6 +108,30 @@ services:
       # Note: `.aws` mount might not be needed later.
       - ~/.aws:/home/senzing/.aws
 
+  redoer:
+    build:
+      context: .
+      dockerfile: Dockerfile.redoer
+    depends_on:
+      - db
+    environment:
+      AWS_ENDPOINT_URL: http://localstack:4566
+      SENZING_ENGINE_CONFIGURATION_JSON: >-
+        {
+          "PIPELINE": {
+            "CONFIGPATH": "/etc/opt/senzing",
+            "LICENSESTRINGBASE64": "${SENZING_LICENSE_BASE64_ENCODED}",
+            "RESOURCEPATH": "/opt/senzing/er/resources",
+            "SUPPORTPATH": "/opt/senzing/data"
+          },
+          "SQL": {
+            "BACKEND": "SQL",
+            "CONNECTION": "postgresql://${POSTGRES_USERNAME:-senzing}:${POSTGRES_PASSWORD:-senzing}@db:5432:${POSTGRES_DB:-G2}/?sslmode=disable"
+          }
+        }
+    volumes:
+      - ~/.aws:/home/senzing/.aws
+
   exporter:
     build:
       context: .

middleware/consumer.py

@@ -17,8 +17,6 @@
     log.error(e)
     sys.exit(1)
 
-# TODO add DLQ logic (see DLG_TAG logging)
-
 Q_URL = os.environ['Q_URL']
 SZ_CONFIG = json.loads(os.environ['SENZING_ENGINE_CONFIGURATION_JSON'])
 

middleware/redoer.py

@@ -0,0 +1,125 @@
+import json
+import os
+import time
+import sys
+import boto3
+import senzing as sz
+
+from loglib import *
+log = retrieve_logger()
+
+try:
+    log.info('Importing senzing_core library . . .')
+    import senzing_core as sz_core
+    log.info('Imported senzing_core successfully.')
+except Exception as e:
+    log.error('Importing senzing_core library failed.')
+    log.error(e)
+    sys.exit(1)
+
+SZ_CONFIG = json.loads(os.environ['SENZING_ENGINE_CONFIGURATION_JSON'])
+
+# How long to wait before attempting next Senzing op.
+WAIT_SECONDS = int(os.environ.get('WAIT_SECONDS', 10))
+
+#-------------------------------------------------------------------------------
+
+def go():
+    '''Starts the Redoer process; runs indefinitely.'''
+
+    sz_eng = None
+    try:
+        sz_factory = sz_core.SzAbstractFactoryCore("ERS", SZ_CONFIG)
+
+        # Init senzing engine object.
+        # Senzing engine object cannot be passed around between functions,
+        # else it will be eagerly cleaned up / destroyed and no longer usable.
+        sz_eng = sz_factory.create_engine()
+        log.info(SZ_TAG + 'Senzing engine object instantiated.')
+    except sz.SzError as sz_err:
+        log.error(SZ_TAG + str(sz_err))
+        sys.exit(1)
+    except Exception as e:
+        log.error(str(e))
+        sys.exit(1)
+
+    log.info('Starting primary loop.')
+
+    # Approach:
+    # - We don't try to both 'get' and 'process' in a single loop; instead we
+    #   'get', then 'continue' to the next loop; the have_rcd flag is used to
+    #   facilitate this.
+    # - Each Senzing call (3 distinct calls) is couched in its own try-except block for
+    #   robustness.
+    tally = None
+    have_rcd = 0
+    rcd = None
+    while 1:
+        try:
+
+            if have_rcd:
+                try:
+                    sz_eng.process_redo_record(rcd)
+                    rcd = None      # <-- TODO this op might not be necessary
+                    have_rcd = 0
+                    log.debug(SZ_TAG + 'Successfully redid one record.')
+                    continue
+                except sz.SzRetryableError as sz_ret_err:
+                    # We'll try to process this record again.
+                    log.error(SZ_TAG + str(sz_ret_err))
+                    time.sleep(WAIT_SECONDS)
+                    continue
+                except sz.SzError as sz_err:
+                    log.error(SZ_TAG + str(sz_err))
+                    sys.exit(1)
+
+            else:
+                try:
+                    tally = sz_eng.count_redo_records()
+                    log.debug(SZ_TAG + 'Current redo count: ' + str(tally))
+                except sz.SzRetryableError as sz_ret_err:
+                    log.error(SZ_TAG + str(sz_ret_err))
+                    time.sleep(WAIT_SECONDS)
+                    continue
+                except sz.SzError as sz_err:
+                    log.error(SZ_TAG + str(sz_err))
+                    sys.exit(1)
+
+                if tally:
+
+                    try:
+                        rcd = sz_eng.get_redo_record()
+                        if rcd:
+                            have_rcd = 1
+                            # At this point, rcd var holds a record, and have_rcd flag
+                            # raised. Will process in the next loop.
+                            log.debug(SZ_TAG + 'Retrieved 1 record via get_redo_record()')
+                        else:
+                            log.debug(SZ_TAG + 'Redo count was greater than 0, but got '
+                                      + 'nothing from get_redo_record')
+                    except sz.SzRetryableError as sz_ret_err:
+                        log.error(SZ_TAG + str(sz_ret_err))
+                    except sz.SzError as sz_err:
+                        log.error(SZ_TAG + str(sz_err))
+                        sys.exit(1)
+
+                else:
+                    log.debug('No redo records. Will wait ' + str(WAIT_SECONDS) + ' seconds.')
+                    time.sleep(WAIT_SECONDS)
+
+        except Exception as e:
+            log.error(str(e))
+            sys.exit(1)
+
+#-------------------------------------------------------------------------------
+
+def main():
+    log.info('====================')
+    log.info(' ------->')
+    log.info('   REDOER    ')
+    log.info('       STARTED')
+    log.info('          ------->')
+    log.info('====================')
+    go()
+
+if __name__ == '__main__': main()