Configuration

Configuration

The SparkPipe service is configured using a simple TOML file located at /etc/sparkpipe/config.toml

Default config.toml

# N3uron SparkPipe configuration file

# Logging configuration
[logging]
  # Log level: DEBUG, INFO, WARN, ERROR
  level  = "info"

  # Log format: JSON, TEXT
  format = "text"

# MQTT configuration
[mqtt]
  # Sparkpug host application ID
  host_id = "sparkpipe"

  # Sparkplug reordering timeout in seconds
  reorder_timeout = 5

# MQTT brokers (add as many as desired)
[[mqtt.brokers]]
  # Connection url: proto://host:port
  # Examples:
  #   Plain TCP     --> tcp://insecure.server.cloud:1883
  #   TLS encrypted --> ssl://secure.server.cloud:8883
  #   Websockets    --> wss://secure.server.cloud:8884
  server      = ""
  
  # Username/password for authentication (leave empty if unused)
  username    = ""
  password    = ""

  # SSL/TLS configuration
  tls_client_certificate     = ""    # Path to the TLS client certificate
  tls_client_certificate_key = ""    # Path to the TLS certificate key
  tls_certificate_authority  = ""    # Path to the TLS Certificate Authority
  tls_insecure_skip_verify   = false # Disable certificate validation (do not use in production) 

# Connectors configuration
[connectors]
  # Name of the specific connector to run.
  #   - kafka
  #   - stdout
  name = "kafka"

# Kafka
[connectors.kafka]
  # Bootstrap servers
  # Example:
  #   servers = [ 
  #     "broker01.kafka.cloud:9092",
  #     "broker02.kafka.cloud:9092",
  #     "broker03.kafka.cloud:9092"
  #   ]
  servers = []

  # SSL/TLS configuration
  tls_enabled                = true
  tls_client_certificate     = ""    # Path to the TLS client certificate
  tls_client_certificate_key = ""    # Path to the TLS certificate key
  tls_certificate_authority  = ""    # Path to the TLS Certificate Authority
  tls_insecure_skip_verify   = false # Disable certificate validation (do not use in production) 

  # Authentication mechanism (NO_AUTH, SASL_PLAIN, SASL_SCRAM, AWS_MSK_IAM)
  authentication = "NO_AUTH"

  # Username/password for SASL_PLAIN and SASL_SCRAM (leave empty if unused)
  username = ""
  password = ""

  # SHA algorithm used in SASL_SCRAM authentication (SHA256 or SHA512)
  algorithm = "SHA512"

  # Topic to publish messages
  topic = "uns-data"

  # Message compression (NONE, GZIP, SNAPPY, LZ4, ZSTD)
  compression = "snappy"

# Stdout
[connectors.stdout]
  # Regex pattern to include metrics
  metric_name_filter = ".*"

Usage

Step 1: Open the configuration file:

sudo nano /etc/sparkpipe/config.toml

Step 2: Add or modify the default settings as desired and save changes (Ctrl+X)

Step 3: After configuring SparkPipe, run the following commands to enable and start the service:

sudo systemctl enable sparkpipe.service
sudo systemctl start sparkpipe.service

Step 4: Verify the service is up and running:

sudo systemctl status sparkpipe.service

Step 5: View the logs produced by the service:

sudo journalctl -u sparkpipe.service -f

Examples

Note:

Configuration examples for each module are provided in their respective sections in Connectors.

Multiple MQTT brokers

The following example shows how to configure SparkPipe to use multiple MQTT brokers. Add as many [[mqtt.brokers]] sections as required.

## N3uron SparkPipe configuration file

# Logging configuration
[logging]
  # Log level: DEBUG, INFO, WARN, ERROR
  level  = "info"

  # Log format: JSON, TEXT
  format = "json"

# MQTT configuration
[mqtt]
  # Sparkpug host application ID
  host_id = "sparkpipe"

  # Sparkplug reordering timeout in seconds
  reorder_timeout = 5

# MQTT brokers
[[mqtt.brokers]]
  # HiveMQ public broker
  server = "ssl://broker.hivemq.com:8883"

[[mqtt.brokers]]
  # Mosquitto public broker
  server                   = "ssl://test.mosquitto.org:8883"
  tls_insecure_skip_verify = true

# Connectors
...

AWS IoT Core with mTLS

This is an example configuration for AWS IoT Core using TLS certificates for authentication. See the official IoT Core documentation on creating and setting up a new connection.

## N3uron SparkPipe configuration file

# Logging configuration
[logging]
  # Log level: DEBUG, INFO, WARN, ERROR
  level  = "info"

  # Log format: JSON, TEXT
  format = "json"

# MQTT configuration
[mqtt]
  # Sparkpug host application ID
  host_id = "sparkpipe"

  # Sparkplug reordering timeout in seconds
  reorder_timeout = 5

# MQTT brokers
[[mqtt.brokers]]
  # AWS IoT Core
  server                     = "ssl://e7paxflom6n2q-ats.iot.us-east-1.amazonaws.com:8883"
  tls_certificate            = "/etc/sparkpipe/certs/703142e50c45a12ac5720aaae40347d56045aa433d71aae6dc5b39d572428f02-certificate.pem.crt"
  tls_certificate_key        = "/etc/sparkpipe/certs/703142e50c45a12ac5720aaae40347d56045aa433d71aae6dc5b39d572428f02-private.pem.key"
  tls_certificate_authority  = "/etc/sparkpipe/certs/AmazonRootCA1.pem"
  
  # Connectors
  ...