Monitoring Interface

Starting from version 0.8.0 the WLDT library integrates a native Monitoring System that allows developers to observe the behaviour and the performance of a Digital Twin (DT) at runtime. The monitoring system is available in the package it.wldt.monitoring and it provides a pluggable, handler-based infrastructure used to collect, route and react to metrics generated by any component of the DT.

The Monitoring System has been designed with the following principles:

• Native: The core components of the library (DT Model, Physical Adapters, Digital Adapters, Augmentation Functions and Storage) automatically generate a rich set of metrics without any additional code to be written by the developer (e.g., execution times, success and error counters)
• Opt-in: No metric is collected and no callback is fired until a configuration and a handler have been attached to the DT. Each component can be individually enabled or disabled
• Decoupled from the monitoring backend: The library does not impose any specific monitoring technology. The developer receives metrics through a MonitoringInterfaceHandler and can forward them to the desired backend (e.g., Prometheus and Grafana, logs, a database, a custom dashboard)
• Extensible: Developers can define and publish their own custom metrics, for example from custom Shadowing Functions, Adapters or Augmentation Functions

Architecture and Core Classes#

The monitoring system is built around three collaborating elements: the MonitoringInterface, the MonitoringInterfaceHandler and the WldtMetricRegistry. Metrics are registered once and then updated in-place as new measurements arrive. Each metric instance accumulates type-specific supporting fields (e.g., delta, minimum and maximum values, cumulative counters) across all the updates.

Each DigitalTwin instance owns exactly one MonitoringInterface, that is automatically injected into all the DT’s components. The components generate metrics through the interface that, if monitoring is active, stores them in the internal registry and notifies the handler implemented by the developer.

MonitoringInterface#

The MonitoringInterface is the central hub of the monitoring system. It can be obtained from the DT instance through the method getMonitoringInterface() and it exposes two primary methods used to push metrics:

• notifyMetric(WldtMetric metric): The standard path. On the first push of a metric name the metric is registered and MonitoringInterfaceHandler.onMetricRegistered() is called with an empty snapshot. If the metric was created with an initial value, onMetricUpdated() is also fired immediately after. On every subsequent push for the same name the stored live instance is updated in-place and onMetricUpdated() is called
• trackCustomMetric(WldtMetric metric): Same pipeline used for developer-defined custom metrics. It bypasses the per-component flags and requires the metric component to be WldtMetricComponent.CUSTOM

The following additional methods are available:

MonitoringInterfaceConfiguration#

The MonitoringInterfaceConfiguration is an immutable class created through a builder. It defines which components of the DT are monitored. Custom metrics always bypass the flags and are always delivered to the handler.

MonitoringInterfaceHandler#

The MonitoringInterfaceHandler is the extension point for the developer. It is an abstract class with two callbacks, both with a no-op default implementation:

public abstract class MonitoringInterfaceHandler {

    /** Fires once when a metric name is observed for the first time.
     *  The metric is an empty (uninitialized) snapshot: it carries the identity fields
     *  (namespace, name, component) but no measured value. */
    public void onMetricRegistered(WldtMetricComponent component, WldtMetric metric) {}

    /** Fires on every update of an already registered metric, including the initial
     *  update if the metric was created with a starting value.
     *  The metric is an independent snapshot copy of the live instance taken at callback time. */
    public void onMetricUpdated(WldtMetricComponent component, WldtMetric metric) {}
}

The parameter component identifies the DT component that emitted the metric and can be used, together with instanceof, to access the typed fields of the metric:

public class MyHandler extends MonitoringInterfaceHandler {

    @Override
    public void onMetricUpdated(WldtMetricComponent component, WldtMetric metric) {
        if (component == WldtMetricComponent.DT_MODEL) {
            if (metric instanceof WldtCounter) {
                WldtCounter c = (WldtCounter) metric;
                if (c.isDeltaAvailable())
                    prometheusCounter.inc(c.getDelta());
            }
            if (metric instanceof WldtTimer) {
                WldtTimer t = (WldtTimer) metric;
                prometheusHistogram.observe(t.getDurationSeconds());
            }
        }
    }
}

Metric Snapshots in Handler Callbacks#

• onMetricRegistered always receives an empty snapshot: the metric exists but it does not carry any measured value yet. This callback is intended for setup operations (e.g., creating the corresponding Prometheus counter before any data arrives)
• onMetricUpdated receives an independent copy of the live metric, capturing all the fields at that instant (current value, delta, min/max, cumulative counters and lastUpdatedMs). Subsequent updates do not affect the copy held by the handler, so it is safe to store it (e.g., in a list)
• If a metric is created with an initial value, both callbacks are fired on the first push: first onMetricRegistered (empty snapshot) and then onMetricUpdated (snapshot of the initial value)

If the current live value of a metric is needed outside a callback, MonitoringInterface.getMetric(fullName) returns the live registry instance directly.

WldtMetricRegistry#

The WldtMetricRegistry is the internal store of the live metric instances (it is not used directly by developers). It keeps a mutable instance for each registered metric name in a ConcurrentHashMap and it is used by the MonitoringInterface to register new metrics and to update the existing ones.

Setup#

The following code shows how to enable the monitoring on a Digital Twin and how to attach a custom handler:

// Create the configuration enabling only the components of interest
MonitoringInterfaceConfiguration config = new MonitoringInterfaceConfiguration.Builder()
        .withDtModelMonitoring()
        .withPhysicalAdapterMonitoring()
        .withDigitalAdapterMonitoring()
        .withCustomNamespace("myapp.dt")
        .build();

// Obtain the Monitoring Interface of the Digital Twin and set configuration and handler
digitalTwin.getMonitoringInterface().setConfiguration(config);
digitalTwin.getMonitoringInterface().setHandler(new MyMonitoringHandler());

Note: The monitoring is active only if both the configuration and the handler have been set. The configuration and the handler should be set before starting the Digital Twin in order to receive the metrics generated since the very first phases of its life cycle.

Metric Types#

All the metrics extend the class WldtMetric and carry the following information:

• digitalTwinId: The id of the DigitalTwin instance that owns the metric (first parameter of all the constructors)
• namespace: Logical grouping prefix (e.g., "core" for the framework metrics or "myapp.sensor" for custom metrics)
• name: Identifier of the metric within the namespace
• component: A WldtMetricComponent value, used for routing and for the flag verification
• timestampMs: Epoch in milliseconds of the creation of the metric
• lastUpdatedMs: Epoch in milliseconds of the most recent update
• isInitialized(): false for metrics created without an initial value, true once a value has been set
• Metadata: Arbitrary key-value tags (addMetadata(key, value), removeMetadata(key), setMetadata(map), getMetadata()) useful to add context to a metric (e.g., the function id or the adapter id)

The full name used to look up a metric in the registry is namespace + "." + name.

All the constructors can be used in two flavors:

• Uninitialized (digitalTwinId, namespace, name, component): Registers the metric slot without a starting value. isInitialized() returns false until the first update
• Initialized (digitalTwinId, namespace, name, component, initialValue): Sets a starting value immediately

Both flavors have an additional overload accepting a Map<String, Object> of metadata as final parameter.

The library provides five types of metrics:

WldtCounter#

Models discrete occurrences that can only increase. It tracks the delta (last increment) and the totalIncrements across all the mutations.

• Mutation methods: update(long newAbsoluteValue) (must be greater than or equal to the current value), increment(), increment(long amount) (amount must be greater than 0)
• Supporting fields: getValue(), getDelta() (null until the first mutation), isDeltaAvailable(), getTotalIncrements()

String dtId = "my-digital-twin";
String namespace = CoreMonitoringUtils.buildCoreNamespace(); // returns "core"

// Registration (first push)
monitoringInterface.notifyMetric(
        new WldtCounter(dtId, namespace, "events_processed", WldtMetricComponent.DT_MODEL, 0L));
// -> fires onMetricRegistered(DT_MODEL, emptySnapshot)
// -> fires onMetricUpdated(DT_MODEL, snapshot{value=0}) because the metric is initialized

// Update (absolute value)
monitoringInterface.notifyMetric(
        new WldtCounter(dtId, namespace, "events_processed", WldtMetricComponent.DT_MODEL, 5L));
// -> fires onMetricUpdated(DT_MODEL, snapshot{value=5, delta=5})

// Update (increment directly on the stored instance)
WldtCounter stored = (WldtCounter) monitoringInterface
        .getMetric(namespace + ".events_processed").get();
stored.increment(3L);
// value == 8, delta == 3

WldtUpDownCounter#

Models discrete entity counts that can increase or decrease. It tracks the signed delta, the peakValue and the troughValue.

• Mutation methods: update(long newAbsoluteValue), increment(), increment(long amount), decrement(), decrement(long amount)
• Supporting fields: getValue(), getDelta(), isDeltaAvailable(), getPeakValue(), getTroughValue(), getTotalUpdates()

// Registration
monitoringInterface.notifyMetric(
        new WldtUpDownCounter(dtId, namespace, "active_connections",
                WldtMetricComponent.PHYSICAL_ADAPTER, 0L));

// Update via absolute value
monitoringInterface.notifyMetric(
        new WldtUpDownCounter(dtId, namespace, "active_connections",
                WldtMetricComponent.PHYSICAL_ADAPTER, 3L));
// delta == +3, peakValue == 3

// Decrement directly on the stored instance
WldtUpDownCounter stored = (WldtUpDownCounter) monitoringInterface
        .getMetric(namespace + ".active_connections").get();
stored.decrement(1L);
// value == 2, delta == -1, troughValue == 2

WldtGauge#

Models a continuously observed numeric value without direction constraints. It tracks the previousValue, the signed delta, the minObserved and the maxObserved values.

• Mutation methods: update(double newValue)
• Supporting fields: getValue(), getPreviousValue() (null before the first update), getDelta() (null before the first update), getMinObserved(), getMaxObserved(), getUpdateCount()

// Registration
monitoringInterface.notifyMetric(
        new WldtGauge(dtId, "myapp.sensor", "temperature", WldtMetricComponent.CUSTOM, 20.0));
// minObserved == maxObserved == 20.0, previousValue == null

// Update
monitoringInterface.notifyMetric(
        new WldtGauge(dtId, "myapp.sensor", "temperature", WldtMetricComponent.CUSTOM, 23.5));
// value == 23.5, previousValue == 20.0, delta == +3.5, maxObserved == 23.5

WldtTimer#

Records the elapsed time of operations in milliseconds. It accumulates the minDurationMs, maxDurationMs, totalDurationMs and observationCount across all the recordings.

• Mutation methods: update(long durationMs), updateSince(long startMs) (equivalent to update(System.currentTimeMillis() - startMs))
• Supporting fields: getDurationMs() (last observation), getDurationSeconds(), getMinDurationMs(), getMaxDurationMs(), getTotalDurationMs(), getObservationCount(), getMeanDurationMs()

// Registration (first measurement)
long startMs = System.currentTimeMillis();
// ... operation ...
monitoringInterface.notifyMetric(
        new WldtTimer(dtId, namespace, "processing_latency_ms",
                WldtMetricComponent.DT_MODEL, System.currentTimeMillis() - startMs));

// Update directly on the stored instance
WldtTimer timer = (WldtTimer) monitoringInterface
        .getMetric(namespace + ".processing_latency_ms").get();
timer.updateSince(startMs);

// Read the statistics
System.out.printf("Latency - last: %dms  min: %dms  max: %dms  mean: %.1fms%n",
        timer.getDurationMs(), timer.getMinDurationMs(),
        timer.getMaxDurationMs(), timer.getMeanDurationMs());

WldtHistogram#

Aggregates observations into a statistical summary. It supports two accumulation modes: single-value observations through observe(double) and pre-aggregated windows through update(count, sum, min, max). It tracks both per-window and cumulative fields.

• Mutation methods: observe(double value), update(long count, double sum, double min, double max)
• Per-window accessors: getCount(), getSum(), getMin(), getMax(), getMean()
• Cumulative accessors: getTotalCount(), getTotalSum(), getGlobalMin(), getGlobalMax(), getWindowCount(), getGlobalMean()

// Registration (first window: count, sum, min, max)
monitoringInterface.notifyMetric(
        new WldtHistogram(dtId, namespace, "message_size_bytes",
                WldtMetricComponent.PHYSICAL_ADAPTER, 10L, 1200.0, 80.0, 160.0));

// Update via single observations on the stored instance
WldtHistogram h = (WldtHistogram) monitoringInterface
        .getMetric(namespace + ".message_size_bytes").get();
h.observe(95.0);
h.observe(200.0);

// Read the global statistics
System.out.printf("Messages - total: %d  global mean: %.1f bytes%n",
        h.getTotalCount(), h.getGlobalMean());

WldtMetricComponent#

The enum WldtMetricComponent is used to route each metric to the DT component that generated it and to verify the flags of the configuration.

Updating Metrics#

Convenience Methods#

The MonitoringInterface exposes a set of high-level methods that allow the developer to update an already registered metric by specifying only the namespace, the name and the new value. The library performs internally the registry lookup, the type validation, the casting, the mutation and the handler notification. Any error (metric not found, wrong type, invalid value) is logged and silently swallowed: the metric updates never affect the normal execution of the program.

Each method also has an overload with a final instanceId parameter used for per-instance metrics (see the next paragraph).

Note: Calling decreaseCounter on a WldtCounter (monotonically increasing) logs an error and has no effect. Use a WldtUpDownCounter for metrics that can decrease.

Instance Id: Use null (or the overload without the parameter) for singleton metrics. For multi-instance components (Adapters, Augmentation Function Handlers and Functions, Storages) the id of the component is used so that each instance has its own metric slot in the registry, retrievable through getMetric(fullName, instanceId).

A complete example of registration followed by updates through the convenience methods is the following:

String dtId = "my-digital-twin";
String namespace = CoreMonitoringUtils.buildCoreNamespace();

// Register once
monitoringInterface.registerMetric(
        new WldtCounter(dtId, namespace, "events_processed", WldtMetricComponent.DT_MODEL, 0L));
monitoringInterface.registerMetric(
        new WldtTimer(dtId, namespace, "processing_latency_ms", WldtMetricComponent.DT_MODEL, 0L));

// Later - increment by 1, by a specific amount or set an absolute value
monitoringInterface.increaseCounter(namespace, "events_processed");
monitoringInterface.increaseCounter(namespace, "events_processed", 5L);
monitoringInterface.updateCounter(namespace, "events_processed", 100L);

// Let the library compute the elapsed time from a start timestamp
long startMs = System.currentTimeMillis();
// ... operation ...
monitoringInterface.updateTimerSince(namespace, "processing_latency_ms", startMs);

Fluent Mutation API#

Every mutation method of a metric returns this, allowing the calls to be chained on the same instance. This is useful when multiple updates have to be applied in sequence or when the result has to be passed immediately to notifyMetric.

String namespace = CoreMonitoringUtils.buildCoreNamespace();

// Counter - chain two increments (value += 5)
((WldtCounter) monitoringInterface.getMetric(namespace + ".events").get())
        .increment(3L)
        .increment(2L);

// UpDownCounter - increment then decrement (net change: +3)
((WldtUpDownCounter) monitoringInterface.getMetric(namespace + ".connections").get())
        .increment(5L)
        .decrement(2L);

// Histogram - record several observations without intermediate variables
((WldtHistogram) monitoringInterface.getMetric(namespace + ".msg_size").get())
        .observe(42.0)
        .observe(55.0)
        .observe(38.0);

// Update a stored metric and notify the handler in a single expression
WldtGauge temperature = (WldtGauge) monitoringInterface
        .getMetric("myapp.sensor.temperature").get();
monitoringInterface.notifyMetric(temperature.update(24.1));

Framework-Native Metrics#

When the flag of a component is enabled in the MonitoringInterfaceConfiguration, the library automatically registers and tracks a standard set of metrics. All the framework-native metrics use the namespace "core" (returned by CoreMonitoringUtils.buildCoreNamespace()), so the full name of a metric is core.<metric_name>. The class CoreMonitoringUtils exposes all the metric names as public string constants.

Most of the metrics follow the same naming convention for each monitored operation: <operation>_exec_time (WldtTimer), <operation>_exec_success_count and <operation>_exec_error_count (WldtCounter).

DT Model Metrics#

Enabled with withDtModelMonitoring(). They track the execution of the callbacks of the Shadowing Function and the calls to the Augmentation Functions.

Life Cycle Metric#

The metric dt_lifecycle_value (WldtUpDownCounter) tracks the current LifeCycleState of the Digital Twin as its numeric ordinal. It is automatically updated by the DigitalTwin on every life cycle transition and it is enabled together with the DT Model monitoring (withDtModelMonitoring()).

Physical Adapter Metrics#

Enabled with withPhysicalAdapterMonitoring().

Per-instance tracking: Each Physical Adapter instance has its own metric slot keyed by the adapter id. In the handler callbacks the source adapter can be identified through the metadata key PhysicalAdapter.METRIC_METADATA_PHYSICAL_ADAPTER_ID_KEY ("pa_id"), and getMetric(fullName, adapterId) can be used for a direct lookup.

Digital Adapter Metrics#

Enabled with withDigitalAdapterMonitoring().

Per-instance tracking: Each Digital Adapter instance has its own metric slot keyed by the adapter id. In the handler callbacks the source adapter can be identified through the metadata key DigitalAdapter.METRIC_METADATA_DIGITAL_ADAPTER_ID_KEY ("da_id"), and getMetric(fullName, adapterId) can be used for a direct lookup.

Augmentation Metrics#

Enabled with withAugmentationMonitoring().

Per-instance tracking: The handler-level metrics (af_handler_registered_*, af_handler_stateful_running_count) are tracked per handler, keyed by the handler id. Use the metadata key AugmentationFunctionHandler.METRIC_METADATA_AF_HANDLER_ID_KEY ("af_handler_id") in the handler callbacks to identify the source handler. The function-level metrics (af_function_*) are tracked per function instance, keyed by the function id, and can be identified through the key AugmentationFunction.METRIC_METADATA_AF_FUNCTION_ID_KEY ("af_function_id"). In both cases getMetric(fullName, id) can be used for a direct lookup.

Storage Metrics#

Enabled with withStorageMonitoring(). The StorageManager, the WldtStorage and the DefaultQueryManager are instrumented automatically (see the Storage Layer page).

Per-instance tracking: Each WldtStorage instance has its own metric slot keyed by the storage id. Use the metadata key WldtStorage.METRIC_METADATA_STORAGE_ID_KEY ("storage_id") in the handler callbacks to identify which storage produced the metric, and getMetric("core.storage_query_success_count", storageId) for a direct lookup.

Augmentation Function Monitoring Integration#

Augmentation Functions participate in the monitoring system through the AugmentationFunction base class, which provides built-in monitoring support (see the Augmentation Functions page for additional details on their structure).

Automatic Injection#

When a function is registered in an AugmentationFunctionHandler, the handler automatically injects the MonitoringInterface by calling setMonitoringInterface(monitoringInterface, digitalTwinId). Developers do not call this method directly. After the injection, three protected fields are available to the subclasses:

Registering Custom Function-Level Metrics#

A function can override the method handleMetricsRegistration() to register its own metrics. The method is called automatically right after the injection of the Monitoring Interface.

public class MyStatelessFunction extends StatelessAugmentationFunction {

    private static final String MY_METRIC = "my_custom_exec_time";

    public MyStatelessFunction(String id) {
        super(id, "My Function", "A function with a custom execution time metric.", "1.0.0");
    }

    @Override
    protected void handleMetricsRegistration() {
        if (monitoringInterface != null && monitoringInterface.isActive(WldtMetricComponent.AUGMENTATION)) {
            Map<String, Object> meta = new HashMap<>();
            meta.put(METRIC_METADATA_AF_FUNCTION_ID_KEY, getId());
            monitoringInterface.registerMetric(
                    new WldtTimer(digitalTwinId, metricsNamespace, MY_METRIC,
                            WldtMetricComponent.AUGMENTATION, meta));
        }
    }

    @Override
    protected AugmentationFunctionResultList run(AugmentationFunctionRequest request)
            throws AugmentationFunctionException {
        long start = System.currentTimeMillis();
        // ... function logic producing a result ...
        monitoringInterface.updateTimerSince(metricsNamespace, MY_METRIC, start);
        return new AugmentationFunctionResultList(result);
    }
}

Tagging Metrics with the Function Id#

The key AugmentationFunction.METRIC_METADATA_AF_FUNCTION_ID_KEY ("af_function_id") is a standard metadata key used to tag function-level metrics with the unique id of the function. Using it when registering custom metrics allows the handler to distinguish the metrics coming from different function instances:

@Override
public void onMetricUpdated(WldtMetricComponent component, WldtMetric metric) {
    if (component == WldtMetricComponent.AUGMENTATION) {
        String functionId = (String) metric.getMetadata()
                .get(AugmentationFunction.METRIC_METADATA_AF_FUNCTION_ID_KEY);
        // use functionId to route the metric to the right Prometheus label, etc.
    }
}

Custom Metrics#

Besides the framework-native metrics, any component (e.g., a custom Shadowing Function or Adapter) can define its own metrics through the MonitoringInterface:

• Create the metric with the component WldtMetricComponent.CUSTOM and a custom namespace (see withCustomNamespace(String))
• Register or push it through trackCustomMetric(WldtMetric) (custom metrics always bypass the per-component flags)
• Update it through the convenience methods or through the stored instance

String customNamespace = "myapp.sensor";

// Register and push a custom gauge
monitoringInterface.trackCustomMetric(
        new WldtGauge(digitalTwinId, customNamespace, "temperature", WldtMetricComponent.CUSTOM, 20.0));

// Later, update it
monitoringInterface.updateGauge(customNamespace, "temperature", 23.5);

Integration with External Monitoring Tools#

The MonitoringInterfaceHandler is the bridge between the WLDT metrics and external monitoring systems. A typical integration, tested with Prometheus and Grafana, consists of the following steps:

1. In onMetricRegistered create the corresponding Prometheus metric (counter, gauge, histogram) using the name, the component and the metadata (e.g., adapter id, handler id, function id) as labels
2. In onMetricUpdated read the typed fields of the snapshot (e.g., getDelta() for counters, getDurationSeconds() for timers) and update the Prometheus metric
3. Expose the Prometheus metrics through an HTTP endpoint and configure Prometheus to scrape it
4. Visualize the data in Grafana. The core library repository contains example dashboards in the prometheus folder, both for the global monitoring of multiple DTs and for the monitoring of a single DT

Since the metrics carry the digitalTwinId, the same handler implementation can be shared across multiple Digital Twins running in the same process while keeping their metrics separated.