Connecter PLCnext Control à Apache Kafka via MQTT – Intégration Enterprise

Contexte technique

Kafka

Apache Kafka est un framework pour l'ingestion, le stockage, le traitement et la redistribution de données. Aujourd'hui, il est largement déployé dans les entreprises du monde entier. Le site officiel de Kafka offre plus d'informations sur son concept et son déploiement. L'une de ses fonctionnalités clés est le grand nombre de connecteurs existants vers d'autres applications et protocoles de communication comme MQTT.

MQTT

MQTT est un protocole de messagerie léger basé sur TCP, souvent utilisé pour la communication IoT grâce à sa robustesse et sa faible empreinte. Les détails sur le standard OASIS MQTT sont disponibles sur son site web.

Ici, vous trouverez un article du Makers Blog sur la compilation croisée de mosquitto pour PLCnext, une implémentation MQTT d'Eclipse. Alternativement, le PLCnext Store propose des applications MQTT prêtes à l'emploi.

Prérequis

• Client MQTT sur le PLCnext (voir section précédente pour les indications d'implémentation)
• Le contrôleur est connecté à un PC/VM
• Broker MQTT sur le PC/VM (ex: mosquitto)
• Instance Kafka sur le PC/VM (voir guide de démarrage rapide Kafka)

Configuration

L'image suivante montre une vue d'ensemble de la configuration que nous allons implémenter pour ingérer les données du contrôleur PLCnext vers Kafka. Bien qu'il soit possible d'utiliser le MQTT Proxy de Confluent pour leur version de Kafka (2), nous nous concentrerons sur la solution plus générique (1). Elle comprend un broker MQTT où le client se connecte et publie des messages, et un connecteur qui s'abonne à un topic du broker, traite les messages et les transmet à Kafka.

Création du connecteur

Dans ce tutoriel, notre connecteur est basé sur le dépôt GitHub evokly/kafka-connect-mqtt, sous licence MIT. D'abord, nous téléchargeons et extrayons le dépôt. Comme la dernière version du dépôt date de fin 2016, nous mettons à jour le fichier build.gradle en remplaçant les anciennes dépendances par leurs nouvelles versions :

ext { kafkaVersion = '2.6.0' }
...
dependencies {
  testCompile group: 'junit', name: 'junit', version: '4.13'
  compile "org.apache.kafka:connect-api:$kafkaVersion"
  compile 'org.eclipse.paho:org.eclipse.paho.client.mqttv3:1.2.5'
  compile 'org.bouncycastle:bcprov-jdk15on:1.67'
  compile 'org.bouncycastle:bcpkix-jdk15on:1.67'
  compile 'org.bouncycastle:bcpg-jdk15on:1.67'
  compile 'commons-io:commons-io:2.8.0'
  compile 'org.slf4j:slf4j-api:1.7.30'
  testCompile 'org.slf4j:slf4j-simple:1.7.30'
}

Dans cet exemple, nous enverrons des messages String simples à Kafka. Nous devons donc modifier la classe Java DumbProcessor.java dans le dossier /kafka-connect-mqtt-master/src/main/java/com/evokly/kafka/connect/mqtt, qui est le processeur de messages par défaut :

@Override
public SourceRecord[] getRecords(String kafkaTopic) {
  return new SourceRecord[]{
    new SourceRecord(
      null, // sourcePartition
      null, // sourceOffset
      kafkaTopic, // topic
      null, // partition
      null, // keySchema
      mTopic, // key
      null, // valueSchema
      mMessage.toString(), // value
      new Long(123L)
    )
  }; // long timestamp
}

Ensuite, nous construisons une archive Java (JAR) contenant les dépendances : ./gradlew clean jar. Nous copions le JAR de sortie kafka-connect-mqtt-1.1-SNAPSHOT.jar (trouvé dans /kafka-connect-mqtt-master/build/libs) vers le répertoire libs de Kafka.

Nous avons aussi besoin d'une copie de l'archive org.eclipse.paho.client.mqttv3-1.2.5.jar dans le répertoire libs de Kafka. Nous pouvons la télécharger ici.

De plus, nous devons créer un fichier de configuration pour le connecteur mqtt.properties dans le dossier config de Kafka. Le fichier contient :

name=mqtt
connector.class=com.evokly.kafka.connect.mqtt.MqttSourceConnector
tasks.max=1
# converters pour messages String simples sans schémas
key.converter = org.apache.kafka.connect.storage.StringConverter
value.converter = org.apache.kafka.connect.storage.StringConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
kafka.topic=test_in
mqtt.client_id=mqtt-kafka-123
mqtt.clean_session=true
mqtt.connection_timeout=30
mqtt.keep_alive_interval=60
mqtt.server_uris=tcp://172.17.0.1:1883
mqtt.topic=test/#
#if we want to use our own processor class
#message_processor_class=com.evokly.kafka.connect.mqtt.sample.OwnProcessor

Test local

Maintenant, nous pouvons tester notre connecteur localement. Allez dans le répertoire de Kafka et démarrez une instance ZooKeeper et Broker :

# Démarrer ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# Démarrer Kafka
bin/kafka-server-start.sh config/server.properties

# Démarrer un broker MQTT (ici un conteneur Docker mosquitto)
sudo docker run -d --name mosquitto -p 1883:1883 eclipse-mosquitto

# Démarrer le connecteur MQTT-Kafka
bin/connect-standalone.sh config/connect-standalone.properties config/mqtt.properties

# Démarrer un consommateur console Kafka
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_in --from-beginning --property print.value=true --property print.timestamp=true

# Publier un message MQTT
mosquitto_pub -h 172.17.0.1 -p 1883 -t test/1 -m test123

Le message apparaît dans le consommateur console.