two containers and connect to kafka

.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@v6
         with:
           context: .
+          file: 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 consumekafka
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          file: Dockerfile.consumekafka
+          push: true
+          tags: |
+            ghcr.io/lsst-dm/consumekafka:${{ steps.vars.outputs.tag }}
+          cache-from: type=gha
+          cache-to: type=gha,mode=max

Dockerfile.consumekafka

@@ -0,0 +1,23 @@
+ARG OBS_LSST_VERSION=w_2024_21
+FROM lsstsqre/centos:7-stack-lsst_distrib-${OBS_LSST_VERSION}
+USER lsst
+
+RUN source loadLSST.bash && mamba install aiokafka httpx
+RUN source loadLSST.bash && pip install kafkit aiokafka httpx
+
+WORKDIR /home/lsst/
+COPY --chown=lsst .  ./obsloctap/
+WORKDIR /home/lsst/obsloctap/
+RUN source /opt/lsst/software/stack/loadLSST.bash && pip install -e .
+
+# 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 -m obsloctap.consumekafka.py" ]

src/obsloctap/process_header.py → src/obsloctap/consumekafka.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()