Merge pull request #3768 from lsst-sqre/tickets/DM-46735

[DM-46735] Implement the repair connector functionality for the Telegraf based connectors in Sasquatch
lsst-sqre · Oct 9, 2024 · b170e0c · b170e0c
2 parents 0736432 + 4882e83
commit b170e0c
Show file tree

Hide file tree

Showing 2 changed files with 49 additions and 0 deletions.
diff --git a/applications/sasquatch/charts/telegraf-kafka-consumer/templates/_helpers.tpl b/applications/sasquatch/charts/telegraf-kafka-consumer/templates/_helpers.tpl
@@ -66,6 +66,36 @@ data:
       max_undelivered_messages = {{ default 10000 .value.max_undelivered_messages }}
       compression_codec = {{ default 3 .value.compression_codec }}
 
+    {{- if .value.repair }}
+    [[inputs.kafka_consumer]]
+      brokers = [
+        "sasquatch-kafka-brokers.sasquatch:9092"
+      ]
+      consumer_group = "telegraf-kafka-consumer-{{ .key }}-repairer"
+      sasl_mechanism = "SCRAM-SHA-512"
+      sasl_password = "$TELEGRAF_PASSWORD"
+      sasl_username = "telegraf"
+      data_format = "avro"
+      avro_schema_registry = "http://sasquatch-schema-registry.sasquatch:8081"
+      avro_timestamp = {{ default "private_efdStamp" .value.timestamp_field | quote }}
+      avro_timestamp_format = {{ default "unix" .value.timestamp_format | quote }}
+      avro_union_mode = {{ default "nullable" .value.union_mode | quote }}
+      avro_field_separator = {{ default "" .value.union_field_separator | quote }}
+      {{- if .value.fields }}
+      avro_fields = {{ .value.fields }}
+      {{- end }}
+      {{- if .value.tags }}
+      avro_tags = {{ .value.tags }}
+      {{- end }}
+      topic_regexps = {{ .value.topicRegexps }}
+      offset = "oldest"
+      precision = {{ default "1us" .value.precision | quote }}
+      max_processing_time = {{ default "5s" .value.max_processing_time | quote }}
+      consumer_fetch_default = {{ default "20MB" .value.consumer_fetch_default | quote }}
+      max_undelivered_messages = {{ default 10000 .value.max_undelivered_messages }}
+      compression_codec = {{ default 3 .value.compression_codec }}
+    {{- end }}
+
     [[inputs.internal]]
       name_prefix = "telegraf_"
       collect_memstats = true

diff --git a/applications/sasquatch/values-usdfprod.yaml b/applications/sasquatch/values-usdfprod.yaml
@@ -143,132 +143,151 @@ telegraf-kafka-consumer:
     # CSC connectors
     maintel:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.MTAOS", "lsst.sal.MTDome", "lsst.sal.MTDomeTrajectory", "lsst.sal.MTPtg" ]
       debug: true
     mtmount:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.MTMount" ]
       debug: true
     comcam:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.CCCamera", "lsst.sal.CCHeaderService", "lsst.sal.CCOODS" ]
       debug: true
     eas:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
          [ "lsst.sal.DIMM", "lsst.sal.ESS", "lsst.sal.DSM", "lsst.sal.EPM", "lsst.sal.HVAC", "lsst.sal.WeatherForecast" ]
       debug: true
     m1m3:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.MTM1M3" ]
       debug: true
     m2:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.MTHexapod", "lsst.sal.MTM2", "lsst.sal.MTRotator" ]
       debug: true
     obssys:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.Scheduler", "lsst.sal.Script", "lsst.sal.ScriptQueue", "lsst.sal.Watcher" ]
       debug: true
     ocps:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.OCPS" ]
       debug: true
     pmd:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.PMD" ]
       debug: true
     calsys:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.ATMonochromator", "lsst.sal.ATWhiteLight", "lsst.sal.CBP", "lsst.sal.Electrometer", "lsst.sal.FiberSpectrograph", "lsst.sal.LinearStage", "lsst.sal.TunableLaser" ]
       debug: true
     mtaircompressor:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.MTAirCompressor" ]
       debug: true
     genericcamera:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.GCHeaderService", "lsst.sal.GenericCamera" ]
       debug: true
     gis:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.GIS" ]
       debug: true
     mtvms:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.MTVMS" ]
       debug: true
     lsstcam:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.MTCamera", "lsst.sal.MTHeaderService", "lsst.sal.MTOODS" ]
       debug: true
     auxtel:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.ATAOS", "lsst.sal.ATDome", "lsst.sal.ATDomeTrajectory", "lsst.sal.ATHexapod", "lsst.sal.ATPneumatics", "lsst.sal.ATPtg", "lsst.sal.ATMCS" ]
       debug: true
     latiss:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.ATCamera", "lsst.sal.ATHeaderService", "lsst.sal.ATOODS", "lsst.sal.ATSpectrograph" ]
       debug: true
     test:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |
         [ "lsst.sal.Test" ]
       debug: true
     lasertracker:
       enabled: true
+      repair: false
       database: "efd"
       timestamp_field: "private_efdStamp"
       topicRegexps: |