MarbleSoft.FlowModule.TraceFinalizer 1.6.0

MarbleSoft.FlowModule.TraceFinalizer

TraceFinalizer closes open flow entries in an existing trace context.

MQTT integration uses MQTTnet 4.0.0.167.

It looks for the currently open flow entry matching the resolved runtime flow id and flow version, sets flowEndTimeUtc and duration_ms, and can optionally close the full trace when all flow entries are closed.

Settings

The settings are defined in TraceFinalizerSettings.cs.

1. Source

• Type: string
• Required: yes
• Validation: min length 1, max length 256
• Default: data.headers.traces.flowChain
• Purpose: input path used to read existing trace context

2. Target

• Type: string
• Required: yes
• Validation: min length 1, max length 256
• Default: data.headers.traces.flowChain
• Purpose: output path used to write trace context

3. Store trace in Key/Value Store

• Type: bool
• Default: false
• Purpose: when enabled, writes the finalized trace to the integrated SQLite Key/Value Store table jobIdToTraces

4. Key/Value Store key template

• Type: string
• Required: when Store trace in Key/Value Store is enabled
• Default: {flowName}_{jobId}
• Purpose: recommended templates use explicit runtime metadata and message-path tokens, e.g. {meta:flowName}{path:data.rpm} or {meta:moduleName}{meta:nodeId}. The legacy placeholders {flowName} and {jobId} are kept only for backward compatibility and should not be used for new configurations.
• Runtime metadata keys resolved from the runtime context / environment variables: flowId, flowName, flowVersion, flowDefinitionId, flowDeploymentId, moduleId, moduleName, moduleVersion, moduleType, nodeId, nodeName, nodeVersion, nodePlatform, traceSource, traceTarget.
• jobId is not runtime metadata; it is a message-level fallback value and should be treated as a separate legacy value when needed.
• Examples: {meta:flowName}{path:data.rpm}, {meta:moduleName}{meta:nodeId}

5. Apply condition

• Type: bool
• Default: false
• Purpose: evaluate the configured condition before finalizing the trace; if it is false, the message is forwarded unchanged

6. Condition left / operator / right

• Type: string
• Purpose: compare a value or message path against another value or path using operators such as ==, !=, >, >=, <, ⇐, contains, startsWith, endsWith

7. Close Trace

• Type: bool
• Required: no
• Default: true
• Purpose: when enabled, endTimeUtc and duration_ms are set if all flow entries are closed

8. MQTT publish settings (optional)

• Enable MQTT Publish: bool, default true
• MQTT Source Property: string path, default data, if empty the full message/trace payload is used
• MQTT Topic: string, required when MQTT is enabled
• MQTT Brokers: comma or line separated host:port list, default localhost:1883
• MQTT Protocol Version: enum V311 or V500
• Use TLS: bool, default true
• SSL Protocol: enum None, Tls12, Tls13
• Allow Untrusted Certificates: bool
• ClientId: optional string, generated automatically when empty
• CA Certificate: optional Crosser Certificate credential selector
• Client Certificate: optional Crosser Certificate credential selector
• MQTT User Credential: optional Crosser UsernamePassword credential selector
• QoS: AtMostOnce, AtLeastOnce, ExactlyOnce
• Retain: bool
• Last Will Topic/Payload/QoS/Retain: available as settings (currently retained for forward-compatible API mapping)

Flow and module metadata are resolved programmatically at runtime. The module attempts runtime values first and then environment-variable fallbacks, for example FLOW_ID, FLOW_VERSION, FLOW_NAME, FLOW_DEFINITION_ID, FLOW_DEPLOYMENT_ID, MODULE_ID.

Runtime behavior

Input lookup:

• Reads trace from configured Source path
• Falls back to legacy paths header.trace and trace when available

If no trace is found:

• Message is forwarded unchanged

If trace exists:

1. Ensures flows list exists
2. Finds the last open matching entry where:

• flowId matches resolved runtime flow id (case-insensitive)
• flowVersion matches resolved runtime flow version (case-insensitive)
• flowEndTimeUtc is empty

3. If found, sets flowEndTimeUtc and calculates duration_ms
4. Updates sequence to at least flows.Count + 1
5. If Close Trace is true and all entries are closed, sets endTimeUtc and trace duration_ms
6. Writes updated trace back to configured Target path only when changes occurred
7. If Store trace in Key/Value Store is enabled, writes the trace to SQLite table jobIdToTraces with key built from template placeholders {flowName} and {jobId}

MQTT behavior:

• MQTT client is initialized on module start when Enable MQTT Publish is true
• Broker failover is applied in configured order from MQTT Brokers
• For each processed message, the configured source payload is published to MQTT Topic with configured QoS/Retain
• On successful publish, the latest flow entry gets mqttOutTopic populated
• MQTT implementation uses MQTTnet 4.0.0.167

Minimal configuration example

Source: data.headers.traces.flowChain Target: data.headers.traces.flowChain Close Trace: true

Output structure excerpt

{
	"data": {
		"headers": {
			"traces": {
				"flowChain": {
					"id": "9b2a59dc-18b4-4acb-8381-8b3c7b915ae7",
					"startTimeUtc": "2026-06-11T11:39:28.7473848Z",
					"endTimeUtc": "2026-06-11T11:39:29.9459869Z",
					"duration_ms": 1199.21,
					"createdByModule": "0d454d32-348f-4383-b610-afd6697b1195",
					"createdByFlow": "02f68cf5-f5c4-48c3-9d3d-62229a378889",
					"name": "JobManagement",
					"flows": [
						{
							"flowId": "02f68cf5-f5c4-48c3-9d3d-62229a378889",
							"flowRunId": "de6a8eee-2c6f-4ddb-877c-2e1c6982d779",
							"flowStartTimeUtc": "2026-06-11T11:39:28.7473940Z",
							"flowEndTimeUtc": "2026-06-11T11:39:28.9459869Z",
							"duration_ms": 198.21,
							"flowVersion": "13",
							"sequence": 1
						}
					]
				}
			}
		}
	}
}

Build and pack

From repository root:

1. Build dotnet build crosser-trace-initializer.sln -c Debug

2. Pack TraceFinalizer dotnet pack MarbleSoft.FlowModule.TraceFinalizer/Module/MarbleSoft.FlowModule.TraceFinalizer.csproj