MarbleSoft.FlowModule.TraceInitializer 1.6.0

MarbleSoft.FlowModule.TraceInitializer

TraceInitializer ensures that each processed message carries a trace flow-chain context.

MQTT integration uses MQTTnet 4.0.0.167.

If a trace context is missing, the module creates one. If a trace context already exists, the module normalizes missing root fields and appends a new open flow entry only when needed.

Settings

The settings are defined in TraceInitializerSettings.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. Read/Delete trace from Key/Value Store

• Type: bool
• Default: false
• Purpose: when enabled, the initializer restores a trace from the integrated SQLite Key/Value Store and deletes the stored entry before it appends its own flow entry

4. Key/Value Store key template

• Type: string
• Required: when Read/Delete trace from 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. Flow Chain Name

• Type: string
• Required: no
• Validation: optional, max length 128
• Default: empty
• Purpose: optional name written to trace flow-chain name field

6. Only if trace exists

• Type: bool
• Default: false
• Purpose: create a trace only when an existing trace chain already contains flow entries; otherwise the message is forwarded unchanged

7. Apply condition

• Type: bool
• Default: false
• Purpose: evaluate the configured condition before creating a trace; if the condition is false, the message is passed through unchanged

8. 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
• Fix: nested message paths are resolved against dictionaries and object graphs before comparison, so conditions like payloadItem.nodeId or payload.items[0].status work reliably when deciding whether a trace should be added

9. FlowTraceName suffix

• Type: string
• Purpose: add a readable suffix to the runtime flow name for the per-flow flowTraceName value, for example "write to OPC" becomes "HEXSRP1693 write to OPC" while the base trace name remains the environment flow name

10. 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

Trace restore behavior:

• Reads trace from configured Source path (with legacy fallbacks header.trace and trace)
• If no trace is found and Read/Delete trace from Key/Value Store is enabled, it builds a key from the configured template and tries to read+delete a persisted trace from SQLite table jobIdToTraces
• If a persisted trace is restored, it is inserted into the source trace path before normal initializer processing continues

When no trace is found:

• Creates id (GUID)
• Creates startTimeUtc (UTC timestamp)
• Sets createdByModule from runtime module id
• Sets createdByFlow from runtime flow id
• Resolves the base trace name from the Flow Chain Name setting or the runtime flow name; the optional FlowTraceNameSuffix is applied only to the individual flow entry flowTraceName value
• Initializes flows list
• Appends a new open flow entry with:
  • sequence = next sequence number
  • flowRunId = GUID
  • flowId = resolved runtime flow id
  • flowVersion = resolved runtime flow version
  • flowStartTimeUtc = current UTC timestamp
  • flowEndTimeUtc = null
  • duration_ms = null

Optional gate conditions:

• CreateTraceOnlyWhenChainExists skips trace creation when no existing trace flow chain is present
• ApplyTraceCondition evaluates the configured rule before a new trace is created; when it evaluates to false, the message is forwarded unchanged

When trace already exists:

• Ensures id, startTimeUtc, createdByModule, createdByFlow, name are populated
• Ensures flows list exists
• Appends a new entry only if the last entry is not already the same open flow (same flowId + flowVersion with empty flowEndTimeUtc)
• Supports fallback read from legacy paths header.trace and trace

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 Flow Chain Name: JobManagement

Output structure excerpt

{
	"data": {
		"headers": {
			"traces": {
				"flowChain": {
					"id": "9b2a59dc-18b4-4acb-8381-8b3c7b915ae7",
					"startTimeUtc": "2026-06-11T11:39:28.7473848Z",
					"endTimeUtc": "",
					"duration_ms": null,
					"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": null,
							"duration_ms": null,
							"flowVersion": "13",
							"sequence": 1
						}
					]
				}
			}
		}
	}
}

Build and pack

From repository root:

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

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