two containers

.github/workflows/ci.yaml

@@ -89,12 +89,24 @@ jobs:
           username: ${{ github.repository_owner }}
           password: ${{ secrets.GITHUB_TOKEN }}
 
-      - name: Build and push
+      - name: Build and push obsloctap
         uses: docker/build-push-action@v5
         with:
           context: .
+          dockerfile: Dockerfile.obsloctap
           push: true
           tags: |
             ghcr.io/lsst-dm/obsloctap:${{ steps.vars.outputs.tag }}
           cache-from: type=gha
           cache-to: type=gha,mode=max
+
+      - name: Build and push consume-kafka
+        uses: docker/build-push-action@v5
+        with:
+          context: .
+          dockerfile: Dockerfile.consume-kafka
+          push: true
+          tags: |
+            ghcr.io/lsst-dm/consume-kafka:${{ steps.vars.outputs.tag }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max

Dockerfile.consume-kafka

@@ -0,0 +1,21 @@
+ARG RUBINENV_VERSION=8.0.0
+FROM lsstsqre/newinstall:${RUBINENV_VERSION}
+ARG OBS_LSST_VERSION
+ENV OBS_LSST_VERSION=${OBS_LSST_VERSION:-w_2024_06}
+USER lsst
+RUN source loadLSST.bash && mamba install aiokafka httpx
+RUN source loadLSST.bash && pip install kafkit
+RUN source loadLSST.bash && eups distrib install -t "${OBS_LSST_VERSION}" obs_lsst
+COPY python/lsst/obsloctap/consume-kafka.py ./obsloctap/
+
+# Environment variables that must be set:
+#   POSTGRES_URL: SQLAlchemy connection URL
+#   KAFKA_BOOTSTRAP: host:port of bootstrap server
+#   KAFKA_PASSWORD: password for SASL_PLAIN authentication
+#   SCHEMA_URL: Kafkit registry schema URL
+# Optional environment variables:
+#   BUCKET_PREFIX: set to "rubin:" at USDF, default is ""
+#   KAFKA_GROUP_ID: name of consumer group, default is "obsloctap-consumer"
+#   KAFKA_USERNAME: username for SASL_PLAIN authentication, default is "obsloctap"
+
+ENTRYPOINT [ "bash", "-c", "source loadLSST.bash; setup obs_lsst; python ./obsloctap/consume-kafka.py" ]

src/obsloctap/process_header.py → src/obsloctap/consume-kafka.py

@@ -34,7 +34,8 @@
 tenant = os.environ.get("BUCKET_TENANT", None)
 kafka_group_id = 1
 
-topic = "lsst.ATHeaderService.logevent_largeFileObjectAvailable"
+# TODO this needs to be LSSTCam but that doe snot exist yet
+topic = "lsst.MTHeaderService.logevent_largeFileObjectAvailable"
 dbhelp = DbHelpProvider().getHelper()