Merge pull request #25 from linuxfoundation/feat/replace-serialization-libs-with-msgspec

emsearcy · web-flow · commit d270e3c32b14 · 2026-01-06T11:33:41.000-08:00
Replace serialization libraries with msgspec for better Decimal handling
diff --git a/manifests/meltano-el-postgres-cronjob.yaml b/manifests/meltano-el-postgres-cronjob.yaml
@@ -0,0 +1,120 @@
+# Copyright The Linux Foundation and each contributor to LFX.
+# SPDX-License-Identifier: MIT
+---
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: meltano-el-postgres-cronjob
+  namespace: v1-sync-helper
+  labels:
+    app.kubernetes.io/name: lfx-v1-sync-helper
+    app.kubernetes.io/component: meltano
+spec:
+  schedule: "*/5 * * * *" # Every 5 minutes
+  concurrencyPolicy: Forbid # Prevent concurrent execution
+  successfulJobsHistoryLimit: 3
+  failedJobsHistoryLimit: 3
+  jobTemplate:
+    metadata:
+      labels:
+        app.kubernetes.io/name: lfx-v1-sync-helper
+        app.kubernetes.io/component: meltano
+    spec:
+      ttlSecondsAfterFinished: 3600 # Clean up job after 1 hour
+      backoffLimit: 1
+      template:
+        metadata:
+          labels:
+            app.kubernetes.io/name: lfx-v1-sync-helper
+            app.kubernetes.io/component: meltano
+        spec:
+          restartPolicy: Never
+          containers:
+            - name: meltano
+              image: ghcr.io/linuxfoundation/lfx-v1-sync-helper/meltano:v0.4.2
+              imagePullPolicy: Always
+              workingDir: /app/meltano
+              args:
+                - el
+                # Catalog from mounted ConfigMap
+                - "--catalog"
+                - "/catalogs/tap-postgres/catalog.json"
+                # State ID for this el job
+                - "--state-id"
+                - platform-db-incremental
+                # Source
+                - tap-postgres
+                # Target
+                - target-nats-kv
+              env:
+                # Environment (prod/staging/dev per meltano.yml "environments" list)
+                - name: MELTANO_ENVIRONMENT
+                  value: "prod"
+                # S3 state backend (prod/staging/dev suffix from OpenTofu workspace names)
+                - name: MELTANO_STATE_BACKEND_URI
+                  value: "s3://lfx-v2-meltano-state-prod"
+                # AWS configuration for DynamoDB access
+                - name: AWS_DEFAULT_REGION
+                  value: "us-west-2"
+                - name: AWS_REGION
+                  value: "us-west-2"
+                # Postgres configuration
+                - name: TAP_POSTGRES_HOST
+                  valueFrom:
+                    secretKeyRef:
+                      name: postgres-credentials-ad-hoc
+                      key: host
+                      optional: true
+                - name: TAP_POSTGRES_PORT
+                  value: "5432"
+                - name: TAP_POSTGRES_DATABASE
+                  value: "sfdc"
+                - name: TAP_POSTGRES_USER
+                  valueFrom:
+                    secretKeyRef:
+                      name: postgres-credentials-ad-hoc
+                      key: username
+                      optional: true
+                - name: TAP_POSTGRES_PASSWORD
+                  valueFrom:
+                    secretKeyRef:
+                      name: postgres-credentials-ad-hoc
+                      key: password
+                      optional: true
+                # Target NATS KV configuration
+                - name: TARGET_NATS_KV_URL
+                  value: "nats://lfx-platform-nats.lfx.svc.cluster.local:4222"
+                - name: TARGET_NATS_KV_BUCKET
+                  value: "v1-objects"
+                - name: TARGET_NATS_KV_REFRESH_MODE
+                  value: "newer"
+                - name: TARGET_NATS_KV_VALIDATE_RECORDS
+                  # (true for PostgreSQL; false for DynamoDB)
+                  value: "true"
+                - name: TARGET_NATS_KV_MSGPACK
+                  value: "true"
+              resources:
+                requests:
+                  memory: "512Mi"
+                  cpu: "250m"
+                limits:
+                  memory: "2Gi"
+                  cpu: "1000m"
+              # Health check - Meltano jobs don't typically have health endpoints
+              # but we can check if the process is running
+              livenessProbe:
+                exec:
+                  command: ["pgrep", "-f", "meltano"]
+                initialDelaySeconds: 30
+                periodSeconds: 60
+                failureThreshold: 3
+              volumeMounts:
+                - name: tap-postgres-catalog
+                  mountPath: /catalogs/tap-postgres
+                  readOnly: true
+          volumes:
+            - name: tap-postgres-catalog
+              configMap:
+                name: tap-postgres-catalog
+          # Use service account with appropriate permissions for AWS
+          serviceAccountName: v1-sync-helper-sa
diff --git a/manifests/meltano-el-postgres.yaml b/manifests/meltano-el-postgres.yaml
@@ -21,7 +21,7 @@ spec:
       restartPolicy: Never
       containers:
         - name: meltano
-          image: ghcr.io/linuxfoundation/lfx-v1-sync-helper/meltano:v0.4.0
+          image: ghcr.io/linuxfoundation/lfx-v1-sync-helper/meltano:v0.4.2
           imagePullPolicy: Always
           workingDir: /app/meltano
           args:
diff --git a/meltano/load/target-nats-kv/README.md b/meltano/load/target-nats-kv/README.md
diff --git a/meltano/load/target-nats-kv/pyproject.toml b/meltano/load/target-nats-kv/pyproject.toml
@@ -9,8 +9,9 @@ requires-python = ">=3.11"
 dependencies = [
     "adjust-precision-for-schema>=0.3.4",
     "jsonschema>=2.6.0",
-    "msgpack>=1.1.0",
+    "msgspec>=0.20.0",
     "nats-py>=2.10.0",
+    "simplejson>=3.17.0",
     "singer-python>=6.1.1",
 ]
 
@@ -38,7 +39,3 @@ follow_untyped_imports = true
 [[tool.mypy.overrides]]
 module = ["adjust_precision_for_schema.*"]
 follow_untyped_imports = true
-
-[[tool.mypy.overrides]]
-module = ["msgpack.*"]
-follow_untyped_imports = true
diff --git a/meltano/load/target-nats-kv/src/target_nats_kv/__init__.py b/meltano/load/target-nats-kv/src/target_nats_kv/__init__.py
@@ -12,12 +12,11 @@
 from pathlib import Path
 
 import jsonschema
-import msgpack
+import msgspec
 import nats
-import simplejson as json
+import simplejson
 import singer
 from adjust_precision_for_schema import adjust_decimal_precision_for_schema
-from jsonschema import Draft4Validator
 from nats.js.kv import KeyValue
 from singer.messages import (
     ActivateVersionMessage,
@@ -33,7 +32,8 @@ def emit_state(state: dict | None) -> None:
     """Emit the state to stdout in JSON format."""
     if state is None:
         return
-    line = json.dumps(state)
+    state_encoder = msgspec.json.Encoder(decimal_format="number")
+    line = state_encoder.encode(state).decode("utf-8")
     logger.debug("Emitting state %s", line)
     sys.stdout.write(f"{line}\n")
     sys.stdout.flush()
@@ -71,12 +71,16 @@ async def persist_messages(
     schemas: dict[str, dict] = {}
     key_properties: dict[str, list[str]] = {}
     bookmarks: dict[str, (list[str] | None)] = {}
-    validators: dict[str, Draft4Validator] = {}
+    validators: dict[str, jsonschema.Draft4Validator] = {}
+    json_encoder = msgspec.json.Encoder(decimal_format="number")
+    msgpack_encoder = msgspec.msgpack.Encoder(decimal_format="number")
+    json_decoder = msgspec.json.Decoder()
+    msgpack_decoder = msgspec.msgpack.Decoder()
 
     for message in next_singer_message():
         try:
             o = singer.parse_message(message)
-        except json.JSONDecodeError:
+        except simplejson.JSONDecodeError:
             logger.error("Unable to parse: %s", repr(message))
             raise
 
@@ -184,11 +188,10 @@ async def persist_messages(
                     # we don't know what format was used when the data was
                     # originally stored.
                     try:
-                        current_record = msgpack.unpackb(current.value, raw=False)
-                    except (msgpack.exceptions.ExtraData, ValueError):
+                        current_record = msgpack_decoder.decode(current.value)
+                    except (msgspec.DecodeError, ValueError):
                         # Fallback to JSON if msgpack fails.
-                        current_value = current.value.decode("utf-8")
-                        current_record = json.loads(current_value)
+                        current_record = json_decoder.decode(current.value)
 
                     # Check if the record was deleted.
                     if "_sdc_deleted_at" in current_record:
@@ -246,9 +249,9 @@ async def persist_messages(
                     if should_update:
                         # Update with revision
                         if use_msgpack:
-                            value = msgpack.packb(o.record)
+                            value = msgpack_encoder.encode(o.record)
                         else:
-                            value = json.dumps(o.record).encode("utf-8")
+                            value = json_encoder.encode(o.record)
                         await kv_client.update(
                             key=key,
                             value=value,
@@ -269,9 +272,9 @@ async def persist_messages(
                 except nats.js.errors.KeyNotFoundError:
                     # Key doesn't exist, create it.
                     if use_msgpack:
-                        value = msgpack.packb(o.record)
+                        value = msgpack_encoder.encode(o.record)
                     else:
-                        value = json.dumps(o.record).encode("utf-8")
+                        value = json_encoder.encode(o.record)
                     await kv_client.create(
                         key=key,
                         value=value,
@@ -280,9 +283,9 @@ async def persist_messages(
                 # User has requested "full" sync, so use "put" without
                 # data checks.
                 if use_msgpack:
-                    value = msgpack.packb(o.record)
+                    value = msgpack_encoder.encode(o.record)
                 else:
-                    value = json.dumps(o.record).encode("utf-8")
+                    value = json_encoder.encode(o.record)
                 await kv_client.put(
                     key=key,
                     value=value,
@@ -296,8 +299,8 @@ async def persist_messages(
             stream = o.stream
             schemas[stream] = o.schema
             adjust_decimal_precision_for_schema(schemas[stream])
+            validators[stream] = jsonschema.Draft4Validator(o.schema)
             bookmarks[stream] = o.bookmark_properties
-            validators[stream] = Draft4Validator(o.schema)
             key_properties[stream] = o.key_properties
         elif isinstance(o, ActivateVersionMessage):
             logger.warning(
@@ -358,8 +361,9 @@ def main() -> None:
     args = parser.parse_args()
 
     if args.config:
-        with open(args.config) as input_json:
-            config = json.load(input_json)
+        config_decoder = msgspec.json.Decoder()
+        with open(args.config, "rb") as input_json:
+            config = config_decoder.decode(input_json.read())
     else:
         config = {}
 
diff --git a/meltano/load/target-nats-kv/uv.lock b/meltano/load/target-nats-kv/uv.lock
