Eventhub (#6)

pelikhan · web-flow · commit 8c3d07e2ed71 · 2023-03-02T23:05:58.000-08:00
* added event hub in arm

* correctly grabing connection string

* add event hub support
diff --git a/.prettierrc b/.prettierrc
@@ -1,5 +1,5 @@
 {
     "arrowParens": "avoid",
-    "semi": "es5",
+    "semi": false,
     "tabWidth": 4
 }
diff --git a/infra/azuredeploy.json b/infra/azuredeploy.json
@@ -42,30 +42,76 @@
         "appInsightsApiVersion": "2020-02-02",
         "webAppApiVersion": "2022-03-01",
         "keyVaultApiVersion": "2022-07-01",
+        "eventHubApiVersion": "2022-01-01-preview",
         "sku": "Free",
         "skuCode": "F1",
         "workerSize": "0",
         "workerSizeId": "0",
         "numberOfWorkers": "1",
         "webAppAlwaysOn": false,
         "webAppLinuxFxVersion": "NODE|18-lts",
+        "eventHubSku": "Basic",
+        "eventHubPartitionCount": 1,
+        "eventHubMessageRetentionInDays": 1,
         "tenantId": "[subscription().tenantId]",
         "hostingPlanName": "[concat(variables('prefix'), 'hosting')]",
         "webAppName": "[concat(variables('prefix'), 'webapp')]",
         "appInsightsName": "[concat(variables('prefix'), 'appinsights')]",
         "storageAccountName": "[concat(variables('prefix'), 'storage')]",
         "keyVaultName": "[concat(variables('prefix'), 'keys')]",
+        "eventHubNamespaceName": "[concat(variables('prefix'), 'ns')]",
+        "eventHubAuthorizationRuleName": "gateway",
+        "telemetryEventHubName": "telemetry",
         "serverFarmId": "[resourceId('Microsoft.Web/serverfarms', variables('hostingPlanName'))]",
         "keyVaultId": "[resourceId('Microsoft.KeyVault/vaults', variables('keyVaultName'))]",
         "storageAccountId": "[resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName'))]",
         "appInsightsId": "[resourceId('Microsoft.Insights/components', variables('appInsightsName'))]",
         "webAppId": "[resourceId('Microsoft.Web/sites', variables('webAppName'))]",
+        "eventHubNamespaceId": "[resourceId('Microsoft.EventHub/namespaces', variables('eventHubNamespaceName'))]",
+        "telemetryEventHubId": "[resourceId('Microsoft.EventHub/namespaces/eventhubs', variables('eventHubNamespaceName'), variables('telemetryEventHubName'))]",
+        "eventHubAuthorizationRuleId": "[resourceId('Microsoft.EventHub/namespaces/authorizationRules', variables('eventHubNamespaceName'), variables('eventHubAuthorizationRuleName'))]",
         "storageConnectionStringSecretName": "storageAccountConnectionString",
+        "eventHubConnectionStringSecretName": "eventHubAccountConnectionString",
         "adminToken": "[concat('admin:', parameters('adminPassword'))]",
         "passwordsSecretName": "passwords",
         "adminConnectionStringSecretName": "adminConnectionString"
     },
     "resources": [
+        //event hub
+        {
+            "type": "Microsoft.EventHub/namespaces",
+            "apiVersion": "[variables('eventHubApiVersion')]",
+            "name": "[variables('eventHubNamespaceName')]",
+            "location": "[variables('location')]",
+            "sku": {
+                "name": "[variables('eventHubSku')]",
+                "tier": "[variables('eventHubSku')]",
+                "capacity": 1
+            },
+            "properties": {
+                "isAutoInflateEnabled": false,
+                "maximumThroughputUnits": 0
+            }
+        },
+        {
+            "type": "Microsoft.EventHub/namespaces/authorizationRules",
+            "apiVersion": "[variables('eventHubApiVersion')]",
+            "name": "[concat(variables('eventHubNamespaceName'), '/', variables('eventHubAuthorizationRuleName'))]",
+            "properties": {
+                "rights": ["Send"]
+            },
+            "dependsOn": ["[variables('eventHubNamespaceId')]"]
+        },
+        {
+            "type": "Microsoft.EventHub/namespaces/eventhubs",
+            "apiVersion": "[variables('eventHubApiVersion')]",
+            "name": "[concat(variables('eventHubNamespaceName'), '/', variables('telemetryEventHubName'))]",
+            "properties": {
+                "messageRetentionInDays": "[variables('eventHubMessageRetentionInDays')]",
+                "partitionCount": "[variables('eventHubPartitionCount')]"
+            },
+            "dependsOn": ["[variables('eventHubNamespaceId')]"]
+        },
         // storage
         {
             "type": "Microsoft.Storage/storageAccounts",
@@ -168,9 +214,7 @@
             "type": "Microsoft.KeyVault/vaults/secrets",
             "apiVersion": "[variables('keyVaultApiVersion')]",
             "name": "[concat(variables('keyVaultName'), '/', variables('adminConnectionStringSecretName'))]",
-            "dependsOn": [
-                "[variables('keyVaultId')]"
-            ],
+            "dependsOn": ["[variables('keyVaultId')]"],
             "properties": {
                 "value": "[concat('WebAppName=', variables('webAppName') ,';AccountName=admin;AccountKey=', parameters('adminPassword'), ';EndPointSuffix=azurewebsites.net')]"
             }
@@ -187,6 +231,18 @@
                 "value": "[concat('DefaultEndpointsProtocol=https;AccountName=', variables('storageAccountName'), ';AccountKey=', listKeys(resourceId('Microsoft.Storage/storageAccounts', variables('storageAccountName')), variables('storageApiVersion')).keys[0].value,';EndpointSuffix=core.windows.net')]"
             }
         },
+        {
+            "type": "Microsoft.KeyVault/vaults/secrets",
+            "apiVersion": "[variables('keyVaultApiVersion')]",
+            "name": "[concat(variables('keyVaultName'), '/', variables('eventHubConnectionStringSecretName'))]",
+            "dependsOn": [
+                "[variables('keyVaultId')]",
+                "[variables('eventHubAuthorizationRuleId')]"
+            ],
+            "properties": {
+                "value": "[listKeys(variables('eventHubAuthorizationRuleId'), variables('eventHubApiVersion')).primaryConnectionString]"
+            }
+        },
         // web
         {
             "type": "Microsoft.Web/serverfarms",
@@ -220,7 +276,9 @@
             },
             "dependsOn": [
                 "[variables('serverFarmId')]",
-                "[variables('appInsightsId')]"
+                "[variables('appInsightsId')]",
+                "[variables('eventHubAuthorizationRuleId')]",
+                "[variables('telemetryEventHubId')]"
             ],
             "identity": {
                 "type": "SystemAssigned"
@@ -242,10 +300,18 @@
                             "name": "DEVS_KEY_VAULT_NAME",
                             "value": "[variables('keyVaultName')]"
                         },
+                        {
+                            "name": "DEVS_EVENT_HUB_NAME",
+                            "value": "[variables('eventHubNamespaceName')]"
+                        },
                         {
                             "name": "DEVS_STORAGE_CONNECTION_STRING_SECRET",
                             "value": "[variables('storageConnectionStringSecretName')]"
                         },
+                        {
+                            "name": "DEVS_EVENT_HUB_CONNECTION_STRING_SECRET",
+                            "value": "[variables('eventHubConnectionStringSecretName')]"
+                        },
                         {
                             "name": "DEVS_PASSWORDS_SECRET",
                             "value": "[variables('passwordsSecretName')]"
diff --git a/package.json b/package.json
@@ -22,6 +22,7 @@
   },
   "dependencies": {
     "@azure/data-tables": "^13.2.0",
+    "@azure/event-hubs": "^5.8.0",
     "@azure/identity": "^3.1.3",
     "@azure/keyvault-secrets": "^4.6.0",
     "@azure/storage-blob": "^12.11.0",
@@ -48,4 +49,4 @@
     "xmlbuilder2": "^3.0.2",
     "zx": "^7.2.0"
   }
-}
+}
diff --git a/src/apidevices.ts b/src/apidevices.ts
@@ -11,6 +11,7 @@ import { DeviceId, DeviceInfo, FromDeviceMessage } from "./schema"
 import { wsskConnString } from "./wssk"
 import { fullDeviceId, pubToDevice, untilFromDevice } from "./devutil"
 import { fwdSockConnSettings } from "./fwdsock"
+import { Telemetry } from "./telemetry"
 
 function displayName(info: DeviceInfo) {
     const devid = info.rowKey
@@ -269,7 +270,7 @@ async function execTelemetryQuery(
             ...q0,
         })
     ).map(r => {
-        const rr: storage.Telemetry = {
+        const rr: Telemetry = {
             ms: r.ms,
             brainId: r.brainId,
             sensorId: r.sensorId,
diff --git a/src/eventhub.ts b/src/eventhub.ts
@@ -0,0 +1,25 @@
+import { EventHubProducerClient } from "@azure/event-hubs"
+import { telemetrySinks } from "./telemetry"
+import { createSecretClient } from "./vault"
+
+export async function setup() {
+    const secrets = createSecretClient()
+    const connectionStringSecretName =
+        process.env["DEVS_EVENT_HUB_CONNECTION_STRING_SECRET"] ||
+        "eventHubAccountConnectionString"
+    const connStrSecret = await secrets.getSecret(connectionStringSecretName)
+    const connStr = connStrSecret.value
+    if (!connStr) throw new Error("event hub connection string is empty")
+
+    const eventHubName = process.env["DEVS_EVENT_HUB_NAME"]
+    if (!eventHubName) throw new Error("event hub name is empty")
+
+    const producer = new EventHubProducerClient(connStr, eventHubName)
+
+    telemetrySinks.push(async (part, entries) => {
+        // Prepare a batch of three events.
+        const batch = await producer.createBatch()
+        entries.forEach(entry => batch.tryAdd({ body: entry }))
+        await producer.sendBatch(batch)
+    })
+}
diff --git a/src/index.ts b/src/index.ts
@@ -6,9 +6,9 @@ import fastifyCors from "@fastify/cors"
 import websocketPlugin from "@fastify/websocket"
 import { throwStatus } from "./util"
 import fastifyStatic from "@fastify/static"
-import { readFile } from "fs/promises"
 
 import * as storage from "./storage"
+import * as eventhub from "./eventhub"
 import * as mq from "./mq"
 import { wsskInit } from "./wssk"
 import { fwdSockInit } from "./fwdsock"
@@ -18,7 +18,8 @@ import { generateOpenApiSpec } from "./swagger/openapi"
 
 async function initAuth(server: FastifyInstance) {
     const secrets = createSecretClient()
-    const passwordSecretName = process.env["DEVS_PASSWORDS_SECRET"] || "passwords"
+    const passwordSecretName =
+        process.env["DEVS_PASSWORDS_SECRET"] || "passwords"
     const passwordsSecret = await secrets.getSecret(passwordSecretName)
     if (!passwordsSecret.value) throw new Error("passwords is empty")
 
@@ -119,6 +120,7 @@ async function main() {
     })
 
     await storage.setup()
+    await eventhub.setup()
     await initAuth(server)
     await wsskInit(server)
     await fwdSockInit(server)
diff --git a/src/jdbr-binfmt.ts b/src/jdbr-binfmt.ts
@@ -1,4 +1,4 @@
-import { Telemetry } from "./storage"
+import { Telemetry } from "./telemetry"
 
 function readU32(s: Buffer, off: number) {
     return s.readUInt32LE(off)
diff --git a/src/storage.ts b/src/storage.ts
@@ -10,6 +10,7 @@ import { randomBytes } from "crypto"
 import { promisify } from "util"
 import { gunzip, gzip } from "zlib"
 import { pubToDevice } from "./devutil"
+import { Telemetry, telemetrySinks, TelemetrySource } from "./telemetry"
 import { DeviceId, DeviceInfo, DeviceStats, zeroDeviceStats } from "./schema"
 import { delay, throwStatus } from "./util"
 import { createSecretClient } from "./vault"
@@ -37,8 +38,7 @@ export function webSiteName() {
 export function selfUrl() {
     // use https://learn.microsoft.com/en-us/azure/app-service/reference-app-settings?tabs=kudu%2Cdotnet
     const hostname = process.env["WEBSITE_HOSTNAME"]
-    if (!hostname)
-        throw new Error("WEBSITE_HOSTNAME not configured")
+    if (!hostname) throw new Error("WEBSITE_HOSTNAME not configured")
     const protocol = /^(0\.|localhost)/i.test(hostname) ? "http" : "https"
     return `${protocol}://${hostname}`
 }
@@ -84,6 +84,8 @@ export async function setup() {
     await telemetryTable.createTable()
     await scriptsBlobs.createIfNotExists()
 
+    telemetrySinks.push(insertTelemetry)
+
     if (false) {
         await scriptsBlobs.createIfNotExists({ access: "blob" })
         await blobClient.setProperties({
@@ -471,22 +473,6 @@ export async function getScriptVersions(scr: ScriptHeader) {
     )
 }
 
-export interface TelemetrySource {
-    brainId: string // "161f314155de245b"
-    sensorId: string // "260f3e4155de245b"
-    srv: string // "temperature"
-    srvIdx: number // typically 0
-}
-
-export interface Telemetry extends TelemetrySource {
-    ms: number // milliseconds since epoch aka Date.now()
-    avg: number // 22.2 C, etc
-    min?: number
-    max?: number
-    nsampl?: number // number of samples
-    dur?: number // sampling time is from [ms-dur, ms]
-}
-
 export type TelemetryQuery = Partial<TelemetrySource> & {
     start?: number
     stop?: number
@@ -587,7 +573,7 @@ export async function queryTelemetry(part: string, query: TelemetryQuery) {
     return res
 }
 
-export async function insertTelemetry(part: string, entries: Telemetry[]) {
+async function insertTelemetry(part: string, entries: Telemetry[]) {
     let acc: TransactionAction[] = []
     for (const e of entries) {
         const actions = keysOfTelemetry(e).map(
diff --git a/src/telemetry.ts b/src/telemetry.ts
@@ -0,0 +1,26 @@
+export interface TelemetrySource {
+    brainId: string // "161f314155de245b"
+    sensorId: string // "260f3e4155de245b"
+    srv: string // "temperature"
+    srvIdx: number // typically 0
+}
+
+export interface Telemetry extends TelemetrySource {
+    ms: number // milliseconds since epoch aka Date.now()
+    avg: number // 22.2 C, etc
+    min?: number
+    max?: number
+    nsampl?: number // number of samples
+    dur?: number // sampling time is from [ms-dur, ms]
+}
+
+export type TelemetrySink = (
+    part: string,
+    entries: Telemetry[]
+) => Promise<void>
+
+export const telemetrySinks: TelemetrySink[] = []
+
+export async function insertTelemetry(part: string, entries: Telemetry[]) {
+    await Promise.all(telemetrySinks.map(sink => sink(part, entries)))
+}
diff --git a/src/wssk.ts b/src/wssk.ts
@@ -12,6 +12,7 @@ import {
 import { runInBg } from "./util"
 import { fullDeviceId, pubFromDevice, subToDevice } from "./devutil"
 import { parseJdbrMessage, toTelemetry } from "./jdbr-binfmt"
+import { insertTelemetry } from "./telemetry"
 
 const JD_AES_CCM_TAG_BYTES = 4
 const JD_AES_CCM_LENGTH_BYTES = 2
@@ -357,7 +358,7 @@ class ConnectedDevice {
                             Date.now() - 20
                         )
                         console.log(telemetry)
-                        await storage.insertTelemetry(this.id.partitionKey, telemetry)
+                        await insertTelemetry(this.id.partitionKey, telemetry)
                     } catch (e: any) {
                         this.log.error(`upload-bin: ${e.stack}`)
                     }
diff --git a/yarn.lock b/yarn.lock

Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`{`
`2`	`2`	`"arrowParens": "avoid",`
`3`		`- "semi": "es5",`
	`3`	`+ "semi": false,`
`4`	`4`	`"tabWidth": 4`
`5`	`5`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-import { Telemetry } from "./storage"`
	`1`	`+import { Telemetry } from "./telemetry"`
`2`	`2`
`3`	`3`	`function readU32(s: Buffer, off: number) {`
`4`	`4`	`return s.readUInt32LE(off)`