development-workflow.md

Development Workflow Reference

Pipeline Stage Categories

Understanding stage categories helps compose valid pipelines. Stages must appear in this order:

Category	Stages	Rules
Source (1, required)	$source	Must be first. One per pipeline.
Stateless Processing	$match, $project, $addFields, $unset, $unwind, $replaceRoot, $redact	Can appear anywhere after source. No state or memory overhead.
Enrichment	$lookup, $https	I/O-bound. Use parallelism setting. Place $https after windows to batch.
Stateful/Window	$tumblingWindow, $hoppingWindow, $sessionWindow	Accumulates state in memory. Monitor memoryUsageBytes.
Validation	$validate	Schema enforcement. Use validationAction: "dlq" (not "error"). Place early to catch bad data.
Custom Code	$function	JavaScript UDFs. Requires SP30+.
Output (1+, required for deployed)	$merge, $emit	Must be last. Required for persistent processors. Sinkless = ephemeral only.

Key ordering principle: Place $match as early as possible (reduces volume for all downstream stages). Place $project after $match (reduces document size). Place $https after windows (batches API calls).

5-Phase Development Lifecycle

Phase 1: Project Setup

Goal: Workspace and connections ready.

1. Discover existing resources:

   • atlas-streams-discover → list-workspaces — see what already exists
   • If workspace exists, inspect-workspace to review config

2. Create workspace (if needed):

   • atlas-streams-build → resource: "workspace"
   • Choose region close to your data sources
   • Start with tier: "SP10" for development
   • includeSampleData: true (default) gives you sample_stream_solar for testing

3. Verify workspace:

   • atlas-streams-discover → inspect-workspace — confirm state and region

Phase 2: Connection Development

Goal: All data sources and sinks connected and verified.

1. Identify required connections:

   • Source connections (Kafka, Cluster change streams, Kinesis, Sample)
   • Sink connections (Cluster for $merge, Kafka for $emit, S3, Kinesis)
   • Enrichment connections (Https for $https, Cluster for $lookup)

2. Create each connection:

   • atlas-streams-build → resource: "connection" for each
   • Let the tool elicit missing sensitive fields (passwords, bootstrap servers)
   • See connection-configs.md for type-specific schemas

3. Verify connections:

   • atlas-streams-discover → list-connections — confirm all created
   • atlas-streams-discover → inspect-connection for each — verify state and config

Phase 3: Processor Development

Goal: Working processor with validated pipeline.

Pre-Deployment Connection Validation (MANDATORY)

BEFORE creating any processor, you MUST validate all connections referenced in your pipeline. This prevents silent failures and confusion about data destinations.

Step 1: List all connections in workspace

atlas-streams-discover → action: "list-connections", workspaceName: "<your-workspace>"

Verify all required connections exist.

Step 2: Inspect EACH connection referenced in pipeline

For EVERY connectionName in your pipeline (source, sink, enrichment), inspect it:

atlas-streams-discover → action: "inspect-connection",
                         workspaceName: "<your-workspace>",
                         resourceName: "<connection-name>"

Verify for each connection:

• Connection exists and state is READY
• Connection type matches intended usage:
  • Cluster: valid for $source (change streams), $merge, $lookup
  • Kafka: valid for $source, $emit
  • S3: valid for $emit only
  • Https: valid for $https enrichment or sink
  • Lambda: valid for $externalFunction only
• Connection name matches actual target (avoid confusion):
  • ⚠️ BAD: connection "atlascluster" → actual target "ClusterRestoreTest"
  • ✅ GOOD: connection "cluster-restore-test" → actual target "ClusterRestoreTest"
• For Cluster connections: verify the clusterName field points to the intended cluster

Step 3: Present validation summary to user

Always show the user what connections will be used:

"Before creating processor '<name>', I've verified your connections:
 - ✅ sample_stream_solar → Sample data (READY)
 - ⚠️ atlascluster → ClusterRestoreTest (READY)
      Warning: Connection name 'atlascluster' doesn't match actual cluster 'ClusterRestoreTest'
 - ✅ open-meteo-api → https://api.open-meteo.com/v1/... (READY)

Proceed with processor creation?"

Step 4: Wait for user confirmation if warnings exist

If any connection name doesn't match its target, ask the user to confirm before proceeding.

Step 5: Only then create the processor

This validation workflow prevents:

• Creating processors with non-existent connections (fails immediately)
• Writing data to unexpected clusters (e.g., "atlascluster" → "ClusterRestoreTest" instead of "AtlasCluster")
• Confusion when verifying output data later

Incremental Pipeline Development

Follow incremental pipeline development — test at each step:

Step 1: Basic connectivity

[
  {"$source": {"connectionName": "my-source"}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step1"}}}
]

Create with autoStart: true. Verify documents flow. Stop processor.

Step 2: Add filtering

[
  {"$source": {"connectionName": "my-source"}},
  {"$match": {"status": "active"}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step2"}}}
]

Modify pipeline (stop → modify-processor → start). Verify filtered output.

Step 3: Add transformations

[
  {"$source": {"connectionName": "my-source"}},
  {"$match": {"status": "active"}},
  {"$addFields": {"processed_at": "$$NOW_NOT_VALID"}},
  {"$project": {"userId": 1, "amount": 1, "processed_at": 1}},
  {"$merge": {"into": {"connectionName": "my-sink", "db": "test", "coll": "step3"}}}
]

Remember: $$NOW is NOT valid in streaming. Use a field from the source document or omit.

Step 4: Add windowing or enrichment (if needed)

Step 5: Add error handling

• Configure DLQ: {"dlq": {"connectionName": "my-sink", "db": "streams_dlq", "coll": "failed_docs"}}
• Add $ifNull for optional enrichment fields
• Set onError: "dlq" on $https stages

Phase 4: Testing & Validation

Goal: Processor verified working correctly.

1. Confirm processor state:

   • atlas-streams-discover → inspect-processor — state should be STARTED

2. Run diagnostics:

   • atlas-streams-discover → diagnose-processor — full health report

3. Verify data flow:

   • Use MongoDB count tool on output collection — documents arriving?
   • Use MongoDB find tool on output collection — data looks correct?
   • Use MongoDB count tool on DLQ collection — any errors?
   • If DLQ has documents, use MongoDB find tool to inspect failure reasons

4. Classify output volume:

   • See output-diagnostics.md for the full decision framework
   • Alert processors: low output is expected
   • Transformation processors: low output is a red flag

Phase 5: Production Deployment

Goal: Processor running at appropriate tier with monitoring.

1. Right-size the tier:

   • See sizing-and-parallelism.md for tier selection
   • Review memoryUsageBytes from diagnostics
   • Consider parallelism needs for $merge, $lookup, $https
   • Upgrade tier: atlas-streams-manage → stop-processor, then start-processor with tier override

2. Ensure DLQ is configured (mandatory for production)

3. Use descriptive processor names (e.g., fraud-detector, order-enricher, iot-rollup)

Debugging Decision Trees

Connection Failures

1. atlas-streams-discover → inspect-connection — check state
2. If Kafka: verify bootstrapServers is a comma-separated string (not array)
3. If Cluster: verify cluster exists in project (atlas-list-clusters)
4. If AWS (S3/Kinesis/Lambda): verify IAM role ARN is registered in Cloud Provider Access
5. If Https: verify URL is reachable and auth headers are in connection config

Processor Startup Failures

1. atlas-streams-discover → diagnose-processor — check state and errors
2. If FAILED: read the error message in diagnostics
3. Common causes:
   • Invalid pipeline syntax (missing $source, missing sink)
   • $$NOW/$$ROOT/$$CURRENT used (not valid in streaming)
   • Kafka $source missing topic field
   • Referenced connection doesn't exist — validate with list-connections first
   • Connection name doesn't match expected target — inspect connection to verify actual cluster/resource
   • OOM — tier too small for pipeline complexity

Processing Errors (Running but DLQ filling up)

1. Use MongoDB find tool on DLQ collection — inspect error messages
2. Common causes:
   • Schema mismatches in source data
   • $https enrichment failures (API down, auth expired)
   • Type errors in $addFields or $project expressions
3. Fix: stop-processor → modify-processor (fix pipeline) → start-processor

Performance Issues (Running but slow)

1. atlas-streams-discover → diagnose-processor — check stats
2. Check memoryUsageBytes — if near 80% of tier RAM, upgrade tier
3. Check if $match is early in pipeline (reduces downstream volume)
4. Check if $https has parallelism setting (increase for I/O-bound enrichment)
5. Check if windows have partitionIdleTimeout (idle Kafka partitions block windows)
6. Consider upgrading tier or increasing stage parallelism

Operational Monitoring Cadence

Daily

• Check processor states via atlas-streams-discover → list-processors
• Verify DLQ collections aren't growing via MongoDB count tool
• Confirm output collections are receiving data

Weekly

• Run diagnose-processor for each production processor
• Review memoryUsageBytes trends — approaching 80%?
• Check connection health across all connections

Monthly

• Evaluate tier appropriateness — over-provisioned or under-provisioned?
• Review DLQ patterns — recurring errors that need pipeline fixes?
• Consider parallelism adjustments based on throughput trends

Troubleshooting

Symptom	Likely cause	Action
Processor FAILED on start	Invalid pipeline syntax, missing connection, $$NOW used	diagnose-processor → read error → fix pipeline
DLQ filling up	Schema mismatch, $https failures, type errors	find on DLQ → fix pipeline or connection
Zero output (transformation)	Connection issue, wrong topic, filter too strict	Check source health → verify connections → check $match
Zero output (alert)	Probably normal — no anomalies detected	Verify with known test event
Windows not closing	Idle Kafka partitions	Add partitionIdleTimeout to $source (e.g., {"size": 30, "unit": "second"})
OOM / processor crash	Tier too small for window state	diagnose-processor → check memoryUsageBytes → upgrade tier
Slow throughput	Low parallelism on I/O stages	Increase parallelism on $merge/$lookup/$https
404 on workspace	Doesn't exist or misspelled	discover → list-workspaces
409 on create	Name already exists	Inspect existing resource or pick new name
402 error on start	No billing configured	Do NOT retry. Add payment method in Atlas → Billing. Use sp.process() in mongosh as free alternative
"processor must be stopped"	Tried to modify running processor	manage → stop-processor first
bootstrapServers format	Passed as array instead of string	Use comma-separated string: "broker1:9092,broker2:9092"
"must choose at least one role"	Cluster connection without dbRoleToExecute	Defaults to readWriteAnyDatabase — or specify custom role
"No cluster named X"	Cluster doesn't exist in project	atlas-list-clusters to verify
IAM role ARN not found	ARN not registered in project	Register via Atlas → Cloud Provider Access
dataProcessRegion format	Wrong region format	See region table above. If unsure, inspect an existing workspace
Processor PROVISIONING for minutes	Restart cycle with exponential backoff	Wait for FAILED state, or stop → restart. Check logs for repeated error
Parallelism exceeded	Tier too small for requested parallelism	Start with higher tier (see sizing-and-parallelism.md)
Networking change needed	Networking is immutable after creation	Delete connection and recreate with new networking config
401 / 403 on API call	Invalid or expired Atlas API credentials	Verify apiClientId/apiClientSecret and project-level permissions
429 rate limit	Too many API calls	Wait and retry; avoid tight loops of discover calls

Pre-Deploy Quality Checklist

Before creating a processor, verify:

Connection Validation (MANDATORY - Always do this first)

• CRITICAL: Call atlas-streams-discover → action: "list-connections" to list all connections in workspace
• CRITICAL: Call atlas-streams-discover → action: "inspect-connection" for EACH connection referenced in pipeline
• CRITICAL: Verify connection names clearly indicate their actual targets (avoid generic names like "atlascluster" pointing to "ClusterRestoreTest")
• CRITICAL: Present connection summary to user: "Connection 'X' → Actual target 'Y'" for each connection
• CRITICAL: Warn user if connection names don't match their targets and ask for confirmation
• All connections are in READY state
• Connection types match usage (Cluster for $source/$merge, Kafka for topics, etc.)

Pipeline Validation

• search-knowledge was called to validate sink/source field names
• Pipeline starts with $source and ends with $merge, $emit, $https, or $externalFunction (async)
• No $$NOW, $$ROOT, or $$CURRENT in the pipeline
• Kafka $source includes a topic field
• Kafka $source with windowed pipeline includes partitionIdleTimeout (prevents windows from stalling on idle partitions)
• HTTPS connections are only used in $https enrichment or sink stages, not in $source
• DLQ is configured (recommended for production)
• $https stages use onError: "dlq" (not "fail")
• $externalFunction stages use onError: "dlq" and execution is explicitly set
• API auth is stored in connection settings, not hardcoded in the pipeline

Post-Deploy Verification Workflow

After creating and starting a processor:

1. atlas-streams-discover → action: "inspect-processor" — confirm state is STARTED
2. atlas-streams-discover → action: "diagnose-processor" — check for errors in the health report
3. Use MongoDB count tool on the DLQ collection — verify no errors accumulating
4. Use MongoDB find tool on the output collection — verify documents are arriving
5. If output is low/zero, classify processor type before assuming a problem (see Debug section)