Edge Xpert proprietary built-in functions

As Edge Xpert Application Service SDK is implemented on the top of EdgeX Foundry Application Functions SDK, Edge Xpert Application Service SDK supports all built-in functions provided by EdgeX Foundry. Moreover, Edge Xpert provides proprietary implementation to offer additional built-in functions that users can specify in their functions pipeline. This document will go through the usage of each proprietary built-in function.

Info

All Edge Xpert proprietary built-in functions can be found in the app-functions-sdk-go-private/pkg/transforms/xpert package.

Export

AWS IoT Core

Factory Method Description

NewAWSIoTCoreMQTTSender(awsIoTCoreConfig AWSIoTCoreMQTTConfig, persistOnError, usingSharedClient bool) This function returns an AWSIoTCoreMQTTSender instance initialized with the options specified in the AWSIoTCoreMQTTConfig.
The parameter usingSharedClient specifies whether to use a shared MQTT client, which will be initialized just once and can be reused by pipeline triggers and pipeline functions to connect to an external MQTT broker. This parameter is only effective when building an Application Service with multiple function-pipelines.

// BaseMqttConfig represents the common configuration for a MQTT client
type BaseMqttConfig struct {
    // BrokerAddress should be set to the complete broker address i.e. mqtts://mosquitto:8883/mybroker
    BrokerAddress string
    // ClientId to connect with the broker with.
    ClientId string
    // AutoReconnect indicated whether or not to retry connection if disconnected
    AutoReconnect bool
    // Topic that you wish to publish to
    Topic string
    // QoS for MQTT Connection
    QoS byte
    // Retain setting for MQTT Connection
    Retain bool
}

type AWSIoTCoreMQTTConfig struct {
    BaseMqttConfig
    // The name of the path in secret provider to retrieve your secrets, only take effect when SecretFormat=PEMBlock
    SecretPath string
    // SkipCertVerify
    SkipCertVerify bool
}

Factory method NewAWSIoTCoreMQTTSender will return an AWSIoTCoreMQTTSender instance. An AWSIoTCoreMQTTSender instance is featured with following methods:

Send: Receives either a string, []byte, or json.Marshaler acceptable types from the previous function in the pipeline and sends it to AWS IoT Core. If no previous function exists, the event that triggered the pipeline, marshaled to JSON, will be used. While persistOnError is true and StoreAndForward is enabled, the data will be stored for a later retry if any error occurs. See Store and Forward for more details.
SendResponse: This function should only be used right after the built-in function ExecuteCoreCommand. This function will send the result of core command execution to AWS IoT Core. The data sent to AWS IoT Core will contain HTTP status code and the execution response of Core Command. An example is shown as below:
```
{
  "Status": 200,
  "ResponseBody": ""
}
```

Note

Following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

clientkey - client private key in PEM format
clientcert - client certificate in PEM format

For further information on configuration, please refer to AWS IoT Core Example.

Azure IoT Hub

Factory Method Description

NewAzureIoTHubMQTTSender(iotHubConfig AzureIOTHubMQTTConfig, persistOnError bool, usingSharedClient bool) This function returns an AzureIoTHubMQTTSender instance initialized with the options specified in the AzureIOTHubMQTTConfig.
The parameter usingSharedClient specifies whether to use a shared MQTT client, which will be initialized just once and can be reused by pipeline triggers and pipeline functions to connect to an external MQTT broker. This parameter is only effective when building an Application Service with multiple function-pipelines.

// BaseMqttConfig represents the common configuration for a MQTT client
type BaseMqttConfig struct {
    // BrokerAddress should be set to the complete broker address i.e. mqtts://mosquitto:8883/mybroker
    BrokerAddress string
    // ClientId to connect with the broker with.
    ClientId string
    // AutoReconnect indicated whether or not to retry connection if disconnected
    AutoReconnect bool
    // Topic that you wish to publish to
    Topic string
    // QoS for MQTT Connection
    QoS byte
    // Retain setting for MQTT Connection
    Retain bool
}

type AzureIoTHubMQTTConfig struct {
    // Common Mqtt Config as embedded field
    BaseMqttConfig
    // The name of the path in secret provider to retrieve your secrets, only take effect when IotHubSecretFormat=PEMBlock
    SecretPath string
    // AuthMode indicates what authentication mode to connect to Azure IOT Hub when running application service. Only clientcert is supported.
    AuthMode string
    // SkipCertVerify
    SkipCertVerify bool
}

Factory method NewAzureIoTHubMQTTSender will return an AzureIoTHubMQTTSender instance. An AzureIoTHubMQTTSender instance is featured with following methods:

Send: Receives either a string, []byte, or json.Marshaler acceptable types from the previous function in the pipeline and sends it to Azure IoT Hub. If no previous function exists, the event that triggered the pipeline, marshaled to JSON, will be used. While persistOnError is true and StoreAndForward is enabled, the data will be stored for a later retry if any error occurs. See Store and Forward for more details.
SendDirectMethodResponse: This function should only be used right after the built-in function ExecuteCoreCommand. This function will send the result of core command execution to Azure IoT Hub as the response of Direct Method invocation.

Note

Following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

clientkey - client private key in PEM format
clientcert - client certificate in PEM format

For further information on configuration, please refer to Azure IoT Hub Example.

IBM

Note

IBM plans to sunset the Watson IoT Platform service on IBM Cloud effective December 1st, 2023 without a direct replacement. As of Edge Xpert v3.0, sending data to IBM Watson IoT Platform is no longer supported.

Factory Method	Description
NewIBMWatsonMQTTSender(mqttConfig IBMWatsonMQTTConfig, persistOnError bool)	This function returns an `IBMWatsonMQTTSender` instance initialized with the options specified in the `IBMWatsonMQTTConfig`.

// BaseMqttConfig represents the common configuration for a MQTT client
type BaseMqttConfig struct {
    // BrokerAddress should be set to the complete broker address i.e. mqtts://mosquitto:8883/mybroker
    BrokerAddress string
    // ClientId to connect with the broker with.
    ClientId string
    // AutoReconnect indicated whether or not to retry connection if disconnected
    AutoReconnect bool
    // Topic that you wish to publish to
    Topic string
    // QoS for MQTT Connection
    QoS byte
    // Retain setting for MQTT Connection
    Retain bool
}

type IBMWatsonMQTTConfig struct {
    BaseMqttConfig
    // The name of the path in secret provider to retrieve your secrets, only take effect when WatsonCertFormat=PEMBlock
    SecretPath string
    // SkipCertVerify
    SkipCertVerify bool
}

Factory method NewIBMWatsonMQTTSender will return an IBMWatsonMQTTSender instance. An IBMWatsonMQTTSender instance is featured with following method:

Send: Receives either a string, []byte, or json.Marshaler acceptable types from the previous function in the pipeline and sends it to IBM Watson IoT Platform. If no previous function exists, the event that triggered the pipeline, marshaled to JSON, will be used. While persistOnError is true and StoreAndForward is enabled, the data will be stored for a later retry if any error occurs. See Store and Forward for more details.

Note

Following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

username - Watson IoT Platform requires the username to be use-token-auth.
password - the unique authentication token that was generated when the device was registered with your Watson IoT Platform.

If the CA root to sign messaging.pem does not exist in your system's CA pool, you may need to provide the following secret in the specified SecretPath of secret provider with the content of messaging.pem:

cacert - content of messaging.pem in PEM format.

For further information on configuration, please refer to IBM Watson IoT Platform Example.

Info

Watson IoT Platform requires device publish the messages to the topic as defined in the format of iot-2/evt/_eventid/fmt/_formatstring, where _eventid specifies the event name that's shown in Watson IoT Platform, and _formatstring is the format of the event, such as JSON.

Regular MQTT

Factory Method Description

NewRegularMQTTSender(mqttConfig RegularMQTTConfig, persistOnError bool, usingSharedClient bool) This function returns a RegularMQTTSender instance initialized with the options specified in the RegularMQTTConfig.
The parameter usingSharedClient specifies whether to use a shared MQTT client, which will be initialized just once and can be reused by pipeline triggers and pipeline functions to connect to an external MQTT broker. This parameter is only effective when building an Application Service with multiple function-pipelines.

// BaseMqttConfig represents the common configuration for a MQTT client
type BaseMqttConfig struct {
    // BrokerAddress should be set to the complete broker address i.e. mqtts://mosquitto:8883/mybroker
    BrokerAddress string
    // ClientId to connect with the broker with.
    ClientId string
    // AutoReconnect indicated whether or not to retry connection if disconnected
    AutoReconnect bool
    // Topic that you wish to publish to
    Topic string
    // QoS for MQTT Connection
    QoS byte
    // Retain setting for MQTT Connection
    Retain bool
}

type RegularMQTTConfig struct {
    BaseMqttConfig
    // The name of the path in secret provider to retrieve secrets
    SecretPath string
    // SkipCertVerify indicates whether to verify the server's certificate chain and host name
    SkipCertVerify bool
    // AuthMode indicates what to use when connecting to the broker. Options are "none", "cacert" , "usernamepassword", "clientcert".
    // If a CA Cert exists in the SecretPath then it will be used for all modes except "none".
    AuthMode string
}

Factory method NewRegularMQTTSender will return a RegularMQTTSender instance. A RegularMQTTSender instance is featured with following method:

Send: Receives either a string, []byte, or json.Marshaler acceptable types from the previous function in the pipeline and sends it to the specified MQTT broker. If no previous function exists, the event that triggered the pipeline, marshaled to JSON, will be used. While persistOnError is true and StoreAndForward is enabled, the data will be stored for a later retry if any error occurs. See Store and Forward for more details.

Note

AuthMode accepts usernamepassword, clientcert, cacert, or none. When AuthMode is usernamepassword, following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

username - specifies the username to authenticate with MQTT broker.
password - specifies the password to authenticate with MQTT broker.

When AuthMode is clientcert, following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

clientkey - specifies the client private key in PEM format.
clientcert - specifies the client certificate in PEM format.

When AuthMode is cacert, the following secret key is mandatory and must exist in the specified SecretPath of secret provider:

cacert - specifies the CA certificate in PEM format.

When AuthMode is none, no secret is required.

InfluxDB

Factory Method	Description
NewInfluxDBWriter(config InfluxDBWriterConfig, persistOnError bool)	This function returns an `influxDBWriter` instance initialized with the options specified in the `InfluxDBWriterConfig`.

type InfluxDBWriterConfig struct {
    // ServerURL specifies the InfluxDB Server address
    ServerURL      string
    // AuthMode specifies what authentication mode to write points to InfluxDB
    AuthMode       string
    // The name of the path in secret provider to retrieve secrets
    SecretPath     string
    // Org specifies the InfluxDB Organization
    Org            string
    // Bucket specifies the InfluxDB Bucket
    Bucket         string
    // Measurement specifies the InfluxDB measurement where EdgeX events shall write to
    Measurement    string
    // ValueType specifies what value type of events shall be written into InfluxDB
    // Valid values: float, integer, string, boolean
    ValueType      string
    // Precision specifies how much timestamp precision is retained in the InfluxDB points
    // Valid values: ns, us, ms, s
    Precision      time.Duration
    // SkipCertVerify specifies whether the Edge Xpert Application Service verifies the server's certificate chain and host name
    SkipCertVerify bool
    // StoreEventTags specifies whether the event tags are exported to InfluxDB
    StoreEventTags bool
    // StoreReadingTags specifies whether the reading tags are exported to InfluxDB
    StoreReadingTags bool
    // FieldKeyPattern specifies the pattern of influxdb field key used to store the reading value 
    FieldKeyPattern string
}

Factory method NewInfluxDBWriter will return an influxDBWriter instance. An influxDBWriter instance is featured with following method:

InfluxDBSyncWrite: Receives either an EdgeX event from the previous function in the pipeline and sends it to InfluxDB. If no previous function exists, the event that triggered the pipeline, marshaled to JSON, will be used. While persistOnError is true and StoreAndForward is enabled, the data will be stored for a later retry if any error occurs. See Store and Forward for more details.

Info

For InfluxDB 2.0, an organization is a workspace for a group of users. All dashboards, tasks, buckets, members, etc., belong to an organization. For InfluxDB 1.8.X, organization is not used, and users can simply ignore this setting. For more details about organizations, please refer to InfluxDB documentation.

Note

AuthMode only accepts token for token authentication or none for no authentication. When AuthMode is token, the following secret key is mandatory and must exist in the specified SecretPath of secret provider:

token - specifies the token value to authenticate with your influxdb server.

For further information on configuration, please refer to InfluxDB Pipeline Function Example.

Kafka

Factory Method	Description
NewKafkaSender(kafkaEndpoint KafkaEndpoint, secretConfig KafkaSecretsConfig, persistOnError bool)	This function returns a `kafkaSender` instance initialized with the options specified in the `KafkaSecretsConfig` and send data to the specified `KafkaEndpoint`.

type KafkaEndpoint struct {
    // ClientID for logging, debugging and auditing purposes
    ClientID string
    // The Kafka broker address
    Address string
    // The port number of the Kafka broker
    Port int
    // The Kafka topic to which the message is sent
    Topic string
    // The partition to which the message is sent
    Partition int32
}

type KafkaSecretsConfig struct {
    // The name of the path in secret provider to retrieve your secrets
    SecretPath string
    // AuthMode specifies what authentication mode to produce messages to Kafka broker
    AuthMode string
    // SkipCertVerify specifies whether the Edge Xpert Application Service verifies the server's certificate chain and host name
    SkipCertVerify bool
}

Factory method NewKafkaSender will return a kafkaSender instance. A kafkaSender instance is featured with following method:

KafkaSend: Receives either a string, []byte, or json.Marshaler acceptable types from the previous function in the pipeline and sends it to Kafka Broker. If no previous function exists, the event that triggered the pipeline, marshaled to JSON, will be used. While persistOnError is true and StoreAndForward is enabled, the data will be stored for a later retry if any error occurs. See Store and Forward for more details.

Note

AuthMode accepts clientcert for TLS authentication, sasl_ssl for Azure Event Hubs Shared Access Signature (SAS), or none for no authentication.

When AuthMode is clientcert, following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

clientkey - specifies the client private key in PEM format.
clientcert - specifies the client certificate in PEM format.
cacert - specifies the CA certificate in PEM format.

If your client private key is decrypted, you must provide the following secret in the specified SecretPath of secret provider:

decryptedpassword - specifies the password to decrypt clientkey secret. This is the pass phrase you provided during the client private key creation.

For further information on configuration, please refer to Kafka Example.

When AuthMode is sasl_ssl, following secret key is mandatory and must exist in the specified SecretPath of secret provider:

azureEventHubsConnectionString - specifies the Azure Event Hubs connection string.

For further information on configuration, please refer to Export to Azure Event Hubs Example.

HTTP

Factory Method	Description
NewHTTPSender(senderConfig HTTPSenderConfig)	This function returns an `HTTPSender` instance initialized with the options specified in the `HTTPSenderConfig`.

// HTTPSenderOptions contains all options available to the sender
type HTTPSenderOptions struct {
    // URL of destination
    URL string
    // MimeType to send to destination
    MimeType string
    // PersistOnError enables use of store & forward loop if true
    PersistOnError bool
    // HTTPHeaderName to use for passing configured secret
    HTTPHeaderName string
    // SecretPath to search for configured secret
    SecretPath string
    // SecretName for configured secret
    SecretName string
    // URLFormatter specifies custom formatting behavior to be applied to configured URL.
    // If nothing specified, default behavior is to attempt to replace placeholders in the
    // form '{some-context-key}' with the values found in the context storage.
    URLFormatter StringValuesFormatter
    // ContinueOnSendError allows execution of subsequent chained senders after errors if true
    ContinueOnSendError bool
    // ReturnInputData enables chaining multiple HTTP senders if true
    ReturnInputData bool
}

type HTTPSenderConfig struct {
    transforms.HTTPSenderOptions
    // HTTPMethod is the method of http request
    HTTPMethod string
    // AuthMode specifies what HTTP authentication mode to use, now supports "none", "header_secret", and
    // "oauth2_clientcredentials".  The default value will be none.
    AuthMode HTTPAuthMode
}

Factory method NewHTTPSender will return an HTTPSender instance. An HTTPSender instance is featured with following method:

HTTPSend: Send data from the previous function to the specified Endpoint via http requests with supported method GET, POST, PUT, PATCH, or DELETE and various authentication mechanism. If no previous function exists, the event that triggered the pipeline will be used. Passing an empty string to the mimetype method will default to application/json. While persistOnError is true and StoreAndForward is enabled, the data will be stored for a later retry if any error occurs. See Store and Forward for more details.

Note

AuthMode accepts usernamepassword, clientcert, cacert, or none. When AuthMode is usernamepassword, following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

username - specifies the username to authenticate with MQTT broker.
password - specifies the password to authenticate with MQTT broker.

When AuthMode is clientcert, following secret keys are mandatory and must exist in the specified SecretPath of secret provider:

clientkey - specifies the client private key in PEM format.
clientcert - specifies the client certificate in PEM format.

When AuthMode is cacert, the following secret key is mandatory and must exist in the specified SecretPath of secret provider:

cacert - specifies the CA certificate in PEM format.

When AuthMode is none, no secret is required.
For further information on configuration, please refer to HTTP Example.

Core Command

type CoreCommandExecutor struct {
    DeviceNameJSONPath    string
    CommandNameJSONPath   string
    BodyJSONPath          string
    RequestMethodJSONPath string
    DefaultRequestMethod  string
    PushEvent             string
    ReturnEvent           string
    ContinueOnError       bool
}

CoreCommandExecutor parses the incoming JSON data to determine the core command, then issues it through the CommandClient. When configuring functions pipeline, an export function can be set right after ExecuteCoreCommand to send the command execution result. The implementation of CoreCommandExecutor uses GJSON library to retrieve values from a JSON document through GJSON path syntax.

Info

To construct CoreCommandExecutor, following parameters are required:

DeviceNameJSONPath - specifies the GJSON path to get Device Name from the incoming JSON document.
CommandNameJSONPath - specifies the GJSON path to get Command Name from the incoming JSON document.
BodyJSONPath - specifies the JSON path to get Request Body from the incoming JSON document.
RequestMethodJSONPath - specifies the JSON path to get Request Method from the incoming JSON document. Valid values: GET, SET, PUT.
DefaultRequestMethod - specifies the default Request Method used to issue core commands. This parameter is effective only when the incoming JSON document does not contain request method information.
PushEvent - specifies yes so that a successful GET will result in an event being pushed to the EdgeX system. Specify no to not push event to the EdgeX system.
ReturnEvent - specifies no so that a successful GET will not return an Event. Specify yes to return the event.
ContinueOnError - specifies true to have the functions pipeline continue after an error occurs so next export function can still be executed.

The default TargetType for data flowing into the function's pipeline is an Edge Xpert event. To receive JSON data, set the TargetType of SDK instance to []byte:

sdk := &appsdk.AppFunctionsSDK{
  ServiceKey: serviceKey, 
  TargetType: common.NewAtomicTargetType(&[]byte{}),
}

Execute - Parses a JSON document to determine a core command, issue it, and return the execution result. If the command is successful, the pipeline function returns the response body received from the Core Command service. If the command fails, an error is returned.

Conversion

Factory Method	Description
NewConversion()	This function returns a Conversion instance that is used to access the conversion functions.

ConvertBoolToIntReading - converts readings from Bool value type to Int8 value type. For a true Bool reading, this function coverts its value to 1 in Int8 type. For a false Bool reading, this functions coverts its value to 0 in Int8 type. This function will return an error and stop the pipeline if a non-Edgex event is received or if any error occurs during conversion.

ConvertBoolToFloatReading - converts readings from Bool value type to Float32 value type. For a true Bool reading, this function converts its value to 1 in Float32 type. For a false Bool reading, this function converts its value to 0 in Float32 type. This function will return an error and stop the pipeline if a non-Edgex event is received or if any error occurs during conversion.

ConvertIntToFloatReading - converts readings from Int8, Int16, Int32, Int64, Uint8, Uint16, Uint32, or Uint64 value type to Float64 value type in e-notation format. This function will return an error and stop the pipeline if a non-Edgex event is received or if any error occurs during conversion.

ConvertFloatToIntReading - converts readings from Float32, or Float64 value type to Int64 value type. This function will return an error and stop the pipeline if a non-Edgex event is received or if any error occurs during conversion.

Filter

Factory Method	Description
NewFilter()	This function returns a `Filter` instance.

FilterByValueMaxMin - removes readings with values outside the maximum and minimum limits defined in the device profile. This function only filters values when Maximum and Minimum are defined in the device profile. If only Maximum is defined, readings will only be compared against the defined maximum value. If only Minimum is defined, readings will only be compared against the defined minimum value. If both are defined, only readings with numeric values that fall between the set maximum and minimum values will be allowed to pass on to the next pipeline function. Please note that the use of this function does not need any parameters. This function will return an error and stop the pipeline if a non-Edgex event is received or if no data is received.

Note

FilterByValueMaxMin function checks the maximum and minimum limits defined in the device profile for each reading.

ScriptTransform

Factory Method	Description
NewScriptTransform(transformScript string)	This function returns a `ScriptTransform` instance initialized with `transformScript`, which will be the target Javascript codes to be executed when `Transform` method is invoked.

Transform - executes Javascript codes as provided during the ScriptTransform instance initialization. When preparing for transformScript, users can expect an embedded javascript object inputObject available for representing the data as passed into the Transform function, and the execution result must be returned if the result needs to be passed to next function in the pipeline. This function will return an error and stop the pipeline if any error occurs when running specified script or if no data is received. Please refer to Export to HTTP endpoints for the sample usage of JavascriptTransform.

SetContextVariable

Factory Method	Description
NewContextVariableSetter(variableName, valueJsonPath string, continueOnError bool)	This function returns a `ContextVariableSetter` instance.

This function can be used to extract value from the data passed into this function and store it in the context for access within the pipeline execution.

The name of context variable will be the value of parameter variableName, and the value of context variable will be the value extracted from the specified JsonPath of data as passed into the function. Set continueOnError to true if users would like the pipeline to continue even when there is error during SetContextVariable. When continueOnError is true and error occurs during SetContextVariable, the function will return incoming data rather than error. Please note that if incoming data does not contain specified JsonPath, the context variable will be set to empty string "".

Variable names are treated as case-insensitive and the following variable names are reserved:

devicename
profilename
sourcename
receivedtopic

Note

The Edge Xpert exclusive export functions (MQTT and HTTP) replace placeholders of the form {context-key-name} in the topic/url with the value found in the context at context-key-name. Key matching is case-insensitive.

ConvertDDATAToEvent

Factory Method	Description
NewSparkplugConverter()	This function returns a `SparkplugConverter` instance.

This function can be used to convert Saprkplug DDATA messages to Edgex events.

Note

The MQTT topic from which the Saprkplug DDATA message comes must conform to the following specification:

<namespace>/<group_id>/DDATA/<edge_node_id>/<device_id>

TransformToParquet

Factory Method	Description
NewParquetTransformer()	This function returns a `ParquetTransformer` instance.

This function can be used to transform Edgex events to Parquet format.

Others

PrintDataToLog - Print the incoming data to log for testing purposes.