Ubuntu EdgeX 실습(3)

앞선 실습에서는 USB Camera Device Service를 이용하여 카메라 데이터를 수집하고, 이미지를 배열 형태로 변환하여 EdgeX Core Data에 저장하는 과정을 진행했다.

이번에는 EdgeX 내부에 저장된 데이터를 외부 시스템으로 전달하는 방법에 대해 알아보았다.

---

1. Application Service란?

EdgeX Foundry에서 Application Service는 EdgeX에서 수집·처리한 데이터를 외부 시스템에서 사용할 수 있도록 전달하는 역할을 한다.

EdgeX에서는 다양한 Device Service를 통해 IoT 장치의 데이터를 수집할 수 있는데, 수집된 데이터가 EdgeX 내부에만 머무르는 것이 아니라 다른 서버나 애플리케이션으로 전달되어야 하는 경우가 있다.

이때 Application Service를 사용할 수 있다.

이번 실습에서는 다음 두 가지 Application Service를 사용했다.

• asc-http: HTTP 방식으로 데이터 Export
• asc-mqtt: MQTT 방식으로 데이터 Export

그리고 두 Application Service가 EdgeX 내부의 데이터를 전달받을 수 있도록 EdgeX Message Bus Trigger를 사용했다.

---

2. Message Bus란?

이번 실습에서 중요한 개념 중 하나가 Message Bus이다.

Message Bus는 EdgeX 내부의 여러 서비스 사이에서 데이터를 전달하는 역할을 한다.

서비스끼리 직접 연결하는 방식이 아니라 중간에 Message Bus를 두고 데이터를 전달하기 때문에 각각의 서비스가 서로 직접 통신하지 않아도 된다.

기본적인 구조는 다음과 같다.

데이터를 생성하는 서비스
        ↓
    Message Bus
        ↓
데이터를 사용하는 서비스

이번 실습에서는 MQTT를 Message Bus로 사용하기 위해 mqtt-bus 서비스를 실행했다.

EdgeX에서는 기본적으로 Redis를 Message Bus로 사용할 수 있으며, 설정에 따라 MQTT나 ZeroMQ 등을 사용할 수도 있다.

이번 실습에서는 MQTT를 사용하기 때문에 다음과 같은 구조가 된다.

Device Service
      ↓
 Core Service
      ↓
 MQTT Message Bus
      ↓
Application Service

---

3. 실습 환경

ds-usb-camera

USB Camera를 EdgeX에서 사용할 수 있도록 관리하는 Device Service이다.

ds-rest

REST API를 이용하여 EdgeX에 데이터를 전달할 수 있도록 한다.

이번 실습에서는 이미지 배열 데이터를 ds-rest를 통해 EdgeX로 전달했다.

asc-http

EdgeX 내부의 데이터를 HTTP 방식으로 외부 서버에 전달한다.

asc-mqtt

EdgeX 내부의 데이터를 MQTT 방식으로 외부 MQTT Broker에 전달한다.

mqtt-bus

EdgeX의 Internal Message Bus에서 MQTT를 사용할 수 있도록 하는 서비스이다.

---

4. Compose Builder 이동

먼저 EdgeX Compose Builder 디렉토리로 이동했다.

cd ~/edgex-compose/compose-builder

이후 Application Service의 환경 설정 파일을 수정했다.

HTTP Export는

asc-http-export.env

MQTT Export는

asc-mqtt-export.env

파일을 사용한다.

---

5. HTTP Export 설정

먼저 HTTP Export 설정을 변경했다.

EDGEX_PROFILE=http-export
WRITABLE_PIPELINE_FUNCTIONS_HTTPEXPORT_PARAMETERS_URL=http://203.237.142.21:8080
WRITABLE_LOGLEVEL=INFO
TRIGGER_TYPE=edgex-messagebus

여기서 중요한 설정이 TRIGGER_TYPE이다.

TRIGGER_TYPE=edgex-messagebus

이번 실습에서는 EdgeX Message Bus를 통해 데이터가 전달되었을 때 HTTP Export가 실행되도록 설정했다.

처음에는 TRIGGER_TYPE=http로 되어 있었는데, 이 상태에서는 이미지 데이터를 EdgeX에 Push해도 HTTP Export가 실행되지 않았다.

확인 후 다음과 같이 수정했다.

TRIGGER_TYPE=http

에서

TRIGGER_TYPE=edgex-messagebus

로 변경했다.

이후 Export 서비스를 다시 실행하자 HTTP Export가 Message Bus를 정상적으로 구독하는 것을 확인할 수 있었다.

---

6. HTTP Export에서 localhost를 사용할 수 없는 이유

HTTP Export의 Endpoint를 설정할 때 주의할 점이 있다.

Docker 컨테이너 환경에서는 localhost가 호스트 컴퓨터를 의미하지 않는다.

예를 들어 Ubuntu에서 Node.js 서버가 다음 주소에서 실행되고 있다고 하자.

localhost:8080

Application Service가 Docker 컨테이너 안에서 실행되는 경우 컨테이너에서 localhost:8080을 요청하면 Ubuntu 호스트의 8080 포트가 아니라 해당 컨테이너의 8080 포트를 바라보게 된다.

따라서 HTTP Export에서는 호스트에서 실행 중인 Node.js 서버의 주소를 사용했다.

http://203.237.142.21:8080

그리고 Node.js 서버는 외부에서 접근할 수 있도록 다음과 같이 실행했다.

server.listen(8080, '0.0.0.0', () => {
    console.log('Server listening on port 8080');
});

---

7. MQTT Export 설정

다음으로 MQTT Export 설정을 변경했다.

EDGEX_PROFILE=mqtt-export
WRITABLE_PIPELINE_FUNCTIONS_MQTTEXPORT_PARAMETERS_BROKERADDRESS=tcp://edgex-mqtt-broker:1883
WRITABLE_PIPELINE_FUNCTIONS_MQTTEXPORT_PARAMETERS_TOPIC=image-array
WRITABLE_LOGLEVEL=INFO

MQTT Broker는 EdgeX에서 실행 중인

edgex-mqtt-broker

를 사용했다.

그리고 MQTT Export 결과를 확인하기 쉽도록 Topic을 다음과 같이 설정했다.

image-array

여기서 edgex/events/#와 image-array는 서로 다른 용도이다.

edgex/events/#는 Application Service가 EdgeX 내부 Message Bus에서 이벤트를 받기 위해 Subscribe하는 Topic이고,

image-array는 MQTT Export가 외부로 데이터를 Publish하는 Topic이다.

즉,

edgex/events/#

↓

Application Service

↓

image-array

와 같은 흐름으로 생각할 수 있다.

---

8. Export 서비스 실행

설정을 완료한 후 다음 명령어를 사용하여 필요한 서비스를 실행했다.

sudo make run no-secty ds-usb-camera ds-rest asc-mqtt asc-http mqtt-bus

실행 후 필요한 서비스들이 정상적으로 실행되는 것을 확인했다.

특히 이번 실습에서 중요한 서비스는 다음 세 가지이다.

mqtt-bus
asc-http
asc-mqtt

mqtt-bus가 Message Bus 역할을 하고,

asc-http와 asc-mqtt가 Message Bus에서 데이터를 받아 각각 HTTP와 MQTT 방식으로 외부에 Export한다.

---

9. MQTT Export 확인을 위한 WebSocket Client

MQTT Export 결과를 확인하기 위해 HiveMQ WebSocket Client를 사용했다.

이번에는 EdgeX MQTT Broker에 WebSocket으로 접속할 수 있도록 설정하고 다음과 같이 연결했다.

Host : localhost
Port : 9001
SSL  : OFF

정상적으로 연결되면 Connected 상태가 표시된다.

위와 같이 WebSocket Client에서 MQTT Broker에 정상적으로 연결된 것을 확인했다.

---

10. MQTT Topic 구독

MQTT Export에서 설정한 Topic은 다음과 같다.

image-array

따라서 HiveMQ WebSocket Client의 Subscription에도 동일한 Topic을 추가했다.

이제 image-array Topic으로 메시지가 Publish되면 WebSocket Client에서 확인할 수 있다.

---

11. 이미지 데이터를 EdgeX로 Push

앞선 실습에서 사용했던 이미지 → 배열 변환 프로그램을 다시 실행했다.

cd ~/edgex-image-to-array
python image_to_array.py

Python 프로그램에서는 이미지를 읽고 NumPy 배열로 변환한다.

image = Image.open(filepath)
image_array = np.array(image)

그리고 배열을 JSON으로 전달할 수 있도록 Python List로 변환한다.

data = {
    "image": image_array.tolist()
}

이렇게 만들어진 데이터는 Device REST의 다음 Endpoint로 전달된다.

http://localhost:59986/api/v2/resource/sample-json/json

실행 결과는 다음과 같았다.

basic.png
status: 200
response:

HTTP Status Code가 200이므로 Device REST에 데이터가 정상적으로 전달된 것을 확인할 수 있었다.

---

12. EdgeX 내부에서 데이터가 전달되는 과정

여기서 중요한 것은 Device REST에 데이터가 전달되었다고 해서 바로 MQTT Export가 실행되는 것이 아니라는 점이다.

EdgeX 내부에서는 다음과 같은 과정이 이루어진다.

image_to_array.py
        ↓
    Device REST
        ↓
   Core Services
        ↓
Internal Message Bus
        ↓
 edgex/events/#
        ↓
Application Service

Device REST를 통해 데이터가 들어오면 Core Service가 데이터를 EdgeX의 내부 데이터 구조로 처리하고 Message Bus에 Publish한다.

그리고 Application Service는 Message Bus를 Subscribe하고 있다.

따라서 새로운 이벤트가 들어오면 Application Service가 해당 데이터를 받아 Pipeline을 실행한다.

---

13. edgex-messagebus Trigger

Application Service가 언제 실행될 것인지를 결정하는 것이 Trigger이다.

Trigger는 쉽게 말하면 Application Service에 "언제 데이터를 받아서 작업을 시작할 것인가?"를 알려주는 설정이라고 생각할 수 있다.

이번 실습에서는

edgex-messagebus

Trigger를 사용했다.

동작 방식은 다음과 같다.

Message Bus에 데이터 Publish
          ↓
Application Service가 Subscribe
          ↓
데이터 수신
          ↓
Pipeline 실행
          ↓
Export

Application Service는 다음 Topic을 Subscribe한다.

edgex/events/#

따라서 해당 Topic에 새로운 데이터가 들어오면 Application Service가 동작하게 된다.

---

14. Pipeline이란?

Application Service에서 또 하나 중요한 개념이 Pipeline이다.

Pipeline은 데이터를 처리하기 위한 여러 함수를 순서대로 연결한 구조라고 생각할 수 있다.

기본적인 형태는 다음과 같다.

Trigger
  ↓
데이터 수신
  ↓
Pipeline
  ↓
데이터 처리
  ↓
Export

Pipeline에서는 필요에 따라 데이터를 필터링하거나 변환할 수 있다.

예를 들어 이미지 데이터를 처리한다면,

EdgeX Event
      ↓
이미지 데이터 추출
      ↓
필요한 데이터 필터링
      ↓
이미지 전처리
      ↓
Object Detection
      ↓
결과 생성
      ↓
HTTP / MQTT Export

와 같은 형태로 확장할 수 있다.

따라서 Application Service는 단순한 데이터 전달 기능뿐만 아니라 EdgeX에서 수집한 데이터를 외부 시스템에서 활용하기 위한 데이터 처리 과정에도 사용할 수 있다.

---

15. MQTT Export 결과

이미지 데이터를 다시 Push하면 Message Bus를 통해 Application Service로 데이터가 전달되고, asc-mqtt가 MQTT Export를 수행한다.

그리고 지정한 Topic인

image-array

로 데이터가 Publish된다.

실제 MQTT 메시지를 확인하면 EdgeX의 데이터 구조가 포함되어 있는 것을 확인할 수 있다.

deviceName   : sample-json
profileName  : sample-json
sourceName   : json
resourceName : json
valueType    : Object
objectValue  : {
    image : [...]
}

특히 objectValue 내부에 실제 이미지 배열이 포함되어 있다.

이미지는 픽셀 단위의 데이터로 표현되기 때문에 다음과 같은 값이 매우 많이 출력된다.

[255,255,255,0]
[255,255,255,0]
[255,255,255,0]
...

처음 터미널에서 [255,255,255,0] 같은 값이 엄청나게 많이 출력되었는데, 이것이 바로 이미지가 배열 데이터로 변환되어 Export되고 있다는 것을 보여주는 부분이었다.

---

16. HTTP Export 결과 확인

MQTT Export뿐만 아니라 HTTP Export도 확인했다.

데이터를 받기 위한 간단한 Node.js 서버를 실행했다.

const http = require('http');

const server = http.createServer((req, res) => {
    let body = '';

    req.on('data', chunk => {
        body += chunk;
    });

    req.on('end', () => {
        console.log('----- Received Data -----');
        console.log(body);
        console.log('-------------------------');

        res.writeHead(200, { 'Content-Type': 'text/plain' });
        res.end('OK');
    });
});

server.listen(8080, '0.0.0.0', () => {
    console.log('Server listening on port 8080');
});

Node.js 서버를 실행한 후 다시 이미지 데이터를 Push했다.

cd ~/edgex-image-to-array
python image_to_array.py

그러자 Node.js 서버에서 다음과 같이 데이터가 출력되었다.

----- Received Data -----
...
[255,255,255,0]
[255,255,255,0]
[255,255,255,0]
...
-------------------------

이미지 배열의 데이터 양이 많기 때문에 터미널에 [255,255,255,0]과 같은 값들이 매우 많이 출력되었다.

이를 통해 HTTP Export도 정상적으로 동작하는 것을 확인했다.

---

17. HTTP Export에서 발생한 문제 해결

이번 실습에서 HTTP Export를 확인하는 과정에서 한 번 문제가 발생했다.

처음 HTTP Export 설정에는 다음과 같이 되어 있었다.

TRIGGER_TYPE=http

이 상태에서 이미지 데이터를 Push했지만 Node.js 서버에서는 Export 데이터가 출력되지 않았다.

원인을 확인한 결과 이번 실습에서는 EdgeX Message Bus를 Trigger로 사용하고 있었기 때문에 HTTP Export 역시 Message Bus Trigger를 사용하도록 변경해야 했다.

따라서 다음과 같이 수정했다.

TRIGGER_TYPE=edgex-messagebus

그리고 Export 서비스를 다시 실행했다.

이후 HTTP Export 로그에서 Message Bus Trigger가 선택되고,

edgex/events/#

Topic을 Subscribe하고 있는 것을 확인할 수 있었다.

다시 이미지 데이터를 Push하자 Node.js 서버에서 실제 이미지 배열 데이터가 출력되었다.

이 과정을 통해 Application Service에서 어떤 Trigger를 사용하는지가 실제 동작에 중요한 영향을 준다는 것을 확인했다.

---

마무리

이번 실습에서는 EdgeX의 "Application Service - Export" 기능을 직접 확인했다.

앞선 실습에서는 이미지를 배열 형태로 변환하여 EdgeX에 저장하는 과정을 진행했다면, 이번에는 그 데이터를 다시 외부 시스템으로 전달하는 과정을 실습했다.

전체적인 흐름은 다음과 같다.

이미지
 ↓
이미지 배열 변환
 ↓
Device REST
 ↓
EdgeX Core Services
 ↓
Internal Message Bus
 ↓
Application Service
 ↓
┌──────────────┬──────────────┐
↓                             ↓
HTTP Export                MQTT Export
↓                             ↓
Node.js                    MQTT Broker
Server                         ↓
                         image-array
                              ↓
                    HiveMQ WebSocket

특히 이번 실습에서는 edgex-messagebus Trigger를 사용하여 Message Bus에 데이터가 들어왔을 때 Application Service가 동작하도록 구성했다.

실제로 MQTT에서는 image-array Topic을 통해 이미지 배열 데이터가 전달되는 것을 확인했고, HTTP에서는 Node.js 서버에서 [255,255,255,0]과 같은 실제 이미지 픽셀 배열이 출력되는 것을 확인했다.

이를 통해 EdgeX 내부에서 관리되는 데이터를 다양한 외부 시스템으로 전달할 수 있다는 것을 확인할 수 있었다.

또한 Application Service의 Pipeline을 이용하면 단순히 데이터를 전달하는 것뿐만 아니라 필터링, 변환, 분석 등의 추가적인 처리 과정을 구성할 수 있다는 것도 알 수 있었다.

이번 실습에서 구성한 데이터 흐름은 이후 TensorFlow Serving을 이용한 Object Detection Pipeline으로 확장할 수 있다.

다음 실습에서는 EdgeX에서 전달된 이미지 데이터를 TensorFlow Serving과 연결하여 Object Detection을 수행하는 Pipeline을 구성해볼 예정이다.