賢慧的雞塊

Description:

we facing the azure iot hub bug today,

after upgrade the azure iot hub gateway to GWV2 version, the dc-traffic-forwarder servcie can no longer retreive data from azure iot hub

the error message from the azure iot hub is:

1
2
3
4
5
6
7

panic: subscribe azure iot event failed, err: protocol error: received flow without next-incoming-id after session established

goroutine 13 [running]:
dc-traffic-forwarder/internal/iothub.(*IotHubBroker).Subscribe.func1.1()
        /go/src/dc-traffic-forwarder/internal/iothub/init.go:90 +0x1f9
created by dc-traffic-forwarder/internal/iothub.(*IotHubBroker).Subscribe.func1
        /go/src/dc-traffic-forwarder/internal/iothub/init.go:48 +0xfb

前言

正當我以為目前的架構可以滿足需求時，登登～需求變更～～～

上面的長官們希望 Worker 可以當做一個共用服務，讓其他團隊可以訂閱 Worker 解析後的設備訊息來做其他應用

這就代表我們不能將資料的順序在 Archiver 那邊處理，而是 Worker 發送出來的設備訊息就需要按照順序了

所以 Worker 不能採用競爭消費的方式爭搶訊息來處理，因為這會導致同一設備的訊息順序錯亂，如下圖

為了解決這個問題，我們需要做到兩件事情

1. 讓同一個 Worker 處理同一個 Device 的訊息
2. 需要一個 Load Balancer 的角色，在 Worker 增加或減少時重新分配設備道不同 Worker 上，且分配的過程中保證順序性

進行到現在的階段，團隊大家對於新版 worker 有了一定的共識，所以先將定案的整體架構畫了出來，之後再根據每個細項實作時再來修改，有最新進度會再更新 ~

最終架構：

經過上次討論結果，最後決定使用 delayed queue 的方式來去實作，所以我們要先驗證假設的條件成立

Worker 可以透過搶訊息方式取得不同設備上傳的資料

目前設備上傳資料的方式可以透過兩個渠道

1. rabbitmq via amqp
2. azure iot hub

在 rabbitmq 的部分，多個 consumer 訂閱相同 topic 時，已經確定他是可以保證兩個 consumer 取得的資料不同了 (透過競爭消費的方式)

但 azure iot hub 的部分，實測如果不手動設定 partition，不同 worker 是有可能收到一樣的訊息的，需要設定 azure iot hub 的 partition 數量與 worker instance 數量一致才能避免這個問題

考慮到想讓 worker 動態擴展越簡單越好，這邊決定新增一個微服務，訂閱 iot hub 的訊息後 forwarding 到 rabbitmq，讓兩個渠道最後的行為一致

承上一篇，在測試完 mongo 與 redis 後，這篇要來討論的是如何將 worker 做水平擴展

需求與限制：

1. 動態擴展：

   由於 iot 的使用情境很多變，上傳的資料量在一天的不同時候差異很大，希望能透過 k8s 內建的擴展機制，自動根據 cpu、memory、net worker traffic 擴展 worker instance 數量

2. 不處理重複資料：

   worker 應該透過某種分配機制，讓設備上傳的資料分配到有空的 worker 去做，而不是兩個 worker 拿到一樣的資料去處理造成性能浪費

3. 資料順序性：

   由於 archiver 計算 recording data 時，需要資料按照資料時間依序傳給 archiver 計算，所以 worker 產出的資料須確保資料順序性

問題拆解：

經過幾天的討論，我們整理出了幾個需要先確認的問題

1. 將 raw data 依照 device id 做 partition 存放在不同的 collection 裡面對於寫入 mongo 速度有多大提升
2. 將 device config 先存放到 redis，worker 改從 redis 找到 device config 不從 postgres 中拿出來放到 memory，這樣子速度有沒有問題
3. worker 水平擴展方式

舊版 Worker 架構：

worker 內架構可以簡單劃成這樣

主要會分為 data thread 與 non-data thread 這兩個處理路徑，data thread 會開啟 10 個，用來處理設備傳輸上來的 data message， non-data thread 會開一個，用來處理設備上傳的 config、上下線資料，以及其他微服務的通知訊息等等

可以發現由於 data thread 會一路將資料從解析、存 redis、存 mongo、發 mqtt 一條龍處理到底，所以當這過程中的某個環節卡住時，data thread 就會無法接收新的資料導致整個 thread 卡住，連帶導致 message queue 裡面的資料堆積太久被丟棄，所以才會掉資料

背景介紹：

目前做的專案是 iot 相關的專案，主要是公司會賣出許多設備，需要將設備產生的資料上雲做即時資料查看、歷史資料保存、警報觸發等功能

整個專案採微服務模式，透過 k8s 部署在公司賣出的一體機(主機電腦)，或是部署在 Azure 上

此篇文章會拿其中一部分的架構出來說明是如何將原本無法水平擴展的微服務一步一步調整成可以動態水平擴展的微服務

專案架構：

問題：

目前在設定 schema 時，希望可以在 edge label 的 property 中記錄下 source vertex label 與 target vertex label 中的 mapping property，來達到 user 在 data explore insert 資料時，自動建立 edge

ex : 有兩個 vertex label， user 和 blog，想要在他們之間建立一個 post 的關聯

所以就會想要在 post 這個 edge label 中新增兩個 property 來代表 user 和 blog 之間建立關聯的規則

可是這會遇到一個問題，在 janusgraph 中，edge label 的 label name 是不能重複的，所以假設今天有另外一組 vertex label 之間的關聯也要叫做 post ，edge label 的 source 跟 target 就有可能會不同

Console

1. 透過 console 連線到 graph server：

   1. remote.yaml

      1
      2
      3
      4
      5
      6
      7
      8
      hosts: [172.16.4.141] # graph server ip
      port: 8182 # graph server port
      serializer: {
          className: org.apache.tinkerpop.gremlin.driver.ser.GryoMessageSerializerV3d0,
          config: {
              serializeResultToString: true
          }
      }
      

      Read more »